提示链模式（Prompt Chaining）

提示链模式（Prompt Chaining），又称流水线模式（Pipeline），是将复杂任务分解为多个简单子任务，每个子任务对应一次 LLM 调用，前一步的输出作为下一步的输入，依次串联执行。

核心思想：分而治之 —— 每步只做一件事，上一步的输出即下一步的输入。

为什么需要提示链？

单次 Prompt 处理复杂任务时，容易出现遗漏、幻觉或格式混乱。提示链通过分步执行来解决这些问题：

• 可控性强：每一步都可以独立验证和调试
• 准确性高：子任务更简单，模型出错概率更低
• 可组合：不同的链条可以灵活复用和重新编排

适用场景

提示链适合确定性、线性的工作流，例如：

实现步骤

1. 定义任务：明确目标，将其拆解为多个子任务
2. 设计提示：为每个子任务编写 Prompt，确保输入输出格式明确
3. 串联执行：按顺序连接各步骤，前一步输出传递给下一步
4. 质量检查（可选）：在关键步骤之间加入校验逻辑（如格式检查、长度限制），不通过则重试或终止
5. 评估优化：检查最终输出，根据效果调整各步骤的 Prompt

流程示意

输入 → [Prompt 1] → 输出₁ → [Prompt 2] → 输出₂ → [Prompt 3] → 最终结果
                        ↓ (可选)
                    [质量检查] —— 不通过 → 重试/终止

代码示例

• 示例 1创意写作流水线 —— 生成大纲 → 撰写故事 → 起标题与总结

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
parser = StrOutputParser()

# 步骤1: 生成一个故事大纲
step1_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个创意写作助手。"),
    ("human", "请为以下主题生成一个简短的三幕故事大纲：{topic}")
])

# 步骤2: 基于大纲写出完整故事
step2_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的故事作家。"),
    ("human", "基于以下故事大纲，写一个300字左右的短篇故事：\n\n{outline}")
])

# 步骤3: 为故事生成总结和标题
step3_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个文学编辑。"),
    ("human", "为以下故事起一个吸引人的标题，并写一句话总结：\n\n{story}")
])

# 构建流水线链
chain1 = step1_prompt | llm | parser
chain2 = step2_prompt | llm | parser
chain3 = step3_prompt | llm | parser


def run_pipeline(topic: str):
    print(f"主题: {topic}\n")

    print("=== 步骤1: 生成大纲 ===")
    outline = chain1.invoke({"topic": topic})
    print(outline, "\n")

    print("=== 步骤2: 撰写故事 ===")
    story = chain2.invoke({"outline": outline})
    print(story, "\n")

    print("=== 步骤3: 标题与总结 ===")
    result = chain3.invoke({"story": story})
    print(result)


if __name__ == "__main__":
    run_pipeline("一只猫在太空中的冒险")

• 示例 2信息提取流水线 —— 从文本提取技术规格 → 转为结构化 JSON

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()

llm = ChatOpenAI(temperature=0)

# --- 提示 1：信息提取 ---
prompt_extract = ChatPromptTemplate.from_template(
    "请从以下文本中提取技术规格：\n\n{text_input}"
)

# --- 提示 2：转为 JSON ---
prompt_transform = ChatPromptTemplate.from_template(
    "请将以下技术规格转为 JSON 格式，包含 'cpu'、'memory' 和 'storage' 三个键：\n\n{specifications}"
)

# --- 用 LCEL 构建链 ---
extraction_chain = prompt_extract | llm | StrOutputParser()

full_chain = (
    {"specifications": extraction_chain}
    | prompt_transform
    | llm
    | StrOutputParser()
)

# --- 运行链 ---
input_text = "新款笔记本配备 3.5GHz 八核处理器、16GB 内存和 1TB NVMe SSD。"

final_result = full_chain.invoke({"text_input": input_text})

print("\n--- 最终 JSON 输出 ---")
print(final_result)

两个示例均使用 LangChain 的 LCEL（| 管道语法）构建链条。

局限性

• 延迟累积：步骤越多，总耗时越长
• 开销较大：每步都要放入大量的上下文，token 开销很大
• 错误传播：前一步的错误会影响后续所有步骤
• 灵活性有限：只适合线性流程，无法根据中间结果动态选择不同路径（需要 → 路由模式）