365 lines
13 KiB
Python
365 lines
13 KiB
Python
"""统一反馈中心的原因池与候选生成。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import OrderedDict
|
|
from typing import Any
|
|
|
|
|
|
_REASON_DEFINITIONS: list[dict[str, Any]] = [
|
|
{
|
|
"code": "owner_mismatch",
|
|
"label": "负责人不匹配",
|
|
"description": "AI 推荐、人工指派或协作责任边界需要复核。",
|
|
"source_types": {"task_breakdown", "task_override"},
|
|
"field_names": {"assignee_id"},
|
|
"confidence": 0.96,
|
|
},
|
|
{
|
|
"code": "capability_mismatch",
|
|
"label": "能力匹配偏差",
|
|
"description": "人员能力、技能覆盖或职责分配与任务要求不一致。",
|
|
"source_types": {"task_breakdown", "task_override"},
|
|
"field_names": {"assignee_id", "required_skill_tag_ids", "required_skill_names"},
|
|
"confidence": 0.91,
|
|
},
|
|
{
|
|
"code": "skill_gap",
|
|
"label": "技能要求不准",
|
|
"description": "任务技能标签、能力要求或技能覆盖范围需要修正。",
|
|
"source_types": {"task_breakdown", "task_override"},
|
|
"field_names": {"required_skill_tag_ids", "required_skill_names"},
|
|
"confidence": 0.94,
|
|
},
|
|
{
|
|
"code": "priority_misalignment",
|
|
"label": "优先级错配",
|
|
"description": "优先级建议与业务紧急度或协作节奏不匹配。",
|
|
"source_types": {"task_breakdown", "task_override"},
|
|
"field_names": {"priority"},
|
|
"confidence": 0.89,
|
|
},
|
|
{
|
|
"code": "dependency_wrong",
|
|
"label": "依赖关系错误",
|
|
"description": "任务之间的依赖、先后顺序或阻塞链路判断有误。",
|
|
"source_types": {"task_breakdown", "task_override"},
|
|
"field_names": {"depends_on_task_keys"},
|
|
"confidence": 0.92,
|
|
},
|
|
{
|
|
"code": "risk_underestimated",
|
|
"label": "风险低估",
|
|
"description": "任务风险等级、风险说明或执行风险识别偏低。",
|
|
"source_types": {"task_breakdown", "task_override", "signal_feedback"},
|
|
"field_names": {"risk_level", "risk_notes", "strategic_classification"},
|
|
"confidence": 0.9,
|
|
},
|
|
{
|
|
"code": "effort_estimation_bias",
|
|
"label": "工时预估偏差",
|
|
"description": "任务预估工时与人工判断差距较大。",
|
|
"source_types": {"task_breakdown", "task_override"},
|
|
"field_names": {"estimated_hours"},
|
|
"confidence": 0.95,
|
|
},
|
|
{
|
|
"code": "scope_overbroad",
|
|
"label": "范围过大",
|
|
"description": "任务标题、描述或业务范围切分不够清晰。",
|
|
"source_types": {"task_breakdown", "task_override"},
|
|
"field_names": {"title", "description"},
|
|
"confidence": 0.82,
|
|
},
|
|
{
|
|
"code": "incomplete_output",
|
|
"label": "输出信息不完整",
|
|
"description": "AI 输出缺少关键字段、上下文或业务约束。",
|
|
"source_types": {"task_breakdown", "task_override", "qa_feedback"},
|
|
"field_names": {"task_object", "qa_feedback"},
|
|
"confidence": 0.78,
|
|
},
|
|
{
|
|
"code": "answer_incorrect",
|
|
"label": "答案不正确",
|
|
"description": "QA 回复存在事实错误、遗漏或推断偏差。",
|
|
"source_types": {"qa_feedback"},
|
|
"field_names": {"qa_feedback"},
|
|
"confidence": 0.93,
|
|
},
|
|
{
|
|
"code": "source_issue",
|
|
"label": "来源问题",
|
|
"description": "QA 的来源引用、检索材料或证据组织存在问题。",
|
|
"source_types": {"qa_feedback"},
|
|
"field_names": {"qa_feedback"},
|
|
"confidence": 0.86,
|
|
},
|
|
{
|
|
"code": "strategic_judgement_mismatch",
|
|
"label": "战略判断偏差",
|
|
"description": "AI 对信号是否战略相关的判断与人工判断不一致。",
|
|
"source_types": {"signal_feedback"},
|
|
"field_names": {"strategic_classification"},
|
|
"confidence": 0.97,
|
|
},
|
|
{
|
|
"code": "signal_noise",
|
|
"label": "噪声信号",
|
|
"description": "信号本身噪声过高或不值得进入后续流程。",
|
|
"source_types": {"signal_feedback"},
|
|
"field_names": {"strategic_classification"},
|
|
"confidence": 0.8,
|
|
},
|
|
{
|
|
"code": "manual_override",
|
|
"label": "人工覆盖",
|
|
"description": "字段由人工介入修改后形成反馈样本。",
|
|
"source_types": {"task_breakdown", "task_override", "qa_feedback", "signal_feedback"},
|
|
"field_names": {"*", "task_object", "qa_feedback", "strategic_classification"},
|
|
"confidence": 0.75,
|
|
},
|
|
{
|
|
"code": "manual_review_needed",
|
|
"label": "需要人工复核",
|
|
"description": "当前样本仍需要人工确认，不足以自动通过。",
|
|
"source_types": {"task_breakdown", "task_override", "qa_feedback", "signal_feedback"},
|
|
"field_names": {"*"},
|
|
"confidence": 0.7,
|
|
},
|
|
]
|
|
|
|
|
|
def _candidate_payload(definition: dict[str, Any], *, source_type: str, field_name: str) -> dict[str, Any]:
|
|
return {
|
|
"code": definition["code"],
|
|
"label": definition["label"],
|
|
"description": definition["description"],
|
|
"source_type": source_type,
|
|
"field_name": field_name,
|
|
"confidence": float(definition.get("confidence") or 0.0),
|
|
}
|
|
|
|
|
|
def list_reason_catalog(
|
|
*,
|
|
source_type: str | None = None,
|
|
field_name: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
items: list[dict[str, Any]] = []
|
|
for definition in _REASON_DEFINITIONS:
|
|
if source_type and source_type not in definition["source_types"]:
|
|
continue
|
|
if field_name and "*" not in definition["field_names"] and field_name not in definition["field_names"]:
|
|
continue
|
|
items.append(
|
|
{
|
|
"code": definition["code"],
|
|
"label": definition["label"],
|
|
"description": definition["description"],
|
|
"source_types": sorted(definition["source_types"]),
|
|
"field_names": sorted(definition["field_names"]),
|
|
"confidence": float(definition.get("confidence") or 0.0),
|
|
}
|
|
)
|
|
return items
|
|
|
|
|
|
def _append_unique(candidate_map: OrderedDict[str, dict[str, Any]], candidate: dict[str, Any]) -> None:
|
|
code = str(candidate.get("code") or "").strip()
|
|
if not code or code in candidate_map:
|
|
return
|
|
candidate_map[code] = candidate
|
|
|
|
|
|
def _normalize_text(value: Any) -> str:
|
|
return str(value or "").strip().lower()
|
|
|
|
|
|
def _is_truthy_change(before_value: Any, after_value: Any) -> bool:
|
|
if before_value is None and after_value is None:
|
|
return False
|
|
return before_value != after_value
|
|
|
|
|
|
def _estimated_hours_direction(before_value: Any, after_value: Any) -> str:
|
|
try:
|
|
before = float(before_value or 0)
|
|
after = float(after_value or 0)
|
|
except (TypeError, ValueError):
|
|
return "unknown"
|
|
if after > before:
|
|
return "increase"
|
|
if after < before:
|
|
return "decrease"
|
|
return "flat"
|
|
|
|
|
|
def build_reason_candidates(
|
|
*,
|
|
source_type: str,
|
|
field_name: str,
|
|
before_value: Any = None,
|
|
after_value: Any = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""基于字段差异构建稳定、可解释的候选原因。"""
|
|
|
|
source_key = _normalize_text(source_type)
|
|
field_key = _normalize_text(field_name)
|
|
candidates: OrderedDict[str, dict[str, Any]] = OrderedDict()
|
|
|
|
def add(code: str, label: str, confidence: float, description: str) -> None:
|
|
_append_unique(
|
|
candidates,
|
|
{
|
|
"code": code,
|
|
"label": label,
|
|
"description": description,
|
|
"source_type": source_key,
|
|
"field_name": field_key,
|
|
"confidence": float(confidence),
|
|
},
|
|
)
|
|
|
|
if source_key == "task_override" or source_key == "task_breakdown":
|
|
if field_key in {"assignee_id", "owner_id"}:
|
|
add(
|
|
"owner_mismatch",
|
|
"负责人不匹配",
|
|
0.96 if _is_truthy_change(before_value, after_value) else 0.88,
|
|
"负责人或责任归属与任务实际执行人不一致。",
|
|
)
|
|
add(
|
|
"capability_mismatch",
|
|
"能力匹配偏差",
|
|
0.91,
|
|
"人员能力、技能覆盖或职责分配与任务要求不一致。",
|
|
)
|
|
if field_key in {"required_skill_tag_ids", "required_skill_names"}:
|
|
add(
|
|
"skill_gap",
|
|
"技能要求不准",
|
|
0.94 if _is_truthy_change(before_value, after_value) else 0.82,
|
|
"任务所需技能标签需要复核。",
|
|
)
|
|
add(
|
|
"capability_mismatch",
|
|
"能力匹配偏差",
|
|
0.9,
|
|
"人员能力、技能覆盖或职责分配与任务要求不一致。",
|
|
)
|
|
if field_key in {"depends_on_task_keys", "dependency", "dependencies"}:
|
|
add(
|
|
"dependency_wrong",
|
|
"依赖关系错误",
|
|
0.92,
|
|
"任务依赖、前置关系或阻塞关系需要修正。",
|
|
)
|
|
if field_key == "priority":
|
|
add(
|
|
"priority_misalignment",
|
|
"优先级错配",
|
|
0.9,
|
|
"优先级建议与业务紧急度或协作节奏不匹配。",
|
|
)
|
|
if field_key == "estimated_hours":
|
|
direction = _estimated_hours_direction(before_value, after_value)
|
|
confidence = 0.95 if direction in {"increase", "decrease"} else 0.8
|
|
add(
|
|
"effort_estimation_bias",
|
|
"工时预估偏差",
|
|
confidence,
|
|
"AI 工时预估与人工判断存在明显差异。",
|
|
)
|
|
if direction == "increase":
|
|
add(
|
|
"scope_overbroad",
|
|
"范围过大",
|
|
0.78,
|
|
"任务范围可能需要进一步拆分。",
|
|
)
|
|
if field_key in {"risk_level", "risk_notes"}:
|
|
add(
|
|
"risk_underestimated",
|
|
"风险低估",
|
|
0.9,
|
|
"任务风险说明可能低估了实际执行复杂度。",
|
|
)
|
|
if field_key in {"title", "description"}:
|
|
add(
|
|
"scope_overbroad",
|
|
"范围过大",
|
|
0.84,
|
|
"任务标题或描述可能覆盖了过多内容。",
|
|
)
|
|
if field_key == "task_object":
|
|
add(
|
|
"incomplete_output",
|
|
"输出信息不完整",
|
|
0.76,
|
|
"模型输出仍可能缺失关键细节，需要人工补全。",
|
|
)
|
|
add("manual_override", "人工覆盖", 0.78, "AI 拆解结果经过人工修改后形成学习样本。")
|
|
elif source_key == "qa_feedback":
|
|
rating = _normalize_text((after_value or {}).get("rating") if isinstance(after_value, dict) else None)
|
|
error_types = []
|
|
if isinstance(after_value, dict):
|
|
error_types = [str(item).strip().lower() for item in after_value.get("error_types") or [] if str(item).strip()]
|
|
if rating == "negative":
|
|
if any(token in {"hallucination", "幻觉", "错误", "wrong"} for token in error_types):
|
|
add(
|
|
"answer_incorrect",
|
|
"答案不正确",
|
|
0.95,
|
|
"反馈显示回答内容存在事实错误或幻觉。",
|
|
)
|
|
add(
|
|
"source_issue",
|
|
"来源问题",
|
|
0.84,
|
|
"回答所依赖的来源或证据组织需要复核。",
|
|
)
|
|
add(
|
|
"incomplete_output",
|
|
"输出信息不完整",
|
|
0.78,
|
|
"回答缺少关键信息或上下文不够完整。",
|
|
)
|
|
else:
|
|
add("manual_review_needed", "需要人工复核", 0.72, "QA 反馈建议保留人工复核以确认样本质量。")
|
|
elif source_key == "signal_feedback":
|
|
predicted = None
|
|
actual = None
|
|
if isinstance(before_value, dict):
|
|
predicted = before_value.get("ai_predicted_strategic")
|
|
if isinstance(after_value, dict):
|
|
actual = after_value.get("was_strategic")
|
|
if predicted is not None and actual is not None and bool(predicted) != bool(actual):
|
|
add(
|
|
"strategic_judgement_mismatch",
|
|
"战略性判断偏差",
|
|
0.97,
|
|
"AI 对信号是否需要进入战略流程的判断与人工结果不一致。",
|
|
)
|
|
add(
|
|
"signal_noise",
|
|
"噪声信号",
|
|
0.82,
|
|
"当前信号可能缺少足够强的业务价值信号。",
|
|
)
|
|
else:
|
|
add("manual_review_needed", "需要人工复核", 0.7, "当前样本需要进一步人工确认。")
|
|
|
|
if not candidates:
|
|
add(
|
|
"manual_review_needed",
|
|
"需要人工复核",
|
|
0.7,
|
|
"当前样本需要进一步人工确认。",
|
|
)
|
|
|
|
return list(candidates.values())
|
|
|
|
|
|
__all__ = ["build_reason_candidates", "list_reason_catalog"]
|