Spaces:
Runtime error
Runtime error
File size: 3,552 Bytes
f062f34 e4a2e5e f062f34 4704dc2 f40dde7 f062f34 e4a2e5e f062f34 e285d1f f062f34 f40dde7 f062f34 f40dde7 f062f34 6f2df81 f40dde7 6f2df81 |
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 |
# server.py
from typing import List, Dict, Optional, Tuple
from fastapi import FastAPI, HTTPException
from flask import Flask, request, render_template_string, jsonify
from pydantic import BaseModel, Field
from sklearn.neighbors import KDTree
import numpy as np
import os
import uvicorn
# Configurable number of scales
NUM_SCALES = 8
# Pydantic models
class Profile(BaseModel):
id: str
scales: List[float] = Field(..., min_items=NUM_SCALES, max_items=NUM_SCALES)
class Content(BaseModel):
id: str
scales: List[float] = Field(..., min_items=NUM_SCALES, max_items=NUM_SCALES)
metadata: Optional[Dict] = None
class Recommendation(BaseModel):
content_id: str
distance: float
# Abstract repository interface
class DataRepository:
def add_user(self, user: Profile): ...
def add_content(self, content: Content): ...
def get_user(self, user_id: str) -> Profile: ...
def query_region(self, mins: List[float], maxs: List[float]) -> List[Content]: ...
# In-memory implementation
class InMemoryRepo(DataRepository):
def __init__(self):
self.users: Dict[str, Profile] = {}
self.contents: Dict[str, Content] = {}
def add_user(self, user: Profile):
self.users[user.id] = user
def add_content(self, content: Content):
self.contents[content.id] = content
def get_user(self, user_id: str) -> Profile:
user = self.users.get(user_id)
if not user:
raise ValueError("User not found")
return user
def query_region(self, mins: List[float], maxs: List[float]) -> List[Content]:
# Filter contents by N-dimensional bounding box
return [
c for c in self.contents.values()
if all(mn <= v <= mx for v, mn, mx in zip(c.scales, mins, maxs))
]
# FastAPI setup
# app = FastAPI()
app = Flask(__name__)
repo = InMemoryRepo()
@app.post("/users", response_model=Profile)
def create_user(user: Profile):
repo.add_user(user)
return user
@app.post("/contents", response_model=Content)
def create_content(content: Content):
repo.add_content(content)
return content
@app.get("/recommendations/{user_id}", response_model=List[Recommendation])
def recommend(user_id: str, region_radius: float = 0.2, top_k: int = 5):
try:
user = repo.get_user(user_id)
except ValueError:
raise HTTPException(status_code=404, detail="User not found")
# Build bounding region
mins = [max(0.0, v - region_radius) for v in user.scales]
maxs = [min(1.0, v + region_radius) for v in user.scales]
# Query region
candidates = repo.query_region(mins, maxs)
if not candidates:
return []
# Build KDTree over candidates to find nearest neighbors
mat = np.array([c.scales for c in candidates])
tree = KDTree(mat)
dist, idx = tree.query(np.array(user.scales).reshape(1, -1), k=min(top_k, len(candidates)))
return [
Recommendation(content_id=candidates[i].id, distance=float(d))
for d, i in zip(dist[0], idx[0])
]
# Sample test data when run directly
# Create random test users and content
repo.add_user(Profile(id="user1", scales=list(np.random.rand(NUM_SCALES))))
for i in range(1, 21):
repo.add_content(Content(
id=f"content{i}",
scales=list(np.random.rand(NUM_SCALES)),
metadata={"title": f"Item {i}"}
))
print(recommend("user1"))
print("ran")
if __name__ == "__main__":
port = int(os.environ.get("PORT", 7860))
app.run(host="0.0.0.0", port=port) |