链式提示:把复杂任务"化整为零"的AI编排艺术

你: 请阅读这篇5000字的技术文档,提取核心观点,
    总结成3条要点,每条不超过50字,
    语气要专业但不失幽默,
    最后给出实践建议...
 
AI: [生成了一堆内容,但要点太啰嗦,语气不对,还漏了实践建议]
 
你: (╯°□°)╯︵ ┻━┻

复杂任务
  ↓
[任务分解]
  ↓
子任务1(Prompt 1) → 输出1
  ↓
子任务2(Prompt 2 + 输出1) → 输出2
  ↓
子任务3(Prompt 3 + 输出2) → 输出3
  ↓
最终结果

❌ 超载的Prompt:
请阅读这篇论文,提取关键发现,评估方法的创新性,
指出潜在局限性,对比已有研究,总结实践价值,
用学术语气撰写,格式化为Markdown表格...
 
问题:
- AI容易遗漏某些要求
- 无法保证每个子任务的质量
- 出错后难以定位问题环节
- 难以调整某个环节的处理逻辑

❌ 单一Prompt方法:
将整篇文档和问题一起扔给AI,让它直接回答
 
问题:
1. 文档太长,AI容易"迷失"
2. 无关内容干扰答案质量
3. 难以验证答案的来源
4. 成本高(处理整个文档)

步骤1: 从文档中提取相关引文
  ↓
步骤2: 基于引文生成答案

你是一个很有帮助的助手。你的任务是根据文档回答问题。
第一步是从文档中提取与问题相关的引文,由####分隔。
请使用<quotes></quotes>输出引文列表。
如果没有找到相关引文,请回应"未找到相关引文!"。
 
####
{{文档内容}}
####
 
用户问题: {{问题}}

文档: [关于Prompt Engineering的维基百科文章]
问题: What are the prompting techniques mentioned in the document?

<quotes>
- Chain-of-thought (CoT) prompting[27]
- Generated knowledge prompting[37]
- Least-to-most prompting[38]
- Self-consistency decoding[39]
- Complexity-based prompting[41]
- Self-refine[42]
- Tree-of-thought prompting[43]
- Maieutic prompting[45]
- Directional-stimulus prompting[46]
- Textual inversion and embeddings[59]
- Using gradient descent to search for prompts[61][62][63][64]
- Prompt injection[65][66][67]
</quotes>

根据从文档中提取的相关引文(由<quotes></quotes>分隔)
和原始文档(由####分隔),请构建对问题的回答。
请确保答案准确、语气友好且有帮助。
 
####
{{文档内容}}
####
 
<quotes>
{{步骤1的输出}}
</quotes>
 
用户问题: {{问题}}

文档中提到的提示技术包括:
 
1. Chain-of-thought (CoT) prompting - 思维链提示
2. Generated knowledge prompting - 生成知识提示
3. Least-to-most prompting - 从简到难提示
4. Self-consistency decoding - 自我一致性解码
5. Complexity-based prompting - 基于复杂度的提示
6. Self-refine - 自我精炼
7. Tree-of-thought prompting - 思维树提示
8. Maieutic prompting - 苏格拉底式提示
9. Directional-stimulus prompting - 定向刺激提示
10. Textual inversion and embeddings - 文本反转和嵌入
11. Using gradient descent to search for prompts - 使用梯度下降搜索提示
12. Prompt injection - 提示注入
 
每一种技术都采用独特的策略来增强或明确与大型语言模型的互动,
以产生期望的结果。

❌ 职责混乱:
提取引文 + 评估相关性 + 生成答案 + 格式化输出
 
✅ 职责清晰:
Prompt 1: 提取引文
Prompt 2: 评估相关性
Prompt 3: 生成答案
Prompt 4: 格式化输出

推荐格式:
- JSON: {"key": "value"}
- XML: <tag>content</tag>
- Markdown: ## 标题\n内容
- 自定义分隔符: ####内容####

// Prompt 1 输出(JSON格式)
{
  "relevant_quotes": [
    "引文1",
    "引文2"
  ],
  "confidence": 0.85
}
 
// Prompt 2 可以直接解析使用

✅ 完善的Prompt:
 
你的任务是提取相关引文。
 
正常情况: 输出<quotes>引文列表</quotes>
找不到引文: 输出"未找到相关引文"
文档为空: 输出"文档为空,无法处理"
格式错误: 输出"文档格式错误"

# 伪代码示例
def prompt_chain(document, question):
    # 步骤1: 提取引文
    quotes = extract_quotes(document, question)
    log("Step 1 - Quotes extracted:", quotes)  # 记录中间结果
 
    if not quotes:
        return "未找到相关信息"
 
    # 步骤2: 生成答案
    answer = generate_answer(document, quotes, question)
    log("Step 2 - Answer generated:", answer)  # 记录结果
 
    return answer

❌ 过度拆分:
步骤1: 分词
步骤2: 去停用词
步骤3: 提取关键词
步骤4: 计算相关性
步骤5: 排序
步骤6: 提取引文
→ 太繁琐,效率低
 
✅ 适度拆分:
步骤1: 提取相关引文(包含上述所有操作)
步骤2: 生成答案
→ 简洁高效

用户需求
  ↓
[需求分析] → 提取关键要素
  ↓
[大纲生成] → 生成内容结构
  ↓
[内容撰写] → 逐节撰写
  ↓
[润色优化] → 改善表达
  ↓
[格式化] → 最终输出

代码提交
  ↓
[语法检查] → 发现语法问题
  ↓
[逻辑分析] → 识别逻辑缺陷
  ↓
[性能评估] → 标注性能瓶颈
  ↓
[安全扫描] → 发现安全漏洞
  ↓
[建议汇总] → 生成审查报告

用户问题
  ↓
[意图识别] → 判断问题类型
  ↓
[知识检索] → 查找相关知识
  ↓
[答案生成] → 生成初步答案
  ↓
[情感优化] → 调整语气风格
  ↓
[个性化] → 根据用户画像定制
  ↓
最终回复

原始数据
  ↓
[数据清洗] → 去除异常值
  ↓
[特征提取] → 识别关键特征
  ↓
[趋势分析] → 发现数据趋势
  ↓
[可视化建议] → 推荐图表类型
  ↓
[洞察总结] → 生成分析报告

def sequential_chain(prompts, initial_input):
    current_input = initial_input
 
    for prompt_template in prompts:
        # 填充当前输入到Prompt模板
        prompt = prompt_template.format(input=current_input)
 
        # 调用LLM
        current_input = llm.generate(prompt)
 
        # 记录中间结果
        log_step(prompt, current_input)
 
    return current_input

prompts = [
    "提取这段文本的关键词: {input}",
    "基于这些关键词,生成3个相关问题: {input}",
    "为每个问题提供简短答案: {input}"
]
 
result = sequential_chain(prompts, "人工智能正在改变世界...")

def conditional_chain(document, question):
    # 步骤1: 提取引文
    quotes = extract_quotes_prompt(document, question)
 
    # 条件判断
    if "未找到" in quotes:
        return "抱歉,文档中没有相关信息"
 
    # 步骤2: 评估引文质量
    quality = evaluate_quotes_prompt(quotes)
 
    if quality < 0.5:
        # 质量不够,尝试扩大搜索范围
        quotes = extract_quotes_with_relaxed_criteria(document, question)
 
    # 步骤3: 生成答案
    answer = generate_answer_prompt(document, quotes, question)
 
    return answer

import asyncio
 
async def parallel_chain(text):
    # 并行执行多个分析任务
    tasks = [
        analyze_sentiment(text),      # 情感分析
        extract_entities(text),       # 实体提取
        summarize_text(text),         # 文本摘要
        detect_language(text)         # 语言检测
    ]
 
    # 等待所有任务完成
    results = await asyncio.gather(*tasks)
 
    # 汇总结果
    final_result = merge_results(results)
 
    return final_result

from langchain.chains import SequentialChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
 
# 定义子链
chain1 = LLMChain(
    llm=OpenAI(),
    prompt=PromptTemplate(
        input_variables=["document", "question"],
        template="从文档中提取与问题相关的引文:\n文档:{document}\n问题:{question}"
    ),
    output_key="quotes"
)
 
chain2 = LLMChain(
    llm=OpenAI(),
    prompt=PromptTemplate(
        input_variables=["document", "quotes", "question"],
        template="基于引文回答问题:\n引文:{quotes}\n问题:{question}"
    ),
    output_key="answer"
)
 
# 组合成顺序链
overall_chain = SequentialChain(
    chains=[chain1, chain2],
    input_variables=["document", "question"],
    output_variables=["answer"]
)
 
# 执行
result = overall_chain({"document": doc, "question": q})
print(result["answer"])

# 创建Prompt模块库
class PromptModules:
    @staticmethod
    def extract_quotes(document, question):
        return f"""
        从文档中提取与问题相关的引文。
 
        文档:
        {document}
 
        问题: {question}
 
        输出格式: <quotes>引文列表</quotes>
        """
 
    @staticmethod
    def generate_answer(document, quotes, question):
        return f"""
        基于引文生成答案。
 
        引文:
        {quotes}
 
        问题: {question}
 
        要求: 准确、简洁、友好
        """
 
# 复用模块
quotes = llm.generate(PromptModules.extract_quotes(doc, q))
answer = llm.generate(PromptModules.generate_answer(doc, quotes, q))

def validated_chain(document, question):
    # 步骤1: 提取引文
    quotes = extract_quotes(document, question)
 
    # 验证: 检查引文格式
    if not is_valid_xml(quotes):
        quotes = fix_xml_format(quotes)
 
    # 验证: 检查引文相关性
    relevance = check_relevance(quotes, question)
    if relevance < 0.6:
        log_warning("引文相关性较低")
 
    # 步骤2: 生成答案
    answer = generate_answer(document, quotes, question)
 
    # 验证: 检查答案完整性
    if len(answer) < 50:
        answer = regenerate_with_more_details(document, quotes, question)
 
    return answer

def chain_with_retry(prompt, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = llm.generate(prompt)
 
            # 验证结果
            if is_valid_result(result):
                return result
 
            log(f"结果验证失败,重试 {attempt + 1}/{max_retries}")
 
        except Exception as e:
            log(f"生成失败: {e}, 重试 {attempt + 1}/{max_retries}")
 
    return "处理失败,请稍后重试"

❌ 问题:
步骤1: 从长文档提取摘要(200字)
步骤2: 基于摘要回答细节问题
→ 摘要可能丢失了回答所需的关键细节
 
✅ 解决:
步骤1: 提取相关段落(保留完整信息)
步骤2: 基于完整段落回答问题

❌ 问题:
步骤1提取的引文有误 → 步骤2基于错误引文生成答案 → 最终答案错误
 
✅ 解决:
- 在关键步骤添加验证
- 保留原始输入,让后续步骤可以"回头看"
- 实现降级策略(如果中间结果质量差,回退到更保守的方法)

❌ 问题:
10个步骤,每步调用一次GPT-4 → 成本是单次调用的10倍
 
✅ 解决:
- 合并不必要的步骤
- 非关键步骤使用更便宜的模型(如GPT-3.5)
- 缓存重复调用的结果

❌ 问题:
步骤1的小错误 → 步骤2放大 → 步骤3继续放大 → 最终结果完全错误
 
✅ 解决:
- 在每个关键步骤进行结果验证
- 设置质量阈值,低于阈值则中止链条
- 实现"熔断机制",检测到异常立即停止

def adaptive_chain(text, question):
    # 评估复杂度
    complexity = assess_complexity(text, question)
 
    if complexity < 0.3:
        # 简单任务,单步处理
        return simple_answer(text, question)
 
    elif complexity < 0.7:
        # 中等任务,两步处理
        quotes = extract_quotes(text, question)
        return generate_answer(text, quotes, question)
 
    else:
        # 复杂任务,完整链条
        quotes = extract_quotes(text, question)
        refined_quotes = refine_quotes(quotes, question)
        answer = generate_detailed_answer(text, refined_quotes, question)
        return polish_answer(answer)

def chain_with_feedback(document, question):
    # 第一轮: 提取引文
    quotes = extract_quotes(document, question)
 
    # 第二轮: 尝试生成答案
    answer = generate_answer(document, quotes, question)
 
    # 检查答案质量
    quality = evaluate_answer_quality(answer, question)
 
    if quality < 0.6:
        # 质量不够,反馈给第一步,重新提取更多引文
        quotes = extract_more_quotes(document, question, current_quotes=quotes)
        answer = generate_answer(document, quotes, question)
 
    return answer

def human_in_the_loop_chain(document, question):
    # 步骤1: AI提取引文
    quotes = extract_quotes(document, question)
 
    # 人工检查点
    if require_human_review(quotes):
        quotes = human_review_and_edit(quotes)
 
    # 步骤2: 基于人工确认的引文生成答案
    answer = generate_answer(document, quotes, question)
 
    return answer

class DocumentQASystem:
    def __init__(self, llm):
        self.llm = llm
        self.cache = {}
 
    def answer_question(self, document, question):
        """主流程"""
        try:
            # 步骤1: 提取引文
            quotes = self._extract_quotes(document, question)
 
            # 验证引文质量
            if not self._validate_quotes(quotes):
                return "无法找到相关信息"
 
            # 步骤2: 生成答案
            answer = self._generate_answer(document, quotes, question)
 
            # 验证答案质量
            if not self._validate_answer(answer):
                # 重试一次
                answer = self._generate_answer(document, quotes, question)
 
            # 步骤3(可选): 润色答案
            final_answer = self._polish_answer(answer)
 
            return final_answer
 
        except Exception as e:
            self._log_error(e)
            return "处理出错,请稍后重试"
 
    def _extract_quotes(self, document, question):
        """步骤1: 提取引文"""
        cache_key = f"quotes_{hash(document)}_{hash(question)}"
 
        if cache_key in self.cache:
            return self.cache[cache_key]
 
        prompt = f"""
        你是一个很有帮助的助手。从文档中提取与问题相关的引文。
 
        文档:
        ####
        {document}
        ####
 
        问题: {question}
 
        输出格式: <quotes>引文列表</quotes>
        如果没有相关引文,输出: 未找到相关引文
        """
 
        quotes = self.llm.generate(prompt)
        self.cache[cache_key] = quotes
 
        return quotes
 
    def _validate_quotes(self, quotes):
        """验证引文质量"""
        if "未找到" in quotes:
            return False
 
        if not quotes.strip():
            return False
 
        # 检查格式
        if not ("<quotes>" in quotes and "</quotes>" in quotes):
            return False
 
        return True
 
    def _generate_answer(self, document, quotes, question):
        """步骤2: 生成答案"""
        prompt = f"""
        根据引文回答问题。要求准确、简洁、友好。
 
        引文:
        {quotes}
 
        问题: {question}
 
        请生成答案:
        """
 
        return self.llm.generate(prompt)
 
    def _validate_answer(self, answer):
        """验证答案质量"""
        if len(answer) < 20:
            return False
 
        if "不知道" in answer or "无法回答" in answer:
            return False
 
        return True
 
    def _polish_answer(self, answer):
        """步骤3: 润色答案(可选)"""
        # 简单处理,实际可以更复杂
        return answer.strip()
 
    def _log_error(self, error):
        """错误日志"""
        print(f"Error: {error}")
 
# 使用示例
system = DocumentQASystem(llm=my_llm)
answer = system.answer_question(
    document=long_document,
    question="What are the main findings?"
)
print(answer)

✅ 适合使用链式提示:
- 多步骤处理任务(数据清洗 → 分析 → 可视化)
- 需要中间验证的任务(提取 → 验证 → 使用)
- 复杂的内容生成(大纲 → 撰写 → 润色)
- 对话系统(意图识别 → 检索 → 生成回复)
 
❌ 不适合使用:
- 简单的单步任务(如文本翻译)
- 对延迟要求极高的场景(多次调用增加延迟)
- 预算非常有限的场景(多次调用增加成本)

✅ 设计阶段:
- [ ] 明确任务目标和约束条件
- [ ] 识别可拆解的子任务
- [ ] 设计每个子任务的输入输出格式
- [ ] 确定子任务之间的依赖关系
- [ ] 考虑异常情况和降级策略
 
✅ 实现阶段:
- [ ] 为每个子任务编写独立的Prompt
- [ ] 实现链条编排逻辑
- [ ] 添加日志和监控
- [ ] 实现重试和错误处理机制
- [ ] 添加缓存优化性能
 
✅ 验证阶段:
- [ ] 测试每个子任务的独立功能
- [ ] 测试完整链条的端到端流程
- [ ] 验证异常情况的处理
- [ ] 评估性能和成本
- [ ] 收集用户反馈并迭代