DeepSeek Harness：全插件化Agent运行时与可回放会话日志实战

做了几年Agent应用，说实话踩过的坑比写出来的代码还多。最近我把自己常用的那套Agent运行骨架重新整理了一遍，抽出了一个内部代号叫 DeepSeek Harness 的轻量级运行时，核心就两件事：全插件化的装配机制，以及一套可回放的会话日志体系。这文章不聊概念，直接把这套框架从设计到落地的完整思路拆开来讲，包括接口怎么定、插件怎么挂、日志怎么录、回放怎么做到位，以及那些只有真正写过才知道的坑。

如果你近期在折腾Agent框架、想把写死的demo工程化、或者被线上Agent的"黑盒行为"搞到头大，这篇文章会比较对胃口。不需要你已经有完整的框架经验，只要写过几个Agent调用就能看懂；如果你已经在做框架设计，里面关于日志回放和插件装配的细节，也值得参考。

1. 为什么要折腾一个Harness层

先说背景。之前做Agent项目，最痛苦的不是模型效果不行，而是代码结构永远处于"能跑但没法维护"的状态。每个Agent脚本里，模型调用、工具函数、提示词、日志输出全混在一起。今天加一个工具要动主流程，明天调一个参数要全局搜索，后天线上出问题只能靠print日志猜。这种状态做demo没问题，一旦要上真实业务，就是灾难现场。

Harness这个词来自硬件测试领域的"测试夹具"：被测设备要接电源、接信号线、接负载，你不能每次手工去搭线，于是做一块转接板，所有连接关系都定义好。Agent的Harness也是这个思路——把模型接口、工具、记忆、日志这些"接线"统一收敛到一个运行容器里，业务代码只关心自己的逻辑，剩下的交给框架。

我最初是想直接找一个现成框架用的，但评估了一圈，发现大部分重型框架的学习成本和定制成本都偏高：光理解它的抽象概念就要花一周，真要改它内部的调度逻辑，又得重新读源码。我需要的是一个足够薄、足够透明、能完全被我掌控的骨架，DeepSeek Harness 就是在这样的诉求下长出来的。

这个Harness层解决的问题，概括起来是三件事：

• 统一接入：不管底层接的是哪一路模型服务、哪种工具协议，在框架内部都是同一个Provider接口，替换实现不碰业务代码。
• 生命周期管理：工具和记忆模块的初始化、启用、停用、销毁，由Harness统一调度，而不是散落在各处。
• 全链路可观测：每一次Agent运行都被记录成结构化事件流，事后可以按帧回放，定位问题不再是"猜"而是"看"。

后面所有内容，都是围绕这三件事展开。

2. 全插件化设计：从接口定义到动态装配

2.1 先定好四类Provider接口

插件化的前提是抽象出稳定的接口。我反复调整过好几版，最后保留了四个最核心的Provider类型：

ModelProvider封装模型服务的调用差异。不管是DeepSeek的聊天补全接口、还是本地部署的推理服务，在框架内部都只有chat(messages, tools, **kwargs)这一个入口。这样做的好处很直接：切换模型供应商，只需要新增一个ModelProvider实现，Agent主流程一个字符都不用改。

ToolProvider封装工具的执行逻辑。工具在Agent里的形态很特殊：既要有给模型看的描述（name/description/parameters），又要有真实执行的代码。ToolProvider把这两部分绑在一起，注册进来就能被模型自动发现和调用。

MemoryProvider封装记忆的存取。短期记忆放会话上下文里，长期记忆要不要写入向量库、要存哪些字段，都交给MemoryProvider。框架本身不感知存储细节。

LoggerProvider是后加的，也是最关键的。刚开始我用普通日志框架凑合，后来发现要支持回放，日志和事件完全是两回事。LoggerProvider专门负责把Agent运行过程中的关键节点记录为结构化事件，这部分在第三章详细说。

每个Provider的生命周期是固定五步：register（注册元信息）、init（初始化资源）、start（开始服务）、stop（停止服务）、dispose（释放资源）。有了这套生命周期，Harness才能在启动时按依赖顺序装配、退出时按逆序清理，避免资源泄漏和悬挂引用。

2.2 插件加载机制：Manifest加入口点

接口定义好之后，接下来是"怎么把插件装上去"。我的方案是：插件目录下的manifest.json声明插件基础信息，Python包的入口点负责告诉Harness"这个插件的实现类在哪里"。

manifest.json承担描述性职责，内容大致是这样：

{ "name": "weather_tool", "version": "1.0.0", "provider_type": "tool", "entry_point": "weather_tool.plugin:WeatherToolProvider", "dependencies": ["http_client"], "config_schema": { "api_key": {"type": "string", "required": true} } }

entry_point指向具体的类，dependencies声明依赖的其他插件，config_schema做配置校验。Harness启动时扫描插件目录，解析Manifest，按依赖关系拓扑排序，然后逐个实例化。

我特意没有用动态import的裸实现，而是要求每个插件类都继承基类并实现抽象方法。这样做的原因很务实：基类能直接提供通用的参数校验、异常包装、事件埋点能力，插件作者只需要填核心逻辑，出错时框架也能统一拦截，不至于让异常裸奔到上层。

2.3 为什么不用写死方案

有人可能会问：工具就那几个，直接在代码里if/else不就行了？我的回答是：如果你只有两个工具、一个模型、跑完就扔，怎么写都行。但Agent项目一定会经历工具数量增长、模型替换、团队协作这几个阶段，到那时候，写死的代码就是技术债的重灾区。

插件化带来的三个实际收益，我在项目里体会很深：

• 并行开发：工具A和工具B完全独立，两个人都可以同时开发、各自测试，最后注册进同一个Harness。
• 故障隔离：某个工具插件崩了，Harness可以把它标记为不可用，其他插件照常服务，而不是整个Agent挂掉。
• 按需裁剪：不同场景加载不同插件集。比如对话机器人不需要代码执行工具，数据分析场景不需要天气服务，通过配置就能组合出不同的Agent实例。

这些都是"写死方案"很难给的灵活性。

3. 可回放会话日志：让每次运行都变成一部现场录像

3.1 日志与事件：两件不同的事

先说一个最容易踩的坑：普通日志和可回放日志，看起来都是"记录东西"，设计思路完全不同。普通日志服务的对象是人，核心诉求是可读性，所以用print、logger.info就够了。可回放日志服务的对象是"状态重建"，核心诉求是完整性和确定性，它记录的不是几句话，而是一系列能还原现场的事件。

DeepSeek Harness 的会话日志设计直接采用了事件溯源（Event Sourcing）的思路，整套系统围绕一个核心原则：所有Agent状态变更都必须通过事件来表达，事件一旦追加就不能修改。这样回放就变简单了，把事件从第一条按顺序执行到最后一条，系统的最终状态自然被还原。

事件的数据结构是JSON Lines格式，每行一个事件，最核心的字段是这样：

{ "event_id": "evt_01JQZ1K3Z8V9F2J5H6Y0X2K4N7", "session_id": "sess_01JQZ1H7NP4B2W9L8T5F6R0D3E1", "seq": 42, "timestamp": "2025-02-18T14:23:05.781Z", "actor": "model", "type": "tool_call", "payload": { "tool_name": "weather_tool", "arguments": {"city": "杭州"}, "raw_output": {"temperature": 22, "condition": "多云"} }, "trace_id": "trc_01JQZ1H7NP4B2W9L8T5F6R0D3E1" }

我特意强调seq（序号），它是单会话内递增、且写死的、不随重放而变化的。timestamp记录的是"事件发生时"的墙上时间，而不是"事件被写入时"的时间，这两个时间在异步场景下会不一致，如果混用，回放时的事件顺序会被插入时间干扰。

3.2 事件颗粒度怎么定

日志设计里，颗粒度是最微妙的问题。事件太细，日志量几倍暴涨，而且大部分记录的是无关紧要的中间噪声；事件太粗，回放时看不到关键决策过程，等于白录。

我自己在项目里摸索出的颗粒度标准是：一个事件必须能够回答"一次状态转换的发生原因、输入、输出、决策依据"。基于这个标准，Harness默认记录了五类事件：

model_request事件特别说明一下。它记录的是未截断的完整请求体，这会让日志体积变大，但它是回放时还原"模型到底看到了什么"的唯一途径。我调试过程中很多次发现，模型的"幻觉输出"其实是因为上下文被前面某轮工具结果污染了，如果没有完整请求记录，这种问题根本无从查起。

3.3 录制端：AOP和解耦

录制端在Harness内部是借助"中间件"机制实现的。每个Agent动作（模型调用、工具执行）在执行前后都会经过中间件管道，LoggerProvider 注册一个中间件，自动在动作发生时发射事件。业务代码和插件代码完全无感知，不需要手动埋点上报。

这里我踩过一个大坑：最初录制工具结果时，直接记录了工具函数的返回对象。听起来没问题，但实际上某些工具返回的对象里包含了不可序列化的类型（比如文件句柄、数据库游标、复杂的对象图），日志文件直接写崩。后来我强制要求每个ToolProvider提供serialize_result方法，把工具输出转换为纯JSON结构后再交给LoggerProvider。可以序列化，是事件能入库的底线。

3.4 回放端：确定性重放

录制只是第一步，真正复杂的是回放。

回放会话的代码不复杂，核心就是按顺序从事件日志中读取事件，然后喂给一个"模拟器"：

def replay_session(session_id, event_stream): harness = create_harness(snapshot_events(event_stream)) # 回放模式下，Harness不再发起真实外部调用， # 而是从事件流中直接读取当时的输出。 for event in event_stream: if event.type == "model_response": harness.feed_model_result(event.payload) elif event.type == "tool_call": harness.feed_tool_result(event.payload["raw_output"]) elif event.type == "user_message": harness.feed_user_input(event.payload["original_input"])

难点在于"确定性"这三个字。Agent运行天然含有不确定性：模型采样结果不稳定、外部API响应有时间差、并发工具调用顺序不固定。回放时如果这些不确定因素原样重现，你会发现重放和原始运行对不上，状态偏移了，后面全乱。

我的定位方式是：回放不追求逐token一致，只追求"关键状态转换一致"。具体做法有三个：

• 模型调用的回放，直接使用事件里记录的raw_output，不重新请求模型。
• 工具调用的回放，直接使用事件里的tool_output，不重新执行工具。
• 系统时间获取会被替换为固定时间源，随机数生成器用事件流里的种子初始化。

这样回放的是"决策路径"，而不是"实现细节"。对于调试、审计、复盘场景，这完全够用；而如果哪次你发现决策路径都对不上，那恰恰说明录制的日志有缺失，是定位问题的好线索。

3.5 日志查询与衍生价值

当所有会话都按结构化事件落盘之后，它的价值远超"回放"本身。我后来基于这些日志，顺带做了几个很实用的能力：

• Token成本追溯：每个会话的模型调用事件带着token用量，统计每个用户、每个功能消耗的额度，不用再靠估算。
• 失败模式聚类：把所有包含异常的会话日志拉出来，按异常类型聚合，就能知道当前系统最脆弱的环节是什么。
• 评测集构建：从正常会话中抽取出user_message -> model_response -> tool_call -> model_response的完整链路，直接构造出评测样本，用来回归测试新版本。

可以说，可回放日志是一笔"越用越赚"的基础设施投资。

4. 实操过程：从零搭起一个最小可运行的Harness

4.1 项目骨架

这一节把前面抽象的设计落到具体代码上，给一个最小可运行的参考实现。

目录结构：

harness/ ├── core/ │ ├── __init__.py │ ├── plugin.py # Provider基类与生命周期 │ ├── manifest.py # Manifest解析与依赖排序 │ ├── recorder.py # 事件录制器 │ ├── middleware.py # 中间件管道 │ └── harness.py # Harness主容器 ├── plugins/ │ ├── deepseek_model/ # 模型Provider插件 │ ├── weather_tool/ # 工具Provider插件 │ └── jsonl_logger/ # 日志Provider插件 └── main.py

Plugin基类我用抽象基类来保证接口约束：

from abc import ABC, abstractmethod from typing import Any class Provider(ABC): name: str = "" def __init__(self, config: dict[str, Any] | None = None): self.config = config or {} @abstractmethod def init(self, context: "HarnessContext") -> None: """初始化和预加载资源。""" @abstractmethod def dispose(self) -> None: """释放所有资源。""" def to_plugin_report(self) -> dict: return {"name": self.name, "status": "unknown"}

ToolProvider和ModelProvider都继承Provider，并扩展各自的方法。比如ToolProvider要额外实现get_tool_schema（返回给模型的工具定义）和execute（实际执行工具逻辑）。

4.2 ModelProvider插件的实现示例

以对接聊天补全接口为例，ModelProvider的最小实现是：

import requests from harness.core.plugin import Provider class DeepSeekModelProvider(Provider): name = "deepseek_chat" def init(self, context): # 这里只做配置校验，不直接建连接。 self.api_base = self.config["api_base"] self.api_key = self.config["api_key"] self.model_name = self.config["model_name"] self.timeout = self.config.get("timeout", 60) def chat(self, messages, tools=None, **kwargs): payload = { "model": self.model_name, "messages": messages, "tools": tools, "temperature": kwargs.get("temperature", 0.2), "stream": False, } resp = requests.post( f"{self.api_base}/chat/completions", json=payload, headers={"Authorization": f"Bearer {self.api_key}"}, timeout=self.timeout, ) resp.raise_for_status() data = resp.json() return data def dispose(self): # 这里实现了连接池管理 pass

这里有个经验：init里只做轻量校验，不要把鉴权请求、模型探测这些动作放到init里。Agent启动是高频操作，如果每次启动都等一次网络往返，整个框架的响应性会拉垮。把真正需要网络的动作推迟到首次chat时再触发，是最实用的优化。

4.3 ToolProvider插件的编写规范

以天气工具为例，写一个完整插件：

import json import requests from harness.core.plugin import Provider class WeatherToolProvider(Provider): name = "weather_tool" def init(self, context): self.api_key = self.config.get("api_key", "") def get_tool_schema(self): return { "type": "function", "function": { "name": self.name, "description": "查询指定城市当前的实时天气。", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名，例如：杭州"} }, "required": ["city"], }, }, } def execute(self, arguments: dict) -> dict: city = arguments["city"] # 模拟真实查询 result = {"temperature": 22, "condition": "多云", "city": city} return result def serialize_result(self, raw_result: dict) -> dict: # 这是回放日志能落盘的关键 return raw_result def dispose(self): pass

get_tool_schema和execute是一对：前者是"模型看到的说明书"，后者是"实际执行的黑盒"。注意序列化问题的处理，我在前面的章节就强调过，这里用接口约束来解决。

4.4 Harness主容器的装配流程

Harness主容器负责把Provider连接起来，并启动中间件管道：

from harness.core.harness import Harness from plugins.deepseek_model.provider import DeepSeekModelProvider from plugins.weather_tool.provider import WeatherToolProvider from plugins.jsonl_logger.provider import JsonlLoggerProvider def build_harness(): harness = Harness( model_provider=DeepSeekModelProvider(config={...}), tool_providers=[WeatherToolProvider(config={"api_key": ""})], logger_provider=JsonlLoggerProvider(config={"log_dir": "./sessions"}), ) harness.init() return harness harness = build_harness() result = harness.run(user_input="杭州今天天气怎么样？")

这个run方法是整个框架的心脏：它负责把用户消息、系统提示词、工具Schema组装成模型请求；拿到模型响应后判断是否需要调用工具；需要则执行工具再回灌给模型；不需要则直接返回最终回复。每一环都被LoggerProvider记录为事件。

框架的核心调度逻辑用自然语言描述就是这样：

收到用户输入，追加到会话上下文；调用ModelProvider获取模型输出；如果模型输出包含工具调用请求，则从ToolProvider池中选出对应工具并执行；工具结果追加到上下文；再次调用模型，直到模型返回最终回复或达到最大轮次。

4.5 一次完整的录制与回放

跑完上面的示例，会在./sessions/{session_id}.jsonl生成一个事件文件。你可以用任意JSONL工具打开它，逐行查看这次运行的完整决策路径。

回放时，我写过一段很简单的CLI工具，逐行读取这个文件，输出可读的对话摘要：

python cli.py replay --session sessions/sess_01JQZ1H7NP4B2W9L8T5F6R0D3E1.jsonl

输出大致长这样：

[1] user_message: 杭州今天天气怎么样？ [2] model_request: 调用工具 weather_tool(city="杭州") [3] tool_call: 返回 {"temperature": 22, "condition": "多云"} [4] model_response: 杭州今天多云，气温22摄氏度，适合出门。

调试线上问题的时候，拿这个摘要和实际运行对比，很快就能定位是模型理解错了、工具返回错了，还是提示词写歪了。

5. 踩坑记录与排查技巧实录

5.1 回放不一致：最隐蔽的坑是时间戳

第一次做回放时，我遇到一个诡异的问题：重放到第20轮，状态和原始记录就分叉了。查了很久，最后发现是插件里直接使用了time.time()作为缓存键的前缀。重放环境下，组件的初始化时间是重放的当前时间，缓存键变了，缓存全部失效，逻辑就走了一条和原始运行完全不同的路。

解决办法是给Harness注入一个Clock对象。所有组件获取时间必须通过context.clock.now()，而不是直接调time.time()。重放模式下，Clock返回的事件流里的历史时间；正常运行模式下，Clock返回真实的当前时间。这个改动花了我半天，但彻底根治了整个类别的bug。

5.2 插件加载顺序：依赖排序不能偷懒

一开始插件注册是简单列表顺序，先后注册谁就看谁先出现在配置里。结果有个工具插件需要在ModelProvider准备好之后才能初始化（因为它要读取模型上下文长度限制），配置顺序写反直接启动失败。后来我实现了Manifest的依赖拓扑排序，dependencies里声明的先加载，同级的按priority字段排序。从那以后，插件加载顺序对使用者透明，配置想怎么写就怎么写。

5.3 异步I/O下的事件乱序

Agent调用模型是串行的，但工具执行有时候我会改成异步并行（批量工具调用），这就带来了乱序问题：工具B先执行完、先返回，但事件流里它在A前面；模型下一轮看到的上下文顺序是对的吗？——不对，如果直接按返回顺序拼接，上下文顺序就会错乱。

我的修复方案是给每个工具执行分配一个request_id，事件流里记录这个ID，同时记录request_started_at和request_finished_at。写入时允许乱序，但读取时按request_id重新组装，确保进入模型上下文的始终是"发起顺序"而不是"完成顺序"。

5.4 日志文件无限膨胀

事务日志还好，它是按会话拆分的，单文件大小可控。真正膨胀的是会话内的model_request事件：每轮都记录完整上下文，10轮对话就能膨胀到几十KB甚至上百KB。后来我加了"增量记录"选项：第一轮记录完整上下文，后续轮次只记录差值（新增的消息和上一次请求的消息），回放时再用还原算法拼回完整快照。进度收益，日志体积直接降了一个数量级。

5.5 模型输出的不确定性会让回归测试失效

给Harness写自动化测试时，我遇到了经典问题：同一份测试用例，模型输出的结果每次都不一样，断言没法写死。网上很多方案是mock掉模型层，但mock掉的模型输出和真实场景差距太大，测试价值有限。

我的折中方案：测试分为两层。第一层是确定性测试，把模型输出替换为事件库中的历史响应（就像回放那样），专门测调度逻辑和工具调用逻辑；第二层是真实模型测试，用固定temperature=0和相同的输入跑几次，用相似度断言而不是原文断言。两层各有分工，确定性测试保证框架逻辑不出错，真实测试保证模型链路通畅。

5.6 插件卸载与热更新

项目上线后，插件更新是不可避免的。最开始更新工具代码只能重启整个服务，代价很大。后来我在Harness里加了热卸载逻辑：先标记插件为停止接收新请求，等当前执行中的任务完成后再卸载，最后加载新版本插件。这个"优雅过渡"机制代码量不大，但稳定性收益非常高。关键点是dispose不能粗暴切断正在运行的执行，否则会造成上下文不一致，连带污染后续所有事件的记录。

6. 如果再做一次，我会怎么选型

Harness做到现在这个程度，回头看很多设计决策其实有更省力的路径。如果要给后来者一个选型清单，我的排序大概是：

• 从零写还是用现成框架：如果只是做一个工具调用的demo，用现成框架没问题。如果你像我一样需要深度控制消息组装逻辑、需要专门设计会话日志、需要插件体系，那自己维护一个薄Harness是划算的。它不复杂，核心代码量只有一两千行。
• 日志方案选现成还是自研：如果有现成的可观测基础设施（比如已经接了链路追踪系统），可以先把事件导出到那边，回放功能后续再加。如果从零开始，像我这样用JSON Lines自研，工作量也不大，而且可控性最好。
• 插件机制用语言原生还是自研规范：Python的入口点机制很成熟，可以直接作为插件发现的基础，但插件之间的依赖描述和生命周期编排，还是需要在框架层自己定义。这一层建议做薄、做稳定，不要在版本迭代中频繁改接口。

我自己在实际使用中最大的体会是：框架的分层不要过早追求完美。最开始我设计了五层抽象，后来砍到四层才够顺手。插件接口多一个抽象，写插件的人就要多理解一个概念；事件类型多一个，调试的人就要多看一种情况。Harness的价值不在抽象多精妙，而在于把高频的、通用的、繁琐的事接住，让真正的业务开发变得轻快。

另外还有一个小技巧：给session_start事件里塞一个extra字段，专门放部署环境、代码版本号、依赖版本号这些元信息。排查问题的时候，回放日志显示行为异常，先看这个字段定位是不是版本变更导致的回归。这个小字段省了我无数次追查线上异常的时间。

这个框架后续还可以扩展的方向，我自己已经在着手做两件事：一是把model_request事件里的关键槽位做索引，支持按"使用了哪个工具""调用了哪个模型"直接检索会话；二是把采集到的历史会话自动标注成评测集，每次模型版本升级前先跑一遍。如果你也在做Agent框架，这两块值得优先规划。