ai_focus_new / main.py
Swapnil-101's picture
Update main.py
1ddb359 verified
raw
history blame
7.11 kB
from huggingface_hub import InferenceClient
import random
from flask import Flask, request, jsonify, make_response
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity
from werkzeug.security import generate_password_hash, check_password_hash
import sqlite3
client = InferenceClient("mistralai/Mistral-7B-Instruct-v0.1")
app = Flask(__name__)
file_path = "mentor.txt"
with open(file_path, "r") as file:
mentors_data = file.read()
@app.route('/')
def home():
return jsonify({"message": "Welcome to the Recommendation API!"})
app.config['JWT_SECRET_KEY'] = 123456
jwt = JWTManager(app)
# Create SQLite database
conn = sqlite3.connect('users.db', check_same_thread=False)
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS users
(id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE, password TEXT)''')
c.execute('''CREATE TABLE IF NOT EXISTS user_details
(id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER UNIQUE, full_name TEXT, email TEXT,
FOREIGN KEY(user_id) REFERENCES users(id))''')
conn.commit()
# Endpoint for user registration
@app.route('/register', methods=['POST'])
def register():
data = request.get_json()
username = data.get('username')
password = data.get('password')
if not username or not password:
return jsonify({"message": "Missing username or password"}), 400
hashed_password = generate_password_hash(password)
try:
c.execute("INSERT INTO users (username, password) VALUES (?, ?)", (username, hashed_password))
conn.commit()
return jsonify({"message": "User created successfully"}), 201
except sqlite3.IntegrityError:
return jsonify({"message": "Username already exists"}), 400
# Endpoint for user login
@app.route('/login', methods=['POST'])
def login():
data = request.get_json()
username = data.get('username')
password = data.get('password')
if not username or not password:
return jsonify({"message": "Missing username or password"}), 400
user = c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
if user and check_password_hash(user[2], password):
access_token = create_access_token(identity=username, expires_delta=False)
return jsonify({"access_token": access_token}), 200
else:
return jsonify({"message": "Invalid username or password"}), 401
# Endpoint for storing additional user details
@app.route('/user_details', methods=['POST'])
@jwt_required()
def add_user_details():
current_user = get_jwt_identity()
data = request.get_json()
first_name = data.get('first_name')
last_name = data.get('last_name')
school_name = data.get('school_name')
bachelors_degree = data.get('bachelors_degree')
masters_degree = data.get('masters_degree')
certification = data.get('certification')
activity = data.get('activity')
country = data.get('country')
if not all([first_name, last_name, school_name, bachelors_degree, masters_degree, certification, activity, country]):
return jsonify({"message": "Missing required fields"}), 400
user = c.execute("SELECT * FROM users WHERE username=?", (current_user,)).fetchone()
if not user:
return jsonify({"message": "User not found"}), 404
user_id = user[0]
try:
c.execute("INSERT INTO user_details (user_id, first_name, last_name, school_name, bachelors_degree, "
"masters_degree, certification, activity, country) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
(user_id, first_name, last_name, school_name, bachelors_degree, masters_degree, certification,
activity, country))
conn.commit()
return jsonify({"message": "User details added successfully"}), 201
except sqlite3.IntegrityError:
return jsonify({"message": "User details already exist"}), 400
def format_prompt(message):
# Generate a random user prompt and bot response pair
user_prompt = "UserPrompt"
bot_response = "BotResponse"
return f"<s>[INST] {user_prompt} [/INST] {bot_response}</s> [INST] {message} [/INST]"
@app.route('/get_course', methods=['POST'])
@jwt_required()
def recommend():
current_user = get_jwt_identity()
temperature = 0.9
max_new_tokens = 256
top_p = 0.95
repetition_penalty = 1.0
content = request.json
user_degree = content.get('degree')
user_stream = content.get('stream')
user_semester = content.get('semester')
generate_kwargs = dict(
temperature=temperature,
max_new_tokens=max_new_tokens,
top_p=top_p,
repetition_penalty=repetition_penalty,
do_sample=True,
seed=42,
)
prompt = f""" prompt:
You need to act like as recommendation engine for course recommendation for a student based on below details.
Degree: {user_degree}
Stream: {user_stream}
Current Semester: {user_semester}
Based on above details recommend the courses that relate to the above details
Note: Output should be list in below format:
[course1, course2, course3,...]
Return only answer not prompt and unnecessary stuff, also dont add any special characters or punctuation marks
"""
formatted_prompt = format_prompt(prompt)
stream = client.text_generation(formatted_prompt, **generate_kwargs, stream=False, details=False, return_full_text=False)
return jsonify({"ans": stream})
@app.route('/get_mentor', methods=['POST'])
@jwt_required()
def mentor():
current_user = get_jwt_identity()
temperature = 0.9
max_new_tokens = 256
top_p = 0.95
repetition_penalty = 1.0
content = request.json
user_degree = content.get('degree')
user_stream = content.get('stream')
user_semester = content.get('semester')
courses = content.get('courses')
temperature = float(temperature)
if temperature < 1e-2:
temperature = 1e-2
top_p = float(top_p)
generate_kwargs = dict(
temperature=temperature,
max_new_tokens=max_new_tokens,
top_p=top_p,
repetition_penalty=repetition_penalty,
do_sample=True,
seed=42,
)
prompt = f""" prompt:
You need to act like as recommendataion engine for mentor recommendation for student based on below details also the list of mentors with their experience is attached.
Degree: {user_degree}
Stream: {user_stream}
Current Semester: {user_semester}
courses opted:{courses}
Mentor list= {mentors_data}
Based on above details recommend the mentor that realtes to above details
Note: Output should be list in below format:
[mentor1,mentor2,mentor3,...]
Return only answer not prompt and unnecessary stuff, also dont add any special characters or punctuation marks
"""
formatted_prompt = format_prompt(prompt)
stream = client.text_generation(formatted_prompt, **generate_kwargs, stream=False, details=False, return_full_text=False)
return jsonify({"ans": stream})
if __name__ == '__main__':
app.run(debug=True)