第 10 章 · 实战:从入门到综合实践(上)
实战概览
7 个递进实战,难度从 ★ 到 ★★★★★。建议按顺序做,每个实战都是下一个的基础。
| # | 难度 | 实战 | 学到 |
|---|---|---|---|
| 1 | ★ | 30 行代码跑通第一个 agent | agent loop 最小形态 |
| 2 | ★★ | 让 agent 会用多工具 | tool use 决策 |
| 3 | ★★★ | 写一个 MCP server | MCP 协议实战 |
| 4 | ★★★ | LangGraph 带反思的研究 agent | 状态机 + 反思循环 |
| 5 | ★★★★★ | 自动追踪目标算法论文的科研 agent | 全栈 agent |
| 6 | ★★★★★ | 自动找候选人并发邮件的招聘 agent | agent + 人工审核 |
| 7 | ★★★★★ | 拆解 OpenClaw / Hermes-agent 架构 | 生产级 agent 设计 |
本章先讲实战 1-4。实战 5-7 在 docs/10-practice/ 目录下各自独立成文。
实战 1 · ★ 30 行代码跑通第一个 agent
目标:跑通最小 agent,感受 agent loop。
场景:用户问天气,agent 调用 mock 天气工具回答。
代码(基于第 3 章简化):
import anthropic
client = anthropic.Anthropic() # 自动读 ANTHROPIC_API_KEY
def get_weather(city: str) -> str:
"""Mock 天气数据。"""
return f"{city} 今天 25 度晴"
def agent(task: str) -> str:
messages = [{"role": "user", "content": task}]
for _ in range(10): # 步数上限
resp = client.messages.create(
model="claude-sonnet-4-5-20250929",
max_tokens=512,
tools=[{
"name": "get_weather",
"description": "查询城市天气。当用户问天气时使用。",
"input_schema": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"]
}
}],
messages=messages
)
if resp.stop_reason == "end_turn":
return resp.content[0].text
# 处理工具调用
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
}]})
return "步数上限"
if __name__ == "__main__":
print(agent("北京今天天气怎么样?"))跑起来:
pip install anthropic
export ANTHROPIC_API_KEY=sk-ant-...
python agent.py你应该看到:agent 调 get_weather(city="北京") → 拿到 "25 度晴" → 回复"北京今天 25 度晴"。
练习:
- 加一个
calculate工具,让 agent 能算数 - 把
get_weather换成真实 API(如 OpenWeatherMap)
实战 2 · ★★ 让 agent 会用多工具
目标:agent 自己决定用哪个工具。
场景:用户问"读 README.md 然后总结",agent 调 read_file 读文件 → 总结回复。
代码:
import anthropic
from pathlib import Path
client = anthropic.Anthropic()
def read_file(path: str) -> str:
"""读本地文件。当用户想看文件内容时使用。"""
try:
return Path(path).read_text(encoding="utf-8")[:5000] # 截断防爆
except Exception as e:
return f"读取失败: {e}"
def calculate(expression: str) -> str:
"""数学计算。当用户做算术时使用。"""
try:
return str(eval(expression))
except Exception as e:
return f"错误: {e}"
def get_weather(city: str) -> str:
return f"{city} 25 度晴"
tools = [
{
"name": "read_file",
"description": "读本地文件内容。当用户想看文件、读 README、查代码时使用。",
"input_schema": {
"type": "object",
"properties": {"path": {"type": "string", "description": "文件路径"}},
"required": ["path"]
}
},
{
"name": "calculate",
"description": "执行数学计算。当用户做加减乘除、算数时使用。",
"input_schema": {
"type": "object",
"properties": {"expression": {"type": "string", "description": "数学表达式"}},
"required": ["expression"]
}
},
{
"name": "get_weather",
"description": "查询城市天气。当用户问天气时使用。",
"input_schema": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"]
}
}
]
TOOL_IMPL = {"read_file": read_file, "calculate": calculate, "get_weather": get_weather}
def agent(task: str) -> str:
messages = [{"role": "user", "content": task}]
for _ in range(10):
resp = client.messages.create(
model="claude-sonnet-4-5-20250929",
max_tokens=1024,
tools=tools,
messages=messages
)
if resp.stop_reason == "end_turn":
return resp.content[0].text
messages.append({"role": "assistant", "content": resp.content})
tool_results = []
for block in resp.content:
if block.type == "tool_use":
result = TOOL_IMPL[block.name](**block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
messages.append({"role": "user", "content": tool_results})
return "步数上限"
if __name__ == "__main__":
print(agent("读一下 README.md 然后告诉我它讲什么"))你应该看到:agent 调 read_file(path="README.md") → 拿到内容 → 总结回复。
练习:
- 加一个
write_file工具,让 agent 能写文件 - 加一个
web_search工具(用 Tavily API) - 故意给一个不存在的路径,看 agent 怎么处理错误
实战 3 · ★★★ 写一个 MCP server
目标:写一个 MCP server,在 Claude Code 里接入。
场景:写一个"笔记管理" MCP server,提供 add_note / list_notes / search_notes 三个工具。
步骤 1:建项目
mkdir notes-mcp && cd notes-mcp
pip install mcp步骤 2:写 server(server.py)
from mcp.server import MCPServer
from datetime import datetime
import json
from pathlib import Path
mcp = MCPServer("notes-mcp")
NOTES_FILE = Path("notes.json")
def load_notes():
if NOTES_FILE.exists():
return json.loads(NOTES_FILE.read_text(encoding="utf-8"))
return []
def save_notes(notes):
NOTES_FILE.write_text(json.dumps(notes, ensure_ascii=False, indent=2), encoding="utf-8")
@mcp.tool()
def add_note(title: str, content: str) -> str:
"""添加一条笔记。当用户想记录想法、写笔记、存信息时使用。
Args:
title: 笔记标题,简短
content: 笔记内容,可长
"""
notes = load_notes()
notes.append({
"id": len(notes) + 1,
"title": title,
"content": content,
"created_at": datetime.now().isoformat()
})
save_notes(notes)
return f"已添加笔记 #{len(notes)}: {title}"
@mcp.tool()
def list_notes() -> str:
"""列出所有笔记。当用户想看已有笔记、回顾记录时使用。"""
notes = load_notes()
if not notes:
return "暂无笔记"
return "\n".join(
f"#{n['id']} [{n['created_at'][:10]}] {n['title']}: {n['content'][:50]}"
for n in notes
)
@mcp.tool()
def search_notes(keyword: str) -> str:
"""搜索笔记。当用户想找某主题的笔记时使用。
Args:
keyword: 搜索关键词
"""
notes = load_notes()
matched = [n for n in notes if keyword.lower() in n["title"].lower()
or keyword.lower() in n["content"].lower()]
if not matched:
return f"没找到含 '{keyword}' 的笔记"
return "\n".join(f"#{n['id']} {n['title']}: {n['content'][:80]}" for n in matched)
if __name__ == "__main__":
mcp.run()步骤 3:本地调试
mcp dev server.py
# 自动打开 MCP Inspector,可以手动调每个工具测试步骤 4:接入 Claude Code
在项目根目录建 .mcp.json:
{
"mcpServers": {
"notes": {
"command": "python",
"args": ["notes-mcp/server.py"],
"cwd": "."
}
}
}启动 Claude Code,它会自动加载这个 MCP server。然后你可以说:
帮我记一条笔记:标题"买菜",内容"西红柿、鸡蛋、牛肉"
列出所有笔记
搜一下"买菜"的笔记Claude 会自动调用对应的 MCP 工具。
练习:
- 加一个
delete_note(id)工具 - 加一个
export_notes(format="markdown")工具 - 把笔记存到 SQLite 而不是 JSON
实战 4 · ★★★ LangGraph 带反思的研究 agent
目标:用 LangGraph 搭一个"会反思"的研究 agent。
场景:用户问一个问题,agent 搜资料 → 写答案 → 自我评估 → 不满意就重搜。
为什么用 LangGraph:复杂状态机(有循环、有条件分支)用 LangGraph 比手写 loop 清晰。
步骤 1:装包
pip install langgraph langchain-anthropic步骤 2:写 agent(research_agent.py)
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage
llm = ChatAnthropic(model="claude-sonnet-4-5-20250929")
# 定义状态
class State(TypedDict):
question: str
research: str
answer: str
critique: str
satisfied: bool
iterations: int
# 节点 1:研究
def research_node(state: State) -> State:
prompt = f"""你是研究助手。针对以下问题,给出关键事实和信息(不要写最终答案):
问题:{state['question']}
"""
resp = llm.invoke([HumanMessage(content=prompt)])
return {"research": resp.content, "iterations": state["iterations"] + 1}
# 节点 2:写答案
def answer_node(state: State) -> State:
prompt = f"""基于以下研究,回答用户问题。答案要清晰、准确、有依据。
问题:{state['question']}
研究:{state['research']}
"""
resp = llm.invoke([HumanMessage(content=prompt)])
return {"answer": resp.content}
# 节点 3:自我评估
def critique_node(state: State) -> State:
prompt = f"""你是严格的审稿人。评估以下答案的质量。
问题:{state['question']}
答案:{state['answer']}
评估维度:
1. 是否回答了问题
2. 是否有事实错误
3. 是否完整
输出 JSON:{{"satisfied": true/false, "critique": "改进建议"}}
如果答案足够好,satisfied=true;否则 false 并给改进建议。
"""
resp = llm.invoke([HumanMessage(content=prompt)])
# 简化解析
satisfied = '"satisfied": true' in resp.content.lower() or '"satisfied":true' in resp.content.lower()
return {"satisfied": satisfied, "critique": resp.content}
# 条件边:决定下一步
def should_continue(state: State) -> str:
if state["satisfied"] or state["iterations"] >= 3:
return END
return "research"
# 构建图
workflow = StateGraph(State)
workflow.add_node("research", research_node)
workflow.add_node("answer", answer_node)
workflow.add_node("critique", critique_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "answer")
workflow.add_edge("answer", "critique")
workflow.add_conditional_edges("critique", should_continue, {
"research": "research",
END: END
})
app = workflow.compile()
# 跑一下
if __name__ == "__main__":
result = app.invoke({
"question": "MCP 协议是什么?解决了什么问题?",
"research": "",
"answer": "",
"critique": "",
"satisfied": False,
"iterations": 0
})
print("=== 最终答案 ===")
print(result["answer"])
print(f"\n(迭代次数:{result['iterations']})")你应该看到:agent 跑研究 → 写答案 → 评估 → 如果不满意,重新研究 → 直到满意或达到 3 次上限。
LangGraph 的好处:
- 状态机清晰:每个节点干一件事,边决定流向
- 可视化:
app.get_graph().draw_mermaid()能生成流程图 - 持久化:加
checkpointer能保存中间状态,断点续跑
练习:
- 加一个真实搜索工具(Tavily / Brave Search),让 research 节点能搜网
- 把 critique 改成调用另一个模型(如 GPT-4o)做交叉评估
- 加一个"rewrite"节点,根据 critique 重写答案而不是重新研究
前人智慧 / Prior Art
本章 4 个实战每个都对应一个真实 Agent 产品的设计选择。理解这些对标,能让你做的练习不只是练手,而是理解行业实践。
实战 1(30 行 agent)对标 Claude Agent SDK 最小示例
本章实战 1 的 30 行代码 agent,对标 Claude Agent SDK 最小示例:
# 实战 1 的手写 loop
for step in range(10):
resp = client.messages.create(...)
if resp.stop_reason == "end_turn":
return resp.content[0].text
# 处理 tool_use# Claude Agent SDK 的等价代码
async for message in query(prompt="...", options=ClaudeAgentOptions(...)):
print(message)差别在 SDK 把 loop / tool_use / error / permission 全封装了。实战 1 的价值在让你看到封装内部--理解了 30 行,才能理解 SDK 在帮你避免写哪 30 行。
实战 2(多工具)对标早期 LangChain Agent
本章实战 2 的多工具 agent,对标早期 LangChain 的 AgentExecutor:
- 同样是 ReAct loop
- 同样是多个 tool + LLM 决策
- 同样是 tool_result 回灌
LangChain 的 AgentExecutor 把这套封装成类,但抽象较重。实战 2 的裸 Anthropic SDK 写法更透明--你能看到每一步在做什么。
工程教训:理解框架前先理解裸实现。直接用 LangChain 会让初学者以为 agent 是黑箱,先写实战 2 这种裸实现,再用 LangChain 就能看清封装的边界。
实战 3(MCP server)对标 awesome-mcp-servers 的入门 server
本章实战 3 的笔记 MCP server,对标 awesome-mcp-servers 里的入门级 server。设计要点:
- 用 fastmcp 的
@mcp.tool()装饰器 - 工具粒度合适(add/list/search 三个)
- description 写"何时使用"
- 错误返回给 LLM
这套设计是 MCP server 的标准范式。读完实战 3 再看 awesome-mcp-servers 里的 server,能快速理解它们的设计。
实战 4(LangGraph 反思)对标 Reflexion 论文
本章实战 4 的 LangGraph 反思 agent,直接对标 Reflexion 论文 的设计:
- research 节点 = Action
- answer 节点 = 输出
- critique 节点 = Reflection
- 不满意重新 research = 经验回灌
Reflexion 论文用 AlfWorld 实验证明这套循环能提升 20+ 个百分点。实战 4 把这套循环工程化到 LangGraph。
LangGraph 的贡献是把 Reflexion 的循环结构显式化为状态机--每个节点是 state 转换,边是条件分支。这种显式化让反思循环可观察、可调试、可持久化。
Andrew Ng Agentic AI 课程的对照
Andrew Ng 的 Agentic AI 课程(DeepLearning.AI,2024)和本教程的实战部分有重叠,但侧重不同:
| Andrew Ng 课程 | 本教程实战 | |
|---|---|---|
| 框架 | LangGraph / LangChain | Anthropic SDK + LangGraph |
| 实战数 | 4 个 | 7 个 |
| 难度梯度 | 平缓 | 从 ★ 到 ★★★★★ |
| 真实场景 | 较少 | 含科研 / 招聘两个真实场景 |
| 评估 | 弱 | 强调 EVALS |
Andrew Ng 课程的强项在系统讲 LangGraph 抽象,本教程的强项在真实任务驱动 + 难度梯度。两者互补。
Aider 工作流的启发
Aider 的工作流(终端 + git + CONVENTIONS.md + TODO.md)启发了本教程实战的设计:
- 实战 3 的 MCP server 用 git 管理
- 实战 5/6 的 agent 都有
.env配置 + README - 实战 7 的架构拆解用文件记录
这种"git 优先 + 文件驱动"的工作流是 Coding Agent 时代的标准实践。做实战时养成这个习惯,比单纯写代码更有价值。
SWE-bench 风格的实战 5/6 验收
本章实战 5(科研 agent)和实战 6(招聘 agent)的验收标准设计参考了 SWE-bench 风格:
- 不只看"输出像不像"
- 看"任务完成度"(找到几篇论文 / 发了几封邮件)
- 看"约束守住没"(成本 / 步数 / 人工确认)
这种"任务完成度"导向的验收比"输出质量"导向更工程化。SWE-bench 用项目测试套件验收,本教程实战用"任务完成度"验收--同一个思路的不同实现。
OpenClaw / Hermes 架构拆解的价值
实战 7 拆解 OpenClaw / Hermes-agent 架构,对标 Anthropic Multi-agent Research System 博客 的思路:
- 不自己造,先看生产级 agent 怎么设计
- 关注 orchestrator-worker 模式
- 关注 subagent 隔离
- 关注 token 成本控制
读生产级 agent 源码是学 Agent 工程的捷径。本教程选 OpenClaw / Hermes 是因为它们开源 + 设计清晰 + 含 skills/memory/loop 完整四件套。
7 个实战的难度梯度设计
本章 7 个实战的难度梯度(★ 到 ★★★★★)不是任意分级,对应 Agent 工程的能力分层:
| 难度 | 能力 | 对应章节 |
|---|---|---|
| ★ | 跑通最小 loop | 第 3 章 Agent Loop |
| ★★ | 多工具决策 | 第 4 章 Tool Use |
| ★★★ | MCP 标准化 | 第 5 章 MCP |
| ★★★ | 状态机 + 反思 | 第 2 章心智模型 |
| ★★★★★ | 全栈真实任务 | 全部章节综合 |
| ★★★★★ | 人工审核回路 | 第 8 章安全 |
| ★★★★★ | 生产级架构拆解 | 全部章节综合 |
这个梯度对应了本教程的章节顺序--每章学完做对应实战,能力逐步叠加。
实战 1-4 让你跑通了 agent 的基本形态。下一篇文章是压轴:
- 实战 5(
docs/10-practice/05-paper-tracker.md):自动追踪目标算法论文的科研 agent —— arxiv 搜 → PDF 解析 → 中文摘要 → 邮件推送 - 实战 6(
docs/10-practice/06-recruiter.md):自动找候选人并发邮件的招聘 agent —— GitHub 搜 → README 解析 → 评分 → 邮件草稿 → 人工审核 - 实战 7(
docs/10-practice/07-openclaw-hermes.md):拆解 OpenClaw / Hermes-agent 架构 —— 看生产级 agent 怎么设计