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
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
|
"""
FastAPI server for the AI Evaluation Visualization Dashboard.
Serves read-only results, JSON logs, and static visuals.
"""
import os
import sys
from pathlib import Path
from typing import Dict, Any, List
import pandas as pd
import json
import math
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse, FileResponse
from fastapi.staticfiles import StaticFiles
# Add project root to path to import core modules
PROJECT_ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from src.config_loader import ConfigLoader
app = FastAPI(
title="AI Agent Evaluation Dashboard",
description="Web GUI for analyzing experiment results",
version="0.2.0"
)
# Mount static files
static_dir = Path(__file__).parent / "static"
if static_dir.exists():
app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")
def get_config():
"""Helper to load config safely."""
try:
config = ConfigLoader(str(PROJECT_ROOT / "config.yaml"))
config.load()
return config
except Exception as e:
print(f"Warning: Could not load config: {e}")
return None
# ============================================================================
# Root Endpoint - Serve HTML
# ============================================================================
@app.get("/", response_class=HTMLResponse)
async def root():
"""Serve the main HTML page."""
index_path = static_dir / "index.html"
if index_path.exists():
return FileResponse(index_path)
return HTMLResponse(content="<h1>index.html not found</h1>", status_code=404)
# ============================================================================
# Results Endpoints
# ============================================================================
@app.get("/api/results/files")
async def get_results_files():
"""List all CSV files in the output directory."""
config = get_config()
if not config:
raise HTTPException(status_code=500, detail="Config not loaded")
output_dir = PROJECT_ROOT / config.output_dir
if not output_dir.exists():
return []
files = []
for p in output_dir.rglob("*.csv"):
files.append(str(p.relative_to(output_dir)))
return sorted(files)
@app.get("/api/results")
async def get_results(file: str = None):
"""Get experiment results as JSON, optionally for a specific file."""
config = get_config()
if not config:
raise HTTPException(status_code=500, detail="Config not loaded")
output_dir = PROJECT_ROOT / config.output_dir
csv_files = []
if file:
file_path = output_dir / file
if not file_path.exists() or ".." in file:
raise HTTPException(status_code=404, detail="File not found")
csv_files.append(file_path)
else:
csv_files = list(output_dir.glob("*.csv")) if output_dir.exists() else []
results = {}
for csv_file in csv_files:
try:
df = pd.read_csv(csv_file)
# Convert NaN values to None for JSON serialization
def clean_value(val):
if isinstance(val, float) and (math.isnan(val) or math.isinf(val)):
return None
return val
# Clean each row
cleaned_data = []
for record in df.to_dict(orient="records"):
cleaned_record = {k: clean_value(v) for k, v in record.items()}
cleaned_data.append(cleaned_record)
results[csv_file.stem] = {
"columns": df.columns.tolist(),
"data": cleaned_data
}
except Exception as e:
results[csv_file.stem] = {"error": str(e)}
return results
@app.get("/api/results/images")
async def list_result_images():
"""List generated visualization images."""
config = get_config()
if not config:
return []
output_dir = PROJECT_ROOT / config.output_dir
viz_dir = output_dir / "viz"
if not viz_dir.exists():
return []
images = []
for img in viz_dir.glob("*"):
if img.suffix.lower() in ['.png', '.jpg', '.jpeg', '.gif', '.svg']:
images.append({
"name": img.name,
"path": str(img.relative_to(PROJECT_ROOT))
})
return images
@app.get("/api/results/images/{image_name}")
async def get_result_image(image_name: str):
"""Serve a specific image."""
config = get_config()
if not config:
raise HTTPException(status_code=500, detail="Config not loaded")
output_dir = PROJECT_ROOT / config.output_dir
viz_dir = output_dir / "viz"
image_path = viz_dir / image_name
if not image_path.exists():
raise HTTPException(status_code=404, detail="Image not found")
return FileResponse(image_path)
# ============================================================================
# Judge Results Endpoints
# ============================================================================
@app.get("/api/judge/files")
async def get_judge_files():
"""Get list of judge result files."""
config = get_config()
if not config:
raise HTTPException(status_code=500, detail="Config not loaded")
judge_log_dir = config._config.get('judge', {}).get('log_dir', 'judge_logs')
judge_dir = PROJECT_ROOT / judge_log_dir
if not judge_dir.exists():
return []
files = []
for p in judge_dir.rglob("*"):
if p.suffix in ['.csv', '.json']:
files.append(str(p.relative_to(judge_dir)))
return sorted(files)
@app.get("/api/judge/results")
async def get_judge_results(file: str = None):
"""Get judge results if available."""
config = get_config()
if not config:
raise HTTPException(status_code=500, detail="Config not loaded")
judge_log_dir = config._config.get('judge', {}).get('log_dir', 'judge_logs')
judge_dir = PROJECT_ROOT / judge_log_dir
if not judge_dir.exists():
return {"message": "No judge results found"}
result_files = []
if file:
file_path = judge_dir / file
if not file_path.exists() or ".." in file:
raise HTTPException(status_code=404, detail="File not found")
result_files.append(file_path)
else:
result_files = list(judge_dir.glob("*.csv")) + list(judge_dir.glob("*.json"))
results = {}
for rf in result_files:
if rf.suffix == '.csv':
try:
df = pd.read_csv(rf)
results[rf.stem] = {
"type": "csv",
"columns": df.columns.tolist(),
"data": df.to_dict(orient="records")
}
except Exception as e:
results[rf.stem] = {"error": str(e)}
elif rf.suffix == '.json':
try:
with open(rf) as f:
results[rf.stem] = {"type": "json", "data": json.load(f)}
except Exception as e:
results[rf.stem] = {"error": str(e)}
return results
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
|
