SA-03 手写生产级ReAct内核

从 30 行原型到生产级：手写一个可靠的 ReAct 执行内核

系列导航：00 系列导航 · 01 单 Agent 总论 · 02 ReAct 原理 · 03 手写内核 · 04 ACI 工具设计 · 05 上下文工程 · 06 两个增强变体 · 07 上线前清单 小提一句： 由于平台每日发布笔记数量限制，目前系列笔记还没全部上传。后续内容会持续更新，大家可以先点个关注、收藏，以免错过~

引子：那段你一定见过的 30 行代码

几乎每篇 Agent 教程都有这么一段：

async def react(task, tools, max_steps=10):
ctx = [f"任务：{task}"]
for step in range(max_steps):
out = await llm(SYSTEM + "\\n".join(ctx))
thought, action = parse_thought_action(out)
ctx.append(out)
if action.name == "finish":
return action.arg
result = await tools[action.name](action.arg)
ctx.append(f"Observation: {result}")
return "未完成"

它能跑通，能在博客里截图。但它离"可以放到线上"差着五个构件。本文就做一件事：逐个把这五个构件补上去，给出一份能直接抄进项目的完整实现。

先说清楚它缺什么：

这五个里，第 2 条和第 5 条最致命，正因为它们不报错，系统会一直"看起来在工作"。

一、架构全景

#mermaid-svg-lBVE9PXlLtqL9655{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-lBVE9PXlLtqL9655 .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-lBVE9PXlLtqL9655 .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-lBVE9PXlLtqL9655 .error-icon{fill:#552222;}#mermaid-svg-lBVE9PXlLtqL9655 .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-lBVE9PXlLtqL9655 .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-lBVE9PXlLtqL9655 .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-lBVE9PXlLtqL9655 .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-lBVE9PXlLtqL9655 .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-lBVE9PXlLtqL9655 .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-lBVE9PXlLtqL9655 .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-lBVE9PXlLtqL9655 .marker{fill:#333333;stroke:#333333;}#mermaid-svg-lBVE9PXlLtqL9655 .marker.cross{stroke:#333333;}#mermaid-svg-lBVE9PXlLtqL9655 svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-lBVE9PXlLtqL9655 p{margin:0;}#mermaid-svg-lBVE9PXlLtqL9655 .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-lBVE9PXlLtqL9655 .cluster-label text{fill:#333;}#mermaid-svg-lBVE9PXlLtqL9655 .cluster-label span{color:#333;}#mermaid-svg-lBVE9PXlLtqL9655 .cluster-label span p{background-color:transparent;}#mermaid-svg-lBVE9PXlLtqL9655 .label text,#mermaid-svg-lBVE9PXlLtqL9655 span{fill:#333;color:#333;}#mermaid-svg-lBVE9PXlLtqL9655 .node rect,#mermaid-svg-lBVE9PXlLtqL9655 .node circle,#mermaid-svg-lBVE9PXlLtqL9655 .node ellipse,#mermaid-svg-lBVE9PXlLtqL9655 .node polygon,#mermaid-svg-lBVE9PXlLtqL9655 .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-lBVE9PXlLtqL9655 .rough-node .label text,#mermaid-svg-lBVE9PXlLtqL9655 .node .label text,#mermaid-svg-lBVE9PXlLtqL9655 .image-shape .label,#mermaid-svg-lBVE9PXlLtqL9655 .icon-shape .label{text-anchor:middle;}#mermaid-svg-lBVE9PXlLtqL9655 .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-lBVE9PXlLtqL9655 .rough-node .label,#mermaid-svg-lBVE9PXlLtqL9655 .node .label,#mermaid-svg-lBVE9PXlLtqL9655 .image-shape .label,#mermaid-svg-lBVE9PXlLtqL9655 .icon-shape .label{text-align:center;}#mermaid-svg-lBVE9PXlLtqL9655 .node.clickable{cursor:pointer;}#mermaid-svg-lBVE9PXlLtqL9655 .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-lBVE9PXlLtqL9655 .arrowheadPath{fill:#333333;}#mermaid-svg-lBVE9PXlLtqL9655 .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-lBVE9PXlLtqL9655 .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-lBVE9PXlLtqL9655 .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-lBVE9PXlLtqL9655 .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-lBVE9PXlLtqL9655 .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-lBVE9PXlLtqL9655 .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-lBVE9PXlLtqL9655 .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-lBVE9PXlLtqL9655 .cluster text{fill:#333;}#mermaid-svg-lBVE9PXlLtqL9655 .cluster span{color:#333;}#mermaid-svg-lBVE9PXlLtqL9655 div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-lBVE9PXlLtqL9655 .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-lBVE9PXlLtqL9655 rect.text{fill:none;stroke-width:0;}#mermaid-svg-lBVE9PXlLtqL9655 .icon-shape,#mermaid-svg-lBVE9PXlLtqL9655 .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-lBVE9PXlLtqL9655 .icon-shape p,#mermaid-svg-lBVE9PXlLtqL9655 .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-lBVE9PXlLtqL9655 .icon-shape .label rect,#mermaid-svg-lBVE9PXlLtqL9655 .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-lBVE9PXlLtqL9655 .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-lBVE9PXlLtqL9655 .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-lBVE9PXlLtqL9655 :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

react() 主循环

是

否

finish

解析失败

重复调用

正常

① 预算检查steps / tokens / 墙钟

超了吗？

返回 RunResult(ok=False, reason)

② build_context()装配 system + 工具 spec + 任务 + 轨迹

③ llm(prompt)

④ parse_output()永不抛异常，失败返回哨兵值

输出分支

RunResult(ok=True)

格式错误作为 Observation 回填

LoopGuard 警告作为 Observation

⑤ ToolRegistry.dispatch()

未知工具 / 缺参数 → 错误字符串side_effect → confirm 钩子wait_for 超时异常捕获 → ToolError，不向上抛

⑥ truncate(obs)append Step，计入 usage

RunResult含完整 steps，可 JSON 序列化

核心设计哲学三条，贯穿下面所有代码：

二、完整实现

下面这份代码约 200 行，只依赖标准库（LLM 调用留一个适配口）。可以直接存成 agent_kernel.py。

2.1 工具注册与运行时

# agent_kernel.py
from __future__ import annotations

import asyncio
import json
import re
import time
from dataclasses import dataclass, field, asdict
from typing import Any, Awaitable, Callable

@dataclass
class Tool:
name: str
description: str
params: dict # JSON Schema（对象级，含 required）
handler: Callable[[dict], Awaitable[str]]
timeout: float = 15.0
side_effect: bool = False # True ⇒ 需授权 / 只读先行
returns: str = "纯文本结果"

class ToolRegistry:
def __init__(self) –> None:
self._tools: dict[str, Tool] = {}

def register(self, **kw):
def deco(fn: Callable[[dict], Awaitable[str]]):
self._tools[kw["name"]] = Tool(handler=fn, **kw)
return fn
return deco

def render(self) –> str:
"""把工具集渲染成提示片段——这就是 ACI 的物理形态（见第 4 篇）"""
return "\\n".join(
f"- {t.name}\\n"
f" 用途: {t.description}\\n"
f" 参数: {json.dumps(t.params, ensure_ascii=False)}\\n"
f" 返回: {t.returns}"
for t in self._tools.values()
)

async def dispatch(self, call: "ToolCall", confirm=None) –> str:
"""执行工具。契约：永不抛异常，只返回字符串（失败以 [ToolError] 开头）"""
t = self._tools.get(call.name)
if t is None:
return (f"[ToolError] 未知工具 '{call.name}'。"
f"可用工具: {', '.join(self._tools)}")

missing = [k for k in t.params.get("required", []) if k not in call.args]
if missing:
return (f"[ToolError] {t.name} 缺少必填参数 {missing}。"
f"参数模式: {json.dumps(t.params, ensure_ascii=False)}")

if t.side_effect and confirm is not None:
allowed = await confirm(t.name, call.args)
if not allowed:
return ("[ToolError] 该动作有副作用且未获授权，已被安全策略拒绝。"
"请改用只读工具达成目标，或要求用户授权。")
try:
result = await asyncio.wait_for(t.handler(call.args), timeout=t.timeout)
except asyncio.TimeoutError:
return f"[ToolError] {t.name} 执行超时（>{t.timeout}s）。请缩小时间窗或换工具。"
except Exception as e: # ★ 绝不向上抛
return f"[ToolError] {t.name} 执行失败: {type(e).__name__}: {e}"
return result if isinstance(result, str) else json.dumps(result, ensure_ascii=False)

dispatch 是整套系统的安全边界所在。 权限、超时、沙箱、审计、限流，全部挂在这里。千万不要把它们分散到各个 handler 里，分散就意味着一定会漏。

2.2 解析层：把格式抖动变成可恢复错误

@dataclass
class ToolCall:
name: str
args: dict = field(default_factory=dict)
raw: str = ""

_JSON_BLOCK = re.compile(r"```(?:json)?\\s*(\\{.*?\\})\\s*```", re.S)

def _balanced_spans(text: str) –> list[tuple[int, int]]:
"""用括号配对找出所有最外层 {…} 的区间（支持嵌套，比正则可靠）"""
spans, stack = [], []
for i, ch in enumerate(text):
if ch == "{":
stack.append(i)
elif ch == "}" and stack:
start = stack.pop()
if not stack:
spans.append((start, i + 1))
return spans

def _candidates(text: str):
for m in _JSON_BLOCK.finditer(text):
yield m.start(), m.group(1)
for s, e in _balanced_spans(text):
yield s, text[s:e]

def parse_output(text: str) –> tuple[str, ToolCall]:
"""拆成 (thought, call)。永不抛异常；全部候选都失败则返回 __unparsable__。"""
for start, payload in _candidates(text):
try:
obj = json.loads(payload)
except json.JSONDecodeError:
continue # 试下一个候选
if not isinstance(obj, dict):
continue
name = obj.get("action") or obj.get("tool") or obj.get("name")
if not isinstance(name, str):
continue
args = obj.get("args") or obj.get("arguments") or obj.get("parameters") or {}
if not isinstance(args, dict):
args = {"_raw": args} # 模型把参数写成字符串/数组时的兜底
return text[:start].strip(), ToolCall(name=name, args=args, raw=payload)

return text.strip(), ToolCall(name="__unparsable__", raw=text)

这一层有三个刻意的设计：

• 容忍三种键名（action / tool / name）和三种参数键（args / arguments / parameters）。模型在不同示例、不同模型家族下偏好不同，"严格格式"只会换来无谓的重试。
• 带 code fence 和不带都认，嵌套对象也认。很多模型会自作主张把 JSON 包进 ```json 块；也有模型直接输出裸 JSON。这里用括号配对而不是正则来找 JSON 片段，正则处理嵌套括号很容易漏（我第一版就踩了这个坑，参数里带一个 {} 就解析失败）。
• 失败不抛，返回哨兵值。主循环拿到 __unparsable__ 后，把"正确格式样例"作为 Observation 回填，模型看到自己的错误输出 + 格式要求，下一次几乎必定修正。这比在代码里正则硬掰要可靠得多。

2.3 预算、追踪与上下文装配

def approx_tokens(s: str) –> int:
"""中英混排的粗略估算：CJK 约 1 token/字，其余约 4 字符/token"""
cjk = sum(1 for ch in s if "\\u4e00" <= ch <= "\\u9fff")
return cjk + (len(s) – cjk + 3) // 4

def truncate(s: str, limit: int = 1200) –> str:
"""按 token 预算截断，头尾各留一部分并显式标注（避免模型以为看到了全部）"""
if approx_tokens(s) <= limit:
return s
cjk = any("\\u4e00" <= c <= "\\u9fff" for c in s)
budget = limit * (1 if cjk else 4)
head, tail = s[: int(budget * 0.6)], s[–int(budget * 0.4):]
return f"{head}\\n…[已截断约 {approx_tokens(s) – limit} tokens]…\\n{tail}"

@dataclass
class Budget:
max_steps: int = 12
max_prompt_tokens: int = 300_000 # 累计，不是单次
max_seconds: float = 180.0
max_consecutive_errors: int = 3

@dataclass
class Usage:
steps: int = 0
prompt_tokens: int = 0
completion_tokens: int = 0
wall: float = 0.0

def exhausted(self, b: Budget) –> str | None:
if self.steps >= b.max_steps: return "max_steps"
if self.prompt_tokens >= b.max_prompt_tokens: return "token_budget"
if self.wall >= b.max_seconds: return "timeout"
return None

@dataclass
class Step:
i: int
thought: str
call: ToolCall
observation: str = ""
ms: int = 0
prompt_tokens: int = 0

@dataclass
class RunResult:
ok: bool
answer: str
stop_reason: str
steps: list[Step]
usage: Usage

def to_json(self) –> str:
return json.dumps(asdict(self), ensure_ascii=False, indent=2)

SYSTEM = """你是一个在循环中工作的 Agent。每一步输出两部分：

1) 思考块，三段式：已知（可溯源到某步 Observation 的事实）/ 缺失 / 本步动作理由。
禁止复述 Observation 原文，禁止写没有信息量的过渡句。
2) 一个 JSON 动作块，形如 {"action": "<工具名>", "args": {"参数名": "参数值"}}

完成任务时输出 {"action": "finish", "args": {"answer": "<结论>"}}。

规则：

– 只依据 Observation 中的事实推理；没有证据就明说证据不足，不要编造。
– 工具报错是事实，必须据此改变策略，不要假装成功。
– 不要重复刚才执行过且结果相同的调用。"""

def build_context(task: str, tools: ToolRegistry, steps: list[Step]) –> str:
parts = [SYSTEM, "# 可用工具\\n" + tools.render(),
f"# 任务\\n{task}", "# 轨迹"]
for s in steps:
parts.append(f"[step {s.i}]")
if s.thought:
parts.append(f"思考: {s.thought}")
parts.append("Action: " + json.dumps(
{"action": s.call.name, "args": s.call.args}, ensure_ascii=False))
if s.observation:
parts.append(f"Observation: {s.observation}")
parts.append(f"[step {len(steps) + 1}]")
return "\\n".join(parts)

2.4 主循环

async def react(
task: str,
tools: ToolRegistry,
*,
budget: Budget = Budget(),
llm: Callable[[str], Awaitable[str]] | None = None,
confirm: Callable[[str, dict], Awaitable[bool]] | None = None,
obs_limit: int = 1200,
) –> RunResult:
llm = llm or default_llm
steps: list[Step] = []
usage = Usage()
t0 = time.perf_counter()
err_streak = 0

while True:
usage.wall = time.perf_counter() – t0
if (reason := usage.exhausted(budget)):
return RunResult(False, "", reason, steps, usage) # ★ 显式失败

prompt = build_context(task, tools, steps)
usage.prompt_tokens += approx_tokens(prompt)
raw = await llm(prompt)
usage.completion_tokens += approx_tokens(raw)

thought, call = parse_output(raw)
st = Step(i=len(steps) + 1, thought=thought, call=call,
prompt_tokens=approx_tokens(prompt))

# ── 分支 A：解析失败 → 格式要求作为 Observation 回填
if call.name == "__unparsable__":
st.observation = (
"[FormatError] 无法从你的输出中解析出 JSON 动作块。"
'请严格输出形如 {"action": "工具名", "args": {"参数": "值"}} 的 JSON。'
)
steps.append(st); usage.steps += 1; err_streak += 1
if err_streak >= budget.max_consecutive_errors:
return RunResult(False, "", "unparsable_exhausted", steps, usage)
continue

# ── 分支 B：finish
if call.name == "finish":
steps.append(st)
return RunResult(True, str(call.args.get("answer", "")), "finish", steps, usage)

# ── 分支 C：循环护栏（与上一次完全相同的调用）
if steps and steps[–1].call.name == call.name and steps[–1].call.args == call.args:
st.observation = (
"[LoopGuard] 你刚刚执行过完全相同的调用且返回相同结果。"
"请更换工具或参数；若信息已足够，直接 finish。"
)
steps.append(st); usage.steps += 1
continue

# ── 分支 D：正常执行
obs = await tools.dispatch(call, confirm=confirm)
st.observation = truncate(obs, obs_limit)
st.ms = int((time.perf_counter() – t0) * 1000)
steps.append(st); usage.steps += 1
err_streak = err_streak + 1 if obs.startswith("[ToolError]") else 0
if err_streak >= budget.max_consecutive_errors:
return RunResult(False, "", "tool_error_exhausted", steps, usage)

以及一个可替换的 LLM 适配口（以 OpenAI 兼容接口为例）：

async def default_llm(prompt: str) –> str:
import os, aiohttp
async with aiohttp.ClientSession() as s:
async with s.post(
f"{os.environ['OPENAI_BASE_URL']}/chat/completions",
headers={"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"},
json={"model": os.environ.get("MODEL", "gpt-4o-mini"),
"messages": [{"role": "user", "content": prompt}],
"temperature": 0}, # ★ Agent 默认 0 温
timeout=aiohttp.ClientTimeout(total=120),
) as r:
r.raise_for_status()
return (await r.json())["choices"][0]["message"]["content"]

生产建议：优先使用厂商的 function calling / tool use 结构化输出而不是正则解析。上面的解析层是兜底，即使接了结构化输出，也要保留它，因为结构化输出同样会产出非法参数。解析层永远不该被信任成唯一防线。

三、五个构件逐个点评

3.1 停止条件：三重保险 + 显式失败

Budget 里有三个硬约束：步数、累计 token、墙钟。为什么三个都要？因为它们防的是不同的失效：

再加一个软约束：max_consecutive_errors：连续 N 次工具报错就认输，避免模型在一个坏工具上反复撞墙。

最重要的一条纪律：ok=False 必须是显式的。

# ❌ 错误示范：静默返回半截结果
for step in range(max_steps):
...
return "根据已有信息，可能是……" # 调用方无法区分成功与失败

# ✅ 正确示范
return RunResult(ok=False, answer="", stop_reason="max_steps", steps=steps, usage=usage)

调用方拿到 ok=False 才知道要降级（转人工 / 换策略 / 退回检索问答）。一个把失败伪装成成功的 Agent，比一个直接报错的 Agent 危险十倍。

3.2 工具运行时：超时、沙箱、副作用授权

三件事全在 dispatch 里，不再重复。补充两个工程要点：

• 超时是每工具独立的，不是全局一个值。查日志 15s、跑测试 120s，混用一个值必然不合适。
• side_effect + confirm 钩子是权限最小化的落点。上线节奏应该是：confirm = lambda n, a: False（全只读）→ dry-run → 白名单工具自动放行 → 全量。这个开关应该能在配置里改，而不是改代码。

3.3 Observation 截断：必须让模型知道被截断了

truncate() 里那句 [已截断约 N tokens] 不是装饰。如果不标注，模型会以为自己看到了全部内容，从而基于"日志里没有这个错误"得出错误结论，一个静默的、无法归因的错误。

截断的进阶做法（第 5 篇详述）：把大段内容落盘，只回一个句柄 + 结构化摘要，模型需要时再用 read_range 工具按需读取。

3.4 循环护栏：一个便宜到不行的高收益机制

if steps and steps[–1].call.name == call.name and steps[–1].call.args == call.args:
→ 回填 [LoopGuard] 警告

不到 5 行，却消灭了最常见的死循环形态。可以扩展成"最近 K 步内出现过相同 (name, args) 即警告"：

def _seen_recently(steps, call, k=3):
return any(s.call.name == call.name and s.call.args == call.args for s in steps[–k:])

# 更强的版本：动作相同且 Observation 也相同 → 几乎确定是死循环

3.5 Trace：可观测性的最小单元

RunResult.to_json() 输出的 trace 至少要能回答六个问题：

1. 这次任务成功了吗？为什么停？（ok / stop_reason）
2. 走了几步？每步花了多久？（steps / ms）
3. 每步模型看到了多少 token？（prompt_tokens）
4. 每步调用了什么工具、传了什么参数？（call）
5. 环境返回了什么？（observation，可能是截断后的）
6. 模型当时是怎么想的？（thought）

把这六项落库，你就有了：成本归因（按工具/按步）、失败模式聚类（按 stop_reason 和 err_streak）、回归测试集（把 trace 存成 fixture）。第 7 篇的六项指标全部可以从这份 trace 里算出来，所以 trace 不是"加分项"，而是 Agent 系统的日志基础设施。

四、跑一个端到端例子

用贯穿系列的 OpsAgent，工具先 mock 掉（真实实现换成 HTTP 调用即可）：

tools = ToolRegistry()

@tools.register(
name="query_metrics",
description="查询服务的监控时序数据。用于确认指标的突增/渐变形态与起始时间点。",
params={"type": "object",
"properties": {"service": {"type": "string"},
"metric": {"type": "string"},
"window": {"type": "string", "default": "30m"}},
"required": ["service", "metric"]},
returns="时间-数值序列（已降采样）",
)
async def query_metrics(args):
return "t-30m:0.10% t-20m:0.10% t-12m:0.11% t-11m:0.90% t-10m:4.20% t-9m:12.0%"

@tools.register(
name="search_logs",
description="按服务与时间检索错误日志。注意：5xx 可能产生于网关层，此时 service 应填 gateway。",
params={"type": "object",
"properties": {"service": {"type": "string"},
"level": {"type": "string", "default": "error"},
"since": {"type": "string"},
"limit": {"type": "integer", "default": 20}},
"required": ["service", "since"]},
returns="日志条目列表（默认截断至 20 条）",
)
async def search_logs(args):
if args["service"] == "gateway":
return "192 条 upstream connect error → order-service；1 条 deploy event: order-service v2.14.0"
return "12 条 ConnectionPoolExhausted（下游视角，非根因）"

@tools.register(
name="read_repo_file",
description="读取代码仓库中指定文件的指定版本内容。路径必须是仓库根的绝对路径。",
params={"type": "object",
"properties": {"path": {"type": "string"},
"ref": {"type": "string"}},
"required": ["path"]},
returns="文件内容（超长自动截断）",
)
async def read_repo_file(args):
return "max_connections: 10 # v2.14.0（上一版本为 50）"

运行与结果：

res = await react(
task="订单服务 5xx 错误率 3 分钟内从 0.1% 涨到 12%，定位根因并给出建议。",
tools=tools,
budget=Budget(max_steps=8, max_seconds=90),
confirm=lambda name, args: False, # 全只读模式上线
)

print(res.ok, res.stop_reason, res.usage.steps, res.usage.prompt_tokens)
print(res.answer)

一条真实 trace（节选）的结构：

{
"ok": true,
"answer": "根因：order-service v2.14.0 将 DB 连接池上限从 50 改为 10，流量下连接耗尽导致 5xx。建议：回滚至 v2.13.7 或将该值调回 ≥50。",
"stop_reason": "finish",
"steps": [
{"i": 1, "thought": "已知：无。缺失：错误率形态与起始点。动作：先用 query_metrics 确认是突增还是渐变。",
"call": {"name": "query_metrics", "args": {"service": "order", "metric": "http_5xx_rate"}},
"observation": "t-30m:0.10% … t-9m:12.0%", "ms": 820, "prompt_tokens": 910},
{"i": 2, "thought": "已知：t-11m 突增，非渐变。缺失：该时间点前后的变更事件。动作：查部署历史。",
"call": {"name": "search_logs", "args": {"service": "gateway", "since": "t-12m"}},
"observation": "192 条 upstream connect error → order-service；1 条 deploy event: v2.14.0", "ms": 1140, "prompt_tokens": 1480}
],
"usage": {"steps": 4, "prompt_tokens": 6240, "completion_tokens": 780, "wall": 6.4}
}

注意 search_logs 的描述里那句"5xx 可能产生于网关层，此时 service 应填 gateway"——这一句话就是 ACI 设计的价值：它直接把模型从"第 3 步查错服务"的常见失败路径上救了回来。第 4 篇整篇都在讲怎么写出这种描述。

五、什么时候不该自研内核

上面这套 200 行能覆盖绝大多数单 Agent 场景。但出现下面任一情况时，应该考虑用成熟框架（LangGraph / OpenAI Agents SDK 等），而不是继续堆自研代码：

• 需要断点续跑（长任务中断后从中间恢复）；
• 需要持久化状态机与人工审批节点的深度集成；
• 需要多实例并发与分布式调度；
• 团队协作，需要统一的 trace 协议与可观测性接入；
• 任务结构已经演变成显式 DAG，此时它其实已经更接近工作流了。

判断标准很简单：如果你开始手写"任务状态持久化"和"节点编排"，你就已经越界到框架该管的事了。 内核要自研，是因为它是你系统的安全边界与观测边界，必须完全可控；框架要复用，是因为它解决的是你已经解决过一遍的通用问题。

六、小结

把 30 行原型补成生产级内核，真正的主线只有一条：循环永远不因模型输出崩溃，只因预算耗尽而终止。

主线下面落两个纪律。错误一律转成 Observation，而不是抛异常；失败一律显式返回 ok=False，不返回半截答案。这两个决定划出了"安静地坏"和"可被观测地工作"的分界，而前者比直接报错危险得多。

其余都是配套。dispatch 把权限、超时、沙箱、审计收在一处，让安全边界只有一个出口；一份可序列化的 trace，让成本归因和失败复盘有据可查。

上一篇 02 ReAct 原理深度解读 下一篇 04 工具即提示：Agent-Computer Interface（ACI）设计方法论。同样的模型，工具描述写得好不好，能把单步成功率从 0.85 拉到 0.98；套第 1 篇的 p^n 曲线，那就是 20 步任务 36% 和 82% 的差别。

参考

• Anthropic, Building Effective Agents, 2024-12（ACI、poka-yoke、工具定义与提示工程同等投入）。
• Yao et al., ReAct: Synergizing Reasoning and Acting in Language Models, ICLR 2023, arXiv:2210.03629.
• OpenAI, Function Calling / Structured Outputs 官方文档（结构化输出与解析兜底的关系）。