Spaces:
Running
Running
File size: 4,200 Bytes
9e50d20 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | """
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"],
))
|