ai-response-according-to-user-preference

Author: sharifajahanshaik

app/agents/voice/automatic/__init__.py

@@ -280,8 +280,10 @@ async def run_normal_mode(args):
         )
     )
 
-    # Personalize the system prompt if a user name is provided
-    system_prompt = get_system_prompt(args.user_name, tts_provider, args.shop_id)
+    # Personalize the system prompt if a user name is provided (with user context from mem0)
+    system_prompt = await get_system_prompt(
+        args.user_name, tts_provider, args.shop_id, args.user_email
+    )
 
     # Configure VAD - use pre-initialized model if available
     global _silero_vad_cache

app/agents/voice/automatic/prompts/__init__.py

@@ -5,7 +5,9 @@
 from app.agents.voice.automatic.prompts.system.performance_directives import (
     get_combined_directives,
 )
-from app.agents.voice.automatic.prompts.system.personalization import append_user_info
+from app.agents.voice.automatic.prompts.system.personalization import (
+    create_personalized_prompt,
+)
 from app.agents.voice.automatic.prompts.system.tool_scope import (
     get_tool_scope_instructions,
 )
@@ -23,12 +25,16 @@
 from app.services.langfuse.prompts import fetch_prompt
 
 
-def get_system_prompt(
-    user_name: str | None, tts_provider: TTSProvider | None, shop_id: str | None
+async def get_system_prompt(
+    user_name: str | None,
+    tts_provider: TTSProvider | None,
+    shop_id: str | None,
+    user_email: str | None = None,
 ) -> str:
     """
     Generates a personalized system prompt based on the user's name and TTS service.
     First attempts to fetch from LangFuse, then falls back to hardcoded prompt.
+    Now includes user context retrieval from memory service.
     """
     langfuse_prompt = None
 
@@ -54,6 +60,7 @@ def get_system_prompt(
 
     if user_name:
         logger.info(f"Personalizing prompt for user: {user_name}")
-        prompt += append_user_info(user_name)
+        # Use create_personalized_prompt which handles mem0 configuration internally
+        prompt += await create_personalized_prompt(user_name, user_email)
 
     return prompt

app/agents/voice/automatic/prompts/system/personalization.py

@@ -1,8 +1,12 @@
-def append_user_info(user_name: str) -> str:
+def append_user_info(user_name: str, user_context: str = None) -> str:
     """
     Appends user personalization instructions to the system prompt.
+
+    Args:
+        user_name: The user's name for personalization
+        user_context: Optional memory-based context about the user's preferences and history
     """
-    return f"""
+    base_instructions = f"""
         USER PERSONALIZATION
         The user's name is {user_name}. Use it only when it adds genuine value to the conversation.
 
@@ -11,4 +15,110 @@ def append_user_info(user_name: str) -> str:
         - In **emotionally significant moments**, such as celebrating a win, expressing empathy, or addressing a concern directly.
 
         Avoid using the name in closing lines, suggestions, or tool-generated follow-ups unless absolutely necessary. Never repeat the name within the same message. Prioritize a warm, natural tone — use the name only when it feels truly warranted in spoken conversation.
+        
+        GREETING GUIDELINES:
+        - Keep initial greetings simple and natural
+        - Don't overwhelm users with multiple options or questions at the start
+    """
+
+    # Add general personalization instructions regardless of whether we have specific user context
+    personalization_instructions = """
+        
+        PERSONALIZATION INSTRUCTIONS:
+        This conversation may include user context from previous interactions. When available, you MUST:
+        
+        1. TOOL PARAMETER PERSONALIZATION
+        - When filling or suggesting tool parameters, prioritize previously used or preferred values.
+        - If the user has configured or used specific settings before, default to those unless they explicitly override them.
+        - Maintain consistency with the user's preferred formats, units, or naming conventions (e.g., currency, date format, region).
+        - Reference successful tool usage patterns from their history when proposing configurations.
+        - When no relevant memory is found, use safe, explainable defaults and clearly state that these can be customized.
+
+        2. SOLUTION PRIORITIZATION
+        - When multiple valid solutions exist, prioritize approaches that worked for this user in the past.
+        - Reference prior successful resolutions or troubleshooting steps they responded well to.
+        - Avoid methods the user has previously rejected, struggled with, or shown disinterest in.
+        - Build upon their existing knowledge level — neither over-explain nor skip crucial context.
+        - When introducing new concepts, connect them to something familiar from their past interactions.
+
+        3. PROACTIVE & CONTEXT-AWARE ASSISTANCE
+        - Anticipate next steps based on user patterns — e.g., follow-up actions they usually take.
+        - Address likely concerns early, before the user asks (e.g., known blockers, compatibility issues).
+        - Recommend optimizations or shortcuts that align with their prior workflow.
+        - Warn about issues they've encountered before or suggest preventive checks.
+        - When appropriate, remind the user why a suggestion matches their past behavior ("Since you usually prefer X, I've set Y accordingly.")
+
+        4. TOOL USAGE GUIDANCE
+        - Always check the conversation context and memory data before applying defaults.
+        - Respect the user's configuration history, even across sessions or devices.
+        - Keep data format consistency (JSON vs YAML, metric vs imperial, etc.) unless instructed otherwise.
+        - Adapt your explanations and verbosity to their known preference — concise vs detailed.
+        - Clearly log or communicate assumptions when inferring parameters from memory.
+
+    """
+
+    if user_context:
+        memory_section = f"""
+        
+        CURRENT USER CONTEXT:
+        {user_context}
+        """
+        return base_instructions + memory_section + personalization_instructions
+
+    return base_instructions
+
+
+async def create_personalized_prompt(user_name: str, user_email: str = None) -> str:
+    """
+    Create a comprehensive personalized prompt by retrieving user context from memory.
+
+    This function checks mem0 configuration internally and creates its own memory service
+    to retrieve user context if available.
+
+    Args:
+        user_name: The user's name
+        user_email: User's email for memory service (optional)
+
+    Returns:
+        Complete personalized prompt string with user context
     """
+    # Import here to avoid circular imports
+    from app.agents.voice.automatic.services.mem0.memory import (
+        ImprovedMem0MemoryService,
+    )
+    from app.core import config
+
+    # Check if mem0 is enabled and configured
+    if (
+        config.MEM0_ENABLED
+        and user_email
+        and user_email.strip()
+        and config.MEM0_API_KEY
+        and config.MEM0_API_KEY.strip()
+    ):
+
+        try:
+            # Create memory service instance internally
+            memory_params = ImprovedMem0MemoryService.InputParams()
+            memory_service = ImprovedMem0MemoryService(
+                api_key=config.MEM0_API_KEY,
+                user_id=user_email,
+                params=memory_params,
+            )
+            user_context_data = await memory_service.get_all_user_context()
+
+            if user_context_data.get("error"):
+                return append_user_info(user_name)
+
+            user_context = user_context_data.get("context", "")
+
+            if user_context:
+                return append_user_info(user_name, user_context)
+            else:
+                return append_user_info(user_name)
+
+        except Exception as e:
+            print(f"Error retrieving user context: {e}")
+            return append_user_info(user_name)
+    else:
+        return append_user_info(user_name)

app/agents/voice/automatic/services/mem0/memory.py

@@ -2,7 +2,7 @@
 import datetime
 import hashlib
 import time
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 from pipecat.frames.frames import Frame, LLMMessagesFrame
 from pipecat.processors.aggregators.openai_llm_context import (
@@ -147,6 +147,63 @@ def _handle_memory_success(self, operation: str):
             )
             self._consecutive_failures = 0
 
+    async def get_all_user_context(
+        self, user_id: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Retrieve all user context from mem0 for LLM personalization.
+        """
+        target_user_id = user_id or getattr(self, "user_id", None)
+
+        if not target_user_id:
+            return {"error": "No user ID available", "context": ""}
+
+        # Check circuit breaker
+        if not self._check_memory_health():
+            logger.info("User context retrieval skipped - circuit breaker OPEN")
+            return {"error": "Memory service unavailable", "context": ""}
+
+        try:
+            # Get ALL memories for the user
+            memories_result = self.memory_client.get_all(user_id=target_user_id)
+
+            # Extract memory text
+            memories = []
+            if isinstance(memories_result, list):
+                for item in memories_result:
+                    if isinstance(item, dict):
+                        memory_text = (
+                            item.get("memory")
+                            or item.get("text")
+                            or item.get("content")
+                            or str(item)
+                        )
+                        if memory_text and memory_text.strip():
+                            memories.append(memory_text.strip())
+
+            formatted_context = (
+                "\n".join(f"- {memory}" for memory in memories) if memories else ""
+            )
+
+            # Handle successful retrieval
+            self._handle_memory_success("context_retrieval")
+
+            return {
+                "user_id": target_user_id,
+                "context": formatted_context,
+                "memory_count": len(memories),
+            }
+
+        except Exception as e:
+            # Handle memory failure
+            self._handle_memory_failure("context_retrieval", e)
+
+            return {
+                "error": f"Context retrieval failed: {str(e)}",
+                "context": "",
+                "user_id": target_user_id,
+            }
+
     def _validate_and_clean_messages(
         self, messages: List[Dict[str, Any]]
     ) -> List[Dict[str, str]]: