ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

BuildingAI实践:模型网关与流程编排打造代码可控的AI应用

BuildingAI实践:模型网关与流程编排打造代码可控的AI应用 年初接一个新项目的时候我一开始给的是那种“包一层API调一调”的demo演示效果很惊艳结果老板来一句“这个逻辑能不能换个模型试试”“这个流程能不能剪一段”我就傻了改prompt还好说想换模型、想改判断分支代码得扒半天测试又不全根本不敢动。那时候我就在想AI项目做到一定复杂度卡脖子的根本不是模型效果而是代码能不能被自己稳稳控制住。后来我把整套思路重新捋了一遍就是围绕着**“灵活、代码可控”的BuildingAI理念**去重构——不把AI当成一个黑盒API塞进业务里而是把模型、流程、参数、输出全拆开全部放在代码能掌控的层面。这套东西做完之后再有人跟我说“换一下模型试试”我不会慌改一个配置项就能跑。今天我就把这套做法里最核心的骨架、设计取舍和一些实测中踩到的坑一起写下来。1. 为什么AI构建必须回到“代码可控”这条路上来先说一个我自己的感受过去AI项目的交付模式基本是“平台化工具在线配置”。平台把模型能力封装得很好看你点点鼠标就能搭一个问答机器人。可真正上线跑起来之后问题的复杂度会迅速超过平台能覆盖的边界。1.1 低代码平台和黑盒API的真实处境低代码平台的最大痛点是模板化。它给你很多现成的控件、节点、prompt槽位但这些控件之间的数据流是平台定义好的。我想在两个节点中间插一步“先判断用户意图再决定要不要走检索”平台支持当然好不支持就得绕路哪怕支持节点多了之后连线密密麻麻出了问题根本不知道是哪条链路带偏的。黑盒API的问题则在于不可观察。API内部做了什么、用了什么prompt模板、检索结果怎么排序、模型在哪个版本上表现稳定你都看不到。而生产环境恰恰需要的是“出问题时我能定位”。有一次线上机器人突然开始胡言乱语排查半天发现是上游模型悄悄升级了版本输出风格全变了。这种“不可控制”会让整个项目变得很被动。1.2 “灵活”在AI项目里到底指什么我后来定义的“灵活”不是能随便问问题这种产品层面的灵活而是工程层面上的可调整性。具体拆开看大概是四件事模型可替换同一套业务代码今天用这个模型明天能换另一个模型甚至换成私有化部署的模型中间不用大改。流程可编排意图识别、知识检索、上下文压缩、答案生成这些步骤可以随时增删、调整顺序像搭积木一样可控。输出可校验模型返回的内容在进到业务系统之前能被拦截、检查格式、强制纠错。策略可定制每个客户、每个场景的prompt和参数都独立配置互不干扰。这四条做到位外部环境再怎么变代码主体都能稳住。1.3 代码可控的三个层次我自己习惯把“代码可控”拆成三层来理解第一层是调用层也就是直接使用模型SDK。第二层是编排层负责把多个AI环节组织成一个完整流程。第三层是配置层把prompt、模型名、温度参数、开关项这些易变的东西提取出来交给外部配置来控制。三层之间各司其职调用层管“怎么连”编排层管“怎么串”配置层管“怎么调”。这样的分层思路是整个BuildingAI实践的基石后面所有设计都是在这个框架里展开的。2. 搭建BuildingAI的最小骨架从模型网关到流程编排有了分层思路之后动手第一个要做的不是接模型而是先把地基打稳。我建议第一步就是写一个统一模型网关让上层代码不直接依赖任何一个具体模型供应商。2.1 模型网关让模型可替换的第一步模型网关的作用相当于给所有模型提供同一个“插座”接口。上层业务只跟这个接口打交道底层换什么模型都不影响。from abc import ABC, abstractmethod from typing import List, Dict, Optional class ChatMessage: def __init__(self, role: str, content: str): self.role role self.content content class ModelGateway(ABC): abstractmethod def chat( self, messages: List[ChatMessage], temperature: Optional[float] None, max_tokens: Optional[int] None, model: Optional[str] None, ) - str: 统一的对话补全接口 pass以Python为例我定义了一个抽象基类任何模型供应商只要实现这个接口就能接入。接口上设计几个关键点messages统一使用ChatMessage对象不直接暴露各厂商的原始结构。temperature和max_tokens是可选的不传就用配置里的默认值。model参数允许在特殊场景下临时指定模型。接着做一个具体实现比如OpenAI的接入import os from openai import OpenAI class OpenAIGateway(ModelGateway): def __init__(self, api_keyNone, base_urlNone): self.client OpenAI( api_keyapi_key or os.getenv(OPENAI_API_KEY), base_urlbase_url or os.getenv(OPENAI_BASE_URL), ) self.default_model os.getenv(OPENAI_MODEL, gpt-4o-mini) def chat(self, messages, temperatureNone, max_tokensNone, modelNone): resp self.client.chat.completions.create( modelmodel or self.default_model, messages[{role: m.role, content: m.content} for m in messages], temperaturetemperature, max_tokensmax_tokens, ) return resp.choices[0].message.content为什么网关层这么重要因为实际项目里“换模型”不是一次性动作而是常态。同一个客户可能今天用A模型跑得挺好明天因为成本原因想换B模型或者同一个应用简单问题走便宜模型复杂问题才走强模型。网关把这类切换从“改业务代码”变成了“加一个实现类”。我后面甚至做了一个简单的路由策略实现同一个网关里按问题类型自动选模型业务层无感。2.2 流程编排把AI能力拆成能被代码控制的小步骤有了网关接下来是编排层。我借鉴了微服务里“编排”的概念把每一次AI交互从“一次大调用”拆成“多个可控小步骤”。举个例子一个企业知识库问答机器人它的处理链路我拆成了这样意图路由先判断用户是想问知识库还是闲聊还是触发某个工具。检索触发如果走知识库则把问题向量化在向量数据库里检索相似段落。上下文组装把检索到的段落和对话历史拼装成一个精炼的上下文。生成回答调用模型生成最终答案。输出校验检查答案是否包含知识库里的关键信息点不含就拒绝并重新生成一次。每一步都可以被单独测试、单独替换。我写编排的时候没有引入重型框架而是定义了一个轻量的Step接口每个步骤只做一件事数据以字典上下文传递。class PipelineStep: def execute(self, ctx: dict) - dict: raise NotImplementedError class Pipeline: def __init__(self, steps: List[PipelineStep]): self.steps steps def run(self, ctx: dict) - dict: for step in self.steps: ctx step.execute(ctx) if ctx.get(abort): break return ctx这样的代码结构看起来平淡但对我来说它解决了最大的问题每个环节都能单独改。比如检索效果不好我只需要替换检索步骤生成步骤根本不用动想加一个敏感词检测步骤往列表里插一段就行。2.3 配置与代码分离不改代码也能调整行为编排层稳定之后最大的麻烦变成“每改一个prompt都要动代码重新部署”。后来我把所有易变项全部提取到了YAML配置里线上调参只需改配置并热加载不用碰代码。# config/models.yaml chat_model: provider: openai model_name: gpt-4o-mini temperature: 0.3 max_tokens: 800 # config/prompts.yaml intent_route: system: | 你是意图识别引擎只输出一个词knowledge / chitchat / tool。 ... qa_generate: system: | 你是企业知识库助手只能根据提供的资料回答 资料不存在时明确说“未找到相关信息”。配置里还习惯放一些开关features: use_vector_search: true use_output_guard: true fallback_to_strong_model: false这些开关让我在灰度发布和降级处理时有了极大的操作空间。有一次线上检索服务抖动我直接把use_vector_search改成false机器人立刻降级成纯模型问答模式保障了基本可用性然后才慢慢排查检索集群的问题。3. 在可观测性与调试能力上做设计而不是等问题出现再补救代码可控的第二层理解是“可以看到、可以重放、可以定位”。很多AI项目给人的印象是“玄学”其实很大程度上是缺少可观测性。我自己吃过亏所以后来把可观测性做进了框架层而不是事后补日志。3.1 结构化日志与追踪让每次请求都有完整轨迹我给每次请求分配一个trace_id这个ID从入口传遍所有步骤。每个步骤执行时都会往结构化日志里写模型名、prompt长度、输出内容、耗时、token用量。后面排查问题时直接按trace_id捞整条链路的日志一眼就能看出问题出在哪一步。import logging import time import uuid class LoggedStep(PipelineStep): def __init__(self, inner_step: PipelineStep): self.inner inner_step self.logger logging.getLogger(buildingai) def execute(self, ctx: dict) - dict: start time.time() trace_id ctx.setdefault(trace_id, uuid.uuid4().hex[:12]) try: result self.inner.execute(ctx) self.logger.info( step_ok, extra{ trace_id: trace_id, step: type(self.inner).__name__, cost_ms: round((time.time() - start) * 1000, 2), }, ) return result except Exception as e: self.logger.error( step_failed, extra{ trace_id: trace_id, step: type(self.inner).__name__, error: str(e), }, ) raise这个装饰式的LoggedStep可以包在任何步骤外层日志也好、追踪也好都不需要侵入业务代码。3.2 输出校验与防御性兜底模型输出是不可控的所以我在生成答案之后加了一道硬校验。具体做法是根据场景定义一些检查规则规则可以很简单比如答案是否以预期格式开头答案是否包含涉敏词JSON字段是否能反序列化成功并且关键字段存在校验失败时不直接返回错误而是进入重新生成通道提示模型刚才的格式有问题让它重试一次重试仍失败就降级为兜底回答。class OutputGuardStep(PipelineStep): def __init__(self, max_retries1): self.max_retries max_retries def execute(self, ctx: dict) - dict: raw_answer ctx[raw_answer] for i in range(self.max_retries 1): problem self._validate(raw_answer) if not problem: ctx[answer] raw_answer return ctx # 把问题反馈给模型让它自己修正 ctx[retry_feedback] problem raw_answer ctx[regenerate_answer](feedbackproblem) ctx[answer] self._fallback_answer() return ctx这类“校验重试兜底”的链路才是生产环境里真正让人放心的设计而不是把希望寄托在模型的“自觉”上。3.3 搭建回归评估集让改动有据可依代码可控也意味着要能回答“这次改动到底有没有变差”。我建了一个轻量的评估集就是几十条有代表性的测试用例每次改同prompt或调配置之后批量跑一遍对比输出质量。用一个脚本统一跑把每条的输入、输出存下来手动或半自动对比新旧版本差异。这个机制不复杂但价值极高它给了我“改起来不怕”的底气因为任何改动都能在几分钟内看到整体影响而不是靠感觉。4. 一个实际场景让对话机器人具备“换脑”能力前面讲了这么多理念用个真实案例串起来会更清楚。我去年做了一个多租户的客服机器人需求是不同客户有不同的模型偏好、不同的知识库、不同的语气要求而且客户可能随时想调整。4.1 需求拆解与代码组织这个项目的核心约束是模型不确定、流程不确定、prompt不确定。按照目前的BuildingAI思路我做了三件事实现OpenAI和Claude两个网关类同时允许按租户配置私有化模型接入。流程编排固定为“路由→检索→组装→生成→校验”但每个步骤都从租户配置里读取参数。每个租户一套独立的prompt模板和模型参数。调用链路上入口根据租户ID加载配置整个后续逻辑全部驱动于配置tenant_config load_tenant_config(tenant_id) gateway ModelGatewayFactory.create(tenant_config[provider]) pipeline build_pipeline(tenant_config) result pipeline.run({question: user_input, tenant_config: tenant_config})这样客户今天说“把模型换成Claude”我只改数据库里provider字段明天说“回答要更简短”我只调temperature和系统prompt模板甚至不需要发版。4.2 实测中遇到的三个真实问题方案跑起来后问题主要集中在几个预料之外的地方。第一个问题是两家模型的输出格式风格差很多同样的JSON要求OpenAI规规矩矩按样例输出Claude有时候会自作主张多包一层markdown代码块直接把解析器搞得崩溃。后来我在输出校验步骤里做了格式清洗并在prompt里加了强约束示例才算稳住。这个事告诉我模型可替换不只是“接口可替换”行为归一化才是真正难点。第二个问题是重试风暴。最开始超时重试没有做退避控制高峰时一个慢请求会触发多次重试反而把模型调用线程打满。后来给网关里统一加了指数退避和最大重试次数限制快速失败比原地等待更合理。第三个问题是token成本失控。上下文太长时成本涨得很快尤其多个步骤都拿全量历史去拼其实很多历史对当前问题没用。我加了一个上下文压缩步骤把历史消息里和当前问题无关的部分过滤掉成本直接降了百分之四五十响应也快了很多。4.3 成本与延迟的可控性提到成本和延迟这也是“可控”的一部分。项目里每次调用都统计token数和耗时实时同步到一个简单面板里。运营同学能看到每个客户每天的消耗趋势有人半夜跑脚本狂调用很快就能发现异常。代码层面对成本的控制主要是通过模型分级简单问题走便宜模型复杂问题才走贵模型。这个分级策略写在配置里运营可以直接调整阈值不需要开发介入。5. 代码可控的边界哪些环节不必自己从头造最后想泼一点冷水代码可控不代表一切都要自己写。实际工程里有些东西直接交给成熟工具和组件比自己造轮子稳得多。5.1 过度开发的信号我自己经历过一个反面案例明明只是做一个内部用的问答工具却花了两周时间去设计插件体系、多租户隔离、可插拔流程。后来复盘发现核心交互就是“问一句话取一段资料生成一段答案”完全不需要那么重的抽象。所以我想提醒一点分层和可配置性是手段不是目的。如果项目只有一条流程、一个模型、不打算变那最朴素的写法反而最稳。什么情况下需要代码可控这套思路我总结下来至少要满足下面之一有多个模型接入需求且可能频繁切换。有多套业务场景prompt和流程差异较大。需要自动化测试和批量回归验证。需要精细化管理成本和延迟灵活做降级。如果一条都不满足简单封装就够了。5.2 适合交给成熟工具的环节检索这一环我建议直接用向量数据库和现成embedding服务自己从头实现语义检索性价比很低。数据导入、切分、索引管理这些脏活累活成熟工具能帮你省一大半时间。流程编排也不需要引入重型的拟人化编排框架一个简单的步骤列表就够等流程复杂到代码写起来明显别扭时再考虑引入可视化编排平台。5.3 代码可控与团队协作的平衡还有一个点很容易被忽视代码可控提高了系统的灵活性但也要求团队里有人能读懂整个链路。低代码平台让业务人员也能拖出demo回到代码之后这个门槛转移到了工程团队身上。所以我通常在项目里保留一份非常朴素的架构说明把“这个流程有几个步骤、每步改哪里、配置项都有什么作用”写得明明白白。这样新成员接手时不至于对着代码发懵。总的来说BuildingAI这条路走下来我最深的感受是AI应用变得可预测、可维护、可演进靠的不是某一个强大模型而是一整套代码层面的掌控力。模型会变、需求会变但只要网关、编排、配置、可观测这几根柱子立住了变化来的时候就不会慌。最后分享一个小习惯每次发布新配置或新prompt之后我都会把线上真实请求和对应的完整上下文存一份脱敏样本隔一段时间翻出来对比很多效果退化都是靠这种“存档重放”的方式提前暴露的。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进