Spaces:
Running
Running
| """ | |
| monitor.py | |
| Lightweight monitoring logger for DocMind AI. | |
| Logs every query, retrieval, LLM response, latency, and errors | |
| to a local JSONL file readable from HuggingFace Logs tab. | |
| """ | |
| import json | |
| import time | |
| import os | |
| from datetime import datetime, timezone | |
| LOG_FILE = "/tmp/docmind_logs.jsonl" | |
| def log_query( | |
| question: str, | |
| answer: str, | |
| sources: list, | |
| latency_ms: float, | |
| model_used: str = "", | |
| chunk_count: int = 0, | |
| error: str = "", | |
| ): | |
| """Log a single query event.""" | |
| entry = { | |
| "timestamp": datetime.now(timezone.utc).isoformat(), | |
| "event": "query", | |
| "question": question[:200], # truncate for log size | |
| "answer_len": len(answer), | |
| "sources": sources, | |
| "latency_ms": round(latency_ms, 1), | |
| "model": model_used, | |
| "chunk_count": chunk_count, | |
| "error": error, | |
| "success": not bool(error), | |
| } | |
| _write(entry) | |
| _print_summary(entry) | |
| def log_ingestion(filename: str, chunk_count: int, latency_ms: float, error: str = ""): | |
| """Log a document ingestion event.""" | |
| entry = { | |
| "timestamp": datetime.now(timezone.utc).isoformat(), | |
| "event": "ingestion", | |
| "filename": filename, | |
| "chunk_count": chunk_count, | |
| "latency_ms": round(latency_ms, 1), | |
| "error": error, | |
| "success": not bool(error), | |
| } | |
| _write(entry) | |
| _print_summary(entry) | |
| def log_startup(): | |
| """Log app startup.""" | |
| entry = { | |
| "timestamp": datetime.now(timezone.utc).isoformat(), | |
| "event": "startup", | |
| "message": "DocMind AI started", | |
| } | |
| _write(entry) | |
| print("[DocMind] App started at {}".format(entry["timestamp"])) | |
| def get_stats() -> dict: | |
| """Read log file and compute summary stats.""" | |
| if not os.path.exists(LOG_FILE): | |
| return { | |
| "total_queries": 0, | |
| "success_rate": 0.0, | |
| "avg_latency_ms": 0.0, | |
| "total_ingestions": 0, | |
| "errors": [], | |
| } | |
| queries = [] | |
| ingestions = [] | |
| errors = [] | |
| with open(LOG_FILE, "r") as f: | |
| for line in f: | |
| try: | |
| entry = json.loads(line.strip()) | |
| if entry["event"] == "query": | |
| queries.append(entry) | |
| if entry.get("error"): | |
| errors.append(entry) | |
| elif entry["event"] == "ingestion": | |
| ingestions.append(entry) | |
| except Exception: | |
| continue | |
| success_count = sum(1 for q in queries if q.get("success")) | |
| latencies = [q["latency_ms"] for q in queries if q.get("latency_ms")] | |
| return { | |
| "total_queries": len(queries), | |
| "success_rate": round(success_count / len(queries) * 100, 1) if queries else 0.0, | |
| "avg_latency_ms": round(sum(latencies) / len(latencies), 1) if latencies else 0.0, | |
| "min_latency_ms": round(min(latencies), 1) if latencies else 0.0, | |
| "max_latency_ms": round(max(latencies), 1) if latencies else 0.0, | |
| "total_ingestions": len(ingestions), | |
| "total_errors": len(errors), | |
| "recent_errors": [e.get("error", "") for e in errors[-3:]], | |
| } | |
| def _write(entry: dict): | |
| try: | |
| with open(LOG_FILE, "a") as f: | |
| f.write(json.dumps(entry) + "\n") | |
| except Exception as e: | |
| print("[DocMind Monitor] Failed to write log: {}".format(e)) | |
| def _print_summary(entry: dict): | |
| """Print to stdout so it shows in HuggingFace Logs tab.""" | |
| if entry["event"] == "query": | |
| status = "OK" if entry["success"] else "ERROR" | |
| print("[DocMind][{}] Q: '{}...' | Latency: {}ms | Model: {} | AnswerLen: {}".format( | |
| status, | |
| entry["question"][:50], | |
| entry["latency_ms"], | |
| entry.get("model", "unknown"), | |
| entry["answer_len"], | |
| )) | |
| elif entry["event"] == "ingestion": | |
| print("[DocMind][INGEST] File: {} | Chunks: {} | Latency: {}ms".format( | |
| entry["filename"], | |
| entry["chunk_count"], | |
| entry["latency_ms"], | |
| )) | |