LangChain LCEL 与 Runnable 实践：把 LLM 应用链路写成可组合协议

LangChain 里最容易被低估的一层，不是 Prompt，也不是 Model，而是 Runnable。

Prompt、Model、OutputParser、Retriever、Tool、普通 Python 函数，只要进入 LangChain 的组合体系，最终都会围绕同一套可运行协议工作：可以 invoke，可以 batch，可以 stream，可以异步执行，可以被追踪，可以接收运行配置，也可以继续和其他步骤组合。

LCEL，也就是 LangChain Expression Language，是这套协议的表达方式。它用 | 把多个 Runnable 串起来，让一段 LLM 应用链路从“手动调用一堆函数”变成“声明一条数据流”。

不过要先说清楚边界：当前 LangChain 的主线已经明显转向 create_agent、middleware、LangGraph 和 LangSmith。LCEL 不是用来替代 Agent 的。它更适合那些边界清楚、流程稳定、步骤可预测的链路，例如分类、摘要、RAG 问答、结构化抽取、数据清洗、批量生成和轻量编排。只要开始出现复杂状态、循环、多 Agent、人工审批、长任务恢复，就应该考虑 LangGraph 或 Agent harness。

为什么需要 Runnable

如果直接写模型调用，代码通常长这样：

from langchain_openai import ChatOpenAI


model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

response = model.invoke("解释一下 LCEL 是什么")

print(response.content)

这能跑，但真实业务不会停在一次模型调用。你很快会加入：

• Prompt 模板。
• 输出解析。
• 检索上下文。
• 多输入字段组装。
• 批量调用。
• 流式输出。
• 超时和重试。
• trace、tags、metadata。
• fallback 模型。
• 线上日志和评估样本。

如果每个地方都手动组织这些步骤，代码会非常快地散掉。Runnable 的价值就是给所有这些步骤一个统一接口。

可以把 Runnable 理解为 LangChain 里的“可运行单元”：

输入 -> Runnable -> 输出

更重要的是，Runnable 可以继续组合：

输入 -> Runnable A -> Runnable B -> Runnable C -> 输出

这就是 LCEL 的基础。

LCEL 的最小链路

最常见的链路是 Prompt、Model、Parser：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个严谨、简洁的 Python 后端工程师。"),
    ("human", "{question}"),
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

chain = prompt | model | parser

answer = chain.invoke({
    "question": "Runnable 在 LangChain 中解决什么问题？"
})

print(answer)

这里的 | 不是普通字符串拼接，而是把左侧输出交给右侧输入。

这条链路可以拆开看：

prompt_value = prompt.invoke({
    "question": "Runnable 在 LangChain 中解决什么问题？"
})

message = model.invoke(prompt_value)

answer = parser.invoke(message)

LCEL 只是把这三步声明成一条可运行链。好处是：组合后的 chain 本身仍然是 Runnable，所以它也可以继续 invoke、batch、stream，也可以传运行配置、接 LangSmith trace，或者作为更大链路的一部分。

Runnable 的统一接口

大多数 Runnable 都支持几类核心方法。

方法	作用	典型场景
invoke	单次同步调用	Web API、普通脚本
ainvoke	单次异步调用	FastAPI、异步任务
batch	批量同步调用	批量摘要、离线处理
abatch	批量异步调用	高并发 I/O 任务
stream	同步流式输出	命令行、SSE
astream	异步流式输出	异步 Web 服务

这就是 Runnable 比“普通函数调用”更适合 LLM 应用的地方。你不是只得到一个能跑的函数，而是得到一个可被 LangChain 生态识别和增强的协议对象。

例如批量调用：

questions = [
    {"question": "什么是 Prompt？"},
    {"question": "什么是 Model？"},
    {"question": "什么是 OutputParser？"},
]

answers = chain.batch(questions)

for answer in answers:
    print(answer)

流式调用：

for chunk in chain.stream({
    "question": "用一段话解释 LCEL 的价值"
}):
    print(chunk, end="", flush=True)

异步调用：

async def answer_question(question: str) -> str:
    return await chain.ainvoke({"question": question})

生产里不要把同步、异步、批处理、流式输出分别写成四套业务逻辑。能用同一条 Runnable 链路承载，就能减少很多重复代码和不一致行为。

RunnableSequence：顺序组合

prompt | model | parser 生成的就是一个顺序链路，底层可以理解为 RunnableSequence。

顺序组合适合这种数据流：

原始输入 -> 构造 Prompt -> 调用模型 -> 解析输出 -> 返回结果

例如文章摘要：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


summary_prompt = ChatPromptTemplate.from_messages([
    (
        "system",
        "你是技术内容编辑。请保留关键事实，不要编造，不要过度扩写。",
    ),
    ("human", "请把下面内容压缩成 5 个要点：\n\n{content}"),
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

summary_chain = summary_prompt | model | StrOutputParser()

summary = summary_chain.invoke({
    "content": "这里放一段很长的业务文档..."
})

顺序链路最重要的设计原则是：每一步输入输出要清楚。不要让某个步骤既负责组装上下文，又负责调模型，又负责解析，又负责写数据库。Runnable 可以组合，但不意味着每个 Runnable 都应该变成大杂烩。

RunnableParallel：并行分支

有时候你希望同一个输入同时走多个分支，然后把结果合并。

LCEL 里可以用 dict 表达并行分支：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

summary_prompt = ChatPromptTemplate.from_template(
    "请总结下面内容，控制在 80 字以内：\n\n{content}"
)

keywords_prompt = ChatPromptTemplate.from_template(
    "请提取下面内容的 5 个关键词，用逗号分隔：\n\n{content}"
)

chain = {
    "summary": summary_prompt | model | parser,
    "keywords": keywords_prompt | model | parser,
}

result = chain.invoke({
    "content": "LangChain LCEL 可以把 Prompt、Model、Parser 组合成可运行链路。"
})

print(result["summary"])
print(result["keywords"])

在 LCEL 中，dict 会被转换为并行 Runnable。它会把同一份输入分发给多个分支，最后返回一个 dict。

并行分支适合：

• 同一文档同时生成摘要、标题、关键词。
• 同一问题同时走多个检索器。
• 同一输入同时做分类和风险判断。
• 多个低耦合模型调用并行执行。

生产里要注意，模型并行不等于免费加速。并行会增加瞬时 token 消耗和 rate limit 压力。如果 provider 对并发限制严格，需要加队列、限速器或批处理策略。

RunnablePassthrough：保留原始输入

很多链路需要一边保留用户原始输入，一边计算中间字段。

例如 RAG 问答里，我们需要：

• 保留 question。
• 用 question 去检索上下文。
• 把 question 和 context 一起送进 Prompt。

可以用 RunnablePassthrough：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI


def fake_retriever(question: str) -> str:
    return "LCEL 是 LangChain 中组合 Runnable 的表达方式。"


prompt = ChatPromptTemplate.from_messages([
    (
        "system",
        """
你是企业知识库助手。
请只根据上下文回答，不知道就说不知道。

上下文：
{context}
""",
    ),
    ("human", "{question}"),
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

rag_chain = (
    {
        "question": RunnablePassthrough(),
        "context": fake_retriever,
    }
    | prompt
    | model
    | StrOutputParser()
)

answer = rag_chain.invoke("LCEL 是什么？")

print(answer)

这里输入是一个字符串。question 分支直接透传原始输入，context 分支调用检索函数。两个字段合并后进入 Prompt。

真实项目里，fake_retriever 会替换成向量库 retriever 或查询服务。它也可以是 Runnable，因此能继续接 batch、trace 和配置。

RunnableLambda：把普通函数放进链路

并不是所有步骤都来自 LangChain 组件。很多业务逻辑只是普通 Python 函数，比如格式化输入、清洗字段、做权限过滤、拼接检索结果。

可以用 RunnableLambda 包一层：

from langchain_core.runnables import RunnableLambda


def normalize_question(payload: dict) -> dict:
    return {
        "question": payload["question"].strip(),
        "locale": payload.get("locale", "zh-CN"),
    }


normalize = RunnableLambda(normalize_question)

然后放进链路：

chain = normalize | prompt | model | parser

result = chain.invoke({
    "question": "  LCEL 有什么用？  ",
})

生产里要注意，RunnableLambda 很方便，也很容易被滥用。不要把大量不可见副作用塞进去，例如写数据库、发消息、扣费、删数据。LCEL 链路最好保持可理解、可重放、可测试。高风险副作用应该放在明确的 service 或 tool 层，并做好幂等、审计和权限。

assign：给输入增加字段

RunnablePassthrough.assign 适合在保留原始输入的同时追加新字段。

from langchain_core.runnables import RunnablePassthrough


def load_user_profile(payload: dict) -> str:
    user_id = payload["user_id"]
    return f"用户 {user_id} 是企业版客户。"


chain = (
    RunnablePassthrough.assign(profile=load_user_profile)
    | prompt
    | model
    | parser
)

result = chain.invoke({
    "user_id": "u_1001",
    "question": "我可以使用高级报表吗？"
})

经过 assign 后，后续 Prompt 可以同时拿到原始字段和新增字段：

prompt = ChatPromptTemplate.from_messages([
    (
        "system",
        "你是产品客服。请结合用户画像回答。\n\n用户画像：{profile}",
    ),
    ("human", "{question}"),
])

这比手动到处复制 dict 更清晰，也更容易在链路图里看出上下文是在哪里被注入的。

RunnableConfig：把运行配置和业务输入分开

业务输入应该进入 chain 的 input；运行配置应该进入 config。

例如：

result = chain.invoke(
    {"question": "如何排查数据库连接池耗尽？"},
    config={
        "tags": ["chat", "backend"],
        "metadata": {
            "tenant_id": "tenant_001",
            "prompt_version": "backend-qa-2026-08-06-v1",
        },
        "run_name": "backend_qa_chain",
    },
)

这样做的好处是，Prompt 不会混入 trace 信息，业务 schema 也不会被观测字段污染。

建议放进 config 的内容：

• tags
• run name
• trace metadata
• tenant id
• user id
• prompt version
• experiment id

不建议放进 Prompt 输入的内容：

• 只用于观测的字段。
• 不应该被模型看到的内部 ID。
• 权限、密钥、成本策略等敏感信息。

配置和输入分开，是 LLM 应用工程化里一个很小但很关键的习惯。

with_config：给链路固定默认配置

如果某条链路有稳定的名称、标签或 metadata，可以使用 with_config：

qa_chain = (
    prompt
    | model
    | parser
).with_config({
    "run_name": "knowledge_base_qa",
    "tags": ["rag", "qa"],
})

调用时仍然可以传入本次请求的动态 metadata：

answer = qa_chain.invoke(
    {"question": "LCEL 和 Agent 有什么区别？"},
    config={
        "metadata": {
            "tenant_id": "tenant_001",
            "request_id": "req_abc",
        }
    },
)

生产系统里，链路名称要稳定。否则 LangSmith、日志系统和指标平台里会出现一堆临时名字，后面做质量分析和成本统计会很痛。

bind：给模型绑定参数

有些 provider 参数不想每次调用都写，可以使用 bind 创建一个带默认参数的新 Runnable。

base_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

json_model = base_model.bind(
    response_format={"type": "json_object"}
)

chain = prompt | json_model | parser

bind 的价值是把调用参数固定到某个链路，而不是把 provider 细节散落在业务代码里。

但要注意 provider 差异。不同模型对参数名、结构化输出、tool calling、流式输出的支持都可能不同。跨模型切换时，要用测试集验证链路行为，而不是只保证代码不报错。

LCEL 与 RAG

LCEL 很适合表达简单 RAG：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI


def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)


prompt = ChatPromptTemplate.from_messages([
    (
        "system",
        """
你是企业知识库问答助手。
请只根据给定上下文回答问题。
如果上下文中没有答案，请明确说不知道。

上下文：
{context}
""",
    ),
    ("human", "{question}"),
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | prompt
    | model
    | StrOutputParser()
)

answer = rag_chain.invoke("数据库迁移上线前要检查什么？")

这条链路很清楚：

问题 -> 检索 -> 格式化文档 -> Prompt -> Model -> 文本答案

如果只是普通知识库问答，这种写法足够清晰。不要一开始就把所有 RAG 都写成复杂 Agent。Agent 适合需要工具选择、循环决策和多步操作的场景；普通 RAG 更需要稳定检索、上下文控制、引用来源和评估。

LCEL 与 Agent 的边界

LCEL 和 Agent 的区别，不是哪个更高级，而是控制方式不同。

LCEL 更适合确定性流程：

• 输入字段已知。
• 步骤顺序固定。
• 每一步做什么由开发者决定。
• 模型主要负责生成或抽取。
• 链路容易测试和回放。

Agent 更适合动态流程：

• 模型需要决定是否调用工具。
• 工具调用次数不固定。
• 中间状态会影响下一步。
• 可能需要 human-in-the-loop。
• 需要短期记忆、middleware、上下文工程。

LangGraph 更适合复杂状态流：

• 多节点。
• 分支。
• 循环。
• 持久化。
• 恢复执行。
• 人工审批。
• 多 Agent 编排。

所以实际选型可以很朴素：

需求	更适合
分类、摘要、抽取、改写	LCEL
普通 RAG 问答	LCEL
模型自己选择工具	Agent
多步状态机和人工审批	LangGraph
多 Agent 协作	LangGraph 或 Deep Agents
简单聊天助手	create_agent 或轻量 LCEL

不要为了“显得智能”把固定流程写成 Agent，也不要为了“代码短”把复杂状态硬塞进 LCEL。

错误处理与 fallback

Runnable 支持组合，也意味着错误会沿着链路传播。

生产里要思考几类错误：

• Prompt 输入缺字段。
• Retriever 查询失败。
• 模型超时或限流。
• OutputParser 解析失败。
• 下游服务不可用。
• 链路中某个普通函数抛异常。

对于低风险链路，可以在服务层做统一捕获：

def run_chain_safely(chain, payload: dict):
    try:
        return chain.invoke(payload)
    except Exception as exc:
        raise RuntimeError("AI 链路执行失败") from exc

但更成熟的做法是按错误类型设计策略：

• 模型超时：重试或切 fallback 模型。
• 限流：进入队列或返回稍后再试。
• 解析失败：记录原始输出摘要，必要时换强模型重试。
• 检索失败：返回无上下文回答或明确提示知识库暂不可用。
• 业务校验失败：拒绝执行高风险动作。

不要把所有异常都吞掉，也不要把原始异常直接暴露给前端。API 层应该返回稳定错误码，内部保留足够 trace 信息。

流式输出的设计

LCEL 链路可以 stream，但不是所有步骤都天然适合流式。

例如：

for chunk in chain.stream({"question": "解释 Runnable 的价值"}):
    print(chunk, end="", flush=True)

如果链路最后是 StrOutputParser，通常能很好地逐步输出文本。

但如果最后是 Pydantic parser，往往要等完整输出结束后才能校验对象。这个时候不要强行把半截 JSON 当成业务结果推给前端。

建议按接口类型拆：

• 聊天展示：使用流式文本。
• Agent 过程：使用事件流展示步骤进展。
• 结构化决策：等待完整对象。
• 后台任务：记录状态，不一定需要 token 级流式。

对前端来说，流式输出不是简单地“不断 append 字符”。还要处理开始、增量、完成、错误、取消、超时和重试。后端最好定义稳定事件协议，而不是把模型 chunk 原样扔出去。

批处理与并发

batch 很适合离线任务：

items = [
    {"content": "第一篇文章内容..."},
    {"content": "第二篇文章内容..."},
    {"content": "第三篇文章内容..."},
]

results = summary_chain.batch(items)

但生产批处理要考虑：

• provider rate limit。
• 单条失败隔离。
• 输入过长截断。
• 重试次数。
• 任务进度记录。
• 成本预算。
• 结果幂等写入。

不要在用户请求里直接跑一个大 batch。Web 请求适合短链路，批量任务应该交给队列和 worker。

如果并发很高，还要把模型调用封装到限速层。LCEL 负责表达链路，不负责替你解决所有资源治理问题。

可测试性

LCEL 链路天然适合拆开测试。

Prompt 可以单独测试：

def test_prompt_variables():
    value = prompt.invoke({
        "question": "什么是 LCEL？"
    })

    messages = value.to_messages()

    assert len(messages) == 2
    assert messages[-1].content == "什么是 LCEL？"

Parser 可以单独测试：

def test_parser():
    assert parser.invoke("hello") == "hello"

普通函数可以单独测试：

def test_format_docs():
    docs = [FakeDoc("A"), FakeDoc("B")]

    assert format_docs(docs) == "A\n\nB"

端到端链路可以用 fake model 测：

from langchain_core.runnables import RunnableLambda


fake_model = RunnableLambda(lambda _: "这是一个固定回答")

test_chain = prompt | fake_model | parser

def test_chain_with_fake_model():
    result = test_chain.invoke({"question": "什么是 Runnable？"})

    assert result == "这是一个固定回答"

关键点是：不要所有测试都真实调用模型。真实模型调用适合做评估集和少量集成测试；单元测试应该尽量稳定、快速、便宜。

目录组织

生产项目里，不建议把 LCEL 链路直接写在 route 里。

可以按职责拆：

internal/
  ai/
    models.py
    prompts/
      qa.py
      summary.py
    parsers/
      qa.py
      summary.py
    retrievers/
      knowledge_base.py
    chains/
      qa.py
      summary.py
    schemas/
      qa.py
    observability.py
  services/
    qa_service.py

示例：

# internal/ai/chains/qa.py
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from internal.ai.models import get_default_chat_model
from internal.ai.prompts.qa import qa_prompt
from internal.ai.retrievers.knowledge_base import get_retriever


def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)


def build_qa_chain():
    retriever = get_retriever()

    return (
        {
            "context": retriever | format_docs,
            "question": RunnablePassthrough(),
        }
        | qa_prompt
        | get_default_chat_model()
        | StrOutputParser()
    ).with_config({
        "run_name": "knowledge_base_qa",
        "tags": ["rag", "qa"],
    })

服务层：

# internal/services/qa_service.py
from internal.ai.chains.qa import build_qa_chain


def answer_question(question: str, tenant_id: str) -> str:
    chain = build_qa_chain()

    return chain.invoke(
        question,
        config={
            "metadata": {
                "tenant_id": tenant_id,
            }
        },
    )

这样 route 层只关心 HTTP，service 层只关心业务入口，AI 模块负责 Prompt、Model、Retriever 和 Parser 的组合。

常见坑

把 LCEL 写得过长

一条链路如果横跨几十行、包含多个 lambda、多个嵌套 dict，就已经不容易维护了。可以把子链路命名拆出来：

context_chain = retriever | format_docs
answer_chain = prompt | model | parser

rag_chain = {
    "context": context_chain,
    "question": RunnablePassthrough(),
} | answer_chain

命名不是形式主义。可读的链路名能让你在 trace 里快速定位问题。

在链路中隐藏副作用

RunnableLambda 可以执行任意 Python 代码，但不代表应该把副作用塞进去。

危险做法包括：

• 在 lambda 里写数据库。
• 在 lambda 里发短信。
• 在 lambda 里扣费。
• 在 lambda 里修改权限。

这些动作应该放在明确的 service、tool 或 workflow 节点里，并配套权限、确认、幂等、审计。

输入输出类型不稳定

LCEL 靠上一步输出对接下一步输入。如果某一步有时返回字符串，有时返回 dict，后续就会出现很难排查的问题。

建议为关键链路写清楚输入输出：

class QAInput(BaseModel):
    question: str
    tenant_id: str


class QAOutput(BaseModel):
    answer: str
    sources: list[str]

即使不把 Pydantic 模型直接接进链路，也应该在服务边界做校验。

把 config 当 Prompt 输入

tenant_id、request_id、experiment_id 这类字段，经常只是为了观测或路由，不一定应该进入 Prompt。除非模型确实需要看到，否则放进 config metadata。

误以为 LCEL 会自动解决成本问题

LCEL 能让链路更清楚，但不会自动帮你控制 token。RAG 上下文过长、批处理过大、并行分支太多，都会让成本上升。成本控制仍然要靠上下文裁剪、限流、缓存、模型分层和评估。

生产 Checklist

上线前至少检查这些点：

• 链路输入 schema 是否稳定。
• Prompt 变量是否有测试覆盖。
• Model 参数是否从配置读取。
• Parser 是否有失败样本测试。
• RAG 上下文是否有 token 上限。
• batch 是否有并发和限速控制。
• 流式接口是否有错误事件。
• config 是否带 trace metadata。
• 高风险副作用是否没有藏在 LCEL 链路里。
• LangSmith 或日志系统是否能看到每一步。
• fallback 策略是否明确。

LCEL 写起来很轻，但生产治理不能轻。

小结

LCEL 的核心价值，是把 LLM 应用里的多个步骤写成可组合、可追踪、可批处理、可流式、可测试的 Runnable 链路。

它最适合稳定流程：Prompt 组装、模型调用、输出解析、RAG 问答、批量处理和轻量并行。它不适合承载复杂状态机、高风险副作用、多 Agent 协作和长任务恢复；这些应该交给 Agent、middleware 或 LangGraph。

如果只记住一句话：LCEL 负责把确定性链路写清楚，Runnable 负责让每个步骤遵守同一套运行协议。把这个边界守住，LangChain 代码就不会在需求变复杂后散成一团。