为什么需要链式调用
你大概遇到过这种情况：
任务：写一份"AI 在医疗领域应用"的调研报告
单次 Prompt：请写一份 5000 字的医疗 AI 调研报告，包含市场现状、
技术趋势、典型企业、风险挑战、未来展望...
结果：
✓ 开头很精彩
✗ 写到第 2000 字开始跑题
✗ 数据来源不明
✗ 企业案例张冠李戴
✗ 前后观点矛盾
问题不是 Prompt 不够好，而是任务太复杂。一次性输出 5000 字，模型需要同时做：调研、组织结构、填充内容、引用数据、保持一致性——任何一个环节出错，整篇都崩。
链式调用的核心思想：
把一个大任务拆成多个小任务，每个小任务只做一件事，前一步的输出作为后一步的输入。
单次调用： [大任务] → [乱七八糟的 5000 字]
链式调用： [拆解大纲] → [逐章调研] → [撰写正文] → [校验修正] → [整合成稿]
每步可控 每步有据 每步聚焦 每步查错 最终成稿
五个设计原则
1. 单一职责
每步只做一件事。反例：一步同时"调研 + 写作 + 校对"。正例：拆成三步。
✗ 一步：请调研 AI 医疗应用并写一篇 3000 字文章并校对错别字
✓ 三步：
Step 1：列出 AI 医疗的 5 个细分方向 + 每方向 3 个典型企业
Step 2：基于 Step 1 的结构，每方向写 600 字
Step 3：检查 Step 2 输出中的事实错误与错别字
2. 显式输入输出
每步必须明确：
- 输入：来自哪一步的什么字段（用 JSON 字段名或 Markdown 区块标识）
- 输出：固定格式（JSON Schema / Markdown 表格 / 纯文本），便于下一步解析
Step 2 输入：Step 1 输出的 directions 数组
Step 2 输出格式：
{
"sections": [
{ "name": "医学影像", "content": "..." },
...
]
}
3. 幂等可重试
任何一步可能失败（超时、格式错误、内容被拦截）。链路设计时必须考虑：
- 失败时能否单独重跑这一步？
- 重跑是否需要前面步骤重新执行？
- 输出是否落盘（便于断点续跑）？
✗ 全链路一气呵成，中间失败全部重来
✓ 每步输出存到文件 / 数据库，失败时从断点续跑
4. 校验关卡
关键步骤后加一个校验步骤，不通过则回退重做。
Step 2（生成）→ Step 2.5（校验）
- 校验字段是否齐全
- 校验字数是否达标
- 校验是否包含禁忌词
- 不通过 → 回退到 Step 2 重新生成（最多 3 次）
5. 渐进式精炼
不要指望一步到位。采用"粗稿 → 精修"模式：
Step 1：生成大纲（200 字）
Step 2：基于大纲写初稿（2000 字，允许粗糙）
Step 3：针对初稿做事实核查与补充数据
Step 4：润色语言风格
Step 5：最终校对
五种经典链路
链路 1：内容生产（最常用）
[选题分析] → [大纲生成] → [分段撰写] → [事实核查] → [润色整合]
↓ ↓ ↓ ↓ ↓
主题价值 结构清晰 内容充实 数据准确 风格统一
适用场景：博客、报告、白皮书、长文案。
链路 2：代码生成
[需求拆解] → [架构设计] → [接口定义] → [实现代码] → [测试用例] → [Review]
↓ ↓ ↓ ↓ ↓ ↓
任务清单 模块划分 API 契约 可运行代码 覆盖率>80% 质量分
适用场景：完整功能开发、API 设计、重构。
链路 3：数据分析
[问题理解] → [数据探索] → [SQL 生成] → [结果解读] → [报告输出]
↓ ↓ ↓ ↓ ↓
业务目标 字段与分布 可执行查询 洞察提炼 决策建议
适用场景：业务分析、BI 报表、数据洞察。
链路 4：多语言翻译
[原文解析] → [术语抽取] → [逐段翻译] → [一致性校验] → [母语润色]
↓ ↓ ↓ ↓ ↓
语意理解 术语表 忠实翻译 术语统一 地道表达
适用场景：文档国际化、产品出海（详见 AI 辅助内容国际化案例）。
链路 5：客服问答
[意图识别] → [知识检索] → [答案生成] → [安全过滤] → [回复用户]
↓ ↓ ↓ ↓ ↓
分类准确 RAG 召回 引用来源 合规检查 用户体验
适用场景：智能客服、企业知识库问答（基于 RAG 架构）。
代码实现
基础版：顺序链
import openai
client = openai.OpenAI()
def step(messages, step_name):
"""执行单步调用"""
print(f"[{step_name}] 开始...")
resp = client.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0.3,
)
output = resp.choices[0].message.content
print(f"[{step_name}] 完成，{len(output)} 字符")
return output
# Step 1：生成大纲
outline = step(
[{"role": "user", "content": "为'AI 医疗应用'写一份调研大纲，5 个章节，每章 3 个要点"}],
"大纲生成"
)
# Step 2：逐章撰写（输入 = Step 1 的大纲）
draft = step(
[{"role": "user", "content": f"基于以下大纲撰写正文，每章 500 字：\n\n{outline}"}],
"正文撰写"
)
# Step 3：事实核查
final = step(
[{"role": "user", "content": f"检查以下内容的事实错误并修正：\n\n{draft}"}],
"事实核查"
)
print(final)
进阶版：带校验与重试
import json
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=2, max=10))
def call_llm(messages, response_format=None):
"""带重试的 LLM 调用"""
kwargs = {"model": "gpt-4o", "messages": messages, "temperature": 0.2}
if response_format:
kwargs["response_format"] = {"type": "json_object"}
resp = client.chat.completions.create(**kwargs)
return resp.choices[0].message.content
def validate_json(text, required_fields):
"""校验输出是否为合法 JSON 且字段齐全"""
try:
data = json.loads(text)
for f in required_fields:
if f not in data:
raise ValueError(f"缺少字段: {f}")
return data
except json.JSONDecodeError:
raise ValueError("输出不是合法 JSON")
# Step 1：结构化大纲
outline_raw = call_llm([
{"role": "system", "content": "你是内容架构师。输出 JSON 格式。"},
{"role": "user", "content": '为"AI 医疗应用"生成大纲，JSON 格式：{"chapters": [{"title": "...", "points": ["..."]}]}'}
])
outline = validate_json(outline_raw, ["chapters"])
# 存盘以便断点续跑
with open("outline.json", "w") as f:
json.dump(outline, f, ensure_ascii=False, indent=2)
# Step 2：逐章生成（每章单独调用，互不影响）
sections = []
for i, ch in enumerate(outline["chapters"]):
section = call_llm([
{"role": "user", "content": f"为章节《{ch['title']}》写 500 字，要点：{','.join(ch['points'])}"}
])
sections.append({"title": ch["title"], "content": section})
# 每章落盘
with open(f"section_{i}.json", "w") as f:
json.dump({"title": ch["title"], "content": section}, f, ensure_ascii=False)
# Step 3：整合 + 润色
final = call_llm([
{"role": "user", "content": f"把以下各章节整合为连贯报告，润色过渡句：\n\n{json.dumps(sections, ensure_ascii=False)}"}
])
工程化：使用 LangChain LCEL
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnablePassthrough
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
# 定义每步
outline_chain = (
ChatPromptTemplate.from_template("为 {topic} 生成 5 章大纲，JSON: {{chapters: [...]}}")
| llm
| JsonOutputParser()
)
draft_chain = (
ChatPromptTemplate.from_template("基于大纲写正文，每章 500 字：\n{outline}")
| llm
| StrOutputParser()
)
polish_chain = (
ChatPromptTemplate.from_template("润色以下报告：\n{draft}")
| llm
| StrOutputParser()
)
# 串联成链
full_chain = (
{"topic": RunnablePassthrough()}
| outline_chain
| (lambda outline: {"outline": json.dumps(outline, ensure_ascii=False)})
| draft_chain
| (lambda draft: {"draft": draft})
| polish_chain
)
result = full_chain.invoke("AI 医疗应用")
常见坑点
坑 1：上下文丢失
Step 1 输出 2000 字 → 全文塞给 Step 2 → Step 2 只关注末尾，开头被忽略
解决：Step 2 的 Prompt 显式指明"请重点关注 Step 1 输出的第 X 部分"，或把上一步输出结构化为 JSON 字段。
坑 2：错误级联
Step 1 输出错别字 → Step 2 基于错别字生成 → Step 3 错上加错
解决：关键步骤后加校验关卡；用 temperature=0 降低随机性；对结构化输出做 schema 校验。
坑 3：成本爆炸
5 步链路 × 每步 4000 token 输入 = 20000 token / 次调用
单次任务 ¥0.5，日均 1000 次 = ¥500/天
解决：
- 简单步骤用小模型（如 GPT-4o-mini 处理格式转换）
- 用缓存（相同输入直接返回历史结果）
- 压缩上一步输出（只传必要字段，不传全文）
坑 4：链路过长
10 步链路 → 任何一步失败影响整体 → 调试困难
解决：链路控制在 3-5 步；每步输出落盘；用 trace ID 串联全链路日志。
坑 5：并行 vs 串行混淆
✗ 把本可并行的步骤串行执行（如 5 个章节逐章生成）→ 慢 5 倍
✗ 把必须串行的步骤并行（如未确定大纲就开始写正文）→ 输出对不上
解决：画依赖图，无依赖的步骤用 Promise.all / asyncio.gather 并行。
与其他技巧的关系
小结
链式调用不是"把一个 Prompt 拆成多个"那么简单，核心是任务工程：
- 想清楚任务由几个子任务组成
- 每个子任务的输入输出契约
- 哪些可以并行、哪些必须串行
- 哪里需要校验、哪里需要重试
掌握链式调用，你就从"写 Prompt"升级到了"设计 AI 工作流"，这是从 Prompt 爱好者到 AI 工程师的关键一跃。