Introduction to the Migration
Pydantic AI and LangChain are two popular frameworks in the Python AI ecosystem, each with distinct philosophies. Pydantic AI emphasizes type safety, structured outputs, and a developer experience rooted in Pydantic's validation model. LangChain, on the other hand, offers a broader ecosystem with extensive integrations, chains, agents, memory systems, and retrieval-augmented generation (RAG) components. Migrating from Pydantic AI to LangChain is a common transition for teams that need richer orchestration capabilities, more tool integrations, or access to LangChain's agent frameworks like LangGraph.
This guide walks through the complete migration process, covering model initialization, structured outputs, tool calling, agents, and best practices to ensure a smooth transition.
Why Migrate from Pydantic AI to LangChain?
Before diving into code, it's important to understand the motivations behind this migration:
- Broader Integration Ecosystem: LangChain supports hundreds of integrations including vector stores, document loaders, retrievers, and LLM providers out of the box.
- Agent Frameworks: LangGraph and LangChain's agent abstractions provide more sophisticated multi-step reasoning and state management.
- Community and Documentation: LangChain has a larger community, more tutorials, and extensive documentation.
- Production Tooling: LangSmith integration provides tracing, evaluation, and monitoring capabilities.
- Composability: LangChain's LCEL (LangChain Expression Language) allows building complex pipelines with streaming and async support.
Setting Up Your Environment
First, install the required LangChain packages. Unlike Pydantic AI, LangChain uses a modular package structure:
pip install langchain langchain-core langchain-openai langchain-anthropic langgraph
pip install pydantic # still used for schema definitions
Set your API keys as environment variables:
import os
os.environ["OPENAI_API_KEY"] = "your-openai-key"
os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-key"
Migrating Model Initialization
Pydantic AI Approach
In Pydantic AI, you define an Agent with a model and a result type:
from pydantic_ai import Agent
from pydantic import BaseModel
class Response(BaseModel):
answer: str
confidence: float
agent = Agent(
"openai:gpt-4o",
result_type=Response,
system_prompt="You are a helpful assistant."
)
result = agent.run_sync("What is the capital of France?")
print(result.data)
LangChain Equivalent
In LangChain, you initialize the model directly and use structured output methods:
from langchain_openai import ChatOpenAI
from pydantic import BaseModel
class Response(BaseModel):
answer: str
confidence: float
model = ChatOpenAI(
model="gpt-4o",
temperature=0
)
structured_model = model.with_structured_output(Response)
result = structured_model.invoke("What is the capital of France?")
print(result)
# Output: Response(answer='Paris', confidence=0.99)
The key difference is that LangChain separates the model from the agent concept. The with_structured_output method wraps the model to return a Pydantic object directly, similar to Pydantic AI's result_type.
Migrating System Prompts and Instructions
Pydantic AI
from pydantic_ai import Agent
agent = Agent(
"openai:gpt-4o",
system_prompt="You are a helpful assistant that answers concisely.",
)
@agent.system_prompt
def dynamic_prompt(ctx) -> str:
return f"Today's date is {ctx.date}"
LangChain Equivalent
LangChain uses prompt templates and message objects:
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant that answers concisely. Today's date is {date}."),
("human", "{question}")
])
model = ChatOpenAI(model="gpt-4o")
chain = prompt | model
response = chain.invoke({
"date": "2025-01-15",
"question": "What is the capital of France?"
})
print(response.content)
The | operator is part of LangChain Expression Language (LCEL), which allows you to compose chains declaratively. This replaces Pydantic AI's decorator-based approach with a more explicit pipeline.
Migrating Tool Calling
Pydantic AI Tools
from pydantic_ai import Agent, RunContext
agent = Agent("openai:gpt-4o")
@agent.tool
def get_weather(ctx: RunContext[str], city: str) -> str:
"""Get the weather for a city."""
# Simulated weather lookup
return f"The weather in {city} is sunny, 22°C"
result = agent.run_sync("What's the weather in Paris?")
print(result.data)
LangChain Tools
LangChain provides the @tool decorator for defining tools:
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
@tool
def get_weather(city: str) -> str:
"""Get the weather for a city."""
return f"The weather in {city} is sunny, 22°C"
model = ChatOpenAI(model="gpt-4o")
model_with_tools = model.bind_tools([get_weather])
response = model_with_tools.invoke("What's the weather in Paris?")
print(response.tool_calls)
# Output: [{'name': 'get_weather', 'args': {'city': 'Paris'}, 'id': 'call_xxx'}]
To execute the tool automatically, you need to build an agent loop. Here is a complete example using LangGraph:
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
@tool
def get_weather(city: str) -> str:
"""Get the weather for a city."""
return f"The weather in {city} is sunny, 22°C"
@tool
def get_population(city: str) -> str:
"""Get the population of a city."""
populations = {"Paris": "2.1 million", "Tokyo": "13.9 million"}
return populations.get(city, "Unknown")
model = ChatOpenAI(model="gpt-4o")
tools = [get_weather, get_population]
agent = create_react_agent(model, tools)
result = agent.invoke({
"messages": [{"role": "user", "content": "What's the weather and population of Paris?"}]
})
for msg in result["messages"]:
print(f"{msg.type}: {msg.content}")
The create_react_agent function from LangGraph replaces Pydantic AI's built-in agent loop. It implements a ReAct-style agent that automatically calls tools and synthesizes responses.
Migrating Dependency Injection
Pydantic AI Dependencies
Pydantic AI uses a typed dependency injection system:
from pydantic_ai import Agent, RunContext
from dataclasses import dataclass
@dataclass
class MyDeps:
db_client: str # simulated database client
api_key: str
agent = Agent("openai:gpt-4o", deps_type=MyDeps)
@agent.tool
def query_db(ctx: RunContext[MyDeps], table: str) -> str:
"""Query the database."""
return f"Queried {table} using {ctx.deps.db_client}"
result = agent.run_sync(
"Get all users from the database",
deps=MyDeps(db_client="postgres://localhost", api_key="secret")
)
LangChain Equivalent
LangChain does not have a built-in dependency injection system like Pydantic AI. Instead, you use closures, partial functions, or configuration objects:
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from dataclasses import dataclass
@dataclass
class MyDeps:
db_client: str
api_key: str
def create_tools(deps: MyDeps):
@tool
def query_db(table: str) -> str:
"""Query the database."""
return f"Queried {table} using {deps.db_client}"
@tool
def call_api(endpoint: str) -> str:
"""Call an external API."""
return f"Called {endpoint} with key {deps.api_key[:4]}..."
return [query_db, call_api]
deps = MyDeps(db_client="postgres://localhost", api_key="secret")
tools = create_tools(deps)
model = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(model, tools)
result = agent.invoke({
"messages": [{"role": "user", "content": "Get all users from the database"}]
})
print(result["messages"][-1].content)
This closure-based approach achieves the same goal: tools have access to shared dependencies without global state.
Migrating Structured Output with Validation
Pydantic AI with Validators
from pydantic_ai import Agent
from pydantic import BaseModel, field_validator
class Answer(BaseModel):
value: int
explanation: str
@field_validator("value")
def check_range(cls, v):
if not 0 <= v <= 100:
raise ValueError("Value must be between 0 and 100")
return v
agent = Agent("openai:gpt-4o", result_type=Answer)
result = agent.run_sync("Give me a percentage for cloud coverage today.")
print(result.data)
LangChain with Structured Output
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, field_validator
class Answer(BaseModel):
value: int
explanation: str
@field_validator("value")
def check_range(cls, v):
if not 0 <= v <= 100:
raise ValueError("Value must be between 0 and 100")
return v
model = ChatOpenAI(model="gpt-4o", temperature=0)
structured_model = model.with_structured_output(Answer)
result = structured_model.invoke("Give me a percentage for cloud coverage today.")
print(result)
LangChain's with_structured_output leverages the same Pydantic validators. By default, it uses function calling under the hood. You can also specify JSON mode:
structured_model = model.with_structured_output(
Answer,
method="json_mode"
)
Migrating Streaming Responses
Pydantic AI Streaming
from pydantic_ai import Agent
agent = Agent("openai:gpt-4o")
with agent.run_stream("Tell me a story about a robot.") as result:
for chunk in result:
print(chunk, end="", flush=True)
LangChain Streaming
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
model = ChatOpenAI(model="gpt-4o", streaming=True)
prompt = ChatPromptTemplate.from_messages([
("human", "Tell me a story about a robot.")
])
chain = prompt | model
for chunk in chain.stream({}):
print(chunk.content, end="", flush=True)
LangChain's streaming works through the stream method on any LCEL chain. You can also stream structured outputs:
structured_model = model.with_structured_output(Answer)
for chunk in structured_model.stream("Give me a percentage for cloud coverage."):
print(chunk)
Migrating Multi-Agent Workflows
Pydantic AI supports multi-agent workflows through programmatic orchestration. LangChain offers LangGraph for this purpose, which is significantly more powerful for complex stateful workflows.
LangGraph Multi-Agent Example
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
model = ChatOpenAI(model="gpt-4o")
@tool
def research_topic(topic: str) -> str:
"""Research a topic and return findings."""
return f"Research findings on {topic}: Key insight found."
@tool
def write_summary(research: str) -> str:
"""Write a summary based on research."""
return f"Summary: {research} - Condensed into key points."
researcher = create_react_agent(model, [research_topic])
writer = create_react_agent(model, [write_summary])
# Step 1: Research
research_result = researcher.invoke({
"messages": [{"role": "user", "content": "Research quantum computing."}]
})
research_output = research_result["messages"][-1].content
# Step 2: Write summary based on research
write_result = writer.invoke({
"messages": [{"role": "user", "content": f"Write a summary of this research: {research_output}"}]
})
final_output = write_result["messages"][-1].content
print(final_output)
For more complex workflows with conditional routing, LangGraph's StateGraph API provides fine-grained control:
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
class State(TypedDict):
messages: Annotated[list, add_messages]
research: str
summary: str
model = ChatOpenAI(model="gpt-4o")
def research_node(state: State) -> State:
response = model.invoke(state["messages"])
return {"research": response.content, "messages": [response]}
def summarize_node(state: State) -> State:
prompt = f"Summarize this: {state['research']}"
response = model.invoke([{"role": "user", "content": prompt}])
return {"summary": response.content}
def should_summarize(state: State) -> str:
if len(state.get("research", "")) > 100:
return "summarize"
return END
graph = StateGraph(State)
graph.add_node("research", research_node)
graph.add_node("summarize", summarize_node)
graph.add_edge(START, "research")
graph.add_conditional_edges("research", should_summarize)
graph.add_edge("summarize", END)
app = graph.compile()
result = app.invoke({"messages": [{"role": "user", "content": "Explain quantum entanglement."}]})
print(result["summary"])
Migrating Message History and Memory
Pydantic AI Message History
from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessage
agent = Agent("openai:gpt-4o")
message_history: list[ModelMessage] = []
result1 = agent.run_sync("My name is Alice.", message_history=message_history)
message_history = result1.all_messages()
result2 = agent.run_sync("What is my name?", message_history=message_history)
print(result2.data) # Should reference Alice
LangChain Message History
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
model = ChatOpenAI(model="gpt-4o")
messages = [
SystemMessage(content="You are a helpful assistant."),
HumanMessage(content="My name is Alice."),
]
response = model.invoke(messages)
messages.append(response) # response is an AIMessage
messages.append(HumanMessage(content="What is my name?"))
response2 = model.invoke(messages)
print(response2.content) # Should reference Alice
For persistent memory across sessions, LangChain provides memory backends:
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
model = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant."),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
chain = prompt | model
chain_with_history = RunnableWithMessageHistory(
chain,
lambda session_id: ChatMessageHistory(session_id=session_id),
input_messages_key="input",
history_messages_key="history",
)
config = {"configurable": {"session_id": "user_123"}}
chain_with_history.invoke({"input": "My name is Alice."}, config=config)
response = chain_with_history.invoke({"input": "What is my name?"}, config=config)
print(response.content)
Best Practices for Migration
1. Migrate Incrementally
Do not attempt to migrate everything at once. Start by migrating individual agents or chains, test them thoroughly, and then move to more complex workflows. This reduces risk and makes debugging easier.
2. Leverage Pydantic for Schemas
LangChain fully supports Pydantic models for structured outputs. Continue using Pydantic BaseModel classes for your schemas — they work seamlessly with with_structured_output.
3. Use LCEL for Composition
Embrace LangChain Expression Language. The pipe operator (|) creates composable, debuggable, and streamable chains. Avoid writing imperative orchestration code when LCEL can express the same logic declaratively:
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_template("Explain {topic} in simple terms.")
model = ChatOpenAI(model="gpt-4o")
parser = StrOutputParser()
chain = prompt | model | parser
result = chain.invoke({"topic": "neural networks"})
print(result)
4. Adopt LangGraph for Complex Agents
If your Pydantic AI application uses multiple agents, conditional logic, or stateful workflows, LangGraph is the right tool. It provides cycles, persistence, human-in-the-loop interrupts, and time travel debugging.
5. Enable LangSmith Tracing
Set up LangSmith early in the migration process. It provides visibility into every LLM call, tool invocation, and chain step:
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-key"
os.environ["LANGCHAIN_PROJECT"] = "migration-project"
6. Handle Async Properly
Pydantic AI uses run for async and run_sync for sync. LangChain uses invoke for sync and ainvoke for async. Make sure to update all call sites:
# Pydantic AI
result = await agent.run("Hello")
# LangChain
result = await chain.ainvoke({"input": "Hello"})
7. Test with Parallel Runs
During migration, run both implementations side by side and compare outputs. This helps catch behavioral differences, especially around tool calling and structured output parsing.
8. Keep Tool Docstrings Clear
LangChain uses tool docstrings as descriptions sent to the LLM. Make sure your docstrings are clear and specific, as they directly influence tool selection behavior:
@tool
def search_products(query: str, max_results: int = 10) -> str:
"""Search the product catalog by keyword.
Args:
query: The search keyword or phrase.
max_results: Maximum number of products to return (default 10).
Returns:
A formatted string listing matching products with names and prices.
"""
# implementation
pass
Common Pitfalls and Solutions
Pitfall 1: Forgetting to Pass Message History
In Pydantic AI, the agent can manage message history internally. In LangChain, you must explicitly manage or configure message history. Use RunnableWithMessageHistory or manually append messages.
Pitfall 2: Tool Argument Schema Mismatches
Pydantic AI infers tool arguments from type hints. LangChain does the same but is stricter about complex types. Stick to primitive types (str, int, float, bool) and simple Pydantic models for tool arguments.
Pitfall 3: Structured Output Method Differences
LangChain's with_structured_output defaults to function calling. If your model does not support function calling, switch to JSON mode or parsing-based approaches:
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
parser = PydanticOutputParser(pydantic_object=Answer)
prompt = ChatPromptTemplate.from_template(
"Answer the question.\n{format_instructions}\n\nQuestion: {question}"
).partial(format_instructions=parser.get_format_instructions())
chain = prompt | model | parser
result = chain.invoke({"question": "What is 2+2?"})
Conclusion
Migrating from Pydantic AI to LangChain opens up a richer ecosystem of integrations, agent frameworks, and production tooling. The core concepts map naturally: Pydantic AI's Agent becomes a LangChain model with with_structured_output, tools use the @tool decorator, and complex agent workflows transition to LangGraph. By migrating incrementally, leveraging LCEL for composition, and enabling LangSmith for observability, you can achieve a smooth transition while gaining access to LangChain's full capabilities. Remember that Pydantic models remain central to schema definition in LangChain, so your existing validation logic carries over directly. With the patterns and best practices outlined in this guide, your migration should be systematic, testable, and low-risk.