How to customize the LLM node. This is my AgentSession:

session = AgentSession(
        # Speech-to-text (STT) is your agent's ears, turning the user's speech into text that the LLM can understand
        # See all available models at <https://docs.livekit.io/agents/models/stt/>
        stt=inference.STT(model="assemblyai/universal-streaming", language="en"),
        # A Large Language Model (LLM) is your agent's brain, processing user input and generating a response
        # See all available models at <https://docs.livekit.io/agents/models/llm/>
        llm=inference.LLM(model="openai/gpt-4.1-mini"),
        # Text-to-speech (TTS) is your agent's voice, turning the LLM's text into speech that the user can hear
        # See all available models as well as voice selections at <https://docs.livekit.io/agents/models/tts/>
        tts=inference.TTS(
            model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"
        ),
        # VAD and turn detection are used to determine when the user is speaking and when the agent should respond
        # See more at <https://docs.livekit.io/agents/build/turns>
        turn_detection=MultilingualModel(),
        vad=ctx.proc.userdata["vad"],
        # allow the LLM to generate a response while waiting for the end of turn
        # See more at <https://docs.livekit.io/agents/build/audio/#preemptive-generation>
        preemptive_generation=True,
    )

async def llm_node(
    self,
    chat_ctx: llm.ChatContext,
    tools: list[FunctionTool],
    model_settings: ModelSettings
) -> AsyncIterable[llm.ChatChunk]:
    # # Insert custom preprocessing here
    # async for chunk in Agent.default.llm_node(self, chat_ctx, tools, model_settings):
    #     # Insert custom postprocessing here
    #     yield chunk
    yield "I'm inevitable and I'm coming for you"

Thanks for reaching out, Rishabh. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

I want full replacement of the LLM and routing to a different LLM provider.

You're asking for a full replacement of the LLM and routing to a different provider. Below is a concise, runnable pattern you can adapt: override

llm_node

in a custom

Agent

, call your provider's streaming API, parse its stream, and yield

llm.ChatChunk

objects so the rest of the pipeline (TTS, tools, etc.) works normally. 1) Custom agent that replaces the LLM and streams from an external provider

from typing import AsyncIterable
import httpx
from livekit.agents import Agent, AgentSession, llm, ChatContext, FunctionTool, ModelSettings
from livekit.plugins import inference

class CustomLLMAgent(Agent):
    async def llm_node(
        self,
        chat_ctx: ChatContext,
        tools: list[FunctionTool],
        model_settings: ModelSettings
    ) -> AsyncIterable[llm.ChatChunk]:
        # Transform chat_ctx.messages to the provider format
        messages = [{"role": m.role, "content": m.content} for m in chat_ctx.messages]

        # Example: call a streaming-compatible API (replace URL/headers/payload)
        headers = {"Authorization": f"Bearer {YOUR_API_KEY}", "Accept": "text/event-stream"}
        payload = {"model": "your-model", "messages": messages, "stream": True}

        async with httpx.AsyncClient(timeout=None) as client:
            async with client.stream("POST", "<https://api.your-llm.com/v1/chat/completions>",
                                     json=payload, headers=headers) as resp:
                resp.raise_for_status()
                async for line in resp.aiter_lines():
                    if not line:
                        continue
                    # handle SSE or JSON-per-line streaming formats
                    # Example for "data: {...}" style (SSE):
                    if line.startswith("data: "):
                        data = line[len("data: "):]
                        if data.strip() == "[DONE]":
                            break
                        obj = json.loads(data)

                        # Extract content piece depending on provider response shape
                        # Adjust these fields to match your provider
                        delta_text = obj.get("choices", [{}])[0].get("delta", {}).get("content", "")

                        if delta_text:
                            yield llm.ChatChunk(
                                delta=llm.ChoiceDelta(content=delta_text, role="assistant"),
                                usage=llm.TokenUsage(input_tokens=0, output_tokens=len(delta_text.split()))
                            )

                # Optionally yield a final chunk or finish signal if needed

2) Start the session with your agent

session = AgentSession(
    stt=inference.STT(model="assemblyai/universal-streaming", language="en"),
    llm=inference.LLM(model="openai/gpt-4.1-mini"),  # llm plugin can remain but is unused
    tts=inference.TTS(model="cartesia/sonic-3", voice="9626c31c-..."),
    turn_detection=MultilingualModel(),
    vad=ctx.proc.userdata["vad"],
    preemptive_generation=True,
)

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

Notes and tips • Do not call `Agent.default.llm_node()`—bypassing it is required for full replacement. • Yield

llm.ChatChunk

objects (or strings) matching the streaming shape downstream expects. The example uses

ChoiceDelta

+

TokenUsage

. • Adjust parsing logic to your provider's stream format (SSE, newline-delimited JSON, etc.). Many providers use "data: {...}" SSE lines or JSON-per-line. • If your provider returns final non-streamed text, wrap it as a single

llm.ChatChunk

. • Keep an eye on timeouts and backpressure for long streams (use httpx streaming as shown). Would you like a tailored example for a specific provider (OpenAI-compatible base_url, Anthropic, or a self-hosted API)? If so, tell me the provider and the streaming format they return (SSE / JSON lines / chunked JSON) and I’ll adapt the code. Sources: Pipeline nodes and hooks | LiveKit docs | Large language models (LLM) | LiveKit Docs | Models | LiveKit Docs