跳到主要内容

Agents 代理

Agents（代理）是 LangChain 最强大的功能，它让 LLM 能够自主决策、调用工具、循环执行，直到完成复杂任务。LangChain 1.0 提供了 create_agent 函数来构建智能代理。

什么是 Agent？​

Agent（代理） 是一个能够"思考-行动-观察"的智能循环系统。与固定的 Chain 不同，Agent 根据任务动态决定执行路径。

Agent vs Chain​

特性ChainAgent
执行顺序预定义的固定流程LLM 动态决策
工具使用不支持或硬编码自动选择和调用
灵活性适合标准化任务适合开放式复杂任务
循环执行不支持支持多轮思考行动

举个例子：

  • Chain：像工厂流水线，每个工位做固定动作
  • Agent：像智能秘书，接到任务后自己规划怎么完成

Agent 的工作原理​

Agent 遵循 ReAct 架构（Reasoning + Acting）：

用户提问 → LLM 思考 → 决定行动 → 执行工具 → 观察结果 →
→ 继续思考 → ... → 给出最终答案

例如用户问"阿里巴巴的创始人是谁，他创办了哪些公司？"：

  1. 思考：需要搜索阿里巴巴创始人信息
  2. 行动：调用搜索工具
  3. 观察：得到结果——马云
  4. 思考：还需要查他创办的其他公司
  5. 行动：再次搜索
  6. 观察：得到淘宝、支付宝等信息
  7. 思考：信息足够，可以回答
  8. 最终答案：整理输出结果

LangChain 与 LangGraph​

在 LangChain 生态中，Agent 的构建主要通过 LangGraph 实现：

  • LangChain：提供模型、工具、提示词等基础组件
  • LangGraph：提供 Agent 的编排框架，支持状态管理、持久化、人工介入等高级功能

LangChain 1.0 将 Agent 创建函数统一到 langchain.agents 模块，简化了 API。

安装依赖​

pip install langchain langgraph langchain-openai

快速开始​

第一个 Agent​

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool

# 1. 定义工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    weather_data = {
        "北京": "晴天，25°C",
        "上海": "多云，22°C",
        "广州": "小雨，28°C"
    }
    return weather_data.get(city, f"没有{city}的天气数据")

@tool
def calculate(expression: str) -> str:
    """执行数学计算，如 '2 + 3 * 4'"""
    try:
        allowed_chars = set('0123456789+-*/.() ')
        if not all(c in allowed_chars for c in expression):
            return "错误：表达式包含非法字符"
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{e}"

# 2. 初始化模型
model = init_chat_model("openai:gpt-4o-mini")

# 3. 创建 Agent
agent = create_agent(
    model=model,
    tools=[get_weather, calculate],
    system_prompt="你是一个智能助手，可以查询天气和进行计算。"
)

# 4. 调用 Agent
result = agent.invoke({
    "messages": [{"role": "user", "content": "北京今天天气怎么样？25度适合穿什么衣服？"}]
})

# 输出最终回答
print(result["messages"][-1].content)

执行过程：

  • Agent 识别需要查询天气
  • 调用 get_weather 工具获取北京天气
  • 根据返回的 25°C 给出穿衣建议
  • 整个过程完全自动

流式输出​

# 流式查看 Agent 执行过程
for event in agent.stream({
    "messages": [{"role": "user", "content": "北京和上海天气对比"}]
}):
    # 注意：节点名是 "model" 而不是 "agent"
    if "model" in event:
        for msg in event["model"]["messages"]:
            if hasattr(msg, "content") and msg.content:
                print(f"[Model]: {msg.content[:100]}...")
    elif "tools" in event:
        for msg in event["tools"]["messages"]:
            print(f"[Tools]: {msg.content[:100]}...")

使用消息对象​

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

model = init_chat_model("openai:gpt-4o-mini")
agent = create_agent(model=model, tools=[get_weather, calculate])

# 使用消息对象
result = agent.invoke({
    "messages": [HumanMessage(content="计算 123 * 456")]
})
print(result["messages"][-1].content)

create_agent 参数​

核心参数​

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4o-mini")

agent = create_agent(
    model=model,                        # 聊天模型（必需）
    tools=[tool1, tool2],               # 工具列表（必需）
    
    # 可选参数
    system_prompt=None,                 # 系统提示词（注意参数名变化）
    checkpointer=None,                  # 检查点存储器（用于记忆）
    store=None,                         # 长期存储
    state_schema=None,                  # 自定义状态（仅支持 TypedDict）
    middleware=None,                    # 中间件列表
    response_format=None,               # 结构化输出格式
    interrupt_before=None,              # 执行前中断的节点
    interrupt_after=None,               # 执行后中断的节点
)

system_prompt 参数​

LangChain 1.0 将 prompt 参数重命名为 system_prompt：

agent = create_agent(
    model=model,
    tools=[get_weather],
    
    # 字符串形式的系统提示
    system_prompt="你是一个专业的天气助手，回答要简洁准确。"
)

# 或使用 SystemMessage
from langchain.messages import SystemMessage

agent = create_agent(
    model=model,
    tools=[get_weather],
    system_prompt=SystemMessage(content="你是天气助手...")
)

从 create_react_agent 迁移​

如果你之前使用 langgraph.prebuilt.create_react_agent，以下是迁移要点：

变更项旧 API (create_react_agent)新 API (create_agent)
导入路径langgraph.prebuiltlangchain.agents
系统提示词参数promptsystem_prompt
预模型钩子pre_model_hookmiddleware 的 before_model 方法
后模型钩子post_model_hookmiddleware 的 after_model 方法
自定义状态支持 Pydantic/Dataclass仅支持 TypedDict
流式输出节点名"agent""model"
运行时上下文config["configurable"]context 参数

导入路径变化​

# 旧代码
from langgraph.prebuilt import create_react_agent

# 新代码
from langchain.agents import create_agent

参数名变化​

# 旧代码
agent = create_react_agent(
    model=model,
    tools=tools,
    prompt="你是一个助手"  # 旧参数名
)

# 新代码
agent = create_agent(
    model=model,
    tools=tools,
    system_prompt="你是一个助手"  # 新参数名
)

流式输出变化​

# 旧代码：节点名是 "agent"
for event in agent.stream(input):
    if "agent" in event:
        # ...

# 新代码：节点名是 "model"
for event in agent.stream(input):
    if "model" in event:
        # ...

工具详解​

工具是 Agent 与外部世界交互的桥梁。好的工具定义直接影响 Agent 的表现。

工具定义最佳实践​

from langchain.tools import tool
from pydantic import BaseModel, Field

# 定义输入模型
class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    max_results: int = Field(default=5, description="最大返回数量")

@tool(args_schema=SearchInput)
def search_products(query: str, max_results: int = 5) -> str:
    """
    搜索商品信息。
    当用户想要购买某类商品或询问商品推荐时使用此工具。
    """
    return f"搜索 '{query}'，返回 {max_results} 个结果"

# 不好的定义示例（不要这样写）
@tool
def search(q: str) -> str:
    """搜索"""  # 描述太模糊，LLM 无法判断何时使用
    return "..."

为什么描述很重要？

LLM 根据工具名称和描述决定是否调用。描述越详细，LLM 越能做出正确选择。

多参数工具​

@tool
def search_hotels(
    location: str,
    check_in: str,
    check_out: str,
    price_level: str = "all"
) -> str:
    """搜索酒店。
    
    当用户需要预订酒店或查询酒店信息时使用此工具。
    
    参数:
        location: 城市或地区名称，如"北京"、"上海"
        check_in: 入住日期，格式"YYYY-MM-DD"
        check_out: 退房日期，格式"YYYY-MM-DD"
        price_level: 价格等级 (budget/mid-range/luxury)，默认为 all
    """
    if price_level == "budget":
        return f"{location}的经济型酒店：如家、汉庭"
    elif price_level == "luxury":
        return f"{location}的豪华酒店：四季、丽思卡尔顿"
    else:
        return f"{location}的中档酒店：全季、亚朵"

错误处理​

工具应该优雅地处理错误，返回字符串而非抛出异常：

@tool
def safe_divide(a: float, b: float) -> str:
    """安全除法，参数为两个数字。"""
    if b == 0:
        return "错误：除数不能为零"
    try:
        result = a / b
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

使用 Middleware 处理工具错误​

LangChain 1.0 使用 middleware 来处理工具错误：

from langchain.agents import create_agent, Middleware
from langchain.tools import tool

@tool
def risky_tool(param: str) -> str:
    """可能失败的工具"""
    if param == "bad":
        raise ValueError("无效参数")
    return f"成功：{param}"

class ErrorHandlingMiddleware(Middleware):
    """错误处理中间件"""
    
    def wrap_tool_call(self, tool_call, state, context):
        """包装工具调用，处理错误"""
        try:
            result = yield tool_call
            return result
        except Exception as e:
            return f"操作失败：{str(e)}。请重试。"

agent = create_agent(
    model=model,
    tools=[risky_tool],
    middleware=[ErrorHandlingMiddleware()]
)

记忆功能​

Agent 默认是无状态的。添加记忆需要使用检查点（checkpoint）进行状态持久化。

内存记忆​

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4o-mini")

# 创建检查点存储
checkpointer = InMemorySaver()

# 创建带记忆的 Agent
agent = create_agent(
    model=model,
    tools=[get_weather, calculate],
    checkpointer=checkpointer
)

# 使用 thread_id 区分不同对话
config = {"configurable": {"thread_id": "conversation_001"}}

# 第一次对话
result1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫张三"}]},
    config=config
)

# 第二次对话（Agent 记得之前的内容）
result2 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字？"}]},
    config=config
)
print(result2["messages"][-1].content)  # "你叫张三"

关键点：

  • thread_id 是区分不同对话的标识
  • 相同的 thread_id 会共享记忆
  • 不同的 thread_id 是完全独立的对话

持久化记忆到 SQLite​

内存记忆在程序重启后会丢失，生产环境建议使用持久化存储：

from langgraph.checkpoint.sqlite import SqliteSaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4o-mini")

# 使用 SQLite 持久化
with SqliteSaver.from_conn_string("agent_checkpoints.db") as checkpointer:
    agent = create_agent(
        model=model,
        tools=[get_weather, calculate],
        checkpointer=checkpointer
    )
    
    config = {"configurable": {"thread_id": "user_123"}}
    
    # 第一次对话
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "记住我喜欢Python"}]},
        config=config
    )

# 程序重启后，记忆仍然存在
with SqliteSaver.from_conn_string("agent_checkpoints.db") as checkpointer:
    agent = create_agent(
        model=model,
        tools=[get_weather, calculate],
        checkpointer=checkpointer
    )
    
    # 仍然记得之前的内容
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "我喜欢什么编程语言？"}]},
        config=config
    )

查看对话历史​

# 获取指定 thread 的历史状态
state = agent.get_state(config)
print(state.values["messages"])

# 获取完整的历史记录
history = list(agent.get_state_history(config))
for state in history:
    print(f"状态 ID: {state.config['configurable']['checkpoint_id']}")

回滚到之前的状态​

# 获取历史状态
states = list(agent.get_state_history(config))

# 回滚到第一个状态
first_state = states[-1]  # 列表是倒序的，最后一个是第一个状态
agent.update_state(first_state.config, first_state.values)

自定义状态​

LangChain 1.0 的 create_agent 只支持 TypedDict 来定义自定义状态，不再支持 Pydantic 模型和 dataclass。

使用 TypedDict 定义状态​

from typing import TypedDict, Annotated
from langchain.agents import create_agent, AgentState

class MyAgentState(TypedDict):
    """扩展的 Agent 状态"""
    messages: Annotated[list, AgentState.__annotations__['messages']]
    user_name: str           # 用户名
    preferences: dict        # 用户偏好
    turn_count: int          # 对话轮数

# 使用自定义状态创建 Agent
agent = create_agent(
    model=model,
    tools=tools,
    state_schema=MyAgentState
)

# 初始化时可以设置自定义字段
result = agent.invoke({
    "messages": [{"role": "user", "content": "你好"}],
    "user_name": "张三",
    "preferences": {"language": "python"},
    "turn_count": 0
})

Middleware 中间件​

LangChain 1.0 引入了 Middleware 机制，替代了之前的 hook 参数。这提供了更灵活的扩展能力。

基本结构​

from langchain.agents import create_agent, Middleware

class MyMiddleware(Middleware):
    """自定义中间件"""
    
    def before_model(self, state, context):
        """模型调用前执行"""
        print("即将调用模型...")
        return state
    
    def after_model(self, result, state, context):
        """模型调用后执行"""
        print("模型调用完成")
        return result
    
    def wrap_tool_call(self, tool_call, state, context):
        """包装工具调用"""
        result = yield tool_call
        return result

agent = create_agent(
    model=model,
    tools=tools,
    middleware=[MyMiddleware()]
)

动态提示词​

使用 @dynamic_prompt 装饰器根据状态动态生成提示词：

from langchain.agents import create_agent, dynamic_prompt

@dynamic_prompt
def get_dynamic_prompt(state, context):
    """根据状态生成动态提示词"""
    user_name = state.get("user_name", "用户")
    turn_count = state.get("turn_count", 0)
    
    return f"""你是一个智能助手。
当前用户：{user_name}
对话轮数：{turn_count}

请根据用户的需求提供帮助。"""

agent = create_agent(
    model=model,
    tools=tools,
    system_prompt=get_dynamic_prompt
)

结构化输出​

LangChain 1.0 提供了两种结构化输出策略：

使用 Pydantic 模型​

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from pydantic import BaseModel, Field

class Person(BaseModel):
    """人物信息"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    occupation: str = Field(description="职业")

model = init_chat_model("openai:gpt-4o-mini")

agent = create_agent(
    model=model,
    tools=[],
    response_format=Person
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "张三今年28岁，是一名软件工程师"}]
})

# 访问结构化输出
print(result["structured_response"])
# Person(name='张三', age=28, occupation='软件工程师')

运行时上下文​

LangChain 1.0 推荐使用 context 参数传递运行时配置，替代旧的 config["configurable"] 模式。

新的 context 参数​

from langchain.agents import create_agent

agent = create_agent(model=model, tools=tools)

# 新方式：使用 context 参数
result = agent.invoke(
    {"messages": [{"role": "user", "content": "你好"}]},
    context={"user_id": "user-123", "preferences": {"language": "zh"}}
)

在工具中访问上下文​

from langchain.tools import tool
from langchain_core.runnables import RunnableConfig

@tool
def get_user_info(config: RunnableConfig) -> str:
    """获取用户信息"""
    # 从 context 获取用户信息
    user_id = config.get("user_id", "unknown")
    return f"当前用户 ID: {user_id}"

实战示例​

智能客服 Agent​

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain.chat_models import init_chat_model
from langchain.tools import tool

model = init_chat_model("openai:gpt-4o-mini")

@tool
def get_order_status(order_id: str) -> str:
    """查询订单状态，order_id 格式为 ORD-XXXXX"""
    orders = {
        "ORD-12345": {"status": "已发货", "tracking": "SF123456"},
        "ORD-67890": {"status": "处理中", "eta": "2天后发货"}
    }
    order = orders.get(order_id)
    return f"订单 {order_id}：{order}" if order else f"未找到订单 {order_id}"

@tool
def initiate_refund(order_id: str, reason: str) -> str:
    """发起退款申请"""
    return f"订单 {order_id} 的退款申请已提交，原因：{reason}"

@tool
def transfer_to_human() -> str:
    """转接人工客服"""
    return "正在为您转接人工客服..."

# 创建客服 Agent
customer_service = create_agent(
    model=model,
    tools=[get_order_status, initiate_refund, transfer_to_human],
    system_prompt="你是专业的客服代表，态度友好，解决客户问题。",
    checkpointer=InMemorySaver()
)

# 使用
config = {"configurable": {"thread_id": "customer_001"}}
result = customer_service.invoke({
    "messages": [{"role": "user", "content": "我的订单 ORD-12345 到哪了？"}]
}, config=config)

print(result["messages"][-1].content)

多工具协作 Agent​

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool

model = init_chat_model("openai:gpt-4o-mini")

@tool
def search_web(query: str) -> str:
    """搜索网络信息"""
    return f"搜索结果：关于 '{query}' 的信息..."

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{e}"

@tool
def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def translate(text: str, target_language: str) -> str:
    """翻译文本到目标语言"""
    return f"[{target_language}] {text}"

# 创建多工具 Agent
agent = create_agent(
    model=model,
    tools=[search_web, calculate, get_current_time, translate],
    system_prompt="你是一个多功能助手，可以搜索、计算、翻译和查询时间。"
)

# 复杂任务示例
result = agent.invoke({
    "messages": [{"role": "user", "content": "现在几点了？顺便帮我算一下 123 * 456"}]
})

调试与监控​

LangSmith 追踪​

import os

# 配置 LangSmith
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls-..."
os.environ["LANGSMITH_PROJECT"] = "my-agent"

# 所有 Agent 调用会自动记录到 LangSmith
result = agent.invoke({"messages": messages})

# 访问 https://smith.langchain.com 查看详细追踪

控制台调试​

from langchain.globals import set_debug
set_debug(True)

# 执行时会打印详细的内部调用信息
result = agent.invoke({"messages": messages})

常见问题排查​

问题可能原因解决方案
Agent 不调用工具工具描述不清晰优化描述，明确使用场景
工具选择错误工具名称或描述相似细化描述，区分场景
无限循环任务超出模型能力简化任务，使用更强的模型
响应格式错误模型不稳定添加输出格式约束

小结​

Agent 是 LangChain 的核心功能，它让 LLM 从"被动回答"进化为"主动执行"：

  • create_agent 是 LangChain 1.0 创建 Agent 的标准方式
  • 工具 是 Agent 与外部世界交互的桥梁，需要清晰描述
  • 记忆 让 Agent 能够保持上下文连贯性
  • Middleware 提供了灵活的扩展机制
  • 结构化输出 让 Agent 返回可控格式的结果

下一步学习​

参考资料​