一、为何需要链式调用？

单次 LLM 调用只能解决简单问题，但实际需求往往涉及多步骤流程：

• 先将用户输入翻译成英文 → 再用英文检索知识库 → 再将结果翻译回中文
• 先提取文章关键词 → 再生成 SEO 标题 → 再打分评估
• 先将代码生成 → 再进行安全审查 → 再格式化输出

Chain（链）正是将这些步骤按顺序串联起来，让每一步的输出自动流入下一步的输入，形成一条完整的数据流水线。

二、LCEL 核心理念：管道操作符 |

LCEL（LangChain Expression Language）利用 | 符号将组件串联起来。你可以把它理解为 Linux 命令行的管道机制：

# Linux 管道：前一个命令的输出成为后一个命令的输入
cat file.txt | grep "error" | sort | uniq

# LCEL 链：原理完全一致
chain = prompt | llm | output_parser
result = chain.invoke({"question": "什么是Python？"})

所有 LCEL 组件都实现了 Runnable 接口，提供统一的调用方式：

三、Step 1：构建简单链

import os
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

# ── 1. 初始化 LLM ─────────────────────────────────────────────
llm = ChatOpenAI(
    model="qwen-plus",
    temperature=0.7,
    openai_api_key=os.getenv("DASHSCOPE_API_KEY"),
    openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# ── 2. 定义 Prompt 模板 ────────────────────────────────────────
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位{domain}领域的权威专家，提供准确且简洁的回答。"),
    ("human", "{question}"),
])

# ── 3. 输出解析器（将 AIMessage 对象转化为纯字符串）────────────
parser = StrOutputParser()

# ── 4. 用 | 组装成链 ──────────────────────────────────────────
# chain 是一个 Runnable 对象，可像调用函数一样使用
chain = prompt | llm | parser

# ── 5. 调用链 ─────────────────────────────────────────────────
result = chain.invoke({
    "domain": "量子物理",
    "question": "用一句话解释量子纠缠"
})
print(result)
# 直接获得字符串，无需 .content

# ── 6. 流式调用（同一个链，无需重写）───────────────────────
print("\n流式输出：")
for chunk in chain.stream({
    "domain": "历史", 
    "question": "秦始皇统一六国的关键因素是什么？"
}):
    print(chunk, end="", flush=True)
print()

数据在链中的流向：

chain.invoke({"domain": "量子物理", "question": "..."})
↓ 进入 prompt
→ 格式化为 [SystemMessage("你是量子物理领域的权威专家..."), HumanMessage("用一句话...")]
↓ 进入 llm
→ 返回 AIMessage(content="量子纠缠是...")
↓ 进入 parser（StrOutputParser）
→ 提取 AIMessage.content，返回纯字符串 "量子纠缠是..."

四、Step 2：顺序链（多步骤流水线）

本文件包含三个渐进式示例，完整演示从两步链到并行链的 LCEL 组合方式。

演示1：两步链（生成文章 → 生成摘要）

核心要点：上一步输出字符串，通过 RunnableLambda 包装为字典后流入下一步。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda

# ── 链1：根据主题生成文章（输出：字符串）─────────────────────
write_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位专业作家，写作时注重逻辑与细节。"),
    ("human", "写一篇关于{topic}的200字短文。"),
])
write_chain = write_prompt | llm | StrOutputParser()

# ── 链2：对文章进行摘要（输入：{"article":"..."}, 输出：字符串）
summary_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个善于总结的助手，用50字以内概括文章核心观点。"),
    ("human", "请摘要以下文章：\n\n{article}"),
])
summary_chain = summary_prompt | llm | StrOutputParser()

def wrap_as_article_dict(article: str) -> dict:
    """将 write_chain 输出的字符串包装为 summary_chain 所需的字典格式。"""
    return {"article": article}

# ── 将两个链串联成完整的两步链 ────────────────────────────────
# 数据流：{"topic":...} → write_chain(字符串) → RunnableLambda(字典) → summary_chain(摘要字符串)
combined_chain = (
    write_chain                              # 步骤1：生成文章（输出：字符串）
    | RunnableLambda(wrap_as_article_dict)   # 中间转换：str → {"article": str}
    | summary_chain                          # 步骤2：生成摘要（输出：字符串）
)

# ── 方式一：使用组合链一次调用（LCEL 推荐用法）───────────────
combined_summary = combined_chain.invoke({"topic": "量子计算的未来"})
print(f"最终摘要：{combined_summary}")

# ── 方式二：分步调用（便于调试查看中间结果）─────────────────
article = write_chain.invoke({"topic": "量子计算的未来"})
print(f"生成的文章：{article}")
summary = summary_chain.invoke({"article": article})
print(f"文章摘要：{summary}")

关键问题：为什么需要 RunnableLambda 做中间转换？

write_chain 输出："量子计算是一种利用量子力学原理..." ← 字符串
summary_chain 期望：{"article": "量子计算是一种利用量子..."} ← 字典
不做转换就直接管道连接 → 报错：KeyError: 'article'
RunnableLambda(wrap_as_article_dict) 把字符串包装成字典 → 正确传递

演示2：多步管道链（翻译 → 并行生成标题 & 情感分析）

核心要点：通过 RunnableLambda 将第一步输出分叉为多个分支，再用 RunnableParallel 并行执行。

from langchain_core.runnables import RunnableLambda, RunnableParallel

# ── 步骤1：翻译（输入：{"english":"..."}, 输出：中文字符串）──
translate_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业翻译，将英文翻译成自然流畅的中文。"),
    ("human", "翻译以下英文文本：\n{english}"),
])
# ── 步骤2：生成标题（输入：{"chinese_text":"..."}, 输出：字符串）
title_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位编辑，根据文章内容生成吸引人的标题（不超过20字）。"),
    ("human", "为以下文章生成一个标题：\n{chinese_text}"),
])
# ── 步骤3：情感分析（输入：{"text":"..."}, 输出：字符串）────
sentiment_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是情感分析专家，判断文本的情感倾向并说明原因。"),
    ("human", "分析以下文本的情感：\n{text}"),
])

translate_chain = translate_prompt | llm | StrOutputParser()
title_chain = title_prompt | llm | StrOutputParser()
sentiment_chain = sentiment_prompt | llm | StrOutputParser()

def split_to_branches(chinese: str) -> dict:
    """将翻译结果分发给后续两个分支所需的字典。"""
    return {"chinese_text": chinese, "text": chinese}

# ── 构建完整管道链 ────────────────────────────────────────────
# 数据流：{"english":...} → translate(中文串) → split(字典) → RunnableParallel → {title, sentiment}
pipeline_chain = (
    translate_chain                                  # 步骤1：翻译（输出：中文字符串）
    | RunnableLambda(split_to_branches)              # 中间转换：str → {"chinese_text":..., "text":...}
    | RunnableParallel(                               # 步骤2&3 并行执行（共享同一输入字典）
        title=title_chain,                           # 使用 {chinese_text} → 生成标题
        sentiment=sentiment_chain                    # 使用 {text} → 情感分析
    )
)

# ── 调用：一次 invoke 触发完整三步流程 ───────────────────────
english_text = "Artificial intelligence is transforming industries..."
result = pipeline_chain.invoke({"english": english_text})
print(f"生成标题：{result['title']}")
print(f"情感分析：{result['sentiment']}")

数据在管道链中的完整流向：

pipeline_chain.invoke({"english": "Artificial intelligence..."})
↓ translate_chain
→ "人工智能正在全球各行各业引发变革..." (中文字符串)
↓ RunnableLambda(split_to_branches)
→ {"chinese_text": "人工智能...", "text": "人工智能..."} (字典)
↓ RunnableParallel(title_chain, sentiment_chain)
  同时执行两个分支
→ {"title": "AI时代的产业变革", "sentiment": "整体情感为中性..."}

演示3：并行链（同时生成多种内容）

核心要点：RunnableParallel 将同一个输入同时传给多条链，所有分支并发执行后合并结果。

from langchain_core.runnables import RunnableParallel, RunnablePassthrough, RunnableLambda

pros_prompt = ChatPromptTemplate.from_messages([
    ("human", "列出{topic}的3个主要优势")
])
cons_prompt = ChatPromptTemplate.from_messages([
    ("human", "列出{topic}的3个主要挑战")
])
summary_prompt2 = ChatPromptTemplate.from_messages([
    ("human", "用两句话总结{topic}")
])

# ── RunnableParallel：同一输入 → 多条链并行 → 合并为字典 ───
parallel_chain = RunnableParallel(
    topic=RunnablePassthrough() | RunnableLambda(lambda x: x["topic"]),
    advantages=pros_prompt | llm | StrOutputParser(),
    challenges=cons_prompt | llm | StrOutputParser(),
    summary=summary_prompt2 | llm | StrOutputParser(),
)

result = parallel_chain.invoke({"topic": "人工智能"})
print(f"优势：{result['advantages']}")
print(f"挑战：{result['challenges']}")
print(f"总结：{result['summary']}")

五、Step 3：结构化输出解析器

LLM 的原始输出是字符串，但在代码中我们通常需要结构化数据（字典、对象）。输出解析器负责这个转换。

JSON 输出解析器

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

# ── 定义期望的输出结构（使用 Pydantic） ─────────────────────────
class BookReview(BaseModel):
    title: str = Field(description="书名")
    author: str = Field(description="作者")
    rating: float = Field(description="评分，1-10分")
    summary: str = Field(description="50字以内的内容简介")
    recommendation: str = Field(description="适合什么类型的读者")

# ── 创建解析器 ────────────────────────────────────────────────
parser = JsonOutputParser(pydantic_object=BookReview)

# ── 包含格式说明的 Prompt ─────────────────────────────────────
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一位博学的书评家。请按照以下格式输出书评：
{format_instructions}"""),
    ("human", "请评价这本书：{book_name}"),
])

llm = ChatOpenAI(
    model="qwen-plus",
    temperature=0.3,      # 结构化输出用低温度，结果更稳定
    openai_api_key=os.getenv("DASHSCOPE_API_KEY"),
    openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# ── 组合成链 ──────────────────────────────────────────────────
chain = prompt | llm | parser

# ── 调用 ─────────────────────────────────────────────────────
result = chain.invoke({
    "book_name": "《三体》刘慈欣",
    "format_instructions": parser.get_format_instructions(),  # 自动生成格式说明
})

# result 是一个 BookReview 对象（或字典，取决于版本）
print(type(result))
print(f"书名：{result.get('title', result.title if hasattr(result, 'title') else 'N/A')}")
print(f"评分：{result.get('rating', '')}")
print(f"简介：{result.get('summary', '')}")

自定义 RunnableLambda（处理任意中间步骤）

from langchain_core.runnables import RunnableLambda

# ── 将任意 Python 函数变成 LCEL 组件 ──────────────────────────
def clean_text(text: str) -> str:
    """清理文本：去除多余空白、特殊字符。"""
    import re
    text = re.sub(r'\s+', ' ', text)  # 合并多余空白
    text = text.strip()
    return text

def truncate_to_100(text: str) -> str:
    """截断到100字。"""
    return text[:100] + "..." if len(text) > 100 else text

# 组合链：prompt → llm → 字符串 → 清理 → 截断
chain = prompt | llm | StrOutputParser() | RunnableLambda(clean_text) | RunnableLambda(truncate_to_100)

六、链的调试技巧

查看中间结果

# 方法一：拆分调用
step1_result = (prompt | llm).invoke({"question": "..."})
print("LLM 原始输出：", step1_result.content)
step2_result = parser.invoke(step1_result)
print("解析后结果：", step2_result)

# 方法二：用 tap 打印中间值
from langchain_core.runnables import RunnablePassthrough

def debug_print(x):
    print(f"[DEBUG] 中间值: {x}")
    return x  # 必须返回原值，不改变数据

chain = prompt | llm | RunnableLambda(debug_print) | parser

错误处理

from langchain_core.runnables import RunnableLambda

def safe_parse(text: str) -> dict:
    """带错误处理的解析函数。"""
    try:
        return parser.invoke(text)
    except Exception as e:
        print(f"解析失败，原始输出：{text}")
        print(f"错误信息：{e}")
        return {"error": str(e), "raw": text}

chain = prompt | llm | StrOutputParser() | RunnableLambda(safe_parse)