源码级别解析 · 源码解析 · 2026
2026-04-12 | 每日技术深度解读
LangChain helps developers build applications powered by LLMs through standard interfaces
LangChain provides abstractions for each component with interchangeable implementations
Agents follow ReAct pattern: Reasoning + Acting
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
# Initialize model
model = ChatOpenAI(
model="gpt-5",
temperature=0.1,
max_tokens=1000
)
# Create agent
agent = create_agent(model, tools=[search, get_weather])Simple agent creation with static model and tools
class AgentAction(Serializable):
"""Represents a request to execute an action by an agent."""
tool: str
"""The name of the `Tool` to execute."""
tool_input: str | dict
"""The input to pass in to the `Tool`."""
log: str
"""Additional information to log about the action."""
type: Literal["AgentAction"] = "AgentAction"Core schema for agent actions and tool execution
Models can be specified via identifier strings or direct instantiation
# Static model
agent = create_agent("openai:gpt-5", tools=tools)
# Dynamic model with middleware
@wrap_model_call
def dynamic_model_selection(request: ModelRequest, handler):
message_count = len(request.state["messages"])
model = advanced_model if message_count > 10 else basic_model
return handler(request.override(model=model))
agent = create_agent(basic_model, tools=tools, middleware=[dynamic_model_selection])Flexible model selection strategies
Tools enable agents to take actions and interact with the world
from langchain.tools import tool
@tool
def search(query: str) -> str:
"""Search for information online."""
return f"Search results for: {query}"
@tool
def get_weather(location: str) -> str:
"""Get weather information for a location."""
return f"Weather in {location}: Sunny, 72°F"Simple tool definitions with type hints and descriptions
Tools can be added or removed based on runtime context and state
class DynamicToolMiddleware(AgentMiddleware):
"""Middleware that registers and handles dynamic tools."""
def wrap_model_call(self, request: ModelRequest, handler):
# Add dynamic tool at runtime
updated = request.override(tools=[*request.tools, calculate_tip])
return handler(updated)
def wrap_tool_call(self, request: ToolCallRequest, handler):
# Handle execution of dynamic tools
if request.tool_call["name"] == "calculate_tip":
return handler(request.override(tool=calculate_tip))
return handler(request)Middleware for runtime tool registration and execution
System prompts shape how agents approach tasks
# Static system prompt
agent = create_agent(
model,
tools,
system_prompt="You are a helpful assistant. Be concise and accurate."
)
# Dynamic system prompt middleware
@dynamic_prompt
def user_role_prompt(request: ModelRequest) -> str:
user_role = request.runtime.context.get("user_role", "user")
base_prompt = "You are a helpful assistant."
if user_role == "expert":
return f"{base_prompt} Provide detailed technical responses."
return base_promptFlexible system prompt configuration
Agents alternate between reasoning steps and tool calls
LangChain agent execution flow following ReAct pattern
LangChain provides multiple approaches for structured responses
from pydantic import BaseModel
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class ContactInfo(BaseModel):
name: str
email: str
phone: str
agent = create_agent(
model="gpt-4.1-mini",
tools=[search_tool],
response_format=ToolStrategy(ContactInfo)
)
result = agent.invoke({
"messages": [{"role": "user", "content": "Extract contact info from: John Doe, [email protected], (555) 123-4567"}]
})Using structured output to enforce response format
Agents maintain context through message state and custom schemas
from langchain.agents import AgentState
from typing import TypedDict
class CustomState(AgentState):
user_preferences: dict
conversation_history: list
current_task: str
# Using middleware with custom state
class CustomMiddleware(AgentMiddleware):
state_schema = CustomState
def before_model(self, state: CustomState, runtime):
# Add context from custom state
return {"user_prefs": state.user_preferences}Extending agent state with custom memory
LangChain supports multiple streaming modes for real-time feedback
# Stream intermediate results
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "Search for AI news"}]},
stream_mode="values"
):
latest_message = chunk["messages"][-1]
if latest_message.content:
print(f"Agent: {latest_message.content}")
elif latest_message.tool_calls:
print(f"Calling: {[tc['name'] for tc in latest_message.tool_calls]}")Real-time agent execution monitoring
Middleware provides powerful extensibility for customizing behavior
@wrap_tool_call
def handle_tool_errors(request, handler):
"""Handle tool execution errors with custom messages."""
try:
return handler(request)
except Exception as e:
return ToolMessage(
content=f"Tool error: Please check your input and try again. ({str(e)})",
tool_call_id=request.tool_call["id"]
)
agent = create_agent(
model="gpt-4.1",
tools=[search, get_weather],
middleware=[handle_tool_errors]
)Custom error handling middleware
LangChain agents can be combined into multi-agent systems
Multi-agent system with specialized subagents
LangChain integrates with various external tools and systems
from langchain.tools import tool
from langchain_community.tools import WikipediaQueryRun
from langchain_community.tools import DuckDuckGoSearchRun
# Pre-built tools
wiki = WikipediaQueryRun()
search = DuckDuckGoSearchRun()
# Custom API tool
@tool
def call_api(endpoint: str, params: dict) -> str:
"""Make API calls to external services."""
response = requests.get(endpoint, params=params)
return response.json()Combining built-in and custom tools
Comprehensive evaluation tools for agent development
# Simple agent test
def test_agent_weather():
"""Test agent weather query functionality."""
result = agent.invoke({
"messages": [{"role": "user", "content": "What's the weather in San Francisco?"}]
})
assert "weather" in result["messages"][-1].content.lower()
assert "san francisco" in result["messages"][-1].content.lower()
print("Weather test passed!")Basic unit testing for agent functionality
Strategies for optimizing agent performance
# Model caching for repeated queries
@wrap_model_call
def model_caching(request, handler):
cache_key = hash(request.messages[-1].content)
if cache_key in cache:
return cache[cache_key]
result = handler(request)
cache[cache_key] = result
return result
# Parallel tool execution
async def parallel_tool_execution(agent, queries):
tasks = [process_query(query) for query in queries]
return await asyncio.gather(*tasks)Caching and parallel execution strategies
Security best practices for agent deployment
Architectural patterns for production deployment
# Production-ready agent configuration
production_agent = create_agent(
model="gpt-4.1",
tools=[production_tools],
middleware=[
rate_limiting_middleware,
authentication_middleware,
monitoring_middleware
],
state_schema=ProductionState,
store=InMemoryStore()
)
# Container-ready entry point
if __name__ == "__main__":
serve_agent(production_agent)Production configuration with security and monitoring
LangChain agents solve real business problems
class CustomerSupportAgent:
"""Customer support automation agent."""
def __init__(self):
self.agent = create_agent(
model="gpt-4.1",
tools=[
lookup_customer_info,
check_order_status,
process_refund,
escalate_to_human
],
system_prompt="You are a helpful customer support assistant."
)
def handle_query(self, query: str) -> str:
"""Process customer support query."""
result = self.agent.invoke({"messages": [{"role": "user", "content": query}]})
return result["messages"][-1].contentSpecialized customer support agent implementation
LangGraph provides advanced orchestration capabilities
Complex workflow orchestration with LangGraph
Key practices for successful agent development
# Good tool design
@tool
def get_user_data(user_id: str) -> dict:
"""Get user profile information.
Args:
user_id: Unique user identifier
Returns:
User profile dictionary
Raises:
ValueError: If user_id is invalid
"""
if not user_id:
raise ValueError("User ID cannot be empty")
# Implementation
return get_user_from_database(user_id)Well-documented tools with clear interfaces
Guide for upgrading from LangChain v0 to v1
Vibrant community and extensive ecosystem
Exciting future developments for LangChain
Multiple resources available for learning LangChain
LangChain provides comprehensive framework for building intelligent applications
感谢阅读！
访问 https://atcfu.com/ai-articles/langchain-agent-platform/ 回顾本文