企业官网建设流程全解析

做 LLM 应用这段时间，我维护过一套基于 LangChain 的问答系统，业务方越来越多，需求也从“能回答就行”变成了“要日志、要限流、要敏感信息过滤、要按租户单独计费”。一开始这些逻辑都是直接往链里塞，结果链上挂了一堆 RunnableLambda，有的负责改 prompt，有的负责记耗时，有的在模型返回后做脱敏，互相之间没有任何生命周期概念。项目名我写作 Langchain_v1.0|middleware中间件，竖线是我自己的命名偏好，意思是给 LangChain v1.0 应用统一挂载一层可插拔的中间件层。这个方案解决的不是模型选型、提示词怎么写，而是链路治理问题，适合那些正被横切逻辑搞得焦头烂额、或者准备在公司内部沉淀一套 LLM 调用规范的团队参考。

1. 为什么要给 LangChain v1.0 加一层 middleware 中间件

1.1 原生链式结构在横切逻辑上的痛点

LangChain 的链式写法本身很舒服，prompt、模型、输出解析器用|串起来，代码看着干净。但一旦你要在每个环节统一加东西，问题就来了。比如记录每次调用的耗时和 token 消耗，你会想到用 CallbackHandler；要在请求发出去之前替换掉输入里的手机号，你可能得在链上插一个 RunnableLambda；要限制某个租户的调用频率，又得在模型调用外面包一层。三个需求来自不同团队，最后链上堆了五六个函数，每个函数只做一点点事，但逻辑散落得到处都是。

我在项目里统计过，最多的一张链上有 7 个 RunnableLambda 包在模型外面，有的叫with_tenant_context，有的叫trace_llm_call，有的叫filter_sensitive_content。它们之间没有统一的上下文，也没有固定的执行顺序约束。新同事接手的时候根本分不清谁先执行、谁后执行，更麻烦的是，这些包装函数在某些链上被重复添加，导致一次请求打了两遍日志。这种问题不是 LangChain 本身的缺陷，而是横切逻辑天生不适合“硬塞进链里”。Web 后端早就遇到过同样的问题，中间件就是标准答案。

1.2 我把中间件能力边界收敛成了四个钩子

参考 Koa 和 Express 的中间件设计，我给这套方案定义了四个钩子：before、after、error、finally。真正的模型调用作为最内层目标，四个钩子围绕它执行。这样设计的原因是，大部分需求都可以归类到这四种时机里：

在这个基础上，中间件只被允许做三件事：修改上下文、观察记录、短路终止。修改上下文指的是改ctx.inputs或ctx.outputs，观察记录是只读地写日志和指标，短路终止则是在 before 阶段直接抛异常或返回默认结果，让后面的中间件和模型调用不再执行。把能力收窄之后，中间件的职责变得非常清楚，不会出现“一个中间件什么都干”的情况。

2. 中间件的核心模型：接口、生命周期与洋葱调度

2.1 极简接口定义与上下文对象

我定义的最小接口长这样：

from dataclasses import dataclass, field from typing import Any, Optional @dataclass class MiddlewareContext: config: dict inputs: dict[str, Any] outputs: Optional[Any] = None error: Optional[BaseException] = None metadata: dict[str, Any] = field(default_factory=dict) class BaseMiddleware: async def before(self, ctx: MiddlewareContext) -> None: pass async def after(self, ctx: MiddlewareContext) -> None: pass async def error(self, ctx: MiddlewareContext) -> None: pass async def finally(self, ctx: MiddlewareContext) -> None: pass

中间件之间唯一的信息通道是MiddlewareContext。它里面有inputs、outputs、error，还有一个metadata字典，专门用来给中间件互相传数据。比如限流中间件在 before 里写入ctx.metadata["request_id"]，后面的审计中间件可以直接拿这个字段做日志关联。如果依赖全局变量或者外部状态，并发一高就会互相污染。

config是从 LangChain Runnable 里带过来的配置对象，业务中间件需要读取调用方参数时，可以从里面取。但注意不要在里面塞业务自定义字段，LangChain 后续版本对 config 的字段管理会更严格，自定义信息尽量放metadata。

2.2 洋葱模型到底在说什么

中间件的执行顺序是典型的洋葱模型。列表里的中间件一层层往里走，先执行所有 before，再执行真正的目标函数，然后按相反顺序执行 after。我见过很多刚接触中间件的同学把它理解成“for 循环按顺序执行”，那 after 就会全部乱掉。

举个例子，注册了 A、B 两个中间件，执行顺序是这样：

这个顺序有一个很实用的效果：内层中间件对ctx.outputs的修改，外层中间件可以拿到。所以我会把“数据转换型”中间件放在内层，把“展示型”中间件放在外层。比如脱敏中间件放内层，日志中间件放外层，日志里记录的就是脱敏后的结果，不会把手机号写进日志。

如果在 before 阶段抛异常，后面的中间件就不会执行，当前层和已经执行过的外层中间件会依次处理 error。每个中间件的 finally 一定会执行，不管成功还是失败，这保证了埋点、资源释放这类需求不漏。

2.3 只包一个 invoke 远远不够

刚开始写中间件管理器的时候，我只包装了invoke和ainvoke，结果一接入流式输出就出问题。LangChain 的astream会把模型的输出拆成多个 chunk 逐个返回，如果中间件只包了一次调用，第一个 chunk 到达时 target 就“结束”了，after 被提前执行，但实际流还没结束。更关键的是，流式场景下很多需求根本拿不到完整结果。

所以在设计里，中间件还要支持一个on_chunk钩子，专门处理流式 chunk：

class BaseMiddleware: async def on_chunk(self, ctx: MiddlewareContext, chunk: Any) -> Any: return chunk

on_chunk的返回值会被透传给下游，所以可以在这一层做流式内容过滤或改写。before 仍然在流式请求开始前执行，after 则要等流完全结束之后再执行。如果你的项目用了流式接口，这个点必须一开始就考虑到，否则后面补会非常痛苦。

3. 落地实现：中间件管理器与 LangChain 接入

3.1 项目组织与核心文件

我用一个独立的 Python 包来放这套东西，目录结构不复杂，核心就几个文件：

langchain_v1_middleware/ __init__.py context.py manager.py middlewares/ __init__.py logging_middleware.py rate_limit_middleware.py redact_middleware.py

context.py放MiddlewareContext和BaseMiddleware，manager.py放调度器和 Runnable 包装逻辑，middlewares/放具体业务中间件。拆开的原因很简单：调度器是通用能力，不应该跟具体业务耦合。比如限流中间件可以换成 Redis 版本，调度器一行都不用改。

3.2 用递归实现洋葱调度

调度器是整套方案的核心。为什么用递归而不是 for 循环？因为 after 需要逆序执行，递归的调用栈天然满足“先进后出”。下面是简化版实现：

from typing import Any, Awaitable, Callable Target = Callable[[], Awaitable[Any]] class MiddlewareManager: def __init__(self): self._middlewares: list[BaseMiddleware] = [] def use(self, middleware: BaseMiddleware) -> "MiddlewareManager": self._middlewares.append(middleware) return self async def dispatch( self, ctx: MiddlewareContext, target: Target, ) -> Any: async def execute(index: int) -> Any: if index == len(self._middlewares): return await target() mw = self._middlewares[index] try: await mw.before(ctx) result = await execute(index + 1) ctx.outputs = result await mw.after(ctx) return ctx.outputs except Exception as exc: ctx.error = exc await mw.error(ctx) raise finally: await mw.finally(ctx) return await execute(0)

注意after里修改ctx.outputs之后，execute返回的是ctx.outputs，这样外层中间件拿到的就是内层修改后的结果。如果直接返回result，脱敏中间件改完输出后日志中间件看到的还是原始内容。

before 阶段抛异常时，异常会进入同一中间件的 except 分支，触发 error，然后继续向上抛，外层中间件也会捕获并执行各自的 error。这样所有已经“经过”的中间件都能感知异常，不会出现错误没人管的情况。

3.3 把包装器变成标准 Runnable

要让 LangChain 的链接受中间件管理，最直接的方式是把链包装成一个新的 Runnable。我用RunnableLambda做了一层包装：

from langchain_core.runnables import Runnable, RunnableLambda class MiddlewareManager: def wrap(self, runnable: Runnable) -> Runnable: async def _invoke(input, config=None): ctx = MiddlewareContext( config=config or {}, inputs={"input": input}, ) return await self.dispatch( ctx, lambda: runnable.ainvoke(input, config=config), ) return RunnableLambda(_invoke)

这里的input对 LangChain 来说不一定只是一个值，它可能是包含多个字段的 dict，比如{"subject": "程序员"}。我在MiddlewareContext.inputs里统一包成{"input": input}，中间件要取原始输入时直接用ctx.inputs["input"]。

RunnableLambda的优点是接入快，但它不会完整保留原始 Runnable 的流式行为。如果你的链对外只提供ainvoke，这个方案完全够用。如果要完整支持stream和astream，建议后面实现一个自定义 Runnable 子类，把ainvoke、astream分别映射到不同的调度器上。这不算复杂，但细节多，可以参考 LangChain Runnable 的协议方法逐步补。

3.4 一个能跑的完整示例

下面用一个模拟模型来演示整套流程，避免涉及具体模型厂商的 SDK。所有能挂到链上的模型 Runnable 用法都一样，所以你只需要把模拟模型换成你实际用的模型实现即可。

import asyncio import time from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableLambda from langchain_v1_middleware.context import BaseMiddleware, MiddlewareContext def fake_model(prompt_value): return "模拟模型回答：这是一个关于程序员的笑话。" class LoggingMiddleware(BaseMiddleware): async def before(self, ctx: MiddlewareContext) -> None: ctx.metadata["started_at"] = time.monotonic() print(f"[before] 输入: {ctx.inputs}") async def after(self, ctx: MiddlewareContext) -> None: elapsed_ms = (time.monotonic() - ctx.metadata["started_at"]) * 1000 print(f"[after] 输出: {ctx.outputs}, 耗时: {elapsed_ms:.1f}ms") class RedactMiddleware(BaseMiddleware): async def after(self, ctx: MiddlewareContext) -> None: text = ctx.outputs if isinstance(text, str): ctx.outputs = text.replace("手机号", "***") manager = MiddlewareManager() manager.use(LoggingMiddleware()) manager.use(RedactMiddleware()) prompt = ChatPromptTemplate.from_template("讲一个关于{subject}的笑话") chain = prompt | RunnableLambda(fake_model) wrapped = manager.wrap(chain) result = asyncio.run(wrapped.ainvoke({"subject": "程序员"})) print(result)

执行顺序是：LoggingMiddleware.before → RedactMiddleware.before（这里没有具体逻辑） → fake_model 执行 → RedactMiddleware.after 把输出里的“手机号”替换成***→ LoggingMiddleware.after 打印脱敏后的输出。这个例子看起来简单，但已经把中间件最核心的生命周期跑通了。后面加限流、租户注入、错误告警，都是在同一个结构里加新的中间件而已。

4. 实战中踩过的坑与排查笔记

4.1 流式输出时 after 被提前执行

我前面提到过这个问题，但实际踩的时候还是花了半天。现象是接口走astream后，日志中间件在第一个 chunk 出来时就打印了“输出完成”，而且打印的内容只有第一个 chunk，不是完整答案。原因就是包装器只处理了ainvoke，流式路径根本没有经过完整生命周期。

排查方法很简单，在日志中间件里同时打印invoke和stream两种调用方式进入的钩子顺序。如果发现after在target还没结束的时候就被触发，说明调度器没有区分普通调用和流式调用。

修复方案并不复杂：在流式模式下，把 target 改成异步生成器，逐 chunk 执行on_chunk，把所有 chunk 收集到一个列表里，等异步生成器耗尽后，再逆序执行 after。after 里看到的ctx.outputs就是完整输出。注意on_chunk的执行顺序应该和中间件注册顺序一致，不能搞成逆序，因为每个 chunk 要先经过外层脱敏，再经过内层业务处理。

4.2 contextvars 在线程池里传不过去

LangChain 在并发批量处理时，有可能把 Runnable 放到线程池里执行。这时候线程的 context 跟主协程的 context 不是同一个。我在中间件里用contextvars.ContextVar保存租户信息，结果经常出现 before 阶段设了值，模型调用阶段读出来是空的。

排查之后发现不是代码顺序问题，而是执行线程变了。解决办法有两个：第一，中间件状态尽量全部放进MiddlewareContext，不依赖contextvars，这是最省心的做法；第二，如果必须用contextvars，可以在包装 Runnable 时用contextvars.copy_context()把当前上下文整体传进去，但要注意线程池的上下文复制逻辑，处理不好反而引入更多坑。

从工程角度看，我强烈建议选第一种。MiddlewareContext本来就是为跨中间件传递信息设计的，把租户 ID、请求 ID、限流计数都放进去，既清晰又不会受线程切换影响。

4.3 重试会让 before 的副作用重复执行

有同事在链外面加了重试装饰器，超时自动重试三次。结果限流中间件的计数一次请求被加了三次，埋点数据也重复了。这属于典型的“before 里有副作用”问题。

我当时把重试逻辑也改造成了中间件，并且放在限流中间件外层。这样每次重试都会经过限流检查，不会出现“第一次请求过了限流，模型超时后重试时绕过限流”的情况。但埋点重复的问题还是存在，因为重试本身也是新的一次调用。

更根本的做法是给每次业务请求生成一个幂等request_id，中间件记录数据时统一带上这个 ID，上报端按 ID 去重。重试请求的request_id不变，即使 before 执行了三次，最终数据只算一次。这个方案也适合消息队列投递、日志归档等场景。

4.4 中间件和 LangChain callbacks 的职责重叠

LangChain 自带 CallbackHandler，很多需求它也能做，比如 inon_llm_start里打印日志、在on_llm_end里统计 token。那为什么不全都用 callbacks？区别在于 callback 本身更偏“观察”，它能看到事件发生，但很难直接改写请求参数、终止调用或修改输出。中间件则允许你对链路做控制。

如果两边同时用，日志会重复。比如on_llm_end里打了模型耗时，中间件的 after 里又打一遍，日志平台会出现两条几乎一样的数据。我最后约定：所有跟可观测性相关的事件，比如 token 统计、模型耗时、调用链追踪，统一走 callbacks；所有跟数据变更和流程控制相关的逻辑，比如脱敏、限流、租户注入、短路返回，统一走中间件。这个分工让两边都不越界，排查问题时也很清晰。

5. 这套中间件在实际项目里省下的工作

5.1 从“到处打补丁”变成“注册即生效”

接入中间件管理器之后，新拦截需求的上线方式完全变了。以前要改链的结构、写单元测试、重新验证所有分支；现在只需要写一个新的中间件类，在统一入口处manager.use(新中间件())就完事。我在项目里大概估算过，一个简单的日志需求从开发到上线，以前至少要半天，现在基本半小时内能搞定，其中大头还是写测试用例。

更重要的是，链上的 RunnableLambda 数量明显减少了。业务链重新变回prompt | model | parser的正常结构，横切逻辑全部收敛到中间件层。新同事维护起来再也不需要关心“为什么这个链上多了一层奇怪的包装”。这对团队来说，比省那点开发时间更有价值。

5.2 后续可以继续扩展的方向

这套结构稳定下来之后，可扩展空间很大。我目前计划做三件事：第一，把中间件注册信息放到配置中心，按租户动态组装中间件列表，实现不同租户走不同的过滤和限流策略；第二，把限流中间件换成分布式实现，用外部存储做滑动窗口，而不是只靠单机内存；第三，把 after/error 阶段的关键事件投递到消息队列，统一用于审计和计费。

另外，如果你们公司在做 Agent 类应用，中间件还可以用来拦截工具调用前的权限检查。方向是一致的：把流程控制从业务代码里抽出来，放到一条清晰的中间件链里。这套方案我实际用了两三个月，稳定性比我预想的好，最大的收获不是代码本身，而是“统一生命周期”这个思路带来的维护效率提升。