智能体将语言模型与工具 结合,创建能够对任务进行推理、决定使用哪些工具并迭代地解决问题的系统。 create_agent 提供了一个生产就绪的智能体实现。 一个 LLM 智能体在循环中运行工具以实现目标 。智能体一直运行直到满足停止条件——即模型发出最终输出或达到迭代限制。
核心组件
智能体的推理引擎是模型 。它可以通过多种方式指定,支持静态和动态模型选择。
静态模型
静态模型在创建智能体时配置一次,并在整个执行过程中保持不变。这是最常见和直接的方法。 要从模型标识符字符串 初始化静态模型: from langchain.agents import create_agent
agent = create_agent(
"gpt-5" ,
tools = tools
)
模型标识符字符串支持自动推断(例如,"gpt-5" 将被推断为 "openai:gpt-5")。请参阅参考 ,查看模型标识符字符串映射的完整列表。
为了更好地控制模型配置,请直接使用提供者包初始化模型实例。在此示例中,我们使用 ChatOpenAI 。请参阅聊天模型 ,了解其他可用的聊天模型类。
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
model = "gpt-5" ,
temperature = 0.1 ,
max_tokens = 1000 ,
timeout = 30
# ... (other params)
)
agent = create_agent(model, tools = tools)
模型实例让您可以完全控制配置。当您需要设置特定的参数 时使用它们,例如 temperature、max_tokens、timeouts、base_url 和其他特定于提供者的设置。请参阅参考 ,查看模型上可用的参数和方法。
动态模型
动态模型在运行时 根据当前状态 和上下文选择。这支持复杂的路由逻辑和成本优化。 要使用动态模型,请使用 @wrap_model_call 装饰器创建中间件,该中间件会修改请求中的模型: from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
basic_model = ChatOpenAI( model = "gpt-4o-mini" )
advanced_model = ChatOpenAI( model = "gpt-4o" )
@wrap_model_call
def 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
request.model = model
return handler(request)
agent = create_agent(
model = basic_model, # Default model
tools = tools,
middleware = [dynamic_model_selection]
)
使用结构化输出时,不支持预绑定模型(已调用 bind_tools 的模型)。如果需要使用结构化输出进行动态模型选择,请确保传递给中间件的模型未预绑定。
工具赋予智能体执行操作的能力。智能体超越了简单的仅模型工具绑定,通过促进以下功能:
按顺序进行多次工具调用(由单个提示触发)
在适当情况下并行工具调用
根据先前结果进行动态工具选择
工具重试逻辑和错误处理
工具调用之间状态的持久化
欲了解更多信息,请参阅工具 。
将工具列表传递给智能体。
from langchain.tools import tool
from langchain.agents import create_agent
@tool
def search ( query : str ) -> str :
"""Search for information."""
return f "Results for: { query } "
@tool
def 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_agent
from langchain.agents.middleware import wrap_tool_call
from langchain_core.messages import ToolMessage
@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 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 = "..."
),
...
]
智能体遵循 ReAct(“推理+行动”)模式,在简短的推理步骤和有针对性的工具调用之间交替进行,并将产生的观察结果馈送到后续决策中,直到它们能够提供最终答案。
提示:识别当前最流行的无线耳机并验证其可用性。 ================================ Human Message =================================
Find the most popular wireless headphones right now and check if they're in stock
推理 :“流行度是时间敏感的,我需要使用提供的搜索工具。”
行动 :调用 search_products("wireless headphones")
================================== Ai Message ==================================
Tool Calls:
search_products (call_abc123)
Call ID: call_abc123
Args:
query: wireless headphones
================================= Tool Message =================================
Found 5 products matching "wireless headphones". Top 5 results: WH-1000XM5, ...
推理 :“我需要在回答之前确认排名第一的商品的可用性。”
行动 :调用 check_inventory("WH-1000XM5")
================================== Ai Message ==================================
Tool Calls:
check_inventory (call_def456)
Call ID: call_def456
Args:
product_id: WH-1000XM5
================================= Tool Message =================================
Product WH-1000XM5: 10 units in stock
推理 :“我已经获得了最受欢迎的型号及其库存状态。现在我可以回答用户的问题了。”
行动 :生成最终答案
================================== Ai Message ==================================
I found wireless headphones (model WH-1000XM5) with 10 units in stock...
系统提示
您可以通过提供提示来塑造智能体处理任务的方式。system_prompt 参数可以作为字符串提供。
agent = create_agent(
model,
tools,
system_prompt = "You are a helpful assistant. Be concise and accurate."
)
当未提供 system_prompt 时,智能体会直接从消息中推断其任务。
动态系统提示
对于更高级的用例,当您需要根据运行时上下文或智能体状态修改系统提示时,可以使用中间件 。 @dynamic_prompt 装饰器创建了一个中间件,可以根据模型请求动态生成系统提示:from typing import TypedDict
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
class Context ( TypedDict ):
user_role: str
@dynamic_prompt
def 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_prompt
agent = create_agent(
model = "gpt-4o" ,
tools = [web_search],
middleware = [user_role_prompt],
context_schema = Context
)
# The system prompt will be set dynamically based on context
result = agent.invoke(
{ "messages" : [{ "role" : "user" , "content" : "Explain machine learning" }]},
context = { "user_role" : "expert" }
)
有关消息类型和格式的更多详细信息,请参阅消息 。有关全面的中间件文档,请参阅中间件 。
您可以通过向其 State 传递更新来调用智能体。所有智能体都在其状态中包含一系列消息 ;要调用智能体,请传递一条新消息
result = agent.invoke(
{ "messages" : [{ "role" : "user" , "content" : "What's the weather in San Francisco?" }]}
)
有关从智能体流式传输步骤和/或标记的信息,请参阅流式传输 指南。 否则,智能体遵循 LangGraph Graph API 并支持所有相关方法。 高级概念
结构化输出
在某些情况下,您可能希望智能体以特定格式返回输出。LangChain 通过 response_format 参数提供结构化输出策略。
ToolStrategy 使用人工工具调用来生成结构化输出。这适用于任何支持工具调用的模型。
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-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)
ProviderStrategy 使用模型提供者原生的结构化输出生成。这更可靠,但仅适用于支持原生结构化输出的提供者(例如 OpenAI)。
from langchain.agents.structured_output import ProviderStrategy
agent = create_agent(
model = "gpt-4o" ,
response_format = ProviderStrategy(ContactInfo)
)
自 langchain 1.0 起,不再支持简单地传递 schema(例如,response_format=ContactInfo)。您必须明确使用 ToolStrategy 或 ProviderStrategy。
智能体通过消息状态自动维护对话历史。您还可以配置智能体使用自定义状态模式,以在对话过程中记住额外信息。 存储在状态中的信息可以被视为智能体的短期记忆 : 自定义状态模式必须将 AgentState 作为 TypedDict 扩展。 定义自定义状态有两种方式:
通过中间件 (首选)
通过 state_schema 在 create_agent 上
通过中间件定义状态
当您的自定义状态需要被特定的中间件钩子和附加到该中间件的工具访问时,请使用中间件定义自定义状态。
from langchain.agents import AgentState
from langchain.agents.middleware import AgentMiddleware
class CustomState ( AgentState ):
user_preferences: dict
class 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 messages
result = agent.invoke({
"messages" : [{ "role" : "user" , "content" : "I prefer technical explanations" }],
"user_preferences" : { "style" : "technical" , "verbosity" : "detailed" },
})
通过 state_schema 定义状态
使用 state_schema 参数作为定义仅在工具中使用的自定义状态的快捷方式。
from langchain.agents import AgentState
class CustomState ( AgentState ):
user_preferences: dict
agent = create_agent(
model,
tools = [tool1, tool2],
state_schema = CustomState
)
# The agent can now track additional state beyond messages
result = agent.invoke({
"messages" : [{ "role" : "user" , "content" : "I prefer technical explanations" }],
"user_preferences" : { "style" : "technical" , "verbosity" : "detailed" },
})
从 langchain 1.0 开始,自定义状态 schema **必须**是 TypedDict 类型。不再支持 Pydantic 模型和数据类。有关更多详细信息,请参阅 v1 迁移指南 。
要了解有关内存的更多信息,请参阅内存 。有关实现跨会话持久化的长期记忆的信息,请参阅长期记忆 。
流式处理
我们已经了解了如何使用 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] } " )
中间件
中间件 为在执行的不同阶段自定义智能体行为提供了强大的可扩展性。您可以使用中间件来
在模型调用前处理状态(例如,消息截断、上下文注入)
修改或验证模型的响应(例如,防护措施、内容过滤)
使用自定义逻辑处理工具执行错误
根据状态或上下文实现动态模型选择
添加自定义日志、监控或分析
中间件无缝集成到智能体的执行图中,允许您在关键点拦截和修改数据流,而无需更改核心智能体逻辑。