Multi-Agent Coordination

1
from pydantic import BaseModel
2
3
4
# Custom events for inter-agent communication
5
class LeadIdentified(BaseModel):
6
    company_name: str
7
    contact_info: dict
8
    interest_level: str
9
10
11
class CompanyResearch(BaseModel):
12
    company_name: str
13
    industry: str
14
    revenue: str
15
    decision_makers: list
16
17
18
async def setup_multi_agent_system(system: VoiceAgentSystem):
19
    # Main sales agent (speaking)
20
    sales_node = SalesNode(
21
        system_prompt="You are a helpful sales representative."
22
    )
23
    sales_bridge = Bridge(sales_node)
24
    system.with_speaking_node(sales_node, bridge=sales_bridge)
25
26
    # Background lead extraction agent
27
    leads_node = LeadExtractionNode()
28
    leads_bridge = Bridge(leads_node)
29
    system.with_node(leads_node, leads_bridge)
30
31
    # Background research agent
32
    research_node = CompanyResearchNode()
33
    research_bridge = Bridge(research_node)
34
    system.with_node(research_node, research_bridge)
35
36
    # Route user input to all agents
37
    sales_bridge.on(UserTranscriptionReceived).map(
38
        sales_node.add_event
39
    )
40
    leads_bridge.on(UserTranscriptionReceived).map(
41
        leads_node.add_event
42
    )
43
44
    # Sales agent responds to user
45
    (
46
        sales_bridge.on(UserStoppedSpeaking)
47
        .stream(sales_node.generate)
48
        .broadcast()
49
    )
50
51
    # Background lead analysis
52
    (
53
        leads_bridge.on(UserStoppedSpeaking)
54
        .stream(leads_node.generate)
55
        .broadcast()
56
    )
57
58
    # Research agent processes identified leads
59
    research_bridge.on(LeadIdentified).map(research_node.add_event)
60
    (
61
        research_bridge.on(LeadIdentified)
62
        .stream(research_node.generate)
63
        .broadcast()
64
    )
65
66
    # Sales agent receives research insights
67
    sales_bridge.on(CompanyResearch).map(sales_node.add_context)

1
# Transfer conversation to technical specialist
2
class TechnicalSupportRequest(BaseModel):
3
    issue_type: str
4
    complexity: str
5
6
7
# Sales agent identifies need for handoff
8
async def process_context(self, context):
9
    if technical_question_detected:
10
        yield AgentHandoff(
11
            target_agent="technical_support",
12
            reason="Customer has technical questions about implementation",
13
        )
14
15
16
# Set up handoff routing
17
tech_bridge.on(TechnicalSupportRequest).map(
18
    lambda _: AgentHandoff(target_agent="technical_support")
19
).broadcast()