LangChain 源码剖析-代理详解(Agent)

代理将语言模型与工具相结合，构建能够推理任务、决定工具使用并迭代推进解决方案的系统。
createAgent() 提供了一个生产就绪的代理实现。
大型语言模型代理通过循环运行工具来实现目标。代理将持续运行，直到满足停止条件——即模型输出最终结果或达到迭代次数限制。

核心组件

静态模型

静态模型在创建代理时配置一次，并在整个执行过程中保持不变。这是最常见和最直接的方法。
要从模型标识符字符串初始化静态模型，请执行以下操作：

from langchain.agents import create_agentagent = create_agent(    "gpt-5",    tools=tools)

为了更好地控制模型配置，请直接使用提供程序包初始化模型实例。在这个例子中，我们使用ChatOpenAI。有关其他可用的聊天模型类，请参阅聊天模型。

from langchain.agents import create_agentfrom langchain_openai import ChatOpenAImodel = ChatOpenAI(    model="gpt-5",    temperature=0.1,    max_tokens=1000,    timeout=30    # ... (other params))agent = create_agent(model, tools=tools)

模型实例使您可以完全控制配置。当您需要设置特定参数时，如温度、max_tokens、超时、base_url和其他特定于提供程序的设置，请使用它们。请参阅参考资料，查看模型上的可用参数和方法。

动态模型

在运行时根据当前状态和上下文选择动态模型。这实现了复杂的路由逻辑和成本优化。
要使用动态模型，请使用@wrap_model_call装饰器创建中间件，该装饰器修改请求中的模型：

from langchain_openai import ChatOpenAIfrom langchain.agents import create_agentfrom langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponsebasic_model = ChatOpenAI(model="gpt-4o-mini")advanced_model = ChatOpenAI(model="gpt-4o")@wrap_model_calldef dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse:    """Choose model based on conversation complexity."""    message_count = len(request.state["messages"])    if message_count > 10:        # Use an advanced model for longer conversations        model = advanced_model    else:        model = basic_model    return handler(request.override(model=model))agent = create_agent(    model=basic_model,  # Default model    tools=tools,    middleware=[dynamic_model_selection])

工具

工具赋予代理人采取行动的能力。代理通过促进以下方面超越了简单的模型工具绑定：
按顺序调用多个工具（由单个提示触发）
在适当的时候并行调用工具
基于先前结果的动态刀具选择
工具重试逻辑和错误处理
跨工具调用的状态持久化

定义工具

将工具列表传递给代理。

from langchain.tools import toolfrom langchain.agents import create_agent@tooldef search(query: str) -> str:    """Search for information."""    return f"Results for: {query}"@tooldef get_weather(location: str) -> str:    """Get weather information for a location."""    return f"Weather in {location}: Sunny, 72°F"agent = create_agent(model, tools=[search, get_weather])

如果提供了一个空的工具列表，则代理将由一个没有工具调用功能的LLM节点组成。

工具错误处理

要自定义如何处理工具错误，请使用@wrap_tool_call装饰器创建中间件：

from langchain.agents import create_agentfrom langchain.agents.middleware import wrap_tool_callfrom langchain.messages import ToolMessage@wrap_tool_calldef handle_tool_errors(request, handler):    """Handle tool execution errors with custom messages."""    try:        return handler(request)    except Exception as e:        # Return a custom error message to the model        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-4o",    tools=[search, get_weather],    middleware=[handle_tool_errors])

当工具发生故障时，代理将返回一条ToolMessage，其中包含自定义错误消息：

[    ...    ToolMessage(        content="Tool error: Please check your input and try again. (division by zero)",        tool_call_id="..."    ),    ...]

系统提示

您可以通过提供提示来塑造代理处理任务的方式， system_prompt参数可以以字符串形式提供：

agent = create_agent(    model,    tools,    system_prompt="You are a helpful assistant. Be concise and accurate.")

当没有提供system_prompt时，代理将直接从消息中推断其任务。
system_prompt参数接受str或SystemMessage。使用SystemMessage可以更好地控制提示结构，这对于Anthropic的提示缓存等特定于提供者的功能非常有用：

from langchain.agents import create_agentfrom langchain.messages import SystemMessage, HumanMessageliterary_agent = create_agent(    model="anthropic:claude-sonnet-4-5",    system_prompt=SystemMessage(        content=[            {                "type": "text",                "text": "You are an AI assistant tasked with analyzing literary works.",            },            {                "type": "text",                "text": "<the entire contents of 'Pride and Prejudice'>",                "cache_control": {"type": "ephemeral"}            }        ]    ))result = literary_agent.invoke(    {"messages": [HumanMessage("Analyze the major themes in 'Pride and Prejudice'.")]})

带有｛“type”：“ephemary”｝的cache_control字段告诉Anthropic缓存该内容块，从而降低使用相同系统提示的重复请求的延迟和成本。

动态系统提示

对于需要根据运行时上下文或代理状态修改系统提示的更高级用例，可以使用中间件。
@dynamic_prompt装饰器创建中间件，根据模型请求生成系统提示：

from typing import TypedDictfrom langchain.agents import create_agentfrom langchain.agents.middleware import dynamic_prompt, ModelRequestclass Context(TypedDict):    user_role: str@dynamic_promptdef user_role_prompt(request: ModelRequest) -> str:    """Generate system prompt based on user role."""    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."    elif user_role == "beginner":        return f"{base_prompt} Explain concepts simply and avoid jargon."    return base_promptagent = create_agent(    model="gpt-4o",    tools=[web_search],    middleware=[user_role_prompt],    context_schema=Context)# The system prompt will be set dynamically based on contextresult = agent.invoke(    {"messages": [{"role": "user", "content": "Explain machine learning"}]},    context={"user_role": "expert"})

调用

您可以通过向代理的状态传递更新来调用代理。所有代理在其状态中都包含一系列消息；要调用代理，请传递一条新消息：

result = agent.invoke(    {"messages": [{"role": "user", "content": "What's the weather in San Francisco?"}]})

高级使用

结构化输出

在某些情况下，您可能希望代理以特定格式返回输出。LangChain通过response_format参数提供结构化输出策略。

工具策略

安装pydantic库

pip install pydanticoruv add pydantic

ToolStrategy使用人工工具调用来生成结构化输出。这适用于支持工具调用的任何模型：

from pydantic import BaseModelfrom langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategyclass ContactInfo(BaseModel):    name: str    email: str    phone: stragent = create_agent(    model="gpt-4o-mini",    tools=[search_tool],    response_format=ToolStrategy(ContactInfo))result = agent.invoke({    "messages": [{"role": "user", "content": "Extract contact info from: John Doe, john@example.com, (555) 123-4567"}]})result["structured_response"]# ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')

供应商战略

ProviderStrategy使用模型提供者的本机结构化输出生成。这更可靠，但仅适用于支持原生结构化输出的提供商（例如OpenAI）：

from langchain.agents.structured_output import ProviderStrategyagent = create_agent(    model="gpt-4o",    response_format=ProviderStrategy(ContactInfo))

记忆

代理通过消息状态自动维护对话历史记录。您还可以将代理配置为使用自定义状态模式，以便在对话期间记住其他信息。
存储在状态中的信息可以被认为是代理的短期记忆

自定义状态架构必须将AgentState扩展为TypedDict。

有两种方法可以定义自定义状态：
通过中间件（首选）
通过create_agent上的state_schema

通过中间件定义状态

当您的自定义状态需要由连接到所述中间件的特定中间件挂钩和工具访问时，使用中间件定义自定义状态。

from langchain.agents import AgentStatefrom langchain.agents.middleware import AgentMiddlewarefrom typing import Anyclass CustomState(AgentState):    user_preferences: dictclass CustomMiddleware(AgentMiddleware):    state_schema = CustomState    tools = [tool1, tool2]    def before_model(self, state: CustomState, runtime) -> dict[str, Any] | None:        ...agent = create_agent(    model,    tools=tools,    middleware=[CustomMiddleware()])# The agent can now track additional state beyond messagesresult = agent.invoke({    "messages": [{"role": "user", "content": "I prefer technical explanations"}],    "user_preferences": {"style": "technical", "verbosity": "detailed"},})

通过state_schema定义状态

使用state_schema参数作为快捷方式来定义仅在工具中使用的自定义状态。

from langchain.agents import AgentStateclassCustomState(AgentState):    user_preferences: dictagent = create_agent(    model,    tools=[tool1, tool2],    state_schema=CustomState)# The agent can now track additional state beyond messagesresult = agent.invoke({    "messages": [{"role": "user", "content": "I prefer technical explanations"}],    "user_preferences": {"style": "technical", "verbosity": "detailed"},})

流媒体

我们已经看到了如何使用invoke调用代理以获得最终响应。如果代理执行多个步骤，这可能需要一段时间。为了显示中间进度，我们可以在消息发生时进行流式传输

for chunk in agent.stream({    "messages": [{"role": "user", "content": "Search for AI news and summarize the findings"}]}, stream_mode="values"):    # Each chunk contains the full state at that point    latest_message = chunk["messages"][-1]    if latest_message.content:        print(f"Agent: {latest_message.content}")    elif latest_message.tool_calls:        print(f"Calling tools: {[tc['name'] for tc in latest_message.tool_calls]}")

中间件

中间件为定制代理在不同执行阶段的行为提供了强大的可扩展性。您可以使用中间件来：

调用模型之前的进程状态（例如，消息修剪、上下文注入）
修改或验证模型的响应（例如护栏、内容过滤）
使用自定义逻辑处理工具执行错误
基于状态或上下文实现动态模型选择
添加自定义日志记录、监控或分析
中间件无缝集成到代理的执行中，允许您在关键点拦截和修改数据流，而无需更改核心代理逻辑。
有关包括@before_model、@after_model和@wrap_tool_call等装饰器的全面中间件文档，请参阅中间件。