⚡ CFM Redis - 跨框架实时通信

基于Redis Pub/Sub的跨框架Agent通信方案。事件驱动，零轮询，通信通道不消耗LLM token。

核心优势

• ⚡ 实时通信 - 消息延迟 < 10ms
• 💰 通信零token - Redis通道本身不消耗LLM token（处理消息时仍需token）
• 🔄 双向通信 - 支持任意框架之间通信
• 💾 消息持久化 - 自动保存历史记录
• 🔍 Agent发现 - 自动发现网络中的Agent

注意：CFM的Redis通道通信不消耗token，但Agent处理消息时仍需调用LLM（会消耗token）。相比传统轮询方案，CFM通过事件驱动大幅减少了不必要的token消耗。

工作原理

Agent A ──publish──→ Redis Server ──subscribe──→ Agent B
                    (本地Redis)                          
Agent B ──publish──→ Redis Server ──subscribe──→ Agent A

安装

1. 安装Redis

# macOS
brew install redis
brew services start redis

# Ubuntu/Debian
sudo apt install redis-server
sudo systemctl start redis

# 验证
redis-cli ping  # 应返回 PONG

2. 安装Python依赖

pip install redis

3. 下载CFM库

mkdir -p ~/.shared/cfm
# 将 cfm_messenger.py 和 cfm_cli.py 放入此目录

核心文件

cfm_messenger.py - 通信库

#!/usr/bin/env python3
"""CFM Messenger - 跨框架通信库"""

import redis
import json
import time
import uuid
from datetime import datetime
from typing import Optional, Dict, Any


class CFMMessenger:
    """跨框架通信器"""
    
    def __init__(self, agent_id: str, redis_host: str = "localhost", redis_port: int = 6379):
        self.agent_id = agent_id
        self.redis_client = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
        self.message_store_key = f"cfm:{agent_id}:messages"
    
    def send(self, to_agent: str, message: str, msg_type: str = "text") -> str:
        """发送消息"""
        msg_id = str(uuid.uuid4())[:8]
        payload = {
            "id": msg_id,
            "from": self.agent_id,
            "to": to_agent,
            "type": msg_type,
            "content": message,
            "timestamp": datetime.now().isoformat()
        }
        
        # 发布到目标agent频道
        target_channel = f"cfm:{to_agent}:inbox"
        self.redis_client.publish(target_channel, json.dumps(payload, ensure_ascii=False))
        
        # 同时存储到双方（便于查询）
        self._store_message(payload, self.agent_id)
        self._store_message(payload, to_agent)
        
        return msg_id
    
    def get_messages(self, limit: int = 50) -> list:
        """获取消息历史"""
        messages = self.redis_client.lrange(self.message_store_key, 0, limit - 1)
        return [json.loads(msg) for msg in messages]
    
    def _store_message(self, msg: dict, agent_id: str = None):
        """持久化消息"""
        store_key = f"cfm:{agent_id or self.agent_id}:messages"
        self.redis_client.lpush(store_key, json.dumps(msg, ensure_ascii=False))
        self.redis_client.ltrim(store_key, 0, 999)
    
    def discover_agents(self) -> list:
        """发现其他Agent"""
        keys = self.redis_client.keys("cfm:*:registered")
        agents = []
        for key in keys:
            agent_id = key.split(":")[1]
            if agent_id != self.agent_id:
                info = self.redis_client.hgetall(key)
                agents.append({"id": agent_id, **info})...