212 lines
6.9 KiB
Python
212 lines
6.9 KiB
Python
"""
|
|
AI Usage data model for tracking AI generation costs and usage.
|
|
|
|
This module defines the AIUsageLog dataclass which represents a single AI usage
|
|
event for tracking costs, tokens used, and generating usage analytics.
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone, date
|
|
from typing import Dict, Any, Optional
|
|
from enum import Enum
|
|
|
|
|
|
class TaskType(str, Enum):
|
|
"""Types of AI tasks that can be tracked."""
|
|
STORY_PROGRESSION = "story_progression"
|
|
COMBAT_NARRATION = "combat_narration"
|
|
QUEST_SELECTION = "quest_selection"
|
|
NPC_DIALOGUE = "npc_dialogue"
|
|
GENERAL = "general"
|
|
|
|
|
|
@dataclass
|
|
class AIUsageLog:
|
|
"""
|
|
Represents a single AI usage event for cost and usage tracking.
|
|
|
|
This dataclass captures all relevant information about an AI API call
|
|
including the user, model used, tokens consumed, and estimated cost.
|
|
Used for:
|
|
- Cost monitoring and budgeting
|
|
- Usage analytics per user/tier
|
|
- Rate limiting enforcement
|
|
- Billing and invoicing (future)
|
|
|
|
Attributes:
|
|
log_id: Unique identifier for this usage log entry
|
|
user_id: User who made the request
|
|
timestamp: When the request was made
|
|
model: Model identifier (e.g., "meta/meta-llama-3-8b-instruct")
|
|
tokens_input: Number of input tokens (prompt)
|
|
tokens_output: Number of output tokens (response)
|
|
tokens_total: Total tokens used (input + output)
|
|
estimated_cost: Estimated cost in USD
|
|
task_type: Type of task (story, combat, quest, npc)
|
|
session_id: Optional game session ID for context
|
|
character_id: Optional character ID for context
|
|
request_duration_ms: How long the request took in milliseconds
|
|
success: Whether the request completed successfully
|
|
error_message: Error message if the request failed
|
|
"""
|
|
|
|
log_id: str
|
|
user_id: str
|
|
timestamp: datetime
|
|
model: str
|
|
tokens_input: int
|
|
tokens_output: int
|
|
tokens_total: int
|
|
estimated_cost: float
|
|
task_type: TaskType
|
|
session_id: Optional[str] = None
|
|
character_id: Optional[str] = None
|
|
request_duration_ms: int = 0
|
|
success: bool = True
|
|
error_message: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""
|
|
Convert usage log to dictionary for storage.
|
|
|
|
Returns:
|
|
Dictionary representation suitable for Appwrite storage
|
|
"""
|
|
return {
|
|
"user_id": self.user_id,
|
|
"timestamp": self.timestamp.isoformat() if isinstance(self.timestamp, datetime) else self.timestamp,
|
|
"model": self.model,
|
|
"tokens_input": self.tokens_input,
|
|
"tokens_output": self.tokens_output,
|
|
"tokens_total": self.tokens_total,
|
|
"estimated_cost": self.estimated_cost,
|
|
"task_type": self.task_type.value if isinstance(self.task_type, TaskType) else self.task_type,
|
|
"session_id": self.session_id,
|
|
"character_id": self.character_id,
|
|
"request_duration_ms": self.request_duration_ms,
|
|
"success": self.success,
|
|
"error_message": self.error_message,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> "AIUsageLog":
|
|
"""
|
|
Create AIUsageLog from dictionary.
|
|
|
|
Args:
|
|
data: Dictionary with usage log data
|
|
|
|
Returns:
|
|
AIUsageLog instance
|
|
"""
|
|
# Parse timestamp
|
|
timestamp = data.get("timestamp")
|
|
if isinstance(timestamp, str):
|
|
timestamp = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
|
|
elif timestamp is None:
|
|
timestamp = datetime.now(timezone.utc)
|
|
|
|
# Parse task type
|
|
task_type = data.get("task_type", "general")
|
|
if isinstance(task_type, str):
|
|
try:
|
|
task_type = TaskType(task_type)
|
|
except ValueError:
|
|
task_type = TaskType.GENERAL
|
|
|
|
return cls(
|
|
log_id=data.get("log_id", ""),
|
|
user_id=data.get("user_id", ""),
|
|
timestamp=timestamp,
|
|
model=data.get("model", ""),
|
|
tokens_input=data.get("tokens_input", 0),
|
|
tokens_output=data.get("tokens_output", 0),
|
|
tokens_total=data.get("tokens_total", 0),
|
|
estimated_cost=data.get("estimated_cost", 0.0),
|
|
task_type=task_type,
|
|
session_id=data.get("session_id"),
|
|
character_id=data.get("character_id"),
|
|
request_duration_ms=data.get("request_duration_ms", 0),
|
|
success=data.get("success", True),
|
|
error_message=data.get("error_message"),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class DailyUsageSummary:
|
|
"""
|
|
Summary of AI usage for a specific day.
|
|
|
|
Used for reporting and rate limiting checks.
|
|
|
|
Attributes:
|
|
date: The date of this summary
|
|
user_id: User ID
|
|
total_requests: Number of AI requests made
|
|
total_tokens: Total tokens consumed
|
|
total_input_tokens: Total input tokens
|
|
total_output_tokens: Total output tokens
|
|
estimated_cost: Total estimated cost in USD
|
|
requests_by_task: Breakdown of requests by task type
|
|
"""
|
|
|
|
date: date
|
|
user_id: str
|
|
total_requests: int
|
|
total_tokens: int
|
|
total_input_tokens: int
|
|
total_output_tokens: int
|
|
estimated_cost: float
|
|
requests_by_task: Dict[str, int] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert summary to dictionary."""
|
|
return {
|
|
"date": self.date.isoformat() if isinstance(self.date, date) else self.date,
|
|
"user_id": self.user_id,
|
|
"total_requests": self.total_requests,
|
|
"total_tokens": self.total_tokens,
|
|
"total_input_tokens": self.total_input_tokens,
|
|
"total_output_tokens": self.total_output_tokens,
|
|
"estimated_cost": self.estimated_cost,
|
|
"requests_by_task": self.requests_by_task,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class MonthlyUsageSummary:
|
|
"""
|
|
Summary of AI usage for a specific month.
|
|
|
|
Used for billing and cost projections.
|
|
|
|
Attributes:
|
|
year: Year
|
|
month: Month (1-12)
|
|
user_id: User ID
|
|
total_requests: Number of AI requests made
|
|
total_tokens: Total tokens consumed
|
|
estimated_cost: Total estimated cost in USD
|
|
daily_breakdown: List of daily summaries
|
|
"""
|
|
|
|
year: int
|
|
month: int
|
|
user_id: str
|
|
total_requests: int
|
|
total_tokens: int
|
|
estimated_cost: float
|
|
daily_breakdown: list = field(default_factory=list)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert summary to dictionary."""
|
|
return {
|
|
"year": self.year,
|
|
"month": self.month,
|
|
"user_id": self.user_id,
|
|
"total_requests": self.total_requests,
|
|
"total_tokens": self.total_tokens,
|
|
"estimated_cost": self.estimated_cost,
|
|
"daily_breakdown": self.daily_breakdown,
|
|
}
|