阶段 4:Tool Calling 与 Agent 学习笔记
阶段 4:Tool Calling 与 Agent 学习笔记
目标:理解 Agent 的本质不是“聊天机器人”,而是 模型 + 工具 + 循环 + 状态管理。
学完这一阶段,你应该能:
- 定义工具 schema;
- 让模型决定是否调用工具;
- 执行工具并把结果回传给模型;
- 设计 Agent Loop 的终止条件与错误处理;
- 区分 Workflow 与 Agent;
- 用 LangGraph 表达有状态流程;
- 理解 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 是你告诉模型:
- 工具叫什么;
- 工具能做什么;
- 需要哪些参数;
- 参数类型是什么;
- 哪些参数必填;
- 参数有什么约束。
示例:定义一个论文检索工具。
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_papers 比 tool1 好 |
| 描述明确 | 告诉模型什么时候应该用这个工具 |
| 参数少而精 | 参数越复杂,模型越容易填错 |
| 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”。
你可以先定义两个工具:
search_papers:检索 PDF 片段;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. 工具结果回传:为什么必须回传给模型
工具执行结果不是直接给用户,而是先回传给模型。
原因是模型还需要做:
- 读取工具结果;
- 判断结果是否足够;
- 必要时继续调用其他工具;
- 整理成自然语言回答;
- 添加引用来源;
- 对用户问题做最终解释。
比如工具返回:
{
"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。
好处:
- 可以恢复中断任务;
- 可以保留多轮对话上下文;
- 可以调试历史步骤;
- 可以做人类审批;
- 可以支持失败重试;
- 可以支持 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 |
生成带引用的回答 |
初学阶段先实现两个:
search_papersget_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
更多推荐
所有评论(0)