File size: 4,592 Bytes
1774186 9a1ff0e 1774186 3edac2b 1774186 e14b333 1774186 f0f3217 1774186 f0f3217 1774186 f0f3217 1774186 9a1ff0e 1774186 9a1ff0e 1774186 9a1ff0e 1774186 9a1ff0e 1774186 |
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 |
import os
import sqlite3
from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import hashlib
from textblob import TextBlob
from groq import Groq
from starlette.middleware.sessions import SessionMiddleware
app = FastAPI()
from fastapi.staticfiles import StaticFiles
app.mount("/static", StaticFiles(directory="static"), name="static")
# Secret key for sessions
app.add_middleware(SessionMiddleware, secret_key=os.urandom(24))
# Initialize the Groq client with your API key
client = Groq(api_key='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')
# Database initialization
DATABASE_PATH = '/app/chat_history.db'
def init_db():
# Ensure the directory exists
os.makedirs(os.path.dirname(DATABASE_PATH), exist_ok=True)
conn = sqlite3.connect(DATABASE_PATH)
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS users
(id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password TEXT NOT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP)''')
c.execute('''CREATE TABLE IF NOT EXISTS chats
(id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
user_message TEXT NOT NULL,
bot_response TEXT NOT NULL,
sentiment_score REAL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users (id))''')
conn.commit()
conn.close()
init_db()
# Helper functions
def hash_password(password):
return hashlib.sha256(password.encode()).hexdigest()
def get_sentiment(text):
analysis = TextBlob(text)
return analysis.sentiment.polarity
def store_chat(user_id, user_message, bot_response, sentiment_score):
conn = sqlite3.connect(DATABASE_PATH) # Use the DATABASE_PATH defined above
c = conn.cursor()
c.execute('''INSERT INTO chats (user_id, user_message, bot_response, sentiment_score)
VALUES (?, ?, ?, ?)''', (user_id, user_message, bot_response, sentiment_score))
conn.commit()
conn.close()
# New generate_response function using Groq API
def generate_response(prompt):
try:
chat_completion = client.chat.completions.create(
messages=[
{
"role": "system",
"content": "You are Amara, a chatbot created to help. Be helpful, truthful, and attentive to emotions. Consider the previous conversation context for personalized responses. Keep it short and concise."
},
{
"role": "user",
"content": prompt,
}
],
model="llama3-70b-8192",
temperature=0.5,
max_tokens=1024,
top_p=1,
stop=None,
stream=False
)
return chat_completion.choices[0].message.content
except Exception as e:
return f"Error generating response: {e}"
# Define request models
class UserMessage(BaseModel):
message: str
@app.get("/")
async def root():
return {"message": "Welcome to Amara, your chatbot!"}
@app.post("/chat")
async def chat_endpoint(request: Request, message: UserMessage):
user_message = message.message
user_id = request.session.get("user_id")
if not user_id:
raise HTTPException(status_code=401, detail="User not logged in")
# Get user's recent chat history for context
conn = sqlite3.connect(DATABASE_PATH) # Use the DATABASE_PATH defined above
c = conn.cursor()
c.execute('''SELECT user_message, bot_response
FROM chats
WHERE user_id = ?
ORDER BY timestamp DESC LIMIT 5''', (user_id,))
recent_chats = c.fetchall()
conn.close()
context = "Previous conversation:\n"
for user_msg, bot_msg in reversed(recent_chats):
context += f"User: {user_msg}\nAmara: {bot_msg}\n"
full_prompt = f"{context}\nUser: {user_message}\nAmara:"
try:
bot_response = generate_response(full_prompt)
sentiment_score = get_sentiment(user_message)
store_chat(user_id, user_message, bot_response, sentiment_score)
return JSONResponse({
'response': bot_response,
'sentiment': sentiment_score
})
except Exception as e:
raise HTTPException(status_code=500, detail=f"I apologize, but I encountered an error: {str(e)}")
|