File size: 3,995 Bytes
6bb0212 189570d 642b6b3 6bb0212 642b6b3 6bb0212 ff4d74f 6bb0212 6756da2 642b6b3 6756da2 642b6b3 6756da2 d2a0c5e 6c7e606 6bb0212 642b6b3 73939b2 6bb0212 7cfde2b 642b6b3 7cfde2b 642b6b3 0491b8c 642b6b3 5a6ca7f 642b6b3 5a6ca7f 6c7e606 6bb0212 0491b8c 642b6b3 | 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 | # hf_demo.py – ARF v4 API with Memory
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import gradio as gr
import numpy as np
from datetime import datetime
# ARF v4 imports
from agentic_reliability_framework.core.governance.risk_engine import RiskEngine
from agentic_reliability_framework.runtime.memory import (
create_faiss_index,
RAGGraphMemory,
MemoryConstants
)
app = FastAPI(title="ARF v4 API with Memory")
# Enable CORS for your frontend
app.add_middleware(
CORSMiddleware,
allow_origins=["https://arf-frontend-sandy.vercel.app"],
allow_methods=["*"],
)
# ---------------------------------------------------------------------------
# Initialize ARF components
# ---------------------------------------------------------------------------
risk_engine = RiskEngine()
# Create FAISS index and memory (using default dimension from constants)
faiss_index = create_faiss_index(dim=MemoryConstants.VECTOR_DIM)
memory = RAGGraphMemory(faiss_index)
# ---------------------------------------------------------------------------
# API Endpoints
# ---------------------------------------------------------------------------
@app.get("/")
async def root():
return {
"service": "ARF OSS API",
"version": "4.0.0",
"status": "operational",
"memory_stats": memory.get_graph_stats() if memory.has_historical_data() else "empty"
}
@app.get("/health")
async def health():
return {"status": "ok", "version": "4.0.0"}
@app.get("/api/v1/get_risk")
async def get_risk():
# Get current system risk (this method exists in v4)
risk_score = risk_engine.get_current_risk()
return {
"system_risk": risk_score.mean,
"status": "critical" if risk_score.mean > 0.8 else "normal"
}
@app.post("/api/v1/incident")
async def store_incident(event_data: dict, analysis: dict):
"""
Store an incident in memory.
event_data should contain at least 'component', 'latency_p99', 'error_rate', etc.
"""
# In a real implementation you would parse a ReliabilityEvent.
# For simplicity, we pass a dict – you may need to adapt.
try:
incident_id = memory.store_incident(event_data, analysis)
return {"incident_id": incident_id}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/v1/memory/similar")
async def find_similar_incidents(action: str, k: int = 5):
"""
Find incidents similar to the given action text.
This uses a simple embedding fallback (random) for OSS.
For production, integrate sentence-transformers.
"""
# Create a dummy event from the action text
class DummyEvent:
def __init__(self, action):
self.component = "user_action"
self.latency_p99 = 0.0
self.error_rate = 0.0
self.throughput = 0
self.cpu_util = 0.0
self.memory_util = 0.0
self.timestamp = datetime.now()
self.severity = "low"
event = DummyEvent(action)
analysis = {"action": action}
similar = memory.find_similar(event, analysis, k=k)
# Convert nodes to dict
results = []
for node in similar:
results.append({
"incident_id": node.incident_id,
"component": node.component,
"severity": node.severity,
"timestamp": node.timestamp,
"metrics": node.metrics,
"agent_analysis": node.agent_analysis,
"similarity_score": node.metadata.get("similarity_score", 0.0)
})
return {"similar": results, "count": len(results)}
@app.get("/api/v1/memory/stats")
async def memory_stats():
return memory.get_graph_stats()
# Optional Gradio interface
iface = gr.Interface(
fn=lambda: f"ARF v4 - Current risk: {risk_engine.get_current_risk().mean:.2f}",
inputs=[],
outputs="text",
title="ARF v4 Demo"
)
app = gr.mount_gradio_app(app, iface, path="/") |