这两年做 Agent 工程化我最大的感受是光有一个能跑通 demo 的工作流远远不够真正让人头疼的是框架耦合、排查困难、扩展两难。团队内部折腾了大半年最终沉淀出一套叫 DeepSeek Harness 的 Agent 运行框架。名字里带 Harness就是想表达“像线束一样把所有模块规整串起来”的意思。它最让我满意的两个工程决策一个是全插件化设计另一个是可回放的会话日志体系。这篇就把这两个核心设计的思路、落地过程、踩坑记录完整拆给大家偏工程实践不是理论课适合正在设计和自研 Agent 编排层的开发者参考。如果你也在纠结为什么同一个 Agent 框架在不同任务上表现落差这么大为什么线上会话现场总是复现不出来为什么想加一个新工具能力却要改动核心流程那这篇内容应该能给你一些启发。后面所有代码和配置思路都是真实项目中逐步迭代出来的版本我把过程尽量还原得详细一些。1. 为什么做全插件化把 Agent 拆成一张插拔表1.1 耦合之痛单体 Agent 的三大症状早期版本的 Agent 内核其实就是一个大函数。模型调用、工具注册、记忆读写、上下文组装、策略路由全部堆在同一个类里甚至不同需求还在同一个文件上叠加 if-else。初期功能少大家很爽一天能改三版。但随着业务接入变多问题逐一暴露。首先是维护困难。加一个“联网搜索”能力要改核心的agent.run()方法调整一下提示词模板也可能影响工具调用逻辑。牵一发动全身测试回归的周期不断拉长。其次是权限边界模糊。所有插件都能访问 Agent 内部的全部状态看似灵活实际上谁都能改上下文出了问题根本定位不到责任人。第三是隔离性差。某个外部工具超时如果异常处理不好直接拖垮整个 Agent 主循环。那个阶段我们做了一个关键决策不再把 Agent 看作“一个函数”而是把它看作“一个宿主环境 一组可插拔能力单元”。主流程只负责调度和装配具体能力全部交给插件。这样从架构层面上把可变部分和稳定部分切开。1.2 插件化的设计哲学能力即插头主流程即骨架全插件化不是简单的“把工具封装成类”而是要求整个 Agent 的每一个可变维度都能通过插件扩展。模型供应商、工具执行器、上下文压缩策略、记忆索引方案、输出校验规则、甚至日志记录器都被定义为插件接口。主流程从一个具体执行者变成了一个骨架它对插件能干什么不做假设只负责在正确的时机调用正确接口。这种设计的好处非常实际。第一新增能力不需要修改核心代码只需要写一个新插件并注册。第二团队可以并行开发一个人做工具插件另一个人做记忆插件互不阻塞。第三线上排查时生效的插件列表就是一份清晰的配置而不是靠猜。第四针对不同业务场景只需要切换不同插件组合就能形成差异化的 Agent 行为框架本身一行不用改。我见过很多团队做插件化做法是在核心代码里预留几个空方法然后让子类覆盖。这种“伪插件化”解决不了扩展问题反而引入继承地狱。真正可用的插件化是定义稳定的扩展点通过组合而非继承来扩展行为。这也是 DeepSeek Harness 在架构层面最坚持的一点。1.3 插件分类能力分层避免一篮子插头把所有能力都塞进同一个“插件”概念在落地时会有问题。比如一个工具插件和一个模型插件它们的生命周期、配置方式、依赖关系差别很大。因此我们很早就做了插件分类按照用途拆成几类模型插件封装各种大模型服务的访问统一输出标准消息格式。工具插件封装外部能力比如搜索、计算、代码执行、数据库查询。记忆插件管理短期上下文、长期向量记忆、会话折叠。策略插件决定下一步动作比如是继续调用工具还是结束会话。验证插件对模型输出做格式或内容校验避免脏数据进入下游。日志插件负责会话数据的采集、序列化、存储。分类之后插件间的依赖关系就清晰了。策略插件可以调用工具插件但工具插件不能反向控制策略。日志插件横切所有插件通过事件总线接收消息不直接依赖任何具体插件实现。这就在插件内部也保持了合理的依赖方向。2. 插件协议与生命周期从“草台插件”到“规整框架”2.1 插件接口最少约定原则插件接口的设计直接决定这个框架是越用越舒服还是越用越拧巴。我们踩过一次大坑一开始接口定义得特别丰富十几个抽象方法每个插件都要实现一堆用不到的方法新人上手就是满屏的NotImplementedError。后来重构时改成最少约定原则插件接口只定义三件事。第一插件能提供什么能力即能力描述。第二插件如何被初始化需要哪些配置。第三插件如何被调用输入输出遵循什么统一协议。用 Python 描述的话核心协议大概长这样class BasePlugin(ABC): name: str abstractmethod def initialize(self, config: dict) - None: 读取插件配置完成初始化准备工作 abstractmethod def capability(self) - PluginCapability: 声明插件提供的能力元信息 async def execute(self, context: AgentContext, payload: dict) - dict: 执行插件逻辑具体签名由各插件类型细化 raise NotImplementedError模型插件、工具插件、记忆插件各自继承BasePlugin并细化execute协议。PluginCapability里记录插件能处理的动作名称、输入模式、输出模式、所需权限。这样主流程在做调度时能直接根据能力元信息判断一个任务应该路由给哪个插件。顺便说一句接口设计里最容易忽略的是“失败语义”。插件执行失败时是重试、降级还是直接终止我们在每个插件返回值里强制带上status字段用枚举表示成功、可重试失败、不可重试失败。这样主流程不需要了解插件内部细节也能做出正确的容错决策。2.2 生命周期管理加载、初始化、启用、卸载插件不是简单的“ import 一下就能用”它有自己的生命周期。DeepSeek Harness 里插件状态机一共五个阶段已注册、已初始化、已启用、已暂停、已卸载。注册阶段插件列表从配置文件中解析框架扫描插件目录检查依赖是否齐备但不执行任何插件代码。初始化阶段执行initialize()加载配置、建立连接、预热缓存。这个阶段做重活。启用阶段把插件挂载到主流程的事件点上开始接收调用。暂停阶段常用于动态摘除某个不稳定插件比如某个工具服务正在升级。卸载阶段释放连接、清理临时文件、注销事件监听。生命周期状态的管理主要是为了保证安全。插件在初始化完成之前绝对不能进入主流程插件在暂停状态下既不能被调用也不能残留任何后台任务。我们为此专门写了一个PluginRegistry负责状态流转和并发安全控制。每次调度前插件都要重新确认自己的状态避免出现“插件已经卸载但任务还在队列里”的竞态问题。2.3 完整示例一个最小插件注册表我把核心的注册表逻辑简化成一个可运行的版本方便说明。这个注册表负责插件加载和按名称路由import asyncio from typing import Dict, Optional class PluginRegistry: def __init__(self): self._plugins: Dict[str, BasePlugin] {} self._states: Dict[str, str] {} def register(self, plugin: BasePlugin, config: dict) - None: if plugin.name in self._plugins: raise ValueError(fplugin {plugin.name} already registered) plugin.initialize(config) self._plugins[plugin.name] plugin self._states[plugin.name] enabled def disable(self, name: str) - None: self._require(name) # 实际项目里这里会通知主流程摘除事件监听 self._states[name] paused def enable(self, name: str) - None: self._require(name) self._states[name] enabled def unregister(self, name: str) - None: self._require(name) plugin self._plugins.pop(name) self._states.pop(name) # 释放插件持有的资源 if hasattr(plugin, shutdown): plugin.shutdown() def get(self, name: str) - BasePlugin: self._require(name) if self._states[name] ! enabled: raise RuntimeError(fplugin {name} is not enabled) return self._plugins[name] def _require(self, name: str) - None: if name not in self._plugins: raise KeyError(fplugin {name} not found) # 简单用法 async def demo(): registry PluginRegistry() registry.register(SearchToolPlugin({api_base: internal://search}), {}) plugin registry.get(search_tool) result await plugin.execute(ctx, {query: postgres deadlock 排查}) print(result)这里的_require保证任何操作都作用于已注册插件get方法保证只有启用状态的插件才会被调度。实际项目里注册表还会记录每个插件的加载耗时、配置哈希、版本号方便后续做配置溯源。3. 可回放会话日志Agent 版的“时光机”3.1 日志的三个层次从文本到事件流Agent 应用排查问题最原始的手段是翻日志文件。但它有天然的劣势文本日志丢失了结构多步推理的因果关系要靠肉眼拼上下文被截断关键变量打印不全想复现一个现场只能靠手工重放而重放过程无法保证与原始输入完全一致。所以我们在日志体系上分了三个层次。第一层是标准应用日志用 loguru 之类的库记录运行日志方便日常看异常。第二层是结构化日志以 JSON 行存储记录每次调用模型的 token 数、延迟、工具名、输入输出摘要。第三层才是重头戏事件流日志。事件流日志不是记“程序打印了什么”而是记“Agent 做了什么”。每一次模型请求、工具执行、状态变化、错误发生都作为一个不可变事件追加到日志流中。这本质上是一种事件溯源思想把 Agent 的一次完整会话建模成一系列有序事件的集合。只要事件流完整整个会话就可以被精确重建。3.2 事件流设计不可变的 append-only 序列每个事件我们设计成这样的结构{ session_id: sess_20241015_abc123, event_id: evt_0000000017, event_type: model_call_started, ts: 1728972345123, parent_event_id: evt_0000000015, trace_id: trc_9f3a, payload: { model: some-model, messages: [...], temperature: 0.3 } }这里有几个关键设计点。parent_event_id是最重要的字段之一。它把事件组织成一棵调用树而不只是一条平铺的链表。比如一个工具调用事件它的父事件是“决策事件”而它内部触发的模型调用事件父事件又是这个工具调用事件。回放时我们可以沿着这棵树做局部展开快速定位某一个决策链路的所有直接影响。trace_id用于关联分布式上下文。在单体部署中可能用不太上但一旦 Agent 服务拆成多进程模型调用和工具执行分布在多个节点上trace_id 就能把所有相关事件重新汇聚到同一条会话流里。每个事件都是一次性写入不做原地修改。如果某个事件数据有误我们追加一个evt_correction事件来标记而不是去修改旧事件。这样保证了日志流的可审计性和可重放性。3.3 回放引擎从事件流重建 Agent 行为回放引擎的核心目标是给定一个session_id把事件流还原成时序明确、上下文完整、可逐步推进的 Agent 运行视图。最简单的回放就是顺序读取事件按时间戳排序后展示。但仅展示不够我们还需要“状态重建”。因为很多 Agent 状态是运行时累积的比如记忆插入、上下文截断、变量变换。如果回放时只看到事件没有重放对应的状态变更还是无法理解为什么 Agent 在某一步做出了奇怪决策。所以在事件流里我们还会埋入state_snapshot事件。它记录关键节点上的状态快照比如当前消息列表长度、token 总消耗、记忆索引状态。回放引擎读取事件流时会维护一个“影子状态”。每处理一个事件就重放它对应的状态变更遇到快照事件则直接校正影子状态。这样做的好处是即使事件流里偶尔缺失一两帧快照也能兜底不至于让后续回放完全失真的。实际运行中事件流偶发丢失很难避免快照机制是我们的保命符。回放引擎还有一个很实用的能力只回放到某个事件为止。输入一个事件 ID引擎从会话开头逐步重放到该事件然后输出这个时间点的完整上下文。这个能力被我们大量用于“事故复盘”用户报告某个会话异常我们不需要看完整流只需要重放到异常前一刻就能拿到 Agent 当时看到的所有上下文。3.4 快照与断点让回放不止于“看”回放不应该只是看历史它还应该能被执行。DeepSeek Harness 的日志里我们把“环境种子”也记录下来。比如模型请求的随机种子、采样参数、工具返回结果。这样回放时可以选择“纯阅读模式”或“重执行模式”。纯阅读模式不会真正调用模型接口只展示事件流和状态变化优点是快、免费、无副作用。重执行模式则会真正调用模型或工具用来验证当前版本的 Agent 在历史会话上的表现对比新旧版本的决策差异。这就是回归测试的基本形态取一批历史会话日志重放到某个断点收集新版本的输出和旧版本做 diff。这个能力刚开始只是为排查设计后来反而成了我们做 Prompt 迭代和模型选型的主要工具。一个候选模型、一套新 Prompt到底表现如何不再靠人工编测试用例直接抓真实线上会话回放验证。就这一点节省的评测时间非常可观。4. 实操过程从零搭一个最小可回放 Harness4.1 工程结构与目录划分理论讲再多不如动手跑一个。我分享一下当前项目的工程结构这是一个精简过但仍能完整运行的版本。模块划分遵循“内核稳定、插件外置、日志独立”的原则harness/ ├── core/ # 内核模块不依赖任何具体插件 │ ├── agent_loop.py # 主循环调度器 │ ├── events.py # 事件定义与事件总线 │ └── context.py # Agent 上下文对象 ├── plugins/ # 插件存放目录 │ ├── model/ │ │ └── local_model.py │ ├── tools/ │ │ └── calculator.py │ └── memory/ │ └── buffer_memory.py ├── logging/ # 日志采集与回放 │ ├── session_logger.py │ └── replay.py ├── configs/ │ └── default.yaml # 插件装配配置 └── main.py # 启动入口核心约束只有一条core目录里的代码不能 import 任何plugins目录里的具体类。所有交互都通过接口进行。这个约束虽然简单但保证了核心的稳定性和可替换性也是全插件化不至于退化成“大杂烩”的底线。4.2 核心调度循环与事件采集主循环的关键逻辑是让 Agent 在“推理/执行/观察”三个环节中循环直到达到终止条件。我们把每一步都设计成“可埋点”的主循环在动作前后分别发出事件插件执行前后也发出事件。这里给出一个精简的事件模块# events.py import time import uuid from dataclasses import dataclass, field from typing import Any, Optional dataclass class Event: session_id: str event_type: str payload: dict[str, Any] parent_event_id: Optional[str] None trace_id: str field(default_factorylambda: uuid.uuid4().hex) event_id: str field(default_factorylambda: uuid.uuid4().hex) ts: int field(default_factorylambda: int(time.time() * 1000))主循环里的一个简版调度逻辑# agent_loop.py class AgentLoop: def __init__(self, registry, logger): self.registry registry self.logger logger async def run(self, session_id: str, user_input: str) - dict: ctx AgentContext(session_idsession_id) ctx.add_user_message(user_input) for step in range(self.max_steps): self.logger.emit( step_started, {step: step, messages_len: len(ctx.messages)}, parent_event_idctx.current_event_id, ) decision await self._make_decision(ctx) if decision.action finish: self.logger.emit(agent_finished, {result: decision.output}) return decision.output tool_plugin self.registry.get(decision.tool_name) tool_result await tool_plugin.execute(ctx, decision.args) ctx.add_tool_result(decision.tool_name, tool_result) self.logger.emit( tool_completed, {tool_name: decision.tool_name, result: tool_result}, parent_event_idctx.current_event_id, )注意每次emit都要带上parent_event_id我用ctx.current_event_id来追踪当前所在的调用树节点。这样后续回放才能完整还原决策链路。4.3 日志采集器异步批量落盘事件采集不能阻塞主流程所以我不建议在emit里直接写文件。我们的做法是所有事件先进入一个内存队列由独立的消费者线程批量落盘。# session_logger.py import json import threading from queue import Queue class SessionLogger: def __init__(self, path): self.path path self.queue Queue(maxsize10000) self._worker threading.Thread(targetself._consume, daemonTrue) self._worker.start() def emit(self, event): self.queue.put(event) def _consume(self): while True: event self.queue.get() line json.dumps(event.__dict__, ensure_asciiFalse) with open(self.path, a, encodingutf-8) as f: f.write(line \n)异步落盘能显著降低对 Agent 主循环的延迟影响。但要注意如果消费者处理速度跟不上队列会积压这时要丢事件还是阻塞需要根据业务取舍。我们在生产环境的选择是日志永远不能阻塞主流程极端情况下宁可丢日志也不能卡住 Agent 响应。这件事看似小事但在高并发下是必须做明确的取舍。4.4 回放脚本最简单的一次全量回放回放端的第一版脚本很简单就是按顺序读取事件流重建会话骨架。# replay.py import json class ReplaySession: def __init__(self, event_file): self.events [] with open(event_file, encodingutf-8) as f: for line in f: event json.loads(line) self.events.append(event) def run(self): steps [] current_step None for evt in self.events: etype evt[event_type] if etype step_started: current_step {step: evt[payload][step], events: []} steps.append(current_step) elif etype tool_completed: current_step[events].append(evt) elif etype model_call_started: current_step[events].append(evt) return steps这个脚本最大的意义不是功能多强而是验证了一个关键假设只要事件流中包含了足够的上下文信息重建一次会话只需要“读文件 按顺序处理”两步不需要依赖原始输入文件也不需要当时进程的内存状态。可回放性从这儿立住了。5. 常见问题与工程避坑实录5.1 插件热加载与类加载器隔离插件化设计做到中后期一个绕不开的问题是热加载。也就是不重启主进程就能更新某个插件代码。理论上Python 可以通过importlib.reload()实现但实际用下来问题很多被 reload 的模块可能被其他模块引用导致新旧代码混用类实例不会自动更新方法全局状态变量在新旧模块间不共享。我们最终的热加载方案是每个插件独占一个独立模块命名空间主进程通过一个桥接类动态转发调用插件更新时创建一个新的命名空间把旧命名空间整体废弃。这个方案不追求运行时无缝切换但保证了不会出现“一半新代码一半旧代码”的诡异状态。如果你只需要按天更新插件完全可以不做热加载每天凌晨重启一次进程更稳妥。5.2 事件日志文件膨胀可回放会话日志的数据量增长非常快。一次完整会话如果模型调用频繁事件条数可能上百每条事件还有消息内容。生产环境跑个把月日志文件轻松上百 GB。针对这个问题我们做了三件套。采样策略对于低价值事件比如心跳、监控类事件按比例采样或不记录。分层存储近期会话事件写高性能存储超过 7 天的自动归档到低频存储。会话级轮转每个会话一个独立目录按会话维度清理而不是按时间维度清理因为“保留完整会话”比“保留一段时间片段”更有价值。5.3 确定性回放的种子问题做回归测试时最怕的是“回放结果不稳定”。模型输出有随机性同样一个 Prompt两次调用结果可能不同。刚开始做回放对比时我们经常因为输出不一致而误判为回归问题。后来在事件流里增加了采样参数记录。回放重执行模式下我们强制固定模型推理的temperature0并且记录随机种子。对于工具类插件它们的输出通常来自真实外部系统本身不随机因此回放时优先使用事件里记录的工具原始结果而不是重新调用工具。这样重执行只对模型部分敏感工具部分完全确定。5.4 性能开销与算力代价有人会问采集这么多日志性能开销到底多大我们的实测数据正常业务负载下事件采集与异步落盘的额外开销约占主流程 CPU 的 1% 到 3%接口调用的 P99 延迟增加约 2 到 5 毫秒。这个代价换来的可观测性、可回放性、可回归测试能力我觉得是相当划算的。但在大规模部署时还是要注意一点事件采集数量应该做成可配置的。调试开发环境全量采集生产环境可以适当降低事件粒度比如只记录决策事件和工具完成事件不记录完整的模型输入输出。模型输入输出往往体积最大但信息密度最低除非有合规审计要求否则建议默认省略。很久不写长文了这一篇聊得比较细。最后说点实在的我在 DeepSeek Harness 这个项目里最大的收获不是某个具体功能写得多炫而是想清楚了一个道理——Agent 框架的复杂度不应该通过堆代码来消化而应该通过约束和结构来消化。插件化是让代码的复杂度可分割回放日志是让运行时的复杂度可观测。这两点如果能想明白自研框架的体验会顺畅很多。如果你也在搞 Agent 框架建议先从日志回放做起哪怕插件化暂时不做先把“会话可回放”这一条价值吃透排查问题的效率直接上一个台阶。