1. 从"能跑"到"能查":Agent 框架的工程化分水岭
做过 Agent 项目的人大概都有过这种体验:Demo 阶段一切丝滑,工具调用、多轮对话、任务规划全都跑得通,可一旦上线或者交给别人复现,问题就来了——某次会话里模型为什么选了那个工具?中间那步参数是怎么拼出来的?同样的输入昨天跑得对今天跑不对,到底差在哪?这些问题靠打印几行日志根本回答不了。
我最近在拆解一套 Agent 框架的工程化设计,核心关注两个点:全插件化和可回放会话日志。前者解决的是"能力怎么装、怎么换、怎么隔离",后者解决的是"过程怎么存、怎么查、怎么重演"。这两件事听起来像是架构层面的抽象话题,但实际落到代码里,直接决定了一个 Agent 框架是玩具还是工具。标题里提到的 DeepSeek Harness,我理解它代表的是一类"把模型能力用工程手段兜住"的思路——模型负责决策,框架负责把决策过程变成可观测、可复现、可替换的工程对象。
这篇文章适合两类人看:一类是正在自己搭 Agent 框架、被状态管理和工具注册搞得头大的开发者;另一类是接手了别人 Agent 项目、面对一堆黑盒调用无从下手的维护者。我会把插件化设计的取舍、会话日志的数据结构、回放机制的实现要点拆开讲,尽量给到能直接抄的思路和代码骨架。需要说明的是,标题只给了一个方向,下面涉及的目录结构、字段设计、参数计算都是基于常见工程实践做的合理补全,不是某个具体项目的内部文档。
2. 全插件化设计:为什么 Agent 的能力必须"可插拔"
2.1 插件化到底在解决什么问题
先想清楚一件事:Agent 和普通程序最大的区别是什么?普通程序的执行路径是写死的,Agent 的执行路径是模型在运行时决定的。这意味着你没法在编译期就知道它会调用哪些能力。如果所有工具、所有模型、所有记忆模块都硬编码在主流程里,那每加一个工具就要改核心代码,每换一个模型就要动一遍调用链,维护成本会指数级上升。
插件化的本质是把"变化的部分"从"稳定的部分"里剥出来。稳定的部分是 Agent 的主循环——接收输入、调用模型、解析动作、执行、回填结果、判断是否结束。变化的部分是具体用什么模型、有哪些工具可用、记忆怎么存、日志往哪写。把这些变化点定义成标准接口,主循环只依赖接口,具体实现通过注册机制注入,这就是插件化的核心。
我见过不少项目嘴上说插件化,实际是把所有工具塞进一个巨大的 if-else 或者字典里,改一个工具要动三个文件。真正的插件化应该做到:新增一个工具只需要新增一个文件加一行注册,不碰主循环;替换一个模型只需要换一个实现类,调用方无感知。
2.2 插件接口的粒度设计
接口粒度是个容易踩坑的地方。设计太粗,一个插件管一堆事,复用性差;设计太细,插件之间依赖复杂,装配起来头疼。我的经验是按"职责单一"来切,Agent 框架里通常需要这几类插件:
| 插件类型 | 职责 | 典型实现 |
|---|---|---|
| Model Provider | 封装模型调用,统一输入输出格式 | 不同模型服务的适配层 |
| Tool | 提供具体能力,声明参数 schema | 搜索、计算、文件读写等 |
| Memory | 管理上下文和历史 | 滑动窗口、向量检索、摘要压缩 |
| Logger | 记录执行过程 | 文件、数据库、消息队列 |
| Parser | 解析模型输出为结构化动作 | JSON 解析、函数调用解析 |
每一类插件定义一个抽象基类或者协议接口,主循环只认接口。这里有个关键决策:接口用抽象基类还是用 Protocol/鸭子类型。抽象基类的好处是强制子类实现必要方法,运行时能提前发现缺失;Protocol 的好处是解耦更彻底,第三方实现不需要继承你的基类。我倾向于核心接口用抽象基类保证契约,扩展点用 Protocol 保持灵活。
2.3 注册与发现机制
插件写好了,怎么让框架知道它的存在?常见有三种方式:
第一种是显式注册,在启动时手动把插件实例注册到注册表里。优点是清晰可控,缺点是新增插件要改启动代码。第二种是装饰器注册,插件类上加个装饰器,导入时自动注册。优点是新增插件零改动,缺点是导入顺序会影响注册结果,调试时容易懵。第三种是配置驱动,通过配置文件声明要加载哪些插件,框架动态导入。优点是部署灵活,缺点是配置和代码容易不同步。
实际项目里我一般用装饰器注册加显式覆盖的组合:默认用装饰器自动注册,需要替换或禁用某个插件时通过配置覆盖。这样日常开发省事,特殊场景又能精确控制。
# 插件注册表的简化实现 class PluginRegistry: def __init__(self): self._tools = {} self._models = {} self._loggers = {} def register_tool(self, name): def decorator(cls): self._tools[name] = cls return cls return decorator def get_tool(self, name): if name not in self._tools: raise KeyError(f"tool {name} not registered") return self._tools[name]() registry = PluginRegistry() @registry.register_tool("calculator") class CalculatorTool: def schema(self): return {"name": "calculator", "params": {"expr": "string"}} def run(self, expr): return eval(expr) # 实际项目务必做安全校验注意:上面用 eval 只是为了演示,真实项目里计算类工具必须做表达式白名单或者用安全的解析库,否则就是一个远程代码执行漏洞。
2.4 插件隔离与生命周期
插件化还有一个容易被忽略的点:隔离。如果两个插件共享全局状态,一个插件的异常可能污染另一个。理想情况下每个插件应该是无状态的,需要状态就通过显式的上下文对象传递。工具类插件尤其要注意,别在实例变量里存会话相关的数据,否则并发场景下必然出问题。
生命周期方面,插件通常有初始化、使用、销毁三个阶段。初始化时做重资源准备(比如建立连接、加载模型),使用时保持轻量,销毁时释放资源。框架应该提供统一的钩子,而不是让每个插件自己管理。我踩过的坑是某个插件在初始化时做了耗时操作,导致整个 Agent 启动变慢,后来改成懒加载才解决。
3. 可回放会话日志:让每一次决策都有据可查
3.1 为什么普通日志不够用
普通日志记录的是"发生了什么",比如"调用了搜索工具,参数是 X,返回了 Y"。但 Agent 调试需要知道的是"为什么发生"——模型当时看到的完整上下文是什么?它输出的原始内容是什么?解析器是怎么把原始输出变成工具调用的?这些信息如果不在日志里,事后根本无法还原决策链路。
可回放会话日志的核心要求是:给定一条日志记录,能够完整重建当时的执行状态,并且可以重新执行一遍得到相同或可对比的结果。这比普通日志的要求高得多,它要求日志记录的是状态快照加事件流,而不是简单的文本消息。
3.2 日志的数据结构设计
我把会话日志设计成事件流的形式,每个事件是一个结构化对象,包含时间戳、事件类型、事件数据和上下文引用。事件类型至少覆盖这几类:
session_start:会话开始,记录初始配置和输入model_request:发给模型的完整请求,包括消息列表和参数model_response:模型的原始返回,不做任何加工action_parsed:解析后的结构化动作tool_call:工具调用的名称、参数、开始时间tool_result:工具返回的结果或错误session_end:会话结束,记录最终状态
关键在于model_request和model_response必须记录原始数据,不能只记解析后的结果。因为解析器本身可能有 bug,如果只记解析结果,你永远不知道是模型输出错了还是解析错了。
# 事件对象的结构 class SessionEvent: def __init__(self, event_type, data, parent_id=None): self.event_id = generate_id() self.event_type = event_type self.timestamp = time.time() self.data = data self.parent_id = parent_id # 支持嵌套调用链parent_id这个字段很关键。Agent 执行过程中可能有嵌套,比如一个工具内部又触发了子 Agent,通过 parent_id 能还原出完整的调用树,而不是一条平铺的时间线。
3.3 回放机制的实现要点
回放有两种模式:重演和重放。重演是重新执行一遍,模型和工具都真实调用,适合验证修复后的效果;重放是不调用外部依赖,直接用日志里的记录模拟返回,适合离线调试和回归测试。
重放的实现要点是依赖拦截。在回放模式下,模型调用和工具调用都被替换成从日志读取的桩。框架需要维护一个游标,按顺序匹配当前调用和日志记录。这里有个细节:如果代码逻辑变了导致调用顺序变了,游标匹配会失败,这时候要给出清晰的错误提示,而不是静默返回错误数据。
class ReplayContext: def __init__(self, events): self.events = [e for e in events if e.event_type in ("model_response", "tool_result")] self.cursor = 0 def next_response(self, expected_type): if self.cursor >= len(self.events): raise ReplayError("no more recorded events") event = self.events[self.cursor] if event.event_type != expected_type: raise ReplayError(f"expected {expected_type}, got {event.event_type}") self.cursor += 1 return event.data提示:回放时建议加一个"宽松模式",允许跳过不匹配的事件并记录警告,这样在代码有小改动时也能跑通,不至于完全卡死。
3.4 日志的存储与检索
日志量大了之后,存储和检索就是问题。我的做法是分层存储:热数据(最近几天的会话)放数据库,支持按会话 ID、时间范围、事件类型检索;冷数据归档到对象存储,需要时再拉取。索引至少要建在会话 ID 和时间戳上,如果经常按工具名排查问题,再给工具名建个索引。
检索能力直接决定排查效率。我常用的几个查询场景:查某个会话的完整事件流、查某个工具在时间范围内的所有调用、查所有失败的模型调用。框架最好内置这些查询接口,而不是让使用者自己写 SQL。
4. 工程化落地的核心环节与实操细节
4.1 目录结构怎么组织
插件化和日志系统都要求清晰的目录结构。我习惯这样组织:
agent/ core/ loop.py # 主循环 context.py # 执行上下文 registry.py # 插件注册表 plugins/ models/ # 模型插件 tools/ # 工具插件 memory/ # 记忆插件 loggers/ # 日志插件 session/ event.py # 事件定义 recorder.py # 记录器 replayer.py # 回放器 config/ default.yaml核心原则是core不依赖plugins的具体实现,只依赖接口。plugins之间尽量不互相依赖,需要协作就通过core提供的上下文。这样任何一层都能单独替换和测试。
4.2 上下文对象的设计
执行上下文是贯穿整个会话的对象,它承载了当前会话的所有状态:消息历史、已注册的插件实例、日志记录器、配置参数。设计上下文时要注意两点:一是不可变优先,每次状态变更产生新的上下文快照,方便回放和对比;二是显式传递,不要用全局变量或者线程本地存储,否则回放时会串数据。
class ExecutionContext: def __init__(self, session_id, config, recorder): self.session_id = session_id self.config = config self.recorder = recorder self.messages = [] self.step_count = 0 def with_message(self, message): new_ctx = copy(self) new_ctx.messages = self.messages + [message] new_ctx.step_count = self.step_count + 1 return new_ctx4.3 主循环的实现骨架
主循环是框架的心脏,它的逻辑应该足够简单,简单到一眼能看懂。复杂逻辑都下沉到插件里。
def run_loop(ctx, user_input): ctx = ctx.with_message({"role": "user", "content": user_input}) ctx.recorder.record("session_start", {"input": user_input}) while ctx.step_count < ctx.config.max_steps: request = build_model_request(ctx) ctx.recorder.record("model_request", request) raw_response = model_plugin.invoke(request) ctx.recorder.record("model_response", {"raw": raw_response}) action = parser.parse(raw_response) ctx.recorder.record("action_parsed", action) if action.type == "finish": ctx.recorder.record("session_end", {"output": action.content}) return action.content result = execute_tool(ctx, action) ctx.recorder.record("tool_result", result) ctx = ctx.with_message({"role": "tool", "content": result}) raise MaxStepsExceeded()这段骨架里,build_model_request、parser.parse、execute_tool都是可替换的,模型插件、解析器插件、工具插件分别负责。主循环不关心具体实现,只负责编排。
4.4 参数计算与配置选择
几个关键参数需要根据实际情况计算,不能拍脑袋定:
最大步数:取决于任务复杂度。简单问答 3-5 步够用,多工具协作任务可能需要 15-20 步。设置太小任务做不完,太大浪费资源还可能陷入循环。我的经验是先设 20,观察实际分布再调整。
上下文窗口:模型能接受的最大 token 数减去预留的输出空间。比如模型窗口 32K,预留 4K 给输出,那历史消息最多占 28K。记忆插件要负责在超限前做压缩或截断。
日志采样率:全量记录日志开销不小,高并发场景可以采样。但调试阶段建议全量,因为出问题的往往就是那几条没被采样的。
回放超时:重演模式下要设超时,防止某个工具卡死。一般设成正常执行时间的 2-3 倍。
5. 常见问题与排查技巧实录
5.1 插件注册冲突怎么排查
最常见的问题是同名插件被注册两次,后注册的覆盖了先注册的,导致行为不符合预期。排查方法是启动时打印注册表快照,对比预期。预防措施是注册时检查重名,除非显式声明允许覆盖,否则直接报错。
| 现象 | 可能原因 | 排查方法 |
|---|---|---|
| 工具行为不符预期 | 插件被覆盖 | 打印注册表,检查重名 |
| 插件未生效 | 导入顺序问题 | 检查装饰器是否执行 |
| 配置不生效 | 配置优先级错误 | 打印最终生效配置 |
5.2 回放对不上的定位思路
回放失败通常有三种原因:日志不完整、代码逻辑变了、外部依赖有随机性。定位时先看游标停在哪一步,对比日志记录和当前调用的差异。如果是随机性导致的(比如模型温度参数不为 0),回放时应该强制用日志里的记录,而不是重新调用。
注意:模型调用如果开了随机采样,重演模式下结果必然不同。要保证可复现,要么把温度设为 0,要么在重演时用日志记录替代真实调用。
5.3 日志膨胀的处理
会话多了之后日志会迅速膨胀。我的处理策略是:结构化字段和原始文本分开存,原始文本压缩后存;设置保留期限,过期自动归档;提供按需加载接口,不要一次性把整个会话读进内存。
5.4 几个我踩过的坑
第一个坑是日志记录时机。一开始我在工具执行完之后才记录,结果工具执行到一半崩溃,日志里什么都没有。后来改成执行前记录调用意图,执行后记录结果,这样即使崩溃也能看到意图。
第二个坑是上下文对象被意外修改。早期版本里上下文是可变对象,某个插件偷偷改了消息历史,导致回放时状态对不上。改成不可变之后问题消失,代价是每次变更要复制,但换来的是可预测性,值得。
第三个坑是回放时的副作用。有些工具调用有副作用,比如写文件、发请求。回放时如果不拦截,会重复产生副作用。解决办法是在回放模式下把所有有副作用的工具替换成桩,只返回日志里的结果。
6. 从工程化视角看 Agent 框架的演进方向
把插件化和可回放日志做扎实之后,会发现很多之前觉得难的问题变得可解了。比如 A/B 测试不同模型效果,只要切换模型插件,日志格式不变,直接对比回放结果就行。比如线上问题复现,把会话日志拉下来本地回放,比在线上加日志重新触发高效得多。比如团队协作,新人接手项目时先看几条回放日志,比读文档快得多。
我个人在实际操作中的体会是,Agent 框架的工程化程度,很大程度上取决于它把多少"隐式行为"变成了"显式对象"。模型决策是隐式的,但请求和响应可以显式记录;工具执行是隐式的,但调用和结果可以显式建模;状态流转是隐式的,但上下文快照可以显式保存。每显式化一层,可观测性和可维护性就上一个台阶。
最后分享一个小技巧:如果你正在维护一个已有的 Agent 项目,不用一次性重构。先加日志记录,把关键节点的事件流跑通,再逐步把硬编码的工具抽成插件。日志先行,插件后补,这样每一步都有可验证的产出,风险可控。等日志和插件都到位了,回放机制几乎是水到渠成的事,因为你需要的数据都已经在了。