在上一篇《开篇：推理引擎之上的运行时系统》里，我们从系统视角算了一笔账，把单次自回归生成扩展成了多步 POMDP 状态转移模型，并且搭起了 mini Agent Runtime 的三块基石：数据契约（Message、ToolCall、Observation）、会话容器（AgentSession）和单步执行引擎（AgentRuntime.step()）。
但在第一篇的末尾，LLMClientProtocol 和 ToolExecutorProtocol 都还只是抽象接口。如果我们现在想让这个 Runtime 真正跑起来，去回答一个像“帮我算一下当前目录下所有 Python 文件的总行数，并算出平均每个文件多少行”这样的问题，中间还缺最关键的两块拼图：
工具是怎么注册并暴露给模型的？ 我们写的明明是一个普通的 Python 函数
def count_lines(path: str) -> int，怎么把它自动变成大模型能看懂的 JSON Schema，又怎么在模型返回一串字符串后，把参数安全地解包并路由回这个函数？模型是怎么决定“调工具”还是“给最终答案”的？ 尤其是在我们自己的
mini vLLM第一版只实现了基础的/v1/chat/completions（当时我们明确把tools和function_calling列在了“暂不支持”列表里）的情况下，怎么让任何一个只会做普通文本补全的模型，稳定地跑通 ReAct（Reasoning + Acting） 闭环？
这一篇，我们就来把这两块拼图亲手补齐。
我们不引入任何第三方框架，只用 Python 标准库里的 inspect、typing、json、re 和 urllib，手写一个基于类型反射的 ToolRegistry，以及一个同时兼容 原生 OpenAI Tool Calling 与 纯文本 ReAct 协议（配 Stop Words 截断） 的双模驱动层。写完这一篇，我们的 mini Agent 就能直接连上之前部署好的 mini vLLM，跑通第一个端到端的真实任务。
1. 这一篇要解决什么问题
很多同学在第一次用大模型做工具调用时，往往会经历两个阶段。
第一个阶段是“手搓 JSON 描述”。每写一个 Python 工具函数，就要去手写一大坨几十行的 JSON Schema 字典，声明 type: "object"、properties、required。只要函数签名改了一个参数名，而 JSON 字典忘了同步改，模型传过来的参数就会在运行时直接触发 TypeError: unexpected keyword argument。
第二个阶段是“迷信原生 tools 字段”。大家习惯了往 OpenAI 接口里塞一个 tools=[...] 参数，然后期待响应里的 message.tool_calls 自动出现结构化数据。但如果你的底层推理引擎（比如我们上一系列刚写完的 mini vLLM）还没有专门针对特定模型编写 Tool Call 解析器，或者你正在使用一个开源基底模型，直接传 tools 字段往往会报错，或者模型根本不按套路出牌。
其实，如果扒开所有大模型服务端的源码看本质：所谓的“原生 Function Calling”，在模型眼里根本不是什么神秘的硬件指令，它依然只是经过特殊模板渲染后的 Prompt 文本，加上特定的停止词（Stop Words / Special Tokens）和输出正则解析而已。
理解了这一点，我们就能设计出一套极其干净的架构：
Python 函数 (带类型注解与 Docstring)
│
▼ inspect.signature + get_type_hints
┌───────────────────────────┐
│ ToolRegistry │ ──► 自动生成标准 JSON Schema + 维护 name -> callable 路由表
└─────────────┬─────────────┘
│
┌────────┴────────┐
│ 模式 A │ 模式 B (通用兜底，直接兼容第一版 mini vLLM)
▼ ▼
原生 tools 字段 ReAct Prompt 注入 + stop=["\nObservation:"]
│ │
└────────┬────────┘
▼
LLM 推理生成 (Thought + Action)
│
▼
ReAct / Native Parser 提取出 ToolCall(name, arguments)
│
▼
ToolRegistry.execute() ──► 参数校验、类型转换、异常捕获 ──► 返回 Observation
无论底层模型服务是否支持原生 tools 参数，对上层的 AgentRuntime.step() 来说，它看到的永远是完全一致的 LLMResponse(content=thought, tool_calls=[...])。
2. ReAct 循环的概率建模与停止词（Stop Words）机理
在动手写解析器之前，我们先从数学和生成机理上搞清楚：为什么 ReAct 能够让一个只会续写文本的自回归模型具备“调用外部工具”的能力？以及为什么如果没有设置好 stop_words，模型就会出现严重的“自问自答幻觉”？
回顾我们在 mini vLLM 里实现的自回归生成过程：给定前缀 Token 序列 C，模型在不断地最大化联合条件概率：
P_θ(Y | C) = Π(t = 1…T) P_θ(yₜ | C, y₁, …, yₜ₋₁)
当我们采用 ReAct 范式时，我们在初始上下文 u 中通过指令规定模型必须严格按照如下语法结构交替生成文本块：
Thoughtₖ ⟶ Actionₖ ⟶ Observationₖ
具体到单步生成轨迹，我们可以把模型在第 k 步要生成的 Token 序列 Y⁽ᵏ⁾ 拆分为三个连续的子序列：
内部推理子序列 Sₖ（对应
Thought:）：模型对当前任务状态与历史观测进行显式分析；动作指令子序列 Aₖ（对应
Action:与Action Input:）：模型决定要调用的工具名与 JSON 参数；伪造观测子序列 Ôₖ（对应
Observation:）：如果推理引擎不主动停下来，模型会顺着语言模型的惯性，自己接着往下编造工具的执行结果！
用条件概率链式法则展开，如果不加干预，模型在单次 generate 调用中生成的联合分布是：
P_θ(Sₖ, Aₖ, Ôₖ | τₖ₋₁) = P_θ(Sₖ | τₖ₋₁) ｜显式推理（Chain-of-Thought）· P_θ(Aₖ | τₖ₋₁, Sₖ) ｜基于推理的动作选择 · P_θ(Ôₖ | τₖ₋₁, Sₖ, Aₖ) ｜必须被截断的幻觉区间！
请特别注意上面公式里的前两项与第三项：
第一项与第二项的乘积 P_θ(Sₖ | τₖ₋₁) · P_θ(Aₖ | τₖ₋₁, Sₖ) 解释了为什么必须先写
Thought再写Action。在 Transformer 的 Causal Attention（因果注意力掩码）下，排在后面的 Token 可以 attend 到排在前面的所有 Token。如果让模型直接输出Action（即直接求 argmax_a P_θ(a | τₖ₋₁)），模型必须在极其有限的单个 Token 前向传播层数（比如 32 层或 80 层矩阵乘法）内完成复杂的多跳逻辑推理；而一旦先自回归生成了长度为 m 的思考序列 Sₖ，这 m 个 Token 的 KV Cache 就成了显式的工作记忆（Scratchpad），使得动作 Aₖ 的条件分布尖锐度大幅提升（条件熵 H(Aₖ | τₖ₋₁, Sₖ) ≪ H(Aₖ | τₖ₋₁)）。第三项 P_θ(Ôₖ | τₖ₋₁, Sₖ, Aₖ) 则是所有手写 ReAct 循环最容易翻车的地方。因为在预训练语料和 Few-shot 示例里，
Action Input: {...}后面总是紧跟着\nObservation: ...，模型在吐完右花括号}后，下一个最高概率的 Token 序列几乎必然是\nObservation:，紧接着它就会自己捏造一个看似合理的查询结果（比如自己编一个天气温度或计算数值），然后顺着自己编的结果直接输出Final Answer:！
怎么在物理层面斩断第三项 P_θ(Ôₖ | …)？
答案就在我们在《从零实现一个 mini vLLM -- 设计 Sampling 系统》里重点讲过的文本级停止条件：stop_words。
当我们向推理引擎发起请求时，必须显式传入：
stop = ["\nObservation:", "Observation:"]
这样一来，底层推理引擎的 Detokenizer 和 StopChecker 一旦检测到增量文本末尾拼出了 Observation:，就会立刻触发停止条件，将 Sequence 标记为 FINISHED (stop_reason="stop")，并把匹配到的 stop_words 从返回文本中截断抹除。
此时控制权完美地停在 Aₖ 结束的那一个字节，交还给我们的 AgentRuntime。随后由 Runtime 在真实的 CPU/沙箱环境中执行函数得到真实的观测 Oₖ^real，再把 \nObservation: + Oₖ^real 拼接到上下文里，开启第 k+1 轮生成。
这就是底层推理引擎的 StopChecker 与上层 Agent Runtime 协同工作的第一个经典切面。
3. 设计 ToolRegistry：从 Python 函数签名到 JSON Schema 的自动反射
搞清楚了原理，我们先来实现第一个核心模块：ToolRegistry。
在设计工具注册器时，我的核心原则是：写工具的人应该只关心写好一个标准的、带类型注解（Type Hints）和文档字符串（Docstring）的原生 Python 函数，绝对不应该为了迁就 Agent 框架去手写任何多余的字典配置。
比如，当我们写下这样一个简单的函数并加上 @ registry.register 装饰器时：
@registry.register
def search_logs(keyword: str, max_lines: int = 20, case_sensitive: bool = False) -> str:
"""在服务器日志文件中检索包含指定关键字的行。"""
...
ToolRegistry 应该在注册的瞬间，利用 Python 标准库的 inspect.signature 和 typing.get_type_hints，自动完成三件事：
提取函数元数据：拿到函数名
search_logs和 docstring 描述；推导参数类型并构造 JSON Schema：遍历每一个参数，将 Python 类型（
str,int,float,bool,list,dict,Literal等）映射为 JSON Schema 类型，并根据参数是否有默认值（param.default is inspect.Parameter.empty）自动判定它是否属于required必填项；构建运行时安全调用包裹器（Safe Invocation Wrapper）：在真正执行工具时，检查必填参数是否缺失、剔除模型幻觉多传的未知参数（或者报错提示）、尝试做轻量级的类型矫正（比如模型把整数
20传成了字符串"20"），并将函数执行过程中的任何 Python 异常捕获封装为Observation(is_error=True)。
我们来看看 tools.py 的完整实现：
from __future__ import annotations
import inspect
import json
import time
import traceback
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union, get_args, get_origin, get_type_hints
from schema import Observation, ToolCall
def _python_type_to_json_schema(py_type: Any) -> Dict[str, Any]:
"""将 Python 类型注解递归转换为 JSON Schema 属性定义"""
if py_type is inspect.Parameter.empty or py_type is Any:
return {"type": "string"}
origin = get_origin(py_type)
args = get_args(py_type)
# 处理 Optional[T] 或 Union[T, None]
if origin is Union:
non_none_args = [a for a in args if a is not type(None)]
if len(non_none_args) == 1:
return _python_type_to_json_schema(non_none_args[0])
return {"type": "string"}
# 处理 Literal["a", "b"] 枚举约束
if origin is Literal:
return {"type": "string", "enum": list(args)}
# 处理 List[T]
if origin in (list, List):
item_type = args[0] if args else Any
return {"type": "array", "items": _python_type_to_json_schema(item_type)}
# 处理 Dict[K, V]
if origin in (dict, Dict):
return {"type": "object"}
# 基础标量类型映射
type_map: Dict[Any, str] = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object",
}
schema_type = type_map.get(py_type, "string")
return {"type": schema_type}
@dataclass
class ToolDefinition:
"""内部工具注册项：持有可调用对象、参数签名与生成的 JSON Schema"""
name: str
description: str
func: Callable[..., Any]
parameters_schema: Dict[str, Any]
sig: inspect.Signature
type_hints: Dict[str, Any]
def to_openai_schema(self) -> Dict[str, Any]:
"""输出兼容 OpenAI / vLLM tools 字段的标准格式"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters_schema,
},
}
class ToolRegistry:
"""工具注册与路由执行中心（实现第一篇定义的 ToolExecutorProtocol）"""
def __init__(self) -> None:
self._tools: Dict[str, ToolDefinition] = {}
def register(
self,
func: Optional[Callable[..., Any]] = None,
*,
name: Optional[str] = None,
description: Optional[str] = None,
) -> Callable[..., Any]:
"""支持 @registry.register 和 @registry.register(name="...") 两种装饰器写法"""
def _decorator(f: Callable[..., Any]) -> Callable[..., Any]:
tool_name = name or f.__name__
tool_desc = (description or inspect.getdoc(f) or f"Execute tool {tool_name}.").strip()
sig = inspect.signature(f)
try:
type_hints = get_type_hints(f)
except Exception:
type_hints = {}
properties: Dict[str, Any] = {}
required: List[str] = []
for param_name, param in sig.parameters.items():
if param_name in ("self", "cls"):
continue
if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
continue
hint = type_hints.get(param_name, param.annotation)
prop_schema = _python_type_to_json_schema(hint)
if param.default is not inspect.Parameter.empty:
prop_schema["default"] = param.default
else:
required.append(param_name)
properties[param_name] = prop_schema
parameters_schema = {
"type": "object",
"properties": properties,
"required": required,
}
self._tools[tool_name] = ToolDefinition(
name=tool_name,
description=tool_desc,
func=f,
parameters_schema=parameters_schema,
sig=sig,
type_hints=type_hints,
)
return f
if func is not None:
return _decorator(func)
return _decorator
def get_schemas(self) -> List[Dict[str, Any]]:
"""获取所有已注册工具的 Schema 列表"""
return [tool.to_openai_schema() for tool in self._tools.values()]
def _coerce_and_validate_args(
self, tool_def: ToolDefinition, raw_args: Dict[str, Any]
) -> Dict[str, Any]:
"""校验必填项、过滤未知参数，并对常见大模型类型偏差做轻量级矫正"""
if not isinstance(raw_args, dict):
raise ValueError(f"Tool arguments must be a JSON object (dict), got {type(raw_args).__name__}.")
required_fields = tool_def.parameters_schema.get("required", [])
missing = [req for req in required_fields if req not in raw_args]
if missing:
raise ValueError(f"Missing required arguments: {missing}. Expected schema: {json.dumps(tool_def.parameters_schema)}")
accepts_kwargs = any(
p.kind == inspect.Parameter.VAR_KEYWORD for p in tool_def.sig.parameters.values()
)
unknown = [k for k in raw_args if k not in tool_def.sig.parameters]
if unknown and not accepts_kwargs:
raise ValueError(
f"Unknown arguments {unknown} for tool '{tool_def.name}'. "
f"Valid parameters are: {list(tool_def.sig.parameters.keys())}"
)
coerced: Dict[str, Any] = {}
for k, v in raw_args.items():
target_type = tool_def.type_hints.get(k)
# 如果模型把 int/float/bool 传成了字符串，尝试安全转换
if target_type is int and isinstance(v, str) and v.strip().lstrip("-").isdigit():
v = int(v.strip())
elif target_type is float and isinstance(v, (str, int)):
try:
v = float(v)
except ValueError:
pass
elif target_type is bool and isinstance(v, str):
if v.lower() in ("true", "1"):
v = True
elif v.lower() in ("false", "0"):
v = False
coerced[k] = v
return coerced
def execute(self, tool_call: ToolCall) -> Observation:
"""
路由并安全执行工具调用。
任何阶段出错都转化为 Observation(is_error=True) 返回给模型自修复，绝不中断主进程。
"""
start_time = time.perf_counter()
tool_name = tool_call.name
# 1. 检查工具是否存在（防止模型幻觉出不存在的函数名）
if tool_name not in self._tools:
elapsed = (time.perf_counter() - start_time) * 1000.0
available = list(self._tools.keys())
return Observation(
call_id=tool_call.id,
tool_name=tool_name,
output=f"Error: Tool '{tool_name}' is not registered. Available tools are: {available}",
is_error=True,
execution_time_ms=elapsed,
)
tool_def = self._tools[tool_name]
# 2. 检查上游 JSON 解析是否已经失败（例如 raw_arguments 不是合法 JSON）
if "_parse_error" in tool_call.arguments:
elapsed = (time.perf_counter() - start_time) * 1000.0
return Observation(
call_id=tool_call.id,
tool_name=tool_name,
output=(
f"Error: Failed to parse Action Input as valid JSON: {tool_call.arguments['_parse_error']}. "
f"Raw input was: {tool_call.raw_arguments!r}. Please output strictly valid JSON."
),
is_error=True,
execution_time_ms=elapsed,
)
# 3. 校验参数并执行目标函数
try:
validated_args = self._coerce_and_validate_args(tool_def, tool_call.arguments)
result = tool_def.func(**validated_args)
if isinstance(result, (dict, list)):
output_str = json.dumps(result, ensure_ascii=False, indent=2)
else:
output_str = str(result)
elapsed = (time.perf_counter() - start_time) * 1000.0
return Observation(
call_id=tool_call.id,
tool_name=tool_name,
output=output_str,
is_error=False,
execution_time_ms=elapsed,
)
except Exception as exc:
elapsed = (time.perf_counter() - start_time) * 1000.0
return Observation(
call_id=tool_call.id,
tool_name=tool_name,
output=f"Error executing tool '{tool_name}': {type(exc).__name__}: {exc}",
is_error=True,
execution_time_ms=elapsed,
)
看上面 ToolRegistry.execute 的设计，我们专门处理了三类最常见的线上故障：
幻觉工具名：如果模型自己编了一个
search_google，我们不仅告诉它这个工具不存在，还把当前可用的工具列表Available tools are: [...]明确塞进报错信息里。模型下一轮看到这段 Observation，马上就会改调正确的工具。JSON 解析失败穿透：当解析器发现模型输出的
Action Input不是合法 JSON 时，不会直接在解析层崩溃，而是塞一个{"_parse_error": ...}传下来，由ToolRegistry统一组装成带原始错误文本的Observation喂回给模型。未知参数与类型偏差矫正：大模型非常喜欢把整数字段写成
"10"，或者多传一个自己猜的字段。_coerce_and_validate_args会自动把能安全转换的基础类型转好，对未知参数给出精确的可用参数名提示。
4. 手写 ReAct 协议适配器与容错解析器
接下来是本篇最核心的部分：如何把 AgentSession 里的结构化消息列表，翻译成模型能看懂的 ReAct Prompt，再把模型吐出来的纯文本，精准地还原成 Thought 和 ToolCall？
我们先来看看 ReAct 模式下模型实际看到的 Prompt 长什么样。
当我们将一组工具 Schema 注入到 System Prompt 时，我们会构造一段清晰的协议指令：
You are a helpful and rigorous AI agent.
You have access to the following tools:
- calculate_expression: 计算数学表达式并返回精确的浮点或整数结果。
Parameters JSON Schema: {"type": "object", "properties": {"expr": {"type": "string"}}, "required": ["expr"]}
- get_word_length: 统计指定字符串的长度与字符分布。
Parameters JSON Schema: {"type": "object", "properties": {"word": {"type": "string"}}, "required": ["word"]}
To use a tool, you MUST strictly follow this exact format:
Thought: Explain your reasoning about what to do next.
Action: <tool_name>
Action Input: {"param1": "value1"}
Wait for the system to provide the result in the following format before continuing:
Observation: <tool_output>
You may repeat the Thought / Action / Action Input / Observation cycle as many times as needed.
When you have gathered enough information and are ready to answer the user's request, output:
Thought: I now know the final answer.
Final Answer: <your comprehensive response to the user>
为什么简单的正则表达式解析一定会翻车？
很多初学者写 ReAct 解析器时，会随手写一行正则：
# 脆弱的写法：遇到嵌套花括号或多行字符串就会截断出错
match = re.search(r"Action:\s*(\w+)\s*Action Input:\s*(\{.*?\})", text, re.DOTALL)
这种非贪婪匹配 \{.*?\} 在遇到下面这种真实工具调用时会瞬间暴毙：
Thought: 我需要先写入一段包含字典结构的配置。
Action: write_json_config
Action Input: {"filename": "app.json", "data": {"host": "127.0.0.1", "port": 8080}}
因为非贪婪匹配 \{.*?\} 会在遇到内部 "port": 8080} 的第一个右花括号 } 时就提前停止！导致提取出来的字符串少了最外层的 }，随后 json.loads 直接报 JSONDecodeError: Unterminated string or object。而如果改成贪婪匹配 \{.*\}，当模型不小心在 JSON 后面多写了一句带花括号的解释时，又会多吞内容。
怎么写出一个真正工业级、能正确处理任意层级嵌套花括号与字符串转义的 Action Input 提取器？
最稳健的做法是结合 括号平衡扫描状态机（Balanced Brace Scanner） 或 Python 标准库 json.JSONDecoder().raw_decode()。raw_decode(s, idx) 是 Python json 模块里一个非常宝藏的底层方法：它会从字符串 s 的指定下标 idx（即第一个 { 出现的位置）开始严格按照 JSON 语法状态机向后扫描，一旦读完一个完整且合法的顶层 JSON 对象，就会立刻返回 (parsed_obj, end_index)，完全不受后续多余文本或内部嵌套括号的干扰！
我们来写出这个兼具 raw_decode 精确解析与 Markdown 代码块剥离能力的 ReActFormatAdapter（位于 react_adapter.py）：
from __future__ import annotations
import json
import re
import uuid
from typing import Any, Dict, List, Optional, Tuple
from schema import Message, Role, ToolCall
REACT_PROTOCOL_TEMPLATE = """
You have access to the following tools:
{tool_descriptions}
To invoke a tool, you MUST strictly follow this format:
Thought: think step-by-step about what information is needed or what action to take.
Action: the exact name of the tool to call (must be one of [{tool_names}]).
Action Input: a single-line or multi-line valid JSON object matching the tool's parameters schema.
After you output `Action Input: ...`, STOP generating immediately. The runtime will execute the tool and append:
Observation: <result of the tool execution>
When you have sufficient information to fulfill the user's request without calling any more tools, output:
Thought: I have all the information needed to answer the user.
Final Answer: <your final response to the user>
"""
class ReActFormatAdapter:
"""
负责将内部 Message/ToolSchema 与文本级 ReAct 格式互相转换。
让不支持原生 function calling 的推理引擎（如第一版 mini vLLM）也能无缝运行 Agent。
"""
STOP_SEQUENCES = ["\nObservation:", "\nObservation：", "Observation:"]
@classmethod
def build_system_prompt(cls, base_prompt: str, tool_schemas: List[Dict[str, Any]]) -> str:
"""将工具定义渲染并注入到 System Prompt 尾部"""
if not tool_schemas:
return base_prompt
lines: List[str] = []
names: List[str] = []
for item in tool_schemas:
fn = item.get("function", item)
name = fn["name"]
desc = fn.get("description", "")
params = json.dumps(fn.get("parameters", {}), ensure_ascii=False)
names.append(name)
lines.append(f"- {name}: {desc}\n Parameters JSON Schema: {params}")
protocol_block = REACT_PROTOCOL_TEMPLATE.format(
tool_descriptions="\n".join(lines),
tool_names=", ".join(names),
)
return base_prompt.rstrip() + "\n" + protocol_block
@classmethod
def format_messages_for_react(
cls, messages: List[Message], tool_schemas: List[Dict[str, Any]]
) -> List[Dict[str, str]]:
"""
将包含 ASSISTANT(tool_calls) 和 TOOL(observation) 的结构化轨迹，
折叠转换为纯 user / assistant 文本消息序列，适配标准 /v1/chat/completions。
"""
formatted: List[Dict[str, str]] = []
for msg in messages:
if msg.role == Role.SYSTEM:
content = cls.build_system_prompt(msg.content or "", tool_schemas)
formatted.append({"role": "system", "content": content})
elif msg.role == Role.USER:
formatted.append({"role": "user", "content": msg.content or ""})
elif msg.role == Role.ASSISTANT:
if msg.tool_calls:
tc = msg.tool_calls[0]
thought_text = (msg.content or "I need to call a tool.").strip()
args_str = tc.raw_arguments or json.dumps(tc.arguments, ensure_ascii=False)
react_turn = (
f"Thought: {thought_text}\n"
f"Action: {tc.name}\n"
f"Action Input: {args_str}"
)
formatted.append({"role": "assistant", "content": react_turn})
else:
text = (msg.content or "").strip()
formatted.append({"role": "assistant", "content": f"Thought: Task complete.\nFinal Answer: {text}"})
elif msg.role == Role.TOOL:
# 将工具观测作为一条 User 角色的环境反馈追加，保持严格的前缀递增
obs_text = f"Observation: {msg.content or ''}"
formatted.append({"role": "user", "content": obs_text})
return formatted
@classmethod
def _extract_json_payload(cls, raw_text: str) -> Tuple[Dict[str, Any], str]:
"""
从 Action Input 后的文本中提取 JSON 对象。
支持剥离 ```json 代码块，并使用 JSONDecoder.raw_decode 处理任意嵌套花括号。
"""
cleaned = raw_text.strip()
# 如果模型把 JSON 包在了 ```json ... ``` 里，先剥离代码块标记
fence_match = re.search(r"```(?:json)?\s*(\{.*)\s*```", cleaned, re.DOTALL)
if fence_match:
cleaned = fence_match.group(1).strip()
brace_idx = cleaned.find("{")
if brace_idx == -1:
return {"_parse_error": "No JSON object '{...}' found in Action Input"}, cleaned
candidate = cleaned[brace_idx:]
decoder = json.JSONDecoder()
try:
obj, end_pos = decoder.raw_decode(candidate)
raw_json_slice = candidate[:end_pos]
if isinstance(obj, dict):
return obj, raw_json_slice
return {"_parse_error": f"Expected JSON object, got {type(obj).__name__}"}, raw_json_slice
except json.JSONDecodeError as exc:
return {"_parse_error": str(exc)}, candidate
@classmethod
def parse_react_output(cls, text: str) -> Tuple[Optional[str], List[ToolCall]]:
"""
解析模型生成的 ReAct 文本，返回 (thought, tool_calls)。
- 如果包含 Action / Action Input：返回提取出的 Thought 和单个 ToolCall；
- 如果包含 Final Answer 或无 Action：返回最终回复文本，tool_calls 为空列表 []。
"""
# 保险起见，如果底层引擎没截断 Observation:，我们在解析层主动截断它防止幻觉渗透
for stop_word in cls.STOP_SEQUENCES:
if stop_word in text:
text = text.split(stop_word, 1)[0]
cleaned = text.strip()
# 1. 优先检查是否触发了 Action 调用
action_match = re.search(
r"Action\s*:\s*([^\n\r]+)\s*[\r\n]+Action Input\s*:\s*(.+)",
cleaned,
re.DOTALL | re.IGNORECASE,
)
if action_match:
prefix_text = cleaned[: action_match.start()].strip()
thought = re.sub(r"^Thought\s*:\s*", "", prefix_text, flags=re.IGNORECASE | re.DOTALL).strip()
tool_name = action_match.group(1).strip().strip("`'\"")
raw_input_part = action_match.group(2).strip()
parsed_args, raw_slice = cls._extract_json_payload(raw_input_part)
call_id = f"call_{uuid.uuid4().hex[:8]}"
return (
thought or None,
[ToolCall(id=call_id, name=tool_name, arguments=parsed_args, raw_arguments=raw_slice)],
)
# 2. 检查是否输出了 Final Answer
final_match = re.search(r"Final Answer\s*:\s*(.+)", cleaned, re.DOTALL | re.IGNORECASE)
if final_match:
final_answer = final_match.group(1).strip()
return final_answer, []
# 3. 兜底逻辑：如果模型既没按格式写 Action，也没写 Final Answer 标签，
# 则剥离可能存在的 Thought: 前缀后，将其整体视为对用户的直接回复。
fallback_text = re.sub(r"^Thought\s*:\s*", "", cleaned, flags=re.IGNORECASE).strip()
return fallback_text, []
注意看 ReActFormatAdapter.format_messages_for_react 里的一个关键设计：
当轨迹每往前推进一步时，之前的 messages[0..k] 转换出来的字典内容是完全静止不变的，新产生的 Assistant (Thought + Action) 和 User (Observation) 只是严格地 append 在列表末尾。这意味着当请求发往我们的 mini vLLM 时，tokenizer.apply_chat_template(messages) 编码出来的 Token ID 序列拥有 100% 稳定的历史前缀，可以直接命中 mini vLLM 的 Prefix Caching！
5. 实现双模 HTTP 客户端：连接 mini vLLM 与 OpenAI 接口
有了 ReActFormatAdapter，我们现在可以实现真正的 OpenAICompatibleClient 了。
为了让它既能连接我们在《mini vLLM》第 20 篇写的本地推理服务（http://127.0.0.1:8000/v1/chat/completions），也能连接任何主流云端 API（如 DeepSeek、Qwen、OpenAI），我们给客户端设计了一个 tool_mode 开关：
tool_mode="react"（默认推荐用于自建mini vLLM）：客户端自动调用上面的ReActFormatAdapter，把工具描述注入 System Prompt，自动在请求体里加上"stop": ["\nObservation:", "Observation:"]，并在拿到模型文本后解析出ToolCall。tool_mode="native"：直接通过 HTTP 请求体的"tools"字段下发 JSON Schema，并从响应的message.tool_calls中读取结构化调用。
我们用 Python 标准库的 urllib.request 来写这个客户端，做到零外部依赖（llm_client.py）：
from __future__ import annotations
import json
import urllib.error
import urllib.request
from typing import Any, Dict, List, Literal, Optional
from react_adapter import ReActFormatAdapter
from schema import Message, TokenUsage, ToolCall
from session import AgentConfig
from runtime import LLMClientProtocol, LLMResponse
class OpenAICompatibleClient(LLMClientProtocol):
"""
兼容 mini vLLM 与标准 OpenAI /v1/chat/completions 的 HTTP 客户端。
支持 "react" (纯文本提示词 + stop words 截断) 与 "native" (原生 tools 字段) 两种模式。
"""
def __init__(
self,
base_url: str = "http://127.0.0.1:8000/v1",
api_key: str = "EMPTY",
tool_mode: Literal["react", "native"] = "react",
timeout_sec: float = 60.0,
) -> None:
self.endpoint = base_url.rstrip("/") + "/chat/completions"
self.api_key = api_key
self.tool_mode = tool_mode
self.timeout_sec = timeout_sec
def generate(
self,
messages: List[Message],
tool_schemas: List[Dict[str, Any]],
config: AgentConfig,
) -> LLMResponse:
if self.tool_mode == "react":
payload_messages = ReActFormatAdapter.format_messages_for_react(messages, tool_schemas)
payload: Dict[str, Any] = {
"model": config.model,
"messages": payload_messages,
"temperature": config.temperature,
"max_tokens": 1024,
"stop": ReActFormatAdapter.STOP_SEQUENCES,
"stream": False,
}
else:
payload_messages = [m.to_openai_dict() for m in messages]
payload = {
"model": config.model,
"messages": payload_messages,
"temperature": config.temperature,
"max_tokens": 1024,
"stream": False,
}
if tool_schemas:
payload["tools"] = tool_schemas
body_bytes = json.dumps(payload, ensure_ascii=False).encode("utf-8")
req = urllib.request.Request(
self.endpoint,
data=body_bytes,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=self.timeout_sec) as resp:
resp_data = json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
err_body = exc.read().decode("utf-8", errors="replace")
raise RuntimeError(f"LLM API HTTP {exc.code}: {err_body}") from exc
except urllib.error.URLError as exc:
raise RuntimeError(f"Failed to connect to LLM server at {self.endpoint}: {exc}") from exc
return self._parse_api_response(resp_data)
def _parse_api_response(self, data: Dict[str, Any]) -> LLMResponse:
choice = data["choices"][0]
msg = choice.get("message", {})
raw_content: Optional[str] = msg.get("content")
usage_dict = data.get("usage", {})
prompt_details = usage_dict.get("prompt_tokens_details") or {}
usage = TokenUsage(
prompt_tokens=usage_dict.get("prompt_tokens", 0),
completion_tokens=usage_dict.get("completion_tokens", 0),
cached_tokens=prompt_details.get("cached_tokens", usage_dict.get("cached_tokens", 0)),
)
# 模式 1：如果服务端返回了原生 tool_calls
native_calls = msg.get("tool_calls")
if native_calls:
parsed_calls: List[ToolCall] = []
for tc in native_calls:
fn = tc.get("function", {})
raw_args = fn.get("arguments", "{}")
try:
args_dict = json.loads(raw_args) if isinstance(raw_args, str) else raw_args
except json.JSONDecodeError as exc:
args_dict = {"_parse_error": str(exc)}
parsed_calls.append(
ToolCall(
id=tc.get("id", "call_native"),
name=fn.get("name", ""),
arguments=args_dict,
raw_arguments=raw_args if isinstance(raw_args, str) else json.dumps(raw_args),
)
)
return LLMResponse(content=raw_content, tool_calls=parsed_calls, usage=usage)
# 模式 2：在 ReAct 模式下，从纯文本 content 中提取 Thought 与 Action
if self.tool_mode == "react" and raw_content:
thought_or_final, react_calls = ReActFormatAdapter.parse_react_output(raw_content)
return LLMResponse(content=thought_or_final, tool_calls=react_calls, usage=usage)
return LLMResponse(content=raw_content, tool_calls=[], usage=usage)
6. 端到端实战验证：跑通多步工具调用与错误自修复闭环
现在，我们已经集齐了所有核心组件：
schema.py&session.py（第一篇实现的轨迹与状态容器）runtime.py（第一篇实现的AgentRuntime单步驱动器）tools.py（本篇实现的反射式ToolRegistry）react_adapter.py&llm_client.py（本篇实现的 ReAct 协议适配器与客户端）
为了验证这套代码不仅能跑通正常的 ReAct 多步调用，还能在模型犯错（例如传错参数类型或遇到 JSON 语法瑕疵）时自动完成 “报错回传 → 观测修正 → 重试成功” 的闭环，我们来写一个可直接运行的验证脚本 demo_minimal_react.py。
在下面的脚本里，我们注册两个真实的本地工具（list_python_files 和 count_file_lines），并提供一个模拟 mini vLLM 行为的本地 Mock Server（你也可以直接把 client 指向正在运行的 mini vLLM 端口 http://127.0.0.1:8000/v1）。特别注意我们在第 2 步故意让模型犯了一个典型的参数名拼写错误（把 file_path 写成了 path），看看我们的 Runtime 是如何优雅自愈的：
import os
from typing import List
from runtime import AgentRuntime, LLMClientProtocol, LLMResponse
from schema import Message, TokenUsage
from session import AgentConfig
from react_adapter import ReActFormatAdapter
from tools import ToolRegistry
# 1. 创建工具注册中心，并用装饰器注册两个纯 Python 函数
registry = ToolRegistry()
@registry.register
def list_python_files(directory: str = ".") -> List[str]:
"""列出指定目录下的所有 .py 源代码文件名。"""
if not os.path.exists(directory):
raise FileNotFoundError(f"Directory '{directory}' does not exist.")
return sorted([f for f in os.listdir(directory) if f.endswith(".py")])
@registry.register
def count_file_lines(file_path: str, ignore_empty: bool = True) -> int:
"""统计单个文件的代码行数。"""
if not os.path.isfile(file_path):
raise FileNotFoundError(f"File '{file_path}' not found.")
with open(file_path, "r", encoding="utf-8") as f:
lines = f.readlines()
if ignore_empty:
lines = [line for line in lines if line.strip()]
return len(lines)
# 2. 构造一个模拟 mini vLLM 文本补全行为的本地演练客户端，展示 4 步完整轨迹
class SimulatedMiniVLLMClient(LLMClientProtocol):
"""模拟底层推理引擎在 ReAct 协议下的逐轮生成输出（含一次参数错误自修复）"""
def __init__(self) -> None:
self._turn = 0
def generate(self, messages: List[Message], tool_schemas: list, config: AgentConfig) -> LLMResponse:
self._turn += 1
react_msgs = ReActFormatAdapter.format_messages_for_react(messages, tool_schemas)
# 模拟每一轮收到 stop=["\nObservation:"] 截断后的模型输出
if self._turn == 1:
raw_text = (
"Thought: 我需要先查看当前目录下有哪些 Python 文件。\n"
"Action: list_python_files\n"
'Action Input: {"directory": "."}'
)
elif self._turn == 2:
# 故意把参数名 file_path 错写成 path，测试 ToolRegistry 的错误捕获与反馈机制
raw_text = (
"Thought: 找到了文件列表，现在我来统计 schema.py 的非空行数。\n"
"Action: count_file_lines\n"
'Action Input: {"path": "schema.py", "ignore_empty": true}'
)
elif self._turn == 3:
# 模型在看到上一轮 Observation 中的 Unknown arguments 报错提示后，自动修正参数名！
raw_text = (
"Thought: 上一步参数名传错了，工具提示正确参数名是 file_path。我重新调用 count_file_lines。\n"
"Action: count_file_lines\n"
'Action Input: {"file_path": "schema.py", "ignore_empty": true}'
)
else:
last_obs = react_msgs[-1]["content"]
raw_text = (
f"Thought: 我已经拿到了准确的行数统计结果（{last_obs}），可以回答用户了。\n"
f"Final Answer: 当前目录下的核心数据结构文件 `schema.py` 共包含 {last_obs.split(':')[-1].strip()} 行非空代码。"
)
thought_or_final, tool_calls = ReActFormatAdapter.parse_react_output(raw_text)
usage = TokenUsage(prompt_tokens=180 * self._turn, completion_tokens=45, cached_tokens=180 * (self._turn - 1))
return LLMResponse(content=thought_or_final, tool_calls=tool_calls, usage=usage)
if __name__ == "__main__":
print("=== 1. 自动反射生成的 Tool JSON Schemas ===")
import json
print(json.dumps(registry.get_schemas(), ensure_ascii=False, indent=2))
print("\n=== 2. 启动 mini Agent Runtime 逐步执行 ===")
runtime = AgentRuntime(llm_client=SimulatedMiniVLLMClient(), tool_executor=registry)
session = runtime.run("帮我看看当前目录下有哪些 Python 文件，并统计 schema.py 的非空代码行数。")
for step in session.steps:
print(f"\n--- Step {step.step_index} (耗时: {step.step_total_ms:.2f} ms) ---")
if step.tool_calls:
tc = step.tool_calls[0]
obs = step.observations[0]
print(f"[Thought] : {step.thought}")
print(f"[Action] : {tc.name}({tc.raw_arguments})")
print(f"[Observation] : (is_error={obs.is_error}) {obs.output}")
else:
print(f"[Final Answer]: { session.final_answer }")
print(f"\n=== 3. 会话终态汇总 ===")
print(f"Session ID : {session.session_id}")
print(f"Final Status : {session.status.value}")
print(f"Total Steps : {session.current_step}")
print(f"Token Usage : Prompt={session.total_usage.prompt_tokens}, "
f"Completion={session.total_usage.completion_tokens}, "
f"Cached={session.total_usage.cached_tokens}")
我们来跑一下这段脚本，看看控制台打印出的真实执行轨迹：
=== 1. 自动反射生成的 Tool JSON Schemas ===
[
{
"type": "function",
"function": {
"name": "list_python_files",
"description": "列出指定目录下的所有 .py 源代码文件名。",
"parameters": {
"type": "object",
"properties": {
"directory": {
"type": "string",
"default": "."
}
},
"required": []
}
}
},
{
"type": "function",
"function": {
"name": "count_file_lines",
"description": "统计单个文件的代码行数。",
"parameters": {
"type": "object",
"properties": {
"file_path": {
"type": "string"
},
"ignore_empty": {
"type": "boolean",
"default": true
}
},
"required": [
"file_path"
]
}
}
}
]
=== 2. 启动 mini Agent Runtime 逐步执行 ===
--- Step 1 (耗时: 0.42 ms) ---
[Thought] : 我需要先查看当前目录下有哪些 Python 文件。
[Action] : list_python_files({"directory": "."})
[Observation] : (is_error=False) [
"demo_minimal_react.py",
"llm_client.py",
"react_adapter.py",
"runtime.py",
"schema.py",
"session.py",
"tools.py"
]
--- Step 2 (耗时: 0.18 ms) ---
[Thought] : 找到了文件列表，现在我来统计 schema.py 的非空行数。
[Action] : count_file_lines({"path": "schema.py", "ignore_empty": true})
[Observation] : (is_error=True) Error executing tool 'count_file_lines': ValueError: Missing required arguments: ['file_path']. Expected schema: {"type": "object", "properties": {"file_path": {"type": "string"}, "ignore_empty": {"type": "boolean", "default": true}}, "required": ["file_path"]}
--- Step 3 (耗时: 0.25 ms) ---
[Thought] : 上一步参数名传错了，工具提示正确参数名是 file_path。我重新调用 count_file_lines。
[Action] : count_file_lines({"file_path": "schema.py", "ignore_empty": true})
[Observation] : (is_error=False) 68
--- Step 4 (耗时: 0.09 ms) ---
[Final Answer]: 当前目录下的核心数据结构文件 `schema.py` 共包含 68 行非空代码。
=== 3. 会话终态汇总 ===
Session ID : sess_8f3a91c04e2b
Final Status : completed
Total Steps : 4
Token Usage : Prompt=1800, Completion=180, Cached=1080
重点观察上面输出的 Step 2 和 Step 3：
在 Step 2 里，模型犯了一个非常典型的幻觉错误——它没有传必填的 file_path，而是自己编了一个 "path": "schema.py"。如果在普通的脚本里，这里直接就会抛异常退出；但在我们的 ToolRegistry.execute 中，这个错误被精准拦截，并生成了一条结构极其清晰的报错信息：
ValueError: Missing required arguments: ['file_path']. Expected schema: ...
这条报错作为 Observation (is_error=True) 被追加进了 AgentSession 的消息轨迹中。于是在 Step 3 里，模型一看到这条 Observation，立刻意识到了自己的参数名拼写错误，主动修正为 {"file_path": "schema.py"} 并顺利拿到了结果 68。
7. 当前实现的局限性与潜在隐患
虽然我们已经用不到 300 行代码跑通了一个支持自动 Schema 反射、支持双模协议、并且具备基础错误自修复能力的 ReAct Agent，但如果拿生产级系统的放大镜来审视，现在的实现还有三个非常明显的软肋：
进程内存态的脆弱性：现在的
AgentSession完全驻留在 Python 进程内存里，且runtime.run()是一个同步阻塞的while循环。如果一个复杂任务要跑 15 步，跑到第 8 步时服务器重启了，或者第 9 步要调用一个危险的delete_database工具，需要暂停下来发消息等人类管理员点击“同意（Approve）”才能继续——现在的while循环根本没法在中途把状态存盘挂起、等几个小时后再从第 9 步原地唤醒。死循环与重复动作陷阱（Action Loop Trap）：有些小参数量的开源模型在遇到工具报错时，不会像上面 Step 3 那样聪明地修改参数，而是会一模一样地把错误的
Action Input重复发十几次，直到把max_steps耗尽。Runtime 需要一种在状态机层面的“重复动作检测与干预机制”。JSON 解析依然处于“事后补救”阶段：我们在
_extract_json_payload里虽然用了raw_decode解决了括号嵌套问题，但如果模型生成的 JSON 本身语法就是坏的（比如字符串里包含了未转义的换行符），我们仍然要浪费一整轮 LLM 推理去让它重生成。
这些问题不需要靠堆砌复杂的业务 if-else 去打补丁，它们指向的是更底层的系统抽象。
小结
这一篇我们把第一篇定义的抽象骨架真正落地成了可运行的最小闭环。
回顾一下我们在这篇里完成的工程模块：
从数学与生成机理上拆解了 ReAct：明确了
Thought作为显式工作记忆降低动作分布条件熵的作用，以及为什么必须通过stop_words = ["\nObservation:"]截断推理引擎的自回归生成，防止模型陷入自问自答的幻觉区间。实现了零配置的
ToolRegistry：基于inspect.signature与typing.get_type_hints自动从原生 Python 函数生成 JSON Schema，并在执行层内置了必填校验、未知参数过滤、轻量级类型矫正与异常转Observation机制。实现了
ReActFormatAdapter与OpenAICompatibleClient：利用json.JSONDecoder().raw_decode解决了嵌套 JSON 提取难题，同时保持了历史消息序列的严格只追加（Append-Only）特性，完美兼容我们自己的mini vLLM与标准 OpenAI 接口。
下一篇文章（第 3 篇），我们会直面刚才提到的第一个核心痛点：《把执行抽象成 Session：从死循环到显式状态机》。
我们会像在 mini vLLM 里设计 Sequence 状态转移一样，为 AgentSession 引入完整的持久化快照（Checkpoint）、重复动作死循环检测（Loop Detector），以及真正可用于生产的 Human-in-the-loop（人工审批中断与跨进程恢复） 机制。
