팀(Teams)
팀(Teams)
이번 섹션에서는 AutoGen으로 멀티에이전트 팀(multi-agent team, 줄여서 팀) 을 만드는 법을 배워요. 팀은 공통의 목표를 이루기 위해 함께 작업하는 에이전트들의 그룹이에요.
먼저 팀을 만들고 실행하는 방법을 보여드릴게요. 그 다음엔 디버깅과 성능 이해에 중요한 팀의 동작을 관찰하는 방법과, 팀의 동작을 제어하는 일반적인 작업들을 설명할게요.
AgentChat은 여러 팀 프리셋을 지원해요.
RoundRobinGroupChat: 참가자들이 라운드로빈 방식으로 차례로 말하는 그룹 채팅을 실행하는 팀입니다(이 페이지에서 다룸).SelectorGroupChat: 각 메시지 후에 ChatCompletion 모델로 다음 발언자를 선택하는 팀입니다 (튜토리얼).MagenticOneGroupChat: 다양한 도메인에 걸쳐 열린 웹·파일 기반 태스크를 푸는 범용 멀티에이전트 시스템입니다(튜토리얼).Swarm:HandoffMessage로 에이전트 간 전환을 신호하는 팀입니다(튜토리얼).
참고: 언제 팀을 써야 하나요?
팀은 협업과 다양한 전문성이 필요한 복잡한 태스크를 위한 거예요. 하지만 단일 에이전트보다 조종하기 위해 더 많은 스캐폴딩(scaffolding)이 필요합니다. AutoGen이 팀 작업을 간단하게 만들어 주긴 하지만, 단순한 태스크에는 단일 에이전트로 시작하고, 단일 에이전트로 부족함이 입증됐을 때 멀티에이전트 팀으로 전환하세요. 팀 기반 접근으로 옮기기 전에, 단일 에이전트를 적절한 도구와 지시로 최적화했는지 확인하세요.
팀 만들기
RoundRobinGroupChat은 모든 에이전트가 같은 컨텍스트를 공유하고 라운드로빈 방식으로 차례로 응답하는, 단순하면서도 효과적인 팀 구성이에요. 각 에이전트는 자기 차례에 응답을 모든 다른 에이전트에게 브로드캐스트해서, 팀 전체가 일관된 컨텍스트를 유지하게 해요.
두 개의 AssistantAgent와, 에이전트 응답에서 특정 단어가 감지되면 팀을 멈추는 TextMentionTermination 조건으로 팀을 만들어 볼게요.
두 에이전트 팀은 리플렉션(reflection) 패턴을 구현해요. 이는 평론가(critic) 에이전트가 주 에이전트의 응답을 평가하는 멀티에이전트 디자인 패턴이에요. 리플렉션 패턴에 대해 더 알고 싶다면 Core API를 참고하세요.
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.base import TaskResult
from autogen_agentchat.conditions import ExternalTermination, TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient
# Create an OpenAI model client.
model_client = OpenAIChatCompletionClient(
model="gpt-4o-2024-08-06",
# api_key="sk-...", # Optional if you have an OPENAI_API_KEY env variable set.
)
# Create the primary agent.
primary_agent = AssistantAgent(
"primary",
model_client=model_client,
system_message="You are a helpful AI assistant.",
)
# Create the critic agent.
critic_agent = AssistantAgent(
"critic",
model_client=model_client,
system_message="Provide constructive feedback. Respond with 'APPROVE' to when your feedbacks are addressed.",
)
# Define a termination condition that stops the task if the critic approves.
text_termination = TextMentionTermination("APPROVE")
# Create a team with the primary and critic agents.
team = RoundRobinGroupChat([primary_agent, critic_agent], termination_condition=text_termination)
팀 실행하기
run() 메서드를 호출해 태스크로 팀을 시작해 봐요.
# Use `asyncio.run(...)` when running in a script.
result = await team.run(task="Write a short poem about the fall season.")
print(result)
팀은 종료 조건이 충족될 때까지 에이전트를 실행해요. 이 경우 에이전트 응답에서 "APPROVE"라는 단어가 감지되어 종료 조건이 충족될 때까지, 팀은 라운드로빈 순서로 에이전트를 실행했어요. 팀이 멈추면 팀 안의 에이전트들이 만든 모든 메시지를 담은 TaskResult 객체를 반환해요.
팀 관찰하기
에이전트의 on_messages_stream 메서드와 비슷하게, 팀이 실행되는 동안 그 메시지를 스트리밍하려면 run_stream() 메서드를 호출하면 돼요. 이 메서드는 팀 안의 에이전트가 생성하는 메시지를 생성(generate)되는 대로 뱉어내는 제너레이터를 반환하는데, 마지막 항목은 TaskResult 객체예요.
# When running inside a script, use a async main function and call it from `asyncio.run(...)`.
await team.reset() # Reset the team for a new task.
async for message in team.run_stream(task="Write a short poem about the fall season."): # type: ignore
if isinstance(message, TaskResult):
print("Stop Reason:", message.stop_reason)
else:
print(message)
위 예제에서 보듯, TaskResult.stop_reason 속성을 확인하면 팀이 멈춘 이유를 알 수 있어요.
Console 메서드는 적절한 포맷으로 메시지를 콘솔에 출력하는 편리한 방법을 제공해요.
await team.reset() # Reset the team for a new task.
await Console(team.run_stream(task="Write a short poem about the fall season.")) # Stream the messages to the console.
팀 리셋하기
reset() 메서드를 호출하면 팀을 리셋할 수 있어요. 이 메서드는 모든 에이전트를 포함한 팀의 상태를 지워요. 각 에이전트의 on_reset 메서드를 호출해서 에이전트의 상태도 지웁니다.
await team.reset() # Reset the team for the next run.
다음 태스크가 이전 태스크와 관련이 없다면 팀을 리셋하는 게 일반적으로 좋아요. 하지만 다음 태스크가 이전 태스크와 관련이 있다면 리셋할 필요 없이 팀을 재개(resume) 할 수 있어요.
팀 멈추기
TextMentionTermination처럼 팀의 내부 상태에 근거해 팀을 멈추는 자동 종료 조건과 별개로, ExternalTermination을 이용해 외부에서 팀을 멈출 수도 있어요.
ExternalTermination에 set을 호출하면 현재 에이전트의 차례가 끝났을 때 팀이 멈춰요. 따라서 팀이 즉시 멈추지 않을 수도 있어요. 이렇게 하면 현재 에이전트가 자기 차례를 마치고 마지막 메시지를 팀에 브로드캐스트한 뒤에 팀이 멈춰서, 팀의 상태가 일관되게 유지돼요.
# Create a new team with an external termination condition.
external_termination = ExternalTermination()
team = RoundRobinGroupChat(
[primary_agent, critic_agent],
termination_condition=external_termination | text_termination, # Use the bitwise OR operator to combine conditions.
)
# Run the team in a background task.
run = asyncio.create_task(Console(team.run_stream(task="Write a short poem about the fall season.")))
# Wait for some time.
await asyncio.sleep(0.1)
# Stop the team.
external_termination.set()
# Wait for the team to finish.
await run
위 출력에서 팀이 외부 종료 조건 충족으로 멈췄지만, 발언 중이던 에이전트는 팀이 멈추기 전에 자기 차례를 마칠 수 있었던 걸 볼 수 있어요.
팀 재개하기
팀은 상태를 유지하는(stateful) 주체라서, 팀을 리셋하지 않는 한 각 실행 후에 대화 히스토리와 컨텍스트를 유지해요.
새 태스크 없이 run()이나 run_stream() 메서드를 다시 호출하면 팀이 중단된 지점부터 계속해서 팀을 재개할 수 있어요. RoundRobinGroupChat은 라운드로빈 순서상 다음 에이전트부터 계속합니다.
await Console(team.run_stream()) # Resume the team to continue the last task.
위 출력에서 팀이 중단된 지점부터 재개됐고, 첫 메시지가 팀이 멈추기 전에 마지막으로 발언한 에이전트의 다음 에이전트로부터 온 걸 볼 수 있어요.
이전 태스크의 컨텍스트를 유지하면서 새 태스크로 팀을 다시 재개해 봐요.
# The new task is to translate the same poem to Chinese Tang-style poetry.
await Console(team.run_stream(task="将这首诗用中文唐诗风格写一遍。"))
팀 중단(Abort)하기
실행 중인 run()이나 run_stream() 호출은, cancellation_token 매개변수로 전달된 CancellationToken을 설정해서 중단할 수 있어요.
팀 멈추기(stop)와 달리 팀 중단(abort) 은 팀을 즉시 멈추고 CancelledError 예외를 던져요.
참고: 팀이 중단되면 호출자는
CancelledError예외를 받아요.
# Create a cancellation token.
cancellation_token = CancellationToken()
# Use another coroutine to run the team.
run = asyncio.create_task(
team.run(
task="Translate the poem to Spanish.",
cancellation_token=cancellation_token,
)
)
# Cancel the run.
cancellation_token.cancel()
try:
result = await run # This will raise a CancelledError.
except asyncio.CancelledError:
print("Task was cancelled.")
단일 에이전트 팀
참고: 버전 0.6.2부터
AssistantAgent를max_tool_iterations와 함께 사용해 여러 번의 도구 호출 반복으로 에이전트를 실행할 수 있어요. 그래서 단지 에이전트를 도구 호출 루프로 돌리고 싶다면 단일 에이전트 팀이 필요 없을 수도 있어요.
종종 단일 에이전트를 팀 구성으로 실행하고 싶을 때가 있어요. 이는 종료 조건이 충족될 때까지 AssistantAgent를 루프로 실행하는 데 유용하죠.
이는 AssistantAgent를 run()이나 run_stream() 메서드로 실행하는 것과 달라요. 그 메서드들은 에이전트를 한 단계만 실행하고 결과를 반환하죠. 단일 단계에 대한 자세한 내용은 AssistantAgent를 참고하세요.
아래는 TextMessageTermination 조건과 함께 RoundRobinGroupChat 팀 구성으로 단일 에이전트를 실행하는 예시예요. 태스크는 도구로 숫자를 10에 도달할 때까지 증가시키는 거예요. 에이전트는 숫자가 10에 도달할 때까지 계속 도구를 호출하고, 그 다음 실행을 멈추게 할 최종 TextMessage를 반환합니다.
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMessageTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
model_client = OpenAIChatCompletionClient(
model="gpt-4o",
# api_key="sk-...", # Optional if you have an OPENAI_API_KEY env variable set.
# Disable parallel tool calls for this example.
parallel_tool_calls=False, # type: ignore
)
# Create a tool for incrementing a number.
def increment_number(number: int) -> int:
"""Increment a number by 1."""
return number + 1
# Create a tool agent that uses the increment_number function.
looped_assistant = AssistantAgent(
"looped_assistant",
model_client=model_client,
tools=[increment_number], # Register the tool.
system_message="You are a helpful AI assistant, use the tool to increment the number.",
)
# Termination condition that stops the task if the agent responds with a text message.
termination_condition = TextMessageTermination("looped_assistant")
# Create a team with the looped assistant agent and the termination condition.
team = RoundRobinGroupChat(
[looped_assistant],
termination_condition=termination_condition,
)
# Run the team with a task and print the messages to the console.
async for message in team.run_stream(task="Increment the number 5 to 10."): # type: ignore
print(type(message).__name__, message)
await model_client.close()
핵심은 종료 조건에 집중하는 거예요. 이 예제에서는 에이전트가 더 이상 ToolCallSummaryMessage를 만들지 않을 때 팀을 멈추는 TextMessageTermination 조건을 쓰고 있어요. 팀은 에이전트가 최종 결과를 담은 TextMessage를 만들 때까지 계속 실행됩니다.
에이전트를 제어하기 위해 다른 종료 조건을 쓸 수도 있어요. 자세한 내용은 Termination Conditions를 참고하세요.