简介这是一套面向B站、抖音直播场景的场控机器人源码适合希望搭建自动化直播间的开发者与运营者。它把弹幕姬、答谢姬、回复姬、点歌姬与多种大模型AI、工作流整合在一起支持弹幕聊天、观众互动管理、数据统计分析、自动点歌、私信处理、AI自动闲聊与直播建议最大特点是可编程控制像搭积木一样配置互动规则打造专属直播风格。压缩包共2000个文件约72.38MB以1350个C头文件、271个C源文件、102个C文件为核心配合少量Python、Go、JS、HTML、CSS及json、xml配置覆盖网络通信、音视频处理与AI调用等模块。目前已有215人学习。借助这套源码读者可研究弹幕协议解析、AI回复与工作流编排的实现思路并在此基础上二次开发构建高粘性粉丝互动与自动化直播管理体系。1. 直播场控机器人到底在控什么从一条弹幕到一次答谢的完整链路做直播的人都有一个共同的体感真正累的不是开播而是开播之后那几百条弹幕里你得一边讲内容、一边记谁送了礼、一边回问题、一边还得接住点歌和上墙的节奏。B站和抖音的直播场景里这套动作靠人肉盯屏撑不过半小时就会开始漏消息。所谓直播场控机器人本质就是把「监听弹幕 → 判断意图 → 触发动作」这条链路自动化弹幕姬负责收消息答谢姬负责礼物和进场回复姬负责关键词应答点歌姬负责把歌名转成播放队列。它不改变直播内容只接管那些重复、机械、又必须及时响应的环节。适合谁一个人开播没有助播的中小主播、做无人值守轮播的运营号、以及想拿直播弹幕流练手消息队列和规则引擎的开发者。下面按我实际搭过的一套方案把每个姬怎么落地、参数怎么调、哪里会翻车讲清楚。2. 弹幕姬把B站和抖音的弹幕接进同一个消息管道2.1 两条平台的接入方式差异B站直播的弹幕走的是 WebSocket 长连接连上直播间对应的房间后服务端会持续推消息包包体是带协议头的二进制格式需要按协议解析出 JSON 再取弹幕内容。抖音这边没有对个人开放的官方弹幕长连接常见做法是走开放平台的直播互动能力或者用浏览器端的事件流做中转把页面里已经渲染出来的弹幕节点变化抓出来再转发到本地。两条链路的技术栈不一样但落地时我一般会统一成同一个内部消息格式后面所有姬都只认这个格式换平台不用改业务代码。统一格式建议至少包含这几个字段平台标识、房间号、用户ID、用户名、消息类型弹幕/礼物/进场/关注、消息内容、时间戳。消息类型这个字段是后面答谢姬和回复姬做分流的关键一开始就定好别等到业务写了一半再回头加。2.2 用 Python 起一个最小弹幕接收服务下面这段是 B站弹幕接入的最小骨架用 websocket-client 连房间按协议头解包。抖音侧因为接口形态不同这里用伪代码位置标注实际替换成你拿到的数据源即可。import websocket import struct import json import zlib # B站弹幕包16字节头 正文 # 头结构4字节总长度 2字节头长度 2字节协议版本 4字节操作码 4字节序列号 HEADER_LEN 16 OP_HEARTBEAT 2 OP_MESSAGE 5 OP_AUTH 7 def parse_packet(data): 拆一个完整包返回操作码和正文 total_len, header_len, ver, op, seq struct.unpack(IHHII, data[:HEADER_LEN]) body data[header_len:total_len] # 版本2是zlib压缩需要先解压再递归拆 if ver 2: body zlib.decompress(body) return op, body def on_message(ws, message): op, body parse_packet(message) if op OP_MESSAGE: # 正文里可能粘了多个包按长度循环切 offset 0 while offset len(body): total_len struct.unpack(I, body[offset:offset4])[0] packet body[offset:offsettotal_len] _, _, _, inner_op, _ struct.unpack(IHHII, packet[:HEADER_LEN]) if inner_op OP_MESSAGE: payload json.loads(packet[HEADER_LEN:total_len].decode(utf-8)) # 统一内部格式 msg { platform: bilibili, room: payload.get(roomid), uid: payload.get(uid), user: payload.get(uname), type: danmaku, content: payload.get(msg, ), ts: payload.get(timestamp) } dispatch(msg) # 交给下游各姬 offset total_len def dispatch(msg): 所有姬的入口按类型分流 if msg[type] danmaku: reply_ji.handle(msg) # 回复姬 song_ji.handle(msg) # 点歌姬 elif msg[type] in (gift, enter): thanks_ji.handle(msg) # 答谢姬逻辑说明parse_packet先按 16 字节头解出总长和操作码版本为 2 时正文是压缩的必须先解压。解压后的正文里可能一次粘了多个消息包所以on_message里要按每个包自己的长度循环切分不能只取第一个。dispatch是分流中枢弹幕类消息进回复姬和点歌姬礼物和进场类进答谢姬。参数说明心跳间隔一般设 30 秒太短会被限流太长连接会被服务端断开重连退避建议从 1 秒起每次翻倍上限 30 秒避免断线后疯狂重连把 IP 打进黑名单。房间号在开播前后可能变化别写死在配置里开播时动态拉一次。提示B站弹幕的 uid 在未登录观众身上可能是 0做用户去重和积分统计时要把这类匿名用户单独处理否则会把所有人算成同一个人。3. 答谢姬与回复姬规则引擎怎么写才不误触发3.1 答谢姬的触发条件与去重答谢姬要处理的是礼物、上舰、进场这三类事件。最容易翻车的地方是重复答谢同一条礼物消息因为重连或补发被推了两次机器人就谢两遍观众看着很傻。我的做法是在答谢姬里维护一个以「平台用户ID事件类型时间窗口」为键的短期去重集合窗口设 5 到 10 秒用内存里的有序集合就行不用上 Redis单机直播量级完全够。答谢文案要分级。小额礼物走统一模板大额礼物和上舰走单独模板并带上用户名进场欢迎则要限流——一个热门直播间每秒可能进几十个人全欢迎会把弹幕刷爆。我一般给进场欢迎设一个每分钟最多 N 条的令牌桶N 取 5 到 10超出就静默丢弃。import time from collections import deque class ThanksJi: def __init__(self, welcome_per_min6): self.seen {} # 去重键 - 过期时间 self.welcome_bucket deque() # 进场欢迎的令牌桶 self.welcome_per_min welcome_per_min def _dedup(self, key, window8): now time.time() if key in self.seen and self.seen[key] now: return False self.seen[key] now window return True def _allow_welcome(self): now time.time() while self.welcome_bucket and now - self.welcome_bucket[0] 60: self.welcome_bucket.popleft() if len(self.welcome_bucket) self.welcome_per_min: return False self.welcome_bucket.append(now) return True def handle(self, msg): key f{msg[platform]}:{msg[uid]}:{msg[type]} if not self._dedup(key): return if msg[type] gift: self.send_thanks(msg) elif msg[type] enter: if self._allow_welcome(): self.send_welcome(msg) def send_thanks(self, msg): # 实际发送走平台发送接口这里只留占位 print(f感谢 {msg[user]} 的礼物) def send_welcome(self, msg): print(f欢迎 {msg[user]} 进入直播间)逻辑说明_dedup用字典记录每个去重键的过期时间窗口内重复的直接返回 False。_allow_welcome是滑动窗口令牌桶只保留最近 60 秒的记录超过上限就拒绝。两个机制分开是因为礼物去重和进场限流解决的是不同问题混在一起写后面很难调。参数说明去重窗口 8 秒是经验值礼物动画本身有延迟窗口太短挡不住补发太长会把观众连续送的小礼物合并掉。进场欢迎上限按你直播间的实际流速调冷场直播间可以放到 20热门场次压到 3 到 5。3.2 回复姬的关键词匹配与优先级回复姬最容易写成一个大 if-else关键词一多就互相打架。正确做法是给规则排优先级精确匹配 前缀匹配 包含匹配同一优先级内按规则注册顺序。比如「点歌」既是点歌姬的触发词也可能被回复姬的包含规则吃掉所以点歌类关键词要在回复姬里显式排除或者把点歌姬的优先级排在回复姬前面。规则配置建议外置成一张表改文案不用动代码字段含义示例pattern匹配串主播多高match_type匹配方式exact / prefix / containspriority优先级数字越小越先10reply回复文案一米八谢谢关心cooldown同一用户冷却秒数30冷却这个字段必须有。没有冷却一个观众连发十遍同一个问题机器人就回十遍弹幕直接失控。冷却按「用户规则」维度记别按全局记否则一个用户触发后所有人都被挡住。class ReplyJi: def __init__(self, rules): # 按优先级排序小的先匹配 self.rules sorted(rules, keylambda r: r[priority]) self.cooldown {} # (uid, pattern) - 下次可触发时间 def handle(self, msg): text msg[content].strip() for rule in self.rules: if not self._match(text, rule): continue key (msg[uid], rule[pattern]) now time.time() if self.cooldown.get(key, 0) now: return # 命中但在冷却直接放弃不再往下匹配 self.cooldown[key] now rule.get(cooldown, 30) self.send(msg, rule[reply]) return # 一条弹幕只触发一条规则 def _match(self, text, rule): mt rule[match_type] if mt exact: return text rule[pattern] if mt prefix: return text.startswith(rule[pattern]) return rule[pattern] in text def send(self, msg, reply): print(f回复 {msg[user]}: {reply})逻辑说明规则先按优先级排序命中一条就 return保证一条弹幕只触发一条回复避免多个规则同时命中刷屏。冷却命中时也直接 return不再尝试后面的规则否则会出现「这条在冷却但下一条规则又命中了」的意外回复。参数说明cooldown 默认 30 秒问答类可以设长一点到 60互动类可以短到 10。priority 建议按 10 递增留空隙方便后面插规则。4. 点歌姬从弹幕文本到播放队列的转换4.1 歌名提取与队列管理点歌姬的输入是「点歌 歌名」这类弹幕输出是一个待播放队列。核心动作是提取歌名、查曲库、入队。曲库可以是一个本地目录也可以是一张歌名到文件路径的映射表。查不到的歌不要直接丢弃要回一条「没找到这首歌」的提示否则观众以为你没理他。队列要设上限比如 20 首满了就提示「队列已满」。还要处理重复点歌同一首歌已经在队列里可以选择忽略、也可以选择置顶看你的直播风格。我一般用忽略加提示避免一个人反复点同一首把队列占满。class SongJi: def __init__(self, library, max_queue20): self.library library # {歌名: 文件路径} self.queue deque() self.max_queue max_queue def handle(self, msg): text msg[content].strip() if not text.startswith(点歌): return name text[2:].strip() if not name: return if name not in self.library: self.reply(msg, f没找到《{name}》换个名字试试) return if len(self.queue) self.max_queue: self.reply(msg, 队列满了等会儿再点) return if name in self.queue: self.reply(msg, f《{name}》已经在队列里了) return self.queue.append((name, msg[user])) self.reply(msg, f已加入《{name}》) def reply(self, msg, text): print(f回复 {msg[user]}: {text})逻辑说明先判断前缀是不是「点歌」再取歌名。歌名查库失败、队列满、重复三种情况分别给不同提示观众能明确知道为什么没点上。队列存的是歌名和点歌人播放时可以把点歌人一起显示出来增加互动感。参数说明max_queue 按你的曲库大小和直播节奏调曲库几百首的话 20 首够用曲库小就压到 10。歌名匹配建议做一次去空格和大小写归一中文歌名还要考虑繁简不然「晴天」和「晴天 」会被当成两首。4.2 播放器联动与切歌信号队列有了还要让播放器真的播。常见做法是机器人进程和播放器进程之间用一个本地文件或本地 socket 通信机器人往队列文件里追加播放器监听文件变化播完一首就取下一首。这种解耦方式的好处是播放器崩了不影响弹幕接收机器人重启也不用管播放状态。切歌信号要带一个唯一标识避免播放器重复消费同一条。我一般用「时间戳歌名」做标识播放器处理完把标识写回一个已处理列表重启后先读这个列表跳过已处理的。注意播放器读取队列文件时要做文件锁或原子替换直接边写边读会出现读到半条记录的情况表现为偶尔播出一首不存在的歌。5. 避坑排查场控机器人上线后最容易翻的五个地方现象一弹幕收着收着就断了重连后重复收到一批旧消息。原因长连接被服务端按心跳超时断开重连时服务端补发了断线期间的消息而你的去重窗口已经过期。 解决把去重窗口从 8 秒拉长到 30 秒以上或者在重连成功后主动丢弃时间戳早于断线时刻的消息。心跳间隔别超过 30 秒。现象二机器人回复延迟越来越高开播一小时后卡成幻灯片。原因去重字典和冷却字典只增不删内存里堆了几万条过期记录每次查找都在大字典里翻。 解决起一个后台线程定期清理过期键或者用带过期时间的缓存结构。清理间隔 60 秒一次就够别在消息处理主路径里做全量扫描。现象三同一个观众送了礼物答谢了两遍甚至三遍。原因礼物消息在平台侧可能同时从多个通道推送或者你的重连逻辑把同一条消息补了两次而去重键里没带足够的信息。 解决去重键要包含平台、用户、事件类型和消息里的唯一标识比如礼物消息自带的 ID不能只用用户类型。窗口内命中就丢弃。现象四点歌姬把「点歌」两个字也当成歌名去查库。原因前缀截取时没做空值判断或者观众发的就是「点歌」后面没跟内容。 解决截取后 strip 一次为空直接返回不查库也不回复。同时把「点歌」本身从曲库里排除防止真有首歌叫这个名字。现象五回复姬在冷场时疯狂回复把正常弹幕淹没了。原因冷却按全局记而不是按用户记或者根本没设冷却一个观众刷屏就触发一片。 解决冷却必须按「用户规则」维度记并且给每条规则单独设冷却时长。另外可以加一个全局频率上限比如机器人每秒最多发 3 条超出排队或丢弃。6. 进阶把场控机器人做成可观测、可回放的系统前面几章讲的是怎么让它跑起来这一章讲怎么让它跑得久。场控机器人最大的问题是它是黑匣子——出问题时你只能看到「它没回复」或者「它回错了」看不到中间发生了什么。我的习惯是从第一天就把所有进出消息落一份结构化日志每条记录包含原始消息、命中的规则、触发的动作、耗时。这份日志平时没用出问题时就是后悔药。落日志用 JSON Lines 格式一行一条方便后面用脚本回放。回放的价值在于你可以把昨天那场直播的弹幕流原样喂给机器人改一版规则跑一遍对比两版的回复差异不用真的开播去试。下面这个回放脚本我用了很久核心就是读日志、重放、对比。import json def replay(log_path, handler): 把日志里的弹幕重新喂给规则引擎输出命中情况 hits [] with open(log_path, r, encodingutf-8) as f: for line in f: record json.loads(line) msg record[raw] # 原始消息 if msg[type] ! danmaku: continue result handler.dry_run(msg) # dry_run 只判断不发送 hits.append({ user: msg[user], content: msg[content], matched: result }) return hits def diff(old_hits, new_hits): 对比两版规则的命中差异 old_map {(h[user], h[content]): h[matched] for h in old_hits} for h in new_hits: key (h[user], h[content]) if old_map.get(key) ! h[matched]: print(f差异: {h[content]} 旧{old_map.get(key)} 新{h[matched]})逻辑说明replay读 JSON Lines 日志只取弹幕类消息调dry_run做纯判断不发送收集命中结果。diff把两版结果按「用户内容」对齐打印出命中不同的条目。dry_run需要你在回复姬和点歌姬里各加一个只返回判断结果、不执行发送的分支改动很小但收益很大。参数说明日志文件按天切分单场直播的日志量通常在几万到几十万行回放一次几秒钟。日志里别存观众敏感信息用户名可以脱敏uid 做哈希合规又够用。除了回放还有两个可观测指标值得盯一是消息处理延迟从收到弹幕到发出回复的时间超过 2 秒观众就能感觉到卡二是规则命中率如果某条规则一周都没命中过要么是关键词写错了要么是这条规则根本不需要该删就删。我现在的习惯是每周看一次命中率报表把零命中的规则清掉规则表越干净出问题时越好排查。最后说个我踩过的坑别在直播进行中改规则。有一次我边播边调回复姬的关键词改完忘了重新加载配置机器人还在用旧规则跑观众问的问题一个没回我对着屏幕纳闷了十分钟。后来我给自己定了个规矩——改规则必须走「改配置 → 本地回放验证 → 重载 → 看日志确认生效」四步一步都不能省。希望帮到你。本文还有配套的精品资源点击获取
