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

发布时间：2026/7/16 17:23:06

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_factory=list) variables: Dict[str, Any] = field(default_factory=dict) metrics: Dict[str, Any] = field(default_factory=dict) 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( messages=list(ctx.messages), variables=dict(ctx.variables), metrics=dict(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_exceptions=True), timeout=self.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( name="domain_classifier", classifier=self._classify_domain, branches={ "code": self._build_code_branch(), "writing": self._build_writing_branch(), "analysis": self._build_analysis_branch(), }, default_branch=ChainNode( "fallback", prompt_template="请以通用助手的方式回应：{user_message}", ), ) # 构建Parallel示例：并行预处理 self.parallel_preprocess = ParallelNode( name="context_enrichment", children=[ ChainNode("retrieve_docs", prompt_template="检索相关文档：{user_message}"), ChainNode("user_profile", prompt_template="分析用户画像特征"), ChainNode("extract_params", prompt_template="提取上下文参数"), ], aggregator=self._aggregate_contexts, timeout=15.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，生产环境可以将这些指标上报到监控系统，实现编排层级的延迟追踪。