How do you integrate a custom llm node into an agent?

from livekit.agents import ModelSettings, llm, FunctionTool, Agent
from typing import AsyncIterable

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

https://docs.livekit.io/agents/build/nodes/#stt-llm-tts-pipeline-nodes Does it replace the llm argument in Agent?

class RAGAgent(Agent):
    def __init__(self, send_text_fn):
        super().__init__(
            instructions="not-needed",
            stt=openai.STT.with_azure(
                model=settings.stt_model_small,
                azure_endpoint=settings.azure_openai_endpoint,
                api_key=settings.azure_openai_api_key,
                api_version=settings.openai_api_version,
                prompt="The audio is in Norwegian. Transcribe it in Norwegian.",
            ),
            llm=openai.LLM.with_azure(
                model=settings.text_model_small,
                azure_endpoint=settings.azure_openai_endpoint,
                api_key=settings.azure_openai_api_key,
                api_version=settings.openai_api_version,
            ),
            tts=openai.TTS.with_azure(
                model=settings.tts_model,
                azure_endpoint=settings.azure_openai_endpoint,
                api_key=settings.azure_openai_api_key,
                api_version=settings.openai_api_version,
                voice="nova",
            )
        )
        self.send_text_fn = send_text_fn

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

The remote does not support streaming, so i'm just supplying the whole message. How would you do that?

You're supplying the full response from a non‑streaming remote. Override

llm_node

to call the remote once, convert its full reply into a single

llm.ChatChunk

, and yield it. Below is a minimal pattern you can follow.

from livekit.agents import llm, FunctionTool, ModelSettings, Agent
from typing import AsyncIterable
import asyncio

class RAGAgent(Agent):
    async def llm_node(
        self,
        chat_ctx: llm.ChatContext,
        tools: list[FunctionTool],
        model_settings: ModelSettings
    ) -> AsyncIterable[llm.ChatChunk]:
        # 1) Prepare input (your preprocessing)
        prompt = chat_ctx.to_prompt_string()  # or build prompt manually

        # 2) Call non-streaming remote LLM once
        full_response_text = await call_non_streaming_llm(prompt, model_settings)

        # 3) Convert to a single ChatChunk and yield
        chunk = llm.ChatChunk(
            text=full_response_text,
            role="assistant",
            done=True
        )
        yield chunk

# Example helper - replace with your remote call
async def call_non_streaming_llm(prompt: str, model_settings: ModelSettings) -> str:
    # synchronous remote call wrapped for async, or use async HTTP client directly
    return await asyncio.to_thread(lambda: my_remote_llm_sync_call(prompt, model_settings))

Notes • Keep

llm

in the Agent constructor if you still want the built‑in LLM available elsewhere; you do not have to use it inside this override. • Ensure the yielded

ChatChunk

has

done=True

(and valid

text

) so downstream nodes treat it as a complete response. • If you need streaming-compatible behavior for consumers, you can split

full_response_text

into multiple chunks and yield them sequentially. Would you like a concrete example for your specific remote API (sync HTTP vs async client) or for splitting a long response into multiple chunks?

According to my livekit imported code, the ChatChunk has these fields:

class ChatChunk(BaseModel):
    id: str
    delta: ChoiceDelta | None = None
    usage: CompletionUsage | None = None

Thanks — you're right about the

ChatChunk

shape. Responding to the last two points: 1) Does the custom

llm_node

replace the

llm

argument? • No. Keep providing an

llm

in the Agent constructor if you want the framework’s default LLM available. Overriding

llm_node

lets you wrap or replace the default behavior. If you call

Agent.default.llm_node(...)

it will use the

llm

you passed; if you do not call it, you must implement the LLM call yourself. 2) How to return a single full response from a non‑streaming remote when

ChatChunk

has

id

,

delta

,

usage

• Make your non‑streaming call, create a single

ChatChunk

with a

ChoiceDelta

that contains the assistant text and role, set an

id

, and yield that chunk. Minimal example matching your model:

import uuid
from typing import AsyncIterable
from livekit.agents import Agent, llm, FunctionTool, ModelSettings
from livekit.agents.llm import ChoiceDelta, CompletionUsage  # adjust import path as in your env

class RAGAgent(Agent):
    async def llm_node(
        self,
        chat_ctx: llm.ChatContext,
        tools: list[FunctionTool],
        model_settings: ModelSettings
    ) -> AsyncIterable[llm.ChatChunk]:
        # 1) build prompt / preprocess
        prompt = chat_ctx.to_prompt_string()

        # 2) call your non-streaming remote (sync or async)
        full_text = await call_non_streaming_llm(prompt, model_settings)

        # 3) wrap into ChoiceDelta and ChatChunk, then yield once
        delta = ChoiceDelta(content=full_text, role="assistant")
        usage = CompletionUsage(  # optional, fill if you have tokens/metrics
            prompt_tokens=0,
            completion_tokens=0,
            total_tokens=0
        )
        chunk = llm.ChatChunk(id=str(uuid.uuid4()), delta=delta, usage=usage)
        yield chunk

Notes • If you need downstream consumers to see streaming, split

full_text

into multiple

ChoiceDelta

contents and yield several `ChatChunk`s in sequence. • You may omit

usage

if you don't have token metrics. • Keep the

llm

in the constructor unless you fully replace LLM usage everywhere. Do you want an example for a specific remote API (sync HTTP vs async client) or code to split the text into multiple chunks? Sources: Pipeline nodes and hooks | LiveKit docs