53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
"""Helpers for recording agent interactions to disk."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
DEFAULT_LOG_PATH = Path("db_agent/logs/query_log.jsonl")
|
|
|
|
|
|
def log_interaction(
|
|
*,
|
|
question: str,
|
|
generated_sql: str,
|
|
summary: str,
|
|
sanitized_sql: Optional[str] = None,
|
|
row_count: Optional[int] = None,
|
|
execution_status: Optional[str] = None,
|
|
execution_error: Optional[str] = None,
|
|
raw_response: Optional[str] = None,
|
|
user_feedback: Optional[str] = None,
|
|
log_path: Path | None = None,
|
|
metadata: Optional[dict[str, Any]] = None,
|
|
) -> Path:
|
|
"""Append a single interaction entry to a JSONL log."""
|
|
|
|
path = log_path or DEFAULT_LOG_PATH
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
entry = {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"question": question,
|
|
"generated_sql": generated_sql,
|
|
"summary": summary,
|
|
"sanitized_sql": sanitized_sql,
|
|
"row_count": row_count,
|
|
"execution_status": execution_status,
|
|
"execution_error": execution_error,
|
|
"raw_response": raw_response,
|
|
"user_feedback": user_feedback,
|
|
}
|
|
if metadata:
|
|
entry["metadata"] = metadata
|
|
# Remove None values for cleaner logs
|
|
entry = {key: value for key, value in entry.items() if value is not None}
|
|
|
|
with path.open("a", encoding="utf-8") as handle:
|
|
handle.write(json.dumps(entry, ensure_ascii=False) + "\n")
|
|
|
|
return path
|