最近把手上一个 Agent 运行框架从“能跑”重构到了“可维护、可调试、可观测”的形态。这个项目对外代号叫 DeepSeek Harness内部一般叫 X-Harness核心目标一句话让基于 DeepSeek 这类对话大模型的 Agent 从一段拼提示词的脚本变成一个可以被上线、监控、回溯、灰度、隔离故障的系统。我接触过的多数 Agent 项目最初都是教程式写法两三个工具函数、几十行上下文拼接、一个 while 循环demo 十分钟搞定。这个问题是被低估的——脚本化的 demo 跑起来很简单但等到工具数量上到两位数、需要加安全校验、加审计、加成本统计时所有的横切逻辑都只能往主循环里塞改一处就牵全身。这篇文章就是这套框架的工程化解剖重点讲两个我认为收益最高的设计全插件化架构以及可回放的会话日志。适合正在写 Agent、又终于受够了“一次性脚本”的开发者如果你是做 Agent 平台或基础设施方向的也值得对照着看方案差异。1. 项目背景为什么 Agent 必须有一个 Harness1.1 脚本式 Agent 的四个痛点在我动手写 Harness 之前线上 Agent 一直是“能跑但不敢动”的状态。总结下来有四个典型问题我认为它们不是单点 bug而是结构性问题编排逻辑和业务逻辑彻底耦合。工具调用直接写在主循环里想加一个“工具耗时统计”都要找到所有调用点漏掉一个分支统计就不准。更别提审计、脱敏这类需要覆盖全链路的横切关注点。中间过程完全不可观测。用户只看到最终答案但 Agent 在这一轮里到底调了哪个工具、传了什么参数、模型中途为什么转向这些事实全都丢失。一旦出问题只能靠日志里的几个 print 猜。失败不可复现。LLM 本身是采样式的temperature 不为 0 时同一个输入两次输出就可能不同线上偶发问题到本地根本稳定复现不出来。没有会话过程的完整记录排查基本靠脑补。扩展成本随规模爆炸。加一个功能不再只改一个函数而是要处理“在哪个环节插桩、怎么传递上下文、怎么保证异常时不影响主链路”这些成本叠加起来超过了 Agent 本身的功能开发。印象最深的一次线上 Agent 在工具返回结果之后开始乱答。我翻遍代码也没看出问题最后才发现是工具返回里带了一段非法字符进入上下文后把后续内容截断了。没有中间事件日志这种问题只能靠反复打印和猜测。1.2 五层架构把会话编排拆成可替换零件Harness 的整体结构我按职责拆成了五层每一层只解决一类问题接入层 HTTP 网关 / 命令行入口 / 消息队列消费者 编排层 Orchestrator会话循环、停止条件、上下文裁剪 扩展层 Plugin System统一钩子分发所有横切能力都走插件 适配层 Model Adapter Tool Registry模型服务与工具函数的统一封装 支撑层 Session Log事件日志、快照、回放引擎独立运行优先级最高的是“扩展层”和“支撑层”。接入层和编排层解决的是会话流程问题而插件化和日志回放解决的是工程化问题——前者让系统易扩展后者让系统可诊断。二者在运行时是正交的编排层只管状态迁移不关心谁在监听Session Log 全程旁路不阻塞任何决策。为什么不用市面上现成的编排框架我的判断是现成框架把“节点图”抽象得足够好但横切关注点依然是侵入式的。你依然要在每个节点代码里手写观测、鉴权、审计逻辑。我要的不是节点级的封装而是事件级的拦截能力所以选择“更薄的编排 更厚的事件总线”。1.3 关键取舍事件驱动加旁路日志架构层面最值得说的是两个取舍。第一编排层用事件驱动而不是简单的函数调用链。每个状态迁移都会广播对应事件插件通过监听事件介入流程。代价是引入了一层间接调用但收益是主链路与扩展逻辑彻底解耦。Agent 的流程本身就是一个状态机接收输入、调用模型、执行工具、判断结束条件每个状态之间的事件点天然存在。第二日志系统作为旁路系统运行不应该影响主链路性能。事件写入走内存队列后台异步落盘队列积压时宁可丢日志也不能拖慢 Agent 响应。后面我会详细展开这个设计。2. 全插件化设计核心扩展点与分发机制2.1 插件协议定义一个最小接口的取舍插件系统首先要解决的是“插件长什么样”。我参考过不少框架的做法最后没有用复杂的抽象基类而是采用了数据类加 Protocol 的思路。核心接口如下# plugin_base.py from dataclasses import dataclass, field from typing import Any, Callable, Dict, Optional dataclass class PluginMeta: name: str version: str description: str enabled: bool True dataclass class HookContext: event: str # 例如 before_llm session_id: str payload: Dict[str, Any] # 事件数据可读可改 state: Dict[str, Any] # 插件共享状态区按命名空间隔离 logger: Any None dataclass class HookResult: action: str continue # continue / skip / override / halt extra: Dict[str, Any] field(default_factorydict) class BasePlugin: meta: PluginMeta def on_register(self, bus): 注册钩子到事件总线 ... def on_unregister(self): 清理资源取消钩子注册 ...为什么不用 ABC 强制子类实现统一方法因为插件的核心是“行为约定”不是继承关系。用数据类和协议插件作者只需要实现 on_register 把自己感兴趣的事件处理函数挂上去框架通过 meta 读取插件元信息解耦程度更高。插件只需要遵循协议不需要理解框架的内部继承结构。插件生命周期分为注册期和运行期。注册期做钩子挂载运行期处理事件。on_unregister 不是可有可无的——热更新插件时状态残留是常见问题清理逻辑必须放在这里。2.2 钩子点设计在哪一步允许插件插一手全插件化的关键不是“有个插件类”而是“有足够多的介入点”。我这边定义了一组核心事件覆盖 Agent 决策全过程事件名触发时机典型插件场景session.start会话创建权限校验、配额预扣、会话初始化user.message用户消息进入输入脱敏、意图改写、敏感词拦截before_llm调用模型前注入上下文、token 预估、路由选择after_llm模型返回后输出校验、schema 纠错、格式修复before_tool_call工具执行前权限拦截、参数修正、限流控制after_tool_call工具执行后结果摘要、成本统计、错误处理completion_end回复结束审计归档、会话评分、缓存写入一次完整会话里的事件顺序大致是session.start - user.message - before_llm - after_llm - before_tool_call - after_tool_call - before_llm - after_llm - completion_end事件总线负责把每个事件按顺序交给已注册的处理器。这里必须考虑三个工程细节优先级、异常隔离、结果动作。看代码class EventBus: def __init__(self): self._handlers: Dict[str, list] {} def register(self, event, handler, priority0, plugin): self._handlers.setdefault(event, []).append({ plugin: plugin, handler: handler, priority: priority, }) self._handlers[event].sort(keylambda h: h[priority]) async def emit(self, event, ctx: HookContext) - HookResult: result HookResult() for h in self._handlers.get(event, []): try: r await h[handler](ctx) if r.action ! continue: result r if r.action in (skip, halt): break except Exception: ctx.logger.exception(plugin hook failed: %s, h[plugin]) return resultaction 字段是我比较得意的设计。continue 表示“我观察完了继续”skip 表示“跳过这次系统默认动作”比如某个安全插件判断当前请求不该调用工具可以直接让后续工具执行被跳过override 表示“用我的返回值替换默认结果”适合做模型回退halt 表示“立刻终止会话”用于安全红线场景。异常隔离也很重要。单个插件抛出异常不应该让整个 Agent 崩溃而是记入错误日志由专门的巡检插件决定后续动作。插件是无辜的它们只是你的扩展扩展不该有能力杀死主进程——除非是显式的 halt 动作。2.3 热加载与配置注入不停机上能力的实现细节插件系统还有一个绕不开的能力热加载。线上 Agent 正在跑现在要加一个敏感词过滤插件总不能重启服务让所有会话中断。Python 场景下用 importlib 可以实现动态加载import importlib.util import sys from pathlib import Path def load_plugin(path: Path): spec importlib.util.spec_from_file_location(path.stem, path) mod importlib.util.module_from_spec(spec) sys.modules[path.stem] mod spec.loader.exec_module(mod) return mod这段代码有两个非常容易踩的坑。第一个是 sys.modules 缓存问题同一文件名的插件二次加载会拿到上一次缓存的旧模块新代码不会生效。稳妥做法是在加载前先清理缓存sys.modules.pop(path.stem, None)。第二个是插件依赖的兄弟模块互相引用问题插件 A import 插件 B 时如果 B 已经在 sys.modules 里有旧版本A 拿到的也是旧的。我的建议是插件之间不要相互 import需要共享能力就走事件总线广播保持解耦。配置注入方面插件不应该自己去读配置文件。正确方式是框架在注册期读取配置注入到 HookContext.state。配置长这样plugin: plugins_dir: ./plugins enabled: audit: { priority: 100 } token_counter: { priority: 10 } sensitive_filter: { priority: 999 }这样每个插件启动时只需要从 state 里拿自己的配置命名空间避免多个插件并发读文件造成 I/O 竞争也避免插件各自定义配置路径导致部署混乱。3. 可回放会话日志数据模型与回放引擎3.1 事件模型日志里必须有哪些事实可回放日志的核心不是“记录结果”而是“记录影响决策的每一个中间事实”。只有把模型输出、工具入参出参、插件事件、会话状态迁移全部落盘才能在事后完整重建当时的决策过程。我定义了下面这些事件类型作为日志的骨架- session.start: 会话 ID、模型配置、启动插件清单 - user.message: 用户原文 - agent.delta: 模型流式输出增量可开关 - agent.complete: 完整回复、token 用量 - tool.call: 工具名、入参 - tool.result: 输出、耗时、是否异常 - plugin.event: 插件自身产生的事件 - session.end: 结束原因、总耗时每条日志事件包含统一字段{ seq: 17, ts: 2025-06-12T14:22:31.891234567Z, event: tool.call, session_id: s_0f8a, payload: {tool: search_stock, args: {code: 600519}} }seq 字段是整个回放系统的基础。它必须是会话内单调递增的序号由事件总线统一分配而不是靠各协程自己生成。否则并发场景下日志写入顺序和实际决策顺序会错位回放出的流程就会和真实情况不一致。时间戳用纳秒级不是为了好看而是为了解决同一毫秒内多个事件的排序问题但这不能替代 seq——网络抖动、时钟回拨都可能让时间戳不可靠。3.2 序列化、脱敏与存储策略日志文件格式我选的是 JSON Lines每行一个独立 JSON 对象。原因很简单便于用 jq、grep 这类工具直接分析不用引入额外解析依赖每个会话一个文件按日期分目录存放清理策略也简单。存储路径组织成logs/日期/session_id.jsonl文件名即会话标识。脱敏是日志系统必须内置的能力。日志会留存一段时间用户消息和工具结果里不可避免会出现手机号、卡号等敏感信息。我的做法是在事件进入日志队列之前经过一个内置 redact 插件# redact.py 示例简化 import re PHONE_PATTERN re.compile(r1[3-9]\d{9}) def redact_text(text: str) - str: text PHONE_PATTERN.sub([电话], text) # 其他规则身份证、卡号、邮箱等 return text这条经验值得记住脱敏一定要在采集点做不能只做在展示层。很多系统是在日志查询界面脱敏原始数据早就完整落盘了这是审计上的硬伤。会话日志本质上是敏感数据源从源头处理才靠谱。事件写入的性能问题后面单独讲这里先提结论走异步队列批量写盘绝不 fsync 每条事件。3.3 回放引擎的两种模式回放引擎是这个系统里含金量最高的部分。它解决的是 Agent 开发调试里最痛苦的问题问题不可复现。我把回放分成两种模式分别应对不同场景模式是否调用真实模型/工具用途离线确定性回放否回归测试、审计、故障复盘在线步进调试否但可暂停改状态开发中逐事件观察、调试状态流转核心原则是回放不是重新执行一遍而是按事件序列把日志内容“还原”到运行时。在回放模式下模型返回和工具结果都不再重新获取而是直接使用日志里记录的内容进行注入。这样保证了确定性——日志里是什么回放结果就是什么。实现上回放引擎就是一个按 seq 推进的事件迭代器class ReplayEngine: def __init__(self, events): self._events events self._index 0 async def step(self): if self._index len(self._events): raise EOFError(replay finished) event self._events[self._index] self._index 1 return event打个比方一次会话就是一部录像录像里已经写死了每个镜头的内容。回放引擎就是播放器你可以快进、暂停、逐帧查看甚至裁剪掉某个插件事件试一次对比结果。这不只是调试工具也是一种审计手段——任何人想复现当时的会话只需要拿到那份 JSONL。3.4 回放与真实运行的差异处理回放模式要重点处理两类差异处理不好就会“回放复现不出线上问题”。第一类是随机性差异。Agent 决策过程中用到随机数的地方不只有 LLM 采样还包括工具自身的随机逻辑、插件里的随机权重。我建议在会话开始时记录一个 RNG seed正常情况下种子只用于调试模式让随机过程可复现。第二类是副作用差异。如果工具调用会写数据库、发消息、调外部系统回放时绝对不能真的执行它。正确做法是在回放模式下把工具标记为 side_effect_free直接返回日志里的 tool.result。日志里记录的入参和出参足够让开发者判断问题不需要真实副作用。实际开发中回放不一致的最大原因往往不是这些而是事件记录时缺了前置事实。比如上下文的裁剪过程如果没有记录完整 prompt回放时模型上下文长度和线上不一致后续决策自然对不上。所以日志系统宁可多记录也不要漏记录关键上下文快照。4. 核心实现与实操记录4.1 从零写一个最小插件写一个实际插件来演示流程。需求统计每次会话的 token 消耗。这个插件不碰主逻辑只注册两个钩子# plugins/token_counter.py from plugin_base import BasePlugin, HookContext, HookResult def estimate_tokens(text): # 粗略估算中文按字英文按词 zh_count sum(1 for ch in text if \u4e00 ch \u9fff) en_words len(text.split()) return int(zh_count * 1.2 en_words * 0.8) class TokenCounter(BasePlugin): meta PluginMeta(nametoken_counter, version1.0.0) def on_register(self, bus): bus.register(before_llm, self.count_request, priority10, plugintoken_counter) bus.register(after_llm, self.count_response, priority10, plugintoken_counter) async def count_request(self, ctx: HookContext): prompt ctx.payload.get(prompt, ) ctx.state[token_counter:req] estimate_tokens(prompt) return HookResult() async def count_response(self, ctx: HookContext): resp ctx.payload.get(response, ) req_tokens ctx.state.get(token_counter:req, 0) resp_tokens estimate_tokens(resp) ctx.logger.info( tokens req%d resp%d total%d, req_tokens, resp_tokens, req_tokens resp_tokens, ) return HookResult()这个插件完整展示了插件开发的三个要点注册期声明钩子、运行期读写 state、通过 logger 而非 print 输出观测信息。配置到 enabled 列表后重启 Harness 即生效。整个过程没有碰任何一条 Agent 主逻辑代码。4.2 模型适配层统一封装模型服务Harness 要接 DeepSeek 这类模型服务适配层的作用是把外部接口差异隔离在后面。统一接口长这样class ModelAdapter: def __init__(self, config): self.endpoint config[endpoint_url] self.api_key config[api_key] self.timeout config.get(timeout, 60) self.max_retries config.get(max_retries, 3) async def chat(self, messages, streamTrue): last_exc None for attempt in range(self.max_retries): try: return await self._request(messages, stream) except TimeoutError as e: last_exc e delay 0.5 * (2 ** attempt) # 指数退避 await asyncio.sleep(delay) raise last_exc工程细节都在这里了。超时要区分 connect_timeout 和 read_timeout——模型流式返回时 read_timeout 尤其关键长时间没有新 token 可能只是思考中粗暴断开会导致大量重试。重试要区分可重试错误和不可重试错误5xx、超时可重试401 鉴权失败、400 参数错误这类就算重试十次结果也一样。不要在 401 上浪费宝贵的调用次数。流式模式下还要处理半包和断流。响应内容按行解析每一段数据对应增量内容这个增量内容在写日志时对应 agent.delta 事件。日志里记录完整回复还是在每个 delta 都记录我建议在配置里提供开关默认关闭 delta 记录以省空间排查问题时再开启。4.3 一次完整的故障回放过程讲一个实际案例。某次线上出问题用户问某只股票行情Agent 直接崩了。常规排查是翻代码、猜原因、等下次复现。用了 Harness 之后流程完全不同。第一步定位会话日志。通过业务侧拿到的 session_id找到对应 JSONL 文件jq -r select(.eventtool.result and .payload.toolsearch_stock) | .seq \ logs/2025-06-12/s_0f8a.jsonl第二步跑回放。回放引擎会把事件按 seq 逐条还原并标记哪些是 from_log、哪些是当前真实执行的插件结果python -m harness.replay --session s_0f8a --trace回放在 after_llm 钩子位置抛出了异常。仔细比对日志发现模型这次返回了一个新格式的 JSON——结构体里多了一个嵌套字段解析器没有兼容。这个问题在真实环境下可能一个月才触发一次因为模型输出的 schema 本来就有概率波动但回放把它稳定复现了。第三步修好解析器再跑一次同一个 session回放通过。问题从“线上偶发”变成了“可复现、可验证、可回归”的普通 bug。对我个人来说这是这套系统带来的最大价值用户不需要再复现一次日志本身就是复现结果。4.4 回放与自动化回归的对接回放不只在出问题时用。沉淀下来的“黄金会话”可以整理成回归集接入 CI。每次修改 prompt、调整插件优先级、升级模型适配层都跑一遍历史回放看有没有破坏原有行为case ReplayCase.load(cases/s_0f8a.jsonl) result harness.run_replay(case) assert result.ok, result.error这些回放用例省去了构造 mock 数据的时间因为每个用例直接来自真实线上会话。你不需要手工拼 Prompt、造工具返回值整个 Agent 决策过程天然带着完整上下文。回放集积累得越久回归覆盖度越高系统对改动的免疫力也越强。5. 常见问题与排查技巧实录5.1 插件加载失败时先查这几处插件系统最常碰到的不是业务逻辑 bug而是加载期问题现象可能原因排查方向ModuleNotFoundError插件依赖缺失、sys.path 没包含插件目录检查插件 requirements确认启动参数钩子注册了但没触发事件名拼写不一致、插件未启用打印总线事件列表核对配置热加载后状态残留sys.modules 缓存了旧模块二次加载前先 del sys.modules[name]插件异常导致主链路中断emit 内异常隔离失效确认是否捕获了 Exception而非只捕获自定义异常另一个容易忽略的坑是把插件的 import 放在全局启动时执行。这样会让热加载机制失效——因为你 import 的是系统模块名后续再动态加载新版本时会被 sys.modules 里的旧缓存覆盖。正确做法是全程用 importlib 独立加载插件之间不要互相 import。5.2 回放不一致的典型原因回放不一致是最容易让人怀疑人生的问题。日志就在这里事件顺序也对但回放结果就是复现不出当时的现场。常见的坑按概率排序seq 错乱。并发协程没有经过统一发号器多个事件同时写入日志顺序与真实决策顺序不一致。这个问题要根治不能靠回放时重新排序。时间戳精度不够。秒级时间戳在并发场景下根本排不出先后统一用纳秒级时间戳。上下文前置事实缺失。日志没记录上下文裁剪前的完整模型输入回放时上下文长度与线上不一致。工具副作用被真实执行。回放模式必须标记工具为 side_effect_free外部系统状态变了回放结果不可能和线上一致。最直接的排查手法是找一次确认回放不一致的会话手工把日志从头到尾读一遍用 seq 排序后看有没有事件跳跃。跳跃意味着采集端丢事件要在采集端找原因而不是回放端。5.3 日志写入的性能优化流式输出场景下agent.delta 事件可能每几十毫秒就产生一条。如果每条事件都立刻写文件并 fsync主链路性能会断崖式下跌。我的实现方案是采集端只负责把事件放进内存队列不做任何文件操作后台异步任务每 200ms 或积攒到 512KB批量写入 JSONL 文件一次队列满时丢弃非关键事件如 agent.delta优先保证 session.start、tool.call、agent.complete 这类核心事件不丢提供“低采样模式”开启后不记录 delta只记录完整的 request 和 response这样一个标准会话的日志文件通常只有几十 KB对磁盘压力很小即使高并发场景也能维持主链路零卡顿。5.4 定位问题的实用命令我常用的三条命令极简但高效# 查看某个会话里所有工具调用 jq -c select(.eventtool.call) | .payload logs/2025-06-12/s_0f8a.jsonl # 查看某个会话里所有异常 jq -c select(.eventtool.result and .payload.error ! null) | .payload logs/2025-06-12/s_0f8a.jsonl # 按事件类型统计 jq -r .event logs/2025-06-12/s_0f8a.jsonl | sort | uniq -c配合回放引擎的 trace 模式已经能覆盖大部分 Agent 排查场景。我很少再为线上 Agent 问题写一次性调试脚本了。6. 最后说几句经验这套 Harness 做完以后我自己收获最大的不是架构有多“漂亮”而是心智模型变了。插件化让每个横切能力都变成一个独立小任务权限校验、成本统计、敏感词过滤各管各的不再互相纠缠会话日志加回放让线上问题从“猜”变成“查”处理时间从小时级降到了分钟级。如果现在让我重做一遍我会把顺序倒过来先设计事件模型再写插件 API。事件模型是整个系统的骨架字段定得全后面的插件和回放都顺字段定漏了后面补字段就像补洞到处都是兼容逻辑。建议你也试试这个开发顺序——先想清楚日志里要留下哪些事实再动手写你的 Agent 循环。