386 lines
13 KiB
Python
386 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
from datetime import datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
import sqlalchemy as sa
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
from fastapi import HTTPException
|
|
from pydantic import ValidationError
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
from starlette.requests import Request
|
|
|
|
from app.gateway.routers import feedback as feedback_router
|
|
from app.gateway.routers.feedback import (
|
|
FeedbackWriteRequest,
|
|
_find_ai_message_events,
|
|
)
|
|
from deerflow.persistence.feedback.model import FeedbackRow
|
|
from deerflow.persistence.feedback.scoring import calculate_answer_quality_score
|
|
from deerflow.persistence.feedback.sql import FeedbackRepository
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def feedback_repository(tmp_path):
|
|
engine = create_async_engine(
|
|
f"sqlite+aiosqlite:///{tmp_path / 'feedback.db'}",
|
|
)
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(FeedbackRow.__table__.create)
|
|
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
try:
|
|
yield FeedbackRepository(session_factory), session_factory
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
def test_feedback_request_enforces_reason_contract() -> None:
|
|
positive = FeedbackWriteRequest(rating=1)
|
|
assert positive.reasons == []
|
|
|
|
negative = FeedbackWriteRequest(
|
|
rating=-1,
|
|
reasons=["incorrect_or_incomplete", "incorrect_or_incomplete"],
|
|
detail="事实不准确",
|
|
)
|
|
assert [reason.value for reason in negative.reasons] == [
|
|
"incorrect_or_incomplete"
|
|
]
|
|
assert negative.normalized_detail() == "事实不准确"
|
|
|
|
with pytest.raises(ValidationError):
|
|
FeedbackWriteRequest(rating=-1)
|
|
with pytest.raises(ValidationError):
|
|
FeedbackWriteRequest(rating=1, reasons=["other"])
|
|
|
|
|
|
def test_answer_quality_score_is_null_without_real_evidence() -> None:
|
|
result = calculate_answer_quality_score([], agent_id="agent-a")
|
|
|
|
assert result["score"] is None
|
|
assert result["task_score"] is None
|
|
assert result["score_band"]["key"] == "insufficient_data"
|
|
assert result["confidence"] == 0
|
|
assert result["confidence_level"] == "none"
|
|
assert all(item["score"] is None for item in result["dimensions"].values())
|
|
|
|
|
|
def test_answer_quality_score_maps_only_documented_reasons() -> None:
|
|
result = calculate_answer_quality_score(
|
|
[
|
|
{"rating": 1, "reasons": []},
|
|
{"rating": -1, "reasons": ["incorrect_or_incomplete"]},
|
|
{"rating": -1, "reasons": ["style_or_tone"]},
|
|
],
|
|
agent_id="agent-a",
|
|
)
|
|
|
|
assert result["score"] == 69.2
|
|
assert result["task_success_rate"] == pytest.approx(1 / 3, abs=0.0001)
|
|
assert result["confidence"] == pytest.approx(3 / 13, abs=0.0001)
|
|
assert result["dimensions"]["correctness"]["negative_count"] == 1
|
|
assert result["dimensions"]["completeness"]["negative_count"] == 1
|
|
assert result["dimensions"]["evidence_grounding"]["negative_count"] == 1
|
|
assert result["dimensions"]["relevance"]["negative_count"] == 0
|
|
assert result["dimensions"]["relevance"]["evidence_count"] == 1
|
|
assert result["reason_breakdown"]["style_or_tone"] == {
|
|
"count": 1,
|
|
"rate": 0.5,
|
|
}
|
|
|
|
|
|
class _AgentScoreRepository:
|
|
def __init__(self) -> None:
|
|
self.agent_ids: list[str] = []
|
|
|
|
async def aggregate_by_agent(self, agent_id: str) -> dict:
|
|
self.agent_ids.append(agent_id)
|
|
return calculate_answer_quality_score([], agent_id=agent_id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_feedback_score_rejects_non_admin(
|
|
monkeypatch,
|
|
) -> None:
|
|
async def current_user(_request):
|
|
return SimpleNamespace(system_role="user")
|
|
|
|
repository = _AgentScoreRepository()
|
|
request = SimpleNamespace(
|
|
app=SimpleNamespace(state=SimpleNamespace(feedback_repo=repository))
|
|
)
|
|
monkeypatch.setattr(feedback_router, "get_optional_user_from_request", current_user)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await feedback_router.get_agent_feedback_score("agent-a", request)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
assert repository.agent_ids == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_feedback_score_allows_admin(monkeypatch) -> None:
|
|
async def current_user(_request):
|
|
return SimpleNamespace(system_role="admin")
|
|
|
|
repository = _AgentScoreRepository()
|
|
request = SimpleNamespace(
|
|
app=SimpleNamespace(state=SimpleNamespace(feedback_repo=repository))
|
|
)
|
|
monkeypatch.setattr(feedback_router, "get_optional_user_from_request", current_user)
|
|
|
|
result = await feedback_router.get_agent_feedback_score(" agent-a ", request)
|
|
|
|
assert result["agent_id"] == "agent-a"
|
|
assert repository.agent_ids == ["agent-a"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_repository_upserts_per_user_thread_message_and_agent(
|
|
feedback_repository,
|
|
) -> None:
|
|
repository, session_factory = feedback_repository
|
|
|
|
first = await repository.upsert(
|
|
run_id="run-1",
|
|
thread_id="thread-1",
|
|
message_id="message-1",
|
|
agent_id="agent-a",
|
|
user_id="user-a",
|
|
rating=-1,
|
|
reasons=["incorrect_or_incomplete"],
|
|
comment="missing evidence",
|
|
)
|
|
updated = await repository.upsert(
|
|
run_id="run-1",
|
|
thread_id="thread-1",
|
|
message_id="message-1",
|
|
agent_id="agent-a",
|
|
user_id="user-a",
|
|
rating=1,
|
|
reasons=[],
|
|
comment=None,
|
|
)
|
|
second_answer = await repository.upsert(
|
|
run_id="run-1",
|
|
thread_id="thread-1",
|
|
message_id="message-2",
|
|
agent_id="agent-a",
|
|
user_id="user-a",
|
|
rating=-1,
|
|
reasons=["unmet_expectations"],
|
|
)
|
|
other_user = await repository.upsert(
|
|
run_id="run-1",
|
|
thread_id="thread-1",
|
|
message_id="message-1",
|
|
agent_id="agent-a",
|
|
user_id="user-b",
|
|
rating=-1,
|
|
reasons=["other"],
|
|
)
|
|
|
|
assert updated["feedback_id"] == first["feedback_id"]
|
|
assert updated["rating"] == 1
|
|
assert second_answer["feedback_id"] != first["feedback_id"]
|
|
assert other_user["feedback_id"] != first["feedback_id"]
|
|
async with session_factory() as session:
|
|
assert await session.scalar(select(func.count()).select_from(FeedbackRow)) == 3
|
|
|
|
score = await repository.aggregate_by_agent("agent-a")
|
|
assert score["sample_count"] == 3
|
|
assert score["positive_count"] == 1
|
|
assert score["negative_count"] == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_run_level_target_remains_idempotent(feedback_repository) -> None:
|
|
repository, _ = feedback_repository
|
|
first = await repository.upsert(
|
|
run_id="run-legacy",
|
|
thread_id="thread-legacy",
|
|
agent_id="lead_agent",
|
|
user_id="user-a",
|
|
rating=1,
|
|
)
|
|
updated = await repository.upsert(
|
|
run_id="run-legacy",
|
|
thread_id="thread-legacy",
|
|
agent_id="lead_agent",
|
|
user_id="user-a",
|
|
rating=-1,
|
|
reasons=["other"],
|
|
)
|
|
assert updated["feedback_id"] == first["feedback_id"]
|
|
|
|
|
|
class _FakeEventStore:
|
|
def __init__(self, rows: list[dict]) -> None:
|
|
self.rows = rows
|
|
|
|
async def list_messages(
|
|
self,
|
|
thread_id: str,
|
|
*,
|
|
limit: int,
|
|
before_seq: int | None,
|
|
) -> list[dict]:
|
|
rows = [row for row in self.rows if row["thread_id"] == thread_id]
|
|
if before_seq is not None:
|
|
rows = [row for row in rows if row["seq"] < before_seq]
|
|
return rows[-limit:]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_batch_message_resolver_uses_serialized_ai_message_id() -> None:
|
|
event_store = _FakeEventStore(
|
|
[
|
|
{
|
|
"thread_id": "thread-1",
|
|
"run_id": "run-1",
|
|
"seq": 1,
|
|
"event_type": "llm.tool.result",
|
|
"content": {"id": "tool-1"},
|
|
},
|
|
{
|
|
"thread_id": "thread-1",
|
|
"run_id": "run-1",
|
|
"seq": 2,
|
|
"event_type": "llm.ai.response",
|
|
"content": {"id": "answer-1", "type": "ai"},
|
|
},
|
|
]
|
|
)
|
|
app = SimpleNamespace(state=SimpleNamespace(run_event_store=event_store))
|
|
request = Request({"type": "http", "app": app})
|
|
|
|
found = await _find_ai_message_events(
|
|
request,
|
|
thread_id="thread-1",
|
|
message_ids={"answer-1", "tool-1", "missing"},
|
|
)
|
|
|
|
assert set(found) == {"answer-1"}
|
|
assert found["answer-1"]["run_id"] == "run-1"
|
|
|
|
|
|
def test_answer_feedback_migration_upgrades_and_downgrades_sqlite(
|
|
tmp_path,
|
|
monkeypatch,
|
|
) -> None:
|
|
engine = sa.create_engine(f"sqlite:///{tmp_path / 'migration.db'}")
|
|
metadata = sa.MetaData()
|
|
runs = sa.Table(
|
|
"runs",
|
|
metadata,
|
|
sa.Column("run_id", sa.String(64), primary_key=True),
|
|
sa.Column("assistant_id", sa.String(128)),
|
|
)
|
|
feedback = sa.Table(
|
|
"feedback",
|
|
metadata,
|
|
sa.Column("feedback_id", sa.String(64), primary_key=True),
|
|
sa.Column("run_id", sa.String(64), nullable=False),
|
|
sa.Column("thread_id", sa.String(64), nullable=False),
|
|
sa.Column("user_id", sa.String(64)),
|
|
sa.Column("message_id", sa.String(64)),
|
|
sa.Column("rating", sa.Integer, nullable=False),
|
|
sa.Column("comment", sa.Text),
|
|
sa.Column("created_at", sa.DateTime, nullable=False),
|
|
sa.Index(
|
|
"uq_feedback_thread_run_user",
|
|
"thread_id",
|
|
"run_id",
|
|
"user_id",
|
|
unique=True,
|
|
),
|
|
)
|
|
metadata.create_all(engine)
|
|
|
|
migration = importlib.import_module(
|
|
"deerflow.persistence.migrations.versions."
|
|
"20260916_02_answer_feedback_scoring"
|
|
)
|
|
with engine.begin() as connection:
|
|
connection.execute(
|
|
runs.insert().values(run_id="run-1", assistant_id="agent-a")
|
|
)
|
|
connection.execute(
|
|
feedback.insert().values(
|
|
feedback_id="feedback-1",
|
|
run_id="run-1",
|
|
thread_id="thread-1",
|
|
user_id="user-1",
|
|
message_id="message-1",
|
|
rating=-1,
|
|
comment="bad",
|
|
created_at=datetime(2026, 9, 16, 12, 0, 0),
|
|
)
|
|
)
|
|
connection.execute(
|
|
feedback.insert(),
|
|
[
|
|
{
|
|
"feedback_id": "anonymous-1",
|
|
"run_id": "run-1",
|
|
"thread_id": "thread-1",
|
|
"user_id": None,
|
|
"message_id": "message-1",
|
|
"rating": -1,
|
|
"comment": "legacy anonymous",
|
|
"created_at": datetime(2026, 9, 16, 12, 1, 0),
|
|
},
|
|
{
|
|
"feedback_id": "anonymous-2",
|
|
"run_id": "run-1",
|
|
"thread_id": "thread-1",
|
|
"user_id": None,
|
|
"message_id": "message-1",
|
|
"rating": 1,
|
|
"comment": None,
|
|
"created_at": datetime(2026, 9, 16, 12, 2, 0),
|
|
},
|
|
],
|
|
)
|
|
operations = Operations(MigrationContext.configure(connection))
|
|
monkeypatch.setattr(migration, "op", operations)
|
|
|
|
migration.upgrade()
|
|
|
|
inspector = sa.inspect(connection)
|
|
columns = {column["name"] for column in inspector.get_columns("feedback")}
|
|
indexes = {index["name"] for index in inspector.get_indexes("feedback")}
|
|
assert {"agent_id", "target_key", "reasons", "updated_at"} <= columns
|
|
assert "uq_feedback_target_key" in indexes
|
|
assert "uq_feedback_thread_run_user" not in indexes
|
|
upgraded = connection.execute(
|
|
sa.text(
|
|
"SELECT agent_id, target_key, reasons, updated_at "
|
|
"FROM feedback WHERE feedback_id = 'feedback-1'"
|
|
)
|
|
).mappings().one()
|
|
assert upgraded["agent_id"] == "agent-a"
|
|
assert len(upgraded["target_key"]) == 64
|
|
assert upgraded["reasons"] == "[]"
|
|
assert upgraded["updated_at"] is not None
|
|
target_keys = connection.execute(
|
|
sa.text("SELECT target_key FROM feedback")
|
|
).scalars().all()
|
|
assert len(target_keys) == 3
|
|
assert len(set(target_keys)) == 3
|
|
|
|
migration.downgrade()
|
|
downgraded_columns = {
|
|
column["name"]
|
|
for column in sa.inspect(connection).get_columns("feedback")
|
|
}
|
|
assert "agent_id" not in downgraded_columns
|
|
assert "target_key" not in downgraded_columns
|
|
|
|
engine.dispose()
|