链式调用原理

单次 LLM 调用的能力是有限的。链式调用 (Prompt Chaining) 将复杂任务分解为多个步骤，每步的输出作为下一步的输入，形成流水线：

[用户输入] → [Step 1: 理解意图] → [Step 2: 检索知识] → [Step 3: 生成答案] → [最终输出]

核心假设：分解后的子任务更简单、更可靠，错误隔离更容易。

链式架构模式

1. 顺序链 (Sequential Chain)

最简单的模式——线性传递：

from typing import Dict, Any
import openai

def llm_call(prompt: str, model: str = "gpt-4") -> str:
    resp = openai.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
    )
    return resp.choices[0].message.content

class SequentialChain:
    def __init__(self):
        self.steps = []
    
    def add_step(self, name: str, prompt_template: str):
        """prompt_template 中用 {prev_output} 引用上一步输出"""
        self.steps.append({"name": name, "template": prompt_template})
        return self
    
    def run(self, initial_input: str) -> Dict[str, str]:
        results = {}
        current_output = initial_input
        for step in self.steps:
            prompt = step["template"].format(prev_output=current_output)
            current_output = llm_call(prompt)
            results[step["name"]] = current_output
            print(f"[{step['name']}] 完成")
        results["final"] = current_output
        return results

# 示例：技术文档翻译流水线
chain = SequentialChain()
chain.add_step("extract_keypoints", 
    "分析以下技术文档，提取关键技术概念和术语（JSON数组格式）：\n{prev_output}")
chain.add_step("translate",
    "将以下内容翻译为中文，保持技术术语准确：\n{prev_output}")
chain.add_step("review",
    "审校以下翻译，修正不准确之处，输出最终版本：\n{prev_output}")

2. 并行链 (Parallel Chain)

多个独立子任务并行执行后汇总：

import asyncio
import aiohttp

async def async_llm_call(prompt: str, model: str = "gpt-4") -> str:
    async with aiohttp.ClientSession() as session:
        # 实际使用 openai async client
        resp = await openai.AsyncOpenAI().chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
        )
        return resp.choices[0].message.content

class ParallelChain:
    def __init__(self):
        self.branches = []
    
    def add_branch(self, name: str, prompt_template: str):
        self.branches.append({"name": name, "template": prompt_template})
        return self
    
    async def run(self, input_text: str) -> Dict[str, str]:
        tasks = []
        for branch in self.branches:
            prompt = branch["template"].format(input=input_text)
            tasks.append(self._run_branch(branch["name"], prompt))
        results = await asyncio.gather(*tasks)
        return dict(results)
    
    async def _run_branch(self, name, prompt):
        output = await async_llm_call(prompt)
        return (name, output)

# 示例：多角度分析
async def analyze_article(article: str):
    chain = ParallelChain()
    chain.add_branch("tech", "从技术角度分析：{input}")
    chain.add_branch("business", "从商业角度分析：{input}")
    chain.add_branch("risk", "从风险角度分析：{input}")
    
    results = await chain.run(article)
    
    # 汇总
    summary = llm_call(
        f"综合以下三个角度的分析，给出总结：\n"
        f"技术：{results['tech']}\n"
        f"商业：{results['business']}\n"
        f"风险：{results['risk']}"
    )
    return summary

3. 条件链 (Conditional Chain)

根据中间结果动态选择下一步：

class ConditionalChain:
    def __init__(self):
        self.routes = {}
    
    def add_route(self, condition_name: str, 
                  condition_fn, prompt_template: str):
        self.routes[condition_name] = {
            "fn": condition_fn,
            "template": prompt_template,
        }
    
    def run(self, input_text: str) -> str:
        # Step 1: 分类
        classification = llm_call(
            f"将以下问题分类为 'code', 'math', 'general' 之一，只输出类别名：\n{input_text}"
        ).strip()
        
        # Step 2: 根据类别路由
        if classification == "code":
            return llm_call(f"作为编程专家回答：\n{input_text}")
        elif classification == "math":
            return llm_call(f"逐步推理以下数学问题：\n{input_text}")
        else:
            return llm_call(f"回答以下问题：\n{input_text}")

4. Map-Reduce 模式

处理长文档或批量数据的标准模式：

class MapReduceChain:
    def __init__(self, chunk_size: int = 3000):
        self.chunk_size = chunk_size
    
    def _chunk_text(self, text: str) -> list[str]:
        words = text.split()
        chunks = []
        current = []
        current_len = 0
        for w in words:
            if current_len + len(w) > self.chunk_size:
                chunks.append(" ".join(current))
                current = [w]
                current_len = len(w)
            else:
                current.append(w)
                current_len += len(w)
        if current:
            chunks.append(" ".join(current))
        return chunks
    
    def run(self, text: str, map_prompt: str, reduce_prompt: str) -> str:
        chunks = self._chunk_text(text)
        print(f"分割为 {len(chunks)} 个块")
        
        # Map: 并行处理每个块
        map_results = []
        for i, chunk in enumerate(chunks):
            result = llm_call(map_prompt.format(chunk=chunk))
            map_results.append(result)
            print(f"  [Map {i+1}/{len(chunks)}] 完成")
        
        # Reduce: 合并结果
        combined = "\n---\n".join(map_results)
        final = llm_call(reduce_prompt.format(intermediate=combined))
        return final

# 示例：长文档摘要
mrc = MapReduceChain(chunk_size=3000)
summary = mrc.run(
    long_document,
    map_prompt="用 200 字摘要以下文本的核心内容：\n{chunk}",
    reduce_prompt="将以下各段摘要整合为一份连贯的总体摘要（500字以内）：\n{intermediate}"
)

LangChain Pipeline 实现

from langchain.chains import LLMChain, SequentialChain
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate

llm = ChatOpenAI(model="gpt-4", temperature=0.3)

# 定义各步骤
extract_prompt = PromptTemplate(
    input_variables=["document"],
    template="从以下文档中提取所有实体和关系（JSON格式）：\n{document}"
)
extract_chain = LLMChain(llm=llm, prompt=extract_prompt, output_key="entities")

verify_prompt = PromptTemplate(
    input_variables=["entities"],
    template="验证以下提取的实体是否准确，修正错误：\n{entities}"
)
verify_chain = LLMChain(llm=llm, prompt=verify_prompt, output_key="verified")

format_prompt = PromptTemplate(
    input_variables=["verified"],
    template="将以下实体信息格式化为 Markdown 表格：\n{verified}"
)
format_chain = LLMChain(llm=llm, prompt=format_prompt, output_key="formatted")

# 组合流水线
pipeline = SequentialChain(
    chains=[extract_chain, verify_chain, format_chain],
    input_variables=["document"],
    output_variables=["formatted"],
    verbose=True,
)

result = pipeline({"document": raw_text})

错误处理与断点续传

生产环境必须处理 LLM 调用失败的情况：

import json
import os
import time
from dataclasses import dataclass, asdict

@dataclass
class Checkpoint:
    step_name: str
    step_index: int
    input_data: str
    output_data: str
    timestamp: str

class ResilientChain:
    def __init__(self, checkpoint_dir: str = ".checkpoints"):
        self.checkpoint_dir = checkpoint_dir
        os.makedirs(checkpoint_dir, exist_ok=True)
        self.steps = []
    
    def add_step(self, name, prompt_template, max_retries=3):
        self.steps.append({
            "name": name,
            "template": prompt_template,
            "max_retries": max_retries,
        })
        return self
    
    def _save_checkpoint(self, run_id: str, step_idx: int, 
                          step_name: str, output: str):
        cp = Checkpoint(
            step_name=step_name,
            step_index=step_idx,
            input_data="",
            output_data=output,
            timestamp=time.strftime("%Y-%m-%d %H:%M:%S"),
        )
        path = os.path.join(self.checkpoint_dir, f"{run_id}_{step_idx}.json")
        with open(path, "w", encoding="utf-8") as f:
            json.dump(asdict(cp), f, ensure_ascii=False, indent=2)
    
    def _load_checkpoint(self, run_id: str) -> int:
        """返回上次完成的步骤索引，无则返回 -1"""
        for i in range(len(self.steps) - 1, -1, -1):
            path = os.path.join(self.checkpoint_dir, f"{run_id}_{i}.json")
            if os.path.exists(path):
                with open(path, "r", encoding="utf-8") as f:
                    data = json.load(f)
                return i, data["output_data"]
        return -1, None
    
    def run(self, initial_input: str, run_id: str = None) -> str:
        if run_id is None:
            run_id = str(int(time.time()))
        
        # 尝试从断点恢复
        last_step, saved_output = self._load_checkpoint(run_id)
        if last_step >= 0:
            print(f"从步骤 {last_step + 1} 恢复 (run_id={run_id})")
            current_output = saved_output
            start_step = last_step + 1
        else:
            current_output = initial_input
            start_step = 0
        
        # 执行剩余步骤
        for i in range(start_step, len(self.steps)):
            step = self.steps[i]
            print(f"[{i+1}/{len(self.steps)}] 执行: {step['name']}")
            
            for attempt in range(step["max_retries"]):
                try:
                    prompt = step["template"].format(prev_output=current_output)
                    current_output = llm_call(prompt)
                    self._save_checkpoint(run_id, i, step["name"], current_output)
                    break
                except Exception as e:
                    print(f"  尝试 {attempt+1} 失败: {e}")
                    if attempt == step["max_retries"] - 1:
                        raise
                    time.sleep(2 ** attempt)  # 指数退避
        
        return current_output

模式选择决策矩阵

场景推荐模式原因
线性处理流程顺序链每步依赖前步，简单可靠
多角度分析并行链子任务独立，可并行加速
多类型问题条件链不同类型需不同处理策略
长文档处理Map-Reduce超出上下文窗口必须分块
多步推理顺序链 + SC每步用 Self-Consistency 提升可靠性
实时对话短链/单步延迟敏感，减少链长度

实战建议

  1. 每步设定明确输出格式：JSON 或结构化文本，避免下游解析失败
  2. 监控每步延迟：链总延迟 = Σ 各步延迟，及时优化瓶颈
  3. 设置全局超时：避免单步卡死导致整个链超时
  4. 可观测性：记录每步输入输出，便于调试和优化
  5. 版本化 prompt 模板：链中任一 prompt 变更都可能影响全局效果

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。