Prompt Chaining 提示链
定义
世界上有一种设计模式叫做Prompt Chaining，它就像一条珍珠项链——每一颗珍珠单独看都很美，但串在一起才能成为传世之作...
Prompt Chaining（提示链）是将复杂任务分解为多个子任务，通过串联多个Prompt逐步完成目标的技术。每个链节处理特定子任务，将结果传递给下一个链节。
核心原理
1. 链式架构
┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
│ Prompt1 │ → │ Prompt2 │ → │ Prompt3 │ → │ Output │
│ 任务1 │ │ 任务2 │ │ 任务3 │ │ 最终 │
└─────────┘ └─────────┘ └─────────┘ └─────────┘
↓ ↓ ↓
Output1 Output2 Output3
↓ ↓ ↓
传递给下一步继续处理
2. 链式类型
| 类型 | 特点 | 适用场景 |
|---|---|---|
| 顺序链 | 按顺序执行，上游输出作为下游输入 | 分析→总结→建议 |
| 并行链 | 多个Prompt同时执行，结果合并 | 多角度分析、多源信息 |
| 条件链 | 根据上一步结果决定下一步 | 分支逻辑、动态流程 |
| 循环链 | 结果反馈重新执行，直到满足条件 | 迭代优化、质量控制 |
3. 关键设计原则
- 单一职责：每个Prompt只完成一个明确子任务
- 清晰接口：定义好上下游数据传递格式
- 错误处理：某链节失败时的降级策略
- 状态管理：跟踪整个链的执行状态
OpenClaw实战应用
案例：文章创作流水线
用户: "帮我写一篇AI发展趋势的文章"
链1 - 选题阶段:
Prompt: "分析当前AI领域热点，列出5个值得写的选题"
Output: ["AGI进展", "多模态发展", "AI Agent", "边缘AI", "AI伦理"]
链2 - 资料收集:
Prompt: "针对'AI Agent'选题，搜索最新动态和技术趋势"
Output: [新闻、研究报告、技术文章...]
链3 - 大纲生成:
Prompt: "基于以下资料，为AI Agent文章生成结构化大纲"
Output: [引言、发展历史、技术原理、应用场景、未来展望]
链4 - 正文写作:
Prompt: "按照以下大纲，撰写完整的AI Agent文章"
Output: [完整文章内容]
链5 - 审核优化:
Prompt: "检查文章质量，提出修改建议"
Output: [优化建议列表]
→ 如需修改，返回链4重新执行
OpenClaw Skill中的链式设计
# SKILL.md - 内容创作技能
## 链1：需求分析
- 提取用户需求关键词
- 确定内容类型和风格
## 链2：资料收集
- 搜索相关信息
- 整理关键要点
## 链3：大纲生成
- 基于资料生成结构
- 确保逻辑连贯
## 链4：内容撰写
- 按大纲填充内容
- 保持风格一致
## 链5：质量检查
- 检查语法错误
- 验证事实准确性
- 优化可读性
代码实现
// OpenClaw 中的提示链实现
async function promptChain(tasks, context) {
let currentResult = context;
for (const task of tasks) {
// 构建当前链节的Prompt
const prompt = buildPrompt(task, currentResult);
// 执行Prompt
const result = await llm.generate(prompt);
// 验证结果
if (!validateResult(result, task.criteria)) {
// 循环优化
for (let i = 0; i < task.max_retries; i++) {
const refined = await llm.generate(
buildRefinePrompt(task, result)
);
if (validateResult(refined, task.criteria)) {
result = refined;
break;
}
}
}
currentResult = {
...currentResult,
[task.name]: result
};
}
return currentResult;
}
代码示例
基础提示链
async def prompt_chain(prompts: list[str]) -> list[str]:
"""顺序执行提示链"""
results = []
for prompt in prompts:
response = await llm.agenerate(prompt)
results.append(response)
return results
# 示例：分析报告流水线
prompts = [
"分析{topic}的市场现状",
"基于以下分析，预测未来趋势：{results[0]}",
"根据预测，提出投资建议：{results[1]}"
]
report = await prompt_chain(prompts)
带条件的动态链
async def conditional_chain(context: dict) -> dict:
"""条件分支提示链"""
# 第一步：分析任务
analysis = await llm.generate(
f"分析以下任务复杂度：{context['task']}"
)
# 根据复杂度决定路径
if analysis.complexity == "high":
# 高复杂度：增加更多链节
steps = ["research", "outline", "draft", "review", "revise"]
else:
# 低复杂度：简化流程
steps = ["outline", "draft"]
# 执行相应链节
result = context
for step in steps:
prompt = build_step_prompt(step, result)
result[step] = await llm.generate(prompt)
return result
LangChain LCEL实现
from langchain.schema import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
# 定义每个链节
prompt1 = ChatPromptTemplate.from_template(
"总结这段文本：{text}"
)
prompt2 = ChatPromptTemplate.from_template(
"基于以下总结提出3个问题：{summary}"
)
prompt3 = ChatPromptTemplate.from_template(
"回答以下问题：{questions}"
)
# 构建链
chain = (
prompt1
| llm
| StrOutputParser()
| (lambda x: {"summary": x}) # 传递结果
| prompt2
| llm
| StrOutputParser()
| (lambda x: {"questions": x})
| prompt3
| llm
| StrOutputParser()
)
# 执行
result = await chain.ainvoke({"text": "原始文本..."})
最佳实践
- ✅ 任务拆分合理：每个子任务复杂度适中
- ✅ 定义清晰接口：上下游数据格式标准化
- ✅ 添加验证环节：确保每步输出质量
- ✅ 实现重试机制：处理临时失败
- ✅ 记录执行日志：便于调试和优化
- ❌ 链节不宜过多：增加延迟和错误风险
- ❌ 避免信息丢失：每步都要传递关键上下文