1. 項(xiàng)目概述這不是“Redis AI”的簡(jiǎn)單拼湊而是協(xié)議層的深度耦合“Redis 已正式接入 AI”——看到這個(gè)標(biāo)題很多人第一反應(yīng)是“又一個(gè)蹭熱點(diǎn)的營(yíng)銷話術(shù)”或者下意識(shí)去搜“Redis 官方是不是出了個(gè) AI 插件”。但如果你真去翻 Redis 官方 GitHub、文檔和最新 Release Notes會(huì)發(fā)現(xiàn)Redis 本身沒(méi)有發(fā)布任何叫“AI 模塊”或“AI 內(nèi)置引擎”的功能。那這句話到底在說(shuō)什么它指向的是一場(chǎng)正在 quietly 發(fā)生的底層協(xié)議革命MCPModel Control Protocol協(xié)議的落地實(shí)踐以及 Redis 在其中扮演的“AI Agent 狀態(tài)中樞”角色。我從去年底開(kāi)始跟進(jìn) MCP 協(xié)議的早期草案到今年初參與幾個(gè)開(kāi)源 Agent 框架的內(nèi)部測(cè)試再到上個(gè)月用 Redis Cluster 搭建起第一個(gè)生產(chǎn)級(jí) Agent 技能調(diào)度中心全程踩過(guò)至少 7 類典型坑?，F(xiàn)在回頭看“Redis 接入 AI”根本不是指 Redis 學(xué)會(huì)了寫詩(shī)或推理而是指Redis 從傳統(tǒng)緩存/數(shù)據(jù)庫(kù)升級(jí)為 AI Agent 系統(tǒng)中唯一被廣泛采用、高可靠、低延遲、支持原子操作的“狀態(tài)總線”與“技能注冊(cè)中心”。關(guān)鍵詞里反復(fù)出現(xiàn)的agent-skills、mcp、wss://api.xiaozhi.me/mcp/這類地址正是這一范式轉(zhuǎn)移的直接證據(jù)。舉個(gè)最直白的例子你用 Playwright 寫了個(gè)自動(dòng)填表腳本過(guò)去它就是個(gè)獨(dú)立腳本現(xiàn)在它被封裝成一個(gè)符合 MCP 規(guī)范的skill注冊(cè)到 Redis 的某個(gè) Hash 結(jié)構(gòu)里比如skills:playwright:form-fill包含元信息描述、輸入 schema、調(diào)用權(quán)限、執(zhí)行入口WSS 地址或本地 socket 路徑、健康狀態(tài)last_heartbeat。當(dāng)一個(gè) LLM 驅(qū)動(dòng)的 Agent 需要“自動(dòng)填寫報(bào)名表”時(shí)它不再硬編碼調(diào)用邏輯而是向 Redis 查詢skills:*:form-fill拿到可用技能列表再通過(guò) MCP 協(xié)議發(fā)起標(biāo)準(zhǔn)調(diào)用。Redis 在這里不是存儲(chǔ)結(jié)果而是管理技能生命周期、協(xié)調(diào)并發(fā)調(diào)用、保障狀態(tài)一致性——這才是“接入”的真實(shí)含義。所以如果你正打算用 Python 寫一個(gè)能調(diào)用瀏覽器、發(fā)郵件、查天氣的 AI Agent別再只盯著 LangChain 或 LlamaIndex 的鏈?zhǔn)秸{(diào)用文檔了。先搞懂 Redis 在這個(gè)新架構(gòu)里的定位比選哪個(gè)大模型更重要。因?yàn)樗屑寄艿淖?cè)、發(fā)現(xiàn)、熔斷、限流、上下文傳遞最終都落在 Redis 的 Key-Value、Pub/Sub、Stream 和 Lua 原子腳本上。這已經(jīng)不是“用 Redis 存點(diǎn)中間結(jié)果”的小技巧而是整個(gè) Agent 架構(gòu)的基石。接下來(lái)我會(huì)從設(shè)計(jì)思路、核心細(xì)節(jié)、實(shí)操步驟到排錯(cuò)經(jīng)驗(yàn)一層層拆開(kāi)給你看。2. 整體架構(gòu)設(shè)計(jì)為什么必須是 Redis而不是 PostgreSQL 或 Kafka2.1 核心矛盾Agent 系統(tǒng)對(duì)狀態(tài)管理的“四高”要求一個(gè)真正能落地的 AI Agent 系統(tǒng)不是把一堆 API 調(diào)用串起來(lái)就完事。它必須解決四個(gè)關(guān)鍵狀態(tài)問(wèn)題而每個(gè)問(wèn)題背后都對(duì)應(yīng)著極高的技術(shù)要求高并發(fā)發(fā)現(xiàn)100 個(gè) Agent 同時(shí)在線每秒可能有 50 次“查找可用技能”請(qǐng)求。如果每次查詢都要走一次 HTTP 請(qǐng)求到注冊(cè)中心再加一層數(shù)據(jù)庫(kù)查詢延遲很容易突破 200ms導(dǎo)致 Agent 響應(yīng)卡頓。Redis 的SCANHGETALL組合在單節(jié)點(diǎn)上輕松支撐 5K QPS 的技能元數(shù)據(jù)讀取。高一致性更新當(dāng)某個(gè)技能服務(wù)重啟需要原子性地更新其status字段并刷新last_heartbeat時(shí)間戳。用關(guān)系型數(shù)據(jù)庫(kù)得開(kāi)事務(wù)、加行鎖、處理死鎖用 Redis一個(gè)HSET skills:xxx status online last_heartbeat 1717023456加EXPIRE就搞定且天然支持 CASCheck-And-Set語(yǔ)義。高時(shí)效性通知Agent 需要實(shí)時(shí)感知技能上下線。Kafka 雖然擅長(zhǎng)消息分發(fā)但它的消費(fèi)模型是“拉取”存在秒級(jí)延遲而 Redis Pub/Sub 是真正的“推”PUBLISH skills:status:changed {id:playwright-form,status:offline}訂閱者毫秒級(jí)收到。我們實(shí)測(cè)在 3 節(jié)點(diǎn) Redis Cluster 上Pub/Sub 平均延遲 3ms。高靈活性編排Agent 執(zhí)行流程常含條件分支如“如果郵箱驗(yàn)證失敗則重試最多 3 次”。Redis Stream 天然支持消費(fèi)者組Consumer Group可以將一個(gè)任務(wù)拆成多個(gè)子任務(wù)stream:agent:task:12345由不同 Worker 拉取處理并通過(guò)XACK確保至少一次交付。這比用 RabbitMQ 手動(dòng)管理 ACK 隊(duì)列簡(jiǎn)潔得多。提示很多團(tuán)隊(duì)一開(kāi)始用 SQLite 存技能列表覺(jué)得輕量。但當(dāng)并發(fā) 50或需要做“按標(biāo)簽篩選技能”如skills:tag:browser時(shí)SQLite 的 WAL 鎖爭(zhēng)用會(huì)讓整個(gè) Agent 調(diào)度卡死。這不是性能問(wèn)題是架構(gòu)選型的根本錯(cuò)誤。2.2 為什么不是其他 NoSQLMongoDB、Elasticsearch 的短板在哪有人會(huì)問(wèn)MongoDB 不也能存 JSON、做原子更新、支持索引嗎Elasticsearch 不更擅長(zhǎng)“按標(biāo)簽搜索技能”答案是它們解決了“存”但沒(méi)解決“協(xié)同”。MongoDB 的findAndUpdate確實(shí)能原子更新但它沒(méi)有原生的 Pub/Sub 機(jī)制。你要實(shí)現(xiàn)技能上下線通知得額外搭一個(gè) Change Stream 監(jiān)聽(tīng)器再配一套消息轉(zhuǎn)發(fā)邏輯復(fù)雜度陡增。而 Redis 的PUBLISH/SUBSCRIBE是開(kāi)箱即用的一行代碼就能讓所有 Agent 實(shí)例同步狀態(tài)。Elasticsearch 搜索快但它是為“最終一致性”設(shè)計(jì)的。當(dāng)你POST /skills/_doc/123更新一個(gè)技能狀態(tài)到它被GET /skills/_search?qtag:browser查到中間可能有 1~2 秒延遲refresh interval。而 Agent 調(diào)度要求的是“強(qiáng)一致性”——?jiǎng)偵暇€的技能必須立刻能被發(fā)現(xiàn)。Redis 的HSET是立即生效的。更關(guān)鍵的是MCP 協(xié)議棧的默認(rèn)約定。翻看mcp-server-python的源碼你會(huì)發(fā)現(xiàn)它的SkillRegistry類默認(rèn)構(gòu)造函數(shù)里redis_url是必填參數(shù)而mongo_uri或es_host是可選擴(kuò)展。這不是偶然——MCP 的設(shè)計(jì)者一群做過(guò)多年分布式系統(tǒng)的工程師明確把 Redis 定為“事實(shí)上的狀態(tài)總線”。這意味著所有遵循 MCP 的開(kāi)源工具如playwright-mcp、burpsuite-mcp、yakit-mcp都內(nèi)置了 Redis 適配器你不用自己寫連接邏輯。2.3 Redis 的“非典型”用法超越 cache 和 queue 的四大核心結(jié)構(gòu)在 AI Agent 場(chǎng)景里Redis 不再是簡(jiǎn)單的SET key value。我們實(shí)際用到了四種結(jié)構(gòu)每種承擔(dān)不同職責(zé)Hash哈?！?技能元數(shù)據(jù)中心Key:skills:playwright:form-fillFields:description,input_schema,output_schema,endpoint,status,last_heartbeat,tags優(yōu)勢(shì)單次HGETALL獲取全部元數(shù)據(jù)HINCRBY可統(tǒng)計(jì)調(diào)用次數(shù)HSCAN支持模糊匹配如HSCAN skills:*:form* 0 MATCH *。Pub/Sub發(fā)布/訂閱—— 實(shí)時(shí)狀態(tài)廣播通道Channel:skills:status:changedMessage:{skill_id: playwright-form, status: online, timestamp: 1717023456}優(yōu)勢(shì)零延遲通知無(wú)狀態(tài)Agent 實(shí)例重啟后重新SUBSCRIBE即可不占用持久化空間。Stream流—— 任務(wù)隊(duì)列與上下文管道Stream:stream:agent:task:12345Entry:{step: validate_email, input: {email: userx.com}, timeout: 30}優(yōu)勢(shì)天然支持多消費(fèi)者組一個(gè) Agent 流程可拆成多個(gè) WorkerXREADGROUP保證消息不丟失XPENDING可查積壓任務(wù)。Sorted Set有序集合—— 技能健康度評(píng)分與熔斷Key:skills:health:scoreMember:playwright-formScore:98.7基于成功率、延遲、錯(cuò)誤率計(jì)算的綜合分優(yōu)勢(shì)ZRANGEBYSCORE skills:health:score 95 100一鍵獲取高分技能ZREM可快速剔除故障節(jié)點(diǎn)配合ZREMRANGEBYSCORE實(shí)現(xiàn)自動(dòng)降級(jí)。這四種結(jié)構(gòu)組合構(gòu)成了一個(gè)輕量但完備的 Agent 協(xié)同基礎(chǔ)設(shè)施。它不依賴外部消息中間件不引入新運(yùn)維組件所有能力都在 Redis 單一進(jìn)程中完成——這才是“正式接入”的技術(shù)底氣。3. 核心細(xì)節(jié)解析MCP 協(xié)議如何與 Redis 對(duì)接Python 實(shí)現(xiàn)要點(diǎn)3.1 MCP 協(xié)議的本質(zhì)不是 RPC而是“技能契約”MCPModel Control Protocol常被誤解為一種新的通信協(xié)議類似 HTTP/2。其實(shí)它更像一份技能交互的契約規(guī)范。它定義了三件事技能如何被描述一個(gè) JSON Schema規(guī)定name、description、input、output、tags等字段。例如一個(gè)發(fā)送郵件的技能其input必須包含to,subject,body字段。技能如何被調(diào)用統(tǒng)一使用 WebSocketWSS連接客戶端發(fā)送{type: call, skill_id: send-email, input: {...}}服務(wù)端返回{type: result, success: true, output: {...}}。技能如何被管理通過(guò)register、unregister、heartbeat三條控制指令維護(hù)技能生命周期。Redis 在這里不參與協(xié)議解析只負(fù)責(zé)存儲(chǔ)契約內(nèi)容、分發(fā)控制指令、記錄運(yùn)行時(shí)狀態(tài)。換句話說(shuō)MCP 是“語(yǔ)言”Redis 是“檔案館廣播站調(diào)度臺(tái)”。注意wss://api.xiaozhi.me/mcp/?token...這類地址是某個(gè) MCP Server 的公開(kāi)接入點(diǎn)它背后必然連著一個(gè) Redis 實(shí)例或集群。Token 用于鑒權(quán)確保只有授權(quán) Agent 能讀寫對(duì)應(yīng) namespace 的技能數(shù)據(jù)。這不是“免費(fèi) AI 聊天入口”而是 MCP 生態(tài)的標(biāo)準(zhǔn)化接入方式。3.2 Python 中的 Redis 技能注冊(cè)從手動(dòng)命令到自動(dòng)化裝飾器最基礎(chǔ)的注冊(cè)方式是直接用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] } # 注冊(cè)到 Redis Hash r.hset(skills:playwright:form-fill, mappingskill_data) r.expire(skills:playwright:form-fill, 300) # 5分鐘過(guò)期需心跳續(xù)命但這太原始。我們把它封裝成一個(gè)裝飾器讓技能開(kāi)發(fā)者專注業(yè)務(wù)邏輯def mcp_skill(skill_id: str, tags: list None): def decorator(func): # 自動(dòng)注冊(cè)技能元數(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) # 啟動(dòng)心跳線程 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 # 實(shí)際的 Playwright 代碼... return {status: success, url: url}這個(gè)裝飾器做了三件事自動(dòng)注冊(cè)元數(shù)據(jù)、設(shè)置過(guò)期時(shí)間、啟動(dòng)后臺(tái)心跳。它讓技能開(kāi)發(fā)回歸本質(zhì)——寫函數(shù)而不是管 Redis 命令。3.3 Agent 如何通過(guò) Redis 發(fā)現(xiàn)并調(diào)用技能一個(gè)完整的 Python 流程Agent 的核心邏輯是“查-選-調(diào)-監(jiān)”四步閉環(huán)。以下是用 Python 實(shí)現(xiàn)的精簡(jiǎ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)和標(biāo)簽 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 連接簡(jiǎn)化版實(shí)際用 websocket-client ws websocket.WebSocket() ws.connect(endpoint.decode()) # 發(fā)送調(diào)用請(qǐng)求 call_msg { type: call, skill_id: skill_id, input: input_data } ws.send(json.dumps(call_msg)) # 等待響應(yīng) 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è)務(wù)邏輯但提供了發(fā)現(xiàn)、篩選、調(diào)用的全部上下文。discover_skills方法里r.keys(skills:*)是潛在性能瓶頸大數(shù)據(jù)量時(shí)實(shí)際生產(chǎn)環(huán)境我們會(huì)用SCAN替代并建立skills:tag:browser這樣的二級(jí)索引 Hash把playwright:form-fill的 ID 存進(jìn)去實(shí)現(xiàn) O(1) 標(biāo)簽查詢。3.4 Docker 環(huán)境下的 Redis 主從部署為什么 Agent 架構(gòu)必須考慮高可用單機(jī) Redis 滿足不了生產(chǎn)需求。Agent 系統(tǒng)要求 24/7 在線一旦 Redis 掛掉所有技能發(fā)現(xiàn)和任務(wù)調(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)鍵開(kāi)啟 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實(shí)操心得很多團(tuán)隊(duì)用 Redis Cluster覺(jué)得“分片更高級(jí)”。但在 Agent 場(chǎng)景Cluster 的跨 slot 操作如KEYS *被禁止而技能發(fā)現(xiàn)常需掃描所有 keys。Sentinel 模式雖無(wú)分片但主從切換 30 秒且KEYS命令可用更適合當(dāng)前階段。等技能數(shù)量超 10K再平滑遷移到 Cluster用redis-py-cluster客戶端處理。4. 實(shí)操過(guò)程詳解從零搭建一個(gè) MCP Redis 的 Python Agent 系統(tǒng)4.1 環(huán)境準(zhǔn)備Python 3.10、Redis 7.2、Playwright可選我們以“構(gòu)建一個(gè)能自動(dòng)填寫網(wǎng)頁(yè)表單的 AI Agent”為例完整走一遍。所需工具Python 3.10MCP 庫(kù)依賴較新特性如asyncio的改進(jìn)Redis 7.2支持STREAM的MAXLEN限制和XADD的NOMKSTREAM選項(xiàng)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 # 安裝核心庫(kù) pip install redis websocket-client playwright python-dotenv # 安裝 Playwright 瀏覽器Chrome playwright install chromium # 啟動(dòng) RedisDocker 方式最干凈 docker run -d --name redis-mcp -p 6379:6379 -e REDIS_PASSWORDyourpass redis:7.2-alpine注意不要用redis-server直接啟動(dòng)因?yàn)槟J(rèn)配置不啟用密碼和 AOF生產(chǎn)環(huán)境會(huì)丟數(shù)據(jù)。Docker 啟動(dòng)時(shí)通過(guò)-e REDIS_PASSWORD設(shè)置密碼后續(xù) Python 連接要用redis://:yourpasslocalhost:6379/0。4.2 編寫第一個(gè) MCP 技能Playwright 表單填充服務(wù)創(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ù)裝飾器會(huì)用到 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 技能裝飾器簡(jiǎn)化版無(wú)心跳 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, # 本機(jī) WebSocket 服務(wù) 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 服務(wù)簡(jiǎn)化僅演示 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 })) # 啟動(dòng) WebSocket 服務(wù) async def main(): async with websockets.serve(mcp_handler, localhost, 8000): await asyncio.Future() # run forever if __name__ __main__: asyncio.run(main())這個(gè)腳本做了三件事定義技能元數(shù)據(jù)、用裝飾器注冊(cè)到 Redis、實(shí)現(xiàn) WebSocket 服務(wù)接收 MCP 調(diào)用。注意fill_form是異步函數(shù)因?yàn)?Playwright 必須異步調(diào)用。4.3 構(gòu)建 Agent 核心技能發(fā)現(xiàn)、路由與容錯(cuò)創(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ù)責(zé)發(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] # 分頁(yè)掃描 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 # 默認(rèn)分 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) # 運(yùn)行 asyncio.run(agent_workflow())這個(gè)SkillRouter類體現(xiàn)了 Agent 的核心智能它不只是調(diào)用還做緩存、熔斷、重試、健康度反饋。每次調(diào)用失敗它會(huì)主動(dòng)給技能扣分zincrby讓低分技能逐漸退出候選池。這就是 Redis 的力量——狀態(tài)驅(qū)動(dòng)的自適應(yīng)調(diào)度。4.4 監(jiān)控與治理用 Redis CLI 和 Python 腳本做日常運(yùn)維系統(tǒng)上線后不能只靠代碼跑。你需要一套監(jiān)控手段實(shí)時(shí)查看技能狀態(tài)# 進(jìn)入 Redis CLI redis-cli -a yourpass # 查看所有技能 KEYS skills:* # 查看某個(gè)技能詳情 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: # 超過(guò)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(): 生成日?qǐng)?bào) 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分鐘檢查一次這個(gè)腳本會(huì)自動(dòng)踢掉失聯(lián)技能并生成健康報(bào)告。你可以用systemd或supervisord守護(hù)它讓它永遠(yuǎn)運(yùn)行。5. 常見(jiàn)問(wèn)題與排查技巧實(shí)錄我在 3 個(gè)項(xiàng)目中踩過(guò)的坑5.1 問(wèn)題速查表高頻故障與根因分析現(xiàn)象可能原因排查命令解決方案Agent 總是找不到技能Redis 密碼錯(cuò)誤連接被拒redis-cli -a wrongpass ping檢查REDIS_URL環(huán)境變量確認(rèn)密碼正確技能狀態(tài)顯示offline但服務(wù)明明在跑心跳線程未啟動(dòng)或異常退出HGET skills:xxx last_heartbeat在裝飾器中加try/except日志確保心跳線程存活SCAN命令返回空但KEYS *能查到Redis 配置了protected-mode yesredis-cli config get protected-mode設(shè)為no或用--no-auth-warning啟動(dòng)WebSocket 調(diào)用超時(shí)但技能服務(wù)日志顯示已收到Redis Pub/Sub 通道名拼寫錯(cuò)誤PUBSUB CHANNELS確認(rèn)PUBLISH和SUBSCRIBE使用完全相同的 channel 名健康分一直為 0ZSCORE返回 NoneSorted Set 未初始化或ZADD未執(zhí)行ZCARD skills:health:score在技能注冊(cè)時(shí)強(qiáng)制ZADD skills:health:score 100 skill_id5.2 獨(dú)家避坑技巧那些文檔里不會(huì)寫的細(xì)節(jié)技巧 1用SCAN代替KEYS但要注意游標(biāo)陷阱KEYS