7. 에이전트 확장성 확보: 멀티 프로바이더(Multi-Provider) 지원 구조 설계

저번 6편 글에서는 단일 파일이었던 에이전트 루프 코드를 모듈화하는 리팩터링을 진행했다.

 

리팩터링을 진행한 가장 중요한 이유 중 하나는 바로 멀티 프로바이더(Multi-Provider) 지원 구조를 갖추기 위해서이다.

에이전트를 서비스나 환경에 맞게 운용하려면 Claude뿐만 아니라 OpenAI의 GPT 모델, 혹은 가성비가 훌륭한 DeepSeek 등 다양한 LLM을 코드 수정 없이 자유롭게 스왑할 수 있어야 한다.

 

프로바이더마다 도구 스펙 포맷, 응답 구조, 메시지 히스토리(특히 Assistant 턴과 Tool Result)를 기록하는 방식이 전부 제각각이다.

이 다른 규격들을 하나의 뼈대로 일관되게 다루기 위해 설계한 실제 멀티 프로바이더 연동 구조를 살펴보자.

 

1. 프로바이더 추상 클래스 설계 (base.py)

가장 먼저 각 프로바이더가 공통으로 따라야 할 인터페이스를 정의하고, 응답 데이터를 정규화할 LLMResponse 구조를 잡는다.

 

# agent/llm/base.py
from abc import ABC, abstractmethod
from dataclasses import dataclass, field

@dataclass
class ToolCall:
    id: str          # 도구 호출 식별자
    name: str
    input: dict

@dataclass
class LLMResponse:
    text: str                    # 정규화된 텍스트 응답
    tool_calls: list             # list[ToolCall]
    stop_reason: str             # 정규화: "tool_use" | "end_turn"
    input_tokens: int
    output_tokens: int
    raw: dict = field(default_factory=dict)   # 원본 응답 저장용

class LLMProvider(ABC):
    @abstractmethod
    def call(self, messages: list, tools: list) -> LLMResponse:
        """messages와 도구 목록을 전달받아 공통 응답 포맷인 LLMResponse로 정규화해 반환한다."""
        raise NotImplementedError

    @abstractmethod
    def append_assistant(self, messages: list, response: LLMResponse) -> None:
        """어시스턴트 턴을 각 프로바이더 스키마에 맞춰 messages에 추가한다."""
        raise NotImplementedError

    @abstractmethod
    def append_tool_results(self, messages: list, results: list) -> None:
        """도구 실행 결과를 각 프로바이더 스키마에 맞춰 messages에 추가한다."""
        raise NotImplementedError

 

2. Anthropic Claude 프로바이더 구현 (claude.py)

Anthropic API의 독자적인 도구 호출 및 메시지 히스토리 규칙을 반영한 구현체이다.

 

# agent/llm/claude.py
import requests
from agent.llm.base import LLMProvider, LLMResponse, ToolCall

class ClaudeProvider(LLMProvider):
    def __init__(self, api_key: str, model: str = "claude-haiku-4-5", max_tokens: int = 500):
        self.api_key = api_key
        self.model = model
        self.max_tokens = max_tokens

    def call(self, messages: list, tools: list) -> LLMResponse:
        headers = {
            "x-api-key": self.api_key,
            "anthropic-version": "2023-06-01",
            "content-type": "application/json",
        }
        body = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "tools": tools,
            "messages": messages,
        }
        response = requests.post("https://api.anthropic.com/v1/messages", headers=headers, json=body, timeout=60)
        if response.status_code != 200:
            raise RuntimeError(f"Claude API {response.status_code}: {response.text}")
        return self._to_response(response.json())

    def _to_response(self, raw: dict) -> LLMResponse:
        text = "".join(b.get("text", "") for b in raw.get("content", []) if b.get("type") == "text")
        tool_calls = [
            ToolCall(id=b["id"], name=b["name"], input=b.get("input", {}))
            for b in raw.get("content", []) if b.get("type") == "tool_use"
        ]
        stop = "tool_use" if raw.get("stop_reason") == "tool_use" else "end_turn"
        usage = raw.get("usage", {})
        return LLMResponse(
            text=text, tool_calls=tool_calls, stop_reason=stop,
            input_tokens=usage.get("input_tokens", 0), output_tokens=usage.get("output_tokens", 0), raw=raw
        )

    def append_assistant(self, messages: list, response: LLMResponse) -> None:
        messages.append({"role": "assistant", "content": response.raw["content"]})

    def append_tool_results(self, messages: list, results: list) -> None:
        content = []
        for tool_call_id, output, is_error in results:
            block = {"type": "tool_result", "tool_use_id": tool_call_id, "content": output}
            if is_error:
                block["is_error"] = True
            content.append(block)
        messages.append({"role": "user", "content": content})

 

3. OpenAI 호환 프로바이더 구현 (openai.py)

OpenAI의 /chat/completions 엔드포인트에 맞춘 프로바이더 구현이다.

OpenAI와 DeepSeek은 API 명세가 호환되므로, 공통 분모가 되는 로직을 여기에 한 번에 작성했다.

 

# agent/llm/openai.py
import json
import requests
from agent.llm.base import LLMProvider, LLMResponse, ToolCall

class OpenAIProvider(LLMProvider):
    def __init__(self, api_key: str, model: str = "gpt-4o-mini", base_url: str = "https://api.openai.com/v1"):
        self.api_key = api_key
        self.model = model
        self.base_url = base_url.rstrip("/")

    def call(self, messages: list, tools: list) -> LLMResponse:
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "content-type": "application/json",
        }
        body = {"model": self.model, "messages": messages}
        if tools:
            body["tools"] = [self._tool_spec(t) for t in tools]

        response = requests.post(f"{self.base_url}/chat/completions", headers=headers, json=body, timeout=60)
        if response.status_code != 200:
            raise RuntimeError(f"OpenAI-호환 API {response.status_code}: {response.text}")
        return self._to_response(response.json())

    @staticmethod
    def _tool_spec(neutral: dict) -> dict:
        return {
            "type": "function",
            "function": {
                "name": neutral["name"],
                "description": neutral["description"],
                "parameters": neutral["input_schema"],
            },
        }

    def _to_response(self, raw: dict) -> LLMResponse:
        choice = raw.get("choices", [{}])[0]
        msg = choice.get("message", {})
        text = msg.get("content") or ""
        tool_calls = []
        for tc in msg.get("tool_calls", []) or []:
            fn = tc.get("function", {})
            args = fn.get("arguments") or "{}"
            try:
                parsed = json.loads(args)
            except json.JSONDecodeError:
                parsed = {}
            tool_calls.append(ToolCall(id=tc["id"], name=fn.get("name", ""), input=parsed))

        stop = "tool_use" if choice.get("finish_reason") == "tool_calls" else "end_turn"
        usage = raw.get("usage", {})
        return LLMResponse(
            text=text, tool_calls=tool_calls, stop_reason=stop,
            input_tokens=usage.get("prompt_tokens", 0), output_tokens=usage.get("completion_tokens", 0), raw=raw
        )

    def append_assistant(self, messages: list, response: LLMResponse) -> None:
        messages.append(response.raw["choices"][0]["message"])

    def append_tool_results(self, messages: list, results: list) -> None:
        for tool_call_id, output, _is_error in results:
            messages.append({"role": "tool", "tool_call_id": tool_call_id, "content": output})

 

4. DeepSeek 프로바이더 구현 (deepseek.py)

DeepSeek은 OpenAI 호환 API 규격을 사용하므로, OpenAIProvider를 상속받아 단 몇 줄의 코드만으로 완벽히 재사용하여 완성한다.

 

# agent/llm/deepseek.py
from agent.llm.openai import OpenAIProvider

class DeepSeekProvider(OpenAIProvider):
    def __init__(self, api_key: str, model: str = "deepseek-chat"):
        super().__init__(api_key, model=model, base_url="https://api.deepseek.com/v1")

 

5. 이름 기반 팩토리 레지스트리 (agent/llm/__init__.py)

마지막으로 외부(UI 등)에서 이름 문자열만 가지고 동적으로 적절한 프로바이더를 생성할 수 있도록 매핑한다.

 

# agent/llm/__init__.py
import os
from agent.llm.base import LLMProvider
from agent.llm.claude import ClaudeProvider
from agent.llm.openai import OpenAIProvider
from agent.llm.deepseek import DeepSeekProvider

_PROVIDERS = {
    "claude":   ("ANTHROPIC_API_KEY", lambda key: ClaudeProvider(key)),
    "openai":   ("OPENAI_API_KEY",    lambda key: OpenAIProvider(key)),
    "deepseek": ("DEEPSEEK_API_KEY",  lambda key: DeepSeekProvider(key)),
}

def available() -> list:
    return list(_PROVIDERS.keys())

def get_provider(name: str) -> LLMProvider:
    entry = _PROVIDERS.get(name)
    if entry is None:
        raise ValueError(f"알 수 없는 프로바이더: {name} (가능: {available()})")

    env_key, factory = entry
    api_key = os.environ.get(env_key)
    if not api_key:
        raise RuntimeError(f"{env_key} 환경 변수가 없습니다.")
    return factory(api_key)

 

6. 정리 및 의의

인터페이스를 추상화하고 각 API 응답 구조를 LLMResponse로 정규화한 덕분에, 코어 에이전트 루프(agent_loop.py)는 호출 대상 모델에 전혀 종속되지 않는 독립성을 갖추게 되었다.

모델을 교체할 때도 루프는 전혀 건드리지 않고 get_provider("openai")나 get_provider("deepseek")를 호출해 주기만 하면 완벽하게 기능이 변경된다.

 

확장성 있는 멀티 프로바이더 뼈대가 안정적으로 구축되었으니, 다음 포스팅부터는 본격적으로 **Phase 2인 언리얼 엔진 통신 및 플러그인 연동**을 진행해보도록 하자.