代理将语言模型与工具 相结合,创建能够推理任务、决定使用哪些工具并迭代地寻求解决方案的系统。
create_agent 提供了一个生产就绪的代理实现。
一个LLM代理通过循环运行工具来实现目标 。
代理会运行直到满足停止条件——即当模型发出最终输出或达到迭代限制时。
核心组件
模型 是你的代理的推理引擎。它可以通过多种方式指定,支持静态和动态模型选择。
静态模型
静态模型在创建代理时配置一次,并在整个执行过程中保持不变。这是最常见和最直接的方法。
要从一个模型标识符字符串 初始化静态模型:
from langchain . agents import create_agent
agent = create_agent ( "openai:gpt-5.4" , tools = tools )
模型标识符字符串支持自动推断(例如,"gpt-5.4" 将被推断为 "openai:gpt-5.4")。请参阅参考文档 以查看完整的模型标识符字符串映射列表。
要对模型配置进行更多控制,可以直接使用提供商包初始化模型实例。在此示例中,我们使用 ChatOpenAI 。有关其他可用的聊天模型类,请参阅聊天模型 。
from langchain . agents import create_agent
from langchain_openai import ChatOpenAI
model = ChatOpenAI (
model = "gpt-5.4" ,
temperature = 0.1 ,
max_tokens = 1000 ,
timeout = 30
# ... (其他参数)
)
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-5.4-mini" )
advanced_model = ChatOpenAI ( model = "gpt-5.4" )
@wrap_model_call
def dynamic_model_selection ( request : ModelRequest , handler ) -> ModelResponse :
"""根据对话复杂性选择模型。"""
message_count = len ( request . state [ " messages " ])
if message_count > 10 :
# 对于较长的对话使用高级模型
model = advanced_model
else :
model = basic_model
return handler ( request . override ( model = model ))
agent = create_agent (
model = basic_model , # 默认模型
tools = tools ,
middleware = [ dynamic_model_selection ]
)
当使用结构化输出时,不支持预绑定模型(已调用 bind_tools 的模型)。如果你需要支持结构化输出的动态模型选择,请确保传递给中间件的模型未被预绑定。
工具赋予代理采取行动的能力。代理超越了简单的仅模型工具绑定,支持:
由单个提示触发的顺序多次工具调用
在适当情况下的并行工具调用
基于先前结果的动态工具选择
工具重试逻辑和错误处理
跨工具调用的状态持久性
更多信息,请参阅工具 。
静态工具
静态工具在创建代理时定义,并在整个执行过程中保持不变。这是最常见和最直接的方法。
要定义具有静态工具的代理,请将工具列表传递给代理。
工具可以指定为普通的 Python 函数或协程 。 工具装饰器 可用于自定义工具名称、描述、参数模式和其他属性。
from langchain . tools import tool
from langchain . agents import create_agent
@tool
def search ( query : str ) -> str :
"""搜索信息。"""
return f "Results for: { query } "
@tool
def get_weather ( location : str ) -> str :
"""获取某个位置的天气信息。"""
return f "Weather in { location } : Sunny, 72°F"
agent = create_agent ( model , tools = [ search , get_weather ])
如果提供空的工具列表,代理将仅包含一个没有工具调用能力的 LLM 节点。
动态工具
使用动态工具,代理可用的工具集在运行时修改,而不是预先全部定义。并非每个工具都适用于每种情况。工具过多可能会使模型不堪重负(上下文过载)并增加错误;工具过少则会限制能力。动态工具选择支持根据认证状态、用户权限、功能标志或对话阶段调整可用工具集。
根据工具是否提前已知,有两种方法:
当所有可能的工具在代理创建时已知时,你可以预注册它们,并根据状态、权限或上下文动态过滤哪些工具暴露给模型。 仅在达到某些对话里程碑后启用高级工具: from langchain . agents import create_agent
from langchain . agents . middleware import wrap_model_call , ModelRequest , ModelResponse
from typing import Callable
@wrap_model_call
def state_based_tools (
request : ModelRequest ,
handler : Callable [[ ModelRequest ], ModelResponse ]
) -> ModelResponse :
"""根据对话状态过滤工具。"""
# 从状态读取:检查用户是否已认证
state = request . state
is_authenticated = state . get ( "authenticated" , False )
message_count = len ( state [ " messages " ])
# 仅在认证后启用敏感工具
if not is_authenticated :
tools = [ t for t in request . tools if t . name . startswith ( "public_" )]
request = request . override ( tools = tools )
elif message_count < 5 :
# 在对话早期限制工具
tools = [ t for t in request . tools if t . name != "advanced_search" ]
request = request . override ( tools = tools )
return handler ( request )
agent = create_agent (
model = "gpt-5.4" ,
tools = [ public_search , private_search , advanced_search ],
middleware = [ state_based_tools ]
)
根据存储中的用户偏好或功能标志过滤工具: from dataclasses import dataclass
from langchain . agents import create_agent
from langchain . agents . middleware import wrap_model_call , ModelRequest , ModelResponse
from typing import Callable
from langgraph . store . memory import InMemoryStore
@dataclass
class Context :
user_id : str
@wrap_model_call
def store_based_tools (
request : ModelRequest ,
handler : Callable [[ ModelRequest ], ModelResponse ]
) -> ModelResponse :
"""根据存储偏好过滤工具。"""
user_id = request . runtime . context . user_id
# 从存储读取:获取用户启用的功能
store = request . runtime . store
feature_flags = store . get (( "features" ,), user_id )
if feature_flags :
enabled_features = feature_flags . value . get ( "enabled_tools" , [])
# 仅包含为此用户启用的工具
tools = [ t for t in request . tools if t . name in enabled_features ]
request = request . override ( tools = tools )
return handler ( request )
agent = create_agent (
model = "gpt-5.4" ,
tools = [ search_tool , analysis_tool , export_tool ],
middleware = [ store_based_tools ],
context_schema = Context ,
store = InMemoryStore ()
)
根据运行时上下文中的用户权限过滤工具: from dataclasses import dataclass
from langchain . agents import create_agent
from langchain . agents . middleware import wrap_model_call , ModelRequest , ModelResponse
from typing import Callable
@dataclass
class Context :
user_role : str
@wrap_model_call
def context_based_tools (
request : ModelRequest ,
handler : Callable [[ ModelRequest ], ModelResponse ]
) -> ModelResponse :
"""根据运行时上下文权限过滤工具。"""
# 从运行时上下文读取:获取用户角色
if request . runtime is None or request . runtime . context is None :
# 如果未提供上下文,默认为查看者(最受限)
user_role = "viewer"
else :
user_role = request . runtime . context . user_role
if user_role == "admin" :
# 管理员获得所有工具
pass
elif user_role == "editor" :
# 编辑者不能删除
tools = [ t for t in request . tools if t . name != "delete_data" ]
request = request . override ( tools = tools )
else :
# 查看者获得只读工具
tools = [ t for t in request . tools if t . name . startswith ( "read_" )]
request = request . override ( tools = tools )
return handler ( request )
agent = create_agent (
model = "gpt-5.4" ,
tools = [ read_data , write_data , delete_data ],
middleware = [ context_based_tools ],
context_schema = Context
)
此方法在以下情况下最佳:
所有可能的工具在编译/启动时已知
你想根据权限、功能标志或对话状态进行过滤
工具是静态的,但其可用性是动态的
更多示例,请参阅动态选择工具 。 当工具在运行时被发现或创建时(例如,从 MCP 服务器加载、基于用户数据生成或从远程注册表获取),你需要同时注册工具并动态处理其执行。 这需要两个中间件钩子:
wrap_model_call - 将动态工具添加到请求中
wrap_tool_call - 处理动态添加工具的执行
from langchain . tools import tool
from langchain . agents import create_agent
from langchain . agents . middleware import AgentMiddleware , ModelRequest , ToolCallRequest
# 一个将在运行时动态添加的工具
@tool
def calculate_tip ( bill_amount : float , tip_percentage : float = 20.0 ) -> str :
"""计算账单的小费金额。"""
tip = bill_amount * ( tip_percentage / 100 )
return f "Tip: $ { tip :.2f } , Total: $ { bill_amount + tip :.2f } "
class DynamicToolMiddleware ( AgentMiddleware ):
"""注册和处理动态工具的中间件。"""
def wrap_model_call ( self , request : ModelRequest , handler ):
# 将动态工具添加到请求中
# 这可以从 MCP 服务器、数据库等加载
updated = request . override ( tools = [ * request . tools , calculate_tip ])
return handler ( updated )
def wrap_tool_call ( self , request : ToolCallRequest , handler ):
# 处理动态工具的执行
if request . tool_call [ " name " ] == "calculate_tip" :
return handler ( request . override ( tool = calculate_tip ))
return handler ( request )
agent = create_agent (
model = "gpt-4o" ,
tools = [ get_weather ], # 这里只注册静态工具
middleware = [ DynamicToolMiddleware ()],
)
# 代理现在可以同时使用 get_weather 和 calculate_tip
result = agent . invoke ({
"messages" : [{ "role" : "user" , "content" : "Calculate a 20% tip on $85" }]
})
此方法在以下情况下最佳:
工具在运行时被发现(例如,从 MCP 服务器)
工具基于用户数据或配置动态生成
你正在与外部工具注册表集成
运行时注册的工具需要 wrap_tool_call 钩子,因为代理需要知道如何执行不在原始工具列表中的工具。没有它,代理将不知道如何调用动态添加的工具。
工具错误处理
要自定义工具错误的处理方式,请使用 @wrap_tool_call 装饰器创建中间件:
from langchain . agents import create_agent
from langchain . agents . middleware import wrap_tool_call
from langchain . messages import ToolMessage
@wrap_tool_call
def handle_tool_errors ( request , handler ):
"""使用自定义消息处理工具执行错误。"""
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-5.4" ,
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 =================================
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 时,代理将直接从消息中推断其任务。
system_prompt 参数接受 str 或 SystemMessage 。使用 SystemMessage 可以让你对提示结构有更多控制,这对于提供商特定功能(如 Anthropic 的提示缓存 )很有用:
from langchain . agents import create_agent
from langchain . messages import SystemMessage , HumanMessage
literary_agent = create_agent (
model = "google_genai:gemini-3.1-pro-preview" ,
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": "ephemeral"} 的 cache_control 字段告诉 Anthropic 缓存该内容块,从而减少使用相同系统提示的重复请求的延迟和成本。
动态系统提示
对于需要根据运行时上下文或代理状态修改系统提示的更高级用例,你可以使用中间件 。
@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-5.4" ,
tools = [ web_search ],
middleware = [ user_role_prompt ],
context_schema = Context
)
# 系统提示将根据上下文动态设置
result = agent . invoke (
{ "messages" : [{ "role" : "user" , "content" : "Explain machine learning" }]},
context = { "user_role" : "expert" }
)
有关消息类型和格式的更多详细信息,请参阅消息 。有关全面的中间件文档,请参阅中间件 。
为代理设置一个可选的 name 。当将代理作为子图添加到多代理系统 中时,这用作节点标识符:
agent = create_agent (
model ,
tools ,
name = "research_assistant"
)
代理名称建议使用 snake_case(例如,research_assistant 而不是 Research Assistant)。一些模型提供商会拒绝包含空格或特殊字符的名称并报错。仅使用字母数字字符、下划线和连字符可确保与所有提供商的兼容性。工具名称 同样适用。
你可以通过向代理的 State 传递更新来调用代理。所有代理在其状态中都包含一个消息序列 ;要调用代理,请传递一条新消息:
result = agent . invoke (
{ "messages" : [{ "role" : "user" , "content" : "What's the weather in San Francisco?" }]}
)
有关从代理流式传输步骤和/或令牌的信息,请参阅流式传输 指南。
否则,代理遵循 LangGraph Graph API 并支持所有相关方法,例如 stream 和 invoke。
高级概念
结构化输出
在某些情况下,你可能希望代理以特定格式返回输出。LangChain 通过 response_format 参数提供了结构化输出的策略。
ToolStrategy 使用人工工具调用来生成结构化输出。这适用于任何支持工具调用的模型。当提供商原生结构化输出(通过 ProviderStrategy )不可用或不可靠时,应使用 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-5.4-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 使用模型提供商的原生结构化输出生成。这更可靠,但仅适用于支持原生结构化输出的提供商:
from langchain . agents . structured_output import ProviderStrategy
agent = create_agent (
model = "gpt-5.4" ,
response_format = ProviderStrategy ( ContactInfo )
)
从 langchain 1.0 开始,简单地传递一个模式(例如,response_format=ContactInfo)如果模型支持原生结构化输出,将默认使用 ProviderStrategy。否则将回退到 ToolStrategy。
代理通过消息状态自动维护对话历史记录。你还可以配置代理使用自定义状态模式,以在对话期间记住额外信息。
存储在状态中的信息可以被视为代理的短期记忆 :
自定义状态模式必须扩展 AgentState 作为 TypedDict。
有两种方法定义自定义状态:
通过中间件 (首选)
通过 create_agent 上的 state_schema
通过中间件定义状态
当你的自定义状态需要被特定的中间件钩子和附加到该中间件的工具访问时,使用中间件定义自定义状态。
from langchain . agents import AgentState
from langchain . agents . middleware import AgentMiddleware
from typing import Any
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 模型和 dataclasses。有关更多详细信息,请参阅 v1 迁移指南 。
要了解更多关于记忆的信息,请参阅记忆 。有关实现跨会话持久化的长期记忆的信息,请参阅长期记忆 。
流式传输
我们已经看到如何使用 invoke 调用代理以获取最终响应。如果代理执行多个步骤,这可能需要一段时间。为了显示中间进度,我们可以在消息发生时将其流式传回。
from langchain . messages import AIMessage , HumanMessage
for chunk in agent . stream ({
"messages" : [{ "role" : "user" , "content" : "Search for AI news and summarize the findings" }]
}, stream_mode = "values" ):
# 每个块包含该点的完整状态
latest_message = chunk [ " messages " ][ - 1 ]
if latest_message . content :
if isinstance ( latest_message , HumanMessage ):
print ( f "User: { latest_message . content } " )
elif isinstance ( latest_message , AIMessage ):
print ( f "Agent: { latest_message . content } " )
elif latest_message . tool_calls :
print ( f "Calling tools: { [ tc [ ' name ' ] for tc in latest_message . tool_calls ] } " )
中间件
中间件 为在执行的不同阶段自定义代理行为提供了强大的可扩展性。你可以使用中间件来:
在调用模型之前处理状态(例如,消息修剪、上下文注入)
修改或验证模型的响应(例如,护栏、内容过滤)
使用自定义逻辑处理工具执行错误
基于状态或上下文实现动态模型选择
添加自定义日志记录、监控或分析
中间件无缝集成到代理的执行中,允许你在关键点拦截和修改数据流,而无需更改核心代理逻辑。
将这些文档连接 到 Claude、VSCode 等,通过 MCP 获取实时答案。