# LangChain Framework

LangChain is a comprehensive framework for building production-ready applications and agents powered by Large Language Models (LLMs). It provides a standard interface for models, prompts, chains, agents, and integrations with 100+ model providers and data sources. The framework enables developers to build complex AI applications through simple composition using the LangChain Expression Language (LCEL), a declarative syntax that makes LLM applications modular, testable, and production-ready. The core package (`langchain-core` v1.1.1) contains the fundamental abstractions, while the main `langchain` package (v1.1.2) provides high-level agent implementations built on LangGraph. Partner integrations (OpenAI, Anthropic, Chroma, Ollama, xAI, DeepSeek, Perplexity, and 15+ others) provide seamless connectivity to external services.

At its core, LangChain implements the Runnable protocol - a universal interface that provides synchronous, asynchronous, batch, and streaming operations for all components. This abstraction allows developers to chain together prompts, models, retrievers, tools, and output parsers using intuitive operators (`|` for sequences, `{}` for parallel execution) while maintaining type safety through Pydantic models and automatic schema inference. The framework's ecosystem includes LangSmith for debugging and observability, LangGraph (integrated as a core dependency) for building stateful agents with human-in-the-loop capabilities and advanced orchestration, and a vast library of partner integrations for seamless model interoperability. LangChain agents are now built on top of LangGraph to provide durable execution, streaming, persistence, and more advanced workflow control.

## Basic Chain: Prompt + Model + Parser

Simple sequential chain combining prompt template, chat model, and output parser.

```python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Initialize model
model = ChatOpenAI(model="gpt-4", temperature=0.7)

# Create prompt template
prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")

# Chain components with | operator
chain = prompt | model | StrOutputParser()

# Invoke synchronously
result = chain.invoke({"topic": "programmers"})
print(result)  # "Why do programmers prefer dark mode? Because light attracts bugs!"

# Stream response
for chunk in chain.stream({"topic": "cats"}):
    print(chunk, end="", flush=True)

# Batch multiple inputs
results = chain.batch([
    {"topic": "dogs"},
    {"topic": "birds"}
])

# Async invocation
import asyncio
async_result = await chain.ainvoke({"topic": "fish"})
```

## Conversational Chain with Message History

Multi-turn conversation using MessagesPlaceholder for dynamic chat history.

```python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-3-5-sonnet-latest")

# Prompt with placeholder for chat history
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful AI assistant. Answer concisely."),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}")
])

chain = prompt | model | StrOutputParser()

# First turn
chat_history = []
response1 = chain.invoke({
    "input": "My name is Alice and I love Python programming",
    "chat_history": chat_history
})
print(response1)  # "Nice to meet you, Alice! Python is a great language..."

# Add to history
chat_history.extend([
    HumanMessage(content="My name is Alice and I love Python programming"),
    AIMessage(content=response1)
])

# Second turn - model remembers context
response2 = chain.invoke({
    "input": "What's my name and what do I like?",
    "chat_history": chat_history
})
print(response2)  # "Your name is Alice and you love Python programming."
```

## Structured Output with Pydantic Models

Extract structured data from LLM responses using Pydantic schemas.

```python
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# Define output structure
class Person(BaseModel):
    name: str = Field(description="Person's full name")
    age: int = Field(description="Person's age in years")
    occupation: str = Field(description="Person's job or profession")
    interests: list[str] = Field(description="List of hobbies or interests")

model = ChatOpenAI(model="gpt-4")
structured_model = model.with_structured_output(Person)

prompt = ChatPromptTemplate.from_template(
    "Extract person information from: {text}"
)

chain = prompt | structured_model

# Returns Pydantic model instance
person = chain.invoke({
    "text": "John Smith is a 34 year old software engineer who enjoys hiking, photography, and playing guitar."
})

print(person.name)  # "John Smith"
print(person.age)  # 34
print(person.interests)  # ["hiking", "photography", "playing guitar"]
print(person.model_dump_json())  # JSON serialization
```

## Tool Calling with @tool Decorator

Define functions that LLMs can invoke for external data access or computation.

```python
from typing import Annotated
from langchain_core.tools import tool
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, ToolMessage

# Define tools with docstrings
@tool
def get_weather(
    location: Annotated[str, "City name or address"],
    units: Annotated[str, "Temperature units: celsius or fahrenheit"] = "celsius"
) -> str:
    """Get current weather for a location."""
    # Simulate API call
    return f"Weather in {location}: 22°{units[0].upper()}, partly cloudy"

@tool
def calculator(expression: str) -> float:
    """Evaluate a mathematical expression safely."""
    return eval(expression)

# Bind tools to model
model = ChatAnthropic(model="claude-3-5-sonnet-latest")
model_with_tools = model.bind_tools([get_weather, calculator])

# Model generates tool calls
response = model_with_tools.invoke([
    HumanMessage(content="What's the weather in Paris and what's 15 * 23?")
])

# Execute tool calls
tool_results = []
for tool_call in response.tool_calls:
    if tool_call["name"] == "get_weather":
        result = get_weather.invoke(tool_call["args"])
    elif tool_call["name"] == "calculator":
        result = calculator.invoke(tool_call["args"])

    tool_results.append(ToolMessage(
        content=str(result),
        tool_call_id=tool_call["id"]
    ))

# Get final response with tool results
final_response = model_with_tools.invoke([
    HumanMessage(content="What's the weather in Paris and what's 15 * 23?"),
    response,
    *tool_results
])
print(final_response.content)  # "The weather in Paris is 22°C and partly cloudy. 15 * 23 equals 345."
```

## RAG: Retrieval-Augmented Generation

Retrieve relevant context from documents before generating responses.

```python
from operator import itemgetter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Setup: Load and index documents
documents = [
    "LangChain is a framework for developing LLM applications.",
    "It provides tools for prompt management, chains, and agents.",
    "LCEL (LangChain Expression Language) enables declarative composition."
]

text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20)
splits = text_splitter.create_documents(documents)

vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# RAG chain
prompt = ChatPromptTemplate.from_template("""
Answer the question based only on the following context:

{context}

Question: {question}
""")

# Method 1: Using RunnablePassthrough.assign
rag_chain = (
    RunnablePassthrough.assign(
        context=itemgetter("question") | retriever | (lambda docs: "\n\n".join(doc.page_content for doc in docs))
    )
    | prompt
    | ChatOpenAI(model="gpt-4")
    | StrOutputParser()
)

# Invoke with question
answer = rag_chain.invoke({"question": "What is LCEL?"})
print(answer)  # "LCEL (LangChain Expression Language) enables declarative composition..."

# Method 2: Using RunnableParallel for explicit control
rag_chain_parallel = RunnableParallel({
    "context": itemgetter("question") | retriever | (lambda docs: "\n\n".join(doc.page_content for doc in docs)),
    "question": itemgetter("question")
}) | prompt | ChatOpenAI(model="gpt-4") | StrOutputParser()
```

## Multi-Step Chain with Intermediate Assignments

Build complex workflows with intermediate processing steps.

```python
from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4")

# Step 1: Analyze query and generate search terms
query_analyzer_prompt = ChatPromptTemplate.from_template(
    "Extract 3 key search terms from: {question}"
)

# Step 2: Use search terms to fetch documents (simulated)
def fetch_documents(data):
    search_terms = data["search_terms"]
    # Simulate document retrieval
    return f"Document about: {search_terms}"

# Step 3: Generate final answer
answer_prompt = ChatPromptTemplate.from_template("""
Question: {question}
Search terms: {search_terms}
Documents: {documents}

Provide a detailed answer:
""")

# Chain with intermediate assignments
chain = (
    RunnablePassthrough.assign(
        search_terms=query_analyzer_prompt | model | StrOutputParser()
    )
    .assign(
        documents=fetch_documents
    )
    | answer_prompt
    | model
    | StrOutputParser()
)

result = chain.invoke({"question": "How does photosynthesis work?"})
# Intermediate values are accessible throughout the chain
```

## Parallel Execution for Multiple Operations

Execute independent operations concurrently using dictionary syntax.

```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4")

# Define multiple independent prompts
joke_prompt = ChatPromptTemplate.from_template("Tell a joke about {topic}")
poem_prompt = ChatPromptTemplate.from_template("Write a haiku about {topic}")
fact_prompt = ChatPromptTemplate.from_template("Share an interesting fact about {topic}")

# Parallel execution using dict
parallel_chain = {
    "joke": joke_prompt | model | StrOutputParser(),
    "poem": poem_prompt | model | StrOutputParser(),
    "fact": fact_prompt | model | StrOutputParser()
}

# All three operations run concurrently
results = parallel_chain.invoke({"topic": "ocean"})
print(results["joke"])  # "Why did the ocean break up with the pond?..."
print(results["poem"])  # "Waves crash on the shore\nDeep blue mystery calls out\nLife beneath the waves"
print(results["fact"])  # "The ocean covers 71% of Earth's surface..."

# Can be part of larger chains
full_chain = (
    parallel_chain
    | ChatPromptTemplate.from_template("Combine these into a creative story:\n\nJoke: {joke}\n\nPoem: {poem}\n\nFact: {fact}")
    | model
    | StrOutputParser()
)
```

## Conditional Branching with RunnableBranch

Route execution based on input conditions or intermediate results.

```python
from langchain_core.runnables import RunnableBranch, RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4")

# Define different prompts for different scenarios
casual_prompt = ChatPromptTemplate.from_template(
    "Give a casual, friendly answer: {question}"
)
technical_prompt = ChatPromptTemplate.from_template(
    "Give a detailed technical explanation: {question}"
)
simple_prompt = ChatPromptTemplate.from_template(
    "Explain to a 10-year-old: {question}"
)

# Routing logic
def classify_question(data):
    question = data["question"].lower()
    if any(word in question for word in ["technical", "explain", "how"]):
        return "technical"
    elif any(word in question for word in ["simple", "basic", "eli5"]):
        return "simple"
    else:
        return "casual"

# Branch based on classification
branch = RunnableBranch(
    (lambda x: classify_question(x) == "technical", technical_prompt | model | StrOutputParser()),
    (lambda x: classify_question(x) == "simple", simple_prompt | model | StrOutputParser()),
    casual_prompt | model | StrOutputParser()  # default
)

chain = RunnablePassthrough.assign(classification=classify_question) | branch

# Different inputs route to different prompts
result1 = chain.invoke({"question": "How does a computer work?"})  # Uses technical_prompt
result2 = chain.invoke({"question": "What's a simple explanation of gravity?"})  # Uses simple_prompt
result3 = chain.invoke({"question": "What's your favorite color?"})  # Uses casual_prompt
```

## Fallbacks and Error Handling

Handle failures gracefully with fallback models or retry logic.

```python
from langchain_core.runnables import RunnableWithFallbacks
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Primary and fallback models
primary_model = ChatOpenAI(model="gpt-4", timeout=5)
fallback_model = ChatAnthropic(model="claude-3-5-sonnet-latest")

prompt = ChatPromptTemplate.from_template("Explain {concept} in one paragraph")

# Chain with fallback
chain = (
    prompt
    | primary_model.with_fallbacks([fallback_model])
    | StrOutputParser()
)

try:
    # If GPT-4 fails (timeout, rate limit, etc), Claude is used automatically
    result = chain.invoke({"concept": "quantum computing"})
    print(result)
except Exception as e:
    print(f"All models failed: {e}")

# Retry with exponential backoff
from langchain_core.runnables import Retry

chain_with_retry = (
    prompt
    | primary_model.with_retry(
        retry_if_exception_type=(TimeoutError,),
        wait_exponential_jitter=True,
        stop_after_attempt=3
    )
    | StrOutputParser()
)
```

## Custom Functions as Runnables

Wrap arbitrary Python functions into the Runnable protocol.

```python
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# Synchronous function
def preprocess_input(data: dict) -> dict:
    """Clean and normalize input data."""
    return {
        "text": data["text"].strip().lower(),
        "metadata": {"processed": True}
    }

# Async function
async def async_postprocess(response: str) -> dict:
    """Asynchronously enrich response."""
    # Simulate async operation
    import asyncio
    await asyncio.sleep(0.1)
    return {
        "response": response,
        "word_count": len(response.split()),
        "timestamp": "2025-10-14T12:00:00Z"
    }

# Create runnables from functions
preprocess = RunnableLambda(preprocess_input)
postprocess = RunnableLambda(async_postprocess)

model = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_template("Summarize: {text}")

# Integrate into chain
chain = (
    preprocess
    | prompt
    | model
    | StrOutputParser()
    | postprocess
)

# Supports all Runnable methods
result = await chain.ainvoke({"text": "  LONG TEXT HERE  "})
print(result)  # {"response": "...", "word_count": 42, "timestamp": "..."}

# Batch processing with custom functions
results = chain.batch([
    {"text": "First document"},
    {"text": "Second document"}
])
```

## JSON Output Parser

Parse and validate JSON responses from LLMs.

```python
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# Define expected structure
class Recipe(BaseModel):
    name: str = Field(description="Recipe name")
    ingredients: list[str] = Field(description="List of ingredients")
    steps: list[str] = Field(description="Cooking steps")
    prep_time: int = Field(description="Preparation time in minutes")

# Create parser
parser = JsonOutputParser(pydantic_object=Recipe)

# Include format instructions in prompt
prompt = ChatPromptTemplate.from_template("""
Generate a recipe for {dish}.

{format_instructions}
""")

chain = (
    prompt.partial(format_instructions=parser.get_format_instructions())
    | ChatOpenAI(model="gpt-4")
    | parser
)

# Returns parsed dictionary
recipe = chain.invoke({"dish": "chocolate chip cookies"})
print(recipe["name"])  # "Classic Chocolate Chip Cookies"
print(recipe["ingredients"])  # ["2 cups flour", "1 cup butter", ...]
print(recipe["prep_time"])  # 45

# Validate against Pydantic model
validated_recipe = Recipe(**recipe)
```

## Configurable Chains with Runtime Parameters

Create chains with parameters that can be configured at runtime.

```python
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Model with configurable alternatives
model = ChatOpenAI(model="gpt-3.5-turbo").configurable_alternatives(
    ConfigurableField(id="model"),
    default_key="openai_gpt35",
    anthropic_claude=ChatAnthropic(model="claude-3-5-sonnet-latest"),
    openai_gpt4=ChatOpenAI(model="gpt-4"),
)

prompt = ChatPromptTemplate.from_template("Explain {topic}")
chain = prompt | model | StrOutputParser()

# Use different models at runtime
result1 = chain.invoke(
    {"topic": "blockchain"},
    config={"configurable": {"model": "openai_gpt35"}}
)

result2 = chain.invoke(
    {"topic": "blockchain"},
    config={"configurable": {"model": "anthropic_claude"}}
)

result3 = chain.invoke(
    {"topic": "blockchain"},
    config={"configurable": {"model": "openai_gpt4"}}
)

# Configurable temperature
model_with_temp = ChatOpenAI(temperature=0.7).configurable_fields(
    temperature=ConfigurableField(
        id="temperature",
        name="LLM Temperature",
        description="The temperature of the LLM"
    )
)

chain_configurable = prompt | model_with_temp | StrOutputParser()

# Adjust temperature at runtime
creative = chain_configurable.invoke(
    {"topic": "poetry"},
    config={"configurable": {"temperature": 1.2}}
)

deterministic = chain_configurable.invoke(
    {"topic": "poetry"},
    config={"configurable": {"temperature": 0.0}}
)
```

## Streaming with Callbacks

Monitor chain execution in real-time with callbacks for logging and debugging.

```python
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# Custom callback handler
class LoggingCallback(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"[LLM START] Prompts: {len(prompts)}")

    def on_llm_end(self, response, **kwargs):
        print(f"[LLM END] Tokens: {response.llm_output.get('token_usage', {})}")

    def on_chain_start(self, serialized, inputs, **kwargs):
        print(f"[CHAIN START] Inputs: {list(inputs.keys())}")

    def on_chain_end(self, outputs, **kwargs):
        print(f"[CHAIN END] Outputs: {type(outputs)}")

model = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_template("Write a story about {topic}")
chain = prompt | model | StrOutputParser()

# Use callback
result = chain.invoke(
    {"topic": "time travel"},
    config={"callbacks": [LoggingCallback()]}
)
# Output:
# [CHAIN START] Inputs: ['topic']
# [LLM START] Prompts: 1
# [LLM END] Tokens: {'prompt_tokens': 15, 'completion_tokens': 150, ...}
# [CHAIN END] Outputs: <class 'str'>

# Streaming with token-by-token callbacks
class StreamingCallback(BaseCallbackHandler):
    def on_llm_new_token(self, token: str, **kwargs):
        print(token, end="", flush=True)

for chunk in chain.stream(
    {"topic": "space exploration"},
    config={"callbacks": [StreamingCallback()]}
):
    pass  # Tokens printed in callback
```

## LangChain Framework Summary

LangChain provides a production-ready platform for building reliable LLM applications and agents through composable abstractions and a vast ecosystem of integrations. The framework excels in common use cases including conversational AI with persistent memory, retrieval-augmented generation (RAG) for grounding responses in private data, autonomous agents with tool calling capabilities, and multi-step workflows with conditional logic. Applications built with LangChain benefit from built-in observability through LangSmith integration, automatic prompt optimization, and the ability to swap models without code changes due to standardized interfaces across 100+ provider integrations. With LangGraph now integrated as a core dependency, LangChain agents provide durable execution, human-in-the-loop workflows, and advanced orchestration capabilities out of the box.

Integration patterns follow a consistent architecture: prompts define input templates with variable substitution, models implement the `BaseChatModel` or `BaseLLM` interface for interoperability, output parsers transform responses into structured data (JSON, Pydantic models, or custom formats), and tools extend model capabilities with function calling for external APIs and databases. Complex applications combine these primitives using LCEL's pipe operator (`|`) for sequential composition and dictionary syntax (`{}`) for parallel execution, with support for streaming responses, batch processing, and async operations built into every component. The framework's type-safe design using Pydantic ensures runtime validation, while the Runnable protocol provides consistent `.invoke()`, `.stream()`, `.batch()`, and `.ainvoke()` methods across all abstractions.