I want to use pydantic-ai as my agent/llm, how can I integrate it with agent session and successfully get STT transcibtion and then send text to TTS

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

1. for stt I plan to use built-in cartesia module with cartesia ink model 2. as for tts I plat to use built-in 11labs module 3. previously I used raw livekit-agents framework, as for now I want to continue using this framework, but to have a custom pydantic ai agent that seamlessly integrates into this setup 4. yep, realtime, like right now in livekit-agents I want to stay on the

livekit-agents

framework, but replace the built-in LLM block with a custom one based on

pydantic-ai

, while keeping everything working seamlessly.

Thanks — this is clear and doable. Below is a concise integration plan plus focused code examples showing how to replace the LLM block with a pydantic-ai agent while keeping Cartesia STT and ElevenLabs TTS in real time. What I’m answering: how to override the LLM node in your existing livekit-agents Agent so pydantic-ai handles generation (streaming if possible), with Cartesia for STT and ElevenLabs for TTS. High-level plan • Keep your existing AgentSession configured with Cartesia STT and ElevenLabs TTS. • Subclass Agent and override

llm_node(...)

so it calls your pydantic-ai runner. • Ensure

llm_node

returns an async iterable that yields

ChatChunk

(or strings) so the pipeline stays real-time compatible. • If pydantic-ai can stream tokens, yield chunks incrementally; if not, yield a single ChatChunk with the full response. Minimal example (adapt to your codebase) 1) pydantic-ai runner (async helper)

# my_pydantic_agent.py
from pydantic_ai import Runner

runner = Runner(...)  # your pydantic-ai setup

async def run_pydantic(prompt: str) -> str:
    # call your pydantic-ai runner and return full text (or provide a streaming API)
    result = await runner.run(prompt)
    return result.data  # adjust to your runner's response shape

2) Custom Agent overriding llm_node

# custom_agent.py
import asyncio
from typing import AsyncIterable
from livekit.agents.llm import ChatChunk, ChatContext, ModelSettings, FunctionTool
from livekit.agents import Agent
from my_pydantic_agent import run_pydantic

class PydanticAgent(Agent):
    def __init__(self, instructions: str = ""):
        super().__init__(instructions=instructions)

    async def llm_node(
        self,
        chat_ctx: ChatContext,
        tools: list[FunctionTool],
        model_settings: ModelSettings,
    ) -> AsyncIterable[ChatChunk | str]:
        # Build a prompt from chat context (system + history + latest user)
        prompt = "\n".join(
            f"{m.role.upper()}: {m.text_content}" for m in chat_ctx.messages if m.text_content
        )

        # If pydantic-ai supports streaming, iterate and yield incremental ChatChunks.
        # If not, await full response and yield a single ChatChunk.
        try:
            response = await run_pydantic(prompt)
        except Exception as e:
            response = f"Error: {str(e)}"

        # Simple non-streaming yield (works if pydantic-ai returns full text)
        yield ChatChunk(text=response)

3) AgentSession wiring (Cartesia STT + ElevenLabs TTS)

# session_setup.py
from livekit.agents import AgentSession
from livekit.plugins import cartesia, elevenlabs

def create_session():
    session = AgentSession(
        stt=cartesia.STT(model="ink-whisper"),    # Cartesia ink model
        tts=elevenlabs.TTS(voice_id="your_voice_id"),
        # vad optional: use silero or built-in VAD if desired
    )
    return session

# usage
session = create_session()
agent = PydanticAgent(instructions="You are a helpful assistant.")
await session.start(agent=agent, room=room)

Streaming notes • If pydantic-ai exposes incremental outputs/tokens, change

run_pydantic

to be an async generator and in

llm_node

iterate and

yield ChatChunk(text=partial)

for each partial piece to preserve low-latency TTS playback. • If pydantic-ai only returns full text, yielding a single

ChatChunk

still works; TTS will synthesize after the full response is ready. Error handling & function tools • Wrap calls to pydantic-ai in try/except and yield an error ChatChunk to avoid pipeline crashes. • If you need function/tool calls, translate pydantic-ai’s function-call outputs into the framework’s `FunctionTool`/`ChatChunk` conventions; embed the tool results back into subsequent prompts as needed. Next step Would you like: • a streaming example using an async generator from pydantic-ai (if your runner supports it), or • a complete ready-to-run example file that plugs into your current livekit-agents project structure? Which would you prefer? Sources: Tool definition and use | LiveKit docs | Pipeline nodes and hooks | LiveKit docs | LLM Output Replacement | LiveKit docs | Models | LiveKit Docs | Agents (Node.js) | livekit/livekit_composite | DeepWiki

Is it okay that I do not have llm inside init of AgentSession? And then initialize PydanticAgent and use it in session start session.start(agent=agent, room=room) ?

Yes — that's fine. You do not need to pass an LLM into AgentSession at construction. Create the AgentSession with STT (Cartesia), TTS (ElevenLabs), VAD, etc., then instantiate your PydanticAgent and pass it to

session.start(agent=agent, room=room)

. The framework expects the agent (which overrides

llm_node

) at start time and will wire it into the real-time pipeline. Quick reminders: • Ensure your agent's

llm_node

returns an async iterable of

ChatChunk

(or strings) so streaming and low-latency TTS work. • Handle errors inside

llm_node

(catch exceptions and yield an error chunk) to avoid pipeline crashes. • If you want incremental TTS playback, make

llm_node

stream partial outputs when pydantic-ai supports it. If you want, I can show the exact minimal code snippet that instantiates AgentSession (no LLM) and calls

session.start(...)

with your PydanticAgent. Which snippet would you prefer: non-streaming or streaming?

streaming

I don't have the answer you're looking for. You could also try asking your question: • in one of the other Slack channels or • to https://deepwiki.com/livekit/livekit_composite which is trained on all LiveKit source code If you find the answer, please post it here to help others!

Here is example from your sources: import logging from dotenv import load_dotenv from livekit.agents import JobContext, JobProcess, AgentServer, cli, Agent, AgentSession from livekit.plugins import openai, deepgram, silero load_dotenv() logger = logging.getLogger("replacing-llm-output") logger.setLevel(logging.INFO) class SimpleAgent(Agent): def __init__(self) -> None: super().__init__( instructions="You are a helpful agent." ) self._llm = openai.LLM.with_groq(model="deepseek-r1-distill-llama-70b") async def on_enter(self): self.session.generate_reply() async def llm_node(self, chat_ctx, tools, model_settings=None): async def process_stream(): async with self._llm.chat(chat_ctx=chat_ctx, tools=tools, tool_choice=None) as stream: async for chunk in stream: if chunk is None: continue content = getattr(chunk.delta, 'content', None) if hasattr(chunk, 'delta') else str(chunk) if content is None: yield chunk continue processed_content = content.replace("<think>", "").replace("</think>", "Okay, I'm ready to respond.") print(f"Original: {content}, Processed: {processed_content}") if processed_content != content: if hasattr(chunk, 'delta') and hasattr(chunk.delta, 'content'): chunk.delta.content = processed_content else: chunk = processed_content yield chunk return process_stream() server = AgentServer() def prewarm(proc: JobProcess): proc.userdata["vad"] = silero.VAD.load() server.setup_fnc = prewarm @breezy-insurance-66772.rtc_session() async def entrypoint(ctx: JobContext): ctx.log_context_fields = {"room": ctx.room.name} session = AgentSession( stt=deepgram.STT(), tts=openai.TTS(), vad=ctx.proc.userdata["vad"], preemptive_generation=True, ) await session.start(agent=SimpleAgent(), room=ctx.room) await ctx.connect() if name == "__main__": cli.run_app(server) In this example you use custom llm agent, who rewrites llm_node func. And adds almost empty

async def on_enter(self):
        self.session.generate_reply()

why do I need on_enter function in my custom agent? How does it work with my custom agent? btw every llm invocation in my whole agents goes tghrough llm_node function?