abundant-father-33863
09/22/2025, 8:03 AMtall-airline-16756
09/22/2025, 8:03 AMtall-airline-16756
09/22/2025, 8:03 AMbefore_llm_cb callback, here’s a friendly example function you can use:
async def before_llm_cb(assistant: VoicePipelineAgent, chat_ctx: llm.ChatContext):
# First, let's check if there are any messages at all
if not chat_ctx.messages:
print("No messages in chat context.")
return # We exit early if there’s nothing to process
# Now, we’ll look through the messages starting from the most recent to find the last user message
last_user_message = None
for msg in reversed(chat_ctx.messages):
if msg.role == "user":
last_user_message = msg.content
break
if last_user_message:
print(f"Last customer message: {last_user_message}")
# Here you can also tweak the chat context or do other useful things
else:
print("No user messages found in chat context.")
This function is designed thoughtfully to make sure you safely access the last user message by:
• First checking if chat_ctx.messages actually contains any messages.
• Going backward from the newest message to find the latest one from the user.
• Gracefully handling situations where no user message is present.
When you set up your VoicePipelineAgent, you can register this callback like this:
agent = VoicePipelineAgent(
vad=silero.VAD.load(),
stt=deepgram.STT(model="nova-2-general"),
llm=openai.LLM(),
tts=cartesia.TTS(),
chat_ctx=initial_ctx,
before_llm_cb=before_llm_cb,
)
This way, you’ll have a reliable and smooth way to manage your conversation history dynamically. If you have any questions or need further help, feel free to ask—I’m here to support you!
Sources: chain-of-thought-tts/README.mdocabundant-father-33863
09/22/2025, 8:39 AMtall-airline-16756
09/22/2025, 8:40 AMbefore_llm_cb callback. You have a few options when returning a value:
• False to skip processing the message altogether.
• None to continue with the normal processing flow.
• True to make sure the message is processed.
Here’s a simple example function that stops processing if the user’s message is "hello":
async def before_llm_cb(assistant: VoicePipelineAgent, chat_ctx: llm.ChatContext):
# First, check if there are any messages
if not chat_ctx.messages:
return False # Skip if no messages
# Find the last user message
last_user_message = None
for msg in reversed(chat_ctx.messages):
if msg.role == "user":
last_user_message = msg.content
break
# If the last message is "hello", stop processing
if last_user_message and last_user_message.lower() == "hello":
print("Skipping LLM response for 'hello'")
return False
return None # Otherwise, continue normal processing
When you’re setting up your VoicePipelineAgent, you can register this callback like this:
agent = VoicePipelineAgent(
vad=silero.VAD.load(),
stt=deepgram.STT(model="nova-2-general"),
llm=openai.LLM(),
tts=cartesia.TTS(),
chat_ctx=initial_ctx,
before_llm_cb=before_llm_cb,
)
This way, you have a handy way to filter out specific messages or add your own custom logic before anything gets sent to the LLM. If you have any questions or want to explore more, feel free to ask—I’m here to help!
Sources: chain-of-thought-tts/README.mdoc | Vision Agent Quickstart | LiveKit Docs | VoicePipelineAgent | LiveKit Docs | Agents 0.x migration guide | LiveKit Docs