Chapter 14: Real-World Architecture Patterns

You have all the building blocks: agents, tasks, tools, handoffs, telephony. Here are five production architectures that show how they snap together. Each pattern is self-contained. Pick the one closest to your use case and adapt it. Every diagram and code snippet here comes from systems that handle real calls.

Pattern 1: Phone Receptionist

The most common voice agent architecture. A caller dials in, an AI receptionist triages the call, collects information, and routes to specialized agents for scheduling, billing, or other workflows. This is the bread-and-butter pattern for any business automation.

When to Use It

• Inbound call handling for service businesses (plumbers, dentists, law firms)
• Any scenario where callers have different intents that require different tools
• When you want to replace or augment a human receptionist

Architecture

                         +------------------+
  Caller ──> SIP ──>     |  FrontDeskAgent  |
                         |  (triage/route)  |
                         +--------+---------+
                                  |
                    +-------------+-------------+
                    |                           |
           +-------v--------+         +--------v-------+
           |  IntakeAgent   |         |  BillingAgent  |
           |  (collect info)|         |  (payments)    |
           +-------+--------+         +----------------+
                   |
            TaskGroup runs:
            +-----------------+
            | CollectName     |
            | CollectPhone    |
            | CollectService  |
            +-----------------+
                   |
           +-------v--------+
           | SchedulerAgent |
           | (book slot)    |
           +----------------+

Key Code

The intake agent uses a TaskGroup to collect required fields in any order. The LLM decides which field to ask for next based on conversation flow, not a rigid script.

from livekit.agents import Agent, AgentTask, RunContext, function_tool, TaskGroup

@dataclass
class NameResult:
    name: str

@dataclass
class PhoneResult:
    phone: str

@dataclass
class ServiceResult:
    service: str

class IntakeAgent(Agent):
    def __init__(self, **kwargs):
        super().__init__(
            instructions=(
                "Collect the caller's name, phone number, and what service "
                "they need. Be conversational -- do not read a checklist."
            ),
            **kwargs,
        )

    async def on_enter(self) -> None:
        group = TaskGroup()
        group.add(CollectNameTask())
        group.add(CollectPhoneTask())
        group.add(CollectServiceTask())

        results = await group.run(self.session)

        # All three fields collected -- store and hand off
        state = self.session.userdata
        state.caller_name = results[CollectNameTask].name
        state.caller_phone = results[CollectPhoneTask].phone
        state.service_needed = results[CollectServiceTask].service

        self.session.update_agent(
            SchedulerAgent(chat_ctx=self.chat_ctx.copy(exclude_instructions=True))
        )

The TaskGroup is the critical piece. It presents all three tasks to the LLM simultaneously. If the caller volunteers their name and service in one sentence, two tasks complete at once. The group waits until all tasks are done, then hands off. No wasted turns.

Pattern 2: Survey Caller

Outbound calling from a list. The agent reads contacts from a CSV or database, dials each one, asks a set of questions, records responses, and cleans up the LiveKit room when the call ends. This pattern is based on LiveKit’s survey_caller example and is useful for appointment reminders, feedback collection, and lead qualification.

When to Use It

• Outbound survey campaigns or appointment confirmations
• Lead qualification where you need to call a list of prospects
• Any batch outbound calling workflow

Architecture

  +------------------+       +-----------------+       +----------------+
  |  CSV / Database  | ───>  |  Dispatch Loop  | ───>  |  LiveKit Room  |
  |  (contact list)  |       |  (async queue)  |       |  (per call)    |
  +------------------+       +--------+--------+       +-------+--------+
                                      |                        |
                              creates room per call      SurveyAgent joins
                                      |                        |
                              +-------v--------+       +-------v--------+
                              |  SIP Outbound  |       |  Record answers|
                              |  (Telnyx dial) |       |  to database   |
                              +----------------+       +-------+--------+
                                                               |
                                                       +-------v--------+
                                                       |  Room cleanup  |
                                                       |  (on call end) |
                                                       +----------------+

Key Code

The dispatch loop creates a room per contact, dials out via SIP, and launches the agent. Room cleanup happens automatically when the call ends.

import csv
import asyncio
from livekit import api

async def dispatch_survey(contacts_csv: str):
    lk = api.LiveKitAPI()

    with open(contacts_csv) as f:
        contacts = list(csv.DictReader(f))

    semaphore = asyncio.Semaphore(5)  # max 5 concurrent calls

    async def call_contact(contact: dict):
        async with semaphore:
            room_name = f"survey-{contact['id']}"

            # Create a dedicated room
            room = await lk.room.create_room(
                api.CreateRoomRequest(name=room_name, empty_timeout=30)
            )

            # Dial out via SIP trunk
            await lk.sip.create_sip_participant(
                api.CreateSIPParticipantRequest(
                    sip_trunk_id="your-trunk-id",
                    sip_call_to=contact["phone"],
                    room_name=room_name,
                    participant_name=contact["name"],
                )
            )

    await asyncio.gather(*(call_contact(c) for c in contacts))

The agent itself is straightforward. It asks each question, records the answer, and disconnects when done.

class SurveyAgent(Agent):
    def __init__(self, questions: list[str], **kwargs):
        super().__init__(
            instructions=(
                "You are conducting a brief phone survey. Ask each question, "
                "wait for a response, and record it. Be polite and concise. "
                "If the person wants to stop, thank them and end the call."
            ),
            **kwargs,
        )
        self._questions = questions
        self._responses: dict[str, str] = {}

    @function_tool()
    async def record_response(
        self, ctx: RunContext, question: str, answer: str
    ):
        """Record the caller's response to a survey question.

        Args:
            question: The question that was asked.
            answer: The caller's response.
        """
        self._responses[question] = answer
        remaining = len(self._questions) - len(self._responses)
        if remaining == 0:
            return "All questions answered. Thank the caller and say goodbye."
        return f"{remaining} questions remaining."

    @function_tool()
    async def end_survey(self, ctx: RunContext):
        """End the survey and disconnect the call."""
        await save_responses(self._responses)  # your persistence logic
        await ctx.session.aclose()

The empty_timeout=30 on the room creation is important. When both the agent and the SIP participant leave, the room automatically deletes itself after 30 seconds. No orphaned rooms, no manual cleanup.

Pattern 3: IVR Navigator

An agent that calls another business’s phone system and navigates its IVR (Interactive Voice Response) menu using DTMF tones. This is useful for automating tasks like checking account balances, making payments, or reaching a specific department on behalf of a user.

When to Use It

• Automating calls to insurance companies, utilities, or government agencies
• Building a “call on my behalf” feature
• Any scenario where a human currently waits on hold navigating phone menus

Architecture

  +-----------+       +----------------+       +------------------+
  |  Your     | ───>  |  IVR Navigator | ───>  |  Target Phone    |
  |  Backend  |       |  Agent         |       |  System (IVR)    |
  +-----------+       +-------+--------+       +--------+---------+
                              |                         |
                       Listens to IVR prompts    Sends DTMF tones
                       via STT                   via SIP INFO
                              |                         |
                       +------v--------+        +-------v---------+
                       | Decision LLM  |        | "Press 1 for    |
                       | (parse menu,  |        |  billing..."    |
                       |  choose option)|       +-----------------+
                       +---------------+

Key Code

The core capability is sending DTMF digits through the SIP connection. LiveKit exposes this via the SIPParticipant API.

from livekit import rtc

class IVRNavigatorAgent(Agent):
    def __init__(self, goal: str, **kwargs):
        super().__init__(
            instructions=(
                f"You are navigating a phone menu system. Your goal: {goal}. "
                "Listen to the menu options carefully. When you hear options, "
                "use the send_dtmf tool to press the right button. "
                "If asked to hold, wait patiently. If you reach a human, "
                "report back what happened."
            ),
            **kwargs,
        )

    @function_tool()
    async def send_dtmf(self, ctx: RunContext, digits: str):
        """Send DTMF tones to navigate the phone menu.

        Args:
            digits: The digits to send (e.g., "1", "0", "1234").
        """
        for participant in ctx.room.remote_participants.values():
            await participant.publish_dtmf(digits)
        return f"Sent DTMF: {digits}. Listen for the next menu prompt."

    @function_tool()
    async def report_result(self, ctx: RunContext, result: str):
        """Report the outcome of the IVR navigation.

        Args:
            result: What happened (reached department, got info, etc.).
        """
        ctx.userdata.ivr_result = result
        await ctx.session.aclose()

The LLM hears the IVR prompts through Deepgram STT just like any other speech. It hears “Press 1 for billing, press 2 for technical support” and decides which digit to send based on its goal. The STT model handles the robotic IVR voice well. Deepgram’s nova-3 was trained on telephony audio and has no trouble with pre-recorded prompts.

One important detail: add a longer silence_timeout to your VAD configuration. IVR systems have long pauses and hold music. You do not want the agent to think the call ended during a 30-second hold.

Pattern 4: Warm Handoff

The AI handles initial triage, collects context, and transfers the caller to a human agent, passing along everything it learned. The human picks up the call with full context instead of asking the caller to repeat themselves. This is the most practical pattern for customer service teams that want AI to handle the first 90 seconds.

When to Use It

• Customer service where AI handles triage but humans handle resolution
• Escalation paths for complex or sensitive issues
• Hybrid teams transitioning from fully human to AI-assisted workflows

Architecture

  Caller
    |
    v
  +----------------+     collect context     +------------------+
  |  TriageAgent   | ──────────────────────> |  Prepare Handoff |
  |  (AI, collects |                         |  (summary + SIP  |
  |   issue details)|                        |   transfer)      |
  +----------------+                         +--------+---------+
                                                      |
                                              SIP REFER to human
                                                      |
                                             +--------v---------+
                                             |  Human Agent      |
                                             |  (sees context on |
                                             |   their screen)   |
                                             +------------------+

Key Code

The triage agent collects the issue, then performs a SIP transfer to the human queue. Before transferring, it pushes context to the human’s dashboard.

import httpx
from livekit.agents import Agent, RunContext, function_tool

class TriageAgent(Agent):
    def __init__(self, **kwargs):
        super().__init__(
            instructions=(
                "You are the first point of contact for customer support. "
                "Collect the customer's name, account number, and a brief "
                "description of their issue. Then transfer to a human agent. "
                "Reassure the caller they will not need to repeat themselves."
            ),
            **kwargs,
        )

    @function_tool()
    async def transfer_to_human(
        self, ctx: RunContext, name: str, account: str, issue_summary: str
    ):
        """Transfer the caller to a human agent with context.

        Args:
            name: Customer's name.
            account: Account number or identifier.
            issue_summary: Brief summary of the issue.
        """
        # Push context to the agent dashboard
        async with httpx.AsyncClient() as client:
            await client.post(
                "https://your-api.com/handoff-context",
                json={
                    "caller_name": name,
                    "account": account,
                    "issue_summary": issue_summary,
                    "transcript": self._get_transcript(),
                    "room_name": ctx.room.name,
                },
            )

        # Transfer via SIP -- caller hears ringing, then the human picks up
        for participant in ctx.room.remote_participants.values():
            await participant.transfer_sip(
                transfer_to="sip:[email protected]"
            )

        return "Transferring now."

    def _get_transcript(self) -> str:
        return "\n".join(
            f"{msg.role}: {msg.text_content}"
            for msg in self.chat_ctx.items
            if msg.text_content
        )

The transfer_sip call performs a SIP REFER, which is the telephony-standard way to transfer a call. The caller hears a brief ring, then the human picks up. From the caller’s perspective, it feels like a normal call transfer.

The context push is what makes this pattern valuable. The human sees the customer’s name, account, issue summary, and full transcript before they even say hello. No “can you tell me your account number again?” That alone cuts average handle time by 30-40 seconds.

Pattern 5: Background Jobs with Voice Control

An agent that kicks off long-running tasks (research, data processing, report generation) and reports back to the caller when they are done. Voice becomes the control interface for work that happens in the background.

When to Use It

• Research tasks that take minutes to complete (web scraping, database analysis)
• Report generation where the caller wants to stay on the line for results
• Any workflow where the agent needs to do async work while keeping the caller engaged

Architecture

  Caller
    |
    v
  +------------------+     kicks off task     +-------------------+
  |  ControllerAgent | ─────────────────────> |  Background Task  |
  |  (voice UI)      |                        |  (async worker)   |
  +--------+---------+                        +--------+----------+
           |                                           |
    keeps caller engaged                       does heavy lifting
    with small talk / updates                  (API calls, DB queries,
           |                                    data processing)
           |                                           |
           +<──────────── result callback ─────────────+
           |
    reads back results

Key Code

The agent launches background work with asyncio.create_task, keeps the caller engaged, and reads back the results when they arrive.

import asyncio
from livekit.agents import Agent, RunContext, function_tool

class ResearchAgent(Agent):
    def __init__(self, **kwargs):
        super().__init__(
            instructions=(
                "You help users by running research tasks. When asked to "
                "research something, start the task and keep the caller "
                "engaged with brief conversation while waiting. When results "
                "arrive, summarize them clearly."
            ),
            **kwargs,
        )
        self._pending_task: asyncio.Task | None = None

    @function_tool()
    async def start_research(self, ctx: RunContext, query: str):
        """Start a background research task.

        Args:
            query: What to research.
        """
        self._pending_task = asyncio.create_task(
            self._do_research(query)
        )
        self._pending_task.add_done_callback(
            lambda t: asyncio.create_task(self._deliver_results(t))
        )
        return (
            "Research started. It will take about a minute. "
            "I will let you know as soon as I have results."
        )

    async def _do_research(self, query: str) -> str:
        """The actual long-running work."""
        import httpx
        async with httpx.AsyncClient() as client:
            # Call your research API, scrape data, query databases
            resp = await client.post(
                "https://your-api.com/research",
                json={"query": query},
                timeout=120,
            )
            return resp.json()["summary"]

    async def _deliver_results(self, task: asyncio.Task) -> None:
        """Called when research completes -- injects results into chat."""
        try:
            result = task.result()
        except Exception as e:
            result = f"Research failed: {e}"

        self.session.generate_reply(
            instructions=(
                f"The research is complete. Here are the results: {result}. "
                "Summarize the key findings for the caller."
            )
        )

    @function_tool()
    async def check_status(self, ctx: RunContext):
        """Check if the research task is still running."""
        if self._pending_task is None:
            return "No research task is running."
        if self._pending_task.done():
            return "Research is complete."
        return "Research is still running."

The pattern hinges on the done_callback. When the background task finishes, it calls session.generate_reply() with instructions that include the results. The agent interrupts whatever small talk it was making and delivers the findings. The caller does not need to ask. Results arrive automatically.

Note that generate_reply is not awaited in the callback. This is the same fire-and-forget pattern from Chapter 6. The reply generation runs asynchronously, and the agent speaks the results as soon as TTS produces the audio.

Choosing the Right Pattern

These patterns are not mutually exclusive. A production system often combines several. A phone receptionist (Pattern 1) might include a warm handoff path (Pattern 4) for escalations. A survey caller (Pattern 2) might use IVR navigation (Pattern 3) to dial through a company directory before reaching the survey target.

Start with the pattern closest to your use case, get it working end-to-end with a single call, then layer in the production hardening from Chapters 10-12. The architecture is the easy part. The hard part is the prompt engineering, error handling, and edge cases that only surface when real humans call your agent.