从零开始的 Agent 教程
本教程分 7 层,每一层都在上一层的代码上加一点东西。示例用 Python 和 Anthropic SDK,思路对任何模型通用。
目录
第 0 层:先搞清楚 Agent 是什么
Agent = LLM + 工具 + 循环
- 普通聊天:你问,模型答,结束。
- Agent:模型自己决定"要不要用工具、用哪个、用完再想想下一步",循环直到任务完成。
核心区别在于控制权:流程不是你写死的,而是模型在运行时动态决定的。
第 1 层:调用一次模型
import anthropic
client = anthropic.Anthropic()
MODEL = "claude-sonnet-5-5"
resp = client.messages.create(
model=MODEL,
max_tokens=1000,
messages=[{"role": "user", "content": "你好,介绍一下你自己"}],
)
print(resp.content[0].text)
要点:
messages是一个列表,每条有role(user/assistant)和content。- 模型没有记忆,每次调用你都要把完整历史传回去。这一点是后面所有内容的基础。
第 2 层:给模型一个工具(Tool Use)
模型只能输出文字,它不能真的"执行"东西。工具调用的机制是:
- 你告诉模型有哪些工具(名字、描述、参数格式)
- 模型输出"我想调用 X,参数是 Y"
- 你的代码真正执行
- 把结果还给模型
tools = [{
"name": "get_weather",
"description": "查询某个城市的当前天气",
"input_schema": {
"type": "object",
"properties": {"city": {"type": "string", "description": "城市名"}},
"required": ["city"],
},
}]
def get_weather(city: str) -> str:
return f"{city}:晴,22°C" # 先用假数据
messages = [{"role": "user", "content": "北京天气怎么样?"}]
resp = client.messages.create(model=MODEL, max_tokens=1000,
tools=tools, messages=messages)
print(resp.stop_reason) # "tool_use",表示模型想调工具
此时 resp.content 里有一个 tool_use 块,包含 id、name、input。你执行它,再把结果回传:
messages.append({"role": "assistant", "content": resp.content})
for block in resp.content:
if block.type == "tool_use":
result = get_weather(**block.input)
messages.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": block.id, # 必须对应
"content": result,
}],
})
final = client.messages.create(model=MODEL, max_tokens=1000,
tools=tools, messages=messages)
print(final.content[0].text)
这一步你已经手动走完了一轮 Agent 流程,下一层把它自动化。
第 3 层:Agent 循环(核心)
把上面的过程包进 while 循环。这就是 Agent 的本体,不到 30 行:
def run_agent(user_input: str, tools: list, tool_fns: dict, max_steps=10):
messages = [{"role": "user", "content": user_input}]
for step in range(max_steps): # 防止死循环
resp = client.messages.create(
model=MODEL, max_tokens=2000,
tools=tools, messages=messages,
)
messages.append({"role": "assistant", "content": resp.content})
if resp.stop_reason != "tool_use": # 模型认为做完了
return "".join(b.text for b in resp.content if b.type == "text")
results = []
for block in resp.content:
if block.type == "tool_use":
try:
out = tool_fns[block.name](**block.input)
except Exception as e:
out = f"工具出错:{e}" # 错误也交给模型,让它自己处理
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": str(out),
})
messages.append({"role": "user", "content": results})
return "达到最大步数,任务未完成"
print(run_agent("北京和上海哪个更热?", tools, {"get_weather": get_weather}))
运行时模型会自己决定调两次 get_weather,再比较作答。你没写"查两个城市"的逻辑。
第 4 层:多个、真实的工具
Agent 的能力上限由工具决定。加几个有用的:
import subprocess, pathlib
def read_file(path: str) -> str:
return pathlib.Path(path).read_text()[:5000]
def write_file(path: str, content: str) -> str:
pathlib.Path(path).write_text(content)
return f"已写入 {path}"
def calculator(expression: str) -> str:
# 演示用;生产环境不要直接 eval 不可信输入
return str(eval(expression, {"__builtins__": {}}))
tool_fns = {"read_file": read_file, "write_file": write_file, "calculator": calculator}
工具的 description 极其重要,模型靠它判断什么时候用。写法建议:
- 说清做什么、什么时候该用、什么时候不该用
- 参数说明具体,带例子
- 返回结果要精简,别把 10 万字塞回上下文
安全提醒:给 Agent 的能力(执行命令、写文件、发邮件)越强,越要加限制,比如白名单目录、危险操作需人工确认。
第 5 层:记忆与上下文管理
Agent 跑得越久,messages 越长,会出现三个问题:成本上升、速度变慢、超出上下文窗口。
短期记忆:压缩历史
def compress(messages, keep_last=6):
if len(messages) <= keep_last + 1:
return messages
old = messages[1:-keep_last]
summary = client.messages.create(
model=MODEL, max_tokens=500,
messages=[{"role": "user",
"content": f"用简洁要点总结以下对话中已完成的事和关键信息:\n{old}"}],
).content[0].text
return [messages[0],
{"role": "user", "content": f"[之前进展摘要]\n{summary}"},
*messages[-keep_last:]]
注意压缩时不要把 tool_use 和对应的 tool_result 拆开,否则 API 会报错。
长期记忆:做成工具,让模型自己存取
memory = {}
def save_memory(key: str, value: str): memory[key] = value; return "已保存"
def recall_memory(key: str): return memory.get(key, "没有这条记忆")
进阶做法是存入向量数据库,按语义检索(即 RAG)。
第 6 层:规划与反思
简单任务直接循环就行,复杂任务需要更有结构。
① 先规划再执行(Plan-and-Execute)
用 system prompt 引导:
SYSTEM = """你是一个任务执行 Agent。
1. 收到复杂任务时,先用 update_plan 工具列出步骤。
2. 每完成一步更新计划状态。
3. 全部完成后,检查结果是否真正满足用户需求,再给出最终答案。"""
resp = client.messages.create(model=MODEL, max_tokens=2000,
system=SYSTEM, tools=tools, messages=messages)
把 update_plan / todo 做成一个工具,是让模型"写下来再照着做"的低成本办法,实际效果很好。
② 反思(Reflection)
让模型检查自己的产出:
def critique(task, answer):
r = client.messages.create(model=MODEL, max_tokens=500, messages=[{
"role": "user",
"content": f"任务:{task}\n答案:{answer}\n请严格指出问题,没问题就回复 OK。"}])
return r.content[0].text
不通过就把批评意见喂回主循环再改一轮,通常限制 2~3 轮。
③ ReAct 模式:思考→行动→观察,不断交替。第 3 层的循环本质上就是 ReAct,现代模型的工具调用已内置了这个模式。
第 7 层:多 Agent 与生产化
多 Agent 协作:把大任务拆给专门的子 Agent,每个有独立的上下文和工具。
def research_agent(topic): # 子 Agent:只有搜索类工具
return run_agent(f"调研:{topic},输出要点", search_tools, search_fns)
def writer_agent(material): # 子 Agent:只有写作类工具
return run_agent(f"根据资料写报告:{material}", write_tools, write_fns)
# 编排者(orchestrator)把子 Agent 当作工具来调用
orchestrator_fns = {"research": research_agent, "write": writer_agent}
好处:每个子 Agent 上下文干净、职责单一。代价:成本和复杂度上升。能用单 Agent 解决就别上多 Agent。
生产环境清单
| 方面 | 要做的事 |
|---|---|
| 可靠性 | 最大步数、超时、重试、错误回传给模型 |
| 安全 | 工具权限最小化、危险操作人工确认、防提示词注入(工具返回的内容不可信) |
| 可观测 | 记录每一步的思考、工具调用、结果、token 用量 |
| 评估 | 准备一组测试任务,改了 prompt 或工具后回归测试 |
| 成本 | 简单步骤用小模型,压缩上下文,缓存重复内容 |
| 人在回路 | 关键节点让用户审批 |
学习路线总结
第1层 调用模型 → 理解 messages、无状态
第2层 工具调用 → 理解"模型决定,代码执行"
第3层 Agent 循环 → 核心,30行代码
第4层 真实工具 → 能力来源,描述要写好
第5层 记忆 → 压缩历史 + 长期存储
第6层 规划/反思 → 处理复杂任务,提升质量
第7层 多Agent/生产 → 扩展与工程化
建议的练习顺序:
- 亲手敲完第 3 层,做一个能读写本地文件的小助手
- 加一个
todo工具,观察它如何自己规划 - 故意让工具报错,看它如何自我修复
- 做一个小评测集,体会"改动是否真的变好"
另外,实际项目中也可以用现成框架(如 Claude Agent SDK、LangGraph)。但先手写一遍,你会明白框架在替你做什么。
想继续深入哪一层?我可以把某一层展开成完整可运行的项目,比如"一个能自己写代码并跑测试的 Agent"。