提示链与多步任务分解

当一个任务太过复杂、无法在单个提示词中完成时，我们需要将其拆分为多个按顺序执行的提示词——这就是提示链（Prompt Chaining）。

什么是提示链？

提示链是将一个复杂任务分解为多个子任务，每个子任务使用独立的提示词，前一步的输出作为后一步的输入。

为什么不用一个长提示词？

实战：构建提示链

示例1：文档分析流水线

一个完整的文档分析任务可以分解为：提取→分析→生成报告。

from openai import OpenAI
import json
client = OpenAI()
def call_llm(system: str, user: str, model: str = "gpt-4o-mini", temperature: float = 0) -> str:
"""封装 LLM 调用"""
response = client.chat.completions.create(
model=model,
temperature=temperature,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user}
]
)
return response.choices[0].message.content
def document_analysis_pipeline(document: str) -> dict:
"""
文档分析提示链：
Step 1: 信息提取 → Step 2: 深度分析 → Step 3: 报告生成
"""
# ======= Step 1: 信息提取 =======
step1_result = call_llm(
system="你是一个信息提取专家。只输出JSON，不包含其他文字。",
user=f"""
从以下文档中提取关键信息，输出JSON格式：
{{
"title": "文档标题或主题",
"key_points": ["要点1", "要点2", "要点3"],
"entities": ["提到的人名/公司/产品"],
"numbers": [{{"metric": "指标名", "value": "值"}}],
"sentiment": "positive/negative/neutral"
}}
文档：
---
{document}
---""",
model="gpt-4o-mini"  # 简单任务用小模型降低成本
)
extracted = json.loads(step1_result)
print(f"[Step 1 完成] 提取了 {len(extracted['key_points'])} 个要点")
# ======= Step 2: 深度分析 =======
step2_result = call_llm(
system="你是一位资深的商业分析师。基于提供的数据进行深入分析。",
user=f"""
基于以下提取的信息，进行深度分析。
提取信息：
{json.dumps(extracted, ensure_ascii=False, indent=2)}
原文参考：
{document[:1000]}
请分析：
1. 核心洞察：最重要的3个发现
2. 趋势判断：基于数据推断的趋势
3. 风险识别：潜在的风险或问题
4. 机会点：可以抓住的机会
以JSON格式输出：
{{
"insights": ["洞察1", "洞察2", "洞察3"],
"trends": ["趋势1", "趋势2"],
"risks": ["风险1", "风险2"],
"opportunities": ["机会1", "机会2"]
}}
""",
model="gpt-4o"  # 深度分析用强模型
)
analysis = json.loads(step2_result)
print(f"[Step 2 完成] 识别了 {len(analysis['insights'])} 个洞察")
# ======= Step 3: 报告生成 =======
step3_result = call_llm(
system="""你是一位专业的商务报告撰写者。
使用清晰、专业的中文撰写报告。使用Markdown格式。""",
user=f"""
基于以下分析结果，撰写一份简洁的分析报告。
基础信息：
{json.dumps(extracted, ensure_ascii=False, indent=2)}
分析结果：
{json.dumps(analysis, ensure_ascii=False, indent=2)}
报告要求：
1. 执行摘要（3-5句话）
2. 关键发现（编号列表）
3. 趋势与展望
4. 风险提示
5. 行动建议（按优先级排序）
语气：专业、简洁、数据驱动
长度：500-800字
""",
model="gpt-4o"
)
print("[Step 3 完成] 报告已生成")
return {
"extracted_info": extracted,
"analysis": analysis,
"report": step3_result
}
# 使用
document = """
2025年第四季度，我们的SaaS平台用户增长15%，达到25万月活用户。
企业客户从120家增长到185家，增速54%。平均客单价从800元/月
提升到950元/月。但客户流失率从2.1%上升到3.5%，主要集中在
中小企业客户群。技术支持工单增加了40%，平均响应时间从4小时
延长到7小时。我们在Q4推出了AI助手功能，采纳率达到60%...
"""
result = document_analysis_pipeline(document)
print(result["report"])

示例2：代码生成 Pipeline

def code_generation_chain(requirement: str) -> dict:
"""
代码生成提示链：
Step 1: 需求分析 → Step 2: 架构设计 → Step 3: 代码实现 → Step 4: 测试生成
"""
# Step 1: 需求分析
analysis = call_llm(
system="你是一位需求分析专家。",
user=f"""
分析以下需求，提取关键信息。
需求描述：{requirement}
以JSON输出：
{{
"feature_name": "功能名称",
"inputs": [{{"name": "参数名", "type": "类型", "required": true/false}}],
"outputs": [{{"name": "返回值", "type": "类型"}}],
"business_rules": ["规则1", "规则2"],
"edge_cases": ["边界情况1", "边界情况2"],
"dependencies": ["依赖库1"]
}}
"""
)
# Step 2: 架构设计
design = call_llm(
system="你是一位软件架构师。",
user=f"""
基于以下需求分析，设计代码架构。
需求分析：{analysis}
请输出：
1. 类/函数设计（名称、职责、接口）
2. 设计模式选择及理由
3. 错误处理策略
4. 数据流说明
以清晰的文字描述，不需要写代码。
"""
)
# Step 3: 代码实现
code = call_llm(
system="""你是一位高级Python开发工程师。
代码要求：Python 3.12+，类型注解，docstring，错误处理，遵循PEP 8。""",
user=f"""
基于以下设计方案实现代码。
需求：{requirement}
需求分析：{analysis}
架构设计：{design}
请输出完整的、可运行的Python代码。
包含所有必要的 import 语句。
"""
)
# Step 4: 测试生成
tests = call_llm(
system="你是一位测试工程师，擅长编写 pytest 测试。",
user=f"""
为以下代码编写单元测试。
代码：
{code}
需求中的边界情况：
{analysis}
要求：
1. 使用 pytest 框架
2. 覆盖正常流程、边界情况和异常情况
3. 每个测试函数有清晰的命名和注释
4. 至少5个测试用例
"""
)
return {
"analysis": analysis,
"design": design,
"code": code,
"tests": tests
}

提示链的设计模式

模式1：线性链（Sequential Chain）

最简单的模式，每步按顺序执行。

模式2：分支链（Branching Chain）

根据中间结果选择不同的后续路径。

def branching_chain(user_input: str) -> str:
"""根据输入类型选择不同的处理流程"""
# Step 1: 分类
category = call_llm(
system="你是一个分类器。只输出类别名称，不要其他文字。",
user=f"""
将以下用户输入分类为：bug_report / feature_request / question
输入：{user_input}
类别："""
).strip().lower()
# Step 2: 根据分类选择不同流程
if category == "bug_report":
return handle_bug_report(user_input)
elif category == "feature_request":
return handle_feature_request(user_input)
else:
return handle_question(user_input)

模式3：并行链（Parallel Chain）

多个独立的子任务同时执行，最后汇总。

import asyncio
async def parallel_analysis(text: str) -> dict:
"""并行执行多个分析任务"""
async def async_llm(system: str, user: str) -> str:
"""异步LLM调用"""
response = await asyncio.to_thread(
call_llm, system, user
)
return response
# 并行执行三个任务
summary_task = async_llm(
"你是一个摘要专家。",
f"用3句话总结以下文本：\n{text}"
)
sentiment_task = async_llm(
"你是一个情感分析器。只输出JSON。",
f"分析情感并输出JSON {{\"sentiment\": \"...\", \"confidence\": 0.x}}：\n{text}"
)
keywords_task = async_llm(
"你是一个关键词提取器。",
f"提取5个关键词，以逗号分隔输出：\n{text}"
)
# 等待所有任务完成
summary, sentiment, keywords = await asyncio.gather(
summary_task, sentiment_task, keywords_task
)
return {
"summary": summary,
"sentiment": json.loads(sentiment),
"keywords": keywords.split("，")
}
# 使用
result = asyncio.run(parallel_analysis("你的文本..."))

模式4：循环链（Loop Chain）

重复执行直到满足条件。

def iterative_refinement(task: str, max_iterations: int = 3) -> str:
"""循环优化直到质量合格"""
current_output = call_llm(
system="你是一位专业的技术写作者。",
user=task
)
for i in range(max_iterations):
# 质量检查
review = call_llm(
system="""你是一位严格的内容审核员。
评估文稿质量，输出JSON:
{"score": 1-10, "issues": ["问题1"], "pass": true/false}""",
user=f"审核以下文稿的质量：\n{current_output}"
)
review_data = json.loads(review)
print(f"第{i+1}轮审核：得分 {review_data['score']}/10")
if review_data.get("pass", False) or review_data["score"] >= 8:
print("质量合格，输出最终结果")
return current_output
# 根据反馈修改
current_output = call_llm(
system="你是一位专业的技术写作者。请根据审核反馈改进文稿。",
user=f"""
原始文稿：
{current_output}
审核反馈：
{json.dumps(review_data, ensure_ascii=False)}
请修改文稿以解决上述问题，输出完整的修改后文稿。
"""
)
return current_output

提示链的成本优化

关键原则：为每个步骤选择"刚好够用"的模型。分类、格式转换等简单任务用 gpt-4o-mini，深度分析和生成用 gpt-4o。

动手练习

练习：设计内容审核 Pipeline

设计一个4步提示链来处理用户生成内容（UGC）： 1. 内容分类（文字/图片描述/链接） 2. 安全检查（是否包含违规内容） 3. 质量评估（评分1-10） 4. 自动标签（添加分类标签）

写出每一步的提示词和数据流。

本章要点

• ✅ 提示链将复杂任务拆分为多个简单步骤，提高可靠性和可调试性
• ✅ 四种设计模式：线性链、分支链、并行链、循环链
• ✅ 每步都应有验证机制，防止错误传播
• ✅ 不同步骤可以使用不同的模型来优化成本
• ✅ 并行执行独立步骤可以显著减少总延迟
• ✅ 中间结果缓存是生产环境中的重要优化手段

下一步：提示词注入攻击与防御 🚀