the transcript is being saved via the <vscode-file...
# ask-aia
aloof-forest-57428
12/23/2025, 7:11 AMthe transcript is being saved via the on_session_end callback which acts like a "webhook" but whats happening is sometimes it is failing since i am using callback at the end
How It Works:
Copy code
python
<@U07SJDE11GF>.rtc_session(
agent_name="emi-reminder-agent-test2",
type=WorkerType.ROOM,
on_session_end=on_session_end, # ← This is the "webhook"
)on_session_end()ctx.make_session_report()CallRepository.queue_call_record()on_session_endon_session_end Copy code
:white_check_mark: SUCCESS: ":floppy_disk: [SessionReport] Transcript persisted for call 7c530579... (10 items)"
:white_check_mark: SUCCESS: ":floppy_disk: [SessionReport] Transcript persisted for call 53e71e23... (17 items)"
:x: FAILURE: ":x: Unable to build session report: Cannot prepare report, no AgentSession was found"session.start()on_session_end Copy code
:white_check_mark: "Transcript persisted... (2 items)" ← Very short
:white_check_mark: "Transcript persisted... (10 items)" ← Reasonable
:white_check_mark: "Transcript persisted... (17 items)" ← Goodon_session_endt
tall-airline-16756
12/23/2025, 7:11 AMThanks for reaching out, Aryan. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
tall-airline-16756
12/23/2025, 7:12 AMThanks — this is a clear write-up of the problem. I’ll respond to the two issues you highlighted and give concrete, low-effort options you can try next.
Acknowledging what you asked: you’re seeing missed or tiny transcripts because
on_session_enduser_input_transcribedconversation_item_addedctx.add_shutdown_callback()session.start(..., record=True)room_finishedagent_state_changeda
aloof-forest-57428
12/23/2025, 7:13 AMwhich is the esiest and sure way of getting all the transcripts properly
t
tall-airline-16756
12/23/2025, 7:13 AMShort answer: enable session recording (
record=Truerecord=True Copy code
await session.start(agent=agent, room=ctx.room, record=True)user_input_transcribedconversation_item_addedis_finalctx.add_shutdown_callback()room_finisheda
aloof-forest-57428
12/23/2025, 8:21 AM@session.on("user_input_transcribed")
async def on_user_transcribed(event):
"""Save user transcript chunks in real-time"""
try:
if hasattr(event, 'is_final') and event.is_final:
transcript_entry = {
"speaker": "user",
"text": event.text if hasattr(event, 'text') else str(event),
"timestamp": datetime.utcnow().isoformat(),
"relative_time": (datetime.utcnow() - agent.call_start_time).total_seconds()
}
transcript_buffer.append(transcript_entry)
emi_data.full_transcript.append(transcript_entry)
# Periodic save every 5 entries
if len(transcript_buffer) >= 5:
record = CallRecordModel(
callId=call_id,
fullTranscript=transcript_buffer.copy(),
updatedAt=datetime.utcnow()
)
await CallRepository.queue_call_record(record)
logger.info(f"💾 [REAL-TIME] Saved transcript chunk ({len(transcript_buffer)} entries)")
transcript_buffer.clear()
except Exception as e:
logger.error(f"❌ Error saving user transcript: {e}")
@session.on("agent_speech_committed")
async def on_agent_speech(event):
"""Save agent speech chunks in real-time"""
try:
if hasattr(event, 'text') and event.text:
transcript_entry = {
"speaker": "assistant",
"text": event.text,
"timestamp": datetime.utcnow().isoformat(),
"relative_time": (datetime.utcnow() - agent.call_start_time).total_seconds()
}
transcript_buffer.append(transcript_entry)
emi_data.full_transcript.append(transcript_entry)
# Periodic save every 5 entries
if len(transcript_buffer) >= 5:
record = CallRecordModel(
callId=call_id,
fullTranscript=transcript_buffer.copy(),
updatedAt=datetime.utcnow()
)
await CallRepository.queue_call_record(record)
logger.info(f"💾 [REAL-TIME] Saved transcript chunk ({len(transcript_buffer)} entries)")
transcript_buffer.clear()
except Exception as e:
logger.error(f"❌ Error saving agent speech: {e}")
# ============================================================================
# STEP: ADD SHUTDOWN CALLBACK FOR LAST TRANSCRIPT SAVE
# ============================================================================
async def save_final_transcript():
"""Save final transcript on shutdown"""
try:
# Save any remaining buffered transcript
if transcript_buffer or emi_data.full_transcript:
final_transcript = emi_data.full_transcript
if final_transcript:
logger.info(f"💾 [SHUTDOWN] Saving final transcript: {len(final_transcript)} entries")
record = CallRecordModel(
callId=call_id,
fullTranscript=final_transcript,
updatedAt=datetime.utcnow(),
callEndTime=datetime.utcnow()
)
await CallRepository.queue_call_record(record)
logger.info(f"✅ [SHUTDOWN] Final transcript saved successfully")
except Exception as e:
logger.error(f"❌ [SHUTDOWN] Error saving final transcript: {e}")
ctx.add_shutdown_callback(save_final_transcript)
logger.info("✅ [AGENT] Shutdown callback registered for transcript backup")
logger.info(f"🎙️ [AGENT] Session recording ENABLED for transcript backup")
await session.start(
agent=agent,
room=ctx.room,
room_input_options=RoomInputOptions(
audio_enabled=True,
text_enabled=False,
),
room_output_options=RoomOutputOptions(
audio_enabled=True,
),
record=True, # ✅ ENABLE SESSION RECORDING for transcript backup
)
logger.info(f"✅ [AGENT] EMI session completed successfully")this is ewhat i wrote rker:✅ Silero VAD configured: 450ms silence threshold, 0.4 activation (confirmation-optimized)
INFOemi agent worker✅ Silero VAD configured: 450ms silence threshold, 0.4 activation (confirmation-optimized)
{"level": "INFO", "name": "emi_agent_worker", "message": "✅ Silero VAD configured: 450ms silence threshold, 0.4 activation (confirmation-optimized)", "pid": 31204, "job_id": "AJ_ttDoUpbK8HX8", "room_id": "RM_e8zk2pouyW38", "timestamp": "2025-12-23T081418.578621+00:00"}
ERRORemi agent worker❌ EMI agent error: Cannot register an async callback with
.on()asyncio.create_task.on()asyncio.create_task.on()asyncio.create_task.on()asyncio.create_task.on()asyncio.create_taskt
tall-airline-16756
12/23/2025, 8:22 AMThanks — this pinpoints the immediate error: the Python SDK’s
.on()ValueError: Cannot register an async callback with .on() Copy code
def _sync_user_transcribed(event):
asyncio.create_task(_handle_user_transcribed(event))
async def _handle_user_transcribed(event):
try:
if getattr(event, "is_final", False):
entry = {
"speaker": "user",
"text": getattr(event, "text", str(event)),
"timestamp": datetime.utcnow().isoformat(),
"relative_time": (datetime.utcnow() - agent.call_start_time).total_seconds(),
}
transcript_buffer.append(entry)
emi_data.full_transcript.append(entry)
if len(transcript_buffer) >= 5:
record = CallRecordModel(callId=call_id, fullTranscript=transcript_buffer.copy(), updatedAt=datetime.utcnow())
await CallRepository.queue_call_record(record)
<http://logger.info|logger.info>(f"💾 [REAL-TIME] Saved transcript chunk ({len(transcript_buffer)} entries)")
transcript_buffer.clear()
except Exception as e:
logger.error(f"❌ Error saving user transcript: {e}")
session.on("user_input_transcribed", _sync_user_transcribed)agent_speech_committedasyncio.create_taskctx.add_shutdown_callback(save_final_transcript)record=Truesession.start(...)agent.call_start_timetranscript_bufferemi_data.full_transcriptasyncio.Lock()agent_speech_committedsave_final_transcripta
aloof-forest-57428
12/24/2025, 8:30 AMtranscript_buffer = [] # Buffer for real-time transcript chunks
transcript_lock = asyncio.Lock() # Protect shared lists from race conditions
# Async handler for user transcript
async def _handle_user_transcribed(event):
"""Async handler to save user transcript chunks"""
try:
if getattr(event, "is_final", False):
# Ensure call_start_time exists
if not hasattr(agent, 'call_start_time') or agent.call_start_time is None:
logger.warning("⚠️ agent.call_start_time not initialized, using current time")
agent.call_start_time = datetime.utcnow()
# ✅ CRITICAL FIX: Extract text correctly from event
# Try transcript first (STT events), then text (agent events), then fallback
text_content = getattr(event, "transcript", None) or getattr(event, "text", None)
if not text_content:
logger.warning(f"⚠️ No transcript/text found in event: {type(event)}")
return
entry = {
"speaker": "user",
"text": text_content,
"timestamp": datetime.utcnow().isoformat(),
"relative_time": (datetime.utcnow() - agent.call_start_time).total_seconds(),
}
async with transcript_lock:
transcript_buffer.append(entry)
emi_data.full_transcript.append(entry)
# Periodic save every 5 entries
if len(transcript_buffer) >= 5:
record = CallRecordModel(
callId=call_id,
fullTranscript=transcript_buffer.copy(),
updatedAt=datetime.utcnow(),
# ✅ Don't include payloadData - already set by API server
)
await CallRepository.queue_call_record(record)
logger.info(f"💾 [REAL-TIME] Saved user transcript chunk ({len(transcript_buffer)} entries)")
transcript_buffer.clear()
except Exception as e:
logger.error(f"❌ Error saving user transcript: {e}")
import traceback
traceback.print_exc()
# Synchronous wrapper for user transcript
def _sync_user_transcribed(event):
"""Synchronous wrapper that spawns async task"""
asyncio.create_task(_handle_user_transcribed(event))
# Async handler for agent speech
async def _handle_agent_speech(event):
"""Async handler to save agent speech chunks"""
try:
# ✅ Extract text from event (agent speech events use 'text' attribute)
text_content = getattr(event, "text", None)
if text_content:
# Ensure call_start_time exists
if not hasattr(agent, 'call_start_time') or agent.call_start_time is None:
logger.warning("⚠️ agent.call_start_time not initialized, using current time")
agent.call_start_time = datetime.utcnow()
entry = {
"speaker": "assistant",
"text": text_content,
"timestamp": datetime.utcnow().isoformat(),
"relative_time": (datetime.utcnow() - agent.call_start_time).total_seconds(),
}
async with transcript_lock:
transcript_buffer.append(entry)
emi_data.full_transcript.append(entry)
# Periodic save every 5 entries
if len(transcript_buffer) >= 5:
record = CallRecordModel(
callId=call_id,
fullTranscript=transcript_buffer.copy(),
updatedAt=datetime.utcnow(),
# ✅ Don't include payloadData - already set by API server
)
await CallRepository.queue_call_record(record)
logger.info(f"💾 [REAL-TIME] Saved agent speech chunk ({len(transcript_buffer)} entries)")
transcript_buffer.clear()
except Exception as e:
logger.error(f"❌ Error saving agent speech: {e}")
import traceback
traceback.print_exc()
# Synchronous wrapper for agent speech
def _sync_agent_speech(event):
"""Synchronous wrapper that spawns async task"""
asyncio.create_task(_handle_agent_speech(event))
# Register synchronous callbacks
session.on("user_input_transcribed", _sync_user_transcribed)
session.on("agent_speech_committed", _sync_agent_speech)
logger.info("✅ [AGENT] Real-time transcript event handlers registered")
# ============================================================================
# STEP: ADD SHUTDOWN CALLBACK FOR LAST TRANSCRIPT SAVE
# ============================================================================
async def save_final_transcript():
"""Save final transcript on shutdown"""
try:
async with transcript_lock:
# Save any remaining buffered transcript
if transcript_buffer or emi_data.full_transcript:
final_transcript = emi_data.full_transcript
if final_transcript:
logger.info(f"💾 [SHUTDOWN] Saving final transcript: {len(final_transcript)} entries")
record = CallRecordModel(
callId=call_id,
fullTranscript=final_transcript,
updatedAt=datetime.utcnow(),
callEndTime=datetime.utcnow(),
# ✅ Include payloadData as backup in case API server data was lost
payloadData={
"language": emi_data.language,
"loanId": emi_data.loan_id,
"phoneNumber": emi_data.phone_number,
"borrowerName": emi_data.borrower_name,
"emiAmount": emi_data.emi_amount,
"dueDate": emi_data.due_date,
},
)
await CallRepository.queue_call_record(record)
logger.info(f"✅ [SHUTDOWN] Final transcript saved successfully")
except Exception as e:
logger.error(f"❌ [SHUTDOWN] Error saving final transcript: {e}")
import traceback
traceback.print_exc()
ctx.add_shutdown_callback(save_final_transcript)
logger.info("✅ [AGENT] Shutdown callback registered for transcript backup")
logger.info(f"🎙️ [AGENT] Session recording ENABLED for transcript backup")
await session.start(
agent=agent,
room=ctx.room,
room_input_options=RoomInputOptions(
audio_enabled=True,
text_enabled=False,
),
room_output_options=RoomOutputOptions(
audio_enabled=True,
),
record=True, # ✅ ENABLE SESSION RECORDING for transcript backup
)
logger.info(f"✅ [AGENT] EMI session completed successfully")
except Exception as e:
logger.error(f"❌ EMI agent error: {e}")
import traceback
traceback.print_exc()
finally:
try:
if 'emi_data' in locals() and hasattr(emi_data, 'full_transcript'):
# Use lock if it exists
if 'transcript_lock' in locals():
async with transcript_lock:
if len(emi_data.full_transcript) > 0:
logger.info(f"💾 [FINALLY] Saving final transcript: {len(emi_data.full_transcript)} entries")
record = CallRecordModel(
callId=call_id,
fullTranscript=emi_data.full_transcript,
updatedAt=datetime.utcnow(),
callEndTime=datetime.utcnow(),
)
await CallRepository.queue_call_record(record)
logger.info(f"✅ [FINALLY] Final transcript saved successfully")
else:
logger.warning(f"⚠️ [FINALLY] No transcript to save (empty)")
else:
# Fallback if lock doesn't exist
if len(emi_data.full_transcript) > 0:
logger.info(f"💾 [FINALLY] Saving final transcript: {len(emi_data.full_transcript)} entries")
record = CallRecordModel(
callId=call_id,
fullTranscript=emi_data.full_transcript,
updatedAt=datetime.utcnow(),
callEndTime=datetime.utcnow(),
)
await CallRepository.queue_call_record(record)
logger.info(f"✅ [FINALLY] Final transcript saved successfully")
else:
logger.warning(f"⚠️ [FINALLY] No transcript to save (empty)")
except Exception as transcript_error:
logger.error(f"❌ [FINALLY] Error saving final transcript: {transcript_error}")
# ✅ CRITICAL FIX: Wait for ALL writes before closing
try:
if db_manager.connected:
queue_size = db_manager.write_queue.qsize()
if queue_size > 0:
logger.info(f"⏳ [CLEANUP] Waiting for {queue_size} pending writes...")
await db_manager.wait_for_writes(timeout=15.0)
logger.info("✅ [CLEANUP] All writes completed")
else:
logger.info("✅ [CLEANUP] No pending writes")
# Small delay to ensure everything is flushed
await asyncio.sleep(0.5)
except Exception as e:
logger.error(f"❌ [CLEANUP] Cleanup error: {e}")
import traceback
traceback.print_exc()
os this is how my transcription hook is working but if call is discpconnectingh in the middle only user transcription is comin
t
tall-airline-16756
12/24/2025, 8:30 AMThanks — you’ve built a solid pipeline. You’re seeing only user transcripts when the call disconnects mid-call because agent speech events sometimes aren’t emitted/committed before the disconnect or your agent’s finalization logic doesn't get time to persist the last agent entries. Below are focused fixes and checks to ensure you capture agent side text as reliably as the user side.
What to check first (quick diagnostics)
• Confirm agent speech events are being emitted before disconnect: enable debug logging for
agent_speech_committedevent.text Copy code
# after appending entry
await CallRepository.queue_call_record(CallRecordModel(callId=call_id, fullTranscript=[entry], updatedAt=datetime.utcnow()))asyncio.create_task Copy code
def _sync_participant_left(event):
asyncio.create_task(_handle_disconnect_event(event))
session.on("participant_disconnected", _sync_participant_left)_handle_disconnect_eventevent.texttextdb_manager.wait_for_writes(timeout=...)record=Truetranscript_lockagent.call_start_time4 Views