Function test_addressing

tests/main/test_msg_routing.py:42–97

Test that an agent is able to address another agent in a message.

(
    test_settings: Settings, prefix: str, address: str, x: int, answer: int
)

)
@pytest.mark.parametrize("x,answer", [(5, 25)])
def test_addressing(
    test_settings: Settings, prefix: str, address: str, x: int, answer: int
):
    """Test that an agent is able to address another agent in a message."""
    set_global(test_settings)

    class BobAgent(ChatAgent):
        def llm_response(
            self, message: Optional[str | ChatDocument] = None
        ) -> Optional[ChatDocument]:
            if (
                isinstance(message, ChatDocument)
                and message.metadata.sender_name == "Alice"
            ):
                return self.create_llm_response(DONE + " " + message.content)

            addr = AT if AT in address else SEND_TO
            # throw in some distracting addresses, to test that
            # only the last one is picked up
            return self.create_llm_response(
                f"Ok {addr}all here {addr}Junk is my question: {address} {x}"
            )

    class AliceAgent(ChatAgent):
        def llm_response(
            self, message: Optional[str | ChatDocument] = None
        ) -> Optional[ChatDocument]:
            # message.content will either be just an an int-string "5"
            # (if prefix != "") or Bob's entire msg otherwise (and hence not an int)
            try:
                y = int(message.content.strip())
            except ValueError:
                return None
            answer = y * y
            return self.create_llm_response(f"{DONE} {answer}")

    bob_config = ChatAgentConfig(name="Bob")

    bob = BobAgent(bob_config)
    bob_task = Task(
        bob,
        interactive=False,
        config=TaskConfig(addressing_prefix=prefix),
    )

    alice_config = ChatAgentConfig(name="Alice")
    alice = AliceAgent(alice_config)
    alice_task = Task(alice, interactive=False)

    bob_task.add_sub_task(alice_task)

    result = bob_task.run()
    if prefix == "" and AT in address:
        assert result is None
    else:
        assert answer == int(result.content.strip())