code refactor

src/app.py

@@ -4,7 +4,7 @@ from contextlib import asynccontextmanager
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
-from src.controllers import api_router, websocket_router
+from src.controllers import api_router
 from src.config import logger, DatabaseConfig
 from src.middlewares import AuthenticationMiddleware
 
@@ -45,5 +45,4 @@ async def check_health():
     return {"response": "Service is healthy!"}
 
 
-app.include_router(api_router, prefix="/api")
-app.include_router(websocket_router, prefix="/ws")
+app.include_router(api_router, prefix="/api")

src/config/_database.py

@@ -2,7 +2,7 @@ import os
 from beanie import init_beanie
 from motor.motor_asyncio import AsyncIOMotorClient
 
-from src.models import User, Session
+from src.models import User, Session, File, VectorStoreRecordId, Conversation, Message
 
 
 class DatabaseConfig:
@@ -10,9 +10,19 @@ class DatabaseConfig:
 
         self.MONGO_URI = os.getenv("MONGO_DB_URI")
         self.MONGO_DB_NAME = os.getenv("MONGO_DB_NAME")
-        self.client = AsyncIOMotorClient(self.MONGO_URI)
+        self.client = AsyncIOMotorClient(self.MONGO_URI, uuidRepresentation="standard")
         self.db = self.client[self.MONGO_DB_NAME]
 
     async def init_beanie(self):
 
-        await init_beanie(database=self.db, document_models=[User, Session])
+        await init_beanie(
+            database=self.db,
+            document_models=[
+                User,
+                Session,
+                Conversation,
+                Message,
+                File,
+                VectorStoreRecordId,
+            ],
+        )

src/controllers/__init__.py

@@ -1,28 +1,23 @@
 from fastapi import APIRouter, Depends
 from fastapi.security import APIKeyHeader
+
 from ._auth_controller import AuthController
-from ._ai_voice_controller import AIVoiceController
-from ._ai_text_controller import AITextController
-from ._pinecone_controller import PineconeController
+from ._file_controller import FileController
+from ._conversation_controller import ConversationController
 
 api_router = APIRouter()
-websocket_router = APIRouter()
 
-_auth_controller = AuthController()
-_ai_voice_controller = AIVoiceController()
-_ai_text_controller = AITextController()
-_pinecone_controller = PineconeController()
+auth_router = AuthController().api_router
+conversation_router = ConversationController().api_router
+file_router = FileController().api_router
 
 API_KEY_HEADER = APIKeyHeader(name="Authorization", auto_error=False)
 
-websocket_router.include_router(_ai_voice_controller.websocket_router, prefix="/v1")
-websocket_router.include_router(_ai_text_controller.websocket_router, prefix="/v1")
-
 api_router.include_router(_auth_controller.api_router, prefix="/v1", dependencies=[Depends(API_KEY_HEADER)])
-api_router.include_router(_ai_voice_controller.api_router, prefix="/v1")
-api_router.include_router(_ai_text_controller.api_router, prefix="/v1")
-api_router.include_router(_pinecone_controller.api_router, prefix="/v1")
+api_router.include_router(auth_router, prefix="/v1", tags=["Authentication"])
+api_router.include_router(conversation_router, prefix="/v1", tags=["Conversation"])
+api_router.include_router(file_router, prefix="/v1", tags=["Semantic Search"])
 
-__all__ = ["api_router", "websocket_router"]
+__all__ = ["api_router"]
 __version__ = "0.1.0"
 __author__ = "Ramanjit Singh"

src/controllers/_ai_text_controller.py

@@ -1,31 +0,0 @@
-from fastapi import APIRouter, WebSocket, HTTPException
-
-from src.config import logger
-from src.services import AITextService
-
-
-class AITextController:
-    def __init__(self):
-        self.service = AITextService
-        self.api_router = APIRouter()
-        self.websocket_router = APIRouter()
-        self.api_router.add_api_route(
-            "/ai/text/session", self.get_session, methods=["GET"]
-        )
-        self.websocket_router.add_websocket_route("/ai/text", self.websocket)
-
-    async def get_session(self):
-        try:
-            async with self.service() as service:
-                return await service.create_openai_session()
-        except Exception as e:
-            logger.error(f"Error while building session: {str(e)}")
-            raise HTTPException(status_code=500, detail=str(e))
-
-    async def websocket(self, websocket: WebSocket):
-        try:
-            async with self.service() as service:
-                return await service.openai_websocket_connection(websocket)
-        except Exception as e:
-            logger.error(f"Error during WebSocket connection: {str(e)}")
-            raise HTTPException(status_code=500, detail=str(e))

src/controllers/_conversation_controller.py

@@ -0,0 +1,74 @@
+from fastapi import (
+    APIRouter,
+    HTTPException,
+    WebSocket,
+    Query,
+    Body,
+    Path,
+    WebSocketDisconnect,
+)
+from pydantic import BaseModel
+from src.config import logger
+from src.services import ConversationService
+
+
+class OfferModel(BaseModel):
+    sdp: str
+    type: str
+
+
+class ConversationController:
+    def __init__(self):
+        self.service = ConversationService
+        self.api_router = APIRouter()
+        self.api_router.add_api_route(
+            "/conversations", self.create_conversation, methods=["POST"]
+        )
+        self.api_router.add_api_route(
+            "/conversations/{conversation_id}/webrtc",
+            self.create_webrtc_connection,
+            methods=["POST"],
+        )
+        self.api_router.add_websocket_route("/conversations", self.conversation)
+
+    async def create_conversation(self, modality: str = Query(...)):
+        try:
+            logger.info(f"Creating conversation with modality: {modality}")
+            user_id = "76ebd74f-7395-455b-bb7a-9e890a51e7a4"
+            async with self.service() as service:
+                return await service.create_conversation(
+                    user_id=user_id, modality=modality
+                )
+        except Exception as e:
+            logger.error(f"Error creating conversation: {str(e)}")
+            raise HTTPException(status_code=500, detail="Failed to create conversation")
+
+    async def create_webrtc_connection(
+        self,
+        conversation_id: str = Path(...),
+        ephemeral_key: str = Body(...),
+        offer: OfferModel = Body(...),
+    ):
+        try:
+            async with self.service() as service:
+                return await service.create_webrtc_connection(
+                    conversation_id=conversation_id,
+                    ephemeral_key=ephemeral_key,
+                    offer=offer.model_dump(),
+                )
+        except Exception as e:
+            logger.error(f"Error creating WebRTC connection: {str(e)}")
+            raise HTTPException(
+                status_code=500, detail="Failed to create WebRTC connection"
+            )
+
+    async def conversation(self, websocket: WebSocket):
+        try:
+            async with self.service() as service:
+                await service.conversation(websocket=websocket)
+        except WebSocketDisconnect:
+            logger.info(f"WebSocket disconnected")
+        except Exception as e:
+            logger.error(f"Error in WebSocket conversation: {str(e)}")
+            await websocket.send_json({"error": "Unexpected error occurred"})
+            await websocket.close()

src/controllers/{_ai_voice_controller.py → _file_controller.py}

@@ -1,31 +1,31 @@
-from fastapi import APIRouter, WebSocket, HTTPException
+from fastapi import APIRouter, HTTPException, UploadFile, File
 
 from src.config import logger
-from src.services import AIVoiceService
+from src.services import FileService
 
 
-class AIVoiceController:
+class FileController:
     def __init__(self):
-        self.service = AIVoiceService
+        self.service = FileService
         self.api_router = APIRouter()
-        self.websocket_router = APIRouter()
+        self.api_router.add_api_route("/files", self.insert_file, methods=["POST"])
         self.api_router.add_api_route(
-            "/ai/voice/session", self.get_session, methods=["GET"]
+            "/files/search", self.semantic_search, methods=["GET"]
         )
-        self.websocket_router.add_websocket_route("/ai/voice", self.websocket)
 
-    async def get_session(self):
+    async def insert_file(self, file: UploadFile = File(...)):
         try:
+            user_id = "76ebd74f-7395-455b-bb7a-9e890a51e7a4"
             async with self.service() as service:
-                return await service.create_openai_session()
+                return await service.insert_file(file=file, user_id=user_id)
         except Exception as e:
-            logger.error(f"Error while building session: {str(e)}")
+            logger.error(f"Error while inserting file: {str(e)}")
             raise HTTPException(status_code=500, detail=str(e))
 
-    async def websocket(self, websocket: WebSocket):
+    async def semantic_search(self, query: str):
         try:
             async with self.service() as service:
-                return await service.openai_websocket_connection(websocket)
+                return await service.semantic_search(query=query)
         except Exception as e:
-            logger.error(f"Error during WebSocket connection: {str(e)}")
+            logger.error(f"Error while searching for similar data: {str(e)}")
             raise HTTPException(status_code=500, detail=str(e))

src/controllers/_pinecone_controller.py

@@ -1,43 +0,0 @@
-from fastapi import APIRouter, HTTPException, UploadFile, Form, File
-
-from src.config import logger
-from src.services import PineconeService
-
-
-class PineconeController:
-    def __init__(self):
-        self.pinecone_service = PineconeService
-        self.api_router = APIRouter()
-        self.api_router.add_api_route(
-            "/pinecone/insert", self.insert_data, methods=["POST"]
-        )
-        self.api_router.add_api_route(
-            "/pinecone/search", self.similarity_search, methods=["POST"]
-        )
-        self.api_router.add_api_route(
-            "/pinecone/delete", self.delete, methods=["DELETE"]
-        )
-
-    async def insert_data(self, file: UploadFile):
-        try:
-            async with self.pinecone_service() as service:
-                return await service.insert_data(file=file)
-        except Exception as e:
-            logger.error(f"Error while inserting data: {str(e)}")
-            raise HTTPException(status_code=500, detail=str(e))
-
-    async def similarity_search(self, query: str):
-        try:
-            async with self.pinecone_service() as service:
-                return await service.similarity_search(query=query)
-        except Exception as e:
-            logger.error(f"Error while searching for similar data: {str(e)}")
-            raise HTTPException(status_code=500, detail=str(e))
-
-    async def delete(self, file_name: str, file_id: str):
-        try:
-            async with self.pinecone_service() as service:
-                return await service.delete(file_name=file_name, file_id=file_id)
-        except Exception as e:
-            logger.error(f"Error while deleting data: {str(e)}")
-            raise HTTPException(status_code=500, detail=str(e))

src/models/__init__.py

@@ -1,9 +1,17 @@
 from ._users import User
 from ._sessions import Session
+from ._conversations import Conversation
+from ._messages import Message
+from ._files import File
+from ._vector_store_record_ids import VectorStoreRecordId
 
 __all__ = [
     "User",
     "Session",
+    "Conversation",
+    "Message",
+    "File",
+    "VectorStoreRecordId",
 ]
 __version__ = "0.1.0"
 __author__ = "Narinder Singh"

src/models/_conversations.py

@@ -0,0 +1,18 @@
+from datetime import datetime
+from typing import Optional
+from beanie import Document, Link
+from uuid import UUID, uuid4
+from pydantic import Field
+
+from ._users import User
+
+
+class Conversation(Document):
+    id: UUID = Field(default_factory=uuid4, alias="_id")
+    user_id: Link[User] = Field(...)
+    summary: Optional[str] = None
+    created_at: datetime = Field(default_factory=datetime.now)
+    updated_at: datetime = Field(default_factory=datetime.now)
+
+    class Settings:
+        name = "conversations"

src/models/_files.py

@@ -0,0 +1,18 @@
+from datetime import datetime
+from beanie import Document, Link
+from uuid import UUID, uuid4
+from pydantic import Field
+
+from ._users import User
+
+
+class File(Document):
+    id: UUID = Field(default_factory=uuid4, alias="_id")
+    user_id: Link[User]
+    file_name: str = Field(...)
+    file_type: str = Field(...)
+    created_at: datetime = Field(default_factory=datetime.now)
+    updated_at: datetime = Field(default_factory=datetime.now)
+
+    class Settings:
+        name = "files"

src/models/_messages.py

@@ -0,0 +1,18 @@
+from datetime import datetime
+from beanie import Document, Link
+from uuid import UUID, uuid4
+from pydantic import Field
+
+from ._conversations import Conversation
+
+
+class Message(Document):
+    id: UUID = Field(default_factory=uuid4, alias="_id")
+    conversation_id: Link[Conversation]
+    role: str = Field(...)
+    content: str = Field(...)
+    created_at: datetime = Field(default_factory=datetime.now)
+    updated_at: datetime = Field(default_factory=datetime.now)
+
+    class Settings:
+        name = "messages"

src/models/_users.py

@@ -1,3 +1,4 @@
+from typing import Annotated
 from pydantic import Field
 from typing import Annotated
 from datetime import datetime

src/models/_vector_store_record_ids.py

@@ -0,0 +1,17 @@
+from datetime import datetime
+from beanie import Document, Link
+from uuid import UUID, uuid4
+from pydantic import Field
+
+from ._files import File
+
+
+class VectorStoreRecordId(Document):
+    id: UUID = Field(default_factory=uuid4, alias="_id")
+    file_id: Link[File]
+    record_id: str = Field(...)
+    created_at: datetime = Field(default_factory=datetime.now)
+    updated_at: datetime = Field(default_factory=datetime.now)
+
+    class Settings:
+        name = "vector_store_record_ids"

src/repositories/__init__.py

@@ -1,12 +1,21 @@
-from ._user_repository import UserRepository
 from ._base_repository import BaseRepository
+from ._user_repository import UserRepository
 from ._session_repository import SessionRepository
+from ._conversation_repository import ConversationRepository
+from ._message_repository import MessageRepository
+from ._file_repository import FileRepository
+from ._vector_store_record_id_repository import VectorStoreRecordIdRepository
+
 
 __all__ = [
     "UserRepository",
     "BaseRepository",
     "SessionRepository",
+    "ConversationRepository",
+    "MessageRepository",
+    "FileRepository",
+    "VectorStoreRecordIdRepository",
 ]
 
 __version__ = "0.1.0"
-__author__ = "Ramanjit Singh"
+__author__ = "Ramanjit Singh"

src/repositories/_conversation_repository.py

@@ -0,0 +1,8 @@
+from src.models import Conversation
+
+from ._base_repository import BaseRepository
+
+
+class ConversationRepository(BaseRepository):
+    def __init__(self):
+        super().__init__(model=Conversation)

src/repositories/_file_repository.py

@@ -0,0 +1,8 @@
+from src.models import File
+
+from ._base_repository import BaseRepository
+
+
+class FileRepository(BaseRepository):
+    def __init__(self):
+        super().__init__(model=File)

src/repositories/_message_repository.py

@@ -0,0 +1,8 @@
+from src.models import Message
+
+from ._base_repository import BaseRepository
+
+
+class MessageRepository(BaseRepository):
+    def __init__(self):
+        super().__init__(model=Message)

src/repositories/_vector_store_record_id_repository.py

@@ -0,0 +1,8 @@
+from src.models import VectorStoreRecordId
+
+from ._base_repository import BaseRepository
+
+
+class VectorStoreRecordIdRepository(BaseRepository):
+    def __init__(self):
+        super().__init__(model=VectorStoreRecordId)

src/services/__init__.py

@@ -1,17 +1,16 @@
 from ._auth_service import AuthService
 from ._user_service import UserService
-from ._ai_text_service import AITextService
+from ._file_service import FileService
 from ._session_service import SessionService
-from ._ai_voice_service import AIVoiceService
-from ._pinecone_service import PineconeService
+from ._conversation_service import ConversationService
+
 
 __all__ = [
+    "FileService",
     "AuthService",
     "UserService",
-    "AITextService",
     "SessionService",
-    "AIVoiceService",
-    "PineconeService",
+    "ConversationService",
 ]
 
 __version__ = "0.1.0"

src/services/_ai_voice_service.py

@@ -1,100 +0,0 @@
-import json
-
-from ._pinecone_service import PineconeService
-from src.config import logger
-from src.utils import OpenAIClient
-
-
-class AIVoiceService:
-    def __init__(self):
-        self.openai_client = OpenAIClient
-        self.pinecone_service = PineconeService
-        self.functions_dictionary = {
-            "get_relevant_information": {
-                "service": self.pinecone_service,
-                "function": "similarity_search",
-            },
-        }
-
-    async def __aenter__(self):
-        return self
-
-    async def __aexit__(self, *args):
-        pass
-
-    async def handle_user_transcript(self, message):
-        logger.info(f"User transcript: {message['transcript']}")
-
-    async def handle_ai_transcript(self, message):
-        if (
-            message["response"]["status"] == "completed"
-            and message["response"]["output"][0]["type"] == "message"
-        ):
-            logger.info(
-                f"Assistant transcript: {message["response"]["output"][0]["content"][0]["transcript"]}"
-            )
-
-    async def handle_ai_function_call(self, data):
-        tool_name = data["name"]
-        arguments = json.loads(data["arguments"])
-        logger.info(f"Function call: {tool_name} \nArguments: {arguments}")
-
-        function_info = self.functions_dictionary.get(tool_name)
-        if not function_info:
-            raise AttributeError(f"Function {tool_name} not found in dictionary")
-
-        service_class = function_info["service"]
-        async with service_class() as service:
-            func = getattr(service, function_info["function"], None)
-            if not func:
-                raise AttributeError(
-                    f"No such function {function_info['function']} in service"
-                )
-            response = await func(arguments["query"])
-            response = f"Context : {response}\n\n# Use this context to respond to the user query : {arguments['query']}"
-            return response
-
-    async def create_openai_session(self):
-        async with self.openai_client() as client:
-            return await client.create_openai_session()
-
-    async def openai_websocket_connection(self, websocket):
-        openai_server_events = {
-            "function_call": "response.function_call_arguments.done",
-            "input_transcription": "conversation.item.input_audio_transcription.completed",
-            "output_transcription": "response.done",
-        }
-        await websocket.accept()
-
-        while True:
-            data = await websocket.receive_text()
-            message = json.loads(data)
-
-            if message["type"] == openai_server_events["function_call"]:
-                response = await self.handle_ai_function_call(message)
-                event_response = {
-                    "type": "conversation.item.create",
-                    "previous_item_id": None,
-                    "item": {
-                        "type": "message",
-                        "role": "user",
-                        "content": [
-                            {
-                                "type": "input_text",
-                                "text": response,
-                            }
-                        ],
-                    },
-                }
-                await websocket.send_text(json.dumps(event_response))
-                event_response_2 = {"type": "response.create"}
-                await websocket.send_text(json.dumps(event_response_2))
-
-            elif message["type"] == openai_server_events["input_transcription"]:
-                await self.handle_user_transcript(message)
-
-            elif message["type"] == openai_server_events["output_transcription"]:
-                await self.handle_ai_transcript(message)
-
-            elif message["type"] == "error":
-                logger.error(f"Error: {message['error']}")

src/services/{_ai_text_service.py → _conversation_service.py}

@@ -1,18 +1,25 @@
 import json
+from fastapi import WebSocket
+from uuid import UUID
+from bson import ObjectId
 
-from ._pinecone_service import PineconeService
-from src.config import logger
+from ._file_service import FileService
 from src.utils import OpenAIClient
+from src.repositories import ConversationRepository, MessageRepository
+from src.models import Conversation, Message
+from src.config import logger
 
 
-class AITextService:
+class ConversationService:
     def __init__(self):
         self.openai_client = OpenAIClient
-        self.pinecone_service = PineconeService
+        self.conversation_repository = ConversationRepository()
+        self.message_repository = MessageRepository()
+        self.file_service = FileService
         self.functions_dictionary = {
             "get_relevant_information": {
-                "service": self.pinecone_service,
-                "function": "similarity_search",
+                "service": self.file_service,
+                "function": "semantic_search",
             },
         }
 
@@ -22,47 +29,63 @@ class AITextService:
     async def __aexit__(self, *args):
         pass
 
-    async def handle_user_message(self, user_message):
-        logger.info(f"User message: {user_message}")
+    async def create_conversation(self, user_id, modality):
+        conversation = await self.conversation_repository.insert_one(
+            Conversation(user_id=user_id)
+        )
 
-    async def handle_ai_response(self, ai_response):
-        logger.info(f"AI response: {ai_response}")
+        text_mode_only = True if modality == "text" else False
+        async with self.openai_client() as client:
+            session_data = await client.create_openai_session(
+                text_mode_only=text_mode_only
+            )
 
-    async def handle_ai_function_call(self, data):
-        tool_name = data["name"]
-        arguments = json.loads(data["arguments"])
-        logger.info(f"Function call: {tool_name} \nArguments: {arguments}")
+        return {
+            "conversation_id": str(conversation.id),
+            "session_data": session_data,
+        }
 
-        function_info = self.functions_dictionary.get(tool_name)
-        if not function_info:
-            raise AttributeError(f"Function {tool_name} not found in dictionary")
+    async def create_webrtc_connection(self, conversation_id, ephemeral_key, offer):
+        async with self.openai_client() as client:
+            return await client.create_webrtc_connection(
+                ephemeral_key=ephemeral_key, offer=offer
+            )
 
-        service_class = function_info["service"]
-        async with service_class() as service:
-            func = getattr(service, function_info["function"], None)
-            if not func:
-                raise AttributeError(
-                    f"No such function {function_info['function']} in service"
-                )
-            response = await func(arguments["query"])
-            response = f"Context : {response}\n\n# Use this context to respond to the user query : {arguments['query']}"
-            return response
+    async def conversation(self, websocket: WebSocket):
+        try:
+            query_params = websocket.query_params
+            conversation_id = query_params.get("conversation_id")
+            modality = query_params.get("modality")
 
-    async def create_openai_session(self):
-        async with self.openai_client() as client:
-            return await client.create_openai_session(text_mode_only=True)
+            await websocket.accept()
 
-    async def openai_websocket_connection(self, websocket):
-        await websocket.accept()
+            if not conversation_id or not modality:
+                await websocket.close(
+                    code=1008, reason="Missing or invalid query parameters"
+                )
+                return
 
+            if modality == "text":
+                await self.handle_text_conversion(websocket, conversation_id)
+            elif modality == "voice":
+                await self.handle_voice_conversion(websocket, conversation_id)
+            else:
+                await websocket.close(code=1008, reason="Unsupported modality")
+                return
+        finally:
+            await self.handle_conversation_summary(conversation_id)
+            # await websocket.close()
+
+    async def handle_text_conversion(self, websocket, conversation_id):
         while True:
             data = await websocket.receive_text()
             message = json.loads(data)
 
             openai_server_events = {
                 "session_created": message["type"] == "session.created",
-                "function_call": message["type"]
-                == "response.function_call_arguments.done",
+                "function_call": (
+                    message["type"] == "response.function_call_arguments.done"
+                ),
                 "ai_response": (
                     message["type"] == "response.done"
                     and message["response"]["status"] == "completed"
@@ -73,7 +96,9 @@ class AITextService:
 
             if openai_server_events["session_created"]:
                 user_query = input("Enter your query: ")
-                await self.handle_user_message(user_message=user_query)
+                await self.handle_user_message(
+                    message_content=user_query, conversation_id=conversation_id
+                )
 
                 event_stage_1 = {
                     "type": "conversation.item.create",
@@ -115,10 +140,14 @@ class AITextService:
 
             if openai_server_events["ai_response"]:
                 ai_response = message["response"]["output"][0]["content"][0]["text"]
-                await self.handle_ai_response(ai_response=ai_response)
+                await self.handle_ai_message(
+                    message_content=ai_response, conversation_id=conversation_id
+                )
 
                 user_query = input("Enter your query: ")
-                await self.handle_user_message(user_message=user_query)
+                await self.handle_user_message(
+                    message_content=user_query, conversation_id=conversation_id
+                )
 
                 event_stage_1 = {
                     "type": "conversation.item.create",
@@ -140,3 +169,103 @@ class AITextService:
 
             elif message["type"] == "error":
                 logger.error(f"Error: {message['error']}")
+
+    async def handle_voice_conversion(self, websocket, conversation_id):
+        while True:
+            data = await websocket.receive_text()
+            message = json.loads(data)
+
+            openai_server_events = {
+                "function_call": (
+                    message["type"] == "response.function_call_arguments.done"
+                ),
+                "input_transcription": (
+                    message["type"]
+                    == "conversation.item.input_audio_transcription.completed"
+                ),
+                "output_transcription": (
+                    message["type"] == "response.done"
+                    and message["response"]["status"] == "completed"
+                    and message["response"]["output"][0]["type"] == "message"
+                ),
+            }
+
+            if openai_server_events["function_call"]:
+                response = await self.handle_ai_function_call(message)
+                event_response = {
+                    "type": "conversation.item.create",
+                    "previous_item_id": None,
+                    "item": {
+                        "type": "message",
+                        "role": "user",
+                        "content": [
+                            {
+                                "type": "input_text",
+                                "text": response,
+                            }
+                        ],
+                    },
+                }
+                await websocket.send_text(json.dumps(event_response))
+                event_response_2 = {"type": "response.create"}
+                await websocket.send_text(json.dumps(event_response_2))
+
+            elif openai_server_events["input_transcription"]:
+                user_message = message["transcript"]
+                await self.handle_user_message(
+                    message_content=user_message, conversation_id=conversation_id
+                )
+
+            elif openai_server_events["output_transcription"]:
+                ai_response = message["response"]["output"][0]["content"][0][
+                    "transcript"
+                ]
+                await self.handle_ai_message(
+                    message_content=ai_response, conversation_id=conversation_id
+                )
+
+            elif message["type"] == "error":
+                logger.error(f"Error: {message['error']}")
+
+    async def handle_user_message(self, message_content, conversation_id):
+        logger.info(f"User Query: {message_content}")
+        message = await self.message_repository.insert_one(
+            Message(
+                conversation_id=conversation_id,
+                role="user",
+                content=message_content,
+            )
+        )
+
+    async def handle_ai_message(self, message_content, conversation_id):
+        logger.info(f"AI Response: {message_content}")
+        message = await self.message_repository.insert_one(
+            Message(
+                conversation_id=conversation_id,
+                role="assistant",
+                content=message_content,
+            )
+        )
+
+    async def handle_conversation_summary(self, conversation_id):
+        pass
+
+    async def handle_ai_function_call(self, data):
+        tool_name = data["name"]
+        arguments = json.loads(data["arguments"])
+        logger.info(f"Function call: {tool_name} \nArguments: {arguments}")
+
+        function_info = self.functions_dictionary.get(tool_name)
+        if not function_info:
+            raise AttributeError(f"Function {tool_name} not found in dictionary")
+
+        service_class = function_info["service"]
+        async with service_class() as service:
+            func = getattr(service, function_info["function"], None)
+            if not func:
+                raise AttributeError(
+                    f"No such function {function_info['function']} in service"
+                )
+            response = await func(arguments["query"])
+            response = f"Context : {response}\n\n# Use this context to respond to the user query : {arguments['query']}"
+            return response

src/services/{_pinecone_service.py → _file_service.py}

@@ -1,15 +1,15 @@
-import os
-import aiofiles
-from asynctempfile import TemporaryDirectory
-
 from fastapi import UploadFile
 
+from src.repositories import FileRepository, VectorStoreRecordIdRepository
+from src.models import File, VectorStoreRecordId
 from src.utils import PineconeClient
 
 
-class PineconeService:
+class FileService:
     def __init__(self):
         self.pinecone_client = PineconeClient
+        self.file_repository = FileRepository()
+        self.vector_store_record_id_repository = VectorStoreRecordIdRepository()
 
     async def __aenter__(self):
         return self
@@ -17,7 +17,7 @@ class PineconeService:
     async def __aexit__(self, *args):
         pass
 
-    async def insert_data(self, file: UploadFile):
+    async def insert_file(self, file: UploadFile, user_id: str):
         try:
             metadata = [{"file_name": file.filename}]
             while content := await file.read():
@@ -25,13 +25,25 @@ class PineconeService:
                     ids = await client.insert_data(
                         text=[content.decode()], metadata=metadata
                     )
+
+            file_data = await self.file_repository.insert_one(
+                File(
+                    user_id=user_id,
+                    file_name=file.filename,
+                    file_type=file.content_type,
+                )
+            )
+
+            for id in ids:
+                await self.vector_store_record_id_repository.insert_one(
+                    VectorStoreRecordId(file_id=file_data.id, record_id=id)
+                )
+
+            return file_data
+
         finally:
             await file.close()
 
-    async def similarity_search(self, query: str):
+    async def semantic_search(self, query):
         async with self.pinecone_client() as client:
             return await client.similarity_search(query=query)
-
-    async def delete(self, file_name: str, file_id: str):
-        async with self.pinecone_client() as client:
-            pass

src/utils/_openai_client.py

@@ -1,5 +1,8 @@
 import os
 import httpx
+from fastapi import HTTPException
+from aiortc import RTCPeerConnection, RTCSessionDescription
+
 from src.config import logger
 
 
@@ -25,13 +28,14 @@ _OPENAI_TOOLS = [
 class OpenAIClient:
     def __init__(self):
         self.session_url = "https://api.openai.com/v1/realtime/sessions"
-        self.model = "gpt-4o-realtime-preview-2024-12-17"
+        self.model = "gpt-4o-mini-realtime-preview-2024-12-17"
+        self.webrtc_url = f"https://api.openai.com/v1/realtime?model={self.model}"
         self.modalities = ["text", "audio"]
         self.voice = "alloy"
         self.transcription_model = "whisper-1"
         self.temperature = 0.8
         self.max_response_output_tokens = 1000
-        self.system_prompt = "You are a helpful AI assistant. Always provide clear and concise responses to the user's questions.Your responses must be very consize and small"
+        self.system_prompt = "You are a helpful AI assistant. Always provide clear and concise responses to the user's questions.Your responses must be very consize and small. Always use your tools to provide the best possible response to the user.Avoid giving long responses to the user's questions.For any query related to Keep Me, use you tool"
         self.tools = _OPENAI_TOOLS
 
     async def __aenter__(self):
@@ -63,5 +67,28 @@ class OpenAIClient:
                 self.session_url, json=payload, headers=headers
             )
             response.raise_for_status()
-            logger.info(f"OpenAI session created: {response.json()}")
             return response.json()
+
+    async def create_webrtc_connection(self, ephemeral_key, offer):
+        peer_connection = RTCPeerConnection()
+        offer_desc = RTCSessionDescription(sdp=offer["sdp"], type=offer["type"])
+        await peer_connection.setRemoteDescription(offer_desc)
+
+        async with httpx.AsyncClient() as client:
+            headers = {
+                "Authorization": f"Bearer {ephemeral_key}",
+                "Content-Type": "application/sdp",
+            }
+            response = await client.post(
+                self.webrtc_url, content=offer_desc.sdp, headers=headers
+            )
+            if not response.is_success:
+                error_data = response.json()
+                error_message = error_data.get("error", {}).get(
+                    "message", response.reason_phrase
+                )
+                raise HTTPException(
+                    status_code=response.status_code, detail=error_message
+                )
+            answer_sdp = response.text
+            return {"answer": {"sdp": answer_sdp, "type": "answer"}}