6장: 실시간 추론 파이프라인 설계

[전체 프롬프트 수신 완료] ──> [프리필 시작] ──> [디코딩 시작] ──> [토큰 출력]
                                                                    
시간: ========= 대기 =========|=== 프리필 ===|=== 디코딩 ====>

[프롬프트 청크 1 도착] ──> [부분 프리필 시작]
[프롬프트 청크 2 도착] ──> [프리필 이어서]
[프롬프트 청크 3 도착] ──> [프리필 완료] ──> [디코딩 시작] ──> [토큰 출력]
 
시간: = 수신+프리필 병렬 =|= 디코딩 ====>

from vllm import AsyncLLMEngine, SamplingParams
 
engine = AsyncLLMEngine.from_engine_args(engine_args)
 
async def stream_inference(prompt_chunks: AsyncIterator[str]):
    """프롬프트를 점진적으로 수신하면서 프리필 진행"""
    sampling_params = SamplingParams(
        temperature=0.7,
        max_tokens=1024,
        stream=True,
    )
 
    request_id = generate_request_id()
 
    # 스트리밍 입력: 청크 단위로 추가
    async for chunk in prompt_chunks:
        await engine.add_request_chunk(
            request_id=request_id,
            prompt_chunk=chunk,
        )
 
    # 입력 완료 신호
    await engine.finalize_request(request_id, sampling_params)
 
    # 출력 토큰 스트리밍
    async for output in engine.generate_stream(request_id):
        yield output.outputs[0].text

[정적 배칭]
배치 1: [요청A(100토큰), 요청B(10토큰), 요청C(50토큰)]
         ─── 요청A 완료까지 대기 ───> 배치 2 시작
 
[연속 배칭]
시점 1: [요청A, 요청B, 요청C] 처리 중
시점 2: 요청B 완료 → [요청A, 요청D, 요청C] (요청D 즉시 추가)
시점 3: 요청C 완료 → [요청A, 요청D, 요청E] (요청E 즉시 추가)

import numpy as np
from redis import Redis
from sentence_transformers import SentenceTransformer
 
class SemanticCache:
    def __init__(
        self,
        similarity_threshold: float = 0.95,
        ttl_seconds: int = 3600,
    ):
        self.model = SentenceTransformer("all-MiniLM-L6-v2")
        self.redis = Redis()
        self.threshold = similarity_threshold
        self.ttl = ttl_seconds
 
    def get(self, query: str) -> str | None:
        """의미적으로 유사한 캐시 엔트리 검색"""
        query_embedding = self.model.encode(query)
 
        # Redis에서 벡터 검색 (RediSearch 모듈 활용)
        results = self.redis.ft("cache_idx").search(
            f"@embedding:[VECTOR_RANGE $radius $vec]",
            query_params={
                "radius": 1 - self.threshold,
                "vec": query_embedding.tobytes(),
            },
        )
 
        if results.total > 0:
            return results.docs[0].response
 
        return None
 
    def put(self, query: str, response: str):
        """쿼리-응답 쌍을 캐시에 저장"""
        embedding = self.model.encode(query)
        cache_key = f"cache:{hash(query)}"
 
        self.redis.hset(cache_key, mapping={
            "query": query,
            "response": response,
            "embedding": embedding.tobytes(),
        })
        self.redis.expire(cache_key, self.ttl)

interface InferenceRequest {
  model: string;
  prompt: string;
  maxTokens: number;
  priority: "low" | "normal" | "high";
}
 
interface BackendServer {
  id: string;
  models: string[];
  currentLoad: number; // 0-1
  gpuMemoryFree: number; // bytes
  avgLatency: number; // ms
  queueDepth: number;
}
 
type RoutingStrategy =
  | "least-connections"
  | "least-latency"
  | "model-affinity"
  | "cost-aware";
 
function routeRequest(
  request: InferenceRequest,
  backends: BackendServer[],
  strategy: RoutingStrategy
): BackendServer {
  // 해당 모델을 지원하는 서버만 필터링
  const eligible = backends.filter(
    (b) => b.models.includes(request.model) && b.currentLoad < 0.95
  );
 
  if (eligible.length === 0) {
    throw new Error("사용 가능한 추론 서버 없음");
  }
 
  switch (strategy) {
    case "least-connections":
      return eligible.reduce((min, b) =>
        b.queueDepth < min.queueDepth ? b : min
      );
 
    case "least-latency":
      return eligible.reduce((min, b) =>
        b.avgLatency < min.avgLatency ? b : min
      );
 
    case "model-affinity":
      // KV 캐시가 이미 로드된 서버 우선
      return (
        eligible.find((b) => b.currentLoad > 0) ??
        eligible[0]
      );
 
    case "cost-aware":
      // 이미 실행 중인 서버 우선 (새 서버 기동 비용 회피)
      return eligible.reduce((best, b) =>
        b.currentLoad > 0 && b.currentLoad < best.currentLoad
          ? b
          : best
      );
  }
}

import asyncio
from dataclasses import dataclass
 
 
@dataclass
class AudioFrame:
    data: bytes
    sample_rate: int
    channels: int
    timestamp_ms: int
 
 
class RealtimeAudioPipeline:
    """오디오 실시간 처리 파이프라인"""
 
    def __init__(self):
        self.vad = VoiceActivityDetector()
        self.stt = StreamingSTT()
        self.llm = StreamingLLM()
        self.tts = StreamingTTS()
 
    async def process(
        self,
        audio_stream: AsyncIterator[AudioFrame],
    ) -> AsyncIterator[AudioFrame]:
        """입력 오디오 스트림을 처리하여 응답 오디오 스트림 생성"""
 
        # 1단계: VAD로 음성 구간 감지
        speech_segments = self.vad.process_stream(audio_stream)
 
        # 2단계: 음성을 텍스트로 변환 (스트리밍)
        text_stream = self.stt.transcribe_stream(speech_segments)
 
        # 3단계: LLM 추론 (스트리밍 입력 + 스트리밍 출력)
        response_text = self.llm.generate_stream(text_stream)
 
        # 4단계: 텍스트를 음성으로 변환 (스트리밍)
        response_audio = self.tts.synthesize_stream(response_text)
 
        async for frame in response_audio:
            yield frame

from pipecat.pipeline import Pipeline
from pipecat.services.openai import OpenAILLMService
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.transports.websocket import WebSocketTransport
 
# 파이프라인 구성: 각 노드가 스트리밍으로 연결
pipeline = Pipeline([
    WebSocketTransport(port=8765),     # 오디오 입력
    DeepgramSTTService(),               # 실시간 음성 인식
    OpenAILLMService(model="gpt-4o"),   # LLM 추론
    ElevenLabsTTSService(),             # 음성 합성
    WebSocketTransport(port=8765),     # 오디오 출력
])
 
# 파이프라인 실행 — 각 노드 간 데이터가 스트리밍으로 흐름
await pipeline.run()

class VideoAnalysisPipeline:
    """비디오 프레임 실시간 분석 파이프라인"""
 
    def __init__(self, target_fps: int = 5):
        self.target_fps = target_fps
        self.frame_interval = 1.0 / target_fps
        self.vision_model = load_vision_model()
 
    async def analyze_stream(
        self,
        video_frames: AsyncIterator[VideoFrame],
    ) -> AsyncIterator[AnalysisResult]:
        """비디오 프레임을 실시간으로 분석"""
        last_analysis_time = 0
 
        async for frame in video_frames:
            current_time = frame.timestamp_ms / 1000
 
            # 프레임 샘플링: 모든 프레임을 분석하면 GPU 과부하
            if current_time - last_analysis_time < self.frame_interval:
                continue
 
            last_analysis_time = current_time
 
            # 비동기 분석 — 이전 프레임 분석 완료를 기다리지 않음
            result = await self.vision_model.analyze(frame)
            yield AnalysisResult(
                timestamp=frame.timestamp_ms,
                objects=result.detected_objects,
                description=result.scene_description,
            )