从零开始学LangChain（四）：Chains 链式调用

从零开始学LangChain（四）：Chains 链式调用

本系列教程将带你从零开始学习LangChain框架，构建强大的AI应用程序。

什么是链？

链（Chain）是LangChain的核心概念，它将多个操作串联起来，形成完整的工作流。

想象一下你要制作一份报告。没有链的时候，你需要手动管理每个步骤，先研究主题，然后创建大纲，写草稿，最后润色。每一步都需要单独处理，容易出错。使用链之后，你可以把这些步骤串联起来，形成清晰的工作流，每一步的输出自动成为下一步的输入，既模块化又易于维护。

链的优势主要体现在几个方面：每个环节独立，易于维护；链可以在不同场景重复使用；小链可以组合成大链；可以单独测试每个环节，方便调试。

LCEL基础：LangChain表达式语言

LCEL（LangChain Expression Language）是LangChain推荐的声明式方式，使用|操作符连接组件。

LCEL由几个核心组件构成。PromptTemplate是提示词模板，LLM/Chat Model是模型，Output Parser是输出解析器，使用|可以组合成链。

数据在链中的流转过程很清晰。输入数据先通过prompt.format()格式化提示词，生成消息对象，然后llm.invoke()调用模型，模型返回AIMessage，最后parser.parse()解析输出，得到最终结果。

创建基础链

简单的三步链

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import StrOutputParser

# 初始化模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# 创建提示词模板
prompt = ChatPromptTemplate.from_template(
    "请用{style}风格解释{topic}，不超过100字"
)

# 创建输出解析器
parser = StrOutputParser()

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

# 调用链
result = chain.invoke({
    "style": "幽默",
    "topic": "递归"
})

print(result)

添加中间处理

使用RunnableLambda添加自定义处理步骤。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import StrOutputParser
from langchain.schema.runnable import RunnableLambda

# 定义中间处理函数
def add_prefix(text: str) -> str:
    """给输出添加前缀"""
    return f"🤖 AI回答：\n{text}"

# 创建链
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("解释{topic}")
parser = StrOutputParser()

chain = (
    prompt
    | llm
    | parser
    | RunnableLambda(add_prefix)  # 添加自定义处理
)

result = chain.invoke({"topic": "生成器"})
print(result)

顺序链

按顺序执行多个步骤，前一步的输出作为后一步的输入。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.8)

# 步骤1：生成故事大纲
outline_prompt = ChatPromptTemplate.from_template(
    "为一部{genre}小说生成故事大纲，包含主角、冲突、结局"
)

# 步骤2：根据大纲写第一章
chapter_prompt = ChatPromptTemplate.from_template(
    "根据以下大纲写小说第一章（500字）：\n\n{outline}"
)

# 步骤3：生成章节标题
title_prompt = ChatPromptTemplate.from_template(
    "为以下内容生成一个吸引人的标题：\n\n{chapter}"
)

# 创建三个链
outline_chain = outline_prompt | llm | StrOutputParser()
chapter_chain = chapter_prompt | llm | StrOutputParser()
title_chain = title_prompt | llm | StrOutputParser()

# 使用LCEL创建顺序链
story_chain = (
    {
        "outline": outline_prompt | llm | StrOutputParser()
    }
    | {
        "chapter": chapter_prompt | llm | StrOutputParser(),
        "outline": lambda x: x["outline"]
    }
    | {
        "title": title_prompt | llm | StrOutputParser(),
        "chapter": lambda x: x["chapter"]
    }
)

并行链

同时执行多个独立的任务，提高效率。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnableParallel

llm = ChatOpenAI(model="gpt-3.5-turbo")

# 创建三个不同视角的分析
technical_prompt = ChatPromptTemplate.from_template(
    "从技术角度分析{product}的优缺点"
)

business_prompt = ChatPromptTemplate.from_template(
    "从商业角度分析{product}的优缺点"
)

user_prompt = ChatPromptTemplate.from_template(
    "从用户角度分析{product}的优缺点"
)

# 创建并行链
analysis_chain = RunnableParallel(
    technical=technical_prompt | llm | StrOutputParser(),
    business=business_prompt | llm | StrOutputParser(),
    user=user_prompt | llm | StrOutputParser()
)

# 执行并行分析
result = analysis_chain.invoke({"product": "ChatGPT"})

print("=== 技术角度 ===")
print(result['technical'])
print("\n=== 商业角度 ===")
print(result['business'])
print("\n=== 用户角度 ===")
print(result['user'])

还可以结合并行和顺序，先并行收集信息，再顺序生成报告。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 先并行收集信息，再顺序生成报告
research_chain = RunnableParallel(
    # 并行执行多个研究任务
    market_analysis=research_prompt | llm | StrOutputParser(),
    tech_review=tech_prompt | llm | StrOutputParser(),
    user_feedback=survey_prompt | llm | StrOutputParser()
) | (
    # 顺序生成综合报告
    lambda x: f"""
# 市场分析
{x['market_analysis']}

# 技术评审
{x['tech_review']}

# 用户反馈
{x['user_feedback']}
"""
) | (
    # 最后润色
    polish_prompt | llm | StrOutputParser()
)

final_report = research_chain.invoke({"product": "AI编程助手"})

条件链

根据条件选择不同的执行路径。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnableBranch, RunnableLambda

llm = ChatOpenAI(model="gpt-3.5-turbo")

# 定义不同类型的处理链
code_chain = ChatPromptTemplate.from_template(
    "解释以下代码：\n\n{input}"
) | llm | StrOutputParser()

question_chain = ChatPromptTemplate.from_template(
    "回答以下问题：\n\n{input}"
) | llm | StrOutputParser()

creative_chain = ChatPromptTemplate.from_template(
    "基于以下主题创作一个小故事：\n\n{input}"
) | llm | StrOutputParser()

# 定义路由函数
def classify_input(inputs):
    """分类输入类型"""
    text = inputs["input"].lower()
    if "def " in text or "class " in text or "import " in text:
        return "code"
    elif "?" in text or "什么是" in text or "如何" in text:
        return "question"
    else:
        return "creative"

# 创建条件分支
branch = RunnableBranch(
    (lambda x: x["type"] == "code", code_chain),
    (lambda x: x["type"] == "question", question_chain),
    creative_chain
)

# 完整的链
full_chain = (
    RunnableLambda(lambda x: {"input": x, "type": classify_input({"input": x})})
    | branch
)

# 测试
print("=== 测试代码 ===")
print(full_chain.invoke("def add(a, b): return a + b"))

print("\n=== 测试问题 ===")
print(full_chain.invoke("什么是Python？"))

print("\n=== 测试创作 ===")
print(full_chain.invoke("未来的世界"))

实战项目：智能文档处理流水线

创建一个完整的文档处理系统。

完整代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnableParallel, RunnablePassthrough

class DocumentPipeline:
    """文档处理流水线"""

    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)
        self.parser = StrOutputParser()

    def create_pipeline(self):
        """创建完整的处理流水线"""

        # 步骤1：摘要生成
        summary_prompt = ChatPromptTemplate.from_template(
            "为以下文本生成简洁的摘要（50字以内）：\n\n{text}"
        )

        # 步骤2：关键词提取
        keywords_prompt = ChatPromptTemplate.from_template(
            "从以下文本中提取5个关键词，用逗号分隔：\n\n{text}"
        )

        # 步骤3：翻译
        translate_prompt = ChatPromptTemplate.from_template(
            "将以下文本翻译成英文：\n\n{summary}"
        )

        # 步骤4：情感分析
        sentiment_prompt = ChatPromptTemplate.from_template(
            "分析以下文本的情感倾向（积极/中性/消极）：\n\n{summary}"
        )

        # 创建并行处理步骤
        processing_step = RunnableParallel(
            summary=summary_prompt | self.llm | self.parser,
            keywords=keywords_prompt | self.llm | self.parser
        )

        # 创建顺序处理步骤
        final_step = RunnablePassthrough.assign(
            translation=lambda x: translate_prompt.format(summary=x['summary'])
            | self.llm
            | self.parser,
            sentiment=lambda x: sentiment_prompt.format(summary=x['summary'])
            | self.llm
            | self.parser
        )

        # 组合成完整流水线
        pipeline = processing_step | final_step

        return pipeline

# 使用示例
pipeline = DocumentPipeline()
doc_chain = pipeline.create_pipeline()

# 处理文档
text = """
LangChain是一个强大的框架，用于开发由大语言模型驱动的应用程序。
它提供了标准化的接口，让开发者可以轻松切换不同的模型提供商，
并支持链式调用、工具集成、记忆管理等高级功能。
使用LangChain，可以快速构建聊天机器人、文档问答系统等AI应用。
"""

result = doc_chain.invoke({"text": text})

print("=== 文档摘要 ===")
print(result['summary'])
print("\n=== 关键词 ===")
print(result['keywords'])
print("\n=== 英文翻译 ===")
print(result['translation'])
print("\n=== 情感分析 ===")
print(result['sentiment'])

常见错误

类型不匹配

症状是出现TypeError: expected string or bytes-like object错误。解决方案是确保每个环节的输入输出类型匹配，prompt输出ChatMessage，llm输出AIMessage，parser输出字符串。

1
2
3
4
5
6
# 确保每个环节的输入输出类型匹配
chain = (
    prompt          # 输出: ChatMessage
    | llm           # 输出: AIMessage
    | parser        # 输出: str  ← 必须有parser或手动提取content
)

变量名冲突

症状是出现ValueError: Duplicate key in dictionary错误。解决方案是使用不同的变量名避免冲突。

1
2
3
4
5
6
7
# 使用不同的变量名
chain = (
    {
        "summary": summary_chain,
        "text": lambda x: x["input"]  # 重命名避免冲突
    }
)

系列导航