Spaces:
Sleeping
Sleeping
| """ | |
| RTS Data Chatbot | |
| ================ | |
| FastAPI backend for Hugging Face Spaces. | |
| Serves a web UI at GET / and the chatbot API at POST /chat. | |
| """ | |
| import os | |
| import json | |
| import io | |
| import re | |
| import time | |
| import traceback | |
| import pandas as pd | |
| from datetime import datetime, timezone, timedelta | |
| from rapidfuzz import process, fuzz | |
| from azure.storage.blob import BlobServiceClient | |
| import requests | |
| from fastapi import FastAPI, Request, HTTPException | |
| from fastapi.responses import JSONResponse, HTMLResponse | |
| from fastapi.staticfiles import StaticFiles | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from pydantic import BaseModel | |
| from dotenv import load_dotenv | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # CONFIGURATION | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| load_dotenv() | |
| AZURE_CONN_STR = os.getenv("AZURE_STORAGE_CONNECTION_STRING", "") | |
| AZURE_CONTAINER = os.getenv("AZURE_CONTAINER_NAME", "chatbottrial") | |
| CSV_BLOB_NAME = os.getenv("CSV_BLOB_NAME", "RTS.csv") | |
| JSON_BLOB_NAME = os.getenv("JSON_BLOB_NAME", "RTS.json") | |
| SARVAM_API_KEY = os.getenv("SARVAM_API_KEY", "") | |
| SARVAM_MODEL = "sarvam-105b" | |
| SARVAM_ENDPOINT = "https://api.sarvam.ai/v1/chat/completions" | |
| SECRET_KEY = os.getenv("SECRET_KEY", "") | |
| PROJECT_NAMES = [p.strip() for p in os.getenv("PROJECT_NAMES", "").split(",") if p.strip()] | |
| FUZZY_COLUMNS = ["Division", "District", "Taluka", "Service", "Department"] | |
| FUZZY_TOP_N = 5 | |
| FUZZY_THRESHOLD = 40 | |
| FUZZY_THRESHOLDS = { | |
| "Division" : 60, | |
| "District" : 60, | |
| "Taluka" : 60, | |
| "Service" : 50, | |
| "Department": 55, | |
| } | |
| LOGS_FILE = os.getenv("LOGS_FILE", "logs.json") | |
| IST = timezone(timedelta(hours=5, minutes=30)) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # LOGGING HELPERS | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def _now_ist() -> datetime: | |
| return datetime.now(IST) | |
| def _format_response_time(elapsed_seconds: float) -> str: | |
| total_seconds = int(round(elapsed_seconds)) | |
| minutes = total_seconds // 60 | |
| seconds = total_seconds % 60 | |
| return f"{minutes:02d}:{seconds:02d}" | |
| def _build_token_dict(usage, details, prompt_details) -> dict: | |
| usage = usage or {} | |
| details = details or {} | |
| prompt_details = prompt_details or {} | |
| return { | |
| "prompt_tokens" : usage.get("prompt_tokens", 0), | |
| "completion_tokens": usage.get("completion_tokens", 0), | |
| "reasoning_tokens" : details.get("reasoning_tokens", 0) or 0, | |
| "cached_tokens" : ( | |
| prompt_details.get("cached_tokens") | |
| or usage.get("cache_read_input_tokens") | |
| or 0 | |
| ), | |
| "total_tokens": usage.get("total_tokens", 0), | |
| } | |
| def read_logs() -> list: | |
| if not os.path.exists(LOGS_FILE): | |
| return [] | |
| try: | |
| with open(LOGS_FILE, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| return data if isinstance(data, list) else [] | |
| except (json.JSONDecodeError, OSError): | |
| return [] | |
| def write_logs(entries: list) -> None: | |
| with open(LOGS_FILE, "w", encoding="utf-8") as f: | |
| json.dump(entries, f, ensure_ascii=False, indent=2) | |
| def append_log(entry: dict) -> None: | |
| entries = read_logs() | |
| entries.append(entry) | |
| write_logs(entries) | |
| def build_log_entry( | |
| *, | |
| project_name, | |
| user_query, | |
| fuzzy_results, | |
| intent, | |
| plan_text, | |
| llm_queries, | |
| final_answer, | |
| call1_tokens, | |
| call2_tokens, | |
| success, | |
| error_code, | |
| error_message, | |
| elapsed_seconds, | |
| ) -> dict: | |
| now = _now_ist() | |
| fuzzy_list = { | |
| col: [{"value": h["value"], "confidence": h["score"]} for h in hits] | |
| for col, hits in fuzzy_results.items() | |
| } | |
| return { | |
| "date" : now.strftime("%d/%m/%Y"), | |
| "time" : now.strftime("%I:%M %p").lstrip("0").upper() + " IST", | |
| "project_name" : project_name, | |
| "user_query" : user_query, | |
| "fuzzy_list" : fuzzy_list, | |
| "intent" : intent, | |
| "plan" : plan_text, | |
| "llm_query" : llm_queries, | |
| "final_answer" : final_answer, | |
| "call1_token_count": call1_tokens, | |
| "call2_token_count": call2_tokens, | |
| "success" : "Success" if success else "Error", | |
| "error_code" : error_code, | |
| "error" : error_message, | |
| "response_time" : _format_response_time(elapsed_seconds), | |
| } | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # 1. DATA LOADING | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def load_blob_text(container_client, blob_name: str) -> str: | |
| blob_client = container_client.get_blob_client(blob_name) | |
| data = blob_client.download_blob().readall() | |
| return data.decode("utf-8") | |
| def load_data(): | |
| if not AZURE_CONN_STR: | |
| raise EnvironmentError( | |
| "AZURE_STORAGE_CONNECTION_STRING is not set. " | |
| "Add it as a Space Secret in Settings." | |
| ) | |
| print("📦 Connecting to Azure Blob Storage …") | |
| service_client = BlobServiceClient.from_connection_string(AZURE_CONN_STR) | |
| container_client = service_client.get_container_client(AZURE_CONTAINER) | |
| print(f" ↓ Downloading {CSV_BLOB_NAME} …") | |
| csv_text = load_blob_text(container_client, CSV_BLOB_NAME) | |
| df = pd.read_csv(io.StringIO(csv_text)) | |
| print(f" ↓ Downloading {JSON_BLOB_NAME} …") | |
| json_text = load_blob_text(container_client, JSON_BLOB_NAME) | |
| column_meta = json.loads(json_text) | |
| print(f"✅ Data loaded — {len(df):,} rows × {len(df.columns)} columns\n") | |
| return df, column_meta | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # 2. SCHEMA BUILDER | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def build_schema_line(column_meta: list) -> str: | |
| return ", ".join( | |
| f"{entry['column_name']}({entry['data_type']})" | |
| for entry in column_meta | |
| ) | |
| def build_column_desc_map(column_meta: list) -> dict: | |
| return {entry["column_name"]: entry["description"] for entry in column_meta} | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # 3. FUZZY SEARCH | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def get_unique_values(df: pd.DataFrame, column: str) -> list: | |
| if column not in df.columns: | |
| return [] | |
| return sorted(df[column].dropna().astype(str).unique().tolist()) | |
| def fuzzy_search_column(query, choices, top_n=FUZZY_TOP_N, threshold=FUZZY_THRESHOLD) -> list: | |
| if not choices or not query.strip(): | |
| return [] | |
| results = process.extract(query, choices, scorer=fuzz.WRatio, limit=top_n) | |
| return [ | |
| {"value": match, "score": round(score, 1)} | |
| for match, score, _ in results | |
| if score >= threshold | |
| ] | |
| def run_fuzzy_searches(query: str, df: pd.DataFrame) -> dict: | |
| results = {} | |
| for col in FUZZY_COLUMNS: | |
| choices = get_unique_values(df, col) | |
| threshold = FUZZY_THRESHOLDS.get(col, FUZZY_THRESHOLD) | |
| hits = fuzzy_search_column(query, choices, threshold=threshold) | |
| results[col] = hits | |
| return results | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # 4. SARVAM LLM INTERFACE | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| _last_token_usage: dict = {} | |
| def call_sarvam(messages, max_tokens=4096, label="", temperature=0.7) -> str: | |
| global _last_token_usage | |
| if not SARVAM_API_KEY: | |
| raise EnvironmentError("SARVAM_API_KEY is not set. Add it as a Space Secret.") | |
| payload = { | |
| "model" : SARVAM_MODEL, | |
| "messages" : messages, | |
| "max_tokens" : max_tokens, | |
| "temperature": temperature, | |
| } | |
| resp = requests.post( | |
| SARVAM_ENDPOINT, | |
| headers={ | |
| "Authorization": f"Bearer {SARVAM_API_KEY}", | |
| "Content-Type" : "application/json", | |
| }, | |
| json=payload, | |
| timeout=120, | |
| ) | |
| resp.raise_for_status() | |
| data = resp.json() | |
| usage = data.get("usage") or {} | |
| details = usage.get("completion_tokens_details") or {} | |
| prompt_details = usage.get("prompt_tokens_details") or {} | |
| token_dict = _build_token_dict(usage, details, prompt_details) | |
| _last_token_usage = token_dict | |
| tag = f" [{label}]" if label else "" | |
| print( | |
| f" 🪙 Tokens{tag} " | |
| f"prompt={token_dict['prompt_tokens']} " | |
| f"completion={token_dict['completion_tokens']} " | |
| f"total={token_dict['total_tokens']}" | |
| ) | |
| try: | |
| choice = data["choices"][0] | |
| message = choice.get("message") or {} | |
| content = message.get("content") | |
| reasoning = message.get("reasoning_content") or "" | |
| if content: | |
| return content.strip() | |
| if reasoning: | |
| for marker in [ | |
| "Final Answer Formulation:", | |
| "Final Answer:", | |
| "final answer:", | |
| "Let's go with", | |
| "Going with", | |
| "**Final Answer**", | |
| ]: | |
| idx = reasoning.rfind(marker) | |
| if idx != -1: | |
| snippet = reasoning[idx + len(marker):].strip() | |
| first_para = snippet.split("\n\n")[0].strip(" *:-\n") | |
| first_para = re.sub(r"^\*{1,2}|^\d+\.\s*", "", first_para).strip() | |
| if first_para: | |
| return first_para | |
| lines = [l.strip() for l in reasoning.splitlines() if l.strip()] | |
| if lines: | |
| return lines[-1].rstrip(" .") + "." | |
| content = choice.get("text") or message.get("text") or "" | |
| if content: | |
| return content.strip() | |
| print(f"\n⚠️ WARNING: LLM returned empty content.\n") | |
| return "" | |
| except (KeyError, IndexError, TypeError) as e: | |
| raise ValueError(f"Could not parse LLM response: {e}") from e | |
| def get_last_token_usage() -> dict: | |
| return dict(_last_token_usage) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # 5. LLM QUERY PLANNING | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| _SCHEMA_LINE = None | |
| SYSTEM_PROMPT_TEMPLATE = """You are a pandas query generator for Maharashtra's RTS (Right to Service) dataset. | |
| Return ONLY valid JSON. No markdown, no explanation, no extra text. | |
| DataFrame `df` schema (column_name(data_type)): | |
| {schema} | |
| Numeric metric columns: TotalReceived, TotalDisposed, Pending, Approved, Rejected, OnTimeDelivery, NotOnTimeDelivery, Total, Online, Offline, PendingatUser, PendingatDepartment | |
| Filter/group columns: Department, Service, Division, District, Taluka, ApplicationSource, Payment_Mode, SubDepartment. | |
| Year columns: `yr` (integer, 2015–2025, start year of financial year) | `Year` (string, e.g. "F.Y. 2024 - 2025") — ALWAYS use `yr` for numeric year filters; NEVER df['Year']=='2025' | |
| Location rule: A place name (e.g. "Baramati", "Pune", "Nashik") may appear in Division, District, OR Taluka. ALWAYS check the fuzzy hints to see which column it matched. If it matched Taluka, filter on Taluka — do NOT assume it is a District. | |
| CRITICAL RESULT RULES — always return rich DataFrames, never bare scalars: | |
| - NEVER use idxmin() or idxmax() alone as the final result. Always return the full row/DataFrame including the metric value. | |
| - For ranking/top-N queries: return a DataFrame with all relevant columns (name, counts, rates, etc.). | |
| - For rate/percentage queries: include numerator, denominator, AND the calculated rate column in the result DataFrame. | |
| - For comparison queries: return all compared groups in one DataFrame. | |
| - result must always be a DataFrame or Series — never a raw scalar string. | |
| - Filter out rows where TotalReceived == 0 before computing rates to avoid division-by-zero. | |
| JSON output format: | |
| {{"intent":"<one line>","plan":"<one line>","queries":[{{"description":"<what it does>","code":"<pandas code; last line must be: result = ...>"}}]}} | |
| Code rules: read-only, no inplace, use df.copy() if needed, string match via .str.contains(x,case=False,na=False) or .isin([...]). | |
| Example — CORRECT code for "service with lowest approval rate": | |
| service_stats = df.groupby('Service')[['TotalReceived', 'Approved']].sum().reset_index() | |
| service_stats = service_stats[service_stats['TotalReceived'] > 0] | |
| service_stats['approval_rate_pct'] = (service_stats['Approved'] / service_stats['TotalReceived'] * 100).round(2) | |
| result = service_stats.nsmallest(5, 'approval_rate_pct')[['Service', 'TotalReceived', 'Approved', 'approval_rate_pct']] | |
| Example — CORRECT code for a single-value lookup: | |
| filtered = df[df['District'].str.contains('Pune', case=False, na=False) & (df['yr'] == 2024)] | |
| result = filtered[['Service', 'TotalReceived', 'Approved', 'Rejected']].groupby('Service').sum().reset_index() | |
| """ | |
| def _build_system_prompt(column_meta: list) -> str: | |
| global _SCHEMA_LINE | |
| if _SCHEMA_LINE is None: | |
| _SCHEMA_LINE = build_schema_line(column_meta) | |
| return SYSTEM_PROMPT_TEMPLATE.format(schema=_SCHEMA_LINE) | |
| def build_planning_prompt(user_query, fuzzy_results, column_desc, df) -> str: | |
| key_columns = ["Division", "District", "Taluka", "Department", "Service"] | |
| top5_fuzzy = {} | |
| for col in key_columns: | |
| hits = fuzzy_results.get(col, []) | |
| choices = get_unique_values(df, col) | |
| if len(hits) < 5: | |
| extra_hits = fuzzy_search_column(user_query, choices, top_n=5, threshold=0) | |
| seen = {h["value"] for h in hits} | |
| for h in extra_hits: | |
| if h["value"] not in seen: | |
| hits.append(h) | |
| seen.add(h["value"]) | |
| if len(hits) == 5: | |
| break | |
| top5_fuzzy[col] = hits[:5] | |
| relevant_cols = set(key_columns) | {"TotalReceived", "Approved", "Rejected", "Pending", "yr", "Year"} | |
| slim_desc = {k: v for k, v in column_desc.items() if k in relevant_cols} | |
| parts = [f"Query: {user_query}"] | |
| parts.append( | |
| "\nFor each of Division, District, Taluka, Department, and Service, here are the top 5 fuzzy matches. " | |
| "If the user query contains a spelling mistake, use the correct option from these fuzzy matches. " | |
| "If none of these matches are suitable, use a contains filter on the column." | |
| ) | |
| parts.append("\nFuzzy matches (col → top 5):") | |
| for col in key_columns: | |
| hits = top5_fuzzy.get(col, []) | |
| vals = ", ".join(f"{h['value']}({h['score']})" for h in hits) | |
| parts.append(f" {col}: {vals}") | |
| if slim_desc: | |
| parts.append("\nRelevant column descriptions:") | |
| for col, desc in slim_desc.items(): | |
| parts.append(f" {col}: {desc}") | |
| return "\n".join(parts) | |
| def get_query_plan(user_query, fuzzy_results, column_meta, df) -> dict: | |
| column_desc = build_column_desc_map(column_meta) | |
| messages = [ | |
| {"role": "system", "content": _build_system_prompt(column_meta)}, | |
| {"role": "user", "content": build_planning_prompt( | |
| user_query, fuzzy_results, column_desc, df | |
| )}, | |
| ] | |
| raw = call_sarvam(messages, max_tokens=4096, label="Query Planning", temperature=0.3) | |
| clean = re.sub(r"```(?:json)?", "", raw).strip().rstrip("`").strip() | |
| try: | |
| return json.loads(clean) | |
| except json.JSONDecodeError: | |
| match = re.search(r"\{.*\}", clean, re.DOTALL) | |
| if match: | |
| return json.loads(match.group()) | |
| raise ValueError(f"LLM did not return valid JSON:\n{raw}") | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # 6. QUERY EXECUTION | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def execute_query(code: str, df: pd.DataFrame): | |
| local_ns = {"df": df.copy(), "pd": pd} | |
| exec(compile(code, "<llm_query>", "exec"), {"__builtins__": {}}, local_ns) # noqa: S102 | |
| if "result" not in local_ns: | |
| raise ValueError("Query code did not define a `result` variable.") | |
| return local_ns["result"] | |
| def execute_plan(plan: dict, df: pd.DataFrame) -> list: | |
| outputs = [] | |
| for q in plan.get("queries", []): | |
| desc = q.get("description", "") | |
| code = q.get("code", "") | |
| try: | |
| result = execute_query(code, df) | |
| if isinstance(result, pd.DataFrame): | |
| result_str = result.to_string(index=False) | |
| result_repr = result.to_dict(orient="records")[:50] | |
| elif isinstance(result, pd.Series): | |
| result_str = result.to_string() | |
| result_repr = result.to_dict() | |
| else: | |
| result_str = str(result) | |
| result_repr = {"value": result} | |
| outputs.append({ | |
| "description": desc, | |
| "code" : code, | |
| "result_str" : result_str, | |
| "result_repr": result_repr, | |
| "error" : None, | |
| }) | |
| except Exception as e: | |
| outputs.append({ | |
| "description": desc, | |
| "code" : code, | |
| "result_str" : "", | |
| "result_repr": None, | |
| "error" : f"{type(e).__name__}: {e}", | |
| }) | |
| return outputs | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # 7. FINAL ANSWER GENERATION (HTML output) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| ANSWER_SYSTEM_PROMPT = """You are a data reporting assistant for Maharashtra's RTS (Right to Service) dataset. | |
| Your job is to convert query results into clean, renderable HTML. | |
| OUTPUT FORMAT — MANDATORY: | |
| - Return ONLY valid HTML. No markdown. No plain text outside HTML tags. No preamble. No explanation. | |
| - Always start with a <p> summary sentence that directly answers the user question. | |
| - If the result has multiple rows/columns, render a styled HTML <table> with <thead> and <tbody>. | |
| - If the result is a single number or name, wrap it in <p> with <strong> around the key value. | |
| - If the result is empty or zero rows, return: <p>No data was found for the given query.</p> | |
| - Numeric columns: right-align using style="text-align:right" on <td> and <th>. | |
| - Percentages: always show 2 decimal places followed by % symbol. | |
| - Large numbers: format with Indian-style commas (e.g. 1,23,456). | |
| - Match the user's language (English/Hindi/Marathi) for the summary sentence. | |
| ALLOWED HTML TAGS ONLY: p, strong, em, table, thead, tbody, tr, th, td, ol, ul, li, br, span | |
| FORBIDDEN: html, head, body, div, script, style tags, CSS classes, any attribute except style on td/th. | |
| """ | |
| def _serialize_result_for_prompt(query_outputs: list) -> str: | |
| parts = [] | |
| for i, out in enumerate(query_outputs, 1): | |
| parts.append(f"--- Result {i}: {out['description']} ---") | |
| if out["error"]: | |
| parts.append(f"ERROR: {out['error']}") | |
| else: | |
| parts.append(out["result_str"] or "(empty — no rows returned)") | |
| return "\n".join(parts) | |
| def generate_final_answer(user_query, plan, query_outputs) -> str: | |
| results_text = _serialize_result_for_prompt(query_outputs) | |
| user_msg = ( | |
| f"User Question: {user_query}\n\n" | |
| f"Query Results:\n{results_text}\n\n" | |
| "Generate a complete HTML answer. " | |
| "Start with a <p> summary sentence. " | |
| "If there are multiple rows, include a <table>." | |
| ) | |
| messages = [ | |
| {"role": "system", "content": ANSWER_SYSTEM_PROMPT}, | |
| {"role": "user", "content": user_msg}, | |
| ] | |
| raw_answer = call_sarvam(messages, max_tokens=4096, label="Final Answer", temperature=0.1) | |
| html = re.sub(r"```(?:html)?", "", raw_answer).strip().rstrip("`").strip() | |
| if not html or not re.search(r"<[a-z]+", html, re.IGNORECASE): | |
| html = _fallback_html(user_query, plan, query_outputs) | |
| return html | |
| def _format_number(val) -> str: | |
| if isinstance(val, float): | |
| if val == int(val): | |
| return f"{int(val):,}" | |
| return f"{val:,.2f}" | |
| if isinstance(val, int): | |
| return f"{val:,}" | |
| return str(val) | |
| def _is_numeric(val) -> bool: | |
| return isinstance(val, (int, float)) | |
| def _fallback_html(user_query, plan, query_outputs) -> str: | |
| parts = [f"<p><strong>Query:</strong> {user_query}</p>"] | |
| for out in query_outputs: | |
| parts.append(f"<p><em>{out['description']}</em></p>") | |
| if out["error"]: | |
| parts.append(f"<p>❌ Error: {out['error']}</p>") | |
| continue | |
| result_repr = out.get("result_repr") | |
| result_str = out.get("result_str", "") | |
| if isinstance(result_repr, list) and result_repr: | |
| headers = list(result_repr[0].keys()) | |
| thead_cells = "".join( | |
| f'<th style="text-align:{"right" if all(_is_numeric(r.get(h)) for r in result_repr) else "left"}">{h}</th>' | |
| for h in headers | |
| ) | |
| tbody_rows = [] | |
| for row in result_repr: | |
| cells = [] | |
| for h in headers: | |
| val = row.get(h, "") | |
| if _is_numeric(val): | |
| cells.append(f'<td style="text-align:right">{_format_number(val)}</td>') | |
| else: | |
| cells.append(f"<td>{val}</td>") | |
| tbody_rows.append(f"<tr>{''.join(cells)}</tr>") | |
| parts.append( | |
| f"<table>" | |
| f"<thead><tr>{thead_cells}</tr></thead>" | |
| f"<tbody>{''.join(tbody_rows)}</tbody>" | |
| f"</table>" | |
| ) | |
| elif isinstance(result_repr, dict): | |
| rows = "".join( | |
| f"<tr><td>{k}</td>" | |
| f'<td style="text-align:right">{_format_number(v) if _is_numeric(v) else v}</td></tr>' | |
| for k, v in result_repr.items() | |
| ) | |
| parts.append( | |
| f"<table>" | |
| f"<thead><tr><th>Key</th><th style='text-align:right'>Value</th></tr></thead>" | |
| f"<tbody>{rows}</tbody>" | |
| f"</table>" | |
| ) | |
| elif result_str: | |
| parts.append(f"<p><strong>{result_str}</strong></p>") | |
| else: | |
| parts.append("<p>No data returned.</p>") | |
| return "\n".join(parts) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # 8. MAIN CHATBOT PIPELINE | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def process_query(user_query, df, column_meta, verbose=False, project_name="") -> str: | |
| start_time = time.monotonic() | |
| fuzzy_results = {} | |
| intent = "" | |
| plan_text = "" | |
| llm_queries = [] | |
| final_answer = "" | |
| call1_tokens = {} | |
| call2_tokens = {} | |
| success = True | |
| error_code = None | |
| error_message = None | |
| try: | |
| if verbose: | |
| print("🔍 Running fuzzy search …") | |
| fuzzy_results = run_fuzzy_searches(user_query, df) | |
| if verbose: | |
| print("\n🧠 Asking LLM to plan queries …") | |
| plan = get_query_plan(user_query, fuzzy_results, column_meta, df) | |
| call1_tokens = get_last_token_usage() | |
| intent = plan.get("intent", "") | |
| plan_text = plan.get("plan", "") | |
| if verbose: | |
| print(f" Intent : {intent}") | |
| print(f" Plan : {plan_text}") | |
| if verbose: | |
| print("\n⚙️ Executing queries on DataFrame …") | |
| query_outputs = execute_plan(plan, df) | |
| llm_queries = [q.get("code", "") for q in plan.get("queries", [])] | |
| if verbose: | |
| print("\n💬 Generating final HTML answer …\n") | |
| final_answer = generate_final_answer(user_query, plan, query_outputs) | |
| call2_tokens = get_last_token_usage() | |
| except Exception as exc: | |
| success = False | |
| error_message = traceback.format_exc() | |
| error_code = type(exc).__name__ | |
| final_answer = f"<p>❌ An error occurred: {exc}</p>" | |
| if verbose: | |
| print(f"\n❌ Error during processing:\n{error_message}\n") | |
| finally: | |
| elapsed = time.monotonic() - start_time | |
| log_entry = build_log_entry( | |
| project_name = project_name, | |
| user_query = user_query, | |
| fuzzy_results = fuzzy_results, | |
| intent = intent, | |
| plan_text = plan_text, | |
| llm_queries = llm_queries, | |
| final_answer = final_answer, | |
| call1_tokens = call1_tokens, | |
| call2_tokens = call2_tokens, | |
| success = success, | |
| error_code = error_code, | |
| error_message = error_message, | |
| elapsed_seconds = elapsed, | |
| ) | |
| try: | |
| append_log(log_entry) | |
| except OSError as log_err: | |
| print(f"⚠️ Could not write log: {log_err}") | |
| if not success: | |
| raise RuntimeError(error_message) | |
| return final_answer | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # FASTAPI APP | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| app = FastAPI(title="RTS Data Chatbot API") | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| # Mount static files (CSS, JS, images) | |
| os.makedirs("static", exist_ok=True) | |
| app.mount("/static", StaticFiles(directory="static"), name="static") | |
| _df = None | |
| _column_meta = None | |
| _startup_error = None | |
| def startup_event(): | |
| global _df, _column_meta, _startup_error | |
| try: | |
| _df, _column_meta = load_data() | |
| print("✅ Data loaded at startup.") | |
| except Exception as e: | |
| _startup_error = str(e) | |
| print(f"❌ Failed to load data at startup: {e}") | |
| _df, _column_meta = None, None | |
| def index(): | |
| """Serve the web UI.""" | |
| with open("static/index.html", "r", encoding="utf-8") as f: | |
| return f.read() | |
| def health(): | |
| if _startup_error: | |
| return {"status": "error", "detail": _startup_error} | |
| if _df is None: | |
| return {"status": "loading"} | |
| return {"status": "ok", "rows": len(_df), "columns": len(_df.columns)} | |
| class ChatRequest(BaseModel): | |
| user_query : str | |
| secret_key : str | |
| project_name: str | |
| def chat_endpoint(request: ChatRequest): | |
| if SECRET_KEY and request.secret_key != SECRET_KEY: | |
| raise HTTPException(status_code=401, detail="Invalid secret key.") | |
| if PROJECT_NAMES and request.project_name not in PROJECT_NAMES: | |
| raise HTTPException(status_code=400, detail="Invalid project name.") | |
| if _df is None or _column_meta is None: | |
| detail = _startup_error or "Data is still loading. Please try again in a moment." | |
| raise HTTPException(status_code=503, detail=detail) | |
| try: | |
| answer = process_query( | |
| request.user_query, | |
| _df, | |
| _column_meta, | |
| verbose=False, | |
| project_name=request.project_name, | |
| ) | |
| return {"answer": answer, "format": "html"} | |
| except Exception as e: | |
| return JSONResponse( | |
| status_code=500, | |
| content={"error": str(e), "traceback": traceback.format_exc()}, | |
| ) | |