Spaces:
Running
Running
| """ | |
| Real Chat with RAG/Graph routing - NO FAKE DATA | |
| Uses existing agents and router system | |
| With semantic caching for API cost savings | |
| SECURED: Uses JWT authentication for user isolation | |
| PROTECTED: Rate limiting and AI security enabled | |
| """ | |
| from fastapi import APIRouter, HTTPException, Depends, Header, Request | |
| from pydantic import BaseModel | |
| from typing import List, Optional | |
| from pathlib import Path | |
| import traceback | |
| from datetime import datetime | |
| import json | |
| # 🔒 SECURITY: Import rate limiter | |
| try: | |
| from core.rate_limiter import check_rate_limit | |
| RATE_LIMITER_AVAILABLE = True | |
| except ImportError: | |
| RATE_LIMITER_AVAILABLE = False | |
| print("Rate limiter not available") | |
| # 🔒 SECURITY: Import AI security filter | |
| try: | |
| from core.ai_security import get_ai_security_filter, detect_prompt_injection | |
| AI_SECURITY_AVAILABLE = True | |
| except ImportError: | |
| AI_SECURITY_AVAILABLE = False | |
| print("AI security filter not available") | |
| from agents.router import route_question | |
| from vector.store_faiss import FaissStore | |
| from graph.query import load_graph, graph_snapshot, revenue_dataframe, get_user_currency | |
| from core.llm import chat, set_requested_model | |
| from core.cache import QueryCache | |
| from config.settings import Settings | |
| from utils.paths import get_user_paths, STORAGE_BASE | |
| from core.security_vault import SecurityVault | |
| from agents.market_context import MarketContextAgent | |
| # Import Tier 1 RAG enhancements | |
| try: | |
| from core.query_decomposer import decompose_query, is_complex_query, merge_results | |
| from core.mmr_search import mmr_rerank | |
| from core.answer_evaluator import evaluate_answer, get_confidence_badge | |
| from core.reranker import rerank, is_reranker_available | |
| RAG_ENHANCEMENTS = True | |
| print("RAG Enhancements loaded: Query Decomposition, MMR, Evaluator, Reranker") | |
| except ImportError as e: | |
| RAG_ENHANCEMENTS = False | |
| print(f" RAG Enhancements not available: {e}") | |
| # Import Tier 2 Advanced RAG (HyDE, Corrective, Self-Reflection) | |
| try: | |
| from core.hyde import should_use_hyde, generate_hypothetical_document_sync | |
| from core.corrective_rag import assess_retrieval_quality, reformulate_query | |
| from core.self_reflection import self_reflect_on_answer, self_rag_pipeline | |
| from core.model_config import get_model, get_model_api_id, get_available_models_api | |
| from core.query_router import get_routing_decision, analyze_query | |
| ADVANCED_RAG = True | |
| print("Advanced RAG loaded: HyDE, Corrective, Self-Reflection, Query Router") | |
| except ImportError as e: | |
| ADVANCED_RAG = False | |
| print(f" Advanced RAG not available: {e}") | |
| # Import Tier 3 Agentic RAG and Multi-RAG | |
| try: | |
| from core.agentic_rag import AgenticRAG, create_agentic_rag_prompt | |
| from core.multi_rag import MultiRAG, RetrievalSource, detect_best_sources, format_multi_rag_context | |
| AGENTIC_RAG = True | |
| print("Agentic RAG loaded: Tool-using agents, Multi-source retrieval, RRF fusion") | |
| except ImportError as e: | |
| AGENTIC_RAG = False | |
| print(f" Agentic RAG not available: {e}") | |
| # Import 5 Unique Mode Engines (Silicon Valley Powerhouses) | |
| try: | |
| from core.mode_engines.analyst_engine import analyst_response_sync | |
| from core.mode_engines.deepthink_engine import deepthink_response_sync | |
| from core.mode_engines.vision_engine import vision_response_sync | |
| from core.mode_engines.predict_engine import predict_response_sync | |
| from core.mode_engines.agent_engine import agent_response_sync | |
| MODE_ENGINES_AVAILABLE = True | |
| print("5 Unique Mode Engines loaded: Analyst, DeepThink, Vision, Predict, Agent") | |
| except ImportError as e: | |
| MODE_ENGINES_AVAILABLE = False | |
| print(f" Mode Engines not fully loaded: {e}") | |
| # Import chart generation for Plotly visualizations | |
| try: | |
| from api.v1.endpoints.charts import ( | |
| generate_revenue_trend_chart, | |
| generate_product_bar_chart, | |
| generate_customer_pie_chart, | |
| generate_prediction_chart, | |
| generate_query_aware_chart, # NEW: Query-aware dynamic charts | |
| get_user_data | |
| ) | |
| CHARTS_AVAILABLE = True | |
| except ImportError: | |
| CHARTS_AVAILABLE = False | |
| try: | |
| from agents.smart_chart import smart_chart | |
| SMART_CHART_AVAILABLE = True | |
| print("Smart Chart loaded in chat.py - LLM-driven charts active") | |
| except ImportError: | |
| SMART_CHART_AVAILABLE = False | |
| print("Smart Chart not available in chat.py") | |
| # Smart MCP - Claude-style auto-selected tools | |
| try: | |
| from core.smart_mcp import smart_mcp_execute, format_mcp_response | |
| SMART_MCP_AVAILABLE = True | |
| print("Smart MCP loaded - Claude-style tool execution active") | |
| except ImportError: | |
| SMART_MCP_AVAILABLE = False | |
| print("Smart MCP not available") | |
| # Import memory engine for chart context storage - USE SHARED SINGLETON | |
| try: | |
| from core.memory_engine import get_shared_memory | |
| _chart_memory = get_shared_memory() # Use singleton, not new instance! | |
| MEMORY_AVAILABLE = True | |
| except ImportError: | |
| _chart_memory = None | |
| MEMORY_AVAILABLE = False | |
| # Import auth dependencies | |
| try: | |
| from database.auth import get_current_user, get_current_user_optional | |
| except ImportError: | |
| # Fallback if auth module not found | |
| async def get_current_user_optional(authorization: Optional[str] = Header(None, alias="Authorization")): | |
| return None | |
| # Import Vector Store (Qdrant) for long-term memory | |
| try: | |
| from services.vector_store import VectorStoreService | |
| vector_store = VectorStoreService() | |
| VECTOR_MEMORY_AVAILABLE = vector_store.is_ready | |
| print(f" Vector Memory (Qdrant) loaded: {VECTOR_MEMORY_AVAILABLE}") | |
| except ImportError as e: | |
| vector_store = None | |
| VECTOR_MEMORY_AVAILABLE = False | |
| print(f" Vector Memory not available: {e}") | |
| # 🏆 Import Smart Suggestions for Competition-Winning Features | |
| try: | |
| from core.smart_suggestions import generate_smart_suggestions, calculate_confidence | |
| SMART_SUGGESTIONS_AVAILABLE = True | |
| print("Smart Suggestions loaded: Dynamic follow-ups and confidence scoring") | |
| except ImportError as e: | |
| SMART_SUGGESTIONS_AVAILABLE = False | |
| print(f" Smart Suggestions not available: {e}") | |
| # Fallback functions | |
| def generate_smart_suggestions(query, response, columns=None, max_suggestions=3): | |
| return [] | |
| def calculate_confidence(response, data_context, columns=None): | |
| return 0.75 | |
| router = APIRouter() | |
| # ============================================================================ | |
| # VISUALIZATION HELPER - Ensures charts work in ALL modes | |
| # ============================================================================ | |
| def append_chart_if_needed(response: str, query: str, user_id: str) -> str: | |
| """ | |
| Universal helper to append a Plotly chart to the response if visualization requested. | |
| Prevents duplication and ensures reliability across all 7 modes. | |
| """ | |
| if not CHARTS_AVAILABLE: | |
| return response | |
| viz_keywords = ['chart', 'graph', 'visualize', 'show', 'display', 'pie', 'bar', 'line', 'trend', 'top', 'breakdown', 'compare', 'distribution', 'performance', 'forecast', 'predict', 'projection', 'versus', 'vs', 'image', 'generate image', 'create chart'] | |
| wants_viz = any(kw in query.lower() for kw in viz_keywords) | |
| has_chart = '```plotly_chart' in response | |
| if wants_viz and not has_chart: | |
| try: | |
| df = get_user_data(user_id) | |
| if df is not None and not df.empty: | |
| # 🏆 Use PURE LLM smart chart generation - 100% autonomous | |
| try: | |
| from agents.smart_chart import generate_smart_chart | |
| chart_result = generate_smart_chart(query, df) | |
| if chart_result and '```plotly_chart' in chart_result: | |
| response += chart_result | |
| print(f" [VIZ] Generated SMART chart for: {query[:50]}...") | |
| return response | |
| except ImportError as ie: | |
| print(f" [VIZ] smart_chart import failed: {ie}") | |
| except Exception as se: | |
| print(f" [VIZ] smart_chart failed: {se}") | |
| # Fallback to generate_query_aware_chart if smart_chart fails | |
| try: | |
| chart_json = generate_query_aware_chart(df, query) | |
| if chart_json and 'error' not in chart_json: | |
| import json as json_lib | |
| import math | |
| # Clean NaN/Infinity values before serialization | |
| def clean_for_json(obj): | |
| if isinstance(obj, dict): | |
| return {k: clean_for_json(v) for k, v in obj.items()} | |
| elif isinstance(obj, list): | |
| return [clean_for_json(item) for item in obj] | |
| elif isinstance(obj, float): | |
| if math.isnan(obj) or math.isinf(obj): | |
| return 0 | |
| return obj | |
| return obj | |
| cleaned_chart = clean_for_json(chart_json) | |
| chart_block = f"\n\n```plotly_chart\n{json_lib.dumps(cleaned_chart, ensure_ascii=False)}\n```" | |
| response += chart_block | |
| print(f" [VIZ] Generated fallback chart for: {query[:50]}...") | |
| except Exception as fallback_err: | |
| print(f" [VIZ] Fallback chart failed: {fallback_err}") | |
| except Exception as chart_err: | |
| import traceback | |
| print(f" [VIZ] Chart generation failed: {chart_err}") | |
| traceback.print_exc() | |
| return response | |
| # Initialize query cache for API cost savings | |
| query_cache = QueryCache( | |
| max_entries=500, | |
| default_ttl=3600, # 1 hour cache | |
| enable_semantic_match=True # Enable semantic similarity matching | |
| ) | |
| # Clear old cache on module load (to clear cached EUR responses) | |
| query_cache.clear_all() | |
| print("Cache cleared on startup - fresh currency settings") | |
| # ============================================================================ | |
| # MODE PROFILES - Speed vs Complexity Configuration | |
| # ============================================================================ | |
| # Each mode has different strengths: some are FAST, some THINK DEEPLY | |
| MODE_PROFILES = { | |
| # FAST MODES - Quick responses for simple queries | |
| "rag": { | |
| "speed": "fast", | |
| "thinking_depth": "standard", | |
| "description": "📚 Document Search - Fast retrieval from your files", | |
| "best_for": ["quick lookups", "data queries", "simple questions"], | |
| "temperature": 0.3, | |
| "max_tokens": 2000, | |
| "uses_graph": False | |
| }, | |
| "chat": { | |
| "speed": "instant", | |
| "thinking_depth": "light", | |
| "description": "💬 Conversational - Instant friendly responses", | |
| "best_for": ["greetings", "small talk", "clarifications"], | |
| "temperature": 0.7, | |
| "max_tokens": 500, | |
| "uses_graph": False | |
| }, | |
| # BALANCED MODES - Good mix of speed and depth | |
| "hybrid": { | |
| "speed": "balanced", | |
| "thinking_depth": "moderate", | |
| "description": "🔀 Hybrid Analysis - Documents + Knowledge Graph", | |
| "best_for": ["complex queries", "multi-source answers", "verified data"], | |
| "temperature": 0.3, | |
| "max_tokens": 3000, | |
| "uses_graph": True | |
| }, | |
| "graph": { | |
| "speed": "balanced", | |
| "thinking_depth": "moderate", | |
| "description": "🕸️ Knowledge Graph - Entity relationships", | |
| "best_for": ["relationships", "connections", "entity queries"], | |
| "temperature": 0.3, | |
| "max_tokens": 2500, | |
| "uses_graph": True | |
| }, | |
| "graphrag": { | |
| "speed": "balanced", | |
| "thinking_depth": "moderate", | |
| "description": "🕸️ GraphRAG - Knowledge Graph Analysis", | |
| "best_for": ["entity relationships", "network analysis", "connections"], | |
| "temperature": 0.3, | |
| "max_tokens": 2500, | |
| "uses_graph": True | |
| }, | |
| # DEEP THINKING MODES - Complex reasoning, takes more time | |
| "agentic": { | |
| "speed": "deep", | |
| "thinking_depth": "advanced", | |
| "description": "🤖 AI Agent - Multi-step reasoning with tools", | |
| "best_for": ["complex analysis", "multi-step queries", "tool usage"], | |
| "temperature": 0.2, | |
| "max_tokens": 4000, | |
| "uses_graph": True | |
| }, | |
| "multirag": { | |
| "speed": "deep", | |
| "thinking_depth": "advanced", | |
| "description": "🔀 Multi-RAG - Cross-file comparison with fusion", | |
| "best_for": ["file comparison", "multi-source", "comprehensive analysis"], | |
| "temperature": 0.2, | |
| "max_tokens": 4000, | |
| "uses_graph": True | |
| }, | |
| "prediction": { | |
| "speed": "deep", | |
| "thinking_depth": "advanced", | |
| "description": "📈 Prediction - Trend forecasting with confidence", | |
| "best_for": ["forecasting", "trends", "future predictions"], | |
| "temperature": 0.2, | |
| "max_tokens": 3500, | |
| "uses_graph": False | |
| }, | |
| "vision": { | |
| "speed": "balanced", | |
| "thinking_depth": "visual", | |
| "description": "👁️ Vision - Image understanding + data extraction", | |
| "best_for": ["image analysis", "document scanning", "chart reading"], | |
| "temperature": 0.3, | |
| "max_tokens": 3000, | |
| "uses_graph": False | |
| } | |
| } | |
| # ============================================================================ | |
| # MCP PROCESSOR - Intelligent Tool Execution Based on Query and Enabled MCPs | |
| # ============================================================================ | |
| def run_enabled_mcps(query: str, user_id: str, enabled_mcps: dict, df=None) -> dict: | |
| """ | |
| Execute enabled MCP tools based on query intent and return enhanced context. | |
| This function analyzes the query and runs appropriate MCPs in sequence, | |
| building up context and insights for the LLM response. | |
| Args: | |
| query: User's question | |
| user_id: User ID for data access | |
| enabled_mcps: Dict of {mcp_name: bool} for enabled tools | |
| df: Optional DataFrame (if already loaded) | |
| Returns: | |
| Dict with: | |
| - mcp_context: Additional context from MCP tools | |
| - mcp_insights: List of insights generated | |
| - mcp_alerts: Any alerts triggered | |
| - tools_used: List of tools that ran | |
| """ | |
| result = { | |
| "mcp_context": "", | |
| "mcp_insights": [], | |
| "mcp_alerts": [], | |
| "tools_used": [], | |
| "data_quality_score": None | |
| } | |
| query_lower = query.lower() | |
| # Get DataFrame if not provided | |
| if df is None: | |
| try: | |
| from api.v1.endpoints.charts import get_user_data | |
| df = get_user_data(user_id) | |
| except: | |
| return result | |
| if df is None or df.empty: | |
| return result | |
| # EXPANDED Keywords to detect MCP relevance (partial matching for plurals) | |
| data_quality_keywords = [ | |
| 'quality', 'validat', 'check', 'clean', 'issue', 'error', 'problem', | |
| 'missing', 'null', 'empty', 'duplicate', 'correct', 'accurate', 'verify' | |
| ] | |
| alert_keywords = [ | |
| 'anomal', 'alert', 'unusual', 'spike', 'drop', 'warning', 'outlier', | |
| 'abnormal', 'unexpected', 'strange', 'weird', 'concern', 'risk', 'critical' | |
| ] | |
| transform_keywords = [ | |
| 'pivot', 'aggregat', 'group', 'transform', 'convert', 'reshape', | |
| 'merge', 'combine', 'split', 'filter', 'sort', 'breakdown', 'by region' | |
| ] | |
| insight_keywords = [ | |
| 'insight', 'trend', 'pattern', 'analyz', 'find', 'discover', 'show', | |
| 'summary', 'overview', 'important', 'significant', 'growth', 'performance' | |
| ] | |
| forecast_keywords = [ | |
| 'predict', 'forecast', 'future', 'next month', 'projection', 'estimate', | |
| 'expect', 'anticipat', 'plan', 'budget', 'target', 'quarterly', 'yearly' | |
| ] | |
| mcp_logs = [] | |
| # ========================================================================= | |
| # DATA VALIDATOR MCP - Check data quality | |
| # ========================================================================= | |
| if enabled_mcps.get('data_validator', False): | |
| is_quality_query = any(kw in query_lower for kw in data_quality_keywords) | |
| if is_quality_query: | |
| try: | |
| from mcp.data_validator import validate_data | |
| validation_result = validate_data(df) | |
| if validation_result.get('success'): | |
| quality_score = validation_result.get('quality_score', 0) | |
| result["data_quality_score"] = quality_score | |
| result["tools_used"].append("Data Validator") | |
| # Add quality context | |
| issues = validation_result.get('issues', []) | |
| if issues: | |
| issue_summary = ", ".join([f"{i['rule']}: {i['message'][:50]}" for i in issues[:3]]) | |
| result["mcp_context"] += f"\n📊 **Data Quality Score: {quality_score}/100**\n" | |
| result["mcp_context"] += f"Issues found: {len(issues)} ({issue_summary})\n" | |
| else: | |
| result["mcp_context"] += f"\n✅ **Data Quality Score: {quality_score}/100** - No issues found\n" | |
| mcp_logs.append(f"✅ Data Validator: Quality score {quality_score}") | |
| except Exception as e: | |
| mcp_logs.append(f"⚠️ Data Validator error: {str(e)[:50]}") | |
| # ========================================================================= | |
| # ALERT ENGINE MCP - Check for anomalies and threshold breaches | |
| # ========================================================================= | |
| if enabled_mcps.get('alert_engine', False): | |
| is_alert_query = any(kw in query_lower for kw in alert_keywords) | |
| # Always run for anomaly detection if enabled | |
| if is_alert_query or enabled_mcps.get('alert_engine', False): | |
| try: | |
| from mcp.alert_engine import evaluate_alerts | |
| # Run with default rules (anomaly detection) | |
| alert_result = evaluate_alerts(df) | |
| if alert_result.get('success'): | |
| alerts = alert_result.get('alerts', []) | |
| result["tools_used"].append("Alert Engine") | |
| # Add critical/high alerts to context | |
| important_alerts = [a for a in alerts if a.get('priority') in ['critical', 'high']] | |
| if important_alerts: | |
| result["mcp_context"] += f"\n🔔 **Alerts Detected ({len(important_alerts)}):**\n" | |
| for alert in important_alerts[:3]: | |
| result["mcp_context"] += f"- [{alert['priority'].upper()}] {alert['title']}: {alert['message'][:80]}\n" | |
| result["mcp_alerts"].append(alert) | |
| mcp_logs.append(f"🔔 Alert Engine: {len(important_alerts)} alerts") | |
| else: | |
| mcp_logs.append(f"✅ Alert Engine: No critical alerts") | |
| except Exception as e: | |
| mcp_logs.append(f"⚠️ Alert Engine error: {str(e)[:50]}") | |
| # ========================================================================= | |
| # INSIGHT ENGINE MCP - Generate business insights | |
| # ========================================================================= | |
| if enabled_mcps.get('insight_engine', False): | |
| is_insight_query = any(kw in query_lower for kw in insight_keywords) | |
| if is_insight_query: | |
| try: | |
| from mcp.insight_engine import generate_insights | |
| # Get basic metrics for insight generation | |
| numeric_cols = df.select_dtypes(include=['int64', 'float64']).columns | |
| # Find revenue-like column | |
| revenue_col = None | |
| for col in numeric_cols: | |
| if any(kw in col.lower() for kw in ['revenue', 'amount', 'total', 'sales', 'price']): | |
| revenue_col = col | |
| break | |
| if revenue_col: | |
| revenue = float(df[revenue_col].sum()) | |
| customers = df['Customer_Name'].nunique() if 'Customer_Name' in df.columns else 0 | |
| orders = len(df) | |
| insight_result = generate_insights( | |
| revenue=revenue, | |
| revenue_previous=revenue * 0.9, | |
| customers=customers, | |
| orders=orders | |
| ) | |
| if insight_result.get('insights'): | |
| result["tools_used"].append("Insight Engine") | |
| top_insights = insight_result['insights'][:3] | |
| result["mcp_context"] += f"\n💡 **Auto-Generated Insights:**\n" | |
| for ins in top_insights: | |
| icon = ins.get('icon', '💡') | |
| message = ins.get('message', '') | |
| result["mcp_context"] += f"- {icon} {message}\n" | |
| result["mcp_insights"].append(ins) | |
| mcp_logs.append(f"💡 Insight Engine: {len(top_insights)} insights") | |
| except Exception as e: | |
| mcp_logs.append(f"⚠️ Insight Engine error: {str(e)[:50]}") | |
| # ========================================================================= | |
| # FORECAST ENGINE MCP - Time-series predictions | |
| # ========================================================================= | |
| if enabled_mcps.get('forecast_engine', False): | |
| is_forecast_query = any(kw in query_lower for kw in forecast_keywords) | |
| if is_forecast_query: | |
| try: | |
| from mcp.forecast_engine import forecast_from_dataframe | |
| # Find date and value columns | |
| date_col = None | |
| value_col = None | |
| for col in df.columns: | |
| if any(kw in col.lower() for kw in ['date', 'time', 'month', 'year', 'period']): | |
| date_col = col | |
| break | |
| for col in df.select_dtypes(include=['int64', 'float64']).columns: | |
| if any(kw in col.lower() for kw in ['revenue', 'amount', 'total', 'sales']): | |
| value_col = col | |
| break | |
| if date_col and value_col: | |
| forecast_result = forecast_from_dataframe(df, date_col, value_col, periods=3) | |
| if forecast_result and 'forecast' in str(forecast_result): | |
| result["tools_used"].append("Forecast Engine") | |
| result["mcp_context"] += f"\n📈 **Forecast Generated:**\n" | |
| result["mcp_context"] += f"Based on {date_col} and {value_col} data\n" | |
| mcp_logs.append(f"📈 Forecast Engine: Generated forecast") | |
| except Exception as e: | |
| mcp_logs.append(f"⚠️ Forecast Engine error: {str(e)[:50]}") | |
| # ========================================================================= | |
| # DATA TRANSFORMER MCP - Pivot/aggregate operations | |
| # ========================================================================= | |
| if enabled_mcps.get('data_transformer', False): | |
| is_transform_query = any(kw in query_lower for kw in transform_keywords) | |
| if is_transform_query: | |
| try: | |
| from mcp.data_transformer import infer_data_schema | |
| schema_result = infer_data_schema(df) | |
| if schema_result.get('success'): | |
| result["tools_used"].append("Data Transformer") | |
| schema = schema_result.get('schema', {}) | |
| result["mcp_context"] += f"\n🔄 **Data Schema Detected:**\n" | |
| result["mcp_context"] += f"- Rows: {schema.get('row_count', 0)}, Columns: {schema.get('column_count', 0)}\n" | |
| mcp_logs.append(f"🔄 Data Transformer: Schema inferred") | |
| except Exception as e: | |
| mcp_logs.append(f"⚠️ Data Transformer error: {str(e)[:50]}") | |
| # ========================================================================= | |
| # ML PREDICTION MCP - Use trained AutoML models for predictions | |
| # ========================================================================= | |
| ml_predict_keywords = [ | |
| 'predict', 'classify', 'forecast', 'estimate', 'what will', 'what would', | |
| 'ml prediction', 'model prediction', 'trained model' | |
| ] | |
| if enabled_mcps.get('ml_prediction', True): # Enabled by default | |
| is_ml_query = any(kw in query_lower for kw in ml_predict_keywords) | |
| if is_ml_query: | |
| try: | |
| from mcp.ml_prediction_mcp import run_ml_prediction, get_ml_model_context | |
| # Get model context | |
| ml_context = get_ml_model_context(user_id) | |
| if ml_context: | |
| result["mcp_context"] += f"\n{ml_context}\n" | |
| result["tools_used"].append("ML Prediction Engine") | |
| # Try to make prediction | |
| pred_result = run_ml_prediction(query, user_id, df) | |
| if pred_result.get('success') and pred_result.get('prediction'): | |
| result["mcp_context"] += pred_result.get('explanation', '') | |
| mcp_logs.append(f"🤖 ML Prediction: {pred_result.get('prediction')}") | |
| else: | |
| mcp_logs.append(f"🤖 ML Model ready for predictions") | |
| except Exception as e: | |
| mcp_logs.append(f"⚠️ ML Prediction error: {str(e)[:50]}") | |
| # Log MCP execution | |
| if mcp_logs: | |
| print(f" MCP Tools Executed:") | |
| for log in mcp_logs: | |
| print(f" {log}") | |
| return result | |
| # ============================================================================ | |
| # MODEL PROFILES - AI Model Speed vs Intelligence | |
| # ============================================================================ | |
| MODEL_PROFILES = { | |
| # FAST MODELS - Quick, efficient responses | |
| "deepseek-chat": { | |
| "speed": "fast", | |
| "intelligence": "high", | |
| "description": "⚡ DeepSeek Chat - Fast & accurate", | |
| "best_for": ["quick queries", "coding", "data analysis"], | |
| "cost": "low" | |
| }, | |
| "mistral-7b": { | |
| "speed": "fast", | |
| "intelligence": "good", | |
| "description": "🚀 Mistral 7B - Lightweight & efficient", | |
| "best_for": ["simple queries", "fast responses"], | |
| "cost": "free" | |
| }, | |
| # BALANCED MODELS - Good mix | |
| "llama-70b": { | |
| "speed": "balanced", | |
| "intelligence": "very_high", | |
| "description": "🦙 Llama 70B - Powerful open-source", | |
| "best_for": ["complex reasoning", "analysis", "explanations"], | |
| "cost": "free" | |
| }, | |
| "gemini-pro": { | |
| "speed": "balanced", | |
| "intelligence": "very_high", | |
| "description": "🔮 Gemini Pro - Google's multimodal AI", | |
| "best_for": ["vision", "complex queries", "multimodal"], | |
| "cost": "low" | |
| }, | |
| # DEEP THINKING MODELS - Most intelligent, slower | |
| "claude-3": { | |
| "speed": "deep", | |
| "intelligence": "exceptional", | |
| "description": "🧠 Claude 3 - Deep reasoning champion", | |
| "best_for": ["complex analysis", "nuanced responses", "research"], | |
| "cost": "medium" | |
| }, | |
| "gpt-4": { | |
| "speed": "deep", | |
| "intelligence": "exceptional", | |
| "description": "🌟 GPT-4 - Industry standard", | |
| "best_for": ["complex tasks", "creative", "reasoning"], | |
| "cost": "high" | |
| } | |
| } | |
| # ============================================================================ | |
| # MCP PROFILES - Tool Speed & Capability | |
| # ============================================================================ | |
| MCP_PROFILES = { | |
| "data_cleaner": { | |
| "speed": "fast", | |
| "capability": "Data preprocessing", | |
| "description": "🧹 Clean and normalize data" | |
| }, | |
| "vectorizer": { | |
| "speed": "balanced", | |
| "capability": "Embedding generation", | |
| "description": "🔢 Generate embeddings for RAG" | |
| }, | |
| "graph_builder": { | |
| "speed": "deep", | |
| "capability": "Knowledge graph construction", | |
| "description": "🕸️ Build entity relationships" | |
| }, | |
| "sql_executor": { | |
| "speed": "fast", | |
| "capability": "Database queries", | |
| "description": "💾 Execute SQL on your data" | |
| }, | |
| "vision_ocr": { | |
| "speed": "balanced", | |
| "capability": "Image text extraction", | |
| "description": "📸 Extract text from images" | |
| } | |
| } | |
| def get_mode_config(mode: str) -> dict: | |
| """Get configuration for a specific mode""" | |
| return MODE_PROFILES.get(mode, MODE_PROFILES["rag"]) | |
| def get_optimal_settings(query: str, mode: str) -> dict: | |
| """Get optimal temperature and token settings based on query complexity""" | |
| config = get_mode_config(mode) | |
| # Detect complex queries that need more thinking | |
| complex_indicators = ['analyze', 'compare', 'explain why', 'predict', 'trend', 'relationship', 'correlation'] | |
| is_complex = any(ind in query.lower() for ind in complex_indicators) | |
| if is_complex and config["speed"] == "fast": | |
| # Boost thinking for complex queries even in fast modes | |
| return { | |
| "temperature": max(0.2, config["temperature"] - 0.1), | |
| "max_tokens": min(4000, config["max_tokens"] + 1000) | |
| } | |
| return { | |
| "temperature": config["temperature"], | |
| "max_tokens": config["max_tokens"] | |
| } | |
| class Message(BaseModel): | |
| role: str | |
| content: str | |
| timestamp: Optional[str] = None | |
| sources: Optional[List[str]] = None | |
| class Config: | |
| extra = "allow" # Allow extra fields like imageData, etc. | |
| class ChatRequest(BaseModel): | |
| user_id: Optional[str] = None | |
| userId: Optional[str] = None | |
| message: str | |
| model: Optional[str] = "llama" | |
| mode: str = "auto" | |
| role: str = "analyst" # User role: executive, manager, analyst, operator | |
| conversationId: Optional[str] = None | |
| compareFiles: Optional[List[str]] = None # For file comparison | |
| attachedFiles: Optional[List[dict]] = None # Newly uploaded files in chat | |
| enabledMcps: Optional[dict] = None # MCP servers toggle: {mcp_name: bool} | |
| conversationHistory: Optional[List[dict]] = None # 🧠 For persistent memory [{role, content}] | |
| class ChatResponse(BaseModel): | |
| message: str | |
| mode: str | |
| sources: Optional[List[str]] = None | |
| conversationId: str | |
| timestamp: str | |
| # 🏆 Competition-winning features | |
| suggestions: Optional[List[str]] = None # Dynamic follow-up suggestions | |
| confidence: Optional[float] = None # Data grounding confidence (0-1) | |
| # 🎯 Direct answer highlighting (ChatGPT/Claude style) | |
| directAnswer: Optional[dict] = None # {value, type, label, trend} | |
| # Streaming response support | |
| from fastapi.responses import StreamingResponse | |
| class StreamingChatRequest(BaseModel): | |
| """Request model for streaming chat endpoint""" | |
| user_id: Optional[str] = None | |
| userId: Optional[str] = None | |
| message: str | |
| mode: str = "rag" | |
| model: str = "deepseek" | |
| conversationId: Optional[str] = None | |
| def load_conversation(user_id: str, conversation_id: str) -> List[Message]: | |
| """Load conversation history with robust error handling""" | |
| try: | |
| paths = get_user_paths(user_id) | |
| history_file = paths["memory"] / f"{conversation_id}.json" | |
| if history_file.exists(): | |
| with open(history_file, 'r', encoding='utf-8') as f: | |
| data = json.load(f) | |
| messages = [] | |
| for msg in data.get("messages", []): | |
| try: | |
| # Extract only the fields we need, ignore extras | |
| messages.append(Message( | |
| role=msg.get("role", "user"), | |
| content=str(msg.get("content", "")), | |
| timestamp=msg.get("timestamp"), | |
| sources=msg.get("sources") | |
| )) | |
| except Exception as msg_err: | |
| print(f" Skip invalid message: {msg_err}") | |
| continue | |
| return messages | |
| return [] | |
| except Exception as e: | |
| print(f" Error loading conversation {conversation_id}: {e}") | |
| return [] | |
| def save_conversation(user_id: str, conversation_id: str, messages: List[Message]): | |
| """Save conversation with robust error handling""" | |
| try: | |
| paths = get_user_paths(user_id) | |
| history_file = paths["memory"] / f"{conversation_id}.json" | |
| # Ensure directory exists | |
| history_file.parent.mkdir(parents=True, exist_ok=True) | |
| # Convert messages to dict safely | |
| message_dicts = [] | |
| for msg in messages: | |
| try: | |
| message_dicts.append({ | |
| "role": msg.role, | |
| "content": str(msg.content) if msg.content else "", | |
| "timestamp": msg.timestamp, | |
| "sources": msg.sources | |
| }) | |
| except Exception as msg_err: | |
| print(f" Skip saving invalid message: {msg_err}") | |
| continue | |
| data = { | |
| "conversation_id": conversation_id, | |
| "user_id": user_id, | |
| "updated_at": datetime.now().isoformat(), | |
| "messages": message_dicts | |
| } | |
| with open(history_file, 'w', encoding='utf-8') as f: | |
| json.dump(data, f, indent=2, ensure_ascii=False) | |
| except Exception as e: | |
| print(f" Error saving conversation {conversation_id}: {e}") | |
| def clean_ai_response(response: str) -> str: | |
| """ | |
| Post-process AI response to remove code blocks, tables, LaTeX. | |
| Ensures clean, professional output even if LLM ignores prompt. | |
| PRODUCTION: Also removes debug footers like "Analysis Mode: X" | |
| """ | |
| import re | |
| if not response: | |
| return response | |
| # IMPORTANT: Preserve plotly_chart blocks (they contain our visualizations!) | |
| # Extract plotly_chart blocks first | |
| import re | |
| plotly_blocks = re.findall(r'```plotly_chart[\s\S]*?```', response) | |
| # Remove ALL OTHER code blocks (python, javascript, etc.) but NOT plotly_chart | |
| # Use negative lookahead to exclude plotly_chart | |
| response = re.sub(r'```(?!plotly_chart)[a-zA-Z]*[\s\S]*?```', '', response) | |
| # Remove LaTeX math | |
| response = re.sub(r'\$\$[\s\S]*?\$\$', '', response) | |
| response = re.sub(r'\$[^$\n]+\$', '', response) | |
| response = re.sub(r'\\\\[a-z]+\{[^}]*\}', '', response) | |
| # Remove markdown tables | |
| lines = response.split('\n') | |
| cleaned_lines = [] | |
| for line in lines: | |
| stripped = line.strip() | |
| # Skip table rows | |
| if stripped.startswith('|') and stripped.endswith('|') and stripped.count('|') >= 2: | |
| continue | |
| if stripped.startswith('|-') or stripped.startswith('|:'): | |
| continue | |
| cleaned_lines.append(line) | |
| response = '\n'.join(cleaned_lines) | |
| # Remove numbered emoji sections (1️⃣, 2️⃣, etc.) | |
| response = re.sub(r'[0-9]️⃣\s*', '', response) | |
| # Remove "Ready to drill deeper?" type endings | |
| response = re.sub(r'(?i)ready to drill deeper\?.*', '', response) | |
| response = re.sub(r'(?i)let me know if you.*', '', response) | |
| # ===================================================================== | |
| # PRODUCTION CLEANUP: Remove ALL debug footers | |
| # ===================================================================== | |
| # Remove "Analysis Mode: X" footers (all variations) | |
| response = re.sub(r'\n*---\n*\*?\*?📈?\s*\*?\*?Analysis Mode:?[^\n]*\*?\*?\n?', '', response, flags=re.IGNORECASE) | |
| response = re.sub(r'\n*---\n*\*?Analysis mode:?\s*\*?\*?[A-Z]+\*?\*?\n?', '', response, flags=re.IGNORECASE) | |
| response = re.sub(r'\*?Analysis Mode:\s*\*?\*?[A-Z]+\*?\*?', '', response, flags=re.IGNORECASE) | |
| # Remove "Accuracy Tier" text | |
| response = re.sub(r'Accuracy Tier:[^\n]*\n?', '', response, flags=re.IGNORECASE) | |
| # Remove "Enterprise Intelligence mode" text | |
| response = re.sub(r'\*?Enterprise Intelligence mode:?[^\n]*\*?\n?', '', response, flags=re.IGNORECASE) | |
| # Remove "Data Grounding:" debug text (but keep actual data grounding content) | |
| response = re.sub(r'🔗\s*\*?\*?Data Grounding:\*?\*?[^\n]*entities found[^\n]*\n?', '', response, flags=re.IGNORECASE) | |
| # Clean up trailing dashes and empty lines | |
| response = re.sub(r'\n*---\s*$', '', response) | |
| response = re.sub(r'(---\s*\n\s*){2,}', '---\n', response) | |
| # Clean up excessive newlines | |
| while '\n\n\n' in response: | |
| response = response.replace('\n\n\n', '\n\n') | |
| return response.strip() | |
| def get_file_metadata(user_id: str) -> dict: | |
| """Get all files with their upload dates and metadata""" | |
| paths = get_user_paths(user_id) | |
| file_info = {} | |
| if paths["files"].exists(): | |
| for file_path in paths["files"].iterdir(): | |
| if file_path.is_file(): | |
| stat = file_path.stat() | |
| file_info[file_path.name] = { | |
| "name": file_path.name, | |
| "uploaded_at": datetime.fromtimestamp(stat.st_mtime).isoformat(), | |
| "size": stat.st_size, | |
| "date": datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d"), | |
| "time": datetime.fromtimestamp(stat.st_mtime).strftime("%H:%M:%S") | |
| } | |
| return file_info | |
| def extract_date_intent(query: str) -> tuple: | |
| """Extract date-based intent from query""" | |
| import re | |
| from datetime import datetime, timedelta | |
| q_lower = query.lower() | |
| # Check for date keywords | |
| date_patterns = [ | |
| (r'(\d{4})-(\d{2})-(\d{2})', 'specific'), # YYYY-MM-DD | |
| (r'(january|february|march|april|may|june|july|august|september|october|november|december)\s+(\d{4})', 'month_year'), | |
| (r'(last|past)\s+(\d+)\s+(day|week|month|year)s?', 'relative'), | |
| (r'(today|yesterday|this week|last week|this month|last month)', 'named') | |
| ] | |
| for pattern, ptype in date_patterns: | |
| match = re.search(pattern, q_lower) | |
| if match: | |
| return ptype, match.groups() | |
| return None, None | |
| def filter_files_by_date(file_metadata: dict, query: str) -> List[str]: | |
| """Filter files based on date mentioned in query""" | |
| date_type, date_parts = extract_date_intent(query) | |
| if not date_type: | |
| return list(file_metadata.keys()) | |
| from datetime import datetime, timedelta | |
| filtered_files = [] | |
| for filename, meta in file_metadata.items(): | |
| file_date = datetime.fromisoformat(meta["uploaded_at"]) | |
| if date_type == 'specific' and date_parts: | |
| target_date = f"{date_parts[0]}-{date_parts[1]}-{date_parts[2]}" | |
| if meta["date"] == target_date: | |
| filtered_files.append(filename) | |
| elif date_type == 'named': | |
| today = datetime.now() | |
| if 'today' in query.lower(): | |
| if file_date.date() == today.date(): | |
| filtered_files.append(filename) | |
| elif 'yesterday' in query.lower(): | |
| if file_date.date() == (today - timedelta(days=1)).date(): | |
| filtered_files.append(filename) | |
| elif 'this week' in query.lower(): | |
| week_start = today - timedelta(days=today.weekday()) | |
| if file_date >= week_start: | |
| filtered_files.append(filename) | |
| elif 'last week' in query.lower(): | |
| week_start = today - timedelta(days=today.weekday() + 7) | |
| week_end = week_start + timedelta(days=7) | |
| if week_start <= file_date < week_end: | |
| filtered_files.append(filename) | |
| else: | |
| filtered_files.append(filename) | |
| return filtered_files if filtered_files else list(file_metadata.keys()) | |
| def rag_search(user_id: str, query: str, k: int = 5, target_files: List[str] = None) -> tuple: | |
| """ | |
| RAG search that ensures ALL uploaded files are represented in results. | |
| For comprehensive multi-file analysis. | |
| """ | |
| try: | |
| paths = get_user_paths(user_id) | |
| Settings.FAISS_DIR = paths["faiss"] | |
| # Load user-specific FAISS store | |
| store = FaissStore.load_or_create(user_id) | |
| # Get file metadata | |
| file_metadata = get_file_metadata(user_id) | |
| # Filter by date if mentioned in query | |
| if target_files is None: | |
| target_files = filter_files_by_date(file_metadata, query) | |
| # IMPROVED: Search with higher k to get more coverage | |
| results = store.search(query, k=25) # Increased for better coverage | |
| if not results: | |
| return "", [] | |
| # CRITICAL: Ensure we include at least some results from EACH file | |
| results_by_file = {} | |
| for r in results: | |
| source = r.get('metadata', {}).get('source', 'Unknown') | |
| if source not in results_by_file: | |
| results_by_file[source] = [] | |
| results_by_file[source].append(r) | |
| # Take top chunks from EACH file (ensures multi-file coverage) | |
| balanced_results = [] | |
| chunks_per_file = max(3, k // len(results_by_file)) if results_by_file else k | |
| for source, file_results in results_by_file.items(): | |
| # Take top chunks from each file | |
| balanced_results.extend(file_results[:chunks_per_file]) | |
| # If we have target files, prioritize them but include others | |
| if target_files: | |
| prioritized = [] | |
| others = [] | |
| for r in balanced_results: | |
| source = r.get('metadata', {}).get('source', '') | |
| if any(tf in source for tf in target_files): | |
| prioritized.append(r) | |
| else: | |
| others.append(r) | |
| balanced_results = prioritized + others[:5] # Keep some from others for context | |
| if not balanced_results: | |
| balanced_results = results[:k] # Fallback to original results | |
| # Apply MMR to diversify results and reduce redundancy | |
| if RAG_ENHANCEMENTS and len(balanced_results) > k: | |
| try: | |
| balanced_results = mmr_rerank(balanced_results, lambda_param=0.7, k=k) | |
| print(f" MMR applied: {len(balanced_results)} diverse results") | |
| except Exception as mmr_err: | |
| print(f"MMR error (non-critical): {mmr_err}") | |
| # Apply Cross-Encoder reranking for better relevance | |
| if RAG_ENHANCEMENTS and is_reranker_available() and len(balanced_results) > 2: | |
| try: | |
| balanced_results = rerank(query, balanced_results, top_k=k, text_key='text') | |
| print(f" Reranker applied: top {len(balanced_results)} by relevance") | |
| except Exception as rerank_err: | |
| print(f"Reranker error (non-critical): {rerank_err}") | |
| context_parts = [] | |
| sources = [] | |
| for i, result in enumerate(balanced_results): | |
| metadata = result.get('metadata', {}) | |
| source_file = metadata.get('source', 'Unknown') | |
| file_date = file_metadata.get(source_file, {}).get('date', 'Unknown date') | |
| file_time = file_metadata.get(source_file, {}).get('time', '') | |
| text_content = result.get('text', '') | |
| context_parts.append(f"[{i+1}] From: {source_file} (Uploaded: {file_date} {file_time})\n{text_content}") | |
| # Return clean source strings instead of objects | |
| sources.append(f"{source_file} ({file_date})") | |
| # Add file summary to context - SHOW ALL FILES | |
| file_summary = f"\n\n## Available Files ({len(file_metadata)}):\n" | |
| for fname, fmeta in file_metadata.items(): | |
| file_summary += f"- {fname} (Uploaded: {fmeta['date']} {fmeta['time']})\n" | |
| context = file_summary + "\n## Relevant Content:\n" + "\n\n".join(context_parts) | |
| return context, list(set(sources)) # Deduplicate sources | |
| except Exception as e: | |
| print(f"RAG error: {e}") | |
| traceback.print_exc() | |
| return "", [] | |
| def graph_query(user_id: str, question: str) -> tuple: | |
| try: | |
| paths = get_user_paths(user_id) | |
| Settings.GRAPH_DIR = paths["graph"] | |
| # Check if graph exists | |
| graph = load_graph(user_id) | |
| if not graph: | |
| return "No knowledge graph available. Please upload and train some data first in Data Hub.", ["Graph Mode - No Data"] | |
| snapshot = graph_snapshot(user_id, max_nodes=50) | |
| revenue_context = "" | |
| if any(kw in question.lower() for kw in ['revenue', 'sales', 'invoice', 'amount', 'customer', 'product', 'currency', 'money', 'total']): | |
| try: | |
| df = revenue_dataframe(user_id) | |
| if df is not None and not df.empty: | |
| # Multi-currency support - group by currency | |
| currency_breakdown = {} | |
| if 'currency' in df.columns: | |
| for curr in df['currency'].unique(): | |
| curr_total = df[df['currency'] == curr]['amount'].sum() if 'amount' in df.columns else 0 | |
| currency_breakdown[curr] = curr_total | |
| else: | |
| # Single currency - use detected currency or default to USD | |
| from utils.currency import detect_currency | |
| detected_currency = detect_currency(df, paths["files"]) | |
| total = df['amount'].sum() if 'amount' in df.columns else 0 | |
| currency_breakdown[detected_currency] = total | |
| num_invoices = len(df) | |
| num_customers = df['customer'].nunique() if 'customer' in df.columns else 0 | |
| # Format currency breakdown | |
| from utils.currency import get_currency_symbol, format_currency, calculate_currency_breakdown | |
| breakdown = calculate_currency_breakdown(currency_breakdown) | |
| revenue_context = f"\n\nRevenue Data from Your Files:\n" | |
| revenue_context += f"- Total Invoices: {num_invoices}\n" | |
| revenue_context += f"- Unique Customers: {num_customers}\n" | |
| revenue_context += f"\n💰 Currency Breakdown ({len(currency_breakdown)} currencies detected):\n" | |
| for item in breakdown['breakdown']: | |
| revenue_context += f" * {item['name']} ({item['currency']}): {item['formatted']}\n" | |
| revenue_context += f" (USD Equivalent: ${item['usd_equivalent']:,.2f})\n" | |
| if len(currency_breakdown) > 1: | |
| revenue_context += f"\n📊 Combined Total (USD Equivalent): {breakdown['total_usd_formatted']}\n" | |
| # Top products by each currency | |
| if 'product' in df.columns and 'currency' in df.columns: | |
| revenue_context += "\n- Top Products by Currency:\n" | |
| for curr in df['currency'].unique(): | |
| curr_df = df[df['currency'] == curr] | |
| top = curr_df.groupby('product')['amount'].sum().sort_values(ascending=False).head(3) | |
| symbol = get_currency_symbol(curr) | |
| revenue_context += f" {symbol} {curr}:\n" | |
| for prod, amt in top.items(): | |
| revenue_context += f" * {prod}: {symbol}{amt:,.2f}\n" | |
| elif 'product' in df.columns: | |
| top_products = df.groupby('product')['amount'].sum().sort_values(ascending=False).head(5) | |
| revenue_context += "\n- Top Products:\n" | |
| primary_currency = list(currency_breakdown.keys())[0] if currency_breakdown else 'USD' | |
| symbol = get_currency_symbol(primary_currency) | |
| for prod, amt in top_products.items(): | |
| revenue_context += f" * {prod}: {symbol}{amt:,.2f}\n" | |
| except Exception as e: | |
| print(f"Revenue error: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| context = snapshot + revenue_context | |
| sources = ["Knowledge Graph Analysis"] | |
| return context, sources | |
| except Exception as e: | |
| print(f"Graph error: {e}") | |
| return "", [] | |
| def hybrid_search(user_id: str, query: str, target_files: List[str] = None) -> tuple: | |
| rag_context, rag_sources = rag_search(user_id, query, k=5, target_files=target_files) | |
| graph_context, graph_sources = graph_query(user_id, query) | |
| context = f"## Document Search (RAG):\n{rag_context}\n\n## Knowledge Graph Analysis:\n{graph_context}" | |
| sources = rag_sources + graph_sources | |
| return context, sources | |
| def vision_analysis(user_id: str, query: str, attached_files: Optional[List[dict]] = None) -> tuple: | |
| """Analyze images using vision capabilities - REAL IMPLEMENTATION""" | |
| try: | |
| if not attached_files: | |
| context = """## Vision Mode | |
| 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 | |
| • 📈 Screenshots → Identify trends""" | |
| return context, ["Vision Mode - No Image"] | |
| # Filter for image files | |
| image_files = [f for f in attached_files if f.get('type', '').startswith('image/')] | |
| if not image_files: | |
| context = "No image files detected. Please upload an image (JPG, PNG, WebP)." | |
| return context, ["Vision Mode - No Images"] | |
| # Get the first image | |
| image_file = image_files[0] | |
| image_path = image_file.get('path', '') | |
| image_name = image_file.get('name', 'image') | |
| if not image_path: | |
| context = "Image file path not found. Please try uploading again." | |
| return context, ["Vision Mode - Path Error"] | |
| # Import and use real vision module | |
| from core.vision import analyze_image, extract_chart_data, extract_table_data | |
| # Determine analysis type based on query | |
| query_lower = query.lower() | |
| if any(word in query_lower for word in ['table', 'extract data', 'rows', 'columns', 'read']): | |
| # Table extraction mode | |
| result = extract_table_data(image_path) | |
| if 'tables' in result: | |
| context = f"## Table Extraction: {image_name}\n\n" | |
| for i, table in enumerate(result.get('tables', [])): | |
| headers = table.get('headers', []) | |
| rows = table.get('rows', []) | |
| if headers: | |
| context += f"| {' | '.join(headers)} |\n" | |
| context += f"| {' | '.join(['---'] * len(headers))} |\n" | |
| for row in rows: | |
| context += f"| {' | '.join(row)} |\n" | |
| context += "\n" | |
| else: | |
| context = f"## Table Extraction: {image_name}\n\n{result.get('raw_analysis', 'No tables found')}" | |
| elif any(word in query_lower for word in ['chart', 'graph', 'plot', 'data points', 'values']): | |
| # Chart analysis mode | |
| result = extract_chart_data(image_path) | |
| if 'data_series' in result: | |
| context = f"## Chart Analysis: {image_name}\n\n" | |
| context += f"**Chart Type:** {result.get('chart_type', 'Unknown')}\n" | |
| context += f"**Title:** {result.get('title', 'N/A')}\n\n" | |
| for series in result.get('data_series', []): | |
| context += f"### {series.get('name', 'Data')}\n" | |
| context += "| Label | Value |\n|------|------|\n" | |
| for point in series.get('data', []): | |
| context += f"| {point.get('label', point.get('x', ''))} | {point.get('value', point.get('y', ''))} |\n" | |
| else: | |
| context = f"## Chart Analysis: {image_name}\n\n{result.get('raw_analysis', 'Analysis not available')}" | |
| else: | |
| # General image analysis | |
| analysis = analyze_image(image_path, query if query else "Describe this image in detail") | |
| context = f"## Image Analysis: {image_name}\n\n{analysis}" | |
| sources = [f"Vision Analysis: {image_name}"] | |
| return context, sources | |
| except Exception as e: | |
| print(f"Vision analysis error: {e}") | |
| traceback.print_exc() | |
| return f"Vision analysis error: {str(e)}", ["Vision Mode - Error"] | |
| # ============================================================================ | |
| # AI BUSINESS ANALYST - PROFESSIONAL MODEL CONFIGURATION | |
| # Smart routing based on query type - like real SaaS analytics products | |
| # ============================================================================ | |
| # ONLY FREE MODELS ON OPENROUTER (No credits required) | |
| # ⚠️ DeepSeek requires payment - REMOVED! | |
| AI_MODELS = { | |
| # 🥇 PRIMARY - Maps to Llama 70B (FREE and powerful) | |
| 'deepseek': 'meta-llama/llama-3.3-70b-instruct:free', # Redirect to free model | |
| 'deepseek-chat': 'meta-llama/llama-3.3-70b-instruct:free', # Alias | |
| # 🎯 FAST - Mistral 7B (FREE) | |
| 'mistral-7b': 'mistralai/mistral-7b-instruct:free', | |
| # 🦙 COMPREHENSIVE - Meta Llama 70B (FREE) | |
| 'llama-70b': 'meta-llama/llama-3.3-70b-instruct:free', | |
| } | |
| # Smart model router - selects best model based on query keywords | |
| def smart_route_model(query: str, selected_model: str) -> str: | |
| """ | |
| Routes to the optimal model based on query type. | |
| User's selected model takes priority. | |
| """ | |
| query_lower = query.lower() | |
| # If user explicitly selected a model, respect their choice | |
| if selected_model in AI_MODELS: | |
| return AI_MODELS[selected_model] | |
| # DEFAULT: Llama 70B (best FREE model for business analysis) | |
| return AI_MODELS['llama-70b'] | |
| # Chart keywords for detecting visualization requests | |
| # Chart keywords for detecting visualization requests | |
| CHART_KEYWORDS = [ | |
| 'chart', 'graph', 'visualize', 'visualization', 'plot', 'trend', | |
| 'show me', 'display', 'bar chart', 'pie chart', 'line chart', | |
| 'revenue trend', 'customer distribution', 'product comparison', | |
| 'img', 'image', 'picture', 'dashboard', | |
| 'violin', 'radar', 'spider', 'funnel', 'gauge', 'heatmap', | |
| 'treemap', 'sunburst', 'bubble', 'scatter', 'box plot', | |
| 'waterfall', 'area chart', 'donut', 'histogram' | |
| ] | |
| def detect_chart_request(query: str) -> str: | |
| """Detect what type of chart the user is asking for""" | |
| query_lower = query.lower() | |
| # Specific chart types (Priority) | |
| if 'violin' in query_lower: return 'violin' | |
| if 'radar' in query_lower or 'spider' in query_lower: return 'radar' | |
| if 'funnel' in query_lower: return 'funnel' | |
| if 'heatmap' in query_lower: return 'heatmap' | |
| if 'treemap' in query_lower: return 'treemap' | |
| if 'sunburst' in query_lower: return 'sunburst' | |
| if 'gauge' in query_lower or 'kpi' in query_lower: return 'gauge' | |
| if 'waterfall' in query_lower: return 'waterfall' | |
| if 'bubble' in query_lower: return 'bubble' | |
| if 'scatter' in query_lower or 'correlation' in query_lower: return 'scatter' | |
| if 'box' in query_lower and 'plot' in query_lower: return 'box' | |
| if 'histogram' in query_lower or 'frequency' in query_lower: return 'histogram' | |
| if 'donut' in query_lower or 'doughnut' in query_lower: return 'donut' | |
| if 'area' in query_lower and 'chart' in query_lower: return 'area' | |
| # General categories | |
| if any(k in query_lower for k in ['trend', 'over time', 'timeline', 'revenue trend', 'line chart']): | |
| return 'line' | |
| elif any(k in query_lower for k in ['bar', 'product', 'compare', 'comparison', 'top product', 'ranking']): | |
| return 'bar' | |
| elif any(k in query_lower for k in ['pie', 'distribution', 'breakdown', 'customer', 'share']): | |
| return 'pie' | |
| elif any(k in query_lower for k in ['predict', 'forecast', 'future', 'next month', 'projection']): | |
| return 'prediction' | |
| # Catch-all for generic "show me a chart" | |
| elif any(k in query_lower for k in CHART_KEYWORDS): | |
| return 'generic' | |
| return None # Not a chart request | |
| async def ai_model_response(user_id: str, query: str, model_key: str, conversation_context: str = "") -> tuple: | |
| """ | |
| Generate response using OpenRouter AI models with RAG context. | |
| This ensures answers are grounded in the user's actual data - no hallucination. | |
| Flow: | |
| 1. Get RAG context from user's trained data | |
| 2. Get Graph context for structured insights | |
| 3. Combine contexts with anti-hallucination prompt | |
| 4. Call OpenRouter API with selected model | |
| 5. Return clean, data-grounded response | |
| """ | |
| import aiohttp | |
| import os | |
| import re | |
| try: | |
| # ===================================================================== | |
| # STEP 0: MEMORY - Extract and save personal info, get user context | |
| # ===================================================================== | |
| user_context = "" | |
| user_name = None | |
| # ===================================================================== | |
| # ChatGPT-LEVEL MEMORY: Read and Write | |
| # ===================================================================== | |
| try: | |
| from core.memory import process_personal_info, get_user_context, get_user_name | |
| # MEMORY WRITE: Save any personal info in the query | |
| saved = process_personal_info(user_id, query) | |
| # MEMORY READ: Get user's stored name directly | |
| user_name = get_user_name(user_id) | |
| # Get full user context for LLM prompt | |
| user_context = get_user_context(user_id) or "" | |
| # If name was just saved, confirm it immediately | |
| if saved and user_name: | |
| query_lower = query.lower() | |
| if any(phrase in query_lower for phrase in ['my name is', 'i am', "i'm", 'call me', 'you can call me']): | |
| return ( | |
| f"Nice to meet you, **{user_name}**! I've saved your name and will remember you.\n\n" | |
| f"How can I help you analyze your business data today?", | |
| ["Memory"] | |
| ) | |
| except Exception as mem_err: | |
| print(f"Memory error: {mem_err}") | |
| import traceback | |
| traceback.print_exc() | |
| # ===================================================================== | |
| # MEMORY READ: Answer identity questions from stored memory | |
| # ===================================================================== | |
| query_lower = query.lower() | |
| if any(phrase in query_lower for phrase in ['my name', 'who am i', 'what is my name', 'tell me my name', 'do you know my name', 'remember my name']): | |
| if user_name: | |
| # Name EXISTS in memory → answer directly | |
| return ( | |
| f"Your name is **{user_name}**.\n\n" | |
| f"How can I help you with your business data today?", | |
| ["Memory"] | |
| ) | |
| else: | |
| # Name NOT in memory → ask politely | |
| return ( | |
| "I don't have your name saved yet. Please tell me your name (e.g., 'My name is Naveen') and I'll remember you!", | |
| ["Memory"] | |
| ) | |
| # Step 1: Get RAG context (document search) | |
| rag_context, rag_sources = rag_search(user_id, query, k=8) | |
| # Step 2: Get Graph context (structured data) | |
| graph_context, graph_sources = graph_query(user_id, query) | |
| # Step 3: Build intelligent multi-file context | |
| # Parse and organize data by file source | |
| data_context = "" | |
| # Build file inventory from sources | |
| all_sources = list(set(rag_sources + graph_sources)) | |
| file_sources = [s for s in all_sources if s.endswith(('.xlsx', '.csv', '.pdf'))] | |
| if file_sources: | |
| data_context += "## 📁 YOUR DATA FILES:\n" | |
| for i, f in enumerate(file_sources[:5], 1): | |
| data_context += f"{i}. {f}\n" | |
| data_context += "\n" | |
| if rag_context and rag_context.strip(): | |
| data_context += f"## 📄 DOCUMENT DATA (from uploaded files):\n{rag_context}\n\n" | |
| if graph_context and graph_context.strip(): | |
| data_context += f"## 📊 STRUCTURED DATA (knowledge graph):\n{graph_context}\n\n" | |
| # Add file-specific notes (domain-agnostic) | |
| data_context += """ | |
| ## 📋 IMPORTANT NOTES: | |
| - Each file above contains your uploaded data | |
| - Use data from ALL relevant files to answer comprehensively | |
| - When comparing files, note which file each number comes from | |
| """ | |
| # Check if we have any data | |
| if not rag_context and not graph_context: | |
| return ( | |
| "⚠️ **No data found in your Data Hub.**\n\n" | |
| "To get insights, please upload your data files:\n" | |
| "1. Go to **Data Hub**\n" | |
| "2. Upload CSV, Excel, PDF or other data files\n" | |
| "3. The AI will learn from your data automatically\n\n" | |
| "Then ask me questions about your data!", | |
| ["No Data Available"] | |
| ) | |
| # Get user's currency setting | |
| currency_symbol, currency_code = get_user_currency(user_id) | |
| # ===================================================================== | |
| # DYNAMIC DOMAIN DETECTION - Works with ANY uploaded data | |
| # ===================================================================== | |
| detected_domain = "Data" | |
| available_metrics = [] | |
| available_dimensions = [] | |
| data_summary = "Your uploaded files" | |
| try: | |
| # Try to detect domain from user's data | |
| from core.schema_intelligence import UniversalSchemaAnalyzer | |
| from api.v1.endpoints.schema_api import _load_user_data | |
| df = _load_user_data(user_id) | |
| if df is not None and not df.empty: | |
| analyzer = UniversalSchemaAnalyzer() | |
| schema = analyzer.analyze_dataframe(df, "data") | |
| detected_domain = schema.domain if schema.domain else "Data" | |
| available_metrics = schema.key_metrics[:10] # Top 10 metrics | |
| available_dimensions = schema.dimensions[:10] # Top 10 dimensions | |
| # Build data summary showing what's available | |
| cols = list(df.columns)[:20] # First 20 columns | |
| rows = len(df) | |
| data_summary = f""" | |
| ## 📊 YOUR DATA SUMMARY | |
| - Domain: {detected_domain} | |
| - Records: {rows:,} rows | |
| - Columns: {', '.join(cols)} | |
| - Metrics (numbers you can analyze): {', '.join(available_metrics) if available_metrics else 'Auto-detected from your data'} | |
| - Dimensions (categories for grouping): {', '.join(available_dimensions) if available_dimensions else 'Auto-detected from your data'} | |
| """ | |
| print(f" Chat domain detected: {detected_domain}, metrics: {available_metrics}, dims: {available_dimensions}") | |
| except Exception as schema_err: | |
| print(f"Schema detection error (non-critical): {schema_err}") | |
| # ===================================================================== | |
| # UNIVERSAL AI ANALYST PROMPT - Works for ANY domain | |
| # ===================================================================== | |
| # Build user memory section if we have stored info about the user | |
| user_memory_section = "" | |
| if user_name: | |
| user_memory_section = f""" | |
| ═══════════════════════════════════════════════════════════════ | |
| USER MEMORY (remember this!) | |
| ═══════════════════════════════════════════════════════════════ | |
| User's Name: {user_name} | |
| {user_context if user_context else ""} | |
| IMPORTANT: Address the user by their name ({user_name}) when appropriate. You remember them from previous conversations. | |
| """ | |
| system_prompt = f"""You are an AI Data Analyst. Your job is to answer questions ONLY from the user's uploaded data. | |
| {user_memory_section} | |
| ═══════════════════════════════════════════════════════════════ | |
| DATA SUMMARY | |
| ═══════════════════════════════════════════════════════════════ | |
| Domain: {detected_domain} | |
| {data_summary} | |
| ═══════════════════════════════════════════════════════════════ | |
| USER'S DATA (from uploaded files) | |
| ═══════════════════════════════════════════════════════════════ | |
| {data_context} | |
| ═══════════════════════════════════════════════════════════════ | |
| QUERY UNDERSTANDING FLOW | |
| ═══════════════════════════════════════════════════════════════ | |
| When user asks a question, follow these EXACT steps: | |
| STEP 1: UNDERSTAND THE QUERY | |
| - What is the user asking for? (sum, count, average, list, comparison, etc.) | |
| - Which column/field are they asking about? | |
| - Are there any filters or conditions? | |
| STEP 2: FIND THE DATA | |
| - Look in "USER'S DATA" section above | |
| - Find the exact column/field mentioned | |
| - If column doesn't exist, say "Column [X] not found in your data" | |
| STEP 3: CALCULATE/EXTRACT | |
| - Perform the requested operation using ONLY values from the data above | |
| - If summing: add up the values shown | |
| - If counting: count the records | |
| - If comparing: extract values to compare | |
| STEP 4: RESPOND | |
| - Give a direct, precise answer with exact numbers from the data | |
| - Use a table for detailed breakdowns | |
| - If data not available, list what IS available | |
| ═══════════════════════════════════════════════════════════════ | |
| EXAMPLES | |
| ═══════════════════════════════════════════════════════════════ | |
| User asks: "What is total salary?" | |
| → Find: Salary column in data | |
| → Calculate: Sum of all Salary values | |
| → Respond: "**Total Salary: [sum from data]**" | |
| User asks: "Show departments" | |
| → Find: Department column in data | |
| → Extract: Unique department names | |
| → Respond: List of departments from data | |
| User asks: "How many students?" | |
| → Find: Student records in data | |
| → Count: Number of records | |
| → Respond: "**Total Students: [count from data]**" | |
| ═══════════════════════════════════════════════════════════════ | |
| STRICT RULES | |
| ═══════════════════════════════════════════════════════════════ | |
| 1. ONLY use data from "USER'S DATA" section - nothing else | |
| 2. If information is NOT in the data, say: "This is not in your uploaded data. Available data: [list columns]" | |
| 3. NEVER invent, estimate, or use outside knowledge | |
| 4. Answer the EXACT question asked - no extra information | |
| 5. Use {currency_symbol} only for monetary values (salary, price, revenue) | |
| CONVERSATION HISTORY: | |
| {conversation_context if conversation_context else "New conversation"} | |
| """ | |
| # Step 4: Call OpenRouter API | |
| api_key = os.getenv("OPENROUTER_API_KEY") | |
| print(f" OpenRouter API key present: {bool(api_key)}, length: {len(api_key) if api_key else 0}") | |
| if not api_key: | |
| return ( | |
| "⚠️ **OpenRouter API key not configured.**\n\n" | |
| "To use AI models, add this to your `.env` file:\n" | |
| "```\nOPENROUTER_API_KEY=your_key_here\n```\n\n" | |
| "Get a free key at: https://openrouter.ai/", | |
| ["Configuration Required"] | |
| ) | |
| model_id = AI_MODELS.get(model_key, AI_MODELS.get('deepseek', 'deepseek/deepseek-chat')) | |
| print(f" Using model: {model_key} -> {model_id}") | |
| payload = { | |
| "model": model_id, | |
| "messages": [ | |
| {"role": "system", "content": system_prompt}, | |
| {"role": "user", "content": query} | |
| ], | |
| "max_tokens": 4096, | |
| "temperature": 0.3, # Low temperature for accuracy | |
| } | |
| headers = { | |
| "Authorization": f"Bearer {api_key}", | |
| "Content-Type": "application/json", | |
| "HTTP-Referer": "https://ai-business-analyst.app", | |
| "X-Title": "AI Business Analyst" | |
| } | |
| async with aiohttp.ClientSession() as session: | |
| async with session.post( | |
| "https://openrouter.ai/api/v1/chat/completions", | |
| json=payload, | |
| headers=headers, | |
| timeout=aiohttp.ClientTimeout(total=60) | |
| ) as response: | |
| if response.status == 200: | |
| data = await response.json() | |
| ai_response = data["choices"][0]["message"]["content"] | |
| # Add model attribution | |
| model_name = model_key.replace('-', ' ').title() | |
| sources = rag_sources + graph_sources + [f"AI: {model_name}"] | |
| # Check if user requested a visualization and add Plotly chart | |
| # BUT SKIP if this is an explanation query (don't generate new chart) | |
| is_explanation_query = any(kw in query.lower() for kw in [ | |
| 'explain this', 'explain the chart', 'what does this chart', | |
| 'describe the chart', 'what does this show', 'explain above', | |
| 'interpret this', 'what is this chart', 'explain what', | |
| 'chart shows', 'shows in one', 'this chart shows', | |
| 'tell me about this chart', 'what does the chart', | |
| 'in one line', 'in one sentence', 'one sentence', | |
| 'what does it show', 'what is shown' | |
| ]) | |
| chart_type = detect_chart_request(query) if not is_explanation_query else None | |
| # CRITICAL: Skip backup chart if LLM already included one in response | |
| has_llm_chart = '```plotly_chart' in ai_response or '```plotly' in ai_response | |
| if chart_type and CHARTS_AVAILABLE and not is_explanation_query and not has_llm_chart: | |
| try: | |
| df = get_user_data(user_id) | |
| if df is not None and not df.empty: | |
| # ========================================================= | |
| # SMART CHART: LLM-driven chart with ALL types supported | |
| # Pie, Donut, Violin, Radar, Treemap, Combo, etc. | |
| # ========================================================= | |
| currency_symbol, _ = get_user_currency(user_id) | |
| chart_json = None | |
| # Check if they specifically want an image | |
| from agents.image_agent import generate_image_dashboard | |
| img_keywords = ['img', 'image', 'picture'] | |
| wants_image = any(kw in query.lower() for kw in img_keywords) | |
| if wants_image: | |
| # Generate PNG markdown | |
| img_markdown = generate_image_dashboard(query, df) | |
| if img_markdown: | |
| ai_response += img_markdown | |
| sources.append("Image Dashboard") | |
| else: | |
| if SMART_CHART_AVAILABLE: | |
| try: | |
| # smart_chart now returns a markdown string with the plotly block | |
| smart_markdown = smart_chart(query=query, df=df) | |
| if smart_markdown: | |
| ai_response += smart_markdown | |
| sources.append("Interactive Chart") | |
| # Create a dummy chart_json so memory context gets saved | |
| chart_json = {"layout": {"title": "Smart Chart Visualization"}} | |
| except Exception as smart_err: | |
| print(f"[CHAT] smart_chart error: {smart_err}") | |
| chart_json = None | |
| # Fallback to legacy if smart_chart failed or wasn't used | |
| if not chart_json: | |
| print(f"[CHAT] Falling back to generate_query_aware_chart") | |
| chart_json = generate_query_aware_chart(df, query) | |
| # Append Plotly chart to response for legacy fallback | |
| if chart_json and 'error' not in chart_json: | |
| import json as json_lib | |
| chart_block = f"\n\n```plotly_chart\n{json_lib.dumps(chart_json)}\n```" | |
| ai_response += chart_block | |
| sources.append("Interactive Chart") | |
| # ========================================================= | |
| # STORE CHART CONTEXT for follow-up explanation | |
| # ========================================================= | |
| if MEMORY_AVAILABLE and _chart_memory: | |
| try: | |
| # For Plotly charts | |
| if chart_json: | |
| chart_title = chart_json.get('layout', {}).get('title', {}) | |
| if isinstance(chart_title, dict): | |
| chart_title = chart_title.get('text', 'Chart') | |
| else: | |
| chart_title = 'Image Dashboard' | |
| # Determine chart type description | |
| chart_descriptions = { | |
| 'trend': 'Monthly Revenue Trend showing revenue over time', | |
| 'bar': 'Revenue by Customer/Product breakdown', | |
| 'pie': 'Revenue distribution by category', | |
| 'prediction': 'Revenue prediction with forecast' | |
| } | |
| chart_desc = chart_descriptions.get(chart_type, 'Data visualization') | |
| _chart_memory.set_last_chart(user_id, { | |
| "type": chart_type, | |
| "title": str(chart_title), | |
| "data_summary": chart_desc, | |
| "timestamp": datetime.now().isoformat() | |
| }) | |
| print(f"[MEMORY] Stored chart context: {chart_type} - {chart_title}") | |
| except Exception as mem_err: | |
| print(f"[MEMORY] Chart context storage failed: {mem_err}") | |
| else: | |
| print(f"All chart types failed: {chart_json}") | |
| except Exception as chart_error: | |
| print(f"Chart generation error: {chart_error}") | |
| import traceback | |
| traceback.print_exc() | |
| # Continue without chart | |
| # Clean up response (strip code, tables, LaTeX) | |
| ai_response = clean_ai_response(ai_response) | |
| # Evaluate answer grounding (optional confidence check) | |
| if RAG_ENHANCEMENTS and data_context: | |
| try: | |
| eval_result = evaluate_answer(ai_response, data_context, query) | |
| if eval_result.get("warning") and eval_result.get("confidence", 100) < 50: | |
| ai_response += f"\n\n---\n{eval_result['warning']}" | |
| print(f" Evaluation: {eval_result['confidence']}% confidence") | |
| except Exception as eval_err: | |
| print(f"Evaluator error (non-critical): {eval_err}") | |
| return ai_response, sources | |
| else: | |
| error = await response.text() | |
| print(f"OpenRouter API error: {response.status} - {error}") | |
| return ( | |
| f"⚠️ **Model Error ({response.status})**\n\n" | |
| f"The model `{model_id}` is not available.\n\n" | |
| "**Please try:** Select a different AI model from the dropdown.\n\n" | |
| "**Working models:** DeepSeek Chat, Mistral 7B, Llama 70B", | |
| ["Model Error"] | |
| ) | |
| except Exception as e: | |
| print(f"AI model error: {e}") | |
| traceback.print_exc() | |
| return ( | |
| f"⚠️ AI model unavailable. Here's your data:\n\n{data_context[:3000]}", | |
| ["Fallback Mode"] | |
| ) | |
| # ============================================================================ | |
| # STREAMING RESPONSE - Real-time word-by-word like ChatGPT | |
| # ============================================================================ | |
| async def stream_openrouter_response( | |
| prompt: str, | |
| model_id: str, | |
| api_key: str, | |
| max_tokens: int = 2000 | |
| ): | |
| """ | |
| Stream response from the AI using litellm. | |
| Yields SSE-formatted chunks for real-time display. | |
| """ | |
| import json as json_module | |
| import os | |
| # We will use litellm for streaming to support ALL configured providers (Nvidia, Groq, etc) | |
| try: | |
| from core.llm import LITELLM_AVAILABLE, litellm | |
| if LITELLM_AVAILABLE: | |
| # Check configured keys to determine the optimal model | |
| nv_key = os.environ.get("NVIDIA_API_KEY") | |
| groq_key = os.environ.get("GROQ_API_KEY") | |
| # Default to OpenRouter if passed | |
| model_to_use = model_id | |
| api_base = None | |
| api_key_to_use = api_key | |
| # Auto-route to Nvidia if available | |
| if nv_key: | |
| model_to_use = "openai/meta/llama-3.1-70b-instruct" | |
| api_base = "https://integrate.api.nvidia.com/v1" | |
| api_key_to_use = nv_key | |
| # Fallback to Groq | |
| elif groq_key: | |
| model_to_use = "groq/llama3-70b-8192" | |
| api_base = None | |
| api_key_to_use = groq_key | |
| if api_key_to_use: | |
| # Use litellm async generator | |
| response = await litellm.acompletion( | |
| model=model_to_use, | |
| messages=[{"role": "user", "content": prompt}], | |
| max_tokens=max_tokens, | |
| temperature=0.3, | |
| stream=True, | |
| api_key=api_key_to_use, | |
| api_base=api_base | |
| ) | |
| async for chunk in response: | |
| if chunk.choices and len(chunk.choices) > 0: | |
| content = chunk.choices[0].delta.content or "" | |
| if content: | |
| yield f"data: {json_module.dumps({'content': content})}\n\n" | |
| yield "data: [DONE]\n\n" | |
| return | |
| else: | |
| yield "data: {\"error\": \"No API key configured for streaming\"}\n\n" | |
| return | |
| else: | |
| yield "data: {\"error\": \"LiteLLM not available\"}\n\n" | |
| return | |
| except Exception as e: | |
| yield f"data: {{\"error\": \"{str(e)[:100]}\"}}\n\n" | |
| async def stream_message( | |
| request: StreamingChatRequest, | |
| x_user_id: Optional[str] = Header(None, alias="X-User-ID"), | |
| authorization: Optional[str] = Header(None, alias="Authorization") | |
| ): | |
| """ | |
| Stream chat response in real-time (like ChatGPT). | |
| Returns Server-Sent Events (SSE) with chunks of text. | |
| Frontend should use EventSource or fetch with stream reader. | |
| """ | |
| import os | |
| # Get user_id | |
| user_id = request.userId or request.user_id or x_user_id | |
| if not user_id: | |
| user_id = "default_user" | |
| query = request.message.strip() | |
| # ⚡ FAST-PATH: Instant responses for greetings/casual queries | |
| query_lower = query.lower() | |
| instant_greetings = { | |
| 'hi': "Hi! 👋 How can I help you analyze your data today?", | |
| 'hii': "Hi! 👋 How can I help you analyze your data today?", | |
| 'hiii': "Hi! 👋 How can I help you analyze your data today?", | |
| 'hello': "Hello! 👋 I'm ready to help with your data analysis. What would you like to know?", | |
| 'hey': "Hey! 👋 What data questions can I answer for you?", | |
| 'howdy': "Howdy! 🤠 Ready to dive into your data. What's on your mind?", | |
| 'yo': "Yo! 👋 Let's analyze some data. What would you like to know?", | |
| 'hola': "¡Hola! 👋 Ready to help with your data analysis!", | |
| 'good morning': "Good morning! ☀️ Ready to help with your data analysis today.", | |
| 'good afternoon': "Good afternoon! 🌤️ How can I assist with your data?", | |
| 'good evening': "Good evening! 🌙 What data insights can I help you find?", | |
| 'thanks': "You're welcome! 😊 Let me know if you need anything else.", | |
| 'thank you': "You're welcome! 😊 Happy to help. Any other questions?", | |
| 'thx': "You're welcome! 😊 Need anything else?", | |
| } | |
| if query_lower in instant_greetings: | |
| import json | |
| async def fast_stream(): | |
| yield f"data: {json.dumps({'content': instant_greetings[query_lower]})}\n\n" | |
| yield "data: [DONE]\n\n" | |
| return StreamingResponse(fast_stream(), media_type="text/event-stream") | |
| # Get RAG context | |
| rag_context, sources = rag_search(user_id, query, k=5) | |
| # Get currency | |
| try: | |
| currency_symbol, _ = get_user_currency(user_id) | |
| except: | |
| currency_symbol = "$" | |
| # Build prompt with context | |
| prompt = f"""You are an AI Data Analyst. Answer based on the user's data. | |
| ## USER'S DATA: | |
| {rag_context[:4000]} | |
| ## RULES: | |
| 1. Use ONLY data from USER'S DATA section | |
| 2. Use {currency_symbol} for currency | |
| 3. Be direct and accurate | |
| 4. Never make up numbers | |
| ## QUESTION: | |
| {query} | |
| ## YOUR RESPONSE:""" | |
| # Get API key and model (allow multiple sources) | |
| api_key = os.getenv("OPENROUTER_API_KEY") or os.getenv("NVIDIA_API_KEY") or os.getenv("GROQ_API_KEY") | |
| if not api_key: | |
| async def error_stream(): | |
| yield "data: {\"error\": \"API key not configured\"}\n\n" | |
| return StreamingResponse(error_stream(), media_type="text/event-stream") | |
| # Respect the client's selected provider; a forced unavailable provider | |
| # made streaming Analyst mode look like it was under maintenance. | |
| effective_model = request.model or "deepseek" | |
| try: | |
| from core.model_config import get_model_api_id | |
| model_id = get_model_api_id(effective_model, fallback="deepseek/deepseek-chat") | |
| except ImportError: | |
| model_id = "deepseek/deepseek-chat" | |
| # The pure LLM-driven visualization task | |
| import asyncio | |
| async def chart_augmented_stream(): | |
| import json | |
| from agents.smart_chart import should_visualize, generate_smart_chart | |
| from agents.image_agent import generate_image_dashboard | |
| from api.v1.endpoints.charts import get_user_data | |
| # 1. Start autonomous chart generation in a background thread if needed | |
| chart_task = None | |
| # 🧠 ORCHESTRATOR AI ROUTING FOR VISUALS | |
| from agents.query_router import route_query | |
| orch_decision = route_query(query) | |
| orch_mode = orch_decision.get("route", "chat") | |
| if orch_mode in ["chart", "image"]: | |
| df = get_user_data(user_id) | |
| if df is not None and not df.empty: | |
| wants_image = (orch_mode == "image") | |
| if wants_image: | |
| chart_task = asyncio.create_task(asyncio.to_thread(generate_image_dashboard, query, df)) | |
| else: | |
| chart_task = asyncio.create_task(asyncio.to_thread(generate_smart_chart, query, df)) | |
| # 2. Stream the text from OpenRouter natively | |
| has_yielded_chart = False | |
| async for chunk in stream_openrouter_response(prompt, model_id, api_key): | |
| if chunk.strip() == "data: [DONE]": | |
| # Wait for chart generation before finishing | |
| if chart_task and not has_yielded_chart: | |
| try: | |
| chart_json_str = await chart_task | |
| if chart_json_str: | |
| # Stream the chart block just like normal text | |
| yield f"data: {json.dumps({'content': chart_json_str})}\n\n" | |
| except Exception as e: | |
| print(f" [VIZ] Autonomous chart failed in stream: {e}") | |
| yield "data: [DONE]\n\n" | |
| return | |
| yield chunk | |
| # If the LLM itself generated a chart, skip our autonomous agent | |
| if "plotly_chart" in chunk: | |
| has_yielded_chart = True | |
| return StreamingResponse( | |
| chart_augmented_stream(), | |
| media_type="text/event-stream", | |
| headers={ | |
| "Cache-Control": "no-cache", | |
| "Connection": "keep-alive", | |
| "X-Accel-Buffering": "no" | |
| } | |
| ) | |
| async def send_message( | |
| http_request: Request, # For rate limiting | |
| request: ChatRequest, | |
| x_user_id: Optional[str] = Header(None, alias="X-User-ID"), | |
| authorization: Optional[str] = Header(None, alias="Authorization") | |
| ): | |
| """Send message - REAL RAG/Graph response with file comparison and memory | |
| SECURED: Uses JWT-based user identification for data isolation | |
| PROTECTED: Rate limiting and prompt injection detection | |
| """ | |
| try: | |
| # 🔒 SECURITY: Rate limiting check | |
| if RATE_LIMITER_AVAILABLE: | |
| # Extract user for rate limiting (use IP if not authenticated yet) | |
| temp_user_id = x_user_id or "anonymous" | |
| await check_rate_limit(http_request, "chat", temp_user_id) | |
| # 🔐 ENTERPRISE AUTH: Get user_id from verified JWT token | |
| # Priority: 1. JWT token (cryptographically verified), 2. X-User-ID header | |
| # NEVER trust request body for user_id (can be manipulated) | |
| user_id = None | |
| # Try to extract from JWT token first (MOST SECURE) | |
| if authorization and authorization.startswith("Bearer "): | |
| try: | |
| # Use core.auth module for proper JWT validation | |
| from core.auth import decode_jwt_token | |
| token = authorization.split(" ")[1] | |
| payload = decode_jwt_token(token) | |
| user_id = payload.get("sub") # Subject is user ID | |
| print(f" Authenticated user from JWT: {user_id}") | |
| except Exception as e: | |
| print(f" JWT decode error: {e}") | |
| # Fallback to legacy decode | |
| try: | |
| from database.auth import decode_jwt | |
| payload = decode_jwt(token) | |
| user_id = payload.get("sub") | |
| except: | |
| pass | |
| # Fallback to X-User-ID header (for authenticated requests where JWT is also sent) | |
| if not user_id and x_user_id and x_user_id not in ["null", "undefined", ""]: | |
| user_id = x_user_id | |
| print(f" User from X-User-ID header: {user_id}") | |
| # Generate guest ID if no authentication (DO NOT use request body user_id) | |
| if not user_id: | |
| import hashlib | |
| ip = http_request.client.host if http_request.client else "unknown" | |
| ua = http_request.headers.get("User-Agent", "unknown")[:100] | |
| fingerprint = hashlib.sha256(f"{ip}:{ua}".encode()).hexdigest()[:12] | |
| user_id = f"guest_{fingerprint}" | |
| print(f" Generated guest user ID: {user_id}") | |
| # 🛡️ SECURITY: Mask any PII before processing | |
| query = SecurityVault.mask_pii(request.message) | |
| mode = request.mode | |
| # Respect the client's selected provider instead of forcing Nemotron. | |
| effective_model = request.model or "deepseek" | |
| # Pass the requested LLM model to the backend engine | |
| if effective_model: | |
| set_requested_model(effective_model) | |
| conversation_id = request.conversationId or f"conv_{int(datetime.now().timestamp())}" | |
| compare_files = request.compareFiles | |
| # ===================================================================== | |
| # ⚡ FAST-PATH: Instant responses for greetings/casual queries | |
| # Skip ALL heavy processing (RAG, Graph, Mode engines) for simple chat | |
| # ===================================================================== | |
| query_lower = query.lower().strip() | |
| # INSTANT GREETINGS - No need for RAG/AI for these | |
| instant_greetings = { | |
| 'hi': "Hi! 👋 How can I help you analyze your data today?", | |
| 'hii': "Hi! 👋 How can I help you analyze your data today?", | |
| 'hiii': "Hi! 👋 How can I help you analyze your data today?", | |
| 'hello': "Hello! 👋 I'm ready to help with your data analysis. What would you like to know?", | |
| 'hey': "Hey! 👋 What data questions can I answer for you?", | |
| 'howdy': "Howdy! 🤠 Ready to dive into your data. What's on your mind?", | |
| 'yo': "Yo! 👋 Let's analyze some data. What would you like to know?", | |
| 'hola': "¡Hola! 👋 Ready to help with your data analysis!", | |
| 'good morning': "Good morning! ☀️ Ready to help with your data analysis today.", | |
| 'good afternoon': "Good afternoon! 🌤️ How can I assist with your data?", | |
| 'good evening': "Good evening! 🌙 What data insights can I help you find?", | |
| 'thanks': "You're welcome! 😊 Let me know if you need anything else.", | |
| 'thank you': "You're welcome! 😊 Happy to help. Any other questions?", | |
| 'thx': "You're welcome! 😊 Need anything else?", | |
| 'ty': "You're welcome! 😊", | |
| 'bye': "Goodbye! 👋 Come back anytime you need data insights!", | |
| 'goodbye': "Goodbye! 👋 Have a great day!", | |
| 'ok': "Great! 👍 What would you like to analyze next?", | |
| 'okay': "Great! 👍 What data question can I help with?", | |
| 'cool': "Glad to help! 😊 What else would you like to know?", | |
| 'nice': "Thanks! 😊 Ready for more data questions!", | |
| 'awesome': "Thanks! 🎉 What else can I help you with?", | |
| 'great': "Glad you think so! 😊 Any other questions?", | |
| } | |
| # Only use fast-path for SHORT messages (5 words or less) that are EXACT matches | |
| word_count = len(query_lower.split()) | |
| if word_count <= 3 and query_lower in instant_greetings: | |
| instant_response = instant_greetings[query_lower] | |
| print(f" FAST-PATH: Instant greeting response for '{query}'") | |
| return ChatResponse( | |
| message=instant_response, | |
| mode="chat", | |
| sources=["Instant Response"], | |
| conversationId=conversation_id, | |
| timestamp=datetime.now().isoformat() | |
| ) | |
| # FAST conversational patterns (EXACT match only for short messages) | |
| fast_patterns = { | |
| 'how are you': "I'm doing great, thanks for asking! 😊 How can I help you with your data today?", | |
| "how's it going": "Going well! 😊 Ready to analyze your data. What would you like to know?", | |
| "what's up": "Ready to help! 💪 What data insights are you looking for?", | |
| 'who are you': "I'm your AI Data Analyst! 🤖 I can analyze your uploaded data, answer questions, create charts, and provide insights. What would you like to know?", | |
| 'what are you': "I'm DataVision AI - your personal data analyst! 📊 Upload data and I'll help you find insights, create visualizations, and answer questions about your data.", | |
| 'what can you do': "I can help you:\n• 📊 Analyze your data and find patterns\n• 📈 Create charts and visualizations\n• 🔍 Answer questions about your data\n• 🎯 Make predictions with ML models\n\nUpload some data and ask me anything!", | |
| "what's your name": "I'm DataVision AI, your personal data analyst! 🤖 How can I help you today?", | |
| 'your name': "I'm DataVision AI! 🤖 Ready to help with your data analysis.", | |
| } | |
| # Only match if query is SHORT (5 words or less) AND matches pattern exactly | |
| if word_count <= 5 and query_lower in fast_patterns: | |
| print(f" FAST-PATH: Quick pattern response for '{query}'") | |
| return ChatResponse( | |
| message=fast_patterns[query_lower], | |
| mode="chat", | |
| sources=["Instant Response"], | |
| conversationId=conversation_id, | |
| timestamp=datetime.now().isoformat() | |
| ) | |
| # 🔒 SECURITY: Check for prompt injection attacks | |
| if AI_SECURITY_AVAILABLE: | |
| is_suspicious, detected_pattern = detect_prompt_injection(query) | |
| if is_suspicious: | |
| print(f" SECURITY: Potential prompt injection detected from user {user_id}: {detected_pattern}") | |
| # Log but don't block - let AI security filter handle it in llm.py | |
| # This provides defense in depth | |
| # ===================================================================== | |
| # USER SELECTED MODE/MODEL - Respect user's choice | |
| # All modes can generate BOTH text AND charts based on query | |
| # ===================================================================== | |
| has_image = request.attachedFiles and len(request.attachedFiles) > 0 | |
| # Detect if user's query wants a visualization (for ANY mode/model) | |
| generate_chart = False | |
| chart_type = None | |
| if ADVANCED_RAG: | |
| try: | |
| routing = get_routing_decision(query, has_image=has_image) | |
| # Detect if we should generate a chart (works for ALL modes) | |
| generate_chart = routing.get('generate_chart', False) | |
| chart_type = routing.get('chart_type') | |
| query_intent = routing.get('intent', 'lookup') | |
| print(f" User selected: mode={mode}, intent={query_intent}") | |
| if generate_chart: | |
| print(f" Will generate {chart_type} chart with response") | |
| except Exception as e: | |
| print(f" Query analysis error: {e}") | |
| # Vision mode enhancement when image attached (user can still use other modes) | |
| if has_image: | |
| print(f" Image attached - mode={mode} will process image") | |
| # Load conversation history for context (user-specific) | |
| history = load_conversation(user_id, conversation_id) | |
| # 🧠 USE FRONTEND-PROVIDED CONVERSATION HISTORY FOR BETTER MEMORY | |
| # This is more reliable than server-side storage for session context | |
| # PERFORMANCE: Skip file I/O if frontend provides history | |
| conversation_context = "" | |
| if request.conversationHistory and len(request.conversationHistory) > 0: | |
| # Fast path: use frontend history directly | |
| conversation_context = "\n\n## Conversation History (for context):\n" | |
| for msg in request.conversationHistory[-5:]: # Last 5 messages (reduced for speed) | |
| role = msg.get('role', 'user').upper() | |
| content = msg.get('content', '')[:300] # Truncate for speed | |
| conversation_context += f"{role}: {content}\n" | |
| else: | |
| # Fallback to server-side history | |
| conversation_context = "" | |
| if history: | |
| recent_messages = history[-10:] # Last 10 messages | |
| conversation_context = "\n\n## Previous Conversation:\n" | |
| for msg in recent_messages: | |
| conversation_context += f"{msg.role.upper()}: {msg.content[:500]}\n" | |
| # 🧠 ENHANCED QDRANT MEMORY: Semantic search for past relevant conversations | |
| if VECTOR_MEMORY_AVAILABLE and vector_store: | |
| try: | |
| semantic_results = vector_store.search_chat_history(user_id=user_id, query=query, limit=3) | |
| if semantic_results: | |
| conversation_context += "\n\n## Long-Term Semantic Memory (Past relevant chats):\n" | |
| for res in semantic_results: | |
| payload = res.get('payload', {}) | |
| role = payload.get('role', 'unknown').upper() | |
| content = payload.get('content', '') | |
| # Only include if not already in recent history to avoid duplication | |
| if content and content not in conversation_context: | |
| conversation_context += f"Past {role}: {content[:300]}\n" | |
| except Exception as e: | |
| print(f" Vector memory retrieval failed: {e}") | |
| # Get user paths for file and memory access | |
| paths = get_user_paths(user_id) | |
| # Build last assistant response for follow-up context | |
| last_assistant_response = "" | |
| # Get last assistant response from history if available | |
| if history: | |
| for msg in reversed(history): | |
| if msg.role == 'assistant': | |
| last_assistant_response = msg.content[:2000] | |
| break | |
| # ===================================================================== | |
| # CRITICAL: If no history but follow-up query, try to get last response | |
| # ===================================================================== | |
| if not last_assistant_response: | |
| try: | |
| # Try to get from chat_history.json as fallback | |
| chat_history_file = paths["memory"] / "chat_history.json" | |
| if chat_history_file.exists(): | |
| with open(chat_history_file, 'r') as f: | |
| chat_history = json.load(f) | |
| # Find last assistant message | |
| for msg in reversed(chat_history): | |
| if msg.get('role') == 'assistant': | |
| last_assistant_response = msg.get('content', '')[:3000] # Increased for context | |
| conversation_context = f"\n\n## Last AI Response:\n{last_assistant_response}\n" | |
| print(f" Found previous response: {len(last_assistant_response)} chars") | |
| break | |
| except Exception as e: | |
| print(f" Could not load chat history: {e}") | |
| # ===================================================================== | |
| # $500K ENTERPRISE MEMORY SYSTEM - ChatGPT-Level Persistent Memory | |
| # ===================================================================== | |
| try: | |
| from core.memory import process_personal_info, get_user_name | |
| query_lower = query.lower().strip() | |
| # STEP 1: MEMORY WRITE - Check if user is providing their name | |
| personal_saved = process_personal_info(user_id, query) | |
| if personal_saved: | |
| print(f" MEMORY WRITE: Saved personal information for user {user_id}") | |
| # STEP 2: MEMORY READ - Get stored name | |
| stored_name = get_user_name(user_id) | |
| print(f" MEMORY READ: Stored name = {stored_name}") | |
| # CASE 1: User PROVIDING name (must check BEFORE asking) | |
| # Patterns: "my name is X", "i am X", "call me X" | |
| name_provide_patterns = ['my name is', 'i am ', "i'm ", 'call me', 'you can call me', 'name is '] | |
| is_providing_name = any(pattern in query_lower for pattern in name_provide_patterns) | |
| if is_providing_name and personal_saved and stored_name: | |
| history.append(Message(role="user", content=query, timestamp=datetime.now().isoformat())) | |
| response_text = ( | |
| f"Nice to meet you, **{stored_name}**!\n\n" | |
| f"I've saved your name to my memory. I'll remember you across all our conversations.\n\n" | |
| f"**Ready to analyze your data.** What would you like to know?" | |
| ) | |
| assistant_msg = Message(role="assistant", content=response_text, timestamp=datetime.now().isoformat()) | |
| history.append(assistant_msg) | |
| save_conversation(user_id, conversation_id, history) | |
| return ChatResponse( | |
| message=response_text, | |
| mode="memory", | |
| sources=["Persistent Memory"], | |
| conversationId=conversation_id, | |
| timestamp=datetime.now().isoformat() | |
| ) | |
| # CASE 2: User ASKING about their name | |
| # Patterns: "what is my name", "who am i", "tell me my name" | |
| name_ask_patterns = ['what is my name', 'what\'s my name', 'who am i', 'tell me my name', | |
| 'do you know my name', 'remember my name', 'do you remember me'] | |
| is_asking_name = any(pattern in query_lower for pattern in name_ask_patterns) | |
| if is_asking_name: | |
| history.append(Message(role="user", content=query, timestamp=datetime.now().isoformat())) | |
| if stored_name: | |
| response_text = ( | |
| f"Your name is **{stored_name}**.\n\n" | |
| f"I remember you from our previous conversations.\n\n" | |
| f"How can I help you with your data today?" | |
| ) | |
| else: | |
| response_text = ( | |
| "I don't have your name saved yet.\n\n" | |
| "Please introduce yourself (e.g., 'My name is Naveen') and I'll remember you for all future conversations." | |
| ) | |
| assistant_msg = Message(role="assistant", content=response_text, timestamp=datetime.now().isoformat()) | |
| history.append(assistant_msg) | |
| save_conversation(user_id, conversation_id, history) | |
| return ChatResponse( | |
| message=response_text, | |
| mode="memory", | |
| sources=["Persistent Memory"], | |
| conversationId=conversation_id, | |
| timestamp=datetime.now().isoformat() | |
| ) | |
| except Exception as mem_err: | |
| print(f"Memory processing error: {mem_err}") | |
| import traceback | |
| traceback.print_exc() | |
| user_msg = Message( | |
| role="user", | |
| content=query, | |
| timestamp=datetime.now().isoformat() | |
| ) | |
| history.append(user_msg) | |
| # Enterprise 4-mode routing | |
| has_image = bool(request.attachedFiles and any(f.get('type', '').startswith('image/') for f in request.attachedFiles)) | |
| print(f" Attached files: {len(request.attachedFiles) if request.attachedFiles else 0}") | |
| print(f" Has image: {has_image}") | |
| print(f" Query: {query[:50]}...") | |
| print(f" Requested mode: {mode}") | |
| # MCP status - which MCPs are enabled | |
| enabled_mcps = request.enabledMcps or { | |
| 'data_cleaner': True, | |
| 'vectorizer': True, | |
| 'graph_builder': True, | |
| 'sql_executor': True, | |
| 'vision_ocr': True, | |
| } | |
| # MCP status logged only in debug mode for performance | |
| # enabled_count = sum(1 for v in enabled_mcps.values() if v) | |
| # print(f" MCP Servers: {enabled_count}/5 enabled") | |
| query_lower = query.lower().strip() | |
| # ===================================================================== | |
| # EMPTY/MINIMAL QUERY - ChatGPT-style: Ask what user wants | |
| # Instead of giving full analysis for empty queries, be conversational | |
| # ===================================================================== | |
| if len(query_lower) < 3 or query_lower in ['', '.', '..', '?', '!', 'ok', 'go']: | |
| print(f" EMPTY/MINIMAL QUERY DETECTED: '{query}' Asking what user wants") | |
| # Get data summary for helpful prompt | |
| try: | |
| from api.v1.endpoints.charts import get_user_data | |
| df = get_user_data(user_id) | |
| if df is not None and not df.empty: | |
| cols = [c for c in df.columns if not c.startswith('_')][:8] | |
| prompt_response = f"""👋 **I'm ready to help!** | |
| I have your data loaded with columns: **{', '.join(cols)}** | |
| **What would you like to know?** For example: | |
| - "What's the total revenue?" | |
| - "Show top 5 customers" | |
| - "Average salary by department" | |
| Just ask your question! 💬""" | |
| else: | |
| prompt_response = """👋 **I'm ready to help!** | |
| Please upload a data file first, then ask me any question about your data. | |
| **What would you like to analyze?** 📊""" | |
| except: | |
| prompt_response = "👋 What would you like to know about your data?" | |
| # Save and return | |
| assistant_msg = Message(role="assistant", content=prompt_response, timestamp=datetime.now().isoformat()) | |
| history.append(assistant_msg) | |
| save_conversation(user_id, conversation_id, history) | |
| return ChatResponse( | |
| message=prompt_response, | |
| mode="chat", | |
| sources=["Assistant"], | |
| conversationId=conversation_id, | |
| timestamp=datetime.now().isoformat() | |
| ) | |
| # BUSINESS QUERY KEYWORDS - These should NEVER be treated as personal chat | |
| business_keywords = [ | |
| 'product', 'customer', 'revenue', 'sales', 'invoice', 'amount', 'total', | |
| 'lowest', 'highest', 'top', 'bottom', 'best', 'worst', 'minimum', 'maximum', | |
| 'performance', 'trend', 'analysis', 'analyze', 'data', 'report', 'show', | |
| 'list', 'display', 'which', 'what is the', 'how much', 'how many', | |
| 'average', 'sum', 'count', 'compare', 'difference', 'graph', 'chart', | |
| 'month', 'year', 'date', 'period', 'quarterly', 'weekly', 'daily' | |
| ] | |
| # Check if this is a business/data query | |
| is_business_query = any(kw in query_lower for kw in business_keywords) | |
| # PERSONAL/GREETING keywords - Only these should go to chat mode | |
| personal_keywords = [ | |
| 'hello', 'hi', 'hey', 'hii', 'hiii', 'howdy', | |
| 'how are you', 'how r u', "how's it going", "what's up", | |
| 'thank you', 'thanks', 'thx', 'ty', | |
| 'bye', 'goodbye', 'see you', 'cya', | |
| 'good morning', 'good evening', 'good night', 'good afternoon', | |
| 'who are you', 'what are you', 'what can you do', 'help me', | |
| 'your name', 'introduce yourself', "what's your name", | |
| 'nice to meet', 'pleased to meet', | |
| 'my name', 'tell me my name', 'what is my name', "what's my name", | |
| 'remember me', 'do you remember', 'who am i' | |
| ] | |
| # EXACT MATCH for very short greetings - these ALWAYS go to chat | |
| exact_greetings = ['hi', 'hello', 'hey', 'hii', 'hiii', 'howdy', 'yo', 'hola', | |
| 'good morning', 'good afternoon', 'good evening', 'thanks', 'thank you'] | |
| is_exact_greeting = query_lower.strip() in exact_greetings | |
| # ===================================================================== | |
| # APP-SPECIFIC HELP QUERIES - How to use the application | |
| # ===================================================================== | |
| upload_help_patterns = [ | |
| 'how to upload', 'how do i upload', 'upload my data', 'uploading data', | |
| 'how to add data', 'add my data', 'import data', 'how to import', | |
| 'where to upload', 'where do i upload', 'upload file', 'upload files' | |
| ] | |
| is_upload_help = any(pattern in query_lower for pattern in upload_help_patterns) | |
| if is_upload_help: | |
| upload_response = """## 📤 How to Upload Your Data | |
| **Step 1:** Click on **"DataHub"** in the left sidebar (or navigation menu) | |
| **Step 2:** In the DataHub page, click the **"Upload"** or **"Add Files"** button | |
| **Step 3:** Select your files: | |
| - ✅ **Supported formats:** CSV, Excel (.xlsx), PDF | |
| - 📄 Drag & drop files or click to browse | |
| **Step 4:** Click **"Train"** to process your data | |
| **Step 5:** Come back here to the **Analyst Chat** and ask questions about your data! | |
| --- | |
| 💡 **Tip:** After uploading, try asking: | |
| - "What data do I have?" | |
| - "Show me revenue trends" | |
| - "Who are my top customers?" | |
| """ | |
| # Save and return immediately | |
| user_msg = Message(role="user", content=query, timestamp=datetime.now().isoformat()) | |
| assistant_msg = Message(role="assistant", content=upload_response, timestamp=datetime.now().isoformat(), sources=["App Help"]) | |
| history.append(user_msg) | |
| history.append(assistant_msg) | |
| save_conversation(user_id, conversation_id, history) | |
| return ChatResponse( | |
| message=upload_response, | |
| mode="chat", | |
| sources=["App Help"], | |
| conversationId=conversation_id, | |
| timestamp=datetime.now().isoformat() | |
| ) | |
| # ===================================================================== | |
| # VAGUE/EXPLORATION QUERIES - ChatGPT-style conversational response | |
| # Instead of giving full analysis, ask what user wants | |
| # ===================================================================== | |
| vague_exploration_patterns = [ | |
| 'tell me about my data', 'what data do i have', 'what can you tell me', | |
| 'analyze my data', 'what do you see', 'what can you do', | |
| 'what can you analyze', 'show me my data', 'what is in my data', | |
| 'help me analyze', 'analyze this', 'what do you have', | |
| 'tell me about this file', 'what is this', 'start analysis', | |
| 'give me insights', 'give me overview', 'summarize my data', | |
| 'what can i ask', 'what questions', 'what should i ask' | |
| ] | |
| is_vague_exploration = any(pattern in query_lower for pattern in vague_exploration_patterns) | |
| # Check if query is purely conversational (NOT a business query) | |
| is_short_message = len(query_lower.split()) <= 5 | |
| is_personal = any(kw in query_lower for kw in personal_keywords) and is_short_message and not is_business_query | |
| # ===================================================================== | |
| # CRITICAL: Detect follow-up queries like "explain that", "what does it mean" | |
| # These should ALWAYS use RAG with previous context, not chat mode! | |
| # ===================================================================== | |
| followup_patterns = [ | |
| 'explain that', 'explain it', 'explain this', 'explain above', | |
| 'what does that', 'what does it', 'what does this', | |
| 'tell me more', 'more about', 'elaborate', 'clarify', | |
| 'in words', 'in simple', 'in detail', | |
| 'what is that', 'what was that', 'meaning of', | |
| # Patterns for "explain in one line" type queries | |
| 'in one line', 'in one sentence', 'in two lines', 'in two sentences', | |
| 'shorter version', 'short version', 'summarize above', 'summarize that', | |
| 'one liner', 'one-liner', 'simple terms', 'simpler', | |
| 'layman terms', 'briefly', 'brief version' | |
| ] | |
| is_followup_query = any(p in query_lower for p in followup_patterns) | |
| if is_followup_query: | |
| is_personal = False # Force NOT personal | |
| is_business_query = True # Force business query | |
| print(f" FOLLOW-UP QUERY DETECTED: '{query[:30]}...' Forcing RAG with context") | |
| print(f" Is personal: {is_personal}, Is exact greeting: {is_exact_greeting}, Is business: {is_business_query}, Is followup: {is_followup_query}") | |
| # Keywords that indicate user wants data analysis, not image analysis | |
| data_query_keywords = ['predict', 'forecast', 'revenue', 'chart', 'visualization', | |
| 'customer', 'product', 'trend', 'sales', 'compare', 'analysis', | |
| 'total', 'average', 'highest', 'lowest', 'best', 'worst'] | |
| is_data_query = any(kw in query_lower for kw in data_query_keywords) | |
| # ======================================== | |
| # 🖼️ AUTO-DETECT IMAGE: Force appropriate mode when image attached | |
| # ======================================== | |
| if has_image and mode not in ['vision']: | |
| # User attached image - auto-switch to vision-capable mode | |
| # Use analyst mode which can handle both images and data | |
| original_mode = mode | |
| mode = 'analyst' # Analyst can handle images via image_context | |
| print(f" AUTO-ROUTE: Image attached with mode '{original_mode}' Switching to '{mode}' for image handling") | |
| # ======================================== | |
| # 🧠 ORCHESTRATOR AI ROUTING | |
| # ======================================== | |
| try: | |
| if mode == "auto" or mode not in ['analyst', 'deep', 'predict', 'agent', 'vision', 'rag', 'graph']: | |
| from agents.query_router import route_query | |
| orch_decision = route_query(query) | |
| orch_mode = orch_decision.get("route", mode) | |
| print(f" ORCHESTRATOR routed query to: {orch_mode} (Confidence: {orch_decision.get('confidence')}%)") | |
| # Map orchestrator routes to actual engine routes | |
| if orch_mode == "chart": | |
| mode = "analyst" # analyst mode handles charts natively | |
| elif orch_mode == "prediction": | |
| mode = "predict" | |
| elif orch_mode == "image": | |
| mode = "analyst" # image generation happens in chat response later | |
| elif orch_mode == "vision": | |
| mode = "vision" | |
| elif orch_mode == "chat": | |
| mode = "chat" | |
| elif orch_mode == "etl": | |
| mode = "etl" | |
| elif orch_mode == "ml_pipeline": | |
| mode = "ml_pipeline" | |
| else: | |
| mode = "rag" | |
| except Exception as e: | |
| print(f" Orchestrator error: {e}") | |
| # ======================================== | |
| # NEW 5 UNIQUE MODE ENGINES (Silicon Valley Powerhouses) | |
| # Each mode has its OWN power - like OpenAI's model lineup | |
| # ======================================== | |
| # Check if we should use the new mode engines | |
| NEW_MODE_ENGINES = ['analyst', 'deep', 'predict', 'agent', 'vision', 'etl', 'ml_pipeline'] # All modes unified | |
| if mode in NEW_MODE_ENGINES and MODE_ENGINES_AVAILABLE: | |
| print(f" USING NEW MODE ENGINE: {mode}") | |
| # Get DataFrame for charts/analysis (SHARED for all modes) | |
| df = None | |
| try: | |
| from api.v1.endpoints.charts import get_user_data | |
| df = get_user_data(user_id) | |
| except Exception as e: | |
| print(f" Failed to load user dataframe: {e}") | |
| # Get context from RAG | |
| try: | |
| from core.rag import rag_search | |
| context, rag_sources = rag_search(user_id, query, k=10) | |
| except Exception as e: | |
| print(f" RAG Search execution failed: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| context = "" | |
| rag_sources = [] | |
| # ========================================================= | |
| # 🖼️ CLAUDE-STYLE: Extract image context for ANY mode | |
| # Images work in all modes, not just Vision | |
| # ========================================================= | |
| image_context = "" | |
| # Convert request.attachedFiles to processed_files format | |
| processed_files = [] | |
| if request.attachedFiles: | |
| for att_file in request.attachedFiles: | |
| processed_files.append({ | |
| 'name': att_file.get('name', 'image'), | |
| 'type': att_file.get('type', 'image/png'), | |
| 'content': att_file.get('content', '') # Base64 data URL | |
| }) | |
| print(f" Prepared {len(processed_files)} files for vision processing") | |
| if processed_files: | |
| for file_info in processed_files: | |
| file_content = file_info.get('content', '') | |
| file_name = file_info.get('name', 'image') | |
| # Check if it's an image (base64 data URL) | |
| if file_content.startswith('data:image'): | |
| print(f" Found image attachment: {file_name}") | |
| try: | |
| from core.vision import analyze_image_with_groq | |
| # Quick vision analysis | |
| analysis = analyze_image_with_groq( | |
| file_content, | |
| f"Describe this image. User asks: {query}" | |
| ) | |
| if not analysis.startswith("❌"): | |
| image_context += f"\n\n## 🖼️ Image Analysis ({file_name})\n{analysis}\n" | |
| print(f" Image analyzed: {len(analysis)} chars") | |
| else: | |
| print(f" Image analysis failed: {analysis[:100]}") | |
| except Exception as e: | |
| print(f" Vision processing error: {e}") | |
| # Combine RAG context with image context | |
| full_context = context | |
| if image_context: | |
| full_context = f"{context}\n\n{image_context}" if context else image_context | |
| try: | |
| if mode == 'analyst': | |
| # 📊 ANALYST - Smart data analysis with auto RAG routing | |
| result = analyst_response_sync(user_id, query, full_context, df=df) | |
| response = result.get('answer', str(result)) if isinstance(result, dict) else str(result) | |
| sources = result.get('sources', ["Analyst Engine"]) if isinstance(result, dict) else ["Analyst Engine"] | |
| print(f" ANALYST ENGINE returned: {len(response)} chars") | |
| elif mode == 'deep': | |
| # 🧠 DEEP THINK - Chain of thought reasoning | |
| result = deepthink_response_sync(user_id, query, full_context, df=df) | |
| response = result.get('answer', str(result)) if isinstance(result, dict) else str(result) | |
| sources = result.get('sources', ["Deep Think Engine"]) if isinstance(result, dict) else ["Deep Think Engine"] | |
| print(f" DEEP THINK ENGINE returned: {len(response)} chars") | |
| elif mode == 'predict': | |
| # 🔮 PREDICT - REAL ML Predictions using trained model | |
| # Use SYNC predict_response to avoid asyncio.run() conflicts | |
| try: | |
| from core.mode_engines.predict_engine import predict_response_sync | |
| # Run SYNC prediction with real ML | |
| result = predict_response_sync(user_id, query, full_context, df) | |
| response = result.get('answer', 'Error making prediction') | |
| if result.get('ml_used'): | |
| sources = result.get('sources', ["Trained ML Model", "ML Prediction"]) | |
| else: | |
| sources = ["Predict Engine", "Data Analysis"] | |
| print(f" PREDICT ENGINE (SYNC ML): {result.get('ml_used')} - {len(response)} chars") | |
| except Exception as pred_err: | |
| print(f" ML Predict error: {pred_err}") | |
| import traceback | |
| traceback.print_exc() | |
| response = "⚠️ Error making prediction. Please try again." | |
| sources = ["Predict Engine", "Error"] | |
| elif mode == 'agent': | |
| # 🤖 AGENT - Full autonomous with web search | |
| result = agent_response_sync(user_id, query, full_context, df=df) | |
| response = result.get('answer', str(result)) if isinstance(result, dict) else str(result) | |
| sources = result.get('sources', ["Agent Engine"]) if isinstance(result, dict) else ["Agent Engine"] | |
| print(f" AGENT ENGINE returned: {len(response)} chars") | |
| elif mode == 'vision': | |
| # 👁️ VISION - Specialized image analysis | |
| result = vision_response_sync(user_id, query, full_context, df=df) | |
| response = result.get('answer', str(result)) if isinstance(result, dict) else str(result) | |
| sources = result.get('sources', ["Vision Engine"]) if isinstance(result, dict) else ["Vision Engine"] | |
| print(f" VISION ENGINE returned: {len(response)} chars") | |
| elif mode == 'etl': | |
| # 🧹 ETL - Autonomous Data Janitor | |
| try: | |
| from agents.data_janitor import generate_etl_script, safe_execute_etl | |
| import os | |
| if df is None or df.empty: | |
| response = "⚠️ I need a dataset to clean! Please upload a CSV first." | |
| sources = ["Data Janitor"] | |
| else: | |
| # 1. Generate ETL Script | |
| print("Generating ETL Script...") | |
| script_code = generate_etl_script(df, query) | |
| # 2. Execute Safely | |
| print("Executing Auto-ETL...") | |
| success, cleaned_df, msg = safe_execute_etl(df, script_code) | |
| if success: | |
| # Save to file for user | |
| script_path = os.path.join(os.getcwd(), 'user_scripts') | |
| os.makedirs(script_path, exist_ok=True) | |
| file_path = os.path.join(script_path, 'clean_data.py') | |
| with open(file_path, 'w') as f: | |
| f.write(script_code) | |
| response = f"## 🧹 Autonomous Data Janitor\n\nI have successfully analyzed your data, written a custom Python Pandas script to clean it, and executed it in memory!\n\n### 🐍 Generated ETL Script\n```python\n{script_code}\n```\n\nYour script is saved at: `{file_path}`" | |
| else: | |
| response = f"⚠️ I tried to write a cleaning script but it failed execution: {msg}\n\n```python\n{script_code}\n```" | |
| sources = ["Data Janitor Auto-ETL"] | |
| except Exception as e: | |
| response = f"⚠️ ETL Pipeline failed: {e}" | |
| sources = ["Data Janitor"] | |
| print(f" ETL Error: {e}") | |
| elif mode == 'ml_pipeline': | |
| from agents.nl_pipeline_agent import handle_nl_pipeline | |
| response_text, sources = handle_nl_pipeline(user_id, query, df=df) | |
| response = response_text | |
| # Add chart if visualization requested - SKIP for engines that add it themselves | |
| # All engines now append their own charts if df is passed | |
| # if mode != 'predict': | |
| # response = append_chart_if_needed(response, query, user_id) | |
| # Save and return | |
| history.append(Message(role="user", content=query, timestamp=datetime.now().isoformat())) | |
| history.append(Message(role="assistant", content=response, timestamp=datetime.now().isoformat(), sources=sources)) | |
| save_conversation(user_id, conversation_id, history) | |
| return ChatResponse( | |
| message=response, | |
| mode=mode, | |
| sources=sources, | |
| conversationId=conversation_id, | |
| timestamp=datetime.now().isoformat() | |
| ) | |
| except Exception as engine_error: | |
| print(f" Mode engine error: {engine_error}") | |
| import traceback | |
| traceback.print_exc() | |
| # Fall through to legacy handling | |
| # Legacy mode mapping for backward compatibility | |
| MODE_MAPPING = { | |
| 'analyst': 'rag', # Fallback if engine fails | |
| 'deep': 'agentic', # Fallback | |
| 'predict': 'prediction', # Fallback | |
| 'agent': 'agentic', # Fallback | |
| } | |
| # Apply legacy mode mapping if needed | |
| original_mode = mode | |
| if mode in MODE_MAPPING and not MODE_ENGINES_AVAILABLE: | |
| mode = MODE_MAPPING[mode] | |
| print(f" LEGACY MODE MAPPING: '{original_mode}' '{mode}'") | |
| # ======================================== | |
| # MODE ROUTING - PRIORITY ORDER IS CRITICAL | |
| # ======================================== | |
| # PRIORITY 0.5: VAGUE EXPLORATION - ChatGPT-style welcome | |
| # Returns immediately with conversational welcome instead of full analysis | |
| if is_vague_exploration: | |
| print(f" VAGUE EXPLORATION DETECTED: Returning ChatGPT-style welcome") | |
| # Get data summary for welcome | |
| try: | |
| from api.v1.endpoints.charts import get_user_data | |
| df = get_user_data(user_id) | |
| if df is not None and not df.empty: | |
| file_count = df['_source_file'].nunique() if '_source_file' in df.columns else 1 | |
| row_count = len(df) | |
| # Generate DYNAMIC examples based on actual columns | |
| numeric_cols = [c for c in df.columns if df[c].dtype in ['int64', 'float64'] and not c.startswith('_')][:3] | |
| text_cols = [c for c in df.columns if df[c].dtype == 'object' and not c.startswith('_') and df[c].nunique() < 50][:3] | |
| # Build dynamic example questions | |
| example_lines = [] | |
| if numeric_cols and text_cols: | |
| example_lines.append(f"- 📊 **Analytics:** \"What's the total {numeric_cols[0]} by {text_cols[0]}?\"") | |
| if numeric_cols: | |
| example_lines.append(f"- 📈 **Trends:** \"Show {numeric_cols[0]} trend over time\"") | |
| if text_cols: | |
| example_lines.append(f"- 🏆 **Rankings:** \"Show top 5 {text_cols[0]}\"") | |
| if numeric_cols: | |
| example_lines.append(f"- 🔮 **Predictions:** \"Forecast {numeric_cols[0]}\"") | |
| if len(text_cols) > 1: | |
| example_lines.append(f"- 📉 **Comparisons:** \"Compare {text_cols[0]} by {text_cols[1]}\"") | |
| examples_text = '\n'.join(example_lines) if example_lines else '- Ask anything about your data!' | |
| welcome_response = f"""👋 **Hi! I'm your AI Data Analyst.** | |
| I can see you've uploaded some data! Let me show you what I found: | |
| 📁 **Your Data:** | |
| - **Files:** {file_count} file(s) | |
| - **Records:** {row_count:,} rows | |
| - **Columns:** {', '.join(cols)} | |
| 🔍 **What I can help with:** | |
| {examples_text} | |
| **What would you like to know about your data?** Just ask naturally! 💬""" | |
| else: | |
| welcome_response = """👋 **Hi! I'm your AI Data Analyst.** | |
| I don't see any uploaded data yet. Please upload a CSV or Excel file first! | |
| 📤 **To get started:** | |
| 1. Click the upload button to add your data file | |
| 2. Ask me anything about your data! | |
| **What types of data do you have?** I can help analyze HR, Sales, Finance, or any structured data! 📊""" | |
| except Exception as e: | |
| print(f" Welcome data error: {e}") | |
| welcome_response = "👋 Hi! I'm ready to help analyze your data. What would you like to know?" | |
| # Save conversation and return immediately | |
| user_msg = Message(role="user", content=query, timestamp=datetime.now().isoformat()) | |
| assistant_msg = Message(role="assistant", content=welcome_response, timestamp=datetime.now().isoformat(), sources=["Welcome Assistant"]) | |
| history.append(user_msg) | |
| history.append(assistant_msg) | |
| save_conversation(user_id, conversation_id, history) | |
| return ChatResponse( | |
| message=welcome_response, | |
| mode="chat", | |
| sources=["Welcome Assistant"], | |
| conversationId=conversation_id, | |
| timestamp=datetime.now().isoformat() | |
| ) | |
| # ======================================== | |
| # MODE ROUTING - PRIORITY ORDER IS CRITICAL | |
| # ======================================== | |
| # PRIORITY 1: EXACT GREETINGS - ALWAYS chat mode (hi, hello, hey) | |
| if is_exact_greeting: | |
| mode = "chat" | |
| print(f" EXACT GREETING DETECTED: '{query_lower}' CHAT mode") | |
| # PRIORITY 2: Personal/conversational queries | |
| elif is_personal and not has_image: | |
| mode = "chat" | |
| print(f" PERSONAL CHAT MODE - Greeting/conversational query detected") | |
| # PRIORITY 3: RESPECT EXPLICIT MODE SELECTION (RAG modes + AI models) | |
| elif mode in ["graphrag", "graph", "hybrid", "rag", "vision", "prediction", "agentic", "multirag"] or mode in AI_MODELS: | |
| # AI MODELS - Direct to OpenRouter | |
| if mode in AI_MODELS: | |
| print(f" AI MODEL SELECTED: {mode} - Will use OpenRouter with RAG context") | |
| # AGENTIC RAG - Uses AI agent with tools | |
| elif mode == "agentic": | |
| print(f" AGENTIC RAG MODE - AI Agent with tools (retrieve, calculate, visualize)") | |
| # MULTI-RAG - Uses multiple retrieval sources with RRF fusion | |
| elif mode == "multirag": | |
| print(f" MULTI-RAG MODE - Multi-source retrieval with RRF fusion") | |
| # SPECIAL CASE: Vision mode selected but query is about DATA (not image) | |
| elif mode == "vision" and not has_image and is_data_query: | |
| mode = "graph" # Redirect to GraphRAG for data analysis | |
| print(f" SMART REDIRECT: Vision mode + data query GRAPH mode for charts/predictions") | |
| elif mode == "vision" and not has_image: | |
| # Vision without image and no data query - still use vision (will show instructions) | |
| print(f" VISION MODE - No image attached, showing instructions") | |
| elif mode == "prediction": | |
| # Prediction mode - uses HYBRID internally but displays as PREDICTION | |
| print(f" PREDICTION MODE - Forecasts and trends analysis (3 accuracy tiers)") | |
| else: | |
| print(f" EXPLICIT MODE SELECTED: {mode.upper()} - Respecting user choice") | |
| # PRIORITY 4: Business queries with auto mode - use intelligent routing | |
| elif is_business_query and mode == "auto": | |
| mode = route_question(query, has_image=has_image, mode="auto") | |
| # Override: If routed to "vision" but no image, use "graph" instead | |
| if mode == "vision" and not has_image: | |
| mode = "graph" | |
| print(f" Visualization request without image Using GRAPH mode to generate charts") | |
| else: | |
| print(f" BUSINESS QUERY - Auto-routed to: {mode}") | |
| # PRIORITY 5: Image attached | |
| elif has_image: | |
| mode = "vision" | |
| print(f" VISION MODE - Image detected") | |
| # PRIORITY 6: Default auto-routing | |
| elif mode == "auto": | |
| mode = route_question(query, has_image=False, mode="auto") | |
| # Override: If routed to "vision" but no image, use "graph" | |
| if mode == "vision" and not has_image: | |
| mode = "graph" | |
| print(f" Auto-routed to mode: {mode}") | |
| # 🔥 CACHE LOOKUP - Save API costs for similar queries | |
| cached_result = None | |
| if is_business_query and mode != "chat": | |
| cached_result = query_cache.get(query, user_id) | |
| if cached_result: | |
| print(f" CACHE HIT - Returning cached response for: {query[:50]}...") | |
| return ChatResponse( | |
| message=cached_result.response, | |
| mode=cached_result.route, | |
| sources=cached_result.sources if cached_result.sources else None, | |
| conversationId=conversation_id, | |
| timestamp=datetime.now().isoformat() | |
| ) | |
| context = "" | |
| sources = [] | |
| # Check for file comparison intent | |
| comparison_keywords = ['compare', 'difference', 'versus', 'vs', 'between', 'contrast'] | |
| is_comparison = any(kw in query.lower() for kw in comparison_keywords) | |
| # Handle temp attached files (for comparison only, NOT trained) | |
| temp_context = "" | |
| if request.attachedFiles: | |
| temp_context = "\n\n## Temporarily Attached Files (NOT in training data):\n" | |
| for att_file in request.attachedFiles: | |
| temp_context += f"- {att_file.get('name', 'Unknown')}: Use for comparison only\n" | |
| temp_context += "\nNote: These files are NOT trained. Only comparing with trained Data Hub files.\n" | |
| # Use agent workflow nodes for proper responses | |
| response = "" | |
| sources = [] | |
| # Get user's currency setting | |
| try: | |
| currency_symbol, currency_code = get_user_currency(user_id) | |
| except: | |
| currency_symbol = "$" | |
| currency_code = "USD" | |
| # ===================================================================== | |
| # MCP PROCESSING - Run enabled MCPs and get enhanced context/insights | |
| # ===================================================================== | |
| mcp_result = None | |
| mcp_context_text = "" | |
| if enabled_mcps and is_business_query: | |
| print(f" Running MCP tools for query: {query[:50]}...") | |
| mcp_result = run_enabled_mcps(query, user_id, enabled_mcps) | |
| if mcp_result and mcp_result.get("tools_used"): | |
| tools_used = mcp_result["tools_used"] | |
| print(f" MCPs used: {', '.join(tools_used)}") | |
| # Add MCP context to enhance the response | |
| mcp_context_text = mcp_result.get("mcp_context", "") | |
| # Add MCP tools used to sources | |
| for tool in tools_used: | |
| if f"MCP: {tool}" not in sources: | |
| sources.append(f"MCP: {tool}") | |
| # ===================================================================== | |
| # FOLLOW-UP QUERIES: Get last response from MAIN conversation history | |
| # ===================================================================== | |
| if is_followup_query: | |
| try: | |
| print(f" FOLLOW-UP QUERY: '{query[:50]}...'") | |
| # Get last assistant response from MAIN history (not ProductionChatHandler!) | |
| last_assistant_response = "" | |
| for msg in reversed(history): | |
| if msg.role == "assistant": | |
| last_assistant_response = msg.content | |
| break | |
| if not last_assistant_response: | |
| print(f" No previous assistant response found in main history") | |
| else: | |
| print(f" Found last response ({len(last_assistant_response)} chars): {last_assistant_response[:100]}...") | |
| # Build direct follow-up prompt - NO ProductionChatHandler | |
| followup_prompt = f"""You are an AI Data Analyst explaining your previous response. | |
| ## YOUR PREVIOUS RESPONSE (what the user is asking about): | |
| --- | |
| {last_assistant_response[:3000]} | |
| --- | |
| ## USER'S FOLLOW-UP QUESTION: | |
| "{query}" | |
| ## INSTRUCTIONS: | |
| - The user wants you to explain, simplify, or clarify YOUR PREVIOUS RESPONSE shown above | |
| - Do NOT generate new data, charts, or analysis | |
| - Do NOT mention heatmaps, revenue, or invoices unless they were in YOUR PREVIOUS RESPONSE | |
| - ONLY explain/rephrase/summarize what you said in YOUR PREVIOUS RESPONSE | |
| - Be concise and direct | |
| ## YOUR EXPLANATION:""" | |
| # Call LLM directly with follow-up context | |
| import aiohttp | |
| import os | |
| api_key = os.getenv("OPENROUTER_API_KEY") | |
| if api_key: | |
| payload = { | |
| "model": "deepseek/deepseek-chat", | |
| "messages": [ | |
| {"role": "system", "content": followup_prompt} | |
| ], | |
| "max_tokens": 1000, | |
| "temperature": 0.3, | |
| } | |
| headers = { | |
| "Authorization": f"Bearer {api_key}", | |
| "Content-Type": "application/json", | |
| } | |
| async with aiohttp.ClientSession() as session: | |
| async with session.post( | |
| "https://openrouter.ai/api/v1/chat/completions", | |
| json=payload, | |
| headers=headers, | |
| timeout=aiohttp.ClientTimeout(total=30) | |
| ) as api_response: | |
| if api_response.status == 200: | |
| data = await api_response.json() | |
| response = data["choices"][0]["message"]["content"] | |
| sources = ["Follow-up Explanation"] | |
| # Add to history | |
| history.append(Message(role="user", content=query, timestamp=datetime.now().isoformat())) | |
| history.append(Message(role="assistant", content=response, timestamp=datetime.now().isoformat())) | |
| save_conversation(user_id, conversation_id, history) | |
| return ChatResponse( | |
| message=response, | |
| mode="rag", | |
| sources=sources, | |
| conversationId=conversation_id, | |
| timestamp=datetime.now().isoformat() | |
| ) | |
| except Exception as fe: | |
| print(f"[FOLLOW-UP] Error: {fe}") | |
| import traceback | |
| traceback.print_exc() | |
| # ======================================== | |
| # 🖼️ LEGACY MODE IMAGE HANDLING - Process images for RAG/Graph/Hybrid modes | |
| # ======================================== | |
| legacy_image_context = "" | |
| if has_image and mode in ["rag", "graph", "graphrag", "hybrid"]: | |
| print(f" Processing image for legacy mode: {mode}") | |
| processed_files = [] | |
| if request.attachedFiles: | |
| for att_file in request.attachedFiles: | |
| processed_files.append({ | |
| 'name': att_file.get('name', 'image'), | |
| 'type': att_file.get('type', 'image/png'), | |
| 'content': att_file.get('content', '') # Base64 data URL | |
| }) | |
| if processed_files: | |
| for file_info in processed_files: | |
| file_content = file_info.get('content', '') | |
| file_name = file_info.get('name', 'image') | |
| if file_content.startswith('data:image'): | |
| try: | |
| from core.vision import analyze_image_with_groq | |
| analysis = analyze_image_with_groq( | |
| file_content, | |
| f"Describe this image in detail. User asks: {query}" | |
| ) | |
| if not analysis.startswith("❌"): | |
| legacy_image_context += f"\n\n## 🖼️ Image Analysis ({file_name})\n{analysis}\n" | |
| print(f" Legacy image analyzed: {len(analysis)} chars") | |
| except Exception as e: | |
| print(f" Legacy vision error: {e}") | |
| if mode == "rag": | |
| from agents.nodes import rag_answer | |
| from agents.state import AgentState | |
| # Add image context to the query if available | |
| modified_query = query | |
| if legacy_image_context: | |
| modified_query = f"{query}\n\n[IMAGE CONTEXT]:{legacy_image_context}" | |
| print(f" Added image context to RAG query") | |
| state = AgentState(company_id=user_id, question=modified_query, route="rag", answer="", context={}) | |
| state = rag_answer(state) | |
| response = state.answer | |
| sources = state.sources | |
| elif mode == "graph" or mode == "graphrag": | |
| from agents.nodes import graph_answer | |
| from agents.state import AgentState | |
| print(f" Calling graph_answer for mode={mode}") | |
| state = AgentState(company_id=user_id, question=query, route="graph", answer="", context={}) | |
| state = graph_answer(state) | |
| response = state.answer | |
| print(f" graph_answer returned: {len(response) if response else 0} chars") | |
| sources = ["Knowledge Graph Analysis"] | |
| mode = "graph" # Normalize to 'graph' for response | |
| elif mode == "hybrid": | |
| from agents.nodes import hybrid_answer | |
| from agents.state import AgentState | |
| state = AgentState(company_id=user_id, question=query, route="hybrid", answer="", context={}) | |
| state = hybrid_answer(state) | |
| response = state.answer | |
| sources = state.sources | |
| elif mode == "prediction": | |
| # PREDICTION MODE - Uses hybrid pipeline with prediction-focused processing | |
| from agents.nodes import hybrid_answer | |
| from agents.state import AgentState | |
| import re | |
| # Add prediction context to the question | |
| prediction_query = f"[PREDICTION MODE - Use 3 accuracy tiers] {query}" | |
| state = AgentState(company_id=user_id, question=prediction_query, route="prediction", answer="", context={}) | |
| state = hybrid_answer(state) | |
| response = state.answer | |
| # DEFINITIVE CLEANUP: Line-by-line filtering to remove ALL hybrid text | |
| cleaned_lines = [] | |
| skip_patterns = ['reasoning type', 'mode weights', 'rag fusion', 'balanced fusion', 'graph-heavy', 'rag 50%', 'graph 50%'] | |
| for line in response.split('\n'): | |
| line_lower = line.lower().strip() | |
| if any(pattern in line_lower for pattern in skip_patterns): | |
| continue | |
| if 'hybrid' in line_lower: | |
| line = re.sub(r'\*?Analysis Mode:\s*\*?HYBRID\*?', '**Analysis Mode: PREDICTION**', line, flags=re.IGNORECASE) | |
| line = re.sub(r'Mode:\s*HYBRID', 'Mode: PREDICTION', line, flags=re.IGNORECASE) | |
| line = re.sub(r'HYBRID', 'PREDICTION', line, flags=re.IGNORECASE) | |
| if line.strip() in ['**', '*', '---', '']: | |
| if cleaned_lines and cleaned_lines[-1].strip() in ['**', '*', '---', '']: | |
| continue | |
| cleaned_lines.append(line) | |
| response = '\n'.join(cleaned_lines) | |
| response = re.sub(r'\n{3,}', '\n\n', response) | |
| response = re.sub(r'(---\s*\n){2,}', '---\n', response) | |
| # PRODUCTION: Clean response without debug footers | |
| # Remove any lingering Analysis Mode text | |
| response = re.sub(r'\n*---\n*\*?\*?Analysis Mode:?[^\n]*\n?', '', response, flags=re.IGNORECASE) | |
| response = re.sub(r'\n*---\n*📈\s*\*?\*?Analysis Mode:?[^\n]*\n?', '', response, flags=re.IGNORECASE) | |
| response = re.sub(r'Accuracy Tier:[^\n]*\n?', '', response, flags=re.IGNORECASE) | |
| sources = ["Prediction Analysis"] + (state.sources if state.sources else []) | |
| elif mode == "agentic": | |
| # AGENTIC RAG - AI Agent with tools | |
| print(f" AGENTIC RAG: Executing agent-based analysis") | |
| # Get base context from RAG | |
| context, rag_sources = rag_search(user_id, query, k=10, target_files=compare_files) | |
| if AGENTIC_RAG: | |
| # Create agent and plan execution | |
| agent = AgenticRAG(user_id=user_id) | |
| plan = agent.plan_execution(query) | |
| plan_str = " → ".join([a.value for a in plan]) | |
| # Use agentic prompt as context | |
| agentic_context = create_agentic_rag_prompt(query, context, [a.value for a in plan]) | |
| # Call LLM with proper signature: (user_id, query, model_key, context) | |
| response, _ = await ai_model_response(user_id, query, "llama-70b", agentic_context) | |
| # Response is returned directly without verbose prefix | |
| else: | |
| # Fallback to standard RAG with proper signature | |
| response, _ = await ai_model_response(user_id, query, "llama-70b", context) | |
| sources = ["Agentic RAG Intelligence"] | |
| elif mode == "multirag": | |
| # MULTI-RAG - Multiple retrieval sources with RRF fusion | |
| print(f" MULTI-RAG: Executing multi-source retrieval with RRF fusion") | |
| if AGENTIC_RAG: | |
| # Detect which sources to use based on query | |
| sources_to_use = detect_best_sources(query) | |
| print(f" Using sources: {[s.value for s in sources_to_use]}") | |
| # Get context from vector search (primary) | |
| context, rag_sources = rag_search(user_id, query, k=10, target_files=compare_files) | |
| # Get context from graph (if relationship query) | |
| graph_context = "" | |
| if RetrievalSource.GRAPH in sources_to_use: | |
| try: | |
| graph_results = graph_query(user_id, query) | |
| graph_context = f"\n\n## From Knowledge Graph:\n{graph_results}" | |
| except: | |
| pass | |
| # Combine contexts with source labels | |
| combined_context = f"## From Vector Search:\n{context}{graph_context}" | |
| # Format response with multi-source info | |
| source_names = [s.value for s in sources_to_use] | |
| # Call LLM with proper signature: (user_id, query, model_key, context) | |
| response, _ = await ai_model_response(user_id, query, "llama-70b", combined_context) | |
| # Response returned directly without verbose prefix | |
| sources = rag_sources + [f"Multi-RAG ({', '.join(source_names)})"] | |
| else: | |
| # Fallback to standard RAG with proper signature | |
| context, sources = rag_search(user_id, query, k=10, target_files=compare_files) | |
| response, _ = await ai_model_response(user_id, query, "llama-70b", context) | |
| elif mode == "vision": | |
| from agents.nodes import vision_answer | |
| from agents.state import AgentState | |
| import base64 | |
| import tempfile | |
| # DEBUG: Print what we received | |
| print(f" VISION MODE: request.attachedFiles = {request.attachedFiles}") | |
| print(f" VISION MODE: Number of files = {len(request.attachedFiles) if request.attachedFiles else 0}") | |
| # Save attached images to temporary files for Gemini Vision | |
| processed_files = [] | |
| if request.attachedFiles: | |
| for file_data in request.attachedFiles: | |
| if file_data.get('type', '').startswith('image/'): | |
| print(f" Processing attached file: {file_data.get('name', 'unknown')}") | |
| print(f" File type: {file_data.get('type', 'unknown')}") | |
| # Check what fields are available | |
| print(f" Available fields: {list(file_data.keys())}") | |
| # Try different ways to get image data | |
| content = None | |
| # Method 1: Base64 content field | |
| if 'content' in file_data: | |
| content = file_data.get('content', '') | |
| # Remove data URL prefix if present | |
| if content.startswith('data:'): | |
| content = content.split(',', 1)[1] if ',' in content else content | |
| print(f" Using 'content' field, length: {len(content)}") | |
| # Method 2: URL field (already uploaded file) | |
| elif 'url' in file_data: | |
| url = file_data.get('url', '') | |
| print(f" Found URL field: {url}") | |
| # If it's a local file path | |
| if url.startswith('/') or url.startswith('C:') or url.startswith('c:'): | |
| try: | |
| with open(url, 'rb') as f: | |
| content = base64.b64encode(f.read()).decode('utf-8') | |
| print(f" Loaded from file path") | |
| except Exception as e: | |
| print(f" Failed to load from path: {e}") | |
| # Method 3: Data field | |
| elif 'data' in file_data: | |
| content = file_data.get('data', '') | |
| if content.startswith('data:'): | |
| content = content.split(',', 1)[1] if ',' in content else content | |
| print(f" Using 'data' field, length: {len(content)}") | |
| if not content: | |
| print(f" No image data found in: {file_data}") | |
| continue | |
| # Decode base64 and save to temp file | |
| try: | |
| image_bytes = base64.b64decode(content) | |
| print(f" Decoded {len(image_bytes)} bytes") | |
| # Create temp file with proper extension | |
| file_ext = file_data.get('type', 'image/png').split('/')[-1] | |
| temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=f'.{file_ext}', mode='wb') | |
| temp_file.write(image_bytes) | |
| temp_file.close() | |
| processed_files.append({ | |
| 'name': file_data.get('name', 'image'), | |
| 'path': temp_file.name, | |
| 'type': file_data.get('type', 'image/png') | |
| }) | |
| print(f" Saved image to: {temp_file.name}") | |
| except Exception as e: | |
| print(f" Failed to decode image: {e}") | |
| print(f" Failed to decode image: {e}") | |
| traceback.print_exc() | |
| state = AgentState(company_id=user_id, question=query, route="vision", answer="", context={"attached_files": processed_files}) | |
| state = vision_answer(state) | |
| response = state.answer | |
| sources = ["Vision Analysis"] | |
| # Clean up temp files | |
| import os | |
| for file in processed_files: | |
| try: | |
| os.unlink(file['path']) | |
| except: | |
| pass | |
| elif mode == "chat": | |
| # Personal/greeting response - conversational mode | |
| context = conversation_context | |
| # AI MODELS VIA OPENROUTER - DeepSeek, Qwen, Nous, Gemini, Llama | |
| elif mode in AI_MODELS: | |
| print(f" AI MODEL MODE: {mode} - Using OpenRouter with RAG context") | |
| response, sources = await ai_model_response( | |
| user_id=user_id, | |
| query=query, | |
| model_key=mode, | |
| conversation_context=conversation_context | |
| ) | |
| print(f" AI model returned: {len(response) if response else 0} chars") | |
| # For agent modes (rag, graph, hybrid, vision), response is already complete | |
| # Only build prompt for chat mode | |
| if mode == "chat": | |
| file_metadata = get_file_metadata(user_id) | |
| file_count = len(file_metadata) | |
| # SHORT prompt to avoid context length issues with Groq | |
| prompt = f"""You are an AI Business Analyst assistant. Be friendly and helpful. | |
| User: {query} | |
| Rules: | |
| - If greeting (hi/hello): Say hello, mention you have {file_count} data files, offer to help with revenue, customers, products, or trends. | |
| - If asked who you are: Briefly explain you are an AI Business Analyst. | |
| - If thanked: Say you are welcome. | |
| - Keep response under 3 sentences. Be warm and natural.""" | |
| response = chat(prompt, max_tokens=200) | |
| # ========================================================================= | |
| # MCP CONTEXT INJECTION - Append MCP insights to response | |
| # ========================================================================= | |
| # Add real-time market context simulation | |
| if is_business_query: | |
| market_context = MarketContextAgent.get_market_context(query) | |
| if market_context: | |
| mcp_context_text = mcp_context_text + "\n\n" + market_context if mcp_context_text else market_context | |
| if mcp_context_text and is_business_query: | |
| # Add MCP insights at the end of the response | |
| response = response + "\n\n---\n" + mcp_context_text | |
| # Add MCP tools used indicator (subtle, Claude-style) | |
| if mcp_result and mcp_result.get("tools_used"): | |
| tools_list = ", ".join(mcp_result["tools_used"]) | |
| response += f"\n\n🔧 *Tools used: {tools_list}*" | |
| # ========================================================================= | |
| # FINAL STEP: Universal Chart Injection for ALL modes | |
| # ========================================================================= | |
| response = append_chart_if_needed(response, query, user_id) | |
| # 🛡️ SECURITY: Apply output guardrails to prevent hallucinations/harmful code | |
| response = SecurityVault.apply_nemo_guardrails(response) | |
| print(f" Sending response: mode={mode}, length={len(response)}") | |
| # 🔥 CACHE STORAGE - Save response for future similar queries | |
| if is_business_query and mode != "chat" and response: | |
| query_cache.set( | |
| query=query, | |
| response=response, | |
| route=mode, | |
| sources=sources if sources else [], | |
| user_id=user_id, | |
| ttl=3600 # 1 hour cache | |
| ) | |
| print(f" Cached response for: {query[:50]}...") | |
| assistant_msg = Message( | |
| role="assistant", | |
| content=response, | |
| timestamp=datetime.now().isoformat(), | |
| sources=sources if sources else None | |
| ) | |
| history.append(assistant_msg) | |
| save_conversation(user_id, conversation_id, history) | |
| # 🧠 ENHANCED QDRANT MEMORY: Save this interaction for long-term recall | |
| if VECTOR_MEMORY_AVAILABLE and vector_store: | |
| try: | |
| # Save user query | |
| vector_store.add_chat_message(user_id=user_id, role="user", content=query, conversation_id=conversation_id) | |
| # Save AI response (truncated to avoid massive embeddings) | |
| vector_store.add_chat_message(user_id=user_id, role="assistant", content=response[:1500], conversation_id=conversation_id) | |
| print("Saved interaction to Qdrant Semantic Memory") | |
| except Exception as e: | |
| print(f" Failed to save to vector memory: {e}") | |
| # 🏆 COMPETITION FEATURE: Generate smart follow-up suggestions | |
| try: | |
| # Get column names for context | |
| data_columns = [] | |
| try: | |
| # Try to get columns from sources or response | |
| import re | |
| col_match = re.search(r'Columns?:\s*\[([^\]]+)\]', str(response)) | |
| if col_match: | |
| data_columns = [c.strip().strip("'\"") for c in col_match.group(1).split(',')] | |
| except: | |
| pass | |
| suggestions = generate_smart_suggestions( | |
| query=query, | |
| response=response[:1000], # Limit context size | |
| columns=data_columns if data_columns else sources, | |
| max_suggestions=3 | |
| ) | |
| confidence = calculate_confidence( | |
| response=response, | |
| data_context=response[:500], # Use response as context | |
| columns=data_columns | |
| ) | |
| print(f" Generated {len(suggestions)} follow-up suggestions, confidence: {confidence:.2f}") | |
| except Exception as e: | |
| print(f" Suggestion generation failed: {e}") | |
| suggestions = [] | |
| confidence = 0.75 | |
| return ChatResponse( | |
| message=response, | |
| mode=mode, | |
| sources=sources if sources else None, | |
| conversationId=conversation_id, | |
| timestamp=datetime.now().isoformat(), | |
| suggestions=suggestions if suggestions else None, | |
| confidence=confidence | |
| ) | |
| except Exception as e: | |
| print(f" Chat endpoint error: {e}") | |
| import traceback as tb | |
| tb.print_exc() | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def get_conversations( | |
| user_id: str, | |
| x_user_id: Optional[str] = Header(None, alias="X-User-ID"), | |
| authorization: Optional[str] = Header(None, alias="Authorization") | |
| ): | |
| """Get all conversations for a user - SECURED""" | |
| try: | |
| # SECURITY: Validate user | |
| if authorization and authorization.startswith("Bearer "): | |
| try: | |
| from database.auth import decode_jwt | |
| token = authorization.split(" ")[1] | |
| payload = decode_jwt(token) | |
| auth_user = payload.get("sub") | |
| if auth_user and auth_user != user_id: | |
| user_id = auth_user | |
| except: | |
| pass | |
| elif x_user_id and x_user_id != user_id: | |
| user_id = x_user_id | |
| paths = get_user_paths(user_id) | |
| conversations = [] | |
| if paths["memory"].exists(): | |
| for file in paths["memory"].glob("*.json"): | |
| try: | |
| with open(file, 'r') as f: | |
| data = json.load(f) | |
| messages = data.get("messages", []) | |
| if messages: | |
| conversations.append({ | |
| "id": data.get("conversation_id"), | |
| "title": messages[0].get("content", "")[:50], | |
| "lastMessage": messages[-1].get("content", "")[:100], | |
| "timestamp": data.get("updated_at"), | |
| "messageCount": len(messages) | |
| }) | |
| except: | |
| pass | |
| conversations.sort(key=lambda x: x.get("timestamp", ""), reverse=True) | |
| return {"conversations": conversations} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def get_conversation_messages( | |
| user_id: str, | |
| conversation_id: str, | |
| x_user_id: Optional[str] = Header(None, alias="X-User-ID"), | |
| authorization: Optional[str] = Header(None, alias="Authorization") | |
| ): | |
| """Get messages for a conversation - SECURED""" | |
| try: | |
| # SECURITY: Use auth header user | |
| if authorization and authorization.startswith("Bearer "): | |
| try: | |
| from database.auth import decode_jwt | |
| token = authorization.split(" ")[1] | |
| payload = decode_jwt(token) | |
| auth_user = payload.get("sub") | |
| if auth_user: | |
| user_id = auth_user | |
| except: | |
| pass | |
| elif x_user_id: | |
| user_id = x_user_id | |
| messages = load_conversation(user_id, conversation_id) | |
| return { | |
| "conversationId": conversation_id, | |
| "messages": [msg.dict() for msg in messages] | |
| } | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def delete_conversation( | |
| user_id: str, | |
| conversation_id: str, | |
| x_user_id: Optional[str] = Header(None, alias="X-User-ID"), | |
| authorization: Optional[str] = Header(None, alias="Authorization") | |
| ): | |
| """Delete a conversation - SECURED""" | |
| try: | |
| # SECURITY: Use auth header user | |
| if authorization and authorization.startswith("Bearer "): | |
| try: | |
| from database.auth import decode_jwt | |
| token = authorization.split(" ")[1] | |
| payload = decode_jwt(token) | |
| auth_user = payload.get("sub") | |
| if auth_user: | |
| user_id = auth_user | |
| except: | |
| pass | |
| elif x_user_id: | |
| user_id = x_user_id | |
| paths = get_user_paths(user_id) | |
| history_file = paths["memory"] / f"{conversation_id}.json" | |
| if history_file.exists(): | |
| history_file.unlink() | |
| return {"success": True, "message": "Deleted"} | |
| else: | |
| raise HTTPException(status_code=404, detail="Not found") | |
| except HTTPException: | |
| raise | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| # ============================================================================ | |
| # MODELS API - Get available AI models for frontend | |
| # ============================================================================ | |
| async def get_available_models(): | |
| """ | |
| Get all available AI models for the frontend model selector. | |
| Returns models grouped by category. | |
| """ | |
| try: | |
| if ADVANCED_RAG: | |
| return get_available_models_api() | |
| # Fallback if advanced RAG not loaded | |
| return { | |
| "models": [ | |
| {"id": "deepseek", "label": "DeepSeek Chat", "description": "Fast • Accurate", "badge": "Best", "category": "general"}, | |
| {"id": "mistral", "label": "Mistral Small", "description": "Fast • Free", "badge": "Free", "category": "general"}, | |
| {"id": "llama", "label": "Llama 3.3 70B", "description": "Comprehensive", "badge": "Free", "category": "general"}, | |
| ], | |
| "default": "deepseek", | |
| "categories": { | |
| "general": ["deepseek", "mistral", "llama"], | |
| "vision": [], | |
| "code": [] | |
| } | |
| } | |
| except Exception as e: | |
| print(f"Error getting models: {e}") | |
| return {"models": [], "default": "deepseek", "categories": {}} | |
| async def health_check(): | |
| """Health check endpoint for monitoring""" | |
| return { | |
| "status": "healthy", | |
| "rag_enhancements": RAG_ENHANCEMENTS, | |
| "advanced_rag": ADVANCED_RAG if 'ADVANCED_RAG' in dir() else False, | |
| "charts_available": CHARTS_AVAILABLE, | |
| } | |
| # ============================================================================= | |
| # V5: CODE EXECUTION — AI Chat v2 | |
| # ============================================================================= | |
| class CodeExecutionRequest(BaseModel): | |
| code: str | |
| user_id: Optional[str] = "default" | |
| async def execute_code( | |
| request: CodeExecutionRequest, | |
| x_user_id: Optional[str] = Header(None, alias="X-User-ID"), | |
| authorization: Optional[str] = Header(None, alias="Authorization") | |
| ): | |
| """ | |
| 🖥️ Execute Python code in a sandboxed environment. | |
| Returns stdout, stderr, matplotlib figures (base64), and DataFrames. | |
| Used by AI Chat v2 for inline code execution. | |
| """ | |
| user_id = x_user_id or request.user_id or "default" | |
| try: | |
| from core.code_executor import execute_code_for_user | |
| result = execute_code_for_user(user_id=user_id, code=request.code) | |
| return {"success": True, **result} | |
| except Exception as e: | |
| return {"success": False, "error": str(e), "stdout": "", "stderr": str(e), "figures": [], "tables": []} | |