企业官网建设流程全解析

如果你在 LangChain 0.x 时代用 callbacks 做过超时管理、日志上报、输出校验这类事，大概率对那段“回调地狱”记忆犹新：想拦一次 LLM 调用就得注册一堆on_llm_start、on_llm_end、on_chain_error，事件靠字符串名硬匹配，跨链传点上下文要绕好几个弯。LangChain v1.0 把 middleware 中间件扶正为一等公民之后，这些“包裹 LLM 调用”的横切逻辑终于有了稳定的第一方抽象，写起来像给函数套了一层编排好的装饰器，逻辑顺序肉眼可见。这篇就围绕 LangChain v1.0 的中间件机制，讲清楚它到底是给谁解决什么问题、底层数据流长什么样，以及我从 0.3 时代迁到 1.0 过程中踩过的坑和沉淀下来的写法。

1. 从回调到中间件：v1.0 为什么把横切逻辑做成头等公民

1.1 callback 时代的三宗罪

我在 0.3 版本里维护过一套多智能体系统，核心诉求其实很朴素：每次调用模型前要注入统一的系统提示词片段，调用结束后要记录 token 消耗，失败时要按策略重试。听起来简单，但用 callbacks 实现时每个诉求都别扭。

第一宗罪是事件类型太松散。on_llm_start里拿到的serialized是一坨字典，想拿到模型名和请求参数得自己解析；on_llm_end里的输出结构在不同版本间还变过。也就是说，回调更像是一种“监听日志流”，而不是结构化的程序接口。

第二宗罪是上下文传递困难。我曾在回调里想拿到当前用户会话 ID，结果只能把会话 ID 塞进run_id前缀里，或者用一个全局变量硬扛。这在单线程测试里没问题，一上并发就乱套。

第三宗罪是组合与顺序不可控。两个回调都想改输入，谁先谁后取决于注册顺序，但回调本身没有一个显式的“管线”概念，排查起来全靠猜。

1.2 v1.0 中间件的设计动机：把控制权还给调用方

LangChain v1.0 中间件机制的核心转变，是把“旁路监听”变成了“管线内显式处理”。中间件不是挂在某个链节点上等待事件广播，而是直接包裹在调用链外面：请求进来先经过before阶段，返回结果再走after阶段，异常则有专门的处理钩子。

这种设计的直接收益有两点：

• 横切逻辑不再依赖字符串匹配的事件名。中间件是确定性函数，有明确的入参和出参，类型系统能帮你兜底。
• 执行顺序完全可预测。before 阶段按注册顺序执行，after 阶段逆序执行，跟 HTTP 框架的中间件模型一致，心智负担小很多。

我个人的理解是：v1.0 团队把“对 LLM 调用的包装”这件事从影子状态提升到正式 API——你不再需要猜测框架内部某个节点“有没有发事件”，而是显式地说“我要在调用前干这个、调用后干那个”。这个思维转变，是整个迁移里最需要适应的一步。

2. 中间件的数据流与结构：一次调用到底经过了什么

2.1 before、after、handle_exception 三个钩子

LangChain v1.0 的中间件核心接口围绕三个生命周期钩子展开。按我的实际使用习惯，可以理解成一段伪代码：

async def call_with_middlewares(chain, request, middlewares): event = request # before 阶段：按注册顺序正序执行 for m in middlewares: event = await m.before(event) try: response = await chain.invoke(event) except Exception as exc: # 异常处理：从后往前找第一个愿意处理的中间件 for m in reversed(middlewares): handled = await m.handle_exception(event, exc) if handled is not None: event = handled break else: raise exc # after 阶段：拿到结果后按注册顺序逆序执行 for m in reversed(middlewares): event = await m.after(event) return event

这里每个中间件必须假设自己拿到的event可能被前面的中间件修改过。举例说明：我写的注入中间件先在before阶段往系统提示词里塞了一段上下文，后面接的输出校验中间件必须兼容这个已经变大的请求结构，否则校验逻辑拿到的字段对不上。

2.2 上下文传递：init_context 与 execution_options

中间件之间如果要共享数据，正确姿势是通过上下文对象，而不是全局变量。LangChain v1.0 提供了上下文的初始化与传递机制，核心玩法是：用init_context先声明上下文结构，调用时通过execution_options往里塞值，中间件侧再按 key 取。

我手头一个模拟项目的写法是这样的：

from langchain_core.middleware import init_context from langchain_core.global_vars import ExecutionOptions # 1. 初始化上下文结构 init_context( "audit", fields={ "user_id": None, "session_id": None, "request_id": None, }, ) # 2. 调用时传入上下文值 response = chain.invoke( request, execution_options=ExecutionOptions( context={"audit.user_id": "u_123", "audit.session_id": "s_456"} ), )

这里有个容易忽略的细节：上下文变量在并发场景下是每次调用独立一份，所以不会出现 A 请求的user_id串到 B 请求身上的问题。这比我在 0.x 时用contextvars自己造的轮子干净得多。

2.3 中间件的注册与顺序设计

中间件注册方式有两种主流姿势：一类是在构建 chain 或 agent 时作为参数传入，另一类是链对象上的.with_middleware()方法。以 agent 为例：

from langchain.agents import create_agent agent = create_agent( model=chat_model, tools=tools, system_prompt=base_prompt, middleware=[audit_middleware, retry_middleware, schema_validate_middleware], )

顺序设计上有一个很重要的原则：先注册的 before 先执行、after 后执行。实际操作里我习惯把“纯观测”类中间件（日志、审计）放在最前面，把“会改输入”的中间件（注入、改写、脱敏）放在后面。原因很简单：如果先让注入中间件改了输入，日志中间件再记录时，看到的就是被加工过的请求，排查原始问题时容易误判。

3. 手写一个生产可用的中间件：从结构化日志到智能重试

3.1 场景一：基于中间件做结构化审计日志

以前用 callback 做审计日志，最痛苦的是跨链把 request_id、session_id 串起来。中间件时代这个事变得非常直白：在 before 阶段记录开始时间和请求快照，在 after 阶段补上耗时和 token 用量。

我常用的模板是：

import time import logging from langchain_core.middleware import Middleware logger = logging.getLogger("audit") class AuditLogMiddleware(Middleware): async def before(self, event): event.state["_start_time"] = time.perf_counter() logger.info( "llm_request_start", extra={ "model": event.model if hasattr(event, "model") else "unknown", "request_id": event.context.get("audit.request_id"), "user_id": event.context.get("audit.user_id"), }, ) return event async def after(self, event): elapsed_ms = (time.perf_counter() - event.state["_start_time"]) * 1000 logger.info( "llm_request_end", extra={ "elapsed_ms": round(elapsed_ms, 2), "completion_tokens": getattr(event.response, "usage", {}).get("completion_tokens", 0), "prompt_tokens": getattr(event.response, "usage", {}).get("prompt_tokens", 0), }, ) return event

这里我特别想分享一个细节：中间件里别用同步的文件 logger 去写盘。我在压测环境里曾经因为日志中间件直接写文件，导致单次调用延迟从 300ms 飙到 2 秒，最后换了异步队列才恢复。审计日志通常要写磁盘，务必用异步 handler 或者丢到消息队列里。

3.2 场景二：带边界条件的自动重试

重试是中间件最典型的价值场景，但也是坑最多的地方。我踩过的教训是：不是所有异常都适合无脑重试。

先说结论，我把异常分成三类处理：

• 瞬时故障（限流 429、连接超时、5xx）：值得重试，但要按指数退避。
• 参数错误（400、上下文超长、Schema 校验失败）：重试一万次也没用，直接抛给上层。
• 流式中断（客户端断连、生成中途出错）：要视情况决定是从头重试还是重新生成。

带重试的中间件核心逻辑长这样：

class RetryMiddleware(Middleware): def __init__(self, max_retries=3, base_delay=1.0): self.max_retries = max_retries self.base_delay = base_delay async def before(self, event): event.state["_retries"] = 0 return event async def handle_exception(self, event, exception): if not self._is_retryable(exception): return None # 返回 None 表示不处理，交给下一个中间件或抛出 retries = event.state.get("_retries", 0) if retries >= self.max_retries: return None delay = self.base_delay * (2 ** retries) event.state["_retries"] = retries + 1 await asyncio.sleep(delay) # 重新调用链本身需要在这里触发，但要注意：善用循环而不是递归 # 这部分我会在外部封装成带重试的执行器，中间件只负责标记次数和退避 return event

_is_retryable里至少要区分RateLimitError、TimeoutError和ValidationError。我把限流错误和超时错误放行，其他一律不重试。还有一点：重试中间件千万别包在“会产生副作用”的链外层，比如链里已经调用了外部 API 写入了数据，重试会导致重复执行。我就遇到过因为重试导致外部工单重复创建的事故，后来把重试范围严格限制在 LLM 调用层——也就是 chain 内部，而不是外面整体包装。

3.3 场景三：输出 schema 校验与 PII 脱敏

生产环境里模型输出往往不可控，我习惯在 after 阶段挂一个输出校验中间件。这里要留意的是，中间件拿到的输出是完整的 LLM 回复对象，不是纯文本，所以校验逻辑要写在与response同构的位置。

class SchemaValidationMiddleware(Middleware): async def after(self, event): content = event.response.content # 简单的 JSON Schema 校验占位 try: parsed = json.loads(content) if not required_keys.issubset(parsed.keys()): raise ValueError(f"missing keys: {required_keys - parsed.keys()}") except json.JSONDecodeError: # 如果模型没返回合法 JSON，可以考虑在这里触发一次修正 # 但注意：after 阶段不再重入调用，只能标记错误或做缝补 event.response.content = "{}" return event

PII 脱敏也适合放在这个阶段：把身份证号、手机号、邮箱这类看似正常的字符串从输出里替换掉。脱敏中间件放 before 还是 after 取决于你的诉求——如果只是防止模型“记住”用户隐私，放 before 阶段对输入做脱敏；如果是防止输出结果泄露隐私，放 after 阶段。我两个都写过，实践下来 after 阶段更安全。

4. 从 0.x callbacks 迁移到 v1.0 中间件的对照与隐藏坑

4.1 事件映射表

很多人迁移第一步就卡在“原来那个on_llm_start现在怎么写”。先给一张我整理的映射关系：

这里要特别提到一点：0.x 里很多“监听”逻辑本质上是只读的，而中间件是允许改写的。迁移时不能用过去的思路写中间件——不是“看到事件然后记一笔”，而是“处理事件并返回给下一个环节”。两者表达力完全不是一个量级。

4.2 迁移中容易踩的四个隐藏坑

第一个坑是事件对象可能不可变。很多中间件试图直接event.response.content = xxx，报错后才发现某些字段是只读的。我在 0.x 时代从来不用担心改不改事件，因为回调本就不该改数据。迁移后要主动查一下中间件拿到的对象哪些字段可变、哪些不可变。

第二个坑是中间件状态的生命周期。如果你想在中间件实例上用self.counter += 1记录次数，在并发下大概率得到令人困惑的结果。原因很简单：同一个中间件实例可能被多个并发请求共享。正确姿势是把“次数”这类可变状态放到event.state或 context 里，而不是中间件实例属性上。

第三个坑是装饰器顺序与注册顺序容易搞混。如果你既用了装饰器写法又用了.with_middleware()，执行顺序会交错，调试时特别头疼。我的经验是：一个项目里统一用一种注册方式，别混搭。

第四个坑是流式场景下的 after 钩子被跳过。如果调用的是stream()而不是invoke()，中间件的 after 阶段处理逻辑要适配流式帧——不是所有中间件在流式下都会原样触发。我遇到过日志中间件在流式模式下打不出结束日志的问题，原因是流式响应的“结束”标志藏在最后一个 chunk 里，需要特殊判断。

4.3 迁移的思维转变：从“监听者”到“拥有者”

这一条算是我自己的经验总结，写出来给大家参考。

0.x 时代的回调是“旁观者”：你注册了一堆 handler，框架觉得“合适的时候”就会调你，但你既不能拦下请求，也不能修改请求，更别说重试了。v1.0 的中间件是“拥有者”：你在管线里，你的代码就是这个请求生命周期的一部分，你有能力修改输入、修改输出、决定要不要把这个异常继续向上抛。

这个转变带了两个连带的影响。第一，中间件的单元测试变得非常简单：直接构造一个事件对象，调用await middleware.before(event)，断言返回结果即可，不需要跑一个完整链。第二，调试心智大幅降低：你可以单步跟踪一条请求依次经过哪些中间件，每一层对事件做了什么修改，像剥洋葱一样清晰。

5. 中间件性能、并发与调试经验：实测下来的几条铁律

5.1 中间件是热路径：别在这里偷懒

中间件包裹着每一次 LLM 调用，它就是你应用的热路径。很多人在中间件里贪图方便，干了些不该干的事，我列个黑名单：

• 在 before/after 里做同步网络请求（比如调外部鉴权 API），会直接阻塞事件循环。
• 在 after 里做大数据量的 JSON 序列化/反序列化，毫秒级开销会被放大到每次调用。
• 用全局 list 组装日志，然后不定时 flush，多线程下丢数据的概率极高。

我压测过一个加了 5 个中间件的 agent，每个中间件只做轻量日志，整体延迟增长大约控制在 10ms 以内；但如果某个中间件里有一次同步外部调用，延迟直接增加一个量级。所以我的纪律是：中间件里的 IO 一律异步化，能做缓存就不要实时算，能轻量就不要携带重上下文。

5.2 中间件与流式输出：容易忽略的适配点

流式是目前 agent 应用的标配，但中间件的 after 钩子在流式模式下不会像 invoke 那样一次性拿到完整结果。实测下来的正确处理姿势是：把流式响应的聚合逻辑放在即时处理之外，比如只做事件标记，真正的内容处理等流结束再统一触发。

我写过一版流式审计中间件，思路是：

class StreamingAuditMiddleware(Middleware): async def before(self, event): event.state["_start_time"] = time.perf_counter() event.state["_chunk_count"] = 0 return event # 注意：这里把 after 拆成流式特有的处理，而不是直接读 response.content async def after(self, event): # 流式下 response 可能是生成器，不能直接切片 # 所以要判断事件类型，区分一次性响应和流式帧 if event.is_streaming: logger.info("stream_finished", extra={ "chunk_count": event.state["_chunk_count"], "elapsed_ms": round((time.perf_counter() - event.state["_start_time"]) * 1000, 2), }) else: logger.info("invoke_finished", extra={ "content_length": len(event.response.content), }) return event

从这能看到一个通用经验：实现中间件前先问自己一句——这个中间件在流式调用下会被怎样触发？如果答案不确定，就在开发环境先用stream()跑一遍用例，再把行为固化下来。

5.3 调试中间件的三板斧

中间件条目一旦超过三个，就需要系统性的调试手段。我常用的三板斧如下。

第一板斧：给每个中间件显式命名，并打印“进入/离开”日志。用日志追踪请求经过的路径，顺序一目了然。

class NamedMiddleware(Middleware): name = "audit" # 显式命名 async def before(self, event): logger.debug("middleware[%s].before", self.name) return event async def after(self, event): logger.debug("middleware[%s].after", self.name) return event

第二板斧：用一个最小复现链在本地单步调试。不要直接在大型 agent 上调试中间件——中间件太多时缓存、顺序、并发交织在一起，很难定位。我一般会单独建一个只含一个模型的 chain，只挂一个要调试的中间件，跑用例，单步看事件变化。

第三板斧：临时加一个“快照中间件”，把进入和离开时的事件内容打印出来。它可以作为最后一道保障，排查“事件是不是被某个中间件改坏了”。我个人习惯把这个快照中间件放在最末尾，专门负责序列化事件并 dump 到日志。

5.4 一个实际的顺序优化案例

最后分享一个我调过的顺序优化真实案例。我那个模拟项目里同时挂了三个中间件：

1. 限流中间件（before 里检查令牌桶）
2. 日志中间件（before/after 记录信息）
3. 输出润色中间件（after 里对回答做排版和裁减）

最初的注册顺序是[限流, 日志, 润色]。测试发现一个问题：当请求被限流拦截时，限流中间件在 before 阶段直接抛了异常，日志中间件的 after 不会执行，所以限流事件不会被记录。解决方法是把日志中间件放在限流之前，再让日志中间件对异常场景做“标记式”处理——这样即使后续阶段报错，也能留下一份完整的审计足迹。

从这个案例可以得到一个通用的顺序判断准则：有拦截/抛错能力的中间件尽量放后面，有审计/观测能力的中间件尽量放前面。观测在前，控制在后，可以让任何“被拦下来”的事件都有迹可循。

实际用下来的体会是，中间件最值钱的地方不是“官方出了一个新功能”，而是它把一个本来要自己造轮子的架构问题——如何干净地把横切逻辑从业务逻辑里剥离出来——用一套统一的模型收编了。如果你正在做多模型、多工具链的项目，或者想给团队沉淀一套可复用的调用治理层，LangChain v1.0 中间件完全够格作为第一方基础设施，而不是临时补丁。