C.W.K.
Stream
Lesson 01 of 05 · published

공통 AI Adapter 인터페이스

~20 min · adapters, abstraction

Level 0Downloader
0 XP0/41 lessons0/11 achievements
0/120 XP to next level120 XP to go0% complete

왜 adapter?

Local이랑 cloud 모델을 app 다시 짜지 않고 swap하고 싶어. 패턴: 모든 provider가 구현하는 narrow interface 정의, 모든 provider가 emit하는 universal chunk type 하나. Route, store, frontend는 interface랑 얘기; provider 전용 코드는 그 뒤에 있어.

최소 인터페이스

  • stream(messages, tools, **opts) — 의미 있는 유일한 method. Universal chunk sequence yield.
  • health_check() — Fallback 라우팅용 빠른 yes/no.
  • list_models() — UI랑 discovery용.

Universal chunk

StreamChunk 한 모양이 yield 필요한 거 다 커버: 텍스트 delta, thinking delta, tool call, timing 가진 final done flag. 각 adapter가 자기 provider의 wire 모양을 universal chunk로 번역.

너무 abstract하지 마

Adapter는 의도적으로 narrow. Route는 NDJSON vs SSE 모름, stream()만 알아. Adapter는 FastAPI route 모름, NDJSON vs SSE만 알아. 한 가지가 경계에 있고 그게 chunk yield 프로토콜이야.

Code

Narrow 인터페이스·python
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import AsyncIterator, Any

@dataclass
class StreamChunk:
    """Universal chunk — 모든 adapter가 이 모양 yield."""
    content: str = ""
    thinking: str = ""
    tool_calls: list = field(default_factory=list)
    done: bool = False
    tokens_per_sec: float = 0.0
    model: str = ""
    raw: Any = None  # provider-전용 payload, debug-only

class AIAdapter(ABC):
    """모든 AI vessel이 구현하는 narrow 계약."""

    @abstractmethod
    async def stream(
        self,
        messages: list[dict],
        tools: list[dict] | None = None,
        **opts: Any,
    ) -> AsyncIterator[StreamChunk]:
        """응답을 universal chunk로 stream."""
        ...

    @abstractmethod
    async def health_check(self) -> bool:
        """빠른 yes/no — provider 지금 살아있어?"""
        ...

    @abstractmethod
    async def list_models(self) -> list[str]:
        """사용 가능한 모델 이름 — UI랑 discovery용."""
        ...

External links

Exercise

자체 프로젝트에 StreamChunkAIAdapter 정의. 위의 셋 이상 method 추가하지 마. 추가하고 싶을 만한 다섯 (token counter, 대화 메모리, retry 로직, rate limiter, 모델 selection) list 만들고 각각 왜 adapter *밖*에 있어야 하는지 한 문장씩.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.