자체 Adapter 만들기 — narrow boundary

Adapter ABC 디자인의 유혹 — 모든 provider feature 노출 (chat(), stream(), embed(), transcribe(), generate_image()). 저항해. Narrow ABC 1 method (stream(messages, tools) -> AsyncIterator[Event]) 가 모든 provider 를 평등하게 + provider-specific 부분이 implementation 으로 내려감.

왜 narrow?

ABC 가 leaky 하면 provider 차이가 routes/store/frontend 까지 transmit. 1 method 면 wire 차이만 implementation 에서 처리, downstream 코드는 provider-agnostic.

Event protocol

Adapter 가 yield 하는 Event 는 typed — TextDelta, ToolCallRequested, ToolCallArguments, Done, Error. Consumer (orchestrator, frontend) 가 type 으로 dispatch. 새 provider 추가 = 새 implementation, 기존 consumer 코드 0 변경.

cwkPippa 의 Adapter ABC

backend/adapters/base.py 가 정확히 이 shape — 1 ABC, 1 streaming method, Event protocol. Codex/Gemini/Ollama vessel 은 backend/variants/ 에서 specialize. Routes, store, frontend 는 모두 Claude shape 가정. Rule 2 — cost is absorbed downstream, not pushed upstream.

Code

Adapter ABC with single stream() method·python

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import AsyncIterator, Literal, Optional

@dataclass
class TextDelta:
    type: Literal["text_delta"] = "text_delta"
    content: str = ""

@dataclass
class ToolCallDelta:
    type: Literal["tool_call_delta"] = "tool_call_delta"
    call_id: str = ""
    name: str = ""
    arguments_delta: str = ""

@dataclass
class ToolCallComplete:
    type: Literal["tool_call_complete"] = "tool_call_complete"
    call_id: str = ""
    name: str = ""
    arguments: str = ""  # complete JSON string

@dataclass
class StreamDone:
    type: Literal["done"] = "done"
    finish_reason: str = "stop"
    usage: Optional[dict] = None

Chunk = TextDelta | ToolCallDelta | ToolCallComplete | StreamDone

OpenAIAdapter implementation·python

class LLMAdapter(ABC):
    @abstractmethod
    async def stream(self, messages, tools=None, model=None) -> AsyncIterator[Chunk]:
        ...

class OpenAIAdapter(LLMAdapter):
    """Raw HTTP adapter — uses httpx, no SDK dependency."""
    BASE_URL = "https://api.openai.com/v1/chat/completions"

    def __init__(self, api_key=None, model="gpt-4.1"):
        self.api_key = api_key or os.environ["OPENAI_API_KEY"]
        self.default_model = model
        self._client = httpx.AsyncClient(
            timeout=httpx.Timeout(connect=10, read=120, write=10, pool=5),
            headers={"Authorization": f"Bearer {self.api_key}",
                     "Content-Type": "application/json"},
        )

    async def stream(self, messages, tools=None, model=None) -> AsyncIterator[Chunk]:
        body = {"model": model or self.default_model, "messages": messages, "stream": True}
        if tools: body["tools"] = tools
        # ... parse SSE, yield TextDelta / ToolCallDelta / StreamDone

class CodexAdapter(LLMAdapter):
    """Uses ~/.codex/auth.json OAuth tokens."""
    def __init__(self, model="gpt-4o"):
        self.default_model = model
    async def stream(self, messages, tools=None, model=None) -> AsyncIterator[Chunk]:
        token = get_bearer_token()  # from auth.json
        adapter = OpenAIAdapter(api_key=token, model=model or self.default_model)
        async for chunk in adapter.stream(messages, tools):
            yield chunk

자체 Adapter 만들기 — narrow boundary

왜 narrow?

Event protocol

cwkPippa 의 Adapter ABC

Code

External links

Exercise

Progress

댓글 0