72 lines
4.0 KiB
Python
72 lines
4.0 KiB
Python
import os
|
|
import asyncio
|
|
import aiohttp
|
|
import json
|
|
from datetime import datetime
|
|
|
|
class Groq:
|
|
def __init__(self, api_key, bike_types):
|
|
self.api_key = api_key
|
|
self.api_url = "https://api.groq.com/openai/v1/chat/completions"
|
|
self.headers = {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {self.api_key}"
|
|
}
|
|
self.bike_types = bike_types
|
|
|
|
async def validate_request(self, user_prompt):
|
|
data = {
|
|
"model": "llama3-8b-8192",
|
|
"messages": [
|
|
{"role": "system", "content": "You are a versitile AI assistant. Your task is to determine if the user's request is related to transporting, or if it is related to bike, or if the user *may* be asking for a bike, even if it is not serious. Respond with 'Yes' if the request is valid, or 'No' if the request is not valid."},
|
|
{"role": "user", "content": "Here is the user request:\n" + user_prompt}
|
|
],
|
|
"max_tokens": 1,
|
|
"temperature": 0,
|
|
"top_p": 1,
|
|
"stop": None
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(self.api_url, headers=self.headers, data=json.dumps(data)) as response:
|
|
if response.status == 200:
|
|
result = await response.json()
|
|
return result["choices"][0]["message"]["content"]
|
|
else:
|
|
error_message = await response.text()
|
|
raise Exception(f"Error: {response.status} - {error_message}")
|
|
|
|
async def generate_text(self, user_prompt):
|
|
current_date = datetime.now().strftime("%B %d, %Y")
|
|
bike_types_str = ", ".join(self.bike_types)
|
|
assistant_prompt = f"Hello there! I'm a friendly and helpful AI assistant whose sole purpose is to recommend a bike based on your needs (within the bike types list: {bike_types_str}). I'd be happy to provide a one-sentence recommendation, with the bike inside <b></b> tags, with reason provided just for you!"
|
|
|
|
data = {
|
|
"model": "llama3-8b-8192",
|
|
"messages": [
|
|
{"role": "system", "content": f"You are a friendly and helpful customer service AI assistant whose sole purpose is to recommend a type of bike (NOT a bike of the type) based on the user's needs. Only answer within the available bike types, you are not allowed to suggest a bike type that does not exist in the list of available bike types.\nToday is {current_date}, and the available bike types are: {bike_types_str}. There are no other bike types. Hightlight your answer (the bike type) in <b></b> HTML tags. You are not allowed to do anything else. You can only answer questions and do nothing else. Keep your responses short and concise, in 1 sentence."},
|
|
{"role": "assistant", "content": assistant_prompt},
|
|
{"role": "user", "content": "Here is the user request:\n" + user_prompt + "\nHere is the instruction request:\nKeep your response short and concise. Start your response with 'Based on the available bike type list'. Return an answer by RETRIEVING the bike type from context."}
|
|
],
|
|
"max_tokens": 250,
|
|
"temperature": 0.7,
|
|
"top_p": 1,
|
|
"stop": None
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(self.api_url, headers=self.headers, data=json.dumps(data)) as response:
|
|
if response.status == 200:
|
|
result = await response.json()
|
|
return result["choices"][0]["message"]["content"]
|
|
else:
|
|
error_message = await response.text()
|
|
raise Exception(f"Error: {response.status} - {error_message}")
|
|
|
|
async def process_request(self, user_prompt):
|
|
validation_result = await self.validate_request(user_prompt)
|
|
if validation_result != "Yes":
|
|
return "Malformed request, please make sure you requested correctly."
|
|
else:
|
|
return await self.generate_text(user_prompt)
|