280 lines
10 KiB
Python
280 lines
10 KiB
Python
from __future__ import annotations
|
||
|
||
from .shared import *
|
||
|
||
class ModelCapabilityValidationMixin:
|
||
def __init__(self, db: AsyncSession, ai_service: AIService):
|
||
self.db = db
|
||
self.ai_service = ai_service
|
||
|
||
async def validate_model_requirement_match(
|
||
self,
|
||
truth_node_id: int,
|
||
model_version_id: int,
|
||
tenant_id: int,
|
||
project_id: Optional[int] = None,
|
||
) -> Dict[str, Any]:
|
||
"""智能匹配校验：评估需求与模型的匹配度
|
||
|
||
Args:
|
||
truth_node_id: 宪章（需求）ID
|
||
model_version_id: 模型版本 ID
|
||
tenant_id: 租户 ID
|
||
|
||
Returns:
|
||
包含匹配度、成本预估、风险提示、能力差距等信息的字典
|
||
"""
|
||
# 获取宪章和模型信息
|
||
truth_node = await self._get_truth_node(truth_node_id, tenant_id)
|
||
model_version = await self._get_model_version(model_version_id)
|
||
|
||
if not truth_node:
|
||
raise ValueError(f"宪章 {truth_node_id} 不存在")
|
||
if not model_version:
|
||
raise ValueError(f"模型版本 {model_version_id} 不存在")
|
||
if project_id is not None:
|
||
await self.validate_truth_node_in_project(
|
||
truth_node_id=truth_node_id,
|
||
tenant_id=tenant_id,
|
||
project_id=project_id,
|
||
)
|
||
|
||
# 格式化模型信息用于 AI 分析
|
||
model_info = self._format_model_info(model_version)
|
||
charter_content = str(truth_node.content or truth_node.title or "")
|
||
|
||
# AI 评估匹配度
|
||
evaluation = await self.ai_service.evaluate_model_requirement(
|
||
charter_content=charter_content,
|
||
model_version_id=model_version_id,
|
||
model_info=json.dumps(model_info, ensure_ascii=False),
|
||
tenant_id=tenant_id,
|
||
db=self.db,
|
||
)
|
||
|
||
# 创建或更新绑定记录
|
||
binding = await self._upsert_binding(
|
||
truth_node_id, model_version_id, evaluation, tenant_id
|
||
)
|
||
|
||
# 检测能力差距
|
||
gaps = await self._detect_capability_gaps(
|
||
truth_node_id, model_version_id, evaluation, tenant_id
|
||
)
|
||
|
||
return {
|
||
"binding_id": binding.id,
|
||
"compatibility_score": evaluation.get("compatibility_score", 0),
|
||
"cost_estimate_usd": evaluation.get("cost_estimate_usd", 0.0),
|
||
"cost_estimate_tokens_per_day": evaluation.get("cost_estimate_tokens_per_day", 0),
|
||
"risk_hints": evaluation.get("risk_hints", []),
|
||
"capability_gaps": gaps,
|
||
"recommendations": evaluation.get("recommendations", []),
|
||
}
|
||
|
||
async def validate_truth_node_in_project(
|
||
self, truth_node_id: int, tenant_id: int, project_id: int
|
||
) -> TruthNode:
|
||
"""校验宪章是否属于指定项目。"""
|
||
truth_node = await self._get_truth_node(truth_node_id, tenant_id)
|
||
if not truth_node:
|
||
raise ValueError(f"宪章 {truth_node_id} 不存在")
|
||
if truth_node.project_id != project_id:
|
||
raise ValueError(
|
||
f"宪章 {truth_node_id} 不属于项目 {project_id}，拒绝跨项目操作"
|
||
)
|
||
return truth_node
|
||
|
||
async def _get_truth_node(
|
||
self, truth_node_id: int, tenant_id: int
|
||
) -> Optional[TruthNode]:
|
||
"""获取宪章"""
|
||
result = await self.db.execute(
|
||
select(TruthNode).where(
|
||
and_(
|
||
TruthNode.id == truth_node_id,
|
||
TruthNode.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
return result.scalar_one_or_none()
|
||
|
||
async def _get_model_version(
|
||
self, model_version_id: int
|
||
) -> Optional[ModelVersion]:
|
||
"""获取模型版本"""
|
||
result = await self.db.execute(
|
||
select(ModelVersion).where(ModelVersion.id == model_version_id)
|
||
)
|
||
return result.scalar_one_or_none()
|
||
|
||
def _format_model_info(self, model: ModelVersion) -> Dict[str, Any]:
|
||
"""格式化模型信息用于 AI 分析"""
|
||
return {
|
||
"name": model.name,
|
||
"provider": model.provider,
|
||
"version": model.version,
|
||
"capability_boundaries": model.capability_boundaries or {},
|
||
"cost_per_1k_tokens": float(model.cost_per_1k_tokens) if model.cost_per_1k_tokens else 0.0,
|
||
}
|
||
|
||
async def _upsert_binding(
|
||
self,
|
||
truth_node_id: int,
|
||
model_version_id: int,
|
||
evaluation: Dict[str, Any],
|
||
tenant_id: int,
|
||
) -> ModelRequirementBinding:
|
||
"""创建或更新模型-需求绑定"""
|
||
# 检查是否已存在绑定
|
||
result = await self.db.execute(
|
||
select(ModelRequirementBinding).where(
|
||
and_(
|
||
ModelRequirementBinding.truth_node_id == truth_node_id,
|
||
ModelRequirementBinding.model_version_id == model_version_id,
|
||
)
|
||
)
|
||
)
|
||
binding = result.scalar_one_or_none()
|
||
|
||
compatibility_score = evaluation.get("compatibility_score", 0)
|
||
cost_estimate = evaluation.get("cost_estimate_usd", 0.0)
|
||
risk_notes = ", ".join(evaluation.get("risk_hints", []))
|
||
|
||
if binding:
|
||
# 更新现有绑定
|
||
binding.compatibility_score = int(compatibility_score)
|
||
binding.estimated_cost = float(cost_estimate)
|
||
binding.risk_notes = risk_notes[:1000] if risk_notes else None # 限制长度
|
||
else:
|
||
# 创建新绑定
|
||
binding = ModelRequirementBinding(
|
||
tenant_id=tenant_id,
|
||
truth_node_id=truth_node_id,
|
||
model_version_id=model_version_id,
|
||
compatibility_score=int(compatibility_score),
|
||
estimated_cost=float(cost_estimate),
|
||
risk_notes=risk_notes[:1000] if risk_notes else None,
|
||
)
|
||
self.db.add(binding)
|
||
|
||
await self.db.commit()
|
||
await self.db.refresh(binding)
|
||
return binding
|
||
|
||
async def _detect_capability_gaps(
|
||
self,
|
||
truth_node_id: int,
|
||
model_version_id: int,
|
||
evaluation: Dict[str, Any],
|
||
tenant_id: int,
|
||
) -> List[Dict[str, Any]]:
|
||
"""检测能力差距并创建预警记录"""
|
||
gaps = []
|
||
|
||
# 检查兼容性评分
|
||
compatibility = evaluation.get("compatibility_score", 100)
|
||
if compatibility < 60:
|
||
gap = await self._create_capability_gap(
|
||
truth_node_id=truth_node_id,
|
||
model_version_id=model_version_id,
|
||
gap_type="capability",
|
||
severity="critical" if compatibility < 40 else "high",
|
||
description=f"模型兼容性评分仅 {compatibility}/100，可能无法满足需求",
|
||
ai_analysis={"compatibility_score": compatibility},
|
||
tenant_id=tenant_id,
|
||
)
|
||
gaps.append(gap)
|
||
|
||
# 检查成本风险
|
||
cost_estimate = evaluation.get("cost_estimate_usd", 0)
|
||
if cost_estimate > 100: # 日成本超过 $100 视为高风险
|
||
gap = await self._create_capability_gap(
|
||
truth_node_id=truth_node_id,
|
||
model_version_id=model_version_id,
|
||
gap_type="cost",
|
||
severity="high" if cost_estimate > 500 else "medium",
|
||
description=f"预估日成本 ${cost_estimate:.2f}，超出常规范围",
|
||
ai_analysis={"cost_estimate_usd": cost_estimate},
|
||
tenant_id=tenant_id,
|
||
)
|
||
gaps.append(gap)
|
||
|
||
# 检查风险提示
|
||
risk_hints = evaluation.get("risk_hints", [])
|
||
if risk_hints:
|
||
gap = await self._create_capability_gap(
|
||
truth_node_id=truth_node_id,
|
||
model_version_id=model_version_id,
|
||
gap_type="risk",
|
||
severity="medium",
|
||
description="检测到潜在风险",
|
||
ai_analysis={"risk_hints": risk_hints},
|
||
tenant_id=tenant_id,
|
||
)
|
||
gaps.append(gap)
|
||
|
||
return gaps
|
||
|
||
async def _create_capability_gap(
|
||
self,
|
||
truth_node_id: int,
|
||
model_version_id: int,
|
||
gap_type: str,
|
||
severity: str,
|
||
description: str,
|
||
ai_analysis: Dict[str, Any],
|
||
tenant_id: int,
|
||
) -> Dict[str, Any]:
|
||
"""创建能力差距记录"""
|
||
# 检查是否已存在相同类型的未解决差距
|
||
existing = await self.db.execute(
|
||
select(ModelCapabilityGap).where(
|
||
and_(
|
||
ModelCapabilityGap.truth_node_id == truth_node_id,
|
||
ModelCapabilityGap.model_version_id == model_version_id,
|
||
ModelCapabilityGap.gap_type == gap_type,
|
||
ModelCapabilityGap.status.in_(["open", "acknowledged"]),
|
||
)
|
||
)
|
||
)
|
||
existing_gap = existing.scalar_one_or_none()
|
||
|
||
if existing_gap:
|
||
# 更新现有差距
|
||
existing_gap.description = description
|
||
existing_gap.ai_analysis = ai_analysis
|
||
existing_gap.severity = severity
|
||
await self.db.commit()
|
||
await self.db.refresh(existing_gap)
|
||
return {
|
||
"id": existing_gap.id,
|
||
"gap_type": existing_gap.gap_type,
|
||
"severity": existing_gap.severity,
|
||
"description": existing_gap.description,
|
||
"status": existing_gap.status,
|
||
}
|
||
else:
|
||
# 创建新差距
|
||
gap = ModelCapabilityGap(
|
||
tenant_id=tenant_id,
|
||
model_version_id=model_version_id,
|
||
truth_node_id=truth_node_id,
|
||
gap_type=gap_type,
|
||
severity=severity,
|
||
description=description,
|
||
ai_analysis=ai_analysis,
|
||
status="open",
|
||
)
|
||
self.db.add(gap)
|
||
await self.db.commit()
|
||
await self.db.refresh(gap)
|
||
return {
|
||
"id": gap.id,
|
||
"gap_type": gap.gap_type,
|
||
"severity": gap.severity,
|
||
"description": gap.description,
|
||
"status": gap.status,
|
||
}
|
||
|