Agent框架工程化：DeepSeek Harness插件化与会话日志回放

做 Agent 框架的工程化改造半年多，我最大的体会是：让模型把话说对不难，难的是让整个 Agent 系统能被“看懂”。一开始我的项目也很朴素——一个 main.py，调模型接口，写点 prompt 模板，再包几个业务函数。代码少的时候没什么问题，但功能一多，改一个工具函数都可能牵连模型调用和会话状态的流转，线上出了问题只能对着控制台日志瞎猜。后来我决定把这一层彻底重做，整理出一套轻量级框架，就叫 DeepSeek Harness。它的设计目标只有两个：全部能力插件化、每次会话日志可回放。这篇把接口设计、实现细节、回放机制和踩过的坑一起聊清楚，给被 Agent 代码复杂度折腾过的开发者一条可以直接抄的工程化路线。

1. 为什么 Agent 应用需要一个独立的 Harness 层，而不是写进 main()

1.1 我在项目里见过的“伪分层”问题

很多 Agent 项目表面上有分层：一个 service 目录、一个 models 目录、一个 tools 目录，但实际调用关系还是乱的。常见的样子是：业务 controller 里直接调模型 SDK，模型返回后在同一函数里解析工具调用，再立刻执行工具，然后再带着工具结果去请求模型。如果我再往里面加会话历史、日志埋点、敏感词过滤、重试逻辑，这个函数的复杂度会迅速膨胀到几百行。

这种写法最大的问题不是代码丑，而是职责边界模糊。模型供应商切换的时候，你要改业务代码；想给某个工具加个观测指标，你要改业务代码；想调整 prompt 里注入系统指令的顺序，你还要改业务代码。也就是说，每一次非业务的需求变更，都在迫使你动业务代码。我做了两个版本之后就意识到，必须把“模型交互的编排过程”从业务代码里剥出去，单独形成一个层，也就是 Harness。

这个 Harness 层类似 Web 框架里的中间件链，它不关心你的业务指令是什么，也不关心模型内部怎么思考，它只负责管理一件事：一次会话从开始到结束，每个环节按什么顺序发生，每个环节能否被外部扩展干预，以及整个过程有没有被完整记录下来。

1.2 Harness 的职责边界：编排、挂载、持久化

我给 DeepSeek Harness 定义了四个核心职责。第一个是请求编排，包括并发控制、超时管理、自动重试、供应商路由。第二个是插件挂载，模型适配器、工具注册器、日志采集器、脱敏过滤器，都以插件形式挂到对应事件上。第三个是上下文管理，维护会话级和任务级的上下文状态，插件可以按需读写。第四个是会话持久化，把每一次事件流转写入可回放的结构化日志。

这四个职责有一个共同点：它们都不是业务逻辑，但几乎每个业务逻辑都会依赖它们。如果你把这些能力写在 main.py 里，那你的 main.py 就成了上帝对象；如果你把它们收敛到 Harness 里，业务代码只需要面向“事件”说话，剩下的交给框架。

为了让你感受这个区别，我举个例子。接入一个新的模型供应商，在普通项目里你可能要新写一个 client 类，然后在业务代码里判断当前用哪个 client。但在 Harness 里，你只需要新增一个模型适配插件，其他插件和业务模块通过事件拿到统一格式的输入输出，完全不知道底层换了家供应商。

1.3 为什么不直接用开源 SDK 的钩子机制

有人会问：现在很多模型 SDK 都有钩子函数，为什么还要自己包一层？我的回答是：SDK 的钩子，覆盖的是“单次模型请求”的生命周期，而 Agent 需要覆盖的是“整个会话”的生命周期。会话里有多轮模型调用、多次工具执行、上下文窗口管理，这些跨请求的状态不是 SDK 钩子能管好的。Harness 把会话而不是请求当作一等公民，日志回放和插件事件都围绕会话展开，这和 SDK 的定位有本质区别。

2. 插件协议与生命周期：把工具、模型、拦截器都装进同一个插槽

2.1 统一插件接口：不让“工具”和“插件”混为一谈

做插件化设计，我踩过的第一个坑就是概念混用。团队里常说的“工具”，通常指 Agent 能调用的外部能力，比如查天气、查数据库；而“插件”指的是扩展系统行为的模块，比如鉴权插件、日志插件。如果这两者不分开，你会发现插件系统里塞满了业务工具，扩展点很快会被污染。

我在 DeepSeek Harness 里做了明确区分：工具是运行时被模型调用的资源，插件是系统事件的订阅者。一个工具可以没有插件身份，直接注册到工具注册表里；而一个插件可以完全不暴露任何工具给模型，只负责监听事件并做一些横切操作。

插件统一实现一个协议类，Python 版本大概长这样：

# plugin_protocol.py from abc import ABC, abstractmethod from typing import Any, Dict, List class PluginMeta: name: str version: str provides: List[str] dependencies: List[str] events: List[str] class HarnessPlugin(ABC): @abstractmethod def meta(self) -> PluginMeta: """返回插件元信息，供框架做加载与依赖解析。""" @abstractmethod async def install(self, host: "PluginHost") -> None: """插件被加载时调用，在这里订阅事件、初始化资源。""" @abstractmethod async def uninstall(self, host: "PluginHost") -> None: """插件被卸载时调用，在这里释放资源、解除订阅。"""

为什么用协议而不是继承一个巨大的接口类？因为不同插件的能力差异很大。有的插件只想监听事件，有的插件需要暴露配置项，有的插件还要和其他插件通信。如果接口里写字数太长，大多数插件会有一堆空实现。用协议类只约束最关键的生命周期方法，其余能力通过宿主对象按需获取，插件之间保持松耦合。

2.2 manifest 声明与依赖解析：插件不是散装的

每个插件目录里除了代码，还会放一个 manifest 文件，声明自己是谁、依赖谁、关心哪些事件。这就避免了“插件之间互相 import”这种失控局面。框架启动时先扫描所有插件目录，读取 manifest，再做拓扑排序，保证依赖方先于被依赖方初始化。

{ "name": "session_store", "version": "1.2.0", "provides": ["session_store"], "dependencies": [], "events": ["session.start", "turn.end", "session.end"], "settings": { "retention_days": 30, "format": "jsonl" } }

依赖解析有个容易忽略的点：插件声明的依赖，最好基于“能力名”而不是“插件名”。比如两个插件都提供了session_store能力，那你只需要依赖session_store，框架会选择合适的提供者；如果你直接写依赖session_store_v1，换实现的时候就要改很多 manifest。这就是为什么 manifest 里我单独保留了provides字段，它承担了服务发现的作用。

事件监听采用发布订阅模型。插件在install时调用host.bus.subscribe(event_type, handler)，框架内部维护一张事件名到处理器列表的映射表。事件触发时按注册顺序调用，但每个 handler 的结果相互隔离，一个插件抛异常不会导致整个调用链中断，而是进入错误处理分支，同时记录一条plugin.error事件。这个设计在后面的排错中帮了大忙。

2.3 核心事件表：一次会话到底拆成了多少个阶段

为了让日志可回放、插件可插拔，我把一次 Agent 会话拆成了下面这些事件阶段：

刚开始我以为事件拆细会影响性能，但实测下来，事件分发走内存队列，单次分发开销在微秒级，和模型单次调用的耗时比几乎可以忽略。真正要注意的反而是：事件定义要稳定，不要今天叫model_chunk，明天叫model.stream.chunk。

3. 会话日志的事件溯源设计：回放能力从哪来

3.1 为什么只存聊天记录根本不够

很多项目记录会话就是一张表：用户说一句，助手回一句。这种快照式数据用来展示没问题，但用来排查问题非常痛苦。你只知道“那一次回答错了”，却不知道模型当时收到了哪些工具结果、流式输出在哪一步中断、token 用到了多少、是不是某个插件的超时把主流程拖垮了。

我在 Harness 里改用了事件溯源的思路：不记录结果，只记录“发生了什么”。每个事件都带有时间戳、序号、事件类型和载荷，回放的时候只需把事件序列重新过一遍，就能还原出当时完整的调用链。这和你行车记录仪只拍过程不拍结论是一个道理，事故发生时，过程才是最有价值的证据。

3.2 日志字段设计与 JSONL 落盘

事件日志格式我最终确定为 JSON Lines，每一行一个 JSON 对象，字段结构如下：

为什么用 JSONL 而不是一个大 JSON 文件？一是追加写入天然适配流式事件，二是即使文件写到一半崩溃，前面的内容也不会丢，三是可以用tail -f直接观察实时日志。我见过有人把所有事件攒在内存里，最后一次性 dump 成一个 JSON，结果应用一崩，内存里的整个会话日志烟消云散。

{"sid":"s_8f1c","trace_id":"t_2a09","seq":1,"ts":1735900000001,"event_type":"session.start","thread_id":"main","payload":{"user_id":"u_01","agent_version":"1.2.0"}} {"sid":"s_8f1c","trace_id":"t_2a09","seq":2,"ts":1735900000100,"event_type":"turn.start","thread_id":"main","payload":{"role":"user","content":"帮我查一下今天的天气"}}

seq字段必须按会话内单调递增生成，不能依赖全局计数器，否则多线程并发会话会互相干扰。实现时我用每个会话一个计数器，写入队列时给事件打上序号，这样即使磁盘上的行顺序因为异步写入有所抖动，回放时也能按seq还原准确顺序。

3.3 流式响应与工具调用怎么记录

模型输出是流式的，一次回答会拆成十几个甚至几十个 chunk。我记录模型输出时并不把这些 chunk 拼成一个完整字符串再存，而是每个 chunk 单独记一条model.stream.chunk事件，载荷里带当前这一小段的 delta。回放时需要完整答案，就把同一trace_id下所有 chunk 按seq拼接起来，顺便还能还原出“用户看到逐字输出”的体验，这对排查“输出到一半突然停了”的问题特别有用。

工具调用是另一类难点。工具入参和出参可能很大，比如一个 SQL 查询返回几千行结果，直接全部塞进日志会撑爆磁盘。我在工具执行前记录入参预览，执行后记录出参预览，同时记录一个内容哈希。需要完整出参时，可以通过哈希去工具侧对应的存储里取。下面是脱敏和预览处理的逻辑片段：

# log_preview.py def make_preview(value, limit=200): text = str(value) if len(text) <= limit: return text return text[:limit] + f"...<truncated, full_len={len(text)}>" def hash_payload(value): import hashlib return hashlib.sha256(str(value).encode("utf-8")).hexdigest()[:16]

另外，流式记录要格外小心中间状态。插件的pre.model.call如果往模型请求里注入了新的系统指令，日志里要记录一次prompt.injection事件，否则回放时你会看到“模型输出和预期不一致”，却找不到是哪个插件改了输入。

3.4 一个最小的回放引擎实现

回放引擎的核心逻辑并不复杂：读取事件文件，按seq排序，然后遍历事件重建状态。下面是一个最小实现：

# replay.py import json from pathlib import Path def replay_session(log_path: Path): events = [] with log_path.open("r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue events.append(json.loads(line)) events.sort(key=lambda e: (e["sid"], e["seq"])) collected_chunks = [] tools_called = [] for ev in events: etype = ev["event_type"] payload = ev.get("payload", {}) if etype == "model.stream.chunk": collected_chunks.append(payload.get("delta", "")) elif etype == "tool.after.exec": tools_called.append({ "tool": payload.get("tool_name"), "arg_preview": payload.get("arg_preview", "")[:80], "ok": payload.get("ok"), }) return { "sid": events[0]["sid"], "full_reply": "".join(collected_chunks), "tools_called": tools_called, "event_count": len(events), }

要注意回放和重跑是两回事。回放只是重演历史事件，不重新调用模型，也不重新执行工具；重跑则是拿着当时的输入再请求一次模型。两者在排查不同问题时各有价值：回放适合看“当时到底发生了什么”，重跑适合验证“现在的模型是不是修好了”。默认回放引擎只做离线重放，这样不产生新成本，也不会因为工具副作用造成二次影响。

4. 实测踩坑：循环依赖、脱敏、流式记录与回放失真

4.1 插件循环依赖：日志插件的经典困境

我第一次给系统加插件化的时候，设计了一个logger插件负责事件收集与落盘，又做了一个tool_monitor插件负责统计工具调用指标。问题来了：tool_monitor依赖logger来记录指标，而logger想订阅tool_monitor产出的中间事件，于是两者在 manifest 里互相声明依赖，启动时拓扑排序直接报错。

这个困境的根子在于我把“日志采集”和“日志落盘”揉成了一个插件。正确的拆法是把采集拆成独立插件metric_collector，它只管在内存里算指标；真正落盘的session_store只依赖事件总线，不反向依赖任何统计插件。能力拆分得越细，依赖环越少。遇到循环依赖别急着加“允许循环”的开关，优先检查是不是职责没拆干净。

4.2 插件升级与事件监听泄漏

插件系统上线之后，我遇到了第二个坑：热更新一个插件后，同一个事件被旧插件和新插件各处理了一次，导致消息重复发送。原因是我卸载插件时只删了提供的能力，没有解绑事件监听。

后来我把卸载流程设计成强制三段式：先取消事件订阅，再释放资源，最后把插件实例从注册表移除。并且每次加载插件前先查重，同一个name + version不允许重复安装。如果你打算做插件热更新，这条建议能帮你少走很多弯路。

4.3 会话日志脱敏：payload 里的 API Key 和用户隐私

日志记录了事件全量 payload，这带来一个副作用：模型请求参数里可能带着用户隐私，工具入参里可能出现密钥串。我在落盘前加了一道脱敏插件，对字段名和内容双重处理：

# sanitize.py import hashlib import re SENSITIVE_KEYS = {"api_key", "token", "secret", "password"} KEY_PATTERN = re.compile(r"sk-[A-Za-z0-9_-]{8,}") def sanitize(payload: dict) -> dict: cleaned = {} for k, v in payload.items(): if k in SENSITIVE_KEYS: cleaned[k] = mask_value(v) elif isinstance(v, str): cleaned[k] = KEY_PATTERN.sub( lambda m: "sk-***" + m.group(0)[-4:], v ) else: cleaned[k] = v return cleaned def mask_value(value) -> str: value = str(value) if len(value) <= 6: return "***" return value[:2] + "***" + value[-2:]

脱敏要在日志写入前执行，只脱敏展示和存储字段，不能改动传给模型和工具的真实内容。我在事件总线里为脱敏插件安排了单独的订阅优先级，让它比其他日志插件更早拿到事件，否则就可能出现“日志插件已经写了，脱敏插件还没运行”的竞态。

4.4 回放结果和当时不一致：非确定性的根源

有段时间我发现，同一个会话事件重放出来的渲染结果，和用户实际看到的不完全一样，尤其是工具结果展示部分。排查后确认，问题出在工具出参被截断预览了，重放时我只能看到预览，展示效果自然比当时差。这是我的预期之内，但在回放界面上我没有标注“此字段为截断预览”，导致误判。

更深层的非确定性问题来自模型本身。同样的输入，temperature 不为 0 时输出每次都可能不同。如果你把回放做成“重新请求模型来复原回答”，那么你得到的根本不是当时的回答，而是一份新的回答。所以我在回放引擎里坚持：承认模型输出的非确定性，回放只还原事件本身。这也是我强烈建议日志里记录模型参数的原因——模型名、版本、temperature、top_p 都记录在post.model.call的 payload 里，拿来做对比分析非常有用。

4.5 异步落盘：性能与完整性的取舍

刚开始日志是同步写的，一次事件写一次磁盘，高并发会话时性能立刻崩掉。后来改成异步批量写：事件先入内存队列，后台线程每 100ms 或每攒够 200 条刷一次盘。这样在程序正常退出、收到终止信号时能把队列里的日志 flush 干净；极端崩溃场景下会丢尾部少量事件，但对排查绝大多数问题来说，已经足够了。为了减少尾部丢失的影响，我在session.end事件里强制做一次同步 flush，保证每个会话的完整日志尽量落盘。

这里有一条经验：别为了让日志“绝对不丢”而牺牲主链路的稳定性。Agent 的核心价值是业务响应，日志是辅助手段。把磁盘抖动导致的超时传递到用户请求上，得不偿失。

5. 工程化红利：bad case 复现、模型热切换与多 Agent 演进

5.1 面对用户反馈，从“猜”变成“查”

上线这套体系之后，我最直观的感受是排查效率变了。以前收到“某次任务没走通”的反馈，我要先找到那台机器的日志、对齐时间点、猜测是哪一步出的问题。现在用户提供会话 ID，我在本地回放器里把事件流拉出来，直接看调用链在哪一步中断、哪个工具返回了异常、模型在哪一个 chunk 之后不再输出。一次 bad case 的定位时间从小时级降到了分钟级。

为了让这个流程更顺手，我还给回放器加了一个 HTML 导出功能，把边事件按时间线渲染成可读的对话卡片，每个工具调用之间插入分隔标记。技术排查时我直接看 JSON 事件流，给非技术同事看的时候就用 HTML 时间线，两边都照顾到了。

5.2 模型供应商切换只需要换一个插件

Harness 里模型适配器同样是一个插件。我日常主要接 DeepSeek 的接口，也测试其他几家供应商，切换时不需要改业务代码，只需要在配置里指定加载哪个 model provider 插件，事件总线上的其他插件毫无感知。更重要的是，切换前后系统自动把同样的用户输入跑两遍并生成对比日志，哪个模型回答更稳定、返回更快一目了然。

我把这种能力称为“模型无关性测试”。如果哪一天有新的模型出来，我只需要写一个适配插件，然后跑一遍已有的会话回放测试集，就能快速评估新模型在当前业务里的表现，而不是临时拉业务代码做手工 demo。

5.3 从单 Agent 到多 Agent 编排的扩展基础

这套插件与事件日志模型，天然适合向多 Agent 方向演进。多个子 Agent 之间的握手，无非是系统内部多了一类事件：agent.delegate.start、agent.message.received、agent.delegate.end。我在 Harness 的事件表里预留了这类扩展点，会话日志里同样能跨 Agent 串联。子任务的执行结果被记录为一个工具调用事件，回放时你可以清楚看到哪个 Agent 在什么时候把任务委托给了谁、结果如何。

我做深一层之后发现，事件溯源对 Agent 系统的价值会越来越大。Agent 越复杂，它执行路径的不确定性就越强。传统的“只记对话、不记过程”思路，在单模型时代勉强够用，到了工具调用、多轮编排、多 Agent 协作的阶段，已经完全跟不上调试需求。

5.4 给同样在做 Agent 工程的开发者几句实在话

这套设计里最值钱的不是某个具体的类或函数，而是“把事件当作会话的一等公民”这个视角。插件化解决的是扩展问题，可回放日志解决的是可观测问题，两者结合起来，才让一个 Agent 项目从“能跑”走向“可控”。

如果只留一条设计原则，我会说：日志永远记录事件，而不是记录结果。结果会过时，会丢失上下文，会被截断，而事件是事实本身。哪怕你的项目现在只有几十行代码，我也建议从今天开始，把每一次模型调用、工具执行、会话切换都用事件的方式记录成结构化日志。养成这个习惯之后，再回头看那些让 Agent 变得不可维护的痛点，你会发现自己已经领先了一个身位。