企业官网建设流程全解析

这段时间在做一个比较复杂的Agent项目，模型本身倒没怎么换，但工程侧的问题差点让我崩溃。先是工具调用逻辑越写越乱，后来加了一堆重试、权限检查、上下文清理的代码，整个调用过程变得像一团乱麻，一个思路没理清就得上百行胶水代码。后来我认真梳理了一下思路，把目光放在了可组合Harness上，核心就是用中间件链的方式重新组织Agent的调用流程。

这个概念听起来偏架构，但落地之后效果立竿见影：每个关注点可以独立成组件，调用链可以随意插拔、复用、测试，Agent的能力边界完全不靠堆代码来实现，而是靠“组合”来扩展。这篇文章我把自己的设计思路和实操过程整理出来，适合正在做Agent工程化、被复杂调用逻辑折磨过的开发者参考，也适合想从单体调用走向模块化控制层的人。

1. 从疯狂堆代码到中间件链，Agent控制层为什么需要重构

1.1 单体Agent调用到底恶心在哪

大多数Agent项目起步都很简单：把用户输入丢给大模型，大模型返回一个JSON，里面写了想调什么工具、参数是什么，然后代码解析JSON再执行。这套流程跑通demo只需要二三十行代码，但一旦进入生产环境，问题就全来了。

我举几个真实场景：需要给每个请求打日志，方便事后排查模型到底怎么想的；工具调用可能失败，网络抖动、参数非法、返回格式异常，都需要重试和兜底；不同的用户能调的工具不一样，需要在调用前做权限校验；上下文不能无限膨胀，得在每次请求之前裁剪或者摘要；遇到上游限流，还得设计退避策略。这些需求如果全部写在Agent调用的主流程里，你会发现这个主流程越长越复杂，最后几乎没法维护。

最典型的反面做法是给主函数不断加参数。今天加一个enable_logging=True，明天加一个max_retries=3，后天又要加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 Harness

3.1 先定义最小的Harness骨架

为了讲清楚实现，我用一个非常精简的Python版示例来演示骨架。这里不绑定任何具体的模型SDK，只用mock_llm_call函数模拟一轮模型交互。实际的LLM调用在这个函数里就行。

# harness.py import asyncio, time, uuid class Harness: def __init__(self, middlewares=None): 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, next=next_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, next=None): 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循环。

以我的生产链为例，顺序是：

1. 日志（最外层，必须记录整个流程）
2. 权限（靠近外层，尽早拦截非授权请求）
3. 限流（紧跟权限，防止恶意/高频请求打到模型）
4. 重试（限流之后，重试可以对限流错误恢复）
5. 记忆管理（重试之后，上下文清理在进入agent循环前）
6. 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工程化，非常建议试一下这个方向，不用管框架大小，从两三个中间件开始搭，感受会很不一样。