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.

Safely truncating chat context while preserving history

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 the recommended pattern:

from livekit.agents import ChatContext, ChatMessage

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

class ConversationManager:
    """Manages chat context with safe truncation."""

    @staticmethod
    def append_to_canonical_history(session: AgentSession, message: ChatMessage):
        """Always append to the canonical history before any truncation."""
        state: SessionState = session.userdata
        state.full_history.append(message)

    @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 canonical history."""
        state: SessionState = session.userdata

        ctx = ChatContext()

        if include_summary and len(state.full_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
        recent_messages = state.full_history[-max_messages:]
        for msg in recent_messages:
            ctx.add_message(role=msg.role, content=msg.content)

        return ctx

    @staticmethod
    def restore_full_context(session: AgentSession) -> ChatContext:
        """Restore full conversation history when needed."""
        state: SessionState = session.userdata
        ctx = ChatContext()
        for msg in state.full_history:
            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 full context restored."""
        # Restore full history for the next agent
        full_ctx = ConversationManager.restore_full_context(context.session)
        followup_agent = FollowUpAgent()
        followup_agent.chat_ctx = full_ctx
        return followup_agent, "Let me connect you with our follow-up team."

AgentTask and TaskGroup with chat context

AgentTask and TaskGroup are useful for structured data collection flows like authentication. However, understanding how chat context propagates to tasks is critical.

The context propagation challenge

A common issue: a user provides their name early in the conversation, but when a TaskGroup runs for authentication, it asks for the name again—even though chat_ctx was passed.

This happens because TaskGroup creates focused sub-conversations for each task. The task's LLM sees its specific instructions and the chat context you provide, but may not recognize that information was already collected if:

1. The relevant messages are truncated from the context
2. The task instructions don't reference the existing data
3. The task is designed to always collect fresh data

Best practices for TaskGroup context

1. Inject session state into task instructions:

from livekit.agents import AgentTask, TaskGroup

def create_auth_tasks(session: AgentSession) -> TaskGroup:
    state: SessionState = session.userdata

    tasks = []

    # Only add tasks for missing information
    if not state.caller_name:
        tasks.append(AgentTask(
            instructions="Ask for the caller's first and last name.",
            output_schema=NameSchema,
        ))

    if not state.date_of_birth:
        tasks.append(AgentTask(
            instructions="Ask for the caller's date of birth for verification.",
            output_schema=DOBSchema,
        ))

    if not state.phone_number:
        tasks.append(AgentTask(
            instructions="Ask for a callback phone number.",
            output_schema=PhoneSchema,
        ))

    # If we already have everything, create a verification-only task
    if not tasks:
        tasks.append(AgentTask(
            instructions=f"""Verify the caller's identity. We have:
- Name: {state.caller_name}
- DOB: {state.date_of_birth}
- Phone: {state.phone_number}
Ask them to confirm this is correct.""",
            output_schema=VerificationSchema,
        ))

    return TaskGroup(tasks=tasks)

2. Pass relevant chat context to tasks:

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

        # Create context with relevant recent messages
        task_ctx = ConversationManager.create_truncated_context(
            context.session,
            max_messages=5,  # Recent messages for context
            include_summary=True  # Include session state summary
        )

        task_group = create_auth_tasks(context.session)

        # Run the task group with the prepared context
        results = await task_group.run(
            context=context,
            chat_ctx=task_ctx
        )

        # Update session state with collected data
        for result in results:
            if hasattr(result, 'name'):
                state.caller_name = result.name
            if hasattr(result, 'dob'):
                state.date_of_birth = result.dob
            # ... etc

3. Use explicit pre-fill for known data:

def create_smart_auth_task(session: AgentSession) -> AgentTask:
    state: SessionState = session.userdata

    known_info = []
    missing_info = []

    if state.caller_name:
        known_info.append(f"Name: {state.caller_name}")
    else:
        missing_info.append("first and last name")

    if state.date_of_birth:
        known_info.append(f"DOB: {state.date_of_birth}")
    else:
        missing_info.append("date of birth")

    instructions = "Collect caller information for authentication.\n"

    if known_info:
        instructions += f"\nWe already have: {', '.join(known_info)}\n"
        instructions += "Confirm this information is correct.\n"

    if missing_info:
        instructions += f"\nWe still need: {', '.join(missing_info)}\n"
        instructions += "Ask for only the missing information.\n"

    return AgentTask(
        instructions=instructions,
        output_schema=AuthDataSchema,
    )

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
    )

Build TaskGroups dynamically from config

def create_intent_task_group(
    config: ClientConfig,
    intent: str,
    selected_option: Optional[str] = None
) -> Optional[TaskGroup]:
    """Create a TaskGroup based on client config for a specific intent."""

    if intent not in config.intent_flows:
        return None

    flow = config.intent_flows[intent]

    # If we need clarification first
    if flow.clarifying_question and not selected_option:
        return TaskGroup(tasks=[
            AgentTask(
                instructions=f"""Ask the caller: "{flow.clarifying_question}"
Valid responses are: {', '.join(flow.clarifying_options or [])}
Determine which option they want.""",
                output_schema=ClarificationSchema,
            )
        ])

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

    if not questions:
        return None

    tasks = []
    for q in questions:
        task_instructions = f"Ask: {q.prompt}"
        if q.response_type == "multiple_choice" and q.options:
            task_instructions += f"\nValid options: {', '.join(q.options)}"

        tasks.append(AgentTask(
            instructions=task_instructions,
            output_schema=create_schema_for_question(q),
        ))

    return TaskGroup(tasks=tasks)

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."

Complete example: Config-driven appointment flow

# Example client configuration (would typically be stored in database)
zocdoc_config = ClientConfig(
    client_id="zocdoc-medical-group",
    client_name="ZocDoc Medical Group",
    custom_greeting="Thank you for calling ZocDoc Medical Group. How can I help you today?",
    warm_transfer=WarmTransferConfig(
        default_number="+15551000001",
        medical_number="+15551000002",
        cosmetic_number="+15551000003",
        error_fallback_number="+15551000099",
    ),
    intent_flows={
        "make_appointment": IntentFlowConfig(
            intent_name="make_appointment",
            clarifying_question="Is this for a medical or cosmetic appointment?",
            clarifying_options=["medical", "cosmetic"],
            questions_by_option={
                "medical": [
                    QuestionConfig(id="symptoms", prompt="What symptoms are you experiencing?", response_type="text"),
                    QuestionConfig(id="duration", prompt="How long have you had these symptoms?", response_type="text"),
                    QuestionConfig(id="urgency", prompt="Is this urgent or can it wait a few days?", response_type="multiple_choice", options=["urgent", "can wait"]),
                ],
                "cosmetic": [
                    QuestionConfig(id="procedure", prompt="What procedure are you interested in?", response_type="text"),
                    QuestionConfig(id="consultation", prompt="Have you had a consultation before?", response_type="yes_no"),
                ],
            },
            warm_transfer_by_option={
                "medical": "+15551000002",
                "cosmetic": "+15551000003",
            }
        )
    }
)

class AppointmentFlowAgent(Agent):
    def __init__(self, config: ClientConfig):
        self.config = config
        super().__init__(
            instructions=f"""You help callers at {config.client_name} schedule appointments.

When a caller wants to make an appointment:
1. Determine if it's medical or cosmetic
2. Ask the relevant screening questions
3. Transfer to the appropriate specialist

Always be professional and empathetic."""
        )

    async def handle_appointment_intent(self, context: RunContext):
        state: SessionState = context.session.userdata

        # Step 1: Get clarification
        clarify_group = create_intent_task_group(self.config, "make_appointment")
        if clarify_group:
            result = await clarify_group.run(context=context)
            selected_option = result[0].option  # "medical" or "cosmetic"
            state.collected_data["appointment_type"] = selected_option

        # Step 2: Run option-specific questions
        questions_group = create_intent_task_group(
            self.config,
            "make_appointment",
            selected_option
        )
        if questions_group:
            results = await questions_group.run(context=context)
            for i, result in enumerate(results):
                question_id = self.config.intent_flows["make_appointment"].questions_by_option[selected_option][i].id
                state.collected_data[question_id] = result.answer

        # Step 3: Transfer with all collected data
        transfer_number = self._get_transfer_number(context, selected_option)
        await initiate_warm_transfer(
            context.session,
            transfer_number,
            metadata=state.collected_data  # Pass collected data to receiving agent
        )

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.