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 会话拆成了下面这些事件阶段事件名触发时机典型订阅者session.start会话建立初始化上下文、创建日志文件turn.start用户消息进入记录输入、计算延迟pre.model.call模型调用前注入 prompt、做鉴权与限流model.stream.chunk模型流式返回片段增量渲染、增量落盘post.model.call模型调用完成统计 token、更新上下文tool.before.exec工具执行前入参校验、配额检查tool.after.exec工具执行完成记录出参、计算耗时turn.end单轮交互结束归档短期上下文session.end会话关闭汇总指标、轮转日志文件刚开始我以为事件拆细会影响性能但实测下来事件分发走内存队列单次分发开销在微秒级和模型单次调用的耗时比几乎可以忽略。真正要注意的反而是事件定义要稳定不要今天叫model_chunk明天叫model.stream.chunk。3. 会话日志的事件溯源设计回放能力从哪来3.1 为什么只存聊天记录根本不够很多项目记录会话就是一张表用户说一句助手回一句。这种快照式数据用来展示没问题但用来排查问题非常痛苦。你只知道“那一次回答错了”却不知道模型当时收到了哪些工具结果、流式输出在哪一步中断、token 用到了多少、是不是某个插件的超时把主流程拖垮了。我在 Harness 里改用了事件溯源的思路不记录结果只记录“发生了什么”。每个事件都带有时间戳、序号、事件类型和载荷回放的时候只需把事件序列重新过一遍就能还原出当时完整的调用链。这和你行车记录仪只拍过程不拍结论是一个道理事故发生时过程才是最有价值的证据。3.2 日志字段设计与 JSONL 落盘事件日志格式我最终确定为 JSON Lines每一行一个 JSON 对象字段结构如下字段含义示例sid会话 IDs_8f1ctrace_id链路追踪 IDt_2a09seq会话内事件序号1, 2, 3...ts毫秒时间戳1735900000001event_type事件类型model.stream.chunkthread_id线程/协程 IDmainpayload事件载荷用户消息、模型增量等为什么用 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, limit200): 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, encodingutf-8) as f: for line in f: line line.strip() if not line: continue events.append(json.loads(line)) events.sort(keylambda 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(rsk-[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 变得不可维护的痛点你会发现自己已经领先了一个身位。