702 lines
24 KiB
Python
702 lines
24 KiB
Python
"""模型能力矩阵核心服务
|
||
|
||
AI 开发团队专属垂直能力矩阵的核心业务逻辑：
|
||
- 模型-需求智能匹配校验
|
||
- 算力成本跟踪与 ROI 核算
|
||
- 能力差距检测与预警
|
||
- 成本优化建议
|
||
"""
|
||
import json
|
||
from datetime import datetime, timedelta
|
||
from typing import Any, Dict, List, Optional, Union
|
||
|
||
from sqlalchemy import select, and_, func, desc
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ..models import (
|
||
ModelVersion,
|
||
ModelRequirementBinding,
|
||
TruthNode,
|
||
Task,
|
||
AICallLog,
|
||
ModelCapabilityGap,
|
||
CostTracking,
|
||
User,
|
||
)
|
||
from ..services.ai_service import AIService
|
||
|
||
|
||
class ModelCapabilityService:
|
||
"""模型能力矩阵核心服务"""
|
||
|
||
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,
|
||
}
|
||
|
||
# ========================================================================
|
||
# 算力成本跟踪与 ROI 核算
|
||
# ========================================================================
|
||
|
||
async def track_cost_and_roi(
|
||
self,
|
||
truth_node_id: int,
|
||
period: str = "weekly",
|
||
tenant_id: int = 1,
|
||
) -> Dict[str, Any]:
|
||
"""算力成本跟踪与 ROI 核算
|
||
|
||
Args:
|
||
truth_node_id: 宪章（需求）ID
|
||
period: 周期类型（daily, weekly, monthly, project）
|
||
tenant_id: 租户 ID
|
||
|
||
Returns:
|
||
包含成本明细、ROI 分析等信息的字典
|
||
"""
|
||
# 获取相关任务和模型绑定
|
||
tasks = await self._get_truth_node_tasks(truth_node_id, tenant_id)
|
||
bindings = await self._get_truth_node_bindings(truth_node_id, tenant_id)
|
||
|
||
if not tasks:
|
||
return {
|
||
"total_tokens": 0,
|
||
"total_cost_usd": 0.0,
|
||
"task_breakdown": [],
|
||
"roi_analysis": {},
|
||
"cost_variance": 0.0,
|
||
}
|
||
|
||
# 汇总 token 使用量和成本
|
||
total_tokens = 0
|
||
total_cost = 0.0
|
||
task_costs = []
|
||
|
||
for task in tasks:
|
||
# 从 AI 调用日志统计 token
|
||
task_tokens = await self._get_task_token_usage(task.id)
|
||
task_cost = await self._calculate_task_cost(task, bindings)
|
||
|
||
total_tokens += task_tokens
|
||
total_cost += task_cost
|
||
task_costs.append({
|
||
"task_id": task.id,
|
||
"task_name": task.title,
|
||
"tokens": task_tokens,
|
||
"cost_usd": round(task_cost, 4),
|
||
})
|
||
|
||
# 获取或创建成本跟踪记录
|
||
cost_tracking = await self._get_or_create_cost_tracking(
|
||
truth_node_id=truth_node_id,
|
||
period=period,
|
||
tenant_id=tenant_id,
|
||
bindings=bindings,
|
||
)
|
||
|
||
# 更新实际数据
|
||
cost_tracking.actual_tokens = total_tokens
|
||
cost_tracking.actual_cost_usd = round(total_cost, 4)
|
||
|
||
# AI 分析 ROI
|
||
truth_node = await self._get_truth_node(truth_node_id, tenant_id)
|
||
roi_analysis = await self._analyze_cost_roi(
|
||
total_cost=total_cost,
|
||
total_tokens=total_tokens,
|
||
task_count=len(tasks),
|
||
truth_node_content=str(truth_node.content or truth_node.title or "") if truth_node else "",
|
||
tenant_id=tenant_id,
|
||
)
|
||
|
||
cost_tracking.roi_score = roi_analysis.get("roi_score")
|
||
cost_tracking.roi_analysis = roi_analysis
|
||
|
||
# 生成优化建议
|
||
optimization = await self._generate_cost_optimization(
|
||
truth_node_id=truth_node_id,
|
||
tenant_id=tenant_id,
|
||
total_cost=total_cost,
|
||
)
|
||
cost_tracking.optimization_suggestions = optimization
|
||
|
||
await self.db.commit()
|
||
await self.db.refresh(cost_tracking)
|
||
|
||
# 计算成本偏差
|
||
cost_variance = total_cost - (cost_tracking.estimated_cost_usd or 0)
|
||
|
||
return {
|
||
"total_tokens": total_tokens,
|
||
"total_cost_usd": round(total_cost, 4),
|
||
"task_breakdown": task_costs,
|
||
"roi_analysis": roi_analysis,
|
||
"cost_variance": round(cost_variance, 4),
|
||
"optimization_suggestions": optimization,
|
||
}
|
||
|
||
async def _get_truth_node_tasks(
|
||
self, truth_node_id: int, tenant_id: int
|
||
) -> List[Task]:
|
||
"""获取宪章下的所有任务"""
|
||
result = await self.db.execute(
|
||
select(Task).where(
|
||
and_(
|
||
Task.truth_node_id == truth_node_id,
|
||
Task.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
return list(result.scalars().all())
|
||
|
||
async def _get_truth_node_bindings(
|
||
self, truth_node_id: int, tenant_id: int
|
||
) -> List[ModelRequirementBinding]:
|
||
"""获取宪章的模型绑定"""
|
||
result = await self.db.execute(
|
||
select(ModelRequirementBinding).where(
|
||
and_(
|
||
ModelRequirementBinding.truth_node_id == truth_node_id,
|
||
ModelRequirementBinding.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
return list(result.scalars().all())
|
||
|
||
async def _get_task_token_usage(self, task_id: int) -> int:
|
||
"""从 AI 调用日志统计任务的 token 使用量"""
|
||
# 注意：需要根据实际日志记录方式调整
|
||
# 这里假设日志中有 task_id 或相关标识
|
||
result = await self.db.execute(
|
||
select(func.sum(AICallLog.total_tokens)).where(
|
||
AICallLog.function_name.contains(f"task_{task_id}")
|
||
)
|
||
)
|
||
total = result.scalar() or 0
|
||
return int(total)
|
||
|
||
async def _calculate_task_cost(
|
||
self, task: Task, bindings: List[ModelRequirementBinding]
|
||
) -> float:
|
||
"""计算任务成本"""
|
||
# 如果有绑定记录，使用绑定的预估成本
|
||
if bindings:
|
||
return sum(float(b.estimated_cost or 0) for b in bindings) / max(len(bindings), 1)
|
||
return 0.0
|
||
|
||
async def _get_or_create_cost_tracking(
|
||
self,
|
||
truth_node_id: int,
|
||
period: str,
|
||
tenant_id: int,
|
||
bindings: List[ModelRequirementBinding],
|
||
) -> CostTracking:
|
||
"""获取或创建成本跟踪记录"""
|
||
# 查找现有记录
|
||
result = await self.db.execute(
|
||
select(CostTracking).where(
|
||
and_(
|
||
CostTracking.truth_node_id == truth_node_id,
|
||
CostTracking.period == period,
|
||
CostTracking.tenant_id == tenant_id,
|
||
)
|
||
).order_by(desc(CostTracking.created_at))
|
||
)
|
||
tracking = result.scalar_one_or_none()
|
||
|
||
if not tracking:
|
||
# 创建新记录
|
||
tracking = CostTracking(
|
||
tenant_id=tenant_id,
|
||
truth_node_id=truth_node_id,
|
||
model_version_id=bindings[0].model_version_id if bindings else None,
|
||
period=period,
|
||
estimated_tokens=0,
|
||
actual_tokens=0,
|
||
estimated_cost_usd=sum(float(b.estimated_cost or 0) for b in bindings),
|
||
actual_cost_usd=0.0,
|
||
roi_score=None,
|
||
roi_analysis={},
|
||
optimization_suggestions={},
|
||
)
|
||
self.db.add(tracking)
|
||
await self.db.commit()
|
||
await self.db.refresh(tracking)
|
||
|
||
return tracking
|
||
|
||
async def _analyze_cost_roi(
|
||
self,
|
||
total_cost: float,
|
||
total_tokens: int,
|
||
task_count: int,
|
||
truth_node_content: str,
|
||
tenant_id: int,
|
||
) -> Dict[str, Any]:
|
||
"""AI 分析成本效益"""
|
||
from ..config.ai_prompts import PROMPT_ANALYZE_COST_ROI
|
||
|
||
prompt = PROMPT_ANALYZE_COST_ROI.format(
|
||
total_cost=f"${total_cost:.2f}",
|
||
total_tokens=f"{total_tokens:,}",
|
||
task_count=task_count,
|
||
truth_node_content=truth_node_content[:2000], # 限制长度
|
||
)
|
||
|
||
try:
|
||
text = await self.ai_service._invoke_with_retry(
|
||
prompt,
|
||
db=self.db,
|
||
tenant_id=tenant_id,
|
||
function_name="analyze_cost_roi",
|
||
)
|
||
return self.ai_service._parse_json(text)
|
||
except Exception:
|
||
# 失败时返回默认分析
|
||
return {
|
||
"roi_score": 5.0,
|
||
"cost_efficiency": "fair",
|
||
"cost_per_task": round(total_cost / max(task_count, 1), 2),
|
||
"optimization_suggestions": [],
|
||
"benchmark_comparison": "无法分析",
|
||
}
|
||
|
||
async def _generate_cost_optimization(
|
||
self, truth_node_id: int, tenant_id: int, total_cost: float
|
||
) -> Dict[str, Any]:
|
||
"""生成成本优化建议"""
|
||
# 简单的优化建议逻辑
|
||
suggestions = []
|
||
|
||
# 检查是否可以使用更便宜的模型
|
||
bindings = await self._get_truth_node_bindings(truth_node_id, tenant_id)
|
||
if bindings:
|
||
for binding in bindings:
|
||
model = await self._get_model_version(binding.model_version_id)
|
||
if model and model.cost_per_1k_tokens and model.cost_per_1k_tokens > 0.001:
|
||
suggestions.append({
|
||
"type": "model_downgrade",
|
||
"priority": "medium",
|
||
"description": f"考虑切换到更便宜的模型替代 {model.name}",
|
||
"estimated_savings_percent": 30,
|
||
})
|
||
|
||
# 检查是否可以启用缓存
|
||
suggestions.append({
|
||
"type": "cache_optimization",
|
||
"priority": "high",
|
||
"description": "启用 Redis 缓存可减少 20-40% 重复调用",
|
||
"estimated_savings_percent": 30,
|
||
})
|
||
|
||
return {
|
||
"suggestions": suggestions,
|
||
"total_potential_savings_percent": 30 if suggestions else 0,
|
||
}
|
||
|
||
# ========================================================================
|
||
# 能力差距管理
|
||
# ========================================================================
|
||
|
||
async def list_capability_gaps(
|
||
self,
|
||
tenant_id: int,
|
||
status: Optional[str] = None,
|
||
severity: Optional[str] = None,
|
||
limit: int = 100,
|
||
) -> Dict[str, Any]:
|
||
"""列出能力差距记录"""
|
||
query = select(ModelCapabilityGap).where(
|
||
ModelCapabilityGap.tenant_id == tenant_id
|
||
)
|
||
|
||
if status:
|
||
query = query.where(ModelCapabilityGap.status == status)
|
||
if severity:
|
||
query = query.where(ModelCapabilityGap.severity == severity)
|
||
|
||
query = query.order_by(desc(ModelCapabilityGap.created_at)).limit(limit)
|
||
|
||
result = await self.db.execute(query)
|
||
gaps = result.scalars().all()
|
||
|
||
return {
|
||
"gaps": [
|
||
{
|
||
"id": g.id,
|
||
"gap_type": g.gap_type,
|
||
"severity": g.severity,
|
||
"description": g.description,
|
||
"ai_analysis": g.ai_analysis,
|
||
"status": g.status,
|
||
"created_at": g.created_at.isoformat() if g.created_at else None,
|
||
"resolved_at": g.resolved_at.isoformat() if g.resolved_at else None,
|
||
}
|
||
for g in gaps
|
||
],
|
||
"total": len(gaps),
|
||
}
|
||
|
||
async def acknowledge_gap(
|
||
self, gap_id: int, tenant_id: int
|
||
) -> Dict[str, Any]:
|
||
"""确认能力差距"""
|
||
result = await self.db.execute(
|
||
select(ModelCapabilityGap).where(
|
||
and_(
|
||
ModelCapabilityGap.id == gap_id,
|
||
ModelCapabilityGap.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
gap = result.scalar_one_or_none()
|
||
|
||
if not gap:
|
||
raise ValueError(f"能力差距 {gap_id} 不存在")
|
||
|
||
gap.status = "acknowledged"
|
||
await self.db.commit()
|
||
|
||
return {"id": gap.id, "status": gap.status}
|
||
|
||
async def resolve_gap(
|
||
self, gap_id: int, tenant_id: int, resolution_notes: Optional[str] = None
|
||
) -> Dict[str, Any]:
|
||
"""解决能力差距"""
|
||
result = await self.db.execute(
|
||
select(ModelCapabilityGap).where(
|
||
and_(
|
||
ModelCapabilityGap.id == gap_id,
|
||
ModelCapabilityGap.tenant_id == tenant_id,
|
||
)
|
||
)
|
||
)
|
||
gap = result.scalar_one_or_none()
|
||
|
||
if not gap:
|
||
raise ValueError(f"能力差距 {gap_id} 不存在")
|
||
|
||
gap.status = "resolved"
|
||
gap.resolved_at = datetime.utcnow()
|
||
gap.resolution_notes = resolution_notes
|
||
await self.db.commit()
|
||
|
||
return {"id": gap.id, "status": gap.status}
|
||
|
||
# ========================================================================
|
||
# 成本跟踪查询
|
||
# ========================================================================
|
||
|
||
async def get_cost_tracking(
|
||
self, truth_node_id: int, period: str = "weekly", tenant_id: int = 1
|
||
) -> Dict[str, Any]:
|
||
"""获取成本跟踪记录"""
|
||
result = await self.db.execute(
|
||
select(CostTracking).where(
|
||
and_(
|
||
CostTracking.truth_node_id == truth_node_id,
|
||
CostTracking.period == period,
|
||
CostTracking.tenant_id == tenant_id,
|
||
)
|
||
).order_by(desc(CostTracking.created_at))
|
||
)
|
||
tracking = result.scalar_one_or_none()
|
||
|
||
if not tracking:
|
||
return {
|
||
"truth_node_id": truth_node_id,
|
||
"period": period,
|
||
"estimated_tokens": 0,
|
||
"actual_tokens": 0,
|
||
"estimated_cost_usd": 0.0,
|
||
"actual_cost_usd": 0.0,
|
||
"roi_score": None,
|
||
"roi_analysis": {},
|
||
"optimization_suggestions": {},
|
||
}
|
||
|
||
return {
|
||
"id": tracking.id,
|
||
"truth_node_id": tracking.truth_node_id,
|
||
"period": tracking.period,
|
||
"estimated_tokens": tracking.estimated_tokens,
|
||
"actual_tokens": tracking.actual_tokens,
|
||
"estimated_cost_usd": float(tracking.estimated_cost_usd),
|
||
"actual_cost_usd": float(tracking.actual_cost_usd),
|
||
"roi_score": tracking.roi_score,
|
||
"roi_analysis": tracking.roi_analysis,
|
||
"optimization_suggestions": tracking.optimization_suggestions,
|
||
}
|