"""Query and Benchmark API routes.
Endpoints:
POST /queries Ask a question against a strategy
GET /queries/{id} Retrieve a past query
POST /benchmarks Run a benchmark (or dry run)
GET /benchmarks/{id} Retrieve experiment results
GET /experiments List all experiments
"""
from fastapi import APIRouter, Query
from fastapi.responses import HTMLResponse
from src.core.exceptions import BenchmarkError, QueryError
from src.core.models import StrategyName
from src.benchmarking import benchmark_service, query_service
from src.benchmarking.models import (
BenchmarkRequest,
BenchmarkResponse,
CostEstimate,
ExperimentDetailResponse,
QueryRequest,
QueryResponse,
QueryDetailResponse,
RetrievedChunk,
StrategyMetrics,
)
router = APIRouter()
# ── Query Endpoints ───────────────────────────────────────────────
@router.post("/queries", response_model=QueryResponse, status_code=201)
async def create_query(request: QueryRequest):
"""Ask a question against a document using a specific chunking strategy.
Pipeline: embed question → vector search → generate answer → store result.
"""
result = query_service.run_query(
document_id=request.document_id,
strategy_name=request.strategy,
question=request.question,
top_k=request.top_k,
neighbor_prev=request.neighbor_prev,
neighbor_next=request.neighbor_next,
corpus_model_id=request.corpus_model_id,
)
return QueryResponse(
query_id=result["query_id"],
document_id=result["document_id"],
strategy=result["strategy"],
question=result["question"],
answer=result["answer"],
retrieved_chunks=[
RetrievedChunk(**chunk) for chunk in result["retrieved_chunks"]
],
expansion_tree=result.get("expansion_tree") or [],
neighbor_prev=result.get("neighbor_prev", 0),
neighbor_next=result.get("neighbor_next", 0),
latency_breakdown=result["latency_breakdown"],
token_usage=result["token_usage"],
created_at=result["created_at"],
)
@router.get("/queries/{query_id}", response_model=QueryDetailResponse)
async def get_query(query_id: str):
"""Retrieve a past query by ID."""
result = query_service.get_query(query_id)
if result is None:
raise QueryError(f"Query not found: {query_id}")
return QueryDetailResponse(
id=result["id"],
document_id=result["document_id"],
strategy_name=result["strategy_name"],
question=result["question"],
answer=result["answer"],
retrieved_chunks=result["retrieved_chunks"],
expansion_tree=result.get("expansion_tree") or [],
latency_breakdown=result["latency_breakdown"],
token_usage=result["token_usage"],
created_at=result["created_at"],
)
# ── Benchmark Endpoints ──────────────────────────────────────────
@router.post("/benchmarks", response_model=BenchmarkResponse, status_code=201)
async def create_benchmark(request: BenchmarkRequest):
"""Run a benchmark comparing multiple strategies on multiple questions.
Can run in dry_run mode to get cost estimate without executing.
"""
# Load questions - prefer questions_file over inline questions
if request.questions_file:
questions = benchmark_service.load_questions(request.questions_file)
elif request.questions and len(request.questions) > 0:
# Validate that questions have required fields
valid_questions = [
q for q in request.questions
if isinstance(q, dict) and "question" in q
]
if not valid_questions:
raise BenchmarkError("Invalid questions: each question must have a 'question' field")
questions = valid_questions
else:
raise BenchmarkError("Either 'questions' or 'questions_file' must be provided")
if not questions:
raise BenchmarkError("No questions to benchmark")
# Dry run - return cost estimate only
if request.dry_run:
estimate = benchmark_service.estimate_cost(
num_questions=len(questions),
num_strategies=len(request.strategies),
)
# Return as BenchmarkResponse with minimal data
return BenchmarkResponse(
experiment_id="dry_run",
document_id=request.document_id,
strategies_used=[s.value for s in request.strategies],
questions_count=len(questions),
aggregate_metrics={},
best_strategy="N/A (dry run)",
total_latency_seconds=0,
estimated_cost_usd=estimate["estimated_cost_usd"],
created_at="N/A",
)
# Run full benchmark
result = benchmark_service.run_benchmark(
document_id=request.document_id,
strategies=request.strategies,
questions=questions,
top_k=request.top_k,
neighbor_prev=request.neighbor_prev,
neighbor_next=request.neighbor_next,
corpus_model_id=request.corpus_model_id,
)
return BenchmarkResponse(
experiment_id=result["experiment_id"],
document_id=result["document_id"],
strategies_used=result["strategies_used"],
questions_count=result["questions_count"],
aggregate_metrics={
k: StrategyMetrics(**v) for k, v in result["aggregate_metrics"].items()
},
best_strategy=result["best_strategy"],
total_latency_seconds=result["total_latency_seconds"],
estimated_cost_usd=result["estimated_cost_usd"],
created_at=result["created_at"],
)
@router.get("/benchmarks/{experiment_id}", response_model=ExperimentDetailResponse)
async def get_benchmark(experiment_id: str):
"""Retrieve experiment results by ID."""
result = benchmark_service.get_experiment(experiment_id)
if result is None:
raise BenchmarkError(f"Experiment not found: {experiment_id}")
# Resolve document filename
from src.storage import sqlite as db
doc = db.get_document(result.get("document_id", ""))
doc_filename = doc.get("filename", "Unknown") if doc else "Deleted"
return ExperimentDetailResponse(
id=result["id"],
document_id=result["document_id"],
document_filename=doc_filename,
benchmark_config=result.get("benchmark_config", {}),
questions=result.get("questions", []),
per_question=result.get("per_question", []),
aggregate_metrics={
k: StrategyMetrics(**v) for k, v in result.get("aggregate_metrics", {}).items()
},
strategies_used=result.get("strategies_used", []),
created_at=result["created_at"],
)
@router.get("/experiments")
async def list_experiments(
document_id: str | None = None,
offset: int = Query(0, ge=0),
limit: int = Query(200, ge=1, le=500),
):
"""List all experiments, optionally filtered by document."""
result = benchmark_service.list_experiments(
document_id=document_id, offset=offset, limit=limit
)
# Enrich with document filenames and best_strategy
from src.storage import sqlite as db
for item in result.get("items", []):
doc = db.get_document(item.get("document_id", ""))
item["document_filename"] = doc.get("filename", "Unknown") if doc else "Deleted"
questions = item.get("questions") or []
item["questions_count"] = (
item.get("benchmark_config", {}).get("num_questions")
or (len(questions) if isinstance(questions, list) else 0)
)
# Calculate best_strategy from aggregate_metrics
aggs = item.get("aggregate_metrics", {})
best_strat, best_score = "N/A", -1
for strat, m in aggs.items():
score = (m.get("avg_context_relevance", 0) * 0.3 +
m.get("avg_answer_similarity", 0) * 0.4 +
m.get("avg_faithfulness", 0) * 0.3)
adjusted = score * (1 - m.get("hallucination_rate", 0))
if adjusted > best_score:
best_score = adjusted
best_strat = strat
item["best_strategy"] = best_strat
# Surface embedding on list even if only in benchmark_config
if not item.get("embedding_model_id"):
cfg = item.get("benchmark_config") or {}
item["embedding_model_id"] = cfg.get("embedding_model_id")
item["embedding_provider"] = cfg.get("embedding_provider")
return result
@router.delete("/experiments/{experiment_id}")
async def delete_experiment(experiment_id: str):
"""Delete an experiment by ID."""
from src.storage import sqlite as db
deleted = db.delete_experiment(experiment_id)
if not deleted:
raise BenchmarkError(f"Experiment not found: {experiment_id}")
return {"deleted": True, "experiment_id": experiment_id}
@router.get("/benchmarks/{experiment_id}/report", response_class=HTMLResponse)
async def get_report(experiment_id: str, view: str = "managerial"):
"""Generate and return an HTML report for an experiment.
Views:
- managerial: Decision-focused, high-level insights (default)
- technical: Full observability with detailed data
"""
from src.benchmarking.report import generate_report
result = benchmark_service.get_experiment(experiment_id)
if result is None:
raise BenchmarkError(f"Experiment not found: {experiment_id}")
# Validate view parameter
if view not in ("managerial", "technical"):
view = "managerial"
# Generate HTML report
html = generate_report(result, view)
return HTMLResponse(content=html)
# ── HTML Report Generator ─────────────────────────────────────────
def _generate_html_report(experiment: dict) -> str:
"""Generate an HTML report from experiment data."""
config = experiment.get("benchmark_config", {})
aggregate = experiment.get("aggregate_metrics", {})
per_question = experiment.get("per_question", [])
strategies = experiment.get("strategies_used", [])
# Build strategy rows for summary table
strategy_rows = ""
for strategy_name in strategies:
metrics = aggregate.get(strategy_name, {})
strategy_rows += f"""
| {strategy_name} |
{metrics.get('avg_context_relevance', 0):.1f} |
{metrics.get('avg_answer_similarity', 0):.1f} |
{metrics.get('avg_faithfulness', 0):.1f} |
{metrics.get('hallucination_rate', 0)*100:.1f}% |
{metrics.get('total_questions', 0)} |
"""
# Build per-question rows
question_rows = ""
for qr in per_question:
q_text = qr.get("question", "")[:80]
q_id = qr.get("question_id", "")
for strategy_name in strategies:
strat_result = qr.get("strategies", {}).get(strategy_name, {})
scores = strat_result.get("scores", {})
answer_preview = strat_result.get("answer", "")[:100]
question_rows += f"""
| {q_id} |
{q_text}... |
{strategy_name} |
{scores.get('context_relevance', 'N/A')} |
{scores.get('answer_similarity', 'N/A')} |
{scores.get('faithfulness', 'N/A')} |
{'✓' if not scores.get('hallucination', False) else '✗'} |
{answer_preview}... |
"""
html = f"""
Benchmark Report - {experiment.get('id', 'Unknown')}
📊 Benchmark Report
Experiment ID
{experiment.get('id', 'N/A')[:16]}...
Document
{experiment.get('document_id', 'N/A')[:16]}...
Questions
{config.get('num_questions', len(per_question))}
Strategies
{len(strategies)}
📈 Summary by Strategy
| Strategy |
Context Relevance |
Answer Similarity |
Faithfulness |
Hallucination Rate |
Questions |
{strategy_rows}
📝 Detailed Results
| Q-ID |
Question |
Strategy |
Context |
Similarity |
Faithfulness |
No Halluc. |
Answer Preview |
{question_rows}
"""
return html