關(guān)與路由器:從概念到Python路由分發(fā)實(shí)現(xiàn))
最近在 Hacker News 上留意到一個(gè)很有意思的開源項(xiàng)目發(fā)布帖Experiential定位是“開源智能體工作流網(wǎng)關(guān)與路由器”。很多人看到這個(gè)標(biāo)題可能第一反應(yīng)是“又來一個(gè) AI 框架”但如果結(jié)合當(dāng)前智能體Agent開發(fā)的實(shí)際痛點(diǎn)來看這個(gè)方向其實(shí)非常值得關(guān)注?,F(xiàn)在的現(xiàn)狀是智能體框架越來越多模型接入方式五花八門企業(yè)內(nèi)部可能同時(shí)跑著好幾個(gè)基于 Dify、Coze 或自研框架搭建的智能體每個(gè)智能體又可能對(duì)接多個(gè)模型供應(yīng)商。調(diào)用關(guān)系一多問題就來了誰來統(tǒng)一入口誰來分發(fā)請(qǐng)求模型切換時(shí)能不能不修改業(yè)務(wù)代碼一個(gè)模型服務(wù)不可用的時(shí)候能不能自動(dòng)把流量切到備用模型Experiential 這類“智能體網(wǎng)關(guān) / 路由器”項(xiàng)目要解決的就是這些問題。它借鑒了網(wǎng)絡(luò)分層中網(wǎng)關(guān)和路由器的思想在網(wǎng)絡(luò)層網(wǎng)關(guān)負(fù)責(zé)連接不同網(wǎng)絡(luò)路由器負(fù)責(zé)根據(jù)地址把數(shù)據(jù)包轉(zhuǎn)發(fā)到正確的下一跳在 AI 應(yīng)用層智能體網(wǎng)關(guān)統(tǒng)一接收上層業(yè)務(wù)請(qǐng)求智能體路由器則根據(jù)意圖、模型能力、成本、可用性等因素把請(qǐng)求轉(zhuǎn)發(fā)到對(duì)應(yīng)的智能體或工作流。本文會(huì)從概念出發(fā)拆解智能體網(wǎng)關(guān)與路由器的核心職責(zé)并結(jié)合一個(gè)可運(yùn)行的 Python 最小示例演示如何自己實(shí)現(xiàn)一套“規(guī)則路由 工作流分發(fā) 降級(jí)容災(zāi)”的簡(jiǎn)化版本。如果你正在做智能體平臺(tái)集成、模型統(tǒng)一接入或者只是想把多個(gè) Agent 的調(diào)用入口收斂起來這篇文章應(yīng)該能給你一個(gè)比較完整的思路。1. 背景與核心概念1.1 從“模型多、智能體多、入口亂”開始過去兩年AI 應(yīng)用的開發(fā)模式發(fā)生了很明顯的變化。最早大家是直接調(diào)用大模型 API寫完 Prompt 就拿結(jié)果后來開始流行 RAG把知識(shí)庫檢索和大模型生成拼在一起再后來出現(xiàn)了智能體模型可以自己規(guī)劃步驟、調(diào)用工具、循環(huán)執(zhí)行任務(wù)直到完成目標(biāo)。到了這個(gè)階段很多團(tuán)隊(duì)的架構(gòu)開始變得復(fù)雜。一個(gè)稍具規(guī)模的項(xiàng)目里可能有下面的情況同時(shí)接入了多家大模型供應(yīng)商包括國內(nèi)和國外的開源模型、商業(yè)模型同一個(gè)模型要服務(wù)于多個(gè)業(yè)務(wù)場(chǎng)景比如客服問答、內(nèi)容生成、數(shù)據(jù)分析團(tuán)隊(duì)基于不同框架搭建了多個(gè)智能體每個(gè)智能體有自己的 Prompt、工具集和工作流部分智能體通過 Dify 這類平臺(tái)發(fā)布部分智能體是自研代碼寫死的調(diào)用方式各不相同。這種“各自為政”的架構(gòu)在剛開始跑通 Demo 時(shí)沒有問題但進(jìn)入生產(chǎn)環(huán)境后很快就會(huì)暴露出幾個(gè)痛點(diǎn)上層業(yè)務(wù)要記住每個(gè)智能體的調(diào)用地址和鑒權(quán)方式模型供應(yīng)商調(diào)整價(jià)格或某天某個(gè)模型服務(wù)不可用時(shí)業(yè)務(wù)代碼要跟著改每個(gè)智能體的調(diào)用量、成功率、延遲數(shù)據(jù)分散在各處根本沒法統(tǒng)一觀測(cè)和治理。1.2 什么是智能體工作流網(wǎng)關(guān)與路由器要理解 Experiential 這類項(xiàng)目可以借用網(wǎng)絡(luò)里的兩個(gè)經(jīng)典設(shè)備來類比網(wǎng)關(guān)和路由器。網(wǎng)關(guān)Gateway負(fù)責(zé)“接入”和“轉(zhuǎn)換”。在傳統(tǒng)網(wǎng)絡(luò)里網(wǎng)關(guān)是連接兩個(gè)不同網(wǎng)絡(luò)的關(guān)口負(fù)責(zé)協(xié)議轉(zhuǎn)換、地址轉(zhuǎn)換和流量接入。在智能體架構(gòu)里網(wǎng)關(guān)就是所有 AI 請(qǐng)求的統(tǒng)一入口。上層業(yè)務(wù)不需要知道背后有幾套智能體也不需要關(guān)心每個(gè)智能體的鑒權(quán)方式只需要把請(qǐng)求發(fā)給網(wǎng)關(guān)由網(wǎng)關(guān)完成后續(xù)的鑒權(quán)、轉(zhuǎn)發(fā)、限流、日志記錄。路由器Router負(fù)責(zé)“決策”和“分發(fā)”。傳統(tǒng)路由器根據(jù) IP 地址和路由表決定數(shù)據(jù)包走哪條鏈路。智能體路由器則根據(jù)請(qǐng)求內(nèi)容、用戶身份、上下文、成本預(yù)算等條件決定這次請(qǐng)求應(yīng)該交給哪個(gè)模型、哪個(gè)智能體、哪條工作流。Experiential 把這兩個(gè)概念合并到了一個(gè)開源項(xiàng)目里既要當(dāng)網(wǎng)關(guān)統(tǒng)一接入和治理又要當(dāng)路由器智能地分發(fā)和調(diào)度。放在實(shí)際業(yè)務(wù)中它的核心價(jià)值可以概括為一句話讓 AI 調(diào)用從“點(diǎn)對(duì)點(diǎn)直連”變成“通過中間層統(tǒng)一調(diào)度”。如果用一個(gè)圖來描述傳統(tǒng)方式是這樣的邏輯業(yè)務(wù)系統(tǒng) - 智能體 A固定調(diào)用模型 X 業(yè)務(wù)系統(tǒng) - 智能體 B固定調(diào)用模型 Y 業(yè)務(wù)系統(tǒng) - 智能體 C通過 Dify 平臺(tái)引入網(wǎng)關(guān)與路由器之后架構(gòu)變成業(yè)務(wù)系統(tǒng) - 智能體網(wǎng)關(guān) - 路由規(guī)則 - 智能體 A / 智能體 B / 智能體 C - 模型 X / 模型 Y / 模型 Z這樣一來上層業(yè)務(wù)只依賴網(wǎng)關(guān)的接口底層模型和智能體的變化被隔離在了網(wǎng)關(guān)層。1.3 與傳統(tǒng) API 網(wǎng)關(guān)的區(qū)別有人可能會(huì)問這不就是 API 網(wǎng)關(guān)嗎Kong、APISIX、Spring Cloud Gateway 不都能做轉(zhuǎn)發(fā)和路由嗎確實(shí)傳統(tǒng)的 API 網(wǎng)關(guān)可以完成請(qǐng)求轉(zhuǎn)發(fā)、鑒權(quán)、限流等基礎(chǔ)能力但智能體網(wǎng)關(guān)路由器和傳統(tǒng) API 網(wǎng)關(guān)有一個(gè)本質(zhì)區(qū)別智能體網(wǎng)關(guān)需要理解“語義”而不只是解析 URL 和 Header。傳統(tǒng) API 網(wǎng)關(guān)的路由規(guī)則通常是基于路徑、方法、Header 等靜態(tài)條件/api/order/** - 訂單服務(wù) /api/user/** - 用戶服務(wù)智能體網(wǎng)關(guān)的輸入是一條自然語言文本它無法只靠 URL 決定轉(zhuǎn)發(fā)給誰。它可能需要解析用戶意圖判斷這是“閑聊”還是“數(shù)據(jù)分析”還是“售后支持”結(jié)合用戶上下文判斷是否要調(diào)用某個(gè)特定工具根據(jù)任務(wù)的復(fù)雜程度選擇模型——簡(jiǎn)單問題走小模型復(fù)雜推理走強(qiáng)模型實(shí)時(shí)感知模型服務(wù)的可用性和延遲動(dòng)態(tài)切換目標(biāo)。所以智能體網(wǎng)關(guān)路由器 傳統(tǒng)網(wǎng)關(guān)的基礎(chǔ)能力 對(duì)文本和任務(wù)的理解能力 對(duì)模型和智能體的動(dòng)態(tài)調(diào)度能力。這也是 Experiential 這類開源項(xiàng)目區(qū)別于普通 API 網(wǎng)關(guān)的地方。2. 為什么需要智能體路由器三大典型場(chǎng)景2.1 模型路由把任務(wù)交給合適的模型大模型的調(diào)用成本差異很大。一個(gè)小參數(shù)模型和頂級(jí)大模型的單次調(diào)用價(jià)格可能相差幾十倍甚至上百倍。如果所有請(qǐng)求都走最強(qiáng)模型成本會(huì)非常難看但如果所有請(qǐng)求都走輕量模型復(fù)雜任務(wù)的效果又達(dá)不到要求。這時(shí)候就需要模型路由。網(wǎng)關(guān)收到請(qǐng)求后先判斷任務(wù)的復(fù)雜度然后做分級(jí)處理。比如簡(jiǎn)單問答、關(guān)鍵詞提取、文本分類路由到輕量模型代碼生成、復(fù)雜推理、長文本分析路由到強(qiáng)模型涉及知識(shí)庫檢索的任務(wù)先走 RAG 鏈路再交給生成模型。這樣可以在效果和成本之間找到更好的平衡點(diǎn)。Experiential 這類路由器可以把模型路由做成規(guī)則化、可配置的能力而不是把判斷邏輯散落在各個(gè)業(yè)務(wù)代碼里。2.2 工作流路由按意圖選擇智能體企業(yè)內(nèi)部往往會(huì)建設(shè)多個(gè)垂直智能體。比如客服智能體處理退換貨、物流查詢、產(chǎn)品咨詢數(shù)據(jù)分析智能體查詢銷售數(shù)據(jù)、生成報(bào)表、分析趨勢(shì)知識(shí)庫問答智能體基于內(nèi)部文檔回答問題內(nèi)容創(chuàng)作智能體生成營銷文案、產(chǎn)品介紹。這些智能體可能由不同團(tuán)隊(duì)開發(fā)和維護(hù)使用不同的技術(shù)棧。如果沒有統(tǒng)一路由層上層業(yè)務(wù)就得自己判斷“這句話該調(diào)用哪個(gè)智能體”判斷邏輯寫死在業(yè)務(wù)代碼里一旦智能體數(shù)量增加代碼就會(huì)變得難以維護(hù)。智能體工作流路由器可以集中管理這些判斷規(guī)則。用戶在對(duì)話框里輸入“幫我寫一段產(chǎn)品介紹”網(wǎng)關(guān)就把請(qǐng)求路由到內(nèi)容創(chuàng)作智能體輸入“這個(gè)月華東區(qū)的銷售額是多少”網(wǎng)關(guān)就路由到數(shù)據(jù)分析智能體輸入“我的訂單什么時(shí)候發(fā)貨”網(wǎng)關(guān)就路由到客服智能體。這種路由策略既可以基于關(guān)鍵詞匹配也可以接入意圖識(shí)別模型還可以結(jié)合用戶的部門、角色、權(quán)限做更細(xì)粒度的分發(fā)。2.3 降級(jí)與容災(zāi)讓服務(wù)不中斷大模型服務(wù)并不總是穩(wěn)定的。某個(gè)模型供應(yīng)商可能因?yàn)樨?fù)載過高返回限流可能因?yàn)榫W(wǎng)絡(luò)問題超時(shí)也可能因?yàn)榘姹旧?jí)導(dǎo)致短暫不可用。如果業(yè)務(wù)代碼直接綁定某一家模型一旦服務(wù)出問題整個(gè)功能就掛了。有了網(wǎng)關(guān)路由器之后可以在路由規(guī)則里配置多個(gè)備用目標(biāo)。當(dāng)主模型調(diào)用失敗或者延遲超過閾值網(wǎng)關(guān)自動(dòng)把請(qǐng)求切換到備用模型或備用智能體。上層業(yè)務(wù)感受到的只是響應(yīng)變慢或結(jié)果來源不同但功能不會(huì)中斷。這也是 Experiential 這類項(xiàng)目很實(shí)用的價(jià)值點(diǎn)。對(duì)生產(chǎn)環(huán)境來說路由層不只是在“分發(fā)流量”更是在“保障可用性”。3. 環(huán)境準(zhǔn)備與整體設(shè)計(jì)3.1 運(yùn)行環(huán)境說明下面我們通過一個(gè)最小示例演示智能體網(wǎng)關(guān)與路由器的核心工作機(jī)制。由于 Experiential 本身是一個(gè)還在迭代中的開源項(xiàng)目不同版本的接口和配置方式可能不同所以我這里不直接綁定它的源碼細(xì)節(jié)而是用 Python 實(shí)現(xiàn)一個(gè)“思路版”的智能體路由網(wǎng)關(guān)讓你先掌握這個(gè)方向的核心原理。示例環(huán)境以常見組合為例操作系統(tǒng)Windows / macOS / Linux 均可Python3.10 或更高版本W(wǎng)eb 框架FastAPI數(shù)據(jù)校驗(yàn)Pydantic v2ASGI 服務(wù)器Uvicorn。這些版本需要根據(jù)你的實(shí)際環(huán)境調(diào)整。如果你本地 Python 版本較低建議先升級(jí)到 3.10 以上避免 Pydantic v2 的兼容性問題。首先創(chuàng)建虛擬環(huán)境并安裝依賴mkdir agent-gateway-demo cd agent-gateway-demo python -m venv venv # Windows venv\Scripts\activate # macOS / Linux source venv/bin/activate pip install fastapi uvicorn pydantic3.2 項(xiàng)目結(jié)構(gòu)我們用一個(gè)單文件加一個(gè)配置文件的輕量結(jié)構(gòu)來演示agent-gateway-demo/ ├── main.py # FastAPI 入口實(shí)現(xiàn)網(wǎng)關(guān)接口與路由邏輯 ├── rules.json # 路由規(guī)則配置 └── README.md # 可選項(xiàng)目說明實(shí)際生產(chǎn)項(xiàng)目中建議把路由邏輯、后端適配器、配置加載拆分成獨(dú)立模塊后面第七節(jié)會(huì)給出工程化建議。3.3 核心抽象在動(dòng)手寫代碼前先明確三個(gè)核心抽象RouteRule路由規(guī)則。它定義了“什么條件的請(qǐng)求該轉(zhuǎn)發(fā)給誰”包含關(guān)鍵詞、目標(biāo)智能體、優(yōu)先級(jí)、降級(jí)目標(biāo)等信息。AgentBackend智能體后端。它描述了一個(gè)可調(diào)用的智能體或模型服務(wù)包含名稱、調(diào)用端點(diǎn)、模型名、權(quán)重、可用狀態(tài)等信息。ExperientialRouter路由器核心。它負(fù)責(zé)解析請(qǐng)求匹配規(guī)則執(zhí)行轉(zhuǎn)發(fā)并在失敗時(shí)觸發(fā)降級(jí)。這三個(gè)抽象合在一起就構(gòu)成了一個(gè)最簡(jiǎn)智能體網(wǎng)關(guān)的骨架。4. 手寫一個(gè)最小可運(yùn)行的路由網(wǎng)關(guān)4.1 定義數(shù)據(jù)模型我們用 Pydantic 定義請(qǐng)求和路由相關(guān)的模型。打開main.py寫入下面的代碼# 文件路徑main.py from typing import List, Optional from pydantic import BaseModel class RouteRule(BaseModel): 路由規(guī)則 - keywords: 命中關(guān)鍵詞列表 - target: 目標(biāo)智能體名稱 - priority: 優(yōu)先級(jí)數(shù)值越小越優(yōu)先 - fallback: 主目標(biāo)不可用時(shí)的備用智能體 name: str keywords: List[str] target: str priority: int 10 fallback: Optional[str] None class AgentBackend(BaseModel): 智能體后端描述 實(shí)際項(xiàng)目中這里會(huì)包含 endpoint、api_key、model 等字段 示例中只保留核心信息 name: str type: str description: str enabled: bool True class GatewayRequest(BaseModel): 網(wǎng)關(guān)統(tǒng)一入?yún)?- user_id: 用戶標(biāo)識(shí)可用于權(quán)限控制 - query: 用戶輸入的自然語言 - session_id: 會(huì)話標(biāo)識(shí)可選 user_id: str query: str session_id: Optional[str] None class RouteResult(BaseModel): 路由結(jié)果返回給上層業(yè)務(wù) request_id: str query: str matched_rule: str target_agent: str reason: str response: str這些模型把“規(guī)則、后端、請(qǐng)求、結(jié)果”統(tǒng)一結(jié)構(gòu)化。Pydantic 在這里有兩個(gè)作用一是定義數(shù)據(jù)約束二是讓后續(xù)代碼可以通過類型提示獲得更好的 IDE 支持。4.2 實(shí)現(xiàn)路由匹配邏輯路由匹配是整個(gè)網(wǎng)關(guān)的核心。最簡(jiǎn)單、也最容易理解的匹配方式是“關(guān)鍵詞打分”遍歷所有規(guī)則統(tǒng)計(jì)每條規(guī)則命中了多少個(gè)關(guān)鍵詞得分最高且大于 0 的規(guī)則勝出。這里要注意優(yōu)先級(jí)的設(shè)計(jì)。當(dāng)兩條規(guī)則得分相同時(shí)需要有一個(gè)穩(wěn)定的規(guī)則來決定勝負(fù)。我們用priority字段表示優(yōu)先級(jí)數(shù)值越小越優(yōu)先這樣規(guī)則的設(shè)計(jì)者可以手動(dòng)把重要規(guī)則放在更靠前的位置。# 文件路徑main.py繼續(xù)追加 class ExperientialRouter: def __init__(self, rules: List[RouteRule], backends: List[AgentBackend]): # 按優(yōu)先級(jí)排序優(yōu)先級(jí)數(shù)值小的排在前面 self.rules sorted(rules, keylambda r: r.priority) self.backends {b.name: b for b in backends} def match_rule(self, query: str) - Optional[RouteRule]: 根據(jù)關(guān)鍵詞打分選擇最合適的規(guī)則 best_rule None best_score 0 for rule in self.rules: score sum(1 for kw in rule.keywords if kw in query) if score 0: continue # 如果打分相同因?yàn)?rules 已經(jīng)按優(yōu)先級(jí)排序 # 先被遍歷到的規(guī)則勝出 if score best_score: best_score score best_rule rule return best_rule if best_score 0 else None def get_available_backend(self, name: str) - Optional[AgentBackend]: 檢查后端是否存在且可用 if name is None: return None backend self.backends.get(name) if backend and backend.enabled: return backend return None def route(self, req: GatewayRequest) - RouteResult: rule self.match_rule(req.query) if rule is None: return RouteResult( request_id, queryreq.query, matched_ruledefault, target_agentdefault_agent, reason未命中任何規(guī)則使用默認(rèn)智能體, response沒有找到匹配的智能體請(qǐng)稍后再試。, ) # 主后端優(yōu)先 backend self.get_available_backend(rule.target) reason f命中規(guī)則 {rule.name}主目標(biāo) {rule.target} # 主后端不可用時(shí)觸發(fā)降級(jí) if backend is None: fallback_backend self.get_available_backend(rule.fallback) if fallback_backend is not None: backend fallback_backend reason f主目標(biāo) {rule.target} 不可用降級(jí)到 {rule.fallback} else: return RouteResult( request_id, queryreq.query, matched_rulerule.name, target_agentnone, reason主目標(biāo)與備用目標(biāo)均不可用, response服務(wù)暫不可用請(qǐng)稍后再試。, ) # 模擬調(diào)用后端智能體 response self.call_backend(backend, req.query) return RouteResult( request_iddemo-request-id, queryreq.query, matched_rulerule.name, target_agentbackend.name, reasonreason, responseresponse, ) def call_backend(self, backend: AgentBackend, query: str) - str: 模擬調(diào)用后端智能體。 真實(shí)場(chǎng)景中這里會(huì)根據(jù) backend.type 調(diào)用不同的 SDK 或 HTTP 接口。 return f[{backend.name}] 收到請(qǐng)求{query}在這個(gè)實(shí)現(xiàn)里call_backend是留給你擴(kuò)展的接口。真實(shí)項(xiàng)目中AgentBackend會(huì)包含endpoint和api_key字段call_backend會(huì)發(fā)起真正的 HTTP 調(diào)用或者調(diào)用第三方 SDK。這里的模擬只是為了把路由鏈路跑通。4.3 準(zhǔn)備路由規(guī)則和后端配置為了便于演示我們把規(guī)則寫成 Python 列表。實(shí)際項(xiàng)目中更推薦把規(guī)則放到 JSON 或 YAML 文件里方便動(dòng)態(tài)修改。# 文件路徑main.py繼續(xù)追加 def load_rules(): return [ RouteRule( namecustomer_service, keywords[退貨, 物流, 訂單, 客服, 發(fā)票], targetcustomer_service_agent, priority1, fallbackcommon_agent, ), RouteRule( namedata_analysis, keywords[銷售, 數(shù)據(jù), 報(bào)表, 趨勢(shì), 統(tǒng)計(jì), 分析], targetdata_analysis_agent, priority2, fallbackcommon_agent, ), RouteRule( namecontent_creation, keywords[文案, 標(biāo)題, 廣告, 產(chǎn)品介紹, 宣傳語], targetcontent_agent, priority3, fallbackcommon_agent, ), ] def load_backends(): return [ AgentBackend(namecustomer_service_agent, typedify, description客服智能體, enabledTrue), AgentBackend(namedata_analysis_agent, typeself_hosted, description數(shù)據(jù)分析智能體, enabledTrue), AgentBackend(namecontent_agent, typeopenai_compatible, description內(nèi)容創(chuàng)作智能體, enabledTrue), AgentBackend(namecommon_agent, typeopenai_compatible, description通用兜底智能體, enabledTrue), ]這里的關(guān)鍵詞是中文的因?yàn)閷?shí)際業(yè)務(wù)場(chǎng)景中中文關(guān)鍵詞匹配是最直觀、最容易理解的路由方式之一。你也可以把它換成拼音、英文或者自定義標(biāo)簽。4.4 暴露 HTTP 接口網(wǎng)關(guān)的核心接口是一個(gè)統(tǒng)一的聊天接口。上層業(yè)務(wù)不管背后是哪個(gè)智能體都往這個(gè)接口發(fā)請(qǐng)求。# 文件路徑main.py繼續(xù)追加 import uuid import time from fastapi import FastAPI, HTTPException app FastAPI(titleExperiential Agent Gateway Demo, version0.1.0) router ExperientialRouter(load_rules(), load_backends()) app.post(/v1/chat, response_modelRouteResult) async def chat(req: GatewayRequest): start time.time() result router.route(req) result.request_id str(uuid.uuid4()) result.reason f耗時(shí) {(time.time() - start) * 1000:.1f}ms return result app.get(/health) async def health(): return {status: ok}/v1/chat接口接收統(tǒng)一請(qǐng)求體返回路由結(jié)果。/health接口用于健康檢查。這樣一個(gè)最簡(jiǎn)單的智能體網(wǎng)關(guān)就成型了。4.5 運(yùn)行與驗(yàn)證在項(xiàng)目目錄下執(zhí)行uvicorn main:app --reload --port 8000啟動(dòng)成功后可以用 curl 測(cè)試curl -X POST http://127.0.0.1:8000/v1/chat \ -H Content-Type: application/json \ -d {user_id: u001, query: 我想查一下我的訂單什么時(shí)候發(fā)貨}預(yù)期返回結(jié)果大致如下{ request_id: xxxx, query: 我想查一下我的訂單什么時(shí)候發(fā)貨, matched_rule: customer_service, target_agent: customer_service_agent, reason: 命中規(guī)則 customer_service主目標(biāo) customer_service_agent耗時(shí) xx ms, response: [customer_service_agent] 收到請(qǐng)求我想查一下我的訂單什么時(shí)候發(fā)貨 }再試一個(gè)數(shù)據(jù)分析的場(chǎng)景curl -X POST http://127.0.0.1:8000/v1/chat \ -H Content-Type: application/json \ -d {user_id: u001, query: 這個(gè)月的銷售數(shù)據(jù)幫我分析一下}請(qǐng)求會(huì)被路由到data_analysis_agent。可以看到同樣是/v1/chat接口網(wǎng)關(guān)根據(jù)不同的輸入文本把它們分發(fā)到了不同的后端智能體。5. 路由策略進(jìn)階上面的示例用了最簡(jiǎn)單的關(guān)鍵詞匹配。實(shí)際生產(chǎn)環(huán)境中Experiential 這類智能體路由器通常需要支持更豐富的路由策略。5.1 關(guān)鍵詞與意圖匹配關(guān)鍵詞匹配的優(yōu)點(diǎn)是簡(jiǎn)單、可解釋、易調(diào)試。你可以明確知道“為什么這條請(qǐng)求走了這個(gè)智能體”。但它也有明顯缺點(diǎn)用戶表達(dá)方式千變?nèi)f化同一種意圖可能有幾十種說法關(guān)鍵詞列表很難枚舉完整。改進(jìn)方案包括維護(hù)同義詞和近義詞表比如“退”“換”“售后”都?xì)w入售后意圖使用正則表達(dá)式匹配更復(fù)雜的模式比如訂單號(hào)格式、日期格式接入輕量級(jí)意圖分類模型把用戶輸入分類到預(yù)定義意圖再做路由。在實(shí)際項(xiàng)目中我傾向于先用關(guān)鍵詞規(guī)則把大部分高頻場(chǎng)景覆蓋掉再逐步引入模型分類這樣既保證了可解釋性又能提高復(fù)雜場(chǎng)景的覆蓋率。5.2 模型能力與成本路由當(dāng)目標(biāo)不是“不同的智能體”而是“不同的模型”時(shí)路由規(guī)則會(huì)變得更細(xì)。比如{ name: simple_task, condition: 任務(wù)類型為簡(jiǎn)單問答, target_model: lightweight-model, cost_per_1k_tokens: 0.001 }你可以按下面幾個(gè)維度設(shè)計(jì)模型路由策略維度示例任務(wù)復(fù)雜度簡(jiǎn)單分類用小模型復(fù)雜推理用大模型上下文長度長文檔用支持長上下文的模型數(shù)據(jù)敏感級(jí)別內(nèi)部數(shù)據(jù)只能走私有化部署的模型成本預(yù)算低預(yù)算場(chǎng)景優(yōu)先使用價(jià)格更低的模型這里的核心思想是把“成本和能力的權(quán)衡”下沉到網(wǎng)關(guān)層讓業(yè)務(wù)方不需要關(guān)心模型價(jià)格。5.3 加權(quán)與故障轉(zhuǎn)移網(wǎng)絡(luò)路由器支持負(fù)載均衡智能體路由器同樣需要。你可以給多個(gè)后端配置權(quán)重class AgentBackend(BaseModel): name: str type: str weight: int 1 # 權(quán)重越高被選中的概率越大當(dāng)同一個(gè)智能體有多套部署實(shí)例時(shí)路由器可以根據(jù)權(quán)重把流量分?jǐn)偟讲煌瑢?shí)例上。同時(shí)結(jié)合健康檢查機(jī)制當(dāng)某個(gè)實(shí)例連續(xù)失敗多次后把它臨時(shí)標(biāo)記為不可用流量自動(dòng)切換到其他實(shí)例。故障轉(zhuǎn)移的順序建議是同集群的其他實(shí)例同供應(yīng)商的其他模型其他供應(yīng)商的等價(jià)模型兜底通用模型。這樣降級(jí)是分層的既能保證可用性又不會(huì)在降級(jí)時(shí)直接犧牲太多效果。5.4 動(dòng)態(tài)更新規(guī)則規(guī)則配置如果寫死在代碼里每次調(diào)整都需要重新發(fā)布服務(wù)。生產(chǎn)環(huán)境更推薦把規(guī)則放到配置文件或配置中心讓路由器支持熱加載。思路是ExperientialRouter保存一份規(guī)則快照后臺(tái)線程或定時(shí)任務(wù)定期讀取最新配置比對(duì)版本號(hào)發(fā)生變化時(shí)自動(dòng)替換。如果規(guī)則變化頻繁你需要重點(diǎn)關(guān)注兩個(gè)問題規(guī)則版本的管理每條規(guī)則建議帶version或updated_at字段路由結(jié)果的可追溯性每個(gè)請(qǐng)求必須記錄命中規(guī)則和版本號(hào)方便排查問題。6. 常見問題與排查思路在實(shí)際開發(fā)中智能體網(wǎng)關(guān)路由器的常見問題主要集中在規(guī)則匹配、后端調(diào)用和配置管理三個(gè)方面。問題現(xiàn)象常見原因解決思路請(qǐng)求都落到了默認(rèn)智能體關(guān)鍵詞覆蓋不全或用戶輸入與關(guān)鍵詞差異太大檢查規(guī)則關(guān)鍵詞是否包含常見說法增加同義詞或引入意圖分類多條規(guī)則都能命中但結(jié)果不穩(wěn)定規(guī)則優(yōu)先級(jí)設(shè)計(jì)不合理打分相同時(shí)順序不確定檢查 priority 設(shè)置明確打分相同時(shí)的決勝規(guī)則主智能體不可用時(shí)沒有自動(dòng)切換沒有配置 fallback或 get_available_backend 檢查邏輯不完整為高可用場(chǎng)景配置備用目標(biāo)并測(cè)試降級(jí)鏈路網(wǎng)關(guān)接口響應(yīng)很慢后端調(diào)用是同步阻塞的且超時(shí)時(shí)間設(shè)置過長在 call_backend 中設(shè)置合理的超時(shí)使用異步調(diào)用修改規(guī)則后不生效規(guī)則列表被加載到內(nèi)存后沒有熱更新增加配置版本檢查和定時(shí)刷新機(jī)制路由結(jié)果無法追溯日志沒有記錄規(guī)則命中和請(qǐng)求響應(yīng)信息為每個(gè)請(qǐng)求生成 request_id記錄 matched_rule、target_agent、耗時(shí)下游智能體鑒權(quán)失敗網(wǎng)關(guān)層沒有正確傳遞或管理下游 API Key把鑒權(quán)配置統(tǒng)一放到網(wǎng)關(guān)層按后端類型分別處理這里要特別提醒一個(gè)容易踩坑的點(diǎn)關(guān)鍵詞匹配的優(yōu)先級(jí)不能太復(fù)雜。不要設(shè)計(jì)超過三個(gè)維度的規(guī)則判斷條件否則規(guī)則之間的相互影響會(huì)非常難排查。先用“關(guān)鍵詞 優(yōu)先級(jí) 降級(jí)目標(biāo)”跑通主鏈路再逐步疊加語義分類、權(quán)重負(fù)載等高級(jí)特性。如果你遇到“網(wǎng)關(guān)里看著規(guī)則沒問題但線上就是不走預(yù)期路由”的情況排查順序建議如下查看請(qǐng)求日志確認(rèn)query原樣進(jìn)入網(wǎng)關(guān)確認(rèn)規(guī)則匹配結(jié)果看命中了哪條規(guī)則、打分是多少確認(rèn)目標(biāo)后端在backends中是否存在且 enabled 為 true確認(rèn)下游調(diào)用是否成功如果失敗是否觸發(fā)了 fallback確認(rèn)返回結(jié)果中reason字段的描述判斷問題出在路由層還是后端層。7. 最佳實(shí)踐與工程建議7.1 配置與代碼分離路由規(guī)則和智能體后端信息不應(yīng)該硬編碼在 Python 文件中。建議使用 JSON 或 YAML 文件單獨(dú)管理部署時(shí)通過環(huán)境變量指定配置文件路徑。示例的rules.json可以這樣組織{ version: 20250301, rules: [ { name: customer_service, keywords: [退貨, 物流, 訂單, 客服], target: customer_service_agent, priority: 1, fallback: common_agent } ], backends: [ { name: customer_service_agent, type: dify, endpoint: https://your-dify-endpoint.example.com, enabled: true } ] }這樣做的好處是規(guī)則調(diào)整不需要發(fā)版運(yùn)維和算法同學(xué)可以直接修改配置文件。配合配置中心可以實(shí)現(xiàn)規(guī)則的熱更新和灰度發(fā)布。7.2 可觀測(cè)性網(wǎng)關(guān)是流量的必經(jīng)之路它是最適合做觀測(cè)的位置。每個(gè)請(qǐng)求建議記錄以下信息請(qǐng)求 ID 和用戶 ID原始輸入文本命中的路由規(guī)則和版本目標(biāo)智能體或模型響應(yīng)耗時(shí)和 Token 消耗是否觸發(fā)了降級(jí)。日志格式建議統(tǒng)一為 JSON方便接入日志平臺(tái){ request_id: xxx, user_id: u001, query: 我的訂單什么時(shí)候發(fā)貨, matched_rule: customer_service, target_agent: customer_service_agent, fallback_used: false, latency_ms: 320, tokens: 128 }有了這些數(shù)據(jù)你才能回答“每個(gè)智能體的調(diào)用量是多少”“哪個(gè)智能體經(jīng)常超時(shí)”“最近一次降級(jí)發(fā)生在什么時(shí)候”這類問題。7.3 安全邊界智能體網(wǎng)關(guān)處于業(yè)務(wù)系統(tǒng)和下游 AI 服務(wù)之間安全設(shè)計(jì)需要覆蓋兩條鏈路。上游鏈路要防止未授權(quán)訪問。網(wǎng)關(guān)接口要做身份認(rèn)證至少要求每個(gè)請(qǐng)求攜帶有效 Token并基于user_id做權(quán)限控制。不同用戶能訪問的智能體可能不同比如管理員賬號(hào)才能路由到數(shù)據(jù)分析智能體。下游鏈路要管理好模型和智能體的憑據(jù)。不要把 API Key 直接暴露給前端所有下游鑒權(quán)信息都應(yīng)該保存在服務(wù)端配置中。同時(shí)對(duì)用戶輸入做敏感信息過濾避免把手機(jī)號(hào)、身份證號(hào)等隱私數(shù)據(jù)直接傳給未經(jīng)授權(quán)的模型服務(wù)。還要注意 Prompt 注入風(fēng)險(xiǎn)。用戶可能通過輸入文本嘗試讓智能體突破系統(tǒng)提示的限制網(wǎng)關(guān)層可以做初步的關(guān)鍵詞攔截和敏感操作確認(rèn)機(jī)制但更完善的治理需要和智能體本身的 Prompt 防護(hù)配合。7.4 性能與成本網(wǎng)關(guān)層不建議做太多耗時(shí)操作。如果要用語義模型做意圖分類建議單獨(dú)部署一個(gè)輕量級(jí)分類服務(wù)路由網(wǎng)關(guān)通過 RPC 調(diào)用它而不是在網(wǎng)關(guān)進(jìn)程內(nèi)加載大模型。否則所有請(qǐng)求的延遲都會(huì)被拖慢。對(duì)于超時(shí)控制建議給下游調(diào)用設(shè)置默認(rèn)超時(shí)時(shí)間比如 10 秒超過閾值直接切換備用目標(biāo)。不要無限制地等待下游響應(yīng)否則網(wǎng)關(guān)的線程池會(huì)被占滿整體吞吐量會(huì)驟降。成本控制方面網(wǎng)關(guān)是統(tǒng)計(jì) Token 消耗的最佳節(jié)點(diǎn)。每個(gè)路由結(jié)果都記錄使用的模型和 Token 數(shù)匯總后可以看到每個(gè)業(yè)務(wù)線、每個(gè)用戶的 AI 調(diào)用成本。這一步對(duì)于成本分?jǐn)偤皖A(yù)算控制非常有用。8. 總結(jié)與下一步學(xué)習(xí)路線通過上面的內(nèi)容我們已經(jīng)把“智能體工作流網(wǎng)關(guān)與路由器”這個(gè)方向的核心邏輯拆解清楚了它借鑒了網(wǎng)絡(luò)層面的網(wǎng)關(guān)與路由器思想在 AI 應(yīng)用層提供一個(gè)統(tǒng)一入口按照規(guī)則把請(qǐng)求分發(fā)給合適的智能體或模型并在異常時(shí)自動(dòng)降級(jí)。我們自己動(dòng)手實(shí)現(xiàn)的最小版本里包含了一個(gè)完整的路由鏈路用戶請(qǐng)求 - 規(guī)則匹配 - 后端選擇 - 調(diào)用結(jié)果返回。雖然代碼很簡(jiǎn)單但它包含了 Experiential 這類項(xiàng)目的核心骨架。你接下來可以在這個(gè)基礎(chǔ)上做幾個(gè)方向的擴(kuò)展第一步把配置文件外部化讓規(guī)則支持熱更新。這樣你就能在不停機(jī)的情況下調(diào)整路由策略。第二步把call_backend改造成真實(shí)的 HTTP 調(diào)用對(duì)接 Dify、開源模型本地部署或者模型供應(yīng)商的標(biāo)準(zhǔn) API。這一步做完你的網(wǎng)關(guān)就可以接入真實(shí)業(yè)務(wù)了。第三步加入可觀測(cè)性為每次請(qǐng)求輸出結(jié)構(gòu)化的 JSON 日志記錄命中規(guī)則、目標(biāo)后端、耗時(shí)和 Token 消耗。第四步學(xué)習(xí)語義路由。當(dāng)你發(fā)現(xiàn)關(guān)鍵詞規(guī)則覆蓋不了越來越多樣的用戶表達(dá)時(shí)可以引入輕量級(jí)的意圖分類模型讓路由器具備理解能力。最后建議你保持關(guān)注 Experiential 這類開源項(xiàng)目的更新。智能體網(wǎng)關(guān)與路由器還是一個(gè)快速演進(jìn)的領(lǐng)域不同項(xiàng)目對(duì)路由策略、插件機(jī)制、工作流編排的取舍各不相同。動(dòng)手跑一個(gè)最小實(shí)現(xiàn)再讀開源項(xiàng)目的源碼會(huì)比只看文檔理解得深得多。如果這篇文章對(duì)你有幫助可以收藏備用后面需要做模型統(tǒng)一接入或智能體編排時(shí)拿出來照著搭建。