사용자 개입

사용자 개입 (Human-in-the-Loop)

앞의 팀(Teams) 섹션에서 에이전트 팀을 만들고, 관찰하고, 제어하는 방법을 봤어요. 이번 섹션에서는 애플리케이션에서 팀과 상호작용하고, 팀에 인간의 피드백을 제공하는 방법에 초점을 맞출게요.

애플리케이션에서 팀과 상호작용하는 방법은 크게 두 가지예요:

팀 실행 중에 — BaseGroupChat.run 또는 run_stream 실행 동안 UserProxyAgent를 통해 피드백 제공
실행이 끝난 뒤 — 다음 run 또는 run_stream 호출에 입력으로 피드백 제공

이 두 방법을 모두 다룰게요.

웹·UI 프레임워크와의 통합을 바로 보려면 다음 링크를 참고하세요:

출처: 공식문서 - Human-in-the-Loop

실행 중 피드백 제공하기

UserProxyAgent는 사용자를 대신해 팀에 피드백을 제공하는 특별한 내장 에이전트예요. 사용자를 위한 프록시(proxy)로 동작하죠.

UserProxyAgent를 사용하려면 인스턴스를 만들고, 팀을 실행하기 전에 팀에 포함시키면 돼요. 그러면 팀이 사용자에게 피드백을 요청할 때가 되면 UserProxyAgent를 호출하기로 결정해요.

예를 들어 RoundRobinGroupChat 팀에서는 UserProxyAgent를 팀에 전달한 순서대로 호출해요. 반면 SelectorGroupChat 팀에서는 셀렉터 프롬프트나 셀렉터 함수가 UserProxyAgent를 언제 호출할지 결정해요.

다음 다이어그램은 팀 실행 중에 UserProxyAgent로 사용자 피드백을 받는 방법을 보여줘요.

human-in-the-loop-user-proxy

굵은 화살표는 팀 실행 중의 제어 흐름을 나타내요. 팀이 UserProxyAgent를 호출하면, 제어가 애플리케이션/사용자에게 넘어가고 피드백을 기다려요. 피드백이 제공되면 제어가 다시 팀으로 돌아와 실행을 계속해요.

실행 중 UserProxyAgent가 호출되면, 사용자가 피드백을 주거나 오류가 날 때까지 팀의 실행을 블로킹해요. 그러면 팀의 진행이 멈추고, 팀이 저장하거나 재개할 수 없는 불안정한 상태가 돼요.

이러한 블로킹 특성 때문에, 이 접근법은 즉각적인 피드백이 필요한 짧은 상호작용 — 예를 들어 버튼 클릭으로 승인/거부를 묻거나, 즉시 주의가 필요해 아니면 태스크가 실패하는 경고 — 에만 쓰는 걸 권장해요.

시(poetry) 생성 태스크에서 RoundRobinGroupChat 안에 UserProxyAgent를 쓰는 예시를 볼게요.

from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Create the agents.
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
assistant = AssistantAgent("assistant", model_client=model_client)
user_proxy = UserProxyAgent("user_proxy", input_func=input)  # Use input() to get user input from console.

# Create the termination condition which will end the conversation when the user says "APPROVE".
termination = TextMentionTermination("APPROVE")

# Create the team.
team = RoundRobinGroupChat([assistant, user_proxy], termination_condition=termination)

# Run the conversation and stream to the console.
stream = team.run_stream(task="Write a 4-line poem about the ocean.")
# Use asyncio.run(...) when running in a script.
await Console(stream)
await model_client.close()

콘솔 출력에서 팀이 user_proxy를 통해 생성된 시(poem)를 승인하도록 사용자 피드백을 요청하는 걸 볼 수 있어요.

UserProxyAgent에 **나만의 입력 함수(input function)**를 제공해서 피드백 과정을 맞춤 설정할 수도 있어요. 예를 들어 팀이 웹 서비스로 실행 중이라면, 웹 소켓 연결에서 메시지를 기다리는 커스텀 입력 함수를 쓸 수 있어요. 다음 코드는 FastAPI 웹 프레임워크를 쓸 때의 커스텀 입력 함수 예시예요.

@app.websocket("/ws/chat")
async def chat(websocket: WebSocket):
    await websocket.accept()

    async def _user_input(prompt: str, cancellation_token: CancellationToken | None) -> str:
        data = await websocket.receive_json() # Wait for user message from websocket.
        message = TextMessage.model_validate(data) # Assume user message is a TextMessage.
        return message.content
    
    # Create user proxy with custom input function
    # Run the team with the user proxy
    # ...

완전한 예시는 AgentChat FastAPI sample을 참고하세요. UserProxyAgent와의 ChainLit 통합은 AgentChat ChainLit sample을 보세요.

다음 실행에 피드백 제공하기

애플리케이션이나 사용자가 **대화형 루프(interactive loop)**에서 에이전트 팀과 상호작용하는 경우가 많아요. 팀이 종료 조건까지 실행되고, 애플리케이션이나 사용자가 피드백을 주면, 그 피드백을 바탕으로 팀이 다시 실행되는 식이죠.

이 접근법은 팀과 애플리케이션/사용자 사이의 **비동기 통신이 있는 지속 세션(persisted session)**에서 유용해요. 팀이 실행을 마치면 애플리케이션이 팀의 상태를 저장하고, 영구 저장소에 넣은 뒤, 피드백이 도착하면 팀을 재개하는 방식이에요.

팀의 상태를 저장·로드하는 방법은 Managing State를 참고하세요. 이 섹션은 피드백 메커니즘에 집중할게요.

이 접근법을 구현하는 방법은 두 가지가 있어요:

최대 턴 수를 설정해서, 팀이 지정한 턴 수 뒤에 항상 멈추게 하는 방법
TextMentionTermination이나 HandoffTermination 같은 종료 조건을 사용해서, 팀의 내부 상태를 바탕으로 팀이 스스로 멈추고 제어권을 돌려주게 하는 방법

두 방법을 함께 써서 원하는 동작을 만들 수도 있어요.

최대 턴 수 사용하기 (Using Max Turns)

이 방법은 최대 턴 수를 설정해서 사용자 입력을 위해 팀을 일시정지시켜요. 예를 들어 max_turns를 1로 설정하면 첫 번째 에이전트가 응답한 뒤 팀이 멈춰요. 챗봇처럼 지속적인 사용자 참여가 필요한 시나리오에서 특히 유용해요.

구현하려면 RoundRobinGroupChat 생성자에서 max_turns 파라미터를 설정하면 돼요.

team = RoundRobinGroupChat([...], max_turns=1)

팀이 멈추면 턴 수는 리셋돼요. 팀을 재개하면 다시 0부터 시작하죠. 하지만 팀의 내부 상태는 보존돼요. 예를 들어 RoundRobinGroupChat은 같은 대화 히스토리를 가진 채 리스트의 다음 에이전트부터 재개해요.

max_turn은 팀 클래스에 특화된 값으로, 현재 RoundRobinGroupChat, SelectorGroupChat, Swarm에서만 지원돼요. 종료 조건과 함께 쓰면 어느 조건이든 먼저 충족될 때 팀이 멈춰요.

시 생성 태스크에서 RoundRobinGroupChat 안에 max_turns(최대 1턴)를 쓰는 예시예요.

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

# Create the agents.
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
assistant = AssistantAgent("assistant", model_client=model_client)

# Create the team setting a maximum number of turns to 1.
team = RoundRobinGroupChat([assistant], max_turns=1)

task = "Write a 4-line poem about the ocean."
while True:
    # Run the conversation and stream to the console.
    stream = team.run_stream(task=task)
    # Use asyncio.run(...) when running in a script.
    await Console(stream)
    # Get the user response.
    task = input("Enter your feedback (type 'exit' to leave): ")
    if task.lower().strip() == "exit":
        break
await model_client.close()

한 에이전트가 응답한 뒤 팀이 즉시 멈추는 걸 볼 수 있어요.

종료 조건 사용하기 (Using Termination Conditions)

이전 섹션들에서 종료 조건의 예시를 여러 번 봤어요. 여기서는 에이전트가 HandoffMessage 메시지를 보낼 때 팀을 멈추는 HandoffTermination에 집중할게요.

handoff 설정이 있는 단일 AssistantAgent로 팀을 만들고, 에이전트가 태스크를 계속 처리할 관련 도구가 없어서 사용자의 추가 입력이 필요한 태스크로 팀을 실행해 볼게요.

AssistantAgent와 함께 쓰는 모델이 handoff 기능을 쓰려면 도구 호출(tool call)을 지원해야 해요.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.base import Handoff
from autogen_agentchat.conditions import HandoffTermination, TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Create an OpenAI model client.
model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    # api_key="sk-...", # Optional if you have an OPENAI_API_KEY env variable set.
)

# Create a lazy assistant agent that always hands off to the user.
lazy_agent = AssistantAgent(
    "lazy_assistant",
    model_client=model_client,
    handoffs=[Handoff(target="user", message="Transfer to user.")],
    system_message="If you cannot complete the task, transfer to user. Otherwise, when finished, respond with 'TERMINATE'.",
)

# Define a termination condition that checks for handoff messages.
handoff_termination = HandoffTermination(target="user")
# Define a termination condition that checks for a specific text mention.
text_termination = TextMentionTermination("TERMINATE")

# Create a single-agent team with the lazy assistant and both termination conditions.
lazy_agent_team = RoundRobinGroupChat([lazy_agent], termination_condition=handoff_termination | text_termination)

# Run the team and stream to the console.
task = "What is the weather in New York?"
await Console(lazy_agent_team.run_stream(task=task), output_stats=True)

handoff 메시지가 감지돼서 팀이 멈춘 걸 볼 수 있어요. 에이전트가 필요한 정보를 제공해서 팀을 계속 진행시켜 볼게요.

await Console(lazy_agent_team.run_stream(task="The weather in New York is sunny."))

사용자가 정보를 제공한 뒤 팀이 계속 진행되는 걸 볼 수 있어요.

Swarm 팀에서 HandoffTermination(target=user)을 쓸 때 팀을 재개하려면, task를 다음에 실행할 에이전트로 target을 설정한 HandoffMessage로 지정해야 해요. 자세한 내용은 Swarm을 참고하세요.

사용자 개입