yuqei/ai-contract-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add ai quality attribution insights

Browse Source
This commit is contained in:
yuqei
2026-06-23 01:38:27 +08:00
parent 963b6d54b0
commit 00f4ebb652
9 changed files with 514 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
project-docs/V2-ai-quality-root-cause-attribution-2026-06-23.md
+61
View File
@@ -0,0 +1,61 @@
# V2 AI 质量问题归因闭环记录
日期：2026-06-23
## 本轮目标
在 AI 质量运营看板基础上，继续把异常按模型、Prompt 模板、知识库和标签类型聚合，帮助管理员找出最常见的质量失败来源。
## 已落地内容
1. 后端质量看板增加归因聚合。
- 返回字段：`attribution`
- 维度：
- `by_model`
- `by_prompt_template`
- `by_knowledge_base`
- `by_label_code`
2. 每个归因项包含：
- `total_runs`：该来源涉及的总调用数。
- `affected_runs`：其中带 warning/error 质量标签的调用数。
- `warning_labels` / `error_labels`
- `pending_labels`
- `fallback_runs`
- `average_quality_score`
- `top_label_codes`
- `root_cause_hint`
3. 归因提示。
- 模型维度：提示检查 Provider 配置、模型延迟、fallback。
- Prompt 维度：提示检查 token、引用约束、模板版本。
- 知识库维度：提示检查文档覆盖、索引质量、来源定位。
- 标签维度：针对 `source_limited`、`fallback_used`、`token_high`、`latency_high` 等给出处理方向。
4. 契约和 SDK 同步。
- OpenAPI 新增 `AiQualityAttribution` 和 `AiQualityAttributionItem`。
- Python SDK 新增对应模型并导出。
- `get_quality_dashboard()` 可直接解析归因结果。
5. 前端看板增强。
- `/admin/ai-quality-dashboard` 新增四个归因区块：
- 模型归因
- Prompt 归因
- 知识库归因
- 标签归因
- 每个区块展示 Top 来源、异常占比、待处理数、常见标签和建议。
## 验收重点
- 访问 `/admin/ai-quality-dashboard` 能看到四组归因区块。
- 当存在 warning/error 标签时，归因区块展示 affected runs、pending labels 和 top label codes。
- 后端 `GET /api/v1/audit/quality-dashboard` 返回 `attribution`。
- SDK `get_quality_dashboard()` 能解析 `dashboard.attribution.by_model` 和 `dashboard.attribution.by_label_code`。
## 下一步建议
进入 AI 质量改进动作闭环：
- 给每个归因项增加“创建改进任务”入口。
- 将问题归因转成待办，例如“补知识库”“调整 Prompt”“切换模型”“修复 Provider 配置”。
- 在看板中追踪改进任务完成后，对应异常率是否下降。
yuqei-ai-platform/packages/ai-openapi/openapi.yaml
+53
View File
@@ -1401,6 +1401,57 @@ components:
properties:
days:
type: integer
AiQualityAttributionItem:
type: object
properties:
dimension:
type: string
enum: [model, prompt_template, knowledge_base, label_code]
key:
type: string
label:
type: string
total_runs:
type: integer
affected_runs:
type: integer
warning_labels:
type: integer
error_labels:
type: integer
pending_labels:
type: integer
fallback_runs:
type: integer
average_quality_score:
type: number
format: float
top_label_codes:
type: array
items:
type: string
root_cause_hint:
type: string
nullable: true
AiQualityAttribution:
type: object
properties:
by_model:
type: array
items:
$ref: "#/components/schemas/AiQualityAttributionItem"
by_prompt_template:
type: array
items:
$ref: "#/components/schemas/AiQualityAttributionItem"
by_knowledge_base:
type: array
items:
$ref: "#/components/schemas/AiQualityAttributionItem"
by_label_code:
type: array
items:
$ref: "#/components/schemas/AiQualityAttributionItem"
AiQualityDashboard:
type: object
properties:
@@ -1414,6 +1465,8 @@ components:
type: array
items:
$ref: "#/components/schemas/AiQualityPendingLabel"
attribution:
$ref: "#/components/schemas/AiQualityAttribution"
AiRun:
type: object
required:
yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/repository.py
+238
View File
@@ -285,6 +285,28 @@ class AiQualityPendingLabel(BaseModel):
quality_score: int = 0
class AiQualityAttributionItem(BaseModel):
dimension: str
key: str
label: str
total_runs: int = 0
affected_runs: int = 0
warning_labels: int = 0
error_labels: int = 0
pending_labels: int = 0
fallback_runs: int = 0
average_quality_score: float = 0.0
top_label_codes: list[str] = Field(default_factory=list)
root_cause_hint: str | None = None
class AiQualityAttribution(BaseModel):
by_model: list[AiQualityAttributionItem] = Field(default_factory=list)
by_prompt_template: list[AiQualityAttributionItem] = Field(default_factory=list)
by_knowledge_base: list[AiQualityAttributionItem] = Field(default_factory=list)
by_label_code: list[AiQualityAttributionItem] = Field(default_factory=list)
class AiQualityDashboardSummary(BaseModel):
days: int
total_runs: int = 0
@@ -306,6 +328,7 @@ class AiQualityDashboard(BaseModel):
summary: AiQualityDashboardSummary
daily_metrics: list[AiQualityDailyMetric] = Field(default_factory=list)
pending_labels: list[AiQualityPendingLabel] = Field(default_factory=list)
attribution: AiQualityAttribution = Field(default_factory=AiQualityAttribution)
class AiPlatformRepository(Protocol):
@@ -1508,6 +1531,7 @@ def _build_quality_dashboard(
for metric in daily.values():
_finalize_quality_rates(metric)
_finalize_quality_rates(summary)
attribution = _build_quality_attribution(runs)
pending_labels = sorted(pending_labels, key=lambda item: item.created_at, reverse=True)[:pending_limit]
ordered_daily = [daily[date] for date in dates if date in daily]
extra_dates = sorted((date for date in daily if date not in dates), reverse=False)
@@ -1516,6 +1540,7 @@ def _build_quality_dashboard(
summary=summary,
daily_metrics=ordered_daily,
pending_labels=pending_labels,
attribution=attribution,
)
@@ -1555,6 +1580,219 @@ def _finalize_quality_rates(metric: AiQualityDailyMetric | AiQualityDashboardSum
metric.latency_issue_rate = round(metric.latency_issue_runs / metric.total_runs, 4)
def _build_quality_attribution(runs: list[AiRunAudit]) -> AiQualityAttribution:
by_model: dict[str, dict[str, object]] = {}
by_prompt_template: dict[str, dict[str, object]] = {}
by_knowledge_base: dict[str, dict[str, object]] = {}
by_label_code: dict[str, dict[str, object]] = {}
for run in runs:
issue_labels = [label for label in run.quality_labels if label.severity in {"warning", "error"}]
model_key = f"{run.provider or 'unknown'} / {run.model or 'unknown'}"
_touch_attribution_group(by_model, "model", model_key, model_key, run, issue_labels)
prompt_key = run.prompt_template_id or "unknown"
_touch_attribution_group(
by_prompt_template,
"prompt_template",
prompt_key,
prompt_key if prompt_key != "unknown" else "未记录 Prompt 模板",
run,
issue_labels,
)
knowledge_base_ids = sorted(
{
source.knowledge_base_id
for source in run.retrieved_sources
if source.knowledge_base_id
}
)
if not knowledge_base_ids and any(label.code in {"source_insufficient", "source_limited"} for label in issue_labels):
knowledge_base_ids = ["no_knowledge_source"]
for knowledge_base_id in knowledge_base_ids:
_touch_attribution_group(
by_knowledge_base,
"knowledge_base",
knowledge_base_id,
"未命中知识来源" if knowledge_base_id == "no_knowledge_source" else knowledge_base_id,
run,
issue_labels,
)
for label in issue_labels:
_touch_label_attribution_group(by_label_code, run, label)
return AiQualityAttribution(
by_model=_finalize_attribution_groups(by_model),
by_prompt_template=_finalize_attribution_groups(by_prompt_template),
by_knowledge_base=_finalize_attribution_groups(by_knowledge_base),
by_label_code=_finalize_attribution_groups(by_label_code),
)
def _touch_attribution_group(
groups: dict[str, dict[str, object]],
dimension: str,
key: str,
label: str,
run: AiRunAudit,
issue_labels: list[AiQualityLabel],
) -> None:
group = groups.setdefault(key, _new_attribution_group(dimension, key, label))
_record_attribution_run(group, run, issue_labels)
def _touch_label_attribution_group(
groups: dict[str, dict[str, object]],
run: AiRunAudit,
label: AiQualityLabel,
) -> None:
group = groups.setdefault(
label.code,
_new_attribution_group("label_code", label.code, _label_title(label)),
)
_record_attribution_run(group, run, [label])
def _new_attribution_group(dimension: str, key: str, label: str) -> dict[str, object]:
return {
"dimension": dimension,
"key": key,
"label": label,
"run_ids": set(),
"affected_run_ids": set(),
"fallback_run_ids": set(),
"warning_labels": 0,
"error_labels": 0,
"pending_labels": 0,
"score_sum": 0,
"score_count": 0,
"label_counts": {},
}
def _record_attribution_run(
group: dict[str, object],
run: AiRunAudit,
issue_labels: list[AiQualityLabel],
) -> None:
run_ids = group["run_ids"]
affected_run_ids = group["affected_run_ids"]
fallback_run_ids = group["fallback_run_ids"]
label_counts = group["label_counts"]
assert isinstance(run_ids, set)
assert isinstance(affected_run_ids, set)
assert isinstance(fallback_run_ids, set)
assert isinstance(label_counts, dict)
if run.id not in run_ids:
run_ids.add(run.id)
group["score_sum"] = int(group["score_sum"]) + run.quality_score
group["score_count"] = int(group["score_count"]) + 1
if issue_labels:
affected_run_ids.add(run.id)
if run.status == "fallback" or any(label.code == "fallback_used" for label in run.quality_labels):
fallback_run_ids.add(run.id)
for label in issue_labels:
if label.severity == "error":
group["error_labels"] = int(group["error_labels"]) + 1
elif label.severity == "warning":
group["warning_labels"] = int(group["warning_labels"]) + 1
if label.review_status == "pending":
group["pending_labels"] = int(group["pending_labels"]) + 1
label_counts[label.code] = int(label_counts.get(label.code, 0)) + 1
def _finalize_attribution_groups(groups: dict[str, dict[str, object]]) -> list[AiQualityAttributionItem]:
items: list[AiQualityAttributionItem] = []
for group in groups.values():
run_ids = group["run_ids"]
affected_run_ids = group["affected_run_ids"]
fallback_run_ids = group["fallback_run_ids"]
label_counts = group["label_counts"]
assert isinstance(run_ids, set)
assert isinstance(affected_run_ids, set)
assert isinstance(fallback_run_ids, set)
assert isinstance(label_counts, dict)
score_count = int(group["score_count"])
top_label_codes = [
str(code)
for code, _count in sorted(
label_counts.items(),
key=lambda item: (-int(item[1]), str(item[0])),
)[:3]
]
item = AiQualityAttributionItem(
dimension=str(group["dimension"]),
key=str(group["key"]),
label=str(group["label"]),
total_runs=len(run_ids),
affected_runs=len(affected_run_ids),
warning_labels=int(group["warning_labels"]),
error_labels=int(group["error_labels"]),
pending_labels=int(group["pending_labels"]),
fallback_runs=len(fallback_run_ids),
average_quality_score=round(int(group["score_sum"]) / score_count, 2) if score_count else 0.0,
top_label_codes=top_label_codes,
root_cause_hint=_root_cause_hint(str(group["dimension"]), str(group["key"]), top_label_codes),
)
if item.affected_runs > 0 or item.fallback_runs > 0 or item.pending_labels > 0:
items.append(item)
return sorted(
items,
key=lambda item: (
item.pending_labels,
item.error_labels,
item.warning_labels,
item.affected_runs,
item.fallback_runs,
-item.average_quality_score,
),
reverse=True,
)[:5]
def _label_title(label: AiQualityLabel) -> str:
return label.title or label.code
def _root_cause_hint(dimension: str, key: str, top_label_codes: list[str]) -> str | None:
top_codes = set(top_label_codes)
if dimension == "model":
if "fallback_used" in top_codes or "provider_failed" in top_codes:
return "模型或 Provider 配置异常集中，优先检查 API Key、模型可用性和超时。"
if "latency_high" in top_codes or "latency_critical" in top_codes:
return "该模型耗时异常集中，优先检查模型延迟、网络和 max_tokens。"
return "该模型相关质量异常较集中，优先对比其他模型的回答质量和成本。"
if dimension == "prompt_template":
if "token_high" in top_codes:
return "Prompt token 偏高，优先压缩模板、减少上下文或拆分任务。"
if "missing_citation" in top_codes:
return "Prompt 对依据引用约束不足，优先加强引用格式和来源要求。"
return "Prompt 模板相关异常集中，优先检查提示词结构和版本发布记录。"
if dimension == "knowledge_base":
if key == "no_knowledge_source":
return "检索未命中知识来源，优先补充知识库、检查切分和索引状态。"
return "该知识库相关异常集中，优先检查文档覆盖、索引质量和来源定位。"
if dimension == "label_code":
hints = {
"source_insufficient": "知识检索没有来源，优先检查知识库索引、检索词和过滤条件。",
"source_limited": "来源数量偏少，优先补充文档或放宽检索召回策略。",
"fallback_used": "模型调用走回退，优先检查模型服务配置和稳定性。",
"provider_failed": "Provider 返回失败，优先检查 API Key、base_url、模型名和额度。",
"token_high": "Token 用量偏高，优先压缩 Prompt 和上下文。",
"token_missing": "Provider 未返回 token，用量和费用统计可能不完整。",
"latency_high": "耗时偏高，优先检查模型延迟、网络和上下文长度。",
"latency_critical": "耗时严重异常，优先设置超时、降级或异步处理。",
"missing_citation": "回答缺少依据，优先加强 RAG 检索和引用约束。",
"run_failed": "调用失败，优先检查错误日志和 provider 状态。",
}
return hints.get(key)
return None
def _update_quality_label_list(
labels: list[AiQualityLabel],
label_code: str,
yuqei-ai-platform/services/ai-platform-api/tests/test_platform_data.py
+4
View File
@@ -183,6 +183,10 @@ def test_quality_dashboard_summarizes_daily_metrics_and_pending_labels() -> None
assert len(dashboard["daily_metrics"]) == 7
assert dashboard["daily_metrics"][-1]["total_runs"] == 1
assert dashboard["pending_labels"]
label_attribution = dashboard["attribution"]["by_label_code"]
assert {item["key"] for item in label_attribution} >= {"fallback_used", "source_limited", "token_missing"}
assert dashboard["attribution"]["by_model"][0]["affected_runs"] == 1
assert dashboard["attribution"]["by_knowledge_base"]
first_pending = dashboard["pending_labels"][0]
update_response = client.patch(
yuqei-ai-sdk-python/src/yuqei_sdk/__init__.py
+4
View File
@@ -1,5 +1,7 @@
from yuqei_sdk.ai_platform import (
AiPlatformClient,
AiQualityAttribution,
AiQualityAttributionItem,
AiQualityDashboard,
AiQualityDashboardSummary,
AiQualityDailyMetric,
@@ -32,6 +34,8 @@ from yuqei_sdk.context import RequestContext
__all__ = [
"AiPlatformClient",
"AiQualityAttribution",
"AiQualityAttributionItem",
"AiQualityDashboard",
"AiQualityDashboardSummary",
"AiQualityDailyMetric",
yuqei-ai-sdk-python/src/yuqei_sdk/ai_platform.py
+23
View File
@@ -299,10 +299,33 @@ class AiQualityPendingLabel(BaseModel):
quality_score: int = 0
class AiQualityAttributionItem(BaseModel):
dimension: str
key: str
label: str
total_runs: int = 0
affected_runs: int = 0
warning_labels: int = 0
error_labels: int = 0
pending_labels: int = 0
fallback_runs: int = 0
average_quality_score: float = 0.0
top_label_codes: list[str] = Field(default_factory=list)
root_cause_hint: str | None = None
class AiQualityAttribution(BaseModel):
by_model: list[AiQualityAttributionItem] = Field(default_factory=list)
by_prompt_template: list[AiQualityAttributionItem] = Field(default_factory=list)
by_knowledge_base: list[AiQualityAttributionItem] = Field(default_factory=list)
by_label_code: list[AiQualityAttributionItem] = Field(default_factory=list)
class AiQualityDashboard(BaseModel):
summary: AiQualityDashboardSummary
daily_metrics: list[AiQualityDailyMetric] = Field(default_factory=list)
pending_labels: list[AiQualityPendingLabel] = Field(default_factory=list)
attribution: AiQualityAttribution = Field(default_factory=AiQualityAttribution)
class AiPlatformClient:
yuqei-ai-sdk-python/tests/test_ai_platform.py
+39
View File
@@ -493,6 +493,42 @@ def test_ai_platform_client_covers_admin_data_endpoints() -> None:
"quality_score": 100,
}
],
"attribution": {
"by_model": [
{
"dimension": "model",
"key": "deepseek / deepseek-chat",
"label": "deepseek / deepseek-chat",
"total_runs": 1,
"affected_runs": 1,
"warning_labels": 1,
"error_labels": 0,
"pending_labels": 1,
"fallback_runs": 0,
"average_quality_score": 90.0,
"top_label_codes": ["source_limited"],
"root_cause_hint": "该模型相关质量异常较集中，优先对比其他模型的回答质量和成本。",
}
],
"by_prompt_template": [],
"by_knowledge_base": [],
"by_label_code": [
{
"dimension": "label_code",
"key": "source_limited",
"label": "Source limited",
"total_runs": 1,
"affected_runs": 1,
"warning_labels": 1,
"error_labels": 0,
"pending_labels": 1,
"fallback_runs": 0,
"average_quality_score": 90.0,
"top_label_codes": ["source_limited"],
"root_cause_hint": "来源数量偏少，优先补充文档或放宽检索召回策略。",
}
],
},
},
)
if request.url.path == "/api/v1/audit/ai-runs/run-1/quality-labels/citation_present":
@@ -628,6 +664,9 @@ def test_ai_platform_client_covers_admin_data_endpoints() -> None:
assert dashboard.summary.pass_rate == 1.0
assert dashboard.daily_metrics[0].date == "2026-06-22"
assert dashboard.pending_labels[0].run_id == "run-1"
assert dashboard.attribution.by_model[0].key == "deepseek / deepseek-chat"
assert dashboard.attribution.by_label_code[0].top_label_codes == ["source_limited"]
assert dashboard.attribution.by_label_code[0].root_cause_hint is not None
updated_audit = client.update_ai_run_quality_label(
"run-1",
"citation_present",
yuqei-contract-platform/apps/contract-web/src/app/admin/ai-quality-dashboard/page.tsx
+65
View File
@@ -37,10 +37,33 @@ type QualityPendingLabel = {
quality_score: number;
};
type QualityAttributionItem = {
dimension: string;
key: string;
label: string;
total_runs: number;
affected_runs: number;
warning_labels: number;
error_labels: number;
pending_labels: number;
fallback_runs: number;
average_quality_score: number;
top_label_codes: string[];
root_cause_hint?: string | null;
};
type QualityAttribution = {
by_model: QualityAttributionItem[];
by_prompt_template: QualityAttributionItem[];
by_knowledge_base: QualityAttributionItem[];
by_label_code: QualityAttributionItem[];
};
type QualityDashboard = {
summary: QualityDailyMetric & { days: number };
daily_metrics: QualityDailyMetric[];
pending_labels: QualityPendingLabel[];
attribution?: QualityAttribution;
};
type LoadState =
@@ -128,6 +151,13 @@ export default function AiQualityDashboardPage() {
<Metric icon={<AlertTriangle size={18} />} label="待处理标签" value={String(dashboard.summary.pending_labels)} />
</div>
<div className="attribution-grid">
<AttributionPanel title="模型归因" items={dashboard.attribution?.by_model ?? []} />
<AttributionPanel title="Prompt 归因" items={dashboard.attribution?.by_prompt_template ?? []} />
<AttributionPanel title="知识库归因" items={dashboard.attribution?.by_knowledge_base ?? []} />
<AttributionPanel title="标签归因" items={dashboard.attribution?.by_label_code ?? []} />
</div>
<div className="dashboard-grid">
<section className="panel">
<div className="panel-header">
@@ -213,6 +243,41 @@ function Metric({ icon, label, value }: { icon: ReactNode; label: string; value:
);
}
function AttributionPanel({ title, items }: { title: string; items: QualityAttributionItem[] }) {
return (
<section className="panel">
<div className="panel-header">
<h2 className="panel-title">{title}</h2>
<span className="tag">Top {items.length}</span>
</div>
<div className="panel-body attribution-list">
{items.length === 0 ? (
<div className="empty-state empty-state-compact">暂无异常归因。</div>
) : (
items.map((item) => (
<div className="attribution-item" key={`${item.dimension}-${item.key}`}>
<div className="list-item-header">
<strong>{item.label}</strong>
<span className="tag tag-warning">{item.affected_runs}/{item.total_runs}</span>
</div>
<div className="trace-meta">
<span>待处理 {item.pending_labels}</span>
<span>warning {item.warning_labels}</span>
<span>error {item.error_labels}</span>
<span>均分 {item.average_quality_score}</span>
</div>
{item.top_label_codes.length > 0 && (
<span className="item-meta">常见标签：{item.top_label_codes.join("、")}</span>
)}
{item.root_cause_hint && <span className="item-meta">{item.root_cause_hint}</span>}
</div>
))
)}
</div>
</section>
);
}
function formatPercent(value: number) {
return `${Math.round(value * 100)}%`;
}
yuqei-contract-platform/apps/contract-web/src/app/globals.css
+27
View File
@@ -519,6 +519,11 @@ select {
background: var(--danger-soft);
}
.empty-state-compact {
min-height: 80px;
padding: 14px;
}
.trace-list {
display: grid;
gap: 12px;
@@ -695,6 +700,28 @@ select {
gap: 16px;
}
.attribution-grid {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(240px, 1fr));
gap: 12px;
margin-bottom: 16px;
}
.attribution-list {
display: grid;
gap: 10px;
}
.attribution-item {
display: grid;
gap: 8px;
min-width: 0;
padding: 12px;
border: 1px solid var(--line);
border-radius: 8px;
background: #fbfcfe;
}
.trend-list {
display: grid;
gap: 10px;