【免费下载链接】ZerolanLiveRobotAI VTuber with LLM, ASR, TTS, OCR, CV and more technologies to live stream or play Minecraft with you.项目地址https://gitcode.com/gh_mirrors/ze/ZerolanLiveRobot点击查看免费下载如果你想在开源项目ZerolanLiveRobotAI VTuber 直播机器人中扩展一个新服务或者注册自定义事件这篇文章将带你从零走完整条链路事件驱动架构、服务注册机制、TypedEventEmitter 监听器以及一个可参考的完整实战示例。只需跟着做你就能让自己的 AI 直播机器人多出一种新能力。一、先搞懂ZerolanLiveRobot 的事件驱动架构ZerolanLiveRobot 是一款基于事件驱动Event-Driven设计模式的 AI VTuber 直播机器人它没有轮询等待而是有事件才行动。理解了这一点扩展新服务就只是填空题。整个事件系统的核心分布在event/目录下的三个文件文件职责event/registry.py事件名注册表EventKeyRegistry所有事件类型必须在这里登记event/event_data.py所有事件数据类继承自BaseEventevent/event_emitter.py核心TypedEventEmitter负责事件派发与监听器执行其中 BaseEvent 是所有事件的基类只需一个type字段标记事件类型class BaseEvent(BaseModel): type: str而emitter是一个全局单例event/event_emitter.py#L250提供两个最关键的 APIemitter.on(event_key)注册监听器。异步函数会作为协程任务执行同步函数则丢进默认 4 个 worker 的线程池执行emitter.emit(event)发送事件触发所有已注册的监听器。 一个直观的例子QQ 服务收到消息后并不是直接调用 LLM而是发出QQMessageEvent事件services/qqbot/napcat.py#L80-L98再由 bot.py 中注册的监听器去处理。服务之间完全解耦这就是扩展的精髓。二、扩展一个新服务的标准流程按照官方约定每个服务放在services/下独立文件夹中并包含专属的config.py。整体流程分 6 步定义配置 → 定义事件 → 实现服务类 → 挂到 ServiceConfig → 在 Context 中实例化 → 在 Bot 中启动并监听下面用一个天气播报服务完整走一遍假设你已 clone 仓库并在自己的工作分支上操作。第 1 步定义服务配置新建services/weather/config.py用 Pydantic 模型声明配置项。description里的说明会直接展示在 WebUI 配置页面上由 common/generator/gradio_gen.py 自动生成from pydantic import BaseModel, Field class WeatherServiceConfig(BaseModel): enable: bool Field(defaultFalse, description是否启用天气播报服务。) city: str Field(default北京, description要播报天气的城市。) interval: int Field(default60, description推送间隔秒。)然后在 services/config.py 的ServiceConfig中挂上它这一步会让你的新服务自动出现在resources/config.yaml的service段里from services.weather.config import WeatherServiceConfig class ServiceConfig(BaseModel): # ... 原有字段 weather: WeatherServiceConfig Field( defaultWeatherServiceConfig(), descriptionConfiguration for the Weather service. )第 2 步注册自定义事件自定义事件分两处登记。先在 event/registry.py 的事件名注册表中新增一个类别命名建议沿用service.xxx.yyy的风格class EventKeyRegistry: # ... class Weather: UPDATE service.weather.update再在 event/event_data.py 中定义事件数据类把要传递的字段声明清楚Pydantic 会自动做校验class WeatherUpdateEvent(BaseEvent): city: str text: str type: str EventKeyRegistry.Weather.UPDATE第 3 步实现服务类新建services/weather/service.py。如果服务需要长期后台运行循环拉取数据、维持连接等建议继承 ThreadRunnable——它在start()时会自动注册到全局可运行组件表中程序退出时能被统一stop()回收import time from loguru import logger from common.concurrent.abs_runnable import ThreadRunnable from event.event_data import WeatherUpdateEvent from event.event_emitter import emitter from services.weather.config import WeatherServiceConfig class WeatherService(ThreadRunnable): def __init__(self, config: WeatherServiceConfig): super().__init__() self._config config def start(self): super().start() while self._activate: text self._fetch_weather() # 换成你自己的真实 API 调用 emitter.emit(WeatherUpdateEvent(cityself._config.city, texttext)) logger.info(f天气推送: {self._config.city} {text}) time.sleep(self._config.interval) def _fetch_weather(self) - str: return 晴26°C def name(self): return WeatherService def stop(self): self._activate False logger.info(WeatherService stopped.)注意服务内部只负责发事件不要在这里直接调用 LLM、TTS 等重逻辑——处理逻辑留给监听器这是保持架构解耦的关键。第 4 步在 Context 中实例化framework/context.py 是资源初始化的唯一入口其类注释明确要求只放资源初始化不放事件逻辑。参考 QQ 服务的写法framework/context.py#L127-L130按配置开关实例化if _config.service.weather.enable: from services.weather.service import WeatherService self.weather WeatherService(_config.service.weather)第 5 步在 Bot 中启动服务线程服务的后台线程在 bot.py 的start()方法中统一拉起。参照 OBS 客户端的启动方式用KillableThread包装成守护线程if self.weather: weather_thread KillableThread(targetself.weather.start, daemonTrue, nameWeatherThread) threads.append(weather_thread)第 6 步注册监听器让事件产生价值最后在 bot.py 的init()方法中用装饰器注册监听器把天气事件交给 LLM 用语音播报出去emitter.on(EventKeyRegistry.Weather.UPDATE) def on_weather_update(event: WeatherUpdateEvent): logger.info(f收到天气事件: {event.city} {event.text}) self.emit_llm_prediction( f{event.city}现在{event.text}。请用一句话在直播间播报一下天气。 )至此事件链闭环完成WeatherService 定时emit→ 监听器收到WeatherUpdateEvent→emit_llm_prediction驱动 LLM 推理 → 触发 TTS 播报bot.py#L486-L487。启动python main.py入口见 main.py并在配置文件中打开service.weather.enable新服务就跑起来了。三、进阶自定义事件的注册规则与注意事项同步监听器 vs 异步监听器TypedEventEmitter在派发时会检查监听器类型event/event_emitter.py#L206-L216async def监听器作为asyncio.Task提交到事件循环适合调用异步管线xxAsyncPipeline普通def监听器提交到线程池默认 4 个 worker适合耗时 IO 操作若只需执行一次用emitter.once(event_key)替代on触发后自动移除。⚠️ 三条实战避坑指南不要阻塞主线程。事件循环运行在主线程上任何同步阻塞调用都会卡死整个机器人README 中明确警告过这一点。注意线程安全。同步监听器运行在线程池中多个监听器可能并发访问同一服务涉及共享状态时请加锁或避免共享可变数据。留意 5 秒超时告警。每个任务默认带 5 秒Timer超时只会在日志中打印 warningevent/event_emitter.py#L37-L60可用于诊断耗时过长的监听器。四、相关文件速查路径作用event/registry.py事件名注册表新事件键在这里登记event/event_data.py事件数据类定义event/event_emitter.pyTypedEventEmitter 核心实现services/config.py所有服务配置的聚合入口framework/context.py服务实例化的唯一入口bot.py服务线程启动与事件监听器注册common/concurrent/abs_runnable.pyThreadRunnable / AsyncRunnable 基类五、总结扩展 ZerolanLiveRobot 的新服务本质上是沿一条清晰的链路做填空config.py声明配置 →event/registry.py登记事件名 →event/event_data.py定义事件类 → 服务类中emitter.emit发事件 →context.py实例化 →bot.py启动线程并注册监听器。由于服务之间通过事件解耦你添加的天气服务、硬件控制服务或任何自定义能力都不会侵入现有代码逻辑。掌握这套事件驱动 服务注册的扩展模式后你就可以基于 LLM、ASR、TTS、OCR、CV 这些现成管线pipeline/目录构建完全属于自己的 AI VTuber 直播机器人了。️赞分享【免费下载链接】ZerolanLiveRobotAI VTuber with LLM, ASR, TTS, OCR, CV and more technologies to live stream or play Minecraft with you.项目地址https://gitcode.com/gh_mirrors/ze/ZerolanLiveRobot点击查看免费下载相关推荐Strapi strapi/openapi 包扩展指南如何开发并注册一个自定义 Routes ProviderStrapi strapi/openapi 包扩展指南如何开发并注册一个自定义 Routes Provider 本文以 Strapi 仓库中 strapi后端CMS前端企业应用Ising-Decoder-SurfaceCode-1-Fast量子纠错领域的革命性表面码解码器Ising Decoder SurfaceCode 1 Fast量子纠错领域的革命性表面码解码器 Ising Decoder SurfaceCode 1 FaStarRailAssistant扩展系统开发者指南如何从零写一个自定义扩展StarRailAssistant扩展系统开发者指南如何从零写一个自定义扩展 StarRailAssistantSRA是一款基于图像识别的 崩坏星穹铁道自创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
【免费下载链接】ZerolanLiveRobotAI VTuber with LLM, ASR, TTS, OCR, CV and more technologies to live stream or play Minecraft with you.项目地址https://gitcode.com/gh_mirrors/ze/ZerolanLiveRobot点击查看免费下载如果你想在开源项目ZerolanLiveRobotAI VTuber 直播机器人中扩展一个新服务或者注册自定义事件这篇文章将带你从零走完整条链路事件驱动架构、服务注册机制、TypedEventEmitter 监听器以及一个可参考的完整实战示例。只需跟着做你就能让自己的 AI 直播机器人多出一种新能力。一、先搞懂ZerolanLiveRobot 的事件驱动架构ZerolanLiveRobot 是一款基于事件驱动Event-Driven设计模式的 AI VTuber 直播机器人它没有轮询等待而是有事件才行动。理解了这一点扩展新服务就只是填空题。整个事件系统的核心分布在event/目录下的三个文件文件职责event/registry.py事件名注册表EventKeyRegistry所有事件类型必须在这里登记event/event_data.py所有事件数据类继承自BaseEventevent/event_emitter.py核心TypedEventEmitter负责事件派发与监听器执行其中 BaseEvent 是所有事件的基类只需一个type字段标记事件类型class BaseEvent(BaseModel): type: str而emitter是一个全局单例event/event_emitter.py#L250提供两个最关键的 APIemitter.on(event_key)注册监听器。异步函数会作为协程任务执行同步函数则丢进默认 4 个 worker 的线程池执行emitter.emit(event)发送事件触发所有已注册的监听器。 一个直观的例子QQ 服务收到消息后并不是直接调用 LLM而是发出QQMessageEvent事件services/qqbot/napcat.py#L80-L98再由 bot.py 中注册的监听器去处理。服务之间完全解耦这就是扩展的精髓。二、扩展一个新服务的标准流程按照官方约定每个服务放在services/下独立文件夹中并包含专属的config.py。整体流程分 6 步定义配置 → 定义事件 → 实现服务类 → 挂到 ServiceConfig → 在 Context 中实例化 → 在 Bot 中启动并监听下面用一个天气播报服务完整走一遍假设你已 clone 仓库并在自己的工作分支上操作。第 1 步定义服务配置新建services/weather/config.py用 Pydantic 模型声明配置项。description里的说明会直接展示在 WebUI 配置页面上由 common/generator/gradio_gen.py 自动生成from pydantic import BaseModel, Field class WeatherServiceConfig(BaseModel): enable: bool Field(defaultFalse, description是否启用天气播报服务。) city: str Field(default北京, description要播报天气的城市。) interval: int Field(default60, description推送间隔秒。)然后在 services/config.py 的ServiceConfig中挂上它这一步会让你的新服务自动出现在resources/config.yaml的service段里from services.weather.config import WeatherServiceConfig class ServiceConfig(BaseModel): # ... 原有字段 weather: WeatherServiceConfig Field( defaultWeatherServiceConfig(), descriptionConfiguration for the Weather service. )第 2 步注册自定义事件自定义事件分两处登记。先在 event/registry.py 的事件名注册表中新增一个类别命名建议沿用service.xxx.yyy的风格class EventKeyRegistry: # ... class Weather: UPDATE service.weather.update再在 event/event_data.py 中定义事件数据类把要传递的字段声明清楚Pydantic 会自动做校验class WeatherUpdateEvent(BaseEvent): city: str text: str type: str EventKeyRegistry.Weather.UPDATE第 3 步实现服务类新建services/weather/service.py。如果服务需要长期后台运行循环拉取数据、维持连接等建议继承 ThreadRunnable——它在start()时会自动注册到全局可运行组件表中程序退出时能被统一stop()回收import time from loguru import logger from common.concurrent.abs_runnable import ThreadRunnable from event.event_data import WeatherUpdateEvent from event.event_emitter import emitter from services.weather.config import WeatherServiceConfig class WeatherService(ThreadRunnable): def __init__(self, config: WeatherServiceConfig): super().__init__() self._config config def start(self): super().start() while self._activate: text self._fetch_weather() # 换成你自己的真实 API 调用 emitter.emit(WeatherUpdateEvent(cityself._config.city, texttext)) logger.info(f天气推送: {self._config.city} {text}) time.sleep(self._config.interval) def _fetch_weather(self) - str: return 晴26°C def name(self): return WeatherService def stop(self): self._activate False logger.info(WeatherService stopped.)注意服务内部只负责发事件不要在这里直接调用 LLM、TTS 等重逻辑——处理逻辑留给监听器这是保持架构解耦的关键。第 4 步在 Context 中实例化framework/context.py 是资源初始化的唯一入口其类注释明确要求只放资源初始化不放事件逻辑。参考 QQ 服务的写法framework/context.py#L127-L130按配置开关实例化if _config.service.weather.enable: from services.weather.service import WeatherService self.weather WeatherService(_config.service.weather)第 5 步在 Bot 中启动服务线程服务的后台线程在 bot.py 的start()方法中统一拉起。参照 OBS 客户端的启动方式用KillableThread包装成守护线程if self.weather: weather_thread KillableThread(targetself.weather.start, daemonTrue, nameWeatherThread) threads.append(weather_thread)第 6 步注册监听器让事件产生价值最后在 bot.py 的init()方法中用装饰器注册监听器把天气事件交给 LLM 用语音播报出去emitter.on(EventKeyRegistry.Weather.UPDATE) def on_weather_update(event: WeatherUpdateEvent): logger.info(f收到天气事件: {event.city} {event.text}) self.emit_llm_prediction( f{event.city}现在{event.text}。请用一句话在直播间播报一下天气。 )至此事件链闭环完成WeatherService 定时emit→ 监听器收到WeatherUpdateEvent→emit_llm_prediction驱动 LLM 推理 → 触发 TTS 播报bot.py#L486-L487。启动python main.py入口见 main.py并在配置文件中打开service.weather.enable新服务就跑起来了。三、进阶自定义事件的注册规则与注意事项同步监听器 vs 异步监听器TypedEventEmitter在派发时会检查监听器类型event/event_emitter.py#L206-L216async def监听器作为asyncio.Task提交到事件循环适合调用异步管线xxAsyncPipeline普通def监听器提交到线程池默认 4 个 worker适合耗时 IO 操作若只需执行一次用emitter.once(event_key)替代on触发后自动移除。⚠️ 三条实战避坑指南不要阻塞主线程。事件循环运行在主线程上任何同步阻塞调用都会卡死整个机器人README 中明确警告过这一点。注意线程安全。同步监听器运行在线程池中多个监听器可能并发访问同一服务涉及共享状态时请加锁或避免共享可变数据。留意 5 秒超时告警。每个任务默认带 5 秒Timer超时只会在日志中打印 warningevent/event_emitter.py#L37-L60可用于诊断耗时过长的监听器。四、相关文件速查路径作用event/registry.py事件名注册表新事件键在这里登记event/event_data.py事件数据类定义event/event_emitter.pyTypedEventEmitter 核心实现services/config.py所有服务配置的聚合入口framework/context.py服务实例化的唯一入口bot.py服务线程启动与事件监听器注册common/concurrent/abs_runnable.pyThreadRunnable / AsyncRunnable 基类五、总结扩展 ZerolanLiveRobot 的新服务本质上是沿一条清晰的链路做填空config.py声明配置 →event/registry.py登记事件名 →event/event_data.py定义事件类 → 服务类中emitter.emit发事件 →context.py实例化 →bot.py启动线程并注册监听器。由于服务之间通过事件解耦你添加的天气服务、硬件控制服务或任何自定义能力都不会侵入现有代码逻辑。掌握这套事件驱动 服务注册的扩展模式后你就可以基于 LLM、ASR、TTS、OCR、CV 这些现成管线pipeline/目录构建完全属于自己的 AI VTuber 直播机器人了。️赞分享【免费下载链接】ZerolanLiveRobotAI VTuber with LLM, ASR, TTS, OCR, CV and more technologies to live stream or play Minecraft with you.项目地址https://gitcode.com/gh_mirrors/ze/ZerolanLiveRobot点击查看免费下载相关推荐Strapi strapi/openapi 包扩展指南如何开发并注册一个自定义 Routes ProviderStrapi strapi/openapi 包扩展指南如何开发并注册一个自定义 Routes Provider 本文以 Strapi 仓库中 strapi后端CMS前端企业应用Ising-Decoder-SurfaceCode-1-Fast量子纠错领域的革命性表面码解码器Ising Decoder SurfaceCode 1 Fast量子纠错领域的革命性表面码解码器 Ising Decoder SurfaceCode 1 FaStarRailAssistant扩展系统开发者指南如何从零写一个自定义扩展StarRailAssistant扩展系统开发者指南如何从零写一个自定义扩展 StarRailAssistantSRA是一款基于图像识别的 崩坏星穹铁道自创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
