https://livekit.io logo
Join Slack
Powered by
When it is best to have hook, post interaction pro...
# ask-ai
m
When it is best to have hook, post interaction processor to write livekit to external memory for threads? I need to store messages that happened in external storage as they appear. How to do it best?
t
Thanks for reaching out, Stanislav. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — this is a great question. You want to reliably capture LiveKit messages as they happen and persist them to external storage (threads/messages), so here’s a concise, practical plan that balances real-time capture, durability, and performance. Acknowledging the goal • You want messages written to external memory immediately as they occur, without blocking the real-time flow. Below are recommended hooks, an architecture pattern, and concrete best practices. Recommended hooks / capture points • For voice / speech flows: use the lifecycle hook 
on_user_turn_completed
to capture the user message once STT (speech-to-text) is complete. This gives you the final message text that should be stored. • For chat / data-channel messages: attach a listener to the room/participant data messages (e.g., 
message_received
) so you capture each data message as it arrives. Architecture pattern (non-blocking, durable) 1. Capture ◦ In the hook/listener, do minimal synchronous work: validate the message, attach metadata (room id, participant id, timestamp, message id). 1. Enqueue ◦ Immediately enqueue the message into a durable queue (Redis stream, SQS, Kafka, or similar). This keeps the real-time path fast and reliable. 1. Background worker ◦ A separate worker consumes the queue and writes the message into your external storage (database, vector store, data lake). 1. Monitoring + retries ◦ On write failure, retry with exponential backoff and move persistent failures to a dead-letter queue for manual inspection. Concrete best-practices and details • Idempotency: Include a unique message identifier (e.g., 
ChatMessage.id
) as a deduplication key when writing so retries don’t create duplicates. • Minimal hook work: Never perform slow network or DB calls inside the lifecycle hook — only enqueue. • Ordering: If strict ordering is required per-thread, keep a sequence number or use a partitioned queue keyed by thread/room so the worker preserves order. • Batching: Background worker can batch writes for throughput, but balance batching latency vs. throughput based on your real-time needs. • Security: Ensure messages enqueued and stored are encrypted in transit and at rest per your security requirements. • Observability: Track queue depth, worker lag, and write error rates. Alert on growth or errors so you don’t lose real-time fidelity. • DLQ & replay: Keep a dead-letter queue and an easy replay mechanism to reprocess failed items. Example flow (high level) 1. `on_user_turn_completed(turn_ctx, new_message)`: ◦ Create a JSON payload with message text, ids, timestamps. ◦ Push to queue (fast, non-blocking). 1. Worker: ◦ Pop messages, dedupe using message id. ◦ Write to external store (and optionally index for search or RAG). ◦ Acknowledge queue entry on success; on permanent failure move to DLQ. When to write synchronously vs. asynchronously • Asynchronous (recommended): Use this for nearly all cases to avoid impacting real-time performance. • Synchronous direct-write: Only consider if you must guarantee the write before continuing and you can accept added latency or blocking behavior. If you’d like, I can: • Suggest a concrete queue + DB stack given your scale (small vs. high throughput). • Provide a short code snippet for the hook and worker in your preferred language. • Help with message schema suggestions (fields to include for RAG/thread reconstruction). Which of those would be most helpful next? Sources: External data and RAG | LiveKit Documentation | Best practices for managing webhook event streams | LiveKit
m
on_user_turn_completed
- great idea for user messages. What to use for agent messages?
t
Thanks — good follow-up. You’re already capturing user messages with `on_user_turn_completed`; to capture agent messages in real time, use the conversation / agent-side hooks that fire when the agent adds items to the conversation. Two reliable options: • Use the 
conversation_item_added
event on the AgentSession (recommended) ◦ This event triggers whenever any new conversation item is added (including agent responses). It gives you the final agent text and metadata so you can persist it the same way you persist user messages. ◦ Example (Python-like pseudocode):
Copy code
@session.on("conversation_item_added")
    async def on_conversation_item_added(item: ChatItem):
        payload = {
            "id": item.id,
            "role": item.role,         # "agent"
            "text": item.text,
            "timestamp": item.timestamp,
            "room_id": item.room_id,
            "thread_id": item.thread_id,   # if applicable
        }
        await enqueue_for_storage(payload)
• Benefits: captures the final message content after any agent-side processing, works for both single-turn and multi-part agent messages, and keeps your persistence flow consistent with user-message handling. • (Optional) Intercept in pipeline nodes such as the 
tts_node
◦ Use this if you need to capture or transform the agent text right before TTS (for logging, redaction, or modifying what’s sent to the voice engine). ◦ Example:
Copy code
async def tts_node(self, text: str, model_settings: ModelSettings) -> AsyncIterator[str]:
        await enqueue_for_storage({"role": "agent", "text": text, "context": "pre-tts"})
        async for fragment in Agent.default.tts_node(self, text, model_settings):
            yield fragment
• Use this only when you specifically need pre-TTS text or want to capture streaming fragments. Best practices (same as for user messages) • Enqueue quickly (non-blocking) and persist in a background worker. • Include 
id
for idempotency and deduplication. • If ordering per thread matters, partition or sequence by thread or room. • Use DLQ/retries, monitoring, and encryption as before. Which would you prefer to implement first — the 
conversation_item_added
listener or a 
tts_node
intercept? I can provide a concrete code sample for your stack (Node/Python) and a suggested message schema for thread reconstruction. Sources: External data and RAG | LiveKit Documentation | Pipeline nodes and hooks | LiveKit Documentation | Redacting PII from agent logs and transcripts | LiveKit
9 Views
Open in Slack
PreviousNext
Powered by