어댑터 패턴

인터페이스 하나로 여러 모델 다루기

어댑터 패턴은 제공자마다 다른 구조를 하나의 인터페이스 뒤에 숨겨. 애플리케이션은 adapter.generate_stream(messages)만 호출하고, 어느 API를 부를지, 요청을 어떻게 만들지, 응답을 어떻게 정규화할지는 어댑터가 알아서 해.

왜 필요한가

앱 코드를 바꾸지 않고 모델 교체 — 한 줄만 바꿔 Gemini에서 OpenAI, 다시 로컬 Ollama로 전환할 수 있어.
폴백 체인 구성 — Gemini가 실패하면 OpenAI를, 그다음 로컬 모델을 시도해. 어댑터 목록이 곧 체인이야.
제공자 A/B 테스트 — 같은 인터페이스 뒤에서 트래픽의 10%를 새 모델로 보낼 수 있어.
프롬프트 형식 변환 — 시스템 프롬프트 위치, 도구 호출 구조, 역할 이름처럼 제공자마다 다른 점을 경계에서 흡수해.

추상화의 경계는 어디에 그을까

좁게 그어. 스트리밍 생성 호출과 필요하다면 도구 호출 반복만 추상화해. File API, 캐싱, 임베딩처럼 모든 제공자 기능을 하나의 인터페이스로 합치려 하지 마. 추상화가 너무 넓으면 새 제공자를 넣을 때마다 억지로 끼워 맞춰야 해. 경계는 generate_stream에 두고 제공자 고유 기능은 고유한 모습으로 살게 해.

Code

ABC와 청크 형식 — 계약·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

앱 코드에서 어댑터 사용하기·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

OpenAI SDK로 OpenAIAdapter를 만들어 GeminiAdapter와 함께 써. OpenAI에 접근할 수 없다면 사용자 메시지를 그대로 되돌리는 EchoAdapter를 만들어도 돼. 둘을 같은 ModelAdapter ABC 뒤에 연결하고, 같은 호출 코드가 두 제공자 모두에서 generate_stream으로 작동하는지 확인하는 테스트를 작성해.