Part 6: Chains（链）

6.1 链的概念

链（Chain）是将多个组件按顺序组合在一起的处理管道。在 LangChain 1.x 中，链主要通过 LCEL 的管道操作符 | 来构建。

---

6.2 使用 LCEL 构建链

简单链 Demo

# demo_simple_chain.py
from dotenv import load_dotenv
load_dotenv()

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

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

# 链1：翻译链
translate_chain = (
    ChatPromptTemplate.from_messages([
        ("system", "你是一个专业翻译。将{source_lang}翻译成{target_lang}。只输出翻译结果。"),
        ("human", "{text}"),
    ])
    | model
    | StrOutputParser()
)

result = translate_chain.invoke({
    "source_lang": "中文",
    "target_lang": "英文",
    "text": "今天天气真好，适合出去散步。",
})
print(f"翻译结果: {result}")

# 链2：摘要链
summary_chain = (
    ChatPromptTemplate.from_template("用一句话总结以下文本：\n{text}")
    | model
    | StrOutputParser()
)

result = summary_chain.invoke({"text": "人工智能是计算机科学的一个分支..."})
print(f"摘要: {result}")

# 链3：情感分析链
sentiment_chain = (
    ChatPromptTemplate.from_template("分析以下文本的情感（正面/负面/中性），只输出一个词：\n{text}")
    | model
    | StrOutputParser()
)

result = sentiment_chain.invoke({"text": "这个产品真的太好用了！"})
print(f"情感: {result}")

复杂链 Demo（RAG 风格）

# demo_complex_chain.py
from dotenv import load_dotenv
load_dotenv()

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

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

# 模拟文档检索（实际项目中使用 Retriever）
def retrieve_documents(query: str) -> list[str]:
    """模拟文档检索"""
    docs = {
        "python": ["Python 是一种解释型编程语言", "Python 由 Guido van Rossum 创建"],
        "java": ["Java 是一种面向对象的语言", "Java 运行在 JVM 上"],
    }
    results = []
    for key, values in docs.items():
        if key in query.lower():
            results.extend(values)
    return results if results else ["未找到相关文档"]

def format_docs(docs: list[str]) -> str:
    """格式化文档列表"""
    return "\n".join([f"[{i+1}] {doc}" for i, doc in enumerate(docs)])

# RAG 风格的链
rag_chain = (
    RunnableParallel(
        question=RunnablePassthrough(),  # 保留原始问题
        context=lambda x: format_docs(retrieve_documents(x)),  # 检索并格式化文档
    )
    | (
        ChatPromptTemplate.from_messages([
            ("system", "根据以下上下文回答问题。如果上下文中没有相关信息，请说明。\n\n上下文：\n{context}"),
            ("human", "{question}"),
        ])
    )
    | model
    | StrOutputParser()
)

result = rag_chain.invoke("Python 是什么？")
print(f"RAG 结果: {result}")

result = rag_chain.invoke("Rust 是什么？")
print(f"RAG 结果（无相关文档）: {result}")

条件链 Demo

# demo_conditional_chain.py
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnableBranch, RunnableLambda

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

# 步骤1：分类器 -- 判断问题类型
classifier_chain = (
    ChatPromptTemplate.from_template(
        "判断以下问题是关于哪个领域的。只输出一个词：编程/数学/其他\n\n问题：{question}"
    )
    | model
    | StrOutputParser()
)

# 步骤2：不同领域的回答链
coding_chain = (
    ChatPromptTemplate.from_template("作为编程专家回答：{question}")
    | model | StrOutputParser()
)
math_chain = (
    ChatPromptTemplate.from_template("作为数学专家回答：{question}")
    | model | StrOutputParser()
)
general_chain = (
    ChatPromptTemplate.from_template("回答以下问题：{question}")
    | model | StrOutputParser()
)

# 步骤3：条件路由
def route_question(classification: str):
    """根据分类结果选择链"""
    if "编程" in classification:
        return coding_chain
    elif "数学" in classification:
        return math_chain
    return general_chain

# 构建完整链
full_chain = (
    RunnableLambda(lambda x: {"question": x, "classification": classifier_chain.invoke(x)})
    | RunnableBranch(
        (lambda x: "编程" in x["classification"],
         lambda x: coding_chain.invoke({"question": x["question"]})),
        (lambda x: "数学" in x["classification"],
         lambda x: math_chain.invoke({"question": x["question"]})),
        lambda x: general_chain.invoke({"question": x["question"]}),
    )
)

result = full_chain.invoke("什么是递归？")
print(f"编程问题: {result}")

result = full_chain.invoke("什么是勾股定理？")
print(f"数学问题: {result}")

---

6.3 链的调用方式

invoke / batch / stream 对比 Demo

# demo_chain_invocation.py
import asyncio
from dotenv import load_dotenv
load_dotenv()

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

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
chain = ChatPromptTemplate.from_template("用一句话解释{concept}") | model | StrOutputParser()

# ===== invoke：单次调用 =====
result = chain.invoke({"concept": "机器学习"})
print(f"invoke: {result}")

# ===== batch：批量调用 =====
results = chain.batch([
    {"concept": "机器学习"},
    {"concept": "深度学习"},
    {"concept": "强化学习"},
])
for i, r in enumerate(results):
    print(f"batch[{i}]: {r}")

# ===== stream：流式调用 =====
print("stream: ", end="")
for chunk in chain.stream({"concept": "人工智能"}):
    print(chunk, end="", flush=True)
print()

# ===== 异步调用 =====
async def async_demo():
    result = await chain.ainvoke({"concept": "区块链"})
    print(f"ainvoke: {result}")

    results = await chain.abatch([{"concept": "云计算"}, {"concept": "边缘计算"}])
    for r in results:
        print(f"abatch: {r}")

    print("astream: ", end="")
    async for chunk in chain.astream({"concept": "物联网"}):
        print(chunk, end="", flush=True)
    print()

asyncio.run(async_demo())

调用方式对比

方式	适用场景	返回值	特点
invoke	单个请求	单个结果	最常用
batch	多个独立请求	结果列表	比循环 invoke 更高效
stream	需要实时显示	逐块输出	适合聊天界面
ainvoke	异步环境	单个结果	高并发场景
abatch	异步批量	结果列表	异步高并发
astream	异步流式	逐块输出	异步聊天界面

---

6.4 链的最佳实践

1. 模块化设计：将复杂的链拆分为多个小链，每个链只做一件事

# 好的实践：模块化
translator = ChatPromptTemplate.from_template("翻译成{lang}：{text}") | model | StrOutputParser()
summarizer = ChatPromptTemplate.from_template("总结：{text}") | model | StrOutputParser()

# 组合使用
full_chain = summarizer | (lambda x: translator.invoke({"lang": "英文", "text": x}))

2. 使用 RunnableParallel 并行化

from langchain_core.runnables import RunnableParallel

# 好的实践：并行执行独立操作
parallel = RunnableParallel(
    summary=summarizer,
    sentiment=sentiment_analyzer,
    keywords=keyword_extractor,
)

3. 添加错误处理

# 好的实践：添加 retry 和 fallback
reliable_chain = my_chain.with_retry(stop_after_attempt=3).with_fallbacks([backup_chain])

4. 使用类型标注

# 好的实践：标注类型
chain = (
    ChatPromptTemplate.from_template("解释{concept}")
    | model
    | StrOutputParser()
).with_types(input_type=dict, output_type=str)

5. 添加配置便于调试

# 好的实践：添加 run_name 和 tags
chain = (
    prompt | model | parser
).with_config(
    run_name="translation_chain",
    tags=["production", "translation"],
    metadata={"version": "2.0"},
)

6. 避免链过长

# 不好的实践：过长的链
bad_chain = step1 | step2 | step3 | step4 | step5 | step6 | step7 | step8

# 好的实践：拆分为子链
sub_chain_1 = step1 | step2 | step3
sub_chain_2 = step4 | step5
sub_chain_3 = step6 | step7 | step8
good_chain = sub_chain_1 | sub_chain_2 | sub_chain_3

---

下一步学习：本教程涵盖了 LangChain 的核心基础（Part 1-6）。后续教程将涵盖：

• Part 7: Retrievers（检索器）与向量存储
• Part 8: Memory（记忆管理）
• Part 9: Agents（智能体）
• Part 10: Tools（工具）
• Part 11: Callbacks（回调系统）
• Part 12: 部署与生产实践