Rust AI CLI 管线化用管道把多个 AI 调用串成处理流水线一、有时候一次 AI 调用真的不够用最近在用 Rust 折腾一个命令行工具功能说起来不复杂用户输入一段需求描述工具自动生成技术方案。刚开始我用了一个很直的做法——直接把用户的输入扔给大模型然后展示返回结果。结果很快就踩坑了。用户的输入往往比较口语化比如我想要一个用户登录功能但模型直接拿这句话去生成方案质量参差不齐。我意识到这里其实需要好几个步骤先让模型分析并结构化用户的需求再根据结构化后的需求生成方案最后对方案做一次合理性检查。这三个步骤如果硬塞到一个 prompt 里模型很容易在某个环节跑偏。于是我开始思考能不能像 Unix 管道那样把多次 AI 调用串成一条处理流水线前一步的输出经过清洗后作为下一步的输入每一步各司其职。这就是本文想聊的主题——如何用 Rust 和 Tokio 异步运行时构建一条可配置、可扩展的 AI 调用管线。二、整体设计把管线画出来在动手写代码之前我先把整个流程画了一张图。这对理清思路帮助很大——尤其是当管线步骤变多的时候没有图很容易把自己绕进去。flowchart TD A[用户输入原始需求] -- B[第一步: 需求结构化] B -- C{结构化结果校验} C --|通过| D[第二步: 方案生成] C --|不通过| A D -- E[中间结果清洗与格式化] E -- F[第三步: 方案合理性检查] F -- G{检查是否通过} G --|通过| H[输出最终方案] G --|存在问题| I[附加检查意见] I -- D H -- J[格式化输出给用户]图中可以看到管线有三个核心步骤结构化、生成、检查。每一步都是一个独立的管道段Pipe Segment有自己的 prompt 配置和模型选择。步骤之间有校验节点如果某一步的输出不符合预期可以触发重试或者回退到上一步。这种设计的好处是每个步骤职责单一。修改某个步骤的 prompt 不会影响其他步骤新增一个步骤也只需要在管线上插一段。三、Rust 实现异步管道的关键代码Rust 的异步生态主要依赖 Tokio而管线化的核心抽象是一个Pipeline结构体它持有一系列Step按顺序执行。先定义每一步的配置use std::sync::Arc; /// 管线中的一个处理步骤 #[derive(Clone)] pub struct PipelineStep { /// 步骤名称用于日志和调试 pub name: String, /// 发送给 AI 的 system prompt pub system_prompt: String, /// 使用的模型名称如 gpt-4o 或 claude-3-opus pub model: String, /// 最大重试次数 pub max_retries: u32, } /// AI 客户端的抽象 trait方便替换不同的模型提供商 #[async_trait::async_trait] pub trait AiClient: Send Sync { /// 向模型发送消息并获取回复 async fn chat(self, system: str, user: str, model: str) - anyhow::ResultString; }然后是管线本身的结构/// AI 调用管线 pub struct AiPipeline { /// 管线中的步骤列表按顺序执行 steps: VecPipelineStep, /// AI 客户端实例 client: Arcdyn AiClient, } impl AiPipeline { /// 创建一条新的管线 pub fn new(client: Arcdyn AiClient) - Self { Self { steps: Vec::new(), client, } } /// 向管线末尾添加一个步骤 pub fn add_step(mut self, step: PipelineStep) - mut Self { self.steps.push(step); self } /// 执行整条管线 pub async fn execute(self, initial_input: str) - anyhow::ResultString { // 用初始输入作为第一个步骤的 user 消息 let mut current_input initial_input.to_string(); for step in self.steps { println!([管线] 正在执行步骤: {}, step.name); // 调用 AI支持重试 let mut last_error None; for attempt in 0..step.max_retries { match self.client .chat(step.system_prompt, current_input, step.model) .await { Ok(output) { // 将当前步骤的输出作为下一步的输入 current_input output; last_error None; break; } Err(e) { eprintln!( [管线] 步骤 {} 第 {} 次尝试失败: {}, step.name, attempt 1, e ); last_error Some(e); // 失败后等待一小段时间再重试 tokio::time::sleep(std::time::Duration::from_secs(1)).await; } } } if let Some(err) last_error { return Err(anyhow::anyhow!( 步骤 {} 在 {} 次重试后仍然失败: {}, step.name, step.max_retries 1, err )); } } Ok(current_input) } }完整的执行入口可以这样用#[tokio::main] async fn main() - anyhow::Result() { // 创建 AI 客户端这里用一个假实现示意实际对接 OpenAI SDK 或 reqwest let client: Arcdyn AiClient Arc::new(MyAiClient::new()?); // 构建管线 let mut pipeline AiPipeline::new(client); pipeline .add_step(PipelineStep { name: 需求结构化.into(), system_prompt: 你是一个需求分析师请将用户的原始需求整理成结构化的功能描述。只输出结构化描述不要任何额外说明。.into(), model: gpt-4o.into(), max_retries: 2, }) .add_step(PipelineStep { name: 方案生成.into(), system_prompt: 你是一个技术架构师根据结构化需求生成完整的技术方案包含技术栈选型、模块划分和核心流程。.into(), model: gpt-4o.into(), max_retries: 2, }) .add_step(PipelineStep { name: 方案审查.into(), system_prompt: 你是一个资深代码审查员请检查技术方案的合理性和潜在问题在方案后附上审查意见。.into(), model: claude-3-opus.into(), max_retries: 1, }); // 执行管线 let result pipeline.execute(我想做一个电商秒杀功能).await?; println!(最终输出:\n{}, result); Ok(()) }关键点在于current_input在每一步之间流转上一步的输出直接变成下一步的输入。整个执行过程是异步的不会阻塞 Tokio 的事件循环。四、边界与取舍管线不是银弹这种管线设计在实际使用中会暴露几个问题值得提前想清楚。延迟累积。每一步 AI 调用都有网络延迟和推理时间三步走下来用户等待的时间可能是单次调用的三倍以上。如果某个步骤用的是慢模型体验会更差。缓解方式是非关键步骤可以考虑用更小更快的模型或者将可以并行执行的步骤改为并发调用。错误传播。管线中间某一步的输出质量差会像滚雪球一样影响后续所有步骤。比如需求结构化这一步把秒杀理解成了秒杀漏洞攻击后面两步就全歪了。我现在的做法是在关键节点加校验——用简单的规则或者再用一次轻量模型调用来判断输出是否合理。成本翻倍。每一次调用都在烧 token三条管线的 token 消耗量是单次调用的三倍。对于高频使用的场景需要考虑缓存——如果用户的输入和之前某次高度相似可以直接复用当时的结果。调试困难。当最终输出不符合预期时很难快速定位是哪个步骤出了问题。我的做法是把每个步骤的中间输出都记录到日志里出问题后逐段排查。五、总结AI 管线化的核心思想其实不复杂把一个大的 AI 任务拆成多个小任务每一步专注做好一件事通过管道串联起来。这和 Unix 哲学做好一件事如出一辙。用 Rust 实现的好处是类型安全和异步性能都有保障。Pipeline结构体的设计让步骤的增删改都很直观后续想加一个步骤间数据转换的中间层也不难。当然这个方案也有明显的代价——延迟和成本都变高了。所以它更适合对质量要求高于响应速度的场景比如代码生成、方案设计这类跑一次等十几秒也能接受的任务。我在自己的项目里用了一段时间直观感受是同样的 prompt拆成三步后的输出质量确实比一步到位要好尤其是一些容易走偏的逻辑推理环节。如果你也在做类似的工具不妨试试把大 prompt 拆开。