Agent框架工程化实践：全插件化架构与可回放会话日志

上个月排查一个多轮 Agent 的线上问题时我彻底受够了面向过程式的会话调试用户在同一轮里问了两个问题模型明明应该先查订单再查物流结果日志里只看到第二次工具调用的返回值第一次调用的入参和中间状态全部丢失。对着控制台翻了一个下午最后靠猜才定位到是一个工具返回的 boolean 字段把状态机带偏了。那一刻我确定Agent 框架真正该解决的从来不是模型多聪明而是出问题之后你能不能在一小时之内复现它、解释它、修正它。这篇文章就把我当时重写的一套内部框架 DeepSeek Harness 的工程化设计完整拆开讲清楚重点是两个核心支柱全插件化架构和可回放会话日志。适合正在做 LLM 应用落地、被 Agent 工程化折磨过、以及想自己设计一套可维护 Agent 运行时的工程师。1. 从 Demo 到生产Agent 框架到底缺什么1.1 能跑通的 Agent 很多能维护的 Agent 很少LLM 具备函数调用能力之后搭一个 Agent demo 的门槛被压到了极低把 API Key 填上写一个 tool 描述配上 system prompt循环里加一次工具结果回填一个会查天气、会算账的 Agent 就诞生了。这也是过去一年里大量团队的常态——demo 仓库热闹生产环境谨慎。但一旦进入生产问题就会密集地暴露出来。工具数量从 2 个涨到 30 个每个工具的参数格式、错误码、返回结构都不统一会话从单轮问答变成多轮带状态的任务中间还要穿插记忆压缩、知识检索、人工审批。这时候如果用命令式代码把流程写死每一次需求变更都会变成一场连锁反应改一个工具的返回结构调度器要跟着改加一种记忆策略状态恢复逻辑要跟着改换一个模型版本解析器可能又要重写。我见过不少项目最后都退化成一个巨型 service 类里塞了几千行 prompt 拼装逻辑的状态。代码能跑但没人敢改。这其实就是工程化缺失的表现系统的行为边界不是通过接口划定的而是靠开发者的自律维持的。1.2 三个让团队最痛的工程问题站在维护者的视角我把当时的痛点归纳成三件事。第一是行为分散。工具调用、模型调用、上下文管理、状态存储散落在不同的函数和类里彼此通过隐式全局变量耦合。想给某个工具追加权限控制得先搞清楚它被多少个地方直接调用。第二是不可观测。模型为什么选择调用这个工具而不是那个工具工具返回后模型为什么立刻终止而不是继续追问这些决策过程如果没有系统性的日志记录就只能靠线上抓取零散日志去考古。第三是不可复现。线上用户报了一个问题本地用同样的 prompt 跑一遍模型给的结果大概率不一样——因为 LLM 本身有随机性加上上下文、工具结果、并发时序都不一致。你无法把一个线上会话在本地原样重建就无法对根因做精确分析。这三个问题不是靠写更多代码能解决的而是要靠架构层面定义边界、定义数据、定义可观测性。1.3 DeepSeek Harness 的定位调度、编排、留痕Harness 这个词在 LLM 工程里通常指执行控制层有点类似测试领域的 fixture 概念它不负责模型本身而是负责把模型、工具、记忆、状态这些零件有序地组织起来。我给框架起的内部代号是 DeepSeek Harness默认对接 DeepSeek 模型接口但设计上模型无关换成其他兼容接口只需要替换一个 Provider 插件。这套框架从第一天起就锁定了两个设计目标一切行为以插件为边界一切决策以日志为证据。插件化解决的是行为分散和难以替换的问题可回放日志解决的是不可观测和不可复现的问题。两个目标相互配合日志记录的是插件之间的协议数据和上下文状态而插件通过统一的上下文体读写接口让日志得以完整捕获每一次决策所需的信息。听起来不复杂但真正把这两个目标落地成代码过程中牵扯到的权衡比想象中多得多。下面我按设计顺序展开。2. 全插件化设计把 Agent 运行时拆成一张可插拔依赖图2.1 先分清哪些东西必须成为插件插件的粒度是整个设计的分水岭。拆得太细光协议就能把你淹没拆得太粗又退回到一个 service 类的老路。我在实际设计中只保留了五类插件这个划分基本覆盖了 Agent 运行时的全部可变点。插件类型职责默认实现ToolPlugin封装外部工具、API、数据检索内置 HTTP 请求、内置搜索ProviderPlugin对接模型接口、处理流式和函数调用DeepSeek Chat 接口实现MemoryPlugin上下文管理、摘要压缩、长期记忆滑动窗口记忆ParserPlugin解析模型输出处理结构化和重试首选工具调用解析器StatePlugin会话状态存储、恢复与迁移SQLite 会话存储这五类插件的共同点是它们都对应一个会变化的行为维度。工具会增删模型会更换记忆策略会迭代输出格式会调整存储方案会迁移。把这些维度做成插件变化就被限制在独立的边界里而调度逻辑、上下文传递、日志记录这些相对稳定的部分则留在内核中不随插件演进而被动修改。2.2 插件协议最小接口加声明式元数据插件本质上就是一个运行期可被动态发现的 Python 模块通过统一的注册函数暴露自己的元数据和能力。我把协议的核心约束定得非常窄每个插件必须声明自己的名字、版本、类型和依赖同时实现对应类型的运行时接口。# plugin_base.py from dataclasses import dataclass, field from typing import Any, Awaitable dataclass class PluginMeta: name: str version: str plugin_type: str # tool / provider / memory / parser / state description: str dependencies: list[str] field(default_factorylist) class ToolPlugin: meta: PluginMeta async def run(self, payload: dict, ctx: AgentContext) - dict: 执行一次工具调用payload 是经过校验的入参。 raise NotImplementedError def to_protocol_spec(self) - dict: 返回给模型的 function-call schema。 raise NotImplementedError这里有两个细节值得解释。第一ToolPlugin 的入参和返回值一律要求是纯 JSON 数据不允许传递 Python 对象。这意味着工具不能偷偷依赖调用方的内存状态所有跨插件的数据交换都走显式序列化——这为后面的日志回放打好了地基。第二ctx是内核注入的 AgentContext它封装了会话状态、日志写入、内存访问能力插件通过它读取上下文而不是直接去访问全局对象。这样设计的意图是让插件变成无状态的纯函数 显式上下文访问。无状态意味着可替换、可测试显式上下文访问意味着所有副作用都对内核可见可以被记录和审计。2.3 插件的加载、注册与依赖校验插件加载我采用了扫描目录 拓扑排序的经典方式。约定每个插件放在独立子目录下都暴露一个register()函数返回插件实例内核扫描到所有模块之后先收集元数据再按依赖关系决定加载顺序。# registry.py import graphlib from importlib import util from pathlib import Path class PluginRegistry: def __init__(self): self._plugins: dict[str, Any] {} def load_from(self, plugin_root: Path) - PluginRegistry: modules [] for py in plugin_root.glob(*/plugin.py): spec util.spec_from_file_location(fplugins.{py.parent.name}, py) mod util.module_from_spec(spec) spec.loader.exec_module(mod) modules.append(mod.register()) by_name {p.meta.name: p for p in modules} graph {name: set(p.meta.dependencies) for name, p in by_name.items()} for name in graphlib.TopologicalSorter(graph).static_order(): plugin by_name[name] for dep in plugin.meta.dependencies: if dep not in self._plugins: raise RuntimeError(f{name} 依赖的 {dep} 未加载) self._plugins[name] plugin return self依赖校验放在加载阶段而不是运行阶段这一点是我反复权衡后坚持的。如果你等插件真正跑起来才发现某个依赖不存在错误往往出现在执行链路深处排查成本会高一个量级。加载阶段做一次拓扑排序和存在性检查最多花几毫秒却能把一类环境不一致的问题提前暴露。另外插件之间的依赖必须指向接口而不是指向具体实现。比如一个 tool 插件需要读会话元数据它依赖的应该是state类型插件而不是某个具体的 SQLite 实现。这样替换实现时不需要改依赖声明。2.4 为什么我不做大配置中心可能有人会问这些变化点用 YAML 配置不也能解决吗配置项可以控制开关、模型名、温度、工具列表看起来也是一种抽象。我的观点是配置化只适合表达参数差异不适合表达行为差异。举一个实际例子。同样是记忆策略滑动窗口和摘要压缩不是靠两个布尔开关就能切换的——两者的算法逻辑、触发时机、失败处理完全不同。如果硬把它们塞进配置项配置本身就会变成一门需要专门学习的 DSL最终比代码还难维护。插件则是把一套完整行为封装成对象注册即可用天然支持独立测试和版本管理。从工程组织上看插件化还有一个隐性收益团队可以按插件拆分工。工具组维护 ToolPlugin提示词组维护 ParserPlugin 和 MemoryPlugin模型接入组维护 ProviderPlugin互不踩代码。这对多人协作的项目来说边界比任何代码规范都更硬。3. 可回放会话日志让每个 Agent 决策都能被重新演绎3.1 一份合格会话日志的数据模型很多人的 Agent 日志就是简单地把每次 LLM 请求和响应打到大盘里美其名曰留痕。但到了排查的时候会发现完整决策链路上的中间信息大量缺失模型是在哪一轮决定调用工具的调用前上下文里已经有哪些消息工具返回了什么样的原始数据解析器又做了哪些清洗我设计的会话日志以事件为基本单位每个事件对应一次原子的运行动作。一段简化的事件结构如下{ event_id: evt_00042, session_id: sess_abc, step: 42, causal_parent: evt_00039, ts: 2025-06-12T10:24:31.082Z, type: tool_call_end, plugin: search_orders, payload: { request_id: req_88, input: {order_id: 202506001}, output: {status: shipped}, error: null, duration_ms: 214 } }这个结构背后有三个原则。第一是全量凡是进入插件边界的数据都要记录尤其是工具入参和出参——它们是模型后续决策的直接依据。第二是有序step字段是一个从 0 递增的逻辑时钟代表会话内的因果次序并不是墙上时间。第三是可关联causal_parent指向触发当前事件的上一个事件这样可以从任意一个异常事件沿着因果链回溯。3.2 事件流加周期快照的存储布局日志文件如果只按时间顺序线性写回放长会话时就会面临一个效率问题为了重建第 10000 步的上下文你可能要重放前 9999 步的所有事件。这在早期测试会话里还能忍受但到了生产长会话就会变得不可接受。我的方案是事件流 周期快照两层布局。每个会话对应一个目录sessions/sess_abc/ manifest.json events.jsonl snapshots/ 00000.json 00050.json 00100.jsonevents.jsonl是追加写入的完整事件流保证不丢任何决策痕迹snapshots/每 N 步我通常设置为每 50 步或每 10 次模型调用写入一份经过序列化的上下文快照。回放时只需要找到距离目标步数最近的一份快照然后重放快照之后的事件即可避免从头遍历整个会话。快照本身也是 JSON 纯数据包括当前系统消息、消息列表、会话状态字典、已注册插件元信息。它既是回放加速的检查点也是崩溃恢复的基石——如果进程中途挂掉下一次可以从快照恢复而不是推倒重来。3.3 回放引擎的核心逻辑回放引擎做的事情本质上是把日志再执行一遍。关键在于它不是真的再去调用外部模型或工具而是把事件当作状态转移函数在本地重建内存态。# replay.py import json from pathlib import Path class ReplaySession: def __init__(self, root: Path): self.root root self._events self._load_events() def _load_events(self): events [] for line in (self.root / events.jsonl).read_text().splitlines(): events.append(json.loads(line)) return events def replay_until(self, step: int): state self._load_snapshot(step) or {messages: [], session_state: {}} for ev in self._events: if ev[step] step: break state self._apply(ev, state) return state def _apply(self, ev: dict, state: dict) - dict: # 根据事件类型更新消息列表、会话状态、工具调用记录 ...这个引擎天然支持两种用法。一种是离线勘探找到一个异常 step把状态导出来看当时模型到底看到了什么另一种是回归测试把历史会话收集成 golden set修改 Prompt 或插件之后批量重放比对同一事件序列下的最终状态差异。回放模式可以按是否真正调用模型分成三类我整理成表格方便大家比较回放模式是否调用外部模型用途离线回放否直接用日志里的模型输出历史问题根因分析、审计Mock 回放否注入固定响应回归测试、依赖变更验证真机重放是重新调用模型Prompt/参数变更对比、模型升级评估3.4 用回放日志做根因分析和回归测试这套机制上线后我们排查问题的节奏发生了质的变化。以前是用户反馈—复现失败—抓日志—靠猜现在是拿到 session_id—本地回放—直接看到出错步骤的完整上下文。有一个案例印象很深。某次线上 Agent 在某一步工具调用后返回了异常以前只能看到模型最终回复了一句暂时无法处理。通过回放我们发现出错前一步的工具返回里有一个字段类型从字符串变成了对象而解析器代码依然按字符串处理于是静默丢弃了关键信息。整个过程只花了半小时其中大部分时间还是在写复现脚本。回归测试的价值同样被低估。Prompt 工程师最怕的事是改了一句话别的任务全崩了。有了回放日志我们可以把线上真实会话抽样成测试集每次改完 Prompt 自动跑一遍 golden set 回放任何行为漂移都会被 diff 出来。这个能力让 Prompt 迭代从靠手感变成靠证据。4. 实现路上的关键权衡与踩坑记录4.1 插件加载顺序与运行时才解析的依赖拓扑排序能解决加载顺序但不能解决所有依赖问题。我在第一版里遇到过一个典型的坑一个 tool 插件依赖 memory 插件但在工具真正执行时memory 插件可能还没完成初始化如果我把依赖写成加载阶段就必须实例化又会导致一些重量级插件比如要建数据库连接的 StatePlugin被过早初始化。最后的解法是区分结构依赖和运行依赖。结构依赖在加载阶段解析——对应元数据里的 dependencies运行依赖则通过 AgentContext 按需获取ProviderPlugin 和 MemoryPlugin 都是延迟实例化的。内核在首次调用对应能力时才触发初始化并把初始化过程也记录成事件确保回放时能重建完整的生命周期。这个设计看似绕了一圈但避免了两难既保证了加载时的显式校验又不会把启动时间拖到不可接受。4.2 事件序列化的三个陷阱要把运行数据完整落盘序列化是逃不过的坎。这里我踩过三个坑逐个说。第一个是函数和方法引用。插件内部为了复用常常会写出self._parser self._wrap_parser(some_callable)这样的代码如果直接把对象 dump 到 JSON自然会失败。我的规则是日志里只存可解析的纯数据函数和类一律存成规范名比如plugins/parser/default.py这样的路径回放时从注册表重新解析。第二个是字节串和二进制内容。工具的返回值偶尔会包含图片字节或文件内容直接塞进 JSON 会爆炸。统一策略是把二进制字段编码成 base64同时增加encoding标记。虽然会膨胀体积但换取的是回放时能还原出所有 payload值这个价。第三个是时间格式。Python 的默认 datetime 序列化结果并不跨语言兼容我统一使用 ISO 8601 字符串并且一律存 UTC。别小看这个细节跨时区调试时它能把人逼疯。还有一个更隐蔽的坑payload 里偶尔会出现不可序列化的自定义对象。我选择直接抛错而不是静默丢弃因为日志丢数据就意味着回放时信息不完整这种静默问题比报错更可怕。4.3 异步调度下的因果顺序怎么还原Agent 运行时内部充满了并发多个模型请求同时发出、多个工具调用并行执行。事件写入的物理顺序和逻辑发生顺序不一定一致这给回放带来一个很头疼的问题——如果只按ts排序回放出来的状态可能与真实执行状态完全不同。我用两把锁解决。一把是写入锁所有事件在到达日志层时都经过单线程的事件循环写入顺序就是调度器提交顺序另一把是因果引用每个事件记录causal_parent指向逻辑上触发它的上一个事件。这样即使物理顺序偶有抖动回放引擎也能根据因果链重建准确的执行次序。在实际实现里我没有做真正的并发重放——回放引擎是纯串行执行状态转移因为目标是稳定可对比而不是模拟性能。串行保证同样的日志永远回放出一模一样的结果这对测试可靠性至关重要。4.4 长会话与日志文件的治理生产环境跑长了日志文件会膨胀到吓人。一个活跃的多轮会话每步平均 5-10 个事件每个事件带完整上下文万步之后轻松超过 100MB。三个约束是必须的轮转、快照压缩、清理钩子。events.jsonl我按 100MB 轮转旧段自动归档到archive/快照不是每步都写而是按步数阈值触发并做 gzip 压缩。另外插件卸载的时候必须执行清理钩子——比如某些工具会起子进程或打开临时文件如果日志系统不感知这些资源长会话跑完总会残留一堆孤儿进程。还有一个容易被忽略的点是脱敏。会话日志里包含用户输入和工具返回这些内容很可能涉及个人信息。我在日志 sink 层做了脱敏管线把手机号、邮箱、地址等模式替换成占位符之后再落盘。这件事一定要在前置链路做等日志已经写进文件再处理又慢又容易漏。5. 接入成本、实测数据与后续扩展5.1 把已有 Agent 迁移到 Harness 要改什么这套框架不是玩具我不希望它只活在 demo 里。为了验证可用性我实际把一个线上多轮客服 Agent 迁移了过来迁移步骤非常收敛大致是四条把现有的工具函数封装成 ToolPlugin实现run和to_protocol_spec两个方法把模型调用层替换成 ProviderPlugin——DeepSeek Harness 默认的 Provider 已经实现了 DeepSeek 函数调用协议大部分情况下不用动把会话状态存储接到 StatePlugin替代原来的 Redis key 散落存储在启动脚本里加载插件目录指定日志目录。迁移后调用侧 API 长这样import asyncio from harness import Harness, ReplaySession, PluginRegistry async def main(): registry PluginRegistry().load_from(Path(./plugins)) harness Harness(registryregistry, log_sinkfile://sessions) agent harness.build_agent(customer_service) reply await agent.run(请帮我查询上周订单的物流状态) print(reply) session_path harness.current_session_path() replay ReplaySession(session_path) trace replay.replay_until(step42) print(trace[session_state]) asyncio.run(main())整个迁移花了两个工作日其中大部分时间在填工具返回格式的历史债插件协议本身几乎没有带来额外负担。5.2 日志写入与回放的实测开销我在做框架时最担心的就是留痕会不会拖垮主流程。实测下来在某个内网压测环境里单条事件的日志序列化和追加写入延迟 p99 约 6ms对动辄几百毫秒的模型调用来说完全可忽略。磁盘占用方面一个包含完整输入输出的会话大约每万步消耗 220MB 空间但这个数值高度依赖模型输出长度和工具 payload 大小。回放速度则明显快于实时在关闭外部调用的离线回放模式下跑完一个 200 步的会话只需约 2 秒大约每秒 100 步左右。主要瓶颈不是事件解析而是快照重建和状态序列化的 Python 开销。这个速度对常规排查和回归测试已经完全够用。有一点要提醒以上数字是在固定机器上的经验值不同机器和不同负载下差异很大。如果你们有性能敏感场景建议把日志写入做成异步队列但要注意队列积压时不能丢事件——我最终用了带背压的有界队列宁可让主流程等一等也不丢一条决策链路上的证据。5.3 接下来想做的几个方向框架跑起来之后我把视线放到了下一步。最想做的有三件事第一是回放可视化把事件流渲染成时间线界面支持在某个 step 上直接展开上下文和工具入参出参这个 UI 对非技术同事排查问题会非常友好第二是回放结果自动 diff 的可视化让 Prompt 团队每周能自动看到行为漂移报告第三是插件仓库的格式规范化让团队内部可以像装 pip 包一样分发已验证的插件减少重复造轮子。另外脱敏能力也想做得更成熟一些。目前的模式匹配只能挡一部分个人数据后续打算引入更细粒度的字段级脱敏配置让每个插件都能声明自己的敏感字段规则。回放日志和插件协议是这套框架里最值得先抄作业的部分。它们看起来是两个独立特性实际上互相成就插件协议提供了清晰的边界日志才能准确地记录边界之间的每一次数据交换回放能力让插件的每次改动都可以被验证插件才能放心地频繁迭代。如果你也要做一个 Agent 运行时我强烈建议把这两件事当成同一个架构问题来设计而不是当成两个后补的增强功能。最后分享一个我从这套框架上获得的额外体会插件协议里运行数据和控制数据一定要分开。运行数据是给模型看和给工具算的数据控制数据是框架自身调度用的元数据。两者混在一个 schema 里前期写着省事后期做日志审计和权限控制时就会处处别扭。我第一版就混得很开心后来重构时花了一整天才拆干净。这个坑越早避开越好。