这段时间在做一个比较复杂的Agent项目模型本身倒没怎么换但工程侧的问题差点让我崩溃。先是工具调用逻辑越写越乱后来加了一堆重试、权限检查、上下文清理的代码整个调用过程变得像一团乱麻一个思路没理清就得上百行胶水代码。后来我认真梳理了一下思路把目光放在了可组合Harness上核心就是用中间件链的方式重新组织Agent的调用流程。这个概念听起来偏架构但落地之后效果立竿见影每个关注点可以独立成组件调用链可以随意插拔、复用、测试Agent的能力边界完全不靠堆代码来实现而是靠“组合”来扩展。这篇文章我把自己的设计思路和实操过程整理出来适合正在做Agent工程化、被复杂调用逻辑折磨过的开发者参考也适合想从单体调用走向模块化控制层的人。1. 从疯狂堆代码到中间件链Agent控制层为什么需要重构1.1 单体Agent调用到底恶心在哪大多数Agent项目起步都很简单把用户输入丢给大模型大模型返回一个JSON里面写了想调什么工具、参数是什么然后代码解析JSON再执行。这套流程跑通demo只需要二三十行代码但一旦进入生产环境问题就全来了。我举几个真实场景需要给每个请求打日志方便事后排查模型到底怎么想的工具调用可能失败网络抖动、参数非法、返回格式异常都需要重试和兜底不同的用户能调的工具不一样需要在调用前做权限校验上下文不能无限膨胀得在每次请求之前裁剪或者摘要遇到上游限流还得设计退避策略。这些需求如果全部写在Agent调用的主流程里你会发现这个主流程越长越复杂最后几乎没法维护。最典型的反面做法是给主函数不断加参数。今天加一个enable_loggingTrue明天加一个max_retries3后天又要加enable_memory标志位。一旦这些开关彼此有依赖代码的可读性就断崖式下降你想理清任何一条执行路径都得从头到尾读一遍。1.2 什么是可组合Harness这里说的Harness并不是测试领域那个测试套件概念而是Agent运行时的一个执行框架。它负责把一次完整的Agent任务包装起来管理从输入到输出的全部过程。你可以把它理解成一个流水线骨架这条流水线上可以挂各种功能模块。而可组合的意思是这个流水线不是写死的而是由很多个小部件临时拼装出来的。每个小部件负责一件事部件和部件之间通过一个共享的上下文对象通信执行顺序由一条链决定。这条链的动态组合能力就是中间件模式的核心价值。拿Web开发来类比非常直观。大家都用过类似Express的路由中间件机制请求从外层流到内层经过日志、鉴权、参数校验、执行业务逻辑再沿原路返回。Agent的Harness完全可以借鉴这个套路只不过我们处理的不是HTTP请求而是“大模型对话轮次”或者“Agent完成任务的一次执行流”。1.3 中间件链对比装饰器和回调钩子的优势有人会问Python里装饰器不也能实现类似功能吗确实可以。装饰器适合静态包裹函数但Agent的执行流是动态的不同场景可能需要完全不同的中间件序列。装饰器写死之后很难在运行时重新排序而中间件链是一个显式的列表你可以根据任务类型、用户权限、当前环境构造不同的链这种灵活度是装饰器给不了的。回调钩子也不够用。钩子通常只提供了“在某一步之后做什么”比如on_tool_call、on_llm_response但中间件链提供的是“整个请求处理流程的控制权”。中间件既可以先处理再放行也可以直接短路返回甚至可以替换最终的输出。这种控制力度回调钩子很难实现。我最初的项目就是踩了这个坑用回调钩子加装饰器攒了一套逻辑结果到了要加“工具执行完需要把摘要写回上下文再问一次模型”这种跨阶段逻辑时钩子根本接不住只能再改回调顺序。换了中间件链之后这类逻辑变成了一个很普通的中间件放在工具执行环节后面即可。2. 中间件链的核心机制搞懂原理你才能自由组装2.1 一个标准中间件长什么样我们先定义最基本的抽象。一个中间件本质上是个函数它接收两个参数一个是上下文对象一个是next函数。调用next()表示让执行流继续往下走返回之后拿到内层的结果再做后处理。代码写出来是这样的# Python示例标准中间件签名 async def logger_middleware(ctx, next): print(f[start] call_id{ctx.call_id}) try: result await next() print(f[end] cost{result.cost_seconds:.2f}s) return result except Exception as e: print(f[error] {e}) raise这里的ctx就是我们所说的上下文对象。在Harness里它不是简单的字典而是一个有明确字段的数据容器。一般至少包括几个部分输入用户消息、系统提示词、模型状态当前对话消息列表、工具定义和工具执行结果、最终输出、以及一些运行时信息重试次数、调用耗时、日志收集器等。而next对应着链中下一个中间件。如果某个中间件不调用next链路就短路了后面的中间件不会再执行。这是实现权限拦截、缓存命中、条件终止的关键。2.2 洋葱模型与线性握手中间件链的执行顺序有两种常见模型。一种是标准的洋葱模型请求从前向后穿透到最核心的处理函数然后再反向穿过每一层。它的特点是前中间件不仅能看到请求还能看到内层结果。类比快递包装最外层的纸箱先拆里层的泡沫再拆最后看到商品检查完毕再一层层包回去。另一种是线性握手模型每个中间件都做完全独立的处理然后传给下一个链尾是最终清洗和收尾。它不适合需要上下行对称处理的场景。对于Agent Harness我强烈建议用洋葱模型因为很多中间件需要有“前置逻辑”和“后置逻辑”。日志中间件要在调用前记录输入调用后记录结果重试中间件在调用前判断次数调用失败后决定是否再执行一次性能监控中间件则天然依赖后处理阶段计算耗时。如果你用线性模型要么把后处理塞到链尾要么就得把每个中间件拆成两个独立节点组合起来不够优雅。实现洋葱模型的关键在于next不再是普通的函数引用而是一个复合了后续中间件的调用闭包。我们可以把中间件列表倒序组合形成嵌套关系。下面这段伪代码描述了核心逻辑# Harness内将中间件组合成嵌套调用 def build_chain(middlewares, final_handler): acc final_handler for mw in reversed(middlewares): acc wrap(mw, acc) # 生成一个新的函数调用mw时会传入acc作为next return acc直观感受就是最后一个注册的中间件是真正贴着模型调用层的第一个注册的中间件是贴着用户的。2.3 上下文对象的设计决定扩展上限很多中间件之间要共享数据。比如鉴权中间件解析出用户身份后续的工具白名单中间件要拿这个身份去判断模型调用中间件生成的对话记录要被历史摘要中间件读取。如果上下文只是一个dict松散的结构会让中间件之间产生隐式耦合改一个字段名全链崩掉。我更推荐用一个轻量级数据类把关键字段显式定义出来再用一个通用extra字典兜底。这样既能获得类型提示又能保留灵活度。class Context: def __init__(self, user_input, system_prompt): self.call_id generate_call_id() self.user_input user_input self.system_prompt system_prompt self.messages [] # 发送给模型的完整消息列表 self.tools [] # 当前可用的工具定义 self.tool_results [] # 每次工具执行的结果 self.final_answer None # 最终输出 self.metadata {} # 额外运行时信息 def add_tool_result(self, result): self.tool_results.append(result) self.messages.append({role: tool, content: result})显式的messages字段是整个链的生命线。中间件通过操作它就能在模型调用前改上下文工具执行中间件往它里面追加结果下一个模型调用轮次才能看见。设计上下文时最容易踩的坑是让中间件各自维护私有状态。比如重试中间件把重试次数存在自己的局部变量里下一个中间件就感知不到。正确做法是写入ctx.metadata[retry_count]这样日志中间件也能记录。3. 实操从零搭一个可组合的Agent Harness3.1 先定义最小的Harness骨架为了讲清楚实现我用一个非常精简的Python版示例来演示骨架。这里不绑定任何具体的模型SDK只用mock_llm_call函数模拟一轮模型交互。实际的LLM调用在这个函数里就行。# harness.py import asyncio, time, uuid class Harness: def __init__(self, middlewaresNone): self.middlewares middlewares or [] async def execute(self, ctx): # 构造整个链条最内层是模型调用 async def final_handler(ctx): # 这里可以换成真实的大模型调用 result await mock_llm_call(ctx.messages, ctx.tools) ctx.final_answer result.get(answer) return ctx.final_answer chain final_handler for mw in reversed(self.middlewares): chain self._wrap(mw, chain) return await chain(ctx) staticmethod def _wrap(mw, next_func): # 生成一个闭包作为中间件接收的next async def compiled(ctx, nextnext_func): return await mw(ctx, next) return compiled在这个设计里每个中间件都是async def middleware(ctx, next)的签名。Harness负责把它们串起来对外部来说使用者只管维护一个middlewares列表。注意一个细节reversed(self.middlewares)意味着middlewares列表的第一个元素是最外层。也就是说列表顺序和业务直觉一致阅读列表时是从用户视角往下读的。3.2 写几个高频必备中间件日志中间件很好写但有一点容易被忽略不光要记录输入输出还要记录call_id和完整的耗时。生产环境里排查Agent问题把同一call_id的所有中间件日志串起来效率提升非常明显。async def logging_middleware(ctx, next): start time.perf_counter() print(f[log] {ctx.call_id} input{ctx.user_input[:50]}...) try: result await next() finally: cost_ms (time.perf_counter() - start) * 1000 print(f[log] {ctx.call_id} cost{cost_ms:.1f}ms result{result}) return result重试中间件是Agent项目里最能救命的一个。LLM调用经常出现偶发失败比如上游限流、JSON解析错误、工具返回格式异常。重试不能盲目做要区分错误类型网络类可以重试参数类重试无意义。另外一定要加最大次数限制和指数退避否则多个Agent实例同时重试会加剧限流。async def retry_middleware(ctx, next): max_retries int(ctx.metadata.get(max_retries, 3)) base_delay 0.5 for attempt in range(1, max_retries 1): try: return await next() except TransientError as e: if attempt max_retries: raise delay base_delay * (2 ** (attempt - 1)) print(f[retry] attempt{attempt} next_delay{delay:.2f}s error{e}) await asyncio.sleep(delay) ctx.metadata[retry_count] attempt # 实际不会到达这里 raise RuntimeError(unreachable)工具白名单中间件是一个典型的“短路型”中间件。它不修改请求只负责检查当前上下文里ctx.tools是否越权。如果在进入模型前就发现某个工具不该被当前用户调用直接抛出权限异常或者从ctx.tools里移除模型就不会生成违法工具调用了。async def permission_middleware(ctx, next): allowed ctx.metadata.get(allowed_tools, set()) ctx.tools [t for t in ctx.tools if t[name] in allowed] return await next()记忆管理中间件稍微复杂一点。它负责在每次调用模型前检查历史消息是否太多如果超长就做一次摘要把摘要注入到system_prompt里。核心点是摘要内容一定不要覆盖原始细节应该保留关键事实型信息比如用户地址、偏好、最近决定等。3.3 一个完整的工具调用循环中间件链里最核心的其实是“工具调用循环”本身。Agent不是一次模型调用就完事的它可能要循环好几轮模型生成工具调用请求执行工具把结果喂回模型模型再继续生成下一步。这个循环通常被封装成Harness内部的一个agent_loop函数。我的做法是把agent_loop也看作一个特殊中间件放在链的最内层。这样重试、日志、白名单都可以直接作用于整个agent循环而不是某一次模型调用。async def agent_loop(ctx, nextNone): max_steps 5 for step in range(max_steps): response await mock_llm_call(ctx.messages, ctx.tools) if response.get(tool_calls): for call in response[tool_calls]: tool_name call[name] args call[arguments] result execute_local_tool(tool_name, args) ctx.add_tool_result(tool_name_result) # 把工具名为前缀也放进result里便于模型理解 else: ctx.final_answer response[answer] break return ctx.final_answer有一点切身体会工具调用循环里的上下文追加顺序极其重要。工具执行完以后要在本轮的assistant消息后面追加tool消息这样模型才能正确关联工具调用和返回结果。如果你把多轮工具执行的结果一股脑全塞到messages的末尾模型经常会产生幻觉把上一个工具的结果当成下一个工具的输入。3.4 组合示例面向不同场景动态拼链考虑到实操我们做两个组合示例。第一个是“带日志的简单问答”适合调试阶段。第二个是“多轮工具调用权限重试”适合生产环境。# 场景1调试简单问答 harness Harness([ logging_middleware, agent_loop, ]) ctx Context(帮我查一下明天的天气, 你是AI助手) await harness.execute(ctx) # 场景2带权限、重试和记忆管理的生产链 harness Harness([ logging_middleware, permission_middleware, memory_management_middleware, retry_middleware, agent_loop, ])看到没有中间件链的价值就在这个列表里。每加一个能力你不需要改动已有代码只需要在列表里插入一个中间件。比如上线后发现模型老是重复调用某个高风险工具那就加一个rate_limit_tool_middleware放在permission_middleware后面瞬间生效。我没有把工具执行逻辑放进单独的中间件原因是agent_loop已经在一个循环里。如果拆成独立的“工具执行中间件”中间件会有一个循环等待外层再次进入自己设计会很别扭。所以建议是工具执行作为agent_loop内部细节而围绕执行的横切能力用中间件包裹。3.5 参数选型如何决定中间件顺序中间件顺序没有绝对标准但有两条经验性原则。第一越靠近外层的中间件越偏“请求级横切关注点”比如日志、性能、限流。第二越靠近内层的中间件越偏“核心业务语义”比如工具调度、上下文组装、Agent循环。以我的生产链为例顺序是日志最外层必须记录整个流程权限靠近外层尽早拦截非授权请求限流紧跟权限防止恶意/高频请求打到模型重试限流之后重试可以对限流错误恢复记忆管理重试之后上下文清理在进入agent循环前agent_loop最内层这里有一个容易搞反的组合重试应该在限流外层还是内层我的答案是重试在限流外层但限流返回的429错误本身由重试捕获。如果重试在限流内层那么限流器本身如果也失败重试就不会被执行了。这个顺序逻辑需要结合自身场景我说的是通用建议。4. 常见问题与排查技巧这些坑我挨个踩过4.1 中间件顺序写反导致上下文被覆盖我曾经把记忆管理中间件放在了agent_loop之后结果每个轮次准备发送给模型的消息列表都是空。原因就是记忆管理中间件在next()之后执行了裁剪逻辑但裁剪之后ctx.messages还没被下一个小伙伴使用于是白白剪掉了。后面改成记忆管理必须在agent_loop之前加载历史在next()之后把新的对话摘要写回到存储里。两者要拆成两个中间件或者在一个中间件里做前置加载、后置回写。排查这类问题的思路很简单在每个中间件里打印ctx.messages的长度和头尾消息跟踪消息列表是怎么变化的。一旦发现某一步跨过之后数据变了就定位到那个中间件。4.2 异常被“吞掉”的问题洋葱模型有个危险行为如果某个中间件内部的try...except捕获了异常但没有重新抛出并且写了return语句就会导致内层后续中间件的代码被跳过而且外层还以为一切正常。这在重试中间件里最容易发生某次执行抛错你捕获后计数并准备重试但忘了return await next()结果链直接返回了None模型输出就没了。我后来给自己定了一个规则中间件代码里只要捕获了异常就必须明确选择“抛出”或“重试后抛出”绝对不允许静默吞掉。调试期可以在捕获分支里加日志但不能直接裸except写死。4.3 性能损耗别把每个中间件做成一次远程调用中间件链本身是内存内函数调用开销很低。但有些开发者喜欢在中间件里调用外部服务比如鉴权中间件请求鉴权服务记忆中间件请求向量库结果整条链被几十次网络往返拖慢。优化手段一般是复用连接、批量获取、预热缓存。比如在权限中间件里把用户权限一次性放进metadata后续链上其他中间件就不要再去查库了。我测试过一个不过分优化的链6个中间件全部内存操作一个Agent循环的额外开销不到1毫秒。而一旦某个中间件引入外部I/O就可能多出几十毫秒甚至更糟。所以尽量让中间件做纯计算和数据变换把IO隔离在链外或者采用批量执行。4.4 调试的可观测性设计Agent项目的调试比普通后端更麻烦因为中间步骤会调用多个工具、产生多轮对话。我强烈建议给Harness内置一个事件收集器每个中间件可以往ctx.events里面追加结构化事件而不是用print。这样后续可以导出成JSON/CTF格式或者对接前端看板。最简单的实现是def emit(ctx, event_name, **kwargs): ctx.metadata.setdefault(events, []).append({ name: event_name, ts: time.time(), **kwargs, })随后在日志中间件的最外层打印ctx.metadata[events]你就能看到一条完整的执行时间线。包括每次工具调用、每个中间件进入/离开时间、模型请求次数等。这个功能哪怕只用JSON打印排查起问题来已经非常高效了。我曾经靠它定位到一个链路退化问题某类任务每次都会进入一个循环里反复请求同一个查询工具事件里可以清楚看到相同的事件出现了10次。4.5 关于可组合性的第三个原则最小上下文传递最后一个经验是在设计中间件时要尽量只传递和中间件职责相关的数据不要所有中间件都把完整的历史消息列表拿过去又拿回来。例如权限中间件只需要ctx.tools和ctx.metadata中的用户ID就不应该去读ctx.messages。这个原则可以降低中间件之间隐式耦合也让测试单个中间件变得轻松。测试上我习惯用真实的Context对象喂给单个中间件配合一个只做“透传”的next桩函数验证它是否按预期修改了上下文。这种方法比全链路集成测试更稳定也能在改动某个中间件时快速回归。5. 后面还能怎么扩展中间件链一旦稳定下来很多后续能力都变成了往列表里加新节点的事。比如给Agent加上“多工具并行调用”能力可以在agent_loop内对工具调用做并发调度这是一个内部优化又比如做多模型路由可以在最外层加一个router_middleware根据输入难度选择调用不同参数的模型再比如做自动评估在链尾加一个evaluator_middleware用一个小模型给大模型的回答打分记录日志。我个人觉得这种可组合Harness最大的意义不一定是性能提升而是终于把Agent做成了一条清晰可见的生产流水线。以前调代码像在扒一团毛线现在每根线头都在自己的位子上问题定位时间至少缩短了一半。如果你正在做Agent工程化非常建议试一下这个方向不用管框架大小从两三个中间件开始搭感受会很不一样。 
长河君 · 2026/10/22 13:36:11 · 阅读 1.2k 用中间件链重构Agent控制层：可组合Harness的设计与实践
这段时间在做一个比较复杂的Agent项目模型本身倒没怎么换但工程侧的问题差点让我崩溃。先是工具调用逻辑越写越乱后来加了一堆重试、权限检查、上下文清理的代码整个调用过程变得像一团乱麻一个思路没理清就得上百行胶水代码。后来我认真梳理了一下思路把目光放在了可组合Harness上核心就是用中间件链的方式重新组织Agent的调用流程。这个概念听起来偏架构但落地之后效果立竿见影每个关注点可以独立成组件调用链可以随意插拔、复用、测试Agent的能力边界完全不靠堆代码来实现而是靠“组合”来扩展。这篇文章我把自己的设计思路和实操过程整理出来适合正在做Agent工程化、被复杂调用逻辑折磨过的开发者参考也适合想从单体调用走向模块化控制层的人。1. 从疯狂堆代码到中间件链Agent控制层为什么需要重构1.1 单体Agent调用到底恶心在哪大多数Agent项目起步都很简单把用户输入丢给大模型大模型返回一个JSON里面写了想调什么工具、参数是什么然后代码解析JSON再执行。这套流程跑通demo只需要二三十行代码但一旦进入生产环境问题就全来了。我举几个真实场景需要给每个请求打日志方便事后排查模型到底怎么想的工具调用可能失败网络抖动、参数非法、返回格式异常都需要重试和兜底不同的用户能调的工具不一样需要在调用前做权限校验上下文不能无限膨胀得在每次请求之前裁剪或者摘要遇到上游限流还得设计退避策略。这些需求如果全部写在Agent调用的主流程里你会发现这个主流程越长越复杂最后几乎没法维护。最典型的反面做法是给主函数不断加参数。今天加一个enable_loggingTrue明天加一个max_retries3后天又要加enable_memory标志位。一旦这些开关彼此有依赖代码的可读性就断崖式下降你想理清任何一条执行路径都得从头到尾读一遍。1.2 什么是可组合Harness这里说的Harness并不是测试领域那个测试套件概念而是Agent运行时的一个执行框架。它负责把一次完整的Agent任务包装起来管理从输入到输出的全部过程。你可以把它理解成一个流水线骨架这条流水线上可以挂各种功能模块。而可组合的意思是这个流水线不是写死的而是由很多个小部件临时拼装出来的。每个小部件负责一件事部件和部件之间通过一个共享的上下文对象通信执行顺序由一条链决定。这条链的动态组合能力就是中间件模式的核心价值。拿Web开发来类比非常直观。大家都用过类似Express的路由中间件机制请求从外层流到内层经过日志、鉴权、参数校验、执行业务逻辑再沿原路返回。Agent的Harness完全可以借鉴这个套路只不过我们处理的不是HTTP请求而是“大模型对话轮次”或者“Agent完成任务的一次执行流”。1.3 中间件链对比装饰器和回调钩子的优势有人会问Python里装饰器不也能实现类似功能吗确实可以。装饰器适合静态包裹函数但Agent的执行流是动态的不同场景可能需要完全不同的中间件序列。装饰器写死之后很难在运行时重新排序而中间件链是一个显式的列表你可以根据任务类型、用户权限、当前环境构造不同的链这种灵活度是装饰器给不了的。回调钩子也不够用。钩子通常只提供了“在某一步之后做什么”比如on_tool_call、on_llm_response但中间件链提供的是“整个请求处理流程的控制权”。中间件既可以先处理再放行也可以直接短路返回甚至可以替换最终的输出。这种控制力度回调钩子很难实现。我最初的项目就是踩了这个坑用回调钩子加装饰器攒了一套逻辑结果到了要加“工具执行完需要把摘要写回上下文再问一次模型”这种跨阶段逻辑时钩子根本接不住只能再改回调顺序。换了中间件链之后这类逻辑变成了一个很普通的中间件放在工具执行环节后面即可。2. 中间件链的核心机制搞懂原理你才能自由组装2.1 一个标准中间件长什么样我们先定义最基本的抽象。一个中间件本质上是个函数它接收两个参数一个是上下文对象一个是next函数。调用next()表示让执行流继续往下走返回之后拿到内层的结果再做后处理。代码写出来是这样的# Python示例标准中间件签名 async def logger_middleware(ctx, next): print(f[start] call_id{ctx.call_id}) try: result await next() print(f[end] cost{result.cost_seconds:.2f}s) return result except Exception as e: print(f[error] {e}) raise这里的ctx就是我们所说的上下文对象。在Harness里它不是简单的字典而是一个有明确字段的数据容器。一般至少包括几个部分输入用户消息、系统提示词、模型状态当前对话消息列表、工具定义和工具执行结果、最终输出、以及一些运行时信息重试次数、调用耗时、日志收集器等。而next对应着链中下一个中间件。如果某个中间件不调用next链路就短路了后面的中间件不会再执行。这是实现权限拦截、缓存命中、条件终止的关键。2.2 洋葱模型与线性握手中间件链的执行顺序有两种常见模型。一种是标准的洋葱模型请求从前向后穿透到最核心的处理函数然后再反向穿过每一层。它的特点是前中间件不仅能看到请求还能看到内层结果。类比快递包装最外层的纸箱先拆里层的泡沫再拆最后看到商品检查完毕再一层层包回去。另一种是线性握手模型每个中间件都做完全独立的处理然后传给下一个链尾是最终清洗和收尾。它不适合需要上下行对称处理的场景。对于Agent Harness我强烈建议用洋葱模型因为很多中间件需要有“前置逻辑”和“后置逻辑”。日志中间件要在调用前记录输入调用后记录结果重试中间件在调用前判断次数调用失败后决定是否再执行一次性能监控中间件则天然依赖后处理阶段计算耗时。如果你用线性模型要么把后处理塞到链尾要么就得把每个中间件拆成两个独立节点组合起来不够优雅。实现洋葱模型的关键在于next不再是普通的函数引用而是一个复合了后续中间件的调用闭包。我们可以把中间件列表倒序组合形成嵌套关系。下面这段伪代码描述了核心逻辑# Harness内将中间件组合成嵌套调用 def build_chain(middlewares, final_handler): acc final_handler for mw in reversed(middlewares): acc wrap(mw, acc) # 生成一个新的函数调用mw时会传入acc作为next return acc直观感受就是最后一个注册的中间件是真正贴着模型调用层的第一个注册的中间件是贴着用户的。2.3 上下文对象的设计决定扩展上限很多中间件之间要共享数据。比如鉴权中间件解析出用户身份后续的工具白名单中间件要拿这个身份去判断模型调用中间件生成的对话记录要被历史摘要中间件读取。如果上下文只是一个dict松散的结构会让中间件之间产生隐式耦合改一个字段名全链崩掉。我更推荐用一个轻量级数据类把关键字段显式定义出来再用一个通用extra字典兜底。这样既能获得类型提示又能保留灵活度。class Context: def __init__(self, user_input, system_prompt): self.call_id generate_call_id() self.user_input user_input self.system_prompt system_prompt self.messages [] # 发送给模型的完整消息列表 self.tools [] # 当前可用的工具定义 self.tool_results [] # 每次工具执行的结果 self.final_answer None # 最终输出 self.metadata {} # 额外运行时信息 def add_tool_result(self, result): self.tool_results.append(result) self.messages.append({role: tool, content: result})显式的messages字段是整个链的生命线。中间件通过操作它就能在模型调用前改上下文工具执行中间件往它里面追加结果下一个模型调用轮次才能看见。设计上下文时最容易踩的坑是让中间件各自维护私有状态。比如重试中间件把重试次数存在自己的局部变量里下一个中间件就感知不到。正确做法是写入ctx.metadata[retry_count]这样日志中间件也能记录。3. 实操从零搭一个可组合的Agent Harness3.1 先定义最小的Harness骨架为了讲清楚实现我用一个非常精简的Python版示例来演示骨架。这里不绑定任何具体的模型SDK只用mock_llm_call函数模拟一轮模型交互。实际的LLM调用在这个函数里就行。# harness.py import asyncio, time, uuid class Harness: def __init__(self, middlewaresNone): self.middlewares middlewares or [] async def execute(self, ctx): # 构造整个链条最内层是模型调用 async def final_handler(ctx): # 这里可以换成真实的大模型调用 result await mock_llm_call(ctx.messages, ctx.tools) ctx.final_answer result.get(answer) return ctx.final_answer chain final_handler for mw in reversed(self.middlewares): chain self._wrap(mw, chain) return await chain(ctx) staticmethod def _wrap(mw, next_func): # 生成一个闭包作为中间件接收的next async def compiled(ctx, nextnext_func): return await mw(ctx, next) return compiled在这个设计里每个中间件都是async def middleware(ctx, next)的签名。Harness负责把它们串起来对外部来说使用者只管维护一个middlewares列表。注意一个细节reversed(self.middlewares)意味着middlewares列表的第一个元素是最外层。也就是说列表顺序和业务直觉一致阅读列表时是从用户视角往下读的。3.2 写几个高频必备中间件日志中间件很好写但有一点容易被忽略不光要记录输入输出还要记录call_id和完整的耗时。生产环境里排查Agent问题把同一call_id的所有中间件日志串起来效率提升非常明显。async def logging_middleware(ctx, next): start time.perf_counter() print(f[log] {ctx.call_id} input{ctx.user_input[:50]}...) try: result await next() finally: cost_ms (time.perf_counter() - start) * 1000 print(f[log] {ctx.call_id} cost{cost_ms:.1f}ms result{result}) return result重试中间件是Agent项目里最能救命的一个。LLM调用经常出现偶发失败比如上游限流、JSON解析错误、工具返回格式异常。重试不能盲目做要区分错误类型网络类可以重试参数类重试无意义。另外一定要加最大次数限制和指数退避否则多个Agent实例同时重试会加剧限流。async def retry_middleware(ctx, next): max_retries int(ctx.metadata.get(max_retries, 3)) base_delay 0.5 for attempt in range(1, max_retries 1): try: return await next() except TransientError as e: if attempt max_retries: raise delay base_delay * (2 ** (attempt - 1)) print(f[retry] attempt{attempt} next_delay{delay:.2f}s error{e}) await asyncio.sleep(delay) ctx.metadata[retry_count] attempt # 实际不会到达这里 raise RuntimeError(unreachable)工具白名单中间件是一个典型的“短路型”中间件。它不修改请求只负责检查当前上下文里ctx.tools是否越权。如果在进入模型前就发现某个工具不该被当前用户调用直接抛出权限异常或者从ctx.tools里移除模型就不会生成违法工具调用了。async def permission_middleware(ctx, next): allowed ctx.metadata.get(allowed_tools, set()) ctx.tools [t for t in ctx.tools if t[name] in allowed] return await next()记忆管理中间件稍微复杂一点。它负责在每次调用模型前检查历史消息是否太多如果超长就做一次摘要把摘要注入到system_prompt里。核心点是摘要内容一定不要覆盖原始细节应该保留关键事实型信息比如用户地址、偏好、最近决定等。3.3 一个完整的工具调用循环中间件链里最核心的其实是“工具调用循环”本身。Agent不是一次模型调用就完事的它可能要循环好几轮模型生成工具调用请求执行工具把结果喂回模型模型再继续生成下一步。这个循环通常被封装成Harness内部的一个agent_loop函数。我的做法是把agent_loop也看作一个特殊中间件放在链的最内层。这样重试、日志、白名单都可以直接作用于整个agent循环而不是某一次模型调用。async def agent_loop(ctx, nextNone): max_steps 5 for step in range(max_steps): response await mock_llm_call(ctx.messages, ctx.tools) if response.get(tool_calls): for call in response[tool_calls]: tool_name call[name] args call[arguments] result execute_local_tool(tool_name, args) ctx.add_tool_result(tool_name_result) # 把工具名为前缀也放进result里便于模型理解 else: ctx.final_answer response[answer] break return ctx.final_answer有一点切身体会工具调用循环里的上下文追加顺序极其重要。工具执行完以后要在本轮的assistant消息后面追加tool消息这样模型才能正确关联工具调用和返回结果。如果你把多轮工具执行的结果一股脑全塞到messages的末尾模型经常会产生幻觉把上一个工具的结果当成下一个工具的输入。3.4 组合示例面向不同场景动态拼链考虑到实操我们做两个组合示例。第一个是“带日志的简单问答”适合调试阶段。第二个是“多轮工具调用权限重试”适合生产环境。# 场景1调试简单问答 harness Harness([ logging_middleware, agent_loop, ]) ctx Context(帮我查一下明天的天气, 你是AI助手) await harness.execute(ctx) # 场景2带权限、重试和记忆管理的生产链 harness Harness([ logging_middleware, permission_middleware, memory_management_middleware, retry_middleware, agent_loop, ])看到没有中间件链的价值就在这个列表里。每加一个能力你不需要改动已有代码只需要在列表里插入一个中间件。比如上线后发现模型老是重复调用某个高风险工具那就加一个rate_limit_tool_middleware放在permission_middleware后面瞬间生效。我没有把工具执行逻辑放进单独的中间件原因是agent_loop已经在一个循环里。如果拆成独立的“工具执行中间件”中间件会有一个循环等待外层再次进入自己设计会很别扭。所以建议是工具执行作为agent_loop内部细节而围绕执行的横切能力用中间件包裹。3.5 参数选型如何决定中间件顺序中间件顺序没有绝对标准但有两条经验性原则。第一越靠近外层的中间件越偏“请求级横切关注点”比如日志、性能、限流。第二越靠近内层的中间件越偏“核心业务语义”比如工具调度、上下文组装、Agent循环。以我的生产链为例顺序是日志最外层必须记录整个流程权限靠近外层尽早拦截非授权请求限流紧跟权限防止恶意/高频请求打到模型重试限流之后重试可以对限流错误恢复记忆管理重试之后上下文清理在进入agent循环前agent_loop最内层这里有一个容易搞反的组合重试应该在限流外层还是内层我的答案是重试在限流外层但限流返回的429错误本身由重试捕获。如果重试在限流内层那么限流器本身如果也失败重试就不会被执行了。这个顺序逻辑需要结合自身场景我说的是通用建议。4. 常见问题与排查技巧这些坑我挨个踩过4.1 中间件顺序写反导致上下文被覆盖我曾经把记忆管理中间件放在了agent_loop之后结果每个轮次准备发送给模型的消息列表都是空。原因就是记忆管理中间件在next()之后执行了裁剪逻辑但裁剪之后ctx.messages还没被下一个小伙伴使用于是白白剪掉了。后面改成记忆管理必须在agent_loop之前加载历史在next()之后把新的对话摘要写回到存储里。两者要拆成两个中间件或者在一个中间件里做前置加载、后置回写。排查这类问题的思路很简单在每个中间件里打印ctx.messages的长度和头尾消息跟踪消息列表是怎么变化的。一旦发现某一步跨过之后数据变了就定位到那个中间件。4.2 异常被“吞掉”的问题洋葱模型有个危险行为如果某个中间件内部的try...except捕获了异常但没有重新抛出并且写了return语句就会导致内层后续中间件的代码被跳过而且外层还以为一切正常。这在重试中间件里最容易发生某次执行抛错你捕获后计数并准备重试但忘了return await next()结果链直接返回了None模型输出就没了。我后来给自己定了一个规则中间件代码里只要捕获了异常就必须明确选择“抛出”或“重试后抛出”绝对不允许静默吞掉。调试期可以在捕获分支里加日志但不能直接裸except写死。4.3 性能损耗别把每个中间件做成一次远程调用中间件链本身是内存内函数调用开销很低。但有些开发者喜欢在中间件里调用外部服务比如鉴权中间件请求鉴权服务记忆中间件请求向量库结果整条链被几十次网络往返拖慢。优化手段一般是复用连接、批量获取、预热缓存。比如在权限中间件里把用户权限一次性放进metadata后续链上其他中间件就不要再去查库了。我测试过一个不过分优化的链6个中间件全部内存操作一个Agent循环的额外开销不到1毫秒。而一旦某个中间件引入外部I/O就可能多出几十毫秒甚至更糟。所以尽量让中间件做纯计算和数据变换把IO隔离在链外或者采用批量执行。4.4 调试的可观测性设计Agent项目的调试比普通后端更麻烦因为中间步骤会调用多个工具、产生多轮对话。我强烈建议给Harness内置一个事件收集器每个中间件可以往ctx.events里面追加结构化事件而不是用print。这样后续可以导出成JSON/CTF格式或者对接前端看板。最简单的实现是def emit(ctx, event_name, **kwargs): ctx.metadata.setdefault(events, []).append({ name: event_name, ts: time.time(), **kwargs, })随后在日志中间件的最外层打印ctx.metadata[events]你就能看到一条完整的执行时间线。包括每次工具调用、每个中间件进入/离开时间、模型请求次数等。这个功能哪怕只用JSON打印排查起问题来已经非常高效了。我曾经靠它定位到一个链路退化问题某类任务每次都会进入一个循环里反复请求同一个查询工具事件里可以清楚看到相同的事件出现了10次。4.5 关于可组合性的第三个原则最小上下文传递最后一个经验是在设计中间件时要尽量只传递和中间件职责相关的数据不要所有中间件都把完整的历史消息列表拿过去又拿回来。例如权限中间件只需要ctx.tools和ctx.metadata中的用户ID就不应该去读ctx.messages。这个原则可以降低中间件之间隐式耦合也让测试单个中间件变得轻松。测试上我习惯用真实的Context对象喂给单个中间件配合一个只做“透传”的next桩函数验证它是否按预期修改了上下文。这种方法比全链路集成测试更稳定也能在改动某个中间件时快速回归。5. 后面还能怎么扩展中间件链一旦稳定下来很多后续能力都变成了往列表里加新节点的事。比如给Agent加上“多工具并行调用”能力可以在agent_loop内对工具调用做并发调度这是一个内部优化又比如做多模型路由可以在最外层加一个router_middleware根据输入难度选择调用不同参数的模型再比如做自动评估在链尾加一个evaluator_middleware用一个小模型给大模型的回答打分记录日志。我个人觉得这种可组合Harness最大的意义不一定是性能提升而是终于把Agent做成了一条清晰可见的生产流水线。以前调代码像在扒一团毛线现在每根线头都在自己的位子上问题定位时间至少缩短了一半。如果你正在做Agent工程化非常建议试一下这个方向不用管框架大小从两三个中间件开始搭感受会很不一样。