04_链(Chain)的使用 - 将多个步骤串联起来.

发表于 更新于 分类于 ， Changyan： 本文字数： 6.8k 阅读时长 ≈ 25 分钟

本课程教你使用LangChain的链（Chain）机制将多个处理步骤串联起来。通过学习LCEL（LangChain Expression Language）语法，你将掌握用管道操作符优雅地组合组件。课程涵盖简单链、顺序链、路由链等多种模式，帮助你构建复杂的多步骤AI应用，避免重复代码，提高开发效率。

🎯 学习目标

  • 理解Chain的概念和价值
  • 掌握LCEL（LangChain Expression Language）语法
  • 学会创建不同类型的链
  • 构建多步骤的AI应用

📖 核心概念

什么是Chain？

生活类比：工厂流水线

想象一个生产手机的工厂：

1
2
3
4
5
6
原材料 → 组装车间 → 质检车间 → 包装车间 → 成品

每个车间：
- 接收上一步的输出
- 完成自己的任务
- 传递给下一步

Chain就是AI应用中的”流水线”：

1
2
3
4
5
6
用户输入 → 组件1处理 → 组件2处理 → 组件3处理 → 最终输出

每个组件：
- 接收输入
- 执行任务（调用AI、处理数据等）
- 输出结果

为什么需要Chain？

问题：复杂任务需要多个步骤

场景：智能文章生成器

1
2
3
4
5
6
7
8
9
10
11
12
# ❌ 不用Chain：代码繁琐
step1_result = llm.invoke("生成文章标题：" + topic)
title = step1_result.content

step2_result = llm.invoke("根据标题生成大纲：" + title)
outline = step2_result.content

step3_result = llm.invoke("根据大纲写文章：" + outline)
content = step3_result.content

step4_result = llm.invoke("润色文章：" + content)
final = step4_result.content

问题：

  • 代码重复
  • 难以维护
  • 不便重用
  • 错误处理复杂
1
2
3
4
5
6
7
8
9
# ✅ 用Chain：简洁优雅
chain = (
    generate_title_chain 
    | generate_outline_chain 
    | write_content_chain 
    | polish_chain
)

final = chain.invoke({"topic": topic})

LCEL：LangChain的”管道语法”

LCEL = LangChain Expression Language

核心思想：使用 | 操作符连接组件

1
2
3
4
5
6
# 就像Unix管道
# cat file.txt | grep "error" | wc -l

# LangChain管道
chain = prompt | llm | parser
result = chain.invoke(input_data)

Chain的类型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
1. 简单链 (Simple Chain)
   输入 → 处理 → 输出
   
2. 顺序链 (Sequential Chain)
   步骤1 → 步骤2 → 步骤3 → ...
   
3. 路由链 (Router Chain)
          ┌→ 路径A
   输入 → 判断 ├→ 路径B
          └→ 路径C
          
4. 并行链 (Parallel Chain)
          ┌→ 任务A ┐
   输入 → ┼→ 任务B ┼→ 汇总 → 输出
          └→ 任务C ┘

💻 代码示例

示例1：LCEL基础

参见：01_lcel_basics.py

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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
"""
LCEL基础：使用管道操作符连接组件
演示：prompt | llm | parser 的完整流程
"""

from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import PydanticOutputParser
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from dotenv import load_dotenv
import os

load_dotenv()


class TranslationResult(BaseModel):
    """翻译结果"""
    original: str = Field(description="原文")
    translated: str = Field(description="译文")
    language: str = Field(description="目标语言")


def example1_simple_chain() -> None:
    """示例1：最简单的链 - prompt | llm"""
    print("=== 示例1：Prompt + LLM ===\n")

    # 创建组件
    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    prompt = ChatPromptTemplate.from_template("讲一个关于{topic}的笑话")

    # 使用 | 创建链
    chain = prompt | llm

    # 调用
    result = chain.invoke({"topic": "程序员"})
    print(f"回答：{result.content}\n")


def example2_full_chain() -> None:
    """示例2：完整链 - prompt | llm | parser"""
    print("=== 示例2：Prompt + LLM + Parser ===\n")

    # 创建组件
    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.3,
    )

    parser = PydanticOutputParser(pydantic_object=TranslationResult)

    prompt = ChatPromptTemplate.from_template("""
把以下文本翻译成{language}。

原文：{text}

{format_instructions}
""")

    # 创建完整链
    chain = (
        prompt
        | llm
        | parser
    )

    # 调用
    result = chain.invoke({
        "text": "人工智能正在改变世界",
        "language": "英文",
        "format_instructions": parser.get_format_instructions()
    })

    print(f"原文：{result.original}")
    print(f"译文：{result.translated}")
    print(f"目标语言：{result.language}\n")


def example3_chaining_multiple() -> None:
    """示例3：串联多个处理步骤"""
    print("=== 示例3：多步骤链 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    # 步骤1：生成故事概要
    step1_prompt = ChatPromptTemplate.from_template(
        "用一句话概括{topic}的故事梗概"
    )
    step1_chain = step1_prompt | llm

    # 步骤2：扩写成完整故事
    step2_prompt = ChatPromptTemplate.from_template(
        "把以下故事概要扩写成一个200字的完整故事：\n\n{story_outline}"
    )
    step2_chain = step2_prompt | llm

    # 串联执行
    print("🔄 步骤1：生成概要...")
    outline_result = step1_chain.invoke({"topic": "勇气"})
    print(f"概要：{outline_result.content}\n")

    print("🔄 步骤2：扩写故事...")
    story_result = step2_chain.invoke(
        {"story_outline": outline_result.content})
    print(f"完整故事：\n{story_result.content}\n")


def example4_batch_processing() -> None:
    """示例4：批量处理"""
    print("=== 示例4：批量处理 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    prompt = ChatPromptTemplate.from_template("把{text}翻译成英文，只返回译文")
    chain = prompt | llm

    # 批量处理多个输入
    texts = [
        {"text": "你好"},
        {"text": "谢谢"},
        {"text": "再见"}
    ]

    print("🔄 批量翻译中...")
    results = chain.batch(texts)

    for i, result in enumerate(results, 1):
        print(f"{i}. {texts[i-1]['text']} → {result.content}")


def main() -> None:
    """主函数"""
    example1_simple_chain()
    print("\n" + "="*60 + "\n")

    example2_full_chain()
    print("\n" + "="*60 + "\n")

    example3_chaining_multiple()
    print("\n" + "="*60 + "\n")

    example4_batch_processing()


if __name__ == "__main__":
    main()

核心知识点：

  • 使用 | 连接组件
  • prompt | llm 模式
  • prompt | llm | parser 完整流程

示例2：顺序链

参见：02_sequential_chain.py

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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
"""
顺序链：多步骤处理
演示：如何将多个步骤串联起来，每步的输出作为下一步的输入
"""

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema.runnable import RunnablePassthrough
from dotenv import load_dotenv
import os

load_dotenv()


def example1_simple_sequential() -> None:
    """示例1：简单的顺序链"""
    print("=== 示例1：简单顺序链 - 文章创作流水线 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    # 步骤1：生成标题
    title_prompt = ChatPromptTemplate.from_template(
        "为主题'{topic}'生成一个吸引人的文章标题，只返回标题文本"
    )

    # 步骤2：生成大纲
    outline_prompt = ChatPromptTemplate.from_template(
        "为文章标题'{title}'生成一个包含3个要点的大纲"
    )

    # 步骤3：写文章
    article_prompt = ChatPromptTemplate.from_template(
        """基于以下大纲写一篇300字的文章：

大纲：
{outline}

要求：内容充实，语言流畅
"""
    )

    # 构建顺序链
    # 方式1：分步执行（便于理解）
    topic = "人工智能的未来"

    print(f"📝 主题：{topic}\n")

    # 步骤1：生成标题
    print("🔄 步骤1：生成标题...")
    title_chain = title_prompt | llm
    title_result = title_chain.invoke({"topic": topic})
    title = title_result.content
    print(f"标题：{title}\n")

    # 步骤2：生成大纲
    print("🔄 步骤2：生成大纲...")
    outline_chain = outline_prompt | llm
    outline_result = outline_chain.invoke({"title": title})
    outline = outline_result.content
    print(f"大纲：\n{outline}\n")

    # 步骤3：写文章
    print("🔄 步骤3：写文章...")
    article_chain = article_prompt | llm
    article_result = article_chain.invoke({"outline": outline})
    article = article_result.content
    print(f"文章：\n{article}\n")


def example2_integrated_chain() -> None:
    """示例2：集成的顺序链"""
    print("\n=== 示例2：集成顺序链 - 一次性执行 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    # 定义各步骤的处理函数
    def extract_content(ai_message):
        """提取AI消息的内容"""
        return ai_message.content

    # 步骤1：标题生成
    title_chain = (
        ChatPromptTemplate.from_template(
            "为主题'{topic}'生成一个吸引人的文章标题，只返回标题"
        )
        | llm
        | extract_content
    )

    # 步骤2：大纲生成（需要使用上一步的结果）
    outline_chain = (
        ChatPromptTemplate.from_template(
            "为标题'{title}'生成一个包含3个要点的大纲"
        )
        | llm
        | extract_content
    )

    # 步骤3：文章生成
    article_chain = (
        ChatPromptTemplate.from_template(
            "根据大纲写一篇200字的文章：\n{outline}"
        )
        | llm
        | extract_content
    )

    # 手动串联执行
    topic = "绿色能源的重要性"
    print(f"📝 主题：{topic}\n")

    print("🔄 执行步骤1：生成标题")
    title = title_chain.invoke({"topic": topic})
    print(f"✅ 标题：{title}\n")

    print("🔄 执行步骤2：生成大纲")
    outline = outline_chain.invoke({"title": title})
    print(f"✅ 大纲：\n{outline}\n")

    print("🔄 执行步骤3：生成文章")
    article = article_chain.invoke({"outline": outline})
    print(f"✅ 文章：\n{article}\n")


def example3_data_transformation() -> None:
    """示例3：数据转换链"""
    print("\n=== 示例3：数据转换链 - 翻译处理流水线 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.3,
    )

    # 步骤1：简化文本
    simplify_prompt = ChatPromptTemplate.from_template(
        "将以下文本简化为更简单的表达，保持核心意思：\n\n{text}"
    )

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

    # 步骤3：总结
    summary_prompt = ChatPromptTemplate.from_template(
        "用一句话总结以下英文内容：\n\n{text}"
    )

    # 定义提取函数
    def extract_content(msg):
        return msg.content

    # 构建完整的转换链
    transform_chain = (
        simplify_prompt
        | llm
        | extract_content
        | (lambda text: {"text": text})
        | translate_prompt
        | llm
        | extract_content
        | (lambda text: {"text": text})
        | summary_prompt
        | llm
        | extract_content
    )

    # 测试
    original_text = """
    量子计算机利用量子力学的叠加态和纠缠特性进行计算，
    相比传统计算机，在处理特定类型的问题时具有指数级的速度优势，
    特别是在密码破解、药物研发和优化问题等领域。
    """

    print(f"原文：{original_text.strip()}\n")
    print("🔄 执行转换流水线...\n")

    result = transform_chain.invoke({"text": original_text})

    print(f"✅ 最终结果：{result}\n")


def example4_preserving_intermediate_results() -> None:
    """示例4：保留中间结果"""
    print("\n=== 示例4：保留所有步骤的中间结果 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    from operator import itemgetter

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

    # 步骤2：情感分析
    sentiment_prompt = ChatPromptTemplate.from_template(
        "分析以下文本的情感（正面/负面/中性）：\n\n{text}"
    )

    # 步骤3：摘要生成
    summary_prompt = ChatPromptTemplate.from_template(
        "为以下文本生成一句话摘要：\n\n{text}"
    )

    # 构建保留中间结果的链
    analysis_chain = {
        "original_text": itemgetter("text"),
        "keywords": keywords_prompt | llm | (lambda x: x.content),
        "sentiment": sentiment_prompt | llm | (lambda x: x.content),
        "summary": summary_prompt | llm | (lambda x: x.content),
    }

    # 测试
    text = """
    这款新产品真是太棒了！设计精美，功能强大，使用起来非常流畅。
    客服态度也很好，物流速度快。强烈推荐给大家！
    """

    print(f"原文：{text.strip()}\n")
    print("🔄 执行分析...\n")

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

    print("✅ 分析结果：")
    print(f"原文：{result['original_text'].strip()}")
    print(f"关键词：{result['keywords']}")
    print(f"情感：{result['sentiment']}")
    print(f"摘要：{result['summary']}")


def example5_real_world_pipeline() -> None:
    """示例5：实际应用 - 内容审核流水线"""
    print("\n\n=== 示例5：实际应用 - 内容审核流水线 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.1,  # 降低温度以获得更稳定的结果
    )

    # 步骤1：内容分类
    classify_prompt = ChatPromptTemplate.from_template(
        """对以下内容进行分类，只返回一个类别：
        
内容：{content}

类别选项：正常/广告/违规/敏感

分类："""
    )

    # 步骤2：风险评估
    risk_prompt = ChatPromptTemplate.from_template(
        """评估以下内容的风险等级（低/中/高）：

内容：{content}
分类：{category}

只返回：低、中、高 之一
风险等级："""
    )

    # 步骤3：处理建议
    suggestion_prompt = ChatPromptTemplate.from_template(
        """基于以下信息给出处理建议：

内容：{content}
分类：{category}
风险等级：{risk}

建议："""
    )

    # 构建审核流水线
    def extract_and_preserve(key):
        """提取内容并保留原有数据"""
        def _extract(input_dict):
            # 调用LLM
            if "content" in input_dict and "category" not in input_dict:
                # 第一步：分类
                chain = classify_prompt | llm
                result = chain.invoke({"content": input_dict["content"]})
                return {
                    "content": input_dict["content"],
                    "category": result.content.strip()
                }
            elif "category" in input_dict and "risk" not in input_dict:
                # 第二步：风险评估
                chain = risk_prompt | llm
                result = chain.invoke({
                    "content": input_dict["content"],
                    "category": input_dict["category"]
                })
                return {
                    **input_dict,
                    "risk": result.content.strip()
                }
            else:
                # 第三步：建议
                chain = suggestion_prompt | llm
                result = chain.invoke(input_dict)
                return {
                    **input_dict,
                    "suggestion": result.content.strip()
                }
        return _extract

    # 测试不同类型的内容
    test_contents = [
        "这是一条正常的产品评价，质量不错",
        "低价促销！！！点击链接立即购买！！！",
    ]

    for i, content in enumerate(test_contents, 1):
        print(f"📋 测试内容 {i}：{content}\n")

        # 执行流水线
        result = {"content": content}

        print("🔄 步骤1：分类...")
        result = extract_and_preserve("category")(result)
        print(f"   分类：{result['category']}")

        print("🔄 步骤2：风险评估...")
        result = extract_and_preserve("risk")(result)
        print(f"   风险等级：{result['risk']}")

        print("🔄 步骤3：生成建议...")
        result = extract_and_preserve("suggestion")(result)
        print(f"   处理建议：{result['suggestion']}")

        print("\n" + "="*60 + "\n")


def main() -> None:
    """主函数"""
    example1_simple_sequential()
    print("\n" + "="*70 + "\n")

    example2_integrated_chain()
    print("\n" + "="*70 + "\n")

    example3_data_transformation()
    print("\n" + "="*70 + "\n")

    example4_preserving_intermediate_results()
    print("\n" + "="*70 + "\n")

    example5_real_world_pipeline()


if __name__ == "__main__":
    main()

核心知识点：

  • 多步骤处理
  • 数据传递
  • 中间结果使用

示例3：路由链

参见：03_router_chain.py

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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
"""
路由链：根据条件选择不同的处理路径
演示：智能路由系统，根据输入类型选择不同的处理链
"""

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import Literal
from dotenv import load_dotenv
import os

load_dotenv()


class IntentClassification(BaseModel):
    """意图分类结果"""
    intent: Literal["技术问题", "业务咨询", "闲聊", "投诉"] = Field(description="问题类型")
    confidence: float = Field(description="置信度", ge=0, le=1)


def example1_simple_router() -> None:
    """示例1：简单的路由逻辑"""
    print("=== 示例1：基于规则的简单路由 ===\n")

    def route_question(question: str) -> str:
        """根据关键词路由问题"""
        question_lower = question.lower()

        # 技术问题关键词
        if any(word in question_lower for word in ["代码", "bug", "错误", "api", "函数"]):
            return "技术支持"

        # 业务问题关键词
        elif any(word in question_lower for word in ["价格", "购买", "订单", "发货", "退款"]):
            return "业务咨询"

        # 投诉关键词
        elif any(word in question_lower for word in ["投诉", "不满意", "差评", "退货"]):
            return "客户投诉"

        # 默认为闲聊
        else:
            return "闲聊"

    # 测试不同类型的问题
    questions = [
        "你好，今天天气怎么样？",
        "我的代码报错了，怎么解决？",
        "这个商品多少钱？",
        "你们的服务太差了，我要投诉！"
    ]

    for question in questions:
        route = route_question(question)
        print(f"问题：{question}")
        print(f"路由到：{route}\n")


def example2_llm_based_router() -> None:
    """示例2：基于LLM的智能路由"""
    print("\n=== 示例2：LLM智能路由 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.1,
    )

    # 意图识别链
    parser = PydanticOutputParser(pydantic_object=IntentClassification)

    intent_prompt = ChatPromptTemplate.from_template(
        """分析以下用户问题的意图类型。

问题：{question}

{format_instructions}
"""
    )

    intent_chain = intent_prompt | llm | parser

    # 测试问题
    questions = [
        "Python中的装饰器怎么使用？",
        "我想购买你们的产品，有优惠吗？",
        "你是谁？",
        "我的订单还没发货，已经3天了！"
    ]

    for question in questions:
        print(f"❓ 问题：{question}")

        try:
            result = intent_chain.invoke({
                "question": question,
                "format_instructions": parser.get_format_instructions()
            })

            print(f"📍 意图：{result.intent}")
            print(f"📊 置信度：{result.confidence:.2f}\n")
        except Exception as e:
            print(f"❌ 分析失败：{e}\n")


def example3_multi_chain_router() -> None:
    """示例3：多链路由系统"""
    print("\n=== 示例3：多链路由系统 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    # 定义不同类型问题的处理链

    # 技术支持链
    tech_prompt = ChatPromptTemplate.from_template(
        """你是一个专业的技术支持工程师。

用户问题：{question}

请提供详细的技术解决方案："""
    )
    tech_chain = tech_prompt | llm

    # 业务咨询链
    sales_prompt = ChatPromptTemplate.from_template(
        """你是一个热情的销售顾问。

客户咨询：{question}

请提供专业的产品介绍和购买建议："""
    )
    sales_chain = sales_prompt | llm

    # 闲聊链
    chat_prompt = ChatPromptTemplate.from_template(
        """你是一个友好的聊天机器人。

用户说：{question}

请轻松愉快地回复："""
    )
    chat_chain = chat_prompt | llm

    # 投诉处理链
    complaint_prompt = ChatPromptTemplate.from_template(
        """你是一个专业的客服主管，善于处理客户投诉。

客户投诉：{question}

请表达歉意并提供解决方案："""
    )
    complaint_chain = complaint_prompt | llm

    # 路由函数
    def route_and_process(question: str) -> str:
        """路由并处理问题"""
        question_lower = question.lower()

        # 选择合适的链
        if any(word in question_lower for word in ["代码", "错误", "bug", "技术"]):
            chain = tech_chain
            chain_name = "技术支持"
        elif any(word in question_lower for word in ["价格", "购买", "产品"]):
            chain = sales_chain
            chain_name = "销售顾问"
        elif any(word in question_lower for word in ["投诉", "不满", "差"]):
            chain = complaint_chain
            chain_name = "投诉处理"
        else:
            chain = chat_chain
            chain_name = "闲聊"

        # 执行选中的链
        result = chain.invoke({"question": question})
        return chain_name, result.content

    # 测试
    test_questions = [
        "我的代码运行时出现了内存溢出错误",
        "你们的旗舰产品多少钱？有什么优势？",
        "你好呀，今天心情怎么样？",
        "我要投诉！你们的服务态度太差了！"
    ]

    for question in test_questions:
        print(f"❓ 用户：{question}")
        chain_name, response = route_and_process(question)
        print(f"🔀 路由到：{chain_name}")
        print(f"💬 回复：{response}\n")
        print("="*60 + "\n")


def example4_dynamic_router() -> None:
    """示例4：动态路由 - 根据用户历史"""
    print("\n=== 示例4：动态路由（考虑用户画像）===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    # 模拟用户画像
    user_profiles = {
        "tech_user": {
            "name": "技术用户",
            "level": "高级",
            "preference": "详细技术方案"
        },
        "normal_user": {
            "name": "普通用户",
            "level": "初级",
            "preference": "简单易懂的说明"
        }
    }

    # 高级技术回答链
    advanced_tech_prompt = ChatPromptTemplate.from_template(
        """你是资深技术专家，用户是高级工程师。

问题：{question}

请提供深入的技术分析和代码示例："""
    )
    advanced_chain = advanced_tech_prompt | llm

    # 简单技术回答链
    simple_tech_prompt = ChatPromptTemplate.from_template(
        """你是技术导师，用户是初学者。

问题：{question}

请用通俗易懂的语言解释，避免复杂术语："""
    )
    simple_chain = simple_tech_prompt | llm

    def route_by_user_profile(question: str, user_type: str) -> str:
        """根据用户画像路由"""
        profile = user_profiles[user_type]

        print(f"👤 用户类型：{profile['name']} ({profile['level']})")
        print(f"❓ 问题：{question}\n")

        if profile["level"] == "高级":
            chain = advanced_chain
            print("🔀 路由到：高级技术支持\n")
        else:
            chain = simple_chain
            print("🔀 路由到：基础技术支持\n")

        result = chain.invoke({"question": question})
        return result.content

    # 同一个问题，不同用户获得不同深度的回答
    question = "什么是装饰器？"

    print("场景1：高级用户提问")
    print("="*60)
    response1 = route_by_user_profile(question, "tech_user")
    print(f"💬 回复：{response1}\n\n")

    print("场景2：普通用户提问")
    print("="*60)
    response2 = route_by_user_profile(question, "normal_user")
    print(f"💬 回复：{response2}\n")


def example5_fallback_router() -> None:
    """示例5：带降级策略的路由"""
    print("\n\n=== 示例5：带降级策略的路由 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    # 专业回答链
    professional_prompt = ChatPromptTemplate.from_template(
        """你是专业顾问，提供详细的专业建议。

问题：{question}

专业回答："""
    )
    professional_chain = professional_prompt | llm

    # 通用回答链（降级方案）
    general_prompt = ChatPromptTemplate.from_template(
        """你是友好的助手。

问题：{question}

回答："""
    )
    general_chain = general_prompt | llm

    def route_with_fallback(question: str, has_expert: bool) -> str:
        """带降级的路由"""
        print(f"❓ 问题：{question}")
        print(f"🔍 专家在线：{'是' if has_expert else '否'}\n")

        try:
            if has_expert:
                print("🔀 路由到：专业顾问")
                result = professional_chain.invoke({"question": question})
            else:
                print("🔀 降级到：通用助手")
                result = general_chain.invoke({"question": question})

            return result.content
        except Exception as e:
            print(f"⚠️ 主链失败，使用降级方案：{e}")
            result = general_chain.invoke({"question": question})
            return result.content

    # 测试
    question = "如何优化数据库查询性能？"

    print("场景1：专家在线")
    print("="*60)
    response1 = route_with_fallback(question, has_expert=True)
    print(f"💬 回复：{response1}\n\n")

    print("场景2：专家离线，降级处理")
    print("="*60)
    response2 = route_with_fallback(question, has_expert=False)
    print(f"💬 回复：{response2}\n")


def main() -> None:
    """主函数"""
    example1_simple_router()
    print("\n" + "="*70 + "\n")

    example2_llm_based_router()
    print("\n" + "="*70 + "\n")

    example3_multi_chain_router()
    print("\n" + "="*70 + "\n")

    example4_dynamic_router()
    print("\n" + "="*70 + "\n")

    example5_fallback_router()


if __name__ == "__main__":
    main()

核心知识点：

  • 根据条件选择路径
  • 动态路由
  • 实际应用场景

示例4：实战应用

参见：04_practical_app.py

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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
"""
实战应用：完整的链式应用
演示：综合运用各种链技术构建实用应用
"""

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List
from dotenv import load_dotenv
import os
import json

load_dotenv()


class ProductReview(BaseModel):
    """产品评论分析结果"""
    sentiment: str = Field(description="情感：正面/负面/中性")
    score: int = Field(description="评分1-5分", ge=1, le=5)
    keywords: List[str] = Field(description="关键词列表")
    summary: str = Field(description="一句话总结")
    suggestions: List[str] = Field(description="改进建议")


class EmailResponse(BaseModel):
    """邮件回复"""
    subject: str = Field(description="邮件主题")
    greeting: str = Field(description="问候语")
    body: str = Field(description="正文内容")
    closing: str = Field(description="结束语")
    signature: str = Field(description="签名")


def example1_review_analysis_pipeline() -> None:
    """示例1：产品评论分析流水线"""
    print("=== 示例1：产品评论分析系统 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.3,
    )

    parser = PydanticOutputParser(pydantic_object=ProductReview)

    # 分析链
    analysis_prompt = ChatPromptTemplate.from_template(
        """分析以下产品评论。

评论内容：
{review}

{format_instructions}
"""
    )

    analysis_chain = analysis_prompt | llm | parser

    # 测试评论
    reviews = [
        """
        这款手机真的很棒！屏幕清晰，拍照效果好，运行流畅。
        唯一的小缺点是电池续航可以再长一点。总体来说非常满意！
        """,
        """
        非常失望。买来第二天就出现死机问题，客服态度也不好。
        质量太差了，不推荐购买。
        """,
        """
        还行吧，符合这个价位的水平。没有特别惊艳，也没有明显缺陷。
        """
    ]

    for i, review in enumerate(reviews, 1):
        print(f"📝 评论 {i}：")
        print(review.strip())
        print("\n🔄 分析中...\n")

        try:
            result = analysis_chain.invoke({
                "review": review,
                "format_instructions": parser.get_format_instructions()
            })

            print("✅ 分析结果：")
            print(f"  情感：{result.sentiment}")
            print(f"  评分：{result.score}分")
            print(f"  关键词：{', '.join(result.keywords)}")
            print(f"  总结：{result.summary}")
            print(f"  改进建议：")
            for suggestion in result.suggestions:
                print(f"    - {suggestion}")
            print("\n" + "="*60 + "\n")
        except Exception as e:
            print(f"❌ 分析失败：{e}\n")


def example2_email_generator() -> None:
    """示例2：智能邮件生成器"""
    print("\n=== 示例2：智能邮件生成器 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    parser = PydanticOutputParser(pydantic_object=EmailResponse)

    email_prompt = ChatPromptTemplate.from_template(
        """生成一封专业的商务邮件。

收件人：{recipient}
目的：{purpose}
要点：{key_points}
语气：{tone}

{format_instructions}
"""
    )

    email_chain = email_prompt | llm | parser

    # 测试场景
    scenarios = [
        {
            "recipient": "张经理",
            "purpose": "请假申请",
            "key_points": "家中有事，需要请假3天，工作已安排好交接",
            "tone": "正式礼貌"
        },
        {
            "recipient": "合作伙伴",
            "purpose": "项目合作邀请",
            "key_points": "介绍公司项目，邀请合作，强调双赢",
            "tone": "专业热情"
        }
    ]

    for i, scenario in enumerate(scenarios, 1):
        print(f"📧 场景 {i}：{scenario['purpose']}")
        print(f"收件人：{scenario['recipient']}\n")
        print("🔄 生成邮件...\n")

        try:
            result = email_chain.invoke({
                **scenario,
                "format_instructions": parser.get_format_instructions()
            })

            print("✅ 生成的邮件：")
            print(f"\n主题：{result.subject}")
            print(f"\n{result.greeting}")
            print(f"\n{result.body}")
            print(f"\n{result.closing}")
            print(f"{result.signature}")
            print("\n" + "="*60 + "\n")
        except Exception as e:
            print(f"❌ 生成失败：{e}\n")


def example3_content_moderation_system() -> None:
    """示例3：内容审核系统"""
    print("\n=== 示例3：智能内容审核系统 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.1,
    )

    # 步骤1：内容分类
    classify_prompt = ChatPromptTemplate.from_template(
        """对以下内容进行分类。

内容：{content}

分类（只返回一个）：正常/广告/敏感/违规

分类："""
    )

    # 步骤2：详细分析
    analyze_prompt = ChatPromptTemplate.from_template(
        """详细分析以下内容。

内容：{content}
初步分类：{category}

分析要点：
1. 是否包含不当信息
2. 是否有广告性质
3. 语言是否文明
4. 建议处理方式

分析结果："""
    )

    # 步骤3：生成审核报告
    report_prompt = ChatPromptTemplate.from_template(
        """生成审核报告。

内容：{content}
分类：{category}
详细分析：{analysis}

生成JSON格式的审核报告，包含：
- status: 通过/警告/拒绝
- reason: 原因说明
- action: 处理建议

报告："""
    )

    def moderate_content(content: str) -> dict:
        """完整的审核流程"""
        print(f"📄 待审核内容：{content}\n")

        # 步骤1：分类
        print("🔄 步骤1：内容分类...")
        category_result = (classify_prompt | llm).invoke({"content": content})
        category = category_result.content.strip()
        print(f"   分类：{category}\n")

        # 步骤2：详细分析
        print("🔄 步骤2：详细分析...")
        analysis_result = (analyze_prompt | llm).invoke({
            "content": content,
            "category": category
        })
        analysis = analysis_result.content
        print(f"   分析：{analysis[:100]}...\n")

        # 步骤3：生成报告
        print("🔄 步骤3：生成审核报告...")
        report_result = (report_prompt | llm).invoke({
            "content": content,
            "category": category,
            "analysis": analysis
        })

        try:
            # 尝试解析JSON
            report = json.loads(report_result.content)
        except:
            # 如果不是JSON，使用文本
            report = {
                "status": category,
                "reason": analysis[:50],
                "action": "人工复审"
            }

        return report

    # 测试内容
    test_contents = [
        "这是一条正常的评论，产品质量不错",
        "超低价！点击链接购买！！！限时优惠！！！",
        "你们的服务真是太差了，垃圾！"
    ]

    for content in test_contents:
        report = moderate_content(content)

        print("✅ 审核报告：")
        print(f"   状态：{report.get('status', '未知')}")
        print(f"   原因：{report.get('reason', '无')}")
        print(f"   处理：{report.get('action', '无')}")
        print("\n" + "="*60 + "\n")


def example4_multi_language_translator() -> None:
    """示例4：多语言翻译器"""
    print("\n=== 示例4：智能多语言翻译器 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.3,
    )

    # 步骤1：语言检测
    detect_prompt = ChatPromptTemplate.from_template(
        """检测以下文本的语言，只返回语言名称（中文/英文/日文/韩文等）。

文本：{text}

语言："""
    )

    # 步骤2：翻译
    translate_prompt = ChatPromptTemplate.from_template(
        """将以下{source_lang}文本翻译成{target_lang}。

原文：{text}

译文："""
    )

    # 步骤3：润色
    polish_prompt = ChatPromptTemplate.from_template(
        """润色以下翻译，使其更加自然流畅。

原始翻译：{translation}

润色后："""
    )

    def translate_pipeline(text: str, target_lang: str) -> dict:
        """完整翻译流程"""
        print(f"📝 原文：{text}")
        print(f"🎯 目标语言：{target_lang}\n")

        # 检测源语言
        print("🔄 检测语言...")
        source_lang_result = (detect_prompt | llm).invoke({"text": text})
        source_lang = source_lang_result.content.strip()
        print(f"   源语言：{source_lang}\n")

        # 翻译
        print("🔄 翻译中...")
        translation_result = (translate_prompt | llm).invoke({
            "source_lang": source_lang,
            "target_lang": target_lang,
            "text": text
        })
        translation = translation_result.content
        print(f"   初步翻译：{translation}\n")

        # 润色
        print("🔄 润色中...")
        polished_result = (polish_prompt | llm).invoke({
            "translation": translation
        })
        polished = polished_result.content

        return {
            "source_lang": source_lang,
            "target_lang": target_lang,
            "original": text,
            "translation": translation,
            "polished": polished
        }

    # 测试
    test_cases = [
        ("人工智能正在改变世界", "英文"),
        ("I love programming", "中文"),
    ]

    for text, target in test_cases:
        result = translate_pipeline(text, target)

        print("✅ 翻译结果：")
        print(f"   原文（{result['source_lang']}）：{result['original']}")
        print(f"   译文（{result['target_lang']}）：{result['polished']}")
        print("\n" + "="*60 + "\n")


def example5_error_handling() -> None:
    """示例5：错误处理和日志"""
    print("\n=== 示例5：带错误处理的链 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    import logging
    from datetime import datetime

    # 配置日志
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)

    def log_step(step_name: str):
        """日志装饰器"""
        def decorator(func):
            def wrapper(*args, **kwargs):
                logger.info(f"[{datetime.now()}] 开始执行：{step_name}")
                try:
                    result = func(*args, **kwargs)
                    logger.info(f"[{datetime.now()}] 完成：{step_name}")
                    return result
                except Exception as e:
                    logger.error(f"[{datetime.now()}] 错误：{step_name} - {e}")
                    raise
            return wrapper
        return decorator

    @log_step("文本分析")
    def analyze_text(text: str) -> str:
        """分析文本"""
        prompt = ChatPromptTemplate.from_template(
            "分析以下文本的主题：\n{text}"
        )
        result = (prompt | llm).invoke({"text": text})
        return result.content

    @log_step("生成摘要")
    def generate_summary(analysis: str) -> str:
        """生成摘要"""
        prompt = ChatPromptTemplate.from_template(
            "基于以下分析生成一句话摘要：\n{analysis}"
        )
        result = (prompt | llm).invoke({"analysis": analysis})
        return result.content

    # 测试
    text = "人工智能技术正在各个领域发挥越来越重要的作用"

    print(f"输入：{text}\n")

    try:
        analysis = analyze_text(text)
        print(f"分析结果：{analysis}\n")

        summary = generate_summary(analysis)
        print(f"摘要：{summary}\n")

    except Exception as e:
        print(f"处理失败：{e}")


def main() -> None:
    """主函数"""
    example1_review_analysis_pipeline()
    print("\n" + "="*70 + "\n")

    example2_email_generator()
    print("\n" + "="*70 + "\n")

    example3_content_moderation_system()
    print("\n" + "="*70 + "\n")

    example4_multi_language_translator()
    print("\n" + "="*70 + "\n")

    example5_error_handling()


if __name__ == "__main__":
    main()

核心知识点：

  • 完整的应用示例
  • 错误处理
  • 日志和调试

🎨 Chain设计模式

模式1：转换链（Transform Chain）

1
2
# 输入转换 → 处理 → 输出转换
chain = input_transform | llm | output_transform

使用场景：

  • 数据预处理
  • 格式转换
  • 结果后处理

模式2：分支链（Branch Chain）

1
2
3
4
5
6
7
8
9
10
# 根据条件选择不同的处理路径
from langchain.chains import LLMChain

def route_logic(input_data):
    if input_data["type"] == "A":
        return chain_a
    elif input_data["type"] == "B":
        return chain_b
    else:
        return chain_default

使用场景：

  • 智能客服（不同类型问题走不同流程）
  • 内容分类处理
  • 多模型调度

模式3：循环链（Loop Chain）

1
2
3
4
5
6
# 重复执行直到满足条件
result = initial_input
for i in range(max_iterations):
    result = chain.invoke(result)
    if is_satisfied(result):
        break

使用场景：

  • 迭代优化
  • 对话系统
  • 逐步求精

模式4：Map-Reduce链

1
2
3
4
5
# 分散处理 → 汇总结果
from langchain.chains import MapReduceChain

# Map：并行处理多个文档
# Reduce：汇总所有结果

使用场景：

  • 批量文档处理
  • 大文本摘要
  • 并行分析

📊 数据流转机制

理解输入输出

1
2
3
4
5
6
# Chain的输入输出都是字典
chain.invoke({
    "key1": "value1",
    "key2": "value2"
})
# 返回：{"output": "result"}

变量传递

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from langchain.schema.runnable import RunnablePassthrough

# 场景：保留原始输入
chain = (
    {
        "original": RunnablePassthrough(),  # 传递原始输入
        "processed": prompt | llm          # 处理后的结果
    }
)

result = chain.invoke("Hello")
# result = {
#     "original": "Hello",
#     "processed": "AI的回复"
# }

变量映射

1
2
3
4
5
6
7
8
9
10
from operator import itemgetter

chain = (
    {
        "topic": itemgetter("user_topic"),    # 重命名变量
        "style": itemgetter("writing_style")
    }
    | prompt 
    | llm
)

🔍 常见问题

Q1: Chain和直接调用有什么区别？

直接调用：

1
2
3
result1 = llm1.invoke(input)
result2 = llm2.invoke(result1)
result3 = llm3.invoke(result2)

Chain：

1
2
chain = llm1 | llm2 | llm3
result = chain.invoke(input)

Chain的优势：

  • 代码更简洁
  • 易于重组和重用
  • 统一的错误处理
  • 支持流式输出
  • 便于测试和调试

Q2: 如何调试Chain？

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 方法1：打印中间结果
def debug_step(x):
    print(f"中间结果：{x}")
    return x

chain = prompt | llm | debug_step | parser

# 方法2：使用callbacks
from langchain.callbacks import StdOutCallbackHandler

chain.invoke(input, config={"callbacks": [StdOutCallbackHandler()]})

# 方法3：分步测试
step1 = prompt.invoke(input)
print(f"Step 1: {step1}")

step2 = llm.invoke(step1)
print(f"Step 2: {step2}")

Q3: Chain性能慢怎么办？

优化策略：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 1. 使用批处理
results = chain.batch([input1, input2, input3])

# 2. 并行执行（无依赖关系的步骤）
from langchain.schema.runnable import RunnableParallel

parallel_chain = RunnableParallel(
    task1=chain1,
    task2=chain2,
    task3=chain3
)

# 3. 使用流式输出（提升用户体验）
for chunk in chain.stream(input):
    print(chunk, end="", flush=True)

# 4. 缓存结果
from langchain.cache import InMemoryCache
import langchain
langchain.llm_cache = InMemoryCache()

Q4: 如何处理Chain中的错误？

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from langchain.schema.runnable import RunnableLambda

def safe_chain_step(func):
    """安全包装Chain步骤"""
    def wrapper(input_data):
        try:
            return func(input_data)
        except Exception as e:
            print(f"错误：{e}")
            return {"error": str(e)}
    return wrapper

# 使用
safe_step = RunnableLambda(safe_chain_step(my_function))
chain = prompt | llm | safe_step | parser

📝 练习任务

练习1：文本处理链

创建一个文本处理链：

  1. 输入：原始文本
  2. 步骤1：提取关键词
  3. 步骤2：生成摘要
  4. 步骤3：翻译成英文
  5. 输出：结构化结果

练习2：智能路由器

创建一个问题路由系统：

  • 技术问题 → 技术助手处理
  • 业务问题 → 业务助手处理
  • 闲聊 → 闲聊助手处理

要求：

  • 自动识别问题类型
  • 路由到对应的处理链
  • 返回统一格式的结果

练习3：文章生成流水线

创建完整的文章生成系统：

  1. 输入主题
  2. 生成标题（3个选项）
  3. 用户选择标题
  4. 生成大纲
  5. 逐段生成内容
  6. 润色和修改
  7. 输出完整文章

🎯 检查清单

完成本课后，你应该能够：

  • 理解Chain的概念和价值
  • 使用LCEL语法创建链
  • 组合多个组件完成复杂任务
  • 理解数据在Chain中如何流转
  • 处理Chain中的错误
  • 调试和优化Chain性能

💡 核心要点

  1. Chain是组件的组合

    • 像搭积木一样组合功能
    • 每个环节专注做好一件事

  2. LCEL让代码更优雅

    • 使用 | 操作符串联
    • 代码更简洁易读

  3. 设计好的Chain

    • 单一职责：每步做好一件事
    • 可重用：组件可以复用
    • 可测试：每步都可单独测试

  4. 性能优化

    • 并行处理无依赖的步骤
    • 使用批处理和缓存
    • 流式输出提升体验

📚 下一步

现在你已经学会了如何串联多个步骤，接下来学习如何让AI记住对话历史：

继续学习： 第5课：记忆管理 →