Class AutomatedUserProxyAgent

examples/ag2/ag2_async_agent.py:54–84

# Define a custom UserProxyAgent that simulates automated user responses
class AutomatedUserProxyAgent(UserProxyAgent):
    def __init__(self, name: str, **kwargs):
        super().__init__(name, **kwargs)
        self.response_count = 0
        self.predefined_responses = [
            "Yes, please generate interview questions for these topics.",
            "The questions look good. Can you make them more specific to senior-level positions?",
            "Perfect! These questions are exactly what we need. Thank you!",
        ]

    async def a_get_human_input(self, prompt: str) -> str:
        # Simulate async processing before responding
        await simulate_async_processing(f"Processing user input #{self.response_count + 1}")

        if self.response_count < len(self.predefined_responses):
            response = self.predefined_responses[self.response_count]
            self.response_count += 1
            print(f"👤 User: {response}")
            return response
        else:
            print("👤 User: TERMINATE")
            return "TERMINATE"

    async def a_receive(
        self,
        message: Union[Dict, str],
        sender,
        request_reply: Optional[bool] = None,
        silent: Optional[bool] = False,
    ):
        await super().a_receive(message, sender, request_reply, silent)


# Define an AssistantAgent that simulates async processing before responding