we have andoutbound agent using sip calls we closing calls i LiveKit Community #ask-ai

we have andoutbound agent using sip calls, we clos...

bitter-appointment-81509

12/15/2025, 8:58 AM

we have andoutbound agent using sip calls, we closing calls in the shutdown function sometimes the preccess is killedmaybe because it takes too much time ( there are a lot of things done during the shutdown)

Copy code

async def on_shutdown(reason: str):
            <http://logger.info|logger.info>("[%s] shutdown: %s", ctx.room.name, reason)
        
            await finish_queue()
            code = dial_info.get("code")
            call_id=dial_info.get("call_id") or ""
        
            if not code:
              m = re.match(r"^room_([^_]+)_", ctx.room.name)
              code = m.group(1) if m else None
            rtdb.reference(f'calls/{code}-appel_agent/{call_id}', app=app).delete()
            <http://logger.info|logger.info>(f"REFERENCE SUPPRIMEE a 'calls/")
            try:
                <http://logger.info|logger.info>("transcription")
                has_content = not await is_transcription_empty()
                <http://logger.info|logger.info>("on_shutdown: has_content=%s", has_content)
            except Exception as e:
                logger.error("on_shutdown: _is_transcription_empty failed: %s", e)
                has_content = False
            <http://logger.info|logger.info>("transcription")
            
            
            if getattr(agent, "was_answering_machine", False):
                <http://logger.info|logger.info>("on_shutdown: answering machine case, pas de résumé ni notif standard.")
                try:
                    async with async_open(log_path, "w", encoding="utf-8") as f:
                        await f.write("")
                        #update_call_status(hardware_code, call_id, "appel-manque")
                except Exception as e:
                    logger.warning("on_shutdown purge transcript failed: %s", e)
        
                try:
                    await ctx.api.room.delete_room(api.DeleteRoomRequest(room=ctx.room.name))
                except Exception as e:
                    logger.error("delete_room error: %s", e)
                return
              
            transcription_complete = ""
            if has_content:
                try:
                    async with async_open(log_path, "r", encoding="utf-8") as f:
                        transcription_complete = await f.read()
                    <http://logger.info|logger.info>("on_shutdown: transcript len=%d", len(transcription_complete))
                except Exception as e:
                    logger.error("on_shutdown: read transcript failed: %s", e)
                    transcription_complete = ""
        
            try:
                contains_key = await contient_mot_clef(log_path, cfg_path)
            except Exception as e:
                logger.error("on_shutdown: contient_mot_clef failed: %s", e)
                contains_key = False
        
            if not has_content or not contains_key:
                <http://logger.info|logger.info>("on_shutdown: skip notify (has_content=%s, contains_key=%s)", has_content, contains_key)
                try:
                    await ctx.api.room.delete_room(api.DeleteRoomRequest(room=ctx.room.name))
                except Exception as e:
                    logger.error("delete_room error: %s", e)
                return
        
        
            try:
                    call_id = call_id or ""
                    tz_paris = ZoneInfo("Europe/Paris")
                    date_iso = datetime.now(timezone.utc).astimezone(tz_paris).isoformat()
                    try:
                        append_rag_entry(code, call_id, date_iso, transcription_complete)
                    except Exception as e:
                        logger.warning("on_shutdown: append_rag_entry failed: %s", e)
            except Exception as e:
                logger.warning("on_shutdown: saving rag entry failed: %s", e)
        
        
            try:
                prenom_val = get_prenom_from_config(cfg_path) or ""
                livana_nom = get_prenom_livana_from_config(cfg_path) or ""
                m = re.match(r"^room_([^_]+)_", ctx.room.name)
                code = m.group(1) if m else None
                
                if "generer_resumer" in globals() and callable(globals()["generer_resumer"]):
                    try:
                        resume_txt = generer_resumer(transcription_complete, prenom_val, livana_nom)
                        if resume_txt:
                            <http://logger.info|logger.info>("on_shutdown: résumé généré (%d chars)", len(resume_txt))
                    except Exception as e:
                        logger.warning("generer_resumer failed: %s", e)
        
                if "generer_centres_interet" in globals() and callable(globals()["generer_centres_interet"]):
                    try:
                        resultat_interets = generer_centres_interet(transcription_complete)
                        if "ecraser_interets_si_differents" in globals() and callable(globals()["ecraser_interets_si_differents"]):
                            ecraser_interets_si_differents(resultat_interets,cfg_path, code)
                    except Exception as e:
                        logger.warning("maj intérêts failed: %s", e)
            except Exception as e:
                logger.warning("post-processing failed: %s", e)
        
            
      
            topic = code
            is_alert = bool(getattr(agent, "action_detected", False))  # si tu l'as implémenté
        
            title = "Probleme detecte" if is_alert else "Appel effectue"
            data_payload = {"alerte": "true"} if is_alert else {"appel_effectue": "true"}
        
            combined_body = (resume_txt.strip() + "\n\n" + transcription_complete.strip()).strip()
            try:
                if "send_notification_to_topic" in globals() and callable(globals()["send_notification_to_topic"]):
                    send_notification_to_topic(
                        topic,
                        title,
                        combined_body,
                        data_payload,  
                        False,          
                        is_alert,       
                        not is_alert, 
                    )
                    <http://logger.info|logger.info>("on_shutdown: notification envoyée (topic=%s, title=%s)", topic, title)
                else:
                    logger.warning("on_shutdown: send_notification_to_topic indisponible")
            except Exception as e:
                logger.error("on_shutdown: erreur envoi notification: %s", e)
        
            # 7) Purge du fichier de transcription
            try:
                async with async_open(log_path, "w", encoding="utf-8") as f:
                    await f.write("")
            except Exception as e:
                logger.warning("on_shutdown: purge transcript failed: %s", e)
        
            # 8) Suppression de la room
            try:
                await ctx.api.room.delete_room(api.DeleteRoomRequest(room=ctx.room.name))
            except Exception as e:
                logger.error("delete_room error: %s", e)

tall-airline-16756

12/15/2025, 8:58 AM

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

tall-airline-16756

12/15/2025, 8:58 AM

Thanks — I looked over your shutdown code and the problem you described (process sometimes killed during shutdown). I’m responding about the shutdown timing and how to make this routine more reliable. Quick diagnosis • The framework waits 60 seconds by default for shutdown hooks; long-running work can trigger a forced kill. • Your on_shutdown performs many potentially slow operations (file I/O, transcription checks, RAG entry, summary generation, interest updates, notifications, room deletion), which can exceed that timeout. What you can do (practical steps) 1. Prioritize critical work first ◦ Keep essential actions that must finish before process exit (room deletion, removing RTDB reference) at the top and run them immediately. ◦ Defer non-critical tasks (RAG append, summary generation, interest updates, full notification body composition, transcript purge) to background processing. 1. Offload long tasks to background workers ◦ Instead of doing heavy processing inside on_shutdown, enqueue a job (message queue, task worker, or simple HTTP job endpoint) with the needed metadata and return quickly from on_shutdown. ◦ Background worker can read the transcript file and run RAG/summary/notification reliably without shutdown time pressure. 1. Make shutdown tasks asynchronous and limited ◦ Ensure all file and network calls are non-blocking (you already use async I/O — keep that). ◦ Add explicit per-step timeouts (asyncio.wait_for) around external operations (e.g., transcription checks, notification send, append_rag_entry) so a single hang won’t block the whole shutdown. 1. Short-circuit when possible ◦ If you determine early that no transcription or no keyword exists, exit immediately (you already do this — move that check up to run before other processing where possible). 1. Increase shutdown timeout if needed ◦ There is a server option

shutdown_process_timeout

you can raise so the runtime waits longer for shutdown hooks. Use this only as a safety net while you refactor to background jobs. 1. Add robust logging and survival guards ◦ Log start/end and durations for each major step to identify which step is exceeding the timeout. ◦ Catch and log exceptions but avoid long retries during shutdown. Targeted, minimal refactor suggestion 1. Move these checks to the very top: ◦ call_id and RTDB deletion ◦ quick transcription-empty check (fast path). If false -> delete room and return. 1. Enqueue background job with: room name, code, call_id, path to transcript, flags (was_answering_machine, action_detected). 2. In on_shutdown: delete room, purge minimal local data, return. 3. Background job does RAG append, summary, interest updates, notifications, transcript purge, and final cleanup. A few clarifying questions • Do you have a task queue or worker infrastructure available (Redis/RQ, Celery, cloud functions, or a separate service) to accept background jobs? • Are you able to increase the

shutdown_process_timeout

temporarily if I suggest code that assumes a longer timeout? • Which step in your logs shows the longest duration currently? If you can share recent shutdown logs with per-step timing, I can point to the exact culprit. If you want, I can: • Show an example pattern to enqueue a background job from on_shutdown. • Draft snippets to apply per-step timeouts (using

asyncio.wait_for

) around your slow calls. Which would you prefer? Sources: Job lifecycle | LiveKit docs | Agents telephony integration | LiveKit docs

bitter-appointment-81509

12/15/2025, 9:03 AM

well it looks likes it comes from the resume generation and so the notification is not sent

tall-airline-16756

12/15/2025, 9:03 AM

Thanks — good to know the resume generation is the likely culprit. Here are targeted, practical options to ensure the notification still gets sent and the process isn’t killed. Short-term fixes (fast to implement) • Run resume generation with a hard timeout so it can’t block shutdown: ◦ Wrap the call with

asyncio.wait_for(generer_resumer(...), timeout=5)

(adjust timeout). ◦ If it times out, log and proceed to send a minimal notification (e.g., use transcription only). • Ensure notification send is prioritized before resume generation: ◦ Compose a short notification using

transcription_complete

(or “Résumé indisponible”) and call

send_notification_to_topic(...)

immediately. ◦ Then run

generer_resumer

in the background or with timeout. Best practice (recommended) • Offload resume generation and other heavy post-processing to a background worker: a. In on_shutdown, quickly gather metadata (room, code, call_id, transcript path, flags). b. Enqueue a job (task queue, message, or HTTP webhook) and return from on_shutdown. c. Worker processes the transcript, generates résumé, updates RAG, sends full notification, and purges transcript. • Keep on_shutdown responsible only for critical cleanup (delete room, remove RTDB reference, enqueue job). Implementation notes you can apply now • Add per-step timing/logging to find exact delays: ◦ Log start/end times around

generer_resumer

append_rag_entry

, and

send_notification_to_topic

. • Use

asyncio.create_task()

to fire-and-forget noncritical async work if you can guarantee a separate process will finish it (but prefer a durable queue). • If you don’t have a worker system, a simple fallback is to POST a job to a short-lived serverless function that does the heavy work. Quick diagnostics I can help with • Share logs showing timestamps around resume generation and notification attempts so I can confirm the order and durations. • Tell me whether you have a task queue or can run a small worker (Redis, Celery, cloud function). If yes, I’ll give a concise enqueue + worker pattern you can paste into your code. Which short-term or best-practice option would you like implementation help with?

2 Views

Open in Slack

Previous Next