Spaces:
Running
Running
File size: 5,706 Bytes
c151c44 ed5b42d c151c44 8fde879 17ea087 8fde879 ed5b42d 17ea087 c151c44 ed5b42d c151c44 525c716 c151c44 ed5b42d c151c44 ed5b42d c151c44 fefb5c9 8fde879 c151c44 fefb5c9 dcefa44 fefb5c9 8fde879 fefb5c9 8fde879 fefb5c9 8fde879 fefb5c9 dcefa44 c151c44 8fde879 c151c44 8fde879 c151c44 525c716 c151c44 8fde879 c151c44 8fde879 c151c44 8fde879 c151c44 8fde879 c151c44 |
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 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 |
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from backend.utils import generate_completions
from backend import config
from backend.database import get_db_connection
import psycopg2
from psycopg2.extras import RealDictCursor
from typing import Union, List, Literal, Optional
import logging
import json
logging.basicConfig(level=logging.INFO)
app = FastAPI()
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allows all origins
allow_credentials=True,
allow_methods=["*"], # Allows all methods
allow_headers=["*"], # Allows all headers
)
# Dependency to get database connection
async def get_db():
conn = await get_db_connection()
try:
yield conn
finally:
conn.close()
class Message(BaseModel):
role: Literal["user", "assistant"]
content: str
class GenerationRequest(BaseModel):
user_id: int
query: Union[str, List[Message]]
class MetadataRequest(BaseModel):
query: str
# Global metadata variables
native_language: Optional[str] = None
target_language: Optional[str] = None
proficiency: Optional[str] = None
@app.get("/")
async def root():
return {"message": "Welcome to the AI Learning Assistant API!"}
@app.post("/extract/metadata")
async def extract_metadata(data: MetadataRequest):
logging.info(f"Query: {data.query}")
try:
response_str = await generate_completions.get_completions(
data.query,
config.language_metadata_extraction_prompt
)
metadata_dict = json.loads(response_str)
# Update globals for other endpoints
globals()['native_language'] = metadata_dict.get('native_language', 'unknown')
globals()['target_language'] = metadata_dict.get('target_language', 'unknown')
globals()['proficiency'] = metadata_dict.get('proficiency_level', 'unknown')
return JSONResponse(
content={
"data": metadata_dict,
"type": "language_metadata",
"status": "success"
},
status_code=200
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/generate/curriculum")
async def generate_curriculum(data: GenerationRequest):
try:
# Use previously extracted metadata
instructions = (
config.curriculum_instructions
.replace("{native_language}", native_language or "unknown")
.replace("{target_language}", target_language or "unknown")
.replace("{proficiency}", proficiency or "unknown")
)
response = await generate_completions.get_completions(
data.query,
instructions
)
return JSONResponse(
content={
"data": response,
"type": "curriculum",
"status": "success"
},
status_code=200
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/generate/flashcards")
async def generate_flashcards(data: GenerationRequest):
try:
# Use previously extracted metadata
instructions = (
config.flashcard_mode_instructions
.replace("{native_language}", native_language or "unknown")
.replace("{target_language}", target_language or "unknown")
.replace("{proficiency}", proficiency or "unknown")
)
response = await generate_completions.get_completions(
data.query,
instructions
)
return JSONResponse(
content={
"data": response,
"type": "flashcards",
"status": "success"
},
status_code=200
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/generate/exercises")
async def generate_exercises(data: GenerationRequest):
try:
# Use previously extracted metadata
instructions = (
config.exercise_mode_instructions
.replace("{native_language}", native_language or "unknown")
.replace("{target_language}", target_language or "unknown")
.replace("{proficiency}", proficiency or "unknown")
)
response = await generate_completions.get_completions(
data.query,
instructions
)
return JSONResponse(
content={
"data": response,
"type": "exercises",
"status": "success"
},
status_code=200
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/generate/simulation")
async def generate_simulation(data: GenerationRequest):
try:
# Use previously extracted metadata
instructions = (
config.simulation_mode_instructions
.replace("{native_language}", native_language or "unknown")
.replace("{target_language}", target_language or "unknown")
.replace("{proficiency}", proficiency or "unknown")
)
response = await generate_completions.get_completions(
data.query,
instructions
)
return JSONResponse(
content={
"data": response,
"type": "simulation",
"status": "success"
},
status_code=200
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) |