AI弹幕助手：从WebSocket实时通信到大模型智能回复的完整技术指南

 传统方案1：基于HTTP轮询的弹幕拉取
import requests
import time

room_id = "123456"
while True:
     每隔3秒向服务器请求一次新弹幕
    resp = requests.get(f"https://api.live.com/room/{room_id}/danmaku?timestamp={int(time.time())}")
    danmakus = resp.json()
    for msg in danmakus:
        print(f"收到弹幕：{msg['content']}")
    time.sleep(3)   轮询间隔，延迟至少3秒

 传统方案2：基于关键词匹配的自动回复
def keyword_reply(content: str) -> str:
    if "你好" in content or "hi" in content.lower():
        return "欢迎来到直播间！"
    elif "666" in content:
        return "感谢支持~"
    elif "价格" in content or "多少钱" in content:
        return "请点击小黄车查看详情"
    else:
        return None   无法识别则不做回复

// 前端：使用WebSocket接收实时弹幕
const ws = new WebSocket('wss://live.bilibili.com/subscription');

ws.onopen = () => {
    console.log('WebSocket连接已建立');
    // 发送心跳包维持连接
    setInterval(() => ws.send(JSON.stringify({type: 'heartbeat'})), 30000);
};

ws.onmessage = (event) => {
    const danmaku = JSON.parse(event.data);
    console.log(`[实时] ${danmaku.user}: ${danmaku.content}`);
    // 将弹幕交给AI处理模块
    aiHandler.process(danmaku);
};

弹幕输入 → WebSocket接收 → 预处理（清洗/归一化）→ LLM推理 → 回复生成 → WebSocket推送

 ai_danmaku_assistant.py
import asyncio
import websockets
import json
from typing import Dict, Optional
import requests

class AIDanmakuAssistant:
    """AI弹幕助手核心类"""
    
    def __init__(self, llm_api_url: str, llm_model: str = "gpt-oss-20b"):
        self.llm_api_url = llm_api_url
        self.llm_model = llm_model
        self.conversation_context: Dict[str, list] = {}   用户对话上下文缓存
        self._cache: Dict[str, str] = {}   高频问题缓存
        
    async def connect_to_platform(self, ws_url: str, room_id: str):
        """通过WebSocket连接直播平台，建立弹幕监听"""
        async with websockets.connect(ws_url) as websocket:
             1. 发送认证信息（token、room_id等）
            await websocket.send(json.dumps({
                "type": "auth",
                "room_id": room_id,
                "token": "your_access_token"
            }))
            
             2. 启动心跳保持连接（每30秒）
            asyncio.create_task(self._heartbeat(websocket))
            
             3. 持续接收弹幕
            async for message in websocket:
                await self._handle_danmaku(message)
    
    async def _heartbeat(self, websocket):
        """维持WebSocket长连接的心跳机制"""
        while True:
            await asyncio.sleep(30)
            await websocket.send(json.dumps({"type": "heartbeat"}))
    
    async def _handle_danmaku(self, raw_message: str):
        """处理单条弹幕：解析 → 去重 → AI推理 → 回复"""
        data = json.loads(raw_message)
        
         过滤非弹幕消息（礼物、进房等）
        if data.get("type") != "danmaku":
            return
        
        user_id = data.get("uid")
        content = data.get("content", "").strip()
        
         去重：避免同一用户短时间内重复处理
        if self._is_duplicate(user_id, content):
            return
        
         调用AI生成回复
        reply = await self._generate_reply(user_id, content)
        
         通过WebSocket发送回复
        if reply:
            await self._send_reply(reply)
    
    async def _generate_reply(self, user_id: str, content: str) -> Optional[str]:
        """调用大语言模型生成智能回复"""
        
         缓存命中：高频问题直接返回
        if content in self._cache:
            return self._cache[content]
        
         构建prompt，注入上下文
        context = self.conversation_context.get(user_id, [])
        prompt = f"""你是一个直播间的AI弹幕助手，风格友好热情。
当前对话上下文：{context[-3:] if context else '无'}
观众说：「{content}」
请生成一句简洁、自然的回复（不超过50字）："""
        
         调用LLM API
        response = requests.post(
            f"{self.llm_api_url}/v1/completions",
            json={
                "model": self.llm_model,
                "prompt": prompt,
                "max_tokens": 100,
                "temperature": 0.7
            },
            timeout=2   2秒超时，保证实时性
        )
        
        if response.status_code == 200:
            reply = response.json()["choices"][0]["text"].strip()
             更新上下文缓存（最多保留10条）
            if user_id not in self.conversation_context:
                self.conversation_context[user_id] = []
            self.conversation_context[user_id].append({"user": content, "assistant": reply})
            if len(self.conversation_context[user_id]) > 10:
                self.conversation_context[user_id].pop(0)
            
             高频问题写入缓存
            if len(self._cache) < 100:
                self._cache[content] = reply
            return reply
        
        return None
    
    def _is_duplicate(self, user_id: str, content: str) -> bool:
        """简单的防重复逻辑"""
         实际生产环境可用Redis记录最近发言
        return False
    
    async def _send_reply(self, reply: str):
        """通过WebSocket发送回复到直播间"""
         调用直播平台发送弹幕的API
         此处省略具体实现
        pass


 启动示例
if __name__ == "__main__":
    assistant = AIDanmakuAssistant(
        llm_api_url="http://localhost:8080",
        llm_model="gpt-oss-20b"
    )
    asyncio.run(assistant.connect_to_platform(
        ws_url="wss://live.bilibili.com/subscription",
        room_id="12345678"
    ))