Update main.py

main.py

@@ -40,7 +40,6 @@ llm = ChatOpenAI(
 # ---------------------------
 app = FastAPI()
 
-# Load MongoDB credentials from environment variables
 MONGO_USER = os.getenv("MONGO_USER")
 MONGO_PASSWORD = os.getenv("MONGO_PASSWORD")
 MONGO_CLUSTER = os.getenv("MONGO_CLUSTER")
@@ -58,10 +57,6 @@ chat_history_collection = database["chat_history"]
 # Classification Helper using LLM (for text messages only)
 # ---------------------------
 def classify_message_content(message: str) -> str:
-    """
-    Use the LLM to classify the provided message as either 'spam' or 'unknown'.
-    The prompt instructs the LLM to respond with only one word.
-    """
     prompt = (
         "Classify the following message as either 'spam' or 'unknown'. "
         "Respond with only one word: spam or unknown.\n"
@@ -93,20 +88,12 @@ class IncomingMessage(BaseModel):
 # Chat History Helper Functions
 # ---------------------------
 def get_chat_history(caller_number: str) -> List[dict]:
-    """
-    Retrieve the conversation history for a given caller from MongoDB.
-    Each entry is a dict with 'role' and 'content' keys.
-    """
     doc = chat_history_collection.find_one({"caller_number": caller_number})
     if doc and "messages" in doc:
         return doc["messages"]
     return []
 
 def update_chat_history(caller_number: str, role: str, content: str):
-    """
-    Append a new message (with role and content) to the chat history for the given caller.
-    If no document exists, one is created (upsert).
-    """
     chat_history_collection.update_one(
         {"caller_number": caller_number},
         {"$push": {"messages": {"role": role, "content": content}}},
@@ -117,20 +104,11 @@ def update_chat_history(caller_number: str, role: str, content: str):
 # Conversation Simulation Functions
 # ---------------------------
 def simulate_text_conversation(caller_number: str, initial_message: str, conversation_type: str = "unknown") -> str:
-    """
-    Simulate a multi-turn text conversation.
-    For unknown texts, use a neutral tone.
-    (Spam texts are handled immediately without simulation.)
-    """
     if conversation_type == "unknown":
         system_prompt = (
             f"You are a call assistant. The unknown caller's text message is '{initial_message}'.\n"
-            "Simulate a multi-turn conversation that unfolds as follows:\n"
-            "1. Immediate reply: 'Unknown text – Bot replying'\n"
-            "2. Follow-up: 'Who is this?'\n"
-            "3. Then: 'What is the purpose of your message?'\n"
-            "4. Finally: 'Please provide more details.'\n"
-            "Return all steps in one message, each on a new line."
+            "Simulate a multi-turn conversation."
+            
         )
     else:
         system_prompt = "You are a call assistant."
@@ -154,21 +132,15 @@ def simulate_text_conversation(caller_number: str, initial_message: str, convers
     return assistant_response
 
 def simulate_call_conversation(caller_number: str, initial_message: str, conversation_type: str = "spam") -> str:
-    """
-    Simulate a multi-turn call conversation.
-    For voice calls, if the caller is not saved, always use the humorous spam conversation.
-    """
     if conversation_type == "spam":
         system_prompt = (
             f"You are HumorBot on a phone call. The caller's number is {caller_number} and the transcribed message is '{initial_message}'.\n"
             "Simulate a multi-turn spam call conversation."
-            
         )
     elif conversation_type == "unknown":
         system_prompt = (
             f"You are a call assistant. The caller's number is {caller_number} and the transcribed message is '{initial_message}'.\n"
             "Simulate a multi-turn unknown call conversation."
-            
         )
     else:
         system_prompt = "You are a call assistant."
@@ -196,10 +168,6 @@ def simulate_call_conversation(caller_number: str, initial_message: str, convers
 # ---------------------------
 @app.post("/contacts", response_model=List[Contact])
 def create_contacts(contacts: List[Contact]):
-    """
-    Save a list of contacts into MongoDB.
-    Email and name are optional, but phone is required and must be unique.
-    """
     contacts_to_insert = []
     for contact in contacts:
         if contacts_collection.find_one({"phone": contact.phone}):
@@ -212,17 +180,11 @@ def create_contacts(contacts: List[Contact]):
 
 @app.get("/contacts", response_model=List[Contact])
 def get_all_contacts():
-    """
-    Retrieve all contacts from MongoDB.
-    """
     contacts = list(contacts_collection.find({}, {"_id": 0}))
     return contacts
 
 @app.get("/contacts/{phone}", response_model=Contact)
 def get_contact(phone: str):
-    """
-    Retrieve a specific contact by phone number.
-    """
     contact = contacts_collection.find_one({"phone": phone}, {"_id": 0})
     if not contact:
         raise HTTPException(status_code=404, detail="Contact not found")
@@ -230,13 +192,6 @@ def get_contact(phone: str):
 
 @app.post("/incoming-message")
 def process_incoming_message(incoming: IncomingMessage):
-    """
-    Process an incoming text message:
-      - If the sender's number is in saved contacts, return a primary message.
-      - Otherwise, use the LLM to classify the message content.
-          • If classified as spam, respond immediately.
-          • If classified as unknown, simulate a neutral multi-turn conversation.
-    """
     if contacts_collection.find_one({"phone": incoming.phone}):
         return {
             "status": "primary",
@@ -258,9 +213,6 @@ def process_incoming_message(incoming: IncomingMessage):
 
 @app.get("/messages/{caller_number}")
 def get_messages(caller_number: str):
-    """
-    Retrieve the conversation history (messages) for a given caller.
-    """
     messages = get_chat_history(caller_number)
     if not messages:
         raise HTTPException(status_code=404, detail="No messages found for this caller")
@@ -268,9 +220,6 @@ def get_messages(caller_number: str):
 
 @app.post("/setup-call-forwarding")
 def setup_call_forwarding():
-    """
-    Simulate call forwarding setup.
-    """
     forwarding_number = "+1-555-123-4567"
     return {"status": "success", "message": f"Setup done! Calls forwarded to {forwarding_number}"}
 
@@ -278,9 +227,6 @@ def setup_call_forwarding():
 # STT and TTS Functions for Voice Calls
 # ---------------------------
 def transcribe_audio(audio_file: bytes) -> str:
-    """
-    Convert incoming audio to text using Groq Whisper v3 (STT).
-    """
     response = groq_client.audio.transcriptions.create(
         file=("audio.m4a", audio_file),
         model="whisper-large-v3",
@@ -289,9 +235,6 @@ def transcribe_audio(audio_file: bytes) -> str:
     return response.text
 
 def text_to_speech(text: str) -> bytes:
-    """
-    Convert text to speech using Cartesia TTS.
-    """
     audio_bytes = cartesia_client.tts.bytes(
         model_id="sonic",
         transcript=text,
@@ -306,14 +249,9 @@ def text_to_speech(text: str) -> bytes:
 
 @app.post("/process-call")
 async def process_call(caller_number: str = Form(...), audio: UploadFile = File(...)):
-    """
-    Process an incoming voice call:
-      - If the caller's number is in saved contacts, immediately return a "Ringing" message.
-      - Otherwise (unsaved caller), always treat the call as spam and simulate a humorous conversation.
-    """
     if contacts_collection.find_one({"phone": caller_number}):
         ringing_text = f"Call from {caller_number} – Ringing"
-        _ = text_to_speech(ringing_text)  # Generate audio if needed
+        _ = text_to_speech(ringing_text)
         return {"status": "success", "message": ringing_text}
     
     try:
@@ -328,7 +266,6 @@ async def process_call(caller_number: str = Form(...), audio: UploadFile = File(
     
     update_chat_history(caller_number, "stt", transcription)
     
-    # For voice calls, unsaved callers are always treated as spam.
     conversation_result = simulate_call_conversation(caller_number, transcription, conversation_type="spam")
     
     try:
@@ -340,13 +277,6 @@ async def process_call(caller_number: str = Form(...), audio: UploadFile = File(
 
 @app.get("/audio-reply/{caller_number}")
 def get_audio_reply(caller_number: str):
-    """
-    Retrieve the latest assistant reply and STT transcription for a given caller.
-    Returns:
-      - stt_response: the transcription (STT)
-      - llm_reply: the assistant reply (LLM)
-      - audio_reply: the TTS audio (WAV) as a base64-encoded string.
-    """
     messages = get_chat_history(caller_number)
     if not messages:
         raise HTTPException(status_code=404, detail="No conversation found for this caller")
@@ -369,6 +299,43 @@ def get_audio_reply(caller_number: str):
         "audio_reply": audio_base64
     }
 
+# ---------------------------
+# New Endpoints for Direct STT, TTS, and LLM Calls
+# ---------------------------
+@app.post("/stt")
+async def stt_endpoint(audio: UploadFile = File(...)):
+    try:
+        audio_bytes = await audio.read()
+        transcription = transcribe_audio(audio_bytes)
+        return {"transcription": transcription}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"STT Error: {str(e)}")
+
+@app.post("/tts")
+def tts_endpoint(text: str = Body(..., embed=True)):
+    try:
+        audio_bytes = text_to_speech(text)
+        audio_base64 = base64.b64encode(audio_bytes).decode("utf-8")
+        return {"audio": audio_base64}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"TTS Error: {str(e)}")
+
+@app.post("/llm")
+def llm_endpoint(message: str = Body(..., embed=True)):
+    try:
+        messages = [
+            {"role": "system", "content": "You are a helpful assistant."},
+            {"role": "user", "content": message}
+        ]
+        response = llm.invoke(messages)
+        if hasattr(response, "content"):
+            reply = response.content
+        else:
+            reply = str(response)
+        return {"reply": reply}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"LLM Error: {str(e)}")
+
 @app.get("/")
 async def root():
     return {"message": "Welcome to the AI Spam Blocker API."}