Skip to Content

04c - Prompt 链与动态组装

本文是《AI Agent 实战手册》第 4 章第 3 节。 上一节：Metaprompt 完整工作流 | 下一节：Prompt 版本管理

概述

单个 prompt 能处理的任务复杂度有上限。当任务涉及多个步骤、条件分支或需要并行处理时，你需要将多个 prompt 串联成链（Prompt Chaining），或者根据运行时上下文动态组装 prompt。本节覆盖三种链式模式（顺序链、条件分支、并行执行）和动态 prompt 组装技术，配合架构图和代码示例。

1. Prompt Chaining 基础

1.1 什么是 Prompt Chaining

Prompt Chaining 是将复杂任务拆分为多个顺序执行的子 prompt，每个 prompt 处理一个子任务，其输出作为下一个 prompt 的输入。

┌──────────┐    输出 A    ┌──────────┐    输出 B    ┌──────────┐
│ Prompt 1 │ ──────────→ │ Prompt 2 │ ──────────→ │ Prompt 3 │
│ 理解需求  │             │ 生成方案  │             │ 审查优化  │
└──────────┘             └──────────┘             └──────────┘

1.2 为什么需要 Chaining

问题单 prompt 方案Chaining 方案
任务太复杂输出质量下降，容易遗漏每步聚焦，质量更高
需要中间验证无法在中途检查每步输出可检查和修正
不同步骤需要不同模型只能用一个模型每步可选最优模型
调试困难黑盒，不知道哪里出错每步可独立调试
上下文窗口不够所有信息挤在一起每步只传递必要信息

2. 顺序链（Sequential Chain）

2.1 模式说明

最基本的 chaining 模式：Prompt A → Prompt B → Prompt C，严格按顺序执行。

2.2 架构图

输入
 │
 ▼
┌─────────────────┐
│  Prompt 1: 分析  │  "分析这段代码的问题"
│  输入: 原始代码   │
│  输出: 问题列表   │
└────────┬────────┘
         │ 问题列表
         ▼
┌─────────────────┐
│  Prompt 2: 方案  │  "为每个问题提供修复方案"
│  输入: 问题列表   │
│  输出: 修复方案   │
└────────┬────────┘
         │ 修复方案
         ▼
┌─────────────────┐
│  Prompt 3: 执行  │  "应用修复方案，输出修改后的代码"
│  输入: 原始代码   │
│      + 修复方案   │
│  输出: 修复后代码  │
└────────┬────────┘
         │
         ▼
       最终输出

2.3 代码示例（TypeScript）

import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
async function sequentialChain(code: string): Promise<string> {
  // Step 1: 分析问题
  const analysis = await client.messages.create({
    model: "claude-sonnet-4-20250514",
    max_tokens: 1024,
    messages: [{
      role: "user",
      content: `分析以下代码的问题，列出每个问题及其严重性：\n\n${code}`
    }]
  });
  const issues = analysis.content[0].type === "text"
    ? analysis.content[0].text : "";
 
  // Step 2: 生成修复方案
  const plan = await client.messages.create({
    model: "claude-sonnet-4-20250514",
    max_tokens: 1024,
    messages: [{
      role: "user",
      content: `针对以下问题列表，为每个问题提供具体的修复方案：\n\n${issues}`
    }]
  });
  const fixes = plan.content[0].type === "text"
    ? plan.content[0].text : "";
 
  // Step 3: 应用修复
  const result = await client.messages.create({
    model: "claude-sonnet-4-20250514",
    max_tokens: 2048,
    messages: [{
      role: "user",
      content: `将以下修复方案应用到代码中，输出完整的修复后代码：
 
原始代码：
${code}
 
修复方案：
${fixes}`
    }]
  });
 
  return result.content[0].type === "text"
    ? result.content[0].text : "";
}

2.4 代码示例（Python）

import anthropic
 
client = anthropic.Anthropic()
 
def sequential_chain(code: str) -> str:
    # Step 1: 分析问题
    analysis = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"分析以下代码的问题，列出每个问题及其严重性：\n\n{code}"
        }]
    )
    issues = analysis.content[0].text
 
    # Step 2: 生成修复方案
    plan = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"针对以下问题列表，为每个问题提供具体的修复方案：\n\n{issues}"
        }]
    )
    fixes = plan.content[0].text
 
    # Step 3: 应用修复
    result = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"""将以下修复方案应用到代码中，输出完整的修复后代码：
 
原始代码：
{code}
 
修复方案：
{fixes}"""
        }]
    )
    return result.content[0].text

3. 条件分支（Conditional Branching）

3.1 模式说明

根据前一步的输出结果，选择不同的后续 prompt 路径。

3.2 架构图

输入
 │
 ▼
┌──────────────────┐
│  Prompt 1: 分类   │  "判断这个请求的类型"
│  输出: 类型标签    │
└────────┬─────────┘
         │
    ┌────┼────┐
    │    │    │
    ▼    ▼    ▼
  Bug  Feature  Question
    │    │    │
    ▼    ▼    ▼
┌──────┐ ┌──────┐ ┌──────┐
│修复链 │ │设计链 │ │问答链 │
└──────┘ └──────┘ └──────┘

3.3 代码示例（TypeScript）

type RequestType = "bug" | "feature" | "question" | "unknown";
 
async function conditionalChain(userRequest: string): Promise<string> {
  // Step 1: 分类
  const classification = await client.messages.create({
    model: "claude-sonnet-4-20250514",
    max_tokens: 50,
    messages: [{
      role: "user",
      content: `将以下请求分类为 bug、feature 或 question，只输出类别名：\n\n${userRequest}`
    }]
  });
  const type = (classification.content[0].type === "text"
    ? classification.content[0].text.trim().toLowerCase()
    : "unknown") as RequestType;
 
  // Step 2: 根据类型选择不同的处理链
  const prompts: Record<RequestType, string> = {
    bug: `你是一个 bug 修复专家。分析以下 bug 报告，提供：
1. 可能的根因
2. 复现步骤
3. 修复建议
 
Bug 报告：${userRequest}`,
 
    feature: `你是一个产品设计师。分析以下功能需求，提供：
1. 用户故事
2. 验收标准
3. 技术可行性评估
 
功能需求：${userRequest}`,
 
    question: `你是一个技术顾问。回答以下技术问题：
1. 直接回答
2. 相关背景知识
3. 推荐资源
 
问题：${userRequest}`,
 
    unknown: `请帮助处理以下请求：\n\n${userRequest}`
  };
 
  const result = await client.messages.create({
    model: "claude-sonnet-4-20250514",
    max_tokens: 2048,
    messages: [{ role: "user", content: prompts[type] }]
  });
 
  return result.content[0].type === "text"
    ? result.content[0].text : "";
}

4. 并行执行（Parallel Execution）

4.1 模式说明

将独立的子任务分发给多个 prompt 并行处理，最后汇总结果。适合子任务之间没有依赖关系的场景。

4.2 架构图

              输入
               │
      ┌────────┼────────┐
      │        │        │
      ▼        ▼        ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Prompt A │ │ Prompt B │ │ Prompt C │
│ 安全审查  │ │ 性能分析  │ │ 可读性   │
└────┬─────┘ └────┬─────┘ └────┬─────┘
     │            │            │
     └────────┬───┘────────────┘
              │
              ▼
      ┌──────────────┐
      │  汇总 Prompt  │  合并三份报告
      └──────────────┘
              │
              ▼
           最终报告

4.3 代码示例（TypeScript）

async function parallelChain(code: string): Promise<string> {
  // 并行执行三个独立的分析任务
  const [security, performance, readability] = await Promise.all([
    client.messages.create({
      model: "claude-sonnet-4-20250514",
      max_tokens: 1024,
      messages: [{
        role: "user",
        content: `从安全角度审查以下代码，列出潜在漏洞：\n\n${code}`
      }]
    }),
    client.messages.create({
      model: "claude-sonnet-4-20250514",
      max_tokens: 1024,
      messages: [{
        role: "user",
        content: `从性能角度分析以下代码，列出优化建议：\n\n${code}`
      }]
    }),
    client.messages.create({
      model: "claude-sonnet-4-20250514",
      max_tokens: 1024,
      messages: [{
        role: "user",
        content: `从可读性角度评估以下代码，列出改进建议：\n\n${code}`
      }]
    })
  ]);
 
  const securityReport = security.content[0].type === "text"
    ? security.content[0].text : "";
  const perfReport = performance.content[0].type === "text"
    ? performance.content[0].text : "";
  const readReport = readability.content[0].type === "text"
    ? readability.content[0].text : "";
 
  // 汇总三份报告
  const summary = await client.messages.create({
    model: "claude-sonnet-4-20250514",
    max_tokens: 2048,
    messages: [{
      role: "user",
      content: `将以下三份代码审查报告合并为一份综合报告，按优先级排序：
 
## 安全审查
${securityReport}
 
## 性能分析
${perfReport}
 
## 可读性评估
${readReport}`
    }]
  });
 
  return summary.content[0].type === "text"
    ? summary.content[0].text : "";
}

4.4 代码示例（Python）

import asyncio
import anthropic
 
async_client = anthropic.AsyncAnthropic()
 
async def parallel_chain(code: str) -> str:
    # 并行执行三个独立分析
    security_task = async_client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"从安全角度审查以下代码，列出潜在漏洞：\n\n{code}"
        }]
    )
    perf_task = async_client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"从性能角度分析以下代码，列出优化建议：\n\n{code}"
        }]
    )
    read_task = async_client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"从可读性角度评估以下代码，列出改进建议：\n\n{code}"
        }]
    )
 
    security, perf, readability = await asyncio.gather(
        security_task, perf_task, read_task
    )
 
    # 汇总
    summary = await async_client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"""将以下三份代码审查报告合并为一份综合报告，按优先级排序：
 
## 安全审查
{security.content[0].text}
 
## 性能分析
{perf.content[0].text}
 
## 可读性评估
{readability.content[0].text}"""
        }]
    )
    return summary.content[0].text

5. 混合模式

实际项目中，三种模式经常组合使用：

输入
 │
 ▼
┌──────────┐
│ 分类器    │  ← 条件分支入口
└────┬─────┘
     │
  ┌──┴──┐
  ▼     ▼
简单    复杂
  │     │
  ▼     ├──────────┐
单步   并行分析     │
处理    │          │
  │    ┌┴┐        │
  │    ▼ ▼        │
  │   A  B        │
  │    │ │        │
  │    └┬┘        │
  │     ▼         │
  │   汇总  ──→ 顺序优化
  │     │
  └──┬──┘
     ▼
   输出

6. 动态 Prompt 组装

6.1 什么是动态组装

动态 Prompt 组装是在运行时根据上下文条件拼接 prompt 的各个部分，而不是使用固定的 prompt 模板。

6.2 组装架构

┌─────────────────────────────────────────┐
│              Prompt 组装器               │
│                                         │
│  ┌──────────┐  ┌──────────┐  ┌────────┐ │
│  │ 角色定义  │  │ 任务指令  │  │ 约束   │ │
│  │ (固定)    │  │ (动态)    │  │ (条件) │ │
│  └────┬─────┘  └────┬─────┘  └───┬────┘ │
│       │             │            │      │
│  ┌────┴─────┐  ┌────┴─────┐  ┌──┴────┐ │
│  │ 上下文    │  │ 用户输入  │  │ 检索   │ │
│  │ (运行时)  │  │ (实时)    │  │ (RAG)  │ │
│  └────┬─────┘  └────┬─────┘  └──┬────┘ │
│       └──────────┬──────────────┘      │
│                  ▼                      │
│          ┌──────────────┐               │
│          │  最终 Prompt  │               │
│          └──────────────┘               │
└─────────────────────────────────────────┘

6.3 代码示例：Prompt Builder 模式（TypeScript）

interface PromptConfig {
  role: string;
  task: string;
  language?: string;
  outputFormat?: "json" | "markdown" | "xml";
  constraints?: string[];
  examples?: Array<{ input: string; output: string }>;
  context?: string;
}
 
function buildPrompt(config: PromptConfig): string {
  const parts: string[] = [];
 
  // 1. 角色定义（始终包含）
  parts.push(`你是${config.role}。`);
 
  // 2. 上下文（如果有）
  if (config.context) {
    parts.push(`\n## 背景信息\n${config.context}`);
  }
 
  // 3. 任务指令（始终包含）
  parts.push(`\n## 任务\n${config.task}`);
 
  // 4. 输出格式（条件包含）
  if (config.outputFormat) {
    const formatInstructions: Record<string, string> = {
      json: "请以 JSON 格式输出，确保格式有效。",
      markdown: "请以 Markdown 格式输出，使用适当的标题和列表。",
      xml: "请以 XML 格式输出，使用语义化的标签名。"
    };
    parts.push(`\n## 输出格式\n${formatInstructions[config.outputFormat]}`);
  }
 
  // 5. 语言约束（条件包含）
  if (config.language) {
    parts.push(`\n请使用 ${config.language} 回答。`);
  }
 
  // 6. 额外约束（条件包含）
  if (config.constraints?.length) {
    parts.push("\n## 约束条件");
    config.constraints.forEach((c, i) => {
      parts.push(`${i + 1}. ${c}`);
    });
  }
 
  // 7. 示例（条件包含，few-shot）
  if (config.examples?.length) {
    parts.push("\n## 示例");
    config.examples.forEach((ex, i) => {
      parts.push(`\n### 示例 ${i + 1}`);
      parts.push(`输入：${ex.input}`);
      parts.push(`输出：${ex.output}`);
    });
  }
 
  return parts.join("\n");
}
 
// 使用示例
const prompt = buildPrompt({
  role: "资深 Python 开发者",
  task: "审查以下代码的安全性",
  outputFormat: "markdown",
  language: "中文",
  constraints: [
    "聚焦 OWASP Top 10 漏洞",
    "每个问题提供修复代码",
    "按严重性排序"
  ],
  context: "这是一个 Flask Web 应用的用户认证模块"
});

6.4 运行时上下文注入

动态组装的核心价值在于运行时注入上下文：

async function dynamicPromptWithRAG(
  userQuery: string,
  userId: string
): Promise<string> {
  // 1. 获取用户上下文
  const userProfile = await getUserProfile(userId);
 
  // 2. 检索相关文档（RAG）
  const relevantDocs = await vectorSearch(userQuery, { limit: 3 });
 
  // 3. 获取对话历史
  const history = await getRecentHistory(userId, { limit: 5 });
 
  // 4. 动态组装 prompt
  const prompt = buildPrompt({
    role: "技术支持助手",
    task: userQuery,
    context: `
用户信息：${userProfile.plan} 用户，使用 ${userProfile.language}
相关文档：
${relevantDocs.map(d => `- ${d.title}: ${d.snippet}`).join("\n")}
最近对话摘要：
${history.map(h => `- ${h.summary}`).join("\n")}
    `.trim(),
    constraints: [
      `使用 ${userProfile.language} 回答`,
      userProfile.plan === "free"
        ? "不要推荐付费功能"
        : "可以推荐高级功能"
    ]
  });
 
  return prompt;
}

7. Chaining 模式选择指南

工具推荐

工具用途价格适用场景
LangChainPrompt chain 编排框架免费开源Python/JS 项目，快速原型
LangGraph有状态的 agent 工作流免费开源复杂条件分支和循环
Anthropic SDK直接 API 调用按 token 计费简单链，不需要框架
Haystack模块化 AI 管线免费开源搜索和 RAG 管线
Semantic Kernel微软 AI 编排框架免费开源.NET/Python 企业项目
DSPy声明式 prompt 管线免费开源需要自动优化的管线

决策矩阵

场景推荐模式原因
步骤间有依赖顺序链后续步骤需要前步输出
需要根据结果走不同路径条件分支不同类型需要不同处理
多个独立分析维度并行执行无依赖，可加速
复杂多步 + 条件混合模式灵活组合
上下文因用户/时间变化动态组装运行时适配
需要自我修正顺序链 + 反馈循环输出审查后重试

实战案例：构建智能客服分流系统

场景

一个 SaaS 产品的客服系统，需要自动分类用户请求并路由到合适的处理流程。

架构

用户消息
    │
    ▼
┌──────────────┐
│ 意图分类器    │  → 分类为：技术问题 / 账单问题 / 功能请求 / 投诉
└──────┬───────┘
       │
  ┌────┼────────┬──────────┐
  ▼    ▼        ▼          ▼
技术   账单     功能请求    投诉
  │    │        │          │
  ▼    ▼        ▼          ▼
┌────┐ ┌────┐  ┌────┐    ┌────┐
│RAG │ │查询 │  │记录 │    │升级 │
│知识 │ │账单 │  │需求 │    │人工 │
│库  │ │系统 │  │系统 │    │    │
└──┬─┘ └──┬─┘  └──┬─┘    └──┬─┘
   │      │       │         │
   ▼      ▼       ▼         ▼
 回复    回复    确认邮件   通知团队

关键代码片段

async function customerServiceRouter(message: string, userId: string) {
  // Step 1: 意图分类（条件分支入口）
  const intent = await classify(message);
 
  // Step 2: 动态组装上下文
  const userContext = await getUserContext(userId);
 
  // Step 3: 根据意图路由
  switch (intent) {
    case "technical":
      // 并行：RAG 检索 + 历史工单查询
      const [docs, tickets] = await Promise.all([
        searchKnowledgeBase(message),
        getRelatedTickets(userId, message)
      ]);
      // 顺序：生成回复 → 质量检查
      const draft = await generateTechResponse(message, docs, tickets, userContext);
      return await qualityCheck(draft);
 
    case "billing":
      const billingInfo = await getBillingInfo(userId);
      return await generateBillingResponse(message, billingInfo);
 
    case "feature_request":
      await logFeatureRequest(message, userId);
      return await generateFeatureAck(message);
 
    case "complaint":
      await escalateToHuman(message, userId);
      return "您的反馈已转交专人处理，我们会在 24 小时内回复您。";
  }
}

案例分析

这个案例组合了三种模式：

  • 条件分支：根据意图分类路由到不同处理链
  • 并行执行：技术问题同时检索知识库和历史工单
  • 顺序链：生成回复后进行质量检查
  • 动态组装：根据用户信息和检索结果组装 prompt

避坑指南

❌ 常见错误

  1. 链太长导致信息丢失

    • 问题：经过 5-6 步传递后，原始信息被逐步稀释
    • 正确做法：每步传递时保留关键原始信息，链长度控制在 3-4 步

  2. 不做中间结果验证

    • 问题：第一步输出错误，后续步骤在错误基础上继续
    • 正确做法：在关键节点添加验证步骤，检查输出格式和内容合理性

  3. 并行任务有隐含依赖

    • 问题：看似独立的任务实际上有顺序依赖，并行执行导致结果不一致
    • 正确做法：仔细分析任务依赖关系，只有真正独立的任务才并行

  4. 动态组装的 prompt 过长

    • 问题：注入太多上下文导致 prompt 超出窗口或注意力稀释
    • 正确做法：对注入的上下文做摘要和截断，只保留最相关的信息

  5. 没有错误处理和降级

    • 问题：链中某一步失败，整个流程崩溃
    • 正确做法：每步添加 try-catch，失败时有降级方案（如跳过可选步骤）

✅ 最佳实践

  1. 每个 prompt 只做一件事，保持单一职责
  2. 在链的关键节点记录日志，方便调试
  3. 为每步设置超时和重试机制
  4. 使用结构化输出（JSON/XML）在步骤间传递数据，避免解析歧义
  5. 先用简单的顺序链验证可行性，再逐步引入并行和条件分支

相关资源与延伸阅读

编排框架

  • LangChain  — 最流行的 LLM 应用框架，内置 Prompt 链编排和条件路由能力
  • LangGraph  — LangChain 的图编排扩展，支持复杂的 Prompt 链拓扑和状态管理
  • CrewAI  — 多 Agent 编排框架，支持 Prompt 链在多个 Agent 间的协作执行

可视化工具

  • ChainForge  — 开源的可视化 Prompt 链实验工具，支持拖拽式构建和多模型对比
  • Flowise  — 低代码 LLM 应用构建工具，可视化编排 Prompt 链

学习资源

监控与调试

  • LangSmith  — Prompt 链执行的可视化追踪和调试平台
  • Langfuse  — 开源替代方案，追踪 Prompt 链的每个步骤和 Token 消耗

参考来源

📖 返回 总览与导航 | 上一节：Metaprompt 完整工作流 | 下一节：Prompt 版本管理

Last updated on
从 Copilot 到 Agent