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) 步骤。

这个循环的工作流程如下:

  1. 思考 (Thought): 模型根据当前任务进行推理,分析问题并制定下一步计划。
  2. 行动 (Action): 模型决定调用一个或多个工具来执行计划。
  3. 观察 (Observation): 模型接收并整合工具执行返回的结果。
  4. 重复: 模型基于新的观察结果进行下一轮的思考,直到任务完成。

ReAct 框架极大地减少了模型的“幻觉”现象,并使得整个决策过程变得可审计、可追溯。开发者可以清晰地看到智能体是如何形成假设(思考),如何通过工具验证假设(行动),以及如何根据反馈更新其计划(观察)的。

以下是 ReAct 循环的简化流程图:

行动 (Action)
观察 (Observation)
完成 (Finish)
用户查询
思考 (Thought)
工具 (Tools)
最终答案

值得注意的是,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 的执行流程:

  1. 第一次 思考 -> 行动:

    • 推理: “‘最受欢迎’是具有时效性的,我需要使用 search_products 工具来查找最新信息。”
    • 行动: 调用 search_products("wireless headphones")
    ================================== Ai Message ==================================
    Tool Calls:
       search_products (call_abc123)
     Call ID: call_abc123
       Args:
         query: wireless headphones
    
  2. 第一次 观察:

    • 工具返回结果。
    ================================= Tool Message =================================
    
    找到 5 款匹配 "wireless headphones" 的产品。排名前5的结果是:WH-1000XM5, ...
    
  3. 第二次 思考 -> 行动:

    • 推理: “我已经找到了排名第一的型号,现在我需要使用 check_inventory 工具来确认它的库存情况。”
    • 行动: 调用 check_inventory("WH-1000XM5")
    ================================== Ai Message ==================================
    Tool Calls:
       check_inventory (call_def456)
     Call ID: call_def456
       Args:
         product_id: WH-1000XM5
    
  4. 第二次 观察:

    • 工具返回库存信息。
    ================================= Tool Message =================================
    
    产品 WH-1000XM5: 库存 10 件
    
  5. 最终 思考 -> 完成:

    • 推理: “我已经获得了最受欢迎的型号及其库存状态,现在可以回答用户的问题了。”
    • 行动: 生成最终答案。
    ================================== 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 是一个可选节点,它在模型被调用之前对状态进行处理。常见的用例包括:消息修剪、对话摘要、上下文注入等。

__start__
pre_model_hook
model
post_model_hook
__end__
tools

这个钩子必须是一个可调用或 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 在模型响应之后、工具执行之前运行。它用于验证、安全护栏或其他后处理逻辑。

__start__
pre_model_hook
model
post_model_hook
__end__
tools

下面是一个过滤掉机密信息的例子:

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 及其相关概念。通过对这份官方文档信息的重构和深化,我们可以总结出以下关键点:

  1. 核心思想: LangChain Agent 的基础是 ReAct 框架,它通过 思考 -> 行动 -> 观察 的循环,将大型语言模型的推理能力与外部工具的执行能力有机结合,实现了任务的自主规划和执行。
  2. 三大基石: 构建一个 Agent 离不开 模型 (Model)工具 (Tools)提示 (Prompt) 这三大核心组件。LangChain 提供了从简单到复杂的多种配置方式,如静态/动态模型选择、列表/ToolNode 式的工具提供、以及字符串/对象/函数式的提示定义,满足不同场景的需求。
  3. 强大的可扩展性: create_agent 的强大之处在于其高度的可扩展性。通过 结构化输出自定义记忆模型调用前后钩子 (Hooks) 以及 中间件 (Middleware) 等高级功能,开发者可以精细地控制 Agent 行为的每一个环节,实现复杂的业务逻辑和安全护栏。
  4. 生产就绪的特性: LangChain Agent 考虑到了生产环境的需求,内置了并行工具调用、错误处理、状态管理等机制。同时,流式输出 (Streaming) 功能使得构建实时、交互式的 Agent 应用成为可能。
  5. 底层引擎: 所有这些功能都构建在 LangGraph 之上,其基于图的执行模型为 Agent 的流程定义和调试提供了极大的灵活性和可见性。

掌握了 create_agent 的使用方法,就等于掌握了开启构建强大、可靠和可控的 AI 智能体大门的钥匙。希望本文能帮助你更好地理解和运用 LangChain Agent,在自己的项目中创造出色的 AI 应用。


Logo

更多推荐