Spaces:
Running
Running
File size: 3,063 Bytes
76a9232 f579b07 76a9232 f579b07 76a9232 f579b07 1b75c1f f579b07 76a9232 f579b07 76a9232 |
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 |
import os
import json
from datetime import datetime
from typing import List, Dict
import requests
from fastapi import FastAPI, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from pydantic import BaseModel
import plotly.graph_objs as go
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from huggingface_hub import AsyncInferenceClient
app = FastAPI()
# Configuration
models = [
"meta-llama/Meta-Llama-3.1-8B-Instruct",
"meta-llama/Meta-Llama-3.1-70B-Instruct",
"meta-llama/Meta-Llama-3-8B-Instruct",
"meta-llama/Meta-Llama-3-70B-Instruct",
"meta-llama/Llama-Guard-3-8B",
"meta-llama/Llama-2-7b-chat-hf",
"meta-llama/Llama-2-13b-chat-hf",
"deepseek-ai/DeepSeek-Coder-V2-Instruct",
"mistralai/Mistral-7B-Instruct-v0.3",
"mistralai/Mixtral-8x7B-Instruct-v0.1",
]
LOG_FILE = "api_logs.json"
CHECK_INTERVAL = 30 # 1 minute
client = AsyncInferenceClient(token=os.environ["HF_INFERENCE_API_TOKEN"])
# Ensure log file exists
if not os.path.exists(LOG_FILE):
with open(LOG_FILE, "w") as f:
json.dump([], f)
class LogEntry(BaseModel):
model: str
success: bool
timestamp: str
failure_message: str
async def check_apis():
results = []
for model in models:
try:
response = await client.chat_completion(
messages=[{"role": "user", "content": "What is the capital of France?"}],
max_tokens=10,
)
success = True
e = 'success'
except Exception as e:
print(e)
success = False
results.append(LogEntry(
model=model,
success=success,
timestamp=datetime.now().isoformat(),
failure_message=str(e)
))
with open(LOG_FILE, "r+") as f:
logs = json.load(f)
logs.extend([result.dict() for result in results])
f.seek(0)
json.dump(logs, f)
@app.on_event("startup")
async def start_scheduler():
scheduler = AsyncIOScheduler()
scheduler.add_job(check_apis, 'interval', minutes=1)
scheduler.start()
@app.get("/")
async def index():
return FileResponse("static/index.html")
@app.get("/api/logs", response_model=List[LogEntry])
async def get_logs():
with open(LOG_FILE, "r") as f:
logs = json.load(f)
return logs
@app.get("/api/chart-data", response_model=Dict[str, Dict[str, List]])
async def get_chart_data():
with open(LOG_FILE, "r") as f:
logs = json.load(f)
chart_data = {}
for log in logs:
model = log['model']
if model not in chart_data:
chart_data[model] = {'x': [], 'y': []}
chart_data[model]['x'].append(log['timestamp'])
chart_data[model]['y'].append(1 if log['success'] else 0)
return chart_data
# Mount the static files directory
app.mount("/static", StaticFiles(directory="static"), name="static")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860) |