一个 Agent 项目跑通 Demo 其实不难难的是让它以工程形态稳定运转。我最近把一套基于 DeepSeek Harness 的实验性 Agent 项目重构了一遍整个改造过程最核心的两个抓手就是全插件化的框架设计以及可回放的会话日志。这两件事听起来不大但它们决定了这个项目能不能从“我本地能跑”进化到“别人也能维护、出问题能查、加需求不心惊胆战”。这篇文章就把这两个核心点掰开揉碎从设计思路、接口细节、落地实现到排查技巧都过一遍。如果你正在做 Agent 应用或者你已经在用一个单体脚本硬扛多个工具调用那这篇复盘应该能给你不少可以直接抄作业的参考。1. 为什么需要 Harness从脚本到工程的三个转折点1.1 脚本能跑但撑不住形态先说说我做重构之前的原始状态。最开始写 Agent思路和写普通脚本没区别一个主循环里面依次塞模型调用、工具调用、条件分支。比如用户输入来了先做意图判断再决定调哪个工具拿到工具结果后拼成上下文再请求一次模型。这个阶段代码量不大跑起来也没啥问题。真正出问题是在“业务开始复杂”之后。我那个项目后面接了三个外部工具做了两轮多模态输入处理还要支持不同客户在不同时间开关不同功能。于是主流程代码里开始出现大量if customer_a and tool_enabled and round 2这种判断。每次加一个功能都要把整个主流程从头到尾读一遍生怕改坏哪个分支。最头疼的是日志和审计需求外部工具调用需要记录参数、耗时、结果摘要模型请求需要记录 token 估算值和返回片段。这些东西如果散落在主流程的各个角落维护就是一场噩梦。这时候我意识到一个问题脚本关注的是“把事跑通”而工程关注的是“在跑通的基础上保证可扩展、可观测、可恢复”。这两个目标需要不同的架构支撑。1.2 Harness 到底在“勒住”什么“Harness”这个词在测试领域指的是“测试脚手架”意思是把被测对象放到一个可以稳定操控的架子上方便观察和驱动。在 Agent 框架里Harness 的含义类似它不是一个具体的业务 Agent而是承载 Agent 运行的那层底座。这个底座负责什么呢按我实际拆分的维度来说至少包括这些配置加载从文件、环境变量、远程配置中心读取参数并做校验。会话生命周期管理创建、持久化、销毁会话上下文。模型调用调度统一封装对 LLM 的请求包括重试、超时、token 控制。工具/插件执行环境管理工具注册、权限校验、执行超时和结果规整。事件分发把用户消息、模型输出、工具结果、异常等关键节点广播给注册的监听器。日志与可观测性结构化记录运行轨迹支持事后回放和分析。换句话说Harness 自己不做业务它只负责把“通用能力”做扎实。业务逻辑全部通过插件方式挂载上去。这样核心底座可以保持稳定而扩展需求不会破坏核心代码。1.3 全插件化和回放日志为什么是第一优先级如果让我给这个项目只选两个架构特性我会毫不犹豫选这两条。全插件化解决的是“扩展能力”的问题。没有插件化每加一个功能就要动核心流程代码风险面越来越大测试范围也越来越大。有了插件化核心流程变成一个稳定的事件分发骨架新功能就是新增一个插件模块互不干扰启停可控。可回放会话日志解决的是“可追责、可复现、可优化”的问题。Agent 的行为是链路式的用户输入 → 意图理解 → 工具调用 → 结果拼接 → 最终回复。任何一个环节出错最终表现都可能很离谱。如果没有完整的执行轨迹记录线上出了诡异问题基本只能靠猜。有了回放日志你可以把当时的执行时间线还原出来逐步查看每一步发生了什么。这两条是相辅相成的插件化决定了系统的扩展边界清晰回放日志决定了系统的可观测性足够深。这两件事做扎实了Agent 框架的工程化才站得住。2. 全插件化设计先把 if-else 请出核心代码2.1 插件化的本质是事件分发我做插件化设计的时候第一件事就是给核心流程减负。原来主流程里那些“前置检查”“后置处理”“分支扩展”全部抽象成事件钩子。什么叫事件钩子就是核心流程在某个关键节点会广播一个事件任何对这个事件感兴趣的插件都可以注册监听并在这个节点前后执行自己的逻辑。举个例子模型调用这个动作我在 Harness 里拆出了这些事件before_model_call模型请求发出之前可以在这里做敏感词过滤、会话历史截断、越狱防护检查。after_model_call模型返回之后可以在这里做输出格式校验、结果脱敏、质量打分。on_message_received用户新消息进入时可以在这里做日志记录、用户画像更新。on_tool_execution工具被调用时可以在这里做权限校验、审计记录。这样一来核心流程不需要知道谁在监听这些事件也不需要知道监听逻辑内部怎么实现。它只需要“广播事件等待结果”。核心代码变成了一个清晰的骨架而业务扩展全部落入插件模块。2.2 插件接口与注册机制接口设计是插件化的灵魂。我在设计插件基类时要求每个插件必须暴露两样东西元信息和生命周期方法。元信息包括插件名、版本、作者、依赖的其他插件、优先级。生命周期方法包括setup和teardown分别用于初始化和清理。然后对外暴露事件监听方法比如handle_message、on_model_call_start等。注册机制我用的是一个简单的注册表模式dataclass class PluginMeta: name: str version: str priority: int 100 dependencies: list[str] field(default_factorylist) class AgentPlugin(ABC): meta: PluginMeta def setup(self, ctx: AgentContext) - None: 插件初始化在框架启动时调用 pass def teardown(self, ctx: AgentContext) - None: 插件清理在框架关闭或插件卸载时调用 pass class PluginRegistry: _plugins: dict[str, AgentPlugin] {} classmethod def register(cls, plugin: AgentPlugin) - None: name plugin.meta.name if name in cls._plugins: raise PluginConflictError(f插件 {name} 已注册) cls._plugins[name] plugin classmethod def get_all_sorted(cls) - list[AgentPlugin]: return sorted(cls._plugins.values(), keylambda p: p.meta.priority)priority这个字段非常关键。插件之间可能对同一个事件都有监听需求这时执行顺序就很重要了。比如一个插件要做用户输入脱敏另一个插件要做敏感词拦截。如果脱敏先执行敏感词拦截看到的就已经是处理过的文本可能漏报。我用的排序规则是数字越小越先执行相同优先级按注册顺序。2.3 上下文对象插件之间的数据传递插件不能直接互相调用否则又会形成依赖耦合。它们之间交换数据的唯一通道是上下文对象AgentContext。我的AgentContext设计遵循两个原则核心状态只读比如session_id、user_id、messages历史列表插件只能读取不能直接修改。插件数据隔离每个插件可以在上下文里申请一个自己的命名空间只能读写自己命名空间下的键值对。这样做的原因很实际如果所有插件都可以随便往上下文里塞键值时间一长就没人知道某个字段是谁写入的、什么时候写入的、格式是什么。全链路可追踪性会瞬间崩塌。dataclass class AgentContext: session_id: str user_id: str messages: list[dict] metadata: dict field(default_factorydict) def __post_init__(self): self._plugin_data {} def plugin_get(self, plugin_name: str, key: str): return self._plugin_data.get(plugin_name, {}).get(key) def plugin_set(self, plugin_name: str, key: str, value) - None: self._plugin_data.setdefault(plugin_name, {})[key] value实际使用中这个隔离设计帮我避免了很多次“插件 A 改了字段让插件 B 崩了”的相互甩锅事件。出了问题直接看哪个插件写的数据非常简单。2.4 配置隔离与热更新每个插件的配置我要求必须独立。框架启动时会读取总的配置文件按插件名拆分配置块传递给对应插件的setup方法。插件不能偷偷读取全局配置里的敏感字段比如数据库密码、第三方密钥。配置更新的支持也是必须的。线上 Agent 运行中经常需要临时调整某个插件的参数比如调高敏感词拦截的严格程度而不需要重启服务。我在框架里加了一个配置重载事件on_config_reload。插件监听这个事件后可以从配置中心拉取最新配置并热更新。不过这里有个经验之谈热更新可以做但要限制范围。我遇到的最常见问题是某个插件热更新后内部状态和已有会话不一致导致后续处理异常。所以我的做法是配置类参数支持热更新但状态类数据比如内存里的缓存、统计计数不支持热更新必须重启才能重置。这个边界一开始就定清楚能省掉很多半夜被叫起来的麻烦。3. 可回放会话日志把 Agent 的执行轨迹变成时间线3.1 记录的不只是对话而是完整事件轨迹普通日志和回放日志最大的区别在于普通日志是给人看的回放日志是给程序读的。普通日志关注“打印了什么信息”回放日志关注“运行的完整状态和时间线”。我设计的会话日志是结构化事件流每条记录就是一个 JSON 对象按时间追加写入文件。核心事件类型包括message_received用户消息到达记录 session_id、时间戳、消息内容。before_model_call模型请求将要发出记录模型名、消息条数、token 估算值。after_model_call模型返回成功记录耗时、返回片段、结束原因。tool_execution某个工具被调用记录工具名、参数、结果摘要、耗时。error任意环节抛出异常记录异常类型、堆栈、触发事件。session_end会话结束记录累计 token 消耗、总耗时。实际写入的 JSON Lines 长这样{event: message_received, ts: 1710000000.123, session: s-001, role: user, content: 查询本周销售额} {event: before_model_call, ts: 1710000000.125, session: s-001, model: deepseek-chat, messages_len: 8, tokens_estimate: 2850} {event: after_model_call, ts: 1710000000.891, session: s-001, duration_ms: 766, response_preview: 好的我来查询本周销售数据, finish_reason: stop} {event: tool_execution, ts: 1710000000.892, session: s-001, tool: report_api, args: {range: week}, result_preview: 本周销售额 23500, duration_ms: 320}每一行都是独立 JSON方便逐行解析。你可能会问为什么不直接用传统的文本日志因为文本日志没法被程序高效解析。逐行 JSON 的好处是可以在不加载全量数据的情况下按时间、会话、事件类型做过滤也能增量同步到分析系统。3.2 存储与写入的工程细节日志写入这块有几个细节不处理就会在关键时刻掉链子。第一是追加写而非覆盖写。每个会话一个独立日志文件文件名包含会话 ID 和时间戳。这样单个会话的日志可以独立回放不互相干扰。第二是缓冲与刷盘策略。高并发场景下每条事件都立刻刷盘会成为瓶颈。我的做法是内存里攒一批事件达到一定条数或时间窗口后批量写入。同时设置一个兜底机制正常关闭时强制刷盘异常退出时恢复读取内存缓冲。第三是多进程/多线程安全。如果一个会话可能被多个 worker 处理日志写入就必须加锁或者通过独立日志进程统一写入。这个我踩过坑后面第五章会细说。第四是数据脱敏。会话日志里可能包含用户个人信息或密钥信息。写入前必须过一遍脱敏规则比如手机号中间四位打码、密钥哈希化。这和普通日志脱敏是同一套机制但回放日志因为内容更完整脱敏要求要更严格。3.3 回放机制模拟回放与重执行回放日志有了回放功能怎么设计我实践下来有两条路线各自适用不同场景。模拟回放把日志里记录的每个事件按时间重新“演”一遍。它不重新调用模型也不重新执行工具只是把当时记录的参数、结果、耗时按时间线加载出来。这种回放适合做 UI 界面调试、产品走查、向非技术同事还原问题场景。因为没有真实外部调用所以回放速度非常快且结果确定一致。重执行回放重新走一遍 Agent 主流程但把模型调用和工具调用替换成 Mock。Mock 的数据来源就是那次会话日志。这种方式适合做回归测试、插件升级验证。比如你改了一个插件的逻辑你可以拿历史会话做重执行回放对比新代码跑出来的结果和线上记录的结果是否一致。两种方式的核心区别用一句话总结就是模拟回放是“看录像”重执行回放是“重新演一遍但这回用了替身演员”。我在框架里同时支持两种模式具体用哪个由回放参数决定。3.4 回放工具的实际效果有了回放能力调试体验完全不同。以前排查线上问题只能看分散的文本日志凭脑补还原现场。现在直接开一个回放终端按时间轴逐步播放每一步显示当前事件的上下文状态。支持断点在某个事件前暂停检查当时的messages和ctx内容。支持 diff重执行回放完成后把新旧两版的事件轨迹逐行对比差异点高亮显示。这个工具模型后来成了我项目里排查问题的主入口。外部服务和模型服务的抖动、插件之间的顺序问题、上下文截断导致的历史遗忘都能在回放时间线上清晰看到。4. 动手实操搭一个最小可跑的插件化 Harness4.1 定义插件基类与注册表这部分我把核心代码抽出来给你一个最小可跑的骨架。为了聚焦架构我用一个假模型函数代替真实大模型调用你跑起来就能看到完整的插件调度和日志记录过程。# harness.py import json import time from dataclasses import dataclass, field from abc import ABC, abstractmethod dataclass class AgentContext: session_id: str messages: list[dict] field(default_factorylist) _plugin_data: dict field(default_factorydict, initFalse) def plugin_get(self, name: str, key: str): return self._plugin_data.get(name, {}).get(key) def plugin_set(self, name: str, key: str, value) - None: self._plugin_data.setdefault(name, {})[key] value class AgentPlugin(ABC): meta: dict def setup(self, ctx: AgentContext) - None: pass def before_model_call(self, ctx: AgentContext) - None: pass def after_model_call(self, ctx: AgentContext) - None: pass class PluginRegistry: _plugins: dict[str, AgentPlugin] {} classmethod def register(cls, plugin: AgentPlugin): cls._plugins[plugin.meta[name]] plugin classmethod def all_plugins(cls): return sorted(cls._plugins.values(), keylambda p: p.meta[priority])这个代码结构很直接AgentPlugin是插件基类PluginRegistry是注册表。每一步注释都很清楚你可以在此基础上扩展更多的事件钩子。4.2 事件分发与模型调用链注入接下来写核心调度器。它要做的事很简单接收用户消息更新上下文按顺序调用每个插件的before_model_call然后调用模型再按顺序调用after_model_call最后记录事件到日志。# harness.py class Harness: def __init__(self, session_id: str): self.session_id session_id self.ctx AgentContext(session_idsession_id) self.log_events [] for plugin in PluginRegistry.all_plugins(): plugin.setup(self.ctx) def _emit(self, event_type: str, **payload): event {event: event_type, ts: time.time(), session: self.session_id, **payload} self.log_events.append(event) print(json.dumps(event, ensure_asciiFalse)) def _fake_model(self, messages): # 模拟模型调用真实环境替换为 LLM API time.sleep(0.1) return 模拟回复内容 def run(self, user_input: str) - str: self.ctx.messages.append({role: user, content: user_input}) self._emit(message_received, roleuser, contentuser_input) for plugin in PluginRegistry.all_plugins(): plugin.before_model_call(self.ctx) self._emit(before_model_call, modelfake-model, messages_lenlen(self.ctx.messages)) response self._fake_model(self.ctx.messages) self._emit(after_model_call, duration_ms100, response_previewresponse[:20]) self.ctx.messages.append({role: assistant, content: response}) for plugin in PluginRegistry.all_plugins(): plugin.after_model_call(self.ctx) return response注意一个设计细节for plugin in ...这段使用了注册表的全量排序每个插件都会收到事件。不管插件内部做什么核心调度器只负责按序广播不做任何业务判断。4.3 插件示例参数统计与调用日志下面写两个最小插件一个统计用户消息长度一个记录模型调用延时。通过这两个示例你可以直观看到插件是如何在不改动核心代码的情况下叠加能力。# plugins.py from harness import AgentPlugin class StatsPlugin(AgentPlugin): meta {name: stats, version: 1.0, priority: 10} def before_model_call(self, ctx): total_len sum(len(m.get(content, )) for m in ctx.messages) ctx.plugin_set(self.meta[name], total_len, total_len) def after_model_call(self, ctx): total ctx.plugin_get(self.meta[name], total_len) print(f[stats] 当前对话累计字符数: {total}) class TimingPlugin(AgentPlugin): meta {name: timing, version: 1.0, priority: 20} def before_model_call(self, ctx): ctx.plugin_set(self.meta[name], t0, time.time()) def after_model_call(self, ctx): t0 ctx.plugin_get(self.meta[name], t0) print(f[timing] 模型调用总耗时: {(time.time() - t0) * 1000:.0f} ms)这两个插件的优先级不同StatsPlugin是 10TimingPlugin是 20。所以消息进来后先执行统计再执行计时。如果你想让计时先执行改一下priority就行。4.4 跑通完整流程并观察日志在主程序里注册插件、创建 Harness、跑一轮对话# main.py from harness import Harness, PluginRegistry from plugins import StatsPlugin, TimingPlugin PluginRegistry.register(StatsPlugin()) PluginRegistry.register(TimingPlugin()) harness Harness(session_ids-001) answer harness.run(请帮我分析这份报告) print(最终答案:, answer)运行后你会看到两件事一是控制台打印的结构化 JSON 事件流二是各插件输出的统计信息。这个骨架虽然简单但插件化调度和事件日志两条主线都已经跑通了。接下来你可以把_fake_model换成真实的 LLM API再按自己的业务加插件即可。5. 常见问题与排查实录插件和回放日志的六个坑5.1 问题速查表我把自己实际踩过、以及帮别人排查过的典型问题整理成了一张速查表遇到类似情况可以直接对着排查。问题现象可能原因排查路径插件加载了但不生效注册表里没有该插件或优先级被更高优先级插件“吃掉”先确认注册表中插件存在再打印插件排序列表检查优先级插件之间的数据读取为 None插件使用了错误的命名空间或者另一插件尚未执行检查plugin_set和plugin_get的插件名是否完全一致回放日志和线上结果不一致事件记录遗漏了某些状态比如上下文截断前的完整消息列表对比线上会话日志和回放时加载的上下文状态回放时模型调用报错日志中缺少模型参数信息比如 model 名、temperature确认before_model_call事件完整记录模型请求参数日志文件中出现残缺行多进程同时写同一文件没有加锁或没有走统一日志进程写入路径改为追加写并加进程锁或改为单日志进程收集日志查询越来越慢单文件日志量过大没有按会话分割或没有索引按会话 ID 和时间戳分文件存储查询时按文件目录索引5.2 场景一插件加载了却不生效这个问题我遇到过三次。最初的怀疑方向是代码逻辑有 bug但检查下来插件逻辑完全正常。后来发现原因出在注册时机插件在main.py里注册但 Harness 的初始化发生在__init__的for plugin in ...之前。如果注册顺序反了Harness 初始化时插件还没进注册表自然就不会被执行。后来我在注册表里加了一个检查函数run()开始前打印当前已注册插件列表和顺序。这个简单的打印动作让类似问题在 10 秒内就能定位。我的经验是日志里一定要加插件列表的启动快照否则插件体系一出问题排查就像无头苍蝇。5.3 场景二回放结果和线上表现不一致重执行回放的最大坑在于“事件记录不完整”。比如我早期只在日志里记录了模型返回的前 20 个字符没记录完整的返回内容。重执行回放时模型调用虽然是 Mock 的但后续插件需要读取模型完整输出做处理读到的却是截断字段。字段存在但内容缺失这种问题最隐蔽。解决办法是日志中记录“足以重建现场”的最小完整状态。对于模型返回完整内容如果太长可以不直接全量存而是存对象存储的路径或者压缩内容。核心原则是回放时需要读取的每一个数据都必须在日志里能找到来源。5.4 场景三回放日志的性能优化全量结构化日志如果写得过于频繁在高并发场景下会成为一个不可忽略的性能损耗。我的优化策略有三层分级采样核心事件工具调用、异常、模型完成全量记录高频低价值事件比如 token 估算更新按比例采样。批量写入事件先进入内存队列每 5 秒或每 500 条批量刷盘一次。独立日志进程业务进程只负责往内存队列放事件一个独立进程负责写入磁盘。这避免了多进程写文件的锁竞争。这套组合下来日志功能对主流程的耗时影响降到可以忽略的水平同时回放所需的完整性没有压缩太多。我个人的实际体会是今天很多 Agent 项目并不缺模型能力缺的是把整个运行过程“结构化地说清楚”的能力。Harness 这套东西的价值不在于它用了多先进的技术而在于它让一个本来混沌的 Agent 运行过程变得有边界、有轨迹、有据可查。你如果正在重构自己的 Agent 项目可以先不要急着堆功能把插件骨架和会话日志这两块打牢。后续每加一个能力都会觉得是在往一个稳定的架子上拼积木而不是往一锅粥里加料。