409 lines
14 KiB
Python
409 lines
14 KiB
Python
"""SQLAlchemy-backed assistant-answer feedback storage.
|
|
|
|
Each method acquires its own short-lived session. Writes are idempotent on a
|
|
SHA-256 target key derived from user/thread/message/agent; the digest avoids
|
|
oversized compound indexes on legacy MySQL utf8mb4 deployments.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import case, delete, func, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from deerflow.persistence.feedback.model import FeedbackRow
|
|
from deerflow.persistence.feedback.scoring import FEEDBACK_REASONS, calculate_answer_quality_score
|
|
from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
|
|
|
|
|
|
class FeedbackRepository:
|
|
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
|
|
self._sf = session_factory
|
|
|
|
@staticmethod
|
|
def _row_to_dict(row: FeedbackRow) -> dict[str, Any]:
|
|
data = row.to_dict()
|
|
for key in ("created_at", "updated_at"):
|
|
value = data.get(key)
|
|
if isinstance(value, datetime):
|
|
data[key] = value.isoformat()
|
|
data["reasons"] = list(data.get("reasons") or [])
|
|
data["detail"] = data.get("comment")
|
|
return data
|
|
|
|
@staticmethod
|
|
def make_target_key(
|
|
*,
|
|
user_id: str | None,
|
|
thread_id: str,
|
|
run_id: str,
|
|
message_id: str | None,
|
|
agent_id: str,
|
|
) -> str:
|
|
"""Build the portable idempotency identity for one answer rating.
|
|
|
|
Historical run-level callers have no message id. Their deterministic
|
|
``run:<id>`` surrogate preserves the old one-rating-per-run behavior.
|
|
"""
|
|
|
|
message_key = message_id or f"run:{run_id}"
|
|
raw = "\x1f".join(
|
|
(
|
|
user_id or "",
|
|
thread_id,
|
|
message_key,
|
|
agent_id or "lead_agent",
|
|
)
|
|
)
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
|
|
|
@staticmethod
|
|
def _normalize_reasons(reasons: list[str] | None) -> list[str]:
|
|
allowed = set(FEEDBACK_REASONS)
|
|
normalized: list[str] = []
|
|
for reason in reasons or []:
|
|
value = str(reason).strip()
|
|
if value in allowed and value not in normalized:
|
|
normalized.append(value)
|
|
return normalized
|
|
|
|
async def create(
|
|
self,
|
|
*,
|
|
run_id: str,
|
|
thread_id: str,
|
|
rating: int,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
message_id: str | None = None,
|
|
agent_id: str = "lead_agent",
|
|
reasons: list[str] | None = None,
|
|
comment: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create or update the same answer target idempotently."""
|
|
|
|
return await self.upsert(
|
|
run_id=run_id,
|
|
thread_id=thread_id,
|
|
rating=rating,
|
|
user_id=user_id,
|
|
message_id=message_id,
|
|
agent_id=agent_id,
|
|
reasons=reasons,
|
|
comment=comment,
|
|
)
|
|
|
|
async def get(
|
|
self,
|
|
feedback_id: str,
|
|
*,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
) -> dict[str, Any] | None:
|
|
resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.get")
|
|
async with self._sf() as session:
|
|
row = await session.get(FeedbackRow, feedback_id)
|
|
if row is None:
|
|
return None
|
|
if resolved_user_id is not None and row.user_id != resolved_user_id:
|
|
return None
|
|
return self._row_to_dict(row)
|
|
|
|
async def get_by_target(
|
|
self,
|
|
*,
|
|
thread_id: str,
|
|
run_id: str,
|
|
message_id: str | None,
|
|
agent_id: str,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
) -> dict[str, Any] | None:
|
|
resolved_user_id = resolve_user_id(
|
|
user_id,
|
|
method_name="FeedbackRepository.get_by_target",
|
|
)
|
|
target_key = self.make_target_key(
|
|
user_id=resolved_user_id,
|
|
thread_id=thread_id,
|
|
run_id=run_id,
|
|
message_id=message_id,
|
|
agent_id=agent_id,
|
|
)
|
|
async with self._sf() as session:
|
|
row = await session.scalar(
|
|
select(FeedbackRow).where(FeedbackRow.target_key == target_key)
|
|
)
|
|
return self._row_to_dict(row) if row is not None else None
|
|
|
|
async def list_by_run(
|
|
self,
|
|
thread_id: str,
|
|
run_id: str,
|
|
*,
|
|
limit: int = 100,
|
|
message_id: str | None = None,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
) -> list[dict[str, Any]]:
|
|
resolved_user_id = resolve_user_id(
|
|
user_id,
|
|
method_name="FeedbackRepository.list_by_run",
|
|
)
|
|
stmt = select(FeedbackRow).where(
|
|
FeedbackRow.thread_id == thread_id,
|
|
FeedbackRow.run_id == run_id,
|
|
)
|
|
if message_id is not None:
|
|
stmt = stmt.where(FeedbackRow.message_id == message_id)
|
|
if resolved_user_id is not None:
|
|
stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
|
|
stmt = stmt.order_by(FeedbackRow.created_at.asc()).limit(limit)
|
|
async with self._sf() as session:
|
|
result = await session.execute(stmt)
|
|
return [self._row_to_dict(row) for row in result.scalars()]
|
|
|
|
async def list_by_thread(
|
|
self,
|
|
thread_id: str,
|
|
*,
|
|
limit: int = 1000,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
) -> list[dict[str, Any]]:
|
|
resolved_user_id = resolve_user_id(
|
|
user_id,
|
|
method_name="FeedbackRepository.list_by_thread",
|
|
)
|
|
stmt = select(FeedbackRow).where(FeedbackRow.thread_id == thread_id)
|
|
if resolved_user_id is not None:
|
|
stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
|
|
stmt = stmt.order_by(FeedbackRow.created_at.asc()).limit(limit)
|
|
async with self._sf() as session:
|
|
result = await session.execute(stmt)
|
|
return [self._row_to_dict(row) for row in result.scalars()]
|
|
|
|
async def list_by_message_ids(
|
|
self,
|
|
*,
|
|
thread_id: str,
|
|
message_ids: list[str],
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
) -> list[dict[str, Any]]:
|
|
"""Load current-user feedback for a bounded visible-message batch."""
|
|
|
|
if not message_ids:
|
|
return []
|
|
resolved_user_id = resolve_user_id(
|
|
user_id,
|
|
method_name="FeedbackRepository.list_by_message_ids",
|
|
)
|
|
stmt = select(FeedbackRow).where(
|
|
FeedbackRow.thread_id == thread_id,
|
|
FeedbackRow.message_id.in_(message_ids),
|
|
)
|
|
if resolved_user_id is not None:
|
|
stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
|
|
async with self._sf() as session:
|
|
result = await session.execute(stmt)
|
|
return [self._row_to_dict(row) for row in result.scalars()]
|
|
|
|
async def delete(
|
|
self,
|
|
feedback_id: str,
|
|
*,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
) -> bool:
|
|
resolved_user_id = resolve_user_id(
|
|
user_id,
|
|
method_name="FeedbackRepository.delete",
|
|
)
|
|
async with self._sf() as session:
|
|
row = await session.get(FeedbackRow, feedback_id)
|
|
if row is None:
|
|
return False
|
|
if resolved_user_id is not None and row.user_id != resolved_user_id:
|
|
return False
|
|
await session.delete(row)
|
|
await session.commit()
|
|
return True
|
|
|
|
async def upsert(
|
|
self,
|
|
*,
|
|
run_id: str,
|
|
thread_id: str,
|
|
rating: int,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
message_id: str | None = None,
|
|
agent_id: str = "lead_agent",
|
|
reasons: list[str] | None = None,
|
|
comment: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create or update feedback for one user/thread/message/agent target."""
|
|
|
|
if rating not in (1, -1):
|
|
raise ValueError(f"rating must be +1 or -1, got {rating}")
|
|
resolved_user_id = resolve_user_id(
|
|
user_id,
|
|
method_name="FeedbackRepository.upsert",
|
|
)
|
|
normalized_agent_id = str(agent_id or "lead_agent")[:128]
|
|
normalized_reasons = self._normalize_reasons(reasons)
|
|
target_key = self.make_target_key(
|
|
user_id=resolved_user_id,
|
|
thread_id=thread_id,
|
|
run_id=run_id,
|
|
message_id=message_id,
|
|
agent_id=normalized_agent_id,
|
|
)
|
|
|
|
# The unique digest handles cross-worker races. On a simultaneous
|
|
# insert, retry as an update in a fresh transaction.
|
|
for attempt in range(2):
|
|
now = datetime.now(UTC)
|
|
async with self._sf() as session:
|
|
row = await session.scalar(
|
|
select(FeedbackRow).where(FeedbackRow.target_key == target_key)
|
|
)
|
|
if row is None:
|
|
row = FeedbackRow(
|
|
feedback_id=str(uuid.uuid4()),
|
|
run_id=run_id,
|
|
thread_id=thread_id,
|
|
user_id=resolved_user_id,
|
|
message_id=message_id,
|
|
agent_id=normalized_agent_id,
|
|
target_key=target_key,
|
|
rating=rating,
|
|
reasons=normalized_reasons,
|
|
comment=comment,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(row)
|
|
else:
|
|
row.run_id = run_id
|
|
row.rating = rating
|
|
row.reasons = normalized_reasons
|
|
row.comment = comment
|
|
row.updated_at = now
|
|
try:
|
|
await session.commit()
|
|
except IntegrityError:
|
|
await session.rollback()
|
|
if attempt == 0:
|
|
continue
|
|
raise
|
|
# Runtime factories use expire_on_commit=False. Avoid a
|
|
# post-commit refresh, unsafe behind a read replica.
|
|
return self._row_to_dict(row)
|
|
raise RuntimeError("unreachable feedback upsert retry state")
|
|
|
|
async def delete_by_run(
|
|
self,
|
|
*,
|
|
thread_id: str,
|
|
run_id: str,
|
|
message_id: str | None = None,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
) -> bool:
|
|
"""Delete this user's feedback for a run or one message in the run."""
|
|
|
|
resolved_user_id = resolve_user_id(
|
|
user_id,
|
|
method_name="FeedbackRepository.delete_by_run",
|
|
)
|
|
conditions = [
|
|
FeedbackRow.thread_id == thread_id,
|
|
FeedbackRow.run_id == run_id,
|
|
FeedbackRow.user_id == resolved_user_id,
|
|
]
|
|
if message_id is not None:
|
|
conditions.append(FeedbackRow.message_id == message_id)
|
|
async with self._sf() as session:
|
|
result = await session.execute(delete(FeedbackRow).where(*conditions))
|
|
await session.commit()
|
|
return bool(result.rowcount)
|
|
|
|
async def delete_by_target(
|
|
self,
|
|
*,
|
|
thread_id: str,
|
|
run_id: str,
|
|
message_id: str,
|
|
agent_id: str,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
) -> bool:
|
|
resolved_user_id = resolve_user_id(
|
|
user_id,
|
|
method_name="FeedbackRepository.delete_by_target",
|
|
)
|
|
target_key = self.make_target_key(
|
|
user_id=resolved_user_id,
|
|
thread_id=thread_id,
|
|
run_id=run_id,
|
|
message_id=message_id,
|
|
agent_id=agent_id,
|
|
)
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
delete(FeedbackRow).where(FeedbackRow.target_key == target_key)
|
|
)
|
|
await session.commit()
|
|
return bool(result.rowcount)
|
|
|
|
async def list_by_thread_grouped(
|
|
self,
|
|
thread_id: str,
|
|
*,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
) -> dict[str, dict[str, Any]]:
|
|
"""Legacy run-keyed view, retaining the newest row for each run."""
|
|
|
|
records = await self.list_by_thread(thread_id, user_id=user_id)
|
|
return {record["run_id"]: record for record in records}
|
|
|
|
async def aggregate_by_run(self, thread_id: str, run_id: str) -> dict[str, Any]:
|
|
"""Aggregate feedback stats for a run using database-side counting."""
|
|
|
|
stmt = select(
|
|
func.count().label("total"),
|
|
func.coalesce(
|
|
func.sum(case((FeedbackRow.rating == 1, 1), else_=0)),
|
|
0,
|
|
).label("positive"),
|
|
func.coalesce(
|
|
func.sum(case((FeedbackRow.rating == -1, 1), else_=0)),
|
|
0,
|
|
).label("negative"),
|
|
).where(
|
|
FeedbackRow.thread_id == thread_id,
|
|
FeedbackRow.run_id == run_id,
|
|
)
|
|
async with self._sf() as session:
|
|
row = (await session.execute(stmt)).one()
|
|
return {
|
|
"run_id": run_id,
|
|
"total": row.total,
|
|
"positive": row.positive,
|
|
"negative": row.negative,
|
|
}
|
|
|
|
async def aggregate_by_agent(self, agent_id: str) -> dict[str, Any]:
|
|
"""Return the documented user-feedback answer-quality score."""
|
|
|
|
normalized_agent_id = str(agent_id or "lead_agent")[:128]
|
|
stmt = select(FeedbackRow.rating, FeedbackRow.reasons).where(
|
|
FeedbackRow.agent_id == normalized_agent_id
|
|
)
|
|
async with self._sf() as session:
|
|
rows = (await session.execute(stmt)).all()
|
|
return calculate_answer_quality_score(
|
|
(
|
|
{"rating": row.rating, "reasons": list(row.reasons or [])}
|
|
for row in rows
|
|
),
|
|
agent_id=normalized_agent_id,
|
|
)
|