# Enterprise Agent Nodes v2.0 - Premium $5M Enterprise Product Quality """ 4 Enterprise AI Modes: RAG, GraphRAG, Hybrid, Vision Features: - LLM-powered query understanding (ChatGPT-quality) - Query classification with confidence scoring - Multi-hop graph reasoning - Hybrid fusion with intelligent weighting - Vision → RAG/Graph pipeline - MCP tool integration - Clean, ChatGPT-style outputs - Dynamic visualization generation $5M Enterprise Product - Query understanding that works like ChatGPT. No hardcoded responses - LLM figures out intent from natural language. """ from core.llm import chat, get_optimal_model from vector.retriever import retrieve from graph.query import query_graph, revenue_dataframe, get_user_currency from agents.state import AgentState from agents.query_classifier import classify_query, QueryType from agents.confidence_scorer import ( get_confidence_scorer, AnswerConfidence, ConfidenceLevel ) import time import json from typing import List, Dict, Optional, Tuple # ============================================================================ # $5M ENHANCEMENT: LLM-Powered Query Dispatcher - ChatGPT-style understanding # ============================================================================ try: from agents.query_dispatcher import ( dispatch_query, get_query_dispatcher, QueryDispatch, QueryIntent as DispatcherIntent, ResponseFormat, VisualizationType, ) QUERY_DISPATCHER_AVAILABLE = True print("✓ Query Dispatcher loaded - LLM-powered query understanding active") except ImportError as e: QUERY_DISPATCHER_AVAILABLE = False print(f"⚠️ Query dispatcher not available: {e}") # Chart generation imports try: from core.charts import ( generate_customer_revenue_chart, generate_product_revenue_chart, generate_monthly_trend_chart, generate_prediction_chart, generate_revenue_bar_chart, generate_pie_chart ) CHARTS_AVAILABLE = True except ImportError: CHARTS_AVAILABLE = False # Interactive Plotly Charts (with hover/tooltips) try: from core.interactive_charts import ( generate_interactive_customer_chart, generate_interactive_product_chart, generate_interactive_trend_chart, generate_interactive_prediction_chart, generate_interactive_comparison_chart ) INTERACTIVE_CHARTS_AVAILABLE = True except ImportError: INTERACTIVE_CHARTS_AVAILABLE = False # Enterprise Engine imports try: from mcp.forecast_engine import ForecastEngine, forecast_from_dataframe from mcp.simulation_engine import SimulationEngine, simulate_scenarios from mcp.insight_engine import InsightEngine, generate_insights from mcp.prediction_engine import PredictionEngine, predict_revenue, predict_sales from mcp.chart_generator import ChartGenerator, generate_forecast_chart, generate_scenario_chart ENGINES_AVAILABLE = True except ImportError: ENGINES_AVAILABLE = False # Memory Pipeline imports (Cache RAG + 4-Layer Memory) try: from core.cache_rag import cache_lookup, cache_store, CacheHitType from core.memory_retrieval import memory_retrieve, memory_store, get_memory_pipeline MEMORY_AVAILABLE = True except ImportError: MEMORY_AVAILABLE = False # ChatGPT-style Answer-Based Chart Generation try: from agents.answer_chart_extractor import extract_chart_data_from_answer ANSWER_CHART_AVAILABLE = True except ImportError: ANSWER_CHART_AVAILABLE = False print("⚠️ Answer chart extractor not available") # ChatGPT Code Interpreter-style Charts (LLM understands any query) try: from agents.smart_chart import smart_chart SMART_CHART_AVAILABLE = True print("✓ Smart Chart loaded - LLM-driven visualization active") except ImportError: SMART_CHART_AVAILABLE = False print("⚠️ Smart Chart not available") # Company Intelligence imports try: from core.company_profile import ( get_company_profile, build_company_context, get_company_terminology, ) COMPANY_PROFILE_AVAILABLE = True except ImportError: COMPANY_PROFILE_AVAILABLE = False print("⚠️ Company profile module not available") # Role Intelligence imports try: from agents.role_templates import ( UserRole, get_role_from_string, get_role_prompt_modifier, format_response_for_role, ) ROLE_TEMPLATES_AVAILABLE = True except ImportError: ROLE_TEMPLATES_AVAILABLE = False print("⚠️ Role templates module not available") # Conversation Intelligence imports try: from core.memory_engine import ReferenceResolver, ShortTermMemory REFERENCE_RESOLVER_AVAILABLE = True # Singleton short-term memory for reference resolution _short_term_memory = ShortTermMemory() except ImportError: REFERENCE_RESOLVER_AVAILABLE = False _short_term_memory = None print("⚠️ Reference resolver not available") # Output Contracts and Chart Gatekeeping imports try: from agents.output_contracts import ( get_contract, get_max_tokens, can_show_chart, assess_data_confidence, format_confidence_response, ConfidenceLevel, ) from agents.chart_gatekeeper import ( should_render_chart, get_chart_decision, suggest_chart_type, ) OUTPUT_CONTRACTS_AVAILABLE = True except ImportError: OUTPUT_CONTRACTS_AVAILABLE = False print("⚠️ Output contracts module not available") # ============================================================================ # MEMORY ENGINE - Conversation Intelligence (ChatGPT-level stateful behavior) # ============================================================================ try: from core.memory_engine import ( ShortTermMemory, ReferenceResolver, MemoryType, get_shared_memory, # Import singleton getter ) # Use SHARED singleton instance - CRITICAL for chart context sharing _short_term_memory = get_shared_memory() _reference_resolver = ReferenceResolver(_short_term_memory) MEMORY_ENGINE_AVAILABLE = True print("✓ Memory engine loaded - using SHARED singleton instance") except ImportError as e: MEMORY_ENGINE_AVAILABLE = False _short_term_memory = None _reference_resolver = None print(f"⚠️ Memory engine not available: {e}") # MCP Router imports (Smart query-based MCP selection) try: from mcp.router import ( get_required_mcps, should_call_mcp, detect_intent, execute_mcp_with_fallback, QueryIntent, ) MCP_ROUTER_AVAILABLE = True except ImportError: MCP_ROUTER_AVAILABLE = False print("⚠️ MCP router not available") # Output Formatter imports (ChatGPT-quality responses) try: from agents.output_formatter import ( format_executive_response, format_factual_response, format_prediction_response, clean_ai_artifacts, clean_for_display, # New v2.0 full cleaning pipeline validate_numbers_in_response, calculate_response_confidence, format_metric_table, format_ranking_table, ) OUTPUT_FORMATTER_AVAILABLE = True except ImportError: OUTPUT_FORMATTER_AVAILABLE = False clean_for_display = None print("⚠️ Output formatter not available") # Query Decomposer imports (LLM-powered query enhancement v2.0) try: from core.query_decomposer import ( decompose_query, llm_decompose_query, classify_query_intent, expand_query_with_schema, QueryIntent, TemporalExpression, is_complex_query, merge_results, ) QUERY_DECOMPOSER_AVAILABLE = True print("✓ Query Decomposer v2.0 loaded - LLM-powered query enhancement active") except ImportError as e: QUERY_DECOMPOSER_AVAILABLE = False print(f"⚠️ Query decomposer not available: {e}") # Visualization Intelligence imports (ChatGPT-level charts) try: from agents.visualization_intelligence import ( decide_visualization, detect_visualization_intent, VisualizationIntent, ChartType, profile_dataframe, extract_count_from_query, detect_visualization_subject, ) from agents.advanced_charts import ( create_chart_from_decision, generate_dynamic_chart, ) from agents.visualization_intelligence import detect_visualization_intent VIZ_INTELLIGENCE_AVAILABLE = True except ImportError: VIZ_INTELLIGENCE_AVAILABLE = False print("⚠️ Visualization intelligence partially available") # $5M ENHANCEMENT: Response Enhancer - ChatGPT-quality responses try: from agents.response_enhancer import ( enhance_full_response, generate_proactive_insight, generate_followup_suggestions, enhance_response_tone, generate_helpful_error_recovery, extract_data_summary_from_response, ) RESPONSE_ENHANCER_AVAILABLE = True print("✓ Response enhancer loaded - ChatGPT-quality responses active") except ImportError as e: RESPONSE_ENHANCER_AVAILABLE = False print(f"⚠️ Response enhancer not available: {e}") # $5M ENHANCEMENT: ChatGPT-Style Handler - Pure LLM-driven responses try: from agents.chatgpt_handler import ( understand_query_with_llm, generate_chatgpt_prompt, should_generate_chart, get_mode_quality_settings, enhance_response_like_chatgpt, ) CHATGPT_HANDLER_AVAILABLE = True print("✓ ChatGPT handler loaded - LLM-driven responses active") except ImportError as e: CHATGPT_HANDLER_AVAILABLE = False print(f"⚠️ ChatGPT handler not available: {e}") # $5M CORE: Truth-Grounded AI - ChatGPT/Claude Level Principles try: from core.truth_grounded_prompt import ( get_truth_grounded_prompt, get_context as get_truth_context, update_context as update_truth_context, resolve_above_reference, validate_response_against_data, get_graceful_error_response, ConversationContext, ) TRUTH_GROUNDED_AVAILABLE = True print("✓ Truth-Grounded AI loaded - ChatGPT/Claude level active") except ImportError as e: TRUTH_GROUNDED_AVAILABLE = False print(f"⚠️ Truth-Grounded AI not available: {e}") # $5M CORE: Production-Grade Business AI Engine (ChatGPT/Claude Architecture) try: from core.business_ai_engine import ( get_memory as get_ai_memory, execute_query_pipeline, should_refuse, self_audit, get_production_system_prompt, ThreeLayerMemory, PipelineResult, QueryIntent, ) BUSINESS_AI_ENGINE_AVAILABLE = True print("✓ Business AI Engine loaded - Production-grade orchestration active") except ImportError as e: BUSINESS_AI_ENGINE_AVAILABLE = False print(f"⚠️ Business AI Engine not available: {e}") # $5M CORE: Production Mode Handler - Query Understanding + Mode-Specific Responses try: from core.mode_handler import ( ProductionModeHandler, understand_query, get_mode_prompt, handle_mode_query, AnalysisMode, MODE_PROMPTS, ) MODE_HANDLER_AVAILABLE = True print("✓ Production Mode Handler loaded - 5 modes active") except ImportError as e: MODE_HANDLER_AVAILABLE = False print(f"⚠️ Mode Handler not available: {e}") # $500K VISUAL INTELLIGENCE ENGINE - Mind Maps, Knowledge Graphs, Smart Visualization try: from core.visual_intelligence import ( VisualIntelligenceEngine, generate_visual_intelligence, generate_mind_map, generate_knowledge_graph, generate_kpi_dashboard, VisualType, ) from core.mermaid_renderer import ( create_mermaid_response, generate_mermaid_visual, create_visual_response, render_ascii_mind_map, render_ascii_relationship_graph, ) from core.premium_visuals import ( generate_premium_visual, generate_mind_map_plotly, generate_knowledge_graph_plotly, generate_kpi_dashboard_plotly, ) VISUAL_INTELLIGENCE_AVAILABLE = True print("✓ Visual Intelligence Engine loaded - Premium Plotly Graphs active") except ImportError as e: VISUAL_INTELLIGENCE_AVAILABLE = False print(f"⚠️ Visual Intelligence not available: {e}") # ============================================================================ # $5M ENHANCEMENT: Dynamic Prompt Builder - ChatGPT-Quality Response Generation # ============================================================================ def _build_dynamic_system_prompt( dispatch: 'QueryDispatch', company_name: str = "Your Company", currency_symbol: str = "₹", has_chart: bool = False ) -> str: """ Build a dynamic system prompt based on LLM query analysis. This is the CORE of ChatGPT-quality responses - the prompt adapts to what the user actually wants, not hardcoded templates. Args: dispatch: QueryDispatch from the LLM-powered dispatcher company_name: Name of the user's company currency_symbol: Currency symbol for formatting has_chart: Whether a chart will be shown Returns: Tailored system prompt for the LLM """ # Base system prompt - always present base = f"""You are a $5M Enterprise AI Business Analyst for {company_name}. You analyze ONLY the uploaded business data - never make up information. Always use {currency_symbol} for currency formatting. """ # Intent-specific instructions intent_instructions = { "factual": """ RESPONSE STYLE: FACTUAL - Give a direct, clear answer - Include the exact number/value - Keep it concise but complete""", "aggregation": """ RESPONSE STYLE: AGGREGATION - Calculate and present the total/average/count - Show the calculation clearly - Use markdown tables for multiple metrics""", "comparison": """ RESPONSE STYLE: COMPARISON - Present both sides fairly - Use a comparison table - Highlight key differences - State a clear winner if applicable""", "trend": """ RESPONSE STYLE: TREND ANALYSIS - Describe the direction (up/down/stable) - Include specific growth percentages - Identify peak and low points - Mention any seasonality patterns""", "ranking": """ RESPONSE STYLE: RANKING - Present as an ordered list or table - Include rank, name, and value - Highlight the top performer - Note any close competitors""", "breakdown": """ RESPONSE STYLE: BREAKDOWN - Show distribution/composition - Include percentages - Use bullet points or table - Highlight largest segments""", "prediction": """ RESPONSE STYLE: PREDICTION/FORECAST - State the prediction clearly - Include confidence level/range - List key assumptions - Note any risks or uncertainties""", "summary": """ RESPONSE STYLE: SUMMARY - Start with executive summary (2-3 sentences) - List key highlights - Include actionable insights - Keep it scannable""", "anomaly": """ RESPONSE STYLE: ANOMALY DETECTION - Identify unusual patterns - Explain why they're unusual - Suggest possible causes - Recommend actions""", } # Format-specific instructions format_instructions = { "one_line": """ CRITICAL FORMAT CONSTRAINT: ONE LINE ONLY Your ENTIRE response must be ONE sentence. No tables, no bullets, no paragraphs. Maximum 30 words.""", "brief": """ FORMAT: BRIEF (2-3 sentences) Keep response concise. No unnecessary elaboration.""", "detailed": """ FORMAT: DETAILED ANALYSIS Provide comprehensive analysis. Include tables, metrics, insights. Be thorough but structured.""", "table": """ FORMAT: TABLE PREFERRED Present data in markdown table format. Include headers and proper alignment.""", } # Build the prompt prompt_parts = [base] # Add intent instructions intent_key = dispatch.intent.value if hasattr(dispatch.intent, 'value') else str(dispatch.intent) if intent_key in intent_instructions: prompt_parts.append(intent_instructions[intent_key]) # Add format instructions format_key = dispatch.response_format.value if hasattr(dispatch.response_format, 'value') else str(dispatch.response_format) if format_key in format_instructions: prompt_parts.append(format_instructions[format_key]) # Add chart context if has_chart: prompt_parts.append(""" CHART INCLUDED: A visualization will be shown. Your text should COMPLEMENT the chart, not repeat all its data. Reference the chart in your response.""") # Add limit context if dispatch.limit: prompt_parts.append(f""" LIMIT: User asked for top/bottom {dispatch.limit} items. Only include {dispatch.limit} items in your response.""") # Anti-hallucination reminder prompt_parts.append(""" CRITICAL RULES: - Use ONLY data from the context provided - Never make up numbers or names - If data is missing, say so clearly - Do not add "As an AI..." phrases - Be confident and direct""") return "\n".join(prompt_parts) def _get_user_context(user_id: str) -> str: """Get personalized context for user""" try: from core.memory import get_user_context return get_user_context(user_id) except: return "" def build_data_context(user_id: str) -> dict: """ 🧠 CHATGPT-STYLE DATA CONTEXT BUILDER Loads user's actual data and builds comprehensive context for LLM. This is the CORE function that makes all modes work dynamically. Returns dict with: - df: The actual DataFrame - schema: Column names and types - sample: First 10 rows as text - stats: Aggregated statistics - context_string: Full text context for LLM prompts """ from api.v1.endpoints.charts import get_user_data result = { "df": None, "schema": {}, "sample": "", "stats": {}, "context_string": "", "row_count": 0, "columns": [] } try: df = get_user_data(user_id) if df is None or df.empty: result["context_string"] = "No data uploaded yet. Please upload files in Data Hub." return result result["df"] = df result["row_count"] = len(df) result["columns"] = list(df.columns) # Build schema info schema_info = {} numeric_cols = [] text_cols = [] for col in df.columns: dtype = str(df[col].dtype) schema_info[col] = dtype if 'int' in dtype or 'float' in dtype: numeric_cols.append(col) else: text_cols.append(col) result["schema"] = schema_info # Build statistics for numeric columns stats = {} for col in numeric_cols: try: stats[col] = { "sum": float(df[col].sum()), "mean": float(df[col].mean()), "min": float(df[col].min()), "max": float(df[col].max()) } except: pass result["stats"] = stats # Build sample rows (first 10) sample_rows = df.head(10).to_string(index=False) result["sample"] = sample_rows # Build full context string for LLM prompts context_parts = [ f"=== USER'S DATA ({len(df)} rows) ===", f"", f"COLUMNS: {', '.join(df.columns)}", f"", f"COLUMN TYPES:", ] for col, dtype in schema_info.items(): context_parts.append(f" - {col}: {dtype}") context_parts.append(f"") context_parts.append(f"NUMERIC COLUMNS: {', '.join(numeric_cols) if numeric_cols else 'None'}") context_parts.append(f"TEXT COLUMNS: {', '.join(text_cols) if text_cols else 'None'}") # Add key statistics if stats: context_parts.append(f"") context_parts.append(f"KEY STATISTICS:") for col, col_stats in stats.items(): context_parts.append(f" - {col}: Total={col_stats['sum']:,.2f}, Avg={col_stats['mean']:,.2f}") # Add sample data context_parts.append(f"") context_parts.append(f"SAMPLE DATA (first 10 rows):") context_parts.append(sample_rows) # Add categorical breakdowns for col in text_cols[:3]: # First 3 categorical columns try: unique_count = df[col].nunique() if 2 <= unique_count <= 20: value_counts = df[col].value_counts().head(10) context_parts.append(f"") context_parts.append(f"{col.upper()} DISTRIBUTION:") for val, count in value_counts.items(): context_parts.append(f" - {val}: {count} records") except: pass result["context_string"] = "\n".join(context_parts) print(f"[DATA CONTEXT] Built context for {user_id}: {len(df)} rows, {len(df.columns)} columns") except Exception as e: print(f"[DATA CONTEXT] Error: {e}") result["context_string"] = f"Error loading data: {str(e)}" return result def _format_currency(amount: float, user_id: str = None) -> str: """Format currency with proper symbol based on user preferences""" try: from core.currency_converter import format_currency, get_currency_symbol if user_id: symbol, code = get_user_currency(user_id) return format_currency(amount, code) except: pass # Default to INR try: from core.currency_converter import format_currency return format_currency(amount, "INR") except: return f"₹{amount:,.2f}" def _build_source_citations(sources: List[str], mode: str) -> str: """Build formatted source citations - simplified to avoid duplication with frontend Sources""" if not sources: return "" # PRODUCTION: No footer needed - frontend shows sources cleanly # Remove "Analysis Mode: X" debug footer for clean professional output return "" def _clean_response_formatting(answer: str, context: str = "") -> str: """ Clean response formatting for premium $5M product quality. Fixes table alignment, removes artifacts, ensures professional output. Now includes: - AI artifact removal ("As an AI...", "I'd be happy to...") - Anti-hallucination number validation """ if not answer: return answer import re # STEP 1: Clean AI artifacts (filler phrases) if OUTPUT_FORMATTER_AVAILABLE: answer = clean_ai_artifacts(answer) # Remove triple+ line breaks (keep max double line breaks) while '\n\n\n' in answer: answer = answer.replace('\n\n\n', '\n\n') # Remove trailing whitespace from each line lines = [line.rstrip() for line in answer.split('\n')] answer = '\n'.join(lines) # Remove double backticks artifacts (but preserve triple backticks for code blocks) answer = re.sub(r'(? str: """ Get recent chat history for context continuity. Returns formatted string of last N messages. """ if not user_id: return "" try: from pathlib import Path from utils.paths import get_user_paths import json paths = get_user_paths(user_id) memory_path = paths.get("memory", Path(f"storage/users/{user_id}/memory")) history_file = memory_path / "chat_history.json" if not history_file.exists(): return "" with open(history_file, 'r') as f: history = json.load(f) # Get last N exchanges recent = history[-limit:] if len(history) > limit else history formatted = [] for msg in recent: role = msg.get('role', 'user') content = msg.get('content', '')[:2000] # Increased limit for better context formatted.append(f"{role.upper()}: {content}") return "\n".join(formatted) except Exception as e: print(f"Chat history error: {e}") return "" def _save_chat_message(user_id: str, role: str, content: str) -> bool: """Save a chat message to history for future context.""" if not user_id: return False try: from pathlib import Path from utils.paths import get_user_paths import json paths = get_user_paths(user_id) memory_path = paths.get("memory", Path(f"storage/users/{user_id}/memory")) memory_path.mkdir(parents=True, exist_ok=True) history_file = memory_path / "chat_history.json" history = [] if history_file.exists(): with open(history_file, 'r') as f: history = json.load(f) # Add new message history.append({ 'role': role, 'content': content[:2000] # Increased limit for better context }) # Keep only last 20 messages history = history[-20:] with open(history_file, 'w') as f: json.dump(history, f, indent=2) return True except Exception as e: print(f"Save chat error: {e}") return False def _is_data_related_query(question: str) -> bool: """ Check if a question relates to business/uploaded data topics. Returns False for general knowledge questions like "What is Python?" """ q_lower = question.lower().strip() # Business data keywords - questions about THEIR data data_keywords = [ 'revenue', 'sales', 'customer', 'product', 'order', 'amount', 'total', 'top', 'bottom', 'highest', 'lowest', 'average', 'sum', 'trend', 'growth', 'profit', 'margin', 'cost', 'price', 'transaction', 'invoice', 'contract', 'deal', 'opportunity', 'my data', 'my file', 'uploaded', 'analyze', 'analysis', 'who', 'which', 'how much', 'how many', 'show me', 'list', 'compare', 'breakdown', 'segment', 'category', 'region', 'chart', 'graph', 'visual', 'forecast', 'predict', 'projection', 'best', 'worst', 'performing', 'underperforming', 'month', 'quarter', 'year', 'date', 'period', 'time', 'name', 'tell me my', 'what is my', 'remember', 'recall' ] # Check if it's about their data if any(kw in q_lower for kw in data_keywords): return True # General knowledge patterns to BLOCK general_patterns = [ 'what is', 'what are', 'who is', 'who are', 'explain', 'define', 'how does', 'how do', 'why is', 'why are', 'tell me about', 'write code', 'write a', 'create a script', 'generate code', 'programming', 'python', 'java', 'javascript', 'html', 'css', 'history of', 'meaning of', 'wikipedia', 'google' ] # If it matches general pattern AND doesn't have data keywords, block it if any(pattern in q_lower for pattern in general_patterns): # Double-check it's not asking about their data if not any(kw in q_lower for kw in ['my', 'our', 'the', 'this', 'uploaded', 'file', 'data', 'revenue', 'customer', 'product']): return False # Short greetings are OK if len(q_lower) < 15: return True # "hi", "hello", "thanks" etc. return True # Default: allow and let the AI figure it out def _extract_and_save_personal_info(question: str, user_id: str) -> Optional[str]: """ Extract personal info from question and save to memory. Returns the extracted name if found, None otherwise. """ if not user_id or not question: return None try: from core.memory import process_personal_info, get_user_context # Try to extract and save personal info was_saved = process_personal_info(user_id, question) if was_saved: # Return the saved context context = get_user_context(user_id) return context return None except Exception as e: print(f"Personal info extraction error: {e}") return None # ============================================================================ # $5M CORE FIX: Build Messages With Context - ChatGPT-Quality Conversation # ============================================================================ # THIS IS THE CRITICAL FIX - Pass actual conversation history to LLM as messages def build_messages_with_context( user_id: str, current_question: str, current_prompt: str, history_limit: int = 10 ) -> list: """ Build a proper message array with conversation history for the LLM. THIS IS THE KEY FIX FOR CONTEXT MEMORY! Instead of: chat("What is revenue?") # LLM has no memory We now do: chat([ {"role": "user", "content": "What is revenue?"}, {"role": "assistant", "content": "Your revenue is $13.8M"}, {"role": "user", "content": "Explain that"} # LLM knows context! ]) Args: user_id: User/session ID current_question: The current user question current_prompt: The full prompt to send (with context, data, etc.) history_limit: Max history messages to include Returns: List of message dicts for the LLM """ messages = [] # Get conversation history history_entries = [] if user_id: try: # Try ShortTermMemory first if MEMORY_ENGINE_AVAILABLE and _short_term_memory: history_entries = _short_term_memory.get_messages(user_id, limit=history_limit) else: # Fallback to file-based history from pathlib import Path from utils.paths import get_user_paths import json paths = get_user_paths(user_id) memory_path = paths.get("memory", Path(f"storage/users/{user_id}/memory")) history_file = memory_path / "chat_history.json" if history_file.exists(): with open(history_file, 'r') as f: history_entries = json.load(f)[-history_limit:] except Exception as e: print(f"[CONTEXT] History load error: {e}") # Add conversation history as message pairs for entry in history_entries: role = entry.get('role', 'user') content = entry.get('content', '') if role in ['user', 'assistant'] and content: # Truncate very long messages to stay within token limits if len(content) > 2000: content = content[:2000] + "... [truncated]" messages.append({ "role": role, "content": content }) # Add the current question/prompt as the final user message messages.append({ "role": "user", "content": current_prompt }) print(f"[CONTEXT] Built {len(messages)} messages ({len(history_entries)} history + 1 current)") return messages def resolve_references_in_query( question: str, user_id: str ) -> tuple: """ Resolve references like "it", "that", "above" using ReferenceResolver. Examples: - "Explain that" -> "Explain the previous revenue analysis" - "What about it?" -> "What about Customer_33?" - "Show more details" -> "Show more details about the monthly trend" Args: question: Original user question user_id: User session ID Returns: (resolved_question, extracted_entities) """ if not MEMORY_ENGINE_AVAILABLE or not _reference_resolver: return question, {} try: # Get session context session_context = None if _short_term_memory: session_context = _short_term_memory.get_context(user_id) if user_id else None # Resolve references resolved, entities = _reference_resolver.resolve( question, session_context=session_context, session_id=user_id ) if resolved != question: print(f"[CONTEXT] Resolved '{question}' -> '{resolved}'") print(f"[CONTEXT] Entities: {entities}") return resolved, entities except Exception as e: print(f"[CONTEXT] Resolution error: {e}") return question, {} def _get_data_topics_summary(user_id: str) -> str: """Get a summary of topics available in user's data for context.""" try: from graph.query import load_graph, get_graph_stats graph = load_graph(user_id) if not graph or graph.number_of_nodes() == 0: return "" stats = get_graph_stats(user_id) if not stats: return "" topics = [] if stats.get('total_nodes', 0) > 0: topics.append(f"{stats.get('total_nodes')} entities") if stats.get('total_edges', 0) > 0: topics.append(f"{stats.get('total_edges')} relationships") return f"Available data: {', '.join(topics)}" if topics else "" except: return "" def _wants_visualization(question: str) -> bool: """Check if user wants a chart/visualization""" viz_keywords = [ # General chart keywords 'chart', 'graph', 'plot', 'visualize', 'visualization', 'show me', 'display', 'draw', 'create a', 'generate', # Specific chart types 'bar chart', 'pie chart', 'line chart', 'trend chart', 'bar', 'pie', 'line', 'area', 'scatter', 'donut', 'treemap', 'heatmap', 'funnel', 'gauge', 'waterfall', 'horizontal bar', 'stacked bar', 'bubble', 'radar', # Visual intents 'breakdown', 'distribution', 'comparison', 'trend', 'top 5', 'top 10', 'top 6', 'top 7', 'top 3', 'top 20', 'bottom 5', 'bottom 10', 'bottom 3', 'by product', 'by customer', 'by category', 'by region' ] q_lower = question.lower() return any(kw in q_lower for kw in viz_keywords) def _wants_prediction(question: str) -> bool: """Check if user wants a prediction/forecast""" pred_keywords = [ 'predict', 'forecast', 'future', 'next month', 'next year', 'projection', 'estimate', 'will be', 'expected', 'growth', 'next 3', 'next 6', 'next 12', 'next week', 'next 7 days', 'next day', 'next two', 'next 2', 'next five', 'next 5', 'tomorrow', 'upcoming', 'for the next', 'in the next', 'days ahead', 'weeks ahead', 'months ahead', 'years ahead', 'customer growth', 'revenue growth' ] q_lower = question.lower() return any(kw in q_lower for kw in pred_keywords) def _wants_simulation(question: str) -> bool: """Check if user wants a what-if simulation""" sim_keywords = [ 'what if', 'what-if', 'simulate', 'simulation', 'scenario', 'if i increase', 'if i decrease', 'if we increase', 'if we decrease', 'price increase', 'price decrease', 'marketing spend', 'marketing increase', 'impact of', 'effect of', 'how would', '10% increase', '20% increase', 'elasticity', 'sensitivity', 'what happens if', 'churn impact' ] q_lower = question.lower() return any(kw in q_lower for kw in sim_keywords) def _wants_insights(question: str) -> bool: """Check if user wants automated insights analysis""" insight_keywords = [ 'insights', 'analyze my', 'analyse my', 'find risks', 'find opportunities', 'what are the risks', 'what are the opportunities', 'health score', 'business health', 'anomalies', 'detect anomalies', 'trends and risks', 'full analysis', 'complete analysis', 'generate insights', 'auto analyze', 'automated analysis', 'assess my', 'evaluate my' ] q_lower = question.lower() return any(kw in q_lower for kw in insight_keywords) # ============================================================================ # CONVERSATION INTELLIGENCE - ChatGPT-Level Follow-Up Handling # ============================================================================ def _detect_followup_type(question: str) -> str: """ Detect the type of follow-up query. Returns: 'explanation', 'conversion', 'comparison', 'refinement', 'hypothetical', 'evaluation', 'summarization', 'reformatting', 'self_explanation', or 'new_query' """ q_lower = question.lower() # ========================================================================== # FIRST: Check for "above" references - these are ALWAYS follow-ups # ========================================================================== above_patterns = ['above', 'that', 'this', 'previous', 'same', 'it '] has_above_reference = any(kw in q_lower for kw in above_patterns) # Reformatting - same content, different format (MUST CHECK EARLY) format_patterns = [ 'in one line', 'in 1 line', 'in a line', 'one line', 'in words', 'in text', 'in simple words', 'in plain words', 'shorter', 'make it shorter', 'more brief', 'briefly', 'as a table', 'in table', 'as list', 'in bullet points' ] if any(kw in q_lower for kw in format_patterns): return 'reformatting' # Explanation requests - reuse existing answer explanation_patterns = [ 'explain', 'why', 'how come', 'what does that mean', 'clarify', 'elaborate', 'what is this', 'tell me more' ] if any(kw in q_lower for kw in explanation_patterns): return 'explanation' # Summarization - reuse existing answer summary_patterns = [ 'summarize', 'summary', 'in short', 'briefly', 'in 2 lines', 'simplify', 'simpler', 'tldr' ] if any(kw in q_lower for kw in summary_patterns): return 'summarization' # Currency conversion - reuse computed values if any(kw in q_lower for kw in ['convert', 'in usd', 'in dollars', 'in euros', 'in inr', 'in rupees']): return 'conversion' # Comparison - needs context if any(kw in q_lower for kw in ['compare', 'versus', 'vs', 'difference between', 'better than', 'worse than']): return 'comparison' # Evaluation - use general knowledge if any(kw in q_lower for kw in ['good or bad', 'is this good', 'is this bad', 'healthy', 'benchmark', 'normal', 'typical']): return 'evaluation' # Hypothetical - what-if scenarios if any(kw in q_lower for kw in ['what if', 'suppose', 'assume', 'hypothetically', 'if we', 'if i']): return 'hypothetical' # Refinement - modify previous query if any(kw in q_lower for kw in ['now only', 'filter by', 'just for', 'only for', 'make it', 'change to', 'monthly', 'weekly']): return 'refinement' # Self-explanation - explain own reasoning (MUST DETECT "why do you think that?") self_explanation_patterns = [ 'why did you', 'how did you', 'explain your', 'reasoning', 'why do you think', 'why do you say', 'how do you know', 'why that', 'why this', 'how did you conclude', 'explain that' ] if any(kw in q_lower for kw in self_explanation_patterns): return 'self_explanation' # If contains 'above' reference but didn't match other patterns, treat as explanation if has_above_reference: return 'explanation' return 'new_query' def _resolve_query_references(question: str, session_id: str) -> tuple: """ Resolve references like 'that', 'this', 'above' using conversation memory. Returns: (resolved_question, metadata) """ if not MEMORY_ENGINE_AVAILABLE or not _reference_resolver: return question, {'resolved': False} try: resolved_query, metadata = _reference_resolver.resolve(question, session_id) if metadata.get('had_references'): print(f"[MEMORY] Resolved references: {metadata.get('resolved')}") return resolved_query, metadata except Exception as e: print(f"[MEMORY] Reference resolution failed: {e}") return question, {'resolved': False, 'error': str(e)} def _get_conversation_context(session_id: str, include_comparison: bool = False) -> str: """ Get formatted conversation history for context. $5M Enhancement: Also includes comparison context when available, so "compare top vs lowest" queries work correctly. Args: session_id: Session identifier include_comparison: If True, also include comparison context Returns: String with last few messages + comparison context for LLM """ if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return "" try: context = _short_term_memory.get_context(session_id) # $5M Enhancement: Add comparison context if available # This fixes "compare top vs lowest" queries failing! if include_comparison: try: comparison_ctx = _short_term_memory.get_comparison_context(session_id) if comparison_ctx: context += "\n" + comparison_ctx print(f"[MEMORY] Added comparison context for follow-up") except Exception as e: print(f"[MEMORY] Comparison context failed: {e}") return context except Exception as e: print(f"[MEMORY] Failed to get context: {e}") return "" def _store_query_result_for_followup(session_id: str, query_type: str, data: list, summary: str = ""): """ Store query results for follow-up comparisons. Call this after "top 5 customers" or "revenue breakdown by product" queries so the user can follow up with "compare top vs lowest". Args: session_id: Session identifier query_type: 'customers', 'products', 'months', etc. data: List of dicts with 'name' and 'value' keys summary: Optional description """ if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return try: _short_term_memory.set_last_query_result(session_id, query_type, data, summary) except Exception as e: print(f"[MEMORY] Failed to store query result: {e}") def _store_conversation_turn(session_id: str, role: str, content: str, metadata: dict = None): """Store a conversation turn in memory.""" if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return try: _short_term_memory.add_message(session_id, role, content, metadata) except Exception as e: print(f"[MEMORY] Failed to store message: {e}") def _is_language_only_query(question: str) -> bool: """ Check if query only requires language transformation (no data re-fetch). These queries can be answered using the previous answer without re-running RAG. """ q_lower = question.lower() language_patterns = [ 'explain', 'summarize', 'simplify', 'rephrase', 'in simple terms', 'what does that mean', 'clarify', 'elaborate', 'in 2 lines', 'briefly', 'is this good', 'is this bad', 'normal or not', 'healthy or not', 'why did you say', 'explain your reasoning' ] return any(pattern in q_lower for pattern in language_patterns) def _detect_response_format(question: str) -> dict: """ Detect user's desired response format and return formatting instructions. This is CRITICAL for ChatGPT-level behavior: respecting format constraints. Examples: - "explain in one line" -> format: "ONE_LINE" - "give me a brief summary" -> format: "BRIEF" - "explain in detail" -> format: "DETAILED" """ q_lower = question.lower() format_info = { "format_type": "NORMAL", "max_lines": None, "instructions": "" } # One-line responses if any(kw in q_lower for kw in ['one line', 'one sentence', 'single line', 'single sentence', '1 line', '1 sentence']): format_info["format_type"] = "ONE_LINE" format_info["max_lines"] = 1 format_info["instructions"] = """ RESPONSE FORMAT: ONE LINE ONLY - Your ENTIRE response must be ONE single sentence - NO tables, NO bullet points, NO multiple paragraphs - Maximum 30 words - Just answer directly in ONE line""" # Two-line responses elif any(kw in q_lower for kw in ['two lines', '2 lines', 'two sentences', '2 sentences']): format_info["format_type"] = "TWO_LINES" format_info["max_lines"] = 2 format_info["instructions"] = """ RESPONSE FORMAT: TWO LINES MAXIMUM - Your ENTIRE response must be 2 sentences or less - NO tables, NO bullet points - Keep it to 50 words maximum""" # Brief responses elif any(kw in q_lower for kw in ['briefly', 'brief', 'short', 'quick', 'concise']): format_info["format_type"] = "BRIEF" format_info["max_lines"] = 3 format_info["instructions"] = """ RESPONSE FORMAT: BRIEF - Keep response to 2-3 sentences - NO unnecessary tables or metrics - Be concise and direct""" # Detailed responses elif any(kw in q_lower for kw in ['detail', 'detailed', 'elaborate', 'comprehensive', 'full analysis']): format_info["format_type"] = "DETAILED" format_info["max_lines"] = None format_info["instructions"] = """ RESPONSE FORMAT: DETAILED - Provide comprehensive analysis - Include relevant tables and metrics - Explain insights thoroughly""" return format_info def _detect_hypothetical(question: str) -> bool: """Detect if query is hypothetical (what-if scenario).""" q_lower = question.lower() return any(kw in q_lower for kw in ['what if', 'suppose', 'assume', 'hypothetically', 'imagine if']) # ============================================================================ # AUTHORITATIVE METRIC TRACKING - Primary metric + Self-explanation # ============================================================================ def _set_primary_metric(session_id: str, metric: str, value: float, currency: str = "INR", formatted: str = None): """Store the authoritative primary metric for this session.""" if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return try: _short_term_memory.set_primary_metric(session_id, metric, value, currency, formatted) except Exception as e: print(f"[MEMORY] Failed to set primary metric: {e}") def _get_primary_metric_context(session_id: str) -> str: """ Get the primary metric context for follow-up queries. Returns formatted context string for LLM. """ if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return "" try: metric_data = _short_term_memory.get_primary_metric(session_id) if metric_data: return f""" AUTHORITATIVE METRIC (use this for 'it', 'that', 'this' references): - Metric: {metric_data.get('metric', 'unknown')} - Value: {metric_data.get('formatted', metric_data.get('value', 'N/A'))} - Currency: {metric_data.get('currency', 'INR')} """ return "" except Exception as e: print(f"[MEMORY] Failed to get primary metric: {e}") return "" def _set_conclusion(session_id: str, conclusion: str, reasons: list, metric: str = None): """Store an evaluative conclusion for self-explanation.""" if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return try: _short_term_memory.set_conclusion(session_id, conclusion, reasons, metric) except Exception as e: print(f"[MEMORY] Failed to set conclusion: {e}") def _get_self_explanation_context(session_id: str) -> str: """ Get the self-explanation context for 'why' questions. Returns formatted context string referencing stored conclusion. """ if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return "" try: conclusion_data = _short_term_memory.get_conclusion(session_id) if conclusion_data: reasons_str = "\n".join([f" - {r}" for r in conclusion_data.get('reasons', [])]) return f""" YOUR PREVIOUS CONCLUSION (explain this reasoning): - Conclusion: {conclusion_data.get('conclusion', 'N/A')} - Metric: {conclusion_data.get('metric', 'the data')} - Reasons: {reasons_str} SELF-EXPLANATION REQUIRED: - Explain WHY you reached that conclusion - Use the reasons listed above - Be confident - DO NOT ask for clarification - DO NOT re-run data queries """ return "" except Exception as e: print(f"[MEMORY] Failed to get conclusion: {e}") return "" def _get_last_answer_context(session_id: str) -> str: """Get the last answer for follow-up transformations.""" # Try short-term memory first if MEMORY_ENGINE_AVAILABLE and _short_term_memory: try: last_answer = _short_term_memory.get_last_answer(session_id) if last_answer: return f""" YOUR PREVIOUS ANSWER (use this for follow-ups): {last_answer[:1500]} """ except Exception as e: print(f"[MEMORY] Short-term memory error: {e}") # Fallback: Read from file-based history try: from pathlib import Path from utils.paths import get_user_paths import json paths = get_user_paths(session_id) memory_path = paths.get("memory", Path(f"storage/users/{session_id}/memory")) history_file = memory_path / "chat_history.json" if history_file.exists(): with open(history_file, 'r') as f: history = json.load(f) # Find last assistant message for msg in reversed(history): if msg.get('role') == 'assistant': last_answer = msg.get('content', '') if last_answer: return f""" YOUR PREVIOUS ANSWER (use this for follow-ups): {last_answer[:1500]} """ break except Exception as e: print(f"[MEMORY] File-based fallback error: {e}") return "" def _extract_primary_metric_from_answer(answer: str, session_id: str, currency_symbol: str = "₹"): """ Extract and store the primary metric from an answer. Called after generating an answer to enable follow-up resolution. """ import re # Look for revenue/total patterns patterns = [ (r'(?:total\s+)?revenue[:\s]+[₹$€£]?([\d,]+(?:\.\d+)?)', 'total_revenue'), (r'(?:total\s+)?sales[:\s]+[₹$€£]?([\d,]+(?:\.\d+)?)', 'total_sales'), (r'(?:total\s+)?orders[:\s]+([\d,]+)', 'total_orders'), (r'(?:average\s+)?order\s+value[:\s]+[₹$€£]?([\d,]+(?:\.\d+)?)', 'average_order_value'), (r'([\d,]+(?:\.\d+)?)\s*(?:customers|clients)', 'total_customers'), ] answer_lower = answer.lower() for pattern, metric_name in patterns: match = re.search(pattern, answer_lower) if match: value_str = match.group(1).replace(',', '') try: value = float(value_str) formatted = f"{currency_symbol}{value:,.2f}" if metric_name != 'total_orders' and metric_name != 'total_customers' else f"{int(value):,}" _set_primary_metric(session_id, metric_name, value, "INR", formatted) return except ValueError: pass # Fallback: Look for any bold metric bold_pattern = r'\*\*([^*]+)[:\s]+[₹$€£]?([\d,]+(?:\.\d+)?)\*\*' match = re.search(bold_pattern, answer) if match: metric_name = match.group(1).strip().lower().replace(' ', '_') value_str = match.group(2).replace(',', '') try: value = float(value_str) formatted = f"{currency_symbol}{value:,.2f}" _set_primary_metric(session_id, metric_name, value, "INR", formatted) except ValueError: pass # ============================================================================ # CURRENCY PREFERENCE HELPERS # ============================================================================ def _set_currency_preference(session_id: str, currency: str): """Store user's preferred output currency.""" if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return try: _short_term_memory.set_currency_preference(session_id, currency) except Exception as e: print(f"[MEMORY] Failed to set currency preference: {e}") def _get_currency_preference(session_id: str) -> str: """Get user's preferred output currency (default INR).""" if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return "INR" try: pref = _short_term_memory.get_currency_preference(session_id) return pref if pref else "INR" except Exception as e: print(f"[MEMORY] Failed to get currency preference: {e}") return "INR" def _detect_currency_from_query(question: str) -> str: """Detect which currency user is asking for and store preference.""" q_lower = question.lower() if any(kw in q_lower for kw in ['to usd', 'in usd', 'in dollars', 'to dollars']): return "USD" elif any(kw in q_lower for kw in ['to eur', 'in eur', 'in euros', 'to euros']): return "EUR" elif any(kw in q_lower for kw in ['to gbp', 'in gbp', 'in pounds', 'to pounds']): return "GBP" elif any(kw in q_lower for kw in ['to inr', 'in inr', 'in rupees', 'to rupees']): return "INR" return None # ============================================================================ # AUTHORITATIVE AGGREGATE HELPERS - Unified truth across modes # ============================================================================ def _store_aggregate(session_id: str, key: str, value): """Store an authoritative aggregate that cannot be contradicted.""" if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return try: _short_term_memory.set_aggregate(session_id, key, value) except Exception as e: print(f"[MEMORY] Failed to store aggregate: {e}") def _get_aggregates_context(session_id: str) -> str: """Get formatted aggregates for LLM prompt.""" if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return "" try: return _short_term_memory.get_aggregates_context(session_id) except Exception as e: print(f"[MEMORY] Failed to get aggregates: {e}") return "" def _store_chart_context(session_id: str, chart_type: str, title: str, data_summary: str, peak: dict = None, low: dict = None): """Store chart context for explanation binding.""" if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return try: _short_term_memory.set_last_chart(session_id, { "type": chart_type, "title": title, "data_summary": data_summary, "peak": peak, "low": low }) except Exception as e: print(f"[MEMORY] Failed to store chart context: {e}") def _get_chart_explanation_context(session_id: str) -> str: """Get chart context for 'explain this chart' queries.""" if not MEMORY_ENGINE_AVAILABLE or not _short_term_memory: return "" try: return _short_term_memory.get_chart_explanation_context(session_id) except Exception as e: print(f"[MEMORY] Failed to get chart context: {e}") return "" def _detect_chart_explanation_query(question: str) -> bool: """Detect if user is asking to explain a chart.""" q_lower = question.lower() patterns = [ 'explain this chart', 'explain the chart', 'what does this chart show', 'explain what this', 'describe the chart', 'interpret this chart', 'what does the chart', 'explain above chart' ] return any(p in q_lower for p in patterns) def _detect_trend_query(question: str) -> bool: """Detect if user is asking about trends.""" q_lower = question.lower() patterns = [ 'trend', 'over time', 'month by month', 'monthly', 'weekly', 'progression', 'growth pattern', 'performance over' ] return any(p in q_lower for p in patterns) def _detect_best_worst_query(question: str) -> bool: """Detect if user is asking about best/worst performance.""" q_lower = question.lower() patterns = [ 'best month', 'worst month', 'highest month', 'lowest month', 'peak month', 'performed best', 'performed worst', 'which month' ] return any(p in q_lower for p in patterns) def _generate_intelligent_chart( question: str, df, currency_symbol: str = "₹", user_role: str = "analyst" ) -> tuple: """ Generate a chart using intelligent visualization decision. This replaces hardcoded chart selection with query-aware selection. Returns: (chart_payload, decision_info) """ print(f"[VIZ] Starting intelligent chart generation for: '{question}'") print(f"[VIZ] VIZ_INTELLIGENCE_AVAILABLE: {VIZ_INTELLIGENCE_AVAILABLE}") if not VIZ_INTELLIGENCE_AVAILABLE: print("[VIZ] ERROR: Visualization intelligence not available!") return None, "Visualization intelligence not available" if df is None: print("[VIZ] ERROR: DataFrame is None!") return None, "No data available for visualization" if df.empty: print("[VIZ] ERROR: DataFrame is empty!") return None, "No data available for visualization" print(f"[VIZ] DataFrame shape: {df.shape}, columns: {list(df.columns)}") # Get visualization decision try: decision = decide_visualization(question, df, user_role) # Extract count from query and set decision.limit dynamically # e.g., "top 3 customers" → limit = 3 from agents.visualization_intelligence import extract_count_from_query extracted_limit = extract_count_from_query(question) decision.limit = extracted_limit print(f"[VIZ] Decision: should_render={decision.should_render}, chart_type={decision.chart_type}, intent={decision.intent}, limit={decision.limit}") print(f"[VIZ] Columns: x={decision.x_column}, y={decision.y_column}, group={decision.group_column}") print(f"[VIZ] Title: {decision.title}, Reason: {decision.reason}") except Exception as e: print(f"[VIZ] ERROR in decide_visualization: {e}") import traceback traceback.print_exc() return None, f"Decision error: {e}" if not decision.should_render: print(f"[VIZ] Not rendering: {decision.reason}") return None, decision.reason # Generate chart from decision try: chart_payload = create_chart_from_decision( decision=decision, df=df, currency_symbol=currency_symbol, user_role=user_role ) print(f"[VIZ] Chart payload generated: {chart_payload is not None}") if chart_payload: print(f"[VIZ] Chart has 'data': {'data' in chart_payload}, has 'layout': {'layout' in chart_payload}") except Exception as e: print(f"[VIZ] ERROR in create_chart_from_decision: {e}") import traceback traceback.print_exc() return None, f"Chart generation error: {e}" if chart_payload and not chart_payload.get("error"): info = f"Generated {decision.chart_type.value} chart: {decision.title}" print(f"[VIZ] SUCCESS: {info}") return chart_payload, info error_msg = chart_payload.get("error", "Chart generation failed") if chart_payload else "No chart payload" print(f"[VIZ] FAILED: {error_msg}") return None, error_msg def _is_followup_query(question: str) -> bool: """Check if this is a follow-up query about previous response/chart""" followup_keywords = [ # Explain patterns 'explain above', 'explain this', 'explain the', 'explain that', 'explain it', 'in words', 'in simple', 'what does it mean', # About patterns 'about above', 'about that', 'about this', 'about it', # Chart patterns 'above graph', 'above chart', 'this graph', 'this chart', 'the graph', 'the chart', 'that chart', 'that graph', 'the bar', 'this bar', 'above bar', # General follow-up 'tell me about', 'explain more', 'more about', 'can you explain', 'previous', 'what does', 'elaborate', 'clarify', # Short references 'that', 'above', 'it' ] q_lower = question.lower().strip() # Check for keyword matches if any(kw in q_lower for kw in followup_keywords): return True # Also detect very short questions that are likely follow-ups if len(q_lower.split()) <= 5 and any(w in q_lower for w in ['that', 'it', 'this', 'above']): return True return False def _extract_chart_type_from_context(conversation_context: str) -> str: """Extract what type of chart was shown in previous response""" context_lower = conversation_context.lower() # Check for customer chart indicators if any(kw in context_lower for kw in ['top customers', 'customer revenue', 'customer_3', 'customer_2', 'customer_1']): return 'customer' # Check for product chart indicators if any(kw in context_lower for kw in ['product revenue', 'product distribution', 'top products', 'product_']): return 'product' # Check for trend chart indicators if any(kw in context_lower for kw in ['monthly revenue', 'trend', '2025-', 'revenue over time']): return 'trend' return 'customer' # Default to customer if unclear def _enhance_followup_question(question: str, conversation_context: str) -> str: """Enhance a follow-up question with context from previous conversation""" if not _is_followup_query(question): return question chart_type = _extract_chart_type_from_context(conversation_context) # Build enhanced question with context if chart_type == 'customer': return f"{question}\n\n[Context: The user is asking about the TOP CUSTOMERS BY REVENUE bar chart that was shown. Explain the customer revenue data with exact customer names and amounts.]" elif chart_type == 'product': return f"{question}\n\n[Context: The user is asking about the PRODUCT DISTRIBUTION chart that was shown. Explain the product revenue data with exact product names and percentages.]" elif chart_type == 'trend': return f"{question}\n\n[Context: The user is asking about the MONTHLY REVENUE TREND chart that was shown. Explain the monthly revenue data with exact months and amounts.]" return question def _extract_prediction_period(question: str) -> dict: """ Extract prediction period from question. Returns dict with: periods: Number of periods unit: 'days', 'weeks', 'months', 'years' display: Human-readable string """ import re q_lower = question.lower() # Word to number mapping word_nums = { 'one': 1, 'two': 2, 'three': 3, 'four': 4, 'five': 5, 'six': 6, 'seven': 7, 'eight': 8, 'nine': 9, 'ten': 10, 'a': 1, 'an': 1 } def extract_number(text): """Extract number from text (digit or word)""" for word, num in word_nums.items(): if word in text: return num match = re.search(r'(\d+)', text) return int(match.group(1)) if match else None # Check for days day_patterns = [r'next\s+(\w+)\s*days?', r'(\d+)\s*days?\s*ahead', r'tomorrow'] for pattern in day_patterns: match = re.search(pattern, q_lower) if match or 'tomorrow' in q_lower: num = extract_number(match.group(0)) if match else 1 num = num or 1 return {'periods': num, 'unit': 'days', 'display': f'{num} day(s)'} # Check for weeks week_patterns = [r'next\s+(\w+)\s*weeks?', r'(\d+)\s*weeks?\s*ahead'] for pattern in week_patterns: match = re.search(pattern, q_lower) if match: num = extract_number(match.group(0)) or 1 return {'periods': num, 'unit': 'weeks', 'display': f'{num} week(s)'} # Check for months month_patterns = [r'next\s+(\w+)\s*months?', r'(\d+)\s*months?\s*ahead'] for pattern in month_patterns: match = re.search(pattern, q_lower) if match: num = extract_number(match.group(0)) or 3 return {'periods': num, 'unit': 'months', 'display': f'{num} month(s)'} # Check for quarters if 'quarter' in q_lower: match = re.search(r'next\s+(\w+)\s*quarters?', q_lower) num = extract_number(match.group(0)) if match else 1 return {'periods': num * 3, 'unit': 'months', 'display': f'{num} quarter(s)'} # Check for years year_patterns = [r'next\s+(\w+)\s*years?', r'(\d+)\s*years?\s*ahead'] for pattern in year_patterns: match = re.search(pattern, q_lower) if match: num = extract_number(match.group(0)) or 1 return {'periods': num * 12, 'unit': 'months', 'display': f'{num} year(s)'} if 'next year' in q_lower: return {'periods': 12, 'unit': 'months', 'display': '1 year'} if 'next quarter' in q_lower: return {'periods': 3, 'unit': 'months', 'display': '1 quarter'} if 'next month' in q_lower: return {'periods': 1, 'unit': 'months', 'display': '1 month'} if 'next week' in q_lower: return {'periods': 1, 'unit': 'weeks', 'display': '1 week'} # Default 3 months return {'periods': 3, 'unit': 'months', 'display': '3 months'} def _format_prediction_response( prediction_result: dict, currency_symbol: str = "₹", data_source: str = "uploaded data" ) -> str: """ Format prediction result into structured AI response. Returns markdown with: - Executive Summary - Forecast Insights - Key Forecast Numbers - Risk Zones - Opportunities - Driver Analysis - Chart Payload (JSON) - Data Source """ response_parts = [] # 1. Executive Summary response_parts.append("### 📌 Executive Summary") insight = prediction_result.get('insight', 'Forecast analysis complete.') response_parts.append(f"{insight}\n") # 2. Forecast Insights response_parts.append("### 🔮 Forecast Insights") trend = prediction_result.get('trend', 'stable') trend_labels = { 'strongly_increasing': '📈 Strong upward momentum detected', 'increasing': '📈 Steady growth trend observed', 'stable': '➡️ Values remain relatively stable', 'decreasing': '📉 Declining trend identified', 'strongly_decreasing': '📉 Significant downward pressure' } response_parts.append(f"- **Trend**: {trend_labels.get(trend, trend)}") seasonality = prediction_result.get('seasonality') if seasonality: response_parts.append(f"- **Seasonality**: {seasonality.capitalize()} pattern detected") model = prediction_result.get('model_used', 'auto') model_names = { 'prophet': 'Facebook Prophet (trend + seasonality)', 'arima': 'ARIMA (auto-regressive)', 'holt_winters': 'Holt-Winters (exponential smoothing)', 'linear': 'Linear regression with smoothing' } response_parts.append(f"- **Model**: {model_names.get(model, model)}\n") # 3. Key Forecast Numbers response_parts.append("### 📈 Key Forecast Numbers") forecast_points = prediction_result.get('forecast_points', []) if forecast_points: # Get specific period forecasts for i, point in enumerate(forecast_points[:6]): value = point.get('value', 0) lower = point.get('lower', value * 0.9) upper = point.get('upper', value * 1.1) date = point.get('date', f'Period +{i+1}') response_parts.append( f"| **{date}** | {currency_symbol}{value:,.0f} | " f"Range: {currency_symbol}{lower:,.0f} - {currency_symbol}{upper:,.0f} |" ) # Accuracy scores accuracy = prediction_result.get('accuracy', {}) if accuracy: mape = accuracy.get('MAPE', 0) rmse = accuracy.get('RMSE', 0) confidence = 100 - mape if mape else 0 response_parts.append(f"\n**Accuracy Score**: {confidence:.0f}% (MAPE: {mape:.1f}%, RMSE: {currency_symbol}{rmse:,.0f})\n") # 4. Risk Zones risks = prediction_result.get('risks', []) if risks: response_parts.append("### ⚠️ Risk Zones") for risk in risks[:5]: response_parts.append(f"- ⚡ {risk}") response_parts.append("") # 5. Opportunities opportunities = prediction_result.get('opportunities', []) if opportunities: response_parts.append("### 🟢 Opportunities") for opp in opportunities[:4]: response_parts.append(f"- {opp}") response_parts.append("") # 6. Driver Analysis explanation = prediction_result.get('explanation', '') if explanation: response_parts.append("### 🧠 Why This Happened") response_parts.append(f"{explanation}\n") # 7. Chart Payload (for frontend) chart_payload = prediction_result.get('chart_payload', {}) if chart_payload: response_parts.append("### 📊 Chart") response_parts.append("```forecast_chart") import json response_parts.append(json.dumps(chart_payload, indent=2)) response_parts.append("```\n") # 8. Data Source response_parts.append("### 📦 Data Source") response_parts.append(f"Analysis based on: **{data_source}**") return "\n".join(response_parts) def _embed_chart_in_response(answer: str, chart_base64: str, chart_title: str) -> str: """Embed a chart image in the response""" # Add chart as embedded image (frontend will render) chart_section = f"\n\n📊 **{chart_title}**\n" chart_section += f"![{chart_title}]({chart_base64})\n" return answer + chart_section def _add_confidence_indicator( answer: str, confidence: AnswerConfidence ) -> str: """Add confidence indicator to answer if needed""" if confidence.should_flag_uncertainty: warning = "\n\n> ⚠️ **Note:** This response has lower confidence. " warning += "Consider uploading more relevant data for better accuracy." return answer + warning return answer def _build_chart_data_summary(df, question: str, currency_symbol: str) -> str: """ Pre-compute chart data summary to include in LLM prompt. This ensures the LLM explanation matches the actual chart data. """ q_lower = question.lower() summary = "" amount_col = 'amount' if 'amount' in df.columns else 'total_amount' # Customer chart data - EXACT data that will be shown in chart if 'customer' in q_lower and 'customer' in df.columns: customer_revenue = df.groupby('customer')[amount_col].sum().sort_values(ascending=False).head(10) customer_orders = df.groupby('customer').size() summary = "\n\n📊 CHART DATA (Use this EXACT data in your explanation):\n" summary += "| Rank | Customer | Revenue | Orders |\n" summary += "|------|----------|---------|--------|\n" for rank, (cust, rev) in enumerate(customer_revenue.items(), 1): orders = customer_orders.get(cust, 0) summary += f"| {rank} | {cust} | {currency_symbol}{rev:,.2f} | {orders} |\n" total_rev = df[amount_col].sum() avg_order = df[amount_col].mean() summary += f"\n**Total Revenue:** {currency_symbol}{total_rev:,.2f}\n" summary += f"**Average Order:** {currency_symbol}{avg_order:,.2f}\n" summary += f"**Top Customer:** {customer_revenue.index[0]} ({currency_symbol}{customer_revenue.iloc[0]:,.2f})\n" summary += f"**Lowest in Top 10:** {customer_revenue.index[-1]} ({currency_symbol}{customer_revenue.iloc[-1]:,.2f})\n" # Product chart data elif 'product' in q_lower and 'product' in df.columns: product_revenue = df.groupby('product')[amount_col].sum().sort_values(ascending=False).head(8) product_counts = df.groupby('product').size() summary = "\n\n📦 CHART DATA (Use this EXACT data in your explanation):\n" summary += "| Rank | Product | Revenue | Units Sold | % Share |\n" summary += "|------|---------|---------|------------|--------|\n" total = product_revenue.sum() for rank, (prod, rev) in enumerate(product_revenue.items(), 1): units = product_counts.get(prod, 0) share = (rev / total) * 100 if total > 0 else 0 summary += f"| {rank} | {prod} | {currency_symbol}{rev:,.2f} | {units} | {share:.1f}% |\n" summary += f"\n**Total Product Revenue:** {currency_symbol}{total:,.2f}\n" summary += f"**Top Product:** {product_revenue.index[0]}\n" summary += f"**Number of Products:** {len(product_revenue)}\n" # Trend/prediction data elif any(kw in q_lower for kw in ['trend', 'month', 'predict', 'forecast']): if 'date' in df.columns: import pandas as pd df_temp = df.copy() df_temp['date_parsed'] = pd.to_datetime(df_temp['date'], errors='coerce') df_dated = df_temp[df_temp['date_parsed'].notna()].copy() if not df_dated.empty: df_dated['month'] = df_dated['date_parsed'].dt.to_period('M') monthly = df_dated.groupby('month')[amount_col].sum() summary = "\n\n📈 CHART DATA (Use this EXACT data in your explanation):\n" summary += "| Month | Revenue |\n" summary += "|-------|--------|\n" for month, rev in monthly.items(): summary += f"| {month} | {currency_symbol}{rev:,.2f} |\n" if len(monthly) >= 2: growth = ((monthly.iloc[-1] - monthly.iloc[0]) / monthly.iloc[0]) * 100 if monthly.iloc[0] > 0 else 0 summary += f"\n**Period Growth:** {growth:+.1f}%\n" summary += f"**Highest Month:** {monthly.idxmax()} ({currency_symbol}{monthly.max():,.2f})\n" summary += f"**Lowest Month:** {monthly.idxmin()} ({currency_symbol}{monthly.min():,.2f})\n" return summary def rag_answer(state: AgentState) -> AgentState: """ 🟦 RAG MODE - Enterprise Document-Based Answers Features: - Adaptive retrieval with confidence scoring - Citation formatting with source locations - Token-aware context building - Clean markdown output - CONVERSATION MEMORY - ChatGPT-level stateful behavior """ start_time = time.time() question = state.question user_id = state.company_id session_id = user_id # Use user_id as session_id for now # ========================================================================= # STEP 1A: CONVERSATION MEMORY - Store user message in session memory # ========================================================================= _store_conversation_turn(session_id, "user", question) # ========================================================================= # STEP 1B: FOLLOW-UP INTELLIGENCE - Detect query type and resolve references # ========================================================================= followup_type = _detect_followup_type(question) original_question = question # Resolve references like "that", "this", "above", "it" resolved_question, resolution_metadata = _resolve_query_references(question, session_id) if resolution_metadata.get('had_references'): print(f"[CONV] Follow-up detected: {followup_type}") print(f"[CONV] Resolved: '{question}' -> '{resolved_question}'") question = resolved_question # ========================================================================= # TRUTH-GROUNDED: Better resolution for "explain above" type queries # ========================================================================= if TRUTH_GROUNDED_AVAILABLE: try: enhanced_q, truth_resolution = resolve_above_reference(session_id, question) if truth_resolution.get("had_reference"): print(f"[TRUTH] Resolved '{question}' -> '{enhanced_q}'") print(f"[TRUTH] Reference type: {truth_resolution.get('reference_type')}") question = enhanced_q except Exception as e: print(f"[TRUTH] Resolution error: {e}") # ========================================================================= # PRODUCTION-GRADE: Execute 7-step Query Pipeline # ========================================================================= pipeline_result = None ai_memory = None if BUSINESS_AI_ENGINE_AVAILABLE: try: # Get 3-layer memory ai_memory = get_ai_memory(session_id) # Add user turn to memory ai_memory.add_turn(role="user", content=original_question) # Execute 7-step pipeline pipeline_result = execute_query_pipeline(question, session_id, ai_memory) # ================================================================= # DISABLED: Early refusal was blocking queries before RAG checks # The existing RAG system already handles data availability properly # ================================================================= # should_refuse_response, refuse_reason = should_refuse(pipeline_result) # if should_refuse_response: # ... (disabled - RAG handles this) print(f"[PIPELINE] Proceeding with intent={pipeline_result.intent.value}") except Exception as e: print(f"[PIPELINE] Error: {e}") # Get conversation context for LLM conversation_context = _get_conversation_context(session_id) # ========================================================================= # STEP 1C: MEMORY - Save user message and extract personal info (legacy) # ========================================================================= _save_chat_message(user_id, "user", original_question) # Extract and save personal info (name, company, role) personal_context = _extract_and_save_personal_info(question, user_id) if personal_context: # User introduced themselves - give a warm response import re name_match = re.search(r"Name:\s*([^\n]+)", personal_context) if name_match: user_name = name_match.group(1).strip() warm_response = f"""Nice to meet you, **{user_name}**! 👋 I'm your AI Business Analyst. I can help you analyze the data you've uploaded. **Ask me about:** • 📊 Revenue and sales trends • 👥 Top customers and products • 📈 Forecasts and predictions What would you like to explore today?""" state.answer = warm_response state.route = "rag" state.sources = [] _save_chat_message(user_id, "assistant", warm_response) _store_conversation_turn(session_id, "assistant", warm_response) return state # ========================================================================= # STEP 2: SMART GUARDRAILS - Allow language-only queries # ========================================================================= # Language-only queries (explain, summarize, evaluate) don't need data check is_language_only = _is_language_only_query(question) is_hypothetical = _detect_hypothetical(question) if not _is_data_related_query(question) and not is_language_only and followup_type == 'new_query': # ================================================================= # HYBRID KNOWLEDGE: General questions get 🌐 AI Knowledge response # ================================================================= try: from core.llm import chat as llm_chat ai_prompt = f"""You are a helpful AI assistant. Answer this general question clearly and concisely. Question: {question} Provide a helpful, accurate response. Format nicely with bullet points if appropriate.""" ai_response = llm_chat(ai_prompt, temperature=0.7, max_tokens=500) # Add source badge to show this is AI knowledge, not user data hybrid_response = f"""🌐 **AI Knowledge** {ai_response} --- *💡 This is general AI knowledge. For analysis of YOUR data, ask questions like "What is my total revenue?" or "Show my top customers."*""" state.answer = hybrid_response state.route = "ai_knowledge" state.sources = ["AI Knowledge"] _save_chat_message(user_id, "assistant", hybrid_response) _store_conversation_turn(session_id, "assistant", hybrid_response) return state except Exception as e: # Fallback if LLM fails import logging logger = logging.getLogger(__name__) logger.error(f"AI Knowledge response failed: {e}") state.answer = f"I can help with that! {question} is an interesting topic. For detailed answers, please try asking about your uploaded data." state.route = "rag" state.sources = [] return state # ========================================================================= # STEP 3: MEMORY - Get chat history for context # ========================================================================= chat_history = _get_chat_history(user_id, limit=5) # Classify query for better retrieval query_analysis = classify_query(question) # Retrieve with hybrid search - INCREASED k for multi-file coverage docs = retrieve(question, k=12, user_id=user_id) if not docs or len(docs) == 0: state.answer = """I don't have any data to analyze yet. **To get started:** • Go to **Data Hub** and upload your business files • Supported formats: CSV, Excel, PDF, Word, Images • Include data with columns like: Customer, Product, Amount, Date Once uploaded, I can help you analyze sales trends, customer insights, and more.""" state.route = "rag" state.sources = [] return state # Build context with metadata - INCREASED to include ALL files context_parts = [] sources = [] similarity_scores = [] sources_with_data = {} # Track data by source file for i, doc in enumerate(docs[:15]): # Process up to 15 for multi-file coverage doc_text = doc.get('text', str(doc))[:1500] source = doc.get('source', doc.get('metadata', {}).get('source', '')) score = doc.get('score', 0.5) if source: if source not in sources: sources.append(source) if source not in sources_with_data: sources_with_data[source] = [] sources_with_data[source].append(doc_text[:500]) similarity_scores.append(score) context_parts.append(f"[Source {i+1}: {source}]\n{doc_text}") # Build organized context showing data from EACH file organized_context = f"## 📁 Data from {len(sources)} Files:\n\n" for src in sources: organized_context += f"### From: {src}\n" for data_snippet in sources_with_data.get(src, [])[:3]: organized_context += f"{data_snippet}\n\n" context = organized_context + "\n---\n\n## All Retrieved Data:\n" + "\n\n---\n\n".join(context_parts[:10]) # Calculate confidence scorer = get_confidence_scorer() rag_conf = scorer.score_rag_confidence( similarity_scores=similarity_scores, sources=sources, query_terms=len(question.split()), matched_terms=len(question.split()) // 2 ) confidence = scorer.build_answer_confidence("rag", rag_conf=rag_conf, sources=[{"source": s} for s in sources]) # Get currency for formatting currency_symbol, currency_code = get_user_currency(user_id) # ========================================================================= # COMPANY INTELLIGENCE - Build company context for aware responses # ========================================================================= company_context_section = "" company_name = "Your Company" if COMPANY_PROFILE_AVAILABLE: try: company_profile = get_company_profile(user_id) if company_profile: company_name = company_profile.company_name company_context_section = build_company_context(user_id) except Exception as e: print(f"⚠️ Company profile loading: {e}") # ========================================================================= # BUILD CONVERSATION-AWARE SYSTEM PROMPT # ========================================================================= # Add conversation context section if available conversation_section = "" if conversation_context: conversation_section = f""" RECENT CONVERSATION (use this for follow-ups): {conversation_context} ---""" # Add aggregates context (AUTHORITATIVE - cannot be contradicted) aggregates_section = _get_aggregates_context(session_id) # Add chart context for "explain this chart" queries chart_section = "" if _detect_chart_explanation_query(question): chart_section = _get_chart_explanation_context(session_id) # Detect and add response format constraints (ONE LINE, BRIEF, etc.) format_info = _detect_response_format(question) format_section = format_info.get("instructions", "") # Add follow-up handling instructions based on query type followup_instructions = "" if followup_type == 'explanation': followup_instructions = """ FOLLOW-UP MODE: EXPLANATION - Explain the previous answer in simpler terms - DO NOT re-query data - use existing values - Be helpful and educational""" elif followup_type == 'summarization': followup_instructions = """ FOLLOW-UP MODE: SUMMARIZATION - Summarize the previous answer briefly - DO NOT re-query data - use existing values - Keep it to 2-3 sentences""" # ========================================================================== # 🧠 CHATGPT-STYLE: Load actual DataFrame context for dynamic understanding # ========================================================================== data_ctx = build_data_context(user_id) data_context_section = data_ctx.get("context_string", "No data available") # Get the last answer for context (critical for follow-ups) last_answer = _get_last_answer_context(session_id) system_prompt = f"""You are DataVision, an AI Data Analyst. ⚠️ CRITICAL - DATA GROUNDING RULES (NEVER VIOLATE): 1. ONLY use information from the DATA PROVIDED BELOW 2. NEVER use outside knowledge, general facts, or industry information 3. NEVER make up or estimate numbers - every number must come from the data 4. If the data doesn't have the answer, say "I don't have data for that" 5. NEVER say "typically", "usually", "in general", or "generally" {company_context_section} ## 📊 YOUR DATA (ONLY USE THIS): {data_context_section} ## DATA SOURCES: {', '.join(sources) if sources else 'No data loaded yet'} ## CURRENCY: {currency_symbol} ({currency_code}) ## CONVERSATION CONTEXT: {conversation_context if conversation_context else 'This is the first message.'} {last_answer} ## RESPONSE RULES: 1. Give DIRECT answers using ONLY the data above 2. COMPUTE actual values from the data - totals, percentages, counts 3. CITE which column/data you used (e.g., "From the Sales column...") 4. Use {currency_symbol} for currency values 5. Format nicely with markdown tables, bullet points, bold 6. If data is missing, say: "Your data doesn't include [X], so I can't answer that" ## 🚫 FORBIDDEN (will cause system failure): - Using general knowledge - Making up numbers - Saying "typically" or "usually" - Answering about topics not in the data ## YOUR RESPONSE (ONLY from data above):""" # ========================================================================= # ROLE INTELLIGENCE - Add role-specific response formatting rules # ========================================================================= if ROLE_TEMPLATES_AVAILABLE: try: user_role = get_role_from_string(state.user_role) role_modifier = get_role_prompt_modifier(user_role) system_prompt += role_modifier except Exception as e: print(f"⚠️ Role template loading: {e}") # Get personalized user context user_context = _get_user_context(user_id) user_context_section = f"\n\nUser Context:\n{user_context}" if user_context else "" # Pre-compute chart data if visualization requested OR follow-up about chart chart_data_section = "" is_followup = _is_followup_query(question) needs_chart_data = _wants_visualization(question) or _wants_prediction(question) or is_followup if needs_chart_data: try: # Use get_user_data to preserve original columns (Department, Salary, etc.) from api.v1.endpoints.charts import get_user_data df_for_chart = get_user_data(user_id) if df_for_chart is not None and not df_for_chart.empty: # For follow-ups, default to customer chart data since that's most common if is_followup: # Always provide customer data for follow-up questions about charts chart_data_section = _build_chart_data_summary(df_for_chart, "customer chart", currency_symbol) else: chart_data_section = _build_chart_data_summary(df_for_chart, question, currency_symbol) except Exception as e: print(f"Chart data pre-computation: {e}") # Enhance question for follow-up queries enhanced_question = question if is_followup: # ================================================================= # CRITICAL FIX: Get ACTUAL previous response, not hardcoded chart # ================================================================= previous_response = "" try: # Get actual last assistant response from chat history from utils.paths import get_user_paths import json paths = get_user_paths(user_id) from pathlib import Path memory_path = paths.get("memory", Path(f"storage/users/{user_id}/memory")) history_file = memory_path / "chat_history.json" if history_file.exists(): with open(history_file, 'r') as f: history = json.load(f) # Find last assistant message for msg in reversed(history): if msg.get('role') == 'assistant': previous_response = msg.get('content', '')[:1000] # First 1000 chars break print(f"[CONTEXT] Found previous response: {len(previous_response)} chars") except Exception as e: print(f"[CONTEXT] Error getting previous response: {e}") if previous_response: enhanced_question = f"""{question} [PREVIOUS AI RESPONSE TO EXPLAIN:] {previous_response} [INSTRUCTION: The user is asking you to EXPLAIN the above response. Answer their specific question about what was shown above. Do NOT generate new charts or analysis - just explain what was already shown.]""" else: enhanced_question = f"""{question} [CONTEXT: The user is referring to a previous response. Please explain based on the conversation context.]""" # Build chat history section chat_history_section = "" if chat_history: chat_history_section = f"\n\n**Recent Conversation:**\n{chat_history}" prompt = f"""Question: {enhanced_question} Query Type: {query_analysis.query_type.value} Aggregation: {query_analysis.aggregation_type or 'N/A'} {user_context_section} {chat_history_section} Retrieved Context: {context} {chart_data_section} IMPORTANT RULES: 1. If CHART DATA is provided above, reference those EXACT numbers and names in your explanation. 2. A chart will be generated AUTOMATICALLY - DO NOT output any JSON, Plotly code, or chart specifications. 3. Just provide a clear text explanation/summary. The visualization system handles charts separately. 4. NEVER include raw JSON objects like {{"type": "...", "data": [...]}} in your response. If the user previously introduced themselves (check Recent Conversation), remember their name and use it warmly. Provide a clear, data-driven answer.""" # =========================================================================== # $5M FIX: BUILD MESSAGES WITH FULL CONVERSATION CONTEXT # This is the key change - LLM now sees entire conversation history # =========================================================================== messages = build_messages_with_context( user_id=user_id, current_question=question, current_prompt=prompt, history_limit=10 ) # =========================================================================== # SMART MODEL ROUTING - Use fast model for simple, reasoning for complex # =========================================================================== optimal_model = get_optimal_model( query_type=query_analysis.query_type.value, context_length=len(context) ) # Use messages array instead of single prompt for context memory answer = chat(messages=messages, system=system_prompt, max_tokens=1500, model=optimal_model) # Clean response with anti-hallucination validation answer = _clean_response_formatting(answer, context=context) # ========================================================================== # VISUALIZATION - Managed universally in chat.py via append_chart_if_needed # ========================================================================== # We no longer generate charts here to prevent duplicates and ensure # orchestration logic in the chat handler. # PLOTLY GENERATION FOR RAG (Migrated to chat.py) # All chart generation is now handled by the universal helper in chat.py # Add citations and confidence answer = _add_confidence_indicator(answer, confidence) answer += _build_source_citations(sources, "rag") # Clean response formatting for premium quality answer = _clean_response_formatting(answer) state.answer = answer state.route = "rag" state.sources = sources state.context["confidence"] = confidence.overall_score state.context["query_type"] = query_analysis.query_type.value # Save assistant response to chat history for memory # CRITICAL: Save enough to include key data for follow-ups _save_chat_message(user_id, "assistant", answer[:2000]) # ========================================================================= # CONVERSATION MEMORY - Store response for follow-up intelligence # ========================================================================= _store_conversation_turn(session_id, "assistant", answer[:1000], { "query_type": query_analysis.query_type.value, "sources": sources[:3], "followup_type": followup_type }) # ========================================================================= # AUTHORITATIVE METRIC EXTRACTION - Store primary metric for follow-ups # ========================================================================= _extract_primary_metric_from_answer(answer, session_id, currency_symbol) # ========================================================================= # $5M ENHANCEMENT: ChatGPT-Quality Response Enhancement # ========================================================================= # Adds proactive insights, follow-up suggestions, and natural tone # ALL using LLM - no hardcoding! if RESPONSE_ENHANCER_AVAILABLE: try: # Extract data summary from answer for insight generation data_summary = extract_data_summary_from_response(answer, currency_symbol) # Add entities from query for relevant suggestions entities = [] q_lower = question.lower() if 'customer' in q_lower: entities.append('customers') if 'product' in q_lower: entities.append('products') if 'revenue' in q_lower or 'sales' in q_lower: entities.append('revenue') # Get query type for context-aware suggestions query_type_str = query_analysis.query_type.value if query_analysis else "general" # Apply full enhancement pipeline enhanced_answer = enhance_full_response( query=question, response=answer, query_type=query_type_str, data_summary=data_summary, entities=entities, add_insight=True, add_suggestions=True, enhance_tone=True, currency_symbol=currency_symbol ) if enhanced_answer and len(enhanced_answer) > len(answer): answer = enhanced_answer state.answer = answer print("[ENHANCE] Response enhanced with insights and suggestions") except Exception as enhance_error: print(f"[ENHANCE] Enhancement failed (non-fatal): {enhance_error}") # ========================================================================= # $5M FIX: SAVE ASSISTANT RESPONSE FOR CONTEXT MEMORY # Without this, the LLM won't know what it said in previous turns! # ========================================================================= _store_conversation_turn(session_id, "assistant", state.answer or answer) _save_chat_message(user_id, "assistant", state.answer or answer) print(f"[CONTEXT] Saved assistant response ({len(state.answer or answer)} chars)") # ========================================================================= # TRUTH-GROUNDED: Update context with WHAT was shown for follow-up queries # ========================================================================= if TRUTH_GROUNDED_AVAILABLE: try: # Detect what topic was discussed topic = None result_type = "text" chart_type = None q_lower = question.lower() if any(t in q_lower for t in ["revenue", "sales", "amount", "total"]): topic = "revenue" elif any(t in q_lower for t in ["customer", "client"]): topic = "customers" elif any(t in q_lower for t in ["product", "item"]): topic = "products" elif any(t in q_lower for t in ["convert", "currency", "inr", "usd"]): topic = "currency_conversion" # Detect if chart was shown if "```plotly_chart" in (state.answer or answer): result_type = "chart" if "bar" in (state.answer or answer).lower() or "bar" in q_lower: chart_type = "bar" elif "line" in (state.answer or answer).lower() or "trend" in q_lower: chart_type = "line" elif "pie" in (state.answer or answer).lower(): chart_type = "pie" elif "|" in (state.answer or answer) and "---" in (state.answer or answer): result_type = "table" update_truth_context( session_id=session_id, topic=topic, result_type=result_type, chart_type=chart_type, answer=(state.answer or answer)[:500] # Store first 500 chars ) print(f"[TRUTH] Updated context: topic={topic}, type={result_type}, chart={chart_type}") except Exception as e: print(f"[TRUTH] Context update error: {e}") # ========================================================================= # PRODUCTION-GRADE: Update ThreeLayerMemory with response metadata # ========================================================================= if BUSINESS_AI_ENGINE_AVAILABLE and ai_memory: try: # Detect topic and metrics topic = None result_type = "text" chart_type = None metrics = {} q_lower = question.lower() if any(t in q_lower for t in ["revenue", "sales", "amount", "total"]): topic = "revenue" elif any(t in q_lower for t in ["customer", "client"]): topic = "customers" elif any(t in q_lower for t in ["product", "item"]): topic = "products" # Detect chart if "```plotly_chart" in (state.answer or answer): result_type = "chart" if "bar" in q_lower: chart_type = "bar" elif "line" in q_lower or "trend" in q_lower: chart_type = "line" elif "|" in (state.answer or answer) and "---" in (state.answer or answer): result_type = "table" # Extract any numbers as potential metrics import re numbers = re.findall(r'[\$₹€£]?([\d,]+\.?\d*)', state.answer or answer) for i, num_str in enumerate(numbers[:3]): try: num = float(num_str.replace(',', '')) metrics[f"value_{i}"] = num except: pass # Add to ThreeLayerMemory ai_memory.add_turn( role="assistant", content=(state.answer or answer)[:500], topic=topic, result_type=result_type, chart_type=chart_type, metrics=metrics ) print(f"[AI-ENGINE] Memory updated: topic={topic}, metrics={len(metrics)}") # SELF-AUDIT before returning (ChatGPT/Claude behavior) if pipeline_result: audit = self_audit(state.answer or answer, ai_memory, pipeline_result) if not audit.passed: print(f"[AUDIT] Warning: {audit.issues}") else: print(f"[AUDIT] Passed: {audit.claims_verified} claims verified") except Exception as e: print(f"[AI-ENGINE] Memory update error: {e}") return state # ============================================================================ # 🟧 GRAPHRAG MODE - Enhanced Knowledge Graph Reasoning # ============================================================================ def graph_answer(state: AgentState) -> AgentState: """ 🟧 GRAPHRAG MODE - Enterprise Graph-Based Analysis Features: - Multi-hop reasoning with path explanations - PageRank-based importance ranking - Complete data tables - Causal chain detection """ start_time = time.time() question = state.question user_id = state.company_id session_id = user_id # Use user_id as session_id for conversation memory # Initialize enterprise context variables to prevent NameError company_name = "Your Company" company_context_section = "" user_context_section = "" graph_insights = "" currency_symbol = "₹" try: from graph.query import get_graph_stats, load_graph, get_graph_analysis from graph.traversal import EnterpriseGraphTraversal, TraversalStrategy graph = load_graph(user_id) if not graph or graph.number_of_nodes() == 0: state.answer = """I don't have a knowledge graph built yet. **To enable GraphRAG:** • Upload business files in **Data Hub** • Include structured data (CSV/Excel) with Customer, Product, Amount columns • The graph will be built automatically **GraphRAG excels at:** • "Which customers buy which products?" • "What are the revenue trends?" • "Show customer-product relationships" • "Why did sales drop?" (causal reasoning)""" state.route = "graph" return state # --- ENTERPRISE CONTEXT LOADING --- if COMPANY_PROFILE_AVAILABLE: try: company_profile = get_company_profile(user_id) if company_profile: company_name = company_profile.company_name company_context_section = build_company_context(user_id) except Exception as e: print(f"⚠️ Company profile loading error: {e}") user_context = _get_user_context(user_id) if user_context: user_context_section = f"\n\nUser Context:\n{user_context}" # Get specialized graph insights for the prompt try: graph_insights = get_graph_analysis(user_id, question) or "No specific relationship patterns detected for this query." except Exception as e: print(f"⚠️ Graph analysis error: {e}") graph_insights = "Error analyzing graph relationships." # Classify query for traversal strategy query_analysis = classify_query(question) # Select traversal strategy based on query type if query_analysis.query_type == QueryType.CAUSAL: strategy = TraversalStrategy.DFS elif query_analysis.query_type == QueryType.RELATIONAL: strategy = TraversalStrategy.BFS else: strategy = TraversalStrategy.PAGERANK # Initialize traversal traversal = EnterpriseGraphTraversal(graph) # Extract entities from question entities = query_analysis.entities_mentioned # Perform traversal if entities found traversal_result = None if entities: traversal_result = traversal.find_paths( source_entities=entities, max_hops=3, strategy=strategy, max_paths=5 ) # Get graph stats stats = get_graph_stats(user_id) # Get comprehensive data - use get_user_data to preserve original columns from api.v1.endpoints.charts import get_user_data # 🧠 CHATGPT-STYLE: Load actual data context dynamically data_ctx = build_data_context(user_id) df = data_ctx.get("df") data_summary = "" if df is not None and not df.empty: currency_symbol, currency_code = get_user_currency(user_id) # Dynamic column detection - works with ANY data domain numeric_cols = df.select_dtypes(include=['int64', 'float64']).columns.tolist() text_cols = df.select_dtypes(include=['object']).columns.tolist() # Find amount column dynamically (Salary, amount, total, revenue, etc.) amount_col = None amount_keywords = ['salary', 'amount', 'total', 'revenue', 'price', 'cost', 'value', 'wage', 'pay'] for col in numeric_cols: if any(kw in col.lower() for kw in amount_keywords): amount_col = col break if not amount_col and numeric_cols: amount_col = numeric_cols[0] # Fallback to first numeric # Find grouping column dynamically (Department, customer, category, etc.) entity_col = None entity_keywords = ['department', 'customer', 'category', 'product', 'region', 'location', 'team', 'client'] for col in text_cols: if any(kw in col.lower() for kw in entity_keywords): entity_col = col break if not entity_col and text_cols: entity_col = text_cols[0] # Fallback to first text column # Calculate summary using detected columns total_value = df[amount_col].sum() if amount_col else 0 num_records = len(df) num_entities = df[entity_col].nunique() if entity_col else 0 data_summary = f""" ## 📊 Data Summary | Metric | Value | |--------|-------| | Total {amount_col or 'Value'} | {currency_symbol}{total_value:,.2f} | | Records | {num_records:,} | | Unique {entity_col or 'Entities'} | {num_entities:,} | ## 🧠 Data Schema: {data_ctx.get('context_string', 'No schema available')} """ # --- PRO DATA SUMMARY GENERATION ($5M Quality) --- query_limit = 10 if VIZ_INTELLIGENCE_AVAILABLE: query_limit = extract_count_from_query(question) # Context-aware extraction using detected columns if entity_col and amount_col: all_entities = df.groupby(entity_col)[amount_col].sum().sort_values(ascending=False) show_count = min(query_limit, len(all_entities)) data_summary += f"\n### Top {show_count} {entity_col} by {amount_col}\n" data_summary += f"| Rank | {entity_col} | {amount_col} | Count |\n|------|----------|---------|--------|\n" for i, (ent, amt) in enumerate(all_entities.head(show_count).items(), 1): count = df[df[entity_col] == ent].shape[0] data_summary += f"| {i} | {ent} | {currency_symbol}{amt:,.2f} | {count} |\n" # --- SMART CONTEXT \u0026 PERSONA ($5M Enterprise Quality) --- q_lower = question.lower() follow_up_indicators = ['above', 'that', 'those', 'chart', 'mentioned', 'previous', 'before', 'there is', 'in it', 'was shown', 'it', 'shown'] is_follow_up = any(w in q_lower for w in follow_up_indicators) or (len(question.split()) < 5 and any(w in q_lower for w in ['how many', 'who', 'what'])) # Redundant Chart Prevention: Only generate new chart if explicitly requested viz_keywords = ['chart', 'graph', 'plot', 'pie', 'bar', 'line', 'visualize', 'show', 'display', 'mindmap', 'mind map', 'knowledge graph', 'sankey', 'network', 'relationship'] explicit_viz_request = any(kw in q_lower for kw in viz_keywords) fact_keywords = ['how many', 'who', 'what', 'is there', 'total', 'count'] is_fact_query = any(kw in q_lower for kw in fact_keywords) should_generate_viz_block = explicit_viz_request if is_follow_up and is_fact_query and not explicit_viz_request: should_generate_viz_block = False # Protect against redundancy # Build Expert System Prompt system_prompt = f"""You are DataVision, an AI Data Analyst. ⚠️ CRITICAL - DATA GROUNDING RULES (NEVER VIOLATE): 1. ONLY use data from summaries and graph insights below 2. NEVER use outside knowledge or general industry facts 3. NEVER make up numbers - every number must be from the data 4. If data doesn't exist, say "I don't have data for that" {company_context_section} ## STRICT RULES: 1. Use EXACT numbers from the data provided 2. If user asks about "above chart", use ONLY previous turn's data 3. NEVER draw charts with text symbols - use markdown tables only 4. Do not say "As an AI..." or use filler phrases """ previous_context = "" if is_follow_up: last_response = _get_chat_history(user_id, limit=2) if last_response: previous_context = f""" ### 🚨 MANDATORY CONTEXT - READ THIS FIRST! 🚨 The user is asking about your PREVIOUS response/chart. PREVIOUS TURN CONTENT: --- {last_response} --- 🛑 DIRECTION: If referring to the chart or "above", use ONLY the numbers in THIS segment. Ignore the Total Dataset Stats if they contradict what was shown in the chart. """ print(f"[GRAPH] Follow-up detected. Engaging Context Lock.") prompt = f"""Question: {question} {user_context_section} {previous_context} ## 📊 RAW DATA SUMMARY (Total Dataset): {data_summary} ## 📈 GRAPH INSIGHTS (Relationships): {graph_insights} **Backend Statistics:** {stats.get('total_nodes', 0)} entities, {stats.get('total_edges', 0)} relationships ⚠️ FINAL INSTRUCTIONS: 1. Provide a one-sentence bolded summary lead. 2. If previous context applies (follow-up), prioritize it for "how many" counts. 3. Use currency {currency_symbol} for all amounts. 4. Charts will be generated AUTOMATICALLY - DO NOT output any JSON, Plotly code, or chart specifications. 5. Just provide clear text analysis. The visualization system handles charts separately. 6. NEVER include raw JSON objects like {{"type": "...", "data": [...]}} in your response.""" # $5M FIX: Build messages with context for GraphRAG messages = build_messages_with_context( user_id=user_id, current_question=question, current_prompt=prompt, history_limit=10 ) # Smart model routing for GraphRAG optimal_model = get_optimal_model( query_type=query_analysis.query_type.value, context_length=len(data_summary) ) answer = chat(messages=messages, system=system_prompt, max_tokens=1500, model=optimal_model) # Clean with anti-hallucination answer = _clean_response_formatting(answer, context=data_summary) # Generate charts if visualization requested # --- PRO-LEVEL VISUAL ORCHESTRATOR ($5M Quality) --- if should_generate_viz_block and df is not None and not df.empty: try: # 1. IDENTIFY SUBJECT \u0026 COLUMNS primary_entity = 'customer' if 'customer' in q_lower else ('product' if 'product' in q_lower else 'customer') revenue_col = 'amount' if 'amount' in df.columns else 'total_amount' # Accuracy fallback: ensure columns exist if primary_entity not in df.columns: for col in df.columns: if primary_entity in col.lower(): primary_entity = col break premium_visual_generated = False # 2. PREMIUM VISUAL: MINDMAPS, KNOWLEDGE GRAPHS, RELATIONSHIPS # Uses new LLM-driven graph_visualizations v2.0 module graph_viz_keywords = ['mindmap', 'mind map', 'knowledge graph', 'network', 'relationship', 'sankey', 'entity map', 'connections', 'radial', 'cluster', 'hierarchy', 'tree diagram'] if any(w in q_lower for w in graph_viz_keywords): try: from agents.graph_visualizations import ( generate_mindmap, generate_knowledge_graph, generate_relationship_diagram, detect_graph_visualization_type, generate_radial_mindmap, generate_entity_cluster, get_best_graph_visualization ) # LLM determines what type of visualization user wants viz_type = detect_graph_visualization_type(question) print(f"[GRAPH VIZ v2.0] Detected visualization type: {viz_type}") if viz_type == 'radial': # NEW: Pro-level radial mindmap chart, explanation = generate_radial_mindmap( graph=graph, query=question, max_depth=3 ) if chart: chart_json = json.dumps(chart, separators=(',', ':')) answer += f"\n\n```plotly_chart\n{chart_json}\n```" answer += f"\n\n{explanation}" premium_visual_generated = True print(f"[GRAPH VIZ v2.0] ✅ Radial Mindmap generated") elif viz_type == 'mindmap': # Generate Plotly treemap (frontend-compatible mindmap) chart, explanation = generate_mindmap( graph=graph, query=question, max_depth=3 ) if chart: chart_json = json.dumps(chart, separators=(',', ':')) answer += f"\n\n```plotly_chart\n{chart_json}\n```" answer += f"\n\n{explanation}" premium_visual_generated = True print(f"[GRAPH VIZ v2.0] ✅ Mindmap treemap generated") elif viz_type == 'cluster': # NEW: Entity clustering visualization chart, explanation = generate_entity_cluster( df=df, query=question, currency_symbol=currency_symbol ) if chart: chart_json = json.dumps(chart, separators=(',', ':')) answer += f"\n\n```plotly_chart\n{chart_json}\n```" answer += f"\n\n{explanation}" premium_visual_generated = True print(f"[GRAPH VIZ v2.0] ✅ Entity Cluster generated") elif viz_type == 'knowledge_graph': # Generate Plotly network diagram chart, explanation = generate_knowledge_graph( graph=graph, query=question, max_nodes=query_limit, currency_symbol=currency_symbol ) if chart: chart_json = json.dumps(chart, separators=(',', ':')) answer += f"\n\n```plotly_chart\n{chart_json}\n```" answer += f"\n\n{explanation}" premium_visual_generated = True print(f"[GRAPH VIZ v2.0] ✅ Knowledge Graph generated") elif viz_type in ['relationship', 'sankey']: # Generate Sankey relationship diagram chart, explanation = generate_relationship_diagram( df=df, query=question, currency_symbol=currency_symbol ) if chart: chart_json = json.dumps(chart, separators=(',', ':')) answer += f"\n\n```plotly_chart\n{chart_json}\n```" answer += f"\n\n{explanation}" premium_visual_generated = True print(f"[GRAPH VIZ] ✅ Relationship Diagram generated") except Exception as e: print(f"⚠️ Graph Visualization failed: {e}") import traceback traceback.print_exc() # 3. SMART CHARTS: Uses LLM to understand query and generate appropriate chart if not premium_visual_generated and SMART_CHART_AVAILABLE: try: chart_result, chart_explanation = smart_chart( query=question, df=df, currency_symbol=currency_symbol ) if chart_result: chart_json = json.dumps(chart_result, separators=(',', ':')) answer += f"\n\n```plotly_chart\n{chart_json}\n```" answer += f"\n\n{chart_explanation}" premium_visual_generated = True print(f"[GRAPH VIZ] ✅ Smart Chart generated") except Exception as e: print(f"⚠️ Smart Chart failed in Graph mode: {e}") # 4. PREDICTIONS (Non-exclusive) if not premium_visual_generated and any(w in q_lower for w in ['predict', 'forecast', 'future', 'trend']): # ... (prediction logic remains focused) ... pass except Exception as e: print(f"⚠️ Pro-Viz Engine global failure: {e}") # Final Summary - PRODUCTION: No debug footers # Remove any Analysis mode text that might have been added if "Analysis mode:" in answer: import re answer = re.sub(r'\n*---\n*\*?Analysis mode:?[^\n]*\*?\n?', '', answer, flags=re.IGNORECASE) # Anti-Artifact cleaning answer = _clean_response_formatting(answer) state.answer = answer state.route = "graph" state.sources = ["Enterprise Knowledge Graph"] except Exception as e: import traceback traceback.print_exc() print(f"[GRAPHRAG] Error: {e}") # Proper error message with debugging help state.answer = f"""**GraphRAG Processing Error** I encountered an issue analyzing the knowledge graph for your query: "{question[:50]}..." **Error Details:** {str(e)[:100]} **To fix this:** 1. Ensure you have uploaded data files in **Data Hub** 2. The graph is auto-built from CSV/Excel files with entity columns 3. Try using **RAG mode** for document-based queries **What GraphRAG needs:** - Structured data with entity columns (Department, Employee, Product, etc.) - The graph is built automatically when you upload structured files""" state.route = "graph" # Store turn for memory _store_conversation_turn(session_id, "assistant", state.answer) _save_chat_message(user_id, "assistant", state.answer) return state # ============================================================================ # 🟪 HYBRID MODE - Intelligent RAG + GraphRAG Fusion # ============================================================================ def hybrid_answer(state: AgentState) -> AgentState: """ 🟪 HYBRID MODE - Combined Document + Graph Analysis Features: - Intelligent query routing - Confidence-weighted fusion - Best of both RAG and Graph - Reasoning type indicator - Cache RAG integration (40-80% API cost savings) - 4-Layer memory context """ start_time = time.time() question = state.question user_id = state.company_id session_id = user_id # Use user_id as session_id # ========================================================================= # CONVERSATION MEMORY - ChatGPT-level stateful behavior # ========================================================================= _store_conversation_turn(session_id, "user", question) # Detect follow-up type and format constraints followup_type = _detect_followup_type(question) format_info = _detect_response_format(question) # Get conversation context for continuity conversation_context = _get_conversation_context(session_id) # Detect if this is prediction mode (from chat.py prefix) is_prediction_mode = "[PREDICTION MODE" in question if is_prediction_mode: # Remove the prefix for cleaner processing question = question.replace("[PREDICTION MODE - Use 3 accuracy tiers] ", "").strip() state.question = question print(f"📈 PREDICTION MODE detected - Will use prediction formatting") # ========================================================================= # CACHE CHECK - Fast path for cached queries # ========================================================================= if MEMORY_AVAILABLE: try: # Check cache first (returns immediately if exact hit) cache_result = memory_retrieve(question, user_id) if cache_result.cache_hit and cache_result.cached_answer: # EXACT CACHE HIT - Return immediately (< 100ms) state.answer = cache_result.cached_answer state.answer += f"\n\n---\n⚡ *Cached response ({cache_result.retrieval_time_ms:.0f}ms)*" state.route = "hybrid" state.context["cache_hit"] = True return state except Exception as cache_err: print(f"Cache lookup error: {cache_err}") # Classify query for routing query_analysis = classify_query(question) # Determine fusion weights based on query type if query_analysis.query_type in [QueryType.CAUSAL, QueryType.RELATIONAL]: weight_graph = 0.7 weight_rag = 0.3 primary_mode = "GraphRAG" elif query_analysis.query_type in [QueryType.FACTUAL, QueryType.AGGREGATION]: weight_graph = 0.3 weight_rag = 0.7 primary_mode = "RAG" else: weight_graph = 0.5 weight_rag = 0.5 primary_mode = "Balanced" # Get RAG context docs = retrieve(question, k=4, user_id=user_id) rag_context = "" sources = [] if docs: for doc in docs[:3]: rag_context += doc.get('text', '')[:600] + "\n\n" source = doc.get('source', doc.get('metadata', {}).get('source', '')) if source and source not in sources: sources.append(source) # Get Graph context graph_context = "" try: from graph.query import get_graph_summary from api.v1.endpoints.charts import get_user_data graph_context = get_graph_summary(user_id) or "" # 🧠 CHATGPT-STYLE: Use build_data_context for dynamic column detection data_ctx = build_data_context(user_id) df = data_ctx.get("df") if df is not None and not df.empty: currency_symbol, _ = get_user_currency(user_id) # Dynamic column detection - works with ANY data domain numeric_cols = df.select_dtypes(include=['int64', 'float64']).columns.tolist() text_cols = df.select_dtypes(include=['object']).columns.tolist() # Find amount column dynamically amount_col = None amount_keywords = ['salary', 'amount', 'total', 'revenue', 'price', 'cost', 'value'] for col in numeric_cols: if any(kw in col.lower() for kw in amount_keywords): amount_col = col break if not amount_col and numeric_cols: amount_col = numeric_cols[0] # Find entity column dynamically entity_col = None entity_keywords = ['department', 'customer', 'category', 'product', 'region', 'team'] for col in text_cols: if any(kw in col.lower() for kw in entity_keywords): entity_col = col break if not entity_col and text_cols: entity_col = text_cols[0] if amount_col: total = df[amount_col].sum() avg = df[amount_col].mean() graph_context += f"\n\n**{amount_col} Summary:** {currency_symbol}{total:,.2f} total, {currency_symbol}{avg:,.2f} avg" # Add top/bottom for entity queries if entity_col and amount_col: top_entity = df.groupby(entity_col)[amount_col].sum().idxmax() low_entity = df.groupby(entity_col)[amount_col].sum().idxmin() graph_context += f"\n**Top {entity_col}:** {top_entity}" graph_context += f"\n**Lowest {entity_col}:** {low_entity}" except Exception as e: pass # Check if we have enough data if not rag_context and not graph_context: state.answer = """I need data to provide hybrid analysis. **To use Hybrid mode:** • Upload files in **Data Hub** (CSV, Excel, PDF) • Hybrid combines document search with knowledge graph • Best for comprehensive business questions""" state.route = "hybrid" state.sources = [] return state currency_symbol, _ = get_user_currency(user_id) system_prompt = f"""You are DataVision, an AI Data Analyst. ⚠️ CRITICAL - DATA GROUNDING (NEVER VIOLATE): 1. ONLY use data from the context provided below 2. NEVER use outside knowledge or general industry facts 3. NEVER make up numbers - every number must be from YOUR data 4. If data doesn't exist, say "I don't have data for that" HYBRID MODE: RAG {weight_rag:.0%} + Graph {weight_graph:.0%} USER CURRENCY: {currency_symbol} ═══════════════════════════════════════════════════════════════ RESPONSE FORMAT ═══════════════════════════════════════════════════════════════ ALWAYS respond with: 1. One-line answer (bold) using {currency_symbol} 2. Table with data breakdown FROM YOUR DATA ONLY EXAMPLE: **Total Revenue: {currency_symbol}[amount from data]** | Metric | Value | |--------|-------| | Total Revenue | {currency_symbol}[from data] | | Customers | [from data] | | Orders | [from data] | ═══════════════════════════════════════════════════════════════ STRICT RULES ═══════════════════════════════════════════════════════════════ 1. Use | table | format for data (UNLESS user asks for brief) 2. Use {currency_symbol} as the currency symbol 3. NEVER say "typically" or "usually" - ONLY real data 3. Use ONLY data from the context provided 4. Be concise - no filler text 5. NEVER make up numbers CURRENCY CONVERSION (MANDATORY): - ₹88 INR = $1 USD (use this EXACT rate) - ₹92 INR = €1 EUR - ₹110 INR = £1 GBP - Show calculation when converting FORMAT CONSTRAINTS (STRICTLY FOLLOW): - "one word" → ONLY one word, NO tables, NO explanation - "one line" → ONE sentence only - "briefly" → 2-3 sentences max ANTI-HALLUCINATION: - ONLY use numbers from context - Do NOT add disclaimers about missing data when data IS provided - NEVER invent or estimate numbers""" # Get personalized user context user_context = _get_user_context(user_id) user_context_section = f"\n\nUser Context:\n{user_context}" if user_context else "" # Build format constraint section format_section = format_info.get("instructions", "") if format_info else "" # Build conversation context section conv_section = f"\n\n**Recent Conversation (for follow-ups):**\n{conversation_context}" if conversation_context else "" prompt = f"""Question: {question} Query Type: {query_analysis.query_type.value} Confidence: {query_analysis.confidence:.2f} {user_context_section} {conv_section} {format_section} **Document Context (RAG):** {rag_context if rag_context else "Limited document data"} **Graph Insights:** {graph_context if graph_context else "Limited graph data"} Based on the user's question and the data provided, give a direct answer. If user asks for "one word" or "briefly", follow that constraint STRICTLY. If this is a follow-up question, use the conversation context above.""" # $5M FIX: Build messages with context for Hybrid mode messages = build_messages_with_context( user_id=user_id, current_question=question, current_prompt=prompt, history_limit=10 ) # Smart model routing for Hybrid mode # Use reasoning model for complex queries (hybrid = more complex) combined_context = f"{rag_context or ''}\n{graph_context or ''}" optimal_model = get_optimal_model( query_type=query_analysis.query_type.value, context_length=len(combined_context) ) answer = chat(messages=messages, system=system_prompt, max_tokens=1500, model=optimal_model) # Clean with anti-hallucination validation answer = _clean_response_formatting(answer, context=combined_context) # ========================================================================= # ENTERPRISE ENGINE INTEGRATION # ========================================================================= if ENGINES_AVAILABLE: try: # Get data for engines - use get_user_data to preserve original columns from api.v1.endpoints.charts import get_user_data df = get_user_data(user_id) if user_id else None # ------------------------------------------------------------- # FORECAST ENGINE - Time-Series Prediction # ------------------------------------------------------------- if _wants_prediction(question) and df is not None and not df.empty: try: # Smart date detection from agents.visualization_intelligence import profile_dataframe profile = profile_dataframe(df) if profile.has_date_column and profile.date_column: date_col = profile.date_column import pandas as pd # revenue_dataframe() returns standardized column names: 'amount', 'date' amount_col = 'amount' # Always 'amount' from revenue_dataframe() print(f"[PREDICTION DEBUG] date_col={date_col}, amount_col={amount_col}, df.columns={list(df.columns)}") if amount_col in df.columns: df_temp = df.copy() df_temp['date_parsed'] = pd.to_datetime(df_temp[date_col], errors='coerce') df_dated = df_temp[df_temp['date_parsed'].notna()].copy() if not df_dated.empty: # Dynamic Granularity & Resampling (Google-style adaptive freq) df_dated.set_index('date_parsed', inplace=True) duration_days = (df_dated.index.max() - df_dated.index.min()).days if duration_days < 32: rule = 'D' period_multiplier = 30 title_suffix = "(Daily)" elif duration_days < 180: rule = 'W' period_multiplier = 4 title_suffix = "(Weekly)" else: rule = 'M' period_multiplier = 1 title_suffix = "(Monthly)" # Resample to ensure uniform time steps (critical for regression) time_data = df_dated[amount_col].resample(rule).sum().fillna(0).reset_index() time_data.columns = ['date', 'value'] time_data['date'] = time_data['date'].dt.strftime('%Y-%m-%d') # Get forecast with scaled periods months_requested = _extract_prediction_period(question) periods = months_requested * period_multiplier # Update title to reflect granularity title = f"Revenue Prediction {title_suffix}" forecast_result = forecast_from_dataframe(time_data, 'date', 'value', periods=periods) if forecast_result.get('success'): # Generate interactive chart if CHARTS_AVAILABLE: # Extract points directly from ForecastEngine result to preserve confidence intervals forecast_items = forecast_result.get('forecast', []) hist_points = [item for item in forecast_items if item.get('type') == 'historical'] pred_points = [item for item in forecast_items if item.get('type') == 'forecast'] chart_payload = generate_forecast_chart({ 'historical_points': hist_points, 'forecast_points': pred_points, 'title': title }) # Embed JSON payload for frontend import json answer += f"\n```forecast_chart\n{json.dumps(chart_payload, separators=(',', ':'))}\n```\n" answer += f"\n\n📈 **Forecast Analysis (AI-Powered)**\n" answer += f"📊 Trend: {forecast_result.get('trend', 'N/A').replace('_', ' ').title()}\n" answer += f"🎯 Confidence: {forecast_result.get('confidence', 'N/A')}\n\n" if forecast_result.get('insights'): answer += "**Forecast Insights:**\n" for insight in forecast_result['insights'][:3]: answer += f"• {insight}\n" else: # SNAPSHOT FALLBACK - No date column found # revenue_dataframe() returns standardized columns: 'amount', 'customer', 'product' amount_col = 'amount' total_revenue = df[amount_col].sum() if amount_col in df.columns else 0 total_customers = df['customer'].nunique() if 'customer' in df.columns else 0 total_orders = len(df) avg_order = total_revenue / total_orders if total_orders > 0 else 0 answer += f"\n\n📊 **Current Business Snapshot**\n" answer += f"| Metric | Value |\n|--------|-------|\n" answer += f"| Total Revenue | {currency_symbol}{total_revenue:,.2f} |\n" answer += f"| Total Customers | {total_customers} |\n" answer += f"| Total Orders | {total_orders} |\n" answer += f"| Avg Order Value | {currency_symbol}{avg_order:,.2f} |\n\n" answer += "📈 **Growth Potential Analysis**\n" answer += f"• Revenue per customer: {currency_symbol}{(total_revenue/total_customers):,.0f}\n" if total_customers > 0 else "" answer += f"• If top 20% customers increase by 10%: +{currency_symbol}{(total_revenue * 0.2 * 0.1):,.0f} potential\n" answer += f"• If customer base grows 15%: +{currency_symbol}{(total_revenue * 0.15):,.0f} projected\n\n" answer += "*Note: Time-series forecasting requires date column. Upload data with dates for trend predictions.*" except Exception as forecast_err: print(f"Forecast engine error: {forecast_err}") # ------------------------------------------------------------- # SIMULATION ENGINE - What-If Scenarios # ------------------------------------------------------------- if _wants_simulation(question) and df is not None and not df.empty: try: # Use schema detector for amount column try: from core.schema_detector import detect_schema schema = detect_schema(df, "simulation") amount_col = schema.best_amount_col entity_cols = schema.best_entity_cols except ImportError: amount_col = 'amount' if 'amount' in df.columns else 'total_amount' entity_cols = [] total_revenue = df[amount_col].sum() if amount_col and amount_col in df.columns else 0 # Find customer count from entity columns if entity_cols: total_customers = df[entity_cols[0]].nunique() else: ent_cols = [c for c in df.columns if df[c].dtype == 'object' and 10 < df[c].nunique() < 500] total_customers = df[ent_cols[0]].nunique() if ent_cols else 100 # Run simulation sim_result = simulate_scenarios( revenue=total_revenue, customers=total_customers, margin=0.2, churn=0.05 ) if sim_result.get('success'): if CHARTS_AVAILABLE: # Generate interactive scenario chart chart_payload = generate_scenario_chart(sim_result) answer += f"\n```forecast_chart\n{json.dumps(chart_payload, separators=(',', ':'))}\n```\n" answer += f"\n\n🎲 **Scenario Simulation Results**\n" answer += f"✅ **Best Strategy:** {sim_result.get('best_scenario', 'N/A')}\n" answer += f"💡 **Recommendation:** {sim_result.get('recommendation', 'N/A')}\n" answer += f"⚠️ **Risk Level:** {sim_result.get('risk_level', 'low').upper()}\n\n" # Show top scenarios scenarios = sim_result.get('scenarios', [])[:4] if scenarios: answer += "| Scenario | Revenue | Change |\n" answer += "|----------|---------|--------|\n" for s in scenarios: change = s.get('change_pct', 0) change_str = f"+{change}%" if change > 0 else f"{change}%" answer += f"| {s.get('name')} | {currency_symbol}{s.get('revenue', 0):,.0f} | {change_str} |\n" if sim_result.get('insights'): answer += "\n**Strategic Insights:**\n" for insight in sim_result['insights'][:3]: answer += f"• {insight}\n" except Exception as sim_err: print(f"Simulation engine error: {sim_err}") # ------------------------------------------------------------- # INSIGHT ENGINE - Automated Analysis # ------------------------------------------------------------- if _wants_insights(question) and df is not None and not df.empty: try: amount_col = 'amount' if 'amount' in df.columns else 'total_amount' total_revenue = df[amount_col].sum() total_customers = df['customer'].nunique() if 'customer' in df.columns else 0 total_orders = len(df) # Get top products and customers for analysis top_products = None top_customers = None if 'product' in df.columns: prod_rev = df.groupby('product')[amount_col].sum().sort_values(ascending=False) top_products = [{'name': p, 'revenue': r} for p, r in prod_rev.head(5).items()] if 'customer' in df.columns: cust_rev = df.groupby('customer')[amount_col].sum().sort_values(ascending=False) top_customers = [{'name': c, 'revenue': r} for c, r in cust_rev.head(5).items()] # Generate insights insight_result = generate_insights( revenue=total_revenue, revenue_previous=total_revenue * 0.9, # Assume 10% growth customers=total_customers, orders=total_orders, churn_rate=0.05, profit_margin=0.2, top_products=top_products, top_customers=top_customers ) if insight_result.get('success'): answer += f"\n\n🔍 **Automated Business Insights**\n" answer += f"📊 Health Score: {insight_result.get('health_score', 0):.0f}/100\n" answer += f"⚠️ Risk Score: {insight_result.get('risk_score', 0):.0f}/100\n" answer += f"💡 Opportunity Score: {insight_result.get('opportunity_score', 0):.0f}/100\n\n" if insight_result.get('top_priority'): answer += f"🎯 **Top Priority:** {insight_result['top_priority']}\n\n" insights = insight_result.get('insights', []) if insights: answer += f"**{len(insights)} Insights Found:**\n" for i in insights[:6]: icon = i.get('icon', '💡') msg = i.get('message', '') answer += f"{icon} {msg}\n" if i.get('recommendation'): answer += f" ↳ *{i['recommendation']}*\n" except Exception as insight_err: print(f"Insight engine error: {insight_err}") except Exception as engine_err: print(f"Engine integration error: {engine_err}") # ========================================================================= # BASIC VISUALIZATION CHARTS (Customer/Product/Trend) # ========================================================================= print(f"[VIZ DEBUG] CHARTS_AVAILABLE={CHARTS_AVAILABLE}, user_id={user_id}") if CHARTS_AVAILABLE: try: # Use get_user_data to preserve original columns (Department, Salary, etc.) from api.v1.endpoints.charts import get_user_data df = get_user_data(user_id) if user_id else None print(f"[VIZ DEBUG] df loaded: rows={len(df) if df is not None else 0}, columns={list(df.columns) if df is not None else []}") if df is not None and not df.empty: q_lower = question.lower() # Check for visualization keywords - comprehensive list viz_keywords = ['chart', 'graph', 'visual', 'plot', 'show', 'display', 'pie', 'bar', 'line', 'area', 'scatter', 'trend', 'breakdown', 'distribution'] wants_viz = any(kw in q_lower for kw in viz_keywords) print(f"[VIZ DEBUG] wants_viz={wants_viz}, matched_keywords={[kw for kw in viz_keywords if kw in q_lower]}") if wants_viz: # Extract count AND chart type from query dynamically from agents.visualization_intelligence import extract_count_from_query, extract_chart_type_from_query from agents.advanced_charts import generate_dynamic_chart chart_limit = extract_count_from_query(question) chart_type = extract_chart_type_from_query(question) # ========================================================= # USE STANDARDIZED COLUMN NAMES FROM revenue_dataframe() # The df returned has: 'amount', 'date', 'customer', 'product' # ========================================================= # Amount column is always 'amount' from revenue_dataframe y_col = 'amount' if 'amount' in df.columns else None # Determine entity based on query - use standardized column names if 'customer' in q_lower: x_col = 'customer' if 'customer' in df.columns else None entity = 'customer' elif 'product' in q_lower: x_col = 'product' if 'product' in df.columns else None entity = 'product' else: # Default to product for "top" queries, customer otherwise if 'top' in q_lower: x_col = 'product' if 'product' in df.columns else ('customer' if 'customer' in df.columns else None) entity = 'product' if 'product' in df.columns else 'customer' else: x_col = 'customer' if 'customer' in df.columns else ('product' if 'product' in df.columns else None) entity = 'customer' if 'customer' in df.columns else 'product' # Determine chart type using intelligent extraction if chart_type == 'auto': # Check direct keywords for chart types if 'pie' in q_lower or 'donut' in q_lower: chart_type = 'pie' elif 'line' in q_lower: chart_type = 'line' elif 'bar' in q_lower: chart_type = 'bar' elif 'area' in q_lower: chart_type = 'area' elif 'scatter' in q_lower: chart_type = 'scatter' elif 'trend' in q_lower or 'over time' in q_lower or 'monthly' in q_lower: chart_type = 'line' elif 'breakdown' in q_lower or 'distribution' in q_lower or 'share' in q_lower: chart_type = 'pie' elif 'compare' in q_lower or 'comparison' in q_lower: chart_type = 'bar' elif 'top' in q_lower or 'best' in q_lower or 'highest' in q_lower: chart_type = 'bar' else: chart_type = 'bar' # Generate dynamic chart using smart_chart for full type support print(f"[CHART DEBUG] Using smart_chart for query: {question[:50]}...") # Use smart_chart which has full chart type detection including pie, violin, radar, etc. if SMART_CHART_AVAILABLE: try: chart, chart_explanation = smart_chart( query=question, df=df, currency_symbol=currency_symbol ) if chart and 'error' not in chart: import json chart_json = json.dumps(chart, separators=(',', ':')) answer += f"\n\n```plotly_chart\n{chart_json}\n```" print(f"[CHART DEBUG] Successfully generated chart via smart_chart") except Exception as chart_err: print(f"[CHART DEBUG] smart_chart failed: {chart_err}") except Exception as viz_err: print(f"Visualization chart error: {viz_err}") # Add reasoning indicator answer += f"\n\n---\n**Reasoning Type:** {primary_mode} Fusion" answer += f"\n**Mode Weights:** RAG {weight_rag:.0%} | Graph {weight_graph:.0%}" answer += _build_source_citations(sources, "hybrid") # ========================================================================= # CACHE STORAGE - Store response for future fast retrieval # ========================================================================= if MEMORY_AVAILABLE: try: memory_store( query=question, answer=answer, workspace_id=user_id, reasoning_type="hybrid" ) except Exception as store_err: print(f"Cache store error: {store_err}") # Handle prediction mode formatting if is_prediction_mode: import re # Replace ALL HYBRID variations (plain, bold, italic, with emoji) hybrid_patterns = [ (r'Analysis Mode:\s*\*?HYBRID\*?', 'Analysis Mode: PREDICTION'), (r'\*Analysis Mode:\s*HYBRID\*', '**Analysis Mode: PREDICTION**'), (r'Mode:\s*HYBRID', 'Mode: PREDICTION'), (r'HYBRID\s*MODE', 'PREDICTION MODE'), (r'Reasoning Type:\s*(?:RAG Fusion|Balanced Fusion|Graph-Heavy)', 'Reasoning Type: Prediction Analysis'), (r'Mode Weights:\s*RAG', 'Prediction Weights: RAG'), ] for pattern, replacement in hybrid_patterns: answer = re.sub(pattern, replacement, answer, flags=re.IGNORECASE) # Remove the entire reasoning/mode weights section (it's hybrid terminology) # Pattern: ---\nReasoning Type: ...\nMode Weights: ...\n--- answer = re.sub( r'---\s*\n\s*Reasoning Type:[^\n]*\n\s*Mode Weights:[^\n]*\n\s*---', '', answer, flags=re.IGNORECASE ) # Also remove standalone reasoning type lines answer = re.sub(r'Reasoning Type:\s*Balanced Fusion[^\n]*\n?', '', answer) answer = re.sub(r'Mode Weights:\s*RAG[^\n]*\n?', '', answer) # Clean up multiple dashes answer = re.sub(r'(---\s*\n\s*){2,}', '---\n', answer) # PRODUCTION: Remove all Analysis Mode footers for clean output answer = re.sub(r'\n*---\n*\*?\*?Analysis Mode:?[^\n]*\n?', '', answer, flags=re.IGNORECASE) answer = re.sub(r'\n*---\n*📈\s*\*?\*?Analysis Mode:?[^\n]*\n?', '', answer, flags=re.IGNORECASE) answer = re.sub(r'Accuracy Tier:[^\n]*\n?', '', answer, flags=re.IGNORECASE) state.answer = answer state.route = "prediction" if is_prediction_mode else "hybrid" state.sources = sources state.context["fusion_weights"] = {"rag": weight_rag, "graph": weight_graph} state.context["primary_mode"] = primary_mode # Store assistant response in conversation memory for follow-ups _store_conversation_turn(session_id, "assistant", answer[:500]) # Store first 500 chars return state # ============================================================================ # 🟩 VISION MODE - Enhanced Image Analysis Pipeline # ============================================================================ def vision_answer(state: AgentState) -> AgentState: """ 🟩 VISION MODE - Enterprise Image Analysis Features: - Chart data extraction to JSON - Table detection and parsing - OCR text extraction - Vision → RAG/Graph pipeline """ start_time = time.time() question = state.question attached_files = state.context.get("attached_files", []) user_id = state.company_id if not attached_files: state.answer = """I need an image to analyze. **How to use Vision mode:** • Drag and drop an image into the chat • Or click the attachment button **I can analyze:** • 📊 Charts and graphs → Extract data points • 📋 Tables → Convert to structured data • 🧾 Invoices and receipts → Extract details • 📄 Documents → OCR and summarize • 📈 Screenshots → Identify trends and metrics Once extracted, I can query the data using RAG or Graph analysis.""" state.route = "vision" return state image_files = [f for f in attached_files if f.get('type', '').startswith('image/')] if not image_files: state.answer = """The attached file doesn't appear to be an image. **Supported formats:** PNG, JPG, JPEG, WebP""" state.route = "vision" return state try: from core.vision import analyze_image try: from mcp.vision_ocr import ( extract_tables_from_image, analyze_chart, vision_to_rag_context ) except ImportError: # Fallback - define stubs def extract_tables_from_image(path, output_format): return {"success": False, "tables": []} def analyze_chart(path): return {"success": False, "chart_data": None} def vision_to_rag_context(path, question): return {"success": False, "ready_for_rag": False} image_file = image_files[0] image_path = image_file.get('path', '') image_name = image_file.get('name', 'image') # Detect analysis type from question q_lower = question.lower() if any(word in q_lower for word in ['table', 'extract data', 'rows', 'columns']): # Table extraction mode table_result = extract_tables_from_image(image_path, output_format="markdown") if table_result.get("success") and table_result.get("tables"): answer = f"## Extracted Tables from {image_name}\n\n" for i, table in enumerate(table_result["tables"], 1): if isinstance(table, str): answer += f"### Table {i}\n{table}\n\n" else: answer += f"### Table {i}\n" answer += f"| {' | '.join(table.get('headers', []))} |\n" answer += f"| {' | '.join(['---'] * len(table.get('headers', [])))} |\n" for row in table.get('rows', []): answer += f"| {' | '.join(row)} |\n" answer += "\n" answer += f"\n*Extracted {len(table_result['tables'])} table(s)*" else: answer = "I couldn't extract tables from this image. Please ensure the image contains clear tabular data." elif any(word in q_lower for word in ['chart', 'graph', 'plot', 'trend', 'data points']): # Chart analysis mode chart_result = analyze_chart(image_path) if chart_result.get("success") and chart_result.get("chart_data"): chart_data = chart_result["chart_data"] answer = f"## 📊 Chart Analysis: {image_name}\n\n" if isinstance(chart_data, dict): if chart_data.get("chart_type"): answer += f"**Type:** {chart_data['chart_type']}\n" if chart_data.get("title"): answer += f"**Title:** {chart_data['title']}\n" if chart_data.get("data_series"): answer += "\n### 📈 Extracted Data\n" for series in chart_data["data_series"]: series_name = series.get('name', 'Series') answer += f"\n**{series_name}:**\n" answer += "| X Value | Y Value |\n" answer += "|---------|--------|\n" for point in series.get("data", []): x_val = point.get('x', 'N/A') y_val = point.get('y', 'N/A') answer += f"| {x_val} | {y_val} |\n" else: answer += chart_result.get("raw_analysis", "Unable to parse chart data") else: # Fall back to general analysis answer = analyze_image(image_path, question) else: # General image analysis analysis = analyze_image(image_path, question) answer = f"## Image Analysis: {image_name}\n\n{analysis}" # Check if user wants to continue with RAG/Graph analysis if any(word in q_lower for word in ['analyze further', 'query', 'insights', 'trends']): rag_context = vision_to_rag_context(image_path, question) if rag_context.get("success") and rag_context.get("ready_for_rag"): answer += "\n\n---\n*📊 Data extracted and ready for further analysis. Ask follow-up questions to explore insights.*" state.answer = answer state.route = "vision" state.sources = [image_name] except Exception as e: import traceback traceback.print_exc() state.answer = f"""I couldn't analyze the image due to an error. **Error:** {str(e)} **Troubleshooting:** • Ensure the image is valid (PNG, JPG, WebP) • Check that the GOOGLE_API_KEY is configured • Try a different image or format""" state.route = "vision" return state # ============================================================================ # FALLBACK MODE # ============================================================================ def fallback_answer(state: AgentState) -> AgentState: """Fallback when no data available""" state.answer = """I don't have any data to work with yet. **Getting Started:** 1. Go to **Data Hub** in the sidebar 2. Upload your business files (CSV, Excel, PDF, Images) 3. Return here to ask questions **Example questions once you have data:** • "What are the total sales?" • "Who are my top customers?" • "Show revenue trends" • "Which products sell the most?" **AI Modes Available:** • 🟦 **RAG** - Document search and retrieval • 🟧 **GraphRAG** - Knowledge graph reasoning • 🟪 **Hybrid** - Combined analysis (best results) • 🟩 **Vision** - Image and chart analysis""" state.route = "fallback" return state # ============================================================================ # AI MODEL MODE (DeepSeek, Mistral, Llama) # ============================================================================ def ai_model_answer(state: AgentState) -> AgentState: """ Handle direct AI model queries (DeepSeek, Mistral, Llama). Uses specified model directly with context from user's data. """ try: from agents.ai_models import ( get_ai_model_config, is_ai_model_mode, get_ai_model_system_prompt, ) question = state.question user_id = state.user_id mode = state.route # The AI model mode (e.g., "deepseek-chat") # Get model configuration model_config = get_ai_model_config(mode) if not model_config: state.answer = f"Unknown AI model: {mode}" return state # Get currency currency_symbol, currency_code = get_user_currency(user_id) # Try to get context from graph/vector context = "" sources = [] # Try graph context first try: from graph.query import query_graph graph_result = query_graph(question, user_id=user_id) if graph_result and graph_result.strip(): context = f"Business Data:\n{graph_result}" sources.append("knowledge_graph") except: pass # Try vector context if graph is empty if not context: try: from vector.retriever import retrieve vector_results = retrieve(question, k=5, user_id=user_id) if vector_results: context = "Document Context:\n" + "\n".join([r.get("content", "") for r in vector_results[:3]]) sources.extend([r.get("source", "document") for r in vector_results[:3]]) except: pass # Build system prompt system_prompt = f"""{get_ai_model_system_prompt()} USER CURRENCY: {currency_symbol} ({currency_code}) MODEL: {model_config.name} {context if context else "No business data loaded. Answer generally or ask user to upload data."}""" # Call LLM with specific model answer = chat( question, system=system_prompt, model=model_config.model_path, max_tokens=model_config.max_tokens, temperature=model_config.temperature, ) # Add model attribution answer += f"\n\n---\n*Powered by **{model_config.name}***" state.answer = answer state.route = mode state.sources = sources if sources else None except Exception as e: import traceback traceback.print_exc() state.answer = f"""I encountered an error with the AI model. **Error:** {str(e)} **Try:** • Switching to a different AI model • Using RAG or Hybrid mode instead • Checking your API configuration""" state.route = state.route or "ai_model" return state