Related Series
- Building an AI VTuber Agent, Part 1: Overview
- Building an AI VTuber Agent, Part 2: LLM and Persona
- Building an AI VTuber Agent, Part 3: Memory
- Building an AI VTuber Agent, Part 4: STT
- Building an AI VTuber Agent, Part 5: TTS
- Building an AI VTuber Agent, Part 6: Live2D
- Building an AI VTuber Agent, Part 7: Streaming Integration
- Building an AI VTuber Agent, Part 8: Operations (current post)
Introduction
At the end of Part 7, I wrote that the series was wrapping up there.
Then I actually ran a stream and changed my mind.
Parts 1 through 7 read more like design documents — a blueprint of well-structured code that fits together neatly in theory. But thirty minutes into the first stream, the TTS server hit 4 GB of memory. The second stream blew the YouTube API quota. On the third, OBS restarted and left the WebSocket permanently disconnected, so an entire day of streaming went by without a single scene transition. On the fourth night, the Twitch bot account got banned.
Writing code and operating code are different disciplines. Working in a dev environment is just the baseline.
What Part 8 covers:
- Real-world failure cases & fixes — four things that actually broke
- Metrics & alerting — catching problems before they become incidents
- Advanced chat interactions — superchats, channel points, and a fast-response cache
- Vision AI integration — one day of teaching the agent to read the game screen
- Final end-to-end deployment architecture
- Cost breakdown & retrospective
1. Real-World Incident Cases & Solutions
Case 1: TTS Server Memory Leak
Symptom: TTS server memory grew linearly with broadcast duration. After a 2-hour stream, the process died with OOM.
Cause: AudioSegment objects created by the TTSOrchestrator from Part 5 were never released by the GC. Python's reference-counting GC doesn't handle circular references.
# 문제 코드 — src/vtuber/tts/orchestrator.py
class TTSOrchestrator:
def __init__(self):
self._active_segments: list[AudioSegment] = [] # 이게 문제
async def process_sentence(self, text: str) -> AudioSegment:
seg = await self._backend.synthesize(text)
self._active_segments.append(seg) # 재생 후에도 해제 안 됨
return seg
This list was originally created to track currently-playing segments, but the code to remove entries after playback finished was never added.
# 수정 코드
import weakref
import gc
class TTSOrchestrator:
def __init__(self):
# weakref으로 변경 — 외부 참조가 없으면 자동 gc
self._active_segments: list[weakref.ref] = []
self._cleanup_interval = 50 # 50 세그먼트마다 정리
self._processed_count = 0
async def process_sentence(self, text: str) -> AudioSegment:
seg = await self._backend.synthesize(text)
self._active_segments.append(weakref.ref(seg))
self._processed_count += 1
if self._processed_count % self._cleanup_interval == 0:
self._cleanup_dead_refs()
return seg
def _cleanup_dead_refs(self) -> None:
before = len(self._active_segments)
self._active_segments = [r for r in self._active_segments if r() is not None]
gc.collect()
after = len(self._active_segments)
logger.info("TTS 세그먼트 gc", extra={"before": before, "after": after})
The more fundamental fix is to remove the _active_segments list entirely. If you actually check whether "tracking currently-playing segments" is necessary, the answer is almost always no.
Case 2: YouTube API Quota Exceeded
Symptom: YouTube chat completely stopped after 1 hour 30 minutes into the stream. HttpError 403 quotaExceeded.
Cause: The Part 7 code did respect pollingIntervalMillis, but every call to from_video_id() at stream start consumed 50 quota units to look up the liveChatId. Combined with repeated lookups during testing, more than half of the 10,000-unit daily quota was gone before the stream even started.
Fix: Cache the liveChatId and reduce the number of lookups.
# src/vtuber/broadcast/youtube_chat.py — 개선
import json
from pathlib import Path
CACHE_FILE = Path(".youtube_livechat_cache.json")
class YouTubeChatReader:
@classmethod
async def from_video_id(
cls,
api_key: str,
video_id: str,
message_queue: asyncio.Queue,
use_cache: bool = True,
) -> "YouTubeChatReader":
"""video_id에서 liveChatId 조회 — 캐시 우선"""
live_chat_id = None
if use_cache and CACHE_FILE.exists():
cached = json.loads(CACHE_FILE.read_text())
if cached.get("video_id") == video_id:
live_chat_id = cached.get("live_chat_id")
logger.info("YouTube liveChatId 캐시 히트", extra={"video_id": video_id})
if not live_chat_id:
youtube = build("youtube", "v3", developerKey=api_key)
resp = 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}")
if use_cache:
CACHE_FILE.write_text(
json.dumps({"video_id": video_id, "live_chat_id": live_chat_id})
)
return cls(api_key, live_chat_id, message_queue)
Also, adjust the polling interval dynamically: back off when chat is quiet, return to the API-recommended interval when messages arrive.
async def _poll_once(self) -> None:
resp = await asyncio.to_thread(self._fetch_messages)
messages = resp.get("items", [])
next_poll_ms = resp.get("pollingIntervalMillis", 5000)
if not messages:
self._empty_poll_count = getattr(self, "_empty_poll_count", 0) + 1
# 빈 응답 5회 연속 → 간격 2배, 최대 30초
if self._empty_poll_count >= 5:
next_poll_ms = min(next_poll_ms * 2, 30_000)
else:
self._empty_poll_count = 0
next_poll_ms = resp.get("pollingIntervalMillis", 5000)
# ... 메시지 처리
self._page_token = resp.get("nextPageToken")
await asyncio.sleep(next_poll_ms / 1000.0)
Case 3: OBS WebSocket Reconnection Failure
Symptom: After OBS restarted, the WebSocket connection dropped, but the _client object remained in an invalid state, causing every OBS call to throw ConnectionRefusedError. For an entire day, the system appeared to be streaming normally — but scenes were never actually switching.
Cause: OBSController._require_client() from Part 7 only checked _client is None. Even after the connection dropped, the _client object still existed, so the check passed.
# 수정된 OBSController — src/vtuber/broadcast/obs_controller.py
class OBSController:
def __init__(self, config: OBSConfig):
self._config = config
self._client = None
self._connected = False
def _require_client(self) -> obs.ReqClient:
if not self._client or not self._connected:
self._reconnect()
return self._client
def _reconnect(self) -> None:
"""재연결 시도 — 최대 3회, 지수 백오프"""
max_attempts = 3
for attempt in range(1, max_attempts + 1):
try:
if self._client:
try:
self._client.disconnect()
except Exception:
pass
self._client = obs.ReqClient(
host=self._config.host,
port=self._config.port,
password=self._config.password,
timeout=self._config.timeout,
)
# 연결 테스트 — GetVersion은 쿼터 소비 없음
self._client.get_version()
self._connected = True
logger.info("OBS 재연결 성공", extra={"attempt": attempt})
return
except Exception as e:
logger.warning(
"OBS 재연결 실패",
extra={"attempt": attempt, "error": str(e)},
)
if attempt < max_attempts:
time.sleep(2 ** attempt) # 2s → 4s
self._connected = False
raise RuntimeError("OBS 재연결 최대 시도 횟수 초과")
def switch_scene(self, scene: BroadcastScene) -> None:
try:
self._require_client().set_current_program_scene(scene.value)
except Exception:
self._connected = False # 즉시 플래그 처리 — 다음 호출에서 재연결 시도
raise
Setting _connected = False immediately on error is what ensures a reconnect attempt on the next call. Swallowing the error lets the broken state persist indefinitely.
Case 4: Twitch Bot Ban
Symptom: On the morning of day 4, the chat connection dropped, and every reconnect attempt was immediately cut. Twitch had suspended the bot account.
Cause: During high chat activity, the bot sent too many responses in a short window. Twitch's chat send rate limit is 20 messages per 30 seconds for regular accounts. ChatRouter had no awareness of this limit when outputting responses to chat.
# src/vtuber/broadcast/twitch_ratelimit.py
import asyncio
import time
class TwitchRateLimiter:
"""Twitch 채팅 전송 rate limit 관리 — 30초에 20개"""
def __init__(self, limit: int = 20, window_sec: float = 30.0):
self._limit = limit
self._window = window_sec
self._timestamps: list[float] = []
self._lock = asyncio.Lock()
async def acquire(self) -> None:
"""rate limit 초과 시 대기"""
async with self._lock:
now = time.monotonic()
self._timestamps = [t for t in self._timestamps if now - t < self._window]
if len(self._timestamps) >= self._limit:
oldest = self._timestamps[0]
wait_sec = self._window - (now - oldest) + 0.1
logger.warning("Twitch rate limit 대기", extra={"wait_sec": round(wait_sec, 2)})
await asyncio.sleep(wait_sec)
self._timestamps.append(time.monotonic())
The real root fix: since all chat responses are already read aloud via TTS, posting text to chat was made selective rather than universal. There was no good reason to echo every response into the chat window.
2. Metrics & Alerting
Failures should be caught before they happen, not after. You can't stare at a monitor throughout the entire stream.
Prometheus Metric Collection
prometheus-fastapi-instrumentator automatically captures HTTP request metrics. Custom metrics are defined separately.
uv add prometheus-fastapi-instrumentator prometheus-client
# src/vtuber/metrics.py
from prometheus_client import Counter, Gauge, Histogram
CHAT_PROCESSED = Counter(
"haru_chat_processed_total",
"처리된 채팅 메시지 수",
["platform", "priority"],
)
CHAT_DROPPED = Counter(
"haru_chat_dropped_total",
"필터링/드롭된 채팅 수",
["reason"],
)
TTS_LATENCY = Histogram(
"haru_tts_latency_seconds",
"TTS 처리 레이턴시",
buckets=[0.1, 0.5, 1, 2, 5, 10],
)
LLM_LATENCY = Histogram(
"haru_llm_latency_seconds",
"LLM 응답 레이턴시",
buckets=[0.5, 1, 2, 5, 10, 30],
)
QUEUE_SIZE = Gauge("haru_chat_queue_size", "현재 채팅 큐 크기")
OBS_CONNECTED = Gauge("haru_obs_connected", "OBS 연결 상태 (1=연결, 0=끊김)")
# src/vtuber/main.py — Prometheus 엔드포인트 등록
from prometheus_fastapi_instrumentator import Instrumentator
from .metrics import *
app = FastAPI(lifespan=lifespan)
Instrumentator().instrument(app).expose(app)
Recording metrics inside ChatRouter._process():
import time
from .metrics import LLM_LATENCY, CHAT_PROCESSED, QUEUE_SIZE
async def _process(self, msg: ChatMessage) -> None:
start = time.monotonic()
# ... 기존 처리 로직 ...
LLM_LATENCY.observe(time.monotonic() - start)
CHAT_PROCESSED.labels(
platform=msg.platform,
priority=str(self._queue._compute_priority(msg)),
).inc()
QUEUE_SIZE.set(self._queue.qsize())
Alerting — Slack Webhook
Grafana Alerts are fine, but receiving only the critical events via Slack is more than enough.
# src/vtuber/alerting.py
import httpx
import os
from datetime import datetime
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL", "")
async def send_alert(message: str, level: str = "warning") -> None:
"""Send an operational alert to Slack."""
if not SLACK_WEBHOOK_URL:
return
emoji = {"info": "ℹ️", "warning": "⚠️", "error": "🚨"}.get(level, "📢")
payload = {
"text": f"{emoji} *하루 운영 알림*\n{message}\n_{datetime.now().strftime('%H:%M:%S')}_"
}
try:
async with httpx.AsyncClient(timeout=5) as client:
resp = await client.post(SLACK_WEBHOOK_URL, json=payload)
resp.raise_for_status()
except Exception as e:
logger.error("슬랙 알림 전송 실패", extra={"error": str(e)})
When to fire alerts is managed by adding a health-monitor loop to BroadcastScheduler.
# src/vtuber/broadcast/scheduler.py — adding _health_monitor
import psutil
from ..alerting import send_alert
class BroadcastScheduler:
async def _health_monitor(self) -> None:
"""Check key metrics every 30 seconds and send alerts."""
while self._running:
await asyncio.sleep(30)
if self._queue.qsize() > 40:
await send_alert(f"채팅 큐 포화 임박: {self._queue.qsize()}개", "warning")
if not self._obs._connected:
await send_alert("OBS 연결 끊김 — 재연결 필요", "error")
mem_pct = psutil.Process().memory_percent()
if mem_pct > 80:
await send_alert(f"메모리 사용률: {mem_pct:.1f}%", "warning")
async def start_broadcast(self) -> None:
# ... 기존 코드 ...
# Start health monitor as a background task
asyncio.create_task(self._health_monitor())
await self._enter_phase(PHASE_ORDER[0])
self._running = True
await self._run_schedule()
3. Advanced Chat Reactions
Through episode 7, Haru responded to all chat messages the same way. Real streamers give special thanks for superchats and perform custom actions for channel point redemptions. We can build the same behavior into Haru.
Superchat / Bits Donation Detection
On Twitch, bits are the unit of donation. Chat messages containing a donation include patterns like Cheer100.
# src/vtuber/broadcast/twitch_chat.py — 도네이션 이벤트 추가
@dataclass
class DonationEvent:
platform: str
username: str
amount: int # bits 수
message: str
tier: str # "small"(1-99) | "medium"(100-999) | "large"(1000+)
class TwitchChatReader(commands.Bot):
async def event_message(self, msg: twitchio.Message) -> None:
if msg.echo:
return
# Bits(치어) 감지
bits = self._extract_bits(msg.content)
if bits > 0:
tier = "large" if bits >= 1000 else "medium" if bits >= 100 else "small"
donation = DonationEvent(
platform="twitch",
username=msg.author.name,
amount=bits,
message=self._strip_cheer_prefix(msg.content),
tier=tier,
)
await self._donation_queue.put(donation)
return # 일반 채팅 큐 대신 도네이션 큐로
# 일반 채팅
# ... 기존 코드 ...
def _extract_bits(self, text: str) -> int:
import re
return sum(int(m) for m in re.findall(r'[Cc]heer(\d+)', text))
def _strip_cheer_prefix(self, text: str) -> str:
import re
return re.sub(r'[Cc]heer\d+\s*', '', text).strip()
When a donation comes in, Haru uses a different response pattern than for regular chat.
# src/vtuber/broadcast/chat_router.py — 도네이션 처리 추가
DONATION_PROMPTS = {
"small": "작은 도네이션을 받았다. 진심으로 감사 인사를 한다. 1-2문장으로.",
"medium": "적지 않은 비츠 도네이션을 받았다. 이름을 부르며 감동적으로 감사한다. 2-3문장.",
"large": "큰 도네이션이다. 매우 놀라고 감동받아서 한 번은 과한 리액션을 한다. 3-4문장.",
}
async def _process_donation(self, donation: DonationEvent) -> None:
extra_instruction = DONATION_PROMPTS[donation.tier]
user_text = (
f"[도네이션] {donation.username}님이 {donation.amount} bits를 보냈다. "
f"메시지: '{donation.message}'. {extra_instruction}"
)
# 채팅 큐를 무시하고 즉시 최우선 처리
messages = [
LLMMessage(role="system", content=self._prompt_builder.build_system_prompt(self._state)),
LLMMessage(role="user", content=user_text),
]
llm_stream = self._llm.chat_stream(messages)
await self._orchestrator.process_llm_stream(llm_stream, "excited")
Channel Point Reward Integration (Twitch EventSub)
Channel point redemptions cannot be received over IRC. You need to use the Twitch EventSub API via WebSocket.
# src/vtuber/broadcast/twitch_eventsub.py
import asyncio
import websockets
import json
import httpx
import os
from dataclasses import dataclass
@dataclass
class ChannelPointRedemption:
username: str
reward_name: str
user_input: str # 유저가 입력한 텍스트 (있을 경우)
# 리워드 이름 → 하루에게 줄 행동 지시
REWARD_ACTIONS: dict[str, str] = {
"하루야 노래해": "짧은 노래 한 구절을 즉흥으로 만들어 부른다.",
"하루 패배 인정": "지금 하고 있는 것에서 자신이 졌다고 인정하는 대사를 한다.",
"오늘의 명언": "즉흥으로 깊어 보이지만 사실 별것 없는 명언을 하나 만든다.",
"하루 화나게 하기": "시청자의 입력에 약간 짜증을 내는 척한다. 과하지 않게.",
}
class TwitchEventSubClient:
"""Twitch EventSub WebSocket — 채널 포인트 리워드 실시간 수신"""
WEBSOCKET_URL = "wss://eventsub.wss.twitch.tv/ws"
def __init__(self, access_token: str, broadcaster_id: str, callback):
self._token = access_token
self._broadcaster_id = broadcaster_id
self._callback = callback
async def run(self) -> None:
async with websockets.connect(self.WEBSOCKET_URL) as ws:
welcome = json.loads(await ws.recv())
session_id = welcome["payload"]["session"]["id"]
await self._subscribe_redemptions(session_id)
async for raw_msg in ws:
await self._handle_message(json.loads(raw_msg))
async def _subscribe_redemptions(self, session_id: str) -> None:
async with httpx.AsyncClient() as client:
resp = await client.post(
"https://api.twitch.tv/helix/eventsub/subscriptions",
headers={
"Authorization": f"Bearer {self._token}",
"Client-Id": os.getenv("TWITCH_CLIENT_ID", ""),
"Content-Type": "application/json",
},
json={
"type": "channel.channel_points_custom_reward_redemption.add",
"version": "1",
"condition": {"broadcaster_user_id": self._broadcaster_id},
"transport": {
"method": "websocket",
"session_id": session_id,
},
},
)
resp.raise_for_status()
async def _handle_message(self, msg: dict) -> None:
if msg.get("metadata", {}).get("message_type") != "notification":
return
event = msg.get("payload", {}).get("event", {})
redemption = ChannelPointRedemption(
username=event.get("user_name", ""),
reward_name=event.get("reward", {}).get("title", ""),
user_input=event.get("user_input", ""),
)
await self._callback(redemption)
Fast-Path Response Cache — No LLM Required
Calling the LLM for short, repetitive messages like "hi", "nice", or "hello" wastes both money and latency. These can be handled by randomly selecting from a pool of pre-defined responses instead.
# src/vtuber/broadcast/response_cache.py
import random
COMMON_GREETINGS = [
"안녕하세요! 와주셔서 감사해요.",
"안녕! 채팅에서 보니 반가워요.",
"오셨어요~ 방송 즐겁게 봐주세요!",
"하이~ 오늘도 잘 부탁해요.",
]
COMMON_COMPLIMENTS = [
"감사해요! 그런 말 들으면 더 열심히 하게 돼요.",
"칭찬 고마워요! 힘이 나요.",
"과찬이세요~ 열심히 할게요!",
]
TEMPLATES: dict[str, list[str]] = {
"greeting": COMMON_GREETINGS,
"compliment": COMMON_COMPLIMENTS,
}
PATTERN_MAP = [
(["안녕", "하이", "hi", "hello", "ㅎㅇ"], "greeting"),
(["잘하", "잘해", "최고", "굿", "gg", "good"], "compliment"),
]
def classify_quick_reply(text: str) -> str | None:
"""짧은 채팅이 패턴에 해당하면 카테고리 반환, 아니면 None"""
if len(text) > 20:
return None # 긴 텍스트는 LLM으로
text_lower = text.lower().strip()
for patterns, category in PATTERN_MAP:
if any(p in text_lower for p in patterns):
return category
return None
def get_quick_reply(category: str) -> str:
return random.choice(TEMPLATES.get(category, [""]))
# ChatRouter._process() — 빠른 응답 선처리
async def _process(self, msg: ChatMessage) -> None:
category = classify_quick_reply(msg.text)
if category:
reply = get_quick_reply(category)
await self._orchestrator.speak_text(reply, self._state.emotion)
return # LLM 호출 없이 종료
# ... 기존 LLM 처리 ...
4. Vision AI Integration — Haru Reads the Game Screen
The goal is to have Haru automatically say things like "I'm nervous, my health is really low right now" during the gaming segment, by looking at the game screen.
The approach: OBS periodically captures the game screen and sends it to a Vision model.
Game Screen Capture
The OBS WebSocket GetSourceScreenshot request returns a Base64-encoded screenshot of a specific source.
# src/vtuber/broadcast/screen_capture.py
import asyncio
from dataclasses import dataclass
@dataclass
class ScreenCapture:
base64_image: str # PNG Base64
source_name: str
timestamp: float
class ScreenCaptureService:
"""OBS WebSocket으로 게임 화면 주기적 캡처"""
def __init__(self, obs_controller, source_name: str = "게임 캡처"):
self._obs = obs_controller
self._source_name = source_name
self._running = False
self._last_capture: ScreenCapture | None = None
async def start(self, interval_sec: float = 5.0) -> None:
self._running = True
while self._running:
try:
resp = await asyncio.to_thread(
self._obs._require_client().get_source_screenshot,
source_name=self._source_name,
image_format="png",
image_width=640, # Vision API용은 640이면 충분
image_height=360,
image_quality=80,
)
# 응답: "data:image/png;base64,iVBOR..."
b64 = resp.image_data.split(",", 1)[1]
self._last_capture = ScreenCapture(
base64_image=b64,
source_name=self._source_name,
timestamp=asyncio.get_event_loop().time(),
)
except Exception as e:
logger.warning("화면 캡처 실패", extra={"error": str(e)})
await asyncio.sleep(interval_sec)
def get_latest(self) -> ScreenCapture | None:
return self._last_capture
def stop(self) -> None:
self._running = False
Vision-Based Game Commentary Generation
# src/vtuber/broadcast/game_commentator.py
import asyncio
import httpx
import os
from .screen_capture import ScreenCaptureService, ScreenCapture
GAME_COMMENT_SYSTEM = """
너는 게임을 하면서 라이브 방송 중인 AI VTuber 하루다.
게임 화면 이미지를 보고 지금 상황에 맞는 짧은 혼잣말을 한다.
- 1~2문장으로 짧게.
- 과장된 리액션도 좋음.
- "제가" 대신 "나"를 써도 됨. 방송 중이라 캐주얼하게.
- 모른다면 "잘 모르겠는데 일단 해봐요" 같은 말도 자연스럽게.
"""
class GameCommentator:
"""게임 화면 → Vision API → 하루 혼잣말"""
def __init__(
self,
capture_service: ScreenCaptureService,
orchestrator, # TTSOrchestrator
comment_interval_sec: float = 15.0,
):
self._capture = capture_service
self._orch = orchestrator
self._interval = comment_interval_sec
self._running = False
self._api_key = os.getenv("OPENAI_API_KEY", "")
async def run(self) -> None:
"""게임 코너 동안 주기적으로 화면 보고 혼잣말"""
self._running = True
while self._running:
await asyncio.sleep(self._interval)
cap = self._capture.get_latest()
if not cap:
continue
comment = await self._get_comment(cap)
if comment:
await self._orch.speak_text(comment, emotion="excited")
async def _get_comment(self, cap: ScreenCapture) -> str | None:
try:
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": f"Bearer {self._api_key}"},
json={
"model": "gpt-4o-mini", # 비용 절감: 4o-mini로 충분
"messages": [
{"role": "system", "content": GAME_COMMENT_SYSTEM},
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{cap.base64_image}",
"detail": "low", # 비용 절감
},
},
{
"type": "text",
"text": "지금 이 화면을 보고 자연스러운 혼잣말을 해줘.",
},
],
},
],
"max_tokens": 100,
},
)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"].strip()
except Exception as e:
logger.warning("Vision API 호출 실패", extra={"error": str(e)})
return None
Vision API cost breakdown:
gpt-4o-mini+detail: "low"images cost roughly $0.001 per call- One call every 15 seconds × 2-hour stream = 480 calls ≈ $0.48
- Using
gpt-4ois about 20× more expensive.gpt-4o-miniis more than sufficient for game commentary.
5. Final System Structure
Here's the complete structure as of part 8.
vtuber-assistant/
├── src/
│ ├── vtuber/ ← 백엔드 (Python/FastAPI)
│ │ ├── broadcast/
│ │ │ ├── obs_controller.py # OBS WebSocket (재연결 로직 강화)
│ │ │ ├── twitch_chat.py # Twitch IRC + 비츠 감지
│ │ │ ├── twitch_eventsub.py ← 8편 추가 (채널 포인트)
│ │ │ ├── twitch_ratelimit.py ← 8편 추가 (rate limiter)
│ │ │ ├── youtube_chat.py # YouTube API 폴링 (캐시 추가)
│ │ │ ├── chat_queue.py # 우선순위 큐
│ │ │ ├── chat_router.py # 도네이션 처리, 빠른 응답 캐시 추가
│ │ │ ├── response_cache.py ← 8편 추가 (LLM 없는 빠른 응답)
│ │ │ ├── scheduler.py # _health_monitor 추가
│ │ │ ├── screen_capture.py ← 8편 추가 (OBS 스크린샷)
│ │ │ └── game_commentator.py ← 8편 추가 (Vision AI)
│ │ ├── api/
│ │ │ ├── broadcast.py
│ │ │ ├── monitor.py
│ │ │ ├── health.py
│ │ │ └── voice.py
│ │ ├── live2d/ ← 6편
│ │ ├── tts/ ← 5편 (메모리 누수 수정)
│ │ ├── stt/ ← 4편
│ │ ├── memory/ ← 3편
│ │ ├── character/ ← 2편
│ │ ├── llm/ ← 2편
│ │ ├── metrics.py ← 8편 추가 (Prometheus 메트릭)
│ │ ├── alerting.py ← 8편 추가 (슬랙 알림)
│ │ ├── logger.py ← 7편
│ │ └── main.py
│ └── frontend/ ← TypeScript (6편)
├── assets/haru/
├── monitoring/
│ ├── prometheus.yml ← 8편 추가
│ └── grafana/
│ └── dashboards/
│ └── haru-dashboard.json
├── docker-compose.yml # prometheus, grafana 서비스 추가
├── Dockerfile
└── pyproject.toml
Docker Compose — Final Version
# docker-compose.yml
services:
haru-agent:
build: .
restart: unless-stopped
ports:
- "8000:8000"
env_file: .env
volumes:
- ./assets:/app/assets
depends_on:
- redis
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 60s
redis:
image: redis:7-alpine
restart: unless-stopped
volumes:
- redis_data:/data
prometheus:
image: prom/prometheus:latest
restart: unless-stopped
ports:
- "9090:9090"
volumes:
- ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
- prometheus_data:/prometheus
command:
- "--config.file=/etc/prometheus/prometheus.yml"
- "--storage.tsdb.retention.time=7d"
grafana:
image: grafana/grafana:latest
restart: unless-stopped
ports:
- "3001:3000" # 3000이 프론트엔드와 겹칠 수 있으니 3001
volumes:
- grafana_data:/var/lib/grafana
- ./monitoring/grafana/dashboards:/etc/grafana/provisioning/dashboards
environment:
- GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD:-admin}
volumes:
redis_data:
prometheus_data:
grafana_data:
# monitoring/prometheus.yml
global:
scrape_interval: 15s
scrape_configs:
- job_name: haru-agent
static_configs:
- targets: ["haru-agent:8000"]
metrics_path: /metrics
6. Cost Breakdown — What Does It Actually Cost?
Rough numbers for a 2-hour stream.
| Item | Cost / session | Notes |
|---|---|---|
| LLM (Claude Sonnet 3.7) | $0.5 – $1.5 | Scales with number of chat responses |
| TTS (XTTS local) | ~$0 | GPU electricity not included |
| STT (Whisper local) | ~$0 | GPU electricity not included |
| YouTube Data API | $0 | Fits within free quota after caching |
| Twitch API | $0 | Free |
| Vision API (gpt-4o-mini) | ~$0.5 | 2-hour gaming segment |
| Total | $1 – $2 / session | Excluding GPU electricity |
LLM cost dominates everything. Ways to reduce it:
- Don't respond to every chat message — apply a cooldown (part 7:
COOLDOWN_SECONDS = 30) - Handle short, repetitive chat with the response cache instead of the LLM (added in part 8)
- Route ordinary chat to Haiku / 4o-mini and important conversations to Sonnet / 4o
Example LLM routing logic:
# ChatRouter._process() — 중요도에 따라 LLM 모델 선택
async def _process(self, msg: ChatMessage) -> None:
# 빠른 응답 선처리 (LLM 없음)
category = classify_quick_reply(msg.text)
if category:
await self._orchestrator.speak_text(get_quick_reply(category), self._state.emotion)
return
# 중요 채팅(구독자/VIP/길이 50자 이상)은 고성능 모델
use_powerful = (
msg.is_sub or msg.is_vip or msg.is_mod or len(msg.text) > 50
)
llm = self._llm_powerful if use_powerful else self._llm_fast
# ... 이하 동일
7. Retrospective
When this series started, the goal was "an AI VTuber assistant that talks, remembers, and lives on screen."
Here's everything built across eight parts:
- Persona: personality defined through LLM prompts, emotion tag parsing (part 2)
- Memory: short-term conversation history + long-term user memory (part 3)
- STT: real-time speech recognition with Whisper (part 4)
- TTS: XTTS-v2 with emotion-driven speed/pitch control, sentence-level streaming (part 5)
- Live2D: lip sync, eye blinking, breathing, mouse gaze tracking (part 6)
- Broadcast automation: OBS WebSocket, Twitch/YouTube chat, BroadcastScheduler (part 7)
- Production hardening: failure handling, metrics, alerting, Vision AI, cost optimization (part 8)
Three things that stood out while writing the series:
Individual components are easy; integration is hard. STT, TTS, and Live2D each had documentation and came together without much trouble. The real problems started when all of them had to run simultaneously and stay in sync with each other.
"Works in dev" and "works during a live stream" are completely different sports. Go live and edge cases pour out that you never anticipated — YouTube API quota, OBS restarts, Twitch bot bans. You can't prevent all of them up front. The practical approach is to build the infrastructure for detecting and responding to failures quickly when they do happen.
AI VTubers are still early. Right now, building a system like Haru requires a significant engineering investment. But the tooling is moving fast. Within one to two years, services will likely exist that replace most of what this series covers with a handful of API calls.
Until then, building it yourself is the fastest path to actually understanding it.
Closing
Everything built from part 1 through part 8 came together and became Haru.
Pay more attention to why each component was designed the way it was than to any individual line of code. Specific libraries may disappear or break their APIs six months from now, but the reasoning behind the design stays relevant.
End of series.