做 Agent 从来不只是把模型接口调通那么简单。DS Harness 这套我内部代号为“H”的轻量运行时核心设计就两句话全插件化可回放会话日志。这篇文章不聊炫酷 demo只做工程化解剖——把 Agent 框架里最容易烂成毛线团的几个部分拆成清晰的边界再给整个运行过程装上一台能倒带的“行车记录仪”。插件化解决的是扩展问题会话日志解决的是复盘问题。如果你正在从“写一个调用模型的脚本”往“搭一套可维护的 Agent 系统”过渡这篇文章应该能帮你少走不少弯路。我会从设计动机讲起逐步拆开插件化架构的事件总线、插件注册表、配置驱动这些关键机制再深入到可回放会话日志的字段设计、回放器实现与缓存策略最后附上一份实操复盘和踩坑清单。全程尽量说人话代码也只是演示最小闭环不搞高大上的抽象层。1. 立项视角为什么需要一套可插拔的 Agent 运行时1.1 Agent 工程化真正难在哪先聊一个几乎所有人都会踩的坑。早先我做 Agent 原型时习惯把所有逻辑塞进同一个调度类模型适配、工具调用、记忆管理、日志输出全部写在主流程里。表面上看代码很“顺”从用户消息到模型回复一条直线写下来几百行搞定但一旦开始加第二个模型供应商、第三个自定义工具、第四种重试策略这个类就会迅速膨胀成维护噩梦。Agent 工程化的难点从来不是“让模型回答一个问题”而是围绕模型构建的系统边界被反复打破。具体来说有几类模型接口不统一。有的供应商用原生 tool calling有的要自己拼 prompt上下文截断逻辑也各有脾气。如果你在主流程里直接调某一家 SDK换模型基本等于重写调度逻辑。工具数量增长是常态。检索、计算、代码执行、数据库查询每个工具都有自己的参数规范和副作用。工具之间还可能互相调用形成复杂的依赖关系。编排策略高频变化。路由策略、重试策略、反思循环、多智能体协作这些属于“思路一变就要调整”的部分如果硬编码调整一次就动一次主流程。可观测性往往被忽略。线上跑出一个离谱回答最痛苦的是手里只有一行 “model returned error” 之类的文本日志完全不知道模型之前看到了什么、工具返回了什么、上下文在哪一步被截断。DS Harness 的设计思路很直接既然是边界问题就把边界定义清楚。凡是可能变化的模块全部做成插件凡是希望被观测的步骤全部发成事件。1.2 DS Harness 的设计取舍先定边界再造轮子全插件化的第一个原则是核心引擎越薄越好。DS Harness 的内核只做两件事事件路由和会话录制。模型适配、工具执行、记忆管理、日志输出、指标采集通通外挂成插件。乍一听有点反直觉有人会问核心引擎什么都不干那这个框架还有什么价值其实把约束定下来本身就是价值。内核不做任何业务判断只保证三件事事件能够有序地流转、插件能够按声明加载、所有事件能够被忠实记录。业务逻辑全部下沉到插件层插件之间互不感知只能通过事件通信。一个典型的功能分类如下插件类型典型示例变化频率模型适配多家模型接入、上下文截断策略中工具技能检索、代码执行、数据库操作高策略编排路由、重试、反思、多智能体协作高观测分析日志、指标、回放、链路追踪中用表格看就很清楚变化频率高的工具和策略恰恰是最应该插件化的部分。而模型适配虽然变化频率中等但因为切换成本最高所以也必须隔离。把这些模块外挂之后换模型、换工具、调策略都不再需要动主流程代价只是新增一个插件或者改一份配置。我不建议一开始就追求“万物皆插件”的极致纯度。DS Harness 的实际做法是先圈定候选边界把演化压力最大的四类模块插件化其余保持简单。如果一个模块半年都不变一次硬塞进插件体系只会增加理解成本。1.3 “可回放会话日志”不是日志是运行现场普通日志记录的是“发生了什么”比如这一行写着 “user message received”、那一行写着 “tool result returned”。这些文本对事后复盘有帮助但无法精确重建运行现场。你没法从普通日志里精确知道工具返回值的完整结构也没法把模型当时的完整请求体重新拼出来。可回放会话日志更进一步它把整个会话当作一条事件流把每一步的关键输入输出原样存入事件记录让事后可以“重放”整场运行。你可以把它理解成飞机上的黑匣子加上飞行模拟器——黑匣子完整记录仪表数据模拟器则能根据这些数据把飞行过程重新演示一遍。这个设计借用了事件溯源的思想。Agent 会话里的所有行为包括用户消息、模型请求、模型响应、工具调用、工具结果、上下文更新、系统状态变更都可以建模为不可变事件。事件一旦追加就不再修改如果某个步骤被发现是错的就用新事件去修正。这样整个会话的因果链条是完整的回放时只需要按照序列逐个重放事件就可以精确还原当时的运行状态。回放能力的价值在几个场景特别突出定位线上问题。模型说了一句胡话直接把当时的会话事件流翻出来看上下文、看工具返回、看模型请求很快能定位是哪个环节出了岔子。回归测试。改了提示词或者插件逻辑之后用历史会话跑一遍回放对比新旧回复观察行为变化是否符合预期。数据积累。事件流本身就是高质量的评测语料可以直接转成评测集或者微调数据。调优基线。同一段会话事件流分别接不同的策略插件对比各自的决策差异比人工翻日志直观得多。2. 全插件化架构的核心设计2.1 插件注册表、生命周期与依赖关系插件化的实现关键在三个部分注册表、生命周期、依赖管理。先给一份足够演示用的最小实现。# plugin_base.py from dataclasses import dataclass, field dataclass class PluginContext: config: dict bus: EventBus recorder: EventRecorder class Plugin: name: str version: str 1.0.0 priority: int 100 def setup(self, ctx: PluginContext) - None: 初始化和订阅事件 def teardown(self) - None: 清理资源允许重复调用注册表使用“名称到工厂函数”的映射而不是直接保存实例。# registry.py PLUGIN_REGISTRY: dict[str, callable] {} def register(name: str, factory: callable) - None: if name in PLUGIN_REGISTRY: raise ValueError(fplugin {name} already registered) PLUGIN_REGISTRY[name] factory def load_all(loading_order: list[str] | None None) - list[Plugin]: names loading_order or list(PLUGIN_REGISTRY) plugins [] for name in names: plugin PLUGIN_REGISTRY[name]() plugin.name getattr(plugin, name, None) or name plugins.append(plugin) return plugins这里有一个容易忽略的细节为什么用工厂函数而不是直接存实例因为插件加载顺序在注册阶段往往还不确定。如果 A 插件启动时需要 B 插件已经存在直接用实例会导致初始化顺序的强耦合。工厂函数把“构造”和“初始化”分离注册表只负责知道怎么造出插件真正的 setup 交给加载器统一调度。生命周期分为四个阶段加载构造对象、配置注入 PluginContext、启动调用 setup、卸载调用 teardown。DS Harness 要求 teardown 必须幂等因为异常退出时可能会重复调用清理逻辑。插件之间不允许互相 import只能通过事件总线通信。这条硬性规则几乎解决了循环依赖问题的全部——你不需要花一晚上梳理依赖图只需要保证每个插件是自洽的模块。优先级字段也不是万能的。它只决定 setup 的调用顺序不影响运行时的消息顺序。如果插件 A 必须处理某类事件的结果才能让插件 B 正常工作不要依赖 priority 去压执行顺序正确做法是让 A 处理后发布新事件B 订阅这个新事件。每个插件只对事件负责不假设其它插件做了什么。2.2 事件总线内核与插件之间唯一的通信方式DS Harness 的所有运行时行为都通过事件总线传递。用户消息、模型请求、模型响应、工具执行、工具结果、上下文更新全部统一为一种事件结构。# event.py dataclass class AgentEvent: seq: int ts: int # 纳秒时间戳 type: str # 事件类型 payload: dict # 事件载荷必须可 JSON 序列化事件总线的实现核心是串行分发。也就是同一时刻只处理一个事件保证 seq 顺序与真实发生顺序一致。这是可回放会话日志能成立的先决条件。# event_bus.py from collections import defaultdict class EventBus: def __init__(self): self._handlers defaultdict(list) self._queue [] self._processing False def subscribe(self, event_type: str, handler: callable) - None: if event_type *: event_type * self._handlers[event_type].append(handler) async def publish(self, event: AgentEvent) - None: self._queue.append(event) if not self._processing: await self._drain() async def _drain(self) - None: self._processing True try: while self._queue: event self._queue.pop(0) for handler in self._handlers.get(event.type, []) self._handlers.get(*, []): await handler(event) finally: self._processing False这个实现还有很多可优化空间但演示足够。核心思想有两个一是订阅方只需要声明自己关心的事件类型不用关心发布者是谁二是通配符订阅“*”给了日志插件一个“全量旁观”的入口方便实现会话录制。事件总线的好处不仅是解耦更重要的是它天然成为可观测性的数据源。日志插件只需要订阅“*”就把整个会话的所有关键动作录下来了。如果不引入总线想要录日志就得在每个调用点手动埋点不仅繁琐还很容易漏掉异常分支。举一个生活化的类比内核就像快递中转站插件是收件人。中转站不关心包裹里装了什么只负责按地址派送。收件人也不关心包裹是从哪里发出的只关心自己订阅的那一类快件。这样新增一个收件人不用改派送系统新增一种包裹也不用逐个通知所有收件人。2.3 配置驱动插件做好自己的隔离插件化如果只管代码扩展不管配置隔离用起来还是难受。DS Harness 的插件配置全部通过 PluginContext.config 注入插件只能读取自己命名空间下的配置项不能直接访问全局配置更不能偷偷读取其它插件的配置段。配置长这样plugins: tool_executor: enabled: true timeout: 30s allowlist: [] session_logger: sink: file://logs/session.jsonl sample_rate: 1.0 model_adapter: provider: demo model_name: demo-model max_tokens: 2048每个插件 setup 的时候拿到的是这个 YAML 里对应自己的字典。这么做的好处很直接配置与代码分离。同一个插件可以用完全不同的配置接入不同场景比如开发环境开全量日志生产环境开采样日志代码一行不用改。测试更容易。构造 PluginContext 时直接塞一份测试配置不需要加载整个框架。避免配置污染。插件 A 写错配置不会影响插件 B排查问题的时候边界清晰。配置校验我建议做在 setup 之前。早期 DS Harness 偷懒等到插件运行时报 KeyError 才发现配置不对排查成本很高。后来在注册表加载阶段就做一层 schemaless 校验类型不对、缺必填项启动时直接报错退出。这个习惯救了很多次后续的调优过程。配置驱动的另一个隐含价值是支持多实例。同一个框架进程里可以同时跑两个 Agent 实例一个用模型 A 全量日志一个用模型 B 采样日志互不干扰。这就是插件化加配置隔离的组合优势。3. 可回放会话日志的实现细节3.1 事件溯源状态变化是果事件才是因可回放会话日志看起来只是“把日志写得结构化一点”但真正做起来就会发现关键问题不是格式而是“应该记录什么”。DS Harness 的答案是记录事件而不是记录状态。传统日志记录的是系统状态的变化结果。比如你在日志里看到一行 “context truncated”你只知道上下文被截断了但不知道截断前是什么样、按照什么规则截断、丢了哪些内容。事件日志记录的是触发状态变化的事实本身当时有哪些消息、总量是多少、截断策略是什么、丢弃了哪些片段。这些事实一旦记录下来状态变化可以被任何人在任何时候重新推导出来。事件溯源的三个核心原则我直接抄下来当纪律事件不可变。已经写入日志的事件不允许修改或删除如果某个事件是错的追加一个新事件来标记修正。事件必须完整。一个事件如果说不清楚它的起因、输入和输出那它对于回放就是无效的。事件必须有序。事件流的顺序就是真相顺序乱掉等于真相被扭曲。这三个原则决定了会话日志的数据模型它是一条 append-only 的事件序列每一条记录都带有递增序号绝对不能改成数据库里随机更新的行。用快照和事件结合的方式控制体积。会话无限长时全量事件流会变得巨大回放速度也会下降。DS Harness 的做法是周期性打一张上下文快照快照记录某个 seq 时刻的完整状态之后的事件从快照开始重放即可不需要从 session 第一天开始跑。3.2 日志结构与字段约定会话日志采用 JSON Lines 格式每行一条事件。选 JSONL 的原因是结构清晰、可追加、方便 grep、方便被各种工具链消费。三个真实事件示例{session_id:s1,seq:1,ts:1710000000123456789,event_type:user_message,payload:{role:user,content:帮我查一下上周的销售数据}} {session_id:s1,seq:2,ts:1710000000123456790,event_type:model_request,payload:{model_name:demo-model,messages:[{role:user,content:帮我查一下上周的销售数据}],tools:[{name:query_sales}]}} {session_id:s1,seq:3,ts:1710000000123456800,event_type:tool_executed,payload:{tool_name:query_sales,arguments:{start_date:2025-01-01,end_date:2025-01-07},result:{rows:10}}}字段约定必须提前定死否则事后想改格式历史日志全废session_id会话标识用于把同一会话的所有事件聚合到一起。长会话建议按天滚动控制单文件体积。seq会话内单调递增的序号。这是回放的核心顺序依据绝不重复绝不小范围重置。ts纳秒时间戳。注意 ts 只作展示参考回放逻辑必须用 seq因为不同机器时钟可能有偏差。event_type事件类型白名单。推荐统一用蛇形命名比如 model_request、tool_executed方便订阅和过滤。payload事件载荷必须能被 json.dumps 直接处理。bytes、datetime、自定义对象都要提前转成基础类型。payload 序列化是最容易翻车的地方。模型返回里可能出现 bytes 字段、自定义异常对象工具结果里可能有 numpy 数组或者日期对象。DS Harness 在写入之前统一走一遍 sanitizer把非 JSON 类型显式转换转换失败就直接报错而不是静默丢弃。宁可在运行时炸出来也不要事后回放时发现 load 不进 JSON。写入模式必须是 append-only带上文件锁。多线程场景下如果各自持有一个文件句柄并发写行会交错JSONL 直接报废。最简单的方案是把日志写入放在事件总线串行分发里天然只有一条写入路径。如果非要异步落盘也要保证写入队列的严格顺序。3.3 回放器的执行流程与缓存策略有了完整的事件流回放器就是一个按 seq 重放事件的执行器。它的工作流分三步读取 JSONL按 seq 排序。虽然写入时已经有序但文件被裁剪或者拷贝时可能乱序回放前先排一遍更稳。跳过快照加载最近一次上下文快照。从快照之后第一个事件开始逐个交给事件总线分发。# replayer.py class Replayer: def __init__(self, harness: Harness, cache: dict[int, dict]): self.harness harness self.cache cache def replay(self, log_path: str) - None: events [] with open(log_path) as f: for line in f: events.append(json.loads(line)) events.sort(keylambda x: x[seq]) for event in events: if event[type] model_request: response self.cache.get(event[seq]) if response is None: raise LookupError(fmissing model response at seq{event[seq]}) self.harness.stub_model_response(response) self.harness.dispatch(event)这里的关键难点是“确定性重放”。Agent 运行时是一个包含随机性和外部依赖的系统直接重放往往得不到与线上一致的结果。三个主要干扰源模型生成过程本身有随机性。同样的输入两次采样可能得到不同的输出。想让回放结果稳定必须把模型真实响应缓存下来回放时直接使用缓存不再真正调用模型。工具调用有副作用。查询数据库会受数据更新影响发送通知真的会发出去。回放时要进入 dry-run 模式工具执行器只返回日志里记录的旧结果不真正执行任何副作用。随机数种子不固定。如果策略里用了随机采样或者 drop 某些 token回放时要固定所有 random 相关的 seed。模型响应缓存怎么落地最简单的方式是在真实运行时就把它写进日志。DS Harness 在 model_request 事件之后紧跟一条 model_response 事件记录完整输出。回放器遇到 model_request 时直接在事件流里查找对应的 model_response不用额外维护缓存文件。缺点是一次回放要把整个会话的事件流都读进内存但对大多数调试场景来说这不是问题。工具副作用也有类似的处理。tool_executed 事件里记录 arguments 和 result回放时工具插件读到事件后直接返回 result不执行真实调用。这样既能复现当时的决策流程又不会在回放时污染线上数据。4. 实操复盘从零实现最小可用 Harness4.1 先把内核跑通事件循环、模型适配、工具协议DS Harness 的最小内核我建议只做三件事维护事件循环、定义模型适配接口、定义工具协议。事件循环是内核的发动机。没有事件循环一切插件都无处挂载。最小实现大概是这个形态# harness.py class Harness: def __init__(self): self.bus EventBus() self.plugins: list[Plugin] [] self.recorder EventRecorder() self._responses: dict[int, dict] {} async def run(self, user_input: str) - str: self.bus.publish(AgentEvent(seq1, tstime_ns(), typeuser_message, payload{role: user, content: user_input})) await self.bus._drain() def dispatch(self, event: AgentEvent) - None: # 内核不处理业务只负责把事件交给订阅者并保证录制 self.recorder.record(event) loop asyncio.get_event_loop() loop.create_task(self.bus.publish(event))这段代码其实只是演示思路。真实实现里用户消息进入内核后内核会发布 user_message 事件模型适配插件订阅到事件后调用模型然后把模型输出作为新的 model_response 事件发布回总线。工具同理。内核完全不感知业务流程只负责“转发 录制”两个动作。模型适配层建议统一成一个接口# model_adapter.py class ModelAdapter(Plugin): async def complete(self, messages: list[dict], tools: list[dict] | None None, **kwargs) - dict: raise NotImplementedError所有供应商的差异都被隔离在这个接口后面。调用方只关心 messages、tools、返回的 response 结构不关心底层是哪个模型、用没用 tool calling。这个抽象虽然简单却解决了 Agent 框架里最常见的“换模型就要改主流程”的问题。工具协议直接采用 JSON Schema 数组这是目前最通用的 tool calling 描述方式。每个工具声明名称、描述、参数结构模型根据这些声明决定是否调用以及传入什么参数。工具插件负责执行并发布 tool_executed 事件。工具执行器的超时、重试、并发控制都属于策略逻辑建议放在工具执行插件内部处理而不是散落到各个工具实现里。4.2 手写第一个日志插件从旁观看清全链路内核跑通之后第一个值得写的插件就是会话日志插件。它不参与业务决策但能立刻让你“看见”整个框架的运行轨迹。# session_logger.py class SessionLogger(Plugin): name session_logger priority 10 def setup(self, ctx: PluginContext) - None: self.sink open(ctx.config[sink], a) ctx.bus.subscribe(*, self.on_event) def on_event(self, event: AgentEvent) - None: self.sink.write(json.dumps(event.__dict__) \n) self.sink.flush() def teardown(self) - None: self.sink.close()把这个插件注册进去跑一次最简单的用户消息日志文件里就会出现完整事件链user_message、model_request、model_response、tool_executed、tool_result、agent_reply。以前排查问题要加一堆 print现在打开 JSONL 文件每一步的输入输出都摆在那里。我自己在这个环节有一个很强的体感插件化的意义不在“能扩展”而在“能旁观”。日志插件不修改任何业务行为只是订阅全部事件就能拿到整个系统的运行全貌。这就是事件总线和全插件化结构带来的红利它让可观测性不再是一个需要侵入业务代码的额外功能而是框架与生俱来的属性。4.3 踩坑实录重放结果不一致的根源在哪里实操中我最头疼的问题就是“重放结果和线上不一致”。明明事件流完整无缺重放出来的分支走向却不一样。排查了一轮之后总结了几个根源第一集合遍历顺序不稳定。Python 里字典和集合的遍历顺序受哈希影响同一个进程内两次运行可能顺序不同。如果策略插件里有for action in available_actions:这种写法而 available_actions 是 set重放时顺序就可能变化导致 Agent 选择不同的下一个动作。修复方式很简单凡是参与决策的集合统一转成有序列表或者显式 sort。第二时间相关的逻辑在重放时“复活”。比如超时重试逻辑回放过程中因为墙钟时间已经过去某些超时判断会触发导致走了和线上不同的重试分支。DS Harness 的修复方案是回放时注入一个假时钟所有超时判断都基于事件里的 ts 推算而不是读当前系统时间。第三模型输出的缓存策略不对。之前只缓存了字符串形式的回复内容没缓存 tool_calls 的结构化参数和 logprobs。结果重放时工具参数解析失败直接让流程崩掉。后来改成缓存完整响应对象问题立刻消失。第四外部服务的副作用污染重放。查询一个实时 API 在重放时返回了不同的结果导致决策路径漂移。修复方式是工具执行器增加 dry-run 模式回放时永远使用日志里记录的旧结果。踩过这些坑之后我把“确定性重放”当成一门纪律来执行一切外部调用必须有缓存或 mock一切随机性来源必须显式固定一切时间判断必须基于事件时间而不是墙钟。做到了这三点重放结果基本上能做到与线上一致。5. 常见问题与排查速查表5.1 插件加载失败与顺序依赖插件体系常见的异常现象集中在加载阶段。直接给一张速查表比长篇大论好使。错误现象触发原因定位手段修复建议插件名字冲突两个插件注册了相同名称检查注册表是否在 register 时做重复检测register 时检测已存在名称并直接报错导入时循环引用插件 A 在模块顶层 import 插件 B观察报错堆栈看循环链插件之间只通过事件通信禁止模块顶层互相 importsetup 才报 KeyError配置缺项运行期才发现打印完整配置段对比文档在加载阶段做配置校验必填项缺失时启动失败初始化顺序不符合预期错误地用 priority 压业务执行顺序在 setup 中打印调用顺序运行期依赖用事件订阅关系表达不要依赖初始化顺序插件系统的核心禁忌有两个一是插件之间互相 import二是插件直接读取全局配置。这两条破戒之后短期看起来方便长期一定会出现“改一个插件影响另一个插件”的连锁事故。5.2 会话日志体积膨胀与采样策略全量事件日志的开销比传统文本日志大很多这是事实。一个长会话跑上几千轮工具调用JSONL 文件轻松到几百 MB。解决思路不是回到传统日志而是分层录制。策略做法适用场景全量录制所有事件原样写入不裁剪短会话、重要线上会话、回归测试样本白名单录制只记录 model_request、model_response、tool_executed 等关键事件数据量可控的常规会话采样录制按 session_id 或时间窗抽样大规模监控场景不需要每个会话都可回放快照 增量定期打上下文快照只保留增量事件超长会话防止从头回放我实测下来的建议是不要一上来就做复杂采样策略。先全量录制一个礼拜观察文件增长速度和实际使用频率再决定裁剪优先级。采样策略越复杂回放时缺失的信息就越多可回放性就越差。只有当你明确知道“哪些事件不需要回放”时才做白名单录制。5.3 回放过程卡住或结果漂移的排查方向回放器本身是一个相对独立的执行器但它复用了插件体系所以问题来源也比较集中。异常现象可能原因排查方法修复建议回放时卡在模型请求模型响应缓存缺失检查 model_response 事件是否存在真实运行时补全缓存回放器缺失时直接报错回放结果与线上不一致集合顺序、随机种子、假时钟未处理对比分支差异点查看决策输入统一集合为有序列表固定 seed注入假时钟回放时工具被真实调用工具执行器没有进入 dry-run检查工具插件实现打日志确认dry-run 模式下直接返回日志里的旧结果事件 seq 跳跃或乱序日志写入多线程并发查看文件里 seq 的单调性日志写入强制走单写者路径加载后按 seq 排序回放结果总是截断快照与增量事件不匹配检查快照 seq 之后是否有缺口用 seq 完整性校验发现缺口时提示重建快照回放最忌讳“大概能跑就行”。如果回放结果偶尔一致、偶尔漂移那说明确定性还没做到位这种状态下的回归测试基本没有参考价值。DS Harness 的做法是给回放器加了一个断言回放结束时会校验关键决策路径上的 seq 集合与线上记录一致不一致就直接失败不给出“看起来正常”的幻觉结论。最后分享一个我认为最有价值的实操经验如果你只做一件事来改进自己的 Agent 框架优先做可回放会话日志。插件化架构带来的扩展性不会立刻体现但一个能完整回放线上会话的日志系统会在你第一次遇到“模型乱说话又找不到原因”的故障时直接救你一次。事件流的设计和日志系统一旦搭好后续加监控、加评测、加回归测试都会变得极其自然。