【LangChain 核心组件指南 | Agent篇】从零到精通:深度解析 create_agent 与 ReAct 智能体构建
本文深度剖析了 LangChain v1-alpha 版本中的核心组件——Agent,旨在为开发者提供一份详实、专业且易于理解的技术指南。文章以 `create_agent()` 工厂函数为核心,系统性地阐述了构建生产级 ReAct (推理+行动) 智能体的完整流程。内容从 Agent 的基本概念与 ReAct 工作原理入手,逐步深入到模型 (Model)、工具 (Tools)、提示 (Promp
Langchain系列文章目录
01-玩转LangChain:从模型调用到Prompt模板与输出解析的完整指南
02-玩转 LangChain Memory 模块:四种记忆类型详解及应用场景全覆盖
03-全面掌握 LangChain:从核心链条构建到动态任务分配的实战指南
04-玩转 LangChain:从文档加载到高效问答系统构建的全程实战
05-玩转 LangChain:深度评估问答系统的三种高效方法(示例生成、手动评估与LLM辅助评估)
06-从 0 到 1 掌握 LangChain Agents:自定义工具 + LLM 打造智能工作流!
07-【深度解析】从GPT-1到GPT-4:ChatGPT背后的核心原理全揭秘
08-【万字长文】MCP深度解析:打通AI与世界的“USB-C”,模型上下文协议原理、实践与未来
Python系列文章目录
PyTorch系列文章目录
机器学习系列文章目录
深度学习系列文章目录
Java系列文章目录
JavaScript系列文章目录
Llamaindex系列文章目录
01-【LlamaIndex核心组件指南 | 模型篇】一文通晓 LlamaIndex 模型层:LLM、Embedding 及多模态应用全景解析
02-【LlamaIndex核心组件指南 | Prompt篇】深度解析LlamaIndex提示模板的设计与实战
03-【LlamaIndex核心组件指南 | 数据加载篇】从原始数据到向量的全链路深度解析
文章目录
摘要
本文深度剖析了 LangChain v1-alpha 版本中的核心组件——Agent,旨在为开发者提供一份详实、专业且易于理解的技术指南。文章以 create_agent()
工厂函数为核心,系统性地阐述了构建生产级 ReAct (推理+行动) 智能体的完整流程。内容从 Agent 的基本概念与 ReAct 工作原理入手,逐步深入到模型 (Model)、工具 (Tools)、提示 (Prompt) 三大核心组件的静态与动态配置方法。此外,本文还详细介绍了结构化输出、自定义记忆、模型调用前后钩子 (Hooks) 以及流式响应等高级配置,并通过丰富的代码示例和 Mermaid 流程图,帮助读者全面掌握 LangChain Agent 的构建技巧与最佳实践。无论您是初学者还是经验丰富的开发者,本文都将为您构建更强大、更可控的 AI 智能体提供坚实的理论基础和实践指导。
一、 LangChain Agent 导论
在大型语言模型(LLM)的应用开发中,我们常常希望模型不仅能进行对话,还能执行具体任务,例如查询数据库、调用 API 或与文件系统交互。LangChain Agent 正是为此而生。它是一种能够将语言模型与外部工具(Tools)相结合的智能系统,使其具备推理任务、决策使用何种工具、并迭代地朝着最终解决方案迈进的能力。
简单来说,Agent 赋予了 LLM “手”和“脚”,让它能够超越纯文本生成,主动与外部世界互动,完成更复杂的任务。
二、 核心原理:ReAct - 协同推理与行动
LangChain 中 create_agent()
函数提供的是一种生产就绪的 ReAct 智能体实现。ReAct 框架源自论文 [ReAct: Synergizing Reasoning and Acting in Language Models],其核心思想是将智能体的行为构建为一系列交错的 思考 (Thought)
-> 行动 (Action)
-> 观察 (Observation)
步骤。
这个循环的工作流程如下:
- 思考 (Thought): 模型根据当前任务进行推理,分析问题并制定下一步计划。
- 行动 (Action): 模型决定调用一个或多个工具来执行计划。
- 观察 (Observation): 模型接收并整合工具执行返回的结果。
- 重复: 模型基于新的观察结果进行下一轮的思考,直到任务完成。
ReAct 框架极大地减少了模型的“幻觉”现象,并使得整个决策过程变得可审计、可追溯。开发者可以清晰地看到智能体是如何形成假设(思考),如何通过工具验证假设(行动),以及如何根据反馈更新其计划(观察)的。
以下是 ReAct 循环的简化流程图:
值得注意的是,create_agent()
构建的 Agent 运行时是基于 LangGraph 的。LangGraph 将 Agent 的执行流程抽象为一个由节点(步骤)和边(连接)组成的图。Agent 在这个图中移动,执行模型节点、工具节点或钩子节点等,从而实现了高度灵活和可定制的执行逻辑。
三、 深度剖析:Agent 的三大核心组件
要构建一个功能完备的 Agent,我们需要理解并配置其三大核心组件:模型(Model)、工具(Tools)和提示(Prompt)。
3.1 模型 (Model):Agent 的“大脑”
模型是 Agent 的推理引擎,负责思考和决策。create_agent()
支持静态和动态两种模型配置方式。
3.1.1 静态模型配置
静态模型在创建 Agent 时一次性配置,并在整个执行过程中保持不变。这是最常见和直接的方法。
(1) 使用模型标识符字符串
最简单的方式是提供一个格式为 provider:model
的字符串。LangChain 会自动推断提供商。
from langchain.agents import create_agent
from langchain_core.tools import tool
# 假设 tools 已经被定义
@tool
def search(query: str) -> str:
"""Search for information."""
return f"Results for: {query}"
tools = [search]
# 使用字符串标识符创建 agent
# "gpt-4o" 会被自动推断为 "openai:gpt-4o"
agent = create_agent(
"openai:gpt-4o",
tools=tools
)
(2) 直接实例化模型类
为了对模型进行更精细的控制(如设置温度、超时、API密钥等),可以直接从提供商的包中实例化一个模型对象。
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# 假设 tools 已经被定义
@tool
def search(query: str) -> str:
"""Search for information."""
return f"Results for: {query}"
tools = [search]
# 实例化模型并进行详细配置
model = ChatOpenAI(
model="gpt-4o",
temperature=0.1,
max_tokens=1000,
timeout=30
)
# 将模型实例传递给 agent
agent = create_agent(model, tools=tools)
这种方式给予了开发者对模型参数的完全控制权,推荐在需要定制化配置的生产环境中使用。
3.1.2 动态模型路由
在某些高级场景下,我们希望 Agent 能在运行时根据当前状态(例如对话的复杂度)动态选择使用哪个模型,以实现成本优化或性能路由。这可以通过向 create_agent
传递一个函数来实现。
该函数接收图状态(AgentState
)和运行时上下文(Runtime
),并返回一个绑定了工具的 BaseChatModel
实例。
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent, AgentState
from langgraph.runtime import Runtime
from langchain_core.tools import tool
# 假设 tools 已经被定义
@tool
def search(query: str) -> str:
"""Search for information."""
return f"Results for: {query}"
tools = [search]
def select_model(state: AgentState, runtime: Runtime) -> ChatOpenAI:
"""根据对话的复杂性选择模型。"""
messages = state["messages"]
message_count = len(messages)
if message_count < 10:
# 对于简短对话,使用更经济的模型
print("--- 使用 gpt-4o-mini ---")
return ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
else:
# 对于长对话,使用更强大的模型
print("--- 使用 gpt-4o ---")
return ChatOpenAI(model="gpt-4o").bind_tools(tools)
# 将模型选择函数传递给 agent
agent = create_agent(select_model, tools=tools)
3.2 工具 (Tools):Agent 的“双手”
工具赋予了 Agent 执行具体行动的能力。LangChain 的 Agent 不仅仅是简单地将工具绑定到模型上,它还提供了更强大的功能:
- 多工具调用: 在单次提示后触发一系列连续的工具调用。
- 并行工具调用: 在适当时机并行执行多个工具。
- 动态工具选择: 根据前一个工具的执行结果来决定下一步调用哪个工具。
- 工具重试与错误处理: 内置的逻辑来处理工具执行失败的情况。
- 状态持久化: 在多次工具调用之间保持状态信息。
你可以通过两种主要方式向 Agent 提供工具。
3.2.1 传递一个工具列表
这是最简单直接的方式。你可以将一个由 LangChain 的 @tool
装饰器创建的函数、任何可调用对象或代表内置工具的字典组成的列表传递给 tools
参数。create_agent
会在内部自动创建一个 ToolNode
。
from langchain_core.tools import tool
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o")
@tool
def search(query: str) -> str:
"""用于搜索信息。"""
return f"关于 '{query}' 的搜索结果..."
@tool
def calculate(expression: str) -> str:
"""用于执行数学计算。"""
# 注意:在生产环境中使用 eval 是不安全的,这里仅为示例
return str(eval(expression))
# 将工具列表直接传递给 agent
agent = create_agent(model, tools=[search, calculate])
# 如果提供一个空列表,Agent 将只包含一个 LLM 节点,不具备工具调用能力
# agent_no_tools = create_agent(model, tools=[])
3.2.2 传递一个已配置的 ToolNode
如果你需要自定义工具节点的行为,例如统一处理工具执行错误,可以直接创建一个 ToolNode
实例并传递给 tools
参数。
from langchain_core.tools import tool
from langchain.agents import create_agent, ToolNode
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o")
# 假设 search 和 calculate 工具已定义
@tool
def search(query: str) -> str:
"""用于搜索信息。"""
return f"关于 '{query}' 的搜索结果..."
@tool
def calculate(expression: str) -> str:
"""用于执行数学计算。"""
# 注意:在生产环境中使用 eval 是不安全的,这里仅为示例
# 模拟一个错误
if "error" in expression:
raise ValueError("Invalid expression")
return str(eval(expression))
# 创建并配置 ToolNode,提供自定义错误处理消息
tool_node = ToolNode(
tools=[search, calculate],
handle_tool_errors="工具执行失败,请检查您的输入并重试。"
)
agent = create_agent(model, tools=tool_node)
# 调用 agent,并传入一个会导致工具出错的输入
result = agent.invoke({"messages": [{"role": "user", "content": "计算 'error'"}]})
# 打印最后几条消息,可以看到自定义的错误 ToolMessage
print(result["messages"][-2:])
当 ToolNode
内部发生错误时,Agent 不会崩溃,而是会向模型返回一个包含自定义错误信息的 ToolMessage
,让模型知道发生了什么并进行下一步决策。
# 上述代码的输出可能包含类似这样的 ToolMessage
[
...,
ToolMessage(content='工具执行失败,请检查您的输入并重试。', tool_call_id='...'),
...
]
3.2.3 ReAct 循环中的工具使用实例
让我们通过一个具体的例子,看看 Agent 是如何在 ReAct 循环中利用工具的。
用户提示: “查找目前最受欢迎的无线耳机,并确认是否有库存。”
Agent 的执行流程:
-
第一次
思考
->行动
:- 推理: “‘最受欢迎’是具有时效性的,我需要使用
search_products
工具来查找最新信息。” - 行动: 调用
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
工具来确认它的库存情况。” - 行动: 调用
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 件库存。
3.3 提示 (Prompt):Agent 的“指令”
通过提供一个 Prompt,你可以塑造 Agent 处理任务的方式和“性格”。prompt
参数可以是一个字符串、一个 SystemMessage
对象,或者一个可调用对象。
(1) 作为字符串
最简单的方式,直接提供一个系统消息字符串。
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o")
tools = [] # 假设 tools 已定义
agent = create_agent(
model,
tools,
prompt="你是一个乐于助人的助手。回答应简洁而准确。"
)
(2) 作为 SystemMessage 对象
使用 SystemMessage
对象,结构更清晰。
from langchain.agents import create_agent
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o")
tools = [] # 假设 tools 已定义
agent = create_agent(
model,
tools,
prompt=SystemMessage(content="你是一名研究助理。请在回答时引用你的信息来源。")
)
(3) 作为可调用对象 (Callable)
对于需要根据当前状态动态生成 Prompt 的复杂场景,可以提供一个函数。
from langchain.agents import create_agent, AgentState
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o")
tools = [] # 假设 tools 已定义
def dynamic_prompt(state: AgentState):
# 从状态中获取用户类型,默认为 'standard'
user_type = state.get("user_type", "standard")
if user_type == "expert":
system_msg = SystemMessage(content="请提供详细的技术性回复。")
else:
system_msg = SystemMessage(content="请提供简单、清晰的解释。")
# 将动态生成的系统消息与历史消息结合
return [system_msg] + state["messages"]
agent = create_agent(model, tools, prompt=dynamic_prompt)
注意:如果不提供 prompt
参数,Agent 将直接根据 messages
中的内容推断其任务。
3.3.1 使用中间件实现高级动态 Prompt
对于需要根据运行时上下文(而不仅仅是 Agent 状态)来修改系统提示的更高级用例,可以使用 @modify_model_request
装饰器创建一个简单的自定义中间件。这对于根据用户角色、会话上下文或其他动态因素来个性化提示特别有用。
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware.types import modify_model_request, ModelRequest
from langgraph.runtime import Runtime
from typing import TypedDict
from langchain_core.tools import tool
tools = [] # 假设 tools 已定义
class Context(TypedDict):
user_role: str
@modify_model_request
def dynamic_system_prompt(state: AgentState, request: ModelRequest, runtime: Runtime[Context]) -> ModelRequest:
# 从运行时上下文中获取 user_role
user_role = runtime.context.get("user_role", "user")
base_prompt = "你是一个乐于助人的助手。"
if user_role == "expert":
prompt = f"{base_prompt} 请提供详细的技术性回复。"
elif user_role == "beginner":
prompt = f"{base_prompt} 请用简单的语言解释概念,避免使用专业术语。"
else:
prompt = base_prompt
# 修改即将发送给模型的请求中的 system_prompt
request.system_prompt = prompt
return request
agent = create_agent(
model="openai:gpt-4o",
tools=tools,
middleware=[dynamic_system_prompt],
)
# 在调用时传入 context,系统提示将根据 user_role 动态设置
result_expert = agent.invoke(
{"messages": [{"role": "user", "content": "解释一下机器学习"}]},
{"context": {"user_role": "expert"}}
)
result_beginner = agent.invoke(
{"messages": [{"role": "user", "content": "解释一下机器学习"}]},
{"context": {"user_role": "beginner"}}
)
四、 进阶配置:解锁 Agent 的高级能力
除了核心组件,create_agent
还提供了一系列高级配置选项,让你的 Agent 更加强大和可控。
4.1 结构化输出 (Structured Output)
有时我们希望 Agent 的最终输出是特定格式的 JSON 或对象,而不是自由文本。通过 response_format
参数,可以轻松实现这一点。你只需提供一个 Pydantic BaseModel
类即可。
from pydantic import BaseModel
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
class ContactInfo(BaseModel):
name: str
email: str
phone: str
model = ChatOpenAI(model="gpt-4o")
# 假设 search_tool 已定义
tools = []
agent = create_agent(
model,
tools=tools,
response_format=ContactInfo
)
result = agent.invoke({
"messages": [{"role": "user", "content": "从这段文本中提取联系人信息:John Doe, john@example.com, (555) 123-4567"}]
})
# 最终结果中会包含一个名为 structured_response 的字段
print(result["structured_response"])
# 输出: ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')
4.2 自定义记忆 (Memory)
Agent 默认通过消息状态自动维护对话历史(短期记忆)。如果你想在对话中跟踪额外的信息(例如用户偏好、会话ID等),可以定义一个自定义的状态模式(State Schema)。
from typing import TypedDict, List
from typing_extensions import Annotated
from langgraph.graph.message import add_messages
from langchain.agents import create_agent, AgentState
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o")
tools = [] # 假设 tools 已定义
# 继承自 AgentState 并添加自定义字段
class CustomAgentState(AgentState):
messages: Annotated[list, add_messages]
user_preferences: dict
agent = create_agent(
model,
tools=tools,
state_schema=CustomAgentState
)
# 现在 Agent 可以跟踪除 messages 之外的自定义状态
result = agent.invoke({
"messages": [{"role": "user", "content": "我更喜欢技术性的解释"}],
"user_preferences": {"style": "technical", "verbosity": "detailed"},
})
# 这个 user_preferences 状态可以在整个对话流程中被访问和更新
print(agent.get_state(result['thread_id']).values['user_preferences'])
# 输出: {'style': 'technical', 'verbosity': 'detailed'}
4.3 模型调用前钩子 (Pre-model Hook)
Pre-model Hook 是一个可选节点,它在模型被调用之前对状态进行处理。常见的用例包括:消息修剪、对话摘要、上下文注入等。
这个钩子必须是一个可调用或 Runnable 对象,接收当前图状态并返回一个状态更新字典。下面是一个修剪消息以适应上下文窗口的例子:
from langchain_core.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o")
tools = [] # 假设 tools 已定义
def trim_messages(state):
"""仅保留最后几条消息以适应上下文窗口。"""
messages = state["messages"]
# 如果消息不多,则不作处理
if len(messages) <= 5:
return {} # 返回空字典表示不修改状态
# 保留第一条(通常是系统消息)和最后4条消息
first_msg = messages[0]
recent_messages = messages[-4:]
new_messages = [first_msg] + recent_messages
# 重要:必须先移除所有旧消息,再添加新消息
return {
"messages": [
RemoveMessage(id=REMOVE_ALL_MESSAGES),
*new_messages
]
}
agent = create_agent(
model,
tools=tools,
pre_model_hook=trim_messages
)
警告:当你在 Pre-model Hook 中返回 messages
时,应通过 RemoveMessage(id=REMOVE_ALL_MESSAGES)
来覆盖整个 messages
列表,以避免消息重复或冲突。
4.4 模型调用后钩子 (Post-model Hook)
与 Pre-model Hook 相对,Post-model Hook 在模型响应之后、工具执行之前运行。它用于验证、安全护栏或其他后处理逻辑。
下面是一个过滤掉机密信息的例子:
from langchain_core.messages import AIMessage, RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o")
tools = [] # 假设 tools 已定义
def validate_response(state):
"""检查模型响应是否违反了策略。"""
messages = state["messages"]
last_message = messages[-1]
if "confidential" in last_message.content.lower():
# 如果检测到敏感词,替换掉模型的原始回复
return {
"messages": [
RemoveMessage(id=REMOVE_ALL_MESSAGES), # 清空所有消息
*messages[:-1], # 添加除最后一条外的所有历史消息
AIMessage(content="我不能分享机密信息。") # 添加一条安全回复
]
}
return {} # 不做任何修改
agent = create_agent(
model,
tools=tools,
post_model_hook=validate_response
)
4.5 流式输出 (Streaming)
对于执行多个步骤的复杂 Agent,一次性等待最终结果可能会耗时很长。为了提供更好的用户体验,我们可以使用 .stream()
方法来流式地获取中间过程的每一步状态。
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
@tool
def search(query: str) -> str:
"""用于搜索信息。"""
import time
time.sleep(2) # 模拟网络延迟
return f"关于 '{query}' 的最新新闻..."
model = ChatOpenAI(model="gpt-4o")
tools = [search]
agent = create_agent(model, tools=tools)
# 使用 stream_mode="values" 来获取每个步骤的完整状态
for chunk in agent.stream({
"messages": [{"role": "user", "content": "搜索关于AI的最新新闻并总结发现"}]
}, stream_mode="values"):
# 每个 chunk 都包含那个时间点的完整状态
latest_message = chunk["messages"][-1]
# 打印模型的思考或最终回答
if latest_message.content:
print(f"Agent: {latest_message.content}")
# 打印工具调用信息
elif latest_message.tool_calls:
tool_names = [tc['name'] for tc in latest_message.tool_calls]
print(f"Agent 正在调用工具: {tool_names}")
通过处理流式输出,你可以在前端实时展示 Agent 的思考过程和行动步骤,极大地提升了应用的交互性。
五、 总结
本文详细介绍了 LangChain 中构建 Agent 的核心函数 create_agent
及其相关概念。通过对这份官方文档信息的重构和深化,我们可以总结出以下关键点:
- 核心思想: LangChain Agent 的基础是 ReAct 框架,它通过
思考 -> 行动 -> 观察
的循环,将大型语言模型的推理能力与外部工具的执行能力有机结合,实现了任务的自主规划和执行。 - 三大基石: 构建一个 Agent 离不开 模型 (Model)、工具 (Tools) 和 提示 (Prompt) 这三大核心组件。LangChain 提供了从简单到复杂的多种配置方式,如静态/动态模型选择、列表/
ToolNode
式的工具提供、以及字符串/对象/函数式的提示定义,满足不同场景的需求。 - 强大的可扩展性:
create_agent
的强大之处在于其高度的可扩展性。通过 结构化输出、自定义记忆、模型调用前后钩子 (Hooks) 以及 中间件 (Middleware) 等高级功能,开发者可以精细地控制 Agent 行为的每一个环节,实现复杂的业务逻辑和安全护栏。 - 生产就绪的特性: LangChain Agent 考虑到了生产环境的需求,内置了并行工具调用、错误处理、状态管理等机制。同时,流式输出 (Streaming) 功能使得构建实时、交互式的 Agent 应用成为可能。
- 底层引擎: 所有这些功能都构建在 LangGraph 之上,其基于图的执行模型为 Agent 的流程定义和调试提供了极大的灵活性和可见性。
掌握了 create_agent
的使用方法,就等于掌握了开启构建强大、可靠和可控的 AI 智能体大门的钥匙。希望本文能帮助你更好地理解和运用 LangChain Agent,在自己的项目中创造出色的 AI 应用。
更多推荐
所有评论(0)