Initialized as basic Fast api server for handling backend request

Author: shivammjft

server/config/llm.py

@@ -0,0 +1,20 @@
+import os
+from langchain_groq import ChatGroq
+from groq import AsyncGroq
+
+client = AsyncGroq(
+    api_key=os.getenv("GROQ_API_KEY"),
+)
+
+
+
+llm = ChatGroq(
+    groq_api_key=os.getenv('GROQ_API_KEY'),
+    model="llama-3.1-70b-versatile",
+    temperature=0,
+)
+
+audiomodel = ChatGroq(
+groq_api_key=os.getenv('GROQ_API_KEY'),
+model="distil-whisper-large-v3-en",
+)

server/main.py

@@ -0,0 +1,14 @@
+from fastapi import FastAPI
+from routers import chat,search
+import os
+from dotenv import load_dotenv
+load_dotenv()
+
+app = FastAPI()
+
+
+app.include_router(chat.router)
+app.include_router(search.router)
+
+
+

server/models/schemas.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class Query(BaseModel):
+    query: str

server/routers/chat.py

@@ -0,0 +1,24 @@
+from fastapi import APIRouter,HTTPException
+from utils.chat import chat_handler
+from models.schemas import Query
+
+router = APIRouter()
+
+
+@router.post("/chat")
+async def chat(query: Query):
+    try:
+        response = await chat_handler(query.query)
+        return {"response": response}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.get("/complete/")
+async def autocomplete(input:str):
+    try:
+        pass
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+    

server/routers/search.py

@@ -0,0 +1,24 @@
+from fastapi import APIRouter,HTTPException
+from models.schemas import Query
+from typing import Optional
+from utils.search import search_handler
+
+router = APIRouter()
+
+
+@router.post("/search")
+async def search(query: Query, search_type: Optional[str] = "text"):
+    try:
+        response = await search_handler(query.query, search_type)
+        return {"response": response}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    
+
+@router.get("/autocomplete")
+async def recommmendation(input:str):
+    try:
+        # response = await auto_recommendation(query.query, search_type)
+        return {"response": "fr"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

server/utils/chat.py

@@ -0,0 +1,49 @@
+import openai
+import os
+from dotenv import load_dotenv
+load_dotenv()
+
+# OpenAI API key setup
+openai.api_key = "YOUR_OPENAI_API_KEY"
+from groq import AsyncGroq
+
+client = AsyncGroq(
+    api_key=os.getenv('GROQ_API_KEY'),
+)
+
+
+
+async def chat_handler(query: str):
+    response = await openai.ChatCompletion.create(
+        model="gpt-4",  # or another model you want to use
+        messages=[
+            {"role": "user", "content": query}
+        ]
+    )
+    return response.choices[0].message['content']
+
+
+
+async def transcribe_audio(file_content: bytes, file_name: str = "audio.m4a") -> str:
+    """
+    Transcribe the given audio file content using Groq API.
+    
+    :param file_content: The audio file content as bytes.
+    :param file_name: The name of the file, default is "audio.m4a".
+    :return: The transcription text.
+    """
+    try:
+        # Create a transcription of the audio file
+        transcription = await client.audio.transcriptions.create(
+            file=(file_name, file_content), 
+            model="distil-whisper-large-v3-en", 
+            prompt="Specify context or spelling", 
+            response_format="json",  
+            language="en", 
+            temperature=0.0 
+        )
+        
+        return transcription.text
+    
+    except Exception as e:
+        raise RuntimeError(f"Error during transcription: {str(e)}")

server/utils/search.py

@@ -0,0 +1,21 @@
+import httpx
+from duckduckgo_search import DDGS
+
+# DuckDuckGo API initialization
+ddgs = DDGS()
+
+async def search_handler(query: str, search_type: str):
+    if search_type == "text":
+        results = list(ddgs.text(query))
+        return results
+    elif search_type == "images":
+        results = list(ddgs.images(query))
+        return results
+    elif search_type == "news":
+        results = list(ddgs.news(query))
+        return results
+    elif search_type == "maps":
+        results = list(ddgs.maps(query))
+        return results
+    else:
+        raise ValueError("Invalid search type")