AutoGen 모델 클라이언트

AutoGen 모델 클라이언트 (Model Clients)

AutoGen을 쓰다 보면 다양한 LLM 제공자를 어떻게 하나의 인터페이스로 다루는지 궁금해지죠. AutoGen은 ChatCompletion API를 쓰기 위한 내장 모델 클라이언트 세트를 제공하고, 모든 모델 클라이언트는 ChatCompletionClient 프로토콜 클래스를 구현합니다.

내장 모델 클라이언트

현재 다음과 같은 내장 모델 클라이언트를 지원해요:

OpenAIChatCompletionClient: OpenAI 모델과 OpenAI API 호환 모델(예: Gemini)용
AzureOpenAIChatCompletionClient: Azure OpenAI 모델용
AzureAIChatCompletionClient: GitHub 모델과 Azure에 호스팅된 모델용
OllamaChatCompletionClient (실험): Ollama에 호스팅된 로컬 모델용
AnthropicChatCompletionClient (실험): Anthropic에 호스팅된 모델용
SKChatCompletionAdapter: Semantic Kernel AI 커넥터용 어댑터

이 모델 클라이언트를 쓰는 방법에 대한 더 자세한 내용은 각 클라이언트의 문서를 참고하세요.

Log Model Calls

AutoGen은 표준 Python logging 모듈로 모델 호출과 응답 같은 이벤트를 기록해요. 로거 이름은 autogen_core.EVENT_LOGGER_NAME이고, 이벤트 타입은 LLMCall입니다.

출처: AutoGen 공식 문서 - Model Clients

import logging

from autogen_core import EVENT_LOGGER_NAME

logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger(EVENT_LOGGER_NAME)
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.INFO)

Call Model Client

모델 클라이언트를 호출하려면 ChatCompletionClient.create 메서드를 씁니다. 이 예제는 OpenAIChatCompletionClient로 OpenAI 모델을 호출해요.

from autogen_core.models import UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(
    model="gpt-4", temperature=0.3
)  # OPENAI_API_KEY 가 환경에 설정되어 있다고 가정.

result = await model_client.create(
    messages=[UserMessage(content="What is the capital of France?", source="user")]
)
print(result)
await model_client.close()

Streaming Tokens

ChatCompletionClient.create_stream 메서드를 쓰면 스트리밍 토큰 청크가 있는 채팅 완성 요청을 만들 수 있어요.

from autogen_core.models import CreateResult, UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4o")  # OPENAI_API_KEY 가 환경에 설정되어 있다고 가정.

messages = [
    UserMessage(
        content="What is the capital of France?",
        source="user",
    )
]
async for chunk in model_client.create_stream(messages=messages):
    print(chunk)

스트리밍 응답의 마지막 응답은 항상 `CreateResult` 타입의 최종 응답이에요.

기본 사용량 응답은 0 값을 반환해요. 사용량을 활성화하려면 `BaseOpenAIChatCompletionClient.create_stream`을 참고하세요.

Structured Output

구조화된 출력은 OpenAIChatCompletionClient와 AzureOpenAIChatCompletionClient의 response_format 필드를 Pydantic BaseModel 클래스로 설정하면 활성화됩니다.

구조화된 출력은 이를 지원하는 모델에서만 사용할 수 있어요. 모델 클라이언트도 구조화된 출력을 지원해야 합니다.
현재 `OpenAIChatCompletionClient`와 `AzureOpenAIChatCompletionClient`가 구조화된 출력을 지원해요.

from typing import Literal

from pydantic import BaseModel


# 에이전트의 응답 형식을 Pydantic base model로 정의한다.
class AgentResponse(BaseModel):
    thoughts: str
    response: Literal["happy", "sad", "neutral"]


# OpenAI 모델을 사용하는 모델 클라이언트를 만든다.
model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    response_format=AgentResponse,  # type: ignore
)

create 메서드의 extra_create_args 파라미터로도 response_format 필드를 설정해 요청별로 구조화된 출력을 구성할 수 있어요.

Caching Model Responses

autogen_ext는 어떤 ChatCompletionClient든 감쌀 수 있는 ChatCompletionCache를 구현합니다. 이 래퍼를 쓰면 같은 프롬프트로 기반 클라이언트를 여러 번 조회할 때 토큰 사용량이 발생하지 않아요.

ChatCompletionCache는 CacheStore 프로토콜을 사용합니다. 유용한 CacheStore 변형으로는 DiskCacheStore와 RedisStore가 구현되어 있어요.

로컬 캐싱에 diskcache를 쓰는 예시입니다:

# pip install -U "autogen-ext[openai, diskcache]"

import asyncio
import tempfile

from autogen_core.models import UserMessage
from autogen_ext.cache_store.diskcache import DiskCacheStore
from autogen_ext.models.cache import CHAT_CACHE_VALUE_TYPE, ChatCompletionCache
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    # 임시 디렉토리에 디스크 캐시 스토어를 만든다.
    cache_store = DiskCacheStore[CHAT_CACHE_VALUE_TYPE](tempfile.gettempdir())

    # 모델 클라이언트와 캐시 래퍼를 만든다.
    model_client = OpenAIChatCompletionClient(model="gpt-4o")
    cached_client = ChatCompletionCache(cache_store, model_client)

    # 첫 번째 호출은 모델로부터 응답을 가져와 캐시에 저장한다.
    result = await cached_client.create(
        messages=[UserMessage(content="What is the capital of France?", source="user")]
    )
    print(result)

    # 두 번째 호출은 캐시에서 응답을 가져온다.
    result = await cached_client.create(
        messages=[UserMessage(content="What is the capital of France?", source="user")]
    )
    print(result)

    await cached_client.close()


asyncio.run(main())

캐시된 응답 전후에 cached_client.total_usage()(또는 model_client.total_usage())를 검사하면 동일한 카운트가 나와야 해요.

캐싱은 cached_client.create나 cached_client.create_stream에 제공된 정확한 인자에 민감해서, tools나 json_output 인자를 바꾸면 캐시 미스가 날 수 있어요.

Build an Agent with a Model Client

ChatCompletion API로 메시지에 응답할 수 있는 간단한 AI 에이전트를 만들어 봅시다.

from dataclasses import dataclass

from autogen_core import MessageContext, RoutedAgent, SingleThreadedAgentRuntime, message_handler
from autogen_core.models import ChatCompletionClient, SystemMessage, UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient


@dataclass
class UserRequest:
    content: str


class SimpleAgent(RoutedAgent):
    def __init__(self, model_client: ChatCompletionClient) -> None:
        super().__init__("Simple Agent")
        self._system_message = SystemMessage(content="You are a helpful AI assistant.")
        self._model_client = model_client

    @message_handler
    async def handle_user_request(self, message: UserRequest, ctx: MessageContext) -> None:
        result = await self._model_client.create(
            messages=[self._system_message, UserMessage(content=message.content, source="user")],
            cancellation_token=ctx.cancellation_token,
        )
        print(result.content[0].content)  # type: ignore


# 런타임을 만들고 에이전트를 등록한다.
from autogen_core import AgentId

model_client = OpenAIChatCompletionClient(
    model="gpt-4o-mini",
    # api_key="sk-...", # OPENAI_API_KEY 가 환경에 설정되어 있으면 선택 사항.
)

runtime = SingleThreadedAgentRuntime()
await SimpleAgent.register(runtime, "simple_agent", lambda: SimpleAgent(model_client=model_client))
runtime.start()

# 에이전트에 메시지를 보낸다.
await runtime.send_message(
    UserRequest(content="What is the capital of France?"),
    AgentId("simple_agent", "default"),
)
await runtime.stop_when_idle()

SimpleAgent 클래스는 메시지를 적절한 핸들러로 자동 라우팅하는 편의를 위한 RoutedAgent 클래스의 하위 클래스예요. 사용자 메시지를 다루는 handle_user_request라는 단일 핸들러가 있고, ChatCompletionClient로 메시지에 응답을 생성해 직접 통신 모델에 따라 사용자에게 응답을 돌려줍니다.

`CancellationToken` 타입의 `cancellation_token`은 비동기 작업을 취소하는 데 씁니다.
메시지 핸들러 내부의 비동기 호출과 연결되어 있어서, 호출자가 핸들러를 취소하는 데 쓸 수 있어요.

위 SimpleAgent는 항상 시스템 메시지와 최신 사용자 메시지만 들어 있는 새 컨텍스트로 응답해요. autogen_core.model_context의 모델 컨텍스트 클래스를 쓰면 에이전트가 이전 대화를 "기억"하게 할 수 있습니다. 자세한 내용은 Model Context 페이지를 참고하세요.

API Keys From Environment Variables

위 예제들에서 api_key 인자로 API 키를 제공할 수 있음을 보여줬어요. 중요한 점은 OpenAI와 Azure OpenAI 클라이언트가 openai 패키지를 사용하는데, 이 패키지는 키가 제공되지 않으면 환경 변수에서 API 키를 자동으로 읽는다는 것이에요.

OpenAI의 경우 OPENAI_API_KEY 환경 변수를 설정하면 됩니다.
Azure OpenAI의 경우 AZURE_OPENAI_API_KEY 환경 변수를 설정하면 됩니다.

또한 Gemini(Beta)의 경우 GEMINI_API_KEY 환경 변수를 설정할 수 있어요.

이 방법은 민감한 API 키를 코드에 포함하지 않아도 되므로 탐색해 보기 좋은 관행이에요.

AutoGen 모델 클라이언트