inference client

app.py

@@ -0,0 +1,109 @@
+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 = 60  # 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
+    status_code: int
+
+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 = response.status_code == 200
+        except requests.RequestException:
+            success = False
+        
+        results.append(LogEntry(
+            model=model,
+            success=success,
+            timestamp=datetime.now().isoformat(),
+            status_code=response.status_code
+        ))
+    
+    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)