Autogen 0.4.5 | Human in the loop with UI - Advise on architecture

[skye0402]

I'm letting a user interact through telegram with Autogen 0.4.5. So far so good. Now I want to add the human-in-the-loop pattern.

In principle there are 2 reasons why the agent team (I use Swarm or SelectorGroupChat) would hand back to the user:

1. The task is completed with a result (this works)
2. The agent has a question to the user

For that I implemented a UserProxyAgent:

        self.user_proxy = UserProxyAgent(
            name="user_proxy",
            description="A human user of our application. To be contacted or queried for information and feedback as needed.",
            input_func=input_by_user
        )     

When it gets called it calls the input_by_user and that would collect the user feedback.

My questions:

1. Of course I want to know what the question is that the agent has! How can I capture this? I believe it should be somewhere in the agent state but not sure how to route it to the input_func.
2. Sending the question or request for user feedback to the user should be possible (once I have point ① clarified). However, the answer of the user is decoupled as it's telegram messenger. I need to feed it back to the agent, but how? I think I can write logic that would state "this user response is related to the agent's question TRUE/FALSE" but even then, how to feed it back in? We're not speaking of a console input here.

I believe this should be a common challenge as soon as a UI is involved? Any hints?

[ekzhu]

For Swarm group chat, you can use HandoffTermination to terminate the run and restart it with your user input as the next task. You can read about it in this example: https://microsoft.github.io/autogen/stable/user-guide/agentchat-user-guide/swarm.html#customer-support-example.

You can also use HandoffTermination for SelectorGroupChat, it also requires the agent to be able to use the handoff feature: https://microsoft.github.io/autogen/stable/user-guide/agentchat-user-guide/tutorial/human-in-the-loop.html#using-termination-conditions.

The use of HandoffTermination means you don't use UserProxyAgent in a group chat, you stop the group chat, give the control back to the applicaiton, and restart it with a new message. The state of the group chat is maintained in memory, or persisted using save_state and load_state.

---

If for some reason, you must use UserProxyAgent with a UI (e.g., your model doesn't support function calling), then your application must be asynchronous. In the input callback of UserProxyAgent, you need to send a message to the frontend, wait on a specific asyncio.Event, when the user message arrives back in a separate coroutine or your fast api handler, you set the asyncio.Event, and obtain the message to be used as response.

If the above sound too complicated to you -- I agree, and I think there should be a more elegant way to address this scenario for SelectorGroupChat and RoundRobinGroupChat when function calling model is not available.

One idea is to have the group chat just terminate when user proxy is selected as the speaker. It could be done by having the group chat exposes a SpeakerSelectedEvent and add a new termination condition, NextSpeakerMatchTermination, that terminates the group chat when a SpeakerSelectedEvent emitts with a matching spaker.

What do you think?

cc @husseinmozannar @victordibia @jackgerrits , consider this instead of using user proxy's input function, which is a bit foot-gun in terms of UI intergration.

[ekzhu]

Similar discussion with an example of using HandoffTermination + TextTermination #4617 (reply in thread)

[victordibia]

I think there are two high level questions here:

• 1. How do I capture the message sent by the previous agent
• 2. How do I make a call to an external entity and return the results while the team is still running .

Getting the Last Message Sent (Previous Agent)

What we likely need is a setup where the message that the user_proxy receives (sent by the team/orchestrator) contains some history of the chat so far. Currently, it does receive the last agent message sent (@ekzhu correct me if this is wrong). However, it does not necessarily pass that to the input_func. Take a look at the UserProxy implementation here.

So how do you solve this? Create a CustomAgent that inspects the last message (which should contain the question from your agent) and then send it to your input function.

A rough sketch below

import time
from typing import AsyncGenerator, List, Sequence, Tuple,Callable

from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.base import Response
from autogen_agentchat.messages import AgentEvent, ChatMessage, TextMessage
from autogen_core import CancellationToken
from autogen_agentchat.ui import Console

async def telegram_input(messages: Sequence[ChatMessage]):
    # this MUST be an async function. 
    telegram_response = f"Telegram response: my feedback is that this is a great idea! \n This is in response to agent message {messages[0].content[:40]}  ..."
    return telegram_response


class CustomAgent(BaseChatAgent):
    def __init__(self, name: str, input_func: Callable):
        super().__init__(name, "A simple agent that counts down.")
        self._input_func = input_func

    @property
    def produced_message_types(self) -> Sequence[type[ChatMessage]]:
        return (TextMessage,)

    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        # Calls the on_messages_stream.
 
        response: Response | None = None
        async for message in self.on_messages_stream(messages, cancellation_token):
            if isinstance(message, Response):
                response = message
        assert response is not None
        return response

    async def on_messages_stream(
        self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken
    ) -> AsyncGenerator[AgentEvent | ChatMessage | Response, None]:
        inner_messages: List[AgentEvent | ChatMessage] = []
        # print("# Messages",len(messages), messages)

        # status update
        status_update_message = TextMessage(content="Getting user feedback", source=self.name)
        yield status_update_message
        inner_messages.append(status_update_message)
        user_input = await self._input_func(messages) # call async function
        print("User input", user_input)
        response_message = TextMessage(content=user_input, source=self.name)
        yield response_message

        time.sleep(2)

        # All this agent does is respond with sounds good for now.
        yield Response(chat_message=TextMessage(content="Sound good, but make it more interesting and movie like", source=self.name), inner_messages=inner_messages)

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        pass

reflector_agent = CustomAgent(name="reflector_agent", input_func=telegram_input) 

writer_agent = AssistantAgent(
    name="writer_agent",
    model_client=OpenAIChatCompletionClient(
        model="gpt-4o-mini", 
    ), 
)

agent_team = RoundRobinGroupChat([writer_agent, reflector_agent], max_turns=5)
stream = agent_team.run_stream(task="Write a 6 line poem about the Andes mountains")
await Console(stream)

Interacting with External Entities During Run

One of the benefits of the updated AutoGen codebase is that it is async. Like @ekzhu , to interact with an external entity, that section of the code must be async. In the autogenstudio code base , when a run is triggered in the UI, a socket connection is opened, and whenever await input_func is called, a message is sent to the UI, and a waits for a response. By wait for a response, .. what I meant is that the input_func monitors a queue which the socket connection writes responses to.

ags_update.mp4

Its complex, but your telegram input function will need to do something similar.

• Spin up a process that is connected to telegram
• Writes messages from telegram to a queue
• your telegram_input_func can send a message to the user on telegram
• waits for a response by reading from the queue

[ekzhu]

Do you think it would be easier to just terminate the group chat, let the application deal with the user input, and then restart the group chat? I feel it would be much easier and developers like @skye0402 won't need to learn too much about async programming.

[victordibia]

That could work too.