AutoGen으로 멀티 에이전트 시스템 구축하기 — Microsoft AI 프레임워크 실전

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from autogen import UserProxyAgent

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",   # ALWAYS / TERMINATE / NEVER
    max_consecutive_auto_reply=10,
    code_execution_config={
        "work_dir": "workspace",
        "use_docker": False,  # True 권장 (보안)
    },
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from autogen import AssistantAgent

assistant = AssistantAgent(
    name="assistant",
    llm_config={
        "model": "gpt-4o",
        "api_key": "YOUR_OPENAI_API_KEY",
        "temperature": 0,
    },
    system_message="""당신은 데이터 분석 전문가입니다.
    Python 코드를 작성해 데이터를 분석하고,
    결과를 명확하게 설명합니다.
    모든 작업이 완료되면 반드시 'TERMINATE'로 대화를 종료합니다.""",
)

1
2
3
4
5

# 대화 시작
user_proxy.initiate_chat(
    assistant,
    message="pandas를 사용해서 다음 CSV 파일을 분석하고 요약 통계를 출력해주세요: sales_data.csv",
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

from autogen import GroupChat, GroupChatManager

# 에이전트 정의
researcher = AssistantAgent(
    name="researcher",
    llm_config=llm_config,
    system_message="""당신은 리서처입니다.
    주어진 주제를 조사하고 핵심 정보를 수집합니다.
    조사 완료 후 'RESEARCH_DONE'을 출력합니다.""",
)

writer = AssistantAgent(
    name="writer",
    llm_config=llm_config,
    system_message="""당신은 기술 작가입니다.
    리서처가 수집한 정보를 바탕으로 블로그 포스트를 작성합니다.
    작성 완료 후 'WRITING_DONE'을 출력합니다.""",
)

reviewer = AssistantAgent(
    name="reviewer",
    llm_config=llm_config,
    system_message="""당신은 편집자입니다.
    작성된 글을 검토하고 개선점을 제안합니다.
    최종 승인 시 'TERMINATE'를 출력합니다.""",
)

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    code_execution_config=False,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
)

# GroupChat 설정
group_chat = GroupChat(
    agents=[user_proxy, researcher, writer, reviewer],
    messages=[],
    max_round=20,
    speaker_selection_method="auto",  # LLM이 다음 발화자를 결정
)

manager = GroupChatManager(
    groupchat=group_chat,
    llm_config=llm_config,
)

# 대화 시작
user_proxy.initiate_chat(
    manager,
    message="AutoGen 프레임워크에 대한 기술 블로그 포스트를 작성해주세요.",
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# Docker 기반 코드 실행 설정
user_proxy = UserProxyAgent(
    name="executor",
    human_input_mode="NEVER",
    code_execution_config={
        "executor": "docker",
        "work_dir": "/tmp/autogen_workspace",
        "timeout": 60,
        "use_docker": "python:3.11-slim",  # 격리된 컨테이너
        "bind_volume": False,  # 호스트 파일시스템 마운트 차단
    },
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from autogen import register_function

# 웹 검색 함수 정의
def search_web(query: str) -> str:
    """웹에서 정보를 검색합니다."""
    # 실제 구현에서는 SerpAPI, Tavily 등을 사용
    results = tavily_client.search(query, max_results=5)
    return "\n".join([r["content"] for r in results["results"]])

# 에이전트에 함수 등록
register_function(
    search_web,
    caller=assistant,      # 함수를 호출할 에이전트
    executor=user_proxy,   # 함수를 실행할 에이전트
    name="search_web",
    description="주어진 쿼리로 인터넷을 검색하고 결과를 반환합니다.",
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

import autogen

llm_config = {"model": "gpt-4o", "api_key": "YOUR_KEY", "temperature": 0}

# 1. 데이터 엔지니어: 데이터 로딩과 전처리
data_engineer = autogen.AssistantAgent(
    name="data_engineer",
    llm_config=llm_config,
    system_message="""데이터 엔지니어로서 CSV/JSON 데이터를 로딩하고
    결측치 처리, 타입 변환 등 전처리 작업을 수행합니다.
    pandas 코드를 작성하며, 완료 시 'DATA_READY'를 출력합니다.""",
)

# 2. 데이터 사이언티스트: 분석과 시각화
data_scientist = autogen.AssistantAgent(
    name="data_scientist",
    llm_config=llm_config,
    system_message="""데이터 사이언티스트로서 전처리된 데이터를 분석하고
    통계 분석, 상관관계 분석, matplotlib 시각화를 수행합니다.
    인사이트 도출 완료 시 'ANALYSIS_DONE'을 출력합니다.""",
)

# 3. 리포트 작성자: 결과 문서화
report_writer = autogen.AssistantAgent(
    name="report_writer",
    llm_config=llm_config,
    system_message="""분석 결과를 바탕으로 경영진이 이해하기 쉬운
    마크다운 형식의 분석 리포트를 작성합니다.
    리포트 완성 후 'TERMINATE'를 출력합니다.""",
)

executor = autogen.UserProxyAgent(
    name="executor",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "analysis_workspace", "use_docker": False},
    is_termination_msg=lambda x: "TERMINATE" in x.get("content", ""),
)

# GroupChat 구성
gc = autogen.GroupChat(
    agents=[executor, data_engineer, data_scientist, report_writer],
    messages=[],
    max_round=30,
    speaker_selection_method="auto",
)

manager = autogen.GroupChatManager(groupchat=gc, llm_config=llm_config)

# 분석 시작
executor.initiate_chat(
    manager,
    message="""
    sales_2025.csv 파일을 분석해주세요.
    - 월별 매출 트렌드 분석
    - 상위 5개 제품 카테고리 분석
    - 전년 대비 성장률 계산
    - 시각화 포함
    결과를 마크다운 리포트로 정리해주세요.
    """,
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# AutoGen 0.4 스타일
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models import OpenAIChatCompletionClient

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

agent = AssistantAgent(name="assistant", model_client=model_client)
team = RoundRobinGroupChat([agent])

# async 기반 실행
import asyncio
asyncio.run(team.run(task="Hello, write a Python hello world."))

1
2
3
4
5
6
7
8

import autogen.runtime_logging

logging_session_id = autogen.runtime_logging.start(
    logger_type="sqlite",
    config={"dbname": "autogen_logs.db"},
)
# ... 에이전트 실행 ...
autogen.runtime_logging.stop()