Implementation:Microsoft Autogen UserProxyAgent

Key	Value
id	Microsoft_Autogen_UserProxyAgent
source	Microsoft_Autogen
category	Agent

Overview

Description

The UserProxyAgent is an agent implementation that represents a human user in a chat system by providing a custom input function. This agent acts as a proxy for human input, allowing seamless integration of human-in-the-loop interactions within multi-agent conversations.

The agent can handle both synchronous and asynchronous input functions, supports cancellation tokens for timeout scenarios, and can respond to handoff messages from other agents. It implements the ChatAgent protocol and extends BaseChatAgent to participate in team-based conversations.

Key capabilities:

• Human Input Integration: Accepts custom input functions to capture user responses
• Handoff Support: Handles HandoffMessage types to enable agent-to-human transfers
• Cancellable Operations: Supports CancellationToken for timeout and cancellation scenarios
• Event Streaming: Emits UserInputRequestedEvent before requesting input
• Context Management: Provides InputRequestContext for tracking input requests
• State Management: Implements save/load state and reset operations

Usage

The UserProxyAgent is used when human input is required in a multi-agent conversation. It is particularly useful in scenarios where:

• Human approval or decision-making is needed
• User feedback is required during agent execution
• Interactive debugging or monitoring of agent behavior is desired

Important Considerations:

• Using UserProxyAgent puts a running team in a blocked state until the user responds
• Input functions should handle timeouts and cancellation appropriately
• For slow human responses, use termination conditions like HandoffTermination or SourceMatchTermination
• The team state can be saved and restored when the user responds

Code Reference

Source Location

• Repository: https://github.com/microsoft/autogen
• File Path: /tmp/kapso_repo_2mr4n2g4/python/packages/autogen-agentchat/src/autogen_agentchat/agents/_user_proxy_agent.py
• Lines: 1-250

Signature

class UserProxyAgent(BaseChatAgent, Component[UserProxyAgentConfig]):
    def __init__(
        self,
        name: str,
        *,
        description: str = "A human user",
        input_func: Optional[InputFuncType] = None,
    ) -> None:
        ...

Import

from autogen_agentchat.agents import UserProxyAgent

I/O Contract

Inputs

Parameter	Type	Required	Description
name	str	Yes	The name of the agent (must be unique within team)
description	str	No	A description of the agent (default: "A human user")
input_func	Optional[InputFuncType]	No	A function that takes a prompt and returns user input. Can be sync (Callable[[str], str]) or async (Callable[[str, Optional[CancellationToken]], Awaitable[str]]). Defaults to cancellable_input.

Outputs

Return Type	Description
Response	Contains either TextMessage or HandoffMessage based on context, with user input as content
AsyncGenerator	When using on_messages_stream, yields UserInputRequestedEvent, then Response

Usage Examples

Simple Usage

import asyncio
from autogen_core import CancellationToken
from autogen_agentchat.agents import UserProxyAgent
from autogen_agentchat.messages import TextMessage


async def simple_user_agent():
    agent = UserProxyAgent("user_proxy")
    response = await asyncio.create_task(
        agent.on_messages(
            [TextMessage(content="What is your name? ", source="user")],
            cancellation_token=CancellationToken(),
        )
    )
    assert isinstance(response.chat_message, TextMessage)
    print(f"Your name is {response.chat_message.content}")

Cancellable Usage

import asyncio
from typing import Any
from autogen_core import CancellationToken
from autogen_agentchat.agents import UserProxyAgent
from autogen_agentchat.messages import TextMessage


token = CancellationToken()
agent = UserProxyAgent("user_proxy")


async def timeout(delay: float):
    await asyncio.sleep(delay)


def cancellation_callback(task: asyncio.Task[Any]):
    token.cancel()


async def cancellable_user_agent():
    try:
        timeout_task = asyncio.create_task(timeout(3))
        timeout_task.add_done_callback(cancellation_callback)
        agent_task = asyncio.create_task(
            agent.on_messages(
                [TextMessage(content="What is your name? ", source="user")],
                cancellation_token=token,
            )
        )
        response = await agent_task
        assert isinstance(response.chat_message, TextMessage)
        print(f"Your name is {response.chat_message.content}")
    except Exception as e:
        print(f"Exception: {e}")

Custom Input Function

from autogen_core import CancellationToken
from autogen_agentchat.agents import UserProxyAgent


async def custom_input_func(prompt: str, cancellation_token: CancellationToken) -> str:
    # Custom logic to get user input (e.g., from web UI, queue, etc.)
    print(f"Prompt: {prompt}")
    return "User response from custom source"


agent = UserProxyAgent(
    "custom_user_proxy",
    description="User proxy with custom input",
    input_func=custom_input_func
)

Using Input Request Context

from autogen_agentchat.agents import UserProxyAgent


async def tracked_input_func(prompt: str, cancellation_token: CancellationToken) -> str:
    # Get the request ID from context
    request_id = UserProxyAgent.InputRequestContext.request_id()
    print(f"Processing input request: {request_id}")
    return input(prompt)


agent = UserProxyAgent(
    "tracked_user_proxy",
    input_func=tracked_input_func
)

Related Pages

• Microsoft_Autogen_ChatAgent_Protocol - The base protocol implemented by UserProxyAgent
• Microsoft_Autogen_AssistantAgent - Another agent implementation that uses LLM for responses
• Microsoft_Autogen_HandoffMessage - Message type used for agent-to-agent handoffs
• Microsoft_Autogen_TextMessage - Standard message type for text communication
• Microsoft_Autogen_CancellationToken - Token used for cancelling operations
• Microsoft_Autogen_BaseChatAgent - Base class extended by UserProxyAgent
• Microsoft_Autogen_UserInputRequestedEvent - Event emitted before requesting user input