Related Series
- AI VTuber Agent Build Log Part 1: Overview
- AI VTuber Agent Build Log Part 2: LLM and Persona
- AI VTuber Agent Build Log Part 3: Memory
- AI VTuber Agent Build Log Part 4: STT
- AI VTuber Agent Build Log Part 5: TTS
- AI VTuber Agent Build Log Part 6: Live2D
- AI VTuber Agent Build Log Part 7: Broadcast Integration (current)
- AI VTuber Agent Build Log Part 8: Production
Introduction
By the end of Part 6, the agent can talk, remember, listen, and move on screen.
But it still can't stream.
Putting it in OBS is about more than just making it visible. It needs to read Twitch and YouTube chat in real time and react to it. Once a stream starts, scenarios should advance automatically, and scene transitions should be driven from code. The whole flow — going live, greeting viewers, moving to a game segment, wrapping up — needs to be automated.
What Part 7 covers:
- Overall broadcast integration architecture
- OBSController — WebSocket v5, scene switching, source control, stream start/stop
- TwitchChatReader — twitchio IRC, badge and subscription parsing
- YouTubeChatReader — Data API v3, polling, quota management
- ChatQueue — priority queue, cooldown, spam filtering
- ChatRouter — chat → LLM + TTS loop
- BroadcastScheduler — phase-based autonomous streaming
- SSE dashboard, JSON logging, systemd/Docker restart
1. Overall Broadcast Integration Architecture
Here's the data flow once broadcast integration is in place.
[Twitch IRC] ──→┐
├→ [ChatQueue] ──→ [ChatRouter] ──→ [LLM] ──→ [TTS] ──→ [Live2D]
[YouTube API] ──→┘ │
└──→ [Twitch chat output]
[BroadcastScheduler]
↕ scene/stream control
[OBSController]
WebSocket v5
BroadcastScheduler is the conductor of the entire flow. When it issues a start-stream command, it tells OBS to switch scenes, spins up the chat readers for each platform, and handles automated lines and transitions phase by phase.
Chat from both platforms flows into ChatQueue, gets sorted by priority, and ChatRouter pulls messages one at a time and sends them through the LLM → TTS pipeline.
Package installation:
uv add obsws-python twitchio google-api-python-client structlog
2. OBSController — WebSocket v5
OBS 28 and later ship with a built-in WebSocket server. obsws-python wraps the v5 protocol so it can be used synchronously from Python.
Config & Scene Enum
# src/vtuber/broadcast/obs_controller.py
from __future__ import annotations
import logging
from dataclasses import dataclass
from enum import Enum
import obsws_python as obs
logger = logging.getLogger(__name__)
@dataclass
class OBSConfig:
host: str = "localhost"
port: int = 4455 # OBS WebSocket default port
password: str = ""
timeout: int = 10
class BroadcastScene(str, Enum):
"""Must match the scene names in OBS exactly — including spaces and capitalization"""
IDLE = "대기화면"
GREETING = "인사화면"
TALK = "잡담화면"
GAME = "게임화면"
ENDING = "마무리화면"
BRB = "잠시자리비움"
Core Implementation
class OBSController:
def __init__(self, config: OBSConfig):
self._config = config
self._client: obs.ReqClient | None = None
def connect(self) -> None:
self._client = obs.ReqClient(
host=self._config.host,
port=self._config.port,
password=self._config.password,
timeout=self._config.timeout,
)
logger.info("OBS WebSocket connected", extra={"host": self._config.host, "port": self._config.port})
def disconnect(self) -> None:
if self._client:
try:
self._client.disconnect()
except Exception:
pass
self._client = None
def _require_client(self) -> obs.ReqClient:
if self._client is None:
self.connect()
return self._client
# ── Scene control ─────────────────────────────────────────────────
def switch_scene(self, scene: BroadcastScene) -> None:
self._require_client().set_current_program_scene(scene.value)
logger.info("Scene switched", extra={"scene": scene.value})
def get_current_scene(self) -> str:
resp = self._require_client().get_current_program_scene()
return resp.current_program_scene_name
# ── Source control ────────────────────────────────────────────────
def set_source_visible(
self,
scene_name: str,
source_name: str,
visible: bool,
) -> None:
"""Show/hide a source in a given scene — v5 requires fetching the scene_item_id first"""
resp = self._require_client().get_scene_item_id(
scene_name=scene_name,
source_name=source_name,
)
self._require_client().set_scene_item_enabled(
scene_name=scene_name,
scene_item_id=resp.scene_item_id,
scene_item_enabled=visible,
)
def set_filter_enabled(
self,
source_name: str,
filter_name: str,
enabled: bool,
) -> None:
"""Enable/disable a source filter — e.g. chroma key"""
self._require_client().set_source_filter_enabled(
source_name=source_name,
filter_name=filter_name,
filter_enabled=enabled,
)
# ── Stream control ────────────────────────────────────────────────
def start_stream(self) -> None:
self._require_client().start_stream()
logger.info("Stream started")
def stop_stream(self) -> None:
self._require_client().stop_stream()
logger.info("Stream stopped")
def is_streaming(self) -> bool:
resp = self._require_client().get_stream_status()
return resp.output_active
def start_recording(self) -> None:
self._require_client().start_record()
def stop_recording(self) -> None:
self._require_client().stop_record()
set_source_visible makes an extra call to get_scene_item_id. OBS WebSocket v5 treats sources as items within a scene, so you need the item ID before you can do anything with them. This is the part that trips up most people migrating from v4.
3. TwitchChatReader
Twitch chat integration uses the twitchio library, which is IRC-based and can parse badge and subscription information.
ChatMessage Common Model
Normalize incoming chat from both platforms into a single shared structure.
# src/vtuber/broadcast/chat_models.py
from __future__ import annotations
from dataclasses import dataclass, field
from enum import IntEnum
class ChatPriority(IntEnum):
"""낮은 값이 높은 우선순위 — asyncio.PriorityQueue 특성"""
VIP_SUB = 0
MOD = 1
SUB = 2
FOLLOWER = 3
NORMAL = 4
@dataclass
class ChatMessage:
platform: str # "twitch" | "youtube"
username: str
text: str
priority: ChatPriority = ChatPriority.NORMAL
is_mod: bool = False
is_sub: bool = False
is_vip: bool = False
is_member: bool = False # YouTube 멤버
raw: dict = field(default_factory=dict)
TwitchChatReader Implementation
# src/vtuber/broadcast/twitch_chat.py
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass
import twitchio
from twitchio.ext import commands
from .chat_models import ChatMessage, ChatPriority
logger = logging.getLogger(__name__)
@dataclass
class TwitchConfig:
token: str # "oauth:..." 형식
bot_nick: str
channels: list[str]
class TwitchChatReader(commands.Bot):
"""Twitch IRC 채팅 수신 → ChatMessage 변환 후 큐에 삽입"""
def __init__(self, config: TwitchConfig, message_queue: asyncio.Queue):
super().__init__(
token=config.token,
nick=config.bot_nick,
prefix="!",
initial_channels=config.channels,
)
self._queue = message_queue
async def event_ready(self) -> None:
logger.info("Twitch 봇 준비", extra={"nick": self.nick})
async def event_message(self, msg: twitchio.Message) -> None:
if msg.echo: # 봇 자신이 보낸 메시지는 스킵
return
parsed = self._parse(msg)
await self._queue.put(parsed)
logger.debug(
"Twitch 채팅 수신",
extra={"user": parsed.username, "priority": int(parsed.priority), "text": parsed.text[:40]},
)
def _parse(self, msg: twitchio.Message) -> ChatMessage:
author = msg.author
badges = author.badges or {}
is_mod = bool(author.is_mod)
is_sub = bool(author.is_subscriber)
is_vip = "vip" in badges
if is_vip and is_sub:
priority = ChatPriority.VIP_SUB
elif is_mod:
priority = ChatPriority.MOD
elif is_sub:
priority = ChatPriority.SUB
else:
priority = ChatPriority.NORMAL
return ChatMessage(
platform="twitch",
username=author.name,
text=msg.content,
priority=priority,
is_mod=is_mod,
is_sub=is_sub,
is_vip=is_vip,
raw={"badges": badges},
)
twitchio requires commands.Bot to run on an asyncio event loop. Launch it from BroadcastScheduler with asyncio.create_task(twitch_reader.start()).
4. YouTubeChatReader
The YouTube Live Chat API is REST-based polling — there is no WebSocket. Poll at an appropriate interval by respecting the pollingIntervalMillis field in the response.
The default project quota is 10,000 units per day. Listing live chat messages costs at least 5 units per request. Poll too aggressively and you will exhaust the quota mid-stream.
# src/vtuber/broadcast/youtube_chat.py
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from .chat_models import ChatMessage, ChatPriority
logger = logging.getLogger(__name__)
@dataclass
class YouTubeConfig:
api_key: str
video_id: str
class YouTubeChatReader:
"""YouTube Live Chat API v3 폴링 기반 채팅 수신"""
def __init__(
self,
api_key: str,
live_chat_id: str,
message_queue: asyncio.Queue,
):
self._api_key = api_key
self._live_chat_id = live_chat_id
self._queue = message_queue
self._page_token: str | None = None
self._running = False
@classmethod
async def from_video_id(
cls,
api_key: str,
video_id: str,
message_queue: asyncio.Queue,
) -> "YouTubeChatReader":
"""video_id → liveChatId 조회"""
youtube = build("youtube", "v3", developerKey=api_key)
resp = await asyncio.to_thread(
lambda: youtube.videos().list(
part="liveStreamingDetails",
id=video_id,
).execute()
)
items = resp.get("items", [])
if not items:
raise ValueError(f"video_id 없음: {video_id}")
live_chat_id = (
items[0]
.get("liveStreamingDetails", {})
.get("activeLiveChatId")
)
if not live_chat_id:
raise ValueError(f"라이브 채팅 없음 (방송 종료 상태?): {video_id}")
logger.info("YouTube liveChatId 조회", extra={"live_chat_id": live_chat_id})
return cls(api_key, live_chat_id, message_queue)
async def run(self) -> None:
self._running = True
youtube = build("youtube", "v3", developerKey=self._api_key)
while self._running:
try:
await self._poll_once(youtube)
except HttpError as e:
if e.resp.status == 403:
logger.error("YouTube API 쿼터 초과", extra={"error": str(e)})
await asyncio.sleep(60)
else:
logger.warning("YouTube API 오류", extra={"error": str(e)})
await asyncio.sleep(5)
except Exception as e:
logger.warning("YouTube 폴링 실패", extra={"error": str(e)})
await asyncio.sleep(5)
async def _poll_once(self, youtube) -> None:
resp = await asyncio.to_thread(
lambda: youtube.liveChatMessages().list(
liveChatId=self._live_chat_id,
part="snippet,authorDetails",
pageToken=self._page_token,
maxResults=200,
).execute()
)
for item in resp.get("items", []):
msg = self._parse_item(item)
if msg:
await self._queue.put(msg)
self._page_token = resp.get("nextPageToken")
next_poll_ms = resp.get("pollingIntervalMillis", 5000)
logger.debug("YouTube 폴링", extra={
"count": len(resp.get("items", [])),
"next_poll_ms": next_poll_ms,
})
await asyncio.sleep(next_poll_ms / 1000.0)
def _parse_item(self, item: dict) -> ChatMessage | None:
snippet = item.get("snippet", {})
author_details = item.get("authorDetails", {})
# textMessageEvent만 처리 — 슈퍼챗·멤버십은 7편에서 스킵
if snippet.get("type") != "textMessageEvent":
return None
text = snippet.get("textMessageDetails", {}).get("messageText", "").strip()
username = author_details.get("displayName", "")
if not text:
return None
is_mod = bool(author_details.get("isChatModerator"))
is_member = bool(author_details.get("isChatSponsor")) # 멤버십
if is_mod:
priority = ChatPriority.MOD
elif is_member:
priority = ChatPriority.SUB
else:
priority = ChatPriority.NORMAL
return ChatMessage(
platform="youtube",
username=username,
text=text,
priority=priority,
is_mod=is_mod,
is_member=is_member,
raw=item,
)
def stop(self) -> None:
self._running = False
pollingIntervalMillis is managed by the YouTube server — it decreases when chat is active and increases when things are quiet. Ignoring this value and polling at a fixed interval will either drain your quota quickly or make chat responses lag behind.
5. ChatQueue — Priority Queue + Cooldown + Spam Filter
When chat floods in, having Haru respond to every message creates two problems: LLM costs blow up, and the TTS queue saturates.
ChatQueue handles three things:
- Priority ordering — VIP_SUB > MOD > SUB > FOLLOWER > NORMAL
- Cooldown — minimum interval between processing messages from the same user
- Spam filter — repeated characters, messages that are too short
# src/vtuber/broadcast/chat_queue.py
from __future__ import annotations
import asyncio
import logging
import re
import time
from collections import defaultdict
from .chat_models import ChatMessage, ChatPriority
logger = logging.getLogger(__name__)
COOLDOWN_SECONDS = 30.0 # 같은 유저 재처리 최소 간격
MIN_TEXT_LENGTH = 2 # 이 길이 미만은 스킵
SPAM_PATTERNS = [
r"^(.)\1{4,}$", # aaaaa... 반복 문자
r"^[\W\d]+$", # 특수문자/숫자만으로 구성
]
class ChatQueue:
"""asyncio.PriorityQueue 래퍼 — 쿨다운 및 스팸 필터 포함"""
def __init__(self, maxsize: int = 50):
self._pq: asyncio.PriorityQueue[tuple[int, float, ChatMessage]] = (
asyncio.PriorityQueue(maxsize=maxsize)
)
self._last_processed: dict[str, float] = defaultdict(float)
async def put(self, msg: ChatMessage) -> bool:
"""필터 통과 시 큐에 삽입. 필터에 걸리면 False 반환."""
if not self._should_process(msg):
return False
if self._pq.full():
logger.warning("ChatQueue 포화 — 메시지 드롭", extra={"user": msg.username})
return False
# 동일 우선순위 내에서는 time.monotonic()으로 FIFO 보장
await self._pq.put((int(msg.priority), time.monotonic(), msg))
return True
async def get(self) -> ChatMessage:
_, _, msg = await self._pq.get()
return msg
def qsize(self) -> int:
return self._pq.qsize()
def task_done(self) -> None:
self._pq.task_done()
def mark_processed(self, msg: ChatMessage) -> None:
"""처리 완료 후 쿨다운 타임스탬프 기록"""
key = f"{msg.platform}:{msg.username}"
self._last_processed[key] = time.monotonic()
def _compute_priority(self, msg: ChatMessage) -> ChatPriority:
return msg.priority
def _should_process(self, msg: ChatMessage) -> bool:
# 쿨다운 확인
key = f"{msg.platform}:{msg.username}"
last_time = self._last_processed[key]
if time.monotonic() - last_time < COOLDOWN_SECONDS:
logger.debug("쿨다운 적용", extra={"user": msg.username})
return False
# 길이 필터
if len(msg.text.strip()) < MIN_TEXT_LENGTH:
return False
# 스팸 패턴 필터
for pattern in SPAM_PATTERNS:
if re.fullmatch(pattern, msg.text.strip(), re.IGNORECASE):
logger.debug("스팸 필터", extra={"user": msg.username, "text": msg.text})
return False
return True
The reason the tuple inserted into PriorityQueue is structured as (priority_int, monotonic_time, msg): when comparing tuples, Python falls back to the second element when the first is equal, so FIFO ordering within the same priority level is guaranteed automatically. Making ChatMessage directly comparable would require defining __lt__, which is more trouble than it's worth.
6. ChatRouter — Chat → LLM + TTS Loop
ChatRouter is the main loop that pulls messages from ChatQueue, sends them to the LLM, and pipes the responses into TTS.
# src/vtuber/broadcast/chat_router.py
from __future__ import annotations
import asyncio
import logging
from ..character.state import CharacterState
from ..llm.client import LLMClient, LLMMessage
from ..tts.orchestrator import TTSOrchestrator
from ..character.prompt_builder import PromptBuilder
from .chat_models import ChatMessage
from .chat_queue import ChatQueue
logger = logging.getLogger(__name__)
CHAT_RESPONSE_SYSTEM_SUFFIX = """
지금은 라이브 방송 중이다.
채팅 메시지에 응답할 때는 1~3문장으로 짧게. 방송 흐름을 방해하지 말 것.
"""
class ChatRouter:
"""ChatQueue에서 메시지를 꺼내 LLM + TTS 파이프라인으로 흘림"""
def __init__(
self,
queue: ChatQueue,
llm: LLMClient,
orchestrator: TTSOrchestrator,
state: CharacterState,
prompt_builder: PromptBuilder,
twitch_bot=None, # 선택: Twitch 채팅창 텍스트 출력용
):
self._queue = queue
self._llm = llm
self._orchestrator = orchestrator
self._state = state
self._prompt_builder = prompt_builder
self._twitch_bot = twitch_bot
self._running = False
async def run(self) -> None:
self._running = True
logger.info("ChatRouter 시작")
while self._running:
try:
msg = await asyncio.wait_for(self._queue.get(), timeout=1.0)
await self._process(msg)
self._queue.mark_processed(msg)
self._queue.task_done()
except asyncio.TimeoutError:
continue
except Exception as e:
logger.exception("ChatRouter 처리 오류", extra={"error": str(e)})
async def _process(self, msg: ChatMessage) -> None:
logger.info(
"채팅 처리",
extra={"user": msg.username, "text": msg.text[:40], "priority": int(msg.priority)},
)
system_prompt = (
self._prompt_builder.build_system_prompt(self._state)
+ CHAT_RESPONSE_SYSTEM_SUFFIX
)
user_text = f"{msg.username}: {msg.text}"
messages = [
LLMMessage(role="system", content=system_prompt),
LLMMessage(role="user", content=user_text),
]
llm_stream = self._llm.chat_stream(messages)
reply_text = await self._orchestrator.process_llm_stream(
llm_stream,
emotion=self._state.current_emotion,
)
# Twitch 채팅창 출력 — 봇이 연결된 경우
if self._twitch_bot and reply_text:
try:
channel = self._twitch_bot.get_channel(
self._twitch_bot.initial_channels[0]
)
if channel:
await channel.send(f"[하루] {reply_text[:200]}")
except Exception as e:
logger.warning("Twitch 채팅 출력 실패", extra={"error": str(e)})
def stop(self) -> None:
self._running = False
The reason reply_text[:200] is used when posting to Twitch chat: Twitch has a 500-character message limit, and even though the prompt instructs Haru to keep responses short, the LLM occasionally generates longer outputs. Capping at 200 characters keeps things safe in virtually all cases.
7. BroadcastScheduler — Phase-Based Autonomous Broadcasting
For a broadcast to run autonomously, it needs scenario phase management. The broadcast progresses through greeting → chat → game → wrap-up phases, switching OBS scenes at each transition and periodically firing automatic utterances.
Phase Definitions
# src/vtuber/broadcast/scheduler.py
from __future__ import annotations
import asyncio
import logging
import time
from dataclasses import dataclass
from .obs_controller import OBSController, BroadcastScene
from .chat_router import ChatRouter
from .chat_queue import ChatQueue
from ..tts.orchestrator import TTSOrchestrator
from ..character.state import CharacterState
logger = logging.getLogger(__name__)
@dataclass
class BroadcastPhase:
name: str
scene: BroadcastScene
duration_sec: float # Phase duration (seconds)
auto_talk_sec: float = 60.0 # Auto-talk interval (0 = disabled)
auto_talk_prompt: str = "" # LLM directive for auto-talk
on_enter: str = "" # Fixed text to speak immediately on phase entry
DEFAULT_PHASES: list[BroadcastPhase] = [
BroadcastPhase(
name="greeting",
scene=BroadcastScene.GREETING,
duration_sec=300, # 5 minutes
auto_talk_sec=0,
on_enter="안녕하세요! 하루예요. 오늘 방송 시작할게요. 잠시 준비할게요!",
),
BroadcastPhase(
name="chat",
scene=BroadcastScene.TALK,
duration_sec=2400, # 40 minutes
auto_talk_sec=90.0,
auto_talk_prompt=(
"라이브 방송 중 잡담 코너다. "
"자연스럽게 혼잣말이나 시청자에게 말을 걸어라. 1~2문장."
),
),
BroadcastPhase(
name="game",
scene=BroadcastScene.GAME,
duration_sec=3600, # 1 hour
auto_talk_sec=30.0,
auto_talk_prompt=(
"게임 코너다. 게임하면서 나올 법한 짧은 혼잣말이나 리액션을 해라. 1문장으로 짧게."
),
),
BroadcastPhase(
name="ending",
scene=BroadcastScene.ENDING,
duration_sec=300, # 5 minutes
auto_talk_sec=0,
on_enter="오늘 방송도 함께해 주셔서 감사해요. 다음에 또 만나요!",
),
]
PHASE_ORDER = [phase.name for phase in DEFAULT_PHASES]
Scheduler Body
class BroadcastScheduler:
def __init__(
self,
obs: OBSController,
queue: ChatQueue,
router: ChatRouter,
orchestrator: TTSOrchestrator,
state: CharacterState,
llm,
phases: list[BroadcastPhase] | None = None,
):
self._obs = obs
self._queue = queue
self._router = router
self._orch = orchestrator
self._state = state
self._llm = llm
self._phases = {p.name: p for p in (phases or DEFAULT_PHASES)}
self._running = False
self._current_phase: str | None = None
self._tasks: list[asyncio.Task] = []
async def start_broadcast(self) -> None:
"""Start the broadcast — turn on the OBS stream and enter the scenario"""
logger.info("Starting broadcast")
self._obs.start_stream()
self._running = True
self._tasks.append(asyncio.create_task(self._router.run()))
await self._run_schedule()
async def stop_broadcast(self) -> None:
self._running = False
self._router.stop()
for task in self._tasks:
task.cancel()
self._tasks.clear()
self._obs.stop_stream()
logger.info("Broadcast ended")
async def _run_schedule(self) -> None:
for phase_name in PHASE_ORDER:
if not self._running:
break
await self._enter_phase(phase_name)
await self.stop_broadcast()
async def _enter_phase(self, phase_name: str) -> None:
phase = self._phases[phase_name]
self._current_phase = phase_name
logger.info("Phase transition", extra={"phase": phase_name, "scene": phase.scene.value})
self._obs.switch_scene(phase.scene)
if phase.on_enter:
await self._orch.speak_text(phase.on_enter, emotion="happy")
start = time.monotonic()
last_auto_talk = start
while time.monotonic() - start < phase.duration_sec and self._running:
await asyncio.sleep(1)
# Auto-talk check — only when not processing chat
if (
phase.auto_talk_sec > 0
and time.monotonic() - last_auto_talk >= phase.auto_talk_sec
and self._queue.qsize() == 0
):
await self._auto_talk(phase)
last_auto_talk = time.monotonic()
async def _auto_talk(self, phase: BroadcastPhase) -> None:
"""Auto-talk to fill broadcast silence when there are no chat messages"""
from ..llm.client import LLMMessage
messages = [
LLMMessage(role="system", content=phase.auto_talk_prompt),
LLMMessage(role="user", content="지금 방송 상황에 맞는 발화를 해줘."),
]
stream = self._llm.chat_stream(messages)
await self._orch.process_llm_stream(stream, emotion=self._state.current_emotion)
logger.debug("Auto-talk fired", extra={"phase": phase.name})
Among the auto-talk trigger conditions, self._queue.qsize() == 0 is critical. If auto-talk fires while chat messages are still queued up, the TTS pipeline gets overloaded. Auto-talk should only kick in when there are no pending chat messages.
8. SSE Monitoring Dashboard & Structured Logging
To inspect system state in real time during a stream, you need somewhere to surface it.
JSON Structured Logging — structlog
Plain text logs from the standard logging module are a pain to analyze during a stream incident. Use structlog to emit JSON instead.
# src/vtuber/logger.py
import logging
import sys
import structlog
def setup_logging(level: str = "INFO") -> None:
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.stdlib.add_log_level,
structlog.stdlib.add_logger_name,
structlog.processors.TimeStamper(fmt="iso"),
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.JSONRenderer(),
],
wrapper_class=structlog.make_filtering_bound_logger(
logging.getLevelName(level)
),
logger_factory=structlog.PrintLoggerFactory(sys.stdout),
cache_logger_on_first_use=True,
)
A call like logger.info("씬 전환", extra={"scene": "게임화면"}) produces JSON output like this:
{"event": "씬 전환", "scene": "게임화면", "level": "info", "logger": "vtuber.broadcast.obs_controller", "timestamp": "2026-05-04T13:00:00Z"}
This makes it straightforward to parse with jq or ship to an ELK stack later.
SSE Dashboard Endpoint
Use FastAPI's StreamingResponse to push stream state to clients.
# src/vtuber/api/monitor.py
from __future__ import annotations
import asyncio
import json
import time
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
router = APIRouter(prefix="/monitor", tags=["monitor"])
_scheduler = None
_queue = None
def init_monitor(scheduler, queue) -> None:
global _scheduler, _queue
_scheduler = scheduler
_queue = queue
@router.get("/stream")
async def monitor_stream():
"""SSE — real-time stream status (2-second interval)"""
async def event_generator():
while True:
data = {
"ts": time.time(),
"is_running": _scheduler._running if _scheduler else False,
"phase": _scheduler._current_phase if _scheduler else None,
"queue_size": _queue.qsize() if _queue else 0,
"obs_scene": None,
}
if _scheduler and _scheduler._obs:
try:
data["obs_scene"] = _scheduler._obs.get_current_scene()
except Exception:
pass
yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"
await asyncio.sleep(2)
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
},
)
Connect from a browser with EventSource("/monitor/stream") and you get a stream status update every two seconds. obs_scene, queue_size, and phase alone are enough to tell at a glance whether the stream is running normally.
systemd Service Registration
You need automatic restart on server reboot or abnormal process exit.
# /etc/systemd/system/haru-agent.service
[Unit]
Description=AI VTuber Haru Agent
After=network.target
[Service]
Type=simple
User=ubuntu
WorkingDirectory=/home/ubuntu/vtuber-assistant
ExecStart=/home/ubuntu/vtuber-assistant/.venv/bin/uvicorn src.vtuber.main:app --host 0.0.0.0 --port 8000
Restart=on-failure
RestartSec=5
StandardOutput=journal
StandardError=journal
Environment=PYTHONPATH=/home/ubuntu/vtuber-assistant
[Install]
WantedBy=multi-user.target
sudo systemctl daemon-reload
sudo systemctl enable haru-agent
sudo systemctl start haru-agent
# check status
sudo systemctl status haru-agent
journalctl -u haru-agent -f # tail logs in real time
Docker restart Policy
If you're running in Docker, use restart: unless-stopped.
# docker-compose.yml
services:
haru-agent:
build: .
restart: unless-stopped
ports:
- "8000:8000"
env_file: .env
volumes:
- ./assets:/app/assets
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 60s
The difference between on-failure and unless-stopped: on-failure restarts only when the exit code is non-zero. unless-stopped restarts unconditionally unless you explicitly stop the container with docker stop. For a streaming server, unless-stopped is the right choice.
9. Putting It All Together — main.py
# src/vtuber/main.py (broadcast-related section)
import asyncio
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from .broadcast.obs_controller import OBSController, OBSConfig
from .broadcast.twitch_chat import TwitchChatReader, TwitchConfig
from .broadcast.youtube_chat import YouTubeChatReader, YouTubeConfig
from .broadcast.chat_queue import ChatQueue
from .broadcast.chat_router import ChatRouter
from .broadcast.scheduler import BroadcastScheduler
from .api.monitor import router as monitor_router, init_monitor
from .logger import setup_logging
@asynccontextmanager
async def lifespan(app: FastAPI):
setup_logging()
queue = ChatQueue(maxsize=50)
obs = OBSController(OBSConfig(
host=os.getenv("OBS_HOST", "localhost"),
port=int(os.getenv("OBS_PORT", "4455")),
password=os.getenv("OBS_PASSWORD", ""),
))
# llm, orchestrator, state, prompt_builder — see the respective posts
# ...
router = ChatRouter(
queue=queue,
llm=llm,
orchestrator=orchestrator,
state=state,
prompt_builder=prompt_builder,
)
scheduler = BroadcastScheduler(
obs=obs,
queue=queue,
router=router,
orchestrator=orchestrator,
state=state,
llm=llm,
)
init_monitor(scheduler, queue)
# Twitch bot — launched as a background task
twitch_config = TwitchConfig(
token=os.getenv("TWITCH_TOKEN", ""),
bot_nick=os.getenv("TWITCH_BOT_NICK", ""),
channels=[os.getenv("TWITCH_CHANNEL", "")],
)
twitch_reader = TwitchChatReader(twitch_config, queue)
asyncio.create_task(twitch_reader.start())
app.state.scheduler = scheduler
app.state.obs = obs
yield
await scheduler.stop_broadcast()
obs.disconnect()
app = FastAPI(lifespan=lifespan)
app.include_router(monitor_router)
The stream is started via an HTTP API call. The YouTube video_id isn't known until just before the stream goes live, so it's accepted through an API endpoint.
# src/vtuber/api/broadcast.py
import asyncio
from fastapi import APIRouter, Request
from ..broadcast.obs_controller import BroadcastScene, OBSController
from ..broadcast.scheduler import BroadcastScheduler
from ..broadcast.youtube_chat import YouTubeChatReader
import os
router = APIRouter(prefix="/broadcast", tags=["broadcast"])
@router.post("/start")
async def start_broadcast(request: Request):
scheduler: BroadcastScheduler = request.app.state.scheduler
asyncio.create_task(scheduler.start_broadcast())
return {"status": "started"}
@router.post("/stop")
async def stop_broadcast(request: Request):
scheduler: BroadcastScheduler = request.app.state.scheduler
await scheduler.stop_broadcast()
return {"status": "stopped"}
@router.post("/youtube/connect")
async def connect_youtube(video_id: str, request: Request):
"""Start the YouTube chat reader — call this once the video_id is available after going live"""
queue = request.app.state.scheduler._queue
youtube_reader = await YouTubeChatReader.from_video_id(
api_key=os.getenv("YOUTUBE_API_KEY", ""),
video_id=video_id,
message_queue=queue,
)
asyncio.create_task(youtube_reader.run())
return {"status": "connected", "video_id": video_id}
@router.post("/scene")
async def switch_scene(scene: str, request: Request):
obs: OBSController = request.app.state.obs
try:
obs.switch_scene(BroadcastScene(scene))
return {"status": "ok", "scene": scene}
except ValueError:
return {"status": "error", "msg": f"알 수 없는 씬: {scene}"}
Wrap-Up
Haru's streaming stack, completed through Part 7:
| Part | Component | Role |
|---|---|---|
| 2 | LLM + Persona | Brain |
| 3 | Memory | Memory |
| 4 | STT | Ears |
| 5 | TTS | Voice |
| 6 | Live2D | Body |
| 7 | Broadcast Integration | Stage |
OBSController, TwitchChatReader, YouTubeChatReader, ChatQueue, ChatRouter, BroadcastScheduler — each component has a clearly separated responsibility. You can swap or extend any one of them without touching the others.
Part 8 will dissect what actually broke when this went live: TTS memory leaks, YouTube API quota exhaustion, OBS reconnection failures, and Twitch bot bans — one by one.
The design is done. Production is another story.