99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
"""STEP 02 — Simulation Panel CRUD + multi-source question intake.
|
||
|
||
Question bank for the AI quality audit. Sources:
|
||
manual 手工添加
|
||
imported 批量导入的真实用户问题
|
||
resolved 从"已答好"的 question_traces 采纳的回归用例
|
||
ai_gen AI 生成的覆盖性问题
|
||
"""
|
||
from fastapi import APIRouter, HTTPException
|
||
|
||
from app.auth import CurrentUser
|
||
from app.config import settings
|
||
from app.contracts import SimulationQuestion, SimulationQuestionUpdate
|
||
from app.db import (
|
||
list_simulation_questions,
|
||
create_simulation_question,
|
||
update_simulation_question,
|
||
delete_simulation_question,
|
||
bulk_create_simulation_questions,
|
||
adopt_hit_question_traces,
|
||
)
|
||
|
||
router = APIRouter()
|
||
_T = settings.default_tenant
|
||
_P = settings.default_project
|
||
|
||
|
||
@router.get("/simulation-panel")
|
||
async def _list(_user: CurrentUser = None):
|
||
return await list_simulation_questions(_T, _P)
|
||
|
||
|
||
@router.post("/simulation-panel")
|
||
async def _create(body: SimulationQuestion, _user: CurrentUser):
|
||
data = body.model_dump()
|
||
data["tenant_id"] = _T
|
||
data["project_id"] = _P
|
||
data["source"] = "manual"
|
||
return await create_simulation_question(data)
|
||
|
||
|
||
@router.post("/simulation-panel/import")
|
||
async def _import(body: dict, _user: CurrentUser):
|
||
"""Bulk import real user questions — one per line, or a list."""
|
||
raw = body.get("text") or body.get("questions") or ""
|
||
texts = raw if isinstance(raw, list) else str(raw).splitlines()
|
||
added = await bulk_create_simulation_questions(_T, _P, texts, "imported")
|
||
return {"added": added}
|
||
|
||
|
||
@router.post("/simulation-panel/adopt-hits")
|
||
async def _adopt_hits(_user: CurrentUser):
|
||
"""Adopt well-answered questions as regression cases (source=resolved)."""
|
||
added = await adopt_hit_question_traces(_T, _P)
|
||
return {"added": added}
|
||
|
||
|
||
@router.post("/simulation-panel/ai-generate")
|
||
async def _ai_generate(body: dict, _user: CurrentUser):
|
||
"""Let the LLM propose representative coverage questions."""
|
||
from app.llm_client import LlmClient
|
||
|
||
llm = LlmClient.from_settings() if settings.llm_api_key else None
|
||
if not llm or not llm.available():
|
||
raise HTTPException(400, "未配置 LLM（Agent 设置里填好 API Key 后可用）")
|
||
|
||
count = max(1, min(int(body.get("count", 10)), 30))
|
||
topic = (body.get("topic") or "城市本地生活知识图谱（地点、区域、路线、夜间体验、标签）").strip()
|
||
try:
|
||
out = llm.chat_json(
|
||
system=(
|
||
"你是知识图谱质量稽查助手。请基于给定主题，生成有代表性、"
|
||
"覆盖不同场景的真实用户中文问题，用于检验知识图谱数据的完善度。"
|
||
'严格返回 JSON：{"questions": ["...", "..."]}'
|
||
),
|
||
user=f"主题：{topic}。生成 {count} 个不同角度、不重复的问题。",
|
||
)
|
||
qs = out.get("questions") or []
|
||
qs = [str(q).strip() for q in qs if str(q).strip()][:count]
|
||
except Exception as e:
|
||
raise HTTPException(400, f"AI 生成失败：{str(e)[:160]}")
|
||
|
||
added = await bulk_create_simulation_questions(_T, _P, qs, "ai_gen")
|
||
return {"added": added, "generated": len(qs)}
|
||
|
||
|
||
@router.patch("/simulation-panel/{q_id}")
|
||
async def _update(q_id: int, body: SimulationQuestionUpdate, _user: CurrentUser):
|
||
row = await update_simulation_question(q_id, body.model_dump(exclude_none=True))
|
||
if not row:
|
||
raise HTTPException(404, "Simulation question not found")
|
||
return row
|
||
|
||
|
||
@router.delete("/simulation-panel/{q_id}")
|
||
async def _delete(q_id: int, _user: CurrentUser):
|
||
await delete_simulation_question(q_id)
|
||
return {"ok": True}
|