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_factorydict) 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( modelgpt-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 None3.2 财报分析链式推理示例# 定义财报分析的链式推理步骤 orchestrator ChainOrchestrator(llm_client) # 步骤1信息提取 orchestrator.add_step(ChainStep( nameextract, step_typeStepType.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_entryTrue) # 步骤2增长分析 orchestrator.add_step(ChainStep( namegrowth_analysis, step_typeStepType.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( nameinvestment_advice, step_typeStepType.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 的推理过程更可控、更可靠。