跟我学 AI Agent : 第 3 课 — Prompt Chaining: 串联分解复杂任务

模块: 模块二: 核心工作流模式 | 难度: ⭐ | 预计时长: 1.5 小时参考: Agentic Design Patterns: A Hands-On Guide to Building Intelligent Systemspi-mono API: agent.prompt(), transformContext()

1. 开场引言

想象这个场景，公司新招了一个实习生。现在给他安排了一个任务，完成「分析市场报告、提取趋势数据、写邮件汇报」。由于没有经验，可能这几件事情他是放在一直完成的，没有清晰的流程。结果他写了一堆，但趋势数据遗漏了，邮件格式也不对。

现在，你将整个工作拆成了三步：先让他只做摘要，确认没问题后让他从摘要里找趋势，最后让他根据趋势写邮件。

这样，每一步都精准了。

这就是 Prompt Chaining 的核心思想：别让 LLM 一次做太多事，拆成链式步骤，前一步的输出是后一步的输入。

💡 核心问题: 为什么单次 Prompt 处理复杂任务容易失败？如何用链式分解提升可靠性？

2. 核心概念

2.1 单次 Prompt 的五大问题

这些是 AI 应用系统中最原始需要面对的问题。当想要在一个 Prompt 把几件事情都要讲明白时，可能就忽略的每件事情的不同点，也就丢失了重点！

2.2 Prompt Chaining 如何解决

核心思路：分而治之，步步为营。

单次 Prompt:  用户 ──[复杂指令]──→ LLM ──[一堆混乱的输出]──→ 结果Prompt Chaining:  用户 ──[步骤1]──→ LLM ──[输出1]──→ LLM ──[输出2]──→ LLM ──[最终结果]         摘要              趋势提取            邮件生成

每一步只做一件事，上下文干净，错误被隔离。

推荐再回去看看 Anthropic 的这篇论文，它可能是所有的起点，下面这张图给我留下了非常深刻的印象：

2.3 设计原则

我们在传统的软件工程中，也会使用类似的设计原则。比如，面向对象的设计方法中，通过封装的方式，将职责按对象进行划分，每个对象只完成自己相应的职责，通过对外的接口提供与外部沟通的渠道。这些原则并没有太大变化。

2.4 典型应用场景

1. 1. 信息处理流水线: URL → 提取文本 → 摘要 → 提取实体 → 查库 → 生成报告
2. 2. 复杂问答: 拆解子问题 → 分别检索 → 综合答案
3. 3. 数据抽取: OCR → 规范化 → 字段提取 → 格式校验 → 补漏
4. 4. 内容生成: 选题 → 大纲 → 分段写作 → 审校润色

3. 架构图解

关键：每一步的输出都经过结构化，下一步只看到精炼过的上下文。

4. 代码实战

4.1 最小版本: 手动链式调用

最直接的方式 —— 多次调用 agent.prompt()：

import { Agent } from"@mariozechner/pi-agent-core";import { getModel } from"@mariozechner/pi-ai";// 用同一个 Agent，逐步推进const agent = newAgent({initialState: {systemPrompt: "你是一个市场分析师。",model: getModel("minimax-cn", "MiniMax-M2.7"),  },});// Step 1: 摘要console.log("=== Step 1: 摘要 ===");const report = `2025年Q1电商报告：直播带货GMV同比增长47%，美妆品类占比32%，3C数码增长放缓仅5%，下沉市场用户增速是一线城市的3倍...`;await agent.prompt(`请将以下市场报告总结为3句话：\n${report}`);// Step 2: 改角色，提取趋势agent.state.systemPrompt = "你是一个行业趋势专家。只输出JSON。";console.log("\n=== Step 2: 趋势提取 ===");await agent.prompt("从上面的总结中提取3个关键趋势，输出JSON格式：\n" +'{"trends": [{"name": "趋势名", "evidence": "数据支撑"}]}');// Step 3: 改角色，写邮件agent.state.systemPrompt = "你是一个商务邮件撰写专家。风格简洁。";console.log("\n=== Step 3: 邮件 ===");await agent.prompt("根据上面的趋势分析，给市场团队写一封简短的趋势汇报邮件。");

问题: 这个版本太手动了 —— 每一步要改 systemPrompt，而且中间产物全堆在 messages 里，越来越臃肿。

4.2 增强版: 结构化传递 + transformContext

用 transformContext() 控制每一步 LLM 看到的上下文：

import { Agent } from"@mariozechner/pi-agent-core";import { getModel } from"@mariozechner/pi-ai";interfaceChainStep {role: string;prompt: string;outputKey: string;}asyncfunctionpromptChain(steps: ChainStep[],initialInput: string): Promise<Map<string, string>> {const results = newMap<string, string>();for (const step of steps) {const agent = newAgent({initialState: {systemPrompt: `你是${step.role}。只输出要求的内容，不要多余解释。`,model: getModel("minimax-cn", "MiniMax-M2.7"),messages: [],      },    });// 构建这一步的输入let input = step.prompt.replace("{input}", initialInput);// 把前面步骤的结果注入for (const [key, value] of results) {      input = input.replace(`{${key}}`, value);    }let output = "";    agent.subscribe((event) => {if (        event.type === "message_update" &&        event.assistantMessageEvent.type === "text_delta"      ) {        output += event.assistantMessageEvent.delta;      }    });await agent.prompt(input);    results.set(step.outputKey, output);console.log(`✅ ${step.role}: ${output.slice(0, 100)}...`);  }return results;}// 定义链式步骤const results = awaitpromptChain(  [    {role: "市场分析师",prompt: "将以下报告总结为3个要点：\n{input}",outputKey: "summary",    },    {role: "数据专家，只输出JSON",prompt: "从以下总结中提取关键趋势（JSON格式）：\n{summary}",outputKey: "trends",    },    {role: "商务邮件撰写专家",prompt:"根据以下趋势写一封简短的趋势汇报邮件：\n{trends}",outputKey: "email",    },  ],  report);console.log("\n📧 最终邮件:\n", results.get("email"));

关键改进:

• • 每步用独立 Agent，messages 不堆积
• • 步骤间通过 Map 传递结构化结果
• • 通过模板变量 {summary}, {trends} 控制上下文注入

4.3 生产级: 带校验和重试

interfaceChainStepValidatedextendsChainStep {validate?: (output: string) =>boolean;maxRetries?: number;}asyncfunctionpromptChainValidated(steps: ChainStepValidated[],initialInput: string): Promise<Map<string, string>> {const results = newMap<string, string>();for (let i = 0; i < steps.length; i++) {const step = steps[i];const maxRetries = step.maxRetries ?? 2;let output = "";let valid = false;for (let attempt = 0; attempt <= maxRetries; attempt++) {const agent = newAgent({initialState: {systemPrompt: `你是${step.role}。`,model: getModel("minimax-cn", "MiniMax-M2.7"),messages: [],        },      });let input = step.prompt.replace("{input}", initialInput);for (const [key, value] of results) {        input = input.replace(`{${key}}`, value);      }// 如果是重试，加入反馈if (attempt > 0) {        input += `\n\n注意：上次输出不符合要求，请重试。`;      }      agent.subscribe((event) => {if (          event.type === "message_update" &&          event.assistantMessageEvent.type === "text_delta"        ) {          output += event.assistantMessageEvent.delta;        }      });await agent.prompt(input);// 校验if (step.validate) {        valid = step.validate(output);if (!valid) {console.log(`⚠️ Step ${i + 1} 校验失败 (attempt ${attempt + 1})`          );          output = "";continue;        }      } else {        valid = true;      }if (valid) break;    }if (!valid && step.validate) {thrownewError(`Step ${i + 1} 在 ${maxRetries} 次重试后仍然校验失败`);    }    results.set(step.outputKey, output);console.log(`✅ Step ${i + 1} (${step.role}): 完成`);  }return results;}// 使用：带 JSON 校验const results = awaitpromptChainValidated(  [    {role: "市场分析师",prompt: "总结以下报告的3个要点：\n{input}",outputKey: "summary",    },    {role: "数据专家，只输出JSON",prompt: "提取趋势，JSON格式：\n{summary}",outputKey: "trends",validate: (output) => {try {const parsed = JSON.parse(output);returnArray.isArray(parsed.trends) && parsed.trends.length > 0;        } catch {returnfalse;        }      },maxRetries: 3,    },    {role: "邮件专家",prompt: "根据趋势写汇报邮件：\n{trends}",outputKey: "email",    },  ],  report);

5. 要点总结

6. 动手练习

练习 1: 基础 (⭐)

用 4.1 的手动方式实现一个三步链：翻译 → 润色 → 格式化。

练习 2: 进阶 (⭐⭐)

用 4.2 的 promptChain 函数实现一个「文章改写流水线」：

1. 1. 分析原文风格
2. 2. 改写为目标风格
3. 3. 校对语法错误

练习 3: 挑战 (⭐⭐⭐)

给 4.3 的版本添加并行分支支持：某些步骤可以同时执行（如同时提取「趋势」和「风险」），最后汇聚。提示：用 Promise.all()。

7. 延伸阅读

• • 🔗 pi-agent-core: https://github.com/badlogic/pi-mono/blob/main/packages/agent/README.md