智能体将语言模型与工具 结合起来,创建可以推理任务、决定使用哪些工具并迭代地朝着解决方案努力的系统。
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 ):
"""使用自定义消息处理工具执行错误。"""
try :
return handler(request)
except Exception as e:
# 向模型返回自定义错误消息
return ToolMessage(
content = f "工具错误:请检查您的输入并重试。( { 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 循环中的工具使用
智能体遵循 ReAct(“推理 + 行动”)模式,在简短的推理步骤与有针对性的工具调用之间交替,并将观察结果反馈到后续决策中,直到能够提供最终答案。
提示:识别当前最受欢迎的无线耳机并验证可用性。 ================================ Human Message =================================
查找现在最受欢迎的无线耳机并检查是否有库存
推理 :“受欢迎程度是时间敏感的,我需要使用提供的搜索工具。”
行动 :调用 search_products("wireless headphones")
================================== Ai Message ==================================
Tool Calls:
search_products (call_abc123)
Call ID: call_abc123
Args:
query: wireless headphones
================================= Tool Message =================================
找到 5 个匹配"wireless headphones"的产品。前 5 个结果: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 =================================
产品 WH-1000XM5:库存 10 件
推理 :“我有了最受欢迎的型号及其库存状态。我现在可以回答用户的问题了。”
行动 :产生最终答案
================================== Ai Message ==================================
我找到了无线耳机(型号 WH-1000XM5),库存 10 件...
系统提示
您可以通过提供提示来塑造智能体处理任务的方式。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 :
"""根据用户角色生成系统提示。"""
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
)
# 系统提示将根据上下文动态设置
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?" }]}
)
有关从智能体流式传输步骤和/或 token,请参阅流式传输 指南。
否则,智能体遵循 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 开始,不再支持简单地传递模式(例如,response_format=ContactInfo)。您必须显式使用 ToolStrategy 或 ProviderStrategy。
智能体通过消息状态自动维护对话历史。您还可以配置智能体使用自定义状态模式来记住对话过程中的其他信息。
存储在状态中的信息可以视为智能体的短期记忆 :
自定义状态模式必须将 AgentState 扩展为 TypedDict。
有两种定义自定义状态的方法:
通过中间件 (推荐)
通过 create_agent 上的 state_schema
通过中间件定义状态
当您的自定义状态需要由特定中间件钩子和附加到该中间件的工具访问时,使用中间件定义自定义状态。
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()]
)
# 智能体现在可以跟踪消息之外的其他状态
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
)
# 智能体现在可以跟踪消息之外的其他状态
result = agent.invoke({
"messages" : [{ "role" : "user" , "content" : "I prefer technical explanations" }],
"user_preferences" : { "style" : "technical" , "verbosity" : "detailed" },
})
从 langchain 1.0 开始,自定义状态模式必须 是 TypedDict 类型。不再支持 Pydantic 模型和 dataclass。有关更多详细信息,请参阅 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] } " )
中间件
中间件 提供了强大的扩展性,用于在执行的不同阶段自定义智能体行为。您可以使用中间件来:
在调用模型之前处理状态(例如,消息修剪、上下文注入)
修改或验证模型的响应(例如,护栏、内容过滤)
使用自定义逻辑处理工具执行错误
基于状态或上下文实现动态模型选择
添加自定义日志记录、监控或分析
中间件无缝集成到智能体的执行图中,允许您在关键点拦截和修改数据流,而无需更改核心智能体逻辑。