AutoGen 에이전트

AutoGen 에이전트 (Agents)

여러 에이전트로 애플리케이션을 만들다 보면 "에이전트가 메시지에 어떻게 응답하는지"를 어디서 정하는지 궁금해지기 마련이에요. AutoGen AgentChat은 프리셋 에이전트 세트를 제공해서, 에이전트가 메시지에 응답하는 방식을 상황에 맞게 고르게 해 줍니다.

모든 에이전트는 공통 속성과 메서드를 공유해요:

  • name: 에이전트의 고유 이름
  • description: 에이전트를 설명하는 텍스트
  • run: 문자열 태스크나 메시지 리스트를 받아 에이전트를 실행하고 TaskResult를 돌려주는 메서드. 에이전트는 상태를 유지하도록 설계되어서, 이 메서드는 완전한 히스토리가 아니라 새로운 메시지를 받아 호출해야 해요.
  • run_stream: run과 같지만, BaseAgentEvent나 BaseChatMessage를 상속한 메시지 이터레이터를 반환하고 마지막 항목으로 TaskResult가 따라와요.

Assistant Agent

AssistantAgent는 언어 모델을 사용하면서 도구도 쓸 수 있는 내장 에이전트예요.

출처: AutoGen 공식 문서 - Agents

`AssistantAgent`는 프로토타이핑과 학습 목적의 "주방 싱크(kitchen sink)" 에이전트예요 — 아주 범용적이죠.
설계 선택을 이해하려면 문서와 구현을 꼭 읽어보세요. 설계를 완전히 이해한 뒤에는 직접 에이전트를 구현하고 싶어질 거예요.
[Custom Agent](../custom-agents.ipynb)를 참고하세요.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import StructuredMessage
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

Getting Result

run 메서드를 쓰면 주어진 태스크로 에이전트를 실행할 수 있어요.

# 스크립트에서 실행할 때는 asyncio.run(agent.run(...)) 를 사용하세요.
result = await agent.run(task="Find information on AutoGen")
print(result.messages)

run 호출은 TaskResult를 돌려주고, 그 안의 messages 속성에 에이전트의 "사고 과정"과 최종 응답이 들어 있어요.

중요한 점은 `run`이 에이전트의 내부 상태를 갱신한다는 거예요 — 메시지를 에이전트의 메시지 히스토리에 추가합니다.
태스크 없이 `run`을 호출하면 현재 상태 기준으로 응답을 생성하게 됩니다.

v0.2 AgentChat과 달리, 도구는 에이전트가 같은 `run` 호출 안에서 직접 실행해요.
기본적으로 에이전트는 도구 호출 결과를 최종 응답으로 돌려줍니다.

Multi-Modal Input

AssistantAgent는 입력을 MultiModalMessage로 제공하면 멀티모달 입력을 처리할 수 있어요.

from io import BytesIO

import PIL
import requests
from autogen_agentchat.messages import MultiModalMessage
from autogen_core import Image

# 랜덤 이미지와 텍스트로 멀티모달 메시지를 만든다.
pil_image = PIL.Image.open(BytesIO(requests.get("https://picsum.photos/300/200").content))
img = Image(pil_image)
multi_modal_message = MultiModalMessage(content=["Can you describe the content of this image?"], source="user")

# 스크립트에서 실행할 때는 asyncio.run(...) 을 사용하세요.
result = await agent.run(task=multi_modal_message)
print(result.messages[-1].content)  # type: ignore

Streaming Messages

run_stream 메서드를 쓰면 에이전트가 생성하는 각 메시지를 흘려보낼 수 있어요. Console로 출력하면 메시지가 나타나는 대로 콘솔에 찍힙니다.

async def assistant_run_stream() -> None:
    # 옵션 1: 스트림에서 각 메시지를 읽기 (이전 예제처럼).
    # async for message in agent.run_stream(task="Find information on AutoGen"):
    #     print(message)

    # 옵션 2: Console로 나타나는 대로 모든 메시지 출력하기.
    await Console(
        agent.run_stream(task="Find information on AutoGen"),
        output_stats=True,
    )

run_stream은 에이전트가 생성한 각 메시지를 산출하는 비동기 제너레이터를 반환하고, 마지막 항목으로 TaskResult가 따라와요. 메시지를 보면 assistant 에이전트가 web_search 도구를 사용해 정보를 모으고, 그 검색 결과를 바탕으로 응답한 걸 확인할 수 있어요.

Using Tools and Workbench

대형 언어 모델(LLM)은 보통 텍스트나 코드 응답만 생성하도록 제한돼 있어요. 하지만 복잡한 태스크는 API나 데이터베이스에서 데이터를 가져오는 것 같은 외부 도구 사용이 필요한 경우가 많죠.

이 한계를 해결하기 위해 현대 LLM은 사용 가능한 도구 스키마(도구와 그 인자에 대한 설명) 목록을 받아서 도구 호출 메시지를 생성할 수 있어요. 이 능력을 Tool Calling 또는 Function Calling이라고 하고, 지능형 에이전트 기반 애플리케이션을 만드는 데 점점 널리 쓰이는 패턴이 되고 있어요. LLM에서의 도구 호출에 대한 더 자세한 내용은 OpenAI와 Anthropic 문서를 참고하세요.

AgentChat에서 AssistantAgent는 도구를 사용해 특정 동작을 수행할 수 있어요. web_search 도구가 바로 그런 도구 중 하나죠. 단일 커스텀 도구는 Python 함수이거나 BaseTool의 하위 클래스일 수 있어요.

반면 Workbench는 상태와 리소스를 공유하는 도구들의 모음이에요.

모델 클라이언트를 도구 및 워크벤치와 직접 쓰는 방법은 Core User Guide의 [Tools](../../core-user-guide/components/tools.ipynb)와 [Workbench](../../core-user-guide/components/workbench.ipynb) 섹션을 참고하세요.

기본적으로 AssistantAgent가 도구를 실행하면, 그 출력을 응답의 ToolCallSummaryMessage에 문자열로 돌려줘요. 도구가 자연어로 된 잘 구성된 문자열을 반환하지 않는다면, AssistantAgent 생성자의 reflect_on_tool_use=True 파라미터를 켜서 모델이 도구 출력을 요약하는 반성(reflection) 단계를 추가할 수 있어요.

Built-in Tools and Workbench

AutoGen Extension은 Assistant Agent와 함께 쓸 수 있는 내장 도구 세트를 제공해요. 사용 가능한 모든 도구는 autogen_ext.tools 네임스페이스 아래 API 문서에서 확인할 수 있어요. 예를 들면:

  • autogen_ext.tools.graphrag: GraphRAG 인덱스를 사용하는 도구
  • autogen_ext.tools.http: HTTP 요청을 만드는 도구
  • autogen_ext.tools.langchain: LangChain 도구를 쓰기 위한 어댑터
  • autogen_ext.tools.mcp: Model Chat Protocol(MCP) 서버를 쓰기 위한 도구와 워크벤치

Function Tool

AssistantAgent는 Python 함수를 자동으로 FunctionTool로 변환해서 에이전트가 도구로 쓸 수 있게 하고, 함수 시그니처와 docstring에서 도구 스키마를 자동으로 생성해요.

from autogen_core.tools import FunctionTool


# Python 함수로 도구를 정의한다.
async def web_search_func(query: str) -> str:
    """Find information on the web"""
    return "AutoGen is a programming framework for building multi-agent applications."


# 도구가 Python 함수라면 이 단계는 AssistantAgent 내부에서 자동으로 수행된다.
web_search_function_tool = FunctionTool(web_search_func, description="Search the web for information")

Model Context Protocol (MCP) Workbench

AssistantAgent는 McpWorkbench를 사용해 Model Context Protocol(MCP) 서버에서 제공되는 도구도 사용할 수 있어요.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import McpWorkbench, StdioServerParams

# mcp-server-fetch에서 fetch 도구를 가져온다.
fetch_mcp_server = StdioServerParams(command="uvx", args=["mcp-server-fetch"])

# MCP 워크벤치를 만든다.
workbench = McpWorkbench(server_params=fetch_mcp_server)

# 워크벤치 도구를 쓰는 에이전트를 만든다.
agent = AssistantAgent(
    name="assistant",
    model_client=OpenAIChatCompletionClient(model="gpt-4o"),
    tools=workbench.get_tools(),  # type: ignore
    system_message="Use tools to solve tasks.",
)

Agent as a Tool

어떤 BaseChatAgent든 AgentTool로 감싸면 도구로 사용할 수 있어요. 이렇게 하면 에이전트가 태스크를 풀기 위해 다른 에이전트를 도구로 호출하는, 모델 주도의 동적인 멀티 에이전트 워크플로우를 만들 수 있어요.

Parallel Tool Calls

일부 모델은 병렬 도구 호출을 지원해서, 여러 도구를 동시에 호출해야 하는 태스크에 유용해요. 기본적으로 모델 클라이언트가 여러 도구 호출을 생성하면 AssistantAgent는 도구를 병렬로 호출합니다.

도구가 서로 간섭할 수 있는 부작용(사이드 이펙트)을 가지거나, 모델 간에 일관된 동작이 필요할 때는 병렬 도구 호출을 끄고 싶을 거예요. 이 설정은 모델 클라이언트 레벨에서 해야 해요.

`AgentTool`이나 `TeamTool`을 쓸 때는 **반드시** 병렬 도구 호출을 꺼야 해요. 에이전트와 팀은 내부 상태를 유지해서 병렬 실행과 충돌할 수 있기 때문에 동시에 실행될 수 없어요.

OpenAIChatCompletionClient와 AzureOpenAIChatCompletionClient에서는 parallel_tool_calls=False로 설정해서 병렬 도구 호출을 끕니다.

model_client_no_parallel_tool_call = OpenAIChatCompletionClient(
    model="gpt-4o",
    parallel_tool_calls=False,  # type: ignore
)
agent_no_parallel_tool_call = AssistantAgent(
    name="assistant",
    model_client=model_client_no_parallel_tool_call,
    tools=[web_search],
    system_message="Use tools to solve tasks.",
)

Tool Iterations

모델 호출 하나 + 도구 호출 하나(또는 병렬 도구 호출)가 하나의 도구 반복입니다. 기본적으로 AssistantAgent는 최대 한 번의 반복만 실행해요.

모델이 도구 호출 생성을 멈추거나 최대 반복 수에 도달할 때까지 여러 번 반복하도록 에이전트를 구성할 수 있어요. 최대 반복 수는 AssistantAgent 생성자의 max_tool_iterations 파라미터로 제어합니다.

agent_loop = AssistantAgent(
    name="assistant_loop",
    model_client=model_client_no_parallel_tool_call,
    tools=[web_search],
    system_message="Use tools to solve tasks.",
    max_tool_iterations=10,  # 루프를 멈추기 전 최대 10회의 도구 호출 반복.
)

Structured Output

구조화된 출력은 애플리케이션이 제공하는 미리 정의된 스키마대로 모델이 구조화된 JSON 텍스트를 반환하게 합니다. JSON 모드와 달리 스키마를 Pydantic BaseModel 클래스로 제공할 수 있고, 출력 검증에도 쓸 수 있어요.

AssistantAgent 생성자의 output_content_type 파라미터에 base model 클래스를 지정하면, 에이전트는 content의 타입이 base model 클래스인 StructuredMessage로 응답합니다.

이렇게 하면 에이전트의 응답을 애플리케이션에 직접 통합하고, 모델 출력을 구조화된 객체로 사용할 수 있어요.

`output_content_type`을 설정하면 기본적으로 에이전트가 도구 사용을 반성하고 도구 호출 결과를 바탕으로 구조화된 출력 메시지를 반환하도록 요구합니다.
이 동작을 끄려면 `reflect_on_tool_use=False`로 명시적으로 설정하세요.

구조화된 출력은 에이전트 응답에 Chain-of-Thought(사고 과정) 추론을 통합하는 데도 유용해요.

from typing import Literal

from pydantic import BaseModel


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


# OpenAI GPT-4o 모델을 사용하는 에이전트를 만든다.
model_client = OpenAIChatCompletionClient(model="gpt-4o")
agent = AssistantAgent(
    "assistant",
    model_client=model_client,
    system_message="You are a helpful assistant.",
    output_content_type=AgentResponse,
)

Streaming Tokens

model_client_stream=True로 설정하면 모델 클라이언트가 생성하는 토큰을 스트리밍할 수 있어요. 그러면 에이전트가 run_stream에서 ModelClientStreamingChunkEvent 메시지를 산출합니다.

이 기능이 동작하려면 기반 모델 API가 토큰 스트리밍을 지원해야 해요. 모델 제공자에 지원 여부를 확인하세요.

model_client = OpenAIChatCompletionClient(model="gpt-4o")

streaming_assistant = AssistantAgent(
    name="assistant",
    model_client=model_client,
    system_message="You are a helpful assistant.",
    model_client_stream=True,  # 토큰 스트리밍 활성화.
)

# 스크립트에서는 비동기 함수와 asyncio.run()을 사용하세요.
async for message in streaming_assistant.run_stream(task="Name two cities in South Korea"):
    print(message)

위 출력에서 스트리밍 청크를 볼 수 있어요. 청크는 모델 클라이언트가 생성하고, 받는 대로 에이전트가 산출합니다. 모든 청크를 이어 붙인 최종 응답은 마지막 청크 직후에 산출돼요.

Using Model Context

AssistantAgent는 model_context 파라미터로 ChatCompletionContext 객체를 전달받을 수 있어요. 이렇게 하면 BufferedChatCompletionContext처럼 서로 다른 모델 컨텍스트를 써서 모델에 보내는 컨텍스트를 제한할 수 있습니다.

기본적으로 AssistantAgent는 전체 대화 히스토리를 모델에 보내는 UnboundedChatCompletionContext를 사용해요. 컨텍스트를 마지막 n개 메시지로 제한하려면 BufferedChatCompletionContext를, 토큰 수로 제한하려면 TokenLimitedChatCompletionContext를 쓸 수 있어요.

from autogen_core.model_context import BufferedChatCompletionContext

# 컨텍스트에서 마지막 5개 메시지만 사용해 응답을 생성하는 에이전트를 만든다.
agent = AssistantAgent(
    name="assistant",
    model_client=model_client,
    tools=[web_search],
    system_message="Use tools to solve tasks.",
    model_context=BufferedChatCompletionContext(buffer_size=5),
)

Other Preset Agents

다음과 같은 프리셋 에이전트도 사용할 수 있어요:

  • UserProxyAgent: 사용자 입력을 받아 응답으로 돌려주는 에이전트
  • CodeExecutorAgent: 코드를 실행할 수 있는 에이전트
  • OpenAIAssistantAgent: OpenAI Assistant가 뒷받침하고 커스텀 도구를 사용할 수 있는 에이전트
  • MultimodalWebSurfer: 웹을 검색하고 웹 페이지를 방문해 정보를 얻는 멀티모달 에이전트
  • FileSurfer: 로컬 파일을 검색하고 탐색하는 에이전트
  • VideoSurfer: 비디오를 보며 정보를 얻는 에이전트

Next Step

AssistantAgent 사용법을 살펴봤으니, 이제 다음 섹션에서 AgentChat의 팀(teams) 기능을 배워볼 수 있어요.

더 알아보기