In-Depth Analysis of Microsoft’s Agent Development Framework AutoGen 0.4: Introduction to AutoGen-Core

from autogen_core import AgentId, MessageContext, RoutedAgent, message_handler

class MyWorkerAgent(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("MyWorkerAgent")
        
    @message_handler
    async def handle_my_message(self, message: MyTextMessage, ctx: MessageContext) -> MyTextMessage:
        print(f"{self.id.key} received a message from {ctx.sender}: {message.content}\n")
        return MyTextMessage(content="OK, Got it!")

class MyManagerAgent(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("MyManagerAgent")
        self.worker_agent_id = AgentId('my_worker_agent', 'worker')

    @message_handler
    async def handle_my_message(self, message: MyTextMessage, ctx: MessageContext) -> None:
        print(f"{self.id.key} received message: {message.content}\n")
        print(f"{self.id.key} sending message to {self.worker_agent_id}...\n")
        response = await self.send_message(message, self.worker_agent_id)
        print(f"{self.id.key} received a message from {self.worker_agent_id}: {response.content}\n")

from autogen_core import SingleThreadedAgentRuntime
import asyncio

async def main():

    # Create runtime and register agent types
    runtime = SingleThreadedAgentRuntime()
    await MyManagerAgent.register(runtime, "my_manager_agent", lambda: MyManagerAgent())
    await MyWorkerAgent.register(runtime, "my_worker_agent", lambda: MyWorkerAgent())

    # Start runtime, send messages, close runtime
    runtime.start() 

    # Create agent_id, send message
    agent_id = AgentId("my_manager_agent", "manager")
    await runtime.send_message(MyTextMessage(content="Hello World!"), agent_id)

    # Close runtime
    await runtime.stop_when_idle() 

asyncio.run(main())

......
@dataclass
class Message:
    content: str

class MyAgent(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("A simple agent")
        self._system_messages = [SystemMessage(content="You are a helpful AI assistant. Please answer using Chinese.")]
        self._model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
        self._model_context = BufferedChatCompletionContext(buffer_size=5)

    @message_handler
    async def handle_user_message(self, message: Message, ctx: MessageContext) -> Message: 
        user_message = UserMessage(content=message.content, source="user")
        await self._model_context.add_message(user_message)
        response = await self._model_client.create(
            self._system_messages + (await self._model_context.get_messages()), cancellation_token=ctx.cancellation_token
        )
        await self._model_context.add_message(AssistantMessage(content=response.content, source=self.metadata["type"]))
        return Message(content=response.content)

...
    # Create agent_id, send message
    message = Message("What is the capital city of China?")
    response = await runtime.send_message(message, AgentId("my_agent", "default"))    print(response.content)