1. 先弄清楚这个 Harness 到底解决什么问题1.1 当 Agent 从“能跑”变成“难维护”我一开始接触 Agent 开发时其实没想太多。无非是把大模型接口接进来写一个循环让模型在“思考、调用工具、拿结果、再思考”之间转几圈。这个阶段代码量很小三五个函数就能跑通效果还挺惊人。可一旦你开始加记忆、加多轮对话、加各种外部工具问题就全来了。最典型的场景是某天线上 Agent 突然回答得牛头不对马嘴。你想排查到底哪一步出了错结果日志里只记了“用户问了什么”和“模型最终回了什么”中间几次工具调用的输入输出、模型临时生成的中间思考、上下文里塞了什么东西全部都是黑盒。你只能靠猜或者打个断点重新跑一遍。可 Agent 的不确定性偏偏就在这里——同样的输入重跑一遍可能走的是完全不同的路径。这时我意识到Agent 项目做大了之后难点早就不是“怎么让模型理解意图”而是工程化问题组件能不能随意插拔、运行过程能不能被完整观测、出问题之后能不能精准复现当时的过程。标题里提到的这套 Harness本质上就是围绕这两个问题做的一是把所有扩展点做成插件化二是把会话日志做成可回放的就像给 Agent 装了一台飞行记录仪。1.2 为什么可回放日志是调试 Agent 的关键先聊日志这件事。普通接口日志记录的是什么请求参数、返回结果、耗时。对无状态服务来说这些信息足够定位大多数问题了。但 Agent 是有状态的而且状态从一次模型调用传到下一次模型调用中间还夹杂着工具返回、用户补充、记忆检索结果。任何一环出了偏差最终表现可能截然不同。可回放日志的思路和行业里常见的 event sourcing 很像不只要记录“最终状态”还要把产生状态的每一个事件都存下来。启动回放时系统按时间顺序把事件一条条“喂”给运行时相当于把当时那轮对话重新演一遍。这样做的好处非常直接。第一复现 Bug 不再依赖运气拿到 session_id 就能还原现场。第二可以用回放日志做回归测试改完插件后把旧日志跑一遍看行为有没有漂移。第三它能当“监控摄像头”用让非技术角色也能看到 Agent 当时每一步在做什么而不是只有一个结论。1.3 全插件化不是赶时髦是解耦刚需插件化架构不是什么新概念很多优秀开源项目都在用。但我一开始在 Agent 框架里用插件化纯粹是被逼的。最早版本的代码工具调用是直接写在主循环里的。今天要加一个搜索插件就在循环里加一个 if明天要更换模型供应商就改主函数里那个请求方法。改到第三次的时候我就知道这条路不能继续走了——因为你会发现每换一次工具主流程代码就要跟着改一遍回归测试范围越来越大团队成员合代码时冲突也越来越频繁。全插件化的目标是把 Agent 运行时的骨架和具体能力彻底分离。骨架只负责调度拿到用户输入分发到模型决定下一步是继续推理还是调用工具把结果再送回模型。至于用哪个模型、挂哪些工具、怎么做记忆、怎么做权限控制全部由插件提供。主流程只认接口不认实现。这样带来的直接收益有三个新功能可以独立开发和测试不污染主干不同团队可以各做各的插件互不等待某个插件出了问题可以单独摘掉整个 Agent 不至于停摆。下一篇我会从接口设计开始把整套做法拆开讲。2. 全插件化设计接口怎么定才扛得住迭代2.1 插件的最小接口一切从上下文对象开始我见过很多插件框架翻车死因不在功能而在接口设计。接口定得太宽插件作者不知道该实现什么空方法满天飞定得太窄核心流程离不开的特性和插件能力出不来。所以我的做法是先定义一个极小的插件基类再按场景加细分接口。# harness/plugin.py class AgentPlugin: name: str unnamed version: str 0.0.1 priority: int 100 def __init__(self, config: dict | None None): self.config config or {} def initialize(self, runtime): 插件初始化此时可以注册回调、建立连接 def shutdown(self): 插件关闭释放资源 def before_turn(self, ctx): 每一轮 Agent 主循环开始前调用 def after_turn(self, ctx): 每一轮 Agent 主循环结束后调用这里的核心是ctx也就是 ExecutionContext。它承载了当前会话的全部状态用户输入、历史消息、模型上下文、工具调用记录、中间变量。插件要读状态从 ctx 里拿要改状态也往 ctx 里写。这样一个对象贯穿始终插件之间就不需要互相感知对方的存在了。你要问为什么前缀是 before 和 after 而不是让插件直接接管主循环原因很简单主循环是 Agent 的地基不能被插件随意改写。插件可以观察、可以干预、可以在边界处加逻辑但没有资格决定整个循环的流程。此时接管权始终在 Harness 手里Agent 的行为才能保持可预测。2.2 插件优先级与生命周期编排有了接口还不够还得回答一个问题多个插件同时监听同一个事件谁先执行我用的是优先级加阶段划分。每个插件声明一个 priority 值数值越小越先执行。比如日志插件优先级设成最高保证在任何业务结果产生之前就开始记录权限检查插件次之工具执行插件排在后面。为什么不能简单按加载顺序来因为配置文件的书写顺序太容易被改动而且 YAML 加载的顺序在多人协作时很难保证直观显式声明优先级反而是一种自我文档化。生命周期也需要管理。initialize 阶段做资源准备shutdown 阶段做资源释放中间是长时间运行的会话服务期。这样设计的意义是幂等性同一个插件可以随配置启停不会有“上次启动的残留状态影响本次运行”的隐患。# config/plugins.yaml plugins: - name: model_adapter priority: 0 config: base_url: http://127.0.0.1:9000/v1 model: demo-model - name: auth_filter priority: 10 config: allow_groups: [ops] - name: session_logger priority: 90 config: storage_dir: ./logs2.3 依赖注入插件之间只认配置不认类很多人写完插件框架下一步会踩的坑是插件之间互相依赖A 插件需要调用 B 插件内部的方法于是直接 import B 的类。这在一开始看着省事但很快会制造出一张无法解开的依赖网。我的解法是“依赖注入 服务注册表”。运行时启动时先把所有插件初始化再把它们按名字注册到一个 Services 容器里。插件之间如果需要协作不直接 import而是从容器里按接口名获取。class ServiceRegistry: def __init__(self): self._services {} def register(self, name, instance): self._services[name] instance def get(self, name): if name not in self._services: raise LookupError(fservice {name} not found) return self._services[name]这样做最大的价值在于隔离。插件 A 获取到的插件 B 实例可以是一个真实实现也可以是一个 mock 实现。在单元测试里这种切换是免费的。同时插件本身的配置全部来自外部初始化时不允许自带硬编码参数这样同一个插件在不同环境本地、测试、预发可以使用完全不同的配置而不需要改代码。还有一个细节很值得分享插件之间永远不要共享可变状态。如果你发现两个插件需要共享某个数据那就把数据提升到 ExecutionContext 里由 Harness 统一管理。插件拿数据时通过 ctx 的 get、set 方法操作这样数据的产生者和消费者之间就只剩下一个松散约定而不是一个强引用关系。3. 可回放会话日志把黑盒变成白盒3.1 日志数据模型事件流加状态快照可回放日志最核心的部分是数据结构的设计。我最后采用的方案是“事件流为主周期性快照为辅”的双层结构。事件流是一条只追加的序列每条事件记录一次原子操作。比如模型调用发起、模型响应返回、工具执行完成、上下文发生变更。所谓只追加就是说已经写入的事件不允许修改和删除。这样的好处是回放时能得到最忠实的历史还原不会出现“日志被后面的人改了”这种说不清的情况。但光有事件流还不够因为一个长会话可能有几百个事件回放时如果从头开始重算效率太低。所以我每累积一定数量的事件就会写入一份完整的状态快照。回放时先定位到最近的快照再重放到目标位置这相当于给回放做索引让时间复杂度从 O(n) 降到 O(log n) 量级。事件的结构大致是这样的dataclass class EventRecord: event_id: str session_id: str parent_event_id: str | None event_type: str timestamp: float payload: dictparent_event_id 是为了把事件串成一条调用链。比如用户提问产生了 model_request 事件中间触发了一次 function_call这个 function_call 的 parent 就是 model_requestfunction_call 结束后产生 function_result它的 parent 又是 function_call。整条链路一目了然排查时可以直接像看调用树一样定位问题。3.2 精确记录每一步模型请求和工具调用都不放过日志要回放第一步是确保每个关键动作都被记录。我这里的做法是在主循环的边界处埋点而不是在业务逻辑里到处加 print。因为主循环是必经之路任何插件执行前和执行后都会经过它。具体来说以下几类事件是必须记录的用户消息进入会话上下文组装完成即将发送给模型模型返回完整响应模型请求调用工具含参数工具返回结果上下文更新后的中间状态每一类事件都要带上足够的 payload。这里最容易被忽视的是上下文本身。很多人在调试时会抓狂模型为什么突然开始胡说八道大概率是前面某次工具调用把一大段噪声文本塞进了上下文。如果日志里不记录模型每次收到的实际消息列表这个锅会一直找不到罪魁祸首。除了记录内容我还会给每个事件加上一个 content_hash。开发时可以和当前内容做比对确认回放过程中状态没有发生漂移。这个字段对定位“日志难复现”类问题帮助极大——你可以很自信地判断到底是事件记录完整但环境变了还是事件记录本身有缺失。3.3 回放器怎么工作复现当时的每一步回放器是整套 Harness 里我最得意的一块。它做的事情可以理解成把日志当作剧本来演运行时是演员逐个事件去执行。具体流程是读取目标会话的事件流按 event_id 排序。找到最近的状态快照恢复 context。从快照后的第一个事件开始逐个派发。每处理完一个事件和日志里的 content_hash 做对比。拿 model_request 事件举例。回放器读到这个事件时先把事件里记录的 prompt 重新交给模型 API得到一个新的响应再和日志里记录的 model_response 对比。如果一致说明这一环没有问题如果不一致说明问题出在模型侧可能模型版本、参数配置、甚至随机温度设置发生了变化。工具事件的回放要单独说明。工具是有副作用的重放时我默认不会真的二次执行外部操作。比如一个发送邮件的工具回放时只会从日志里还原“它返回了什么”而不是真的再发一封邮件。这就需要事件记录足够完整把入参和出参都原样保存。这是回放安全性的底线同时也是这类设计比“按原样重跑一遍”更实用的原因。4. 实操从零搭一个最小可运行版本4.1 项目骨架与目录结构概念说了这么多还是上一份能直接跑的示例更有说服力。我清理了一套缩略版核心模块只有三个调度器、插件管理器、日志记录器。harness-demo/ ├── harness/ │ ├── __init__.py │ ├── runtime.py │ ├── plugin.py │ ├── registry.py │ ├── events.py │ └── replay.py ├── plugins/ │ ├── model_client.py │ ├── tool_executor.py │ └── session_logger.py ├── config/ │ └── plugins.yaml └── main.py我建议先跑通这条最小链路再逐步加复杂插件。因为这个骨架一旦建立后续新增插件只需要三步定义插件类、在配置文件里注册、重启运行时。4.2 主循环与调度器实现主循环是整套框架最核心的部分它负责把一次用户输入转化成若干轮模型调用和工具调用。# harness/runtime.py class AgentRuntime: def __init__(self, plugins): self.plugins plugins self.registry ServiceRegistry() self.events EventLog() async def initialize(self): # 先初始化所有插件 for plugin in sorted(self.plugins, keylambda p: p.priority): plugin.initialize(self) self.registry.register(plugin.name, plugin) async def run_turn(self, user_input: str, session_id: str): ctx ExecutionContext(user_inputuser_input, session_idsession_id) for plugin in self.plugins: plugin.before_turn(ctx) while ctx.should_continue(): model_resp await self._call_model(ctx) if model_resp.tool_calls: for call in model_resp.tool_calls: result await self._dispatch_tool(call) ctx.add_tool_result(result) else: ctx.finish(model_resp.content) for plugin in self.plugins: plugin.after_turn(ctx) return ctx.get_final_message()这里的思路是主循环不直接依赖任何具体的模型客户端或工具实现。_call_model从 registry 里拿 model_adapter_dispatch_tool从 registry 里拿对应的 tool executor。这样即使模型供应商整个换了主循环也不会有任何改动。4.3 事件日志和回放器的代码事件日志模块在初始化时注册进插件列表。由于它的 priority 最高它会先于所有业务插件拿到 before_turn 的通知。# harness/events.py class EventLog: def __init__(self, storage): self.storage storage def record(self, event_type, payload, session_id, parent_idNone): event EventRecord( event_iduuid4().hex, session_idsession_id, parent_event_idparent_id, event_typeevent_type, timestamptime.time(), payloadpayload, ) self.storage.append(event) return event回放器的骨架更直接# harness/replay.py async def replay_session(session_id, log_store, runtime_factory): events log_store.get_events(session_id) state await restore_nearest_snapshot(session_id) runtime await runtime_factory.create() for event in events: await replay_event(runtime, event) verify_hash(event)回放时最关键的一点是工具调用不要真的执行副作用。我在 replay_event 里做了一个判断如果事件类型是 tool_execution就从登录里直接恢复函数返回值如果事件类型是 model_request才真正调用一次模型并把新结果和原始结果对比。4.4 用一份会话日志验证回放成果验证逻辑不复杂。我本地启动一个模拟环境让 Agent 完成一次带工具调用的对话比如查询某个指标。结束后导出日志清空上下文调用 replay看它是否成功还原出每个步骤日志里记录的唯一 ID 是否一一对应模型响应是不是和原来一致工具参数是否被原样读出来final answer 是否一致这个 demo 里我用的是一个非常小的模型适配层输入输出都是简单 JSON。跑通之后把同一套日志丢给一个什么都没装的空环境也能得出完整调用链那一刻你会真切感觉到“黑盒变白盒”是什么体验。5. 踩过的坑和排查技巧实录5.1 常见问题速查表在实际开发这套东西的过程中我踩过的坑远比顺利的地方多。这里整理了一份速查表供参考问题现象常见原因解决建议回放时模型响应和日志不一致模型服务带随机性或上下文内容不完整记录上下文时使用 content_hash回放时先验证内容一致插件明明注册了但没有被调用优先级排序错误导致被其他插件阻断检查插件优先级注意 before_turn 中被 return 拦截日志事件顺序错乱异步回调里没有携带 parent_event_id在事件初始化时强制要求传入父事件 ID不要默认置空慢日志阻塞主流程直接同步写入存储日志写入放进异步队列丢日志不丢主流程二次开发后重放旧日志失败事件 schema 变更给事件类型加版本号做兼容解析另一个强烈建议是定期抽查日志的完整性。我早期就碰到过一种情况模型响应太长超过了存储的单条大小限制结果这条事件被静默丢弃导致一次回放链条中间断裂。后来在写入接口里加了一个“写入后立即回读”的校验及时发现此类问题。5.2 几个值得坚持的实操习惯最后分享几个我后来才悟到的习惯希望你能少走弯路。第一不要太早做插件框架的抽象。可以先让功能以“战术实现”的方式跑通当你发现自己第三次在复制粘贴同一段代码时再提取抽象也不迟。我最初的模型适配层就是写在主函数里的直到第二个模型客户端接入才意识到必须拆成插件。过早抽象会让你所有判断都建立在猜之上。第二日志记录的字段宁多勿缺。早期我为了省钱把不用的字段都省了只记录核心参数。后来调试一个工具执行超时的 Bug发现少了发起时间戳字段根本没法判断是网络慢还是处理慢。从此我的原则是事件记录多存几个字段存储成本可以压缩但排查时缺失的字段是补不回来的。第三回放器一定要支持“在指定事件处暂停”。这不是需求文档里写的功能是我有一天排查问题时有感而发。定位到某个事件后你希望先检查当时的上下文再决定要不要继续重放。回放器支持断点之后它的角色从一个“日志查看器”变成了真正的交互式调试工具价值完全不同。5.3 后续扩展的三个方向这套 Harness 现在还能继续进化我已经在盘算三个方向。一是支持多会话并行回放用于批量回归。比如把一周的所有线上 Session 全部回放一遍自动找出上下文构造异常或者工具接口不稳的会话这就比人工抽查高效太多。二是把日志事件关联到更细粒度的链路追踪。目前的 parent_id 只能还原函数调用关系但没办法辨别哪个环节消耗了主要的 token 和时间。下一步打算给每次模型调用加上 token 预估把成本数据也变成事件的一部分。三是插件市场。当插件接口稳定后团队之外的开发者也可以按统一规范写插件放上仓库供别人直接安装。这一步一旦走通框架就从一个人的内部工具变成了一个真正的生态起点。我现在的实现还只是最小骨架但方向上我看得很清楚。