-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathrequirements.txt
More file actions
72 lines (60 loc) · 2.15 KB
/
requirements.txt
File metadata and controls
72 lines (60 loc) · 2.15 KB
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
import os
import json
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from openai import OpenAI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi.dependencies import Depends
# Environment Variables:
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
# Database Configuration (if applicable):
DATABASE_URL = os.environ.get("DATABASE_URL")
# Application Configuration:
DEBUG = os.environ.get("DEBUG", False)
PORT = int(os.environ.get("PORT", 5000))
# Create the FastAPI Application
app = FastAPI()
# Data Models:
class PromptRequest(BaseModel):
model: str
prompt: str
# Define a route for generating responses
@app.post("/generate_response")
async def generate_response(request: Request):
# Parse the incoming request body as JSON
request_data = await request.json()
# Validate the request data
try:
prompt = PromptRequest(**request_data)
except ValueError as e:
raise HTTPException(status_code=400, detail=f"Invalid request data: {e}")
# Use OpenAI API to generate a response
try:
openai = OpenAI(api_key=OPENAI_API_KEY)
response = await openai.completions.create(
model=prompt.model,
prompt=prompt.prompt,
)
return JSONResponse({"response": response.choices[0].text})
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error processing request: {e}")
# Database Connection (if applicable):
if DATABASE_URL:
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Define a function to get a database session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Example database interaction (if applicable):
@app.get("/get_response_history")
async def get_response_history(db: SessionLocal = Depends(get_db)):
# Implement logic to retrieve response history from the database
# Example:
responses = db.query(ResponseHistory).all()
return JSONResponse({"responses": responses})