1. 从“管道”到“流水线”：Middleware在LangChain中的角色再认识

如果你用过LangChain，大概率已经写过类似chain.invoke({"input": "..."})这样的代码。表面上看，这只是一次简单的调用，但在LangChain内部，这背后其实是一条精心设计的“流水线”。这条流水线由一个个可复用的“组件”（如LLM、提示词模板、输出解析器）串联而成，而Middleware（中间件），就是这条流水线上可以随时插入、拔除的“质检站”或“加工站”。它不改变流水线的主体结构，却能对流过它的每一个“工件”（即请求和响应）进行观察、记录、修改甚至拦截。很多人把LangChain当作一个快速拼接LLM应用的工具包，却忽略了Middleware这个能让你深入掌控流程、实现企业级需求（如日志、监控、缓存、重试）的关键设计。今天，我们就来彻底拆解LangChain的Middleware机制，从概念到源码，从内置中间件到自定义实现，让你不仅能“用”，更能“驾驭”这条流水线。

2. 核心概念拆解：Middleware到底是什么？

在深入代码之前，我们必须先统一认知：在LangChain的语境下，Middleware到底是什么？它和装饰器、回调函数有什么区别？

2.1 定义与类比：流水线上的“拦截过滤器”

官方文档可能将Middleware描述为一种在调用链执行前后注入逻辑的机制。这个描述没错，但太抽象。我更愿意用一个更形象的类比：Web开发中的中间件。

在FastAPI或Express.js中，中间件函数可以接收请求（request）和响应（response）对象，在请求到达核心处理逻辑之前，或响应返回给客户端之后，执行一些通用操作，比如身份验证、日志记录、请求耗时统计。LangChain的Middleware与此高度相似。

• 请求（Request）： 对应LangChain中调用一个可运行对象（Runnable）时的输入字典。例如，{"input": "请总结这篇文章"},{"question": "什么是机器学习？", "context": "..."}。
• 响应（Response）： 对应调用完成后的输出。可能是一个字符串、一个字典，或一个包含生成内容、令牌使用量等元数据的AIMessage对象。
• 流水线（Pipeline）： 就是你的RunnableSequence（链）或任何一个实现了Runnable接口的对象。

Middleware就是挂载在这条流水线上的一个钩子（hook）。当数据（请求）流经这个钩子时，你可以做三件事：

1. 观察（Observe）： 记录请求内容、响应内容、耗时、令牌数。这是最常见的用途，用于监控和调试。
2. 修改（Transform）： 在请求到达核心逻辑前修改它（例如，为所有用户输入自动添加一个系统指令）；或在响应返回给用户前修改它（例如，对AI生成的内容进行敏感词过滤）。
3. 拦截（Intercept）： 基于某些条件（如输入包含敏感词、用户权限不足）提前返回一个响应，阻止请求继续向下游传递。

它与回调函数（Callbacks）的主要区别在于作用域和侵入性。回调函数通常与单次调用绑定，用于在特定事件（如on_llm_start,on_chain_end）发生时触发，更侧重于观测和日志，修改能力有限。而Middleware是绑定在Runnable对象本身的，一旦配置，所有流经该Runnable的调用都会自动应用中间件逻辑，它更底层，能力更强，可以直接操作请求/响应的原始数据。

2.2 LangChain中的Middleware实现载体：RunnableConfig

在代码层面，Middleware的功能是通过RunnableConfig这个配置对象来承载和传递的。RunnableConfig是一个类似字典的结构，你可以在调用任何Runnable时通过config参数传入。

其中，与Middleware直接相关的配置项是configurable字典，以及更具体的callbacks和middleware字段（在不同版本中可能有差异，但思想一致）。Middleware的实现类需要定义如何处理config中的这些信息。

关键点在于：Middleware的逻辑执行，依赖于Runnable内部对config的解析和应用。并不是你随便写个函数就能当中间件，它需要遵循LangChain定义的协议。

3. 内置Middleware实战：以日志和监控为例

LangChain提供了一些开箱即用的中间件，让我们先通过两个最实用的例子来感受一下。

3.1 使用LoggingMiddleware进行结构化日志记录

想象一下，你的AI客服在线上跑了一个月，突然有用户投诉回答有误。你怎么回溯当时发生了什么？原始的print语句散落在各处，难以关联。LoggingMiddleware就是为了解决这个问题。

import logging from langchain_core.runnables import RunnableLambda, RunnablePassthrough from langchain_core.middleware import LoggingMiddleware # 1. 设置一个结构化日志器 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # 2. 定义一个简单的链 def add_prefix(x: dict) -> dict: return {"output": f"Processed: {x['input']}"} chain = RunnablePassthrough() | RunnableLambda(add_prefix) # 3. 为链绑定LoggingMiddleware # 注意：这里演示的是概念。实际中，LoggingMiddleware可能需要以特定方式集成。 # 更常见的模式是在创建Runnable时通过`config`指定，或使用`with_config`方法。 # 以下是一个模拟其作用的示例： from langchain_core.runnables import ConfigurableFieldSpec, RunnableConfig from typing import Any, Dict class SimpleLogMiddleware: """一个简化的日志中间件示例""" def __init__(self, logger): self.logger = logger def __call__(self, request: Dict[str, Any], config: RunnableConfig): # 请求前日志 self.logger.info(f"Request received: {request}") # 这里本应调用真正的处理逻辑，我们模拟一个响应 # 实际中，你会调用下一个中间件或最终的runnable response = {"output": f"Mock response for {request.get('input')}"} # 响应后日志 self.logger.info(f"Response generated: {response}") return response # 使用自定义的简单中间件 logged_chain = chain.with_config(middleware=[SimpleLogMiddleware(logger)]) result = logged_chain.invoke({"input": "test query"}) print(result)

在实际的LoggingMiddleware中，它会自动捕获每次调用的输入、输出、开始时间、结束时间、错误信息等，并以结构化的格式（如JSON）输出到指定的日志处理器。这对于后续用ELK（Elasticsearch, Logstash, Kibana）或Datadog等工具进行聚合分析至关重要。

注意：LoggingMiddleware的具体API可能随版本变化。核心思想是，它通过with_config或装饰器模式嵌入到Runnable中，自动拦截流量并记录日志。

3.2 集成OpenAIMonitoringMiddleware实现深度LLM监控

如果你主要使用OpenAI的模型，那么OpenAIMonitoringMiddleware是一个神器。它需要与langchain-openai和phoenix（一个可观测性平台）配合使用。

# 假设已安装 langchain-openai, openai, arize-phoenix import os from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_community.callbacks.manager import get_openai_callback # 注意：以下导入路径为示例，实际请参考最新文档 # from langchain.middleware import OpenAIMonitoringMiddleware # 1. 启动Phoenix会话（用于可视化管理） import phoenix as px px.launch_app() # 2. 创建链 prompt = ChatPromptTemplate.from_template("用一句话解释{concept}") llm = ChatOpenAI(model="gpt-3.5-turbo") chain = prompt | llm | StrOutputParser() # 3. 传统方式：使用get_openai_callback获取token消耗（这只是回调，不是中间件） with get_openai_callback() as cb: result = chain.invoke({"concept": "量子计算"}) print(f"结果: {result}") print(f"消耗: {cb}") # 4. 中间件方式（概念性）：将链的调用自动追踪并发送到Phoenix # monitored_chain = chain.with_config(middleware=[OpenAIMonitoringMiddleware()]) # 调用monitored_chain.invoke(...)后，可以在Phoenix UI中看到详细的追踪信息，包括： # - 提示词（Prompt）和补全（Completion） # - Token使用量（Prompt Tokens, Completion Tokens, Total Tokens） # - 耗时（Latency） # - 模型名称和版本 # - 用户自定义的标签（如session_id, user_id）

这个中间件的强大之处在于，它提供了生产环境级别的可观测性。你可以在一个统一的仪表板里看到所有LLM调用的性能指标、成本分布，甚至可以对提示词和生成结果进行聚类分析，找出效果不佳的案例进行优化。它把LLM从一个黑盒变成了一个可度量、可分析的系统组件。

4. 从零构建自定义Middleware：实现请求修改与响应过滤

内置中间件虽好，但真实业务场景千奇百怪。这时，你需要自己动手打造中间件。我们来实现两个经典场景。

4.1 场景一：为所有用户输入自动注入系统指令（请求修改）

很多应用希望用户输入能隐含一些上下文，但又不想让用户每次手动输入。比如，一个翻译机器人，我们希望它默认将中文翻译成英文。

from typing import Any, Dict, Optional from langchain_core.runnables import Runnable, RunnableConfig from langchain_core.runnables.config import run_in_executor class SystemPromptInjectionMiddleware: """自动向LLM请求注入系统提示词的中间件""" def __init__(self, system_prompt: str): self.system_prompt = system_prompt async def ainvoke(self, input: Dict[str, Any], config: Optional[RunnableConfig] = None, **kwargs: Any) -> Dict[str, Any]: # 异步调用入口 return await run_in_executor(None, self._invoke_sync, input, config, **kwargs) def _invoke_sync(self, input: Dict[str, Any], config: Optional[RunnableConfig] = None, **kwargs: Any) -> Dict[str, Any]: # 核心逻辑：修改请求 # 假设我们的链的输入期望一个 `messages` 列表（OpenAI格式） # 如果输入是简单字典，我们将其转换为包含系统消息的messages if "messages" not in input: # 如果输入是文本，包装成用户消息 user_input = input.get("input", input.get("question", str(input))) new_messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": user_input} ] modified_input = {"messages": new_messages} else: # 如果已有messages，确保系统提示在最前面 messages = input["messages"] # 检查是否已有系统消息，如果没有则插入 if not any(msg.get("role") == "system" for msg in messages): messages.insert(0, {"role": "system", "content": self.system_prompt}) modified_input = input # 关键点：中间件需要调用链的下一环。 # 这里我们需要获取到原始的Runnable。通常，中间件是包装在Runnable外面的。 # 更标准的做法是继承或使用`Runnable.bind`或装饰器模式。 # 以下演示一个简化的、不完整的流程，实际应用需结合`Runnable`接口。 print(f"Middleware修改后的输入: {modified_input}") # 在实际中间件中，这里应该调用 `self.next.invoke(modified_input, config, **kwargs)` # 为了示例，我们直接返回一个模拟的修改后输入 return modified_input # 如何使用？我们需要一个能应用中间件的Runnable。 # 通常，我们会创建一个装饰器函数或使用`RunnableLambda`包装。 from langchain_core.runnables import RunnableLambda def inject_system_prompt(system_prompt: str): """一个返回应用了中间件的Runnable的工厂函数""" def wrapper(input_data: Dict): # 这是一个非常简化的实现，实际中间件逻辑比这复杂。 # 它模拟了修改输入的过程。 user_input = input_data.get("input", "") return { "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_input} ] } return RunnableLambda(wrapper) # 创建业务链 from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser llm = ChatOpenAI(model="gpt-3.5-turbo") parser = StrOutputParser() # 组合：先经过“中间件”处理输入，再交给LLM和解析器 business_chain = inject_system_prompt("你是一个专业的英文翻译，将中文翻译成地道英文。") | llm | parser # 测试 result = business_chain.invoke({"input": "今天的天气真好"}) print(f"翻译结果: {result}") # 预期输出关于天气好的英文句子

这个例子揭示了自定义中间件的核心：在invoke或ainvoke方法中，拦截输入，进行修改，然后（通常）将修改后的输入传递给被包装的原始Runnable（通过self.next或类似机制）。上面的示例是一个高度简化的概念演示，真实实现需要正确处理Runnable接口和config的传递。

4.2 场景二：对AI输出进行实时敏感词过滤（响应修改）

内容安全是AI应用上线的红线。我们可以在最终响应返回给用户前，加一道过滤中间件。

import re from typing import Any, Dict class ContentFilterMiddleware: """响应内容过滤中间件""" def __init__(self, blocked_words: list): self.blocked_words = blocked_words # 构建正则模式，匹配词汇（简单示例，实际可能更复杂） self.pattern = re.compile('|'.join(map(re.escape, blocked_words)), re.IGNORECASE) self.replacement = "[内容已过滤]" def filter_text(self, text: str) -> str: """过滤文本中的敏感词""" return self.pattern.sub(self.replacement, text) def _invoke_sync(self, input: Dict[str, Any], config, **kwargs): # 假设上一个环节（如LLM）返回的结果字典中有个 `content` 字段 # 1. 先调用下游链得到原始响应 # 这里需要调用 self.next.invoke(...)，我们模拟一个响应 raw_response = {"content": "这是一个包含不良信息和正常内容的测试句子。"} # 2. 修改响应 if "content" in raw_response and isinstance(raw_response["content"], str): raw_response["content"] = self.filter_text(raw_response["content"]) # 也可能需要处理其他字段，如 `generations` 列表中的内容 return raw_response # 模拟使用 filter_middleware = ContentFilterMiddleware(blocked_words=["不良信息", "测试"]) test_input = {} # 模拟调用 output = filter_middleware._invoke_sync(test_input, None) print(f"过滤后输出: {output}")

在实际集成中，你需要将这个ContentFilterMiddleware包装到你的链上，确保LLM的输出在返回前必须经过它。这比在应用层做过滤更彻底，因为它作用于LangChain框架层，无论你的调用来自API、CLI还是其他入口，都能确保过滤生效。

实操心得：编写自定义中间件时，最大的坑在于正确处理异步（async）和同步（sync）接口。LangChain的Runnable同时支持invoke（同步）和ainvoke（异步）。你的中间件最好也实现这两个方法，或者继承某个基类（如Runnable本身）来确保兼容性。另外，不要忘记传递RunnableConfig，里面可能包含了重要的上下文信息，如回调函数、并发限制配置等。

5. Middleware的集成模式：装饰器、包装器与配置注入

理解了如何写一个中间件，下一步就是如何将它“安装”到你的链上。LangChain提供了几种模式。

5.1 使用with_config方法进行临时配置

这是最灵活的方式，允许你在单次调用时动态添加中间件逻辑。不过，标准的config参数主要用于传递callbacks,tags,metadata等，直接传递完整的中间件对象可能不是标准做法。更常见的模式是将中间件逻辑封装成Runnable，然后通过管道（|）组合。

from langchain_core.runnables import RunnableLambda def logging_middleware(runnable): """一个将普通Runnable包装成带日志功能的Runnable的装饰器（工厂函数）""" def wrapper(input_data, config=None, **kwargs): print(f"[LOG] 输入: {input_data}") # 调用原始的runnable result = runnable.invoke(input_data, config, **kwargs) print(f"[LOG] 输出: {result}") return result # 返回一个新的RunnableLambda return RunnableLambda(wrapper) # 原始链 simple_chain = RunnableLambda(lambda x: {"answer": f"Hello {x['name']}"}) # 应用中间件 logged_chain = logging_middleware(simple_chain) result = logged_chain.invoke({"name": "World"}) print(result)

5.2 创建可配置的Middleware Chain（推荐）

更优雅和强大的方式是利用Runnable本身的组合性。既然每个Runnable都可以通过|连接，那么中间件本身也可以是一个Runnable。我们可以创建一个专门负责某种处理的“中间件Runnable”，然后把它插入到链的合适位置。

from langchain_core.runnables import RunnableSerializable from typing import Any, Dict class FilterRunnable(RunnableSerializable[Dict, Dict]): """一个可序列化的、专门用于过滤的Runnable""" def __init__(self, blocked_words: list): super().__init__() self.blocked_words = blocked_words def _invoke(self, input: Dict[str, Any], config, **kwargs) -> Dict[str, Any]: # 这里实现过滤逻辑 content = input.get("content", "") for word in self.blocked_words: if word in content: content = content.replace(word, "[FILTERED]") return {"content": content} # 通常还需要实现 invoke, ainvoke, batch 等方法，这里省略简化 # 构建一个完整的、包含“中间件”的链 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser prompt = ChatPromptTemplate.from_template("回答：{query}") llm = ChatOpenAI(model="gpt-3.5-turbo") parser = StrOutputParser() # 定义过滤逻辑（作为一个Runnable） profanity_filter = RunnableLambda( lambda x: {"content": x["content"].replace("不良词", "***")} if isinstance(x, dict) and "content" in x else x ) # 组装链：提示词 -> LLM -> 解析器 -> 过滤 full_chain = prompt | llm | parser | profanity_filter # 调用 result = full_chain.invoke({"query": "请说一句包含'不良词'的话。"}) print(result)

在这种模式下，profanity_filter就是一个标准的Runnable，它作为链的一环，承担了中间件的职责。这种方式的优点是清晰、可测试、易于组合。你可以像搭积木一样，把认证、日志、过滤、缓存等中间件Runnable插入到业务链的任何位置。

5.3 全局Middleware与Lambda层的权衡

你可能会想，有没有一种方式可以为整个应用的所有LangChain调用全局添加一个中间件？比如全局日志。在LangChain的当前设计中，没有像Django或FastAPI那样的全局中间件注册表。通常有两条路：

1. 封装基础Runnable：创建一个工厂函数，用来生成你的LLM或基础链，在这个工厂函数内部就为其包装上通用的中间件。所有业务代码都使用这个工厂函数创建的对象。

   def create_logged_llm(model_name): llm = ChatOpenAI(model=model_name) # 返回一个包装了日志功能的LLM Runnable return RunnableLambda(lambda x: (print(f"LLM Input: {x}"), llm.invoke(x))[1]) # 简化示例

2. 使用回调函数进行全局观测：对于纯粹的观测需求（不修改请求/响应），使用callbacks可能是更标准的选择。你可以在初始化时设置一个全局的CallbackManager，它会自动附加到所有调用中。

   from langchain_core.callbacks import CallbackManager, StdOutCallbackHandler manager = CallbackManager([StdOutCallbackHandler()]) llm = ChatOpenAI(model="gpt-3.5-turbo", callback_manager=manager) # 此后所有llm.invoke调用，StdOutCallbackHandler都会打印事件日志

选择哪种方式，取决于你的需求是修改还是观测，以及你希望控制的粒度是全局还是局部。

6. 高级应用与性能考量

当中间件变得复杂或数量增多时，就需要考虑一些高级话题了。

6.1 中间件执行顺序与责任链模式

如果你为同一个Runnable添加了多个中间件（例如，先日志、再认证、最后过滤），它们的执行顺序至关重要。这通常遵循责任链模式（Chain of Responsibility）。

在包装器模式下，最后添加的中间件最先执行其“请求前”逻辑，但最后执行其“响应后”逻辑（类似于栈，先进后出）。理解这一点对于调试至关重要。例如，如果认证中间件拒绝了请求，后续的日志中间件可能就记录不到LLM的响应。

# 概念性代码，展示顺序 def middleware_a(next_fn): def wrapper(input): print("A: before") result = next_fn(input) # 调用下一个中间件或核心逻辑 print("A: after") return result return wrapper def middleware_b(next_fn): def wrapper(input): print("B: before") result = next_fn(input) print("B: after") return result return wrapper def core_logic(input): print("Core: processing") return f"Processed {input}" # 组合：A包裹B，B包裹核心逻辑 chain = middleware_a(middleware_b(core_logic)) chain("data") # 输出： # A: before # B: before # Core: processing # B: after # A: after

在LangChain的Runnable组合中，顺序就是管道（|）的顺序。数据从左流向右，每个环节依次处理。

6.2 异步（Async）中间件的正确实现

生产环境多为异步。你的中间件必须完美支持异步，否则会成为性能瓶颈。

import asyncio from langchain_core.runnables import Runnable class AsyncLoggingMiddleware(Runnable): """一个支持异步的日志中间件""" def __init__(self, runnable: Runnable): self.runnable = runnable async def ainvoke(self, input, config=None, **kwargs): print(f"[Async Middleware] 异步请求开始: {input}") start_time = asyncio.get_event_loop().time() try: # 调用下游的异步接口 result = await self.runnable.ainvoke(input, config, **kwargs) end_time = asyncio.get_event_loop().time() print(f"[Async Middleware] 异步请求成功，耗时{end_time-start_time:.2f}s: {result}") return result except Exception as e: print(f"[Async Middleware] 异步请求失败: {e}") raise # 使用 async def main(): async_chain = AsyncLoggingMiddleware(some_async_runnable) await async_chain.ainvoke({"input": "test"}) asyncio.run(main())

关键点：

1. 中间件类实现ainvoke方法。
2. 内部使用await调用下游的ainvoke。
3. 使用异步友好的方式记录时间（如asyncio.get_event_loop().time()）。

6.3 性能开销与采样率

每个中间件都会增加额外的函数调用和逻辑处理，带来性能开销。在高并发场景下，需要谨慎评估。

• 轻量级操作：如添加一个请求ID、增加简单的标签，开销可忽略。
• 重量级操作：如写入数据库、调用外部API进行内容安全审核、计算复杂的特征，开销可能很大。

优化策略：

• 异步化：所有I/O操作（网络、磁盘）必须使用异步，避免阻塞事件循环。
• 采样：并非每次调用都需要全量日志。可以为日志中间件设置采样率（如1%），只记录少量请求用于监控和调试。
• 批处理：对于需要发送到外部系统的监控数据，可以在内存中缓冲，定期批量发送，而不是每次调用都发起网络请求。
• 开关配置：通过环境变量或配置中心控制中间件的开启/关闭，在压测或故障排查时可以临时关闭非关键中间件。

7. 排查Middleware失效的典型场景

即使代码写对了，中间件也可能不生效。以下是几个常见的坑。

场景一：中间件被意外“绕过”

# 错误示例：直接调用了被包装对象的方法，而不是包装器的方法 class MyMiddleware: def __init__(self, wrapped): self.wrapped = wrapped def invoke(self, input): # 这里应该包含中间件逻辑 return self.wrapped.invoke(input) # 如果直接返回这个，中间件逻辑就没执行 # 正确做法：在调用self.wrapped.invoke前后执行你的逻辑 def invoke(self, input): print("Before") result = self.wrapped.invoke(input) # 核心调用 print("After") return result

场景二：配置（Config）未正确传递中间件可能需要读取RunnableConfig中的信息（如用户ID、会话ID）。如果你在包装或调用时丢失了config参数，中间件就获取不到这些信息。

# 在自定义中间件的invoke方法中，务必接收并传递config def invoke(self, input, config=None, **kwargs): # 你的逻辑 result = self.next.invoke(input, config, **kwargs) # 注意传递config return result

场景三：异步/同步方法不匹配如果你用chain.invoke()（同步）调用，但你的中间件只实现了ainvoke()（异步），那么中间件逻辑可能不会执行。确保你的中间件同时覆盖了invoke和ainvoke方法，或者继承的基类已经做了默认处理。

场景四：Middleware在复杂链中的位置错误在chain_a | chain_b这样的组合中，如果你把中间件只加在chain_a上，那么流经chain_b的数据就不会被处理。你需要仔细考虑中间件应该作用于整个流水线，还是某个特定环节。通常，全局性的中间件（如日志）应该包装在最外层的链上。