자체 Adapter 만들기 — narrow boundary · GPT Wire Quest

Adapter ABC에 chat(), stream(), embed(), transcribe(), generate_image()처럼 provider 기능을 모두 노출하고 싶어질 수 있어. 하지만 stream(messages, tools) -> AsyncIterator[Event] 하나만 둔 좁은 ABC가 provider 차이를 각 구현 안에 가둬.

경계가 좁아야 하는 이유

ABC가 provider별 세부 기능을 드러내면 그 차이가 route, store, frontend까지 퍼져. method 하나로 제한하면 wire 차이는 implementation이 처리하고 나머지 코드는 provider를 몰라도 돼.

type이 붙은 event protocol

Adapter는 TextDelta, ToolCallRequested, ToolCallArguments, Done, Error 같은 event를 yield해. orchestrator와 frontend는 type으로 분기해 처리해. 새 provider를 추가할 때 새 구현만 만들고 기존 consumer 코드는 바꾸지 않는 게 목표야.

cwkPippa의 Adapter ABC

backend/adapters/base.py는 ABC 하나, streaming method 하나, Event protocol 하나로 이 경계를 만든다. Codex, Gemini, Ollama vessel은 backend/variants/에서 specialize하고 route, store, frontend는 공통 Claude shape를 사용해. provider 차이를 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

경계가 좁아야 하는 이유

type이 붙은 event protocol

cwkPippa의 Adapter ABC

Code

External links

Exercise

Progress

댓글 0