阶段 4:Tool Calling 与 Agent 学习笔记

目标:理解 Agent 的本质不是“聊天机器人”,而是 模型 + 工具 + 循环 + 状态管理
学完这一阶段,你应该能:

  1. 定义工具 schema;
  2. 让模型决定是否调用工具;
  3. 执行工具并把结果回传给模型;
  4. 设计 Agent Loop 的终止条件与错误处理;
  5. 区分 Workflow 与 Agent;
  6. 用 LangGraph 表达有状态流程;
  7. 理解 short-term memory、state、persistence 的关系。

1. 先建立核心认知:Agent 不是“会聊天”,而是“会调用工具完成任务”

普通 LLM 应用通常是:

用户问题 -> 模型生成回答 -> 返回用户

Agent 应用通常是:

用户目标
  -> 模型判断下一步
  -> 选择工具
  -> 程序执行工具
  -> 工具结果回传模型
  -> 模型继续判断
  -> 直到完成任务或停止

所以 Agent 的核心不是“模型很聪明”,而是:

LLM + Tools + Loop + State + Guardrails

其中:

组成 作用
LLM 负责理解任务、规划下一步、决定是否调用工具
Tools 外部能力,例如查数据库、查文档、搜索、发邮件、调用业务 API
Loop 让模型可以多轮决策:调用工具 -> 看结果 -> 再决定
State 保存任务过程中的上下文、工具结果、中间状态
Guardrails 权限、安全、错误处理、终止条件、人工确认

面试时可以这样说:

Agent 本质上是一个带工具调用能力的决策循环。模型负责根据当前状态决定下一步动作,应用程序负责真正执行工具,并把执行结果回传给模型。为了稳定运行,需要有状态管理、循环终止条件、错误处理和安全边界。


2. Tool Calling 是什么

Tool Calling,也叫 Function Calling,是让模型“请求调用外部函数”的机制。

注意:模型不会真的执行函数

模型做的是输出类似这样的结构化请求:

{
  "name": "search_papers",
  "arguments": {
    "query": "RAG chunking strategy",
    "top_k": 5
  }
}

真正执行的是你的后端程序:

result = search_papers(query="RAG chunking strategy", top_k=5)

执行完后,你需要把结果再传回模型:

{
  "tool_name": "search_papers",
  "result": [
    {
      "title": "RAG Survey",
      "snippet": "Chunk size affects retrieval..."
    }
  ]
}

模型再基于工具结果生成最终回答。


3. 为什么要用 Tool Calling

3.1 解决模型“不知道最新信息”的问题

模型参数里没有实时数据库、业务订单、用户权限、最新论文、最新库存等信息。

工具可以让模型访问:

  • 数据库;
  • 文件检索系统;
  • 搜索 API;
  • 公司内部系统;
  • 日历、邮件、工单、订单系统。

3.2 解决模型“不能执行动作”的问题

模型本身不能:

  • 发邮件;
  • 创建订单;
  • 修改数据库;
  • 调用支付接口;
  • 提交表单;
  • 创建日程;
  • 执行业务工作流。

Tool Calling 把这些动作交给程序执行。

3.3 降低幻觉

没有工具时,模型可能会“猜”。

有工具时,模型可以先查:

用户问:这篇论文主要讲什么?
Agent:调用 PDF 检索工具
工具:返回相关片段
Agent:基于片段回答,并附引用

这就把回答从“模型凭记忆生成”变成“基于外部证据生成”。


4. 工具 schema:模型如何知道一个工具怎么用

工具 schema 是你告诉模型:

  1. 工具叫什么;
  2. 工具能做什么;
  3. 需要哪些参数;
  4. 参数类型是什么;
  5. 哪些参数必填;
  6. 参数有什么约束。

示例:定义一个论文检索工具。

tools = [
    {
        "type": "function",
        "name": "search_papers",
        "description": "Search relevant PDF chunks from the user's paper knowledge base.",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The user's search query."
                },
                "top_k": {
                    "type": "integer",
                    "description": "Number of relevant chunks to return.",
                    "minimum": 1,
                    "maximum": 10
                }
            },
            "required": ["query", "top_k"],
            "additionalProperties": False
        },
        "strict": True
    }
]

4.1 schema 设计原则

原则 说明
工具名清晰 search_paperstool1
描述明确 告诉模型什么时候应该用这个工具
参数少而精 参数越复杂,模型越容易填错
enum 限制选项 例如 source_type: ["paper", "note", "web"]
必填字段明确 避免模型漏参数
关闭额外字段 additionalProperties: false
使用 strict 模式 尽量让参数严格符合 schema

5. Tool Calling 的完整执行流程

完整流程可以理解为 6 步:

1. 用户输入任务
2. 程序把用户输入 + 工具列表发给模型
3. 模型判断是否需要工具
4. 如果需要,模型返回 function_call
5. 程序解析参数并执行本地函数
6. 程序把 function_call_output 回传给模型
7. 模型基于工具结果继续回答或继续调用工具

5.1 伪代码

messages = [{"role": "user", "content": "帮我查一下 RAG chunk size 怎么设计"}]

while True:
    response = call_llm(messages, tools=tools)

    if response.has_tool_call():
        tool_call = response.tool_call
        name = tool_call.name
        args = json.loads(tool_call.arguments)

        result = execute_tool(name, args)

        messages.append(response.tool_call_message)
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(result, ensure_ascii=False)
        })

        continue

    else:
        return response.final_answer

6. 最小示例:定义两个工具

假设你做的是“论文 / PDF 知识库问答 Agent”。

你可以先定义两个工具:

  1. search_papers:检索 PDF 片段;
  2. get_paper_metadata:根据 paper_id 查询论文元信息。

6.1 工具 1:检索论文片段

def search_papers(query: str, top_k: int = 5) -> dict:
    """
    从向量数据库中检索最相关的论文片段。
    """
    # 这里实际项目中会调用 Chroma / FAISS / Milvus
    return {
        "query": query,
        "results": [
            {
                "paper_id": "rag_001",
                "title": "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks",
                "page": 3,
                "chunk": "RAG combines parametric memory with non-parametric memory..."
            }
        ]
    }

6.2 工具 2:查询论文元信息

def get_paper_metadata(paper_id: str) -> dict:
    """
    根据 paper_id 获取论文标题、作者、年份等元信息。
    """
    metadata_db = {
        "rag_001": {
            "title": "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks",
            "authors": ["Patrick Lewis", "Ethan Perez", "Aleksandra Piktus"],
            "year": 2020,
            "source": "PDF knowledge base"
        }
    }
    return metadata_db.get(paper_id, {"error": "paper not found"})

6.3 工具路由函数

def execute_tool(name: str, args: dict) -> dict:
    if name == "search_papers":
        return search_papers(**args)

    if name == "get_paper_metadata":
        return get_paper_metadata(**args)

    return {
        "error": "unknown_tool",
        "message": f"Tool {name} is not registered."
    }

7. 工具结果回传:为什么必须回传给模型

工具执行结果不是直接给用户,而是先回传给模型。

原因是模型还需要做:

  1. 读取工具结果;
  2. 判断结果是否足够;
  3. 必要时继续调用其他工具;
  4. 整理成自然语言回答;
  5. 添加引用来源;
  6. 对用户问题做最终解释。

比如工具返回:

{
  "results": [
    {
      "title": "RAG Survey",
      "page": 5,
      "chunk": "Smaller chunks improve precision but may lose context..."
    }
  ]
}

模型最终回答应该是:

论文中提到,较小的 chunk 通常能提高检索精度,但可能丢失上下文。
因此实际项目中常用 300-800 tokens 的 chunk,并设置一定 overlap。
来源:RAG Survey, page 5

8. Agent Loop:Agent 的循环控制

Agent Loop 是阶段 4 最核心的内容。

一个典型 Agent Loop 是:

while not done:
    让模型根据当前 state 决定下一步
    如果模型要调用工具:
        执行工具
        把结果写入 state
    如果模型要回答:
        输出最终答案
        done = True

8.1 Agent Loop 里必须有终止条件

如果没有终止条件,Agent 可能会无限调用工具。

常见终止条件:

终止条件 说明
模型返回 final answer 最常见
达到最大步数 例如最多 5 次工具调用
达到最大 token / 成本 防止费用失控
工具连续失败 例如同一工具失败 3 次停止
用户请求取消 前端或后端中断
需要人工确认 敏感操作暂停

示例:

MAX_STEPS = 5

for step in range(MAX_STEPS):
    response = call_llm(state)

    if response.type == "final_answer":
        return response.content

    if response.type == "tool_call":
        result = execute_tool(response.name, response.args)
        state["tool_results"].append(result)

return "任务未能在最大步骤内完成,请缩小问题范围或稍后重试。"

9. 错误处理:Agent 稳不稳定,关键看这里

Agent 系统中最容易出问题的地方不是“模型不会回答”,而是工具调用链路。

9.1 常见错误

错误类型 例子 处理方式
参数缺失 少了 paper_id 返回参数错误,让模型修正
参数类型错误 top_k="five" schema 限制 + Pydantic 校验
工具不存在 模型调用未注册工具 工具白名单
工具超时 检索服务卡住 timeout + retry
工具返回空 没查到文档 让模型换查询词或说明未找到
权限不足 用户查无权限文件 返回权限错误,不能泄露数据
敏感操作 发邮件、删数据 人工确认

9.2 工具错误结果也要结构化

不要只返回:

出错了

应该返回:

{
  "ok": false,
  "error_code": "VALIDATION_ERROR",
  "message": "top_k must be between 1 and 10",
  "retryable": true
}

这样模型更容易判断下一步:

  • 参数错:重新调用;
  • 权限错:停止;
  • 超时:重试;
  • 没结果:换 query;
  • 敏感操作:请求用户确认。

10. Workflow vs Agent:这是面试高频考点

10.1 Workflow 是什么

Workflow 是固定流程。

例如:

上传 PDF
  -> 解析 PDF
  -> 切块
  -> embedding
  -> 入库
  -> 返回成功

流程基本固定,不需要模型自由决策。

特点:

  • 路径预先写死;
  • 顺序明确;
  • 适合稳定业务流程;
  • 可控性强;
  • 易测试、易排查。

10.2 Agent 是什么

Agent 是动态决策流程。

例如:

用户问:帮我分析这几篇论文的技术路线,并总结它们的区别
  -> 模型先检索论文
  -> 发现结果不够
  -> 再查 metadata
  -> 再检索某一篇论文的实验部分
  -> 再组织对比表
  -> 最后回答

特点:

  • 路径不是完全固定;
  • 模型决定下一步;
  • 可能多次调用工具;
  • 适合开放式复杂任务;
  • 需要更强的安全和终止控制。

10.3 对比表

维度 Workflow Agent
流程 固定 动态
谁决定下一步 程序 模型 + 程序约束
可控性 中等,需要 guardrails
灵活性 一般
适合场景 确定性业务流程 开放式任务、多工具决策
风险 工具误用、循环失控、成本失控
测试难度 较低 较高

10.4 面试回答模板

Workflow 更适合流程明确、路径固定的任务,比如 PDF 入库、订单状态查询。Agent 更适合目标明确但路径不固定的任务,比如让系统自己决定先检索、再查数据库、再调用分析工具。实际生产中我不会盲目使用 Agent,而是优先用 Workflow 承载确定流程,只在需要动态决策的局部引入 Agent。


11. LangGraph 是什么

LangGraph 是一个用于构建有状态 Agent / Workflow 的图编排框架。

它把应用拆成三个核心元素:

State:共享状态
Node:执行逻辑
Edge:决定流转

可以简单理解:

State 是数据
Node 是步骤
Edge 是下一步怎么走

11.1 为什么需要 LangGraph

普通 Agent Loop 你可以自己用 while 写,但复杂后会遇到问题:

  • 分支越来越多;
  • 状态散落在代码里;
  • 多工具调用难管理;
  • 重试和错误恢复复杂;
  • 需要人工审批;
  • 需要中断后恢复;
  • 需要调试每一步;
  • 需要多 Agent 协作。

LangGraph 的价值是把流程显式建模成图,让每一步可观察、可控制、可持久化。


12. LangGraph 三个核心概念

12.1 State

State 是整个图运行时共享的数据结构。

例如论文问答 Agent 的 state:

from typing import TypedDict, List, Optional

class AgentState(TypedDict):
    question: str
    query: Optional[str]
    retrieved_chunks: List[dict]
    answer: Optional[str]
    error: Optional[str]
    step_count: int

State 可以保存:

  • 用户问题;
  • 历史消息;
  • 检索结果;
  • 工具返回值;
  • 中间分析;
  • 错误信息;
  • 当前步骤数;
  • 是否需要人工确认。

12.2 Node

Node 是一个函数,负责执行一步逻辑。

def retrieve_node(state: AgentState) -> AgentState:
    chunks = search_papers(state["question"], top_k=5)
    return {
        "retrieved_chunks": chunks,
        "step_count": state["step_count"] + 1
    }

Node 可以是:

  • 普通 Python 函数;
  • LLM 调用;
  • 工具调用;
  • RAG 检索;
  • 数据库查询;
  • 规则判断;
  • 人工审批。

12.3 Edge

Edge 决定节点之间怎么流转。

固定 Edge:

A -> B -> C

条件 Edge:

retrieve -> 如果结果足够 -> answer
retrieve -> 如果结果不足 -> rewrite_query

13. LangGraph 最小流程示例

下面是一个“检索 -> 回答”的简单 workflow。

from typing import TypedDict, List, Optional
from langgraph.graph import StateGraph, START, END

class RAGState(TypedDict):
    question: str
    chunks: List[dict]
    answer: Optional[str]

def retrieve(state: RAGState):
    chunks = search_papers(state["question"], top_k=5)
    return {"chunks": chunks}

def answer(state: RAGState):
    prompt = f"""
    请基于以下片段回答问题。
    问题:{state["question"]}
    片段:{state["chunks"]}
    """
    final_answer = call_llm(prompt)
    return {"answer": final_answer}

builder = StateGraph(RAGState)

builder.add_node("retrieve", retrieve)
builder.add_node("answer", answer)

builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "answer")
builder.add_edge("answer", END)

graph = builder.compile()

result = graph.invoke({
    "question": "RAG 为什么能减少幻觉?",
    "chunks": [],
    "answer": None
})

print(result["answer"])

14. 带条件分支的 LangGraph 示例

比如:

  • 如果检索结果为空,重写 query;
  • 如果检索结果足够,直接回答。
class RAGState(TypedDict):
    question: str
    query: str
    chunks: List[dict]
    answer: Optional[str]
    retry_count: int

def retrieve(state: RAGState):
    chunks = search_papers(state["query"], top_k=5)
    return {"chunks": chunks}

def rewrite_query(state: RAGState):
    new_query = call_llm(f"请改写这个检索 query:{state['query']}")
    return {
        "query": new_query,
        "retry_count": state["retry_count"] + 1
    }

def answer(state: RAGState):
    answer = call_llm(f"基于片段回答:{state['chunks']}")
    return {"answer": answer}

def route_after_retrieve(state: RAGState):
    if len(state["chunks"]) > 0:
        return "answer"

    if state["retry_count"] >= 2:
        return "answer"

    return "rewrite_query"

builder = StateGraph(RAGState)

builder.add_node("retrieve", retrieve)
builder.add_node("rewrite_query", rewrite_query)
builder.add_node("answer", answer)

builder.add_edge(START, "retrieve")
builder.add_conditional_edges(
    "retrieve",
    route_after_retrieve,
    {
        "answer": "answer",
        "rewrite_query": "rewrite_query"
    }
)
builder.add_edge("rewrite_query", "retrieve")
builder.add_edge("answer", END)

graph = builder.compile()

这个例子已经体现了 Agent 的核心特征:

检索 -> 判断 -> 不够就改写 -> 再检索 -> 回答

15. Memory、State、Persistence 的区别

这三个概念很容易混。

15.1 State

State 是当前运行过程中的状态。

例如:

{
  "question": "RAG chunk 怎么设计?",
  "chunks": [...],
  "retry_count": 1
}

它回答的是:

当前这次任务运行到哪里了?中间结果是什么?

15.2 Memory

Memory 是模型或 Agent 能记住的上下文。

分两类:

类型 含义 示例
Short-term memory 单次会话 / 单个 thread 内的记忆 当前对话历史、刚刚上传的文件
Long-term memory 跨会话保存的信息 用户偏好、长期项目背景、常用设置

15.3 Persistence

Persistence 是“把 state / memory 保存下来”的机制。

例如保存到:

  • 内存;
  • SQLite;
  • Postgres;
  • Redis;
  • 文件;
  • LangGraph checkpointer。

它回答的是:

程序重启、请求中断、用户下次回来时,状态还能不能恢复?

15.4 三者关系

State = 当前状态数据
Memory = 可被后续交互使用的上下文
Persistence = 保存和恢复状态/记忆的机制

16. Short-term memory

Short-term memory 是会话级记忆。

比如同一个用户连续问:

用户:帮我总结这篇 RAG 论文
助手:总结如下...
用户:它的方法和传统检索有什么区别?

第二个问题里的“它”依赖前文。

如果没有 short-term memory,模型不知道“它”指哪篇论文。

在 LangGraph 中,short-term memory 通常作为 state 的一部分,并通过 checkpointer 按 thread_id 保存。


17. State Persistence

State Persistence 是把 graph 的运行状态保存为 checkpoint。

好处:

  1. 可以恢复中断任务;
  2. 可以保留多轮对话上下文;
  3. 可以调试历史步骤;
  4. 可以做人类审批;
  5. 可以支持失败重试;
  6. 可以支持 time travel / replay。

示例思路:

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()

graph = builder.compile(checkpointer=checkpointer)

config = {
    "configurable": {
        "thread_id": "user_123_session_001"
    }
}

graph.invoke(
    {"messages": [{"role": "user", "content": "你好,帮我分析论文"}]},
    config=config
)

重点:

  • thread_id 用来区分不同会话;
  • 同一个 thread_id 可以恢复之前状态;
  • 生产环境不要只用内存 checkpointer,要考虑持久化存储。

18. 阶段 4 推荐小项目:多工具论文 Agent

你可以在阶段 3 的 PDF RAG 项目上升级。

18.1 原来的 RAG 系统

用户问题
  -> 检索 PDF chunk
  -> 拼 prompt
  -> 模型回答

18.2 升级后的 Agent 系统

用户问题
  -> Agent 判断是否需要检索
  -> 调用 search_papers
  -> 如需要,调用 get_paper_metadata
  -> 如结果不够,改写 query 再检索
  -> 生成带引用答案
  -> 保存本轮 state

18.3 建议工具

工具 功能
search_papers 从向量库检索相关 chunk
get_paper_metadata 查询论文标题、作者、年份
list_uploaded_papers 列出知识库已有论文
summarize_paper_section 总结某篇论文某个章节
compare_papers 对比多篇论文
generate_citation_answer 生成带引用的回答

初学阶段先实现两个:

  1. search_papers
  2. get_paper_metadata

19. 项目架构建议

app/
  main.py                      # FastAPI 入口
  agent/
    graph.py                   # LangGraph 流程
    state.py                   # AgentState 定义
    nodes.py                   # retrieve / answer / rewrite 等节点
    tools.py                   # 工具定义和执行
    prompts.py                 # prompt 模板
  rag/
    loader.py                  # PDF 加载
    splitter.py                # 文档切块
    embeddings.py              # 向量化
    vectorstore.py             # Chroma / FAISS
    retriever.py               # 检索逻辑
  schemas/
    request.py                 # Pydantic 请求体
    response.py                # Pydantic 响应体
  config.py                    # 环境变量

20. 面试高频问题与回答

Q1:Tool Calling 和普通 prompt 有什么区别?

普通 prompt 是让模型直接生成答案;Tool Calling 是让模型在需要外部能力时输出结构化工具调用请求,由程序执行工具,再把结果回传给模型。它适合需要查数据库、检索文档、调用业务系统或执行动作的场景。


Q2:模型会不会真的执行函数?

不会。模型只会生成工具调用请求,包括工具名和参数。真正执行函数的是应用程序。应用程序执行后,需要把工具结果回传给模型。


Q3:为什么 Agent 需要循环?

因为很多任务不是一次工具调用就能完成。模型需要根据工具返回结果继续判断下一步,比如检索结果不足时改写 query,再检索;查到 paper_id 后再查 metadata;最后整合回答。


Q4:Agent 如何防止无限循环?

可以设置最大步数、最大工具调用次数、最大成本、工具连续失败次数、超时时间,并要求模型在满足条件时输出 final answer。同时对工具错误做结构化返回,让模型知道什么时候可以重试,什么时候必须停止。


Q5:Workflow 和 Agent 的区别是什么?

Workflow 是固定路径,适合确定性流程;Agent 是动态路径,适合开放式任务。生产中应该优先使用 Workflow 保证稳定性,只在需要模型动态决策的地方引入 Agent。


Q6:LangGraph 中 State、Node、Edge 分别是什么?

State 是共享状态,保存当前任务的数据;Node 是执行逻辑的函数,可以是 LLM 调用、工具调用或普通代码;Edge 决定节点之间如何流转,可以是固定边,也可以是条件分支。


Q7:Short-term memory 和 long-term memory 有什么区别?

Short-term memory 是单个会话或 thread 内的记忆,比如当前对话历史。Long-term memory 是跨会话保存的信息,比如用户偏好、长期项目背景或历史事实。


Q8:Persistence 有什么用?

Persistence 用来保存 graph state。它可以支持多轮对话恢复、中断恢复、人类审批、错误重试、历史调试和 replay。没有 persistence,服务重启或请求中断后状态就丢失。


21. 你这一阶段最应该掌握的最小闭环

不用一开始就做复杂多 Agent。

你只需要先完成这个闭环:

用户问题
  -> 模型判断调用 search_papers
  -> 程序执行 search_papers
  -> 工具结果回传模型
  -> 模型生成带引用回答

然后再升级:

检索为空
  -> rewrite_query
  -> 再检索
  -> 最多重试 2 次
  -> 回答或说明未找到

最后再用 LangGraph 表达:

START
  -> decide
  -> retrieve
  -> route
      -> answer
      -> rewrite_query -> retrieve
  -> END

22. 学习检查清单

学完本阶段后,你应该能回答:

  • Tool Calling 是什么?
  • 工具 schema 包含哪些部分?
  • 为什么模型不直接执行工具?
  • 工具调用结果为什么要回传给模型?
  • Agent Loop 的基本结构是什么?
  • 如何设计终止条件?
  • 如何处理工具错误?
  • Workflow 和 Agent 的区别是什么?
  • LangGraph 的 State / Node / Edge 是什么?
  • 什么是 short-term memory?
  • 什么是 state persistence?
  • 如何把阶段 3 的 RAG 项目升级为 Agent 项目?

23. 阶段 4 学习顺序建议

建议按照这个顺序学:

1. 先理解 Tool Calling 原理
2. 手写两个 Python 工具
3. 写工具 schema
4. 写一个简单 Agent Loop
5. 加入错误处理和最大步数
6. 把 RAG 检索包装成工具
7. 学 LangGraph 的 State / Node / Edge
8. 用 LangGraph 重写 Agent Loop
9. 加 short-term memory
10. 加 persistence
Logo

欢迎加入DeepSeek 技术社区。在这里,你可以找到志同道合的朋友,共同探索AI技术的奥秘。

更多推荐