Chapter 7: Multi-Agent Workflows

A single agent can handle a surprising amount of complexity. But at some point, you will find yourself stuffing too many responsibilities into one set of instructions. The agent that greets callers also tries to book appointments, handle billing questions, and enforce compliance rules. Its instructions grow to three pages. It starts confusing contexts. It forgets which tools are relevant.

The solution is the same one that works in human organizations: specialization. Split the work across multiple agents, each with a focused role, and hand off between them as the conversation evolves.

What follows is a practical guide to building multi-agent voice systems with LiveKit: when to split agents, how to hand off between them, and how to share state across an entire session.

When to Use Agents vs Tasks

Before reaching for multiple agents, understand the distinction between agents and tasks. Both are units of work in a LiveKit workflow, but they serve different purposes.

Tasks are short-lived. They take temporary control of the session, accomplish a specific objective, and return a typed result. They do not persist after completion. Think of a task as a function call that can talk to the user.

Agents are long-lived. They define reasoning behavior, carry their own instructions and tools, and remain in control of the session until they explicitly hand off to another agent. Think of an agent as a role that the system inhabits.

The rule of thumb: use tasks for data collection, use agents for different modes of operation.

For example, in a phone receptionist system:

• IntakeAgent (agent): greets callers, determines intent, routes them
• SchedulerAgent (agent): handles the entire booking flow with calendar tools
• BillingAgent (agent): answers account questions with payment API access
• GetEmailTask (task): collects and validates an email address, returns it as a typed result

The agents represent different modes of operation. The task is a reusable subroutine that any agent can invoke.

Agent Handoffs

There are two ways to transfer control between agents.

Via Tool Return

The most natural approach: define a function tool that returns a new Agent instance. The LLM decides when to call the tool based on the conversation context.

from livekit.agents import Agent, RunContext, function_tool

class FrontDeskAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You are a front desk receptionist. Greet callers and route them."
        )

    @function_tool()
    async def transfer_to_scheduling(self, context: RunContext):
        """Transfer the caller to the scheduling department."""
        return SchedulerAgent(chat_ctx=self.chat_ctx), "Transferring you to scheduling"

    @function_tool()
    async def transfer_to_billing(self, context: RunContext):
        """Transfer the caller to billing for payment questions."""
        return BillingAgent(chat_ctx=self.chat_ctx), "Transferring you to billing"

The tuple return (Agent, str) is important. The string becomes the tool output that the LLM sees, which lets the new agent understand why the handoff happened. The LLM decides to call transfer_to_scheduling based on what the caller said. You do not need to write routing logic yourself.

Via update_agent()

For programmatic handoffs where you want explicit control, use session.update_agent():

# Inside any async context with access to the session
session.update_agent(SchedulerAgent())

This is useful when the handoff decision comes from business logic rather than the LLM. For example, after a task completes and you want to route based on the result:

@function_tool()
async def collect_issue_type(self, context: RunContext, issue: str):
    """Record the caller's issue type."""
    if issue == "billing":
        context.session.update_agent(BillingAgent())
    elif issue == "technical":
        context.session.update_agent(TechnicalAgent())

Context Preservation

By default, a new agent starts with a blank conversation history. It only sees its own instructions. This is often what you want: a fresh start for a new phase of the conversation. But sometimes the new agent needs to know what was discussed.

Pass chat_ctx to preserve the conversation:

# Preserve full history (including previous agent's instructions)
return SchedulerAgent(chat_ctx=self.chat_ctx)

# Preserve conversation turns only (exclude previous instructions)
return SchedulerAgent(chat_ctx=self.chat_ctx.copy(exclude_instructions=True))

Use exclude_instructions=True when the new agent has its own instructions and you do not want the previous agent’s system prompt polluting its context.

When to reset context: Reset when agents have fundamentally different roles and the prior conversation would confuse the new agent. A billing agent does not need to see the small talk from the greeting phase.

When to preserve context: Preserve when the new agent needs to reference what was discussed. A scheduler needs to know the caller mentioned “Thursday afternoon” before the handoff happened.

For long conversations, consider summarizing context before handoff instead of passing the entire history. LiveKit provides patterns for this (see the summarizing context section in the official docs).

The Observer Pattern

Sometimes you want a second intelligence monitoring the conversation without participating in it: not a second agent that takes turns speaking, but a background process that listens, analyzes, and occasionally injects guidance.

This is the observer pattern. A background agent subscribes to conversation events, runs analysis with a separate LLM (potentially a slower, more capable model), and modifies the active agent’s context with system messages when it has something to say.

How It Works

The observer listens to the conversation_item_added event on the session. Every time a user or assistant message is committed to the chat history, the observer receives it. It accumulates messages, runs periodic analysis, and injects system messages into the active agent’s chat_ctx when it detects something worth flagging.

Use Cases

• Safety monitoring: Flag when a caller mentions self-harm, threats, or other sensitive topics
• Compliance: Ensure the agent follows required disclosures or does not make prohibited claims
• Quality coaching: Suggest better responses when the agent misses an opportunity
• Sentiment tracking: Detect caller frustration and inject de-escalation hints

Full Example

This example implements a quality observer inspired by the Doheny Surf Desk pattern. It monitors conversation flow and injects coaching hints when needed.

import asyncio
from livekit.agents import Agent, AgentSession, ChatContext, inference


class QualityObserver:
    """Background observer that monitors conversation quality."""

    def __init__(self, session: AgentSession):
        self._session = session
        self._message_buffer: list[dict] = []
        self._analysis_interval = 5  # analyze every 5 messages
        self._running = True

        # Use a separate, potentially smarter model for analysis
        self._llm = inference.LLM("openai/gpt-4.1")

        # Subscribe to conversation events
        session.on("conversation_item_added", self._on_conversation_item)

    def _on_conversation_item(self, event) -> None:
        """Called each time a message is added to the conversation."""
        item = event.item
        if item.role in ("user", "assistant"):
            self._message_buffer.append({
                "role": item.role,
                "content": item.text_content or "",
            })

            if len(self._message_buffer) >= self._analysis_interval:
                asyncio.create_task(self._analyze())

    async def _analyze(self) -> None:
        """Run background analysis on accumulated messages."""
        if not self._message_buffer:
            return

        # Build an analysis prompt
        transcript = "\n".join(
            f"{m['role'].upper()}: {m['content']}" for m in self._message_buffer
        )
        self._message_buffer.clear()

        analysis_ctx = ChatContext()
        analysis_ctx.add_message(
            role="system",
            content=(
                "You are a conversation quality monitor for a voice AI receptionist. "
                "Analyze the following transcript segment. If you detect any of these "
                "issues, respond with a brief, actionable hint. Otherwise respond with "
                "'NO_ISSUES'.\n\n"
                "Issues to watch for:\n"
                "- Agent gave incorrect or speculative information\n"
                "- Caller sounds frustrated and agent did not acknowledge it\n"
                "- Agent forgot to collect required information\n"
                "- Safety concern (threats, self-harm mentions)\n"
                "- Agent is being too verbose for a voice conversation"
            ),
        )
        analysis_ctx.add_message(role="user", content=transcript)

        response = await self._llm.chat(chat_ctx=analysis_ctx).collect()
        result = (response.text or "").strip()

        if result and result != "NO_ISSUES":
            await self._inject_hint(result)

    async def _inject_hint(self, hint: str) -> None:
        """Inject a system message into the active agent's context."""
        agent = self._session.agent
        if agent is None:
            return

        chat_ctx = agent.chat_ctx.copy()
        chat_ctx.add_message(
            role="system",
            content=f"[QUALITY HINT]: {hint}",
        )
        await agent.update_chat_ctx(chat_ctx)

To wire it up, create the observer after starting the session:

# server = AgentServer() defined earlier
@server.rtc_session()
async def entrypoint(ctx: JobContext):
    await ctx.connect()

    session = AgentSession(
        stt=inference.STT("deepgram/nova-3"),
        llm=inference.LLM("openai/gpt-4.1-mini"),
        tts=inference.TTS("cartesia/sonic-3"),
        vad=silero.VAD.load(),
    )

    await session.start(agent=FrontDeskAgent(), room=ctx.room)

    # Start the observer -- it runs in the background
    observer = QualityObserver(session)

The observer never speaks. It never takes a turn. It quietly watches and nudges the active agent when needed. The active agent sees the [QUALITY HINT] as a system message in its context and can adjust its behavior accordingly.

Session Userdata for Shared State

When multiple agents collaborate in a single session, they need shared state. A scheduler agent books a slot, and the confirmation agent needs to know which slot was booked. An intake agent collects a name, and every subsequent agent needs to use it.

LiveKit solves this with session userdata, a typed object that persists across agent transfers.

Defining the State

Use a dataclass to define your session-wide state:

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class SessionState:
    caller_name: Optional[str] = None
    caller_phone: Optional[str] = None
    caller_email: Optional[str] = None
    intent: Optional[str] = None
    booked_slot: Optional[str] = None
    notes: list[str] = field(default_factory=list)

Creating a Typed Session

Pass the dataclass as a type parameter and initial value to AgentSession:

session = AgentSession[SessionState](
    userdata=SessionState(),
    stt=inference.STT("deepgram/nova-3"),
    llm=inference.LLM("openai/gpt-4.1-mini"),
    tts=inference.TTS("cartesia/sonic-3"),
    vad=silero.VAD.load(),
)

Accessing State

Inside any agent or tool, access userdata through the session or the RunContext:

class IntakeAgent(Agent):
    @function_tool()
    async def record_name(self, ctx: RunContext[SessionState], name: str):
        """Record the caller's name."""
        ctx.userdata.caller_name = name

class SchedulerAgent(Agent):
    async def on_enter(self) -> None:
        name = self.session.userdata.caller_name
        await self.session.generate_reply(
            instructions=f"Greet {name} and help them book an appointment."
        )

The key insight: userdata survives agent handoffs. When IntakeAgent hands off to SchedulerAgent, the SessionState object is the same instance. No serialization, no passing data through constructors. Every agent reads and writes to the same shared state.

Building a Multi-Agent Phone Receptionist

Let us put everything together with a complete example: a phone receptionist with three specialized agents.

The Architecture

Caller
  |
  v
FrontDeskAgent  -->  IntakeAgent  -->  SchedulerAgent
  (routes)          (collects info)    (books appointment)
                         |
                    uses TaskGroup
                    for data collection

Session State

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ReceptionistState:
    caller_name: Optional[str] = None
    caller_phone: Optional[str] = None
    service_needed: Optional[str] = None
    preferred_date: Optional[str] = None
    booked_slot: Optional[str] = None

FrontDeskAgent

The greeter and router. Its only job is to figure out what the caller needs and send them to the right agent.

from livekit.agents import Agent, RunContext, function_tool

class FrontDeskAgent(Agent):
    def __init__(self, **kwargs):
        super().__init__(
            instructions=(
                "You are the front desk receptionist at a plumbing company. "
                "Greet callers warmly. Determine if they need to book an appointment "
                "or just have a general question. Keep it brief -- this is a phone call."
            ),
            **kwargs,
        )

    async def on_enter(self) -> None:
        self.session.generate_reply(
            instructions="Greet the caller and ask how you can help them today."
        )

    @function_tool()
    async def route_to_booking(self, context: RunContext[ReceptionistState]):
        """Route the caller to appointment booking when they want to schedule service."""
        return IntakeAgent(
            chat_ctx=self.chat_ctx.copy(exclude_instructions=True)
        ), "Routing to intake"

    @function_tool()
    async def answer_general_question(self, context: RunContext[ReceptionistState], answer: str):
        """Answer a general question about business hours, location, or services.

        Args:
            answer: Your answer to the caller's question.
        """
        return f"Answered: {answer}"

IntakeAgent

Collects caller details before booking. Uses tools to record each piece of information and hands off to scheduling once everything is gathered.

class IntakeAgent(Agent):
    def __init__(self, **kwargs):
        super().__init__(
            instructions=(
                "You are collecting information before booking an appointment. "
                "You need the caller's name, phone number, and what service they need. "
                "Ask for each piece naturally in conversation. Do not read a list of "
                "questions -- have a conversation."
            ),
            **kwargs,
        )

    async def on_enter(self) -> None:
        self.session.generate_reply(
            instructions=(
                "Let the caller know you will need a few details before booking. "
                "Start by asking for their name."
            )
        )

    @function_tool()
    async def record_name(self, ctx: RunContext[ReceptionistState], name: str):
        """Record the caller's name.

        Args:
            name: The caller's full name.
        """
        ctx.userdata.caller_name = name
        return self._check_complete()

    @function_tool()
    async def record_phone(self, ctx: RunContext[ReceptionistState], phone: str):
        """Record the caller's phone number.

        Args:
            phone: The caller's phone number.
        """
        ctx.userdata.caller_phone = phone
        return self._check_complete()

    @function_tool()
    async def record_service(self, ctx: RunContext[ReceptionistState], service: str):
        """Record what service the caller needs.

        Args:
            service: Description of the service needed.
        """
        ctx.userdata.service_needed = service
        return self._check_complete()

    def _check_complete(self):
        ud = self.session.userdata
        if ud.caller_name and ud.caller_phone and ud.service_needed:
            return SchedulerAgent(
                chat_ctx=self.chat_ctx.copy(exclude_instructions=True)
            )
        return None

SchedulerAgent

Books the appointment. Has access to calendar tools and knows the caller’s details from userdata.

import datetime

AVAILABLE_SLOTS = [
    "Monday 9:00 AM", "Monday 2:00 PM",
    "Tuesday 10:00 AM", "Tuesday 3:00 PM",
    "Wednesday 9:00 AM", "Wednesday 1:00 PM",
]

class SchedulerAgent(Agent):
    def __init__(self, **kwargs):
        super().__init__(
            instructions=(
                "You are booking an appointment. You know the caller's details already. "
                "Check availability, offer options, and confirm the booking. "
                "Be concise -- say times naturally like 'Monday at 9 in the morning'."
            ),
            **kwargs,
        )

    async def on_enter(self) -> None:
        ud = self.session.userdata
        self.session.generate_reply(
            instructions=(
                f"The caller is {ud.caller_name} and they need {ud.service_needed}. "
                f"Let them know you can help with scheduling and ask when works for them."
            )
        )

    @function_tool()
    async def check_availability(self, ctx: RunContext[ReceptionistState]):
        """Check available appointment slots."""
        return f"Available slots: {', '.join(AVAILABLE_SLOTS)}"

    @function_tool()
    async def book_appointment(
        self, ctx: RunContext[ReceptionistState], slot: str
    ):
        """Book the appointment at the selected time.

        Args:
            slot: The selected time slot.
        """
        if slot not in AVAILABLE_SLOTS:
            return f"Sorry, {slot} is not available. Available: {', '.join(AVAILABLE_SLOTS)}"

        ctx.userdata.booked_slot = slot
        return (
            f"Appointment booked for {ctx.userdata.caller_name} on {slot} "
            f"for {ctx.userdata.service_needed}. Confirmation will be sent to "
            f"{ctx.userdata.caller_phone}."
        )

Wiring It Together

from livekit.agents import AgentServer, AgentSession, JobContext, cli, inference
from livekit.plugins import silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel

server = AgentServer()

@server.rtc_session()
async def receptionist(ctx: JobContext):
    await ctx.connect()

    session = AgentSession[ReceptionistState](
        userdata=ReceptionistState(),
        stt=inference.STT("deepgram/nova-3"),
        llm=inference.LLM("openai/gpt-4.1-mini"),
        tts=inference.TTS("cartesia/sonic-3"),
        vad=silero.VAD.load(),
        turn_detection=MultilingualModel(),
    )

    await session.start(agent=FrontDeskAgent(), room=ctx.room)

if __name__ == "__main__":
    cli.run_app(server)

The caller experiences a seamless conversation. Behind the scenes, three agents take turns, each with focused instructions and tools, passing state through userdata and conversation history through chat_ctx.

Best Practices

Keep Agents Focused

One responsibility per agent. If an agent’s instructions exceed a paragraph, it is probably doing too much. Split it. A focused agent with five lines of instructions will outperform a Swiss-army-knife agent with fifty lines every time.

Announce Handoffs Naturally

Callers do not know about your agent architecture, and they should not have to. When handing off, have the current agent say something natural:

@function_tool()
async def transfer_to_scheduling(self, context: RunContext):
    """Transfer to scheduling when the caller wants to book."""
    await self.session.generate_reply(
        instructions="Let the caller know you're connecting them with scheduling."
    )
    return SchedulerAgent(chat_ctx=self.chat_ctx.copy(exclude_instructions=True))

The on_enter hook on the receiving agent then introduces itself. From the caller’s perspective, it feels like being transferred to a different department, familiar and expected.

Handle the “Back” Case

What happens when the caller says “actually, I had another question” after being transferred to scheduling? If you do not handle this, the scheduler agent will try to answer general questions with its limited instructions.

Add a “go back” tool to every non-root agent:

class SchedulerAgent(Agent):
    @function_tool()
    async def return_to_front_desk(self, context: RunContext):
        """Return to the front desk if the caller has a different question."""
        return FrontDeskAgent(
            chat_ctx=self.chat_ctx.copy(exclude_instructions=True)
        ), "Returning to front desk"

Test Each Agent Independently

Multi-agent systems are hard to debug when tested as a whole. Test each agent in isolation first. LiveKit’s testing framework lets you run an AgentSession without a LiveKit room, feeding it synthetic user input:

async with AgentSession(llm=test_llm, userdata=ReceptionistState()) as sess:
    await sess.start(IntakeAgent())
    result = await sess.run(user_input="My name is Sarah")
    result.expect.next_event().is_function_call(
        name="record_name", arguments={"name": "Sarah"}
    )

Test the handoff chain only after each individual agent works correctly. When something breaks in the chain, you will know the problem is in the handoff logic, not in the agent itself.

Mind the Context Window

Each handoff is an opportunity to trim context. A conversation that has been through three agents might have accumulated hundreds of turns. Use exclude_instructions=True at minimum. For longer conversations, summarize before handing off, or truncate to the last few turns:

# Keep only the last 6 items in the context
trimmed_ctx = self.chat_ctx.copy(exclude_instructions=True)
trimmed_ctx.truncate(max_items=6)
return NextAgent(chat_ctx=trimmed_ctx)

Summary

Multi-agent workflows let you decompose complex voice interactions into manageable, testable pieces. The core building blocks are:

• Agents for distinct modes of operation with their own instructions and tools
• Tasks for short-lived data collection that returns typed results
• Tool-based handoffs for LLM-driven routing decisions
• update_agent() for programmatic routing
• chat_ctx for preserving conversation history across handoffs
• userdata for sharing typed state across the entire session
• Observers for background monitoring without blocking the conversation

Start simple. One agent is fine until it is not. When you notice an agent struggling with too many responsibilities, split it. The framework makes handoffs cheap and context preservation straightforward. The hard part is not the code. It is deciding where to draw the boundaries between agents.