Agent工程化：全插件化架构与可回放日志实现

开篇当“跑通Demo”不再是终点Agent的工程化才是真正的分水岭如果你最近半年在折腾 Agent 开发大概率会有一种“拆分一时爽落地火葬场”的体会。模型的能力上限大家几乎拉平了真正拉开差距的是外层那套工程骨架插件怎么挂载、日志怎么留痕、一次带状态的会话怎么复现。我以前也觉得这些是“面向 API 的胶水活”直到在一个跨平台系统模拟项目里被多轮工具调用的状态错乱坑到怀疑人生才决定认真造一套能撑住复杂业务编排的 Agent 执行框架——也就是后来演化成 DeepSeek Harness 的全插件化架构。这篇文章不聊模型算法只聊我在做这个框架时踩过的坑、重新设计过的插件协议以及最让我满意的一块可回放会话日志。这套内容适合谁看如果你正在从单轮 Prompt 走向多轮 Agent 任务或者团队协作时发现“明明代码一样跑出来的分支行为却不同”那么这篇笔记应该能帮你省下一两周的试错时间。我尽量用工程上能直接落地的说法把设计取舍、代码骨架和排查经验都摊开讲。1. 整体设计与思路拆解为什么非要把 Agent 拆成“壳 插件”1.1 Agent 工程化的尴尬现状单机脚本可以生产环境不行很多人第一次把 Agent 写进业务时路径几乎一样先在一个 Python 脚本里定义run()循环里调模型、调工具、拼接上下文跑通一个 Demo 后倍感欣慰。但一旦接入真实场景比如需求是“让 Agent 自动处理业务工单”局面立刻变得复杂工单类型会不断新增每加一种都要改主循环的 if-else不同任务需要不同的工具集、不同的提示词模板、不同的后处理逻辑出了问题时日志里只有 “tool called” 这种片段的假象你根本不知道模型当时到底看到了什么样的历史状态。第一个痛点本质上是“变异性”没有被隔离。工具、提示模板、后处理逻辑、上下文裁剪策略这些被硬编码在同一个执行循环里每加一个需求就让主代码膨胀一圈。第二个痛点是“状态”没有被显式建模。Agent 的每一步输出都依赖于完整的对话上下文与工具返回结果可是常规的 log 方案只记录“发生了什么”不记录“为什么是这一步”于是复现问题几乎全靠运气。我在模拟项目 X 里最早一版也是这种“主循环 分支控制”的路子两周后代码量冲到四千多行几乎没法维护。后来下决心重构才有了 DeepSeek Harness 的原型。1.2 Harness 的定位不绑死任何模型只负责编排、隔离和留痕把 Agent 比喻成一个人的话主循环是“神经系统”工具是“手脚”提示词是“思维模板”。DeepSeek Harness 要做的不是替你把手脚造好而是把神经系统搭成一个标准插座你把手脚插进来它就帮你统一管理信号、记录动作、复现行为。这套架构的核心原则就三个执行与能力解耦模型调用、工具执行、记忆检索、输出格式化全部抽象成独立的插件单元主流程不感知具体实现。协议优先于实现所有插件只依赖接口约定不依赖彼此的内部数据结构。一切留痕可回放会话内每一个事件模型的输入、模型的输出、工具的调用参数、工具的结果、分支的触发条件都按时间序列记录下来并且能“重放”成一份与原会话完全等价的执行轨迹。其中前两条解决“能不能扩展”第三条解决“出了问题怎么查”。这三条原则叠加之后Agent 系统才真正变成一个工程系统而不是一段魔法脚本。1.3 为什么选择“全插件化”而不是“策略模式”或“微服务”有人会问拆模块的方式很多为什么要致力于“全插件化”我也想聊一下对比过程。策略模式的问题是接口太细。策略模式适合“算法可以替换”的场景但 Agent 的可变点不仅包含算法还包含数据来源、中间处理器、上下文组装方式、后处理逻辑等多种类别若每个可变点都定义一个策略接口抽象层级会迅速失控。微服务又太重了工具调用之间的延迟、部署成本、数据一致性处理对一个几十个插件的规模来说是杀鸡用牛刀。全插件化的本质是“统一生命周期 统一数据契约 统一挂载点”。每个插件有自己的生命周期初始化、运行、销毁输入输出是结构化的统一消息体挂载方式只有“声明式注册”一种。主流程根本不需要关心你是一个 API 工具还是一个知识库检索器你的身份只是“一个处理器”。而且插件化还有一个意外收获团队协作时的冲突大幅减少。以前改主循环代码每个人都在动同一个文件合并起来痛不欲生。改成插件之后每个人贡献的只是一个独立文件或独立目录只要接口没变代码合并基本可以做到无感知。这个收益在多人并行开发时尤其明显。2. 核心细节解析与实操要点把“插件协议”和“会话日志”设计到能用的程度2.1 插件协议的最小完备集接口抽象得越少越好设计插件系统时最忌讳一上来就定义一堆抽象基类。我第一版设计里有BaseToolPlugin、BasePromptPlugin、BaseOutputPlugin每个基类二十多个抽象方法看起来“面向未来”实际上每个插件开发者都在骂。后来我砍掉了绝大多数方法只保留一个通用接口from dataclasses import dataclass, field from typing import Any, AsyncIterator, Optional dataclass class Event: type: str # model_input, model_output, tool_call, tool_result, flow_branch payload: dict[str, Any] seq: int # 全局自增序号用于回放时的顺序还原 timestamp: float trace_id: str # 会话ID用于关联同一会话的全部事件 parent_seq: Optional[int] None # 父事件序号用于构建因果链 class HarnessPlugin: name: str base version: str 0.0.0 async def on_init(self, context: dict[str, Any]) - None: pass async def on_event(self, event: Event, context: dict[str, Any]) - AsyncIterator[Event]: # 这是唯一最重要的方法。插件接收一个事件处理后产出零个或多个新事件 if False: yield event async def on_shutdown(self) - None: pass这个设计中有几个关键点值得展开讲。第一为什么只留on_event一个核心方法因为 Agent 的整个执行过程就是事件流用户输入是一个事件模型输出是一个事件工具调用是一个事件。把一切抽象成事件之后插件之间的所有交互都变成了对事件流的“监听—处理—产出”操作。你不需要关心上游是谁也不需要关心下游是谁你只处理你认识的事件类型。第二上下文context是什么它是会话级别的共享字典存活于整个会话生命周期插件可以在里面读写临时数据。比如工具执行插件可以把“某个 API 的鉴权 token”放进去后续重试逻辑就能在同一个上下文里拿到凭证。但这里必须加一个纪律不要在 context 里存大对象否则会拖垮回放时的内存。我们后来专门限制了 context 的序列化大小超过阈值自动告警。第三版本号为什么必须保留因为插件更新之后历史会话日志里记录的“当时执行语义”可能已经不存在了。有了版本号回放器就能判断这个日志记录的是 v1.0.3 的插件行为当前代码是 v2.0.0要回放的话应当先切换到对应版本或者至少给出警告。没有版本信息的日志数周之后基本是废纸。2.2 插件注册机制声明式配置文件比代码注册更实用在程序里写register(MyPlugin())当然可行但当插件数量超过 30 个时纯代码注册的维护成本直线上升。更深层的问题在于代码注册把“哪些插件生效”这件事写死在了源码里而实际运维中经常需要按场景切换插件组合。我在 DeepSeek Harness 里采用的是“声明式注册 目录扫描”的机制。每个插件目录下有一个manifest.yaml描述插件名称、版本、入口类、默认配置、依赖的其他插件。主程序启动时扫描指定目录按 manifest 顺序加载。# tools/web_search/manifest.yaml name: web_search version: 1.2.0 entry: plugin.WebSearchPlugin config: max_results: 5 timeout_seconds: 10 depends_on: - http_client - result_normalizer这种做法有四个直接好处新增插件不需要改动主程序代码扔一个目录进去即可可以通过环境变量或外部配置覆盖 manifest 中的默认参数不需要重新发布依赖关系显式声明启动时就能检测环依赖而不是运行时才炸运维人员不需要懂 Python 就能完成插件启停。有一个细节值得提醒插件扫描顺序和依赖解析必须稳定如果扫描依赖了字典序或者文件系统返回的随机顺序那么同一套配置在不同机器上可能产生不同的行为。我们后来的做法是先收集全部 manifest做拓扑排序再按排序结果加载。这样一台新机器启动后插件加载顺序永远一致。2.3 会话日志的设计目标不只是“日志”而是“时间机器”把日志做成“可回放”不是加几行 print 就能交差的。回放的核心要求是任意时刻的会话状态都能被精确重建。这意味着你要记录的不是“用户说了什么、助手回了什么”这种粗粒度对话而是每一轮模型调用时“发送给模型的完整消息数组”、当时生效的插件配置、工具调用的原始参数与完整返回。实现回放的技术栈可以拆成四个部分事件捕获层所有插件的输入输出统一封装成 Event由执行引擎自动落盘持久化层事件流写入本地文件或消息队列按 trace_id 分片索引层生成时间戳、seq、事件类型三个维度的索引回放器按 seq 逐个重放事件并驱动一套“回放专用上下文”还原执行现场。日志格式我用了 JSON Lines每行一个事件不要小看这个简单的选择。JSON Lines 天然支持追加写不需要昂贵的事务机制也方便用标准流式工具逐行处理。更重要的是它很容易做“校验”回放器读到的每一行即使在恶意编辑的情况下也能被及时发现因为 JSON 结构破坏了就能立刻识别出来。2.4 回放会话日志从“看到了”到“重现了”之间差了三个关键能力只把事件存下来离“可回放”还差得远。要从日志中真正还原执行现场我认为必须具备三个能力。能力一确定性重放。同一份日志无论用哪台机器哪个时间点去重放最终得到的事件序列应当完全一致。如果重放过程中有任何随机性比如采样 top_p 随机、时间戳参与排序、集合遍历顺序不确定回放就失去了对比价值。解决办法是把随机种子、采样参数temperature、top_p全部固化进行日志重放时值只来自日志而不是来自当前环境。能力二分支复现。Agent 执行过程中经常有 if-else 逻辑比如“如果工具返回错误则重试最多三次”。日志必须记录每个分支判定时的输入条件和判定结果。只记录“重试了三次”是不够的要记录“为什么重试”也就是把判定条件和上下文片段一起写入日志。能力三外部副作用隔离。重放时最怕的是调用真实的工具 API既慢又贵还容易污染数据。我的方案是给回放器集成一个“mock 模式”日志中的 tool_call 事件按原参数重放时直接返回日志中记录的 tool_result而不是真的去调用远端 API。这样回放过程完全离线、确定、无副作用。这个设计听着简单实际操作时才发现难点在于如何从日志中正确匹配“哪次调用对应哪次返回”。解决办法是在 tool_call 事件里写入 call_idtool_result 事件里也写入相同的 call_id用这个 ID 关联。3. 实操过程与核心环节实现从零搭建一个带可回放日志的 Agent Harness3.1 最小骨架先让事件流跑起来做这种架构最好的策略是“先搭骨架后填肉”。我的做法是先用 200 行左右代码把事件流引擎搭出来不急着写任何业务插件先把模型的 Mock 调用跑通。import asyncio import json import uuid from datetime import datetime, timezone class HarnessEngine: def __init__(self, plugins: list[HarnessPlugin], event_storeNone): self.plugins plugins self.context: dict[str, Any] {} self.seq 0 self.event_store event_store or InMemoryEventStore() self.trace_id uuid.uuid4().hex async def emit(self, type: str, payload: dict[str, Any], parent_seq: Optional[int] None) - Event: self.seq 1 event Event( typetype, payloadpayload, seqself.seq, timestampdatetime.now(timezone.utc).timestamp(), trace_idself.trace_id, parent_seqparent_seq, ) await self.event_store.append(event) for plugin in self.plugins: async for new_event in plugin.on_event(event, self.context): await self.publish(new_event) return event async def publish(self, event: Event) - None: await self.event_store.append(event) for plugin in self.plugins: async for new_event in plugin.on_event(event, self.context): await self.publish(new_event) async def run(self, user_input: str) - None: await self.emit(user_input, {text: user_input}) # 这里通常由 orchestrator 插件决定如何调用模型 # 骨架只保证事件流能流转具体调度放给插件 await asyncio.sleep(0)这段代码虽然只有三十几行但它体现了架构里最重要的一点引擎不感知业务。它不知道“模型是什么”“工具有哪些”它只负责把事件分发给所有插件再把插件产生的后续事件继续广播出去。业务逻辑全部下沉到插件层。跑通这个骨架后我们做了一个全链路 Mock用户输入事件发起一个 orchestrator 插件模拟生成模型回复另一个 logger 插件把回复写进控制台。整个过程没接真实模型但事件流的拓扑结构已经显现出来了。3.2 插件的真正挂载事件广播与“谁监听谁”的协作模式有了引擎之后写插件就非常舒服了。比如我想加一个“请求日志审计”插件只需要监听model_input和model_output事件把完整载荷快照存到另外一张表里。再比如我想加一个“敏感词过滤”插件只需要监听tool_result事件在向模型回传之前做拦截与替换。这种模式的威力在“组合”时最能体现同样的引擎只需要替换插件目录就能从“客服助手”平滑切换成“代码审查助手”。你不需要重写主流程只需要准备另一组插件。当然事件广播也有副作用所有插件都会收到所有事件插件内部必须自行过滤自己关注的事件类型。这会造成一些无效遍历但对中小规模系统完全够用。如果将来插件数量超过一百、事件吞吐量非常高可以考虑改成“按事件类型订阅”的发布订阅模型。不过在现阶段这个复杂度完全没必要提前引入。3.3 日志落盘的工程细节从进程崩溃中保住最后一条事件可回放日志的价值建立在“可靠落盘”之上。如果进程在写日志时崩溃最后几条事件丢了那么回放就会得到一个残缺的现场问题定位难度瞬间增大。为避免这个情况我采用了两级缓冲策略第一级内存中按 trace_id 分片的事件缓冲区事件先写入内存定时或定量落盘第二级落盘前先写入一个.pending文件写完再原子 rename 成正式日志文件。import json import os import aiofiles class FileEventStore: def __init__(self, log_dir: str, flush_every: int 50): self.log_dir log_dir self.flush_every flush_every self.buffer: list[Event] [] self.pending_writer None async def append(self, event: Event) - None: self.buffer.append(event) if len(self.buffer) self.flush_every: await self.flush() async def flush(self) - None: if not self.buffer: return trace_id self.buffer[0].trace_id pending_path os.path.join(self.log_dir, f{trace_id}.pending) final_path os.path.join(self.log_dir, f{trace_id}.jsonl) async with aiofiles.open(pending_path, a) as f: for event in self.buffer: await f.write(json.dumps(event.__dict__) \n) os.rename(pending_path, final_path) self.buffer.clear()这里最容易被忽视的是数据库原子替换原子性语义在普通文件系统上的应用只有当文件完整写入后才执行 rename对读取方而言日志文件才从“不存在”跃迁为“完整存在”。如果有多个事件存储消费者同时读文件这个技巧能保证它们不会读到半截文件。还有一个细节事件类型为tool_result的记录往往体积庞大完整返回可能几十 KB如果不做裁剪日志文件会快速膨胀。我的策略是“全量记录 可选裁剪”。默认全量写但在配置里允许针对特定插件设置record_size_limit超过限制时截断并在 payload 里标记truncated: true。排查时如果发现被截断再去对应的业务日志里找完整数据。3.4 回放器的实现把日志变成可执行剧本回放器是我整个框架里最得意的一块。它的核心逻辑非常直观读取 JSONL 文件逐条重放事件在重放时把 tool_call 事件替换为工具仿真执行mock然后把每一步产生的状态与原始日志中的事件状态进行对比。import json from pathlib import Path class Replayer: def __init__(self, plugins: list[HarnessPlugin], mock_engine: MockToolEngine): self.plugins plugins self.mock_engine mock_engine self.context: dict[str, Any] {} async def replay(self, log_path: Path) - None: events [] with open(log_path, r, encodingutf-8) as f: for line in f: line line.strip() if not line: continue data json.loads(line) events.append(Event(**data)) # 按 seq 排序保证乱序写入也能正确回放 events.sort(keylambda e: e.seq) for event in events: if event.type tool_call: # 不调用真实工具从日志中找回放结果 mock_result self.find_mock_result(events, event) fake_event Event( typetool_result, payloadmock_result, seqevent.seq 1, # 模拟真实执行时的 next seq timestampevent.timestamp, # 保留原时间便于对比 trace_idevent.trace_id, parent_seqevent.seq, ) for plugin in self.plugins: async for new_event in plugin.on_event(fake_event, self.context): self.context self.merge_context(self.context, new_event) continue for plugin in self.plugins: async for new_event in plugin.on_event(event, self.context): pass # 这里可插入断言比较当前 context 与日志中记录的 context 快照回放器有两个设计点需要特别说明。第一是parent_seq的作用。回放本身是线性遍历似乎不需要 parent_seq但实际工作中经常要回答“这个工具调用是因哪条模型输出触发的”有了 parent_seq 才能构建因果链而不是只有时间上的先后。哪怕回放器不强制使用它分析工具和可视化面板一定用得到。第二是“上下文合并”的实现。插件在回放过程中可能会有自我状态更新比如“调用次数 1”如果重放后不把最新值合并回 context后续插件看到的次数就会错。合并策略我们采用了简单却有效的做法每个插件显式声明要保留在 context 中的 key 白名单其余键全量丢弃。这看起来像是一种限制实际上是为了防止插件误把一个内部临时变量覆盖到全局共享上下文中。3.5 关键决策回顾为什么要给每个事件分配全局 seq日志回放时会遇到一个很现实的问题多条事件可能同时产生比如多个插件并行触发落盘时它们的相对顺序可能与真实发生顺序不一致。如果只依赖时间戳排序毫秒级的时间精度在并发场景其实是不够用的同一毫秒内可能产生几十条事件排序结果不稳定。解决办法就是全局自增 seq。所有事件在产生瞬间就被分配一个序号seq是整个会话中的严格全序落盘顺序即使打乱也能据此重建执行序列。回放器先 sort by seq再逐条执行就保证了重放路径等价于原始执行路径。这个机制是“可回放”的基石没有它回放就只是个好看的噱头。4. 常见问题与排查技巧实录4.1 插件加载顺序导致的“幽灵冲突”有一次我们把一个新的上下文增强插件接入系统后突然发现所有工具调用都出现 401 鉴权错误。刚开始以为是鉴权插件出了问题排查了半天最后发现是上下文增强插件在初始化时把 context 里的 token 字段覆盖成了空字符串。问题不在新插件本身而在它加载顺序早于鉴权插件导致鉴权插件后来写入的 token 被覆盖。这个案例给我的教训是插件之间的 context 写入权必须显式化。后来我们在 manifest 里增加了context_keys声明启动时检测到两个插件声明同一个 key 且写入模式都是 “override” 时直接拒绝启动。宁可启动失败也不要运行时“随缘覆盖”。4.2 回放时的时序失真回放环境里的时间基准与原始时间不一致回放早期我在比较重放结果时发现输出内容完全正确但时间戳全是重放时刻的真实时间不是原始时间。这个问题乍看不影响正确性但一旦要做“耗时对比分析”比如定位慢在哪一步重放数据就不可用了。后来我调整了回放策略默认使用原日志里的timestamp字段驱动作业计划也就是说模拟执行时保留原事件时间只有插件内部发起新的外部请求时才用当前时间。这样既能隔离真实请求又能保留原始时间视角。4.3 幂等性重放与外部依赖的边界回放器虽然 mock 了工具调用但插件内部如果有发送 HTTP 请求的逻辑比如错误告警模块重放时依然可能触发真实外部调用。这既污染外部系统又破坏回放确定性。最终我们加了一层“网络策略白名单”重放模式下插件只允许访问白名单内的地址通常是完全离线地址其他网络请求一律被拦截并记录。这个策略听起来简单但真正实现时才发现需要侵入插件的 HTTP 客户端封装层统一注入一个过滤器。4.4 日志文件损坏与异常恢复日志写入通常很可靠但磁盘故障、手动删改还是可能导致 JSONL 文件损坏。我们在回放器里加了一道“容错读取”流程逐行解析遇到损坏行时先尝试跳过并在报告中标注“从第 N 行起跳过 X 条事件”。这样至少能回放到故障点前面的完整事件流。更进一步我们开发了一个小工具能对损坏日志做“局部修复”——如果只是某个 JSON 的结尾被截断就尝试补齐括号后继续解析。不过这条修复路径只作为兜底手段绝不值得依赖。5. 实战中的几点心得与后续演进在实际使用 DeepSeek Harness 的过程里我最大的感受是插件化的难点从来不在写插件而在约定插件之间的数据边界。只要把 context 的读写权、事件类型的语义、版本兼容策略定清楚后面全是体力活。可回放日志也是如此难点不在于记录事件而在于记录足够的上下文信息来支撑事后的因果推断。有一些很受用的经验不要追求插件框架功能大而全够用即可。我在 0.1 版本设计了七个抽象基类后来删到只剩一个HarnessPlugin开发效率反而提高。事件类型的设计要面向“语义”而不是“来源”。不要造openai_response这类事件要造model_output这类事件。后者能在不改变插件代码的前提下替换底层模型。日志回放要尽早做不要等项目大了再补。对一个三十行骨架做回放只需要一天对一个三千行系统做回放可能需要三周。后续这个框架最值得扩展的方向有两个一是把回放日志与缺陷分析工具打通比如自动对比“重放结果”和“原始执行结果”的差异直接圈出行为漂移点二是把插件状态的一致性校验做得更细引入类似数据库快照的概念每 N 个事件做一次 context 全量快照回放时就能从最近快照快速重建而不是必须从头跑一遍。架构这条路说到底是不断在“灵活性”和“可控性”之间找平衡。全插件化给了灵活性可回放日志给了可控性两者一结合Agent 系统才真正进入了“可工程化”的周期。后面无论模型怎么换、工具怎么加这套底座都能稳稳托住这大概是我在这类项目里最值回票价的一笔投入。