第 06 课：链（Chains）- LCEL 与流程编排

课程目标

通过本节课，你将学到：

• ✅ 什么是 LCEL（LangChain Expression Language）
• ✅ 如何使用 | 操作符连接组件
• ✅ 常用 Chain 类型
• ✅ 复杂流程编排

---

一、LCEL 基础

1.1 什么是 LCEL？

LCEL（LangChain Expression Language） 是 LangChain 的声明式语言，用于组合和编排组件。

核心特点：

• ✅ 使用 | 操作符连接组件
• ✅ 支持流式处理
• ✅ 自动追踪和调试
• ✅ 统一的接口

1.2 Hello LCEL

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import StrOutputParser

# 1. 定义组件
model = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("给我讲一个关于{topic}的笑话")

# 2. 使用 | 连接组件
chain = prompt | model | StrOutputParser()

# 3. 调用
result = chain.invoke({"topic": "程序员"})
print(result)

解析：

1. prompt：生成消息
2. model：调用 LLM
3. StrOutputParser()：提取字符串内容

1.3 为什么使用 LCEL？

对比传统方式：

# ❌ 传统方式：手动调用
prompt_text = prompt.format(topic="程序员")
response = model.invoke(prompt_text)
result = StrOutputParser().parse(response.content)

# ✅ LCEL 方式：简洁优雅
result = chain.invoke({"topic": "程序员"})

---

二、常用 Chain 类型

2.1 简单链（Simple Chain）

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

# 创建链
model = ChatOpenAI()
prompt = ChatPromptTemplate.from_template("翻译成英文：{text}")

chain = prompt | model

# 调用
result = chain.invoke({"text": "你好"})
print(result.content)

2.2 带输出解析的链

from langchain.schema import StrOutputParser

chain = prompt | model | StrOutputParser()

result = chain.invoke({"text": "你好"})
print(result)  # 直接是字符串，不再是 AIMessage

2.3 并行链（Parallel Chain）

from langchain.schema.runnable import RunnableParallel

# 定义多个分支
chain = RunnableParallel(
    joke=prompt | model,
    poem=ChatPromptTemplate.from_template("写一首关于{topic}的诗") | model
)

# 同时执行
result = chain.invoke({"topic": "春天"})
print(result["joke"].content)
print(result["poem"].content)

2.4 条件链（Branching Chain）

from langchain.schema.runnable import RunnableBranch

# 定义条件分支
branch = RunnableBranch(
    (lambda x: x["type"] == "joke", joke_chain),
    (lambda x: x["type"] == "poem", poem_chain),
    default_chain  # 默认分支
)

result = branch.invoke({"type": "joke", "topic": "程序员"})

---

三、复杂流程编排

3.1 多步骤链

from langchain.schema.runnable import RunnablePassthrough

# 步骤1：生成主题
step1 = ChatPromptTemplate.from_template("给我一个{category}的主题") | model

# 步骤2：基于主题生成内容
step2 = ChatPromptTemplate.from_template("写一篇关于{topic}的文章") | model

# 组合链
full_chain = (
    {"topic": step1, "category": RunnablePassthrough()}
    | step2
)

result = full_chain.invoke({"category": "科技"})

3.2 带记忆的链

from langchain.memory import ConversationBufferMemory
from langchain.schema.runnable import RunnableMap

# 创建记忆
memory = ConversationBufferMemory(return_messages=True)

# 带记忆的链
chain_with_memory = (
    RunnableMap({
        "input": lambda x: x["input"],
        "history": lambda x: memory.load_memory_variables({})["history"]
    })
    | ChatPromptTemplate.from_messages([
        ("system", "你是一个助手"),
        ("history", "{history}"),
        ("human", "{input}")
    ])
    | model
)

# 第一次对话
result1 = chain_with_memory.invoke({"input": "我叫张三"})
memory.save_context({"input": "我叫张三"}, {"output": result1.content})

# 第二次对话
result2 = chain_with_memory.invoke({"input": "我叫什么？"})
print(result2.content)  # "你叫张三"

3.3 错误处理

from langchain.schema.runnable import RunnableLambda

def handle_error(x):
    return f"发生错误：{x}"

# 带错误处理的链
safe_chain = (
    prompt
    | model
    | StrOutputParser()
    .with_fallbacks([RunnableLambda(handle_error)])
)

try:
    result = safe_chain.invoke({"topic": "程序员"})
except Exception as e:
    print(f"错误: {e}")

---

四、实战案例

4.1 文档摘要链

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import StrOutputParser

# 初始化
model = ChatOpenAI(model="gpt-3.5-turbo")

# 步骤1：提取关键信息
extract_prompt = ChatPromptTemplate.from_template(
    "从以下文本中提取关键信息：\n\n{text}"
)

# 步骤2：生成摘要
summarize_prompt = ChatPromptTemplate.from_template(
    "基于以下关键信息，生成一个简洁的摘要：\n\n{key_info}"
)

# 组合链
summary_chain = (
    {"text": RunnablePassthrough()}
    | extract_prompt
    | model
    | StrOutputParser()
    | (lambda x: {"key_info": x})
    | summarize_prompt
    | model
    | StrOutputParser()
)

# 使用
long_text = "..."  # 长文本
summary = summary_chain.invoke(long_text)
print(summary)

4.2 多语言翻译链

from langchain.schema.runnable import RunnableParallel

# 定义翻译链
def create_translation_chain(target_lang):
    prompt = ChatPromptTemplate.from_template(
        "翻译以下文本到{language}：\n\n{text}"
    )
    return prompt | model | StrOutputParser()

# 并行翻译
multi_translate = RunnableParallel(
    english=create_translation_chain("英文"),
    japanese=create_translation_chain("日文"),
    french=create_translation_chain("法文")
)

# 使用
result = multi_translate.invoke({"text": "你好", "language": ""})
print(result["english"])  # Hello
print(result["japanese"])  # こんにちは

---

五、本课小结

核心要点

1. LCEL：

   • 使用 | 连接组件
   • 声明式、简洁

2. 常用 Chain：

   • 简单链、并行链、条件链

3. 流程编排：

   • 多步骤链
   • 带记忆的链
   • 错误处理

下节预告

下一课，我们将学习 记忆（Memory），包括：

• 对话记忆类型
• 向量存储记忆
• 持久化方案

---

🎯 课后练习

1. 基础练习：创建一个翻译链，支持中文→英文→日文。
2. 进阶练习：创建一个带错误处理的文档处理链。
3. 挑战练习：创建一个条件链，根据输入长度选择不同的处理方式。

---

上一课：05 - 输出解析 | 下一课：07 - 记忆