Agent Harness 工程核心架构拆解：300 行代码实现 ReAct 与 MCP 上下文管理-平芜编程栈

1. 为什么你的 Agent Demo 一上生产就崩

很多人第一次写 Agent，都是从一个while True循环开始的：调模型、解析工具调用、执行工具、把结果塞回消息列表、再调模型。本地跑个天气查询、算个加减法，顺得不行。可一旦任务变成十几步、工具变成七八个、还要跑几个小时，问题就全冒出来了——跑到第 6 步忘了第 2 步查过什么、进程一崩全部重来、工具报错就卡死、上下文越堆越长最后模型开始胡言乱语。

这不是模型不行，是 Harness 不行。Agent Harness 说白了就是包在模型外面那一整套运行时基础设施：它负责驱动 ReAct 循环、管理 MCP 工具调用、压缩和组装上下文、把每一步状态落盘、在出错时重试或降级。模型是发动机，Harness 是底盘、变速箱和刹车。发动机再强，底盘散架照样开不动。

这篇就按工程落地的思路，把 Harness 拆成四个核心模块——ReAct 循环、MCP 工具调用、上下文管理、状态持久化——给你一份 300 行左右能直接跑的 Python 骨架，再演示多轮任务下崩溃恢复的验证动作。适合已经写过 Demo、想把它变成"能托管跑"的开发者。代码依赖只有标准库加一个模型 SDK，你可以照着抄、改、扩展。

2. 前置准备：模型接入与 Key 获取

在写循环之前，先把模型调用这条链路打通。Harness 的所有模块最终都要落到"发一次请求、拿一次回复"上，所以先把接入层固定下来。

我这边统一走 TaoToken 的兼容接口，它同时提供对话模型和编码类模型，OpenAI SDK 直接改base_url就能用，省得为不同厂商写适配层。官网入口在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 基址是 https://taotoken.net/api 。

拿 Key 的路径很直接：进控制台 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，在 API Keys 页面 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 新建一个密钥，复制出来存到环境变量里。别硬编码进代码，后面状态持久化会把配置一起落盘，密钥泄露就麻烦了。

export TAOTOKEN_API_KEY="sk-你的密钥" export TAOTOKEN_BASE_URL="https://taotoken.net/api"

装依赖就一个：

pip install openai

如果你后面要跑长任务、做编码类 Agent，可以顺带了解下 Coding Plan https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ，它针对多轮工具调用场景做了额度上的适配，比按次调用更划算。接入细节和参数说明在文档 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 里，遇到 401/404 先翻这里。

3. 300 行骨架：四个模块怎么拼

先给整体结构，再逐块填。整个 Harness 就是一个类，内部持有四个子模块，主循环只做一件事：把当前状态喂给模型，拿到动作，执行，写回状态。

# harness.py import os, json, time, uuid, hashlib from dataclasses import dataclass, field, asdict from typing import Any, Callable from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ.get("TAOTOKEN_BASE_URL", "https://taotoken.net/api"), ) MODEL = "gpt-4o-mini" # 换成你账号下可用的模型名 STATE_FILE = "agent_state.jsonl"

3.1 ReAct 循环：心脏只有四步

ReAct 的本质是 Thought → Action → Observation → 再 Thought 的交替。别把它想复杂，循环体就是"问模型要下一步动作，执行，把结果追加进消息，再问"。

@dataclass class Step: idx: int thought: str action: str | None action_input: dict | None observation: str | None class ReActLoop: def __init__(self, tools: "ToolRegistry", ctx: "ContextManager", store: "StateStore"): self.tools = tools self.ctx = ctx self.store = store self.max_steps = 12 def run(self, task: str, resume_from: int = 0): self.ctx.reset(task) step_idx = resume_from while step_idx < self.max_steps: messages = self.ctx.build_messages() resp = client.chat.completions.create( model=MODEL, messages=messages, tools=self.tools.schemas(), tool_choice="auto", ) msg = resp.choices[0].message # 没有工具调用 = 模型认为任务结束 if not msg.tool_calls: self.store.append({"type": "final", "content": msg.content}) return msg.content for call in msg.tool_calls: name = call.function.name args = json.loads(call.function.arguments or "{}") obs = self.tools.invoke(name, args) step = Step(step_idx, msg.content or "", name, args, obs) self.store.append({"type": "step", **asdict(step)}) self.ctx.push_step(step) step_idx += 1 return "达到最大步数，任务未完成"

关键点：每一步都先store.append落盘，再更新内存上下文。顺序反了，崩溃时就会丢步。

3.2 MCP 工具调用：注册表 + 统一 Schema

工具不要散落在各处。用一个注册表集中管理，每个工具声明名字、描述、参数 schema 和实现函数。MCP 的价值就在于把"工具怎么描述、怎么调"标准化，这里先用本地注册表模拟，后面接真实 MCP Server 时把invoke换成协议调用即可。

class ToolRegistry: def __init__(self): self._tools: dict[str, dict] = {} def register(self, name: str, desc: str, params: dict, fn: Callable): self._tools[name] = {"desc": desc, "params": params, "fn": fn} def schemas(self): return [{ "type": "function", "function": { "name": n, "description": t["desc"], "parameters": t["params"], }, } for n, t in self._tools.items()] def invoke(self, name: str, args: dict) -> str: if name not in self._tools: return f"[error] 未知工具: {name}" try: return str(self._tools[name]["fn"](**args)) except Exception as e: return f"[error] {type(e).__name__}: {e}"

工具描述要写得像给新人看的文档：说清什么时候用、参数什么格式、边界在哪。工具数量控制在 5 个以内，多了模型选错概率直线上升。

3.3 上下文管理：静态在前，动态在后

上下文不是把所有消息一股脑塞进去。分三层：系统提示词（几乎不变，放最前，利于缓存）、压缩后的历史摘要、最近几步的原始记录。

class ContextManager: def __init__(self, system_prompt: str, keep_recent: int = 6): self.system_prompt = system_prompt self.keep_recent = keep_recent self.task = "" self.summary = "" self.recent: list[Step] = [] def reset(self, task: str): self.task = task self.summary = "" self.recent = [] def push_step(self, step: Step): self.recent.append(step) if len(self.recent) > self.keep_recent: self._compress() def _compress(self): old = self.recent[: -self.keep_recent] lines = [f"步骤{s.idx}: 调用{s.action}({s.action_input}) -> {s.observation}" for s in old] self.summary += "\n" + "\n".join(lines) self.recent = self.recent[-self.keep_recent:] def build_messages(self): msgs = [{"role": "system", "content": self.system_prompt}] if self.summary: msgs.append({"role": "system", "content": f"历史摘要:\n{self.summary}"}) msgs.append({"role": "user", "content": f"任务: {self.task}"}) for s in self.recent: if s.action: msgs.append({"role": "assistant", "content": f"Thought: {s.thought}\nAction: {s.action}"}) msgs.append({"role": "tool", "content": s.observation or ""}) return msgs

压缩策略是 Harness 里最需要调参的地方。太激进丢关键信息，太保守等于没压。先用"保留最近 6 步 + 更早的转成一行摘要"起步，跑起来再调。

3.4 状态持久化：JSONL 追加写，崩溃可恢复

这是最容易被忽略、却最救命的一块。每步一行 JSON 追加写磁盘，进程崩了，重启读回最后一行接着跑。

class StateStore: def __init__(self, path: str): self.path = path def append(self, record: dict): record["ts"] = time.time() with open(self.path, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") f.flush() os.fsync(f.fileno()) # 强制落盘，别省 def load_steps(self) -> list[dict]: if not os.path.exists(self.path): return [] with open(self.path, encoding="utf-8") as f: return [json.loads(l) for l in f if l.strip()] def last_step_idx(self) -> int: steps = [r for r in self.load_steps() if r.get("type") == "step"] return steps[-1]["idx"] + 1 if steps else 0

os.fsync那行别删。很多人只flush，结果断电或 kill -9 时缓冲区数据没落盘，恢复出来还是缺步。

4. 跑起来：验证请求与崩溃恢复

把四个模块拼起来，注册两个工具，跑一个多步任务。

def make_harness(): tools = ToolRegistry() tools.register("add", "计算两数之和", { "type": "object", "properties": {"a": {"type": "number"}, "b": {"type": "number"}}, "required": ["a", "b"], }, lambda a, b: a + b) tools.register("now", "返回当前时间戳", {"type": "object", "properties": {}}, lambda: time.time()) ctx = ContextManager("你是一个会使用工具的助手，逐步推理并调用工具。") store = StateStore(STATE_FILE) return ReActLoop(tools, ctx, store), store if __name__ == "__main__": loop, store = make_harness() resume = store.last_step_idx() print(f"从第 {resume} 步恢复") print(loop.run("先算 12+30，再告诉我当前时间戳", resume_from=resume))

第一次跑，你会看到agent_state.jsonl里一行行追加：

{"type":"step","idx":0,"thought":"先做加法","action":"add","action_input":{"a":12,"b":30},"observation":"42","ts":...} {"type":"step","idx":1,"thought":"再取时间","action":"now","action_input":{},"observation":"1735...","ts":...} {"type":"final","content":"12+30=42，当前时间戳是 1735...","ts":...}

验证崩溃恢复：在任务跑到一半时按 Ctrl+C 杀掉进程，然后重新执行python harness.py。因为last_step_idx()读回了已完成步数，循环会从断点继续，而不是从头再来。这就是事件溯源的价值——状态在磁盘上，内存只是缓存。

想单独验证模型连通性，可以先用模型对话 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=chat&utm_campaign=rewrite 发一条消息确认 Key 和模型名没问题，再跑 Harness，能省掉一半排障时间。

5. 常见报错排查

401 Unauthorized：Key 没读到或写错。检查echo $TAOTOKEN_API_KEY有没有值，注意别把引号也复制进去。

404 model not found：模型名写错，或者你账号下没开这个模型。换成控制台里列出的可用模型名。

工具调用参数解析失败：json.loads抛异常，多半是模型返回了非标准 JSON。在invoke外面包一层 try，把原始字符串也记进 observation，让模型看到自己错在哪，下一轮能自我纠正。

恢复后重复执行工具：说明append和push_step顺序反了，或者last_step_idx算错。记住先落盘再更新内存。

上下文爆炸：keep_recent设太大，或摘要没生效。打印build_messages()的 token 估算，超过模型窗口一半就该压。

循环停不下来：模型一直调工具不返回 final。加max_steps兜底，超了强制结束并落盘，别让它无限跑。

6. 下一步怎么扩展

这份骨架刻意做小，方便你读懂每一行。真要上生产，往三个方向加：一是把ToolRegistry.invoke换成真实 MCP 协议调用，工具就能跨进程、跨服务复用；二是给StateStore加事件回放，支持从任意历史步 fork 出新分支做调试；三是加权限门控，危险工具（写文件、发请求）执行前先过一道白名单校验。

代码骨架和接入配置都在这了，剩下的就是拿你自己的任务去跑、去踩坑。Harness 这东西，看十篇不如自己崩一次再恢复一次来得实在。