From 6b8f5d069bd159ccd375995105c8feb2e2c43041 Mon Sep 17 00:00:00 2001 From: Yera All Date: Mon, 13 Apr 2026 11:59:43 +0500 Subject: [PATCH] feat(ai): add rolling conversation summary memory --- .env.example | 6 +- .env.production.template | 6 +- deployment/docker-compose.yml | 2 +- scripts/local_stack.py | 6 +- scripts/materialize_ivr_prompts.py | 4 +- scripts/materialize_tts_prompt.py | 4 +- services/ai_orchestrator_service/app.py | 128 ++++++- services/ai_orchestrator_service/voice.py | 67 +++- services/shared/ai_context_summary.py | 434 ++++++++++++++++++++++ services/shared/sql_init.py | 9 + services/shared/sql_models.py | 2 + tests/test_ai_context_summary.py | 99 +++++ tests/test_ai_orchestrator_service.py | 120 ++++++ tests/test_sql_init_compat.py | 39 ++ 14 files changed, 900 insertions(+), 26 deletions(-) create mode 100644 services/shared/ai_context_summary.py create mode 100644 tests/test_ai_context_summary.py diff --git a/.env.example b/.env.example index b5a59ad..3673b1b 100644 --- a/.env.example +++ b/.env.example @@ -40,12 +40,12 @@ AI_CUSTOMER_PERSONA_MODE=operator_humanlike AI_DISCLOSURE_MODE=hidden AI_TELEGRAM_ENABLED=0 AI_TELEGRAM_ALWAYS_REPLY=0 -AI_TELEGRAM_MAX_CONTEXT_MESSAGES=20 +AI_TELEGRAM_MAX_CONTEXT_MESSAGES=12 AI_TELEGRAM_MAX_KB_RESULTS=3 AI_TELEGRAM_CONFIDENCE_HANDOFF_THRESHOLD=0.65 AI_WHATSAPP_ENABLED=0 AI_WHATSAPP_ALWAYS_REPLY=0 -AI_WHATSAPP_MAX_CONTEXT_MESSAGES=20 +AI_WHATSAPP_MAX_CONTEXT_MESSAGES=12 AI_WHATSAPP_MAX_KB_RESULTS=3 AI_WHATSAPP_CONFIDENCE_HANDOFF_THRESHOLD=0.65 AI_VOICE_RUNTIME_SERVICE_URL=http://localhost:8018 @@ -102,7 +102,7 @@ AI_VOICE_VAD_MIN_SPEECH_MS=300 AI_VOICE_VAD_TRAILING_SILENCE_MS=400 AI_VOICE_TURN_MAX_MS=10000 AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS=15 -AI_VOICE_MAX_CONTEXT_SEGMENTS=8 +AI_VOICE_MAX_CONTEXT_SEGMENTS=12 AI_VOICE_HANDOFF_TIMEOUT_SECONDS=8 AI_VOICE_RUNTIME_TRUSTED_SERVICE_SUBJECTS=svc:ai-voice-runtime diff --git a/.env.production.template b/.env.production.template index 4c67e74..ad4be9e 100644 --- a/.env.production.template +++ b/.env.production.template @@ -36,12 +36,12 @@ AI_CUSTOMER_PERSONA_MODE=operator_humanlike AI_DISCLOSURE_MODE=hidden AI_TELEGRAM_ENABLED=0 AI_TELEGRAM_ALWAYS_REPLY=0 -AI_TELEGRAM_MAX_CONTEXT_MESSAGES=20 +AI_TELEGRAM_MAX_CONTEXT_MESSAGES=12 AI_TELEGRAM_MAX_KB_RESULTS=3 AI_TELEGRAM_CONFIDENCE_HANDOFF_THRESHOLD=0.65 AI_WHATSAPP_ENABLED=0 AI_WHATSAPP_ALWAYS_REPLY=0 -AI_WHATSAPP_MAX_CONTEXT_MESSAGES=20 +AI_WHATSAPP_MAX_CONTEXT_MESSAGES=12 AI_WHATSAPP_MAX_KB_RESULTS=3 AI_WHATSAPP_CONFIDENCE_HANDOFF_THRESHOLD=0.65 AI_VOICE_ENABLED=0 @@ -98,7 +98,7 @@ AI_VOICE_VAD_MIN_SPEECH_MS=300 AI_VOICE_VAD_TRAILING_SILENCE_MS=400 AI_VOICE_TURN_MAX_MS=10000 AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS=15 -AI_VOICE_MAX_CONTEXT_SEGMENTS=8 +AI_VOICE_MAX_CONTEXT_SEGMENTS=12 AI_VOICE_HANDOFF_TIMEOUT_SECONDS=8 AI_VOICE_RUNTIME_TRUSTED_SERVICE_SUBJECTS=svc:ai-voice-runtime diff --git a/deployment/docker-compose.yml b/deployment/docker-compose.yml index 234a636..ea37f12 100644 --- a/deployment/docker-compose.yml +++ b/deployment/docker-compose.yml @@ -46,7 +46,7 @@ x-app-env: &app_env AI_VOICE_V2_EMOTIVE_ACK_RU_ONLY: "1" AI_VOICE_TURN_MAX_MS: "10000" AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS: "15" - AI_VOICE_MAX_CONTEXT_SEGMENTS: "8" + AI_VOICE_MAX_CONTEXT_SEGMENTS: "12" AI_VOICE_HANDOFF_TIMEOUT_SECONDS: "8" AI_VOICE_RUNTIME_TRUSTED_SERVICE_SUBJECTS: svc:ai-voice-runtime diff --git a/scripts/local_stack.py b/scripts/local_stack.py index 20b88aa..a875fe6 100644 --- a/scripts/local_stack.py +++ b/scripts/local_stack.py @@ -292,7 +292,7 @@ def spawn_service(spec: dict[str, Any], runtime_dir: Path, data_dir: Path, base_ env["AI_TIMEOUT_SECONDS"] = env.get("AI_TIMEOUT_SECONDS", "20") env["AI_TELEGRAM_ENABLED"] = env.get("AI_TELEGRAM_ENABLED", "0") env["AI_TELEGRAM_ALWAYS_REPLY"] = env.get("AI_TELEGRAM_ALWAYS_REPLY", "0") - env["AI_TELEGRAM_MAX_CONTEXT_MESSAGES"] = env.get("AI_TELEGRAM_MAX_CONTEXT_MESSAGES", "20") + env["AI_TELEGRAM_MAX_CONTEXT_MESSAGES"] = env.get("AI_TELEGRAM_MAX_CONTEXT_MESSAGES", "12") env["AI_TELEGRAM_MAX_KB_RESULTS"] = env.get("AI_TELEGRAM_MAX_KB_RESULTS", "3") env["AI_TELEGRAM_CONFIDENCE_HANDOFF_THRESHOLD"] = env.get( "AI_TELEGRAM_CONFIDENCE_HANDOFF_THRESHOLD", @@ -300,7 +300,7 @@ def spawn_service(spec: dict[str, Any], runtime_dir: Path, data_dir: Path, base_ ) env["AI_WHATSAPP_ENABLED"] = env.get("AI_WHATSAPP_ENABLED", "0") env["AI_WHATSAPP_ALWAYS_REPLY"] = env.get("AI_WHATSAPP_ALWAYS_REPLY", "0") - env["AI_WHATSAPP_MAX_CONTEXT_MESSAGES"] = env.get("AI_WHATSAPP_MAX_CONTEXT_MESSAGES", "20") + env["AI_WHATSAPP_MAX_CONTEXT_MESSAGES"] = env.get("AI_WHATSAPP_MAX_CONTEXT_MESSAGES", "12") env["AI_WHATSAPP_MAX_KB_RESULTS"] = env.get("AI_WHATSAPP_MAX_KB_RESULTS", "3") env["AI_WHATSAPP_CONFIDENCE_HANDOFF_THRESHOLD"] = env.get( "AI_WHATSAPP_CONFIDENCE_HANDOFF_THRESHOLD", @@ -330,7 +330,7 @@ def spawn_service(spec: dict[str, Any], runtime_dir: Path, data_dir: Path, base_ env["AI_VOICE_V2_PARTIAL_ASR"] = env.get("AI_VOICE_V2_PARTIAL_ASR", "1") env["AI_VOICE_V2_EMOTIVE_ACK_ENABLED"] = env.get("AI_VOICE_V2_EMOTIVE_ACK_ENABLED", "1") env["AI_VOICE_V2_EMOTIVE_ACK_RU_ONLY"] = env.get("AI_VOICE_V2_EMOTIVE_ACK_RU_ONLY", "1") - env["AI_VOICE_MAX_CONTEXT_SEGMENTS"] = env.get("AI_VOICE_MAX_CONTEXT_SEGMENTS", "8") + env["AI_VOICE_MAX_CONTEXT_SEGMENTS"] = env.get("AI_VOICE_MAX_CONTEXT_SEGMENTS", "12") env["AI_VOICE_HANDOFF_TIMEOUT_SECONDS"] = env.get("AI_VOICE_HANDOFF_TIMEOUT_SECONDS", "8") env["AI_VOICE_VAD_TRAILING_SILENCE_MS"] = env.get("AI_VOICE_VAD_TRAILING_SILENCE_MS", "400") env["AI_VOICE_RUNTIME_TRUSTED_SERVICE_SUBJECTS"] = env.get( diff --git a/scripts/materialize_ivr_prompts.py b/scripts/materialize_ivr_prompts.py index 1add495..33515d3 100644 --- a/scripts/materialize_ivr_prompts.py +++ b/scripts/materialize_ivr_prompts.py @@ -23,7 +23,7 @@ from services.ai_voice_runtime_service.providers.tts import build_tts_provider def _default_provider_name() -> str: - return os.getenv("AI_VOICE_TTS_PROVIDER", "openai").strip() or "openai" + return os.getenv("AI_VOICE_TTS_PROVIDER", "yandex").strip() or "yandex" def _parse_args() -> argparse.Namespace: @@ -39,7 +39,7 @@ def _parse_args() -> argparse.Namespace: parser.add_argument( "--provider", default=_default_provider_name(), - help="TTS provider name. Defaults to AI_VOICE_TTS_PROVIDER or openai.", + help="TTS provider name. Defaults to AI_VOICE_TTS_PROVIDER or yandex.", ) parser.add_argument("--sample-rate", type=int, default=8000, help="Target WAV sample rate. Defaults to 8000.") parser.add_argument("--default-language", default="ru", help="Fallback language for ambiguous prompts. Defaults to ru.") diff --git a/scripts/materialize_tts_prompt.py b/scripts/materialize_tts_prompt.py index b929dd5..22de4db 100644 --- a/scripts/materialize_tts_prompt.py +++ b/scripts/materialize_tts_prompt.py @@ -16,7 +16,7 @@ from services.ai_voice_runtime_service.providers.tts import build_tts_provider def _parse_args() -> argparse.Namespace: - default_provider = os.getenv("AI_VOICE_TTS_PROVIDER", "openai").strip() or "openai" + default_provider = os.getenv("AI_VOICE_TTS_PROVIDER", "yandex").strip() or "yandex" parser = argparse.ArgumentParser( description="Generate TTS once, cache it on the server, and materialize a WAV prompt file.", ) @@ -26,7 +26,7 @@ def _parse_args() -> argparse.Namespace: parser.add_argument( "--provider", default=default_provider, - help="TTS provider name. Defaults to AI_VOICE_TTS_PROVIDER or openai.", + help="TTS provider name. Defaults to AI_VOICE_TTS_PROVIDER or yandex.", ) parser.add_argument("--sample-rate", type=int, default=8000, help="Target WAV sample rate. Defaults to 8000.") return parser.parse_args() diff --git a/services/ai_orchestrator_service/app.py b/services/ai_orchestrator_service/app.py index ee20bb7..a46572e 100644 --- a/services/ai_orchestrator_service/app.py +++ b/services/ai_orchestrator_service/app.py @@ -16,6 +16,12 @@ from sqlalchemy import select from services.shared.core import Role, new_id, utc_now_iso from services.shared.db import get_session +from services.shared.ai_context_summary import ( + dump_context_summary, + render_context_summary_text, + update_context_summary_from_assistant_turn, + update_context_summary_from_user_turn, +) from services.shared.kb_localization import normalize_kb_language from services.shared.kb_search import search_kb_rows from services.shared.models import ( @@ -1467,11 +1473,11 @@ def _ai_whatsapp_always_reply() -> bool: def _ai_max_context_messages() -> int: - return max(4, _int_env("AI_TELEGRAM_MAX_CONTEXT_MESSAGES", 20)) + return max(4, _int_env("AI_TELEGRAM_MAX_CONTEXT_MESSAGES", 12)) def _ai_whatsapp_max_context_messages() -> int: - return max(4, _int_env("AI_WHATSAPP_MAX_CONTEXT_MESSAGES", 20)) + return max(4, _int_env("AI_WHATSAPP_MAX_CONTEXT_MESSAGES", 12)) def _ai_max_kb_results() -> int: @@ -2293,6 +2299,7 @@ def _openai_prompt( messages: list[Any], kb_results: list[KBArticleRow], language: str, + conversation_summary_text: str = "", channel_label: str = "Telegram", channel_key: str = "telegram", ) -> list[dict[str, str]]: @@ -2338,6 +2345,7 @@ def _openai_prompt( "chat_id": thread.chat_id, "display_name": thread.display_name, }, + "conversation_summary": conversation_summary_text, "kb_results": kb_context, "history": history, } @@ -2383,6 +2391,7 @@ def _openai_compatible_decision( messages: list[Any], kb_results: list[KBArticleRow], language: str, + conversation_summary_text: str = "", channel_label: str = "Telegram", channel_key: str = "telegram", ) -> dict[str, Any]: @@ -2394,6 +2403,7 @@ def _openai_compatible_decision( messages=messages, kb_results=kb_results, language=language, + conversation_summary_text=conversation_summary_text, channel_label=channel_label, channel_key=channel_key, ) @@ -2563,6 +2573,7 @@ def _decide_reply( messages: list[Any], kb_results: list[KBArticleRow], language: str, + conversation_summary_text: str = "", channel_label: str = "Telegram", channel_key: str = "telegram", ) -> dict[str, Any]: @@ -2577,6 +2588,7 @@ def _decide_reply( messages=messages, kb_results=kb_results, language=language, + conversation_summary_text=conversation_summary_text, channel_label=channel_label, channel_key=channel_key, ) @@ -2594,6 +2606,51 @@ def _decide_reply( return _sanitize_decision(raw, fallback_language=language) +def _update_ai_session_context_summary_from_user_turn( + ai_session: AISessionRow, + *, + channel: str, + language: str, + customer_name: str | None, + text: str, + now: str, +) -> str: + summary = update_context_summary_from_user_turn( + ai_session.context_summary_json, + channel=channel, + language=language, + customer_name=customer_name, + text=text, + now=now, + ) + ai_session.context_summary_json = dump_context_summary(summary) + ai_session.context_summary_updated_at = now + return render_context_summary_text(summary) + + +def _update_ai_session_context_summary_from_assistant_turn( + ai_session: AISessionRow, + *, + language: str, + customer_name: str | None, + reply_text: str, + decision: dict[str, Any], + now: str, +) -> None: + summary = update_context_summary_from_assistant_turn( + ai_session.context_summary_json, + language=language, + customer_name=customer_name, + reply_text=reply_text, + decision_intent=str(decision.get("intent") or "").strip() or None, + needs_handoff=bool(decision.get("needs_handoff")), + handoff_reason=str(decision.get("handoff_reason") or "").strip() or None, + now=now, + ) + ai_session.context_summary_json = dump_context_summary(summary) + ai_session.context_summary_updated_at = now + + def _update_thread_after_close(session, thread: TelegramThreadRow, when: str) -> None: thread.status = "closed" thread.ai_state = "closed" @@ -2692,6 +2749,7 @@ def _process_job(job_id: str) -> dict[str, Any]: ai_session.customer_id = customer_id ai_session.last_user_message_id = trigger_message.message_id ai_session.updated_at = utc_now_iso() + user_turn_now = ai_session.updated_at _record_turn( session, session_id=ai_session.session_id, @@ -2702,6 +2760,14 @@ def _process_job(job_id: str) -> dict[str, Any]: text=trigger_message.text, payload={"message_id": trigger_message.message_id, "author_type": trigger_message.author_type}, ) + conversation_summary_text = _update_ai_session_context_summary_from_user_turn( + ai_session, + channel="telegram", + language=ai_session.language or "ru", + customer_name=customer.display_name if customer else thread.display_name, + text=trigger_message.text, + now=user_turn_now, + ) session.commit() messages = _last_messages(session, thread.thread_id, _ai_max_context_messages()) @@ -2713,6 +2779,7 @@ def _process_job(job_id: str) -> dict[str, Any]: messages=messages, kb_results=kb_results, language=ai_session.language or "ru", + conversation_summary_text=conversation_summary_text, ) decision = _apply_always_reply_mode(decision, last_user_text=trigger_message.text) _record_turn( @@ -2728,9 +2795,8 @@ def _process_job(job_id: str) -> dict[str, Any]: finish_reason=decision["_finish_reason"], latency_ms=decision["_latency_ms"], ) - ai_session.updated_at = utc_now_iso() - ai_session.summary_text = decision["reply_text"] or (decision["handoff_reason"] or ai_session.summary_text) - session.commit() + decision_now = utc_now_iso() + ai_session.updated_at = decision_now needs_handoff = False if not _ai_telegram_always_reply(): @@ -2768,6 +2834,15 @@ def _process_job(job_id: str) -> dict[str, Any]: ai_session.status = "handoff_required" ai_session.handoff_reason = reason ai_session.updated_at = utc_now_iso() + ai_session.summary_text = decision["reply_text"] or (reason or ai_session.summary_text) + _update_ai_session_context_summary_from_assistant_turn( + ai_session, + language=decision["language"], + customer_name=customer.display_name if customer else thread.display_name, + reply_text=decision["reply_text"] or reason, + decision=decision, + now=ai_session.updated_at, + ) _mark_job_done(session, job) session.commit() return {"ok": True, "status": "handoff_required", "job_id": job.job_id} @@ -2799,6 +2874,17 @@ def _process_job(job_id: str) -> dict[str, Any]: return resolved raise + ai_session.updated_at = utc_now_iso() + ai_session.summary_text = decision["reply_text"] or (decision["handoff_reason"] or ai_session.summary_text) + _update_ai_session_context_summary_from_assistant_turn( + ai_session, + language=decision["language"], + customer_name=customer.display_name if customer else thread.display_name, + reply_text=decision["reply_text"], + decision=decision, + now=ai_session.updated_at, + ) + if decision["case_action"] == "escalate": _interaction_request( "POST", @@ -2926,6 +3012,7 @@ def _process_whatsapp_job(job_id: str) -> dict[str, Any]: ai_session.customer_id = customer_id ai_session.last_user_message_id = trigger_message.message_id ai_session.updated_at = utc_now_iso() + user_turn_now = ai_session.updated_at _record_turn( session, session_id=ai_session.session_id, @@ -2936,6 +3023,14 @@ def _process_whatsapp_job(job_id: str) -> dict[str, Any]: text=trigger_message.text, payload={"message_id": trigger_message.message_id, "author_type": trigger_message.author_type}, ) + conversation_summary_text = _update_ai_session_context_summary_from_user_turn( + ai_session, + channel="whatsapp", + language=ai_session.language or "ru", + customer_name=customer.display_name if customer else thread.display_name, + text=trigger_message.text, + now=user_turn_now, + ) session.commit() messages = _last_whatsapp_messages(session, thread.thread_id, _ai_whatsapp_max_context_messages()) @@ -2951,6 +3046,7 @@ def _process_whatsapp_job(job_id: str) -> dict[str, Any]: messages=messages, kb_results=kb_results, language=ai_session.language or "ru", + conversation_summary_text=conversation_summary_text, channel_label="WhatsApp", channel_key="whatsapp", ) @@ -2974,8 +3070,6 @@ def _process_whatsapp_job(job_id: str) -> dict[str, Any]: latency_ms=decision["_latency_ms"], ) ai_session.updated_at = utc_now_iso() - ai_session.summary_text = decision["reply_text"] or (decision["handoff_reason"] or ai_session.summary_text) - session.commit() needs_handoff = False if not _ai_whatsapp_always_reply(): @@ -3013,6 +3107,15 @@ def _process_whatsapp_job(job_id: str) -> dict[str, Any]: ai_session.status = "handoff_required" ai_session.handoff_reason = reason ai_session.updated_at = utc_now_iso() + ai_session.summary_text = decision["reply_text"] or (reason or ai_session.summary_text) + _update_ai_session_context_summary_from_assistant_turn( + ai_session, + language=decision["language"], + customer_name=customer.display_name if customer else thread.display_name, + reply_text=decision["reply_text"] or reason, + decision=decision, + now=ai_session.updated_at, + ) _mark_job_done(session, job) session.commit() return {"ok": True, "status": "handoff_required", "job_id": job.job_id} @@ -3044,6 +3147,17 @@ def _process_whatsapp_job(job_id: str) -> dict[str, Any]: return resolved raise + ai_session.updated_at = utc_now_iso() + ai_session.summary_text = decision["reply_text"] or (decision["handoff_reason"] or ai_session.summary_text) + _update_ai_session_context_summary_from_assistant_turn( + ai_session, + language=decision["language"], + customer_name=customer.display_name if customer else thread.display_name, + reply_text=decision["reply_text"], + decision=decision, + now=ai_session.updated_at, + ) + if decision["case_action"] == "escalate": _interaction_request( "POST", diff --git a/services/ai_orchestrator_service/voice.py b/services/ai_orchestrator_service/voice.py index 30005ee..482a7ed 100644 --- a/services/ai_orchestrator_service/voice.py +++ b/services/ai_orchestrator_service/voice.py @@ -8,6 +8,14 @@ from typing import Any from fastapi import HTTPException from sqlalchemy import select +from services.shared.ai_context_summary import ( + context_summary_context_texts, + dump_context_summary, + load_context_summary, + render_context_summary_text, + update_context_summary_from_assistant_turn, + update_context_summary_from_user_turn, +) from services.shared.core import new_id, utc_now_iso from services.shared.db import get_session from services.shared.models import VoiceAIStartIn, VoiceAIStartOut, VoiceAITurnDecisionOut, VoiceAITurnIn, VoiceStartResult @@ -39,7 +47,7 @@ def _app(): def _voice_max_context_segments() -> int: app = _app() - return max(6, app._int_env("AI_VOICE_MAX_CONTEXT_SEGMENTS", 8)) + return max(6, app._int_env("AI_VOICE_MAX_CONTEXT_SEGMENTS", 12)) def _voice_disclosure_prefix(language: str) -> str: @@ -787,12 +795,29 @@ def _voice_text_key(text: str | None) -> str: def _voice_recent_caller_texts( transcript_window: list[VoiceTranscriptSegmentRow], *, - limit: int = 8, + limit: int = 12, ) -> list[str]: caller_texts = [str(segment.text or "").strip() for segment in transcript_window if segment.speaker == "caller"] return caller_texts[-max(limit, 1) :] +def _voice_context_texts_with_summary( + transcript_window: list[VoiceTranscriptSegmentRow], + context_summary: str | dict[str, Any] | None, +) -> list[str]: + result = context_summary_context_texts(context_summary) + result.extend(_voice_recent_caller_texts(transcript_window, limit=12)) + deduped: list[str] = [] + seen: set[str] = set() + for text in result: + normalized = _voice_text_key(text) + if not normalized or normalized in seen: + continue + seen.add(normalized) + deduped.append(str(text).strip()) + return deduped[-12:] + + def _voice_is_low_signal_caller_text(text: str | None) -> bool: normalized = _voice_text_key(text) if not normalized: @@ -1311,6 +1336,7 @@ def _voice_postprocess_reply_text( language: str, transcript_text: str, transcript_window: list[VoiceTranscriptSegmentRow], + context_summary: str | dict[str, Any] | None = None, reply_text: str, kb_results: list[Any], needs_handoff: bool, @@ -1318,7 +1344,7 @@ def _voice_postprocess_reply_text( normalized_reply = str(reply_text or "").strip() if not normalized_reply or needs_handoff: return normalized_reply - caller_texts = _voice_recent_caller_texts(transcript_window) + caller_texts = _voice_context_texts_with_summary(transcript_window, context_summary) prior_caller_texts = _voice_caller_context_before_current(caller_texts, transcript_text) active_topic_texts = _voice_service_context_texts(prior_caller_texts or caller_texts) active_topic_prompt = _voice_topic_prompt(language, active_topic_texts) if active_topic_texts else None @@ -1435,6 +1461,7 @@ def _voice_llm_prompt_messages( interaction: Interaction, transcript_text: str, transcript_window: list[VoiceTranscriptSegmentRow], + conversation_summary_text: str = "", kb_results: list[Any], name_value: str | None, name_status: str | None, @@ -1449,7 +1476,7 @@ def _voice_llm_prompt_messages( "interrupted": bool(segment.barge_in_interrupted), "created_at": segment.created_at, } - for segment in transcript_window[-6:] + for segment in transcript_window[-12:] ] kb_context = [ { @@ -1477,6 +1504,7 @@ def _voice_llm_prompt_messages( "last_user_text": transcript_text, "language": language, }, + "conversation_summary": conversation_summary_text, "kb_results": kb_context, "history": history, } @@ -1518,6 +1546,7 @@ def _voice_llm_decision( interaction: Interaction, transcript_text: str, transcript_window: list[VoiceTranscriptSegmentRow], + conversation_summary_text: str = "", kb_results: list[Any], name_value: str | None, name_status: str | None, @@ -1535,6 +1564,7 @@ def _voice_llm_decision( interaction=interaction, transcript_text=transcript_text, transcript_window=transcript_window, + conversation_summary_text=conversation_summary_text, kb_results=kb_results, name_value=name_value, name_status=name_status, @@ -1749,6 +1779,7 @@ def _voice_decision( interaction: Interaction, transcript_text: str, transcript_window: list[VoiceTranscriptSegmentRow], + context_summary: str | dict[str, Any] | None = None, kb_results: list[Any], disclosure_required: bool, customer_name_value: str | None = None, @@ -1758,10 +1789,11 @@ def _voice_decision( app = _app() normalized = str(transcript_text or "").strip() lower_text = normalized.lower() - caller_texts = _voice_recent_caller_texts(transcript_window) + caller_texts = _voice_context_texts_with_summary(transcript_window, context_summary) model = app._ai_model() v2_metadata = _voice_v2_metadata(transcript_text, request_metadata) reply_phase = _voice_reply_phase(request_metadata) + conversation_summary_text = render_context_summary_text(context_summary) if reply_phase == "early_plan": return _voice_early_plan( @@ -1854,6 +1886,7 @@ def _voice_decision( interaction=interaction, transcript_text=transcript_text, transcript_window=transcript_window, + conversation_summary_text=conversation_summary_text, kb_results=kb_results, name_value=customer_name_value or (customer.display_name if customer else None), name_status=customer_name_status, @@ -2415,6 +2448,16 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe source=effective_name_source, resolved_at=effective_name_resolved_at, ) + user_context_summary = update_context_summary_from_user_turn( + ai_session.context_summary_json, + channel="voice", + language=ai_session.language or "ru", + customer_name=effective_name_value or (customer.display_name if customer else None), + text=payload.transcript_text, + now=now, + ) + ai_session.context_summary_json = dump_context_summary(user_context_summary) + ai_session.context_summary_updated_at = now kb_results = [] if not early_plan_only: kb_results = app._kb_search( @@ -2429,6 +2472,7 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe interaction=interaction, transcript_text=payload.transcript_text, transcript_window=transcript_window, + context_summary=ai_session.context_summary_json, kb_results=kb_results, disclosure_required=disclosure_required, customer_name_value=effective_name_value, @@ -2439,6 +2483,7 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe language=decision["language"], transcript_text=payload.transcript_text, transcript_window=transcript_window, + context_summary=ai_session.context_summary_json, reply_text=str(decision.get("reply_text") or ""), kb_results=kb_results, needs_handoff=bool(decision.get("needs_handoff")), @@ -2546,6 +2591,18 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe voice_session.handoff_reason = decision["handoff_reason"] ai_session.status = "handoff_required" if decision["needs_handoff"] else "active" ai_session.handoff_reason = decision["handoff_reason"] + updated_context_summary = update_context_summary_from_assistant_turn( + ai_session.context_summary_json, + language=decision["language"], + customer_name=effective_name_value or (customer.display_name if customer else None), + reply_text=decision["reply_text"] or decision["handoff_reason"] or "", + decision_intent=decision["intent"], + needs_handoff=bool(decision["needs_handoff"]), + handoff_reason=decision["handoff_reason"], + now=now, + ) + ai_session.context_summary_json = dump_context_summary(updated_context_summary) + ai_session.context_summary_updated_at = now ai_session.summary_text = decision["summary_text"] or decision["reply_text"] or ai_session.summary_text ai_session.updated_at = now session.commit() diff --git a/services/shared/ai_context_summary.py b/services/shared/ai_context_summary.py new file mode 100644 index 0000000..5c5232d --- /dev/null +++ b/services/shared/ai_context_summary.py @@ -0,0 +1,434 @@ +import json +import re +from typing import Any + + +_LOW_SIGNAL_TEXTS = { + "", + "ага", + "алло", + "да", + "добрый", + "добрый день", + "здравствуйте", + "ладно", + "неа", + "нет", + "ой", + "ок", + "okay", + "понял", + "поняла", + "привет", + "слышу", + "слышно", + "угу", + "хм", + "хорошо", + "hi", + "hello", + "ясно", +} + +_CITY_ALIASES = { + "алма ата": "Алма-Ата", + "алмат": "Алмата", + "астан": "Астана", + "актау": "Актау", + "актоб": "Актобе", + "атырау": "Атырау", + "караганд": "Караганда", + "кокшетау": "Кокшетау", + "костан": "Костанай", + "кызылорд": "Кызылорда", + "орал": "Орал", + "павлодар": "Павлодар", + "петропавл": "Петропавловск", + "семей": "Семей", + "талдыкорган": "Талдыкорган", + "тар": "Тараз", + "туркестан": "Туркестан", + "уральск": "Уральск", + "шымкент": "Шымкент", +} + +_SERVICE_HINTS = ( + (("график", "работает", "часы работы", "режим работы", "jumys uaqyty"), "график работы", "schedule"), + (("адрес", "филиал", "офис", "отделени", "мкр", "микрорайон", "район", "мекен"), "филиал или адрес", "address"), + (("тариф", "цена", "стоимость", "оплат", "баға", "сколько стоит"), "тариф или стоимость", "price"), + (("статус", "заявк", "заказ", "өтінім", "обращени"), "статус заявки", "status"), + (("не работает", "ошибка", "проблем", "сбой", "интернет", "связь"), "техническая проблема", "problem"), +) + +_INTENT_SLOT_DEFAULTS = { + "address": ["city_or_branch"], + "price": ["service_or_tariff"], + "problem": ["problem_details"], + "schedule": ["city_or_branch"], + "status": ["request_number_or_phone"], +} + + +def _default_summary() -> dict[str, Any]: + return { + "version": 1, + "channel": None, + "language": None, + "customer_name": None, + "active_intent": None, + "active_request_text": None, + "scenario_state": None, + "confirmed_facts": { + "branch_hint": None, + "city": None, + "request_number": None, + "service_hint": None, + }, + "requested_clarifications": [], + "open_slots": [], + "updated_at": None, + } + + +def _text_key(value: str | None) -> str: + compact = re.sub(r"[^\w\s-]+", " ", str(value or "").lower(), flags=re.UNICODE) + return re.sub(r"\s+", " ", compact).strip() + + +def load_context_summary(raw: str | dict[str, Any] | None) -> dict[str, Any]: + summary = _default_summary() + payload: dict[str, Any] = {} + if isinstance(raw, dict): + payload = raw + elif isinstance(raw, str) and raw.strip(): + try: + parsed = json.loads(raw) + except json.JSONDecodeError: + parsed = {} + if isinstance(parsed, dict): + payload = parsed + for key in ("channel", "language", "customer_name", "active_intent", "active_request_text", "scenario_state", "updated_at"): + value = payload.get(key) + if value not in {None, ""}: + summary[key] = str(value).strip() + facts = payload.get("confirmed_facts") + if isinstance(facts, dict): + for key in summary["confirmed_facts"]: + value = facts.get(key) + if value not in {None, ""}: + summary["confirmed_facts"][key] = str(value).strip() + clarifications = payload.get("requested_clarifications") + if isinstance(clarifications, list): + summary["requested_clarifications"] = [str(item).strip() for item in clarifications if str(item).strip()] + open_slots = payload.get("open_slots") + if isinstance(open_slots, list): + summary["open_slots"] = [str(item).strip() for item in open_slots if str(item).strip()] + return summary + + +def dump_context_summary(summary: dict[str, Any]) -> str: + normalized = load_context_summary(summary) + return json.dumps(normalized, ensure_ascii=False, sort_keys=True) + + +def _is_low_signal_text(text: str | None) -> bool: + normalized = _text_key(text) + return normalized in _LOW_SIGNAL_TEXTS + + +def _canonical_city(candidate: str) -> str | None: + normalized = _text_key(candidate) + if not normalized: + return None + for alias, canonical in _CITY_ALIASES.items(): + if alias in normalized: + return canonical + tokens = [token for token in re.split(r"[\s-]+", normalized) if token] + if not tokens: + return None + trimmed = " ".join(tokens[:3]) + if len(trimmed) < 3: + return None + return " ".join(token.capitalize() for token in trimmed.split()) + + +def _extract_city(text: str | None) -> str | None: + source = str(text or "").strip() + if not source: + return None + lowered = _text_key(source) + for alias, canonical in _CITY_ALIASES.items(): + if alias in lowered: + return canonical + match = re.search( + r"(?:в городе|город(?:е)?|қала(?:да|сы)?|qala(?:da)?)\s+([A-Za-zА-Яа-яЁёІіҢңҒғҮүҰұҚқӨөҺһ\- ]{2,40})", + source, + flags=re.IGNORECASE, + ) + if not match: + return None + candidate = re.split(r"[,.!?;:]", match.group(1), maxsplit=1)[0].strip() + return _canonical_city(candidate) + + +def _extract_request_number(text: str | None) -> str | None: + normalized = str(text or "").strip() + if not normalized: + return None + match = re.search(r"\b[А-ЯA-Z]{0,3}\d{5,}\b", normalized, flags=re.IGNORECASE) + if match: + return match.group(0).strip() + match = re.search(r"\b\d{6,}\b", normalized) + if match: + return match.group(0).strip() + return None + + +def _extract_branch_hint(text: str | None) -> str | None: + normalized = _text_key(text) + if not normalized: + return None + if not any(marker in normalized for marker in ("адрес", "филиал", "офис", "отделени", "район", "мкр", "микрорайон", "мекен")): + return None + compact = re.sub(r"\s+", " ", str(text or "").strip()) + return compact[:160] if compact else None + + +def _extract_service_hint(text: str | None) -> tuple[str | None, str | None]: + normalized = _text_key(text) + if not normalized: + return None, None + for markers, label, intent in _SERVICE_HINTS: + if any(marker in normalized for marker in markers): + return label, intent + return None, None + + +def _intent_open_slots(intent: str | None, facts: dict[str, Any]) -> list[str]: + normalized_intent = str(intent or "").strip().lower() + city = str(facts.get("city") or "").strip() + branch_hint = str(facts.get("branch_hint") or "").strip() + service_hint = str(facts.get("service_hint") or "").strip() + request_number = str(facts.get("request_number") or "").strip() + if normalized_intent in {"schedule", "address"}: + if not city and not branch_hint: + return ["city_or_branch"] + if city and not branch_hint: + return ["branch_or_address"] + return [] + if normalized_intent == "price": + return [] if service_hint else ["service_or_tariff"] + if normalized_intent == "status": + return [] if request_number else ["request_number_or_phone"] + if normalized_intent == "problem": + return [] if service_hint else ["problem_details"] + return list(_INTENT_SLOT_DEFAULTS.get(normalized_intent, [])) + + +def _requested_clarifications_from_reply(text: str | None) -> list[str]: + normalized = _text_key(text) + if not normalized: + return [] + if any(marker in normalized for marker in ("город", "в городе", "қала")): + if any(marker in normalized for marker in ("филиал", "адрес", "офис", "отделени")): + return ["branch_or_address"] + return ["city_or_branch"] + if any(marker in normalized for marker in ("филиал", "адрес", "офис", "отделени", "район")): + return ["branch_or_address"] + if any(marker in normalized for marker in ("тариф", "стоимость", "цена", "услуга", "баға")): + return ["service_or_tariff"] + if any(marker in normalized for marker in ("номер заявки", "номер заказа", "телефон", "өтінім")): + return ["request_number_or_phone"] + if any(marker in normalized for marker in ("что именно", "уточните", "подскажите точнее")): + return ["goal"] + return [] + + +def _merge_open_slots( + current_slots: list[str], + *, + clarified_slots: list[str], + fallback_slots: list[str], +) -> tuple[list[str], list[str]]: + specific_current_slots = [slot for slot in current_slots if slot and slot != "goal"] + specific_fallback_slots = [slot for slot in fallback_slots if slot and slot != "goal"] + if clarified_slots: + if clarified_slots == ["goal"]: + preserved_slots = specific_current_slots or specific_fallback_slots + if preserved_slots: + return preserved_slots, preserved_slots + return clarified_slots, clarified_slots + if specific_current_slots and specific_fallback_slots and set(specific_current_slots) == set(specific_fallback_slots): + return specific_current_slots, specific_current_slots + return fallback_slots, current_slots + + +def _looks_like_request_statement(text: str | None) -> bool: + normalized = _text_key(text) + if not normalized: + return False + markers = ( + "интересует", + "как работает", + "какой график", + "мне надо", + "мне нужен", + "нужно", + "подключ", + "подскажите", + "узнать", + "хочу", + ) + return any(marker in normalized for marker in markers) + + +def render_context_summary_text(raw: str | dict[str, Any] | None) -> str: + summary = load_context_summary(raw) + parts: list[str] = [] + if summary["customer_name"]: + parts.append(f"Customer name: {summary['customer_name']}") + if summary["language"]: + parts.append(f"Language: {summary['language']}") + if summary["active_intent"]: + parts.append(f"Active intent: {summary['active_intent']}") + if summary["active_request_text"]: + parts.append(f"Active request: {summary['active_request_text']}") + facts = summary["confirmed_facts"] + fact_parts = [] + if facts.get("city"): + fact_parts.append(f"city={facts['city']}") + if facts.get("branch_hint"): + fact_parts.append(f"branch={facts['branch_hint']}") + if facts.get("service_hint"): + fact_parts.append(f"service={facts['service_hint']}") + if facts.get("request_number"): + fact_parts.append(f"request_number={facts['request_number']}") + if fact_parts: + parts.append("Confirmed facts: " + ", ".join(fact_parts)) + if summary["open_slots"]: + parts.append("Open slots: " + ", ".join(summary["open_slots"])) + if summary["scenario_state"]: + parts.append(f"Scenario state: {summary['scenario_state']}") + return " | ".join(parts) + + +def context_summary_context_texts(raw: str | dict[str, Any] | None) -> list[str]: + summary = load_context_summary(raw) + result: list[str] = [] + if summary["active_request_text"]: + result.append(summary["active_request_text"]) + facts = summary["confirmed_facts"] + if facts.get("city"): + result.append(f"город {facts['city']}") + if facts.get("branch_hint"): + result.append(facts["branch_hint"]) + if facts.get("service_hint"): + result.append(facts["service_hint"]) + if summary["active_intent"] == "schedule": + result.append("график работы") + elif summary["active_intent"] == "address": + result.append("филиал или адрес") + elif summary["active_intent"] == "price": + result.append("тариф или стоимость") + elif summary["active_intent"] == "status": + result.append("статус заявки") + elif summary["active_intent"] == "problem": + result.append("техническая проблема") + seen: set[str] = set() + deduped: list[str] = [] + for item in result: + normalized = _text_key(item) + if not normalized or normalized in seen: + continue + seen.add(normalized) + deduped.append(str(item).strip()) + return deduped + + +def update_context_summary_from_user_turn( + raw: str | dict[str, Any] | None, + *, + channel: str, + language: str | None, + customer_name: str | None, + text: str | None, + now: str | None = None, +) -> dict[str, Any]: + summary = load_context_summary(raw) + summary["channel"] = str(channel or summary["channel"] or "").strip() or summary["channel"] + if language: + summary["language"] = str(language).strip() + if customer_name: + summary["customer_name"] = str(customer_name).strip() + compact_text = re.sub(r"\s+", " ", str(text or "").strip()) + if not compact_text or _is_low_signal_text(compact_text): + if now: + summary["updated_at"] = now + return summary + city = _extract_city(compact_text) + branch_hint = _extract_branch_hint(compact_text) + request_number = _extract_request_number(compact_text) + service_hint, inferred_intent = _extract_service_hint(compact_text) + should_refresh_request_text = ( + bool(city or branch_hint or request_number or service_hint) + or _looks_like_request_statement(compact_text) + or not str(summary.get("active_request_text") or "").strip() + ) + if should_refresh_request_text: + summary["active_request_text"] = compact_text[:500] + if city: + summary["confirmed_facts"]["city"] = city + if branch_hint: + summary["confirmed_facts"]["branch_hint"] = branch_hint + if request_number: + summary["confirmed_facts"]["request_number"] = request_number + if service_hint: + summary["confirmed_facts"]["service_hint"] = service_hint + if inferred_intent: + summary["active_intent"] = inferred_intent + summary["scenario_state"] = f"{inferred_intent}_in_progress" + summary["open_slots"] = _intent_open_slots(summary["active_intent"], summary["confirmed_facts"]) + if now: + summary["updated_at"] = now + return summary + + +def update_context_summary_from_assistant_turn( + raw: str | dict[str, Any] | None, + *, + language: str | None, + customer_name: str | None, + reply_text: str | None, + decision_intent: str | None, + needs_handoff: bool, + handoff_reason: str | None, + now: str | None = None, +) -> dict[str, Any]: + summary = load_context_summary(raw) + if language: + summary["language"] = str(language).strip() + if customer_name: + summary["customer_name"] = str(customer_name).strip() + if decision_intent and decision_intent not in {"clarification", "kb_answer", "unknown"}: + summary["active_intent"] = str(decision_intent).strip() + if needs_handoff: + summary["scenario_state"] = "handoff_requested" + summary["open_slots"] = [] + summary["requested_clarifications"] = [] + else: + clarifications = _requested_clarifications_from_reply(reply_text) + fallback_slots = _intent_open_slots(summary["active_intent"], summary["confirmed_facts"]) + merged_open_slots, merged_requested_clarifications = _merge_open_slots( + list(summary["open_slots"]), + clarified_slots=clarifications, + fallback_slots=fallback_slots, + ) + summary["open_slots"] = merged_open_slots + summary["requested_clarifications"] = merged_requested_clarifications + if summary["active_intent"]: + summary["scenario_state"] = f"{summary['active_intent']}_in_progress" + if handoff_reason and needs_handoff: + summary["active_request_text"] = summary["active_request_text"] or str(handoff_reason).strip() + if now: + summary["updated_at"] = now + return summary diff --git a/services/shared/sql_init.py b/services/shared/sql_init.py index f97509b..faa539d 100644 --- a/services/shared/sql_init.py +++ b/services/shared/sql_init.py @@ -375,11 +375,20 @@ def _apply_runtime_schema_compatibility() -> None: if "ai_sessions" in table_names: columns = _table_columns(inspector, "ai_sessions") _add_column_if_missing(conn, columns, "ai_sessions", "call_id", "VARCHAR(128)") + _add_column_if_missing(conn, columns, "ai_sessions", "context_summary_json", "TEXT DEFAULT '{}'") + _add_column_if_missing(conn, columns, "ai_sessions", "context_summary_updated_at", "VARCHAR(64)") indexes = _table_indexes(inspector, "ai_sessions") if "idx_ai_sessions_call_id" not in indexes: conn.execute( text("CREATE INDEX IF NOT EXISTS idx_ai_sessions_call_id ON ai_sessions(call_id)") ) + if "idx_ai_sessions_context_summary_updated_at" not in indexes: + conn.execute( + text( + "CREATE INDEX IF NOT EXISTS idx_ai_sessions_context_summary_updated_at " + "ON ai_sessions(context_summary_updated_at)" + ) + ) if "voice_events" in table_names: columns = _table_columns(inspector, "voice_events") diff --git a/services/shared/sql_models.py b/services/shared/sql_models.py index 6baf172..58d04aa 100644 --- a/services/shared/sql_models.py +++ b/services/shared/sql_models.py @@ -372,6 +372,8 @@ class AISessionRow(Base): language: Mapped[str | None] = mapped_column(String(16), nullable=True, index=True) status: Mapped[str] = mapped_column(String(32), index=True, default="active") summary_text: Mapped[str] = mapped_column(Text, default="") + context_summary_json: Mapped[str] = mapped_column(Text, default="{}") + context_summary_updated_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) last_user_message_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) last_ai_message_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) handoff_reason: Mapped[str | None] = mapped_column(Text, nullable=True) diff --git a/tests/test_ai_context_summary.py b/tests/test_ai_context_summary.py new file mode 100644 index 0000000..8c195dd --- /dev/null +++ b/tests/test_ai_context_summary.py @@ -0,0 +1,99 @@ +import json + +from services.shared.ai_context_summary import ( + render_context_summary_text, + update_context_summary_from_assistant_turn, + update_context_summary_from_user_turn, +) + + +def test_context_summary_tracks_city_and_branch_slot_for_schedule(): + summary = update_context_summary_from_user_turn( + None, + channel="voice", + language="ru", + customer_name="Ернур", + text="Мне нужен график работы в городе Алмата.", + now="2026-04-12T00:00:00+00:00", + ) + + assert summary["channel"] == "voice" + assert summary["customer_name"] == "Ернур" + assert summary["active_intent"] == "schedule" + assert summary["confirmed_facts"]["city"] == "Алмата" + assert summary["open_slots"] == ["branch_or_address"] + + +def test_context_summary_does_not_overwrite_meaningful_request_with_low_signal(): + summary = update_context_summary_from_user_turn( + None, + channel="voice", + language="ru", + customer_name="Ернур", + text="Мне нужен график работы в городе Алмата.", + now="2026-04-12T00:00:00+00:00", + ) + + updated = update_context_summary_from_user_turn( + json.dumps(summary, ensure_ascii=False), + channel="voice", + language="ru", + customer_name="Ернур", + text="Алло", + now="2026-04-12T00:00:02+00:00", + ) + + assert updated["active_request_text"] == "Мне нужен график работы в городе Алмата." + assert updated["confirmed_facts"]["city"] == "Алмата" + + +def test_context_summary_assistant_turn_keeps_slots_and_renders_text(): + summary = update_context_summary_from_user_turn( + None, + channel="telegram", + language="ru", + customer_name="Клиент", + text="Хочу узнать график работы в городе Алмата", + now="2026-04-12T00:00:00+00:00", + ) + + updated = update_context_summary_from_assistant_turn( + json.dumps(summary, ensure_ascii=False), + language="ru", + customer_name="Клиент", + reply_text="Подскажите, какой филиал или адрес в этом городе вас интересует?", + decision_intent="clarification", + needs_handoff=False, + handoff_reason=None, + now="2026-04-12T00:00:03+00:00", + ) + + rendered = render_context_summary_text(updated) + assert updated["open_slots"] == ["branch_or_address"] + assert "city=Алмата" in rendered + assert "график работы" in rendered + + +def test_context_summary_generic_clarification_does_not_downgrade_specific_slot(): + summary = update_context_summary_from_user_turn( + None, + channel="telegram", + language="ru", + customer_name="Клиент", + text="Хочу узнать график работы в городе Алмата", + now="2026-04-12T00:00:00+00:00", + ) + + updated = update_context_summary_from_assistant_turn( + json.dumps(summary, ensure_ascii=False), + language="ru", + customer_name="Клиент", + reply_text="Уточните, пожалуйста, что именно нужно проверить или подсказать.", + decision_intent="clarification", + needs_handoff=False, + handoff_reason=None, + now="2026-04-12T00:00:03+00:00", + ) + + assert updated["open_slots"] == ["branch_or_address"] + assert updated["requested_clarifications"] == ["branch_or_address"] diff --git a/tests/test_ai_orchestrator_service.py b/tests/test_ai_orchestrator_service.py index 506b179..c5c5c2b 100644 --- a/tests/test_ai_orchestrator_service.py +++ b/tests/test_ai_orchestrator_service.py @@ -1590,6 +1590,71 @@ def test_turn_voice_session_early_plan_does_not_persist_partial_turns(monkeypatc session.close() +def test_voice_llm_prompt_includes_context_summary_and_uses_12_segments(): + transcript_window = [ + SimpleNamespace( + speaker="caller" if index % 2 == 0 else "assistant", + text=f"segment {index}", + sequence_no=index, + source_type="voice_policy", + barge_in_interrupted=False, + created_at=utc_now_iso(), + ) + for index in range(1, 16) + ] + + messages = voice_module._voice_llm_prompt_messages( + language="ru", + customer=None, + interaction=SimpleNamespace( + interaction_id="int_voice_prompt", + status="open", + queue_id="que_voice", + subject="schedule", + customer_id="cus_voice_prompt", + ), + transcript_text="Мне нужен график работы", + transcript_window=transcript_window, + conversation_summary_text="Customer name: Ернур | Active intent: schedule | Confirmed facts: city=Алмата", + kb_results=[], + name_value="Ернур", + name_status="name_obtained", + ) + + payload = json.loads(messages[1]["content"]) + assert payload["conversation_summary"].startswith("Customer name: Ернур") + assert len(payload["history"]) == 12 + assert payload["history"][0]["sequence_no"] == 4 + + +def test_voice_postprocess_reply_uses_summary_context_when_raw_window_lost_topic(): + reply_text = voice_module._voice_postprocess_reply_text( + language="ru", + transcript_text="Сколько раз повторять тебе?", + transcript_window=[ + SimpleNamespace(speaker="caller", text="Сколько раз повторять тебе?", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()), + ], + context_summary=json.dumps( + { + "customer_name": "Ернур", + "active_intent": "schedule", + "active_request_text": "Мне нужен график работы в городе Алмата", + "confirmed_facts": {"city": "Алмата", "branch_hint": None, "service_hint": "график работы", "request_number": None}, + "open_slots": ["branch_or_address"], + }, + ensure_ascii=False, + ), + reply_text="Пожалуйста, уточните, какая услуга вас интересует, чтобы я мог подсказать тариф.", + kb_results=[], + needs_handoff=False, + ) + + normalized = reply_text.lower() + assert "тариф" not in normalized + assert "услуг" not in normalized + assert "филиал" in normalized or "адрес" in normalized + + def test_ai_enqueue_creates_outbound_ai_reply_and_delivery_flow(monkeypatch): monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1") monkeypatch.setenv("AI_PROVIDER", "stub") @@ -1639,6 +1704,61 @@ def test_ai_enqueue_creates_outbound_ai_reply_and_delivery_flow(monkeypatch): assert delivered.json()[-1]["telegram_message_id_external"] == "5551" +def test_ai_enqueue_persists_internal_context_summary(monkeypatch): + monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1") + monkeypatch.setenv("AI_PROVIDER", "stub") + monkeypatch.setattr(telegram_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None) + monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None) + + telegram_client = TestClient(telegram_app) + interaction_client = TestClient(interaction_app) + ai_client = TestClient(ai_app) + patch_ai_internal_calls(monkeypatch, telegram_client, interaction_client) + + created = create_inbound_thread( + telegram_client, + f"chat_ai_context_summary_{new_id('chat')}", + "Хочу узнать график работы в городе Алмата", + ) + enqueue = ai_client.post( + f"/ai/telegram/threads/{created['thread_id']}/enqueue", + headers=admin_headers(), + json={"trigger_message_id": created["message_id"]}, + ) + + assert enqueue.status_code == 200 + assert enqueue.json()["status"] in {"done", "handoff_required", "running"} + + deadline = time.time() + 2.0 + last_summary = None + while time.time() < deadline: + session = get_session() + try: + thread = session.execute( + select(TelegramThreadRow).where(TelegramThreadRow.thread_id == created["thread_id"]) + ).scalar_one() + if not thread.ai_session_id: + time.sleep(0.05) + continue + ai_session = session.execute( + select(AISessionRow).where(AISessionRow.session_id == thread.ai_session_id) + ).scalar_one() + last_summary = json.loads(ai_session.context_summary_json or "{}") + if ( + last_summary.get("active_intent") == "schedule" + and last_summary.get("confirmed_facts", {}).get("city") == "Алмата" + ): + break + finally: + session.close() + time.sleep(0.05) + + assert last_summary is not None + assert last_summary["active_intent"] == "schedule" + assert last_summary["confirmed_facts"]["city"] == "Алмата" + assert last_summary["open_slots"] == ["branch_or_address"] + + def test_ai_enqueue_relaxed_kb_search_answers_phrase_query(monkeypatch): monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1") monkeypatch.setenv("AI_PROVIDER", "stub") diff --git a/tests/test_sql_init_compat.py b/tests/test_sql_init_compat.py index f92b723..da50a1f 100644 --- a/tests/test_sql_init_compat.py +++ b/tests/test_sql_init_compat.py @@ -64,3 +64,42 @@ def test_init_sql_schema_backfills_whatsapp_message_runtime_columns(): assert "author_type" in columns assert "ix_whatsapp_messages_chat_external_unique" in indexes assert "idx_whatsapp_messages_next_delivery_attempt_at" in indexes + + +def test_init_sql_schema_backfills_ai_session_context_summary_columns(): + with engine.begin() as conn: + conn.execute(text("DROP TABLE IF EXISTS ai_sessions")) + conn.execute( + text( + """ + CREATE TABLE ai_sessions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id VARCHAR(64) NOT NULL UNIQUE, + channel VARCHAR(32) NOT NULL, + thread_id VARCHAR(64) NULL, + interaction_id VARCHAR(64) NULL, + customer_id VARCHAR(64) NULL, + agent_profile VARCHAR(64) NOT NULL, + language VARCHAR(16) NULL, + status VARCHAR(32) NOT NULL, + summary_text TEXT NOT NULL, + last_user_message_id VARCHAR(64) NULL, + last_ai_message_id VARCHAR(64) NULL, + handoff_reason TEXT NULL, + created_at VARCHAR(64) NOT NULL, + updated_at VARCHAR(64) NOT NULL, + closed_at VARCHAR(64) NULL + ) + """ + ) + ) + + init_sql_schema() + + inspector = inspect(engine) + columns = {item["name"] for item in inspector.get_columns("ai_sessions")} + indexes = {item["name"] for item in inspector.get_indexes("ai_sessions")} + + assert "context_summary_json" in columns + assert "context_summary_updated_at" in columns + assert "idx_ai_sessions_context_summary_updated_at" in indexes