做Agent应用做得久了你会越来越认同一件事Agent框架的工程化难度从来不在“怎么调用大模型”那一层而在“怎么把模型调用、工具执行、上下文状态、日志审计串成一套真正能上线、能迭代、能排障的工程系统”。我最近基于DeepSeek模型完整搭了一套Agent运行框架代号DeepSeek Harness。这套框架最核心的两个设计压箱底全插件化的能力装配以及可回放的会话日志。这篇文章就把这套框架从设计动机到落地代码完整解剖一遍包括为什么用插件化而不用传统模块化、会话日志为什么要做成可回放的、核心代码怎么组织、关键参数怎么取舍最后再把我实际踩过的几个坑一并交底。1. 项目背景与整体设计思路1.1 为什么是DeepSeek模型底座选型的三个工程维度选底座模型我从来不看榜单分数只看三个工程维度可控性、成本、结构化输出能力。DeepSeek在这三个维度上的综合表现最均衡所以才拿它作为Harness的默认底座。首先是可控性。Agent框架意味着模型要反复执行工具调用而且往往要处理企业内网数据。DeepSeek支持私有化部署数据不出内网这对工具类、企业内部类Agent应用是硬性要求。其次是成本。Agent和普通聊天不一样一轮对话可能要触发好几次模型推理token消耗是乘数级的推理成本直接决定了一个Agent功能能不能规模化跑起来。最后是结构化输出能力。Agent的命门是function calling——模型能不能稳定地输出结构化的tools_call决定了整个Harness的工具调度能不能成立。DeepSeek在这块的输出格式稳定度实测下来是够用的而且它对复杂指令的遵循能力在长链路工具调用场景下没有明显掉链子。不过要说明一点Harness本身并不绑定DeepSeek。模型适配层在框架里是独立的一层底层换模型只需要换一个adapter实现。其实这正是插件化思路的第一步——连模型本身都是一种可替换的组件。1.2 从胶水代码到框架三个被逼出来的设计目标最开始我写Agent应用也是典型的胶水代码一个handle_message函数里面先调模型再判断要不要走工具走完工具再把结果拼回去所有逻辑全揉在一起。这种代码跑demo没问题一旦认真用起来三个问题立刻暴露。第一个是扩展性。想加一个“物流查询”工具就得在handle_message里加一个if 查物流 in user_message的分支再加一个“库存查询”又得加一个分支。分支越来越多到最后没人敢动主流程因为任何一行改动都可能影响所有历史功能。第二个是调试性。Agent跑完七步工具调用第三步返回了空结果最终输出就开始胡说八道。这时候没有任何中间日志根本不知道问题是模型理解错了还是工具返回错了还是上下文拼接出了问题。纯靠肉眼盯着控制台猜效率极低。第三个是审计性。Agent对外提供服务之后一旦出了问题怎么证明是模型的问题、工具的问题还是系统故意输出了违规内容普通文本日志只记录了“发生了什么”没有记录“当时完整的状态和因果链路”根本经不起复盘。这三个痛点逼出了Harness的两个核心设计目标能力全部插件化让系统边界清晰、可装配会话日志可回放让任何一次运行都能被完整还原。1.3 Harness整体架构编排核心、插件注册中心与事件总线Harness的整体架构是分层的从外到内大致是接入层、编排核心、插件注册中心、模型适配层、存储层外加一条贯穿全程的事件总线。接入层负责会话API接收用户消息、返回Agent回复编排核心是HarnessRuntime负责执行主循环——把用户消息送进插件链、调用模型推理、处理工具调用、把最终回复送回用户插件注册中心管理所有插件按声明式配置装配模型适配层是对DeepSeek API的统一封装处理流式输出和工具调用解析存储层是会话日志库记录所有事件并支持回放。这个结构可以用一个很朴素的类比来理解像一支标准化的后厨团队。接入层是前台编排核心是后厨总调度插件是各个工位模型是灶台日志是全流程的监控摄像头。每个工位只干自己那一件事流程由后厨统一调度所有操作都有录像出了任何问题都能倒带重看。2. 全插件化设计让Agent的能力边界可生长、可治理2.1 模块与插件的本质区别为什么Agent能力必须运行时可装配很多同学会问模块化也能解耦为什么非要做成插件化“模块”和“插件”的本质区别在于装配时机。模块化是在编译期或导入期就确定关系的一个模块import另一个模块这是硬依赖。插件的装配发生在运行时通过注册中心动态绑定依赖关系是软的、由框架注入。打个比方模块是焊死在电路板上的元件插件是效果器踏板——踩一脚就接入再踩一脚就走旁路。Agent的能力面恰恰是高度动态的。今天加一个联网搜索明天改一个记忆策略后天加一个内容过滤。插件化意味着这些能力变更只需要动配置、动注册不需要改主流程代码。更重要的一点是可测试性跑单元测试时只加载必要插件不加载真实工具插件而加载一个mock工具插件做回归时又换回真实插件。这在模块化体系下是很难做到的。2.2 插件基类与生命周期五个阶段定义清楚插件的边界插件不能是“想怎么写就怎么写”必须有一份严格的生命周期契约。Harness里定义了BasePlugin抽象基类每个插件都遵循同样的生命周期注册、加载、就绪、执行、卸载。先看上下文对象和插件基类的代码from abc import ABC, abstractmethod from dataclasses import dataclass, field dataclass class HarnessContext: session_id: str config: dict state: dict field(default_factorydict) tool_results: list field(default_factorylist) messages: list field(default_factorylist) class BasePlugin(ABC): name: str base version: str 1.0.0 priority: int 100 async def on_load(self, ctx: HarnessContext) - None: 插件被装配时执行做资源初始化 async def on_unload(self, ctx: HarnessContext) - None: 插件被摘除时执行做资源释放 abstractmethod async def before_step(self, ctx: HarnessContext, user_message: str) - tuple[str, bool]: 模型推理前执行返回处理后的消息和是否阻断后续插件 ... abstractmethod async def after_step(self, ctx: HarnessContext, response: str) - str: 模型输出后执行返回处理后的响应文本 ...生命周期五阶段的职责划分注册插件实例被放入注册中心框架校验name和version是否合法。加载根据配置实例化插件对象此时还没有任何资源被占用。就绪调用on_load插件在这里初始化自己的资源比如打开数据库连接、加载词表、初始化内存索引。执行每一轮对话都会调用before_step和after_step这是插件真正干活的阶段。卸载调用on_unload释放资源。系统关闭时框架会逆序执行所有插件的卸载。这个生命周期设计的核心思想是把插件的“有状态”压缩到最小范围。插件内部尽量保持无状态如果确实需要状态必须显式挂在HarnessContext上按session_id分桶。这一点在后面“常见问题”里会重点展开。2.3 插件注册中心与YAML声明式装配改配置就能加能力注册中心是整个插件化设计的枢纽。它维护一个插件实例表并且按priority排序决定执行顺序。class PluginRegistry: def __init__(self): self._plugins: dict[str, BasePlugin] {} self._order: list[str] [] def register(self, plugin: BasePlugin) - None: self._plugins[plugin.name] plugin self._order.append(plugin.name) self._order.sort(keylambda n: self._plugins[n].priority) def get(self, name: str) - BasePlugin: return self._plugins[name] def snapshot(self) - list[BasePlugin]: return [self._plugins[n] for n in self._order]注册中心本身不决定装哪些插件它只负责“装配”这件事。真正决定装哪些插件的是一份YAML配置plugins: history_memory: enabled: true priority: 10 params: max_turns: 20 safe_filter: enabled: true priority: 20 web_search: enabled: true priority: 30 params: timeout: 5启动时框架读取配置逐个实例化启用的插件并注册。要停用一个能力把enabled改成false要调整执行顺序改priority。主流程代码一行都不用动。这就是“声明式装配”——系统的能力边界通过描述文件表达而不是散落在代码里。这里有一个很重要的设计决策Harness第一版不追求“热插拔”也就是运行时不动态加载/卸载插件。原因很简单——热插拔需要处理插件状态迁移、运行中资源释放、并发请求中断等一系列复杂问题在单机版里收益很小。配置化装配已经解决了90%的可扩展性诉求剩下的复杂度不值得引入。2.4 插件组合原则哪些逻辑适合插件化哪些必须留在主流程插件虽然灵活但不是什么逻辑都适合塞进去。我自己总结了一份边界清单适合插件化的逻辑横向切面日志统计、限流、内容过滤、记忆注入。独立专业能力工具执行、知识库检索、回复格式化。可替换的策略重试策略、摘要策略、上下文裁剪策略。不适合插件化的逻辑模型协议适配这是框架核心插件化会引入不稳定因素。主流程状态机用户消息如何进入、最终回复如何返回这些是骨架不能动。会话路由与生命周期管理属于基础设施插件化会让排障变得复杂。判断标准很简单插件只处理“能力”不处理“流程”。如果一段逻辑会影响主流程的走向、中断或者跳转它就应该留在编排核心如果一段逻辑只是“在某个节点做一件事”那才适合变成插件。3. 可回放会话日志把黑盒变成时间旅行调试器3.1 普通日志与可回放日志的本质区别从拍照片到全程录像很多人把“打日志”理解为“print”这是Agent框架里最大的误区。普通日志相当于事后拍照片——每张照片记录了一个瞬间发生了什么但照片之间没有关联当时的上下文状态、因果顺序全部丢失。可回放会话日志是全程录像。它不只记录“发生了什么”还记录“当时的状态是什么”“这个事件是由哪个事件触发的”“执行完以后状态变成了什么”。有了这些信息你可以把任意一场会话像视频一样倒回、逐帧播放、在某一步暂停查看状态。这个区别对Agent框架来说特别关键。因为Agent的行为不是单一请求而是一连串推理、工具调用、再推理的循环。中间任何一步的状态偏差都会被后续步骤放大。普通日志只能告诉你“第三步挂了”可回放日志能告诉你“第三步为什么挂——因为第二步返回里的某个字段被插件改了”。3.2 会话日志数据模型事件、序号与状态增量可回放的底层支撑是数据模型。Harness的会话日志采用三层结构Session会话、Turn一轮交互、Event事件。事件是核心它的结构如下dataclass class AgentEvent: session_id: str seq: int # 会话内全局自增序号 ts: float # 毫秒时间戳 kind: str # user_message / model_request / tool_call / tool_result / final_response / state_delta payload: dict # 事件内容如请求体、工具名与入参、响应文本 state_delta: dict # 执行前后的状态变化 parent_seq: int | None # 关联的父事件用于链路追踪为什么需要seq和parent_seq因为纯靠时间戳无法表达因果顺序——同一毫秒内完全可能发生多个事件而Agent的因果链是关键是“模型生成了工具调用请求”才导致“工具执行”是“工具返回结果”才导致“模型下一轮推理”。parent_seq把这层父子关系显式记录下来回放时可以构建出完整的调用树。事件存储用JSONL格式一行一个事件天然支持追加写入{session_id: s_01, seq: 1, ts: 1690000000123.4, kind: user_message, payload: {text: 帮我查一下订单状态}, state_delta: {}, parent_seq: null} {session_id: s_01, seq: 2, ts: 1690000000156.7, kind: model_request, payload: {messages_len: 5}, state_delta: {}, parent_seq: 1} {session_id: s_01, seq: 3, ts: 1690000000234.5, kind: tool_call, payload: {tool: query_order, args: {order_id: A123}}, state_delta: {}, parent_seq: 2}这里有一个细节为什么事件里不保存全量状态而只保存state_delta因为全量状态太大而且大部分字段在每步之间根本没变。记录增量既能压缩体积又能在回放时一步步重建出任意时刻的完整状态——这正是“录像”的本质。3.3 回放执行器observe与restore两种模式事件数据有了回放执行器就是把这串数据重新“播放”出来的引擎。class SessionReplay: def __init__(self, events: list[AgentEvent]): self.events sorted(events, keylambda e: e.seq) def step(self, seq: int) - AgentEvent: return self.events[seq - 1] def run(self, mode: str observe): for event in self.events: if mode observe: self.observe(event) elif mode restore: self.restore(event) def observe(self, event: AgentEvent) - None: 只读地遍历事件恢复当时的输入输出 print(f[{event.seq}] {event.kind}: {event.payload}) def restore(self, event: AgentEvent) - None: 重建状态机恢复到某一时刻后继续执行 self.state.apply(event.state_delta) if event.kind tool_call: # 还原模式下不真正调用外部API而是读取录音数据 recorded_result self.recorded_results.get(event.seq) self.context.inject_tool_result(recorded_result)回放模式分两种observe模式只读地按顺序遍历所有事件把当时的输入、输出、状态变化打印出来。适合审计和排障——不重新执行任何逻辑只是“倒带看录像”。restore模式把状态恢复到某个时间点然后让框架重新执行后续步骤。适合复现问题——比如用户报“第四步开始回答得不对”就把状态恢复到第三步结束时用当时的上下文重新跑看问题出在哪里。restore模式有个关键约束幂等性。重放时绝对不能重新调用外部API——外部数据已经变了重新调用得到的结果和当初不一样回放就失真了。所以在还原模式下工具执行器必须被注入一个“录音数据源”所有工具调用结果都从历史事件里读取而不是重新发起真实请求。这一点是回放系统真正能落地的分水岭。3.4 回放日志的四类工程应用排障、回归、成本与审计可回放日志不是摆设它在Harness里至少有四类明确的工程用途线上问题排查。用户说“刚才那轮回复不对”拿到session_id把会话日志回放一遍看是工具返回错了、模型理解错了还是插件改坏了上下文。整个排查过程从“猜”变成了“看”。回归测试。每次框架升级、插件调整之后把历史会话作为测试集跑一遍回放对比新旧行为差异——某个插件改动有没有影响其他插件的执行结果一跑便知。成本与Token分析。从model_request事件里提取每次请求的token数就能精确统计每一场会话消耗了多少token、哪一个工具调用占用了大头然后针对性优化prompt或裁剪逻辑。合规审计。完整的因果链、不可变的事件序列、可校验的状态增量对外提供证据链条时非常有用。4. 实操过程与核心环节实现Harness核心代码走读4.1 技术选型Python asyncio、JSONL与SQLite的组合Harness的技术栈非常简单Python 3.11、asyncio、openai库走DeepSeek既兼容的接口、PyYAML、aiosqlite。为什么选Python asyncio而不是别的因为Agent的整个流程几乎全是IO等待——等待模型推理返回、等待工具调用返回、等待日志落盘。协程在这种场景下性价比最高单机就能扛千级并发又不用像多进程那样处理共享内存和队列的复杂度。存储层为什么是JSONL SQLite的组合JSONL负责记录全量事件追加写、天然适配日志场景SQLite只存会话的索引信息——session_id、起始时间、事件数量、最终状态方便快速定位一场会话在哪个文件、哪一段。单体阶段完全够用没必要上消息队列或时序数据库。工程上有一个原则我一直遵守不为不存在的复杂度买单。环境准备很简单python -m venv .venv source .venv/bin/activate pip install openai1.30.0 httpx pyyaml aiosqlite4.2 Harness核心运行循环从消息进入到最终回复的完整链路核心是handle_message方法它完成整个Agent主流程async def handle_message(self, session_id: str, user_message: str) - str: ctx self._get_context(session_id) # 1. before_step让每个插件有机会改写输入或注入信息 for plugin in self._registry.snapshot(): user_message, blocked await plugin.before_step(ctx, user_message) if blocked: return self._build_blocked_response(user_message, plugin) # 2. 调用模型适配层拿到本轮回复 response await self._model_adapter.chat( messagesctx.messages, toolsctx.get_available_tools(), ) # 3. 处理工具调用模型可能要求执行多个工具 while response.tool_calls: for tc in response.tool_calls: result await self._execute_tool(tc) ctx.tool_results.append(result) self._record_event(tool_result, { name: tc.function.name, result: result, }) response await self._model_adapter.chat( messagesctx.messages self._tool_result_messages(), ) # 4. after_step每个插件处理最终回复 final_text response.content or for plugin in reversed(self._registry.snapshot()): final_text await plugin.after_step(ctx, final_text) self._record_event(final_response, {text: final_text}) return final_text几个关键点值得展开。before_step先按升序遍历插件每个插件可以改写用户消息也可以阻断执行。设计上把“是否阻断”作为返回值而不是异常是为了让过滤逻辑用一种非常直接的方式表达——内容不合规就直接短路不再消耗模型推理。while response.tool_calls是工具调用的核心循环。模型一次返回可能要求执行多个工具执行完工具之后模型又可能基于结果发起新的工具调用所以必须用while兜底直到模型不再请求工具为止。为什么要给工具调用链设置最大深度因为模型在复杂任务中可能陷入工具调用的死循环——调了A工具、再调B工具、又调A工具。没有阈值约束token会被无限消耗。after_step用reversed遍历这是经典的洋葱模型。before_step按priority从小到大执行after_step按从大到小执行——先执行的后处理越晚声明的插件离用户越近。4.3 SessionRecorder与回放工具记录与还原的落地代码会话记录器是最贴近“可回放”这个目标的组件class SessionRecorder: def __init__(self, log_path: str, index_db: str, flush_every: int 50): self._writer JsonlWriter(log_path, flush_everyflush_every) self._index SqliteIndex(index_db) async def append(self, event: AgentEvent) - None: self._writer.write(event) if self._writer.should_flush(): await self._writer.flush() await self._index.upsert_session(event) async def replay(self, session_id: str) - SessionReplay: events await self._writer.read_session(session_id) return SessionReplay(events)flush_every50的意思是攒够50条事件或超过2秒就批量刷一次盘。为什么要这样因为每条事件都立刻调用flush会把磁盘IO变成整个系统最大的瓶颈。批量刷盘在故障场景下最多丢失最后50条事件对Agent排障来说完全可以接受但性能收益是数量级的。4.4 关键参数参考表超时、深度、并发与刷盘Harness里最值得调优的参数我整理了一张表参数推荐值说明每轮模型调用超时60秒留足流式输出的余量工具调用链最大深度5防止模型陷入工具调用死循环会话并发数上限100单机协程加SQLite写入可控日志刷盘阈值50条 / 2秒平衡可观测性与性能上下文保留轮数20轮超出后由记忆插件做摘要压缩回放最大会话数5000SQLite索引上限超出走归档这些参数不是拍脑袋定的。比如工具调用链深度设为5是因为实测大多数Agent任务在3到4次工具调用内就能收敛超过5次基本说明模型在绕路太浅会打断复杂任务太深会放大token成本。比如上下文保留轮数设为20轮是因为超出后请求的token数急剧上升但收益递减——这时候应该触发记忆插件的摘要压缩而不是继续塞原始消息。5. 常见问题与排查技巧实录实测五类高频坑5.1 插件状态串扰一场会话的数据串到另一场这是一个非常隐蔽的坑。现象是A用户的消息在B用户的会话里出现了两个会话的数据串了。排查方式把两个会话的日志回放出来看state_delta。结果发现某个插件在before_step里改动了一个类属性列表而这个列表是所有会话共享的。根因是插件把可变状态放在了实例属性甚至类属性上# 错误做法所有会话共享一份history class HistoryPlugin(BasePlugin): def __init__(self): self.history [] # 正确做法状态必须挂在ctx.state上按session_id分桶 class HistoryPlugin(BasePlugin): async def before_step(self, ctx, user_message): bucket ctx.state.setdefault(self.name, {}) history bucket.setdefault(ctx.session_id, [])教训是插件内部不允许持有可变的长生命周期状态。任何跨轮次、跨会话的数据都必须放在HarnessContext里由框架统一管理和隔离。5.2 回放结果与线上不一致幂等性被破坏现象重放一场历史会话结果和线上当初完全不一样甚至出现了线上没有的错误。根因回放时某个插件真的又调了一次外部API或数据库——比如搜索插件重新执行了网络请求而搜索结果的排序已经变了。解决回放系统必须默认走“录音数据”。在restore模式下工具执行器被替换成ReadonlyStore所有工具调用都从历史事件里读取当时的结果绝不发起真实请求。同时给插件上下文注入replay_mode标志插件检测到该标志时主动禁用副作用操作。这个设计是整个回放系统可信度的地基——如果回放结果不能精确还原当初那回放就没有意义。5.3 上下文被插件塞爆工具返回不做压缩的代价现象Agent跑了几轮之后请求token数飙升响应越来越慢单轮成本也肉眼可见地涨。排查方法看会话日志里model_request事件的messages_len曲线发现某次工具调用之后消息体积突然暴涨。根因工具插件把完整工具返回原样拼进了模型消息。比如一次网页搜索返回了几万字的原文全被塞进了上下文。解决在工具插件返回结果给模型之前做截断。Harness里专门写了一个ContextTrimmer工具按比例截断超长文本或者提取关键段落再送进模型。另外针对“上下文窗口被撑爆”的问题记忆插件会在轮数超过阈值后触发摘要压缩把历史消息压缩成一段摘要再继续。5.4 日志写入拖慢主流程IO与事件循环的拉扯现象单机压到100并发时整体延迟从300ms涨到了800ms。排查方法用profiler看热点发现SessionRecorder的写入占了大量时间。根因初版实现里每条事件都同步写盘然后立刻flush磁盘IO阻塞了事件循环。解决改成异步批量写入每50条或2秒刷一次盘事件先写内存缓冲由独立任务负责落盘。在内存缓冲区接近满的时候降级——只保留关键事件类型丢弃统计类事件保证主流程不因日志排队而卡死。这个坑的教训是日志系统是给主流程服务的不是反过来。日志性能再差也不能拖垮业务延迟。5.5 插件顺序导致行为不可预期优先级与洋葱模型现象两个插件都想改最终回复文本——一个是格式化一个是内容过滤——执行顺序不同结果截然不同。根因没有明确约定插件的执行顺序各插件声明的优先级混乱。解决Harness明确了两条规则。before_step按priority升序执行after_step按priority降序执行优先级通过YAML配置调整不写死在代码里。同时在会话日志里记录每个插件实际执行的顺序排查时逐帧核对能清楚看到哪个插件先处理、哪个插件后处理。5.6 常见问题速查表五个坑一句话版问题根因解法会话数据串扰插件持有共享可变状态状态挂ctx.state按session_id分桶回放结果不一致回放时重新调用了外部API回放模式注入ReadonlyStore上下文被塞爆工具返回未做压缩截断超长文本、触发摘要插件日志拖慢主流程同步写盘阻塞事件循环异步批量写入 flush阈值插件顺序不可控没有优先级契约before升序、after降序、YAML配置priority写在最后一点实际体会这套Harness框架不是一版设计出来的是改了五轮之后沉淀下来的。最早连日志都没有出问题只能靠肉眼盯控制台后来加了插件机制能力边界清晰了最后补上回放系统才真正敢放开手迭代。现在每次改完插件第一件事就是跑一遍历史会话回放看行为有没有漂移这套习惯帮我挡掉了大量回归问题。如果让我重写一遍我开工第一天就会定三件事按事件模型设计日志、插件不持有任何全局状态、回放默认走录音数据。这三个决定越早做后面的工程化改造就越省力。DeepSeek Harness这套做法不是什么银弹它只是把Agent开发的“感觉”变成了“可观察、可控制、可重现”的工程过程。下一步我打算做可视化回放界面把事件流渲染成时间轴再叠加一个Shadow灰度模式——新插件先并行走一段对比行为差异再灰度上线。工程化这条路远没到终点。