为什么需要 Prompt Chaining
我们都知道，一个 Prompt 塞太多东西，效果往往不好。就像让一个人同时做翻译、校对、排版、配图，不如让四个人各做一步。
Prompt Chaining 的核心思想：把一个复杂任务拆成多个简单步骤，每步用一个专门的 Prompt，前一步的输出作为后一步的输入。
输入 → [Prompt 1] → 中间结果 1 → [Prompt 2] → 中间结果 2 → [Prompt 3] → 最终输出这就是软件工程中的”管道”（Pipeline）思想，应用到了提示工程中。
三种基本模式
模式一：顺序链（Sequential Chain）
最简单的模式，步骤按顺序执行：
步骤 1 → 步骤 2 → 步骤 3 → 输出import anthropic
client = anthropic.Anthropic()
def call_llm(prompt: str, context: str = "") -> str:
"""调用 LLM"""
messages = [{"role": "user", "content": f"{context}\n\n{prompt}" if context else prompt}]
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=messages
)
return response.content[0].text
def sequential_chain(user_input: str) -> str:
"""顺序链：需求分析 → 代码生成 → 代码审查"""
# 步骤 1：需求分析
step1 = call_llm(
prompt=f"""分析以下需求，输出结构化的功能规格：
- 功能列表
- 输入输出定义
- 边界条件
- 错误场景
需求：{user_input}""")
# 步骤 2：代码生成（基于步骤 1 的输出）
step2 = call_llm(
prompt="根据以下功能规格，生成 Python 代码。包含类型注解和文档字符串。",
context=f"功能规格：\n{step1}")
# 步骤 3：代码审查（基于步骤 2 的输出）
step3 = call_llm(
prompt="审查以下代码，指出问题并给出修复后的完整代码。",
context=f"待审查代码：\n{step2}")
return step3模式二：并行链（Parallel Chain）
多个步骤可以同时执行，最后合并结果：
┌→ 步骤 A ─┐
输入 ──→├→ 步骤 B ──├→ 合并 → 输出
└→ 步骤 C ─┘import asyncio
import anthropic
async_client = anthropic.AsyncAnthropic()
async def call_llm_async(prompt: str) -> str:
"""异步调用 LLM"""
response = await async_client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
messages=[{"role": "user", "content": prompt}]
)
return response.content[0].text
async def parallel_chain(code: str) -> str:
"""并行链：同时进行多维度代码审查"""
# 并行执行三个审查任务
security_task = call_llm_async(
f"从安全角度审查以下代码，列出所有安全问题：\n{code}")
performance_task = call_llm_async(
f"从性能角度审查以下代码，列出所有性能问题：\n{code}")
style_task = call_llm_async(
f"从代码风格角度审查以下代码，列出所有风格问题：\n{code}")
# 等待所有任务完成
security, performance, style = await asyncio.gather(
security_task, performance_task, style_task
)
# 合并结果
merge_result = await call_llm_async(
f"""请将以下三个维度的代码审查结果合并为一份完整的审查报告，
按严重程度排序：
安全审查：
{security}
性能审查：
{performance}
风格审查：
{style}""")
return merge_result并行链的优势：速度快。三个审查任务同时执行，总耗时约等于最慢的那个，而不是三个之和。
模式三：条件链（Conditional Chain）
根据中间结果决定下一步走哪条路：
输入 → [分类] → 类型 A → [处理 A] → 输出
类型 B → [处理 B] → 输出
类型 C → [处理 C] → 输出def conditional_chain(user_input: str) -> str:
"""条件链：根据问题类型选择不同处理流程"""
# 步骤 1：分类
category = call_llm(
f"""将以下用户请求分类为以下类型之一，只输出类型名称：
- bug_report（Bug 报告）
- feature_request（功能请求）
- question（技术问题）
- code_review（代码审查）
用户请求：{user_input}""")
category = category.strip().lower()
# 步骤 2：根据分类走不同路径
if "bug" in category:
return call_llm(
f"""作为 Bug 分析专家，请：
1. 分析 Bug 的可能原因
2. 提供复现步骤
3. 给出修复建议
Bug 描述：{user_input}""")
elif "feature" in category:
return call_llm(
f"""作为产品经理，请：
1. 分析需求的合理性
2. 评估实现难度（高/中/低）
3. 给出实现方案建议
功能请求：{user_input}""")
elif "review" in category:
return call_llm(
f"""作为高级工程师，请进行全面代码审查：
1. 逻辑正确性
2. 性能问题
3. 安全问题
4. 改进建议
代码：{user_input}""")
else:
return call_llm(
f"""作为技术顾问，请回答以下技术问题。
给出清晰、准确、有代码示例的回答。
问题：{user_input}""")实战案例：内容生产管道
我们来构建一个完整的博客文章生产管道：
主题 → [大纲生成] → [内容撰写] → [SEO 优化] → [质量检查] → 成品def content_pipeline(topic: str, target_audience: str = "中级开发者") -> dict:
"""博客文章生产管道"""
# 第 1 步：生成大纲
outline = call_llm(f"""为以下主题生成博客文章大纲：
主题：{topic}
目标读者：{target_audience}
要求：
- 3-5 个主要章节
- 每个章节 2-3 个子话题
- 包含代码示例的位置标记
- 预估每个章节的字数
输出 JSON 格式。""")
# 第 2 步：逐章节撰写内容
sections = parse_outline(outline)
full_content = ""
for section in sections:
section_content = call_llm(f"""根据以下大纲章节撰写内容：
文章主题：{topic}
目标读者：{target_audience}
当前章节：{section['title']}
子话题：{section['subtopics']}
已写内容摘要：{summarize(full_content) if full_content else '这是第一个章节'}
要求：
- 使用"我们"的口吻
- 技术术语保留英文
- 包含代码示例
- 自然过渡到下一章节""")
full_content += section_content + "\n\n"
# 第 3 步：SEO 优化
seo_content = call_llm(f"""对以下文章进行 SEO 优化：
{full_content}
请：
1. 优化标题（包含关键词，60 字符以内）
2. 生成 meta description（155 字符以内）
3. 在正文中自然插入关键词
4. 优化小标题结构（H2/H3）
5. 添加内链建议
输出优化后的完整文章。""")
# 第 4 步：质量检查
quality_check = call_llm(f"""对以下文章进行质量检查：
{seo_content}
检查项：
1. 事实准确性（代码示例是否正确）
2. 逻辑连贯性
3. 语法和拼写
4. 可读性评分（1-10）
5. 需要修改的地方
输出检查报告和修改建议。""")
return {
"outline": outline,
"content": seo_content,
"quality_report": quality_check
}步骤间的数据传递
Prompt Chaining 中最关键的问题之一是：如何在步骤之间传递数据。
方式一：全文传递
把上一步的完整输出传给下一步。简单但可能超出 context window：
step2_result = call_llm(f"上一步的输出：\n{step1_result}\n\n请基于此...")方式二：摘要传递
对上一步的输出做摘要，只传关键信息：
summary = call_llm(f"请用 3 句话总结以下内容的关键信息：\n{step1_result}")
step2_result = call_llm(f"背景信息：{summary}\n\n请...")方式三：结构化传递
让每一步输出结构化数据（JSON），下一步只取需要的字段：
# 步骤 1 输出 JSON
step1_result = call_llm("...请输出 JSON 格式...")
data = json.loads(step1_result)
# 步骤 2 只取需要的字段
step2_result = call_llm(f"""
功能列表：{data['features']}
约束条件：{data['constraints']}
请生成代码...""")推荐方式三。结构化数据传递最可靠，也最节省 token。
错误处理
链式调用中，任何一步失败都会影响后续步骤。我们需要健壮的错误处理：
import time
def robust_chain_step(prompt: str, max_retries: int = 3, validate_fn=None) -> str:
"""带重试和验证的链步骤"""
for attempt in range(max_retries):
try:
result = call_llm(prompt)
# 自定义验证
if validate_fn and not validate_fn(result):
raise ValueError(f"验证失败：输出不符合预期格式")
return result
except Exception as e:
if attempt < max_retries - 1:
# 在重试的 prompt 中加入错误信息
prompt = f"""{prompt}
注意：上次尝试失败了，错误信息：{str(e)}
请确保输出格式正确。"""
time.sleep(1)
else:
raise
def validate_json(text: str) -> bool:
"""验证输出是否为有效 JSON"""
import json
try:
json.loads(text)
return True
except json.JSONDecodeError:
return False
# 使用
result = robust_chain_step(
"请输出 JSON 格式的分析结果...",
validate_fn=validate_json
)降级策略
当某一步反复失败时，可以降级处理：
def chain_with_fallback(user_input: str) -> str:
"""带降级策略的链"""
try:
# 尝试完整的 3 步链
step1 = robust_chain_step("分析需求...")
step2 = robust_chain_step("生成代码...", context=step1)
step3 = robust_chain_step("审查代码...", context=step2)
return step3
except Exception:
# 降级：用单个 Prompt 完成所有工作
return call_llm(f"""请完成以下任务：
1. 分析需求
2. 生成代码
3. 自我审查
需求：{user_input}""")实战案例：代码审查管道
async def code_review_pipeline(code: str, language: str = "Python") -> dict:
"""完整的代码审查管道"""
# 第 1 步：代码理解（顺序）
understanding = await call_llm_async(f"""分析以下 {language} 代码：
1. 这段代码的功能是什么？
2. 主要的类/函数有哪些？
3. 数据流是怎样的？
代码：
```{language.lower()}
{code}
```""")
# 第 2 步：多维度审查（并行）
reviews = await asyncio.gather(
call_llm_async(f"""安全审查：
代码功能：{understanding}
代码：{code}
检查 SQL 注入、XSS、认证问题等。"""),
call_llm_async(f"""性能审查：
代码功能：{understanding}
代码：{code}
检查时间复杂度、内存使用、N+1 查询等。"""),
call_llm_async(f"""可维护性审查：
代码功能：{understanding}
代码：{code}
检查命名、注释、复杂度、重复代码等。""")
)
# 第 3 步：生成修复建议（顺序）
fixes = await call_llm_async(f"""基于以下审查结果，生成具体的修复代码：
安全问题：{reviews[0]}
性能问题：{reviews[1]}
可维护性问题：{reviews[2]}
原始代码：{code}
请输出修复后的完整代码，并用注释标注每处修改。""")
return {
"understanding": understanding,
"security_review": reviews[0],
"performance_review": reviews[1],
"maintainability_review": reviews[2],
"fixed_code": fixes
}设计原则
1. 单一职责
每个 Prompt 只做一件事。不要在一个 Prompt 里又分析又生成又审查。
2. 明确的输入输出契约
每个步骤的输入和输出格式要明确定义，最好用 JSON。
3. 可观测性
记录每一步的输入、输出和耗时，方便调试：
import logging
def logged_step(name: str, prompt: str) -> str:
"""带日志的链步骤"""
logging.info(f"[{name}] 开始执行")
start = time.time()
result = call_llm(prompt)
elapsed = time.time() - start
logging.info(f"[{name}] 完成，耗时 {elapsed:.1f}s，输出 {len(result)} 字符")
return result4. 适度拆分
不是拆得越细越好。每增加一步，就增加一次 API 调用的延迟和成本。一般 3-5 步是比较合理的范围。
| 步骤数 | 适用场景 |
|---|---|
| 2 步 | 简单的”生成 + 优化”流程 |
| 3-4 步 | 大多数生产场景 |
| 5-7 步 | 复杂的内容生产或分析管道 |
| 7+ 步 | 考虑是否过度拆分了 |
总结
Prompt Chaining 是构建可靠 AI 应用的基础模式。它把”一个超级 Prompt”变成”多个专业 Prompt 的协作”，每一步都更可控、更可调试。
三种模式各有适用场景：顺序链适合有依赖关系的步骤，并行链适合独立的多维度分析，条件链适合需要分支处理的场景。
好的工程不是写一个万能的函数，而是把复杂问题拆成简单的步骤，让每一步都做到最好。Prompt Chaining 就是提示工程的”Unix 哲学”。
相关文章
评论
加载中...
评论
加载中...