1766 lines
77 KiB
Python
1766 lines
77 KiB
Python
"""Prompt asset read model for platform AI Ops."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import time
|
||
import string
|
||
from datetime import datetime, timedelta
|
||
from importlib import import_module
|
||
from typing import Any
|
||
|
||
from sqlalchemy import and_, func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ..config.ai_prompt_registry import get_prompt_registry, get_prompt_registry_entry
|
||
from ..config.ai_prompt_eval_rubrics import get_prompt_eval_rubric
|
||
from ..models import (
|
||
AICallLog,
|
||
OptimizerCandidate,
|
||
PMFlowAIFeedbackEvent,
|
||
PromptGoldenCase,
|
||
PromptRelease,
|
||
PromptReleaseCheck,
|
||
PromptReplayCaseResult,
|
||
PromptReplayRun,
|
||
)
|
||
from ..utils.timezone import now_cn_naive
|
||
from .llm_service import llm_service
|
||
from .prompt_release import PromptReleaseService
|
||
|
||
|
||
RELEASE_CHECK_TYPES = ("variable_contract", "schema_smoke", "golden_replay")
|
||
QUALITY_WINDOWS_HOURS = {"24h": 24, "7d": 168, "30d": 720}
|
||
REPLAY_MODES = {"contract_smoke", "compare", "llm_compare"}
|
||
GOLDEN_CASE_TYPES = {"contract", "regression", "risk", "edge", "business"}
|
||
GOLDEN_CASE_PRIORITIES = {"blocking", "important", "observation"}
|
||
|
||
|
||
def _hash_text(text: str) -> str:
|
||
if not text:
|
||
return ""
|
||
return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]
|
||
|
||
|
||
def _preview(text: str, limit: int = 420) -> str:
|
||
value = str(text or "").strip()
|
||
if len(value) <= limit:
|
||
return value
|
||
return f"{value[:limit].rstrip()}..."
|
||
|
||
|
||
def _template_variables(text: str) -> list[str]:
|
||
variables: set[str] = set()
|
||
formatter = string.Formatter()
|
||
try:
|
||
for _, field_name, _, _ in formatter.parse(str(text or "")):
|
||
if not field_name:
|
||
continue
|
||
head = str(field_name).split(".", 1)[0].split("[", 1)[0].strip()
|
||
if head and head.isidentifier():
|
||
variables.add(head)
|
||
except ValueError:
|
||
return []
|
||
return sorted(variables)
|
||
|
||
|
||
def _template_variables_or_error(text: str) -> list[str]:
|
||
variables: set[str] = set()
|
||
formatter = string.Formatter()
|
||
for _, field_name, _, _ in formatter.parse(str(text or "")):
|
||
if not field_name:
|
||
continue
|
||
head = str(field_name).split(".", 1)[0].split("[", 1)[0].strip()
|
||
if head and head.isidentifier():
|
||
variables.add(head)
|
||
return sorted(variables)
|
||
|
||
|
||
class PromptAssetService:
|
||
"""Combines static registry metadata with tenant prompt release state."""
|
||
|
||
def __init__(self, db: AsyncSession) -> None:
|
||
self.db = db
|
||
|
||
async def list_assets(
|
||
self,
|
||
*,
|
||
tenant_id: int | None,
|
||
family: str | None = None,
|
||
runtime_source: str | None = None,
|
||
editable_policy: str | None = None,
|
||
keyword: str | None = None,
|
||
) -> dict[str, Any]:
|
||
releases = await self._release_index(tenant_id=tenant_id)
|
||
items = []
|
||
for entry in get_prompt_registry():
|
||
if family and str(entry.get("prompt_family") or "") != family:
|
||
continue
|
||
if runtime_source and str(entry.get("runtime_source") or "") != runtime_source:
|
||
continue
|
||
if editable_policy and str(entry.get("editable_policy") or "") != editable_policy:
|
||
continue
|
||
haystack = " ".join(
|
||
[
|
||
str(entry.get("prompt_key") or ""),
|
||
str(entry.get("business_scenario") or ""),
|
||
str(entry.get("module") or ""),
|
||
str(entry.get("responsibility_scope") or ""),
|
||
]
|
||
).lower()
|
||
if keyword and str(keyword).strip().lower() not in haystack:
|
||
continue
|
||
items.append(self._asset_payload(entry, releases=releases, include_body=False))
|
||
|
||
family_counts: dict[str, int] = {}
|
||
policy_counts: dict[str, int] = {}
|
||
source_counts: dict[str, int] = {}
|
||
for item in items:
|
||
family_counts[item["prompt_family"]] = family_counts.get(item["prompt_family"], 0) + 1
|
||
policy_counts[item["editable_policy"]] = policy_counts.get(item["editable_policy"], 0) + 1
|
||
source_counts[item["runtime_source"]] = source_counts.get(item["runtime_source"], 0) + 1
|
||
|
||
return {
|
||
"items": items,
|
||
"total": len(items),
|
||
"summary": {
|
||
"total_assets": len(items),
|
||
"active_release_count": sum(1 for item in items if item.get("active_release")),
|
||
"needs_migration_count": sum(1 for item in items if item.get("editable_policy") == "needs_migration"),
|
||
"guarded_count": sum(1 for item in items if item.get("editable_policy") == "guarded_release"),
|
||
"family_counts": family_counts,
|
||
"editable_policy_counts": policy_counts,
|
||
"runtime_source_counts": source_counts,
|
||
},
|
||
}
|
||
|
||
async def get_asset(self, *, tenant_id: int, prompt_key: str) -> dict[str, Any]:
|
||
entry = get_prompt_registry_entry(prompt_key)
|
||
if entry is None:
|
||
raise ValueError("prompt asset not found")
|
||
releases = await self._release_index(tenant_id=tenant_id, prompt_key=prompt_key, include_body=True)
|
||
payload = self._asset_payload(entry, releases=releases, include_body=True)
|
||
payload["quality_summary"] = await self.get_quality_summary(
|
||
tenant_id=tenant_id,
|
||
prompt_key=prompt_key,
|
||
window="7d",
|
||
)
|
||
return payload
|
||
|
||
async def create_release(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
body_snapshot: str,
|
||
created_by: int | None,
|
||
notes: str | None = None,
|
||
) -> dict[str, Any]:
|
||
entry = self._releaseable_entry(prompt_key)
|
||
default_template = self._default_template(str(entry.get("prompt_key") or ""))
|
||
self._validate_release_body(default_template=default_template, body_snapshot=body_snapshot)
|
||
return await PromptReleaseService(self.db).create_release(
|
||
tenant_id=tenant_id,
|
||
prompt_key=str(entry.get("prompt_key") or "").strip(),
|
||
body_snapshot=body_snapshot,
|
||
created_by=created_by,
|
||
notes=notes,
|
||
)
|
||
|
||
async def list_releases(self, *, tenant_id: int, prompt_key: str) -> dict[str, Any]:
|
||
entry = get_prompt_registry_entry(prompt_key)
|
||
if entry is None:
|
||
raise ValueError("prompt asset not found")
|
||
return await PromptReleaseService(self.db).list_releases(tenant_id=tenant_id, prompt_key=prompt_key)
|
||
|
||
async def activate_release(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
release_id: int,
|
||
actor_id: int | None,
|
||
) -> dict[str, Any]:
|
||
entry = self._releaseable_entry(prompt_key)
|
||
await self._assert_activation_allowed(
|
||
tenant_id=tenant_id,
|
||
prompt_key=prompt_key,
|
||
release_id=release_id,
|
||
entry=entry,
|
||
)
|
||
return await PromptReleaseService(self.db).activate_release(
|
||
tenant_id=tenant_id,
|
||
prompt_key=prompt_key,
|
||
release_id=release_id,
|
||
actor_id=actor_id,
|
||
)
|
||
|
||
async def rollback_release(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
release_id: int,
|
||
actor_id: int | None,
|
||
) -> dict[str, Any]:
|
||
entry = self._releaseable_entry(prompt_key)
|
||
await self._assert_activation_allowed(
|
||
tenant_id=tenant_id,
|
||
prompt_key=prompt_key,
|
||
release_id=release_id,
|
||
entry=entry,
|
||
)
|
||
return await PromptReleaseService(self.db).rollback_release(
|
||
tenant_id=tenant_id,
|
||
prompt_key=prompt_key,
|
||
release_id=release_id,
|
||
actor_id=actor_id,
|
||
)
|
||
|
||
async def list_release_checks(self, *, tenant_id: int, prompt_key: str, release_id: int) -> dict[str, Any]:
|
||
release = await self._get_release(tenant_id=tenant_id, prompt_key=prompt_key, release_id=release_id)
|
||
rows = (
|
||
await self.db.execute(
|
||
select(PromptReleaseCheck)
|
||
.where(
|
||
PromptReleaseCheck.tenant_id == int(tenant_id),
|
||
PromptReleaseCheck.prompt_key == str(prompt_key or "").strip(),
|
||
PromptReleaseCheck.release_id == int(release_id),
|
||
)
|
||
.order_by(PromptReleaseCheck.id.desc())
|
||
)
|
||
).scalars().all()
|
||
items = [self._check_payload(row) for row in rows]
|
||
return {
|
||
"release_id": int(release.id),
|
||
"prompt_key": str(release.prompt_key or ""),
|
||
"items": items,
|
||
"total": len(items),
|
||
"gate": self._gate_summary_from_checks(items),
|
||
}
|
||
|
||
async def run_release_checks(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
release_id: int,
|
||
actor_id: int | None,
|
||
reason: str | None = None,
|
||
) -> dict[str, Any]:
|
||
entry = self._releaseable_entry(prompt_key)
|
||
release = await self._get_release(tenant_id=tenant_id, prompt_key=prompt_key, release_id=release_id)
|
||
default_template = self._default_template(str(entry.get("prompt_key") or ""))
|
||
active_case_count = await self._active_golden_case_count(
|
||
tenant_id=tenant_id,
|
||
prompt_key=prompt_key,
|
||
)
|
||
latest_replay = await self._latest_replay_for_release(
|
||
tenant_id=tenant_id,
|
||
release_id=int(release_id),
|
||
)
|
||
check_specs = self._build_release_check_specs(
|
||
tenant_id=tenant_id,
|
||
entry=entry,
|
||
release=release,
|
||
default_template=default_template,
|
||
reason=reason,
|
||
active_case_count=active_case_count,
|
||
latest_replay=latest_replay,
|
||
)
|
||
rows: list[PromptReleaseCheck] = []
|
||
completed_at = now_cn_naive().isoformat()
|
||
for spec in check_specs:
|
||
row = PromptReleaseCheck(
|
||
tenant_id=int(tenant_id),
|
||
prompt_key=str(prompt_key or "").strip(),
|
||
release_id=int(release_id),
|
||
check_type=spec["check_type"],
|
||
status=spec["status"],
|
||
severity=spec["severity"],
|
||
summary=spec["summary"],
|
||
details_json=spec.get("details_json") or {},
|
||
created_by=actor_id,
|
||
completed_at=completed_at,
|
||
)
|
||
self.db.add(row)
|
||
rows.append(row)
|
||
await self.db.commit()
|
||
for row in rows:
|
||
await self.db.refresh(row)
|
||
items = [self._check_payload(row) for row in rows]
|
||
return {
|
||
"release_id": int(release.id),
|
||
"prompt_key": str(release.prompt_key or ""),
|
||
"items": items,
|
||
"total": len(items),
|
||
"gate": self._gate_summary_from_checks(items),
|
||
}
|
||
|
||
async def get_quality_summary(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
window: str = "7d",
|
||
status: str | None = None,
|
||
release_id: int | None = None,
|
||
feedback_tenant_id: int | None = None,
|
||
include_all_pmflow_feedback: bool = False,
|
||
) -> dict[str, Any]:
|
||
if get_prompt_registry_entry(prompt_key) is None:
|
||
raise ValueError("prompt asset not found")
|
||
hours = QUALITY_WINDOWS_HOURS.get(str(window or "7d").strip(), QUALITY_WINDOWS_HOURS["7d"])
|
||
filters = [
|
||
AICallLog.tenant_id == int(tenant_id),
|
||
AICallLog.prompt_key == str(prompt_key or "").strip(),
|
||
AICallLog.created_at >= datetime.now() - timedelta(hours=hours),
|
||
]
|
||
if status and status != "all":
|
||
filters.append(AICallLog.status == status)
|
||
if release_id is not None:
|
||
filters.append(AICallLog.prompt_release_id == int(release_id))
|
||
where_clause = and_(*filters)
|
||
summary_stmt = select(
|
||
func.count(AICallLog.id),
|
||
func.sum(AICallLog.cost),
|
||
func.avg(AICallLog.latency_ms),
|
||
).where(where_clause)
|
||
total, cost, latency = (await self.db.execute(summary_stmt)).one()
|
||
total_calls = int(total or 0)
|
||
rows = (
|
||
await self.db.execute(
|
||
select(
|
||
AICallLog.status,
|
||
AICallLog.prompt_source,
|
||
AICallLog.prompt_release_id,
|
||
func.count(AICallLog.id),
|
||
)
|
||
.where(where_clause)
|
||
.group_by(AICallLog.status, AICallLog.prompt_source, AICallLog.prompt_release_id)
|
||
)
|
||
).all()
|
||
failed_calls = 0
|
||
fallback_calls = 0
|
||
active_release_calls = 0
|
||
default_fallback_calls = 0
|
||
release_breakdown: dict[str, dict[str, Any]] = {}
|
||
for row_status, prompt_source, row_release_id, count in rows:
|
||
amount = int(count or 0)
|
||
if self._is_failed_status(str(row_status or "")):
|
||
failed_calls += amount
|
||
source = str(prompt_source or "unknown")
|
||
if source == "active_release":
|
||
active_release_calls += amount
|
||
if source == "default_fallback":
|
||
default_fallback_calls += amount
|
||
fallback_calls += amount
|
||
release_key = str(row_release_id) if row_release_id is not None else source
|
||
bucket = release_breakdown.setdefault(
|
||
release_key,
|
||
{
|
||
"prompt_release_id": int(row_release_id) if row_release_id is not None else None,
|
||
"prompt_source": source,
|
||
"total_calls": 0,
|
||
"failed_calls": 0,
|
||
},
|
||
)
|
||
bucket["total_calls"] += amount
|
||
if self._is_failed_status(str(row_status or "")):
|
||
bucket["failed_calls"] += amount
|
||
recent_failures = [
|
||
self._call_log_summary(row)
|
||
for row in (
|
||
await self.db.execute(
|
||
select(AICallLog)
|
||
.where(where_clause)
|
||
.order_by(AICallLog.created_at.desc(), AICallLog.id.desc())
|
||
.limit(20)
|
||
)
|
||
).scalars().all()
|
||
if self._is_failed_status(str(row.status or ""))
|
||
][:5]
|
||
feedback_summary = await self._pmflow_feedback_summary(
|
||
tenant_id=None if include_all_pmflow_feedback else feedback_tenant_id if feedback_tenant_id is not None else tenant_id,
|
||
prompt_key=str(prompt_key or "").strip(),
|
||
since=datetime.now() - timedelta(hours=hours),
|
||
release_id=release_id,
|
||
)
|
||
optimizer_candidates = await self._prompt_optimizer_candidates(
|
||
tenant_id=None if include_all_pmflow_feedback else feedback_tenant_id if feedback_tenant_id is not None else tenant_id,
|
||
prompt_key=str(prompt_key or "").strip(),
|
||
)
|
||
return {
|
||
"prompt_key": str(prompt_key or "").strip(),
|
||
"window": str(window or "7d"),
|
||
"total_calls": total_calls,
|
||
"failed_calls": failed_calls,
|
||
"failure_rate": round(failed_calls / total_calls, 4) if total_calls else 0,
|
||
"fallback_calls": fallback_calls,
|
||
"active_release_calls": active_release_calls,
|
||
"default_fallback_calls": default_fallback_calls,
|
||
"average_latency_ms": round(float(latency), 2) if latency is not None else None,
|
||
"estimated_cost": float(cost or 0),
|
||
"recent_failures": recent_failures,
|
||
"release_breakdown": list(release_breakdown.values()),
|
||
"pmflow_feedback": feedback_summary,
|
||
"optimizer_candidates": optimizer_candidates,
|
||
}
|
||
|
||
async def list_golden_cases(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
status: str | None = None,
|
||
) -> dict[str, Any]:
|
||
if get_prompt_registry_entry(prompt_key) is None:
|
||
raise ValueError("prompt asset not found")
|
||
stmt = select(PromptGoldenCase).where(
|
||
PromptGoldenCase.tenant_id == int(tenant_id),
|
||
PromptGoldenCase.prompt_key == str(prompt_key or "").strip(),
|
||
)
|
||
if status and status != "all":
|
||
stmt = stmt.where(PromptGoldenCase.status == status)
|
||
rows = list((await self.db.execute(stmt.order_by(PromptGoldenCase.id.desc()))).scalars().all())
|
||
return {
|
||
"prompt_key": str(prompt_key or "").strip(),
|
||
"items": [self._golden_case_payload(row) for row in rows],
|
||
"total": len(rows),
|
||
"active_count": sum(1 for row in rows if str(row.status or "") == "active"),
|
||
}
|
||
|
||
async def create_golden_case(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
case_name: str,
|
||
input_payload: dict[str, Any],
|
||
expected_contract: dict[str, Any] | None = None,
|
||
expected_contains: str | None = None,
|
||
case_type: str = "regression",
|
||
priority: str = "blocking",
|
||
scenario_key: str | None = None,
|
||
tags: list[str] | None = None,
|
||
source: str = "manual",
|
||
source_ai_call_log_id: int | None = None,
|
||
created_by: int | None = None,
|
||
notes: str | None = None,
|
||
) -> dict[str, Any]:
|
||
if get_prompt_registry_entry(prompt_key) is None:
|
||
raise ValueError("prompt asset not found")
|
||
name = str(case_name or "").strip()
|
||
if not name:
|
||
raise ValueError("case_name is required")
|
||
if not isinstance(input_payload, dict):
|
||
raise ValueError("input_payload must be a JSON object")
|
||
effective_case_type = str(case_type or "regression").strip() or "regression"
|
||
if effective_case_type not in GOLDEN_CASE_TYPES:
|
||
raise ValueError("unsupported golden case type")
|
||
effective_priority = str(priority or "blocking").strip() or "blocking"
|
||
if effective_priority not in GOLDEN_CASE_PRIORITIES:
|
||
raise ValueError("unsupported golden case priority")
|
||
row = PromptGoldenCase(
|
||
tenant_id=int(tenant_id),
|
||
prompt_key=str(prompt_key or "").strip(),
|
||
case_name=name,
|
||
status="active",
|
||
input_payload=input_payload,
|
||
expected_contract=expected_contract or {},
|
||
expected_contains=str(expected_contains or "").strip() or None,
|
||
source=str(source or "manual").strip() or "manual",
|
||
source_ai_call_log_id=source_ai_call_log_id,
|
||
case_type=effective_case_type,
|
||
priority=effective_priority,
|
||
scenario_key=str(scenario_key or "").strip() or None,
|
||
tags=[str(item).strip() for item in list(tags or []) if str(item).strip()],
|
||
created_by=created_by,
|
||
notes=str(notes or "").strip() or None,
|
||
)
|
||
self.db.add(row)
|
||
await self.db.commit()
|
||
await self.db.refresh(row)
|
||
return self._golden_case_payload(row)
|
||
|
||
async def create_golden_case_from_call(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
ai_call_log_id: int,
|
||
created_by: int | None = None,
|
||
case_name: str | None = None,
|
||
notes: str | None = None,
|
||
) -> dict[str, Any]:
|
||
if get_prompt_registry_entry(prompt_key) is None:
|
||
raise ValueError("prompt asset not found")
|
||
row = (
|
||
await self.db.execute(
|
||
select(AICallLog).where(
|
||
AICallLog.tenant_id == int(tenant_id),
|
||
AICallLog.id == int(ai_call_log_id),
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if row is None:
|
||
raise ValueError("AI call log not found")
|
||
if str(row.prompt_key or "").strip() != str(prompt_key or "").strip():
|
||
raise ValueError("AI call log is not linked to this prompt_key")
|
||
status_text = str(row.status or "").lower()
|
||
priority = "blocking" if "structured_invalid" in status_text or "invalid" in status_text else "important"
|
||
return await self.create_golden_case(
|
||
tenant_id=tenant_id,
|
||
prompt_key=prompt_key,
|
||
case_name=case_name or f"trace-{row.trace_id or row.id}",
|
||
input_payload={
|
||
"ai_call_log_id": int(row.id),
|
||
"trace_id": row.trace_id,
|
||
"function_name": row.function_name,
|
||
"prompt_release_id": row.prompt_release_id,
|
||
"prompt_source": row.prompt_source,
|
||
"prompt_preview": _preview(row.prompt or "", limit=1000),
|
||
},
|
||
expected_contract={
|
||
"status": row.status,
|
||
"response_preview": _preview(row.response or "", limit=1000),
|
||
},
|
||
source="ai_call_log",
|
||
source_ai_call_log_id=int(row.id),
|
||
case_type="regression",
|
||
priority=priority,
|
||
scenario_key=str(row.function_name or "").strip() or None,
|
||
tags=["from_ai_call_log", status_text] if status_text else ["from_ai_call_log"],
|
||
created_by=created_by,
|
||
notes=notes,
|
||
)
|
||
|
||
async def create_replay_run(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
release_id: int,
|
||
created_by: int | None,
|
||
mode: str = "contract_smoke",
|
||
case_status: str = "active",
|
||
max_cases: int = 20,
|
||
include_llm: bool = False,
|
||
judge_enabled: bool = True,
|
||
reason: str | None = None,
|
||
) -> dict[str, Any]:
|
||
entry = self._releaseable_entry(prompt_key)
|
||
release = await self._get_release(tenant_id=tenant_id, prompt_key=prompt_key, release_id=release_id)
|
||
run_mode = str(mode or "contract_smoke").strip() or "contract_smoke"
|
||
if include_llm and run_mode in {"contract_smoke", "compare"}:
|
||
run_mode = "llm_compare"
|
||
if run_mode not in REPLAY_MODES:
|
||
raise ValueError("unsupported replay mode")
|
||
if include_llm and int(max_cases or 20) > 50:
|
||
raise ValueError("LLM replay max_cases cannot exceed 50")
|
||
limit = max(1, min(int(max_cases or 20), 50 if include_llm else 100))
|
||
cases = await self._active_golden_cases(
|
||
tenant_id=tenant_id,
|
||
prompt_key=prompt_key,
|
||
status=case_status,
|
||
limit=limit,
|
||
)
|
||
if not cases:
|
||
raise ValueError("no active golden cases for replay")
|
||
baseline = await self._active_release_for_prompt(tenant_id=tenant_id, prompt_key=prompt_key)
|
||
started_at = now_cn_naive().isoformat()
|
||
run = PromptReplayRun(
|
||
tenant_id=int(tenant_id),
|
||
prompt_key=str(prompt_key or "").strip(),
|
||
release_id=int(release.id),
|
||
baseline_release_id=int(baseline.id) if baseline is not None else None,
|
||
run_status="running",
|
||
run_mode=run_mode,
|
||
case_count=0,
|
||
passed_count=0,
|
||
failed_count=0,
|
||
warning_count=0,
|
||
summary="回放执行中",
|
||
risk_flags_json=[],
|
||
created_by=created_by,
|
||
started_at=started_at,
|
||
)
|
||
self.db.add(run)
|
||
await self.db.flush()
|
||
|
||
passed = 0
|
||
failed = 0
|
||
warning = 0
|
||
risk_flags: list[str] = []
|
||
for golden_case in cases:
|
||
started = time.perf_counter()
|
||
result = await self._evaluate_replay_case(
|
||
tenant_id=tenant_id,
|
||
release=release,
|
||
baseline=baseline,
|
||
golden_case=golden_case,
|
||
entry=entry,
|
||
include_llm=include_llm or run_mode == "llm_compare",
|
||
judge_enabled=judge_enabled,
|
||
run_id=int(run.id),
|
||
)
|
||
latency_ms = max(0, int((time.perf_counter() - started) * 1000))
|
||
status = self._effective_replay_status(str(result["status"]), priority=str(golden_case.priority or "blocking"))
|
||
if status == "passed":
|
||
passed += 1
|
||
elif status == "failed":
|
||
failed += 1
|
||
else:
|
||
warning += 1
|
||
risk_flags.extend([str(item) for item in result.get("risk_flags") or [] if str(item).strip()])
|
||
self.db.add(
|
||
PromptReplayCaseResult(
|
||
tenant_id=int(tenant_id),
|
||
run_id=int(run.id),
|
||
golden_case_id=int(golden_case.id),
|
||
prompt_key=str(prompt_key or "").strip(),
|
||
release_id=int(release.id),
|
||
baseline_output=result.get("baseline_output"),
|
||
candidate_output=result.get("candidate_output"),
|
||
status=status,
|
||
contract_result_json=result.get("contract_result_json") or {},
|
||
diff_summary=result.get("diff_summary"),
|
||
risk_flags_json=result.get("risk_flags") or [],
|
||
baseline_ai_call_log_id=result.get("baseline_ai_call_log_id"),
|
||
candidate_ai_call_log_id=result.get("candidate_ai_call_log_id"),
|
||
judge_score=result.get("judge_score"),
|
||
judge_label=result.get("judge_label"),
|
||
judge_reason=result.get("judge_reason"),
|
||
latency_ms=latency_ms,
|
||
estimated_cost=float(result.get("estimated_cost") or 0.0),
|
||
error_summary=result.get("error_summary"),
|
||
)
|
||
)
|
||
golden_case.last_run_status = status
|
||
if status == "failed":
|
||
golden_case.last_failed_at = now_cn_naive().isoformat()
|
||
unique_risks = sorted(set(risk_flags))
|
||
completed_at = now_cn_naive().isoformat()
|
||
run.run_status = "completed"
|
||
run.case_count = len(cases)
|
||
run.passed_count = passed
|
||
run.failed_count = failed
|
||
run.warning_count = warning
|
||
run.score_delta = 0.0
|
||
run.risk_flags_json = unique_risks
|
||
if failed:
|
||
run.summary = f"回放完成：{failed} 个样本阻塞，{warning} 个样本需复核。"
|
||
elif warning:
|
||
run.summary = f"回放完成：无阻塞失败，{warning} 个样本需人工复核。"
|
||
elif include_llm or run_mode == "llm_compare":
|
||
run.summary = "LLM 回放完成：所有样本通过评估，建议进入发布评审。"
|
||
else:
|
||
run.summary = "回放完成：所有样本通过确定性契约冒烟。"
|
||
run.completed_at = completed_at
|
||
await self.db.commit()
|
||
await self.db.refresh(run)
|
||
return await self.get_replay_run(tenant_id=tenant_id, prompt_key=prompt_key, run_id=int(run.id))
|
||
|
||
async def list_replay_runs(self, *, tenant_id: int, prompt_key: str, release_id: int) -> dict[str, Any]:
|
||
release = await self._get_release(tenant_id=tenant_id, prompt_key=prompt_key, release_id=release_id)
|
||
rows = list(
|
||
(
|
||
await self.db.execute(
|
||
select(PromptReplayRun)
|
||
.where(
|
||
PromptReplayRun.tenant_id == int(tenant_id),
|
||
PromptReplayRun.prompt_key == str(prompt_key or "").strip(),
|
||
PromptReplayRun.release_id == int(release.id),
|
||
)
|
||
.order_by(PromptReplayRun.id.desc())
|
||
)
|
||
).scalars().all()
|
||
)
|
||
return {
|
||
"prompt_key": str(prompt_key or "").strip(),
|
||
"release_id": int(release.id),
|
||
"items": [self._replay_run_payload(row, include_results=False) for row in rows],
|
||
"total": len(rows),
|
||
}
|
||
|
||
async def get_replay_run(self, *, tenant_id: int, prompt_key: str, run_id: int) -> dict[str, Any]:
|
||
run = (
|
||
await self.db.execute(
|
||
select(PromptReplayRun).where(
|
||
PromptReplayRun.tenant_id == int(tenant_id),
|
||
PromptReplayRun.prompt_key == str(prompt_key or "").strip(),
|
||
PromptReplayRun.id == int(run_id),
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if run is None:
|
||
raise ValueError("prompt replay run not found")
|
||
results = list(
|
||
(
|
||
await self.db.execute(
|
||
select(PromptReplayCaseResult)
|
||
.where(
|
||
PromptReplayCaseResult.tenant_id == int(tenant_id),
|
||
PromptReplayCaseResult.run_id == int(run.id),
|
||
)
|
||
.order_by(PromptReplayCaseResult.id.asc())
|
||
)
|
||
).scalars().all()
|
||
)
|
||
return self._replay_run_payload(run, include_results=True, results=results)
|
||
|
||
async def get_release_watch_summary(self, *, tenant_id: int, prompt_key: str, release_id: int) -> dict[str, Any]:
|
||
release = await self._get_release(tenant_id=tenant_id, prompt_key=prompt_key, release_id=release_id)
|
||
quality_24h = await self.get_quality_summary(tenant_id=tenant_id, prompt_key=prompt_key, window="24h", release_id=int(release.id))
|
||
quality_7d = await self.get_quality_summary(tenant_id=tenant_id, prompt_key=prompt_key, window="7d", release_id=int(release.id))
|
||
failure_rate = float(quality_24h.get("failure_rate") or 0)
|
||
fallback_rate = round(float(quality_24h.get("default_fallback_calls") or 0) / max(1, int(quality_24h.get("total_calls") or 0)), 4)
|
||
structured_invalid_count = int(
|
||
(
|
||
await self.db.execute(
|
||
select(func.count(AICallLog.id)).where(
|
||
AICallLog.tenant_id == int(tenant_id),
|
||
AICallLog.prompt_key == str(prompt_key or "").strip(),
|
||
AICallLog.prompt_release_id == int(release.id),
|
||
AICallLog.status == "structured_invalid",
|
||
AICallLog.created_at >= datetime.now() - timedelta(hours=24),
|
||
)
|
||
)
|
||
).scalar()
|
||
or 0
|
||
)
|
||
if int(quality_24h.get("total_calls") or 0) <= 0:
|
||
recommendation = "watch"
|
||
summary = "观察中，暂无 release 级运行样本。"
|
||
elif failure_rate >= 0.2 or structured_invalid_count >= 3:
|
||
recommendation = "rollback_suggested"
|
||
summary = "24h 失败或结构化异常偏高，建议复核并准备回滚。"
|
||
elif failure_rate >= 0.08 or fallback_rate >= 0.15 or structured_invalid_count > 0:
|
||
recommendation = "needs_review"
|
||
summary = "已出现可疑失败、结构化异常或默认回退，建议复核失败样本并补充 golden case。"
|
||
else:
|
||
recommendation = "healthy"
|
||
summary = "当前观察窗口未发现明显异常。"
|
||
return {
|
||
"prompt_key": str(prompt_key or "").strip(),
|
||
"release_id": int(release.id),
|
||
"release_version": int(release.version or 0),
|
||
"status": str(release.status or ""),
|
||
"windows": {"24h": quality_24h, "7d": quality_7d},
|
||
"structured_invalid_24h": structured_invalid_count,
|
||
"fallback_rate_24h": fallback_rate,
|
||
"recommendation": recommendation,
|
||
"summary": summary,
|
||
"recent_failures": quality_24h.get("recent_failures") or [],
|
||
}
|
||
|
||
async def _release_index(self, *, tenant_id: int, prompt_key: str | None = None, include_body: bool = False) -> dict[str, Any]:
|
||
stmt = select(PromptRelease).where(PromptRelease.tenant_id == int(tenant_id))
|
||
if prompt_key:
|
||
stmt = stmt.where(PromptRelease.prompt_key == str(prompt_key).strip())
|
||
rows = list((await self.db.execute(stmt.order_by(PromptRelease.prompt_key.asc(), PromptRelease.version.desc()))).scalars().all())
|
||
checks_by_release = await self._latest_checks_by_release(
|
||
tenant_id=tenant_id,
|
||
release_ids=[int(row.id) for row in rows],
|
||
)
|
||
replay_by_release = await self._latest_replay_by_release(
|
||
tenant_id=tenant_id,
|
||
release_ids=[int(row.id) for row in rows],
|
||
)
|
||
by_key: dict[str, dict[str, Any]] = {}
|
||
for row in rows:
|
||
key = str(row.prompt_key or "").strip()
|
||
if not key:
|
||
continue
|
||
bucket = by_key.setdefault(key, {"releases": [], "active": None})
|
||
snapshot = self._release_payload(
|
||
row,
|
||
include_body=include_body,
|
||
checks=checks_by_release.get(int(row.id), []),
|
||
latest_replay=replay_by_release.get(int(row.id)),
|
||
)
|
||
bucket["releases"].append(snapshot)
|
||
if row.status == "active" and bucket["active"] is None:
|
||
bucket["active"] = snapshot
|
||
return by_key
|
||
|
||
def _asset_payload(self, entry: dict[str, Any], *, releases: dict[str, Any], include_body: bool) -> dict[str, Any]:
|
||
prompt_key = str(entry.get("prompt_key") or "").strip()
|
||
default_template = self._default_template(prompt_key)
|
||
release_bucket = releases.get(prompt_key, {})
|
||
active_release = release_bucket.get("active")
|
||
all_releases = list(release_bucket.get("releases") or [])
|
||
payload = {
|
||
"prompt_key": prompt_key,
|
||
"module": str(entry.get("module") or ""),
|
||
"business_scenario": str(entry.get("business_scenario") or ""),
|
||
"prompt_family": str(entry.get("prompt_family") or ""),
|
||
"responsibility_scope": str(entry.get("responsibility_scope") or ""),
|
||
"priority": str(entry.get("priority") or ""),
|
||
"audience": str(entry.get("audience") or ""),
|
||
"risk_level": str(entry.get("risk_level") or ""),
|
||
"runtime_source": str(entry.get("runtime_source") or ""),
|
||
"editable_policy": str(entry.get("editable_policy") or ""),
|
||
"input_contract": str(entry.get("input_contract") or ""),
|
||
"output_contract": str(entry.get("output_contract") or ""),
|
||
"call_sites": list(entry.get("call_sites") or []),
|
||
"default_template_available": bool(default_template),
|
||
"default_template_hash": _hash_text(default_template),
|
||
"default_template_preview": _preview(default_template),
|
||
"template_variables": _template_variables(default_template),
|
||
"active_release": active_release,
|
||
"release_count": len(all_releases),
|
||
"draft_release_count": sum(1 for item in all_releases if item.get("status") == "draft"),
|
||
"latest_release": all_releases[0] if all_releases else None,
|
||
"migration_status": self._migration_status(entry, default_template),
|
||
}
|
||
if include_body:
|
||
payload["default_template_body"] = default_template
|
||
payload["releases"] = all_releases
|
||
return payload
|
||
|
||
@staticmethod
|
||
def _release_payload(
|
||
row: PromptRelease,
|
||
*,
|
||
include_body: bool = False,
|
||
checks: list[dict[str, Any]] | None = None,
|
||
latest_replay: dict[str, Any] | None = None,
|
||
) -> dict[str, Any]:
|
||
body = str(row.body_snapshot or "")
|
||
check_items = list(checks or [])
|
||
payload = {
|
||
"id": int(row.id),
|
||
"prompt_key": str(row.prompt_key or ""),
|
||
"version": int(row.version or 0),
|
||
"status": str(row.status or ""),
|
||
"notes": row.notes,
|
||
"body_hash": _hash_text(body),
|
||
"body_preview": _preview(body),
|
||
"created_by": int(row.created_by) if row.created_by is not None else None,
|
||
"activated_by": int(row.activated_by) if row.activated_by is not None else None,
|
||
"activated_at": row.activated_at,
|
||
"release_gate": PromptAssetService._gate_summary_from_checks(check_items),
|
||
"latest_checks": check_items,
|
||
"latest_replay_run": latest_replay,
|
||
}
|
||
if include_body:
|
||
payload["body_snapshot"] = body
|
||
return payload
|
||
|
||
def _releaseable_entry(self, prompt_key: str) -> dict[str, Any]:
|
||
entry = get_prompt_registry_entry(str(prompt_key or "").strip())
|
||
if entry is None:
|
||
raise ValueError("prompt asset not found")
|
||
policy = str(entry.get("editable_policy") or "")
|
||
if policy == "needs_migration":
|
||
raise ValueError("prompt call sites must be migrated before release operations")
|
||
if policy not in {"managed_release", "guarded_release"}:
|
||
raise ValueError("prompt asset is not editable")
|
||
if str(entry.get("runtime_source") or "") == "inline_discovered":
|
||
raise ValueError("inline prompt must be migrated before release operations")
|
||
if not self._default_template(str(entry.get("prompt_key") or "")):
|
||
raise ValueError("prompt asset has no default template")
|
||
return entry
|
||
|
||
async def _assert_activation_allowed(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
release_id: int,
|
||
entry: dict[str, Any],
|
||
) -> None:
|
||
release = await self._get_release(tenant_id=tenant_id, prompt_key=prompt_key, release_id=release_id)
|
||
if str(entry.get("editable_policy") or "") != "guarded_release":
|
||
return
|
||
checks = await self._latest_checks_for_release(
|
||
tenant_id=tenant_id,
|
||
release_id=int(release.id),
|
||
)
|
||
gate = self._gate_summary_from_checks(checks)
|
||
if gate.get("status") != "passed":
|
||
raise ValueError(f"guarded prompt release is blocked by gate: {gate.get('summary') or 'checks not passed'}")
|
||
|
||
async def _get_release(self, *, tenant_id: int, prompt_key: str, release_id: int) -> PromptRelease:
|
||
row = (
|
||
await self.db.execute(
|
||
select(PromptRelease).where(
|
||
PromptRelease.tenant_id == int(tenant_id),
|
||
PromptRelease.prompt_key == str(prompt_key or "").strip(),
|
||
PromptRelease.id == int(release_id),
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if row is None:
|
||
raise ValueError("prompt release not found")
|
||
return row
|
||
|
||
async def _latest_checks_by_release(self, *, tenant_id: int, release_ids: list[int]) -> dict[int, list[dict[str, Any]]]:
|
||
ids = [int(item) for item in release_ids if item is not None]
|
||
if not ids:
|
||
return {}
|
||
rows = (
|
||
await self.db.execute(
|
||
select(PromptReleaseCheck)
|
||
.where(
|
||
PromptReleaseCheck.tenant_id == int(tenant_id),
|
||
PromptReleaseCheck.release_id.in_(ids),
|
||
)
|
||
.order_by(PromptReleaseCheck.release_id.asc(), PromptReleaseCheck.check_type.asc(), PromptReleaseCheck.id.desc())
|
||
)
|
||
).scalars().all()
|
||
by_release: dict[int, dict[str, dict[str, Any]]] = {}
|
||
for row in rows:
|
||
release_id = int(row.release_id)
|
||
bucket = by_release.setdefault(release_id, {})
|
||
check_type = str(row.check_type or "")
|
||
if check_type not in bucket:
|
||
bucket[check_type] = self._check_payload(row)
|
||
return {release_id: list(checks.values()) for release_id, checks in by_release.items()}
|
||
|
||
async def _latest_checks_for_release(self, *, tenant_id: int, release_id: int) -> list[dict[str, Any]]:
|
||
grouped = await self._latest_checks_by_release(tenant_id=tenant_id, release_ids=[int(release_id)])
|
||
return grouped.get(int(release_id), [])
|
||
|
||
async def _latest_replay_by_release(self, *, tenant_id: int, release_ids: list[int]) -> dict[int, dict[str, Any]]:
|
||
ids = [int(item) for item in release_ids if item is not None]
|
||
if not ids:
|
||
return {}
|
||
rows = list(
|
||
(
|
||
await self.db.execute(
|
||
select(PromptReplayRun)
|
||
.where(
|
||
PromptReplayRun.tenant_id == int(tenant_id),
|
||
PromptReplayRun.release_id.in_(ids),
|
||
)
|
||
.order_by(PromptReplayRun.release_id.asc(), PromptReplayRun.id.desc())
|
||
)
|
||
).scalars().all()
|
||
)
|
||
latest: dict[int, dict[str, Any]] = {}
|
||
for row in rows:
|
||
release_id = int(row.release_id)
|
||
if release_id not in latest:
|
||
latest[release_id] = self._replay_run_payload(row, include_results=False)
|
||
return latest
|
||
|
||
async def _latest_replay_for_release(self, *, tenant_id: int, release_id: int) -> dict[str, Any] | None:
|
||
grouped = await self._latest_replay_by_release(tenant_id=tenant_id, release_ids=[int(release_id)])
|
||
return grouped.get(int(release_id))
|
||
|
||
async def _pmflow_feedback_summary(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
since: datetime,
|
||
release_id: int | None,
|
||
) -> dict[str, Any]:
|
||
filters = [
|
||
PMFlowAIFeedbackEvent.prompt_key == str(prompt_key or "").strip(),
|
||
PMFlowAIFeedbackEvent.created_at >= since,
|
||
]
|
||
if tenant_id is not None:
|
||
filters.append(PMFlowAIFeedbackEvent.tenant_id == int(tenant_id))
|
||
if release_id is not None:
|
||
filters.append(PMFlowAIFeedbackEvent.prompt_release_id == int(release_id))
|
||
rows = list(
|
||
(
|
||
await self.db.execute(
|
||
select(PMFlowAIFeedbackEvent)
|
||
.where(and_(*filters))
|
||
.order_by(PMFlowAIFeedbackEvent.created_at.desc(), PMFlowAIFeedbackEvent.id.desc())
|
||
.limit(200)
|
||
)
|
||
).scalars().all()
|
||
)
|
||
total = len(rows)
|
||
positive = sum(1 for row in rows if row.rating == "positive")
|
||
negative = sum(1 for row in rows if row.rating == "negative")
|
||
reason_counts: dict[str, int] = {}
|
||
answer_kind_counts: dict[str, int] = {}
|
||
for row in rows:
|
||
answer_kind_counts[row.answer_kind] = answer_kind_counts.get(row.answer_kind, 0) + 1
|
||
for tag in row.reason_tags or []:
|
||
label = str(tag or "").strip()
|
||
if label:
|
||
reason_counts[label] = reason_counts.get(label, 0) + 1
|
||
return {
|
||
"total_feedback": total,
|
||
"positive_feedback": positive,
|
||
"negative_feedback": negative,
|
||
"negative_rate": round(negative / total, 4) if total else 0,
|
||
"reason_tag_counts": [
|
||
{"tag": tag, "count": count}
|
||
for tag, count in sorted(reason_counts.items(), key=lambda item: (-item[1], item[0]))
|
||
],
|
||
"answer_kind_counts": [
|
||
{"answer_kind": answer_kind, "count": count}
|
||
for answer_kind, count in sorted(answer_kind_counts.items(), key=lambda item: (-item[1], item[0]))
|
||
],
|
||
"recent_feedback": [self._pmflow_feedback_payload(row) for row in rows[:8]],
|
||
}
|
||
|
||
async def _prompt_optimizer_candidates(self, *, tenant_id: int | None, prompt_key: str) -> list[dict[str, Any]]:
|
||
filters = [
|
||
OptimizerCandidate.target_type == "prompt_key",
|
||
OptimizerCandidate.target_ref == str(prompt_key or "").strip(),
|
||
OptimizerCandidate.status == "draft",
|
||
]
|
||
if tenant_id is not None:
|
||
filters.append(OptimizerCandidate.tenant_id == int(tenant_id))
|
||
rows = list(
|
||
(
|
||
await self.db.execute(
|
||
select(OptimizerCandidate)
|
||
.where(and_(*filters))
|
||
.order_by(OptimizerCandidate.created_at.desc(), OptimizerCandidate.id.desc())
|
||
.limit(8)
|
||
)
|
||
).scalars().all()
|
||
)
|
||
return [
|
||
{
|
||
"id": int(row.id),
|
||
"tenant_id": int(row.tenant_id),
|
||
"candidate_type": row.candidate_type,
|
||
"target_type": row.target_type,
|
||
"target_ref": row.target_ref,
|
||
"status": row.status,
|
||
"suggestion": row.suggestion or {},
|
||
"evidence": row.evidence or {},
|
||
"notes": row.notes,
|
||
"created_at": row.created_at.isoformat() if row.created_at else None,
|
||
}
|
||
for row in rows
|
||
]
|
||
|
||
@staticmethod
|
||
def _pmflow_feedback_payload(row: PMFlowAIFeedbackEvent) -> dict[str, Any]:
|
||
metadata = row.feedback_metadata if isinstance(row.feedback_metadata, dict) else {}
|
||
prompt_optimization = metadata.get("promptOptimization") if isinstance(metadata.get("promptOptimization"), dict) else {}
|
||
return {
|
||
"id": int(row.id),
|
||
"tenant_id": int(row.tenant_id),
|
||
"project_id": int(row.project_id),
|
||
"target_type": row.target_type,
|
||
"target_id": row.target_id,
|
||
"answer_kind": row.answer_kind,
|
||
"rating": row.rating,
|
||
"reason_tags": list(row.reason_tags or []),
|
||
"comment": row.comment,
|
||
"scene_key": row.scene_key,
|
||
"prompt_release_id": int(row.prompt_release_id) if row.prompt_release_id is not None else None,
|
||
"runtime_profile_id": row.runtime_profile_id,
|
||
"optimizer_candidate_id": prompt_optimization.get("candidateId"),
|
||
"created_at": row.created_at.isoformat() if row.created_at else None,
|
||
}
|
||
|
||
async def _active_golden_cases(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
status: str | None,
|
||
limit: int,
|
||
) -> list[PromptGoldenCase]:
|
||
stmt = (
|
||
select(PromptGoldenCase)
|
||
.where(
|
||
PromptGoldenCase.tenant_id == int(tenant_id),
|
||
PromptGoldenCase.prompt_key == str(prompt_key or "").strip(),
|
||
)
|
||
.order_by(PromptGoldenCase.id.asc())
|
||
.limit(max(1, min(int(limit or 20), 100)))
|
||
)
|
||
effective_status = str(status or "active").strip()
|
||
if effective_status and effective_status != "all":
|
||
stmt = stmt.where(PromptGoldenCase.status == effective_status)
|
||
return list((await self.db.execute(stmt)).scalars().all())
|
||
|
||
async def _active_release_for_prompt(self, *, tenant_id: int, prompt_key: str) -> PromptRelease | None:
|
||
return (
|
||
await self.db.execute(
|
||
select(PromptRelease).where(
|
||
PromptRelease.tenant_id == int(tenant_id),
|
||
PromptRelease.prompt_key == str(prompt_key or "").strip(),
|
||
PromptRelease.status == "active",
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
def _build_release_check_specs(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
entry: dict[str, Any],
|
||
release: PromptRelease,
|
||
default_template: str,
|
||
reason: str | None,
|
||
active_case_count: int,
|
||
latest_replay: dict[str, Any] | None,
|
||
) -> list[dict[str, Any]]:
|
||
release_body = str(release.body_snapshot or "")
|
||
specs: list[dict[str, Any]] = []
|
||
|
||
try:
|
||
default_vars = set(_template_variables_or_error(default_template))
|
||
release_vars = set(_template_variables_or_error(release_body))
|
||
unknown_vars = sorted(release_vars - default_vars)
|
||
if unknown_vars:
|
||
specs.append(
|
||
{
|
||
"check_type": "variable_contract",
|
||
"status": "failed",
|
||
"severity": "blocking",
|
||
"summary": f"模板引用了未登记变量：{', '.join(unknown_vars)}",
|
||
"details_json": {"default_variables": sorted(default_vars), "release_variables": sorted(release_vars), "unknown_variables": unknown_vars},
|
||
}
|
||
)
|
||
else:
|
||
specs.append(
|
||
{
|
||
"check_type": "variable_contract",
|
||
"status": "passed",
|
||
"severity": "info",
|
||
"summary": "变量契约通过，未引入默认模板之外的新变量。",
|
||
"details_json": {"default_variables": sorted(default_vars), "release_variables": sorted(release_vars), "reason": reason},
|
||
}
|
||
)
|
||
except ValueError as exc:
|
||
specs.append(
|
||
{
|
||
"check_type": "variable_contract",
|
||
"status": "failed",
|
||
"severity": "blocking",
|
||
"summary": f"模板语法错误：{exc}",
|
||
"details_json": {"reason": reason},
|
||
}
|
||
)
|
||
|
||
contract_text = " ".join(
|
||
[
|
||
str(entry.get("input_contract") or ""),
|
||
str(entry.get("output_contract") or ""),
|
||
str(entry.get("prompt_key") or ""),
|
||
]
|
||
).lower()
|
||
structured = "json" in contract_text or "结构化" in contract_text
|
||
if not structured:
|
||
specs.append(
|
||
{
|
||
"check_type": "schema_smoke",
|
||
"status": "skipped",
|
||
"severity": "info",
|
||
"summary": "该 prompt 未声明结构化 JSON 输出，schema smoke 本轮跳过。",
|
||
"details_json": {"reason": reason},
|
||
}
|
||
)
|
||
elif "json" in release_body.lower() or "json" in contract_text:
|
||
specs.append(
|
||
{
|
||
"check_type": "schema_smoke",
|
||
"status": "passed",
|
||
"severity": "info",
|
||
"summary": "结构化输出契约已声明 JSON/schema 约束。",
|
||
"details_json": {"output_contract": str(entry.get("output_contract") or ""), "reason": reason},
|
||
}
|
||
)
|
||
else:
|
||
specs.append(
|
||
{
|
||
"check_type": "schema_smoke",
|
||
"status": "failed",
|
||
"severity": "blocking",
|
||
"summary": "结构化 prompt 缺少 JSON/schema 输出约束。",
|
||
"details_json": {"output_contract": str(entry.get("output_contract") or ""), "reason": reason},
|
||
}
|
||
)
|
||
|
||
_ = tenant_id
|
||
replay_status = str((latest_replay or {}).get("run_status") or "")
|
||
replay_failed_count = int((latest_replay or {}).get("failed_count") or 0)
|
||
replay_case_count = int((latest_replay or {}).get("case_count") or 0)
|
||
replay_risks = list((latest_replay or {}).get("risk_flags_json") or [])
|
||
replay_passed = replay_status == "completed" and replay_case_count > 0 and replay_failed_count == 0
|
||
if active_case_count <= 0:
|
||
specs.append(
|
||
{
|
||
"check_type": "golden_replay",
|
||
"status": "skipped",
|
||
"severity": "blocking" if str(entry.get("editable_policy") or "") == "guarded_release" else "warning",
|
||
"summary": "尚未接入 active golden case，不能把 replay 结果伪造成通过。",
|
||
"details_json": {"case_count": 0, "reason": reason},
|
||
}
|
||
)
|
||
elif replay_passed:
|
||
specs.append(
|
||
{
|
||
"check_type": "golden_replay",
|
||
"status": "passed",
|
||
"severity": "info",
|
||
"summary": f"最近一次 replay 已通过：{replay_case_count} 个样本，{int((latest_replay or {}).get('warning_count') or 0)} 个需复核，无阻塞失败。",
|
||
"details_json": {
|
||
"case_count": active_case_count,
|
||
"run_id": (latest_replay or {}).get("id"),
|
||
"mode": (latest_replay or {}).get("run_mode") or "contract_smoke",
|
||
"risk_flags": replay_risks,
|
||
"reason": reason,
|
||
},
|
||
}
|
||
)
|
||
else:
|
||
if latest_replay:
|
||
summary = str(latest_replay.get("summary") or "最近一次 replay 未通过。")
|
||
details = {
|
||
"case_count": active_case_count,
|
||
"run_id": latest_replay.get("id"),
|
||
"run_status": latest_replay.get("run_status"),
|
||
"failed_count": replay_failed_count,
|
||
"risk_flags": replay_risks,
|
||
"reason": reason,
|
||
}
|
||
status = "failed" if replay_failed_count > 0 or replay_risks else "skipped"
|
||
else:
|
||
summary = "已有 active golden case，但尚未执行 release replay run。"
|
||
details = {"case_count": active_case_count, "reason": reason}
|
||
status = "skipped"
|
||
specs.append(
|
||
{
|
||
"check_type": "golden_replay",
|
||
"status": status,
|
||
"severity": "blocking" if str(entry.get("editable_policy") or "") == "guarded_release" else "warning",
|
||
"summary": summary,
|
||
"details_json": details,
|
||
}
|
||
)
|
||
return specs
|
||
|
||
async def _active_golden_case_count(self, *, tenant_id: int, prompt_key: str) -> int:
|
||
return int(
|
||
(
|
||
await self.db.execute(
|
||
select(func.count(PromptGoldenCase.id)).where(
|
||
PromptGoldenCase.tenant_id == int(tenant_id),
|
||
PromptGoldenCase.prompt_key == str(prompt_key or "").strip(),
|
||
PromptGoldenCase.status == "active",
|
||
)
|
||
)
|
||
).scalar()
|
||
or 0
|
||
)
|
||
|
||
@staticmethod
|
||
def _check_payload(row: PromptReleaseCheck) -> dict[str, Any]:
|
||
return {
|
||
"id": int(row.id),
|
||
"prompt_key": str(row.prompt_key or ""),
|
||
"release_id": int(row.release_id),
|
||
"check_type": str(row.check_type or ""),
|
||
"status": str(row.status or ""),
|
||
"severity": str(row.severity or ""),
|
||
"summary": str(row.summary or ""),
|
||
"details_json": row.details_json or {},
|
||
"created_by": int(row.created_by) if row.created_by is not None else None,
|
||
"created_at": row.created_at.isoformat() if getattr(row, "created_at", None) is not None else None,
|
||
"completed_at": row.completed_at,
|
||
}
|
||
|
||
@staticmethod
|
||
def _golden_case_payload(row: PromptGoldenCase) -> dict[str, Any]:
|
||
return {
|
||
"id": int(row.id),
|
||
"prompt_key": str(row.prompt_key or ""),
|
||
"case_name": str(row.case_name or ""),
|
||
"status": str(row.status or ""),
|
||
"input_payload": row.input_payload or {},
|
||
"expected_contract": row.expected_contract or {},
|
||
"expected_contains": row.expected_contains,
|
||
"source": str(row.source or ""),
|
||
"source_ai_call_log_id": int(row.source_ai_call_log_id) if row.source_ai_call_log_id is not None else None,
|
||
"case_type": str(row.case_type or "regression"),
|
||
"priority": str(row.priority or "blocking"),
|
||
"scenario_key": row.scenario_key,
|
||
"tags": row.tags or [],
|
||
"last_run_status": row.last_run_status,
|
||
"last_failed_at": row.last_failed_at,
|
||
"created_by": int(row.created_by) if row.created_by is not None else None,
|
||
"notes": row.notes,
|
||
"created_at": row.created_at.isoformat() if getattr(row, "created_at", None) is not None else None,
|
||
}
|
||
|
||
async def _evaluate_replay_case(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
release: PromptRelease,
|
||
baseline: PromptRelease | None,
|
||
golden_case: PromptGoldenCase,
|
||
entry: dict[str, Any],
|
||
include_llm: bool,
|
||
judge_enabled: bool,
|
||
run_id: int,
|
||
) -> dict[str, Any]:
|
||
payload = golden_case.input_payload if isinstance(golden_case.input_payload, dict) else {}
|
||
candidate_template = str(release.body_snapshot or "")
|
||
baseline_template = str(baseline.body_snapshot or "") if baseline is not None else ""
|
||
candidate_output = ""
|
||
baseline_output = ""
|
||
failures: list[str] = []
|
||
warnings: list[str] = []
|
||
risks: list[str] = []
|
||
try:
|
||
candidate_output = candidate_template.format(**payload).strip()
|
||
except Exception as exc:
|
||
failures.append(f"candidate_render_failed: {exc}")
|
||
risks.append("candidate_render_failed")
|
||
if baseline_template:
|
||
try:
|
||
baseline_output = baseline_template.format(**payload).strip()
|
||
except Exception as exc:
|
||
warnings.append(f"baseline_render_failed: {exc}")
|
||
risks.append("baseline_render_failed")
|
||
baseline_ai_call_log_id = None
|
||
candidate_ai_call_log_id = None
|
||
judge_score = None
|
||
judge_label = None
|
||
judge_reason = None
|
||
estimated_cost = 0.0
|
||
if include_llm and not failures:
|
||
baseline_model_output = baseline_output
|
||
candidate_model_output = candidate_output
|
||
try:
|
||
if baseline_output:
|
||
baseline_model_output, baseline_ai_call_log_id = await self._run_llm_replay_prompt(
|
||
tenant_id=tenant_id,
|
||
prompt_key=str(release.prompt_key or ""),
|
||
release_id=int(baseline.id) if baseline is not None else None,
|
||
prompt=baseline_output,
|
||
function_name="prompt_replay_baseline",
|
||
run_id=run_id,
|
||
)
|
||
candidate_model_output, candidate_ai_call_log_id = await self._run_llm_replay_prompt(
|
||
tenant_id=tenant_id,
|
||
prompt_key=str(release.prompt_key or ""),
|
||
release_id=int(release.id),
|
||
prompt=candidate_output,
|
||
function_name="prompt_replay_candidate",
|
||
run_id=run_id,
|
||
)
|
||
baseline_output = baseline_model_output
|
||
candidate_output = candidate_model_output
|
||
if judge_enabled:
|
||
judge = await self._judge_replay_case(
|
||
tenant_id=tenant_id,
|
||
entry=entry,
|
||
golden_case=golden_case,
|
||
baseline_output=baseline_output,
|
||
candidate_output=candidate_output,
|
||
run_id=run_id,
|
||
)
|
||
judge_score = judge.get("score")
|
||
judge_label = str(judge.get("label") or "manual_review")
|
||
judge_reason = str(judge.get("reason") or "")
|
||
risks.extend([str(item) for item in judge.get("risk_flags") or [] if str(item).strip()])
|
||
decision = str(judge.get("decision") or "")
|
||
if decision == "block_release" or judge_label in {"worse", "risky", "invalid"}:
|
||
failures.append(judge_reason or "LLM judge blocked this replay case")
|
||
elif decision in {"needs_revision", "manual_review"} or judge_label == "same":
|
||
warnings.append(judge_reason or "LLM judge requires manual review")
|
||
except Exception as exc:
|
||
failures.append(f"llm_replay_failed: {exc}")
|
||
risks.append("llm_replay_failed")
|
||
contract_result = self._evaluate_expected_contract(
|
||
expected_contract=golden_case.expected_contract if isinstance(golden_case.expected_contract, dict) else {},
|
||
expected_contains=golden_case.expected_contains,
|
||
input_payload=payload,
|
||
candidate_output=candidate_output,
|
||
entry=entry,
|
||
)
|
||
failures.extend(contract_result["failures"])
|
||
warnings.extend(contract_result["warnings"])
|
||
risks.extend(contract_result["risk_flags"])
|
||
status = "failed" if failures else "warning" if warnings else "passed"
|
||
return {
|
||
"status": status,
|
||
"baseline_output": baseline_output,
|
||
"candidate_output": candidate_output,
|
||
"contract_result_json": {
|
||
"case_name": golden_case.case_name,
|
||
"case_type": str(golden_case.case_type or "regression"),
|
||
"priority": str(golden_case.priority or "blocking"),
|
||
"failures": failures,
|
||
"warnings": warnings,
|
||
"mode": "llm_compare" if include_llm else "contract_smoke_no_llm",
|
||
"judge": {
|
||
"score": judge_score,
|
||
"label": judge_label,
|
||
"reason": judge_reason,
|
||
},
|
||
},
|
||
"diff_summary": self._diff_summary(baseline_output=baseline_output, candidate_output=candidate_output),
|
||
"risk_flags": sorted(set(risks)),
|
||
"baseline_ai_call_log_id": baseline_ai_call_log_id,
|
||
"candidate_ai_call_log_id": candidate_ai_call_log_id,
|
||
"judge_score": judge_score,
|
||
"judge_label": judge_label,
|
||
"judge_reason": judge_reason,
|
||
"estimated_cost": estimated_cost,
|
||
"error_summary": "；".join(failures or warnings)[:800] or None,
|
||
}
|
||
|
||
@staticmethod
|
||
def _effective_replay_status(raw_status: str, *, priority: str) -> str:
|
||
status = str(raw_status or "").strip() or "warning"
|
||
case_priority = str(priority or "blocking").strip()
|
||
if status != "failed":
|
||
return status
|
||
if case_priority == "blocking":
|
||
return "failed"
|
||
return "warning"
|
||
|
||
async def _run_llm_replay_prompt(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
release_id: int | None,
|
||
prompt: str,
|
||
function_name: str,
|
||
run_id: int,
|
||
) -> tuple[str, int | None]:
|
||
routing_meta = {
|
||
"prompt_key": prompt_key,
|
||
"prompt_release_id": release_id,
|
||
"prompt_source": "replay_candidate" if function_name.endswith("candidate") else "replay_baseline",
|
||
"tier_alias": "low",
|
||
"run_id": f"prompt-replay-{run_id}-{function_name}",
|
||
"scene_key": "prompt_replay",
|
||
}
|
||
output = await llm_service.generate(
|
||
prompt,
|
||
db=self.db,
|
||
tenant_id=tenant_id,
|
||
function_name=function_name,
|
||
temperature=0.0,
|
||
max_tokens=1200,
|
||
routing_meta=routing_meta,
|
||
)
|
||
call_id = await self._latest_ai_call_log_id(
|
||
tenant_id=tenant_id,
|
||
prompt_key=prompt_key,
|
||
release_id=release_id,
|
||
function_name=function_name,
|
||
)
|
||
return str(output or "").strip(), call_id
|
||
|
||
async def _judge_replay_case(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
entry: dict[str, Any],
|
||
golden_case: PromptGoldenCase,
|
||
baseline_output: str,
|
||
candidate_output: str,
|
||
run_id: int,
|
||
) -> dict[str, Any]:
|
||
rubric = get_prompt_eval_rubric(
|
||
prompt_key=str(entry.get("prompt_key") or ""),
|
||
prompt_family=str(entry.get("prompt_family") or ""),
|
||
)
|
||
judge_prompt = "\n".join(
|
||
[
|
||
"你是 PMFlow 平台 Prompt 发布评估器。只输出 JSON，不输出解释性正文。",
|
||
"评估候选 Prompt 输出相对 baseline 是否适合发布。",
|
||
f"Prompt Key: {entry.get('prompt_key')}",
|
||
f"业务场景: {entry.get('business_scenario')}",
|
||
f"输入契约: {entry.get('input_contract')}",
|
||
f"输出契约: {entry.get('output_contract')}",
|
||
f"Rubric: {json.dumps(rubric, ensure_ascii=False)}",
|
||
f"Golden Case: {golden_case.case_name}",
|
||
f"Expected Contract: {json.dumps(golden_case.expected_contract or {}, ensure_ascii=False)}",
|
||
f"Baseline Output:\n{baseline_output or '无 baseline'}",
|
||
f"Candidate Output:\n{candidate_output}",
|
||
"输出 JSON schema: {\"score\":0.0,\"label\":\"better|same|worse|risky|invalid\",\"decision\":\"recommend_release|needs_revision|block_release|manual_review\",\"reason\":\"...\",\"risk_flags\":[],\"dimension_scores\":{\"contract_adherence\":0,\"business_correctness\":0,\"evidence_grounding\":0,\"safety_boundary\":0,\"operator_readability\":0}}",
|
||
]
|
||
)
|
||
try:
|
||
result = await llm_service.generate_structured_object(
|
||
judge_prompt,
|
||
db=self.db,
|
||
tenant_id=tenant_id,
|
||
function_name="prompt_replay_judge",
|
||
temperature=0.0,
|
||
max_tokens=900,
|
||
routing_meta={
|
||
"prompt_key": str(entry.get("prompt_key") or ""),
|
||
"prompt_source": "replay_judge",
|
||
"tier_alias": "structured",
|
||
"run_id": f"prompt-replay-{run_id}-judge",
|
||
"scene_key": "prompt_replay",
|
||
},
|
||
)
|
||
payload = result.payload if isinstance(result.payload, dict) else {}
|
||
except Exception as exc:
|
||
return {
|
||
"score": None,
|
||
"label": "manual_review",
|
||
"decision": "manual_review",
|
||
"reason": f"Judge 结构化评估失败：{exc}",
|
||
"risk_flags": ["judge_invalid"],
|
||
}
|
||
return self._normalize_judge_payload(payload)
|
||
|
||
@staticmethod
|
||
def _normalize_judge_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
label = str(payload.get("label") or "manual_review").strip()
|
||
if label not in {"better", "same", "worse", "risky", "invalid"}:
|
||
label = "manual_review"
|
||
decision = str(payload.get("decision") or "manual_review").strip()
|
||
if decision not in {"recommend_release", "needs_revision", "block_release", "manual_review"}:
|
||
decision = "manual_review"
|
||
try:
|
||
score = float(payload.get("score")) if payload.get("score") is not None else None
|
||
except (TypeError, ValueError):
|
||
score = None
|
||
return {
|
||
"score": score,
|
||
"label": label,
|
||
"decision": decision,
|
||
"reason": str(payload.get("reason") or "").strip()[:1200],
|
||
"risk_flags": [str(item).strip() for item in list(payload.get("risk_flags") or []) if str(item).strip()],
|
||
"dimension_scores": payload.get("dimension_scores") if isinstance(payload.get("dimension_scores"), dict) else {},
|
||
}
|
||
|
||
async def _latest_ai_call_log_id(
|
||
self,
|
||
*,
|
||
tenant_id: int,
|
||
prompt_key: str,
|
||
release_id: int | None,
|
||
function_name: str,
|
||
) -> int | None:
|
||
filters = [
|
||
AICallLog.tenant_id == int(tenant_id),
|
||
AICallLog.prompt_key == str(prompt_key or "").strip(),
|
||
AICallLog.function_name == str(function_name or "").strip(),
|
||
]
|
||
if release_id is not None:
|
||
filters.append(AICallLog.prompt_release_id == int(release_id))
|
||
row = (
|
||
await self.db.execute(
|
||
select(AICallLog)
|
||
.where(and_(*filters))
|
||
.order_by(AICallLog.id.desc())
|
||
.limit(1)
|
||
)
|
||
).scalar_one_or_none()
|
||
return int(row.id) if row is not None else None
|
||
|
||
@staticmethod
|
||
def _evaluate_expected_contract(
|
||
*,
|
||
expected_contract: dict[str, Any],
|
||
expected_contains: str | None,
|
||
input_payload: dict[str, Any],
|
||
candidate_output: str,
|
||
entry: dict[str, Any],
|
||
) -> dict[str, list[str]]:
|
||
failures: list[str] = []
|
||
warnings: list[str] = []
|
||
risk_flags: list[str] = []
|
||
required_inputs = expected_contract.get("required_input_keys") or expected_contract.get("required_variables") or []
|
||
if isinstance(required_inputs, str):
|
||
required_inputs = [required_inputs]
|
||
if isinstance(required_inputs, list):
|
||
missing = [str(item) for item in required_inputs if str(item) not in input_payload]
|
||
if missing:
|
||
failures.append(f"缺少输入变量：{', '.join(missing)}")
|
||
risk_flags.append("missing_input_variable")
|
||
required_contains = expected_contract.get("required_prompt_contains") or expected_contract.get("required_contains") or []
|
||
if isinstance(required_contains, str):
|
||
required_contains = [required_contains]
|
||
if isinstance(required_contains, list):
|
||
missing_text = [str(item) for item in required_contains if str(item) and str(item) not in candidate_output]
|
||
if missing_text:
|
||
failures.append(f"候选渲染缺少期望片段：{', '.join(missing_text[:5])}")
|
||
risk_flags.append("missing_required_prompt_fragment")
|
||
if expected_contains and str(expected_contains).strip() not in candidate_output:
|
||
failures.append("候选渲染未包含 expected_contains。")
|
||
risk_flags.append("missing_expected_contains")
|
||
forbidden_terms = expected_contract.get("forbidden_terms") or []
|
||
if isinstance(forbidden_terms, str):
|
||
forbidden_terms = [forbidden_terms]
|
||
if isinstance(forbidden_terms, list):
|
||
hits = [str(item) for item in forbidden_terms if str(item) and str(item) in candidate_output]
|
||
if hits:
|
||
failures.append(f"候选渲染包含禁止片段：{', '.join(hits[:5])}")
|
||
risk_flags.append("forbidden_prompt_fragment")
|
||
output_required = expected_contract.get("required") or expected_contract.get("required_keys")
|
||
if output_required:
|
||
warnings.append("该 golden case 声明了输出字段契约；contract smoke 不执行 LLM，只能记录为人工复核项。")
|
||
risk_flags.append("unsupported_output_contract")
|
||
if str(entry.get("output_contract") or "").lower().find("json") >= 0 and "json" not in candidate_output.lower():
|
||
warnings.append("资产声明 JSON 输出，但候选 prompt 渲染未显式包含 JSON 约束。")
|
||
risk_flags.append("json_contract_weak")
|
||
return {"failures": failures, "warnings": warnings, "risk_flags": risk_flags}
|
||
|
||
@staticmethod
|
||
def _diff_summary(*, baseline_output: str, candidate_output: str) -> str:
|
||
if not baseline_output:
|
||
return "无 baseline release，仅验证候选渲染。"
|
||
if baseline_output == candidate_output:
|
||
return "候选渲染与 baseline 一致。"
|
||
return f"候选渲染与 baseline 不同：baseline {len(baseline_output)} 字符，candidate {len(candidate_output)} 字符。"
|
||
|
||
@staticmethod
|
||
def _replay_run_payload(
|
||
row: PromptReplayRun,
|
||
*,
|
||
include_results: bool,
|
||
results: list[PromptReplayCaseResult] | None = None,
|
||
) -> dict[str, Any]:
|
||
payload = {
|
||
"id": int(row.id),
|
||
"prompt_key": str(row.prompt_key or ""),
|
||
"release_id": int(row.release_id),
|
||
"baseline_release_id": int(row.baseline_release_id) if row.baseline_release_id is not None else None,
|
||
"run_status": str(row.run_status or ""),
|
||
"run_mode": str(row.run_mode or ""),
|
||
"case_count": int(row.case_count or 0),
|
||
"passed_count": int(row.passed_count or 0),
|
||
"failed_count": int(row.failed_count or 0),
|
||
"warning_count": int(row.warning_count or 0),
|
||
"score_delta": float(row.score_delta or 0),
|
||
"summary": row.summary,
|
||
"risk_flags_json": row.risk_flags_json or [],
|
||
"created_by": int(row.created_by) if row.created_by is not None else None,
|
||
"started_at": row.started_at,
|
||
"completed_at": row.completed_at,
|
||
"created_at": row.created_at.isoformat() if getattr(row, "created_at", None) is not None else None,
|
||
}
|
||
if include_results:
|
||
payload["results"] = [
|
||
{
|
||
"id": int(item.id),
|
||
"run_id": int(item.run_id),
|
||
"golden_case_id": int(item.golden_case_id),
|
||
"prompt_key": str(item.prompt_key or ""),
|
||
"release_id": int(item.release_id),
|
||
"baseline_output": item.baseline_output,
|
||
"candidate_output": item.candidate_output,
|
||
"status": str(item.status or ""),
|
||
"contract_result_json": item.contract_result_json or {},
|
||
"diff_summary": item.diff_summary,
|
||
"risk_flags_json": item.risk_flags_json or [],
|
||
"baseline_ai_call_log_id": int(item.baseline_ai_call_log_id) if item.baseline_ai_call_log_id is not None else None,
|
||
"candidate_ai_call_log_id": int(item.candidate_ai_call_log_id) if item.candidate_ai_call_log_id is not None else None,
|
||
"judge_score": float(item.judge_score) if item.judge_score is not None else None,
|
||
"judge_label": item.judge_label,
|
||
"judge_reason": item.judge_reason,
|
||
"latency_ms": item.latency_ms,
|
||
"estimated_cost": float(item.estimated_cost or 0),
|
||
"error_summary": item.error_summary,
|
||
"created_at": item.created_at.isoformat() if getattr(item, "created_at", None) is not None else None,
|
||
}
|
||
for item in list(results or [])
|
||
]
|
||
return payload
|
||
|
||
@staticmethod
|
||
def _call_log_summary(row: AICallLog) -> dict[str, Any]:
|
||
return {
|
||
"id": int(row.id),
|
||
"trace_id": row.trace_id,
|
||
"status": row.status,
|
||
"function_name": row.function_name,
|
||
"prompt_release_id": int(row.prompt_release_id) if row.prompt_release_id is not None else None,
|
||
"prompt_source": row.prompt_source,
|
||
"latency_ms": row.latency_ms,
|
||
"estimated_cost": float(row.cost or 0),
|
||
"created_at": row.created_at.isoformat() if row.created_at else None,
|
||
"error_summary": _preview(row.response or "", limit=240),
|
||
}
|
||
|
||
@staticmethod
|
||
def _is_failed_status(status: str) -> bool:
|
||
text = str(status or "").strip().lower()
|
||
return text not in {"success", "ok", "success_cache_hit", "success_failover"} and (
|
||
"fail" in text or "error" in text or "invalid" in text or "empty" in text
|
||
)
|
||
|
||
@staticmethod
|
||
def _gate_summary_from_checks(items: list[dict[str, Any]]) -> dict[str, Any]:
|
||
latest: dict[str, dict[str, Any]] = {}
|
||
for item in items:
|
||
check_type = str(item.get("check_type") or "")
|
||
if check_type and check_type not in latest:
|
||
latest[check_type] = item
|
||
if not latest:
|
||
return {
|
||
"status": "not_run",
|
||
"summary": "尚未运行发布门禁检查。",
|
||
"blocking_reasons": ["尚未运行发布门禁检查。"],
|
||
"required_checks": list(RELEASE_CHECK_TYPES),
|
||
}
|
||
blocking_reasons: list[str] = []
|
||
for check_type in RELEASE_CHECK_TYPES:
|
||
item = latest.get(check_type)
|
||
if item is None:
|
||
blocking_reasons.append(f"{check_type} 尚未运行。")
|
||
continue
|
||
if item.get("status") == "failed":
|
||
blocking_reasons.append(str(item.get("summary") or f"{check_type} 未通过。"))
|
||
elif item.get("status") == "skipped" and item.get("severity") == "blocking":
|
||
blocking_reasons.append(str(item.get("summary") or f"{check_type} 被跳过。"))
|
||
if blocking_reasons:
|
||
return {
|
||
"status": "blocked",
|
||
"summary": "；".join(blocking_reasons),
|
||
"blocking_reasons": blocking_reasons,
|
||
"required_checks": list(RELEASE_CHECK_TYPES),
|
||
}
|
||
return {
|
||
"status": "passed",
|
||
"summary": "发布门禁检查已通过。",
|
||
"blocking_reasons": [],
|
||
"required_checks": list(RELEASE_CHECK_TYPES),
|
||
}
|
||
|
||
@staticmethod
|
||
def _validate_release_body(*, default_template: str, body_snapshot: str) -> None:
|
||
body = str(body_snapshot or "").strip()
|
||
if not body:
|
||
raise ValueError("body_snapshot is required")
|
||
try:
|
||
default_vars = set(_template_variables_or_error(default_template))
|
||
release_vars = set(_template_variables_or_error(body))
|
||
except ValueError as exc:
|
||
raise ValueError(f"invalid prompt template syntax: {exc}") from exc
|
||
unknown_vars = sorted(release_vars - default_vars)
|
||
if unknown_vars:
|
||
raise ValueError(f"release template references unknown variables: {', '.join(unknown_vars)}")
|
||
|
||
@staticmethod
|
||
def _default_template(prompt_key: str) -> str:
|
||
try:
|
||
ai_prompts = import_module("app.config.ai_prompts")
|
||
except Exception:
|
||
return ""
|
||
value = getattr(ai_prompts, str(prompt_key or "").strip(), "")
|
||
return str(value or "").strip() if isinstance(value, str) else ""
|
||
|
||
@staticmethod
|
||
def _migration_status(entry: dict[str, Any], default_template: str) -> str:
|
||
policy = str(entry.get("editable_policy") or "")
|
||
if policy == "needs_migration":
|
||
return "inline_prompt_needs_runtime_migration"
|
||
if not default_template:
|
||
return "definition_only"
|
||
if policy == "guarded_release":
|
||
return "guarded_runtime_configurable"
|
||
return "runtime_configurable"
|