va/int-layer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Optimization

Browse Source
This commit is contained in:
Anibal Angulo
2026-02-20 15:59:19 +00:00
parent ade4689ab7
commit 383efed319
12 changed files with 168 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

148
src/capa_de_integracion/services/conversation.py
View File

@@ -1,5 +1,6 @@
"""Conversation manager service for orchestrating user conversations."""
import asyncio
import logging
import re
from datetime import UTC, datetime, timedelta
@@ -22,6 +23,14 @@ from capa_de_integracion.services.storage.redis import RedisService
logger = logging.getLogger(__name__)
# Keep references to background tasks to prevent garbage collection
_background_tasks: set[asyncio.Task[None]] = set()
def get_background_tasks() -> set[asyncio.Task[None]]:
"""Return the set of pending background tasks (for graceful shutdown)."""
return _background_tasks
MSG_EMPTY_MESSAGE = "Message cannot be empty"
@@ -88,16 +97,16 @@ class ConversationManagerService:
# Step 1: Validate message is not empty
self._validate_message(request.mensaje)
# Step 2: Apply DLP security
obfuscated_message = await self.dlp_service.get_obfuscated_string(
request.mensaje,
self.settings.dlp_template_complete_flow,
# Step 2+3: Apply DLP security and obtain session in parallel
telefono = request.usuario.telefono
obfuscated_message, session = await asyncio.gather(
self.dlp_service.get_obfuscated_string(
request.mensaje,
self.settings.dlp_template_complete_flow,
),
self._obtain_or_create_session(telefono),
)
request.mensaje = obfuscated_message
telefono = request.usuario.telefono
# Step 3: Obtain or create session
session = await self._obtain_or_create_session(telefono)
# Step 4: Try quick reply path first
response = await self._handle_quick_reply_path(request, session)
@@ -131,6 +140,8 @@ class ConversationManagerService:
# Try Firestore if Redis miss
session = await self.firestore_service.get_session_by_phone(telefono)
if session:
# Cache to Redis for subsequent requests
await self.redis_service.save_session(session)
return session
# Create new session if both miss
@@ -165,27 +176,31 @@ class ConversationManagerService:
canal: Communication channel
"""
# Save user entry
# Save user and assistant entries in parallel.
# Use a single timestamp for both, but offset the assistant entry by 1µs
# to avoid Firestore document ID collision (save_entry uses isoformat()
# as the document ID).
now = datetime.now(UTC)
user_entry = ConversationEntry(
entity="user",
type=entry_type,
timestamp=datetime.now(UTC),
timestamp=now,
text=user_text,
parameters=None,
canal=canal,
)
await self.firestore_service.save_entry(session_id, user_entry)
# Save assistant entry
assistant_entry = ConversationEntry(
entity="assistant",
type=entry_type,
timestamp=datetime.now(UTC),
timestamp=now + timedelta(microseconds=1),
text=assistant_text,
parameters=None,
canal=canal,
)
await self.firestore_service.save_entry(session_id, assistant_entry)
await asyncio.gather(
self.firestore_service.save_entry(session_id, user_entry),
self.firestore_service.save_entry(session_id, assistant_entry),
)
async def _update_session_after_turn(
self,
@@ -204,8 +219,10 @@ class ConversationManagerService:
"""
session.last_message = last_message
session.last_modified = datetime.now(UTC)
await self.firestore_service.save_session(session)
await self.redis_service.save_session(session)
await asyncio.gather(
self.firestore_service.save_session(session),
self.redis_service.save_session(session),
)
async def _handle_quick_reply_path(
self,
@@ -253,17 +270,25 @@ class ConversationManagerService:
response.query_result.response_text if response.query_result else ""
) or ""
# Save conversation turn
await self._save_conversation_turn(
session_id=session.session_id,
user_text=request.mensaje,
assistant_text=response_text,
entry_type="CONVERSACION",
canal=getattr(request, "canal", None),
)
# Fire-and-forget: persist conversation turn and update session
async def _post_response() -> None:
try:
await asyncio.gather(
self._save_conversation_turn(
session_id=session.session_id,
user_text=request.mensaje,
assistant_text=response_text,
entry_type="CONVERSACION",
canal=getattr(request, "canal", None),
),
self._update_session_after_turn(session, response_text),
)
except Exception:
logger.exception("Error in quick-reply post-response work")
# Update session
await self._update_session_after_turn(session, response_text)
task = asyncio.create_task(_post_response())
_background_tasks.add(task)
task.add_done_callback(_background_tasks.discard)
return response
@@ -292,13 +317,19 @@ class ConversationManagerService:
telefono,
)
# Load conversation history only if session is older than threshold
# (optimization: new/recent sessions don't need history context)
# Load conversation history and notifications in parallel
session_age = datetime.now(UTC) - session.created_at
if session_age > timedelta(minutes=self.SESSION_RESET_THRESHOLD_MINUTES):
entries = await self.firestore_service.get_entries(
session.session_id,
limit=self.settings.conversation_context_message_limit,
load_history = session_age > timedelta(
minutes=self.SESSION_RESET_THRESHOLD_MINUTES,
)
if load_history:
entries, notifications = await asyncio.gather(
self.firestore_service.get_entries(
session.session_id,
limit=self.settings.conversation_context_message_limit,
),
self._get_active_notifications(telefono),
)
logger.info(
"Session is %s minutes old. Loaded %s conversation entries.",
@@ -307,13 +338,12 @@ class ConversationManagerService:
)
else:
entries = []
notifications = await self._get_active_notifications(telefono)
logger.info(
"Session is only %s minutes old. Skipping history load.",
session_age.total_seconds() / 60,
)
# Retrieve active notifications for this user
notifications = await self._get_active_notifications(telefono)
logger.info("Retrieved %s active notifications", len(notifications))
# Prepare current user message
@@ -344,27 +374,8 @@ class ConversationManagerService:
assistant_response[:100],
)
# Save conversation turn
await self._save_conversation_turn(
session_id=session.session_id,
user_text=request.mensaje,
assistant_text=assistant_response,
entry_type="LLM",
canal=getattr(request, "canal", None),
)
logger.info("Saved user message and assistant response to Firestore")
# Update session
await self._update_session_after_turn(session, assistant_response)
logger.info("Updated session in Firestore and Redis")
# Mark notifications as processed if any were included
if notifications:
await self._mark_notifications_as_processed(telefono)
logger.info("Marked %s notifications as processed", len(notifications))
# Return response object
return DetectIntentResponse(
# Build response object first, then fire-and-forget persistence
response = DetectIntentResponse(
responseId=str(uuid4()),
queryResult=QueryResult(
responseText=assistant_response,
@@ -373,6 +384,31 @@ class ConversationManagerService:
quick_replies=None,
)
# Fire-and-forget: persist conversation and update session
async def _post_response() -> None:
try:
coros = [
self._save_conversation_turn(
session_id=session.session_id,
user_text=request.mensaje,
assistant_text=assistant_response,
entry_type="LLM",
canal=getattr(request, "canal", None),
),
self._update_session_after_turn(session, assistant_response),
]
if notifications:
coros.append(self._mark_notifications_as_processed(telefono))
await asyncio.gather(*coros)
except Exception:
logger.exception("Error in post-response background work")
task = asyncio.create_task(_post_response())
_background_tasks.add(task)
task.add_done_callback(_background_tasks.discard)
return response
def _is_pantalla_context_valid(self, last_modified: datetime) -> bool:
"""Check if pantallaContexto is still valid (not stale)."""
time_diff = datetime.now(UTC) - last_modified