
## 基础操作

在进入 LCEL 与 Chain 之前，先快速回顾一下 Python 列表的基础操作。列表是 Python 中最常用的数据结构之一，理解它的创建、索引、切片与追加操作，有助于后续阅读 LCEL 示例中的数据结构处理逻辑。

**1. 创建列表**

列表使用方括号 `[]` 创建，也可以用 `list()` 从其他可迭代对象转换：

```python
# 创建空列表
empty_list = []

# 创建包含多个元素的列表
fruits = ["apple", "banana", "cherry"]

# 使用 list() 从其他可迭代对象创建列表
numbers = list(range(5))

print(fruits)  # ['apple', 'banana', 'cherry']
print(numbers) # [0, 1, 2, 3, 4]
```

**2. 通过索引访问元素**

列表索引从 `0` 开始，负数索引表示从末尾倒数：

```python
fruits = ["apple", "banana", "cherry"]

# 访问第一个元素
print(fruits[0])   # apple

# 访问最后一个元素
print(fruits[-1])  # cherry
```

**3. 切片操作**

切片用于获取列表中的一部分元素，格式为 `start:stop:step`，其中 `stop` 位置不包含在内：

```python
nums = [10, 20, 30, 40, 50]

# 获取前三个元素
print(nums[:3])   # [10, 20, 30]

# 获取索引 1 到 3 的元素
print(nums[1:4])  # [20, 30, 40]

# 步长为 2 取元素
print(nums[::2])  # [10, 30, 50]
```

**4. 追加元素**

可以使用 `append()`、`extend()` 和 `insert()` 向列表添加元素：

```python
fruits = ["apple", "banana"]

# 在列表末尾追加单个元素
fruits.append("cherry")
print(fruits)  # ['apple', 'banana', 'cherry']

# 一次追加多个元素
fruits.extend(["orange", "grape"])
print(fruits)  # ['apple', 'banana', 'cherry', 'orange', 'grape']

# 在指定位置插入元素
fruits.insert(1, "pear")
print(fruits)  # ['apple', 'pear', 'banana', 'cherry', 'orange', 'grape']
```

## 1. 为什么需要了解 LCEL 与 Chain

LangChain 是目前最主流的大模型应用开发框架之一。早期的 LangChain 通过 LLMChain、SequentialChain 等类来组织流程，虽然功能丰富，但链条一长，代码就变得冗长、难以复用和调试。

LCEL（LangChain Expression Language）的出现改变了这一局面。它提供了一套声明式的表达式语法，让我们可以用管道符 `|` 把不同的组件串联起来，像搭积木一样组合 Prompt、模型、解析器和工具调用。

本文围绕三个核心主题展开：

- **LCEL 表达式语法**：理解 Runnable 协议与管道语法。
- **Chain 的构建**：用 LCEL 搭建复杂链，并掌握回调机制。
- **Output Parser 的使用**：把模型输出解析成结构化数据，对接下游业务。

学完本文后，你将能够用简洁、可维护的方式构建生产级的 LLM 应用链。

## 2. LCEL 表达式语法入门

### 2.1 核心概念：Runnable 协议

LCEL 的核心是 Runnable 协议。在 LangChain 中，Prompt、ChatModel、Retriever、OutputParser 等组件都实现了 Runnable 接口，因此具备统一的能力：

- `invoke`：同步调用，输入单个值，返回单个结果。
- `batch`：批量调用，输入一组值，返回一组结果。
- `stream`：流式调用，逐步返回中间结果。
- `ainvoke / abatch / astream`：对应的异步版本。

统一接口的好处是显而易见的：无论组件是模型还是解析器，调用方式完全一致，切换底层实现时业务代码无需改动。

### 2.2 管道操作符：串联一切

LCEL 最直观的语法就是管道符 `|`。它把左侧组件的输出作为右侧组件的输入，形成一条数据流：

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

prompt = ChatPromptTemplate.from_template("用一句话介绍{topic}")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

chain = prompt | model | parser

result = chain.invoke({"topic": "LCEL"})
print(result)
```

数据流向如下：

```mermaid
flowchart LR
  A["输入字典"] --> B["ChatPromptTemplate"]
  B --> C["ChatOpenAI"]
  C --> D["StrOutputParser"]
  D --> E["输出字符串"]
```

### 2.3 RunnablePassthrough：透传与并行

当需要原样传递某些数据，或者并行执行多个分支时，可以使用 `RunnablePassthrough`：

```python
from langchain_core.runnables import RunnablePassthrough

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | parser
)

chain.invoke("什么是 RAG？")
```

上面的代码中，`retriever` 接收输入并检索上下文，而 `RunnablePassthrough` 把用户问题原样透传给 prompt 的 `question` 字段。

### 2.4 RunnableLambda：注入自定义逻辑

当内置组件满足不了需求时，可以用 `RunnableLambda` 把普通函数包装成 Runnable：

```python
from langchain_core.runnables import RunnableLambda

def add_question_mark(text: str) -> str:
    return text.rstrip("?") + "?"

question_maker = RunnableLambda(add_question_mark)

chain = prompt | model | parser | question_maker
```

`RunnableLambda` 让任意 Python 函数都能融入 LCEL 管道，极大提升了灵活性。

### 2.5 常用操作符速查

| 操作符/方法 | 作用 | 示例 |
|---|---|---|
| `\|` | 串联组件 | `prompt \| model` |
| `.with_retry()` | 失败重试 | `model.with_retry(retry_if_exception_type=(RateLimitError,))` |
| `.bind()` | 给模型绑定工具或参数 | `model.bind(tools=[...])` |
| `.with_fallbacks()` | 设置降级方案 | `model.with_fallbacks([fallback_model])` |
| `.assign()` | 在字典中新增键 | `RunnablePassthrough.assign(count=...)` |
| `RunnableParallel` | 并行执行多个分支 | `RunnableParallel(a=..., b=...)` |

## 3. Chain 的构建与类型

### 3.1 从 LLMChain 到 LCEL Chain

老版本 LangChain 使用 LLMChain 类构建链，代码冗长且不利于扩展：

```python
# 旧写法（不推荐）
from langchain.chains import LLMChain

chain = LLMChain(llm=model, prompt=prompt)
result = chain.run(topic="Transformer")
```

而 LCEL 写法简洁直观，并且天然支持流式、批量与异步调用。新项目建议全部采用 LCEL 构建。

### 3.2 构建带检索的 RAG 链

一个典型的 RAG（检索增强生成）链条如下：

```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_chroma import Chroma

# 假设已经构建好向量库
vectorstore = Chroma(...)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

system_template = """根据以下上下文回答问题：
{context}

问题：{question}
"""
prompt = ChatPromptTemplate.from_messages([
    ("system", system_template),
    ("human", "{question}"),
])

rag_chain = (
    {
        "context": retriever | (lambda docs: "\n\n".join(d.page_content for d in docs)),
        "question": RunnablePassthrough(),
    }
    | prompt
    | model
    | StrOutputParser()
)

answer = rag_chain.invoke("LangChain 如何处理文档分块？")
```

### 3.3 构建 Agent 链

结合工具调用的 Agent 是 LangChain 的核心能力之一。现代 LangChain 使用 `create_react_agent` 构建 Agent：

```python
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

model = ChatOpenAI(model="gpt-4o-mini")

@tool
def search(query: str) -> str:
    """搜索与 query 相关的信息。"""
    return f"关于「{query}」的搜索结果..."

agent = create_react_agent(model, tools=[search])

response = agent.invoke({"messages": [("user", "帮我查一下今天的新闻")]})
```

这里的 Agent 底层也是由 Runnable 组合而成，因此同样支持流式输出与回调监听。

### 3.4 Chain 调试技巧

调试 LCEL 链时，可以在任意管道节点后插入一个打印中间结果的 RunnableLambda：

```python
def debug_print(x):
    print(f"[DEBUG] {x}")
    return x

chain = prompt | model | RunnableLambda(debug_print) | parser
```

这样就能清楚看到每个环节的输入输出，定位问题更容易。

## 4. Callbacks 回调机制

### 4.1 回调的作用

Callbacks（回调）让你能够在链的执行过程中插入自定义逻辑，比如：

- 记录每次 LLM 调用的 token 消耗。
- 监控链执行耗时，定位性能瓶颈。
- 把中间过程写入日志或监控系统。
- 做成本统计与告警。

LangChain 的回调系统覆盖了几乎所有关键事件，包括模型调用开始/结束、工具执行、链路开始/结束、重试等。

### 4.2 常用回调处理器

LangChain 内置了一些实用的回调处理器：

```python
from langchain_community.callbacks import get_openai_callback

with get_openai_callback() as cb:
    result = chain.invoke({"topic": "回调机制"})
    print(f"总花费: ${cb.total_cost}")
    print(f"总 token: {cb.total_tokens}")
    print(f"提示 token: {cb.prompt_tokens}")
    print(f"补全 token: {cb.completion_tokens}")
```

`get_openai_callback` 用于统计 OpenAI 模型的 token 与费用，是日常开发中最常用的回调之一。

### 4.3 自定义回调处理器

通过继承 `BaseCallbackHandler`，可以创建自己的回调处理器：

```python
from langchain_core.callbacks import BaseCallbackHandler

class MyCallbackHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"[LLM START] 调用模型，prompts 数量: {len(prompts)}")

    def on_llm_end(self, response, **kwargs):
        generation = response.generations[0][0]
        print(f"[LLM END] 完成，输出长度: {len(generation.text)}")

    def on_chain_start(self, serialized, inputs, **kwargs):
        print(f"[CHAIN START] 链开始执行")

    def on_chain_end(self, outputs, **kwargs):
        print(f"[CHAIN END] 链执行完成")

handler = MyCallbackHandler()
result = chain.invoke({"topic": "回调"}, config={"callbacks": [handler]})
```

回调处理器通过 `config` 参数传入，不会污染链的全局状态，更适合按需启用。

### 4.4 全局回调注册

如果希望所有链都启用某个回调，可以使用 `langchain.debug` 或手动设置：

```python
# 打开全局调试模式，自动打印每一步的输入输出
import langchain
langchain.debug = True
```

全局调试模式适合开发阶段快速排查问题，但生产环境应关闭，避免日志过多影响性能。

### 4.5 回调与流式输出结合

回调同样适用于流式场景：

```python
handler = MyCallbackHandler()
for chunk in chain.stream({"topic": "流式输出"}, config={"callbacks": [handler]}):
    print(chunk, end="", flush=True)
```

此时回调处理器会在模型和链的开始/结束节点触发，而流式输出则逐步返回给用户，两者互不干扰。

## 5. Output Parser 的使用

### 5.1 为什么需要 Output Parser

大模型的输出本质上是自由文本。实际业务中，我们往往需要**结构化数据**，比如 JSON、列表、枚举值，以便程序直接消费。Output Parser 的作用就是把模型输出从文本转换成目标格式。

### 5.2 字符串解析器：StrOutputParser

最简单的解析器，把 ChatModel 的 `AIMessage` 对象转换为纯字符串：

```python
from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()
chain = prompt | model | parser
result = chain.invoke({"topic": "Python"})
print(type(result))  # <class 'str'>
```

### 5.3 逗号分隔解析器：CommaSeparatedListOutputParser

适合让模型输出列表类数据，例如标签、关键词等：

```python
from langchain_core.output_parsers import CommaSeparatedListOutputParser

parser = CommaSeparatedListOutputParser()
# 通过 get_format_instructions 获取格式说明，并注入到 prompt 中
format_instructions = parser.get_format_instructions()

prompt = ChatPromptTemplate.from_template(
    "列出 5 个关于 {topic} 的关键词。\n{format_instructions}"
)

chain = prompt | model | parser
result = chain.invoke({"topic": "机器学习", "format_instructions": format_instructions})
print(result)  # ['神经网络', '过拟合', ...]
```

### 5.4 JSON 解析器：JsonOutputParser

当需要稳定的 JSON 输出时，使用 `JsonOutputParser`：

```python
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

class Person(BaseModel):
    name: str = Field(description="人物姓名")
    age: int = Field(description="人物年龄")
    city: str = Field(description="居住城市")

parser = JsonOutputParser(pydantic_object=Person)
format_instructions = parser.get_format_instructions()

prompt = ChatPromptTemplate.from_template(
    """从下面文本中提取人物信息：
{text}

输出格式要求：
{format_instructions}
"""
)

chain = prompt | model | parser
result = chain.invoke({
    "text": "李明今年 28 岁，目前在北京工作。",
    "format_instructions": format_instructions,
})
print(result)  # {'name': '李明', 'age': 28, 'city': '北京'}
```

### 5.5 Pydantic 结构化输出：with_structured_output

对于支持结构化输出的模型，推荐使用 `with_structured_output`，它比手动拼格式说明更可靠、更省 token，并且能自动处理 JSON Schema 的生成与校验：

```python
from pydantic import BaseModel, Field
from typing import List

class Article(BaseModel):
    title: str = Field(description="文章标题")
    tags: List[str] = Field(description="文章标签列表")
    summary: str = Field(description="100 字以内的摘要")

structured_model = model.with_structured_output(Article)
chain = prompt | structured_model

result = chain.invoke({"topic": "LCEL 表达式"})
print(result.title)
print(result.tags)
print(result.summary)
```

返回的 `result` 直接就是 `Article` 对象，字段类型得到了 Pydantic 的保障。

### 5.6 自动修正解析器：OutputFixingParser

模型偶尔会输出不符合格式的内容。`OutputFixingParser` 会在解析失败时自动调用模型修正输出：

```python
from langchain_core.output_parsers import PydanticOutputParser, OutputFixingParser

base_parser = PydanticOutputParser(pydantic_object=Person)
fixing_parser = OutputFixingParser.from_llm(parser=base_parser, llm=model)

chain = prompt | model | fixing_parser
```

这样即使模型输出略微偏离格式，系统也会自动修复，提升链路健壮性。

### 5.7 常用解析器对比

| 解析器 | 输出类型 | 适用场景 |
|---|---|---|
| `StrOutputParser` | 字符串 | 纯文本生成 |
| `CommaSeparatedListOutputParser` | 列表 | 关键词、标签提取 |
| `JsonOutputParser` | 字典 | JSON 数据抽取 |
| `PydanticOutputParser` | Pydantic 对象 | 严格类型校验 |
| `with_structured_output` | Pydantic 对象 | 模型原生结构化输出 |
| `OutputFixingParser` | 任意 | 解析失败自动修复 |

## 6. 完整实战：构建一个组合链

下面把前面所学整合成一个完整示例：输入一个主题，先检索相关资料，再生成结构化文章信息，并在控制台打印 token 消耗。

```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_community.callbacks import get_openai_callback
from pydantic import BaseModel, Field
from typing import List

class ArticleInfo(BaseModel):
    title: str = Field(description="文章标题")
    outline: List[str] = Field(description="文章大纲，每个元素是一级标题")
    keywords: List[str] = Field(description="3-5 个关键词")

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位资深技术博主，擅长写通俗易懂的教程。"),
    ("human", "请针对「{topic}」生成文章标题、大纲和关键词。"),
])

chain = (
    {"topic": RunnablePassthrough()}
    | prompt
    | model.with_structured_output(ArticleInfo)
)

with get_openai_callback() as cb:
    result = chain.invoke("主流 Agent 框架对比")
    print(f"标题：{result.title}")
    print(f"大纲：{result.outline}")
    print(f"关键词：{result.keywords}")
    print(f"本次消耗 token：{cb.total_tokens}，费用：${cb.total_cost}")
```

整个链路清晰可读，token 统计与结构化输出一气呵成。

## 7. 最佳实践与常见坑

**最佳实践**

- 优先使用 LCEL 构建新链路，避免遗留的 LLMChain 写法。
- 用 `with_structured_output` 获取结构化数据，比手动拼 JSON 格式说明更稳定。
- 回调只通过 `config` 局部传入，避免全局污染。
- 在开发环境打开 `langchain.debug`，生产环境务必关闭。

**常见坑**

- 忘记在 prompt 中注入 `format_instructions`，导致模型输出不符合解析器预期。
- 用 `|` 串联时，左侧输出结构与右侧输入结构不匹配，导致运行时报错。
- 回调处理器中抛异常会中断整条链，需要在回调内部做好异常保护。
- 过度依赖 `OutputFixingParser` 会隐藏模型输出的质量问题，应该同时关注 prompt 的优化。

## 8. 总结

LCEL 为 LangChain 提供了统一、简洁、可组合的链构建方式。通过管道符、RunnablePassthrough、RunnableLambda 等语法，你可以轻松搭建从简单问答到复杂 Agent 的各种链路。回调机制则为链路注入了可观测性，让 token 统计、性能监控和日志记录变得简单。而 Output Parser 解决了大模型输出不可控的核心痛点，让 LLM 应用能够真正走进业务流程。


在进入 LCEL 与 Chain 之前，先快速回顾一下 Python 列表的基础操作。列表是 Python 中最常用的数据结构之一，理解它的创建、索引、切片与追加操作，有助于后续阅读 LCEL 示例中的数据结构处理逻辑。

LangChain 实战：LCEL 表达式语法、Chain 构建、Callbacks 与 Output Parser 全解析

人工智能

结构化数据

大模型应用

机器学习

数据结构

代码解释

数据流

Python

Agent

JSON

2026上云采购 | AI焕新·智启新局

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

LangChain 实战：LCEL 表达式语法、Chain 构建、Callbacks 与 Output Parser 全解析-腾讯云开发者社区-腾讯云

LangChain 实战：LCEL 表达式语法、Chain 构建、Callbacks 与 Output Parser 全解析

LangChain 实战：LCEL 表达式语法、Chain 构建、Callbacks 与 Output Parser 全解析

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐