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