简介这是一套面向B站、抖音直播场景的场控机器人源码适合希望搭建自动化直播间的主播、运营者及二次开发者。它把弹幕姬、答谢姬、回复姬、点歌姬与多种大模型AI、工作流整合在一起支持弹幕聊天、观众互动管理、数据统计分析、自动点歌、私信处理、AI自动闲聊与直播建议核心亮点是可编程控制像搭积木一样配置互动规则打造专属直播风格。压缩包共2000个文件以1350个C头文件、271个C源文件、102个C文件为主另有txt、md说明文档、html与css前端页面、js脚本、json配置及少量Python、Go代码整体约72.38MB结构完整便于二次开发。目前已有215人学习下载。借助这套源码读者可快速理解直播互动系统的模块划分与通信逻辑参考AI接入与工作流编排方式并在此基础上定制弹幕规则、答谢策略与点歌流程构建高粘性粉丝互动环境。1. 从一条弹幕到一场直播场控机器人到底在替谁干活凌晨一点直播间在线人数掉到两位数主播还在硬撑。这时候屏幕上飘过一条“主播唱首歌”紧接着又一条“刚才那个问题能再讲一遍吗”再然后是一串“666”。如果全靠人盯主播要么分心要么漏掉。B站/抖音直播场控机器人要解决的就是把这堆重复、琐碎、有时效性的互动动作自动化弹幕姬负责抓取和分发消息答谢姬负责礼物和上舰的感谢回复姬负责关键词应答点歌姬负责把点歌请求排进队列。它适合个人主播、小团队运营也适合做直播数据采集和二次开发的工程师。热词里“弹幕姬”“直播数据”“抖音福袋自动抢脚本”都指向同一个底层能力——实时拿到直播间事件流再按规则触发动作。这篇笔记不讲空泛概念只讲怎么把这条链路跑通、参数怎么调、哪里容易翻车。2. 弹幕姬把直播间事件流接进本地程序2.1 为什么选 WebSocket 而不是轮询直播间的弹幕、礼物、进场、上舰本质上是服务端持续推送的事件流。用 HTTP 轮询去拉延迟高、请求量大还容易被限流。常见做法是走 WebSocket 长连接B站有公开的弹幕长连接协议抖音侧则更多依赖开放平台或页面侧的事件回调。选型时先确认一件事你要的是“只读”还是“可写”。只读弹幕做数据分析和答谢WebSocket 足够要发弹幕、禁言、上房管就得走带鉴权的开放接口权限和风控完全是另一套。我一般会把弹幕姬拆成三层连接层负责握手和心跳解析层负责把二进制或 JSON 包转成统一事件对象分发层负责按事件类型投递给下游。这样后面加答谢姬、回复姬时不用动连接层。2.2 最小可跑的弹幕接收脚本下面这段是 B站弹幕长连接的简化版重点看握手包和心跳实际项目里要把房间号、cookie 换成自己的。import websocket import json import struct import zlib import threading import time ROOM_ID 123456 # 换成你的真实房间号 COOKIE SESSDATAxxx # 登录态只读弹幕可省略部分字段 # B站弹幕包头4字节总长 2字节头长 2字节协议版本 4字节操作码 4字节序列 HEADER_STRUCT struct.Struct(IHHII) def pack_body(body: str, op: int) - bytes: data body.encode(utf-8) header HEADER_STRUCT.pack( HEADER_STRUCT.size len(data), # 总长度 HEADER_STRUCT.size, # 头长度固定 16 1, # 协议版本1 表示 body 不压缩 op, # 操作码7 进房2 心跳 1 # 序列号固定 1 ) return header data def on_message(ws, message): # 服务端可能一次推多个包需要循环拆 while message: total, head_len, ver, op, seq HEADER_STRUCT.unpack(message[:16]) body message[head_len:total] if ver 2: body zlib.decompress(body) # 压缩包要解 if op 5: # 通知类消息 payload json.loads(body.decode(utf-8)) for cmd in payload.get(cmd, ).split(|): if cmd.startswith(DANMU_MSG): text payload[info][1] uid payload[info][2][0] print(f[弹幕] uid{uid} text{text}) message message[total:] def on_open(ws): ws.send(pack_body(json.dumps({roomid: ROOM_ID}), 7), opcode0x2) def heartbeat(): while True: time.sleep(30) ws.send(pack_body(, 2), opcode0x2) threading.Thread(targetheartbeat, daemonTrue).start() ws websocket.WebSocketApp( wss://broadcastlv.chat.bilibili.com/sub, on_messageon_message, on_openon_open, header{Cookie: COOKIE} ) ws.run_forever()逻辑说明pack_body按协议拼包头操作码 7 是进房2 是心跳。on_message里先解包头拿到总长再按总长切出当前包剩余字节继续循环这是处理粘包的关键。参数上心跳间隔常见 30 秒太短浪费连接太长会被服务端断开。ver 2表示 body 是 zlib 压缩必须解压后再解析 JSON。提示只读弹幕时 cookie 可以简化但涉及用户维度的统计建议带上登录态否则部分字段会缺失。2.3 事件统一成对象后再分发拿到原始 JSON 后不要直接在各处解析先转成统一结构比如{type, uid, nickname, text, ts, raw}。这样答谢姬和回复姬都只认这个结构换平台时只改解析层。抖音侧如果走开放平台事件字段名不同但映射逻辑一样。这一步做扎实后面加功能就是加订阅者不是改主流程。3. 答谢姬与回复姬规则引擎比 if-else 更抗造3.1 规则该长什么样答谢姬要处理的是“谁送了什么该说什么”。回复姬要处理的是“谁说了什么该回什么”。如果全写成 if-else规则一多就变成玄学。常见做法是抽一张规则表每条规则包含匹配条件、优先级、冷却时间、动作模板。匹配条件可以是事件类型、关键词、正则、用户等级、礼物价值区间。优先级决定多条命中时谁先执行冷却时间防止同一个人刷屏被反复答谢。字段含义示例event_type事件类型danmu / gift / guardmatch匹配表达式正则点歌\s*(.)priority优先级数字越小越先10cooldown同一用户冷却秒数60action动作模板感谢 {nickname} 的 {gift_name}3.2 用 Python 写一个可扩展的规则引擎import re import time from dataclasses import dataclass, field from typing import Callable, Optional dataclass class Rule: name: str event_type: str pattern: Optional[str] None priority: int 100 cooldown: int 0 action: Optional[Callable] None _last_fire: dict field(default_factorydict) def match(self, event: dict) - Optional[re.Match]: if event[type] ! self.event_type: return None if self.pattern is None: return True return re.search(self.pattern, event.get(text, )) def can_fire(self, uid: str) - bool: now time.time() last self._last_fire.get(uid, 0) if now - last self.cooldown: return False self._last_fire[uid] now return True class RuleEngine: def __init__(self): self.rules [] def register(self, rule: Rule): self.rules.append(rule) self.rules.sort(keylambda r: r.priority) def dispatch(self, event: dict): for rule in self.rules: m rule.match(event) if m and rule.can_fire(event[uid]): rule.action(event, m) break # 命中高优先级后不再往下避免重复回复逻辑说明Rule.match先比事件类型再跑正则。can_fire用字典记录每个用户上次触发时间实现冷却。dispatch按优先级排序后逐条尝试命中即停。参数上priority建议把“点歌”“提问”这类需要即时响应的放前面把“666”这类泛匹配放后面。cooldown对答谢类建议 30 到 60 秒对关键词回复可以设 10 秒。3.3 点歌姬的队列与去重点歌姬本质是一个带优先级的队列。用户发“点歌 晴天”回复姬命中后把歌名推进队列点歌姬按顺序消费。要处理三个问题同一首歌短时间被多人点去重合并队列满了要拒绝并提示歌曲不存在要有兜底回复。我一般用collections.deque加一个set做去重队列长度设 20 到 50超出就回复“当前排队较多稍后再点”。注意回复姬发弹幕有频率限制连续快速发送容易触发风控动作之间加 1 到 2 秒随机间隔更稳。4. 避坑与排查那些让机器人当场翻车的细节4.1 连接频繁断开日志里全是重连现象弹幕姬跑十几分钟就断重连后丢一段消息。原因通常是心跳没发或发得太晚服务端判定连接空闲。解决把心跳间隔压到 30 秒以内并在on_close里做指数退避重连第一次 1 秒之后翻倍上限 30 秒。重连后要重新进房否则收不到消息。4.2 弹幕解析乱码或 JSON 报错现象json.loads抛异常或者中文变乱码。原因是没处理压缩包或者一次收到多个包只解了第一个。解决严格按包头总长切包循环处理ver 2时先zlib.decompress。另外注意 body 可能是多个 JSON 用换行拼接要按行拆。4.3 答谢重复刷屏被禁言现象同一个人连送礼物机器人连发多条感谢然后账号被限制。原因是冷却只按规则维度做没按用户维度。解决冷却字典的 key 用uid rule.name并且发送动作之间加随机延迟。礼物连击可以合并成一条“感谢 xx 的 N 个礼物”。4.4 关键词误触发现象正常聊天里出现“点”字就被当成点歌。原因是正则太宽。解决用更严格的模式比如^点歌\s(.)$并要求消息以关键词开头。上线前拿历史弹幕跑一遍回归看误触发率。4.5 多平台字段对不上现象B站跑通的规则换到抖音就失效。原因是事件字段名和结构不同。解决在解析层做适配输出统一事件对象规则引擎只认统一结构。平台差异全部收敛在解析层不要散落到规则里。5. 进阶把场控机器人做成可观测、可回放的系统跑到这一步功能基本齐了但真正决定它能不能长期用的是可观测性。我习惯给每个事件打一个trace_id从接收到规则命中再到动作发送全链路记一条结构化日志。这样出问题时能回答三个问题消息有没有收到、命中了哪条规则、动作有没有发出去。日志用 JSON 行格式方便后面用脚本统计。回放能力同样重要。把原始事件流落盘格式就用{ts, raw}一行一条。想调规则时不用真开播直接读文件回放看新规则会不会误触发。下面这个小脚本就是回放入口import json from rule_engine import RuleEngine, Rule engine RuleEngine() engine.register(Rule( name点歌, event_typedanmu, patternr^点歌\s(.)$, priority10, cooldown10, actionlambda e, m: print(f点歌: {m.group(1)} from {e[nickname]}) )) with open(events.jsonl, r, encodingutf-8) as f: for line in f: event json.loads(line) engine.dispatch(event)参数上回放时把冷却时间临时调成 0才能看到所有命中正式跑再恢复。日志里重点看两个指标规则命中率和动作发送成功率。命中率突然掉多半是平台字段变了发送成功率掉多半是风控或权限问题。最后说个我自己的习惯每次改规则前先拿最近三天的弹幕回放一遍确认误触发和漏触发都在可接受范围再上线。直播是实时的没有后悔药回放就是唯一的黑匣子。这套东西不复杂难的是把边界和冷却想清楚。希望帮到你。本文还有配套的精品资源点击获取