链式提示——把复杂任务拆成多步对话

一、一个问题引发的血案

假设你让大模型做这件事：

你把300条评论一股脑塞进去，prompt写了800字。结果呢？模型经常漏掉第三条痛点；JSON的key有时是中文有时是英文；字数限制形同虚设。

这不是模型不行——是你的prompt在逼模型“一心多用”。

Ja va开发者都懂一个道理：一个方法如果超过50行还在干三件事，code review绝对过不了。你不会写一个 parseValidatePersistAndNotify() 方法，为什么要求模型在一个token窗口里同时做提取、排序、总结、格式化？

这正是**链式提示（Chain Prompting）**要解决的问题。

二、Ja va类比：从上帝方法到责任链

2.1 反模式：上帝Prompt

// 你绝不会这么写Ja va代码 public String processReviews(String allReviews) { // 提取痛点 List<String> painPoints = extractPainPoints(allReviews); // 排序 painPoints.sort(bySeverity()); // 总结 String summary = summarize(painPoints); // 格式化JSON return toJson(summary); }

但如果这四个操作全挤在一个方法里、没有任何分界、还期望返回值严格符合规范——这就是大多数“超级prompt”的真实写照。

2.2 正解：链式调用

// 这才是工程化的写法 public ReportJson processReviews(String allReviews) { List<PainPoint> points = painPointExtractor.extract(allReviews); List<PainPoint> sorted = severitySorter.sort(points); String summary = reportSummarizer.summarize(sorted, 200); return jsonFormatter.format(summary); }

链式提示干的就是这件事：把一次巨型LLM调用拆成多次小型调用，每次只让模型干一件事，前一步的输出是后一步的输入。

三、链式提示的三个核心原则

原则1：单一职责

每一步prompt只描述一个任务。不要出现“请提取并总结”——那是两步。

错误示例：

请阅读评论，提取痛点并按严重程度排序，然后写总结。

正确示例（第一步）：

以下是一批产品评论。请提取每个用户提到的具体痛点。只输出痛点列表，不要排序，不要总结。 格式：每行一个痛点，格式为 "- [严重程度:高/中/低] 痛点描述"

原则2：输出契约

每一步的输出格式必须在prompt中显式约定。因为下一步的prompt会直接引用上一步的输出，格式不一致会导致链断裂。

技巧：让模型输出markdown结构化内容（标题、列表、代码块），比JSON更容错。除非最后一步确实需要JSON，否则中间步骤用markdown。

原则3：上下文传递

把上一步输出原样注入下一步prompt。不要手动编辑——你编辑了就引入了新的变量，出了问题没法定位是哪个环节的锅。

四、完整代码实现

下面是一个完整的链式提示实现。场景：分析用户对一款SaaS产品的反馈评论，输出结构化分析报告。

"""链式提示完整示例 —— 评论分析流水线 依赖: pip install openai""" import json import os from openai import OpenAI client = OpenAI( api_key=os.getenv("DASHSCOPE_API_KEY"), base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", ) MODEL = "qwen-plus" def call_llm(system_prompt: str, user_prompt: str, temperature: float = 0.3) -> str: """单步LLM调用封装。相当于Ja va里的RestTemplate.postForObject()""" response = client.chat.completions.create( model=MODEL, temperature=temperature, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], ) return response.choices[0].message.content # ---- 步骤1: 提取痛点 ---- def step1_extract_pain_points(reviews: str) -> str: """类比Ja va:PainPointExtractor.extract(String reviews) -> List 单一职责：只提取，不排序不总结""" system = """你是一个用户反馈分析师。你的唯一职责是从评论中提取用户提到的痛点。 输出规则： - 每行一个痛点，格式 "- [严重程度:高/中/低] 痛点描述" - 严重程度标准：高=用户表示要弃用/退款，中=功能缺失/体验差，低=小建议/优化需求 - 不要输出任何其他内容，不要总结，不要排序""" user = f"以下是一批用户评论，请提取所有痛点： {reviews}" return call_llm(system, user, temperature=0.1) # ---- 步骤2: 按严重程度排序 ---- def step2_sort_by_severity(pain_points_raw: str) -> str: """类比Ja va:SeveritySorter.sort(List) -> List 输入来自步骤1的输出，不做任何修改直接传入""" system = """你负责将痛点列表按严重程度排序。 规则： - 高严重度排最前面，中其次，低最后 - 同一严重度内，按原文顺序保留 - 输出格式与输入格式完全相同，只改变顺序 - 不要添加任何解释""" user = f"请将以下痛点按严重程度排序（高>中>低），保持输出格式不变： {pain_points_raw}" return call_llm(system, user, temperature=0.0) # ---- 步骤3: 生成中文总结 ---- def step3_summarize(sorted_pain_points: str, max_chars: int = 200) -> str: """类比Ja va:ReportSummarizer.summarize(List, int maxChars) -> String""" system = f"""你是一个报告撰写专家。请根据以下痛点分析结果，生成一段不超过{max_chars}字的中文总结。 要求： - 先概括最严重的问题（如果存在高严重度痛点） - 再提中等痛点中的共性问题 - 最后简要提及改进建议方向 - 语言简洁，不废话""" user = f"痛点分析结果： {sorted_pain_points} 请生成不超过{max_chars}字的总结。" return call_llm(system, user, temperature=0.5) # ---- 步骤4: 格式化输出 ---- def step4_format_output( original_reviews: str, pain_points: str, summary: str, ) -> dict: """类比Ja va:ReportFormatter.format(...) -> ReportJson 最后一步才做JSON格式化，由代码主导而非模型主导""" return { "report_time": "2026-05-23", "total_reviews": len([r for r in original_reviews.split(" ") if r.strip()]), "pain_points": pain_points, "summary": summary, "metadata": { "pipeline": "chain_prompting_v1", "steps": ["extract", "sort", "summarize", "format"], }, } # ---- 主流水线 ---- def run_analysis_pipeline(reviews: str) -> dict: """编排整个链。类比Ja va的@Service主方法""" print("[Step 1/4] 提取痛点...") raw_pain_points = step1_extract_pain_points(reviews) print(f"提取结果: {raw_pain_points[:200]}... ") print("[Step 2/4] 按严重程度排序...") sorted_points = step2_sort_by_severity(raw_pain_points) print(f"排序结果: {sorted_points[:200]}... ") print("[Step 3/4] 生成总结...") summary = step3_summarize(sorted_points, max_chars=200) print(f"总结: {summary} ") print("[Step 4/4] 格式化输出...") result = step4_format_output(reviews, sorted_points, summary) return result if __name__ == "__main__": # 模拟用户评论 sample_reviews = """界面太卡了，每次点按钮要等3秒，严重影响效率。 数据导出功能到现在都没有，每次都要手动复制粘贴，太麻烦了。 功能挺全的但我需要批量导入Excel，现有的CSV导入经常乱码。 如果你们下个月还不支持批量导出我就换别家的产品了。 暗色模式什么时候上线？ 晚上用太刺眼了。 客服响应倒是挺快的，点赞。 报告功能生成的图表配色太丑了，能不能换个设计师。 API文档全是英文的，国内用户看着费劲。 价格涨了30%但是功能没什么变化，性价比越来越低。 移动端适配基本没有，手机上根本没法用。""" report = run_analysis_pipeline(sample_reviews) print("=" * 50) print("最终报告:") print(json.dumps(report, ensure_ascii=False, indent=2))

运行结果

[Step 1/4] 提取痛点... 提取结果: - [严重程度:高] 界面太卡，每次点按钮要等3秒，严重影响效率 - [严重程度:高] 如果下个月还不支持批量导出就换产品 - [严重程度:中] 缺少数据导出功能 - [严重程度:中] CSV导入经常乱码 - [严重程度:中] API文档全是英文 ... [Step 2/4] 按严重程度排序... 排序结果: - [严重程度:高] 界面太卡，每次点按钮要等3秒，严重影响效率 - [严重程度:高] 如果下个月还不支持批量导出就换产品 - [严重程度:中] 缺少数据导出功能 - [严重程度:中] CSV导入经常乱码 ... [Step 3/4] 生成总结... 总结: 用户最不满的是性能卡顿和关键功能缺失，已有流失风险。中等痛点集中在数据导入导出不完善和本地化不足。建议优先优化响应速度并尽快上线批量导出，同时补充中文文档和移动端适配。 [Step 4/4] 格式化输出...

五、什么时候该用链式提示

并非所有场景都需要链。判断标准很简单：

场景 | 是否用链 | 原因 ------ | --------- | ------ 简单问答(“什么是REST”) | ❌ 不用 | 一步就能回答好 翻译+润色 | ⚠️ 可选 | 两步可以提升质量，但一步也够用 提取+分类+排序+总结+格式化 | ✅ 必须用 | 多维度任务，单prompt必丢信息 需要中间结果验证的场景 | ✅ 必须用 | 链式天然支持断点检查 不同步骤需要不同temperature | ✅ 必须用 | 提取要低温(0.1)，总结要中温(0.5)

关键判断： 问自己“我写Ja va会把这些逻辑放在一个方法里吗？”如果答案是不会——就该拆。

六、进阶技巧

6.1 分支与条件

链不一定是线性的。你可以根据中间结果决定下一步走向，就像if-else：

pain_points = step1_extract_pain_points(reviews) high_severity_count = pain_points.count("[严重程度:高]") if high_severity_count >= 3: # 走危机处理分支 result = crisis_analysis(pain_points) else: # 走常规分析分支 result = routine_analysis(pain_points)

6.2 并行步骤

相互独立的步骤可以并行执行，缩短总耗时：

import concurrent.futures with concurrent.futures.ThreadPoolExecutor() as executor: future_sentiment = executor.submit(analyze_sentiment, reviews) future_pain_points = executor.submit(extract_pain_points, reviews) future_keywords = executor.submit(extract_keywords, reviews) sentiment = future_sentiment.result() pain_points = future_pain_points.result() keywords = future_keywords.result()

6.3 重试与降级

链式调用中任何一步都可能失败（API超时、输出格式异常）。需要加防护：

def step_with_retry(step_fn, input_data, max_retries=2): """类比Ja va的@Retryable注解""" for attempt in range(max_retries + 1): try: result = step_fn(input_data) if result and len(result) > 10: # 基本校验 return result except Exception as e: if attempt == max_retries: raise print(f"步骤失败，重试 {attempt + 1}/{max_retries}: {e}") raise RuntimeError("所有重试均失败")

七、常见坑与解法

坑1：中间输出格式漂移

模型偶尔会“自作主张”在输出里加一段解释性文字，破坏格式约定。

解法： temperature设为0或0.1，system prompt里加上“不要添加任何解释，只输出指定格式的内容”。

坑2：错误传播放大

步骤2基于步骤1的输出，如果步骤1就错了，后续全错。

解法： 关键链路的步骤1加人工审核断点，或者设一个格式校验器，不符合预期就重试。

坑3：延迟叠加

4步链式调用 = 4次API往返，总延迟是单次调用的4倍。

解法： 用6.2的并行化减少延迟；非关键步骤用更快的模型（如qwen-turbo替代qwen-plus）。

八、总结

链式提示的本质是用工程化的思维组织提示词。你把一个复杂任务拆成单一职责的小步骤，每步独立测试、独立优化、出错独立重试——这和写Ja va代码时拆Service、拆方法的思路完全一致。

核心原则就三个：

1. 单一职责——每步只干一件事
2. 输出契约——每步约定好输出格式
3. 上下文传递——上一步输出原样传给下一步