Spaces:
Configuration error
Configuration error
File size: 6,043 Bytes
f898356 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 | """
Request Tracker - Track all AI requests with metadata
------------------------------------------------------
Saves structured data about each AI request including:
- What triggered it
- Game context
- Request/response data
- Timing information
"""
import json
from pathlib import Path
from datetime import datetime
from typing import Dict, Any, Optional, List
class RequestTracker:
"""Tracks all AI requests with full context and metadata."""
def __init__(self, session_dir: Path):
self.session_dir = session_dir
self.requests_file = session_dir / "requests.json"
self.requests: List[Dict[str, Any]] = []
self.load()
def load(self):
"""Load existing requests from file."""
if self.requests_file.exists():
try:
with open(self.requests_file, 'r', encoding='utf-8') as f:
data = json.load(f)
self.requests = data.get("requests", [])
except Exception as e:
print(f"Warning: Could not load requests: {e}")
self.requests = []
def add_request(
self,
player_name: str,
trigger: str,
game_phase: str,
current_player: str,
prompt_data: Dict[str, Any],
response_data: Optional[Dict[str, Any]] = None,
metadata: Optional[Dict[str, Any]] = None
) -> str:
"""
Add a new AI request.
Args:
player_name: Name of the player/agent
trigger: What caused this request (e.g., "turn_start", "state_change")
game_phase: Current game phase (e.g., "SETUP_FIRST_ROUND", "MAIN_GAME")
current_player: Who's turn it is
prompt_data: The prompt sent to LLM
response_data: The response from LLM (optional, can be added later)
metadata: Additional metadata (timing, tokens, etc.)
Returns:
Request ID
"""
request_id = f"req_{len(self.requests) + 1:04d}"
timestamp = datetime.now().isoformat()
request = {
"request_id": request_id,
"timestamp": timestamp,
"player_name": player_name,
"trigger": trigger,
"game_context": {
"phase": game_phase,
"current_player": current_player,
},
"prompt": prompt_data,
"response": response_data,
"raw_response": None, # Store unprocessed LLM response
"metadata": metadata or {},
"is_new": True # Flag for UI to show as new
}
self.requests.append(request)
self.save()
return request_id
def update_response(self, request_id: str, response_data: Dict[str, Any], raw_response: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None):
"""Update request with response data."""
for req in self.requests:
if req["request_id"] == request_id:
req["response"] = response_data
if raw_response:
req["raw_response"] = raw_response
if metadata:
req["metadata"].update(metadata)
req["response_timestamp"] = datetime.now().isoformat()
self.save()
return True
return False
def mark_as_viewed(self, request_id: str):
"""Mark request as viewed (not new anymore)."""
for req in self.requests:
if req["request_id"] == request_id:
req["is_new"] = False
self.save()
return True
return False
def mark_all_as_viewed(self):
"""Mark all requests as viewed."""
for req in self.requests:
req["is_new"] = False
self.save()
def get_new_count(self) -> int:
"""Get count of new (unviewed) requests."""
return sum(1 for req in self.requests if req.get("is_new", False))
def get_player_requests(self, player_name: str) -> List[Dict[str, Any]]:
"""Get all requests for a specific player."""
return [req for req in self.requests if req["player_name"] == player_name]
def get_new_player_requests(self, player_name: str) -> List[Dict[str, Any]]:
"""Get new requests for a specific player."""
return [req for req in self.requests
if req["player_name"] == player_name and req.get("is_new", False)]
def save(self):
"""Save requests to file."""
try:
with open(self.requests_file, 'w', encoding='utf-8') as f:
json.dump({
"requests": self.requests,
"total_count": len(self.requests),
"last_updated": datetime.now().isoformat()
}, f, indent=2, ensure_ascii=False)
except Exception as e:
print(f"Error saving requests: {e}")
def get_all_requests(self) -> List[Dict[str, Any]]:
"""Get all requests."""
return self.requests
def get_stats(self) -> Dict[str, Any]:
"""Get statistics about requests."""
players = {}
for req in self.requests:
player = req["player_name"]
if player not in players:
players[player] = {
"total": 0,
"new": 0,
"phases": set()
}
players[player]["total"] += 1
if req.get("is_new", False):
players[player]["new"] += 1
players[player]["phases"].add(req["game_context"]["phase"])
# Convert sets to lists for JSON serialization
for player_data in players.values():
player_data["phases"] = list(player_data["phases"])
return {
"total_requests": len(self.requests),
"new_requests": self.get_new_count(),
"players": players
}
|