提示词链编排：复杂任务的分治策略

原创灵阙教研团队

A 推荐提升最佳实践 | 约 7 分钟阅读更新于 2026-02-28

AI 导读

提示词链编排：复杂任务的分治策略 Chain-of-Prompts、条件路由、错误传播与 LangChain LCEL 实现 | 2026-02 一、为什么需要提示词链单个提示词在面对复杂任务时会遇到瓶颈：上下文窗口不够用、指令过于复杂导致遵循率下降、不同子任务需要不同的模型或参数。提示词链（Chain of Prompts）将复杂任务分解为多个简单步骤，每个步骤使用专门优化的提示词。...

提示词链编排：复杂任务的分治策略

Chain-of-Prompts、条件路由、错误传播与 LangChain LCEL 实现 | 2026-02

一、为什么需要提示词链

单个提示词在面对复杂任务时会遇到瓶颈：上下文窗口不够用、指令过于复杂导致遵循率下降、不同子任务需要不同的模型或参数。提示词链（Chain of Prompts）将复杂任务分解为多个简单步骤，每个步骤使用专门优化的提示词。

Single Prompt (fragile):
  Complex Question -> [Mega Prompt] -> Answer (often wrong)

Chain of Prompts (robust):
  Complex Question -> [Classify] -> [Retrieve] -> [Synthesize] -> [Validate] -> Answer
                         |             |              |              |
                      Simple,        Focused,       Specialized,   Quality
                      reliable       efficient      accurate       checked

二、链式编排模式

2.1 基础模式

模式	描述	适用场景
顺序链	A -> B -> C	固定流程
条件链	A -> if(X) B else C	分类后分支
并行链	A -> [B, C] -> D	独立子任务
循环链	A -> B -> (check) -> A	自我修正
树形链	A -> [B -> D, C -> E] -> F	复杂分解
Map-Reduce	Split -> [Process...] -> Merge	批量处理

2.2 架构图

Pattern 1: Sequential Chain
  Input -> [Step 1] -> [Step 2] -> [Step 3] -> Output

Pattern 2: Conditional Chain (Router)
  Input -> [Classifier] --"tech"--> [Tech Support]
                        --"billing"--> [Billing Agent]
                        --"general"--> [General FAQ]

Pattern 3: Parallel Chain
  Input -> [Extract Entities] ---|
        -> [Analyze Sentiment] --+--> [Merge & Synthesize] -> Output
        -> [Detect Language]   ---|

Pattern 4: Self-Correcting Loop
  Input -> [Generate] -> [Validate] --pass--> Output
                             |
                           fail
                             |
                             v
                       [Fix with feedback] ---> [Validate]
                       (max 3 iterations)

Pattern 5: Map-Reduce
  Long Document -> [Split into chunks]
                        |
               +--------+--------+
               |        |        |
          [Summarize] [Summarize] [Summarize]
               |        |        |
               +--------+--------+
                        |
                   [Merge summaries] -> Final Summary

三、LangChain LCEL 实现

3.1 LCEL 基础

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_openai import ChatOpenAI

# LCEL: LangChain Expression Language
# Uses the pipe (|) operator to chain components

# Simple chain: prompt | model | parser
classify_prompt = ChatPromptTemplate.from_messages([
    ("system", "Classify the user query into: tech_support, billing, general"),
    ("user", "{query}"),
])

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

# This creates a Runnable chain
classify_chain = classify_prompt | model | parser

# Execute
result = await classify_chain.ainvoke({"query": "My invoice is wrong"})
# result: "billing"

3.2 条件路由

from langchain_core.runnables import RunnableLambda, RunnablePassthrough

# Define specialized chains for each category
tech_chain = (
    ChatPromptTemplate.from_messages([
        ("system", "You are a technical support specialist. Diagnose and solve the issue."),
        ("user", "{query}"),
    ])
    | ChatOpenAI(model="gpt-4o", temperature=0.3)
    | StrOutputParser()
)

billing_chain = (
    ChatPromptTemplate.from_messages([
        ("system", "You are a billing specialist. Help with invoice and payment issues."),
        ("user", "{query}"),
    ])
    | ChatOpenAI(model="gpt-4o-mini", temperature=0.3)  # Cheaper model for billing
    | StrOutputParser()
)

general_chain = (
    ChatPromptTemplate.from_messages([
        ("system", "You are a helpful general assistant."),
        ("user", "{query}"),
    ])
    | ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
    | StrOutputParser()
)

# Router function
def route(info: dict) -> object:
    category = info["category"].strip().lower()
    if "tech" in category:
        return tech_chain
    elif "billing" in category:
        return billing_chain
    else:
        return general_chain

# Complete routing chain
full_chain = (
    {"category": classify_chain, "query": RunnablePassthrough()}
    | RunnableLambda(route)
)

# Execute
result = await full_chain.ainvoke({"query": "My invoice is wrong"})

3.3 并行链

from langchain_core.runnables import RunnableParallel

# Run multiple analyses in parallel
parallel_analysis = RunnableParallel(
    entities=(
        ChatPromptTemplate.from_messages([
            ("system", "Extract named entities (people, companies, locations)."),
            ("user", "{text}"),
        ])
        | model
        | JsonOutputParser()
    ),
    sentiment=(
        ChatPromptTemplate.from_messages([
            ("system", "Analyze sentiment. Output: positive/negative/neutral with score."),
            ("user", "{text}"),
        ])
        | model
        | JsonOutputParser()
    ),
    topics=(
        ChatPromptTemplate.from_messages([
            ("system", "Extract main topics (max 5)."),
            ("user", "{text}"),
        ])
        | model
        | JsonOutputParser()
    ),
)

# Merge results
synthesis_chain = (
    parallel_analysis
    | ChatPromptTemplate.from_messages([
        ("system", "Synthesize the analysis into a comprehensive report."),
        ("user", "Entities: {entities}\nSentiment: {sentiment}\nTopics: {topics}"),
    ])
    | model
    | StrOutputParser()
)

report = await synthesis_chain.ainvoke({"text": article_text})

四、自纠错循环

4.1 验证-修复循环

from pydantic import BaseModel, ValidationError

class AnalysisResult(BaseModel):
    summary: str
    key_points: list[str]
    confidence: float

async def generate_with_validation(
    query: str, max_retries: int = 3,
) -> AnalysisResult:
    """Generate structured output with self-correction loop."""
    generate_prompt = ChatPromptTemplate.from_messages([
        ("system", "Analyze the query and provide structured analysis as JSON."),
        ("user", "{query}"),
    ])

    fix_prompt = ChatPromptTemplate.from_messages([
        ("system", "Fix the JSON output based on the validation error."),
        ("user", "Original output:\n{output}\n\nError:\n{error}\n\nFix it:"),
    ])

    generate_chain = generate_prompt | model | StrOutputParser()
    fix_chain = fix_prompt | model | StrOutputParser()

    output = await generate_chain.ainvoke({"query": query})

    for attempt in range(max_retries):
        try:
            parsed = json.loads(output)
            return AnalysisResult.model_validate(parsed)
        except (json.JSONDecodeError, ValidationError) as e:
            if attempt == max_retries - 1:
                raise
            output = await fix_chain.ainvoke({
                "output": output, "error": str(e),
            })

    raise ValueError("Failed to generate valid output")

4.2 质量门禁

async def chain_with_quality_gate(query: str) -> str:
    """Chain with intermediate quality check."""
    # Step 1: Generate draft
    draft_chain = (
        ChatPromptTemplate.from_messages([
            ("system", "Write a detailed analysis."),
            ("user", "{query}"),
        ])
        | ChatOpenAI(model="gpt-4o")
        | StrOutputParser()
    )

    # Step 2: Quality check (cheaper model as judge)
    quality_chain = (
        ChatPromptTemplate.from_messages([
            ("system", """Rate this analysis on a scale of 1-10.
Output JSON: {"score": N, "issues": ["issue1", ...]}
Score >= 7 = pass. Below 7 = needs improvement."""),
            ("user", "Query: {query}\n\nAnalysis: {draft}"),
        ])
        | ChatOpenAI(model="gpt-4o-mini", temperature=0)
        | JsonOutputParser()
    )

    # Step 3: Improve if needed
    improve_chain = (
        ChatPromptTemplate.from_messages([
            ("system", "Improve this analysis based on the feedback."),
            ("user", "Original: {draft}\n\nIssues: {issues}\n\nImproved version:"),
        ])
        | ChatOpenAI(model="gpt-4o")
        | StrOutputParser()
    )

    draft = await draft_chain.ainvoke({"query": query})
    quality = await quality_chain.ainvoke({"query": query, "draft": draft})

    if quality["score"] >= 7:
        return draft
    else:
        improved = await improve_chain.ainvoke({
            "draft": draft,
            "issues": "\n".join(quality["issues"]),
        })
        return improved

五、Map-Reduce 模式

5.1 文档摘要

from langchain_text_splitters import RecursiveCharacterTextSplitter

async def map_reduce_summary(document: str) -> str:
    """Summarize a long document using map-reduce."""
    # Split document into chunks
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=4000, chunk_overlap=200,
    )
    chunks = splitter.split_text(document)

    # Map: Summarize each chunk
    map_chain = (
        ChatPromptTemplate.from_messages([
            ("system", "Summarize this text section concisely. Keep key facts and numbers."),
            ("user", "{chunk}"),
        ])
        | ChatOpenAI(model="gpt-4o-mini")
        | StrOutputParser()
    )

    # Process chunks in parallel
    import asyncio
    summaries = await asyncio.gather(*[
        map_chain.ainvoke({"chunk": chunk}) for chunk in chunks
    ])

    # Reduce: Merge summaries
    reduce_chain = (
        ChatPromptTemplate.from_messages([
            ("system", """Merge these section summaries into one coherent summary.
Eliminate redundancy. Preserve key facts. Keep under 500 words."""),
            ("user", "Section summaries:\n\n{summaries}"),
        ])
        | ChatOpenAI(model="gpt-4o")
        | StrOutputParser()
    )

    combined = "\n\n---\n\n".join(
        f"Section {i+1}: {s}" for i, s in enumerate(summaries)
    )
    final = await reduce_chain.ainvoke({"summaries": combined})
    return final

六、错误传播与处理

6.1 错误处理策略

策略	描述	适用场景
Fail-fast	第一个错误即停止	高精度要求
Fallback	失败时用备选链	高可用要求
Retry	失败时重试	临时性错误
Skip	跳过失败步骤	可选步骤
Default	返回默认值	非关键步骤

6.2 LCEL 错误处理

from langchain_core.runnables import RunnableConfig

# Fallback chains
primary_chain = (
    prompt | ChatOpenAI(model="gpt-4o") | parser
)
fallback_chain = (
    prompt | ChatOpenAI(model="gpt-4o-mini") | parser
)

# Use with_fallbacks for automatic failover
robust_chain = primary_chain.with_fallbacks([fallback_chain])

# Use with_retry for transient errors
retrying_chain = primary_chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
    retry_if_exception_type=(TimeoutError, ConnectionError),
)

七、性能优化

7.1 优化策略

策略	效果	实现方式
并行执行	减少总延迟	RunnableParallel
流式输出	减少感知延迟	.astream()
模型分级	减少成本	简单步骤用小模型
缓存	减少重复调用	LangChain Cache
批处理	提高吞吐量	.abatch()

7.2 模型分级策略

# Use different models for different chain steps
CHAIN_CONFIG = {
    "classify": {"model": "gpt-4o-mini", "temp": 0, "reason": "Simple classification"},
    "retrieve": {"model": None, "reason": "No LLM needed, vector search"},
    "synthesize": {"model": "gpt-4o", "temp": 0.3, "reason": "Complex generation"},
    "validate": {"model": "gpt-4o-mini", "temp": 0, "reason": "Binary judgment"},
    "format": {"model": "gpt-4o-mini", "temp": 0, "reason": "Formatting only"},
}

# Cost estimate per 1000 queries:
# All gpt-4o:    $25 (5 calls x $5 each)
# Model grading:  $8 (1 x $5 + 4 x $0.75)
# Savings: 68%

八、总结

提示词链编排是将 LLM 从"单次调用"提升为"可组合系统"的关键技术。核心设计原则：

单一职责：每个链节点只做一件事
模型匹配：按复杂度选择模型，不要所有步骤都用最贵的
错误隔离：每个节点独立错误处理，不让一个失败拖垮整条链
可观测：每个节点记录输入/输出/延迟/成本
并行优先：能并行的步骤不要串行

Maurice | [email protected]

深度加工（NotebookLM 生成）

基于本文内容生成的 PPT 大纲、博客摘要、短视频脚本与 Deep Dive 播客，用于多场景复用

PPT 大纲（5-8 张幻灯片）点击展开

提示词链编排：复杂任务的分治策略 — ppt

基于上传的文章，我为您整理了一份关于**“提示词链编排”**的 PPT 大纲，共 7 张幻灯片。内容已使用 Markdown 格式输出，重点已加粗标记：

提示词链编排：复杂任务的分治策略

单提示词的局限性：单个提示词在处理复杂任务时，容易遇到上下文窗口不足、指令过于复杂导致模型遵循率下降的瓶颈 [1]。
分治思想：提示词链（Chain of Prompts）将复杂问题拆解为多个简单的子步骤，为每个子任务配备专门优化的提示词 [1]。
提升系统鲁棒性：相较于脆弱的单次大提示词调用，链式编排通过“分类-检索-综合-验证”的流水线，产出更加高效、准确和可靠 [1]。

常见的五种链式编排模式

顺序链（Sequential Chain）：适用于 A -> B -> C 的固定线性执行流程 [1]。
条件链（Conditional Chain）：先经过分类器进行判断，随后根据条件分支路由到不同专业的处理流程 [1]。
并行链（Parallel Chain）：并发处理独立的子任务（如同时提取实体、情感和主题），最后再将结果统一合并 [1-3]。
循环链（Self-Correcting Loop）：包含生成、检查与自我修复机制的反馈循环 [1, 4]。
Map-Reduce：将大任务（如长文档）拆分、分别并发处理后再合并汇总的批量模式 [1, 4]。

LangChain LCEL 架构与代码实现

管道式语法：LangChain 表达式语言 (LCEL) 使用管道符 (|) 将提示词 (Prompt)、大模型 (Model) 与解析器 (Parser) 无缝串联为可执行链 [4]。
实现动态路由：利用 RunnableLambda 函数根据分类结果，将请求精准分发到特定的专业链（如技术支持、账单处理或通用问答）[2, 5]。
实现并发分析：通过 RunnableParallel 工具，可以同时调度多个大模型任务，极大提升复杂信息提取的效率 [2]。

提升输出稳定性的自纠错循环

验证-修复机制：通过 Pydantic 等工具验证输出结构，一旦发生解析错误或字段缺失，立刻将错误信息反馈给模型进行循环修复（可设定最大重试次数）[3, 6]。
引入质量门禁：在最终输出前增加一个质量检查节点，可使用成本较低的小模型作为裁判，为草稿输出进行打分并指出问题 [6, 7]。
条件驱动改进：若裁判打分低于预设阈值（例如低于7分），则自动触发改进链，根据反馈意见优化并生成新版本 [6, 7]。

Map-Reduce 长文档处理模式

文档自动化切分：面对超长文档，首先使用文本切分器将其切分为带有重叠区（Overlap）的多个文本块，避免超出模型的上下文窗口 [7, 8]。
Map 并行摘要：为每个文本块并行调用摘要链（推荐使用小模型降低成本），提取关键事实与核心数据 [8]。
Reduce 总结归纳：使用推理能力更强的大模型，将所有局部摘要合并，消除冗余，输出最终的高质量连贯总结 [8]。

错误传播与自动处理策略

常用容错策略：根据具体场景可采用 Fail-fast（失败即停止）、Fallback（备选方案）、Retry（失败重试）、Skip（跳过）或 Default（返回默认值）[8, 9]。
自动故障转移：在 LCEL 中可使用 with_fallbacks 方法为主链路配置备用模型，保障高可用性 [9]。
应对瞬态错误：利用 with_retry 可以针对网络超时、连接断开等临时异常进行指数退避重试 [9]。

系统性能优化与核心设计原则

模型分级降低成本：根据节点任务复杂度匹配不同模型（如简单分类和格式化用小模型，复杂生成用大模型），估算可节省高达 68% 的成本 [9, 10]。
降低系统延迟：通过并行执行 (RunnableParallel) 降低绝对总延迟，或利用流式输出 (.astream()) 降低用户感知的等待时间 [9]。
五大核心设计原则：坚持节点单一职责、合理进行模型匹配、保障节点间的错误隔离、系统具备高可观测性，并且能并行不串行 [10]。

博客摘要 + 核心看点点击展开

提示词链编排：复杂任务的分治策略 — summary

SEO 友好博客摘要

面对大模型处理复杂任务时的性能瓶颈，单一提示词往往力不从心[1]。“提示词链（Chain-of-Prompts）”采用分治策略，将复杂任务拆解为易于管理的简单步骤[1]。本文深入解析了顺序链、条件路由、并行处理、自纠错循环及 Map-Reduce 五大核心编排模式[1, 2]，并结合 LangChain LCEL 框架提供了直接可用的代码实现[2-4]。此外，文章还详细拆解了错误容灾机制与“模型分级”性能优化策略[5-7]，助您以更低成本构建高可用、高鲁棒性的 LLM 组合系统。

核心看点