종료 조건(Termination)
종료 조건(Termination)
이전 섹션에서 에이전트를 정의하고, 태스크를 해결할 수 있는 팀으로 조직하는 방법을 살펴봤어요. 하지만 실행은 영원히 계속될 수 있고, 많은 경우 그걸 언제 멈출지(when to stop) 알아야 해요. 이 역할을 담당하는 게 바로 종료 조건(termination condition) 이에요.
AgentChat은 기본 TerminationCondition 클래스와 그를 상속한 여러 구현체를 제공해 여러 종료 조건을 지원해요.
종료 조건은 마지막으로 조건이 호출된 이후의 BaseAgentEvent 또는 BaseChatMessage 객체 시퀀스를 받고, 대화를 종료해야 한다면 StopMessage를, 아니면 None을 반환하는 callable이에요. 종료 조건에 도달하면, 다시 사용하기 전에 reset()을 호출해 리셋해야 해요.
종료 조건에 대해 알아둬야 할 중요한 점들:
- 상태를 유지하지만 각 실행(
run()이나run_stream())이 끝나면 자동으로 리셋돼요. - AND와 OR 연산자로 결합할 수 있어요.
참고: 그룹 채팅 팀(즉
RoundRobinGroupChat,SelectorGroupChat,Swarm)에서는 각 에이전트가 응답할 때마다 종료 조건이 호출돼요. 응답이 여러 내부 메시지를 담을 수 있지만, 팀은 단일 응답의 모든 메시지에 대해 종료 조건을 한 번만 호출합니다. 따라서 조건은 마지막 호출 이후의 메시지 "델타 시퀀스(delta sequence)"로 호출돼요.
내장 종료 조건들:
MaxMessageTermination: 에이전트 메시지와 태스크 메시지를 포함해 지정된 수의 메시지가 생산된 후 멈춥니다.TextMentionTermination: 메시지에서 특정 텍스트·문자열(예: "TERMINATE")이 언급되면 멈춥니다.TokenUsageTermination: 특정 수의 prompt 또는 completion 토큰이 사용되면 멈춥니다. 에이전트가 메시지에서 토큰 사용량을 보고해야 해요.TimeoutTermination: 지정된 시간(초) 후 멈춥니다.HandoffTermination: 특정 대상으로의 핸드오프가 요청되면 멈춥니다. 핸드오프 메시지는Swarm같은 패턴을 만드는 데 쓸 수 있어요. 이는 에이전트가 사용자에게 핸드오프할 때 실행을 멈추고 애플리케이션·사용자가 입력을 제공하게 하려는 경우에 유용해요.SourceMatchTermination: 특정 에이전트가 응답한 후 멈춥니다.ExternalTermination: 실행 외부에서 종료를 프로그래밍 방식으로 제어할 수 있게 해줍니다. UI 통합(예: 채팅 인터페이스의 "정지" 버튼)에 유용해요.StopMessageTermination: 에이전트가StopMessage를 생산하면 멈춥니다.TextMessageTermination: 에이전트가TextMessage를 생산하면 멈춥니다.FunctionCallTermination: 일치하는 이름의FunctionExecutionResult를 담은ToolCallExecutionEvent를 에이전트가 생산하면 멈춥니다.FunctionalTermination: 마지막 메시지 델타 시퀀스에서 함수 표현식이True로 평가되면 멈춥니다. 내장 조건으로 다루지 않는 커스텀 종료 조건을 빠르게 만들 때 유용해요.
기본 사용법
종료 조건의 특성을 보여주기 위해, 텍스트 생성 담당의 주(primary) 에이전트와 생성된 텍스트를 검토·피드백하는 평론가(critic) 에이전트, 두 에이전트로 구성된 팀을 만들 거예요.
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
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",
temperature=1,
# 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 for every message. Respond with 'APPROVE' to when your feedbacks are addressed.",
)
종료 조건이 각 run 또는 run_stream 호출 후 자동으로 리셋되어, 팀이 중단된 지점부터 대화를 재개할 수 있게 하는 방식을 살펴봐요.
max_msg_termination = MaxMessageTermination(max_messages=3)
round_robin_team = RoundRobinGroupChat([primary_agent, critic_agent], termination_condition=max_msg_termination)
# Use asyncio.run(...) if you are running this script as a standalone script.
await Console(round_robin_team.run_stream(task="Write a unique, Haiku about the weather in Paris"))
대화는 최대 메시지 수에 도달한 후 멈췄어요. 주 에이전트가 피드백에 응답할 기회를 얻지 못했으니, 대화를 계속해 봐요.
# Use asyncio.run(...) if you are running this script as a standalone script.
await Console(round_robin_team.run_stream())
팀은 중단된 지점부터 계속되어, 주 에이전트가 피드백에 응답할 수 있었어요.
종료 조건 결합하기
종료 조건이 AND(&)와 OR(|) 연산자로 결합되어 더 복잡한 종료 로직을 만드는 방법을 보여드릴게요. 예를 들어 10개의 메시지가 생성되거나 평론가 에이전트가 메시지를 승인할 때 멈추는 팀을 만들어 볼게요.
max_msg_termination = MaxMessageTermination(max_messages=10)
text_termination = TextMentionTermination("APPROVE")
combined_termination = max_msg_termination | text_termination
round_robin_team = RoundRobinGroupChat([primary_agent, critic_agent], termination_condition=combined_termination)
# Use asyncio.run(...) if you are running this script as a standalone script.
await Console(round_robin_team.run_stream(task="Write a unique, Haiku about the weather in Paris"))
대화는 평론가 에이전트가 메시지를 승인한 후 멈췄어요. 물론 10개의 메시지가 생성됐다면 그때도 멈출 수 있었죠.
반대로 두 조건이 모두 충족될 때만 실행을 멈추고 싶다면 AND(&) 연산자를 쓰면 돼요.
combined_termination = max_msg_termination & text_termination
커스텀 종료 조건
내장 종료 조건은 대부분의 사용 사례에 충분해요. 하지만 기존 조건에 맞지 않는 커스텀 종료 조건을 구현해야 하는 경우가 있을 수 있어요. TerminationCondition 클래스를 상속하면 만들 수 있습니다.
이 예제에서는 특정 함수 호출이 이루어질 때 대화를 멈추는 커스텀 종료 조건을 만들어 볼게요.
from typing import Sequence
from autogen_agentchat.base import TerminatedException, TerminationCondition
from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage, StopMessage, ToolCallExecutionEvent
from autogen_core import Component
from pydantic import BaseModel
from typing_extensions import Self
class FunctionCallTerminationConfig(BaseModel):
"""Configuration for the termination condition to allow for serialization
and deserialization of the component.
"""
function_name: str
class FunctionCallTermination(TerminationCondition, Component[FunctionCallTerminationConfig]):
"""Terminate the conversation if a FunctionExecutionResult with a specific name is received."""
component_config_schema = FunctionCallTerminationConfig
component_provider_override = "autogen_agentchat.conditions.FunctionCallTermination"
"""The schema for the component configuration."""
def __init__(self, function_name: str) -> None:
self._terminated = False
self._function_name = function_name
@property
def terminated(self) -> bool:
return self._terminated
async def __call__(self, messages: Sequence[BaseAgentEvent | BaseChatMessage]) -> StopMessage | None:
if self._terminated:
raise TerminatedException("Termination condition has already been reached")
for message in messages:
if isinstance(message, ToolCallExecutionEvent):
for execution in message.content:
if execution.name == self._function_name:
self._terminated = True
return StopMessage(
content=f"Function '{self._function_name}' was executed.",
source="FunctionCallTermination",
)
return None
async def reset(self) -> None:
self._terminated = False
def _to_config(self) -> FunctionCallTerminationConfig:
return FunctionCallTerminationConfig(
function_name=self._function_name,
)
@classmethod
def _from_config(cls, config: FunctionCallTerminationConfig) -> Self:
return cls(
function_name=config.function_name,
)
이 새 종료 조건을 사용해, 평론가 에이전트가 approve 함수 호출로 메시지를 승인할 때 대화를 멈추게 해볼게요.
먼저 평론가 에이전트가 메시지를 승인할 때 호출되는 간단한 함수를 만들어요.
def approve() -> None:
"""Approve the message when all feedbacks have been addressed."""
pass
그 다음 에이전트들을 만들어요. 평론가 에이전트에는 approve 도구가 장착돼 있어요.
from autogen_agentchat.agents import AssistantAgent
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",
temperature=1,
# 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 with the approve function as a tool.
critic_agent = AssistantAgent(
"critic",
model_client=model_client,
tools=[approve], # Register the approve function as a tool.
system_message="Provide constructive feedback. Use the approve tool to approve when all feedbacks are addressed.",
)
이제 종료 조건과 팀을 만들어요. 시를 쓰는 태스크로 팀을 실행합니다.
function_call_termination = FunctionCallTermination(function_name="approve")
round_robin_team = RoundRobinGroupChat([primary_agent, critic_agent], termination_condition=function_call_termination)
# Use asyncio.run(...) if you are running this script as a standalone script.
await Console(round_robin_team.run_stream(task="Write a unique, Haiku about the weather in Paris"))
await model_client.close()
평론가 에이전트가 approve 함수 호출로 메시지를 승인했을 때 대화가 멈춘 걸 볼 수 있어요.