Multi-Agent Orchestration

1
from smallestai.atoms.agent.nodes import Node, OutputAgentNode
2
3
4
class RouterNode(Node):
5
    """Routes conversations to the appropriate specialist."""
6
    
7
    def __init__(self, sales_agent, support_agent, billing_agent):
8
        super().__init__(name="router")
9
        self.sales = sales_agent
10
        self.support = support_agent
11
        self.billing = billing_agent
12
13
    async def process_event(self, event):
14
        # Simple keyword-based routing
15
        if hasattr(event, "content"):
16
            content = event.content.lower()
17
            
18
            if any(word in content for word in ["buy", "price", "demo"]):
19
                event.metadata["route"] = "sales"
20
            elif any(word in content for word in ["broken", "help", "issue"]):
21
                event.metadata["route"] = "support"
22
            elif any(word in content for word in ["invoice", "payment", "bill"]):
23
                event.metadata["route"] = "billing"
24
            else:
25
                event.metadata["route"] = "support"  # Default
26
        
27
        await self.send_event(event)
28
29
30
class SalesAgent(OutputAgentNode):
31
    """Handles sales inquiries."""
32
    
33
    def __init__(self):
34
        super().__init__(name="sales-agent")
35
    
36
    async def process_event(self, event):
37
        if event.metadata.get("route") == "sales":
38
            await super().process_event(event)
39
        else:
40
            await self.send_event(event)
41
42
43
class SupportAgent(OutputAgentNode):
44
    """Handles support requests."""
45
    
46
    def __init__(self):
47
        super().__init__(name="support-agent")
48
    
49
    async def process_event(self, event):
50
        if event.metadata.get("route") == "support":
51
            await super().process_event(event)
52
        else:
53
            await self.send_event(event)

1
async def setup(session: AgentSession):
2
    # Create agents
3
    sales = SalesAgent()
4
    support = SupportAgent()
5
    billing = BillingAgent()
6
    router = RouterNode(sales, support, billing)
7
    
8
    # Add to session
9
    session.add_node(router)
10
    session.add_node(sales)
11
    session.add_node(support)
12
    session.add_node(billing)
13
    
14
    # Connect the graph
15
    session.add_edge(router, sales)
16
    session.add_edge(router, support)
17
    session.add_edge(router, billing)
18
    
19
    await session.start()
20
    await session.wait_until_complete()

1
class GreeterAgent(OutputAgentNode):
2
    """Greets users and hands off to scheduler."""
3
    
4
    async def generate_response(self):
5
        # Get last user message
6
        user_messages = [m for m in self.context.messages if m["role"] == "user"]
7
        user_message = user_messages[-1]["content"] if user_messages else ""
8
        
9
        if "appointment" in user_message.lower():
10
            # Trigger handoff
11
            await self.send_event(HandoffEvent(
12
                target="scheduler",
13
                reason="User wants to schedule",
14
                context={"user_name": self.user_name}
15
            ))
16
            yield "Let me transfer you to our scheduling assistant."
17
            return
18
        
19
        # Normal greeting
20
        yield "Hello! How can I help you today?"
21
22
23
class SchedulerAgent(OutputAgentNode):
24
    """Handles appointment scheduling."""
25
    
26
    async def process_event(self, event):
27
        if isinstance(event, HandoffEvent) and event.target == "scheduler":
28
            # Received handoff, take over
29
            self.handoff_context = event.context
30
            
31
        await super().process_event(event)

1
class HandoffEvent(SDKEvent):
2
    """Custom event for agent handoffs."""
3
    type: str = "internal.handoff"
4
    target: str  # Target agent name
5
    reason: str  # Why the handoff is happening
6
    context: dict = {}  # Shared data
7
8
9
class PrequalAgent(OutputAgentNode):
10
    def __init__(self):
11
        super().__init__(name="prequal-agent")
12
        self.collected_data = {}
13
14
    async def generate_response(self):
15
        # Check if we have all needed info
16
        if self._qualification_complete():
17
            await self.send_event(HandoffEvent(
18
                target="closer",
19
                reason="Qualification complete",
20
                context={
21
                    "lead_name": self.collected_data["name"],
22
                    "company": self.collected_data["company"],
23
                    "budget": self.collected_data["budget"],
24
                    "timeline": self.collected_data["timeline"]
25
                }
26
            ))
27
            yield "Great! Let me connect you with our sales team."
28
            return
29
        
30
        # Continue qualification...

1
from smallestai.atoms.agent.nodes import Node
2
from smallestai.atoms.agent.clients.openai import OpenAIClient
3
4
class SmartRouterNode(Node):
5
    def __init__(self):
6
        super().__init__(name="smart-router")
7
        self.llm = OpenAIClient(model="gpt-4o-mini")
8
9
    async def process_event(self, event):
10
        if not hasattr(event, "content"):
11
            await self.send_event(event)
12
            return
13
        
14
        # Ask LLM to classify
15
        response = await self.llm.chat(
16
            messages=[
17
                {
18
                    "role": "system",
19
                    "content": """Classify the user intent. Respond with one word:
20
- SALES: pricing, demos, purchasing
21
- SUPPORT: issues, bugs, help
22
- BILLING: invoices, payments, refunds
23
- GENERAL: everything else"""
24
                },
25
                {"role": "user", "content": event.content}
26
            ]
27
        )
28
        
29
        intent = response.content.strip().upper()
30
        event.metadata["route"] = intent.lower()
31
        
32
        await self.send_event(event)

1
class FallbackAgent(OutputAgentNode):
2
    """Handles requests that don't match any specialist."""
3
    
4
    async def process_event(self, event):
5
        route = event.metadata.get("route")
6
        
7
        # Only handle if no other agent claimed it
8
        if route == "general" or route is None:
9
            await super().process_event(event)
10
        else:
11
            await self.send_event(event)

1
class LoggingRouterNode(Node):
2
    async def process_event(self, event):
3
        route = self._determine_route(event)
4
        event.metadata["route"] = route
5
        
6
        logger.info(
7
            f"Routing to {route}",
8
            extra={
9
                "event_type": event.type,
10
                "route": route,
11
                "session_id": self.session_id
12
            }
13
        )
14
        
15
        await self.send_event(event)