본문 바로가기
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을 다시 안 짜고 갈아끼우고 싶은 거야. 패턴은 이래. 모든 provider가 구현하는 좁은 interface를 하나 정하고, 모든 provider가 뱉는 universal chunk type을 하나 정해. Route도 store도 frontend도 그 interface만 상대하고, provider 전용 코드는 전부 그 뒤에 숨어.

최소 인터페이스

  • stream(messages, tools, **opts) — 진짜 의미 있는 유일한 method야. Universal chunk를 줄줄이 yield해.
  • health_check() — Fallback 라우팅용으로 빠르게 예/아니오만 답해.
  • list_models() — UI랑 discovery에 써.

Universal chunk

StreamChunk 한 모양으로 yield할 게 다 덮여. 텍스트 delta, thinking delta, tool call, 그리고 timing을 실은 final done flag까지. 각 adapter는 자기 provider가 쓰는 wire 모양을 이 universal chunk로 번역하는 일만 해.

너무 abstract하지 마

Adapter는 일부러 좁게 뒀어. Route는 NDJSON인지 SSE인지 몰라. stream()만 알아. Adapter는 FastAPI route를 몰라. NDJSON이냐 SSE냐만 알고. 경계에 놓인 건 딱 하나, chunk를 yield하는 프로토콜뿐이야.

그리고 이게 다형성이 실제로 일하는 모습이야. 인터페이스는 하나인데 그 뒤에 선 vessel은 여럿이고, 부르는 쪽은 어느 놈이 답했는지 알 필요가 없어. OOP 교과서가 도형 클래스로 설명하던 그 얘기를, 여기선 Claude랑 Ollama가 같은 stream() 뒤에 나란히 서서 증명해. 상속 도표를 외워서 얻는 게 아니라, cloud가 죽은 밤에 답이 계속 나오는 걸로 체감하는 거지.

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

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

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