In an the LLM response of a pipeline agent, is there a way I can perform a regex on the string, and retry for a new llm response if a certain condition is met?

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

how does this interact with preemptive generation?

You're asking how a custom regex-based retry of an LLM response would interact with preemptive generation. Good question — here’s a concise explanation and practical guidance. Short answer • Preemptive generation starts generation early from interim STT transcripts and can be cancelled if the final transcript differs. Your regex-retry logic can run on the generated text, but you need to decide whether to act on preemptive outputs, final outputs, or both. Preemptive responses may be canceled or replaced when the final transcript arrives. How they interact (practical points) • Timing: Preemptive outputs arrive earlier and may be provisional. If you run your regex on a preemptive response and trigger a retry, that retry might be wasted if the preemptive response is later canceled and a new generation is started from the final transcript. • Correctness: Validating only final responses avoids false positives caused by incomplete interim transcripts. Validating preemptive responses can reduce latency when the interim is already good enough, but increases complexity. • Cancellation: The platform cancels preemptive generation when the final transcript changes significantly. Your retry logic must handle cancellations (clean up in-flight retries and not leak state). • Retry target: Decide whether retries re-run the LLM on the same (interim) context or wait for/use the final transcript. Retrying on interim context risks repeating the same mistake; retrying after final transcript is more robust. • Retry limits and backoff: Always enforce a max attempts and simple backoff to avoid loops and resource waste, especially with preemptive flows. • Prompt adjustments: For effective retries, modify the prompt (e.g., explicit constraint or instruction) so subsequent generations are less likely to match the regex again. Recommended approaches • Conservative (safer): Only run regex + retries on the final confirmed transcript. Simpler and avoids wasted retries. • Latency-optimizing: Run regex on preemptive output; if it fails, attempt one quick retry on the interim; on final transcript, revalidate and, if needed, run a final retry using the final transcript. Ensure cancellation and cleanup logic. • Hybrid: Validate preemptive output; if it passes regex, deliver immediately; if it fails, hold briefly and prefer final-transcript retries before additional attempts. Would you like a short code sketch showing: • (A) validating only final responses, or • (B) validating preemptive responses with cancellation/cleanup and a max-retry policy? Which option fits your needs? Sources: Agent speech and audio | LiveKit docs | examples/voice_agents | livekit/livekit_composite | DeepWiki | Pipeline nodes and hooks | LiveKit docs

elaborate on how I would do the following: "Re-invoke the LLM generation (for example, call the same

llm_node

/ session generate method again) to get a new response.". write me an example llm_node function that implements the validation logic I am looking for

Thanks — I’ll show a concise example

llm_node

that buffers the assistant response, runs a regex validation, and retries up to a max number of attempts. This sketch assumes an async agent framework where

self.llm.chat(...)

yields stream chunks, you can add messages to

chat_ctx

, and you can yield chunks downstream. Adjust names to your codebase as needed.

import re
import asyncio

MAX_RETRIES = 3
VALIDATION_REGEX = re.compile(r"your_valid_pattern")  # match valid output; invert if you prefer invalid-pattern detection
RETRY_INSTRUCTION = "Please regenerate the response without using the disallowed pattern."

async def llm_node(self, chat_ctx, tools, model_settings=None):
    """
    Intercepts streaming LLM output, buffers full text, validates with regex,
    and retries generation up to MAX_RETRIES if validation fails.
    Yields the final accepted stream chunks downstream.
    """
    attempt = 0

    while attempt < MAX_RETRIES:
        attempt += 1
        buffered = ""
        stream_cancelled = False

        # Open a chat stream for this attempt
        async with self.llm.chat(chat_ctx=chat_ctx, tools=tools, model_settings=model_settings) as stream:
            try:
                # Stream chunks and buffer content
                async for chunk in stream:
                    # Example: chunk.delta.content contains incremental text when present
                    if hasattr(chunk, "delta") and getattr(chunk.delta, "content", None):
                        buffered += chunk.delta.content
                    # Forward streaming chunks downstream so UI/tts can start
                    yield chunk

            except asyncio.CancelledError:
                # Stream was cancelled externally (e.g., preemptive generation canceled).
                stream_cancelled = True
            except Exception:
                # If any stream error, decide to break/raise or retry
                stream_cancelled = True

        # If stream was cancelled, treat as non-final and retry only if appropriate
        if stream_cancelled:
            # Optionally do not count cancelled attempts against retry budget
            # attempt -= 1
            # Small delay to avoid tight loop
            await asyncio.sleep(0.1)
            continue

        # Validate the buffered final text
        valid = bool(VALIDATION_REGEX.search(buffered))
        if valid:
            # Accept: we've already forwarded the chunks during streaming, so just exit
            return
        else:
            # Validation failed: decide whether to retry
            if attempt >= MAX_RETRIES:
                # Reached limit — you can yield a final error message or raise to stop
                error_msg = "Unable to generate a valid response after retries."
                # Optionally yield a final assistant chunk with the error
                yield self._make_error_chunk(error_msg)
                return

            # Prepare chat context for retry:
            # 1) Add the failed assistant output as context (optional)
            chat_ctx.add_message(role="assistant", content=buffered)
            # 2) Add an instruction telling the model to avoid the invalid pattern
            chat_ctx.add_message(role="system", content=RETRY_INSTRUCTION)

            # Small backoff before retrying
            await asyncio.sleep(0.2)
            # Loop will re-invoke generation (create a new chat stream) with updated context

    # End while

Notes and customization suggestions • Validation logic:

VALIDATION_REGEX

above matches valid outputs. If you want to detect invalid patterns, invert the check:

invalid = bool(INVALID_REGEX.search(buffered))

. • Prompt shaping: Adding a

system

or

assistant

message before retry helps the model correct the output. • Streaming vs buffering tradeoff: This example streams chunks as they arrive (low latency) but still buffers full text to validate. If you prefer to withhold streaming until validation passes, buffer without yielding until after validation. • Cancel handling: Preemptive generation may cancel streams — this example treats cancellations as non-final and retries. Adjust whether cancelled attempts decrement retry budget. • Helper: replace

self._make_error_chunk(...)

with how your agent constructs final chunks/messages. If you want, I can adapt this to: • Hold streaming until a valid response is confirmed, or • Validate preemptive interim transcripts differently from final transcripts, or • Show a version that uses an invalid-pattern check (instead of valid-pattern). Which variant do you want?

How will this affect perceived latency by the user? Will there be a delay in streaming the llm output into the tts node?

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!

I only care to check the first 5 characters of the response for a forbidden pattern. Change your implementation to do this. It should stop and retry generation if the first 5 characters are forbidden