LangChain 中 Runnable 与 Middleware 的定位及选型问题

[hanbaobao003]

目前 LangChain 提供了多种内置中间件，也支持自定义中间件。部分过去使用 Runnable 实现的流程，现在似乎也可以通过 Middleware 完成。
例如，若 Agent 执行前必须进行 RAG 检索，过去可以将 RAG 封装为 Runnable，通过链式调用接入流程；现在也可以将该逻辑放入 before_agent 中间件。
想请教：

1. Runnable 是否正在被逐步弱化？
2. Runnable 与 Middleware 分别适用于哪些场景？
3. 在生产项目中，应该根据什么标准选择二者？

[webup]

感谢提问。这个问题把 Runnable、Middleware、LangGraph 和 Deep Agents 这几个容易混在一起的层次问到了一起。

直接回答你关心的三点：

1. Runnable 没有被弱化。它仍然是 LangChain 最基础的执行与组合协议，模型、Retriever、Chain、LangGraph，以及最终创建出来的 Agent，都通过这套接口运行。
2. Runnable 和 Middleware 不是二选一。Runnable 适合表达明确的数据流和业务步骤；Middleware 适合给 Agent 增加生命周期能力。由模型决定是否执行的动作做成 Tool；标准 Agent Loop 表达不了的复杂编排，再用 LangGraph。
3. 生产项目通常从 LangChain 的 create_agent() 加预制 Middleware 开始。任务需要规划、文件系统、上下文管理、Skills 和子 Agent 等深度能力时，可以用 Deep Agents；需要自己控制复杂拓扑、循环和恢复边界时，再直接使用 LangGraph。

Runnable、LangGraph 和 Agent 的关系

Runnable 是统一执行接口，提供 invoke、ainvoke、stream、batch 等调用方式。Prompt、模型、Retriever 和 Chain 都可以是 Runnable。

LangGraph 负责把执行单元编排成图。图里的节点既可以放 Runnable，也可以放普通 Python 函数：

from langchain_core.runnables import RunnableLambda

retrieve_node = RunnableLambda(
    lambda state: {
        "docs": retriever.invoke(state["question"])
    }
)

def format_node(state):
    return {
        "context": "\n\n".join(
            doc.page_content for doc in state["docs"]
        )
    }

builder.add_node("retrieve", retrieve_node)  # Runnable
builder.add_node("format", format_node)      # 普通 Python 函数

LangGraph 经过 compile() 后得到的 CompiledGraph 仍然符合 Runnable 接口，所以可以继续调用 invoke()、stream()，也能作为一个节点放进更大的图里。

LangChain 的 create_agent() 底层就是用 LangGraph 构建标准的模型与工具循环，因此它返回的 Agent 对象本身也是 Runnable。

几层关系可以这样看：

• Runnable 定义统一的执行和组合方式
• LangGraph 用 Runnable 或 Python 函数搭建流程
• LangChain 在 LangGraph 上预制了常见的 Agent Loop
• Middleware 在 Agent Loop 的生命周期位置增加能力

所以 Middleware 不会替代 Runnable，它们本来就不在同一层。Runnable API；LangGraph 官方说明

Middleware 负责什么

Middleware 管的是“Agent 运行到这里时，需要额外做什么”。比如：

• before_agent：输入校验、加载运行上下文
• before_model：整理消息、准备模型上下文
• wrap_model_call：重试、缓存、模型降级
• wrap_tool_call：工具权限、审计、异常处理
• after_agent：结果检查、运行记录

同一段逻辑有时既能写成 Runnable，也能放进 Middleware，主要看它想表达什么。

比如“先检索，再生成报告”是明确的业务步骤，用 Runnable 或 LangGraph 更直观；“每次模型调用失败后重试两次”是 Agent 的统一行为，用 Middleware 更合适。

LangChain 现在的预制 Middleware 已经覆盖了大部分常见需求，包括重试与降级、上下文管理、HITL、调用限制、To-do、工具选择、Filesystem、File Search、Subagent、Shell、Rubric 等。Filesystem 和 Subagent 也已经是标准预制能力。常规 Agent 先用官方预制；缺少业务规则再写自定义 Middleware；标准 Agent Loop 表达不了时，再下沉到 LangGraph。官方预制 Middleware 列表

Deep Agents 现在解决什么问题

Deep Agents 的价值不在于独占某几个 Middleware，而在于按深度任务的需要，把一批能力装配好了。

研究或编码 Agent 往往要同时处理任务规划、文件系统、上下文压缩、Skills、子 Agent、长期记忆，以及配合这些工具的系统提示词。只用 LangChain 也能搭，但需要自己选择组件、安排 Middleware 顺序、准备 Backend，再处理这些能力之间的配合。

Deep Agents 已经做了这层装配，更适合执行时间长、步骤多、中间产物也多的任务。普通问答或简单工具调用没必要直接上 Deep Agents，create_agent() 加几个 Middleware 通常更轻。

Deep Agents 中的部分 Middleware 也可以单独拿出来给普通 LangChain Agent 使用。Skills 就是一个典型例子：如果只想增加 Skills，可以使用 deepagents.middleware.SkillsMiddleware，不必把应用整体切换到 create_deep_agent()。

从选型上看，我会按这个顺序来：

• 常规 Agent 优先用 LangChain
• 深度、多步骤、自主执行任务使用 Deep Agents
• LangChain 的标准 Agent Loop 表达不了，需要低层编排、持久化和精细恢复控制时，使用 LangGraph

这三者经常会叠在一起，因为 LangChain 和 Deep Agents 底层本来就运行在 LangGraph 上。官方产品分层说明

回到 RAG 这个例子

RAG 放在哪里，要看检索在流程中扮演什么角色。

1. 每次都要先检索

如果每次生成前都必须检索，我更倾向于把 RAG 放在 Agent 外部，用 Runnable 把依赖关系写清楚：

from langchain_core.runnables import RunnableLambda

def prepare_agent_input(question: str):
    docs = retriever.invoke(question)
    context = "\n\n".join(doc.page_content for doc in docs)
    return {
        "messages": [{
            "role": "user",
            "content": f"参考资料：\n{context}\n\n问题：{question}",
        }]
    }

rag_agent = RunnableLambda(prepare_agent_input) | agent
result = rag_agent.invoke("Runnable 和 Middleware 有什么区别？")

这里的流程很明确：先检索，再调用 Agent。Retriever 可以单独测试、缓存，也能被其他应用复用。这属于 2-Step RAG。

2. 检索是 Agent 的启动上下文

如果检索只属于某个 Agent，是它每次启动时都要加载的上下文，放进 before_agent 也合理。Middleware 负责检索并更新 State，再通过动态 Prompt 把结果交给模型。

有一点容易忽略：before_agent 是每次 invocation 执行一次，不是整个 thread 只执行一次。如果应用每轮对话都会重新调用 Agent，它也会每轮重新检索。自定义 Middleware 文档

3. 由模型判断要不要检索

如果要让模型自己判断是否检索，把 Retriever 做成 Tool，交给 create_agent() 的标准模型与工具循环就可以了。这已经属于 Agentic RAG，但流程还比较简单，不一定需要自己搭 LangGraph。

4. RAG 自己就有分支和循环

如果检索后还要评估文档相关性，结果不好时改写问题并重新检索，RAG 本身已经是一套工作流。把这些逻辑都藏进 Middleware 会比较绕，直接用 LangGraph 更清楚。

LangChain 官方的 Custom Agentic RAG 教程 就是这套流程：

• generate_query_or_respond：决定直接回答还是检索
• retrieve：执行 Retriever Tool
• grade_documents：判断检索结果是否相关
• rewrite_question：结果不理想时改写问题并重新检索
• generate_answer：结果合格后生成答案

这里的条件路由和循环直接写在图里。图经过 compile() 后仍然是 Runnable，可以继续调用 invoke()、stream()，也可以嵌入更大的业务流程。

5. 基于 Deep Agents 构建 RAG

如果 RAG 只是“模型按需调用一次 Retriever”，普通 LangChain Agent 已经够用。Deep Agents 更适合检索之外还要做任务规划、并行调研、保存中间材料或汇总长报告的场景。实现方式并不特殊，仍然是把 Retriever 包成 Tool，再交给 create_deep_agent()：

from langchain.tools import tool
from deepagents import create_deep_agent

@tool
def retrieve_knowledge(query: str) -> str:
    """检索内部知识库，返回相关片段及来源。"""
    docs = retriever.invoke(query)
    return "\n\n".join(
        f"[{i}] {doc.page_content}\n"
        f"source: {doc.metadata.get('source', 'unknown')}"
        for i, doc in enumerate(docs, start=1)
    )

research_subagent = {
    "name": "knowledge-researcher",
    "description": "检索知识库并整理有来源的材料",
    "system_prompt": "先检索再整理，保留来源编号，不要补写资料中没有的事实。",
    "tools": [retrieve_knowledge],
}

agent = create_deep_agent(
    model="openai:gpt-5.4",
    tools=[retrieve_knowledge],
    subagents=[research_subagent],
    system_prompt=(
        "回答知识库问题时先检索；复杂问题拆分后交给子 Agent，"
        "最后根据检索结果合并答案，并保留来源编号。"
    ),
)

这时主 Agent 可以先列计划，把不同检索主题交给子 Agent，再借助文件系统保存和合并中间材料。官方的 Deep Research 教程 展示了相同的组织方式，只是示例里的检索源是 Web Search；换成向量库 Retriever、SQL、知识图谱或企业搜索 API，整体结构不变。

把 Deep Agents 也算进来，实际可以这样选：

• 固定执行“检索 → 生成”：用 Runnable，属于 2-Step RAG
• 检索是 Agent 每次启动时都要加载的上下文：用 before_agent Middleware
• 模型只需要自主决定是否调用 Retriever：把 Retriever 做成 Tool
• 还要做文档评分、问题改写和循环重试：用 LangGraph 构建 Custom Agentic RAG
• 检索之外还要结合任务规划、文件系统和子 Agent：把 Retriever 作为 Tool 交给 Deep Agents

实际项目里，我会先把 Retriever 保持成独立组件，再根据流程复杂度选择接法。这样以后从固定 RAG 改成 Agentic RAG，或者继续升级成自定义 LangGraph，都不会把检索逻辑绑死在某一层。