者指南：如何擴展一個新服務并注冊自定義事件（附完整實戰(zhàn)示例）)
【免費下載鏈接】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 直播機器人中擴展一個新服務或者注冊自定義事件這篇文章將帶你從零走完整條鏈路事件驅(qū)動架構(gòu)、服務注冊機制、TypedEventEmitter 監(jiān)聽器以及一個可參考的完整實戰(zhàn)示例。只需跟著做你就能讓自己的 AI 直播機器人多出一種新能力。一、先搞懂ZerolanLiveRobot 的事件驅(qū)動架構(gòu)ZerolanLiveRobot 是一款基于事件驅(qū)動Event-Driven設(shè)計模式的 AI VTuber 直播機器人它沒有輪詢等待而是有事件才行動。理解了這一點擴展新服務就只是填空題。整個事件系統(tǒng)的核心分布在event/目錄下的三個文件文件職責event/registry.py事件名注冊表EventKeyRegistry所有事件類型必須在這里登記event/event_data.py所有事件數(shù)據(jù)類繼承自BaseEventevent/event_emitter.py核心TypedEventEmitter負責事件派發(fā)與監(jiān)聽器執(zhí)行其中 BaseEvent 是所有事件的基類只需一個type字段標記事件類型class BaseEvent(BaseModel): type: str而emitter是一個全局單例event/event_emitter.py#L250提供兩個最關(guān)鍵的 APIemitter.on(event_key)注冊監(jiān)聽器。異步函數(shù)會作為協(xié)程任務執(zhí)行同步函數(shù)則丟進默認 4 個 worker 的線程池執(zhí)行emitter.emit(event)發(fā)送事件觸發(fā)所有已注冊的監(jiān)聽器。 一個直觀的例子QQ 服務收到消息后并不是直接調(diào)用 LLM而是發(fā)出QQMessageEvent事件services/qqbot/napcat.py#L80-L98再由 bot.py 中注冊的監(jiān)聽器去處理。服務之間完全解耦這就是擴展的精髓。二、擴展一個新服務的標準流程按照官方約定每個服務放在services/下獨立文件夾中并包含專屬的config.py。整體流程分 6 步定義配置 → 定義事件 → 實現(xiàn)服務類 → 掛到 ServiceConfig → 在 Context 中實例化 → 在 Bot 中啟動并監(jiān)聽下面用一個天氣播報服務完整走一遍假設(shè)你已 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中掛上它這一步會讓你的新服務自動出現(xiàn)在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 中定義事件數(shù)據(jù)類把要傳遞的字段聲明清楚Pydantic 會自動做校驗class WeatherUpdateEvent(BaseEvent): city: str text: str type: str EventKeyRegistry.Weather.UPDATE第 3 步實現(xiàn)服務類新建services/weather/service.py。如果服務需要長期后臺運行循環(huán)拉取數(shù)據(jù)、維持連接等建議繼承 ThreadRunnable——它在start()時會自動注冊到全局可運行組件表中程序退出時能被統(tǒng)一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 調(diào)用 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.)注意服務內(nèi)部只負責發(fā)事件不要在這里直接調(diào)用 LLM、TTS 等重邏輯——處理邏輯留給監(jiān)聽器這是保持架構(gòu)解耦的關(guān)鍵。第 4 步在 Context 中實例化framework/context.py 是資源初始化的唯一入口其類注釋明確要求只放資源初始化不放事件邏輯。參考 QQ 服務的寫法framework/context.py#L127-L130按配置開關(guān)實例化if _config.service.weather.enable: from services.weather.service import WeatherService self.weather WeatherService(_config.service.weather)第 5 步在 Bot 中啟動服務線程服務的后臺線程在 bot.py 的start()方法中統(tǒng)一拉起。參照 OBS 客戶端的啟動方式用KillableThread包裝成守護線程if self.weather: weather_thread KillableThread(targetself.weather.start, daemonTrue, nameWeatherThread) threads.append(weather_thread)第 6 步注冊監(jiān)聽器讓事件產(chǎn)生價值最后在 bot.py 的init()方法中用裝飾器注冊監(jiān)聽器把天氣事件交給 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}現(xiàn)在{event.text}。請用一句話在直播間播報一下天氣。 )至此事件鏈閉環(huán)完成WeatherService 定時emit→ 監(jiān)聽器收到WeatherUpdateEvent→emit_llm_prediction驅(qū)動 LLM 推理 → 觸發(fā) TTS 播報bot.py#L486-L487。啟動python main.py入口見 main.py并在配置文件中打開service.weather.enable新服務就跑起來了。三、進階自定義事件的注冊規(guī)則與注意事項同步監(jiān)聽器 vs 異步監(jiān)聽器TypedEventEmitter在派發(fā)時會檢查監(jiān)聽器類型event/event_emitter.py#L206-L216async def監(jiān)聽器作為asyncio.Task提交到事件循環(huán)適合調(diào)用異步管線xxAsyncPipeline普通def監(jiān)聽器提交到線程池默認 4 個 worker適合耗時 IO 操作若只需執(zhí)行一次用emitter.once(event_key)替代on觸發(fā)后自動移除。?? 三條實戰(zhàn)避坑指南不要阻塞主線程。事件循環(huán)運行在主線程上任何同步阻塞調(diào)用都會卡死整個機器人README 中明確警告過這一點。注意線程安全。同步監(jiān)聽器運行在線程池中多個監(jiān)聽器可能并發(fā)訪問同一服務涉及共享狀態(tài)時請加鎖或避免共享可變數(shù)據(jù)。留意 5 秒超時告警。每個任務默認帶 5 秒Timer超時只會在日志中打印 warningevent/event_emitter.py#L37-L60可用于診斷耗時過長的監(jiān)聽器。四、相關(guān)文件速查路徑作用event/registry.py事件名注冊表新事件鍵在這里登記event/event_data.py事件數(shù)據(jù)類定義event/event_emitter.pyTypedEventEmitter 核心實現(xiàn)services/config.py所有服務配置的聚合入口framework/context.py服務實例化的唯一入口bot.py服務線程啟動與事件監(jiān)聽器注冊common/concurrent/abs_runnable.pyThreadRunnable / AsyncRunnable 基類五、總結(jié)擴展 ZerolanLiveRobot 的新服務本質(zhì)上是沿一條清晰的鏈路做填空config.py聲明配置 →event/registry.py登記事件名 →event/event_data.py定義事件類 → 服務類中emitter.emit發(fā)事件 →context.py實例化 →bot.py啟動線程并注冊監(jiān)聽器。由于服務之間通過事件解耦你添加的天氣服務、硬件控制服務或任何自定義能力都不會侵入現(xiàn)有代碼邏輯。掌握這套事件驅(qū)動 服務注冊的擴展模式后你就可以基于 LLM、ASR、TTS、OCR、CV 這些現(xiàn)成管線pipeline/目錄構(gòu)建完全屬于自己的 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點擊查看免費下載相關(guān)推薦Strapi strapi/openapi 包擴展指南如何開發(fā)并注冊一個自定義 Routes ProviderStrapi strapi/openapi 包擴展指南如何開發(fā)并注冊一個自定義 Routes Provider 本文以 Strapi 倉庫中 strapi后端CMS前端企業(yè)應用Ising-Decoder-SurfaceCode-1-Fast量子糾錯領(lǐng)域的革命性表面碼解碼器Ising Decoder SurfaceCode 1 Fast量子糾錯領(lǐng)域的革命性表面碼解碼器 Ising Decoder SurfaceCode 1 FaStarRailAssistant擴展系統(tǒng)開發(fā)者指南如何從零寫一個自定義擴展StarRailAssistant擴展系統(tǒng)開發(fā)者指南如何從零寫一個自定義擴展 StarRailAssistantSRA是一款基于圖像識別的 崩壞星穹鐵道自創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 ZerolanLiveRobot開發(fā)者指南：如何擴展一個新服務并注冊自定義事件（附完整實戰(zhàn)示例）
【免費下載鏈接】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 直播機器人中擴展一個新服務或者注冊自定義事件這篇文章將帶你從零走完整條鏈路事件驅(qū)動架構(gòu)、服務注冊機制、TypedEventEmitter 監(jiān)聽器以及一個可參考的完整實戰(zhàn)示例。只需跟著做你就能讓自己的 AI 直播機器人多出一種新能力。一、先搞懂ZerolanLiveRobot 的事件驅(qū)動架構(gòu)ZerolanLiveRobot 是一款基于事件驅(qū)動Event-Driven設(shè)計模式的 AI VTuber 直播機器人它沒有輪詢等待而是有事件才行動。理解了這一點擴展新服務就只是填空題。整個事件系統(tǒng)的核心分布在event/目錄下的三個文件文件職責event/registry.py事件名注冊表EventKeyRegistry所有事件類型必須在這里登記event/event_data.py所有事件數(shù)據(jù)類繼承自BaseEventevent/event_emitter.py核心TypedEventEmitter負責事件派發(fā)與監(jiān)聽器執(zhí)行其中 BaseEvent 是所有事件的基類只需一個type字段標記事件類型class BaseEvent(BaseModel): type: str而emitter是一個全局單例event/event_emitter.py#L250提供兩個最關(guān)鍵的 APIemitter.on(event_key)注冊監(jiān)聽器。異步函數(shù)會作為協(xié)程任務執(zhí)行同步函數(shù)則丟進默認 4 個 worker 的線程池執(zhí)行emitter.emit(event)發(fā)送事件觸發(fā)所有已注冊的監(jiān)聽器。 一個直觀的例子QQ 服務收到消息后并不是直接調(diào)用 LLM而是發(fā)出QQMessageEvent事件services/qqbot/napcat.py#L80-L98再由 bot.py 中注冊的監(jiān)聽器去處理。服務之間完全解耦這就是擴展的精髓。二、擴展一個新服務的標準流程按照官方約定每個服務放在services/下獨立文件夾中并包含專屬的config.py。整體流程分 6 步定義配置 → 定義事件 → 實現(xiàn)服務類 → 掛到 ServiceConfig → 在 Context 中實例化 → 在 Bot 中啟動并監(jiān)聽下面用一個天氣播報服務完整走一遍假設(shè)你已 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中掛上它這一步會讓你的新服務自動出現(xiàn)在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 中定義事件數(shù)據(jù)類把要傳遞的字段聲明清楚Pydantic 會自動做校驗class WeatherUpdateEvent(BaseEvent): city: str text: str type: str EventKeyRegistry.Weather.UPDATE第 3 步實現(xiàn)服務類新建services/weather/service.py。如果服務需要長期后臺運行循環(huán)拉取數(shù)據(jù)、維持連接等建議繼承 ThreadRunnable——它在start()時會自動注冊到全局可運行組件表中程序退出時能被統(tǒng)一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 調(diào)用 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.)注意服務內(nèi)部只負責發(fā)事件不要在這里直接調(diào)用 LLM、TTS 等重邏輯——處理邏輯留給監(jiān)聽器這是保持架構(gòu)解耦的關(guān)鍵。第 4 步在 Context 中實例化framework/context.py 是資源初始化的唯一入口其類注釋明確要求只放資源初始化不放事件邏輯。參考 QQ 服務的寫法framework/context.py#L127-L130按配置開關(guān)實例化if _config.service.weather.enable: from services.weather.service import WeatherService self.weather WeatherService(_config.service.weather)第 5 步在 Bot 中啟動服務線程服務的后臺線程在 bot.py 的start()方法中統(tǒng)一拉起。參照 OBS 客戶端的啟動方式用KillableThread包裝成守護線程if self.weather: weather_thread KillableThread(targetself.weather.start, daemonTrue, nameWeatherThread) threads.append(weather_thread)第 6 步注冊監(jiān)聽器讓事件產(chǎn)生價值最后在 bot.py 的init()方法中用裝飾器注冊監(jiān)聽器把天氣事件交給 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}現(xiàn)在{event.text}。請用一句話在直播間播報一下天氣。 )至此事件鏈閉環(huán)完成WeatherService 定時emit→ 監(jiān)聽器收到WeatherUpdateEvent→emit_llm_prediction驅(qū)動 LLM 推理 → 觸發(fā) TTS 播報bot.py#L486-L487。啟動python main.py入口見 main.py并在配置文件中打開service.weather.enable新服務就跑起來了。三、進階自定義事件的注冊規(guī)則與注意事項同步監(jiān)聽器 vs 異步監(jiān)聽器TypedEventEmitter在派發(fā)時會檢查監(jiān)聽器類型event/event_emitter.py#L206-L216async def監(jiān)聽器作為asyncio.Task提交到事件循環(huán)適合調(diào)用異步管線xxAsyncPipeline普通def監(jiān)聽器提交到線程池默認 4 個 worker適合耗時 IO 操作若只需執(zhí)行一次用emitter.once(event_key)替代on觸發(fā)后自動移除。?? 三條實戰(zhàn)避坑指南不要阻塞主線程。事件循環(huán)運行在主線程上任何同步阻塞調(diào)用都會卡死整個機器人README 中明確警告過這一點。注意線程安全。同步監(jiān)聽器運行在線程池中多個監(jiān)聽器可能并發(fā)訪問同一服務涉及共享狀態(tài)時請加鎖或避免共享可變數(shù)據(jù)。留意 5 秒超時告警。每個任務默認帶 5 秒Timer超時只會在日志中打印 warningevent/event_emitter.py#L37-L60可用于診斷耗時過長的監(jiān)聽器。四、相關(guān)文件速查路徑作用event/registry.py事件名注冊表新事件鍵在這里登記event/event_data.py事件數(shù)據(jù)類定義event/event_emitter.pyTypedEventEmitter 核心實現(xiàn)services/config.py所有服務配置的聚合入口framework/context.py服務實例化的唯一入口bot.py服務線程啟動與事件監(jiān)聽器注冊common/concurrent/abs_runnable.pyThreadRunnable / AsyncRunnable 基類五、總結(jié)擴展 ZerolanLiveRobot 的新服務本質(zhì)上是沿一條清晰的鏈路做填空config.py聲明配置 →event/registry.py登記事件名 →event/event_data.py定義事件類 → 服務類中emitter.emit發(fā)事件 →context.py實例化 →bot.py啟動線程并注冊監(jiān)聽器。由于服務之間通過事件解耦你添加的天氣服務、硬件控制服務或任何自定義能力都不會侵入現(xiàn)有代碼邏輯。掌握這套事件驅(qū)動 服務注冊的擴展模式后你就可以基于 LLM、ASR、TTS、OCR、CV 這些現(xiàn)成管線pipeline/目錄構(gòu)建完全屬于自己的 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點擊查看免費下載相關(guān)推薦Strapi strapi/openapi 包擴展指南如何開發(fā)并注冊一個自定義 Routes ProviderStrapi strapi/openapi 包擴展指南如何開發(fā)并注冊一個自定義 Routes Provider 本文以 Strapi 倉庫中 strapi后端CMS前端企業(yè)應用Ising-Decoder-SurfaceCode-1-Fast量子糾錯領(lǐng)域的革命性表面碼解碼器Ising Decoder SurfaceCode 1 Fast量子糾錯領(lǐng)域的革命性表面碼解碼器 Ising Decoder SurfaceCode 1 FaStarRailAssistant擴展系統(tǒng)開發(fā)者指南如何從零寫一個自定義擴展StarRailAssistant擴展系統(tǒng)開發(fā)者指南如何從零寫一個自定義擴展 StarRailAssistantSRA是一款基于圖像識別的 崩壞星穹鐵道自創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考