from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from typing import List, Dict, Any, Optional from datetime import datetime import uuid import os from dotenv import load_dotenv # Load environment variables load_dotenv() # Import your existing services and models from services.scenario_generator import ScenarioGenerator from services.roleplay_engine import RoleplayEngine from services.groq_service import GroqService from models.scenario import RoleplayScenario app = FastAPI( title="AI Roleplay Service", description="Simple AI service for roleplay functionality", version="1.0.0" ) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], # Configure this for production allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Initialize services scenario_generator = ScenarioGenerator() roleplay_engine = RoleplayEngine() groq_service = GroqService() # Pydantic models for request/response class SkillData(BaseModel): skill_id: str skill_name: str class RoleplayData(BaseModel): category: str objective: str learner_role: str = Field(..., alias="Learner role") additional_info: str company_policies: str = Field("", alias="Constraints/Policies") skills_for_roleplay: List[SkillData] class RoleplayRequest(BaseModel): client_id: str session_id: str roleplay_data: RoleplayData query: str class RoleplayResponse(BaseModel): session_id: str response: str class CharacterDetails(BaseModel): name: str personality: str goals: str background: str emotional_state: str class ScenarioSetup(BaseModel): context: str environment: str constraints: str class PreviewResponse(BaseModel): scenario_id: str category: str objective: str learner_role: str ai_role: str skills_to_assess: List[str] scenario_setup: ScenarioSetup character_details: CharacterDetails conversation_starter: str success_criteria: Dict[str, str] difficulty_level: str background_info: str class ScenarioPreviewResponse(BaseModel): response: str @app.get("/") async def root(): """Health check endpoint""" return {"message": "AI Roleplay Service", "status": "running"} @app.get("/health") async def health_check(): """Health check""" try: api_key = os.getenv("GROQ_API_KEY") if not api_key: return {"status": "error", "message": "GROQ_API_KEY not configured"} return {"status": "healthy", "timestamp": datetime.now().isoformat()} except Exception as e: raise HTTPException(status_code=500, detail=f"Health check failed: {str(e)}") @app.post("/roleplay", response_model=RoleplayResponse) async def handle_roleplay(request: RoleplayRequest): """Handle roleplay request - both first time and ongoing conversation""" try: session_id = request.session_id print(f"Processing request for session: {session_id}") # Check if scenario already exists for this session scenario = load_scenario_for_session(session_id) print(f"Scenario found: {scenario is not None}") if not scenario: # First request - create scenario and start session print("Creating new scenario...") scenario = create_scenario_from_request(request) if not scenario: raise HTTPException(status_code=500, detail="Failed to create scenario") print("Starting new session...") # Start session with opening AI message roleplay_engine.start_session(scenario) # Return the conversation starter as the response return RoleplayResponse( session_id=session_id, response=scenario.conversation_starter ) else: # Ongoing conversation - scenario exists, just add user response print(f"Adding learner response: {request.query}") # Check if conversation file exists conversation_history = roleplay_engine.get_conversation_history(session_id) print(f"Conversation history exists: {conversation_history is not None}") if conversation_history is None: # Scenario exists but no conversation started yet - start it print("Starting conversation for existing scenario...") roleplay_engine.start_session(scenario) # Return the conversation starter return RoleplayResponse( session_id=session_id, response=scenario.conversation_starter ) # Add learner response and get AI reply try: ai_response = roleplay_engine.add_learner_response( session_id, scenario, request.query ) print(f"AI response generated: {ai_response is not None}") if not ai_response: raise HTTPException(status_code=500, detail="Failed to generate AI response") return RoleplayResponse( session_id=session_id, response=ai_response ) except Exception as e: print(f"Error in add_learner_response: {str(e)}") raise except HTTPException: raise except Exception as e: print(f"Error in handle_roleplay: {str(e)}") print(f"Error type: {type(e)}") import traceback traceback.print_exc() raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}") except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Error processing roleplay: {str(e)}") def create_scenario_from_request(request: RoleplayRequest) -> Optional[RoleplayScenario]: """Create scenario from the roleplay request data""" try: roleplay_data = request.roleplay_data # Extract skill names for assessment skills_to_assess = [skill.skill_name for skill in roleplay_data.skills_for_roleplay] # Prepare details dictionary details = { "background": roleplay_data.additional_info, "constraints": roleplay_data.company_policies, "environment": "Roleplay conversation", "difficulty_level": "Medium" } # Determine AI role based on category and context ai_role = generate_ai_role(roleplay_data.category, roleplay_data.additional_info) # Create scenario using existing generator scenario = scenario_generator.create_scenario( category=roleplay_data.category, objective=roleplay_data.objective, details=details, ai_role=ai_role, learner_role=roleplay_data.learner_role, skills_to_assess=skills_to_assess ) # Override the scenario ID with session ID for consistency if scenario: scenario.id = request.session_id # Save with session ID as the scenario ID scenario_generator.json_handler.save_scenario(scenario.to_dict()) return scenario except Exception as e: print(f"Error creating scenario: {e}") return None def generate_ai_role(category: str, additional_info: str) -> str: """Generate appropriate AI role based on category and context""" category_roles = { "sales": "Potential Customer", "customer service": "Customer with Issue", "leadership": "Team Member", "negotiation": "Negotiation Partner", "technical support": "User with Technical Problem" } base_role = category_roles.get(category.lower(), "Conversation Partner") # Add context-specific details if available if "enterprise" in additional_info.lower(): return f"Enterprise {base_role}" elif "budget" in additional_info.lower() or "price" in additional_info.lower(): return f"Budget-Conscious {base_role}" elif "frustrated" in additional_info.lower() or "complaint" in additional_info.lower(): return f"Frustrated {base_role}" return base_role def load_scenario_for_session(session_id: str) -> Optional[RoleplayScenario]: """Load scenario for existing session""" try: # The scenario should be saved with session_id as the scenario_id return scenario_generator.load_scenario(session_id) except Exception as e: print(f"Error loading scenario for session {session_id}: {e}") return None def format_scenario_for_preview(scenario: RoleplayScenario) -> PreviewResponse: """Convert scenario to formatted preview response""" try: # Extract character details character_details = CharacterDetails( name=scenario.ai_character.get('name', 'AI Character'), personality=scenario.ai_character.get('personality', 'Not specified'), goals=scenario.ai_character.get('goals', 'Not specified'), background=scenario.ai_character.get('background', 'Not specified'), emotional_state=scenario.ai_character.get('emotional_state', 'Not specified') ) # Extract scenario setup scenario_setup = ScenarioSetup( context=scenario.scenario_setup.get('context', 'Not specified'), environment=scenario.scenario_setup.get('environment', 'Not specified'), constraints=scenario.scenario_setup.get('constraints', 'Not specified') ) # Format success criteria success_criteria = {} if isinstance(scenario.success_criteria, dict): success_criteria = scenario.success_criteria else: success_criteria = {"general": str(scenario.success_criteria)} # Create preview response return PreviewResponse( scenario_id=scenario.id, category=scenario.category, objective=scenario.objective, learner_role=scenario.learner_role, ai_role=scenario.ai_role, skills_to_assess=scenario.skills_to_assess, scenario_setup=scenario_setup, character_details=character_details, conversation_starter=scenario.conversation_starter, success_criteria=success_criteria, difficulty_level=scenario.details.get('difficulty_level', 'Medium'), background_info=scenario.details.get('background', 'Not specified') ) except Exception as e: print(f"Error formatting scenario for preview: {e}") raise def format_scenario_as_paragraph(scenario: RoleplayScenario) -> str: """Convert scenario to natural paragraph text using LLM""" try: # Prepare structured data for LLM scenario_data = { "category": scenario.category, "objective": scenario.objective, "learner_role": scenario.learner_role, "ai_role": scenario.ai_role, "skills_to_assess": scenario.skills_to_assess, "context": scenario.scenario_setup.get('context', 'Not specified'), "environment": scenario.scenario_setup.get('environment', 'Not specified'), "constraints": scenario.scenario_setup.get('constraints', 'Not specified'), "ai_character_name": scenario.ai_character.get('name', 'AI Character'), "ai_personality": scenario.ai_character.get('personality', 'Not specified'), "ai_background": scenario.ai_character.get('background', 'Not specified'), "ai_goals": scenario.ai_character.get('goals', 'Not specified'), "ai_emotional_state": scenario.ai_character.get('emotional_state', 'Not specified'), "conversation_starter": scenario.conversation_starter, "success_criteria": scenario.success_criteria, "background_info": scenario.details.get('background', 'Not specified'), "difficulty_level": scenario.details.get('difficulty_level', 'Medium') } # Create LLM prompt prompt = f"""Convert the following roleplay scenario data into natural, flowing paragraphs without any headings, bullet points, or markdown formatting. Write it as a cohesive narrative description that explains everything about the scenario in a conversational manner. Scenario Data: {scenario_data} Requirements: - Write in natural paragraphs only (no headings, no bullet points, no markdown) - Make it flow like a story or description - Include all the important information about the scenario, roles, characters, objectives, and success criteria - Keep it engaging and easy to understand - Don't use any formatting symbols like **, ##, -, etc. - Write it as if you're explaining the scenario to someone in person Write the complete scenario description in paragraph form:""" # Use Groq service to generate paragraph response paragraph_response = groq_service.get_completion(prompt) if not paragraph_response: # Fallback to simple format if LLM fails return f"This is a {scenario.category} roleplay scenario where you play the role of {scenario.learner_role} and interact with {scenario.ai_character.get('name', 'an AI character')} who is a {scenario.ai_role}. {scenario.objective} The scenario takes place in {scenario.scenario_setup.get('environment', 'a roleplay setting')}. {scenario.scenario_setup.get('context', '')} The conversation will begin with: {scenario.conversation_starter}" return paragraph_response.strip() except Exception as e: print(f"Error formatting scenario as paragraph with LLM: {e}") # Fallback to simple format return f"This is a {scenario.category} roleplay scenario where you play the role of {scenario.learner_role} and interact with {scenario.ai_character.get('name', 'an AI character')} who is a {scenario.ai_role}. {scenario.objective} The conversation will begin with: {scenario.conversation_starter}" @app.post("/roleplay_scenario", response_model=ScenarioPreviewResponse) async def get_roleplay_scenario_preview(request: RoleplayRequest): """Generate and return formatted preview data as paragraph text""" try: print(f"Processing preview request for session: {request.session_id}") # First check if scenario already exists for this session existing_scenario = load_scenario_for_session(request.session_id) if existing_scenario: print("Using existing scenario for preview") paragraph_response = format_scenario_as_paragraph(existing_scenario) return ScenarioPreviewResponse(response=paragraph_response) # Create new scenario from request data print("Creating new scenario for preview") scenario = create_scenario_from_request(request) if not scenario: raise HTTPException(status_code=500, detail="Failed to create scenario for preview") # Format and return paragraph preview data paragraph_response = format_scenario_as_paragraph(scenario) print("Preview paragraph generated successfully") return ScenarioPreviewResponse(response=paragraph_response) except HTTPException: raise except Exception as e: print(f"Error in get_roleplay_scenario_preview: {str(e)}") import traceback traceback.print_exc() raise HTTPException(status_code=500, detail=f"Error generating preview: {str(e)}") @app.post("/end_session") async def end_session(session_id: str): """End a roleplay session""" try: success = roleplay_engine.end_session(session_id) if success: return {"message": "Session ended successfully", "session_id": session_id} else: raise HTTPException(status_code=404, detail="Session not found") except Exception as e: raise HTTPException(status_code=500, detail=f"Error ending session: {str(e)}") @app.get("/session/{session_id}/history") async def get_session_history(session_id: str): """Get conversation history for a session""" try: history = roleplay_engine.get_conversation_history(session_id) if history is None: raise HTTPException(status_code=404, detail="Session not found") return {"session_id": session_id, "conversation_history": history} except Exception as e: raise HTTPException(status_code=500, detail=f"Error retrieving history: {str(e)}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)