feat: Made the SystemPrompt dynamic

Author: naynisinghal1008

.env.example

@@ -95,3 +95,12 @@ HITL_ACTIONS=create,update,delete,pause
 
 LIGHTHOUSE_APP_URL='http://localhost:5173'
 ENABLE_ALL_METRICS_FROM_CKH='true'
+
+
+# LangFuse Configuration
+ENABLE_LANGFUSE_PROMPTS=true
+LANGFUSE_SECRET_KEY=''
+LANGFUSE_PUBLIC_KEY=''
+LANGFUSE_BASEURL=https://us.cloud.langfuse.com
+LANGFUSE_SYSTEM_PROMPT_NAME=AUTOMATIC_VOICE_LANGFUSE_PROMPT
+LANGFUSE_SYSTEM_PROMPT_LABEL=automatic_system_langfuse_prompt

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

@@ -2,6 +2,7 @@
 from app.core.logger import logger
 from app.core.config import ENABLE_SEARCH_GROUNDING, HITL_ENABLE
 from app.agents.voice.automatic.types import TTSProvider
+from app.agents.voice.automatic.services.langfuse import langfuse_service
 
 SYSTEM_PROMPT = f"""
     SYSTEM ROLE
@@ -183,11 +184,39 @@ def get_tts_based_instructions(tts_provider: TTSProvider | None) -> str:
         """
     return ""
 
+def _process_template_variables(prompt_content: str) -> str:
+    """
+    Replace template variables in the prompt with actual values.
+    
+    Args:
+        prompt_content: The prompt content with template variables
+        
+    Returns:
+        str: Prompt content with template variables replaced
+    """
+    # Replace {current_time} with actual current date
+    current_time = datetime.datetime.now().strftime("%B %d, %Y")
+    prompt_content = prompt_content.replace("{current_time}", current_time)
+    
+    return prompt_content
+
+
 def get_system_prompt(user_name: str | None, tts_provider: TTSProvider | 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.
     """
-    prompt = SYSTEM_PROMPT
+    # Try to fetch prompt from LangFuse first
+    langfuse_prompt = langfuse_service.fetch_prompt()
+    
+    if langfuse_prompt:
+        logger.info("Using dynamic prompt from LangFuse")
+        prompt = _process_template_variables(langfuse_prompt)
+    else:
+        logger.info("Using fallback hardcoded prompt")
+        prompt = SYSTEM_PROMPT
+    
+    # Append dynamic components that are always added locally
     prompt += get_tts_based_instructions(tts_provider)
     prompt += get_internet_search_instructions()
     prompt += get_hitl_security_instructions()

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

@@ -0,0 +1,91 @@
+"""
+LangFuse service for dynamic prompt management.
+Handles fetching prompts from LangFuse with error handling.
+"""
+
+from typing import Optional
+from langfuse import Langfuse
+from app.core.logger import logger
+from app.core.config import (
+    ENABLE_LANGFUSE_PROMPTS,
+    LANGFUSE_SECRET_KEY,
+    LANGFUSE_PUBLIC_KEY,
+    LANGFUSE_BASEURL,
+    LANGFUSE_SYSTEM_PROMPT_NAME,
+    LANGFUSE_SYSTEM_PROMPT_LABEL
+)
+
+
+class LangFuseService:
+    """Service for managing LangFuse prompt operations."""
+    
+    def __init__(self):
+        self.client: Optional[Langfuse] = None
+        self.initialized = False
+        
+        if ENABLE_LANGFUSE_PROMPTS:
+            self._initialize_client()
+    
+    def _initialize_client(self) -> None:
+        """Initialize the LangFuse client with error handling."""
+        try:
+            if not LANGFUSE_SECRET_KEY or not LANGFUSE_PUBLIC_KEY:
+                logger.warning("LangFuse credentials not found, using fallback prompts")
+                return
+                
+            self.client = Langfuse(
+                secret_key=LANGFUSE_SECRET_KEY,
+                public_key=LANGFUSE_PUBLIC_KEY,
+                host=LANGFUSE_BASEURL
+            )
+            self.initialized = True
+            logger.info("LangFuse client initialized successfully")
+            
+        except Exception as e:
+            logger.error(f"Failed to initialize LangFuse client: {e}")
+            self.client = None
+    
+    def fetch_prompt(
+        self, 
+        prompt_name: Optional[str] = None, 
+        label: Optional[str] = None
+    ) -> Optional[str]:
+        """
+        Fetch prompt from LangFuse with error handling.
+        
+        Args:
+            prompt_name: Name of the prompt (defaults to config value)
+            label: Label for the prompt (defaults to config value)
+            
+        Returns:
+            str: Prompt content if successful, None if failed
+        """
+        if not ENABLE_LANGFUSE_PROMPTS or not self.initialized or not self.client:
+            logger.debug("LangFuse not enabled or not initialized, using fallback")
+            return None
+        
+        # Use default values if not provided
+        prompt_name = prompt_name or LANGFUSE_SYSTEM_PROMPT_NAME
+        label = label or LANGFUSE_SYSTEM_PROMPT_LABEL
+        
+        # Fetch from LangFuse
+        try:
+            logger.info(f"Fetching prompt '{prompt_name}' with label '{label}' from LangFuse")
+            
+            # Get prompt from LangFuse
+            prompt = self.client.get_prompt(prompt_name, label=label)
+            
+            if prompt and hasattr(prompt, 'prompt'):
+                prompt_content = prompt.prompt
+                logger.info(f"Successfully fetched prompt from LangFuse (length: {len(prompt_content)} chars)")
+                return prompt_content
+            else:
+                logger.warning(f"Prompt '{prompt_name}' with label '{label}' not found in LangFuse")
+                return None
+                
+        except Exception as e:
+            logger.error(f"Error fetching prompt from LangFuse: {e}")
+            return None
+        
+# Global instance
+langfuse_service = LangFuseService()

app/core/config.py

@@ -161,3 +161,16 @@ def get_required_env(var_name: str) -> str:
 # HITL Actions Configuration
 _hitl_actions_str = os.environ.get("HITL_ACTIONS", "delete")
 HITL_ACTIONS = [action.strip().lower() for action in _hitl_actions_str.split(",") if action.strip()]
+
+
+# LangFuse Configuration
+ENABLE_LANGFUSE_PROMPTS = os.environ.get("ENABLE_LANGFUSE_PROMPTS", "true").lower() == "false"
+LANGFUSE_SECRET_KEY = os.environ.get("LANGFUSE_SECRET_KEY", "")
+LANGFUSE_PUBLIC_KEY = os.environ.get("LANGFUSE_PUBLIC_KEY", "")
+LANGFUSE_BASEURL = os.environ.get("LANGFUSE_BASEURL", "https://us.cloud.langfuse.com")
+LANGFUSE_SYSTEM_PROMPT_NAME = os.environ.get("LANGFUSE_SYSTEM_PROMPT_NAME", "AUTOMATIC_VOICE_LANGFUSE_PROMPT")
+LANGFUSE_SYSTEM_PROMPT_LABEL = os.environ.get("LANGFUSE_SYSTEM_PROMPT_LABEL", "automatic_system_langfuse_prompt")
+
+logger.info(f"LangFuse prompts enabled: {ENABLE_LANGFUSE_PROMPTS}")
+if ENABLE_LANGFUSE_PROMPTS:
+    logger.info(f"LangFuse system prompt: {LANGFUSE_SYSTEM_PROMPT_NAME} (label: {LANGFUSE_SYSTEM_PROMPT_LABEL})")

memory-bank/progress.md

@@ -21,7 +21,15 @@
 - **Enhanced Banner Management:** The current banner management tool supports basic functionality for login and payment page announcements. Future enhancements could include support for more banner types, customization options (colors, icons), scheduling (start/end dates), and targeting specific user segments.
 - **Additional Shop Configuration Tools:** Building on the shop configuration utilities, more tools could be developed to manage other aspects of shop configuration, such as theme settings, payment options, or shipping methods.
 
-## 3. Known Issues & Risks
+## 3. Recent Improvements (Latest)
+
+- **Dynamic Prompt Management:** Implemented LangFuse integration to make system prompts dynamic instead of hardcoded. Created `LangFuseService` class for fetching prompts from external LangFuse platform.
+- **External Prompt Configuration:** Added comprehensive LangFuse configuration in `config.py` with environment variables for credentials, prompt names, and labels, enabling prompt management outside the codebase.
+- **Graceful Fallback System:** Modified `get_system_prompt()` to prioritize LangFuse prompts with automatic fallback to hardcoded `SYSTEM_PROMPT` if LangFuse is unavailable or fails.
+- **Template Variable Processing:** Added `{{current_time}}` template variable support in LangFuse prompts with dynamic replacement during prompt processing.
+- **No-Code Prompt Updates:** System prompts can now be updated through LangFuse interface without code changes or application deployments.
+
+## 4. Known Issues & Risks
 
 - **Sensitive Data Exposure:** As documented in `activeContext.md`, there is a known risk of exposing sensitive information (from tool schemas and results) to the LLM. This remains the most significant risk.
 - **Dependency on Remote Server:** While error handling has been improved, the agent's tooling capability is still entirely dependent on the availability and correctness of the remote MCP server.