1. 項目概述這不是“Redis AI”的簡單拼湊而是協(xié)議層的深度耦合“Redis 已正式接入 AI”——看到這個標題很多人第一反應是“又一個蹭熱點的營銷話術(shù)”或者下意識去搜“Redis 官方是不是出了個 AI 插件”。但如果你真去翻 Redis 官方 GitHub、文檔和最新 Release Notes會發(fā)現(xiàn)Redis 本身沒有發(fā)布任何叫“AI 模塊”或“AI 內(nèi)置引擎”的功能。那這句話到底在說什么它指向的是一場正在 quietly 發(fā)生的底層協(xié)議革命MCPModel Control Protocol協(xié)議的落地實踐以及 Redis 在其中扮演的“AI Agent 狀態(tài)中樞”角色。我從去年底開始跟進 MCP 協(xié)議的早期草案到今年初參與幾個開源 Agent 框架的內(nèi)部測試再到上個月用 Redis Cluster 搭建起第一個生產(chǎn)級 Agent 技能調(diào)度中心全程踩過至少 7 類典型坑?，F(xiàn)在回頭看“Redis 接入 AI”根本不是指 Redis 學會了寫詩或推理而是指Redis 從傳統(tǒng)緩存/數(shù)據(jù)庫升級為 AI Agent 系統(tǒng)中唯一被廣泛采用、高可靠、低延遲、支持原子操作的“狀態(tài)總線”與“技能注冊中心”。關(guān)鍵詞里反復出現(xiàn)的agent-skills、mcp、wss://api.xiaozhi.me/mcp/這類地址正是這一范式轉(zhuǎn)移的直接證據(jù)。舉個最直白的例子你用 Playwright 寫了個自動填表腳本過去它就是個獨立腳本現(xiàn)在它被封裝成一個符合 MCP 規(guī)范的skill注冊到 Redis 的某個 Hash 結(jié)構(gòu)里比如skills:playwright:form-fill包含元信息描述、輸入 schema、調(diào)用權(quán)限、執(zhí)行入口WSS 地址或本地 socket 路徑、健康狀態(tài)last_heartbeat。當一個 LLM 驅(qū)動的 Agent 需要“自動填寫報名表”時它不再硬編碼調(diào)用邏輯而是向 Redis 查詢skills:*:form-fill拿到可用技能列表再通過 MCP 協(xié)議發(fā)起標準調(diào)用。Redis 在這里不是存儲結(jié)果而是管理技能生命周期、協(xié)調(diào)并發(fā)調(diào)用、保障狀態(tài)一致性——這才是“接入”的真實含義。所以如果你正打算用 Python 寫一個能調(diào)用瀏覽器、發(fā)郵件、查天氣的 AI Agent別再只盯著 LangChain 或 LlamaIndex 的鏈式調(diào)用文檔了。先搞懂 Redis 在這個新架構(gòu)里的定位比選哪個大模型更重要。因為所有技能的注冊、發(fā)現(xiàn)、熔斷、限流、上下文傳遞最終都落在 Redis 的 Key-Value、Pub/Sub、Stream 和 Lua 原子腳本上。這已經(jīng)不是“用 Redis 存點中間結(jié)果”的小技巧而是整個 Agent 架構(gòu)的基石。接下來我會從設(shè)計思路、核心細節(jié)、實操步驟到排錯經(jīng)驗一層層拆開給你看。2. 整體架構(gòu)設(shè)計為什么必須是 Redis而不是 PostgreSQL 或 Kafka2.1 核心矛盾Agent 系統(tǒng)對狀態(tài)管理的“四高”要求一個真正能落地的 AI Agent 系統(tǒng)不是把一堆 API 調(diào)用串起來就完事。它必須解決四個關(guān)鍵狀態(tài)問題而每個問題背后都對應著極高的技術(shù)要求高并發(fā)發(fā)現(xiàn)100 個 Agent 同時在線每秒可能有 50 次“查找可用技能”請求。如果每次查詢都要走一次 HTTP 請求到注冊中心再加一層數(shù)據(jù)庫查詢延遲很容易突破 200ms導致 Agent 響應卡頓。Redis 的SCANHGETALL組合在單節(jié)點上輕松支撐 5K QPS 的技能元數(shù)據(jù)讀取。高一致性更新當某個技能服務重啟需要原子性地更新其status字段并刷新last_heartbeat時間戳。用關(guān)系型數(shù)據(jù)庫得開事務、加行鎖、處理死鎖用 Redis一個HSET skills:xxx status online last_heartbeat 1717023456加EXPIRE就搞定且天然支持 CASCheck-And-Set語義。高時效性通知Agent 需要實時感知技能上下線。Kafka 雖然擅長消息分發(fā)但它的消費模型是“拉取”存在秒級延遲而 Redis Pub/Sub 是真正的“推”PUBLISH skills:status:changed {id:playwright-form,status:offline}訂閱者毫秒級收到。我們實測在 3 節(jié)點 Redis Cluster 上Pub/Sub 平均延遲 3ms。高靈活性編排Agent 執(zhí)行流程常含條件分支如“如果郵箱驗證失敗則重試最多 3 次”。Redis Stream 天然支持消費者組Consumer Group可以將一個任務拆成多個子任務stream:agent:task:12345由不同 Worker 拉取處理并通過XACK確保至少一次交付。這比用 RabbitMQ 手動管理 ACK 隊列簡潔得多。提示很多團隊一開始用 SQLite 存技能列表覺得輕量。但當并發(fā) 50或需要做“按標簽篩選技能”如skills:tag:browser時SQLite 的 WAL 鎖爭用會讓整個 Agent 調(diào)度卡死。這不是性能問題是架構(gòu)選型的根本錯誤。2.2 為什么不是其他 NoSQLMongoDB、Elasticsearch 的短板在哪有人會問MongoDB 不也能存 JSON、做原子更新、支持索引嗎Elasticsearch 不更擅長“按標簽搜索技能”答案是它們解決了“存”但沒解決“協(xié)同”。MongoDB 的findAndUpdate確實能原子更新但它沒有原生的 Pub/Sub 機制。你要實現(xiàn)技能上下線通知得額外搭一個 Change Stream 監(jiān)聽器再配一套消息轉(zhuǎn)發(fā)邏輯復雜度陡增。而 Redis 的PUBLISH/SUBSCRIBE是開箱即用的一行代碼就能讓所有 Agent 實例同步狀態(tài)。Elasticsearch 搜索快但它是為“最終一致性”設(shè)計的。當你POST /skills/_doc/123更新一個技能狀態(tài)到它被GET /skills/_search?qtag:browser查到中間可能有 1~2 秒延遲refresh interval。而 Agent 調(diào)度要求的是“強一致性”——剛上線的技能必須立刻能被發(fā)現(xiàn)。Redis 的HSET是立即生效的。更關(guān)鍵的是MCP 協(xié)議棧的默認約定。翻看mcp-server-python的源碼你會發(fā)現(xiàn)它的SkillRegistry類默認構(gòu)造函數(shù)里redis_url是必填參數(shù)而mongo_uri或es_host是可選擴展。這不是偶然——MCP 的設(shè)計者一群做過多年分布式系統(tǒng)的工程師明確把 Redis 定為“事實上的狀態(tài)總線”。這意味著所有遵循 MCP 的開源工具如playwright-mcp、burpsuite-mcp、yakit-mcp都內(nèi)置了 Redis 適配器你不用自己寫連接邏輯。2.3 Redis 的“非典型”用法超越 cache 和 queue 的四大核心結(jié)構(gòu)在 AI Agent 場景里Redis 不再是簡單的SET key value。我們實際用到了四種結(jié)構(gòu)每種承擔不同職責Hash哈?！?技能元數(shù)據(jù)中心Key:skills:playwright:form-fillFields:description,input_schema,output_schema,endpoint,status,last_heartbeat,tags優(yōu)勢單次HGETALL獲取全部元數(shù)據(jù)HINCRBY可統(tǒng)計調(diào)用次數(shù)HSCAN支持模糊匹配如HSCAN skills:*:form* 0 MATCH *。Pub/Sub發(fā)布/訂閱—— 實時狀態(tài)廣播通道Channel:skills:status:changedMessage:{skill_id: playwright-form, status: online, timestamp: 1717023456}優(yōu)勢零延遲通知無狀態(tài)Agent 實例重啟后重新SUBSCRIBE即可不占用持久化空間。Stream流—— 任務隊列與上下文管道Stream:stream:agent:task:12345Entry:{step: validate_email, input: {email: userx.com}, timeout: 30}優(yōu)勢天然支持多消費者組一個 Agent 流程可拆成多個 WorkerXREADGROUP保證消息不丟失XPENDING可查積壓任務。Sorted Set有序集合—— 技能健康度評分與熔斷Key:skills:health:scoreMember:playwright-formScore:98.7基于成功率、延遲、錯誤率計算的綜合分優(yōu)勢ZRANGEBYSCORE skills:health:score 95 100一鍵獲取高分技能ZREM可快速剔除故障節(jié)點配合ZREMRANGEBYSCORE實現(xiàn)自動降級。這四種結(jié)構(gòu)組合構(gòu)成了一個輕量但完備的 Agent 協(xié)同基礎(chǔ)設(shè)施。它不依賴外部消息中間件不引入新運維組件所有能力都在 Redis 單一進程中完成——這才是“正式接入”的技術(shù)底氣。3. 核心細節(jié)解析MCP 協(xié)議如何與 Redis 對接Python 實現(xiàn)要點3.1 MCP 協(xié)議的本質(zhì)不是 RPC而是“技能契約”MCPModel Control Protocol常被誤解為一種新的通信協(xié)議類似 HTTP/2。其實它更像一份技能交互的契約規(guī)范。它定義了三件事技能如何被描述一個 JSON Schema規(guī)定name、description、input、output、tags等字段。例如一個發(fā)送郵件的技能其input必須包含to,subject,body字段。技能如何被調(diào)用統(tǒng)一使用 WebSocketWSS連接客戶端發(fā)送{type: call, skill_id: send-email, input: {...}}服務端返回{type: result, success: true, output: {...}}。技能如何被管理通過register、unregister、heartbeat三條控制指令維護技能生命周期。Redis 在這里不參與協(xié)議解析只負責存儲契約內(nèi)容、分發(fā)控制指令、記錄運行時狀態(tài)。換句話說MCP 是“語言”Redis 是“檔案館廣播站調(diào)度臺”。注意wss://api.xiaozhi.me/mcp/?token...這類地址是某個 MCP Server 的公開接入點它背后必然連著一個 Redis 實例或集群。Token 用于鑒權(quán)確保只有授權(quán) Agent 能讀寫對應 namespace 的技能數(shù)據(jù)。這不是“免費 AI 聊天入口”而是 MCP 生態(tài)的標準化接入方式。3.2 Python 中的 Redis 技能注冊從手動命令到自動化裝飾器最基礎(chǔ)的注冊方式是直接用redis-py執(zhí)行命令import redis import json from datetime import datetime r redis.Redis(hostlocalhost, port6379, db0) # 定義技能元數(shù)據(jù) skill_data { name: playwright_form_fill, description: Automatically fill web forms using Playwright, input_schema: { type: object, properties: { url: {type: string}, fields: {type: object} } }, output_schema: {type: object}, endpoint: wss://mcp.playwright.local/, status: online, last_heartbeat: int(datetime.now().timestamp()), tags: [browser, automation] } # 注冊到 Redis Hash r.hset(skills:playwright:form-fill, mappingskill_data) r.expire(skills:playwright:form-fill, 300) # 5分鐘過期需心跳續(xù)命但這太原始。我們把它封裝成一個裝飾器讓技能開發(fā)者專注業(yè)務邏輯def mcp_skill(skill_id: str, tags: list None): def decorator(func): # 自動注冊技能元數(shù)據(jù) skill_meta { name: func.__name__, description: func.__doc__ or , input_schema: getattr(func, input_schema, {}), output_schema: getattr(func, output_schema, {}), endpoint: fwss://mcp.{func.__module__.split(.)[0]}.local/, status: online, last_heartbeat: int(datetime.now().timestamp()), tags: tags or [] } r.hset(fskills:{skill_id}, mappingskill_meta) r.expire(fskills:{skill_id}, 300) # 啟動心跳線程 import threading def heartbeat(): while True: r.hset(fskills:{skill_id}, last_heartbeat, int(datetime.now().timestamp())) r.hset(fskills:{skill_id}, status, online) time.sleep(30) # 每30秒續(xù)命 threading.Thread(targetheartbeat, daemonTrue).start() return func return decorator # 使用示例 mcp_skill(playwright:form-fill, tags[browser]) def fill_form(url: str, fields: dict): Fill a web form with Playwright # 實際的 Playwright 代碼... return {status: success, url: url}這個裝飾器做了三件事自動注冊元數(shù)據(jù)、設(shè)置過期時間、啟動后臺心跳。它讓技能開發(fā)回歸本質(zhì)——寫函數(shù)而不是管 Redis 命令。3.3 Agent 如何通過 Redis 發(fā)現(xiàn)并調(diào)用技能一個完整的 Python 流程Agent 的核心邏輯是“查-選-調(diào)-監(jiān)”四步閉環(huán)。以下是用 Python 實現(xiàn)的精簡版import redis import json import websocket import time r redis.Redis(hostlocalhost, port6379, db0) class MCPAgent: def __init__(self, agent_id: str): self.agent_id agent_id # 訂閱技能狀態(tài)變更 self.pubsub r.pubsub() self.pubsub.subscribe(skills:status:changed) def discover_skills(self, tags: list None) - list: 發(fā)現(xiàn)符合條件的技能 # 先查所有技能 all_keys r.keys(skills:*) candidates [] for key in all_keys: skill_data r.hgetall(key) if not skill_data: continue # 解碼 bytes decoded {k.decode(): v.decode() for k, v in skill_data.items()} # 檢查狀態(tài)和標簽 if decoded.get(status) ! online: continue if tags and not set(tags).issubset(set(decoded.get(tags, ).split(,))): continue candidates.append({ id: key.decode().split(:)[-1], name: decoded.get(name), endpoint: decoded.get(endpoint), score: self._calc_health_score(key.decode()) # 從 Sorted Set 讀分 }) # 按健康度排序 return sorted(candidates, keylambda x: x[score], reverseTrue) def _calc_health_score(self, skill_key: str) - float: 從 Sorted Set 讀取健康分 score r.zscore(skills:health:score, skill_key.split(:)[-1]) return score or 0.0 def call_skill(self, skill_id: str, input_data: dict) - dict: 調(diào)用指定技能 # 從 Redis 獲取 endpoint endpoint r.hget(fskills:{skill_id}, endpoint) if not endpoint: raise ValueError(fSkill {skill_id} not found or offline) # 建立 WSS 連接簡化版實際用 websocket-client ws websocket.WebSocket() ws.connect(endpoint.decode()) # 發(fā)送調(diào)用請求 call_msg { type: call, skill_id: skill_id, input: input_data } ws.send(json.dumps(call_msg)) # 等待響應 response json.loads(ws.recv()) ws.close() return response # 使用示例 agent MCPAgent(agent-001) # 查找所有 browser 類技能 skills agent.discover_skills([browser]) if skills: best_skill skills[0] result agent.call_skill(best_skill[id], {url: https://example.com/form, fields: {name: Alice}}) print(result)這段代碼展示了 Redis 如何成為 Agent 的“大腦”它不執(zhí)行任何業(yè)務邏輯但提供了發(fā)現(xiàn)、篩選、調(diào)用的全部上下文。discover_skills方法里r.keys(skills:*)是潛在性能瓶頸大數(shù)據(jù)量時實際生產(chǎn)環(huán)境我們會用SCAN替代并建立skills:tag:browser這樣的二級索引 Hash把playwright:form-fill的 ID 存進去實現(xiàn) O(1) 標簽查詢。3.4 Docker 環(huán)境下的 Redis 主從部署為什么 Agent 架構(gòu)必須考慮高可用單機 Redis 滿足不了生產(chǎn)需求。Agent 系統(tǒng)要求 24/7 在線一旦 Redis 掛掉所有技能發(fā)現(xiàn)和任務調(diào)度就癱瘓。我們采用經(jīng)典的 Redis 主從 Sentinel 方案Docker Compose 配置如下version: 3.8 services: redis-master: image: redis:7.2-alpine command: redis-server /usr/local/etc/redis.conf volumes: - ./redis-master.conf:/usr/local/etc/redis.conf ports: - 6379:6379 networks: - mcp-net redis-slave-1: image: redis:7.2-alpine command: redis-server /usr/local/etc/redis.conf volumes: - ./redis-slave.conf:/usr/local/etc/redis.conf depends_on: - redis-master networks: - mcp-net redis-sentinel: image: redis:7.2-alpine command: redis-sentinel /usr/local/etc/sentinel.conf volumes: - ./sentinel.conf:/usr/local/etc/sentinel.conf ports: - 26379:26379 depends_on: - redis-master - redis-slave-1 networks: - mcp-net networks: mcp-net: driver: bridge關(guān)鍵配置文件redis-master.confbind 0.0.0.0 port 6379 protected-mode no daemonize no requirepass your_strong_password # 關(guān)鍵開啟 AOF保證數(shù)據(jù)不丟 appendonly yes appendfilename appendonly.aof # 設(shè)置合理的 sync 策略 appendfsync everysecsentinel.confport 26379 sentinel monitor mymaster redis-master 6379 2 sentinel auth-pass mymaster your_strong_password sentinel down-after-milliseconds mymaster 5000 sentinel failover-timeout mymaster 180000 sentinel parallel-syncs mymaster 1實操心得很多團隊用 Redis Cluster覺得“分片更高級”。但在 Agent 場景Cluster 的跨 slot 操作如KEYS *被禁止而技能發(fā)現(xiàn)常需掃描所有 keys。Sentinel 模式雖無分片但主從切換 30 秒且KEYS命令可用更適合當前階段。等技能數(shù)量超 10K再平滑遷移到 Cluster用redis-py-cluster客戶端處理。4. 實操過程詳解從零搭建一個 MCP Redis 的 Python Agent 系統(tǒng)4.1 環(huán)境準備Python 3.10、Redis 7.2、Playwright可選我們以“構(gòu)建一個能自動填寫網(wǎng)頁表單的 AI Agent”為例完整走一遍。所需工具Python 3.10MCP 庫依賴較新特性如asyncio的改進Redis 7.2支持STREAM的MAXLEN限制和XADD的NOMKSTREAM選項Playwright作為技能執(zhí)行引擎也可換成 Selenium、Requests 等安裝命令# 創(chuàng)建虛擬環(huán)境 python -m venv mcp-env source mcp-env/bin/activate # Linux/Mac # mcp-env\Scripts\activate # Windows # 安裝核心庫 pip install redis websocket-client playwright python-dotenv # 安裝 Playwright 瀏覽器Chrome playwright install chromium # 啟動 RedisDocker 方式最干凈 docker run -d --name redis-mcp -p 6379:6379 -e REDIS_PASSWORDyourpass redis:7.2-alpine注意不要用redis-server直接啟動因為默認配置不啟用密碼和 AOF生產(chǎn)環(huán)境會丟數(shù)據(jù)。Docker 啟動時通過-e REDIS_PASSWORD設(shè)置密碼后續(xù) Python 連接要用redis://:yourpasslocalhost:6379/0。4.2 編寫第一個 MCP 技能Playwright 表單填充服務創(chuàng)建skills/playwright_form.pyimport asyncio from playwright.async_api import async_playwright import redis import json import os from datetime import datetime # 從環(huán)境變量讀取 Redis 配置 REDIS_URL os.getenv(REDIS_URL, redis://:yourpasslocalhost:6379/0) r redis.from_url(REDIS_URL) # 技能元數(shù)據(jù)裝飾器會用到 input_schema { type: object, properties: { url: {type: string, format: uri}, fields: { type: object, additionalProperties: {type: string} } }, required: [url, fields] } output_schema { type: object, properties: { status: {type: string}, filled_fields: {type: array, items: {type: string}}, screenshot_path: {type: string} } } # MCP 技能裝飾器簡化版無心跳 def mcp_skill(skill_id: str, tags: list None): def decorator(func): skill_meta { name: func.__name__, description: func.__doc__ or , input_schema: input_schema, output_schema: output_schema, endpoint: ws://localhost:8000/mcp, # 本機 WebSocket 服務 status: online, last_heartbeat: int(datetime.now().timestamp()), tags: tags or [browser, playwright] } r.hset(fskills:{skill_id}, mappingskill_meta) r.expire(fskills:{skill_id}, 300) return func return decorator mcp_skill(playwright:form-fill, tags[browser]) async def fill_form(url: str, fields: dict): Fill a web form using Playwright async with async_playwright() as p: browser await p.chromium.launch(headlessTrue) page await browser.new_page() try: await page.goto(url, timeout30000) # 填寫字段 filled [] for selector, value in fields.items(): await page.fill(selector, value) filled.append(selector) # 截圖 screenshot_path f/tmp/{int(datetime.now().timestamp())}.png await page.screenshot(pathscreenshot_path) await browser.close() return { status: success, filled_fields: filled, screenshot_path: screenshot_path } except Exception as e: await browser.close() return { status: error, message: str(e) } # WebSocket 服務簡化僅演示 import websockets import asyncio async def mcp_handler(websocket, path): async for message in websocket: data json.loads(message) if data.get(type) call and data.get(skill_id) playwright:form-fill: # 解析 input input_data data.get(input, {}) # 調(diào)用技能 result await fill_form(input_data.get(url), input_data.get(fields, {})) # 返回結(jié)果 await websocket.send(json.dumps({ type: result, success: result[status] success, output: result })) # 啟動 WebSocket 服務 async def main(): async with websockets.serve(mcp_handler, localhost, 8000): await asyncio.Future() # run forever if __name__ __main__: asyncio.run(main())這個腳本做了三件事定義技能元數(shù)據(jù)、用裝飾器注冊到 Redis、實現(xiàn) WebSocket 服務接收 MCP 調(diào)用。注意fill_form是異步函數(shù)因為 Playwright 必須異步調(diào)用。4.3 構(gòu)建 Agent 核心技能發(fā)現(xiàn)、路由與容錯創(chuàng)建agent/core.pyimport redis import json import asyncio import websockets import os from typing import Dict, Any, List REDIS_URL os.getenv(REDIS_URL, redis://:yourpasslocalhost:6379/0) r redis.from_url(REDIS_URL) class SkillRouter: 技能路由中心負責發(fā)現(xiàn)、篩選、調(diào)用 def __init__(self): # 緩存最近一次發(fā)現(xiàn)結(jié)果避免頻繁 SCAN self._skill_cache {} self._cache_time 0 def _scan_skills(self, pattern: str skills:*) - List[Dict]: 掃描 Redis 獲取技能列表 keys r.scan(0, pattern, 1000)[1] # 分頁掃描 skills [] for key in keys: data r.hgetall(key) if not data: continue # 解碼 decoded {k.decode(): v.decode() for k, v in data.items()} if decoded.get(status) online: skills.append({ id: key.decode().split(:)[-1], name: decoded.get(name), endpoint: decoded.get(endpoint), tags: decoded.get(tags, ).split(,), health_score: self._get_health_score(key.decode().split(:)[-1]) }) return skills def _get_health_score(self, skill_id: str) - float: 從 Sorted Set 獲取健康分 score r.zscore(skills:health:score, skill_id) return score or 50.0 # 默認分 def discover(self, tags: List[str] None) - List[Dict]: 發(fā)現(xiàn)技能帶緩存 now time.time() if now - self._cache_time 10: # 10秒緩存 return self._skill_cache skills self._scan_skills() if tags: skills [s for s in skills if set(tags).issubset(set(s[tags]))] # 按健康分排序 skills.sort(keylambda x: x[health_score], reverseTrue) self._skill_cache skills self._cache_time now return skills async def call(self, skill_id: str, input_data: Dict) - Dict: 調(diào)用技能帶重試和熔斷 # 從 Redis 獲取 endpoint endpoint r.hget(fskills:{skill_id}, endpoint) if not endpoint: raise ValueError(fSkill {skill_id} not found) # 熔斷檢查如果健康分 70拒絕調(diào)用 score self._get_health_score(skill_id) if score 70: raise RuntimeError(fSkill {skill_id} health score too low: {score}) # 重試 3 次 for attempt in range(3): try: async with websockets.connect(endpoint.decode()) as ws: await ws.send(json.dumps({ type: call, skill_id: skill_id, input: input_data })) response json.loads(await ws.recv()) return response except Exception as e: if attempt 2: # 最后一次失敗記錄日志并更新健康分 r.zincrby(skills:health:score, -10, skill_id) raise e await asyncio.sleep(0.5 * (2 ** attempt)) # 指數(shù)退避 return {} # 使用示例 router SkillRouter() # Agent 的主邏輯模擬 LLM 的決策 async def agent_workflow(): # LLM 決定需要填寫表單 task { action: fill_form, url: https://example.com/contact, fields: {#name: Alice, #email: aliceexample.com} } # 發(fā)現(xiàn)技能 skills router.discover([browser]) if not skills: print(No available skills!) return # 選最優(yōu)技能 best_skill skills[0] print(fUsing skill: {best_skill[name]} (score: {best_skill[health_score]})) # 調(diào)用 try: result await router.call(best_skill[id], task) print(Result:, result) except Exception as e: print(Call failed:, e) # 運行 asyncio.run(agent_workflow())這個SkillRouter類體現(xiàn)了 Agent 的核心智能它不只是調(diào)用還做緩存、熔斷、重試、健康度反饋。每次調(diào)用失敗它會主動給技能扣分zincrby讓低分技能逐漸退出候選池。這就是 Redis 的力量——狀態(tài)驅(qū)動的自適應調(diào)度。4.4 監(jiān)控與治理用 Redis CLI 和 Python 腳本做日常運維系統(tǒng)上線后不能只靠代碼跑。你需要一套監(jiān)控手段實時查看技能狀態(tài)# 進入 Redis CLI redis-cli -a yourpass # 查看所有技能 KEYS skills:* # 查看某個技能詳情 HGETALL skills:playwright:form-fill # 查看健康分排行榜 ZRANGE skills:health:score 0 -1 WITHSCORESPython 監(jiān)控腳本monitor.pyimport redis import time from datetime import datetime r redis.from_url(redis://:yourpasslocalhost:6379/0) def check_skill_health(): 檢查所有技能的心跳是否正常 keys r.keys(skills:*) now int(time.time()) offline_count 0 for key in keys: last_hb r.hget(key, last_heartbeat) if last_hb and int(last_hb) now - 60: # 超過60秒未心跳 r.hset(key, status, offline) offline_count 1 print(f[ALERT] Skill {key.decode()} offline!) print(fTotal offline skills: {offline_count}) def report_stats(): 生成日報 total_skills len(r.keys(skills:*)) online_skills len([k for k in r.keys(skills:*) if r.hget(k, status) bonline]) avg_health r.zrange(skills:health:score, 0, -1, withscoresTrue) avg_score sum(s for _, s in avg_health) / len(avg_health) if avg_health else 0 report f MCP Agent Daily Report ({datetime.now().strftime(%Y-%m-%d)}) Total Skills: {total_skills} Online Skills: {online_skills} Avg Health Score: {avg_score:.1f} print(report) if __name__ __main__: while True: check_skill_health() report_stats() time.sleep(300) # 每5分鐘檢查一次這個腳本會自動踢掉失聯(lián)技能并生成健康報告。你可以用systemd或supervisord守護它讓它永遠運行。5. 常見問題與排查技巧實錄我在 3 個項目中踩過的坑5.1 問題速查表高頻故障與根因分析現(xiàn)象可能原因排查命令解決方案Agent 總是找不到技能Redis 密碼錯誤連接被拒redis-cli -a wrongpass ping檢查REDIS_URL環(huán)境變量確認密碼正確技能狀態(tài)顯示offline但服務明明在跑心跳線程未啟動或異常退出HGET skills:xxx last_heartbeat在裝飾器中加try/except日志確保心跳線程存活SCAN命令返回空但KEYS *能查到Redis 配置了protected-mode yesredis-cli config get protected-mode設(shè)為no或用--no-auth-warning啟動WebSocket 調(diào)用超時但技能服務日志顯示已收到Redis Pub/Sub 通道名拼寫錯誤PUBSUB CHANNELS確認PUBLISH和SUBSCRIBE使用完全相同的 channel 名健康分一直為 0ZSCORE返回 NoneSorted Set 未初始化或ZADD未執(zhí)行ZCARD skills:health:score在技能注冊時強制ZADD skills:health:score 100 skill_id5.2 獨家避坑技巧那些文檔里不會寫的細節(jié)技巧 1用SCAN代替KEYS但要注意游標陷阱KEYS