mirror of
http://100.103.83.12:3003/fegger/pv-agent.git
synced 2026-09-17 15:46:23 +00:00
feat(agent): add index bootstrap and answer feedback
This commit is contained in:
+179
@@ -0,0 +1,179 @@
|
||||
"""Persistente, strukturierte Interaktions- und Bewertungsprotokolle."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from .config import Config
|
||||
|
||||
SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS interactions (
|
||||
request_id TEXT PRIMARY KEY,
|
||||
created_at INTEGER NOT NULL,
|
||||
question TEXT,
|
||||
answer TEXT,
|
||||
status TEXT NOT NULL,
|
||||
verified INTEGER NOT NULL,
|
||||
refused INTEGER NOT NULL,
|
||||
citations_json TEXT NOT NULL,
|
||||
sources_json TEXT NOT NULL,
|
||||
conflicts_json TEXT NOT NULL,
|
||||
planned_queries_json TEXT NOT NULL,
|
||||
model TEXT NOT NULL,
|
||||
latency_ms INTEGER NOT NULL,
|
||||
n_context INTEGER NOT NULL,
|
||||
regenerations INTEGER NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS ratings (
|
||||
request_id TEXT PRIMARY KEY,
|
||||
created_at INTEGER NOT NULL,
|
||||
updated_at INTEGER NOT NULL,
|
||||
rating TEXT NOT NULL CHECK (rating IN ('up', 'down')),
|
||||
feedback TEXT,
|
||||
FOREIGN KEY (request_id) REFERENCES interactions(request_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS interactions_created_at_idx
|
||||
ON interactions(created_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS ratings_updated_at_idx
|
||||
ON ratings(updated_at DESC);
|
||||
"""
|
||||
|
||||
|
||||
class AuditStore:
|
||||
def __init__(self, cfg: Config):
|
||||
self.cfg = cfg
|
||||
path = Path(cfg.audit_db_path)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
self._lock = threading.RLock()
|
||||
self._con = sqlite3.connect(path, check_same_thread=False)
|
||||
self._con.row_factory = sqlite3.Row
|
||||
self._con.execute("PRAGMA foreign_keys = ON")
|
||||
self._con.execute("PRAGMA journal_mode = WAL")
|
||||
self._con.executescript(SCHEMA)
|
||||
self._delete_expired()
|
||||
|
||||
def close(self) -> None:
|
||||
with self._lock:
|
||||
self._con.close()
|
||||
|
||||
def _delete_expired(self) -> None:
|
||||
if self.cfg.audit_retention_days <= 0:
|
||||
return
|
||||
cutoff = int(time.time()) - self.cfg.audit_retention_days * 86_400
|
||||
with self._lock:
|
||||
self._con.execute("DELETE FROM interactions WHERE created_at < ?", (cutoff,))
|
||||
self._con.commit()
|
||||
|
||||
def record_interaction(self, payload: dict) -> None:
|
||||
now = int(time.time())
|
||||
include_content = self.cfg.audit_log_content
|
||||
question = payload.get("question") if include_content else None
|
||||
answer = payload.get("answer") if include_content else None
|
||||
sources = payload.get("sources", []) if include_content else []
|
||||
conflicts = payload.get("conflicts", []) if include_content else []
|
||||
planned_queries = payload.get("planned_queries", []) if include_content else []
|
||||
values = (
|
||||
payload["request_id"],
|
||||
now,
|
||||
question,
|
||||
answer,
|
||||
payload["status"],
|
||||
int(bool(payload["verified"])),
|
||||
int(bool(payload["refused"])),
|
||||
json.dumps(payload.get("citations", []), ensure_ascii=False),
|
||||
json.dumps(sources, ensure_ascii=False),
|
||||
json.dumps(conflicts, ensure_ascii=False),
|
||||
json.dumps(planned_queries, ensure_ascii=False),
|
||||
payload.get("model", ""),
|
||||
int(payload.get("latency_ms", 0)),
|
||||
int(payload.get("n_context", 0)),
|
||||
int(payload.get("regenerations", 0)),
|
||||
)
|
||||
with self._lock:
|
||||
self._con.execute(
|
||||
"INSERT OR REPLACE INTO interactions("
|
||||
"request_id, created_at, question, answer, status, verified, refused, "
|
||||
"citations_json, sources_json, conflicts_json, planned_queries_json, "
|
||||
"model, latency_ms, n_context, regenerations"
|
||||
") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||
values,
|
||||
)
|
||||
self._con.commit()
|
||||
if self.cfg.audit_stdout:
|
||||
if include_content:
|
||||
event = {"event": "agent_interaction", "created_at": now, **payload}
|
||||
else:
|
||||
event = {
|
||||
"event": "agent_interaction",
|
||||
"created_at": now,
|
||||
"request_id": payload["request_id"],
|
||||
"status": payload["status"],
|
||||
"verified": payload["verified"],
|
||||
"refused": payload["refused"],
|
||||
"citations": payload.get("citations", []),
|
||||
"model": payload.get("model", ""),
|
||||
"latency_ms": payload.get("latency_ms", 0),
|
||||
"n_context": payload.get("n_context", 0),
|
||||
"regenerations": payload.get("regenerations", 0),
|
||||
}
|
||||
print("AUDIT " + json.dumps(event, ensure_ascii=False), flush=True)
|
||||
|
||||
def record_rating(self, request_id: str, rating: str, feedback: str | None) -> None:
|
||||
now = int(time.time())
|
||||
stored_feedback = feedback if self.cfg.audit_log_content else None
|
||||
with self._lock:
|
||||
exists = self._con.execute(
|
||||
"SELECT 1 FROM interactions WHERE request_id = ?", (request_id,)
|
||||
).fetchone()
|
||||
if exists is None:
|
||||
raise KeyError(request_id)
|
||||
self._con.execute(
|
||||
"INSERT INTO ratings(request_id, created_at, updated_at, rating, feedback) "
|
||||
"VALUES (?, ?, ?, ?, ?) "
|
||||
"ON CONFLICT(request_id) DO UPDATE SET "
|
||||
"updated_at=excluded.updated_at, rating=excluded.rating, "
|
||||
"feedback=excluded.feedback",
|
||||
(request_id, now, now, rating, stored_feedback),
|
||||
)
|
||||
self._con.commit()
|
||||
if self.cfg.audit_stdout:
|
||||
print(
|
||||
"AUDIT "
|
||||
+ json.dumps(
|
||||
{
|
||||
"event": "agent_rating",
|
||||
"created_at": now,
|
||||
"request_id": request_id,
|
||||
"rating": rating,
|
||||
"feedback": stored_feedback,
|
||||
},
|
||||
ensure_ascii=False,
|
||||
),
|
||||
flush=True,
|
||||
)
|
||||
|
||||
def recent(self, limit: int = 20) -> list[dict]:
|
||||
with self._lock:
|
||||
rows = self._con.execute(
|
||||
"SELECT i.*, r.rating, r.feedback, r.updated_at AS rating_updated_at "
|
||||
"FROM interactions i LEFT JOIN ratings r USING(request_id) "
|
||||
"ORDER BY i.created_at DESC LIMIT ?",
|
||||
(limit,),
|
||||
).fetchall()
|
||||
out = []
|
||||
for row in rows:
|
||||
item = dict(row)
|
||||
for field in (
|
||||
"citations_json",
|
||||
"sources_json",
|
||||
"conflicts_json",
|
||||
"planned_queries_json",
|
||||
):
|
||||
item[field.removesuffix("_json")] = json.loads(item.pop(field))
|
||||
item["verified"] = bool(item["verified"])
|
||||
item["refused"] = bool(item["refused"])
|
||||
out.append(item)
|
||||
return out
|
||||
Reference in New Issue
Block a user