|
| 1 | +"""FastAPI Backend for Epistemic Forge. |
| 2 | +
|
| 3 | +Provides a RESTful Enterprise-Ready API to integrate the Neuro-Symbolic |
| 4 | +engine into any larger MLOps or business workflow. |
| 5 | +""" |
| 6 | +from fastapi import FastAPI, HTTPException, BackgroundTasks |
| 7 | +from pydantic import BaseModel |
| 8 | +from typing import List, Optional, Dict, Any |
| 9 | +from epistemic_forge.models import ProjectSpec |
| 10 | +from epistemic_forge.pipeline.arsenal_run import run_pipeline |
| 11 | + |
| 12 | +app = FastAPI( |
| 13 | + title="Epistemic Forge API", |
| 14 | + description="Neuro-Symbolic State Machine API for Deterministic Reasoning.", |
| 15 | + version="1.0.0-rc1" |
| 16 | +) |
| 17 | + |
| 18 | +class ForgeRequest(BaseModel): |
| 19 | + title: str |
| 20 | + question: str |
| 21 | + domain: str = "hybrid" |
| 22 | + target_model: str = "openai/gpt-4o-mini" |
| 23 | + keywords: List[str] = [] |
| 24 | + api_key: Optional[str] = None |
| 25 | + api_base: Optional[str] = None |
| 26 | + budget_tokens: int = 15000 |
| 27 | + |
| 28 | +class ForgeResponse(BaseModel): |
| 29 | + status: str |
| 30 | + score: float |
| 31 | + claims: List[Dict[str, Any]] |
| 32 | + final_memo: str |
| 33 | + peer_review: Dict[str, Any] |
| 34 | + |
| 35 | +@app.post("/api/v1/forge", response_model=ForgeResponse) |
| 36 | +async def generate_claim_lattice(req: ForgeRequest): |
| 37 | + """Executes the full L0-L6 pipeline and returns a structured lattice and synthesis.""" |
| 38 | + try: |
| 39 | + spec = ProjectSpec( |
| 40 | + title=req.title, |
| 41 | + question=req.question, |
| 42 | + domain=req.domain, |
| 43 | + target_model=req.target_model, |
| 44 | + keywords=req.keywords, |
| 45 | + api_key=req.api_key, |
| 46 | + api_base=req.api_base, |
| 47 | + budget_tokens=req.budget_tokens |
| 48 | + ) |
| 49 | + |
| 50 | + result = run_pipeline( |
| 51 | + title=spec.title, |
| 52 | + question=spec.question, |
| 53 | + domain=spec.domain.value |
| 54 | + ) |
| 55 | + |
| 56 | + final_memo = "" |
| 57 | + for art in getattr(result, "artifacts", []): |
| 58 | + if art.name == "Final Synthesis Memo": |
| 59 | + final_memo = art.content |
| 60 | + break |
| 61 | + |
| 62 | + # Handle claims safely |
| 63 | + raw_claims = getattr(result, "claims", []) |
| 64 | + claims_list = [c.model_dump() if hasattr(c, "model_dump") else c for c in raw_claims] |
| 65 | + |
| 66 | + return ForgeResponse( |
| 67 | + status="success", |
| 68 | + score=getattr(result, "final_score", 0.0), |
| 69 | + claims=claims_list, |
| 70 | + final_memo=final_memo, |
| 71 | + peer_review=getattr(result, "peer_review", {}) |
| 72 | + ) |
| 73 | + |
| 74 | + except Exception as e: |
| 75 | + raise HTTPException(status_code=500, detail=str(e)) |
| 76 | + |
| 77 | +@app.get("/health") |
| 78 | +async def health_check(): |
| 79 | + return {"status": "operational", "engine": "neuro-symbolic-v1"} |
0 commit comments