踐:模型網(wǎng)關(guān)與流程編排打造代碼可控的AI應(yīng)用)
年初接一個(gè)新項(xiàng)目的時(shí)候我一開始給的是那種“包一層API調(diào)一調(diào)”的demo演示效果很驚艷結(jié)果老板來一句“這個(gè)邏輯能不能換個(gè)模型試試”“這個(gè)流程能不能剪一段”我就傻了改prompt還好說想換模型、想改判斷分支代碼得扒半天測(cè)試又不全根本不敢動(dòng)。那時(shí)候我就在想AI項(xiàng)目做到一定復(fù)雜度卡脖子的根本不是模型效果而是代碼能不能被自己穩(wěn)穩(wěn)控制住。后來我把整套思路重新捋了一遍就是圍繞著**“靈活、代碼可控”的BuildingAI理念**去重構(gòu)——不把AI當(dāng)成一個(gè)黑盒API塞進(jìn)業(yè)務(wù)里而是把模型、流程、參數(shù)、輸出全拆開全部放在代碼能掌控的層面。這套東西做完之后再有人跟我說“換一下模型試試”我不會(huì)慌改一個(gè)配置項(xiàng)就能跑。今天我就把這套做法里最核心的骨架、設(shè)計(jì)取舍和一些實(shí)測(cè)中踩到的坑一起寫下來。1. 為什么AI構(gòu)建必須回到“代碼可控”這條路上來先說一個(gè)我自己的感受過去AI項(xiàng)目的交付模式基本是“平臺(tái)化工具在線配置”。平臺(tái)把模型能力封裝得很好看你點(diǎn)點(diǎn)鼠標(biāo)就能搭一個(gè)問答機(jī)器人??烧嬲暇€跑起來之后問題的復(fù)雜度會(huì)迅速超過平臺(tái)能覆蓋的邊界。1.1 低代碼平臺(tái)和黑盒API的真實(shí)處境低代碼平臺(tái)的最大痛點(diǎn)是模板化。它給你很多現(xiàn)成的控件、節(jié)點(diǎn)、prompt槽位但這些控件之間的數(shù)據(jù)流是平臺(tái)定義好的。我想在兩個(gè)節(jié)點(diǎn)中間插一步“先判斷用戶意圖再?zèng)Q定要不要走檢索”平臺(tái)支持當(dāng)然好不支持就得繞路哪怕支持節(jié)點(diǎn)多了之后連線密密麻麻出了問題根本不知道是哪條鏈路帶偏的。黑盒API的問題則在于不可觀察。API內(nèi)部做了什么、用了什么prompt模板、檢索結(jié)果怎么排序、模型在哪個(gè)版本上表現(xiàn)穩(wěn)定你都看不到。而生產(chǎn)環(huán)境恰恰需要的是“出問題時(shí)我能定位”。有一次線上機(jī)器人突然開始胡言亂語排查半天發(fā)現(xiàn)是上游模型悄悄升級(jí)了版本輸出風(fēng)格全變了。這種“不可控制”會(huì)讓整個(gè)項(xiàng)目變得很被動(dòng)。1.2 “靈活”在AI項(xiàng)目里到底指什么我后來定義的“靈活”不是能隨便問問題這種產(chǎn)品層面的靈活而是工程層面上的可調(diào)整性。具體拆開看大概是四件事模型可替換同一套業(yè)務(wù)代碼今天用這個(gè)模型明天能換另一個(gè)模型甚至換成私有化部署的模型中間不用大改。流程可編排意圖識(shí)別、知識(shí)檢索、上下文壓縮、答案生成這些步驟可以隨時(shí)增刪、調(diào)整順序像搭積木一樣可控。輸出可校驗(yàn)?zāi)P头祷氐膬?nèi)容在進(jìn)到業(yè)務(wù)系統(tǒng)之前能被攔截、檢查格式、強(qiáng)制糾錯(cuò)。策略可定制每個(gè)客戶、每個(gè)場(chǎng)景的prompt和參數(shù)都獨(dú)立配置互不干擾。這四條做到位外部環(huán)境再怎么變代碼主體都能穩(wěn)住。1.3 代碼可控的三個(gè)層次我自己習(xí)慣把“代碼可控”拆成三層來理解第一層是調(diào)用層也就是直接使用模型SDK。第二層是編排層負(fù)責(zé)把多個(gè)AI環(huán)節(jié)組織成一個(gè)完整流程。第三層是配置層把prompt、模型名、溫度參數(shù)、開關(guān)項(xiàng)這些易變的東西提取出來交給外部配置來控制。三層之間各司其職調(diào)用層管“怎么連”編排層管“怎么串”配置層管“怎么調(diào)”。這樣的分層思路是整個(gè)BuildingAI實(shí)踐的基石后面所有設(shè)計(jì)都是在這個(gè)框架里展開的。2. 搭建BuildingAI的最小骨架從模型網(wǎng)關(guān)到流程編排有了分層思路之后動(dòng)手第一個(gè)要做的不是接模型而是先把地基打穩(wěn)。我建議第一步就是寫一個(gè)統(tǒng)一模型網(wǎng)關(guān)讓上層代碼不直接依賴任何一個(gè)具體模型供應(yīng)商。2.1 模型網(wǎng)關(guān)讓模型可替換的第一步模型網(wǎng)關(guān)的作用相當(dāng)于給所有模型提供同一個(gè)“插座”接口。上層業(yè)務(wù)只跟這個(gè)接口打交道底層換什么模型都不影響。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: 統(tǒng)一的對(duì)話補(bǔ)全接口 pass以Python為例我定義了一個(gè)抽象基類任何模型供應(yīng)商只要實(shí)現(xiàn)這個(gè)接口就能接入。接口上設(shè)計(jì)幾個(gè)關(guān)鍵點(diǎn)messages統(tǒng)一使用ChatMessage對(duì)象不直接暴露各廠商的原始結(jié)構(gòu)。temperature和max_tokens是可選的不傳就用配置里的默認(rèn)值。model參數(shù)允許在特殊場(chǎng)景下臨時(shí)指定模型。接著做一個(gè)具體實(shí)現(xiàn)比如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為什么網(wǎng)關(guān)層這么重要因?yàn)閷?shí)際項(xiàng)目里“換模型”不是一次性動(dòng)作而是常態(tài)。同一個(gè)客戶可能今天用A模型跑得挺好明天因?yàn)槌杀驹蛳霌QB模型或者同一個(gè)應(yīng)用簡(jiǎn)單問題走便宜模型復(fù)雜問題才走強(qiáng)模型。網(wǎng)關(guān)把這類切換從“改業(yè)務(wù)代碼”變成了“加一個(gè)實(shí)現(xiàn)類”。我后面甚至做了一個(gè)簡(jiǎn)單的路由策略實(shí)現(xiàn)同一個(gè)網(wǎng)關(guān)里按問題類型自動(dòng)選模型業(yè)務(wù)層無感。2.2 流程編排把AI能力拆成能被代碼控制的小步驟有了網(wǎng)關(guān)接下來是編排層。我借鑒了微服務(wù)里“編排”的概念把每一次AI交互從“一次大調(diào)用”拆成“多個(gè)可控小步驟”。舉個(gè)例子一個(gè)企業(yè)知識(shí)庫問答機(jī)器人它的處理鏈路我拆成了這樣意圖路由先判斷用戶是想問知識(shí)庫還是閑聊還是觸發(fā)某個(gè)工具。檢索觸發(fā)如果走知識(shí)庫則把問題向量化在向量數(shù)據(jù)庫里檢索相似段落。上下文組裝把檢索到的段落和對(duì)話歷史拼裝成一個(gè)精煉的上下文。生成回答調(diào)用模型生成最終答案。輸出校驗(yàn)檢查答案是否包含知識(shí)庫里的關(guān)鍵信息點(diǎn)不含就拒絕并重新生成一次。每一步都可以被單獨(dú)測(cè)試、單獨(dú)替換。我寫編排的時(shí)候沒有引入重型框架而是定義了一個(gè)輕量的Step接口每個(gè)步驟只做一件事數(shù)據(jù)以字典上下文傳遞。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這樣的代碼結(jié)構(gòu)看起來平淡但對(duì)我來說它解決了最大的問題每個(gè)環(huán)節(jié)都能單獨(dú)改。比如檢索效果不好我只需要替換檢索步驟生成步驟根本不用動(dòng)想加一個(gè)敏感詞檢測(cè)步驟往列表里插一段就行。2.3 配置與代碼分離不改代碼也能調(diào)整行為編排層穩(wěn)定之后最大的麻煩變成“每改一個(gè)prompt都要?jiǎng)哟a重新部署”。后來我把所有易變項(xiàng)全部提取到了YAML配置里線上調(diào)參只需改配置并熱加載不用碰代碼。# config/models.yaml chat_model: provider: openai model_name: gpt-4o-mini temperature: 0.3 max_tokens: 800 # config/prompts.yaml intent_route: system: | 你是意圖識(shí)別引擎只輸出一個(gè)詞knowledge / chitchat / tool。 ... qa_generate: system: | 你是企業(yè)知識(shí)庫助手只能根據(jù)提供的資料回答 資料不存在時(shí)明確說“未找到相關(guān)信息”。配置里還習(xí)慣放一些開關(guān)features: use_vector_search: true use_output_guard: true fallback_to_strong_model: false這些開關(guān)讓我在灰度發(fā)布和降級(jí)處理時(shí)有了極大的操作空間。有一次線上檢索服務(wù)抖動(dòng)我直接把use_vector_search改成false機(jī)器人立刻降級(jí)成純模型問答模式保障了基本可用性然后才慢慢排查檢索集群的問題。3. 在可觀測(cè)性與調(diào)試能力上做設(shè)計(jì)而不是等問題出現(xiàn)再補(bǔ)救代碼可控的第二層理解是“可以看到、可以重放、可以定位”。很多AI項(xiàng)目給人的印象是“玄學(xué)”其實(shí)很大程度上是缺少可觀測(cè)性。我自己吃過虧所以后來把可觀測(cè)性做進(jìn)了框架層而不是事后補(bǔ)日志。3.1 結(jié)構(gòu)化日志與追蹤讓每次請(qǐng)求都有完整軌跡我給每次請(qǐng)求分配一個(gè)trace_id這個(gè)ID從入口傳遍所有步驟。每個(gè)步驟執(zhí)行時(shí)都會(huì)往結(jié)構(gòu)化日志里寫模型名、prompt長(zhǎng)度、輸出內(nèi)容、耗時(shí)、token用量。后面排查問題時(shí)直接按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這個(gè)裝飾式的LoggedStep可以包在任何步驟外層日志也好、追蹤也好都不需要侵入業(yè)務(wù)代碼。3.2 輸出校驗(yàn)與防御性兜底模型輸出是不可控的所以我在生成答案之后加了一道硬校驗(yàn)。具體做法是根據(jù)場(chǎng)景定義一些檢查規(guī)則規(guī)則可以很簡(jiǎn)單比如答案是否以預(yù)期格式開頭答案是否包含涉敏詞JSON字段是否能反序列化成功并且關(guān)鍵字段存在校驗(yàn)失敗時(shí)不直接返回錯(cuò)誤而是進(jìn)入重新生成通道提示模型剛才的格式有問題讓它重試一次重試仍失敗就降級(jí)為兜底回答。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這類“校驗(yàn)重試兜底”的鏈路才是生產(chǎn)環(huán)境里真正讓人放心的設(shè)計(jì)而不是把希望寄托在模型的“自覺”上。3.3 搭建回歸評(píng)估集讓改動(dòng)有據(jù)可依代碼可控也意味著要能回答“這次改動(dòng)到底有沒有變差”。我建了一個(gè)輕量的評(píng)估集就是幾十條有代表性的測(cè)試用例每次改同prompt或調(diào)配置之后批量跑一遍對(duì)比輸出質(zhì)量。用一個(gè)腳本統(tǒng)一跑把每條的輸入、輸出存下來手動(dòng)或半自動(dòng)對(duì)比新舊版本差異。這個(gè)機(jī)制不復(fù)雜但價(jià)值極高它給了我“改起來不怕”的底氣因?yàn)槿魏胃膭?dòng)都能在幾分鐘內(nèi)看到整體影響而不是靠感覺。4. 一個(gè)實(shí)際場(chǎng)景讓對(duì)話機(jī)器人具備“換腦”能力前面講了這么多理念用個(gè)真實(shí)案例串起來會(huì)更清楚。我去年做了一個(gè)多租戶的客服機(jī)器人需求是不同客戶有不同的模型偏好、不同的知識(shí)庫、不同的語氣要求而且客戶可能隨時(shí)想調(diào)整。4.1 需求拆解與代碼組織這個(gè)項(xiàng)目的核心約束是模型不確定、流程不確定、prompt不確定。按照目前的BuildingAI思路我做了三件事實(shí)現(xiàn)OpenAI和Claude兩個(gè)網(wǎng)關(guān)類同時(shí)允許按租戶配置私有化模型接入。流程編排固定為“路由→檢索→組裝→生成→校驗(yàn)”但每個(gè)步驟都從租戶配置里讀取參數(shù)。每個(gè)租戶一套獨(dú)立的prompt模板和模型參數(shù)。調(diào)用鏈路上入口根據(jù)租戶ID加載配置整個(gè)后續(xù)邏輯全部驅(qū)動(dòng)于配置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”我只改數(shù)據(jù)庫里provider字段明天說“回答要更簡(jiǎn)短”我只調(diào)temperature和系統(tǒng)prompt模板甚至不需要發(fā)版。4.2 實(shí)測(cè)中遇到的三個(gè)真實(shí)問題方案跑起來后問題主要集中在幾個(gè)預(yù)料之外的地方。第一個(gè)問題是兩家模型的輸出格式風(fēng)格差很多同樣的JSON要求OpenAI規(guī)規(guī)矩矩按樣例輸出Claude有時(shí)候會(huì)自作主張多包一層markdown代碼塊直接把解析器搞得崩潰。后來我在輸出校驗(yàn)步驟里做了格式清洗并在prompt里加了強(qiáng)約束示例才算穩(wěn)住。這個(gè)事告訴我模型可替換不只是“接口可替換”行為歸一化才是真正難點(diǎn)。第二個(gè)問題是重試風(fēng)暴。最開始超時(shí)重試沒有做退避控制高峰時(shí)一個(gè)慢請(qǐng)求會(huì)觸發(fā)多次重試反而把模型調(diào)用線程打滿。后來給網(wǎng)關(guān)里統(tǒng)一加了指數(shù)退避和最大重試次數(shù)限制快速失敗比原地等待更合理。第三個(gè)問題是token成本失控。上下文太長(zhǎng)時(shí)成本漲得很快尤其多個(gè)步驟都拿全量歷史去拼其實(shí)很多歷史對(duì)當(dāng)前問題沒用。我加了一個(gè)上下文壓縮步驟把歷史消息里和當(dāng)前問題無關(guān)的部分過濾掉成本直接降了百分之四五十響應(yīng)也快了很多。4.3 成本與延遲的可控性提到成本和延遲這也是“可控”的一部分。項(xiàng)目里每次調(diào)用都統(tǒng)計(jì)token數(shù)和耗時(shí)實(shí)時(shí)同步到一個(gè)簡(jiǎn)單面板里。運(yùn)營同學(xué)能看到每個(gè)客戶每天的消耗趨勢(shì)有人半夜跑腳本狂調(diào)用很快就能發(fā)現(xiàn)異常。代碼層面對(duì)成本的控制主要是通過模型分級(jí)簡(jiǎn)單問題走便宜模型復(fù)雜問題才走貴模型。這個(gè)分級(jí)策略寫在配置里運(yùn)營可以直接調(diào)整閾值不需要開發(fā)介入。5. 代碼可控的邊界哪些環(huán)節(jié)不必自己從頭造最后想潑一點(diǎn)冷水代碼可控不代表一切都要自己寫。實(shí)際工程里有些東西直接交給成熟工具和組件比自己造輪子穩(wěn)得多。5.1 過度開發(fā)的信號(hào)我自己經(jīng)歷過一個(gè)反面案例明明只是做一個(gè)內(nèi)部用的問答工具卻花了兩周時(shí)間去設(shè)計(jì)插件體系、多租戶隔離、可插拔流程。后來復(fù)盤發(fā)現(xiàn)核心交互就是“問一句話取一段資料生成一段答案”完全不需要那么重的抽象。所以我想提醒一點(diǎn)分層和可配置性是手段不是目的。如果項(xiàng)目只有一條流程、一個(gè)模型、不打算變那最樸素的寫法反而最穩(wěn)。什么情況下需要代碼可控這套思路我總結(jié)下來至少要滿足下面之一有多個(gè)模型接入需求且可能頻繁切換。有多套業(yè)務(wù)場(chǎng)景prompt和流程差異較大。需要自動(dòng)化測(cè)試和批量回歸驗(yàn)證。需要精細(xì)化管理成本和延遲靈活做降級(jí)。如果一條都不滿足簡(jiǎn)單封裝就夠了。5.2 適合交給成熟工具的環(huán)節(jié)檢索這一環(huán)我建議直接用向量數(shù)據(jù)庫和現(xiàn)成embedding服務(wù)自己從頭實(shí)現(xiàn)語義檢索性價(jià)比很低。數(shù)據(jù)導(dǎo)入、切分、索引管理這些臟活累活成熟工具能幫你省一大半時(shí)間。流程編排也不需要引入重型的擬人化編排框架一個(gè)簡(jiǎn)單的步驟列表就夠等流程復(fù)雜到代碼寫起來明顯別扭時(shí)再考慮引入可視化編排平臺(tái)。5.3 代碼可控與團(tuán)隊(duì)協(xié)作的平衡還有一個(gè)點(diǎn)很容易被忽視代碼可控提高了系統(tǒng)的靈活性但也要求團(tuán)隊(duì)里有人能讀懂整個(gè)鏈路。低代碼平臺(tái)讓業(yè)務(wù)人員也能拖出demo回到代碼之后這個(gè)門檻轉(zhuǎn)移到了工程團(tuán)隊(duì)身上。所以我通常在項(xiàng)目里保留一份非常樸素的架構(gòu)說明把“這個(gè)流程有幾個(gè)步驟、每步改哪里、配置項(xiàng)都有什么作用”寫得明明白白。這樣新成員接手時(shí)不至于對(duì)著代碼發(fā)懵??偟膩碚fBuildingAI這條路走下來我最深的感受是AI應(yīng)用變得可預(yù)測(cè)、可維護(hù)、可演進(jìn)靠的不是某一個(gè)強(qiáng)大模型而是一整套代碼層面的掌控力。模型會(huì)變、需求會(huì)變但只要網(wǎng)關(guān)、編排、配置、可觀測(cè)這幾根柱子立住了變化來的時(shí)候就不會(huì)慌。最后分享一個(gè)小習(xí)慣每次發(fā)布新配置或新prompt之后我都會(huì)把線上真實(shí)請(qǐng)求和對(duì)應(yīng)的完整上下文存一份脫敏樣本隔一段時(shí)間翻出來對(duì)比很多效果退化都是靠這種“存檔重放”的方式提前暴露的。