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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 。拿 Key 的路径很直接进控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 在 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 新建一个密钥复制出来存到环境变量里。别硬编码进代码后面状态持久化会把配置一起落盘密钥泄露就麻烦了。export TAOTOKEN_API_KEYsk-你的密钥 export TAOTOKEN_BASE_URLhttps://taotoken.net/api装依赖就一个pip install openai如果你后面要跑长任务、做编码类 Agent可以顺带了解下 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 它针对多轮工具调用场景做了额度上的适配比按次调用更划算。接入细节和参数说明在文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里遇到 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_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api), ) MODEL gpt-4o-mini # 换成你账号下可用的模型名 STATE_FILE agent_state.jsonl3.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( modelMODEL, messagesmessages, toolsself.tools.schemas(), tool_choiceauto, ) 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: fThought: {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, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \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, encodingutf-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 0os.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(先算 1230再告诉我当前时间戳, resume_fromresume))第一次跑你会看到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:123042当前时间戳是 1735...,ts:...}验证崩溃恢复在任务跑到一半时按 CtrlC 杀掉进程然后重新执行python harness.py。因为last_step_idx()读回了已完成步数循环会从断点继续而不是从头再来。这就是事件溯源的价值——状态在磁盘上内存只是缓存。想单独验证模型连通性可以先用模型对话 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 发一条消息确认 Key 和模型名没问题再跑 Harness能省掉一半排障时间。5. 常见报错排查401 UnauthorizedKey 没读到或写错。检查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 这东西看十篇不如自己崩一次再恢复一次来得实在。