Enhance roleplay API to support audio messages and add comprehensive test cases

requirements.txt

@@ -11,4 +11,5 @@ langchain
 langgraph-swarm
 langchain-google-genai
 python-dotenv
-loguru
+loguru
+python-multipart

sessions.json

@@ -3,8 +3,8 @@
     "id": "82a6779d-ad13-4edd-a046-575e563a4348",
     "name": "New Conversation",
     "created_at": "2025-08-21T11:57:23.992279",
-    "last_message": "I would like a coffee",
-    "message_count": 7
+    "last_message": "[Audio message]",
+    "message_count": 37
   },
   {
     "id": "4fbf6c50-6054-4f3d-ac4e-d8281c306d72",

src/apis/routes/chat_route.py

@@ -1,4 +1,13 @@
-from fastapi import APIRouter, status, Depends, BackgroundTasks, HTTPException
+from fastapi import (
+    APIRouter,
+    status,
+    Depends,
+    BackgroundTasks,
+    HTTPException,
+    File,
+    UploadFile,
+    Form,
+)
 from fastapi.responses import JSONResponse
 from src.utils.logger import logger
 from src.agents.role_play.func import create_agents
@@ -9,6 +18,7 @@ import json
 import os
 import uuid
 from datetime import datetime
+import base64
 
 router = APIRouter(prefix="/ai", tags=["AI"])
 
@@ -113,22 +123,103 @@ async def list_scenarios():
 
 
 @router.post("/roleplay", status_code=status.HTTP_200_OK)
-async def roleplay(request: RoleplayRequest):
-    """Send a message to the roleplay agent"""
-    scenario = request.scenario
-    if not scenario:
+async def roleplay(
+    session_id: str = Form(
+        ..., description="Session ID for tracking user interactions"
+    ),
+    scenario: str = Form(
+        ..., description="The scenario for the roleplay as JSON string"
+    ),
+    text_message: Optional[str] = Form(None, description="Text message from user"),
+    audio_file: Optional[UploadFile] = File(None, description="Audio file from user"),
+):
+    """Send a message (text or audio) to the roleplay agent"""
+
+    # Validate that at least one input is provided
+    if not text_message and not audio_file:
+        raise HTTPException(
+            status_code=400, detail="Either text_message or audio_file must be provided"
+        )
+
+    # Parse scenario from JSON string
+    try:
+        scenario_dict = json.loads(scenario)
+    except json.JSONDecodeError:
+        raise HTTPException(status_code=400, detail="Invalid scenario JSON format")
+
+    if not scenario_dict:
         raise HTTPException(status_code=400, detail="Scenario not provided")
-    response = await create_agents(scenario).ainvoke(
-        {
-            "messages": [request.query],
-        },
-        {"configurable": {"thread_id": request.session_id}},
-    )
 
-    # Update session with last message
-    update_session_last_message(request.session_id, request.query)
+    # Prepare message content
+    message_content = []
+
+    # Handle text input
+    if text_message:
+        message_content.append({"type": "text", "text": text_message})
+
+    # Handle audio input
+    if audio_file:
+        try:
+            # Read audio file content
+            audio_data = await audio_file.read()
 
-    return JSONResponse(content=response["messages"][-1].content)
+            # Convert to base64
+            audio_base64 = base64.b64encode(audio_data).decode("utf-8")
+
+            # Determine mime type based on file extension
+            file_extension = (
+                audio_file.filename.split(".")[-1].lower()
+                if audio_file.filename
+                else "wav"
+            )
+            mime_type_map = {
+                "wav": "audio/wav",
+                "mp3": "audio/mpeg",
+                "ogg": "audio/ogg",
+                "webm": "audio/webm",
+                "m4a": "audio/mp4",
+            }
+            mime_type = mime_type_map.get(file_extension, "audio/wav")
+
+            message_content.append(
+                {
+                    "type": "audio",
+                    "source_type": "base64",
+                    "data": audio_base64,
+                    "mime_type": mime_type,
+                }
+            )
+
+        except Exception as e:
+            logger.error(f"Error processing audio file: {str(e)}")
+            raise HTTPException(
+                status_code=400, detail=f"Error processing audio file: {str(e)}"
+            )
+
+    # Create message in the required format
+    message = {"role": "user", "content": message_content}
+
+    try:
+        response = await create_agents(scenario_dict).ainvoke(
+            {
+                "messages": [message],
+            },
+            {"configurable": {"thread_id": session_id}},
+        )
+
+        # Update session with last message (use text if available, otherwise indicate audio)
+        last_message = text_message if text_message else "[Audio message]"
+        update_session_last_message(session_id, last_message)
+        
+        # Extract AI response content
+        ai_response = response["messages"][-1].content
+        logger.info(f"AI response: {ai_response}")
+        
+        return JSONResponse(content={"response": ai_response})
+
+    except Exception as e:
+        logger.error(f"Error in roleplay: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
 
 
 @router.post("/get-messages", status_code=status.HTTP_200_OK)

test_audio_api.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python3
+"""
+Test script for audio roleplay API
+"""
+
+import requests
+import json
+import base64
+
+# API configuration
+API_BASE_URL = "http://localhost:8000"
+# API_BASE_URL = "https://abao77-run-code-api.hf.space"
+
+def test_text_message():
+    """Test sending text message only"""
+    print("Testing text message...")
+    
+    scenario = {
+        "scenario_title": "Restaurant Order",
+        "scenario_description": "Order food at a restaurant",
+        "scenario_context": "You are at a restaurant and want to order food",
+        "your_role": "Customer",
+        "key_vocabulary": ["menu", "order", "bill", "table"]
+    }
+    
+    data = {
+        "session_id": "test-session-123",
+        "scenario": json.dumps(scenario),
+        "text_message": "Hello, I'd like to see the menu please."
+    }
+    
+    try:
+        response = requests.post(f"{API_BASE_URL}/api/ai/roleplay", data=data)
+        if response.ok:
+            result = response.json()
+            print("✅ Text message test successful!")
+            print(f"Response: {result.get('response', 'No response')}")
+        else:
+            print(f"❌ Text message test failed: {response.status_code}")
+            print(f"Error: {response.text}")
+    except Exception as e:
+        print(f"❌ Text message test error: {e}")
+
+def test_audio_message():
+    """Test sending audio file"""
+    print("\nTesting audio message...")
+    
+    scenario = {
+        "scenario_title": "Restaurant Order",
+        "scenario_description": "Order food at a restaurant", 
+        "scenario_context": "You are at a restaurant and want to order food",
+        "your_role": "Customer",
+        "key_vocabulary": ["menu", "order", "bill", "table"]
+    }
+    
+    # Create a dummy audio file (in real scenario, this would be actual audio)
+    dummy_audio_data = b"fake_audio_data_for_testing"
+    
+    data = {
+        "session_id": "test-session-456",
+        "scenario": json.dumps(scenario)
+    }
+    
+    files = {
+        "audio_file": ("test_audio.wav", dummy_audio_data, "audio/wav")
+    }
+    
+    try:
+        response = requests.post(f"{API_BASE_URL}/api/ai/roleplay", data=data, files=files)
+        if response.ok:
+            result = response.json()
+            print("✅ Audio message test successful!")
+            print(f"Response: {result.get('response', 'No response')}")
+        else:
+            print(f"❌ Audio message test failed: {response.status_code}")
+            print(f"Error: {response.text}")
+    except Exception as e:
+        print(f"❌ Audio message test error: {e}")
+
+def test_combined_message():
+    """Test sending both text and audio"""
+    print("\nTesting combined text + audio message...")
+    
+    scenario = {
+        "scenario_title": "Restaurant Order",
+        "scenario_description": "Order food at a restaurant",
+        "scenario_context": "You are at a restaurant and want to order food", 
+        "your_role": "Customer",
+        "key_vocabulary": ["menu", "order", "bill", "table"]
+    }
+    
+    dummy_audio_data = b"fake_audio_data_for_testing"
+    
+    data = {
+        "session_id": "test-session-789",
+        "scenario": json.dumps(scenario),
+        "text_message": "I have a question about the menu"
+    }
+    
+    files = {
+        "audio_file": ("question.wav", dummy_audio_data, "audio/wav")
+    }
+    
+    try:
+        response = requests.post(f"{API_BASE_URL}/api/ai/roleplay", data=data, files=files)
+        if response.ok:
+            result = response.json()
+            print("✅ Combined message test successful!")
+            print(f"Response: {result.get('response', 'No response')}")
+        else:
+            print(f"❌ Combined message test failed: {response.status_code}")
+            print(f"Error: {response.text}")
+    except Exception as e:
+        print(f"❌ Combined message test error: {e}")
+
+if __name__ == "__main__":
+    print("🧪 Testing Audio Roleplay API")
+    print("=" * 50)
+    
+    test_text_message()
+    test_audio_message() 
+    test_combined_message()
+    
+    print("\n" + "=" * 50)
+    print("🏁 Testing completed!")