Spaces:
Sleeping
Sleeping
| from core.constants import TOKEN_RATES | |
| from database.db import DatabaseManager | |
| from schemas.enums import DecisionStage, MissionStatus | |
| from schemas.models import EvidenceScore | |
| class ConfidenceEngine: | |
| """Calculates objective confidence metrics based on evidence attributes.""" | |
| def calculate_confidence( | |
| credibility: float, freshness: float, authority: float, agreement: float, conflict: float | |
| ) -> EvidenceScore: | |
| weighted_score = (credibility * 0.30) + (freshness * 0.20) + (authority * 0.25) + (agreement * 0.25) | |
| penalty = conflict * 0.35 | |
| overall = max(0.0, min(100.0, weighted_score - penalty)) | |
| return EvidenceScore( | |
| credibility=credibility, | |
| freshness=freshness, | |
| authority=authority, | |
| agreement=agreement, | |
| conflict=conflict, | |
| overall_confidence=round(overall, 2), | |
| ) | |
| class TokenCostEngine: | |
| """Estimates and records operational token usage and cost metrics.""" | |
| RATES = TOKEN_RATES | |
| async def track_usage( | |
| cls, | |
| db: DatabaseManager, | |
| mission_id: str, | |
| agent_id: str, | |
| prompt_tokens: int = 0, | |
| completion_tokens: int = 0, | |
| reasoning_tokens: int = 0, | |
| vision_tokens: int = 0, | |
| ) -> float: | |
| cost = ( | |
| (prompt_tokens * cls.RATES["prompt"]) | |
| + (completion_tokens * cls.RATES["completion"]) | |
| + (reasoning_tokens * cls.RATES["reasoning"]) | |
| + (vision_tokens * cls.RATES["vision"]) | |
| ) | |
| await db.record_tokens( | |
| mission_id=mission_id, | |
| agent_id=agent_id, | |
| prompt_tokens=prompt_tokens, | |
| completion_tokens=completion_tokens, | |
| reasoning_tokens=reasoning_tokens, | |
| vision_tokens=vision_tokens, | |
| cost_usd=cost, | |
| ) | |
| await db.save_mission( | |
| mission_id=mission_id, topic="", status=MissionStatus.IN_PROGRESS, stage=DecisionStage.EXECUTE, total_cost=cost | |
| ) | |
| return cost | |