Spaces:
Sleeping
Sleeping
File size: 2,944 Bytes
f8a3ca2 | 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 | """Direct (non-agent) Bayesian compute endpoints.
These are useful both as a fallback (when no API key is configured) and as a
fast path the frontend can call directly when the user is hand-driving the
workbench instead of talking to the Agent.
"""
from __future__ import annotations
from fastapi import APIRouter, HTTPException
from app.api.schemas import (
BayesComputeRequest,
BayesResultModel,
SensitivityPointModel,
SensitivityRequest,
SensitivityResponse,
bayes_result_to_model,
)
from app.domain.bayesian import compute, kde, sensitivity_analysis
from app.domain.report import render_markdown_report
router = APIRouter(prefix="/api/bayesian", tags=["bayesian"])
@router.post("/compute", response_model=BayesResultModel)
async def post_compute(req: BayesComputeRequest) -> BayesResultModel:
try:
result = compute(req.data, req.judgments, req.R)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
return bayes_result_to_model(result)
@router.post("/sensitivity", response_model=SensitivityResponse)
async def post_sensitivity(req: SensitivityRequest) -> SensitivityResponse:
try:
points = sensitivity_analysis(req.data, req.judgments, req.r_values)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
return SensitivityResponse(
points=[
SensitivityPointModel(R=p["R"], result=bayes_result_to_model(p["result"]))
for p in points
]
)
@router.post("/kde")
async def post_kde(req: BayesComputeRequest) -> dict:
"""Return the KDE curve for the dataset (200 points). Used by the
PriorPosterior / KDE chart on the frontend."""
try:
pts = kde(req.data, n_points=200)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
return {"points": pts}
@router.post("/report")
async def post_report(req: BayesComputeRequest) -> dict:
"""Compute + render a structured Markdown report in one call."""
try:
result = compute(req.data, req.judgments, req.R)
sweep = sensitivity_analysis(req.data, req.judgments, [2.0, 5.0, 10.0, 20.0, 50.0])
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
sens_rows = [
{
"R": p["R"],
"posterior_mean": p["result"].posterior.stats.mean,
"posterior_std": p["result"].posterior.stats.std,
"ci95_lower": p["result"].posterior.stats.ci95_lower,
"ci95_upper": p["result"].posterior.stats.ci95_upper,
}
for p in sweep
]
md = render_markdown_report(
result,
scenario_name=req.scenario_name or "未命名情景",
reference_case=req.reference_case or "",
sensitivity_rows=sens_rows,
)
return {"markdown": md, "result": bayes_result_to_model(result).model_dump()}
|