Routed Agent

The engine/agent module in AgentOpera provides a core set of agent implementations that serve as the foundation for building agent-based systems. These components enable asynchronous messaging, type-based message routing, and managing the lifecycle of agents within the engine runtime.

Agent Identity

AgentId

An AgentId uniquely identifies an agent instance within an agent engine, including distributed engines. It functions as the "address" of the agent instance for receiving messages.

from agentopera.engine.types.agent import AgentId, AgentType

# Create an AgentId
agent_id = AgentId("assistant", "user_session_123")
# or
agent_type = AgentType("assistant")
agent_id = AgentId(agent_type, "user_session_123")

# Convert to string representation
agent_id_str = str(agent_id)  # "assistant/user_session_123" 

# Create from string
agent_id = AgentId.from_str("assistant/user_session_123")

An AgentId consists of two components:

1. Agent Type: Identifies the category of agent and associates it with a specific factory function

2. Agent Key: Instance identifier for the given agent type

An AgentId follows the string format: Agent_Type/Agent_Key. Both types and keys must contain only alphanumeric characters (a-z, 0-9) or underscores (_) and cannot start with a number.

AgentType

AgentType represents a category of agents that share the same implementation or behavior patterns.

from agentopera.engine.types.agent import AgentType

agent_type = AgentType("assistant")

Agent types are typically defined directly in application code, while agent keys are often generated dynamically based on messages delivered to agents.

RoutedAgent

RoutedAgent provides type-based message routing through decorators:

from agentopera.engine.agent import RoutedAgent, message_handler
from agentopera.engine.types.msg_context import MessageContext
from dataclasses import dataclass

@dataclass
class GreetingMessage:
    content: str

class MyGreetingAgent(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("Greeting agent that handles greeting messages")
        
    @message_handler
    async def handle_greeting(self, message: GreetingMessage, ctx: MessageContext) -> str:
        return f"Hello! You said: {message.content}"

RoutedAgent enables you to:

• Define message handlers with the @message_handler decorator

• Route messages to appropriate handlers based on message type

The message handling in RoutedAgent follows these steps:

1. When a message arrives, the agent examines the message type

2. The agent looks up registered handlers that can process this message type

3. The first matching handler is called with the message and context

Agent Lifecycle

Agents in AgentOpera follow a managed lifecycle:

1. Registration: An agent type is registered with the engine using the register method

2. Instantiation: The engine creates an agent instance when a message is delivered to that agent ID

3. Message Handling: The agent processes messages through its message handlers

Integration with Chat Agents

The engine module agents are designed to be used with chatflow agents. Since chatflow agents like AssistantAgent are created directly by the application and not by the engine runtime, they need to be wrapped with an engine-managed agent like RoutedAgent:

from agentopera.engine.agent import RoutedAgent, message_handler
from agentopera.engine.types.msg_context import MessageContext
from agentopera.chatflow.agent import AssistantAgent
from agentopera.chatflow.messages import TextMessage

class WorkerAgent(RoutedAgent):
    def __init__(self, name: str) -> None:
        super().__init__(f"Wrapper for assistant agent {name}")
        # Create the chatflow agent
        self._assistant = AssistantAgent(name, model_client=some_model_client)
        
    @message_handler
    async def handle_text_message(self, message: TextMessage, ctx: MessageContext) -> TextMessage:
        # Delegate to chatflow agent
        response = await self._assistant.on_messages([message], ctx.cancellation_token)
        return response.chat_message

With this pattern, the engine can manage the wrapper agent, which delegates message processing to the chatflow agent.