https://livekit.io logo
Join Slack
Powered by
how to get the llm generated response before it is...
# ask-ai
a
how to get the llm generated response before it is going to tts using session.on event
✅ 1
t
Thanks for reaching out, Harshita. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — you want to capture the LLM-generated text right before it’s sent to TTS using the session flow. I’m responding to that specific part of your question. Short answer: implement a custom Agent and override its 
llm_node
to intercept the streaming LLM output. Inside 
llm_node
you can read, modify, log, or forward each chunk before yielding it onward to the rest of the pipeline (including TTS). What this looks like (conceptual steps): 1. Create a custom Agent class that inherits from 
Agent
. 2. Override the 
llm_node
method. Inside it, stream the LLM results and inspect each chunk’s content. 3. Take whatever action you need (store the text, modify it, emit a session event, etc.), then 
yield
the chunk so the pipeline continues to TTS. 4. Use your custom agent when creating the 
AgentSession
. Example pattern (Python-like pseudocode you can adapt):
Copy code
from livekit.agents import Agent, ModelSettings, llm

class MyAgent(Agent):
    async def llm_node(self, chat_ctx: llm.ChatContext, tools: list[llm.FunctionTool], model_settings: ModelSettings) -> llm.ChatChunk:
        # Stream LLM output
        async for chunk in self.llm.chat(chat_ctx=chat_ctx, tools=tools, tool_choice=None):
            # Extract text from the streaming chunk
            content = None
            if hasattr(chunk, "delta"):
                content = getattr(chunk.delta, "content", None)
            else:
                content = str(chunk)

            if content:
                # Example actions:
                # - log or save the content
                # - emit a session event (session.on handler) with the content
                # - modify content before TTS (if desired)
                print("LLM generated content:", content)
                # If you want to notify the session layer, call whatever session/event API you have here

            # Yield the chunk so TTS receives the (possibly modified) output
            yield chunk
Notes and tips • The 
llm_node
gets the stream as it’s produced; this is the single best place to intercept text before TTS. • If you need to trigger a 
session.on
style event, emit that event from inside 
llm_node
when you see content (how to emit depends on your session/event API — if you share which SDK or session object you’re using, I can show a concrete example). • If you want to modify the text before TTS, change the chunk contents before yielding. Be careful to preserve chunk format expected downstream. If you’d like, tell me: • Which SDK/language you’re using (Python, TypeScript, etc.). • The exact session/event API you want to call (
session.on(...)
handler shape or method). I’ll provide a concrete code example showing how to emit a session event with the LLM text before it reaches TTS. Sources: Modifying LLM output before TTS | LiveKit Documentation | Pipeline nodes and hooks | LiveKit Documentation
a
can't I directly use session.on events ?
t
Thanks — I’m answering the part where you asked whether you can use 
session.on
events directly to capture the LLM response before it goes to TTS. Short answer: you cannot rely on 
session.on
alone to intercept the raw LLM stream before TTS. The point in the pipeline where the LLM produces text is inside the agent flow (the 
llm_node
). 
session.on
events are useful for observing higher‑level session state and events, but they don’t provide a guaranteed hook into the streaming LLM chunks before TTS consumes them. What you should do instead • Use a custom Agent and override 
llm_node
to read the streaming chunks from the LLM. That is the place where you can inspect, modify, log, or emit your own events with the generated text before yielding chunks onward to TTS. • From inside 
llm_node
you can emit a session-level event (call whatever 
session.emit
/ event API your SDK exposes) so your existing 
session.on
handlers receive the LLM text in real time. This keeps 
session.on
for handling the event while 
llm_node
is the interceptor. Why this approach • 
llm_node
is the earliest reliable point that receives the text stream from the LLM. • 
session.on
handlers run at the session/event layer; without an explicit emit from the agent layer they won’t receive the raw streaming chunks before TTS consumes them. If you want, I can give a concrete example that: 1. Overrides 
llm_node
in a custom Agent, 2. Emits a session event with each chunk’s text (so your 
session.on("llm_chunk", ...)
handler receives it), and 3. Optionally modifies the chunk before yielding so TTS gets the altered text. Tell me which SDK/language you’re using (Python or TypeScript) and the shape of your 
session
event API (e.g., how you emit events), and I’ll provide the exact code you can drop into your project.
a
suppose i have an agent in entrypoint .. based on what agent is saying before going to tts(cartesia) I want to change the emotion of the agent options [angry, excited, happy] based on the words in agent llm text. so this should be dynamic emotion change based on what agent is going to speek next. no function tool should be involved here
t
Thanks — I’m responding to the part where you asked how to change the agent’s TTS emotion (angry, excited, happy) dynamically based on the LLM text right before it’s sent to Cartesia, with no function tools involved. Short answer: yes — do this by overriding the Agent’s llm_node to intercept the LLM stream, detect emotion from the text, and call your session TTS update method (for example 
self.session.tts.update_options(emotion=...)
) so the next TTS utterance uses the chosen emotion. You cannot rely on 
session.on
alone to catch the raw streaming chunks before TTS — 
llm_node
is the reliable interception point. Concrete plan 1. Override 
llm_node
in a custom Agent subclass. 2. Stream LLM chunks, buffer until a sensible utterance boundary (sentence end or newline). 3. Run simple keyword/regex matching or an emotion classifier on the buffered utterance to pick one of [angry, excited, happy]. 4. Call the session’s TTS update (e.g., 
self.session.tts.update_options(emotion=detected)
) so the next utterance is synthesized with that emotion. 5. Yield the original (or modified) chunks so the pipeline continues to TTS. Important notes • Emotion changes take effect for the next TTS utterance; they don’t reliably change mid-sentence. Design your buffer/utterance detection accordingly. • No function tools are required — all work happens inside 
llm_node
. • Use a conservative sentence boundary strategy (punctuation/newline) to avoid switching emotion too often. • If you prefer ML-based emotion classification, you can call a local classifier inside 
llm_node
instead of keyword matching. If you want, I can provide a ready-to-drop example in either Python or TypeScript: • Python example will show buffering, keyword detection, and 
self.session.tts.update_options(...)
. • TypeScript example will show the same pattern for a JS Agent and how to emit session events if you want 
session.on
to also receive the detected emotion. Which language would you like the example in (Python or TypeScript)? Also tell me how you’d like to detect emotion (simple keywords or a classifier), and I’ll give the exact code you can use. Sources: Cartesia TTS | LiveKit Documentation | Hume TTS plugin guide | LiveKit Documentation | Pipeline nodes and hooks | LiveKit Documentation | LLM Output Replacement | LiveKit Documentation
a
can you tell generate a dummy example code
t
Thanks — I’ll show a simple, dummy example you can drop in to test dynamic emotion switching. This example uses Python, overrides 
llm_node
to inspect streaming LLM text, detects emotion with keyword matching (angry, excited, happy), updates the session TTS options via 
self.session.tts.update_options(...)
, and yields chunks so the pipeline continues to Cartesia TTS. Notes: • Emotion changes apply to the next utterance (not mid-sentence). • This is a minimal, illustrative example — adapt buffering, detection logic, and TTS option names to your codebase and Cartesia settings. Example (Python)
Copy code
import re
from typing import AsyncIterable
from livekit.agents import Agent, AgentSession, ModelSettings, llm

class DummyEmotiveAgent(Agent):
    async def llm_node(
        self,
        chat_ctx: llm.ChatContext,
        tools: list[llm.FunctionTool],
        model_settings: ModelSettings,
    ) -> AsyncIterable[llm.ChatChunk]:
        # Simple keyword lists for emotions
        emotion_map = {
            "angry": ["angry", "furious", "outraged", "annoyed", "frustrated"],
            "excited": ["excited", "amazing", "incredible", "wow", "thrilled", "awesome"],
            "happy": ["happy", "glad", "great", "wonderful", "pleased", "fantastic"],
        }

        def detect_emotion(text: str) -> str | None:
            t = text.lower()
            for emotion, kws in emotion_map.items():
                for kw in kws:
                    if re.search(rf"\b{re.escape(kw)}\b", t):
                        return emotion
            return None

        # Stream the LLM as usual
        stream = self.llm.chat(chat_ctx=chat_ctx, tools=tools)
        buffer = ""

        async for chunk in stream:
            # If the chunk has streaming content, append to buffer
            content = None
            if hasattr(chunk, "delta") and getattr(chunk.delta, "content", None):
                content = chunk.delta.content
            elif hasattr(chunk, "content"):
                content = chunk.content

            if content:
                buffer += content

                # Check for a sentence boundary (simple heuristic)
                if any(p in buffer for p in (".", "?", "!", "\n")):
                    # Extract up to last end punctuation
                    last_end = max((buffer.rfind(c) for c in ".?!\n" if c in buffer))
                    utterance = buffer[: last_end + 1].strip()

                    # Detect emotion for this utterance
                    detected = detect_emotion(utterance)
                    if detected:
                        # Update session TTS options so the NEXT utterance uses this emotion
                        # (adjust the method name / params if your SDK differs)
                        try:
                            self.session.tts.update_options(emotion=detected)
                            print(f"Set TTS emotion -> {detected} for: {utterance[:60]}...")
                        except Exception as e:
                            print("Failed to update TTS options:", e)

                    # Clear processed part of buffer
                    buffer = buffer[last_end + 1 :]

            # Yield the chunk onward to TTS (unaltered here)
            yield chunk

        # After stream ends, check remaining buffer
        if buffer.strip():
            detected = detect_emotion(buffer)
            if detected:
                try:
                    self.session.tts.update_options(emotion=detected)
                    print(f"Set TTS emotion -> {detected} for trailing text...")
                except Exception as e:
                    print("Failed to update TTS options for trailing text:", e)
            # Optionally yield a final synthetic chunk if required by your pipeline

# Entrypoint: create your AgentSession with the custom agent
session = AgentSession(
    agent=DummyEmotiveAgent(),
    # stt=..., (if needed)
    llm=llm.InferenceLLM(model="openai/gpt-4o"),  # example
    tts=llm.InferenceTTS(
        model="cartesia/sonic-3",
        voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc",
        extra_kwargs={"emotion": "neutral"},  # initial emotion
    ),
    # other session configs...
)

# Start the session in your app as you normally would
# session.run() or equivalent per your framework
If you want: • A TypeScript/Node.js version of this example. • A slightly more robust utterance detector (e.g., pause-based VAD or shorter buffer thresholds). • Replacement of keyword matching with a small classifier (I can show a lightweight approach). Which would help you next?
a
llm_node is not running when i include it in entrypoint
Copy code
async def llm_node(self,chat_ctx: llm.ChatContext,tools: list[llm.FunctionTool],model_settings: ModelSettings,) -> AsyncIterable[llm.ChatChunk]:
    <http://logger.info|logger.info>(f"LLM Node :::")
    emotion_map = {
        "angry": ["can not", "warning"],
        "excited": ["excited", "amazing", "incredible", "wow", "thrilled", "awesome"],
        "happy": ["happy", "glad", "great", "wonderful", "pleased", "fantastic"],
    }

    def detect_emotion(text: str) -> str | None:
        <http://logger.info|logger.info>(f"detecting emotion for : {text}")
        t = text.lower()
        for emotion, kws in emotion_map.items():
            for kw in kws:
                if re.search(rf"\b{re.escape(kw)}\b", t):
                    <http://logger.info|logger.info>(f"detected emotion: {emotion}")
                    return emotion
        return None

    stream = self.llm.chat(chat_ctx=chat_ctx, tools=tools)
    buffer = ""

    async for chunk in stream:
        # If the chunk has streaming content, append to buffer
        content = None
        if hasattr(chunk, "delta") and getattr(chunk.delta, "content", None):
            content = chunk.delta.content
        elif hasattr(chunk, "content"):
            content = chunk.content

        if content:
            buffer += content

            # Check for a sentence boundary (simple heuristic)
            if any(p in buffer for p in (".", "?", "!", "\n")):
                # Extract up to last end punctuation
                last_end = max((buffer.rfind(c) for c in ".?!\n" if c in buffer))
                utterance = buffer[: last_end + 1].strip()

                # Detect emotion for this utterance
                detected = detect_emotion(utterance)
                if detected:
                    # Update session TTS options so the NEXT utterance uses this emotion
                    # (adjust the method name / params if your SDK differs)
                    try:
                        self.session.tts.update_options(emotion=detected)
                        print(f"Set TTS emotion -> {detected} for: {utterance[:60]}...")
                    except Exception as e:
                        print("Failed to update TTS options:", e)

                # Clear processed part of buffer
                buffer = buffer[last_end + 1:]

        # Yield the chunk onward to TTS (unaltered here)
        yield chunk
t
Thanks — I understand you’ve added 
llm_node
in your entrypoint but it’s not running. I’ll help you troubleshoot and give a minimal reproducible pattern so you can confirm where the problem is. Quick checklist (most common causes) • Confirm your custom Agent class is actually being used when creating the session (the session must be constructed with your agent instance). • Verify your Agent class correctly inherits from the Agent base class in the same SDK version you’re running. • Make sure the method signature exactly matches the expected one for your SDK (async, same parameter names/types, and return type). • Ensure the pipeline path actually routes through 
llm_node
(some configurations or shortcuts may bypass agent nodes). • Add logging at Agent construction and the entrypoint that creates the AgentSession so you can confirm the runtime is instantiating your class. Targeted questions to narrow this down 1. How are you creating the AgentSession / entrypoint? Please paste the code where you instantiate your Agent and create the session. 2. Which SDK and version are you running (Python package name and version)? 3. Are there any errors or warnings in logs when the session starts? 4. Are you using a custom entrypoint or any middleware that might bypass agent nodes? Minimal reproducible example (pattern to follow) 1. Define a custom Agent subclass with an obvious log in 
__init__
and 
llm_node
. 2. Instantiate that agent and pass it to AgentSession. 3. Start the session and exercise the pipeline so the LLM produces output. Example skeleton (adapt to your imports / SDK):
Copy code
import asyncio
import logging
from livekit.agents import Agent, AgentSession, ModelSettings, llm  # adjust imports if needed

logging.basicConfig(level=<http://logging.INFO|logging.INFO>)
logger = logging.getLogger("test")

class MyAgent(Agent):
    def __init__(self):
        super().__init__(instructions="Hello")
        <http://logger.info|logger.info>("MyAgent constructed")

    async def llm_node(self, chat_ctx: llm.ChatContext, tools: list[llm.FunctionTool], model_settings: ModelSettings):
        <http://logger.info|logger.info>("llm_node STARTED")
        stream = self.llm.chat(chat_ctx=chat_ctx, tools=tools)
        async for chunk in stream:
            # simple log to prove we got chunks
            content = getattr(chunk.delta, "content", None) if hasattr(chunk, "delta") else getattr(chunk, "content", None)
            if content:
                <http://logger.info|logger.info>("llm chunk content: %s", content[:80])
            yield chunk
        <http://logger.info|logger.info>("llm_node FINISHED")

async def run_session():
    agent = MyAgent()

    session = AgentSession(
        agent=agent,
        llm=llm.InferenceLLM(model="openai/gpt-4o"),  # example
        tts=llm.InferenceTTS(model="cartesia/sonic-3", voice="...", extra_kwargs={"emotion":"neutral"}),
    )

    <http://logger.info|logger.info>("Starting session run")
    await session.run()  # or whatever entry method your framework uses
    <http://logger.info|logger.info>("Session run complete")

if __name__ == "__main__":
    asyncio.run(run_session())
What to inspect after running this: • Do you see "MyAgent constructed" in logs? If not, your agent object wasn’t instantiated. • Do you see "llm_node STARTED"? If yes, llm_node is running and the issue is in your content handling logic. • If you never see "llm_node STARTED" but you do see session start logs, then the agent nodes are being bypassed — share the session config and I can check for reasons. Additional debugging tips • Temporarily remove buffering logic and simply log and yield every chunk to confirm it runs. • If your SDK offers a “pipeline” or graph visualization/debug mode, enable it to confirm 
llm_node
is included. • Confirm there aren’t multiple Agent classes and you’re editing a different class than the one used at runtime. • If you use dependency injection or an entrypoint factory, ensure it returns your Agent subclass instance. If you share: • The code where you create AgentSession (the instantiation), • Any startup logs you have, I’ll review it and point out exactly what to change. If you prefer, paste a small snippet of your entrypoint and I’ll produce a corrected version you can run immediately.
a
im getting this error "message": "Error in _llm_inference_task\nTraceback (most recent call last):\n File \"/Users/harshita.patil/whizdm-dev/livekit-agent/.venv/lib/python3.11/site-packages/livekit/agents/utils/log.py\", line 17, in async_fn_logs\n return await fn(*args, **kwargs)\n ^^^^^^^^^^^^^^^^^^^^^^^^^\n File \"/Users/harshita.patil/whizdm-dev/livekit-agent/.venv/lib/python3.11/site-packages/opentelemetry/util/_decorator.py\", line 71, in async_wrapper\n return await func(*args, **kwargs) # type: ignore\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File \"/Users/harshita.patil/whizdm-dev/livekit-agent/.venv/lib/python3.11/site-packages/livekit/agents/voice/generation.py\", line 138, in _llm_inference_task\n async for chunk in llm_node:\n File \"/Users/harshita.patil/whizdm-dev/livekit-agent/agents/factory/call_actions_factory.py\", line 300, in llm_node\n stream = self.llm.chat(chat_ctx=chat_ctx, tools=tools)\n ^^^^^^^^^^^^^\nAttributeError: 'NotGiven' object has no attribute 'chat'", "level": "ERROR", "name": "livekit.agents", "exc_info": "Traceback (most recent call last):\n File \"/Users/harshita.patil/whizdm-dev/livekit-agent/.venv/lib/python3.11/site-packages/livekit/agents/utils/log.py\", line 17, in async_fn_logs\n return await fn(*args, **kwargs)\n ^^^^^^^^^^^^^^^^^^^^^^^^^\n File \"/Users/harshita.patil/whizdm-dev/livekit-agent/.venv/lib/python3.11/site-packages/opentelemetry/util/_decorator.py\", line 71, in async_wrapper\n return await func(*args, **kwargs) # type: ignore\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File \"/Users/harshita.patil/whizdm-dev/livekit-agent/.venv/lib/python3.11/site-packages/livekit/agents/voice/generation.py\", line 138, in _llm_inference_task\n async for chunk in llm_node:\n File \"/Users/harshita.patil/whizdm-dev/livekit-agent/agents/factory/call_actions_factory.py\", line 300, in llm_node\n stream = self.llm.chat(chat_ctx=chat_ctx, tools=tools)\n ^^^^^^^^^^^^^\nAttributeError: 'NotGiven' object has no attribute 'chat'", "pid": 15120, "job_id": "AJ_WjLHz5omi85j", "room_id": "RM_AmSqjKvp5suP", "timestamp": "2026-03-16T063734.978739+00:00"} {
t
Thanks — I see the error message and I’m here to help. You’re hitting: "AttributeError: 'NotGiven' object has no attribute 'chat'" This means 
self.llm
inside your 
llm_node
is not an LLM instance; it’s a placeholder value (
NotGiven
) used by the SDK when no LLM was supplied to the agent/session. In short: your Agent doesn’t have a real 
llm
assigned at runtime, so calling 
self.llm.chat(...)
fails. What to check and how to fix it 1. Ensure your AgentSession is constructed with an actual LLM object (not omitted). ◦ When creating the session, pass a proper LLM instance (e.g., 
llm.InferenceLLM(...)
), and ensure that instance is attached to the session that uses your Agent. ◦ Example:
Copy code
session = AgentSession(
         agent=MyAgent(),
         llm=llm.InferenceLLM(model="openai/gpt-4o"),
         tts=llm.InferenceTTS(...),
         ...
     )
• If you omit 
llm=...
, the SDK will leave 
self.llm
as 
NotGiven
. 1. Confirm you’re using the same Agent instance in the session ◦ Make sure the Agent you subclass and edit is the one you pass into 
AgentSession(agent=...)
. ◦ Add a constructor log in your Agent to confirm instantiation (helps rule out multiple classes/instances). 1. Verify SDK usage pattern / signatures ◦ Ensure the SDK version you’re running expects 
llm
passed to 
AgentSession
(API may differ across versions). ◦ If your project uses dependency injection or a factory to create sessions, confirm the factory is supplying the LLM field. 1. Quick minimal test to validate LLM presence ◦ Add a small check at the top of your `llm_node`:
Copy code
if getattr(self, "llm", None) is None or getattr(self, "llm", None).__class__.__name__ == "NotGiven":
         logger.error("llm not set on agent — check AgentSession creation")
         return
• This will surface the problem early and avoid the AttributeError stack trace. Corrected minimal pattern (complete flow) • Agent subclass with llm_node that uses self.llm.chat • AgentSession that includes llm=llm.InferenceLLM(...)
Copy code
class MyAgent(Agent):
    async def llm_node(self, chat_ctx: llm.ChatContext, tools: list[llm.FunctionTool], model_settings: ModelSettings):
        <http://logger.info|logger.info>("llm_node started")
        # quick guard
        if getattr(self, "llm", None) is None or getattr(self, "llm").__class__.__name__ == "NotGiven":
            logger.error("Agent has no llm configured")
            return

        stream = self.llm.chat(chat_ctx=chat_ctx, tools=tools)
        async for chunk in stream:
            # process and yield chunk
            yield chunk

# Session creation — ensure llm is provided
session = AgentSession(
    agent=MyAgent(),
    llm=llm.InferenceLLM(model="openai/gpt-4o"),   # required
    tts=llm.InferenceTTS(model="cartesia/sonic-3", voice="...", extra_kwargs={"emotion":"neutral"}),
)
await session.run()
If you still see 
NotGiven
after passing 
llm=...
, tell me: • The code you use to instantiate AgentSession (paste the snippet). • The version of the LiveKit agents package (pip show livekit-agents or pyproject/requirements). • Any factories or wrappers used to build the session (they may override/omit fields). If you paste your AgentSession creation snippet, I’ll check it and give an exact corrected example you can run.
a
give githug documentation of these latencies
t
Do you mean GitHub (not “githug”)? I’ll assume GitHub — here are the best places to find LiveKit agent latency info and related code. Official docs (latency/metrics) • Understand and improve agent latency: https://livekit.com/blog/understand-and-improve-agent-latency • Build your first AI voice agent (includes target latencies): https://livekit.com/blog/build-your-first-ai-voice-agent-python • Agent observability / per-turn latency metrics: https://docs.livekit.io/deploy/observability/insights/ • Metrics reference / per-turn latency fields: https://docs.livekit.io/deploy/observability/data/#per-turn-latency Relevant GitHub repos (examples & implementation) • livekit/agents (core agent SDK and examples): https://github.com/livekit/agents • livekit/agents-js (JavaScript SDK & examples): https://github.com/livekit/agents-js • livekit/knowledge_base (examples, recipes): https://github.com/livekit/knowledge_base Notes • There isn’t a single GitHub “latency docs” page; latency guidance and metric names (e2e_latency, llm_node_ttft, tts_node_ttfb, etc.) are documented in the LiveKit docs and blog posts linked above. Use the GitHub repos for code examples and the docs for metric definitions and target numbers. Would you like direct links to specific files in the GitHub repos (e.g., the agent telemetry/metrics implementation) or a short snippet showing how to read per-turn latency from 
ChatMessage.metrics
? Sources: Understand and Improve Agent Latency | LiveKit | Build Your First AI Voice Agent in Python: Complete Tutorial | LiveKit | Build Your First AI Voice Agent in Python: Complete Tutorial | LiveKit, with interactive elements
4 Views
Open in Slack
PreviousNext
Powered by