PMFlow/backend/tests/test_reply_iteration_api.py

284 lines
9.7 KiB
Python

from __future__ import annotations
import os
from types import SimpleNamespace
import pytest
import pytest_asyncio
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from sqlalchemy import JSON
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
os.environ["DATABASE_URL"] = "postgresql+asyncpg://tester:tester@localhost:5432/pmflow_test"
from app.api.ai_steward_api import routes_feedback_center, routes_reply_iteration, routes_training_pipeline
from app.core.database import get_db
from app.dependencies.auth import get_current_user
from app.dependencies.tenant import get_tenant_id
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.models.training_pipeline import (
TrainedModelArtifact,
TrainedModelDeployment,
TrainingDataset,
TrainingDatasetItem,
TrainingEvaluation,
TrainingJob,
TrainingJobRun,
)
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__,
TrainingDataset.__table__,
TrainingDatasetItem.__table__,
TrainingJob.__table__,
TrainingJobRun.__table__,
TrainingEvaluation.__table__,
TrainedModelArtifact.__table__,
TrainedModelDeployment.__table__,
):
for column in table.columns:
if column.name in {
"before_value",
"after_value",
"ai_snapshot",
"reason_candidates",
"sources",
"domains_used",
"routing_snapshot",
"evidence",
"source_snapshot",
"tags",
"metadata_json",
"findings",
"llm_evidence",
"suggestion",
"scope_json",
"linked_context_nodes",
"result_summary",
"selection_rules",
"channel_scope",
"redacted_snapshot",
"metrics_json",
"evaluation_summary",
"effective_scope",
}:
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__,
TrainingDataset.__table__,
TrainingDatasetItem.__table__,
TrainingJob.__table__,
TrainingJobRun.__table__,
TrainingEvaluation.__table__,
TrainedModelArtifact.__table__,
TrainedModelDeployment.__table__,
],
)
)
maker = async_sessionmaker(engine, expire_on_commit=False)
async with maker() as session:
yield session
await engine.dispose()
def _build_app(db_session):
app = FastAPI()
app.include_router(routes_feedback_center.router)
app.include_router(routes_reply_iteration.router)
app.include_router(routes_training_pipeline.router)
async def _fake_get_db():
yield db_session
app.dependency_overrides[get_db] = _fake_get_db
app.dependency_overrides[get_tenant_id] = lambda: 1
app.dependency_overrides[get_current_user] = lambda: SimpleNamespace(
id=9001,
tenant_id=1,
role="ADMIN",
name="管理员",
username="admin",
)
return app
@pytest.mark.asyncio
async def test_reply_iteration_and_training_admin_routes(db_session) -> None:
qa_record = QARecord(
tenant_id=1,
question="这个回复为什么不对",
answer="让我先看一下。",
sources=[],
intent="question",
domains_used=[],
asked_via="feishu",
project_id=7,
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()
feedback = FeedbackCenterRecord(
tenant_id=1,
project_id=7,
source_type="qa_feedback",
source_id=501,
object_type="qa_record",
object_id=qa_record.id,
field_name="qa_feedback",
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"},
review_status="accepted",
)
db_session.add(feedback)
await db_session.commit()
app = _build_app(db_session)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
promote_resp = await client.post(
f"/ai-steward/feedback-center/items/{feedback.id}/promote",
json={
"asset_type": "failure",
"prompt_key": "PROMPT_FEISHU_DIRECT_QA",
"workflow_type": "query.read_only",
},
)
assert promote_resp.status_code == 200
asset_id = promote_resp.json()["data"]["asset"]["id"]
judge_resp = await client.post(
"/ai-steward/reply-iteration/judge-runs",
json={"asset_id": asset_id},
)
assert judge_resp.status_code == 200
judge_id = judge_resp.json()["data"]["id"]
candidate_resp = await client.post(
"/ai-steward/reply-iteration/candidates",
json={"judge_run_id": judge_id},
)
assert candidate_resp.status_code == 200
release_resp = await client.post(
"/ai-steward/reply-iteration/prompt-releases",
json={
"prompt_key": "PROMPT_FEISHU_DIRECT_QA",
"body_snapshot": "【试验版】{question}",
"notes": "试验候选",
},
)
assert release_resp.status_code == 200
candidate_release_id = release_resp.json()["data"]["id"]
activate_resp = await client.post(
f"/ai-steward/reply-iteration/prompt-releases/{candidate_release_id}/activate",
)
assert activate_resp.status_code == 200
trial_resp = await client.post(
"/ai-steward/reply-iteration/trials",
json={
"title": "回复质量试验",
"hypothesis": "试验 prompt 能减少过程话术",
"target_metric": "accepted_rate",
"prompt_key": "PROMPT_FEISHU_DIRECT_QA",
"baseline_release_id": candidate_release_id,
"candidate_release_id": candidate_release_id,
"scope_json": {"asked_via": ["feishu"]},
},
)
assert trial_resp.status_code == 200
trial_id = trial_resp.json()["data"]["id"]
trial_decision_resp = await client.post(
f"/ai-steward/reply-iteration/trials/{trial_id}/decision",
json={"decision": "accepted", "result_summary": {"accepted_rate": 0.91}},
)
assert trial_decision_resp.status_code == 200
dataset_resp = await client.post(
"/ai-steward/training/datasets",
json={
"dataset_type": "eval",
"version": "eval-v1",
"asset_ids": [asset_id],
"pair_ids": [],
"selection_rules": {"scope": "reply_iteration"},
"notes": "评测集",
},
)
assert dataset_resp.status_code == 200
dataset_id = dataset_resp.json()["data"]["id"]
job_resp = await client.post(
"/ai-steward/training/jobs",
json={
"dataset_id": dataset_id,
"job_type": "sft",
"base_model_family": "glm",
"target_slot": "high",
"provider_backend": "openclaw-training",
},
)
assert job_resp.status_code == 200
job_id = job_resp.json()["data"]["id"]
artifact_id = job_resp.json()["data"]["artifact"]["id"]
eval_resp = await client.get(f"/ai-steward/training/jobs/{job_id}/evaluations")
promote_artifact_resp = await client.post(f"/ai-steward/training/artifacts/{artifact_id}/promote")
deployment_id = promote_artifact_resp.json()["data"]["id"]
deployment_decision_resp = await client.post(
f"/ai-steward/training/deployments/{deployment_id}/decision",
json={"decision": "accepted", "notes": "允许候选部署"},
)
assert judge_resp.json()["code"] == 0
assert candidate_resp.json()["data"]["status"] == "draft"
assert trial_decision_resp.json()["data"]["decision"] == "accepted"
assert eval_resp.json()["data"]["items"]
assert deployment_decision_resp.json()["data"]["deployment_status"] == "active"