Skip to content

LangGraph Agent 详细解读

📚 概述

本文档详细解读 LangGraph 中的 Agent(智能体) 架构。这是构建智能 AI 应用的核心模式,通过让 AI 模型自主决策、调用工具、并基于结果继续推理,实现复杂任务的自动化处理。

🎯 核心概念

什么是 Agent?

Agent(智能体)是一种能够自主决策和行动的 AI 系统。在 LangGraph 中,Agent 具有以下特点:

  1. 自主决策

    • AI 模型自己决定是否需要调用工具
    • 根据工具返回的结果决定下一步行动
    • 可以连续调用多个工具完成复杂任务
  2. 工具调用

    • 调用外部函数获取信息或执行操作
    • 将工具结果传回模型继续处理
    • 支持多轮工具调用
  3. 循环推理

    • 模型可以基于工具结果继续思考
    • 决定是继续调用工具还是直接回答
    • 形成 "思考-行动-观察" 的循环

ReAct 架构

ReAct(Reasoning + Acting)是一种经典的 Agent 架构模式:

  • Reason(推理):模型分析问题,决定需要调用什么工具
  • Act(行动):调用选定的工具
  • Observe(观察):获取工具返回结果
  • 循环:基于观察结果,决定继续行动还是给出最终答案

与 Router 的区别

在之前的 Router 模式中:

  • 模型调用工具后,直接返回 ToolMessage 给用户
  • 这是一次性的工具调用,没有循环

在 Agent 模式中:

  • 模型调用工具后,ToolMessage 被传回模型
  • 模型可以基于结果决定:
    • 继续调用其他工具
    • 或者直接回答用户

关键差异:工具结果的流向

Router:  用户 → 模型 → 工具 → 用户
Agent:   用户 → 模型 → 工具 → 模型 → ... → 用户
                    ↑____________↓
                      循环反馈

🎭 实战案例:数学计算 Agent

我们将构建一个数学计算助手,演示 Agent 的完整工作流程。

需求: 给定一个多步骤数学问题,Agent 需要:

  1. 理解问题需要哪些计算步骤
  2. 依次调用相应的工具(加法、乘法、除法)
  3. 基于每一步结果,决定下一步操作
  4. 最终给出答案

系统架构图

用户输入: "Add 3 and 4. Multiply the output by 2. Divide the output by 5"

   [assistant] 分析:需要先做加法

    (tools_condition 判断:需要调用工具)

    [tools] 调用 add(3, 4) → 返回 7

   [assistant] 收到结果 7,分析:需要乘以 2

    (tools_condition 判断:需要调用工具)

    [tools] 调用 multiply(7, 2) → 返回 14

   [assistant] 收到结果 14,分析:需要除以 5

    (tools_condition 判断:需要调用工具)

    [tools] 调用 divide(14, 5) → 返回 2.8

   [assistant] 收到结果 2.8,生成最终答案

    (tools_condition 判断:不需要工具)

       END → 返回结果给用户

🔧 代码实现详解

1. 定义工具函数

python
def multiply(a: int, b: int) -> int:
    """Multiply a and b.

    Args:
        a: first int
        b: second int
    """
    return a * b

def add(a: int, b: int) -> int:
    """Adds a and b.

    Args:
        a: first int
        b: second int
    """
    return a + b

def divide(a: int, b: int) -> float:
    """Divide a and b.

    Args:
        a: first int
        b: second int
    """
    return a / b

tools = [add, multiply, divide]

Python 知识点:函数文档字符串(Docstring)

函数的文档字符串不仅仅是注释,它是 Python 的一等公民:

python
def add(a: int, b: int) -> int:
    """Adds a and b.    # ← 简短描述

    Args:               # ← 参数说明
        a: first int
        b: second int
    """
    return a + b

# 可以通过 __doc__ 属性访问
print(add.__doc__)

LangChain 的魔法: LangChain 会解析这些文档字符串,并将它们作为工具描述传递给 LLM:

python
# LLM 会看到类似这样的工具描述:
# Tool: add
# Description: Adds a and b.
# Parameters:
#   - a: first int
#   - b: second int

这就是为什么 LLM 能够"理解"工具的作用!


2. 绑定工具到模型

python
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools(tools, parallel_tool_calls=False)

关键参数:parallel_tool_calls=False

这个参数控制模型是否可以同时调用多个工具:

python
# parallel_tool_calls=True(默认)
# 模型可能同时调用:add(3, 4) 和 multiply(7, 2)
# 适合:并行、独立的任务

# parallel_tool_calls=False
# 模型必须依次调用:先 add(3, 4),等结果,再调用 multiply
# 适合:数学计算等需要顺序执行的任务

为什么数学计算要设置为 False?

数学计算通常有依赖关系:

第一步:3 + 4 = 7
第二步:7 × 2 = 14  ← 依赖第一步的结果!
第三步:14 ÷ 5 = 2.8  ← 依赖第二步的结果!

如果允许并行调用,模型可能在不知道第一步结果的情况下就尝试执行第二步,导致错误。


3. 定义状态

python
from langgraph.graph import MessagesState
from langchain_core.messages import HumanMessage, SystemMessage

# System message
sys_msg = SystemMessage(content="You are a helpful assistant tasked with performing arithmetic on a set of inputs.")

LangGraph 知识点:MessagesState

MessagesState 是 LangGraph 内置的状态类,专门用于聊天场景:

python
class MessagesState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

核心特性:

  1. 自动消息管理:维护对话历史
  2. 消息追加:使用 add_messages reducer 自动追加新消息
  3. 类型安全:支持不同类型的消息(HumanMessage、AIMessage、ToolMessage 等)

消息类型:

python
from langchain_core.messages import (
    HumanMessage,    # 用户消息
    AIMessage,       # AI 回复
    SystemMessage,   # 系统提示
    ToolMessage      # 工具返回结果
)

4. 定义 Assistant 节点

python
def assistant(state: MessagesState):
   return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}

代码解析:

python
[sys_msg] + state["messages"]
# 构建完整的消息列表:
# [
#   SystemMessage("You are a helpful assistant..."),  # 系统提示
#   HumanMessage("Add 3 and 4..."),                   # 用户问题
#   AIMessage(tool_calls=[...]),                       # AI 的工具调用
#   ToolMessage(content="7"),                          # 工具返回结果
#   ...
# ]

为什么每次都加上 sys_msg?

每次调用 LLM 时,都需要提供完整的上下文:

  • 系统提示定义 AI 的角色和行为
  • 历史消息提供对话上下文
  • 这样 AI 才能理解当前处于对话的哪个阶段

返回格式:

python
return {"messages": [new_message]}
# 返回的消息会被追加到 state["messages"] 中
# 因为 MessagesState 使用了 add_messages reducer

5. 构建 Agent 图

python
from langgraph.graph import START, StateGraph
from langgraph.prebuilt import tools_condition
from langgraph.prebuilt import ToolNode

# 创建图
builder = StateGraph(MessagesState)

# 添加节点
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))

# 添加边
builder.add_edge(START, "assistant")

# 条件边:根据 assistant 的输出决定路由
builder.add_conditional_edges(
    "assistant",
    tools_condition,  # 条件函数
)

# 关键:从 tools 回到 assistant 形成循环
builder.add_edge("tools", "assistant")

# 编译
react_graph = builder.compile()

LangGraph 知识点:ToolNode

ToolNode 是 LangGraph 提供的预构建节点,用于执行工具调用:

python
ToolNode(tools)
# 自动处理:
# 1. 解析 AIMessage 中的 tool_calls
# 2. 调用对应的工具函数
# 3. 将结果包装成 ToolMessage
# 4. 返回给下一个节点

等价的手动实现:

python
def tools_node(state: MessagesState):
    # 获取最后一条消息(AI 的工具调用)
    last_message = state["messages"][-1]

    # 执行所有工具调用
    tool_messages = []
    for tool_call in last_message.tool_calls:
        # 找到对应的工具
        tool = next(t for t in tools if t.name == tool_call["name"])
        # 调用工具
        result = tool.invoke(tool_call["args"])
        # 包装成 ToolMessage
        tool_messages.append(ToolMessage(
            content=str(result),
            tool_call_id=tool_call["id"]
        ))

    return {"messages": tool_messages}

LangGraph 知识点:tools_condition

tools_condition 是预构建的条件函数,用于判断是否需要调用工具:

python
def tools_condition(state: MessagesState) -> str:
    # 获取最后一条消息
    last_message = state["messages"][-1]

    # 如果是 AIMessage 且包含 tool_calls
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"  # 路由到 tools 节点
    else:
        return END  # 结束图执行

条件边的工作流程:

python
builder.add_conditional_edges("assistant", tools_condition)

# 完整流程:
# assistant 节点执行完毕
#    ↓
# tools_condition 检查输出
#    ↓
# 如果有 tool_calls → 路由到 "tools"
# 如果没有 tool_calls → 路由到 END

Agent 循环的关键:

python
builder.add_edge("tools", "assistant")  # ← 这条边创建了循环!

# 完整循环:
# assistant → (有工具调用) → tools → assistant → ...
#          → (无工具调用) → END

6. 执行 Agent

python
messages = [HumanMessage(content="Add 3 and 4. Multiply the output by 2. Divide the output by 5")]
messages = react_graph.invoke({"messages": messages})

执行过程详解:

步骤 1:assistant 节点
输入:[HumanMessage("Add 3 and 4. Multiply the output by 2. Divide the output by 5")]
输出:AIMessage(tool_calls=[ToolCall(name="add", args={"a": 3, "b": 4})])
tools_condition:检测到 tool_calls → 路由到 "tools"

步骤 2:tools 节点
输入:[HumanMessage(...), AIMessage(tool_calls=[...])]
执行:add(3, 4) = 7
输出:ToolMessage(content="7", tool_call_id="...")
自动路由:→ "assistant"

步骤 3:assistant 节点
输入:[HumanMessage(...), AIMessage(...), ToolMessage("7")]
分析:看到结果是 7,需要乘以 2
输出:AIMessage(tool_calls=[ToolCall(name="multiply", args={"a": 7, "b": 2})])
tools_condition:检测到 tool_calls → 路由到 "tools"

步骤 4:tools 节点
执行:multiply(7, 2) = 14
输出:ToolMessage(content="14")
自动路由:→ "assistant"

步骤 5:assistant 节点
输入:包含 ToolMessage("14") 的完整对话历史
分析:看到结果是 14,需要除以 5
输出:AIMessage(tool_calls=[ToolCall(name="divide", args={"a": 14, "b": 5})])
tools_condition:检测到 tool_calls → 路由到 "tools"

步骤 6:tools 节点
执行:divide(14, 5) = 2.8
输出:ToolMessage(content="2.8")
自动路由:→ "assistant"

步骤 7:assistant 节点
输入:包含所有历史消息和最终结果 2.8
分析:所有计算完成,可以给出最终答案
输出:AIMessage(content="The final result is 2.8")  # 注意:没有 tool_calls
tools_condition:没有 tool_calls → 路由到 END

图执行结束

7. 查看输出

python
for m in messages['messages']:
    m.pretty_print()

输出解析:

================================ Human Message =================================
Add 3 and 4. Multiply the output by 2. Divide the output by 5

================================== Ai Message ==================================
Tool Calls:
  add (call_i8zDfMTdvmIG34w4VBA3m93Z)
 Call ID: call_i8zDfMTdvmIG34w4VBA3m93Z
  Args:
    a: 3
    b: 4

解释:

  • AI 理解了任务,决定先调用 add 工具
  • Call ID 是唯一标识符,用于追踪工具调用和结果的对应关系
  • Args 是传递给工具的参数
================================= Tool Message =================================
Name: add
7

解释:

  • 工具执行结果:3 + 4 = 7
  • 这个结果会被传回 assistant 节点
================================== Ai Message ==================================
Tool Calls:
  multiply (call_nE62D40lrGQC7b67nVOzqGYY)
 Call ID: call_nE62D40lrGQC7b67nVOzqGYY
  Args:
    a: 7
    b: 2

解释:

  • AI 收到结果 7,决定下一步调用 multiply
  • 注意参数 a: 7 来自上一步的结果!
================================= Tool Message =================================
Name: multiply
14

解释: 7 × 2 = 14

================================== Ai Message ==================================
Tool Calls:
  divide (call_6Q9SjxD2VnYJqEBXFt7O1moe)
 Call ID: call_6Q9SjxD2VnYJqEBXFt7O1moe
  Args:
    a: 14
    b: 5

解释:

  • AI 继续推理,调用 divide
  • 参数 a: 14 来自上一步结果
================================= Tool Message =================================
Name: divide
2.8

解释: 14 ÷ 5 = 2.8

================================== Ai Message ==================================
The final result after performing the operations (3 + 4) × 2 ÷ 5 is 2.8.

解释:

  • 所有计算完成,AI 给出最终答案
  • 这条消息没有 tool_calls,所以图执行结束

关键观察点:

  1. AI 自主决定了每一步的工具调用
  2. 每次都基于前一步的结果进行推理
  3. 最终给出了自然语言的回答
  4. 整个过程是自动化的,无需人工干预

🎓 核心知识点总结

LangGraph 特有概念

1. Agent vs Router

特性RouterAgent
工具调用次数一次多次(循环)
工具结果流向直接返回用户返回模型继续处理
决策次数一次多次(每次循环都决策)
适用场景简单分类、单次查询复杂任务、多步骤推理

2. ReAct 循环

python
while True:
    # Reason: AI 分析当前状态,决定行动
    ai_message = assistant(state)

    # 判断:需要工具吗?
    if no_tool_calls(ai_message):
        break  # 不需要,返回答案

    # Act: 调用工具
    tool_result = tools(state)

    # Observe: 更新状态,将结果传回 AI
    state = update_state(tool_result)

3. 条件边与循环

python
# 条件边:根据输出动态路由
builder.add_conditional_edges("assistant", tools_condition)

# 返回边:创建循环
builder.add_edge("tools", "assistant")

# 组合效果:
# assistant → [有工具调用] → tools → assistant → ...
#          → [无工具调用] → END

4. MessagesState

python
class MessagesState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# 特点:
# 1. 自动维护对话历史
# 2. 支持多种消息类型
# 3. 使用 add_messages reducer 追加消息

5. ToolNode

python
ToolNode(tools)

# 自动处理:
# - 解析 tool_calls
# - 调用函数
# - 包装结果为 ToolMessage
# - 追踪 tool_call_id

Python 特有知识点

1. 函数类型注解

python
def add(a: int, b: int) -> int:
    """Adds a and b."""
    return a + b

# a: int        ← 参数类型注解
# -> int        ← 返回类型注解
# """..."""     ← 文档字符串

作用:

  • IDE 提供智能提示
  • 类型检查工具(mypy)验证代码
  • LangChain 解析用于工具描述

2. Docstring 规范(Google 风格)

python
def multiply(a: int, b: int) -> int:
    """Multiply a and b.          # 简短描述(必须)

    Args:                          # 参数列表(可选)
        a: first int
        b: second int

    Returns:                       # 返回值(可选)
        The product of a and b

    Raises:                        # 异常(可选)
        ValueError: If a or b is negative
    """
    return a * b

3. 列表拼接

python
[sys_msg] + state["messages"]

# 等价于:
result = []
result.append(sys_msg)
result.extend(state["messages"])

4. 对象属性检查

python
hasattr(obj, "attribute_name")

# 示例:
hasattr(last_message, "tool_calls")
# 检查 last_message 对象是否有 tool_calls 属性

💡 最佳实践

1. 何时使用 Agent?

适用场景:

  • 需要多步骤推理的任务(如数学计算、数据分析)
  • 需要根据中间结果动态调整策略
  • 工具调用顺序不固定,需要 AI 自主决策
  • 任务复杂度高,需要灵活应对

不适用场景:

  • 简单的单次工具调用(用 Router 即可)
  • 固定流程的任务(用普通图即可)
  • 不需要循环推理的场景

2. 并行工具调用的选择

python
# 场景 1:数学计算 → parallel_tool_calls=False
llm_with_tools = llm.bind_tools([add, multiply, divide], parallel_tool_calls=False)
# 原因:步骤有依赖关系,必须顺序执行

# 场景 2:信息收集 → parallel_tool_calls=True(默认)
llm_with_tools = llm.bind_tools([search_web, query_db, call_api])
# 原因:三个工具可以同时调用,提高效率

3. 系统提示的重要性

python
# ✅ 好的系统提示
sys_msg = SystemMessage(content="""
You are a helpful assistant tasked with performing arithmetic on a set of inputs.
- Break down complex calculations into steps
- Use the appropriate tool for each operation
- Show your reasoning
""")

# ❌ 不好的系统提示
sys_msg = SystemMessage(content="You are helpful.")
# 太模糊,AI 可能不知道如何使用工具

4. 防止无限循环

Agent 有可能陷入无限循环,需要设置限制:

python
# 方法 1:使用 recursion_limit
react_graph = builder.compile(recursion_limit=10)
# 最多执行 10 次循环

# 方法 2:在状态中计数
class MessagesStateWithCount(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    iterations: int

def assistant(state: MessagesStateWithCount):
    if state.get("iterations", 0) > 10:
        return {"messages": [AIMessage(content="Reached maximum iterations")]}
    # ... 正常逻辑
    return {
        "messages": [llm_with_tools.invoke(...)],
        "iterations": state.get("iterations", 0) + 1
    }

5. 工具函数设计原则

python
# ✅ 好的工具设计
def search_database(query: str, limit: int = 10) -> list[dict]:
    """Search the database for matching records.

    Args:
        query: Search query string
        limit: Maximum number of results (default: 10)

    Returns:
        List of matching records
    """
    # 实现...
    return results

# 特点:
# - 清晰的文档字符串
# - 类型注解
# - 默认参数值
# - 返回结构化数据
python
# ❌ 不好的工具设计
def search(q):  # 没有类型注解
    """搜索"""  # 文档不清楚
    # AI 不知道如何使用这个工具

🚀 进阶技巧

1. 带记忆的 Agent

python
class MessagesStateWithMemory(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    user_preferences: dict  # 存储用户偏好
    conversation_summary: str  # 对话摘要

def assistant(state: MessagesStateWithMemory):
    # 可以访问历史偏好
    preferences = state.get("user_preferences", {})

    # 构建带上下文的提示
    context = f"User preferences: {preferences}\n"
    prompt = context + "\n".join([m.content for m in state["messages"]])

    response = llm_with_tools.invoke(prompt)
    return {"messages": [response]}

2. 动态工具选择

python
def assistant(state: MessagesState):
    # 根据对话内容动态选择可用工具
    last_message = state["messages"][-1].content

    if "weather" in last_message:
        tools = [get_weather, get_forecast]
    elif "database" in last_message:
        tools = [query_db, insert_db, update_db]
    else:
        tools = all_tools

    llm_with_selected_tools = llm.bind_tools(tools)
    response = llm_with_selected_tools.invoke(state["messages"])
    return {"messages": [response]}

3. 错误处理

python
def tools_node(state: MessagesState):
    last_message = state["messages"][-1]
    tool_messages = []

    for tool_call in last_message.tool_calls:
        try:
            tool = next(t for t in tools if t.name == tool_call["name"])
            result = tool.invoke(tool_call["args"])
            tool_messages.append(ToolMessage(
                content=str(result),
                tool_call_id=tool_call["id"]
            ))
        except Exception as e:
            # 返回错误信息给 AI
            tool_messages.append(ToolMessage(
                content=f"Error: {str(e)}",
                tool_call_id=tool_call["id"],
                status="error"
            ))

    return {"messages": tool_messages}

4. 流式输出

python
# 使用 stream 方法查看实时执行过程
for chunk in react_graph.stream({"messages": [HumanMessage("Calculate...")]}):
    print(chunk)

# 输出:
# {'assistant': {'messages': [AIMessage(tool_calls=[...])]}}
# {'tools': {'messages': [ToolMessage(content="7")]}}
# {'assistant': {'messages': [AIMessage(tool_calls=[...])]}}
# ...

📊 Agent 变体对比

1. 基础 Agent(本教程)

python
builder.add_edge("tools", "assistant")  # 工具结果总是回到 assistant

特点: 简单、直接,适合大多数场景

2. 条件性返回 Agent

python
def should_continue(state: MessagesState):
    last_message = state["messages"][-1]
    # 可以根据工具结果决定是否继续
    if "error" in last_message.content:
        return END
    return "assistant"

builder.add_conditional_edges("tools", should_continue, ["assistant"])

特点: 可以提前终止,处理错误情况

3. 多路径 Agent

python
def route_after_tool(state: MessagesState):
    last_message = state["messages"][-1]

    if "final" in last_message.content:
        return "summarize"  # 去总结节点
    elif "error" in last_message.content:
        return "error_handler"  # 去错误处理节点
    else:
        return "assistant"  # 继续推理

builder.add_conditional_edges("tools", route_after_tool,
                             ["assistant", "summarize", "error_handler"])

特点: 更复杂的控制流,适合大型应用


🎯 实际应用案例

案例 1:研究助手

python
tools = [
    search_web,          # 搜索网络
    read_paper,          # 读取论文
    summarize_text,      # 总结文本
    save_notes           # 保存笔记
]

# 用户:Research recent advances in quantum computing
# Agent 工作流:
# 1. search_web("quantum computing 2024")
# 2. read_paper(paper_url_1)
# 3. summarize_text(paper_content_1)
# 4. read_paper(paper_url_2)
# 5. summarize_text(paper_content_2)
# 6. save_notes(combined_summary)
# 7. 返回研究总结

案例 2:数据分析助手

python
tools = [
    query_database,      # 查询数据库
    calculate_stats,     # 计算统计
    create_chart,        # 生成图表
    export_report        # 导出报告
]

# 用户:Analyze sales data for Q4 and create a report
# Agent 工作流:
# 1. query_database("SELECT * FROM sales WHERE quarter = 'Q4'")
# 2. calculate_stats(sales_data)
# 3. create_chart(stats, type="bar")
# 4. create_chart(stats, type="pie")
# 5. export_report(charts, stats)
# 6. 返回报告链接

案例 3:客户支持助手

python
tools = [
    search_knowledge_base,  # 搜索知识库
    check_order_status,     # 查询订单
    create_ticket,          # 创建工单
    send_email              # 发送邮件
]

# 用户:I haven't received my order #12345
# Agent 工作流:
# 1. check_order_status("12345")
# 2. search_knowledge_base("delayed shipping")
# 3. 发现问题:物流延误
# 4. create_ticket(order_id="12345", issue="shipping_delay")
# 5. send_email(user, "Your order is delayed, ticket created")
# 6. 返回解决方案和工单号

🔍 常见问题

Q1: Agent 什么时候会停止循环?

答:assistant 节点返回的 AIMessage 不包含 tool_calls 时:

python
# 会继续循环
AIMessage(tool_calls=[ToolCall(name="add", ...)])

# 会停止(进入 END)
AIMessage(content="The answer is 42")

Q2: 如何让 Agent 调用特定工具?

答: 通过系统提示引导:

python
sys_msg = SystemMessage(content="""
You are a helpful assistant.
When the user asks about weather, ALWAYS use the get_weather tool.
When the user asks about time, ALWAYS use the get_time tool.
""")

或者使用 tool_choice 参数:

python
# 强制使用特定工具
llm_with_tools = llm.bind_tools(tools, tool_choice="get_weather")

# 强制调用任意工具(不能直接回答)
llm_with_tools = llm.bind_tools(tools, tool_choice="any")

# 禁止调用工具
llm_with_tools = llm.bind_tools(tools, tool_choice="none")

Q3: 如何查看 Agent 的执行过程?

方法 1:使用 stream

python
for chunk in graph.stream(input_data):
    print(chunk)

方法 2:使用 LangSmith

python
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "my-agent"

# 在 LangSmith 网站查看详细 trace

方法 3:自定义日志

python
def assistant(state: MessagesState):
    print(f"[Assistant] Processing {len(state['messages'])} messages")
    response = llm_with_tools.invoke(state["messages"])
    print(f"[Assistant] Response: {response}")
    return {"messages": [response]}

Q4: 工具调用失败怎么办?

答: 在 ToolNode 中处理异常:

python
def custom_tools_node(state: MessagesState):
    last_message = state["messages"][-1]
    tool_messages = []

    for tool_call in last_message.tool_calls:
        try:
            result = execute_tool(tool_call)
            tool_messages.append(ToolMessage(
                content=str(result),
                tool_call_id=tool_call["id"]
            ))
        except Exception as e:
            # 将错误返回给 AI,让它重试或换方法
            tool_messages.append(ToolMessage(
                content=f"Error: {str(e)}. Please try a different approach.",
                tool_call_id=tool_call["id"]
            ))

    return {"messages": tool_messages}

Q5: 如何限制 Agent 的执行成本?

方法 1:限制循环次数

python
graph = builder.compile(recursion_limit=5)

方法 2:限制 token 使用

python
class TokenTrackingState(MessagesState):
    total_tokens: int

def assistant(state: TokenTrackingState):
    if state.get("total_tokens", 0) > 10000:
        return {"messages": [AIMessage(content="Token limit reached")]}

    # 调用 LLM 并记录 token
    response = llm_with_tools.invoke(state["messages"])
    tokens_used = response.response_metadata.get("token_usage", {}).get("total_tokens", 0)

    return {
        "messages": [response],
        "total_tokens": state.get("total_tokens", 0) + tokens_used
    }

📖 扩展阅读


总结:Agent 是 LangGraph 的核心模式,通过 "推理-行动-观察" 的循环,让 AI 能够自主完成复杂任务。掌握 Agent 的构建方法,是开发强大 AI 应用的基础!

基于 MIT 许可证发布。内容版权归作者所有。