Handled user_name and merchant_id in db calls

Author: BurriKalyan

app/agents/voice/automatic/__init__.py

@@ -192,7 +192,7 @@ async def on_function_calls_started(service, function_calls):
         # If not found in memory, try loading from database
         if not existing_conversation:
             try:
-                existing_conversation = await conversation_manager.load_conversation_from_db(args.session_id)
+                existing_conversation = await conversation_manager.load_conversation_from_db(args.session_id, args.user_name, args.merchant_id)
                 if existing_conversation:
                     logger.info(f"Loaded conversation from database for session {args.session_id}")
             except Exception as e:
@@ -262,7 +262,7 @@ async def on_function_calls_started(service, function_calls):
 
     # Add processors for conversation tracking
     user_message_capture = UserMessageCaptureProcessor(args.session_id)
-    tool_call_processor = LLMSpyProcessor(rtvi, args.session_id)
+    tool_call_processor = LLMSpyProcessor(rtvi, args.session_id, args.user_name, args.merchant_id)
 
     # Build pipeline components
     pipeline_components = [

app/agents/voice/automatic/conversation_manager.py

@@ -64,7 +64,7 @@ async def stop(self):
             await self._cleanup_task
         logger.info("ConversationManager stopped")
 
-    def get_or_create_conversation(self, session_id: str) -> ConversationDebugData:
+    def get_or_create_conversation(self, session_id: str, user_id: Optional[str] = None, merchant_id: Optional[str] = None) -> ConversationDebugData:
         """Get existing conversation or create a new one for the session"""
         with self._lock:
             # Update access time
@@ -80,6 +80,8 @@ def get_or_create_conversation(self, session_id: str) -> ConversationDebugData:
                 conversation = ConversationDebugData(
                     session_id=session_id,
                     conversation_id=conversation_id,
+                    user_id=user_id,
+                    merchant_id=merchant_id,
                     metadata={
                         "created_by": "conversation_manager",
                         "max_turns": self.max_turns_per_session
@@ -346,11 +348,11 @@ async def _save_conversation_to_db(self, conversation: 'ConversationDebugData',
                 create_conversation, get_conversation_by_session, save_message, save_tool_call
             )
 
-            # Use fallback values if user context not provided
+            # Use values from conversation object if not provided as parameters
             if not user_id:
-                user_id = "unknown_user"  # This should be improved to get from session context
+                user_id = conversation.user_id or "unknown_user"
             if not merchant_id:
-                merchant_id = "unknown_merchant"  # This should be improved to get from session context
+                merchant_id = conversation.merchant_id or "unknown_merchant"
 
             # Check if conversation already exists in database
             existing_conv = await get_conversation_by_session(conversation.session_id, user_id, merchant_id)

app/agents/voice/automatic/processors/llm_spy.py

@@ -4,6 +4,7 @@
 """
 
 import time
+from typing import Optional
 
 from pipecat.frames.frames import (
     Frame, 
@@ -54,10 +55,12 @@ class LLMSpyProcessor(FrameProcessor):
     4. Processes highlight text for timing correlation
     """
 
-    def __init__(self, rtvi: RTVIProcessor, session_id: str, name: str = "LLMSpyProcessor"):
+    def __init__(self, rtvi: RTVIProcessor, session_id: str, user_id: Optional[str] = None, merchant_id: Optional[str] = None, name: str = "LLMSpyProcessor"):
         super().__init__(name=name)
         self._rtvi = rtvi
         self._session_id = session_id
+        self._user_id = user_id
+        self._merchant_id = merchant_id
 
         # LLM response collection
         self._accumulated_text = ""
@@ -81,6 +84,8 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
             user_content = _session_user_messages.get(self._session_id, "[Inferred from voice]")
 
             # Start conversation turn via ConversationManager with actual user message
+            # Ensure conversation is created with user_id and merchant_id
+            self._conversation_manager.get_or_create_conversation(self._session_id, self._user_id, self._merchant_id)
             event = await self._conversation_manager.start_turn_with_events(self._session_id, user_content)
             if event:
                 await self._emit_rtvi_event(event)
@@ -205,4 +210,3 @@ async def _emit_chart_components(self, function_name: str) -> None:
             pass
         except Exception as e:
             logger.error(f"Error emitting chart components for session {self._session_id}: {e}")
-

app/agents/voice/automatic/services/conversation.py

@@ -140,6 +140,8 @@ class ConversationDebugData(BaseModel):
     """Complete conversation data for debugging"""
     session_id: str
     conversation_id: str
+    merchant_id: Optional[str] = None
+    user_id: Optional[str] = None
     turns: List[ConversationTurn] = Field(default_factory=list)
     summary: Optional[ConversationSummary] = None
     metadata: Optional[Dict[str, Any]] = None