AI Engineering Academy · 课时

使用 LCEL 构建链

使用管道运算符将提示模板、LLM 和输出解析器连接成 Runnable 序列，以同步和异步方式调用它，并检查中间输出。

第 2 / 4 课13 个步骤

使用 LCEL 构建链 是 CoddyKit 上的免费 AI Engineering Academy 课时。这是第 2 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。

什么是 LCEL？

LCEL（LangChain 表达式语言）是将 LangChain 组件组合成流水线的声明式方式。您无需实例化冗长的链类，而是使用 | 运算符连接可运行对象。LCEL 链会自动支持流式传输、异步处理、批处理和回退机制——只需使用管道语法，便可免费获得所有这些能力。

您的第一个 LCEL 链

最简单的 LCEL 链由 PromptTemplate、ChatModel 和 OutputParser 组成。每个组件都是可运行对象，管道运算符负责将它们连接起来。当您调用 chain.invoke() 时，输入会按顺序流经每个步骤，每一步的输出都会成为下一步的输入。

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

prompt = ChatPromptTemplate.from_template(
    'Write a product description for: {product}'
)
model = ChatOpenAI(model='gpt-4o-mini', temperature=0.7)
parser = StrOutputParser()

chain = prompt | model | parser

result = chain.invoke({'product': 'noise-cancelling headphones'})
print(result)

检查链的内部结构

LCEL 链会透明地公开自身结构。chain.steps 会按顺序列出各个组件。您可以调用 chain.input_schema 和 chain.output_schema，检查输入和输出的数据类型。invoke() 调用还接受一个 config 参数，用于向单次调用传递运行名称、标签和回调，而无需修改链的定义。

# Inspect chain structure
print('Steps:', chain.steps)
print('Input schema:', chain.input_schema.schema())
print('Output schema:', chain.output_schema.schema())

# Named run for observability
result = chain.invoke(
    {'product': 'coffee maker'},
    config={'run_name': 'product-desc-run', 'tags': ['production']}
)

同步调用与异步调用

LCEL 链为所有方法都提供同步和异步版本。invoke() 会阻塞调用线程。ainvoke() 是对应的异步方法，在 FastAPI 处理程序和其他异步上下文中应使用它，以避免阻塞事件循环。异步版本的语义完全相同，接口上的差异仅在于使用了 await。

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

chain = ChatPromptTemplate.from_template('Explain {topic}') | ChatOpenAI() | StrOutputParser()

async def main():
    # Async invocation
    result = await chain.ainvoke({'topic': 'quantum computing'})
    print(result)

asyncio.run(main())

让令牌流经链

LCEL 链会将流式传输一路传递下去。调用 chain.stream() 即可在模型生成令牌时接收它们。StrOutputParser 支持流式处理，会直接传递单个字符串块，而不是等待完整响应。这能让用户立即获得反馈，而不必长时间等待后才看到完整答案。

chain = (
    ChatPromptTemplate.from_template('Tell me about {topic} in detail.')
    | ChatOpenAI(model='gpt-4o-mini')
    | StrOutputParser()
)

# Print each token as it arrives
for chunk in chain.stream({'topic': 'neural networks'}):
    print(chunk, end='', flush=True)
print()  # newline at end

# Async streaming
async def stream_async():
    async for chunk in chain.astream({'topic': 'transformers'}):
        print(chunk, end='', flush=True)

批量处理多个输入

chain.batch() 接受输入字典列表，并以可配置的并发度处理它们。在内部，LangChain 会在线程中并行运行这些调用，因此批处理比循环调用 invoke() 快得多。max_concurrency 参数会限制并行 API 调用的数量，使其保持在速率限制范围内。结果返回的顺序与输入相同。

products = [
    {'product': 'laptop'},
    {'product': 'tablet'},
    {'product': 'smartwatch'},
    {'product': 'earbuds'},
]

# Process all 4 concurrently
results = chain.batch(products, config={'max_concurrency': 4})

for product, description in zip(products, results):
    print(f'{product["product"]}: {description[:80]}...')

RunnablePassthrough 与 RunnableLambda

RunnablePassthrough 会原样转发输入，适合在传递转换后数据的同时保留原始上下文。RunnableLambda 会将任意 Python 函数封装为可运行对象，让您可以将任意逻辑插入 LCEL 链中。两者结合使用，可以在链的步骤之间处理数据，而无需编写完整的自定义可运行对象类。

from langchain_core.runnables import RunnablePassthrough, RunnableLambda

def word_count(text: str) -> str:
    count = len(text.split())
    return f'{text}\n\n[Word count: {count}]'

chain = (
    ChatPromptTemplate.from_template('Write about {topic}')
    | ChatOpenAI(model='gpt-4o-mini')
    | StrOutputParser()
    | RunnableLambda(word_count)  # Add word count to output
)

result = chain.invoke({'topic': 'climate change'})
print(result)

使用 RunnablePassthrough 传递额外上下文

在 RAG 中，一种常见模式是将检索到的文档和用户问题同时传递给最终提示词。使用 RunnablePassthrough.assign() 可以向输入字典添加计算字段，同时保留原有键。这样，随着上下文对象在链中流转，您可以不断丰富它，使提示词模板能够访问所有数据。

from langchain_core.runnables import RunnablePassthrough

# Fake retriever for illustration
def retrieve_docs(question: str):
    return ['Doc 1 content', 'Doc 2 content']

retriever = RunnableLambda(retrieve_docs)

rag_chain = (
    RunnablePassthrough.assign(context=retriever)  # adds 'context' key
    | ChatPromptTemplate.from_template(
        'Context: {context}\n\nAnswer: {question}'
    )
    | ChatOpenAI()
    | StrOutputParser()
)

result = rag_chain.invoke({'question': 'What is machine learning?'})

为链添加回退机制

LCEL 链支持回退机制：如果主链引发异常，就尝试使用备用链。调用 chain.with_fallbacks([backup_chain]) 可以注册一个或多个回退链。LangChain 会先尝试主链，捕获异常后，再按顺序尝试每个回退链。这是优雅处理服务提供商故障或模型错误的标准方式。

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

primary_chain = (
    ChatPromptTemplate.from_template('Answer: {question}')
    | ChatOpenAI(model='gpt-4o')
    | StrOutputParser()
)

fallback_chain = (
    ChatPromptTemplate.from_template('Answer: {question}')
    | ChatAnthropic(model='claude-3-haiku-20240307')
    | StrOutputParser()
)

robust_chain = primary_chain.with_fallbacks([fallback_chain])
# If OpenAI fails, automatically tries Anthropic

使用 intermediate_steps 进行调试

借助 invoke() 的配置和 verbose=True 选项，调试 LCEL 链会更加容易。您还可以对任意链使用 .with_config({'verbose': True})，记录每个步骤的输入和输出。对于结构化调试，可以全局使用 set_verbose(True) 包装链，或使用 LangSmith 追踪，在可视化轨迹查看器中捕获每个中间值。

from langchain.globals import set_verbose, set_debug

# Verbose: logs LLM inputs/outputs
set_verbose(True)

# Debug: logs everything including prompts, parsers, and runnables
set_debug(True)

result = chain.invoke({'product': 'smart watch'})

# Or per-chain verbose config
chain_debug = chain.with_config({'verbose': True})
result = chain_debug.invoke({'product': 'laptop'})

运行时可配置的链

LCEL 链可以通过 ConfigurableField 支持运行时配置。您可以将模型参数——温度、模型名称、max_tokens——公开为可配置选项，而不必将它们硬编码。在调用时，传入 Configurable 配置即可覆盖默认值。这样，一个链就能为使用不同设置的多个用户提供服务，也可以用于对模型配置进行 A/B 测试。

from langchain_core.runnables import ConfigurableField

llm = ChatOpenAI(model='gpt-4o-mini').configurable_fields(
    temperature=ConfigurableField(
        id='temperature',
        name='LLM Temperature',
        description='Controls output creativity'
    )
)

chain = ChatPromptTemplate.from_template('Write about {topic}') | llm | StrOutputParser()

# Low temperature for factual output
factual = chain.invoke(
    {'topic': 'history of Rome'},
    config={'configurable': {'temperature': 0.0}}
)

快速检查

测试您对使用 LCEL 构建链的理解。

在交互式课程中回答

课程回顾

本课中，您学习了以下内容：LCEL 管道组合使用 | 将可运行对象连接成整洁、易读的流水线；RunnablePassthrough 和 RunnableLambda让您无需脱离 LCEL 范式即可插入数据处理步骤；回退机制会在链执行失败时自动尝试备用链，从而增强链的稳定性。接下来，我们将学习分支链和并行链，以实现更复杂的路由逻辑。

免费开始

用 AI 导师学习 Python — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程: 30
课程: 120

免费开始学习查看完整课程

常见问题解答

「使用 LCEL 构建链」课时是免费的吗？

是的 — 「使用 LCEL 构建链」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。

「使用 LCEL 构建链」这节课中我会学到什么？

使用管道运算符将提示模板、LLM 和输出解析器连接成 Runnable 序列，以同步和异步方式调用它，并检查中间输出。你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Engineering Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 2 节课，共 4 节。

「使用 LCEL 构建链」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Engineering Academy 课中编写并运行代码吗？

能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Engineering Academy