PMFlow/backend/tests/test_reply_iteration_servic...

259 lines
8.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from __future__ import annotations
import os
import pytest
import pytest_asyncio
from sqlalchemy import JSON
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
os.environ["DATABASE_URL"] = "sqlite+aiosqlite:///:memory:"
from app.models.base import Base
from app.models.experiment import Experiment
from app.models.feedback_center import FeedbackCenterRecord
from app.models.qa_record import QARecord
from app.models.reply_iteration import (
JudgeRun,
OptimizerCandidate,
PromptRelease,
QualityAsset,
QualityAssetPair,
)
from app.services.prompt_release.service import PromptReleaseService
from app.services.reply_iteration.service import ReplyIterationService
def _use_sqlite_json_types() -> None:
for table in (
FeedbackCenterRecord.__table__,
QARecord.__table__,
QualityAsset.__table__,
QualityAssetPair.__table__,
JudgeRun.__table__,
OptimizerCandidate.__table__,
PromptRelease.__table__,
Experiment.__table__,
):
for column in table.columns:
if column.name in {
"before_value",
"after_value",
"ai_snapshot",
"reason_candidates",
"sources",
"domains_used",
"routing_snapshot",
"citations",
"sources_used",
"evidence_pack",
"tool_trace",
"retrieval_metrics",
"evidence",
"source_snapshot",
"tags",
"metadata_json",
"findings",
"llm_evidence",
"suggestion",
"scope_json",
"linked_context_nodes",
"result_summary",
}:
column.type = JSON()
@pytest_asyncio.fixture
async def db_session():
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
_use_sqlite_json_types()
async with engine.begin() as conn:
await conn.run_sync(
lambda sync_conn: Base.metadata.create_all(
sync_conn,
tables=[
FeedbackCenterRecord.__table__,
QARecord.__table__,
QualityAsset.__table__,
QualityAssetPair.__table__,
JudgeRun.__table__,
OptimizerCandidate.__table__,
PromptRelease.__table__,
Experiment.__table__,
],
)
)
maker = async_sessionmaker(engine, expire_on_commit=False)
async with maker() as session:
yield session
await engine.dispose()
@pytest.mark.asyncio
async def test_promote_feedback_record_builds_asset_and_preference_pair(db_session) -> None:
qa_record = QARecord(
tenant_id=1,
question="这个回复为什么不对",
answer="让我先看一下你的情况,再帮你判断。",
sources=[],
intent="question",
domains_used=[],
response_time_ms=120,
asked_by=None,
asked_via="feishu",
project_id=8,
turn_index=1,
routing_snapshot={
"planned_tier": "low",
"executed_tier": "high",
"route_source": "default",
"route_rule_name": "engine_high_answer",
},
)
db_session.add(qa_record)
await db_session.flush()
failure_record = FeedbackCenterRecord(
tenant_id=1,
project_id=8,
source_type="qa_feedback",
source_id=101,
object_type="qa_record",
object_id=qa_record.id,
field_name="qa_feedback",
source_title="飞书问答反馈",
object_title="问答修正",
before_value={"question": qa_record.question, "answer": qa_record.answer},
after_value={"rating": "negative", "correct_answer": "上一条回复走的是 high 路由。"},
ai_snapshot={"prompt_key": "PROMPT_FEISHU_DIRECT_QA"},
reason_code="answer_incorrect",
review_status="accepted",
)
golden_record = FeedbackCenterRecord(
tenant_id=1,
project_id=8,
source_type="qa_feedback",
source_id=102,
object_type="qa_record",
object_id=qa_record.id,
field_name="qa_feedback",
source_title="飞书问答反馈",
object_title="问答修正",
before_value={"question": qa_record.question, "answer": qa_record.answer},
after_value={"rating": "positive", "correct_answer": "这条回复实际走的是 high,来源于默认规则。"},
ai_snapshot={"prompt_key": "PROMPT_FEISHU_DIRECT_QA"},
reason_code="answer_incorrect",
review_status="accepted",
)
db_session.add_all([failure_record, golden_record])
await db_session.commit()
service = ReplyIterationService(db_session)
failure_result = await service.promote_feedback_record(
tenant_id=1,
record_id=failure_record.id,
actor_id=9001,
asset_type="failure",
prompt_key="PROMPT_FEISHU_DIRECT_QA",
workflow_type="query.read_only",
)
golden_result = await service.promote_feedback_record(
tenant_id=1,
record_id=golden_record.id,
actor_id=9001,
asset_type="golden",
prompt_key="PROMPT_FEISHU_DIRECT_QA",
workflow_type="query.read_only",
pair_with_asset_id=failure_result["asset"]["id"],
pair_role="chosen",
pair_reason="人工确认修正版明显优于原答复",
)
assert failure_result["asset"]["asset_type"] == "failure"
assert failure_result["asset"]["qa_record_id"] == qa_record.id
assert failure_result["asset"]["planned_tier"] == "low"
assert failure_result["asset"]["executed_tier"] == "high"
assert failure_result["asset"]["route_rule_name"] == "engine_high_answer"
assert failure_result["pair"] is None
assert golden_result["asset"]["asset_type"] == "golden"
assert golden_result["pair"] is not None
assert golden_result["pair"]["chosen_asset_id"] == golden_result["asset"]["id"]
assert golden_result["pair"]["rejected_asset_id"] == failure_result["asset"]["id"]
@pytest.mark.asyncio
async def test_judge_run_and_optimizer_candidate_follow_bucket_rules(db_session) -> None:
asset = QualityAsset(
tenant_id=1,
project_id=3,
asset_type="failure",
status="approved",
prompt_key="PROMPT_FEISHU_DIRECT_QA",
workflow_type="query.read_only",
asked_via="feishu",
planned_tier="low",
executed_tier="low",
route_rule_name="engine_low_fallback",
question_text="这个回复为什么不对",
answer_text="让我先看一下,再通过 API 帮你创建。",
corrected_answer="直接说明这条回复走的是 low,并给出路由来源。",
source_snapshot={"reason_code": "answer_incorrect"},
)
db_session.add(asset)
await db_session.commit()
service = ReplyIterationService(db_session)
judge = await service.create_judge_run(
tenant_id=1,
asset_id=asset.id,
actor_id=9001,
)
candidate = await service.create_optimizer_candidate(
tenant_id=1,
judge_run_id=judge["id"],
actor_id=9001,
)
assert judge["bucket_key"] == "PROMPT_FEISHU_DIRECT_QA|query.read_only|feishu|low|low"
finding_codes = {item["code"] for item in judge["findings"]}
assert "process_talk_prefix" in finding_codes
assert "execution_claim_prefix" in finding_codes
assert candidate["target_type"] == "renderer_policy"
assert candidate["status"] == "draft"
assert candidate["target_ref"] == "reply_renderer"
@pytest.mark.asyncio
async def test_prompt_release_service_prefers_active_release_and_falls_back_on_format_error(db_session) -> None:
service = PromptReleaseService(db_session)
release = await service.create_release(
tenant_id=1,
prompt_key="PROMPT_FEISHU_DIRECT_QA",
body_snapshot="【定制版】问题:{question}",
created_by=9001,
notes="定制回复语气",
)
await service.activate_release(
tenant_id=1,
release_id=release["id"],
actor_id=9001,
)
rendered = await service.render_prompt(
tenant_id=1,
prompt_key="PROMPT_FEISHU_DIRECT_QA",
fallback_template="默认:{question}",
question="系统现在走哪个 tier",
)
broken = await service.render_prompt(
tenant_id=1,
prompt_key="PROMPT_FEISHU_DIRECT_QA",
fallback_template="默认:{question}",
question="系统现在走哪个 tier",
_override_template="坏模板:{missing_key}",
)
assert rendered == "【定制版】问题:系统现在走哪个 tier"
assert broken == "默认:系统现在走哪个 tier"