Adapter 패턴

한 interface, 많은 모델

Adapter 패턴이 provider-specific 모양을 single interface 뒤에 숨김. Application 코드는 adapter.generate_stream(messages) 호출; adapter 가 어느 API hit, request 어떻게 포맷, response 어떻게 normalize 알고 있음.

왜 신경 써

App 코드 안 바꾸고 모델 swap — 한 줄 바꿔서 Gemini → OpenAI → 로컬 Ollama flip.
Fallback chain 빌드 — Gemini 가 에러 나면 OpenAI, 그 다음 로컬 시도. Chain 이 adapter 의 list.
Provider A/B 테스트 — 같은 interface 뒤에 새 모델로 트래픽 10% 라우팅.
Prompt 포맷 번역 — provider 별 quirk (system prompt 위치, tool-call 모양, role 이름) 위에 abstract.

Abstraction line 어디 그리나

좁게. Streaming generate 호출 (그리고 maybe tool loop) 만 abstract. 모든 provider feature — file API, caching, embedding — 을 unified interface 로 abstract 시도 X. Over-broad abstraction 비용은 새 provider 마다 shoehorn 필요. Seam 을 generate_stream 에 두고 provider-specific feature 가 provider-specific 으로 살게.

Code

ABC + chunk type — 계약·python

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

@dataclass
class StreamChunk:
    text: str = ''
    function_call: Optional[dict] = None
    finish_reason: Optional[str] = None
    usage: Optional[dict] = None

class ModelAdapter(ABC):
    @abstractmethod
    async def generate_stream(
        self,
        messages: list[dict],
        tools: Optional[list[dict]] = None,
        **kw,
    ) -> AsyncIterator[StreamChunk]:
        ...

    @abstractmethod
    def name(self) -> str:
        ...

    @abstractmethod
    def supports_tools(self) -> bool:
        ...

    @abstractmethod
    def supports_vision(self) -> bool:
        ...

Gemini 구현·python

from google import genai
from google.genai import types

class GeminiAdapter(ModelAdapter):
    def __init__(self, api_key: str, model: str = 'gemini-2.5-flash'):
        self.client = genai.Client(api_key=api_key)
        self.model = model

    def name(self):           return f'gemini:{self.model}'
    def supports_tools(self): return True
    def supports_vision(self): return True

    async def generate_stream(self, messages, tools=None, **kw):
        contents = self._convert_messages(messages)
        config = types.GenerateContentConfig()
        if tools:
            config = types.GenerateContentConfig(
                tools=[types.Tool(function_declarations=tools)],
            )

        async for chunk in await self.client.aio.models.generate_content_stream(
            model=self.model, contents=contents, config=config,
        ):
            sc = StreamChunk(text=chunk.text or '')
            if chunk.usage_metadata:
                sc.usage = {
                    'prompt': chunk.usage_metadata.prompt_token_count,
                    'completion': chunk.usage_metadata.candidates_token_count,
                }
            cand = chunk.candidates[0] if chunk.candidates else None
            if cand:
                sc.finish_reason = cand.finish_reason.name if cand.finish_reason else None
            yield sc

    def _convert_messages(self, messages):
        # OpenAI-style messages -> Gemini contents
        out = []
        for m in messages:
            role = 'model' if m['role'] == 'assistant' else m['role']
            out.append(types.Content(
                role=role,
                parts=[types.Part(text=m['content'])],
            ))
        return out

App 코드에서 adapter 사용·python

import os
adapter: ModelAdapter = GeminiAdapter(api_key=os.environ['GEMINI_API_KEY'])

async for chunk in adapter.generate_stream(
    messages=[
        {'role': 'system', 'content': 'You are helpful.'},  # adapter handles role conversion
        {'role': 'user',   'content': 'Hello!'},
    ],
):
    if chunk.text:
        print(chunk.text, end='', flush=True)
    if chunk.usage:
        print(f'\n[usage: {chunk.usage}]')

Exercise

OpenAIAdapter 를 GeminiAdapter 와 함께 OpenAI SDK 사용해서 빌드 (또는, OpenAI 접근 없으면 user 메시지 그냥 echo 하는 EchoAdapter 빌드). 둘 다 같은 ModelAdapter ABC 뒤에 wire. 둘 다 generate_stream 호출하고 같은 caller 코드가 두 provider 가로질러 동작 확인하는 테스트 작성.