Building multi-agent architectures with LiveKit agents

1
from livekit.agents import Agent, function_tool, RunContext
2

3
class IntakeAgent(Agent):
4
    def __init__(self):
5
        super().__init__(
6
            instructions="You gather initial information from callers."
7
        )
8

9
    @function_tool()
10
    async def transfer_to_billing(self, context: RunContext) -> tuple[Agent, str]:
11
        """Transfer the caller to the billing specialist."""
12
        return BillingAgent(), "Transferring you to our billing department."
13

14
    @function_tool()
15
    async def transfer_to_scheduling(self, context: RunContext) -> tuple[Agent, str]:
16
        """Transfer the caller to scheduling."""
17
        return SchedulingAgent(), "Let me connect you with scheduling."

1
from dataclasses import dataclass, field
2
from typing import Optional, List
3
from livekit.agents import Agent, AgentSession, function_tool, RunContext
4

5
@dataclass
6
class SessionState:
7
    # Deterministic memory - always reliable
8
    caller_name: Optional[str] = None
9
    date_of_birth: Optional[str] = None
10
    phone_number: Optional[str] = None
11
    account_id: Optional[str] = None
12
    intent: Optional[str] = None
13
    authenticated: bool = False
14
    collected_data: dict = field(default_factory=dict)
15

16
    # Canonical conversation history - never truncated
17
    full_history: List[dict] = field(default_factory=list)
18

19
class IntakeAgent(Agent):
20
    @function_tool()
21
    async def save_caller_identity(
22
        self,
23
        context: RunContext,
24
        name: str,
25
        date_of_birth: str,
26
        phone: str
27
    ):
28
        """Save verified caller identity information."""
29
        state: SessionState = context.session.userdata
30
        state.caller_name = name
31
        state.date_of_birth = date_of_birth
32
        state.phone_number = phone
33
        return f"Thank you {name}, I've saved your information."

1
class SchedulingAgent(Agent):
2
    async def check_context(self, context: RunContext):
3
        # Access full conversation history from the session
4
        full_history = context.session.history
5

6
        # Access current agent's chat context
7
        current_ctx = self.chat_ctx

1
from livekit.agents import ChatContext, ChatMessage
2

3
@dataclass
4
class SessionState:
5
    # ... other fields ...
6
    full_history: List[ChatMessage] = field(default_factory=list)
7

8
class ConversationManager:
9
    """Custom helper for managing chat context with safe truncation.
10
    This is not part of the SDK - implement it in your application as needed.
11
    """
12

13
    @staticmethod
14
    def create_truncated_context(
15
        session: AgentSession,
16
        max_messages: int = 10,
17
        include_summary: bool = True
18
    ) -> ChatContext:
19
        """Create a truncated context for an agent while preserving key information."""
20
        state: SessionState = session.userdata
21

22
        ctx = ChatContext()
23

24
        if include_summary and len(session.history) > max_messages:
25
            # Add a summary message with key session state
26
            summary = f"""Previous conversation summary:
27
- Caller: {state.caller_name or 'Unknown'}
28
- Account: {state.account_id or 'Not identified'}
29
- Intent: {state.intent or 'Not determined'}
30
- Key data collected: {state.collected_data}
31
"""
32
            ctx.add_message(role="system", content=summary)
33

34
        # Add recent messages from session history
35
        recent_messages = list(session.history)[-max_messages:]
36
        for msg in recent_messages:
37
            ctx.add_message(role=msg.role, content=msg.content)
38

39
        return ctx

1
class SchedulingAgent(Agent):
2
    def __init__(self):
3
        super().__init__(
4
            instructions="You help schedule appointments."
5
        )
6

7
    @function_tool()
8
    async def transfer_to_followup(self, context: RunContext) -> tuple[Agent, str]:
9
        """Transfer to follow-up agent with truncated context including summary."""
10
        # Create truncated context with summary for the next agent
11
        truncated_ctx = ConversationManager.create_truncated_context(
12
            context.session,
13
            max_messages=10,
14
            include_summary=True
15
        )
16
        followup_agent = FollowUpAgent()
17
        followup_agent.chat_ctx = truncated_ctx
18
        return followup_agent, "Let me connect you with our follow-up team."

1
from livekit.agents import AgentTask
2
from dataclasses import dataclass
3

4
@dataclass
5
class NameResult:
6
    first_name: str
7
    last_name: str
8

9
class CollectNameTask(AgentTask[NameResult]):
10
    def __init__(self):
11
        super().__init__(
12
            instructions="Ask for the caller's first and last name."
13
        )
14

15
    @function_tool
16
    async def set_name(self, context: RunContext, first_name: str, last_name: str):
17
        """Save the caller's name."""
18
        self.complete(NameResult(first_name=first_name, last_name=last_name))
19
        return "Got it, thank you."
20

21
@dataclass
22
class DOBResult:
23
    date_of_birth: str
24

25
class CollectDOBTask(AgentTask[DOBResult]):
26
    def __init__(self):
27
        super().__init__(
28
            instructions="Ask for the caller's date of birth for verification."
29
        )
30

31
    @function_tool
32
    async def set_dob(self, context: RunContext, date_of_birth: str):
33
        """Save the date of birth."""
34
        self.complete(DOBResult(date_of_birth=date_of_birth))
35
        return "Thank you for confirming."
36

37
@dataclass
38
class PhoneResult:
39
    phone_number: str
40

41
class CollectPhoneTask(AgentTask[PhoneResult]):
42
    def __init__(self):
43
        super().__init__(
44
            instructions="Ask for a callback phone number."
45
        )
46

47
    @function_tool
48
    async def set_phone(self, context: RunContext, phone_number: str):
49
        """Save the phone number."""
50
        self.complete(PhoneResult(phone_number=phone_number))
51
        return "I've saved your number."

1
from livekit.agents import TaskGroup
2

3
class AuthenticationAgent(Agent):
4
    async def run_auth_flow(self, context: RunContext):
5
        state: SessionState = context.session.userdata
6

7
        # Create task group with chat context
8
        task_group = TaskGroup(chat_ctx=self.chat_ctx)
9

10
        # Add tasks using lambdas - only for missing information
11
        if not state.caller_name:
12
            task_group.add(
13
                lambda: CollectNameTask(),
14
                id="collect_name",
15
                description="Collect caller's name"
16
            )
17

18
        if not state.date_of_birth:
19
            task_group.add(
20
                lambda: CollectDOBTask(),
21
                id="collect_dob",
22
                description="Collect date of birth"
23
            )
24

25
        if not state.phone_number:
26
            task_group.add(
27
                lambda: CollectPhoneTask(),
28
                id="collect_phone",
29
                description="Collect phone number"
30
            )
31

32
        # Await the task group directly
33
        results = await task_group
34

35
        # Update session state with collected data
36
        for result in results:
37
            if isinstance(result, NameResult):
38
                state.caller_name = f"{result.first_name} {result.last_name}"
39
            elif isinstance(result, DOBResult):
40
                state.date_of_birth = result.date_of_birth
41
            elif isinstance(result, PhoneResult):
42
                state.phone_number = result.phone_number

1
class ContextAwareAgent(Agent):
2
    async def check_previous_context(self, context: RunContext):
3
        # Access full conversation history from session
4
        full_history = context.session.history
5

6
        # Access current agent's chat context
7
        current_ctx = self.chat_ctx
8

9
        # Check if information was already mentioned
10
        for msg in full_history:
11
            if "name" in msg.content.lower():
12
                # Name was discussed earlier
13
                pass

1
from dataclasses import dataclass
2
from typing import List, Optional, Dict, Any
3

4
@dataclass
5
class WarmTransferConfig:
6
    default_number: str
7
    medical_number: Optional[str] = None
8
    cosmetic_number: Optional[str] = None
9
    error_fallback_number: Optional[str] = None
10

11
@dataclass
12
class QuestionConfig:
13
    id: str
14
    prompt: str
15
    response_type: str  # "text", "yes_no", "multiple_choice"
16
    options: Optional[List[str]] = None
17
    required: bool = True
18

19
@dataclass
20
class IntentFlowConfig:
21
    intent_name: str
22
    clarifying_question: Optional[str] = None  # e.g., "Is this cosmetic or medical?"
23
    clarifying_options: Optional[List[str]] = None
24
    questions_by_option: Dict[str, List[QuestionConfig]] = field(default_factory=dict)
25
    warm_transfer_by_option: Dict[str, str] = field(default_factory=dict)
26

27
@dataclass
28
class ClientConfig:
29
    client_id: str
30
    client_name: str
31
    warm_transfer: WarmTransferConfig
32
    intent_flows: Dict[str, IntentFlowConfig] = field(default_factory=dict)
33
    custom_greeting: Optional[str] = None
34
    timezone: str = "UTC"

1
import json
2
from pathlib import Path
3

4
class ConfigLoader:
5
    _configs: Dict[str, ClientConfig] = {}
6

7
    @classmethod
8
    async def load_client_config(cls, client_id: str) -> ClientConfig:
9
        """Load client config from database, file, or cache."""
10
        if client_id in cls._configs:
11
            return cls._configs[client_id]
12

13
        # Example: Load from JSON file (replace with your data source)
14
        config_path = Path(f"configs/clients/{client_id}.json")
15
        if config_path.exists():
16
            data = json.loads(config_path.read_text())
17
            config = ClientConfig(**data)
18
            cls._configs[client_id] = config
19
            return config
20

21
        # Return default config if not found
22
        return ClientConfig(
23
            client_id=client_id,
24
            client_name="Default",
25
            warm_transfer=WarmTransferConfig(default_number="+15551234567")
26
        )
27

28
# In your entrypoint
29
async def entrypoint(ctx: JobContext):
30
    # Extract client ID from room metadata or participant attributes
31
    client_id = ctx.room.metadata.get("client_id", "default")
32

33
    # Load configuration before starting the session
34
    client_config = await ConfigLoader.load_client_config(client_id)
35

36
    # Initialize session state with config
37
    session_state = SessionState()
38
    session_state.client_config = client_config
39

40
    session = AgentSession()
41
    session.userdata = session_state
42

43
    # Start with config-aware agent
44
    await session.start(
45
        agent=create_intake_agent(client_config),
46
        room=ctx.room
47
    )

1
@dataclass
2
class ClarificationResult:
3
    option: str
4

5
class ClarificationTask(AgentTask[ClarificationResult]):
6
    def __init__(self, question: str, options: List[str]):
7
        super().__init__(
8
            instructions=f"""Ask the caller: "{question}"
9
Valid responses are: {', '.join(options)}
10
Determine which option they want."""
11
        )
12
        self.options = options
13

14
    @function_tool
15
    async def select_option(self, context: RunContext, option: str):
16
        """Record the caller's selection."""
17
        self.complete(ClarificationResult(option=option))
18
        return f"Got it, you selected {option}."
19

20
@dataclass
21
class QuestionResult:
22
    question_id: str
23
    answer: str
24

25
class DynamicQuestionTask(AgentTask[QuestionResult]):
26
    def __init__(self, question_id: str, prompt: str, options: Optional[List[str]] = None):
27
        instructions = f"Ask: {prompt}"
28
        if options:
29
            instructions += f"\nValid options: {', '.join(options)}"
30
        super().__init__(instructions=instructions)
31
        self.question_id = question_id
32

33
    @function_tool
34
    async def record_answer(self, context: RunContext, answer: str):
35
        """Record the caller's answer."""
36
        self.complete(QuestionResult(question_id=self.question_id, answer=answer))
37
        return "Thank you."

1
def add_intent_tasks(
2
    task_group: TaskGroup,
3
    config: ClientConfig,
4
    intent: str,
5
    selected_option: Optional[str] = None
6
) -> bool:
7
    """Add tasks to a TaskGroup based on client config for a specific intent.
8
    Returns True if tasks were added."""
9

10
    if intent not in config.intent_flows:
11
        return False
12

13
    flow = config.intent_flows[intent]
14

15
    # If we need clarification first
16
    if flow.clarifying_question and not selected_option:
17
        task_group.add(
18
            lambda: ClarificationTask(
19
                flow.clarifying_question,
20
                flow.clarifying_options or []
21
            ),
22
            id="clarification",
23
            description="Get clarification from caller"
24
        )
25
        return True
26

27
    # Add tasks for the selected option
28
    questions = flow.questions_by_option.get(selected_option, [])
29

30
    if not questions:
31
        return False
32

33
    for q in questions:
34
        # Use default argument to capture loop variable
35
        task_group.add(
36
            lambda q=q: DynamicQuestionTask(
37
                question_id=q.id,
38
                prompt=q.prompt,
39
                options=q.options if q.response_type == "multiple_choice" else None
40
            ),
41
            id=q.id,
42
            description=f"Ask about {q.id}"
43
        )
44

45
    return True

1
class DynamicRoutingAgent(Agent):
2
    def __init__(self, config: ClientConfig):
3
        self.config = config
4
        super().__init__(
5
            instructions=self._build_instructions()
6
        )
7

8
    def _build_instructions(self) -> str:
9
        base = f"You are an assistant for {self.config.client_name}."
10
        if self.config.custom_greeting:
11
            base += f"\n\nUse this greeting: {self.config.custom_greeting}"
12
        return base
13

14
    def _get_transfer_number(self, context: RunContext, option: str) -> str:
15
        """Get the appropriate warm transfer number based on context."""
16
        state: SessionState = context.session.userdata
17
        flow = self.config.intent_flows.get(state.intent, {})
18

19
        # Check for option-specific number
20
        if option and flow:
21
            specific_number = flow.warm_transfer_by_option.get(option)
22
            if specific_number:
23
                return specific_number
24

25
        # Fall back to category numbers
26
        wt = self.config.warm_transfer
27
        if option == "medical" and wt.medical_number:
28
            return wt.medical_number
29
        if option == "cosmetic" and wt.cosmetic_number:
30
            return wt.cosmetic_number
31

32
        return wt.default_number
33

34
    @function_tool()
35
    async def transfer_to_specialist(
36
        self,
37
        context: RunContext,
38
        option: str
39
    ) -> str:
40
        """Transfer the caller to the appropriate specialist."""
41
        transfer_number = self._get_transfer_number(context, option)
42

43
        # Initiate SIP transfer (implementation depends on your setup)
44
        await initiate_warm_transfer(context.session, transfer_number)
45

46
        return f"Transferring you now. Please hold."
47

48
    @function_tool()
49
    async def handle_error_transfer(self, context: RunContext) -> str:
50
        """Transfer to fallback number on error or timeout."""
51
        fallback = self.config.warm_transfer.error_fallback_number
52
        number = fallback or self.config.warm_transfer.default_number
53

54
        await initiate_warm_transfer(context.session, number)
55
        return "Let me connect you with someone who can help."