Prompt Engineering 与 Agent 工作流：多步骤推理的链式编排实践

一、单次 Prompt 的推理极限：为什么复杂任务总是"跑偏"

当用户向 AI 提出一个需要多步推理的复杂任务时（如"分析这份财报，找出增长最快的业务线，评估其可持续性，并给出投资建议"），单次 Prompt 的输出质量急剧下降。原因在于：大模型的注意力在长输出中逐渐衰减，前半部分的推理逻辑可能在后半部分被遗忘；多个子任务的输出格式和标准不同，混合在一个回复中容易互相干扰；错误在推理链中累积——第一步的偏差会导致后续所有步骤偏离正确方向。链式编排（Chain-of-Thought Orchestration）将复杂任务拆解为独立的子步骤，每步专注一个子任务，前一步的输出作为后一步的输入，通过结构化的推理链提升整体质量。

二、链式推理的编排模型

链式推理有三种编排模式：线性链（Sequential Chain）、分支链（Branching Chain）和循环链（Looping Chain）。线性链按固定顺序执行，适合步骤确定的任务；分支链根据中间结果选择不同路径，适合条件性任务；循环链允许重复执行直到满足终止条件，适合迭代优化任务。

graph TD
    A[复杂任务输入] --> B[任务分解器<br/>拆解为子步骤]
    B --> C[步骤1：信息提取<br/>从原始数据中提取关键信息]
    C --> D[步骤2：分析推理<br/>基于提取信息进行推理]
    D --> E{需要分支?}
    E -->|数据增长型| F[步骤3a：增长分析]
    E -->|利润导向型| G[步骤3b：利润分析]
    F --> H[步骤4：综合评估]
    G --> H
    H --> I{质量达标?}
    I -->|否| J[循环：补充分析<br/>回到步骤2]
    I -->|是| K[步骤5：生成结论]

    style B fill:#e1f5fe
    style D fill:#c8e6c9
    style H fill:#fff3e0

每个子步骤使用独立的、精心设计的 Prompt，而非将所有指令塞入一个 Prompt。子步骤的 Prompt 更短、更聚焦，模型的输出质量显著提升。步骤间的数据传递通过结构化格式（JSON Schema）约束，确保前一步的输出能被后一步正确解析。

三、链式编排的工程实现

3.1 链式推理框架

from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional
from enum import Enum
import json

class StepType(Enum):
    SEQUENTIAL = "sequential"   # 线性步骤
    BRANCHING = "branching"    # 分支步骤
    LOOPING = "looping"        # 循环步骤

@dataclass
class ChainStep:
    """链式推理步骤：包含 Prompt 模板、输入输出 Schema 和执行逻辑"""
    name: str
    step_type: StepType
    prompt_template: str
    input_schema: Dict[str, Any]     # 输入 JSON Schema
    output_schema: Dict[str, Any]    # 输出 JSON Schema
    max_retries: int = 2
    condition: Optional[Callable[[Dict], bool]] = None  # 分支条件
    next_steps: Optional[Dict[str, str]] = None  # 分支映射：条件值 → 步骤名

@dataclass
class ChainContext:
    """链式推理上下文：在步骤间传递数据"""
    steps_output: Dict[str, Any] = field(default_factory=dict)
    current_step: str = ""
    iteration_count: int = 0
    max_iterations: int = 5

class ChainOrchestrator:
    """
    链式推理编排器：按定义的步骤链执行多步推理

    设计考量：编排器的核心职责是步骤调度和数据传递，
    而非推理本身。每步推理由 LLM 完成，编排器负责：
    1. 将上一步的输出注入当前步骤的 Prompt
    2. 校验当前步骤的输出是否符合 Schema
    3. 根据条件选择下一步骤
    4. 处理循环和重试逻辑
    """

    def __init__(self, llm_client):
        self.llm_client = llm_client
        self._steps: Dict[str, ChainStep] = {}
        self._entry_step: Optional[str] = None

    def add_step(self, step: ChainStep, is_entry: bool = False):
        """添加步骤到链中"""
        self._steps[step.name] = step
        if is_entry:
            self._entry_step = step.name

    async def execute(self, initial_input: Dict[str, Any]) -> Dict[str, Any]:
        """执行链式推理"""
        if not self._entry_step:
            raise ValueError("未设置入口步骤")

        context = ChainContext()
        context.steps_output["input"] = initial_input
        current_step_name = self._entry_step

        while current_step_name:
            step = self._steps.get(current_step_name)
            if step is None:
                break

            context.current_step = current_step_name

            # 构建当前步骤的 Prompt
            prompt = self._build_prompt(step, context)

            # 执行推理，带重试
            output = await self._execute_step(step, prompt)

            # 校验输出
            if not self._validate_output(output, step.output_schema):
                # 输出不符合 Schema，重试
                for retry in range(step.max_retries):
                    output = await self._execute_step(step, prompt)
                    if self._validate_output(output, step.output_schema):
                        break

            # 保存输出到上下文
            context.steps_output[current_step_name] = output

            # 确定下一步骤
            current_step_name = self._get_next_step(step, output, context)

            # 循环安全检查
            if step.step_type == StepType.LOOPING:
                context.iteration_count += 1
                if context.iteration_count >= context.max_iterations:
                    break

        return context.steps_output

    def _build_prompt(self, step: ChainStep, context: ChainContext) -> str:
        """构建步骤 Prompt：将上下文数据注入模板"""
        prompt = step.prompt_template

        # 替换模板中的变量引用，如 {{input.query}} → 实际值
        for step_name, output in context.steps_output.items():
            if isinstance(output, dict):
                for key, value in output.items():
                    placeholder = f"{{{{{step_name}.{key}}}}}"
                    prompt = prompt.replace(placeholder, str(value))

        return prompt

    async def _execute_step(self, step: ChainStep, prompt: str) -> Dict:
        """执行单个步骤的 LLM 推理"""
        response = await self.llm_client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "请严格按照 JSON Schema 输出结果。"},
                {"role": "user", "content": prompt},
            ],
            response_format={"type": "json_object"},
        )

        try:
            return json.loads(response.choices[0].message.content)
        except json.JSONDecodeError:
            return {"error": "输出解析失败", "raw": response.choices[0].message.content}

    def _validate_output(self, output: Dict, schema: Dict) -> bool:
        """校验输出是否符合 Schema"""
        try:
            import jsonschema
            jsonschema.validate(output, schema)
            return True
        except jsonschema.ValidationError:
            return False

    def _get_next_step(
        self, step: ChainStep, output: Dict, context: ChainContext
    ) -> Optional[str]:
        """确定下一步骤"""
        if step.step_type == StepType.BRANCHING and step.next_steps:
            # 分支步骤：根据条件选择路径
            if step.condition:
                branch_key = step.condition(output)
                return step.next_steps.get(str(branch_key))
            return None

        # 线性步骤：返回下一个步骤（按添加顺序）
        step_names = list(self._steps.keys())
        current_idx = step_names.index(step.name)
        if current_idx + 1 < len(step_names):
            return step_names[current_idx + 1]

        return None

3.2 财报分析链式推理示例

# 定义财报分析的链式推理步骤

orchestrator = ChainOrchestrator(llm_client)

# 步骤1：信息提取
orchestrator.add_step(ChainStep(
    name="extract",
    step_type=StepType.SEQUENTIAL,
    prompt_template="""
    从以下财报文本中提取关键财务数据：
    {{input.report_text}}

    请提取以下信息：
    - 各业务线的收入与增长率
    - 毛利率与净利率
    - 现金流状况
    - 重大风险提示
    """,
    input_schema={"type": "object", "properties": {"report_text": {"type": "string"}}},
    output_schema={
        "type": "object",
        "properties": {
            "business_lines": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string"},
                        "revenue": {"type": "number"},
                        "growth_rate": {"type": "number"},
                    },
                },
            },
            "margins": {"type": "object"},
            "cash_flow": {"type": "object"},
            "risks": {"type": "array", "items": {"type": "string"}},
        },
    },
), is_entry=True)

# 步骤2：增长分析
orchestrator.add_step(ChainStep(
    name="growth_analysis",
    step_type=StepType.SEQUENTIAL,
    prompt_template="""
    基于以下提取的财务数据，分析增长最快的业务线：
    {{extract.business_lines}}

    请评估：
    1. 增长最快的业务线及其增长驱动因素
    2. 增长的可持续性（市场空间、竞争格局、护城河）
    3. 潜在风险因素
    """,
    input_schema={"type": "object"},
    output_schema={
        "type": "object",
        "properties": {
            "fastest_growing": {"type": "string"},
            "growth_drivers": {"type": "array", "items": {"type": "string"}},
            "sustainability_score": {"type": "number", "minimum": 0, "maximum": 10},
            "risks": {"type": "array", "items": {"type": "string"}},
        },
    },
))

# 步骤3：投资建议
orchestrator.add_step(ChainStep(
    name="investment_advice",
    step_type=StepType.SEQUENTIAL,
    prompt_template="""
    基于以下分析结果，给出投资建议：
    增长分析：{{growth_analysis}}
    风险提示：{{extract.risks}}

    请给出：
    1. 综合评级（买入/持有/观望）
    2. 核心理由（不超过 3 条）
    3. 关键风险提示
    4. 建议关注的时间节点
    """,
    input_schema={"type": "object"},
    output_schema={
        "type": "object",
        "properties": {
            "rating": {"type": "string", "enum": ["买入", "持有", "观望"]},
            "reasons": {"type": "array", "items": {"type": "string"}},
            "key_risks": {"type": "array", "items": {"type": "string"}},
            "watch_dates": {"type": "array", "items": {"type": "string"}},
        },
    },
))

四、链式编排的边界与权衡

链式推理的延迟是线性增长的——每增加一个步骤，总延迟增加一次 LLM 调用的时间（通常 1-3 秒）。5 步链式推理的总延迟可能达到 5-15 秒，对于实时交互场景不可接受。缓解手段包括：并行执行无依赖的步骤（如同时提取多个维度的信息）、使用更快的模型处理简单步骤（如用 GPT-4o-mini 做信息提取，GPT-4o 做深度推理）。

步骤间的数据传递存在信息损失。每个步骤只将结构化输出传递给下一步，原始文本中的细微信息可能在提取过程中丢失。对于需要完整上下文的步骤，应将原始输入作为额外上下文传入，而非仅依赖前一步的结构化输出。

循环链的终止条件设计需要谨慎。如果终止条件过于宽松，循环可能无限执行；过于严格，则可能在质量不达标时就终止。建议设置最大迭代次数（通常 3-5 次），并在每次迭代时评估输出质量，质量不再提升时提前终止。

五、总结

链式编排将复杂的多步推理任务拆解为独立的子步骤，通过结构化的数据传递和条件调度提升输出质量。核心实践包括：线性链处理步骤确定的任务，分支链处理条件性任务，循环链处理迭代优化任务；JSON Schema 约束步骤间的数据格式；重试机制应对输出格式异常；并行执行和模型分层优化延迟。链式编排不是让 AI 更聪明，而是让 AI 的推理过程更可控、更可靠。