c81b0fc1
 tangwang
scripts/evaluatio...
|
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
|
"""FastAPI app for the search evaluation UI (static frontend + JSON APIs)."""
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from .api_models import BatchEvalRequest, SearchEvalRequest
from .constants import DEFAULT_QUERY_FILE
from .framework import SearchEvaluationFramework
_STATIC_DIR = Path(__file__).resolve().parent / "static"
def create_web_app(framework: SearchEvaluationFramework, query_file: Path = DEFAULT_QUERY_FILE) -> FastAPI:
app = FastAPI(title="Search Evaluation UI", version="1.0.0")
app.mount(
"/static",
StaticFiles(directory=str(_STATIC_DIR)),
name="static",
)
index_path = _STATIC_DIR / "index.html"
@app.get("/", response_class=HTMLResponse)
def home() -> str:
return index_path.read_text(encoding="utf-8")
@app.get("/api/queries")
def api_queries() -> Dict[str, Any]:
return {"queries": framework.queries_from_file(query_file)}
@app.post("/api/search-eval")
def api_search_eval(request: SearchEvalRequest) -> Dict[str, Any]:
return framework.evaluate_live_query(
query=request.query,
top_k=request.top_k,
auto_annotate=request.auto_annotate,
language=request.language,
)
@app.post("/api/batch-eval")
def api_batch_eval(request: BatchEvalRequest) -> Dict[str, Any]:
queries = request.queries or framework.queries_from_file(query_file)
if not queries:
raise HTTPException(status_code=400, detail="No queries provided")
return framework.batch_evaluate(
queries=queries,
top_k=request.top_k,
auto_annotate=request.auto_annotate,
language=request.language,
force_refresh_labels=request.force_refresh_labels,
)
@app.get("/api/history")
def api_history() -> Dict[str, Any]:
return {"history": framework.store.list_batch_runs(limit=20)}
@app.get("/api/history/{batch_id}/report")
def api_history_report(batch_id: str) -> Dict[str, Any]:
row = framework.store.get_batch_run(batch_id)
if row is None:
raise HTTPException(status_code=404, detail="Unknown batch_id")
report_path = Path(row["report_markdown_path"]).resolve()
root = framework.artifact_root.resolve()
try:
report_path.relative_to(root)
except ValueError:
raise HTTPException(status_code=403, detail="Report path is outside artifact root")
if not report_path.is_file():
raise HTTPException(status_code=404, detail="Report file not found")
return {
"batch_id": row["batch_id"],
"created_at": row["created_at"],
"tenant_id": row["tenant_id"],
"report_markdown_path": str(report_path),
"markdown": report_path.read_text(encoding="utf-8"),
}
return app
|