最近一个多月我一直在折腾一件事把平时随手写的 Agent 脚本重构成一个能持续迭代、能回归测试、能回放排障的工程化框架。内部代号就叫 DeepSeek Harness底座用 DeepSeek 模型核心只干两件事——把 Agent 的能力全部插件化以及把每一轮会话完整记录并支持回放。这篇文章就是把这次的工程化解剖记录写下来围绕全插件化设计与可回放会话日志展开适合正在做 Agent 应用、却被“改一处代码就要重跑一遍流程”折磨的开发者参考。先说结论Agent 项目做到后面真正的瓶颈不是“模型不够聪明”而是工程化太脆弱。脚本能跑通 demo但撑不起持续迭代。我踩过不少坑之后把所有可变的环节抽象成插件把每一次执行变成可回放的事件流整个项目的维护成本才真正降下来。1. 先对齐概念Agent 框架为什么需要一层 Harness1.1 Agent 项目从原型到工程化之间的断层大多数人写 Agent 都是从一段脚本开始的读配置、调模型接口、写工具函数、循环处理。这套东西跑一个 Demo 没问题但一旦要加新工具、换记忆策略、调整推理前的提示词处理问题就来了——业务逻辑和模型调用全揉在一起改一个搜索工具的参数格式可能要连带改主循环、改日志、改测试。我见过不少 Agent 项目代码量两三千行但函数之间互相“背靠背”引用全局变量到处传递上下文。这种结构的典型症状是新增一个插件能力需要动五个文件回放一次线上问题只能靠肉眼翻终端输出想跑回归测试发现没有稳定的输入输出记录。这不是代码质量问题而是缺少一层隔离层。Harness 就是这层隔离层。它本身不是 Agent而是 Agent 的“测试床”和“黑匣子”。模型、工具、记忆、观测全部通过统一接口接入运行时的每一次交互都被结构化记录。这样 Agent 的业务逻辑不再是“一坨代码”而是一组可以独立替换、独立验证的组件。1.2 DeepSeek Harness 的定位它是 Agent 的陪跑测试场我在设计 DeepSeek Harness 时最初的想法很简单能不能让 Agent 像软件工程里的服务一样被启动、被调用、被观测、被回滚。后来逐步清晰成三个能力边界。第一统一执行环境。不管 Agent 内部用多少个模型调用、多少次工具调用Harness 提供一个标准化的 Runner 来协调这些动作。插件只关心自己的输入输出不关心全局调度。第二统一生命周期管理。每个插件有 init、start、stop、dispose 这些钩子和主程序的生命周期解耦热更新某个插件时不影响整体运行。第三统一可观测性。跑了多少轮、用了哪些工具、上下文在哪个节点发生变化全部以事件形式落盘之后可以按时间轴回放。这套定位直接影响了技术选型。我没有选择引入重量级的编排框架而是自己写了一个轻量内核——因为编排框架的抽象层级太高日志格式和插件的生命周期绑定得很死定制起来的成本反而更高。1.3 为什么底座选 DeepSeek 模型Harness 可以对接任何模型我实际用的是 DeepSeek。原因很简单兼容 OpenAI 风格的 Chat Completions 接口迁移成本低上下文窗口足够大适合做多轮工具调用的 Agent 场景推理模型在函数调用这类结构化输出上表现稳定并且开源权重可以私有部署对做 Harness 这种偏基础设施的项目来说不会被某个封闭平台的接口变化绑架。当然模型选型不是重点。重点是Harness 的设计必须与模型弱耦合。也就是把模型调用封装成标准的 LLMProvider 接口上层插件和日志系统完全不知道背后是 DeepSeek 还是别的模型。这一点在做可回放日志时尤其重要——因为回放的时候我们可能用同一个模型重新跑也可能用完全不同的模型做对比验证。2. 全插件化设计把 Agent 拆成可插拔的积木2.1 插件化的目标与边界做插件化最怕的是“为了插件而插件”搞出一堆抽象接口实际用途只是心理安慰。我给 DeepSeek Harness 定的插件化原则只有一句话凡是可能独立变化的能力都做成插件凡是变化概率低、且与主流程强耦合的部分留在内核。按这个标准我划出了四个扩展点ToolPlugin工具调用比如搜索、HTTP 请求、数据库查询、MemoryPlugin短期记忆窗口、长期向量记忆、StrategyPlugin推理循环策略、重试策略、停止条件、ProcessPlugin前处理和后处理比如敏感信息脱敏、输出格式化。这四个扩展点覆盖了 Agent 项目里 90% 的定制需求。边界划清楚之后好处非常明显。团队里有人要做新的搜索工具他只需要实现 ToolPlugin 接口并注册完全不需要理解 Runner 内部怎么调度有人要换记忆策略改的是 MemoryPlugin 实现别的地方一行不动。插件化本质上是在代码层面划出“高内聚、低耦合”的物理边界而不是花架子。2.2 插件接口设计四个核心抽象我定义插件接口时刻意保持了最小化。每个插件只需要实现四个生命周期方法init 做初始化、preprocess 处理进入模型前的数据、postprocess 处理模型返回后的结果、dispose 做资源清理。核心代码大致是这样# harness_core/plugin.py from typing import Any, Optional class Plugin: 所有插件的基类。子类只需要按需覆写对应方法。 name: str base async def init(self, ctx: dict[str, Any]) - None: 初始化资源比如加载模型配置、建立连接池。 async def preprocess( self, messages: list[dict[str, str]], ctx: dict[str, Any], ) - list[dict[str, str]]: 进入模型前对 messages 做处理可原样返回。 async def postprocess( self, result: dict[str, Any], ctx: dict[str, Any], ) - dict[str, Any]: 模型返回后对结果做处理可原样返回。 async def dispose(self) - None: 释放资源保证可重复初始化。ToolPlugin 和 MemoryPlugin 都是这个基类的语义化扩展。ToolPlugin 多了 execute 方法返回结构化输出MemoryPlugin 多了 recall 和 store 方法负责上下文记忆的读写。这样设计的好处是插件之间天然解耦每个插件只处理自己关心的那一段数据流Runner 拿到结果后按顺序组装不会出现插件 A 改了消息格式、插件 B 拿到手就崩溃的问题。2.3 插件注册与依赖管理解决谁在用谁的问题有了接口下一步就是注册和装配。我在 Harness 里实现的注册机制非常轻量本质上是一个字典加一个装饰器# harness_core/registry.py _PLUGIN_REGISTRY: dict[str, type[Plugin]] {} def register_plugin(name: str): def wrapper(cls: type[Plugin]) - type[Plugin]: cls.name name _PLUGIN_REGISTRY[name] cls return cls return wrapper def get_plugin_class(name: str) - type[Plugin]: if name not in _PLUGIN_REGISTRY: raise KeyError(fplugin not registered: {name}) return _PLUGIN_REGISTRY[name]注册之后插件之间的依赖通过配置文件和依赖注入容器解决而不是在代码里直接 new 对象。我的做法是每个插件的 init 方法接收一个 ctx 字典里面预置了 logger、config、共享的 model_client 等核心对象。插件如果需要依赖另一个插件在配置里声明依赖项启动顺序由容器根据拓扑排序自动决定。依赖注入看起来多了一层实际收益很大。它让插件之间无法“偷偷”互相引用所有依赖关系都显式可见。排查问题的时候不用在全局搜某个对象是谁在哪个文件里创建的看配置和启动日志就能定位。2.4 插件生命周期与热插拔思路一个 Agent 进程可能持续运行很久插件的生命周期管理必须严谨。我的实现是Runner 启动时按配置顺序加载插件先调 init再进入就绪状态每次任务运行时插件实例是复用还是新建由插件的类型决定——无状态插件直接复用单例有状态插件比如带本地缓存的可以声明为“每次任务重建”。这里有一个我踩过的坑如果所有插件都做成单例状态会在不同 session 之间串扰。比如短时记忆插件在 A 用户的会话里存了上下文结果 B 用户的请求也拿得到。后来我强制插件声明自己的 scope默认是 session也就是每个会话新建实例只有声明为 singleton 的插件才跨会话复用。热插拔则在实例层操作更新配置后旧实例执行 dispose新实例执行 init进程不重启。2.5 一个插件示例可回放的记忆清空开关说一个具体例子。做客服机器人时经常需要测试“多轮对话后用户重开话题”的场景。我在 Harness 里写了一个 MemoryResetPlugin它监听消息里的特殊指令一旦检测到“/reset”前缀就清空当前 session 的短时记忆然后返回一条系统提示给模型# plugins/memory_reset.py from harness_core.plugin import Plugin register_plugin(memory.reset) class MemoryResetPlugin(Plugin): async def preprocess(self, messages, ctx): if messages and messages[-1].get(content, ).startswith(/reset): ctx[memory].clear() messages[-1][content] 用户已重置上下文请重新开始对话 return messages这个插件只有十几行但解决了一个很实际的测试需求。因为它是插件所以可以在测试环境启用、生产环境关闭也可以在回放日志时单独打开用来验证“重置记忆后模型是否真的不记得之前的内容”。这就是插件化带来的好处——能力是一块积木放哪里、什么时候放由配置决定而不是改代码。3. 可回放会话日志Agent 调试的行车记录仪3.1 为什么普通日志不够用Agent 调试最大的痛点是“不可复现”。大模型输出有随机性外部工具返回有时效性多轮对话的上下文又有强依赖。普通日志只能记录“某时刻发生了什么”无法回答“为什么这时候模型会这么选”。我早期用 loguru 打日志确实能看到 token 用量、工具调用结果但排查问题时只能靠猜——因为日志没有保存“当时模型看到的完整 messages”没有保存“每个工具调用的中间状态”也没有保存“上下文在哪一步开始漂移”。回放会话日志就是为了解决这个问题它不仅仅是日志而是一个完整的状态机快照序列可以像倒带一样把一次会话的每一步还原出来。3.2 会话日志的数据设计事件流加状态快照我把日志设计成 JSONL 格式每一行是一个事件对象。事件类型分为六类session_start、model_request、model_response、tool_call、tool_result、session_end外加一个 error 事件。每个事件的核心字段如下session_id会话唯一 ID用于关联同一轮对话的所有事件。event_seq单调递增的事件序号用于重建时间顺序。event_type事件类型。parent_seq父事件序号比如 tool_result 的父事件是 tool_call。timestamp事件发生时间精确到毫秒。payload事件的具体内容比如请求的 messages、工具参数、模型返回文本。fingerprint当前上下文的哈希指纹用于快速对比状态差异。payload 不需要额外压缩因为工具调用的参数通常不太大真正要控制体积的是 messages 里的历史文本。我的做法是每条消息单独记录一个简化的 message_id如果消息内容特别长给它打一个内容哈希回放时按需加载原文。3.3 回放机制离线回放与半在线回放回放器加载日志后有两种工作模式。离线回放模式不真正调用模型只按照日志事件流重新“走一遍”Agent 的执行流程校验插件处理的幂等性、事件顺序的合法性、工具参数的格式是否匹配。这种模式适合做回归测试——代码改动了把历史日志全部离线跑一遍能快速发现破坏性变更。半在线回放模式则重新调用模型但固定了随机种子和温度参数输出会和线上略有差异但整体流程可比对。半在线回放主要用于验证“修改插件逻辑后模型在相同上下文下是否表现更好”对比维度包括工具调用次数、是否提前终止、关键节点输出是否一致。回放器的核心代码如下# harness_core/replayer.py import json class Replayer: def __init__(self, log_path, runner_builder): self.log_path log_path self.runner_builder runner_builder def load_events(self): events [] with open(self.log_path, r, encodingutf-8) as f: for line in f: if line.strip(): events.append(json.loads(line)) return events def replay_offline(self): events self.load_events() for ev in events: # 离线模式只校验事件结构和指纹不调用模型 if ev[event_type] session_start: print(freplay session {ev[session_id]} start) elif ev[event_type] tool_call: self._validate_tool_params(ev[payload]) def replay_online(self, dry_runFalse): events self.load_events() runner self.runner_builder() for ev in events: if ev[event_type] model_request: resp runner.complete(ev[payload][messages]) self._compare(ev[payload], resp)离线回放的价值在跑回归时体现得最明显。我有一次把所有插件从同步实现改成异步实现心里没底直接拉了三天的历史日志做离线回放发现有两个工具调用顺序异常改完再跑一遍才放心上线。3.4 上下文指纹与分支比对回放日志里最有价值的东西我认为是上下文指纹。它的计算方式很简单把当前会话的所有上下文状态做规范化 JSON 序列化然后取 SHA256 哈希的前 16 位。规范化很关键因为 JSON 的键顺序、空格差异都会影响哈希值。我的做法如下# harness_core/fingerprint.py import hashlib import json def fingerprint(state: dict) - str: canonical json.dumps( state, ensure_asciiFalse, sort_keysTrue, separators(,, :), ) return hashlib.sha256(canonical.encode(utf-8)).hexdigest()[:16]把指纹埋到每个事件的 payload 里之后回放时就能直接对比两条分支的差异轨迹。比如线上某次会话失败了我在日志里发现工具 A 调用前后的指纹都正常但工具 B 调用后的指纹发生变化说明问题出在 B 的返回内容上。通过指纹可以从数百条日志事件里快速定位到“从哪一步开始上下文发生漂移”不用把整个 messages 都打印出来逐字对比。3.5 日志的安全与体积控制会话日志里最容易踩的坑是脱敏不彻底。工具调用参数里经常携带用户手机号、API Key、内部 token如果原样写进日志回放时等于把敏感数据多复制了一份。我的规则很简单凡是注册为 secret 的字段在写入 payload 之前统一替换为***回放器遇到***时不尝试还原只在报告中标记“此字段已脱敏”。体积控制方面我做了三层。第一层是采样默认所有会话都记录但工具 result 的大文本内容可以做截断只保留前 2000 字符和内容哈希。第二层是滚动压缩按天拆分日志文件超过一周的老日志用 gzip 压缩存储。第三层是归档用于做长期回归测试的日志单独归档不走滚动删除。这套策略跑下来单日十万条事件大约 2GB 原始 JSON压缩后大概 300MB在可接受范围内。4. 实操搭建 DeepSeek Harness 的最小闭环4.1 环境准备与项目结构开始动手前先准备好环境。我用的 Python 3.10依赖只有两个核心包一个请求库httpx做模型 API 调用一个标准库 json 做序列化。没有引入大型框架因为内置功能足够。项目结构如下harness-demo/ ├── config/ │ └── agent.yaml ├── harness_core/ │ ├── __init__.py │ ├── plugin.py │ ├── registry.py │ ├── runner.py │ ├── logger.py │ ├── replayer.py │ └── fingerprint.py ├── plugins/ │ ├── tool_search.py │ └── memory_reset.py └── main.py4.2 核心执行器 Runner 的实现Runner 是整个 Harness 的心脏。它从配置里加载插件列表按依赖顺序实例化插件然后进入主循环。每一次用户输入Runner 先执行所有插件的 preprocess再调用模型模型返回后执行 postprocess如果有工具调用把结果追加到 messages 里循环往复直到终止条件满足。Runner 的关键代码# harness_core/runner.py import asyncio from typing import Any class AgentRunner: def __init__(self, model_client, plugins, logger): self.model_client model_client self.plugins plugins self.logger logger async def run(self, user_input: str) - dict[str, Any]: session_id self.logger.new_session() messages [{role: user, content: user_input}] for plugin in self.plugins: messages await plugin.preprocess(messages, {session_id: session_id}) step 0 while step self.max_steps: self.logger.log(model_request, { messages: messages, fingerprint: fingerprint({messages: messages}), }) response await self.model_client.complete(messages) self.logger.log(model_response, { content: response.content, tool_calls: response.tool_calls, }) if response.tool_calls: for call in response.tool_calls: self.logger.log(tool_call, call) result await self._execute_tool(call) self.logger.log(tool_result, result) messages.append({ role: tool, name: call[name], content: result[content], }) else: break step 1 self.logger.log(session_end, {final_result: response.content}) return {session_id: session_id, result: response.content}注意这里把session_id绑定在了 logger 上而不是 Runner 上。因为一个 Runner 可以服务多个会话但一个 logger 实例负责一个 session 的日志流。实际项目里我倾向于把 logger 也线程隔离。4.3 日志记录模块让每一步都留下痕迹日志模块的核心是 SessionLogger。它维护一个循环缓冲区批量写入磁盘避免每次事件都打开关闭文件。实现很简单但有一个关键细节事件写入前必须先做脱敏和规范化。# harness_core/logger.py import json import time import uuid SECRET_FIELDS {api_key, token, password, phone} def _sanitize(payload: dict) - dict: if isinstance(payload, dict): return { k: (*** if k in SECRET_FIELDS else _sanitize(v)) for k, v in payload.items() } if isinstance(payload, list): return [_sanitize(item) for item in payload] return payload class SessionLogger: def __init__(self, path: str): self.path path self._buffer [] self._seq 0 def new_session(self) - str: session_id uuid.uuid4().hex[:12] self.log(session_start, {session_id: session_id}) return session_id def log(self, event_type: str, payload: dict, parent_seq: int 0) - int: self._seq 1 event { session_id: self._session_id, event_seq: self._seq, parent_seq: parent_seq, event_type: event_type, timestamp: time.time(), payload: _sanitize(payload), } self._buffer.append(event) if len(self._buffer) 50: self.flush() return self._seq def flush(self): with open(self.path, a, encodingutf-8) as f: for ev in self._buffer: f.write(json.dumps(ev, ensure_asciiFalse) \n) self._buffer.clear()这里有个细节值得注意每次 log 之前把整条事件做一次脱敏成本很低但能从源头保证日志文件里绝对不会出现明文敏感字段。一旦日志已经落盘再来清洗就麻烦得多。4.4 回放器的对比策略到底比什么回放器不只要“重新跑一遍”更要能给出差异报告。我把对比分成三个层次。第一层是流程结构对比。比对事件序列是否一致包括事件类型顺序、工具调用次数、是否在相同位置触发错误。第二层是上下文指纹对比。比对每次 model_request 前后的 fingerprint定位上下文在哪一步开始偏离。第三层是结果语义对比。对模型输出做关键词级或向量级对比标记“虽然内容不同但语义等价”和“语义已经漂移”两种情况。语义对比我用了最简单的办法把两段文本分别做归一化去掉标点、转小写、用 jieba 分词后排序算一个基于 token 重叠的相似度分数。分数高于阈值就认为是语义等价。这个方案不完美但做回归测试足够因为回归测试关注的是“差异是否可接受”不是“差异是否为零”。4.5 用实际场景验证一个带搜索和记忆的客服机器人为了验证这套设计我做了一个很简化的客服机器人 Demo。它有两个插件搜索插件负责根据用户问题检索本地 FAQ记忆插件负责保存和召回最近 5 轮对话摘要。运行流程是用户提问、搜索插件介入、模型结合搜索结果生成答案。第一次运行时日志显示搜索插件调用的关键词是“退款”但实际 FAQ 分类用的是“售后”导致检索结果为空。模型没有拿到任何信息只能泛泛而谈。这个 bug 在普通日志里很容易被忽略因为跑了半天才发现搜索结果为空。但有了可回放日志我看到 tool_result 事件的 payload 里返回了一个空列表配合上下文指纹很快就判断出问题不在模型而在搜索插件的参数映射。修复搜索插件的映射逻辑后我把这次失败会话的日志用离线模式回放确认工具参数格式和事件顺序都正确然后再做半在线回放对比发现模型这次拿到了 FAQ 内容回答质量明显提升。整个过程不到半小时这在没有回放日志之前是不可想象的。5. 常见问题与避坑实录5.1 插件顺序依赖导致的不稳定插件化最大的坑是插件之间隐式的顺序依赖。比如记忆插件必须在搜索插件之前执行因为搜索插件要依赖记忆里的用户偏好来改写查询词。这种依赖如果只写在文档里不写在配置和代码里迟早出问题。我的解法是插件声明自己的依赖列表容器启动时按拓扑排序初始化。一旦有人调整配置导致循环依赖启动直接报错绝不“碰运气”运行。另外日志里的每个 preprocess 事件都记录当前插件名称回放时如果发现顺序不对一眼就能看到是哪一步先动了。5.2 回放时上下文指纹对不上指纹用于定位漂移节点但有时明明没改逻辑指纹却对不上。我排查后发现原因是模型返回的 tool_calls 里带了随机生成的 call_id我把它也塞进了状态上下文导致每次回放指纹都不同。解决方法是在做上下文规范化时剔除掉随机字段只保留影响语义的字段比如 role、content、tool_name。这个经验教训是指纹的稳定性取决于你画的“上下文边界”是不是只包含真正影响模型决策的数据。5.3 外部 API 不可重放的问题回放日志的目标是可复现但外部 API 做不到。某次排查时发现回放过程中调用了同一搜索服务返回结果和线上完全不一样导致后续所有对比都失去意义。解决方法是对强时效性的外部调用在日志里记录返回结果的快照半在线回放时默认使用快照结果只有显式指定“忽略快照”才重新调用外部 API。5.4 插件状态泄漏这个问题前面提过但值得再强调一次。在 Harness 初期我把所有插件都实现为单例跑自动化测试时经常出现“上一个 scenario 的记忆污染了下一个 scenario”。我后来的强制规则是插件必须声明 scope 为 session 或 singleton默认 session。这让测试隔离变得非常简单因为每次 Runner 构建新 session就会重新 new 一批插件实例。5.5 高并发下的日志写入性能Agent 服务一旦上了量日志写入很容易成为瓶颈。我一开始每条事件单独打开文件写入压测到每秒 200 条事件时I/O 排队明显。优化方案是批量写入加缓冲每攒 50 条或每 500ms flush 一次单条日志写入量下降了一个数量级。另外JSON 序列化确实有开销但相比网络请求和模型推理来说可以忽略不要为了省序列化时间牺牲事件的结构化程度。5.6 常见问题速查表现象原因解决方案回放事件顺序错乱session_id 未绑定到日志实例让 logger 持有当前 session_id事件中自动携带指纹频繁变化随机字段进入上下文规范化剔除 call_id、timestamp 等非语义字段外部 API 重放不一致未使用响应快照日志记录 tool_result 完整快照回放时复用插件间状态串扰单例插件跨 session 复用强制 session scope每次会话新建实例日志文件过大未做采样和归档长文本截断、按天滚动、gzip 归档敏感字段泄露未在写入前脱敏维护脱敏字段表log 函数统一过滤6. 写在最后的个人心得我在搭建 DeepSeek Harness 的过程中最大的体会是工程化的本质不是加功能而是降低“改动一处后需要验证全部”的成本。全插件化设计让组件边界清晰可回放会话日志让每一步都有据可查这两者叠加之后开发 Agent 的节奏从“小心翼翼改代码、满屏打日志”变成了“大胆改插件、跑一遍回放就知道有没有破坏行为”。如果你也想给自己的 Agent 项目做类似改造我的建议是不要一上来就追求完美抽象。先做两件事把工具和记忆从主循环里拆出去让它们走接口把每次模型调用和工具调用写成结构化事件日志。做完这两步再考虑插件注册、依赖注入、回放对比这些进阶能力。脚手架搭好后后面就是水到渠成的事。这个框架后续我还会继续扩展比如把回放结果接入自动化 CI每次提交代码自动跑回归让 Agent 应用也能享受传统软件工程的测试保障。