1. 定义工具
AgentRag 学习 AI与LLM
1. 定义工具
AgentRag 学习 AI与LLM

1. 定义工具

Agent 搭建

一、AI Agent 核心概念

1.1 Agent 的历史脉络

AI Agent 并不是大语言模型的专属概念。它的思想根源可以追溯到 20 世纪 50 年代的符号主义 AI：

时期 范式 核心思想 代表
1950s-1980s 符号推理 Agent 基于逻辑和知识库进行符号操作 专家系统、SOAR
1980s-2000s 反应式 Agent 基于感知-动作规则对环境做出反应 行为树、有限状态机
1990s-2010s BDI Agent 基于信念(Belief)-欲望(Desire)-意图(Intention)模型推理 PRS、Jason
2010s 强化学习 Agent 通过环境交互和奖励信号学习策略 DQN、AlphaGo
2022-至今 LLM Agent 以语言模型为核心推理引擎，通过自然语言与环境交互 AutoGPT、LangGraph Agent

关键转折：LLM Agent 的核心突破在于用自然语言作为”思维语言”——Agent 的推理过程不再需要形式逻辑或结构化规则，而是利用了 LLM 在预训练中习得的常识推理能力。

1.2 什么是 LLM Agent

LLM Agent 是一个以语言模型为核心推理引擎、通过工具调用与环境交互的智能系统。它与传统 Chain 的本质区别在于：

维度 Chain Agent
控制流 预定义、确定性 动态、由 LLM 自主决定
工具使用 无或固定 灵活选择、动态调用
决策能力 无 每步都需做出决策
适应性 无 可根据中间结果调整策略
复杂度 低 高（需要循环、路由、错误处理）

1.3 Agent 的数学定义

可以将 Agent 形式化为一个函数：

Agent(input, state_t, tools) → (action, state_{t+1})

其中：
  action ∈ {FINAL_ANSWER(text), TOOL_CALL(name, params)}
  state_t 为当前状态（消息历史、中间结果等）

Agent 在每一步中自主决定：

  1. 是否已经有足够信息给出最终答案（FINAL_ANSWER）
  2. 还是需要调用某个工具获取更多信息（TOOL_CALL）

这个决策-执行循环持续进行，直到 Agent 决定终止。

二、ReAct 模式

2.1 ReAct 的历史与理论基础

ReAct（Reasoning + Acting）由 Yao 等人于 2022 年在《ReAct: Synergizing Reasoning and Acting in Language Models》论文中提出。这篇论文的关键贡献是证明了推理和行动可以相互增强：

  • 推理指导行动：通过思考”我需要知道什么”来决定调用哪个工具
  • 行动反馈推理：工具返回的结果更新了状态，促使下一轮推理更准确

论文的实验表明，ReAct 在 HotpotQA（多跳推理）和 Fever（事实验证）任务上显著优于纯 CoT（Chain-of-Thought）和纯 Act-only 方法，同时减少了幻觉——因为外部工具提供了可验证的事实依据。

2.2 ReAct 循环的深层原理

Thought → Action → Observation → Thought → Action → ... → Final Answer

为什么这个循环有效？

关键在于 LLM 的”上下文学习”（In-Context Learning）能力。当 LLM 在自己的上下文中看到之前的 Thought-Action-Observation 序列后，它能：

  1. 从观察中提取关键信息：解析工具返回结果中有用的部分
  2. 检测信息缺口：判断当前信息是否足以回答问题
  3. 调整搜索方向：如果当前工具没返回有用信息，换一个工具或参数

这本质上是将 LLM 作为一个隐式的世界模型来使用——它通过自然语言在上下文中维护了对任务状态的认知。

2.3 ReAct 循环流程序列图

迭代 1:
  [用户] "北京今天多少度？"
  [Agent Thought] 我需要查询北京的实时天气
  [Agent Action] → get_weather(city="北京")
  [Observation] "北京 25°C，晴"

迭代 2:
  [Agent Thought] 已获得天气数据，可以给出答案
  [Agent Action] → FINAL_ANSWER: "北京今天 25°C，晴天"

2.4 LangGraph 实现 ReAct Agent

方式一：使用 create_agent（高层 API）

from langchain.agents import create_agent

agent = create_agent(
    model="gpt-4o",
    tools=[search_tool, calculator_tool, weather_tool],
    prompt="你是一个有用的助手，可以使用工具来回答问题。",
    checkpointer=MemorySaver(),
)
result = agent.invoke({"messages": [{"role": "user", "content": "北京今天多少度？"}]})

方式二：手动构建 StateGraph（深入理解底层机制）

from langgraph.graph import StateGraph, START, END
from langchain.messages import SystemMessage, HumanMessage, ToolMessage
from langchain.chat_models import init_chat_model
from typing import Literal, Annotated, TypedDict
import operator


from langchain.tools import tool

@tool
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

tools = [add, multiply]
tools_by_name = {t.name: t for t in tools}

# 2. 初始化模型并绑定工具
# bind_tools 将工具定义转为 JSON Schema 注入到 system message 中
model = init_chat_model("gpt-4o", temperature=0)
model_with_tools = model.bind_tools(tools)

# 3. 定义状态
# 使用 Annotated + operator.add 确保消息列表在节点间自动追加而非覆盖
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

# 4. LLM 调用节点 —— Agent 的"思考"环节
def llm_call(state: AgentState):
    response = model_with_tools.invoke(
        [SystemMessage(content="你是一个数学助手。")] + state["messages"]
    )
    return {"messages": [response]}

# 5. 工具执行节点 —— Agent 的"行动"环节
def tool_node(state: AgentState):
    last_message = state["messages"][-1]
    results = []
    for tool_call in last_message.tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        results.append(ToolMessage(
            content=str(observation),
            tool_call_id=tool_call["id"]
        ))
    return {"messages": results}

# 6. 路由判断 —— 决定继续循环还是终止
def should_continue(state: AgentState) -> Literal["tool_node", END]:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"
    return END

# 7. 构建有向循环图
builder = StateGraph(AgentState)
builder.add_node("llm_call", llm_call)
builder.add_node("tool_node", tool_node)
builder.add_edge(START, "llm_call")
builder.add_conditional_edges("llm_call", should_continue, {
    "tool_node": "tool_node", END: END
})
builder.add_edge("tool_node", "llm_call")  # 工具结果返回 LLM

agent = builder.compile()

# 8. 调用
result = agent.invoke(
    {"messages": [HumanMessage(content="3乘以4再加5等于多少？")]}
)
for m in result["messages"]:
    m.pretty_print()

2.5 工具调用（Tool Calling）的底层机制

bind_tools 到底做了什么？

model.bind_tools(tools) 将每个工具的定义转换为 JSON Schema 格式，注入到 API 请求的 tools 参数中：

{
  "tools": [{
    "type": "function",
    "function": {
      "name": "get_weather",
      "description": "获取指定城市的天气信息",
      "parameters": {
        "type": "object",
        "properties": {
          "city": {"type": "string", "description": "城市名称"}
        },
        "required": ["city"]
      }
    }
  }]
}

当 LLM 决定调用工具时，它的输出不是普通文本，而是结构化的 tool_call 对象：

{
  "role": "assistant",
  "content": null,
  "tool_calls": [{
    "id": "call_abc123",
    "type": "function",
    "function": {
      "name": "get_weather",
      "arguments": "{\"city\": \"北京\"}"
    }
  }]
}

LLM 如何学会调用工具？ 模型在微调阶段被训练了”在看到特定工具描述后输出结构化调用”的能力。这不是 prompt engineering 的结果，而是模型参数中内化的功能。

2.6 ReAct 的优缺点

优点 缺点
灵活适应不确定任务 可能进入死循环（无限调用同一工具）
推理过程完全可解释（可在 UI 展示 Thought） 复杂任务效率低（每步都需调用 LLM）
适合交互式探索任务 缺乏全局规划（只考虑下一步，不考虑全局最优）
工具使用直观 Token 消耗大（每轮循环都在积累上下文）
出错后可以自我纠正 对长任务容易迷失方向（上下文漂移）

三、Plan-and-Execute 模式

3.1 理论基础：分层任务分解

Plan-and-Execute 模式借鉴了 AI 规划理论中的 **HTN（Hierarchical Task Network）**思想——将高层目标逐步分解为可直接执行的原语操作。与 ReAct 的”走一步看一步”不同，Plan-Execute 先做全局规划再执行。

用户任务 → Planner（全局分解为步骤序列）→ Executor（逐步执行）→ 结果汇总

Plan-Execute vs ReAct 的本质区别：

维度 ReAct Plan-Execute
规划粒度 每步规划下一步 一次性规划全部
适用任务 探索式（{不确定性高}） 确定性（步骤可预知）
容错性 高（每步可纠偏） 低（依赖计划质量）
效率 低（频繁 LLM 调用） 高（规划一次，执行 N 步）
可中断性 差 好（可在步骤间中断）

3.2 实现示例

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated, Literal
import operator
import json

class PlanStep(TypedDict):
    step: int
    description: str
    tool: str
    status: str  # pending, in_progress, completed

class PlanExecuteState(TypedDict):
    task: str
    plan: list[PlanStep]
    current_step: int
    results: Annotated[list, operator.add]
    messages: list

def planner(state: PlanExecuteState):
    """全局规划：将任务分解为步骤序列"""
    prompt = f"""你是一个任务规划专家。将以下任务拆解为可独立执行的步骤。
    每个步骤必须明确使用的工具。

    任务: {state["task"]}

    以 JSON 格式返回：{{"steps": [{{"step": 1, "description": "...", "tool": "..."}}]}}"""

    response = llm.invoke(prompt)
    plan = json.loads(response.content)["steps"]
    return {"plan": plan, "current_step": 0, "results": []}

def executor(state: PlanExecuteState):
    """逐步执行：每次执行当前步骤"""
    step = state["plan"][state["current_step"]]
    result = execute_step(step, state["results"])
    step["status"] = "completed"
    return {
        "results": [result],
        "current_step": state["current_step"] + 1
    }

def summarizer(state: PlanExecuteState):
    """汇总所有步骤结果，生成最终输出"""
    summary_prompt = f"汇总以下步骤结果：\n{state['results']}\n生成最终回答。"
    response = llm.invoke(summary_prompt)
    return {"messages": [response]}

def should_execute(state: PlanExecuteState) -> Literal["executor", "summarizer"]:
    if state["current_step"] < len(state["plan"]):
        return "executor"
    return "summarizer"

builder = StateGraph(PlanExecuteState)
builder.add_node("planner", planner)
builder.add_node("executor", executor)
builder.add_node("summarizer", summarizer)
builder.add_edge(START, "planner")
builder.add_edge("planner", "executor")
builder.add_conditional_edges("executor", should_execute, {
    "executor": "executor", "summarizer": "summarizer"
})
builder.add_edge("summarizer", END)

plan_execute_agent = builder.compile()

3.3 Replan（重新规划）

纯 Plan-Execute 的重大缺陷是：如果某步执行失败，后续步骤可能全部无效。Replan 变体在执行过程中监控结果质量，必要时重新触发规划器生成新计划。

四、其他重要 Agent 范式

4.1 Self-Refine（自我反思）

理论基础：LLM 具有”自我批评”能力——给模型机会审视自己的输出，通常能发现并修正错误。

生成回答 → 自我检查（反馈: "这段回答不完整，缺少...") → 改进回答 → ... → 最终回答

适用场景：写作、代码生成、翻译等可以迭代优化的任务。

关键论文：Madaan et al., “Self-Refine: Iterative Refinement with Self-Feedback” (2023)

4.2 Tree-of-Thoughts（思维树）

理论基础：将推理过程建模为树状搜索，使用 BFS/DFS 探索多条推理路径。每步生成多个候选”下一步思路”，LLM 对每个候选评分，选择最有前景的路径继续探索。

适用于需要创造性思维或”灵光一现”的任务——包括数学证明、创意写作构思、谜题解决。

关键论文：Yao et al., “Tree of Thoughts: Deliberate Problem Solving with LLMs” (2023)

4.3 Reflexion

Reflexion 在 ReAct 基础上增加了”长期记忆”机制——将失败的经验（为什么上次做错了）存储为文本记忆，在后续类似任务中注入到 prompt 中以避免重复犯同样的错误。

4.4 范式对比总结

范式 推理方式 执行方式 记忆 最佳场景
ReAct 逐步推理 逐步执行 上下文内 探索式交互
Plan-Execute 全局规划 按计划执行 无 多步骤确定性任务
Self-Refine 迭代反馈 迭代改进 无 内容生成
ToT 树状搜索 最佳路径执行 无 创造性推理
Reflexion 逐步+反思 逐步执行 长期记忆 需要经验的重复任务

五、LangGraph 深度解析

5.1 设计哲学

LangGraph 的核心抽象是基于有向图的计算模型。与传统的 DAG 流水线不同，LangGraph 支持循环边，使其天然适合 Agent 的循环推理模式。

为什么选择图而非链？

Chain（如 LangChain 的 LCEL）是线性的：A → B → C → D。它适合确定性流水线（如 RAG），但无法表达”如果 A 产生 X 去 B，产生 Y 去 C，然后都回到 A”的循环逻辑。Agent 的核心特征是循环决策，这恰好是图模型的优势。

5.2 核心概念

概念 类型 说明
StateGraph 容器 有状态图，定义 Agent 的工作流拓扑结构
Node 节点 图中的执行单元，接收 State 返回 State 的更新
Edge 边 确定性的数据流路径（A 之后永远是 B）
Conditional Edge 边 分支路由（A 之后根据条件去 B 或 C 或 D）
State 数据 图的状态对象，通过 Reducer 函数在各节点间传递和合并
Checkpointer 持久化 在每个 Superstep 后保存状态快照
Store 持久化 长期记忆存储（跨会话、跨线程）
Superstep 执行概念 一次从当前节点到下一个节点（或 END）的执行为一个 Superstep

5.3 持久执行（Durable Execution）

LangGraph 的持久执行机制是其与传统工作流引擎的关键区别：

节点A 执行 → 保存检查点到 Checkpointer
节点B 执行 → 保存检查点到 Checkpointer
节点B 中断（崩溃/人工暂停）
--- 恢复后 ---
从节点B 的检查点恢复 → 继续执行节点C

三种持久化模式：

模式 行为 适用场景
async（默认） 后台异步写检查点，不阻塞执行 高吞吐，允许少量数据丢失
sync 同步写检查点，阻塞执行直到持久化完成 金融等不允许任何丢失的场景
exit 仅在图完全结束时写一次检查点 不需要中间恢复的场景

5.4 状态设计原理

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
import operator

class AgentState(TypedDict):
    # 使用 add_messages reducer：新消息追加到列表末尾
    messages: Annotated[list, add_messages]

    # 使用 operator.add reducer：列表合并
    intermediate_steps: Annotated[list, operator.add]

    # 无 Annotated = 默认覆盖行为：每次节点返回都会覆盖
    tool_call_count: int
    phase: str

Reducer 机制是 LangGraph 状态管理的关键创新。每个字段可以定义独立的数据合并策略：

  • Annotated[list, add_messages] → 新消息追加
  • Annotated[list, operator.add] → 两个列表拼接
  • 无 Annotated → 覆盖（新值替换旧值）

5.5 节点粒度设计的权衡

粗粒度节点（合并多个操作）:
  优点：代码简洁，检查点少
  缺点：失败后重跑更多工作，不透明

细粒度节点（每个操作单独节点）:
  优点：失败后重跑少，可观测性高，可独立配置重试策略
  缺点：图结构更复杂，检查点更多

推荐原则：对外部服务调用（API、数据库）、LLM 调用、关键决策点使用独立节点。

5.6 容错与重试机制

from langgraph.errors import NodeError
from langgraph.types import Command, RetryPolicy
from langgraph.graph import StateGraph, START

class State(TypedDict):
    status: str

def charge_payment(state: State) -> State:
    raise RuntimeError("payment timeout")

def payment_error_handler(state: State, error: NodeError) -> Command:
    """Saga 补偿模式：支付失败后取消预留"""
    return Command(
        update={"status": f"compensated: {error.error}"},
        goto="rollback",
    )

graph = (
    StateGraph(State)
    .add_node(
        "charge_payment",
        charge_payment,
        retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
        error_handler=payment_error_handler,
    )
    .add_node("rollback", rollback_node)
    .add_edge(START, "charge_payment")
    .add_edge("charge_payment", "rollback")
    .compile()
)

5.7 时间旅行（Time Travel）

LangGraph 支持从历史检查点重放执行——对于调试和”撤销-重做”功能非常有用。

# 获取检查点历史
history = list(graph.get_state_history(config))

# 找到"写笑话之前"的检查点
before_joke = next(s for s in history if s.next == ("write_joke",))

# 从该点重新执行
replay_result = graph.invoke(None, before_joke.config)

5.8 子图与 Agent 组合

from langgraph.graph import StateGraph, MessagesState

def create_sub_agent(model, *, name, tools, prompt, checkpointer=None):
    """包装子 Agent 为独立子图，确保命名空间隔离"""
    agent = create_agent(
        model=model, name=name,
        tools=tools, prompt=prompt,
        checkpointer=checkpointer,
    )
    return (
        StateGraph(MessagesState)
        .add_node(name, agent)
        .add_edge("__start__", name)
        .compile()
    )

命名空间隔离的重要性：多个子 Agent 在同一个线程中运行时，如果没有命名空间隔离，它们的内部状态（消息历史）会相互污染。每个被 StateGraph 包装的子 Agent 自动获得独立的检查点命名空间。

5.9 流式输出

# 消息级流式：每当产生新消息就输出（适合聊天 UI）
for message, metadata in graph.astream(
    input={"messages": [{"role": "user", "content": "你好"}]},
    stream_mode="messages",
):
    print(message.content, end="")

# 值级流式：每个节点执行完毕输出完整状态（适合调试）
for chunk in graph.astream(
    input={"messages": [{"role": "user", "content": "3+5等于多少？"}]},
    stream_mode="values",
):
    print(chunk)

5.10 Human-in-the-Loop（人机协作）

from langgraph.types import interrupt

def approval_node(state: AgentState):
    """在关键操作前暂停，等待人类审批"""
    action = state["pending_action"]
    approval = interrupt(f"是否确认执行 {action}？(yes/no)")
    if approval.lower() != "yes":
        return {"messages": ["操作已取消"]}
    return execute_action(action)

# 在 approval_node 之前自动暂停
graph = builder.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["approval_node"]
)

六、Agent 开发实践

6.1 Agent vs Chain 决策指南

场景 选择 原因
固定流程 Chain 无决策需求，确定性流水线更简单可靠
多轮交互 + 工具 Agent 需要动态决策何时使用哪个工具
简单文档问答 Chain (RAG) RAG 流程固定，用户意图明确
复杂多步推理 Agent 需要多轮工具调用和信息整合
需要不确定次数循环 Agent Chain 无法表达循环

6.2 Agent 开发最佳实践

  1. 渐进式构建：先用 Chain 验证核心流程，确认需要动态决策后再升级为 Agent
  2. 工具描述即接口协议：LLM 完全依赖工具描述来决策调用，模糊描述 = 错误调用
  3. 循环上限：设置 max_iterations=5~10，防止死循环消耗大量 Token
  4. 完整日志：记录每一步的 Thought/Action/Observation，方便审计和调试
  5. 错误优雅降级：捕获工具异常，返回有意义的错误信息（而非 traceback）给 LLM
  6. 流式设计：实时展示推理过程，提升用户信任感和体验
  7. A/B 测试 System Prompt：不同的系统指令对 Agent 行为影响巨大

6.3 常见问题诊断

症状 可能原因 治疗方案
死循环 LLM 找不到正确答案，不断试同一工具 设置 max_iterations，添加”如果三次调用同一工具无进展则换个策略”的Prompt
工具参数错误 工具描述不清楚，或参数名歧义 使用 Pydantic schema + Field(description=…) 明确每个参数
调用不存在的工具 幻觉 在 System Prompt 中明确”你只能使用以下工具”
过早终止 LLM 过于”保守” Prompt 中添加”如果你不确定，请多调用一次工具确认”
Token 爆炸 历史消息过长 实现滑动窗口或摘要压缩
多轮后”遗忘”任务 上下文漂移 定期重述当前目标（”我们正在尝试解决…”）

七、Agent 安全与可控性

7.1 工具沙箱化

所有 Agent 调用的工具都应该在受限环境中执行。核心原则是”最小权限原则”：

  • 网络工具：限制域名白名单、超时时间
  • 文件工具：限制可访问的目录范围
  • 数据库工具：使用只读账号，限制返回行数
  • 代码执行工具：在 Docker 容器或沙箱中执行

7.2 权限分级

Level 0: 只读工具（搜索、查询、读取）—— 无需确认
Level 1: 非破坏性写入（保存草稿、创建标签）—— 需要告知用户
Level 2: 破坏性操作（删除、发送、付费）—— 需要人类明确确认

7.3 Human-in-the-Loop 的三种模式

模式 说明 适用场景
interrupt_before 在指定节点前自动暂停 高风险操作（如发送邮件前）
interrupt_after 在指定节点后自动暂停 审查结果（如生成内容后人工审核）
动态中断 程序代码中调用 interrupt() 根据运行时条件决定是否暂停

相关条目

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注