- 教程
- 文档
- AI Agent
- 人工智能
- 大模型
【免费下载链接】awesome-agentic-ai-zh
A trilingual (繁中 / English / 简中) learning roadmap for agentic AI: from LLM basics to multi-agent systems, with 240+ curated resources and hands-on examples. 中文 AI agent 學習地圖。
本文以 awesome-agentic-ai-zh 仓库 Stage 3 的「练习 3:从零实现 ReAct(不用 framework)」为骨架,完整拆解如何用约 70 行 Python 亲手写出「思考 → 调用工具 → 观察结果 → 再思考」的 Agent 循环(对应 Stage 3 — 工具使用与第一个 Agent Loop)。读完本文,你将掌握 OpenAI 与 Anthropic 两套 Tool Use 消息协议的差异、messages累积与tool_use_id配对机制、max_iter安全阀设计,以及基于 AST allowlist 的防eval注入安全计算器,并能直接运行本仓库的 starter 与离线 Mock 测试完成验证。
为什么必须从零写一次 ReAct
ReAct(Reasoning + Acting)是现代 Agent 的基础 pattern,其核心就是一条朴素的循环:
while not done: thought = LLM 看完目前 context、讲出下一步要做什么 action = LLM 调用一个 tool observation = tool 执行结果、喂回去给 LLMLangGraph、CrewAI 这类框架把这个 loop 藏在了框架内部。只有自己亲手写过一次,你才能真正回答下面四个问题:
- 为什么 messages array 一直长:因为每一轮的 assistant 回复和 tool 执行结果都必须追加进历史,LLM 才有完整的上下文决定下一步;如果丢掉任何一环,模型就"失忆"了。
- tool_use_id 跟 tool_result 怎么配对:一次回复里可能同时发起多个工具调用,每个调用有唯一 ID,工具结果必须回带该 ID,模型才知道哪个结果对应哪个调用。
- stop_reason 为什么是
tool_use或end_turn:模型一回合的输出要么是"请求调用工具"(需要继续循环),要么是"给出最终答案"(循环结束),两个终止信号对应两种分支。 - max_iter 为什么是 safety net:工具结果写得不好时模型可能无限调用工具,迭代上限是必须存在的硬性保险。
本练习的 starter.py 用约 70 行 Python 把这些全部交代清楚;Stage 3 章节 还给出了对应的 13 行最小循环骨架,供对照理解。学习模式遵循 docs/HOW_TO_USE.md 的"每次只改一件事"方法:先运行 starter 与测试,再做一个小改动,用测试结果验证你的理解。
怎么跑:两条运行路径
练习目录提供两条并行的运行路径,共享同一套工具定义与循环逻辑,只在 SDK 与消息协议上不同(详见下文源码走查)。依赖统一声明在 requirements.txt:
openai>=3.5,<4 anthropic>=1.1,<2 # Only Anthropic starters need this package.Path A(默认、本机免费):Ollama + qwen2.5:3b
pip install -r requirements.txt ollama pull qwen2.5:3b ollama serve python starter.py- 模型默认值:
MODEL = os.environ.get("MODEL", "qwen2.5:3b")(starter.py),可用环境变量MODEL覆盖。 - 客户端构造:
OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")(starter.py),即通过 Ollama 提供的 OpenAI 兼容端点访问本机模型,api_key 仅为占位。 - 预算:$0 API 费用;不包含硬件、内存与电力成本,Ollama 首次 pull 模型需要下载空间(Stage 3–6 默认 tag
qwen2.5:3b约 1.9 GB,核查日期 2026-08-31,详见 examples/README.zh-Hans.md)。
Path B(Anthropic、云端比较):Claude
pip install -r requirements.txt $env:ANTHROPIC_API_KEY = "your-key" python starter_anthropic.py- 模型默认值:
MODEL = os.environ.get("MODEL", "claude-haiku-4-5-20251001")(starter_anthropic.py)。 - 预算:每次先预留$0.05。实际费用按
输入 tokens × $1 / 1,000,000 + 输出 tokens × $5 / 1,000,000计算,Tool Use 还会加入 prompt tokens;价格查核日:2026-08-27。云端调用可能使用额度或产生费用,运行前查看当天官方 pricing 页面并设置上限,不要把 key 写进程序或 commit(examples/README.zh-Hans.md)。
两条路径均需注意:Windows 的 cp950 控制台无法直接输出中文与 emoji,因此两个 starter 都通过sys.stdout.reconfigure(encoding="utf-8", errors="replace")强制 UTF-8 输出(starter.py)。
预期看到(Path A、本机)
❓ 问题:'台北人口' 除以 '纽约人口'、答案保留 4 位小数。 ------------------------------------------------------------ [step 0] thought: 我先查台北人口... tool: lookup_fact({'query': '台北人口'}) → 2602000 [step 1] thought: 接着查纽约人口... tool: lookup_fact({'query': '纽约人口'}) → 8336000 [step 2] thought: 计算比例... tool: calculator({'expression': '2602000 / 8336000'}) → 0.3121... [step 3] thought: 答案是 0.3122 ------------------------------------------------------------ ✅ 最终答案:台北人口除以纽约人口约 0.3122 共 4 轮 ✅ 练习 3 通过 — ReAct loop 自己连用了 lookup_fact 跟 calculator这展示了 ReAct 的核心价值:单一 LLM 调用无法可靠完成的"查两个事实 + 精确计算"任务,通过多轮工具调用即可拆解完成。
不花钱验证程序逻辑:离线 Mock 测试
运行真实模型之前,先跑离线测试验证循环逻辑(两条路径各自对应一个测试文件):
python test.py # 验 Path A (Ollama) starter.py 逻辑 python test_anthropic.py # 验 Path B (Anthropic) starter_anthropic.py 逻辑两条 test 都用unittest.mock、不打真 API、$0/run。区别在于 Mock 响应的形状:test.py用 OpenAI-compat response shape(choices[0].message.tool_calls+finish_reason),test_anthropic.py用 Anthropic content blocks(content中type="text"与type="tool_use"块 +stop_reason)。
以 test.py 为例,其通过SimpleNamespace构造假响应对象、用MagicMock取代 OpenAI client,例如test_react_loop_multi_step用side_effect依次注入"查台北 → 查纽约 → 算比例 → stop 给答案"四轮假响应,断言最终答案含0.3122、总步数为 4、工具调用序列恰为["lookup_fact", "lookup_fact", "calculator"](test.py)。README 中给出的预期输出:
✅ test_calculator_basic ✅ test_calculator_rejects_eval_injection ✅ test_lookup_fact ✅ test_react_loop_single_tool_call ✅ test_react_loop_multi_step ✅ test_react_loop_respects_max_iter 🎉 全部通过 — 你的 ReAct loop 逻辑正确源码中实际还包含两组额外断言:test_invalid_tool_call_is_returned_as_data(非法 JSON、未知工具名、多余参数、参数类型错误、空字符串参数均返回error:前缀)与test_non_stop_terminal_reason_is_not_a_final_answer(finish_reason == "length"时final必须为None且truncated为True,防止把截断的半截话当作答案);test_anthropic.py 则额外验证了错误 tool_result 会带is_error: true标记回传(test_anthropic.py)。
Stage 3 章节给出的练习 3 完成条件(stages/03-tool-use-and-hello-agent.zh-Hans.md)是:测试能证明"没有 tool call 就停止"与"超过MAX_STEPS会报错"——这正是test_react_loop_single_tool_call与test_react_loop_respects_max_iter所覆盖的。
程序结构走查(源码级)
README 用一张表格概括了程序的分段结构,本文结合源码逐段展开:
| 段 | 行 | 在做什么 |
|---|---|---|
tool_calculator | ~30-40 | 安全的计算器(whitelist 过滤、避免eval漏洞) |
tool_lookup_fact | ~42-50 | 假事实库(教学用、避免依赖外部 API) |
TOOLS_SPEC | ~52-75 | tool schema 给 LLM 看 |
TOOL_IMPL | ~77-80 | name → callable 对应表(dispatch) |
react_loop | ~85-130 | 主循环、含 max_iter safety、messages累积、tool result 接回去 |
安全计算器:AST operator allowlist 而非 eval
tool_calculator及其底层_evaluate_arithmetic(starter.py)是练习的安全重点,实现思路是把模型文本解析成 AST,只允许白名单内的算子求值,绝不把模型文字当代码执行。具体防御包括四层:
- 长度上限:表达式长度超过
MAX_EXPRESSION_LENGTH = 200直接报错(starter.py)。 - AST 规模上限:
ast.walk统计节点数,超过MAX_AST_NODES = 50拒绝(防止超长嵌套表达式耗尽资源)。 - 深度上限:递归求值时深度超过
MAX_AST_DEPTH = 12拒绝。 - 数值边界:中间值与结果都须满足
_within_bounds——整数绝对值不超过MAX_ABS_NUMBER = 1_000_000_000_000,浮点数必须math.isfinite且绝对值在界内(starter.py),杜绝10**1000这类溢出。
白名单只放行ast.Constant(int/float)、ast.UAdd、ast.USub、ast.BinOp下的Add/Sub/Mult/Div,且显式拦截除零(starter.py);其他任何节点(包括**、//、函数调用、属性访问)一律抛出ValueError,由tool_calculator转成error: ...字符串返回给模型(starter.py)。对应测试test_calculator_rejects_unsafe_expressions一次性验证了2 ** 3、7 // 2、10 ** 1000、__import__('os').system('ls')、1 / 0等注入向量全部被拒(test.py)。README 特别强调:不要靠字符 whitelist 或ast.literal_eval做算术——字符过滤可被编码绕过,literal_eval只解析不执行但无法约束表达式复杂度,必须用明确的 AST operator allowlist 并同时限制输入长度、AST 深度、节点数与数字/结果大小。
假事实库:教学用、零外部依赖
tool_lookup_fact内置三条假事实(台北人口 2602000、纽约人口 8336000、光速 299792458 m/s),未命中时返回unknown: ...前缀(starter.py)。这样练习不依赖任何外部 API 或数据库,任何机器都能复现。
TOOLS_SPEC:两套 SDK 的 schema 差异
TOOLS_SPEC是喂给 LLM 的工具说明。注意两个文件格式不同:
- Path A(OpenAI-compatible,starter.py):工具包裹在
{"type": "function", "function": {"name": ..., "description": ..., "parameters": {...}}}里,参数用properties+required声明,并设置"additionalProperties": False。 - Path B(Anthropic,starter_anthropic.py):直接是
{"name": ..., "description": ..., "input_schema": {...}}顶层结构。
两边参数对象结构相似,但外层包装与键名(parametersvsinput_schema)不同——不要直接复制旗标名称,这也是本项目刻意保留两条路径的原因之一。
TOOL_IMPL 与 execute_tool:白名单派发 + 输入校验
TOOL_IMPL是name → callable的字典映射(dispatch 表):calculator与lookup_fact分别用 lambda 取出inp["expression"]/inp["query"]调用实现(starter.py)。
在派发之前,execute_tool扮演了"不信任输入"的校验层(starter.py):
- 工具名不在
TOOL_IMPL中 → 返回error: tool not allowed: ...; - 参数必须能
json.loads成对象(Path A 的 arguments 是字符串),否则报 JSON 错误; - 参数必须是 dict,且恰好只包含 schema 声明的那一个字段(
set(args) != {field}即拒绝多余/缺失字段); - 字段值必须是非空字符串;
- 实现抛出
KeyError/TypeError/ValueError时统一转成error: invalid arguments: ...。
返回值是三元组(args, observation, is_error),其中observation.startswith("error:")会作为 is_error 标记,供 Anthropic 路径在 tool_result 块上设置is_error。
react_loop:主循环与两套消息协议
react_loop(question, max_iter=6, client=None)返回{final, trace, steps, ...}(starter.py),每一轮:
- 携带完整
messages(含tools=TOOLS_SPEC)调用 LLM; - 取出 assistant 文本与 tool_calls;
- 把 assistant 的完整回复(含 tool_calls)追加进 messages——OpenAI 格式下需重建
{"role": "assistant", "content": ..., "tool_calls": [{"id", "type", "function": {"name", "arguments"}}]}; - 若没有 tool_calls:
finish_reason == "stop"时返回最终答案;否则返回terminal_reason并标记truncated(如length); - 若有 tool_calls:逐个执行
execute_tool,把{"role": "tool", "tool_call_id": tc.id, "content": obs}追加回 messages——这就是 tool_use_id 与 tool_result 的配对方式; - 循环跑满
max_iter仍未收尾则返回final=None, truncated=True。
Path B 的对应实现(starter_anthropic.py)协议差异清晰可见:
- 终止信号是
stop_reason:end_turn表示完成,max_tokens表示截断(同样不当作最终答案); - assistant 回复整体原样追加:
messages.append({"role": "assistant", "content": resp.content}); - 工具结果以 content block 形式回传:
{"type": "tool_result", "tool_use_id": call.id, "content": obs},错误时加"is_error": True,并作为user 消息追加:messages.append({"role": "user", "content": tool_results})。
trace只记录可观察的 assistant 文本、工具名、输入与结果({step, assistant_text, tool, tool_input, obs}),不记录私有 Chain-of-Thought——这是仓库刻意维持的日志契约(stages/03-tool-use-and-hello-agent.zh-Hans.md)。
常见坑与防御
README 列出四个最经典的坑,源码均有对应防御:
- 忘记把 assistant response 加进 messages→ 下一轮 LLM 看不到自己上一轮讲过什么,会 loop forever。防御:
messages.append(assistant_entry)在每次调用后立即执行(starter.py)。 - tool_result 没带
tool_use_id→ LLM 无法配对哪个 result 对应哪个 call。防御:OpenAI 路径用"tool_call_id": tc.id,Anthropic 路径用"tool_use_id": call.id。 while True没 max_iter→ 工具结果写得不好时 LLM 无限调用。防御:for step in range(max_iter)硬性封顶,跑满即truncated=True返回;test_react_loop_respects_max_iter用永不收尾的假响应验证了这一点(test.py)。- 未过滤的 eval→ 模型文本可能造成 RCE。防御:AST operator allowlist + 长度/深度/节点数/数值边界四重限制(见上文安全计算器小节)。
换模型与扩展方向
换模型
Path B 预设用固定 IDclaude-haiku-4-5-20251001,想比较 sonnet 时:
$env:MODEL = "claude-sonnet-5"; python starter_anthropic.py或在starter_anthropic.py改MODEL = ...那行(Path A 同样支持$env:MODEL覆盖,例如切到更大尺寸的本地模型)。注意:换模型会影响工具调用格式的遵循程度与答案质量,属于 docs/HOW_TO_USE.md 六步循环里"每次只改一件事"的独立变量。
加更多 tool
在TOOLS_SPEC+TOOL_IMPL各补一个 entry 即可:schema 描述新工具的参数,TOOL_IMPL登记 name → callable,TOOL_FIELDS声明参数键名,execute_tool的校验逻辑自动生效。
加 streaming
把client.messages.create(...)换成with client.messages.stream(...) as s:,边跑边打印 token,观察工具调用文本的流式输出。
加 prompt cache
在system=或tools=上带cache_control={"type":"ephemeral"},重复 call 可省约 90% 输入 token(成本结构中的输入侧开销,与 Path B 的计费方式直接相关)。
接框架
用 LangGraph 或 Pydantic AI 重写同一任务,对照观察框架如何把这 70 行循环"藏"进声明式 API——这正是 Stage 4 — Agent Frameworks 的内容衔接点。
延伸学习资源
本练习是章节式深度教材的入口而非替代品。仓库 Stage 3 章节的「精选 Projects」 列出了从零实现方向的延伸资源(如 pguso/ai-agents-from-scratch、arunpshankar/react-from-scratch 等项目及其维护状态与许可证说明),建议在完成本练习的 mock 测试与一次 live call 之后,再对照阅读 ReAct 原论文(Yao et al. 2022)的第 3 节,加深对 loop 设计动机的理解。完成检查:能解释为什么 messages 一直长、tool_use_id 如何配对、stop_reason 两个值各代表什么、max_iter 为什么是 safety net——做到这四点,本练习的核心目标即达成。
- 教程
- 文档
- AI Agent
- 人工智能
- 大模型
【免费下载链接】awesome-agentic-ai-zh
A trilingual (繁中 / English / 简中) learning roadmap for agentic AI: from LLM basics to multi-agent systems, with 240+ curated resources and hands-on examples. 中文 AI agent 學習地圖。
相关推荐
从零手写 ReAct Agent Loop:awesome-agentic-ai-zh 练习 3 的 70 行实现与双 SDK 路径验证
从零手写 ReAct Agent Loop:awesome agentic ai zh 练习 3 的 70 行实现与双 SDK 路径验证 本篇文章围绕 awes
教程文档AI Agent人工智能大模型MinIO 集群监控实战:Prometheus 接入到告警落地,一条链路讲透
MinIO 集群监控实战:Prometheus 接入到告警落地,一条链路讲透 MinIO 是 S3 兼容的分布式对象存储,跑上生产之后总得有人盯着它。我印象最深
教程文档AI Agent人工智能大模型AReaL 分布式训练调试指南:FSDP2/TP/CP/EP 场景下的 Hang、OOM 与通信错误排查实战
AReaL 分布式训练调试指南:FSDP2/TP/CP/EP 场景下的 Hang、OOM 与通信错误排查实战 AReaL 的分布式训练基于 PyTorch 原生
教程文档AI Agent人工智能大模型
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考