Prompt编排的工程化抽象：Chain、Router与Parallel三种模式的架构设计

Prompt编排的工程化抽象Chain、Router与Parallel三种模式的架构设计一、Prompt从单次调用到流水线的演进压力大模型应用从Demo走向产品的过程中单次Prompt调用的局限性迅速暴露。一个典型的Agent应用需要执行多步推理先理解用户意图再检索相关知识然后生成结构化输出最后验证结果的正确性。每一步都可能依赖上一步的结果也可能与上一步并行推进。Prompt编排的本质是将LLM调用从单次问询抽象为可组合的计算图。这个图由节点Prompt调用或工具调用和边数据流向与控制流向构成。三种核心编排模式——Chain链式、Router路由和Parallel并行——覆盖了90%以上的Agent编排需求。工程化的难点在于如何在保证编排灵活性的同时维持流式输出的低延迟体验如何在Router分支中处理不确定性如何在Parallel模式下管理多个LLM调用的Token预算。二、三种编排模式的数据流与适用场景Chain模式适用于确定性顺序强依赖的场景。上一个Prompt的输出直接作为下一个Prompt的输入形成严格的前后依赖。典型场景是多步推理Agent先规划任务分解再逐步执行子任务最后汇总结论。Router模式的价值在于将单一重度Prompt拆分为多个轻量专用Prompt。一个意图分类器通常用小模型或规则引擎决定走向哪个专业分支。这样做的好处有两个每个分支的Prompt可以精确定制且不相关的分支不会消耗Token。Parallel模式的核心挑战是同步等待。多个并行调用中任何一个完成得慢整个编排的端到端延迟就是那个最慢调用的耗时。因此Parallel模式更适合一个慢调用加上多个快调用的场景——主生成任务占用主要时间检索、画像、提取等预处理任务在背景并行完成。四、编排流水线的生产级抽象# prompt_pipeline.py - Prompt编排流水线引擎 from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Any, AsyncIterator, Dict, List, Optional import asyncio import time dataclass class PipelineContext: 在编排节点间传递的上下文 messages: List[Dict[str, str]] field(default_factorylist) variables: Dict[str, Any] field(default_factorydict) metrics: Dict[str, Any] field(default_factorydict) class PipelineNode(ABC): 编排节点的抽象基类 def __init__(self, name: str, model: str gpt-4o): self.name name self.model model abstractmethod async def execute( self, ctx: PipelineContext ) - PipelineContext: ... class ChainNode(PipelineNode): 链式节点——严格顺序执行 def __init__(self, name: str, prompt_template: str, **kwargs): super().__init__(name, **kwargs) self.prompt_template prompt_template self.next_node: Optional[ChainNode] None def then(self, next_node: ChainNode) - ChainNode: 链式连接语法糖 self.next_node next_node return next_node async def execute( self, ctx: PipelineContext ) - PipelineContext: start time.time() # 渲染Prompt模板 prompt self.prompt_template.format(**ctx.variables) ctx.messages.append({role: user, content: prompt}) # 调用LLM实际实现会接入模型API response await self._call_llm(ctx.messages) ctx.messages.append({role: assistant, content: response}) ctx.variables[f{self.name}_output] response # 记录链式延迟 ctx.metrics[f{self.name}_latency] time.time() - start # 如果存在下一个节点继续执行 if self.next_node: return await self.next_node.execute(ctx) return ctx class RouterNode(PipelineNode): 路由节点——根据分类结果分发到不同分支 def __init__( self, name: str, classifier: callable, branches: Dict[str, PipelineNode], default_branch: Optional[PipelineNode] None, **kwargs ): super().__init__(name, **kwargs) self.classifier classifier self.branches branches self.default_branch default_branch async def execute( self, ctx: PipelineContext ) - PipelineContext: start time.time() # 执行分类——决定路由方向 # 分类器可以是轻量模型调用或规则引擎 category await self.classifier(ctx) ctx.variables[route_category] category # 路由到对应分支 target self.branches.get( category, self.default_branch ) if target is None: raise ValueError( f未知路由类别: {category} f可用分支: {list(self.branches.keys())} ) ctx.metrics[f{self.name}_routing_latency] ( time.time() - start ) return await target.execute(ctx) class ParallelNode(PipelineNode): 并行节点——多个子节点同时执行 def __init__( self, name: str, children: List[PipelineNode], aggregator: callable, timeout: float 30.0, **kwargs ): super().__init__(name, **kwargs) self.children children self.aggregator aggregator self.timeout timeout async def execute( self, ctx: PipelineContext ) - PipelineContext: start time.time() # 并行执行所有子节点 tasks [] for child in self.children: # 每个子节点使用独立的上下文副本 child_ctx PipelineContext( messageslist(ctx.messages), variablesdict(ctx.variables), metricsdict(ctx.metrics), ) task asyncio.create_task( self._execute_with_timeout(child, child_ctx) ) tasks.append(task) # 等待所有任务完成带总超时 try: results await asyncio.wait_for( asyncio.gather(*tasks, return_exceptionsTrue), timeoutself.timeout, ) except asyncio.TimeoutError: # 超时处理——记录日志并使用降级上下文 ctx.metrics[f{self.name}_timeout] True results [] # 聚合所有结果 child_contexts [] for i, result in enumerate(results): if isinstance(result, Exception): ctx.metrics[f{self.name}_child_{i}_error] str(result) else: child_contexts.append(result) # 调用聚合器合并上下文 aggregated await self.aggregator(ctx, child_contexts) ctx.metrics[f{self.name}_total_latency] time.time() - start return aggregated async def _execute_with_timeout( self, child: PipelineNode, child_ctx: PipelineContext ): return await child.execute(child_ctx) # 使用示例构建一个完整的Agent编排流水线 class AgentPipeline: Agent级别的编排流水线 def __init__(self): # 构建Chain示例意图→决策→行动 self.intent_analysis ChainNode( intent_analysis, prompt_template分析用户意图{user_message}, ) self.decision ChainNode( decision, prompt_template( 基于意图分析结果决定需要调用的工具。 意图{intent_analysis_output} ), ) self.action ChainNode( action, prompt_template( 执行决策{decision_output} ), ) # 组装链式结构 self.intent_analysis.then(self.decision).then(self.action) # 构建Router示例按领域分类分发 self.domain_router RouterNode( namedomain_classifier, classifierself._classify_domain, branches{ code: self._build_code_branch(), writing: self._build_writing_branch(), analysis: self._build_analysis_branch(), }, default_branchChainNode( fallback, prompt_template请以通用助手的方式回应{user_message}, ), ) # 构建Parallel示例并行预处理 self.parallel_preprocess ParallelNode( namecontext_enrichment, children[ ChainNode(retrieve_docs, prompt_template检索相关文档{user_message}), ChainNode(user_profile, prompt_template分析用户画像特征), ChainNode(extract_params, prompt_template提取上下文参数), ], aggregatorself._aggregate_contexts, timeout15.0, # 15秒并行超时 ) async def run( self, user_message: str ) - AsyncIterator[str]: 运行完整流水线并支持流式输出 ctx PipelineContext() ctx.variables[user_message] user_message # 第一步并行预处理上下文 ctx await self.parallel_preprocess.execute(ctx) # 第二步路由到领域分支 ctx await self.domain_router.execute(ctx) # 第三步通过链式执行完成任务 ctx await self.intent_analysis.execute(ctx) # 流式返回最终结果 async for token in self._stream_output(ctx): yield token async def _classify_domain(self, ctx: PipelineContext) - str: # 简化的分类器实现 msg ctx.variables.get(user_message, ).lower() if any(kw in msg for kw in [代码, 编程, 函数]): return code elif any(kw in msg for kw in [写, 文章, 文档]): return writing return analysis async def _aggregate_contexts( self, main_ctx: PipelineContext, child_contexts: List[PipelineContext] ) - PipelineContext: 聚合并行子节点的结果 for child_ctx in child_contexts: main_ctx.variables.update(child_ctx.variables) main_ctx.metrics.update(child_ctx.metrics) return main_ctx async def _stream_output( self, ctx: PipelineContext ) - AsyncIterator[str]: # 流式输出最终结果 final_message ctx.messages[-1][content] for token in final_message.split(): yield token def _build_code_branch(self) - ChainNode: return ChainNode( code_gen, prompt_template作为代码专家{user_message}, ) def _build_writing_branch(self) - ChainNode: return ChainNode( writing, prompt_template作为写作专家{user_message}, ) def _build_analysis_branch(self) - ChainNode: return ChainNode( analysis, prompt_template作为分析专家{user_message}, )这个实现的三个关键设计决策。第一所有编排节点共享PipelineContext但ParallelNode为每个子节点创建副本以避免数据竞争。第二Router的分类器是一个可注入的函数而非硬编码的Prompt——这样可以用规则引擎替代LLM分类节省Token成本。第三每个节点独立记录metrics生产环境可以将这些指标上报到监控系统实现编排层级的延迟追踪。四、编排模式的性能边界与陷阱三种模式的组合使用需要注意成本爆炸的问题。Chain模式中每一步的延迟线性累加。一个3步的Chain如果每步延迟2秒总延迟是6秒。User的流式体验中首Token必须等到第一步完成后才能出现。优化策略是将不依赖前一步结果的节点放入一个ParallelNode中并发执行。Router模式的核心风险是分类错误。分类器的准确率每降低10%就会导致10%的请求进入错误分支——错误分支的回复质量通常远低于正确分支。建议在Router后加入一个置信度检查节点如果分类置信度低于阈值走默认通用分支。Parallel模式最大的陷阱是Token浪费。三个并行Nodes各自调用LLM成本是单个调用的3倍。如果一个并行分支的结果最终未被聚合器使用那个调用的成本就是纯浪费。因此聚合器的设计必须保证每个并行结果都有明确的消费路径。五、总结Prompt编排的三种模式构建了Agent应用的骨架。Chain负责确定性顺序执行是大多数Agent的默认模式。Router负责根据输入特征选择最优路径用小成本避免大模型的泛化困境。Parallel负责加速无依赖的子任务是降低端到端延迟的核心手段。工程落地的建议从Chain入手搭建基础编排能力当识别到明确的意图分歧点时引入Router当端到端延迟超过用户体验阈值时引入Parallel。三种模式的组合形成一张有向无环图DAG这是生产级Agent编排的正确抽象层次。