Spaces:
Running
Running
| # 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'(?<!`)`{2}(?!`)', '', answer) | |
| # Remove excessive ** bold markers that break formatting | |
| answer = re.sub(r'\*\*\s*\*\*', '', answer) | |
| # Fix broken table separators (ensure proper | alignment) | |
| answer = re.sub(r'\| *\|', '| |', answer) | |
| # Remove empty table rows | |
| answer = re.sub(r'\n\|\s*\|\s*\|\s*\|\s*\|?\s*\n', '\n', answer) | |
| # Ensure consistent table header separators | |
| answer = re.sub(r'\|[-:]+\|', lambda m: '|' + '-' * (len(m.group(0)) - 2) + '|', answer) | |
| # SELECTIVE CLEANUP - Remove code blocks EXCEPT plotly_chart (our visualizations!) | |
| # Use negative lookahead to preserve plotly_chart blocks | |
| answer = re.sub(r'```(?!plotly_chart|forecast_chart)[a-zA-Z]*[\s\S]*?```', '', answer) | |
| # Remove LaTeX math blocks (these cause rendering issues) | |
| answer = re.sub(r'\$\$[\s\S]*?\$\$', '', answer) # Display math | |
| answer = re.sub(r'\$[^$\n]+\$', '', answer) # Inline math | |
| answer = re.sub(r'\\\\[a-z]+\{[^}]*\}', '', answer) # LaTeX commands | |
| # Remove hallucinated markdown image links (LLM tries to make its own images) | |
| # Pattern:  including attachment:// scheme | |
| answer = re.sub(r'!\[.*?\]\(.*?\)', '', answer) | |
| # Also remove bare attachment:// links and [text](attachment://...) links | |
| answer = re.sub(r'\[.*?\]\(attachment://[^)]*\)', '', answer) | |
| answer = re.sub(r'attachment://[^\s\)]+', '', answer) | |
| # PRESERVE VALID TABLES - Only remove incomplete/broken table rows | |
| # Do NOT remove complete tables, as they are essential for data presentation | |
| # Remove numbered emoji sections (1️⃣, 2️⃣, etc.) | |
| answer = re.sub(r'[0-9]️⃣\s*', '', answer) | |
| # Clean up emoji spacing | |
| answer = re.sub(r'([📊📈💰🎯💡👥📦✅⚠️🔗])\s{2,}', r'\1 ', answer) | |
| # Remove "Error:" messages that leaked through | |
| answer = re.sub(r'Error:\s*\n?', '', answer) | |
| # Remove leading/trailing whitespace from entire response | |
| answer = answer.strip() | |
| # Ensure proper spacing after headers | |
| answer = re.sub(r'(#{1,3}\s+[^\n]+)\n{3,}', r'\1\n\n', answer) | |
| # Ensure proper spacing before headers | |
| answer = re.sub(r'\n{4,}(#{1,3}\s+)', r'\n\n\1', answer) | |
| # Remove any remaining incomplete table rows (single | at end of line) | |
| answer = re.sub(r'\n\|[^|\n]*$', '', answer) | |
| # STEP 2: Anti-hallucination validation (if context provided) | |
| if context and OUTPUT_FORMATTER_AVAILABLE: | |
| answer = validate_numbers_in_response(answer, context) | |
| return answer | |
| # ============================================================================ | |
| # MEMORY & ANTI-HALLUCINATION HELPERS | |
| # ============================================================================ | |
| def _get_chat_history(user_id: str, limit: int = 5) -> 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"\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 | |