03_LangChain核心组件 - 模型、提示词、输出解析器

本课程深入讲解LangChain的三大核心组件:语言模型(LLM)、提示词模板(Prompt)、输出解析器(Parser)。你将学会如何创建可重用的提示词模板、设计有效的提示词、使用Pydantic获取结构化输出。课程通过大量实例展示组件的单独使用和组合应用,为后续构建复杂应用打下坚实基础。

🎯 学习目标

  • 深入理解LangChain的三大核心组件
  • 掌握提示词工程的基本技巧
  • 学会使用输出解析器获取结构化数据
  • 理解组件之间如何协作

📚 核心概念

组件概览:三个好朋友

想象你在经营一家智能咨询公司:

1
2
3
4
5
6
7
8
9
10
11
1. 模型 (LLM) = 聪明的顾问
- 有知识,会思考
- 但需要明确的指示

2. 提示词 (Prompt) = 给顾问的工作说明书
- 告诉顾问:任务是什么、怎么做、输出什么格式
- 说明书写得好,顾问工作效果好

3. 输出解析器 (Parser) = 秘书
- 把顾问的回答整理成标准格式
- 便于后续处理和使用

1. 模型 (LLM) - 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
# 1. 聊天模型 (ChatModel) - 最常用
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
model="qwen-plus",
temperature=0.7
)

# 特点:
# - 支持多轮对话
# - 理解角色(系统、用户、助手)
# - 适合大多数场景
# 2. 文本补全模型 (LLM) - 简单场景
from langchain_openai import OpenAI

llm = OpenAI(
model="text-davinci-003",
temperature=0.7
)

# 特点:
# - 只是续写文本
# - 不记录对话历史
# - 适合简单的文本生成

模型的调用方式

1
2
3
4
5
6
7
8
9
10
11
12
# 方式1: invoke - 单次调用
response = llm.invoke("你好")
print(response.content)

# 方式2: batch - 批量调用
responses = llm.batch(["问题1", "问题2", "问题3"])
for resp in responses:
print(resp.content)

# 方式3: stream - 流式输出(像打字机)
for chunk in llm.stream("讲一个故事"):
print(chunk.content, end="", flush=True)

2. 提示词 (Prompt) - 沟通的艺术

为什么需要提示词模板?

没有模板(硬编码):

1
2
3
4
5
# ❌ 不好的做法
response = llm.invoke(f"把{text}翻译成英文")
response = llm.invoke(f"把{text}翻译成日文")
response = llm.invoke(f"把{text}翻译成法文")
# 问题:重复代码,难以维护

使用模板:

1
2
3
4
5
6
7
8
9
10
11
# ✅ 好的做法
from langchain.prompts import PromptTemplate

template = PromptTemplate(
input_variables=["text", "language"],
template="把{text}翻译成{language}"
)

# 重用模板
prompt1 = template.format(text="你好", language="英文")
prompt2 = template.format(text="你好", language="日文")

提示词的组成部分

1
2
3
4
5
6
7
8
9
[系统角色] - 定义AI的身份和行为规范

[任务描述] - 说明要做什么

[输入数据] - 具体的数据

[输出格式] - 期望的输出格式

[示例(可选)] - 给AI看例子

聊天提示词模板

1
2
3
4
5
6
7
8
9
from langchain.prompts import ChatPromptTemplate

# 多角色对话
chat_template = ChatPromptTemplate.from_messages([
("system", "你是一个专业的{role}"),
("human", "请帮我{task}"),
("ai", "好的,我会帮你{task}"),
("human", "{user_input}")
])

3. 输出解析器 (Parser) - 结构化输出

为什么需要解析器?

问题:AI的输出是文本,不是数据

1
2
3
4
5
6
7
8
9
10
# AI返回:
"北京今天晴天,温度25度,适合出行"

# 但你需要:
{
"city": "北京",
"weather": "晴天",
"temperature": 25,
"advice": "适合出行"
}

解析器的类型

  1. 简单解析器 - 提取基本数据
  2. JSON解析器 - 获取JSON格式
  3. Pydantic解析器 - 获取类型安全的对象
  4. 列表解析器 - 获取列表

💻 代码示例

示例1:提示词模板基础

参见:01_prompt_template.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
"""
提示词模板基础
演示:如何创建和使用提示词模板
"""

from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

load_dotenv()


def example1_basic_template() -> None:
"""示例1:基础模板使用"""
print("=== 示例1:基础模板 ===\n")

# 创建模板
template = PromptTemplate(
input_variables=["product", "language"],
template="请为{product}写一个{language}版本的产品介绍,不超过50字"
)

# 使用模板
prompt1 = template.format(product="智能手表", language="中文")
prompt2 = template.format(product="智能手表", language="英文")

print("生成的提示词1:")
print(prompt1)
print("\n生成的提示词2:")
print(prompt2)
print()


def example2_with_llm() -> None:
"""示例2:模板配合LLM使用"""
print("=== 示例2:模板配合LLM ===\n")

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

# 创建翻译模板
translate_template = PromptTemplate(
input_variables=["text", "target_language"],
template="""
你是一个专业的翻译员。

任务:把以下文本翻译成{target_language}
原文:{text}

翻译:
"""
)

# 测试翻译
texts = [
"人工智能正在改变世界",
"今天天气真好",
"我喜欢编程"
]

for text in texts:
prompt = translate_template.format(
text=text,
target_language="英文"
)
response = llm.invoke(prompt)
print(f"原文:{text}")
print(f"翻译:{response.content}\n")


def example3_reusable_template() -> 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.8,
)

# 创建一个通用的内容生成模板
content_template = PromptTemplate(
input_variables=["role", "task", "style", "length"],
template="""
你是一个{role}。

任务:{task}
风格:{style}
长度:{length}

请开始:
"""
)

# 场景1:写诗
prompt1 = content_template.format(
role="诗人",
task="写一首关于秋天的诗",
style="优美抒情",
length="4句"
)
response1 = llm.invoke(prompt1)
print("📝 场景1:写诗")
print(response1.content)
print()

# 场景2:写代码
prompt2 = content_template.format(
role="Python工程师",
task="写一个计算斐波那契数列的函数",
style="简洁高效,有注释",
length="不超过15行"
)
response2 = llm.invoke(prompt2)
print("💻 场景2:写代码")
print(response2.content)
print()


def example4_template_from_string() -> None:
"""示例4:从字符串创建模板"""
print("=== 示例4:快速创建模板 ===\n")

# 方式1:使用from_template(更简洁)
template1 = PromptTemplate.from_template(
"给我讲一个关于{topic}的{length}故事"
)

prompt = template1.format(topic="勇气", length="短小")
print("生成的提示词:")
print(prompt)
print()

# 方式2:带验证的创建
template2 = PromptTemplate(
input_variables=["topic", "length"],
template="给我讲一个关于{topic}的{length}故事",
validate_template=True # 验证模板格式
)

print("✅ 模板验证通过")


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

example2_with_llm()
print("\n" + "="*50 + "\n")

example3_reusable_template()
print("\n" + "="*50 + "\n")

example4_template_from_string()


if __name__ == "__main__":
main()

核心知识点:

  • 创建和使用提示词模板
  • 变量替换
  • 重用模板

示例2:聊天提示词模板

参见:02_chat_prompt.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
"""
聊天提示词模板
演示:如何使用ChatPromptTemplate处理多角色对话
"""

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage, SystemMessage
from dotenv import load_dotenv
import os

load_dotenv()


def example1_basic_chat_template() -> None:
"""示例1:基础聊天模板"""
print("=== 示例1:基础聊天模板 ===\n")

# 创建聊天模板
chat_template = ChatPromptTemplate.from_messages([
("system", "你是一个{role},你的特点是{characteristic}"),
("human", "{user_input}")
])

# 格式化模板
messages = chat_template.format_messages(
role="幽默的助手",
characteristic="说话风趣,喜欢用比喻",
user_input="什么是云计算?"
)

print("生成的消息:")
for msg in messages:
print(f"{msg.__class__.__name__}: {msg.content}")
print()


def example2_chat_with_llm() -> None:
"""示例2:聊天模板配合LLM"""
print("=== 示例2:聊天模板配合LLM ===\n")

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

# 场景:智能客服
customer_service_template = ChatPromptTemplate.from_messages([
("system", """
你是一个专业的客服人员,工作在{company}公司。
你的职责是:
1. 友好地回答客户问题
2. 提供准确的信息
3. 保持礼貌和耐心
4. 回答要简洁明了
"""),
("human", "客户问题:{question}")
])

# 测试不同问题
questions = [
"你们的退货政策是什么?",
"订单什么时候能发货?",
"如何联系人工客服?"
]

for question in questions:
messages = customer_service_template.format_messages(
company="科技商城",
question=question
)
response = llm.invoke(messages)
print(f"客户:{question}")
print(f"客服:{response.content}\n")


def example3_multi_turn_conversation() -> 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,
)

# 创建支持历史对话的模板
chat_template = ChatPromptTemplate.from_messages([
("system", "你是一个Python编程导师,擅长解释编程概念"),
MessagesPlaceholder(variable_name="chat_history"), # 历史对话占位符
("human", "{user_input}")
])

# 模拟多轮对话
chat_history = []

# 第一轮
user_input1 = "什么是列表推导式?"
messages = chat_template.format_messages(
chat_history=chat_history,
user_input=user_input1
)
response1 = llm.invoke(messages)
print(f"学生:{user_input1}")
print(f"导师:{response1.content}\n")

# 更新历史
chat_history.append(HumanMessage(content=user_input1))
chat_history.append(AIMessage(content=response1.content))

# 第二轮(基于上下文)
user_input2 = "能给我举个例子吗?"
messages = chat_template.format_messages(
chat_history=chat_history,
user_input=user_input2
)
response2 = llm.invoke(messages)
print(f"学生:{user_input2}")
print(f"导师:{response2.content}\n")


def example4_role_based_template() -> 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.9, # 提高创意
)

# 角色扮演模板
roleplay_template = ChatPromptTemplate.from_messages([
("system", "你现在扮演{character},请完全进入角色"),
("human", "场景:{scenario}"),
("human", "{action}")
])

# 测试不同角色
scenarios = [
{
"character": "一个侦探",
"scenario": "在案发现场调查",
"action": "发现了一个可疑的脚印,你会怎么做?"
},
{
"character": "一位厨师",
"scenario": "在厨房准备晚餐",
"action": "客人突然说对海鲜过敏,你会怎么调整菜单?"
}
]

for scenario in scenarios:
messages = roleplay_template.format_messages(**scenario)
response = llm.invoke(messages)
print(f"角色:{scenario['character']}")
print(f"场景:{scenario['scenario']}")
print(f"情况:{scenario['action']}")
print(f"回应:{response.content}\n")


def example5_template_composition() -> None:
"""示例5:模板组合"""
print("=== 示例5:组合多个模板 ===\n")

# 系统角色模板
system_template = "你是一个{profession},专长是{specialty}"

# 任务模板
task_template = """
任务类型:{task_type}
具体要求:{requirements}
"""

# 组合成完整模板
full_template = ChatPromptTemplate.from_messages([
("system", system_template),
("human", task_template)
])

messages = full_template.format_messages(
profession="数据分析师",
specialty="用数据讲故事",
task_type="分析销售数据",
requirements="找出销售下降的原因,用简单的语言解释"
)

print("组合后的消息:")
for msg in messages:
print(f"\n{msg.__class__.__name__}:")
print(msg.content)


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

example2_chat_with_llm()
print("\n" + "="*50 + "\n")

example3_multi_turn_conversation()
print("\n" + "="*50 + "\n")

example4_role_based_template()
print("\n" + "="*50 + "\n")

example5_template_composition()


if __name__ == "__main__":
main()

核心知识点:

  • 多角色对话
  • 系统提示词设置
  • 对话历史管理

示例3:输出解析器

参见:03_output_parser.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
"""
输出解析器
演示:如何获取结构化的输出
"""

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

load_dotenv()


# 定义数据模型
class ProductReview(BaseModel):
"""产品评论分析结果"""
product_name: str = Field(description="产品名称")
sentiment: str = Field(description="情感:正面/负面/中性")
score: int = Field(description="评分1-5", ge=1, le=5)
keywords: List[str] = Field(description="关键词列表")
summary: str = Field(description="一句话总结")


class PersonInfo(BaseModel):
"""人物信息"""
name: str = Field(description="姓名")
age: int = Field(description="年龄")
occupation: str = Field(description="职业")
hobbies: List[str] = Field(description="爱好列表")


def example1_pydantic_parser() -> None:
"""示例1:使用Pydantic解析器"""
print("=== 示例1:Pydantic解析器 ===\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)

# 创建提示词模板
template = PromptTemplate(
template="""
分析以下产品评论。

评论:{review}

{format_instructions}
""",
input_variables=["review"],
partial_variables={
"format_instructions": parser.get_format_instructions()}
)

# 测试评论
review = "这个手机太棒了!拍照清晰,运行流畅,电池续航也很好。强烈推荐!"

# 生成提示词
prompt = template.format(review=review)
print("📝 发送给AI的提示词:")
print(prompt)
print("\n" + "="*50 + "\n")

# 调用LLM
response = llm.invoke(prompt)
print("🤖 AI的原始回复:")
print(response.content)
print("\n" + "="*50 + "\n")

try:
# 解析输出
result = parser.parse(response.content)
print("✅ 解析成功!\n")
print(f"产品名称:{result.product_name}")
print(f"情感:{result.sentiment}")
print(f"评分:{result.score}")
print(f"关键词:{', '.join(result.keywords)}")
print(f"总结:{result.summary}")
except ValidationError as e:
print(f"❌ 解析失败:{e}")


def example2_list_parser() -> 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 = CommaSeparatedListOutputParser()

# 创建提示词
template = PromptTemplate(
template="""
列出{count}个{category}。

{format_instructions}
""",
input_variables=["count", "category"],
partial_variables={
"format_instructions": parser.get_format_instructions()}
)

# 测试
prompt = template.format(count="5", category="编程语言")
print(f"📝 提示词:{prompt}\n")

response = llm.invoke(prompt)
print(f"🤖 AI回复:{response.content}\n")

# 解析为列表
result = parser.parse(response.content)
print("✅ 解析结果(列表):")
for i, item in enumerate(result, 1):
print(f"{i}. {item}")


def example3_complex_structure() -> 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,
)

# 创建解析器
parser = PydanticOutputParser(pydantic_object=PersonInfo)

# 创建提示词
template = PromptTemplate(
template="""
根据以下描述,提取人物信息。

描述:{description}

{format_instructions}
""",
input_variables=["description"],
partial_variables={
"format_instructions": parser.get_format_instructions()}
)

# 测试
description = "张三是一名30岁的软件工程师,他喜欢编程、阅读和跑步。"
prompt = template.format(description=description)

response = llm.invoke(prompt)
print(f"🤖 AI回复:{response.content}\n")

try:
result = parser.parse(response.content)
print("✅ 解析成功!\n")
print(f"姓名:{result.name}")
print(f"年龄:{result.age}")
print(f"职业:{result.occupation}")
print(f"爱好:{', '.join(result.hobbies)}")
except ValidationError as e:
print(f"❌ 解析失败:{e}")


def example4_error_handling() -> 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,
)

parser = PydanticOutputParser(pydantic_object=ProductReview)

template = PromptTemplate(
template="""
分析以下产品评论。

评论:{review}

{format_instructions}

重要:请务必严格按照JSON格式输出,不要有任何其他文字。
""",
input_variables=["review"],
partial_variables={
"format_instructions": parser.get_format_instructions()}
)

review = "还行吧,凑合用"
prompt = template.format(review=review)

max_retries = 3
for attempt in range(max_retries):
try:
print(f"🔄 尝试 {attempt + 1}/{max_retries}")
response = llm.invoke(prompt)
result = parser.parse(response.content)

print("✅ 解析成功!")
print(json.dumps(result.model_dump(), ensure_ascii=False, indent=2))
break

except ValidationError as e:
print(f"❌ 解析失败:{e}")
if attempt < max_retries - 1:
print("重试中...\n")
else:
print("达到最大重试次数,使用降级方案")
print(f"原始文本:{response.content}")


def example5_custom_parser() -> None:
"""示例5:自定义简单解析器"""
print("\n=== 示例5:自定义解析器 ===\n")

class SimpleKeyValueParser:
"""简单的键值对解析器"""

def parse(self, text: str) -> dict:
"""解析键值对格式的文本"""
result = {}
lines = text.strip().split('\n')
for line in lines:
if ':' in line:
key, value = line.split(':', 1)
result[key.strip()] = value.strip()
return result

def get_format_instructions(self) -> str:
"""返回格式说明"""
return """
请按照以下格式输出,每行一个键值对:
键: 值
键: 值
"""

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

parser = SimpleKeyValueParser()

template = PromptTemplate(
template="""
总结这本书的信息。

书名:{book_name}

{format_instructions}

输出示例:
书名: xxx
作者: xxx
类型: xxx
出版年份: xxx
""",
input_variables=["book_name"],
partial_variables={
"format_instructions": parser.get_format_instructions()}
)

prompt = template.format(book_name="三体")
response = llm.invoke(prompt)

print(f"🤖 AI回复:\n{response.content}\n")

result = parser.parse(response.content)
print("✅ 解析结果(字典):")
for key, value in result.items():
print(f" {key}: {value}")


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

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

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

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

example5_custom_parser()


if __name__ == "__main__":
main()

核心知识点:

  • 使用Pydantic定义输出结构
  • 自动解析AI输出
  • 类型验证

示例4:完整流程

参见:04_complete_flow.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
"""
完整流程:组件协作
演示:模型 + 提示词 + 解析器的完整应用
"""

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

load_dotenv()


# 数据模型
class EmailAnalysis(BaseModel):
"""邮件分析结果"""
sender: str = Field(description="发件人")
category: str = Field(description="邮件类别:工作/个人/营销/垃圾")
priority: str = Field(description="优先级:高/中/低")
requires_reply: bool = Field(description="是否需要回复")
key_points: List[str] = Field(description="关键要点列表")
suggested_reply: str = Field(description="建议回复内容")


class ArticleOutline(BaseModel):
"""文章大纲"""
title: str = Field(description="文章标题")
introduction: str = Field(description="引言")
sections: List[str] = Field(description="主要章节标题列表")
conclusion: str = Field(description="结论")
estimated_words: int = Field(description="预计字数")


def example1_email_assistant() -> None:
"""示例1:智能邮件助手"""
print("=== 示例1:智能邮件助手 ===\n")

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

# 2. 创建解析器
parser = PydanticOutputParser(pydantic_object=EmailAnalysis)

# 3. 创建提示词模板
template = ChatPromptTemplate.from_messages([
("system", """
你是一个智能邮件助手,帮助用户分析和处理邮件。
你需要:
1. 识别邮件类别和优先级
2. 提取关键信息
3. 判断是否需要回复
4. 建议回复内容
"""),
("human", """
分析以下邮件:

发件人:{sender}
主题:{subject}
内容:{content}

{format_instructions}
""")
])

# 4. 准备数据
email_data = {
"sender": "张经理",
"subject": "关于下周项目评审会议",
"content": """
你好,

下周一下午3点我们需要召开项目评审会议,请准备以下材料:
1. 项目进度报告
2. 技术方案文档
3. 风险评估

请确认能否参加,谢谢!
"""
}

# 5. 组合流程
messages = template.format_messages(
**email_data,
format_instructions=parser.get_format_instructions()
)

# 6. 调用LLM
print("🔄 正在分析邮件...")
response = llm.invoke(messages)

# 7. 解析结果
try:
result = parser.parse(response.content)

print("✅ 分析完成!\n")
print("="*60)
print(f"📧 发件人:{result.sender}")
print(f"📁 类别:{result.category}")
print(f"⭐ 优先级:{result.priority}")
print(f"💬 需要回复:{'是' if result.requires_reply else '否'}")
print("\n📌 关键要点:")
for i, point in enumerate(result.key_points, 1):
print(f" {i}. {point}")
print(f"\n💡 建议回复:\n{result.suggested_reply}")
print("="*60)

except Exception as e:
print(f"❌ 解析失败:{e}")


def example2_article_generator() -> None:
"""示例2:文章大纲生成器"""
print("\n=== 示例2:文章大纲生成器 ===\n")

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

# 2. 创建解析器
parser = PydanticOutputParser(pydantic_object=ArticleOutline)

# 3. 创建提示词模板
template = ChatPromptTemplate.from_messages([
("system", """
你是一个专业的内容策划师,擅长规划文章结构。
你的大纲需要:
1. 逻辑清晰
2. 层次分明
3. 内容充实
4. 适合目标读者
"""),
("human", """
为以下主题创建文章大纲:

主题:{topic}
目标读者:{audience}
文章风格:{style}
期望长度:{length}

{format_instructions}
""")
])

# 4. 准备数据
article_info = {
"topic": "如何开始学习Python",
"audience": "编程初学者",
"style": "通俗易懂,循序渐进",
"length": "2000字左右"
}

# 5. 生成大纲
messages = template.format_messages(
**article_info,
format_instructions=parser.get_format_instructions()
)

print("🔄 正在生成文章大纲...")
response = llm.invoke(messages)

# 6. 解析和展示
try:
result = parser.parse(response.content)

print("✅ 大纲生成完成!\n")
print("="*60)
print(f"📝 标题:{result.title}\n")
print(f"📖 引言:\n{result.introduction}\n")
print("📑 章节:")
for i, section in enumerate(result.sections, 1):
print(f" {i}. {section}")
print(f"\n🎬 结论:\n{result.conclusion}\n")
print(f"📊 预计字数:{result.estimated_words}字")
print("="*60)

except Exception as e:
print(f"❌ 解析失败:{e}")
print(f"原始输出:\n{response.content}")


def example3_chain_pattern() -> None:
"""示例3:使用LCEL链式调用"""
print("\n=== 示例3:链式调用模式 ===\n")

# 1. 创建组件
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=EmailAnalysis)

template = ChatPromptTemplate.from_messages([
("system", "你是邮件助手"),
("human", "分析邮件:{email}\n\n{format_instructions}")
])

# 2. 使用 | 操作符链接组件
chain = template | llm | parser

# 3. 一行调用
email = """
发件人:李总
内容:紧急!客户要求明天提交方案,请今晚加班完成。
"""

print("🔄 使用链式调用处理...")

try:
result = chain.invoke({
"email": email,
"format_instructions": parser.get_format_instructions()
})

print("✅ 处理完成!\n")
print(f"类别:{result.category}")
print(f"优先级:{result.priority}")
print(f"关键要点:{', '.join(result.key_points)}")

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


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

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

example3_chain_pattern()


if __name__ == "__main__":
main()

核心知识点:

  • 组件组合使用
  • 链式调用
  • 实际应用

🎨 提示词工程最佳实践

技巧1:明确角色和任务

❌ 不好的提示词:

1
"写一篇文章"

✅ 好的提示词:

1
2
3
4
5
6
7
8
你是一位资深的技术博客作者,擅长把复杂的技术概念用通俗易懂的语言解释给初学者。

任务:写一篇关于"什么是Docker"的技术文章
要求:
1. 使用生活中的类比解释概念
2. 包含实际使用场景
3. 语言通俗易懂
4. 长度500-800字

技巧2:提供示例(Few-shot Learning)

1
2
3
4
5
6
7
8
9
10
11
12
template = """
任务:分类用户评论的情感(正面/负面/中性)

示例:
评论:"这个产品太棒了!" → 正面
评论:"质量一般,价格偏贵" → 负面
评论:"还可以,符合预期" → 中性

现在分类这条评论:
评论:"{comment}"
分类:
"""

技巧3:明确输出格式

1
2
3
4
5
6
7
8
9
10
11
12
13
template = """
分析以下产品评论,输出JSON格式:

评论:{review}

输出格式(严格按照此格式):
{{
"sentiment": "正面/负面/中性",
"score": 1-5分,
"keywords": ["关键词1", "关键词2"],
"summary": "一句话总结"
}}
"""

技巧4:使用约束和限制

1
2
3
4
5
6
7
8
9
10
11
12
template = """
生成一个Python函数,满足以下要求:

功能:{function_description}

约束:
1. 必须有类型注解
2. 必须有docstring
3. 必须有错误处理
4. 代码不超过30行
5. 只使用Python标准库
"""

📊 组件协作流程图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
用户输入

[提示词模板]
↓ 格式化
完整提示词

[语言模型]
↓ 生成
AI文本输出

[输出解析器]
↓ 解析
结构化数据

应用程序使用

🔍 常见问题

Q1: 什么时候用PromptTemplate,什么时候用ChatPromptTemplate?

PromptTemplate: 简单的单轮问答

1
2
# 适合:翻译、摘要、分类等单一任务
PromptTemplate(template="把{text}翻译成英文")

ChatPromptTemplate: 需要角色、多轮对话

1
2
# 适合:客服、助手、需要上下文的任务
ChatPromptTemplate.from_messages([...])

Q2: 如何调试提示词?

1
2
3
4
5
6
7
# 1. 打印格式化后的提示词
prompt = template.format(...)
print(prompt)

# 2. 测试不同的措辞
# 3. 查看AI的输出是否符合预期
# 4. 逐步调整提示词

Q3: 输出解析失败怎么办?

1
2
3
4
5
6
7
8
9
10
from langchain.output_parsers import PydanticOutputParser
from pydantic import ValidationError

try:
result = parser.parse(ai_output)
except ValidationError as e:
print(f"解析失败:{e}")
# 策略1:重新生成
# 策略2:降级为文本处理
# 策略3:提示AI修正格式

Q4: 如何提高输出的稳定性?

1
2
3
4
5
6
7
8
9
10
11
# 1. 降低temperature
llm = ChatOpenAI(temperature=0.1)

# 2. 在提示词中明确格式
template = """
务必严格按照以下JSON格式输出,不要有任何其他内容:
{{"name": "...", "age": ...}}
"""

# 3. 使用更强的模型
llm = ChatOpenAI(model="qwen-max")

📝 练习任务

练习1:创建自己的提示词模板

创建一个”智能邮件助手”:

  • 输入:收件人、主题、要点
  • 输出:正式的商务邮件

要求:

  • 使用PromptTemplate
  • 可以自定义语气(正式/友好/简洁)

练习2:使用输出解析器

创建一个”天气查询助手”:

  • 输入:城市名称
  • 输出:结构化的天气信息(使用Pydantic)

要求:

  • 定义WeatherInfo模型
  • 使用PydanticOutputParser
  • 处理解析错误

练习3:组合应用

创建一个”产品评论分析器”:

  • 读取产品评论
  • 分析情感、提取关键词
  • 输出结构化报告

要求:

  • 使用ChatPromptTemplate
  • 使用PydanticOutputParser
  • 完整的错误处理

🎯 检查清单

完成本课后,你应该能够:

  • 理解模型、提示词、解析器的作用
  • 创建和使用提示词模板
  • 设计有效的提示词
  • 使用输出解析器获取结构化数据
  • 组合多个组件完成任务
  • 调试和优化提示词

💡 核心要点

  1. 提示词是AI应用的核心

    • 提示词质量直接影响输出质量
    • 投入时间优化提示词是值得的
  2. 使用模板提高可维护性

    • 避免硬编码
    • 便于重用和修改
  3. 结构化输出更可靠

    • 使用Pydantic定义数据结构
    • 类型安全,便于后续处理
  4. 组件化思维

    • 每个组件做好一件事
    • 组合起来实现复杂功能

📚 下一步

现在你已经掌握了LangChain的核心组件,接下来学习如何把它们串联起来:


继续学习: 第4课:链(Chain)的使用 →