Building multi-agent architectures with LiveKit agents

When designing a multi-agent architecture with LiveKit, developers often ask:

• How do I transfer control between agents?
• How do I share context without making agents stateful?
• How should I manage session state vs LLM chat context?
• How do I build dynamic, per-client configurable flows?

Based on internal development and customer implementations, this guide covers recommended patterns for each of these challenges.

Transferring between agents with function_tool

The @function_tool decorator defines tools that an LLM can invoke. For multi-agent architectures, you can use tools to hand off control to another agent:

from livekit.agents import Agent, function_tool, RunContext

class IntakeAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You gather initial information from callers."
        )

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

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

When the LLM decides a handoff is appropriate, it calls the tool and returns both the new agent and a message to speak during the transition. The framework handles the orchestration automatically.

See the tools documentation for more on defining tools and the multi-agent handoff guide for handoff patterns.

Session state vs LLM chat context

A critical design decision in multi-agent systems is how to manage state. There are two complementary approaches:

1. Session state (userdata): Deterministic, structured data stored on the session
2. LLM chat context (chat_ctx): The conversation history passed to the LLM

Recommended pattern: Session state as primary memory

For production systems, treat session.userdata as your primary source of truth for important data like:

• Caller identity (name, DOB, phone number)
• Account information
• Intent classification
• Collected form data
• Authentication status

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

@dataclass
class SessionState:
    # Deterministic memory - always reliable
    caller_name: Optional[str] = None
    date_of_birth: Optional[str] = None
    phone_number: Optional[str] = None
    account_id: Optional[str] = None
    intent: Optional[str] = None
    authenticated: bool = False
    collected_data: dict = field(default_factory=dict)

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

class IntakeAgent(Agent):
    @function_tool()
    async def save_caller_identity(
        self,
        context: RunContext,
        name: str,
        date_of_birth: str,
        phone: str
    ):
        """Save verified caller identity information."""
        state: SessionState = context.session.userdata
        state.caller_name = name
        state.date_of_birth = date_of_birth
        state.phone_number = phone
        return f"Thank you {name}, I've saved your information."

The LLM's chat context should be treated as working memory that can be truncated for performance, while userdata serves as persistent memory that survives across agents and context truncation.

Accessing conversation history

The SDK provides session.history to access the full conversation history:

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

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

Custom helper pattern: truncated context with summaries

Note: The following ConversationManager is a custom helper pattern you can implement in your application. It is not part of the LiveKit SDK.

When you truncate chat_ctx for an agent (e.g., a specialized scheduling agent that doesn't need the full history), you risk losing information for subsequent agents. Here's a helper pattern you can use:

from livekit.agents import ChatContext, ChatMessage

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

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

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

        ctx = ChatContext()

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

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

        return ctx

Use this pattern in your handoff logic:

class SchedulingAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You help schedule appointments."
        )

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

AgentTask and TaskGroup for structured data collection

AgentTask and TaskGroup are useful for structured data collection flows like authentication or surveys. To use them correctly, you need to create custom task classes that define their result types.

Creating custom task classes

The AgentTask class is generic and requires subclassing. Define your result type and call self.complete(result) when the task finishes:

from livekit.agents import AgentTask
from dataclasses import dataclass

@dataclass
class NameResult:
    first_name: str
    last_name: str

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

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

@dataclass
class DOBResult:
    date_of_birth: str

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

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

@dataclass
class PhoneResult:
    phone_number: str

class CollectPhoneTask(AgentTask[PhoneResult]):
    def __init__(self):
        super().__init__(
            instructions="Ask for a callback phone number."
        )

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

Using TaskGroup with lambdas

TaskGroup collects tasks using the add() method with lambda functions. Pass chat_ctx in the constructor, then await the group directly:

from livekit.agents import TaskGroup

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

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

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

        if not state.date_of_birth:
            task_group.add(
                lambda: CollectDOBTask(),
                id="collect_dob",
                description="Collect date of birth"
            )

        if not state.phone_number:
            task_group.add(
                lambda: CollectPhoneTask(),
                id="collect_phone",
                description="Collect phone number"
            )

        # Await the task group directly
        results = await task_group

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

Accessing conversation history

Use session.history for full conversation history, or self.chat_ctx inside agents:

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

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

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

See the survey agent example and drive-thru example for canonical implementations of these patterns.

Dynamic per-client routing and flows

For multi-tenant systems, you often need different conversation flows per client. Here's a pattern for config-driven dynamic routing.

Define your client configuration schema

from dataclasses import dataclass
from typing import List, Optional, Dict, Any

@dataclass
class WarmTransferConfig:
    default_number: str
    medical_number: Optional[str] = None
    cosmetic_number: Optional[str] = None
    error_fallback_number: Optional[str] = None

@dataclass
class QuestionConfig:
    id: str
    prompt: str
    response_type: str  # "text", "yes_no", "multiple_choice"
    options: Optional[List[str]] = None
    required: bool = True

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

@dataclass
class ClientConfig:
    client_id: str
    client_name: str
    warm_transfer: WarmTransferConfig
    intent_flows: Dict[str, IntentFlowConfig] = field(default_factory=dict)
    custom_greeting: Optional[str] = None
    timezone: str = "UTC"

Load configuration at call start

import json
from pathlib import Path

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

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

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

        # Return default config if not found
        return ClientConfig(
            client_id=client_id,
            client_name="Default",
            warm_transfer=WarmTransferConfig(default_number="+15551234567")
        )

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

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

    # Initialize session state with config
    session_state = SessionState()
    session_state.client_config = client_config

    session = AgentSession()
    session.userdata = session_state

    # Start with config-aware agent
    await session.start(
        agent=create_intake_agent(client_config),
        room=ctx.room
    )

Define custom task classes for dynamic questions

First, define task classes for your question types:

@dataclass
class ClarificationResult:
    option: str

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

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

@dataclass
class QuestionResult:
    question_id: str
    answer: str

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

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

Build TaskGroups dynamically from config

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

    if intent not in config.intent_flows:
        return False

    flow = config.intent_flows[intent]

    # If we need clarification first
    if flow.clarifying_question and not selected_option:
        task_group.add(
            lambda: ClarificationTask(
                flow.clarifying_question,
                flow.clarifying_options or []
            ),
            id="clarification",
            description="Get clarification from caller"
        )
        return True

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

    if not questions:
        return False

    for q in questions:
        # Use default argument to capture loop variable
        task_group.add(
            lambda q=q: DynamicQuestionTask(
                question_id=q.id,
                prompt=q.prompt,
                options=q.options if q.response_type == "multiple_choice" else None
            ),
            id=q.id,
            description=f"Ask about {q.id}"
        )

    return True

Set warm transfer numbers dynamically

class DynamicRoutingAgent(Agent):
    def __init__(self, config: ClientConfig):
        self.config = config
        super().__init__(
            instructions=self._build_instructions()
        )

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

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

        # Check for option-specific number
        if option and flow:
            specific_number = flow.warm_transfer_by_option.get(option)
            if specific_number:
                return specific_number

        # Fall back to category numbers
        wt = self.config.warm_transfer
        if option == "medical" and wt.medical_number:
            return wt.medical_number
        if option == "cosmetic" and wt.cosmetic_number:
            return wt.cosmetic_number

        return wt.default_number

    @function_tool()
    async def transfer_to_specialist(
        self,
        context: RunContext,
        option: str
    ) -> str:
        """Transfer the caller to the appropriate specialist."""
        transfer_number = self._get_transfer_number(context, option)

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

        return f"Transferring you now. Please hold."

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

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

Key takeaways

1. Use userdata as primary memory: Store deterministic data (identity, account info, collected form data) in session.userdata, not just in chat context.

2. Maintain a canonical history: Always append to a full conversation history in userdata before truncating chat_ctx. This lets you restore context when needed.

3. Be explicit with TaskGroup context: Don't assume TaskGroup will automatically see earlier conversation. Inject known data into task instructions and only ask for missing information.

4. Design for truncation: When handing off to specialized agents, create truncated contexts that include a summary of session state plus recent messages.

5. Use config-driven flows: For multi-tenant systems, externalize flow logic into configuration that can vary per client.

6. Set transfer numbers dynamically: Use session state and client config to determine the appropriate transfer destination at runtime.