AutoGen 0.4 HumanInLoop Support

[asheeshgarg]

What feature would you like to be added?

AutoGen 0.4 HumanInLoop Support

Why is this needed?

To support interruption for with human in loop

[ekzhu]

I converted this into a discussion due to the open-ended question.

Can you be more specific about your requirement? There are many discussions on this topic, and you can also read about the documentation.

[asheeshgarg]

Thanks @ekzhu is there a example for this in 0.4 that I can follow I don't see this pattern. Thanks for your guidance

[victordibia]

@asheeshgarg

In AgentChat (assuming you are exploring the AgentChat api), you can create an agent that always requests human input when a message is received. Documentation on building this is here.

import asyncio


class UserProxyAgent(BaseChatAgent):
    def __init__(self, name: str) -> None:
        super().__init__(name, "A human user.")

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

    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        user_input = await asyncio.get_event_loop().run_in_executor(None, input, "Enter your response: ")
        return Response(chat_message=TextMessage(content=user_input, source=self.name))

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


async def run_user_proxy_agent() -> None:
    user_proxy_agent = UserProxyAgent(name="user_proxy_agent")
    response = await user_proxy_agent.on_messages([], CancellationToken())
    print(response.chat_message.content)


# Use asyncio.run(run_user_proxy_agent()) when running in a script.
await run_user_proxy_agent()

You can explore something similar using the core api (its a bit more involved), but see this guide.

[asheeshgarg]

@victordibia thanks for the pointers will try this out