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 services.ollama_service import OllamaService from services.skill_analyzer import SkillAnalyzer 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 AI services first groq_service = GroqService() ollama_service = OllamaService() # Initialize other services with AI service dependency injection scenario_generator = ScenarioGenerator(groq_service, ollama_service) roleplay_engine = RoleplayEngine(groq_service, ollama_service) skill_analyzer = SkillAnalyzer(groq_service, ollama_service) # 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] difficulty_level: str = "Easy" # Default value isAdmin: int # Default to admin groqRoleplay: int # Default value class TokenCount(BaseModel): input: int output: int total: int class TokenCounts(BaseModel): preview: TokenCount conversation: TokenCount assessment: TokenCount service_used: str # "groq" or "ollama" class RoleplayRequest(BaseModel): client_id: str session_id: str roleplay_data: RoleplayData query: str class RoleplayResponse(BaseModel): session_id: str response: str token_counts: TokenCounts 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 scenario_intro: str conversation_starter: str success_criteria: Dict[str, str] difficulty_level: str background_info: str class ScenarioPreviewResponse(BaseModel): slides: List[Dict[str, str]] token_counts: TokenCounts class EndSessionResponse(BaseModel): message: str session_id: str assessment: Optional[Dict[str, Any]] = None token_counts: TokenCounts class EndSessionRequest(BaseModel): session_id: str roleplay_data: RoleplayData @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: # Check both Groq and Ollama services health_status = {"timestamp": datetime.now().isoformat()} # Check Groq api_key = os.getenv("GROQ_API_KEY") if not api_key: health_status["groq"] = {"status": "error", "message": "GROQ_API_KEY not configured"} else: health_status["groq"] = {"status": "healthy"} # Check Ollama ollama_available = ollama_service.health_check() health_status["ollama"] = {"status": "healthy" if ollama_available else "unavailable"} # Overall status overall_healthy = (api_key is not None) or ollama_available health_status["overall"] = "healthy" if overall_healthy else "error" return health_status 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 use_groq = request.roleplay_data.groqRoleplay == 1 print(f"DEBUG: Received session_id from PHP: '{session_id}' (type: {type(session_id)})") print(f"DEBUG: Received client_id: '{request.client_id}'") print(f"DEBUG: request.roleplay_data.isAdmin: {request.roleplay_data.isAdmin}") print(f"DEBUG: Using {'Groq' if use_groq else 'Ollama'} for AI service (groqRoleplay={request.roleplay_data.groqRoleplay})") print(f"Processing request for session: {session_id}") # Check if scenario already exists for this session scenario = load_scenario_for_session(session_id, request.roleplay_data.isAdmin) 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, use_groq) if not scenario: raise HTTPException(status_code=500, detail="Failed to create scenario") print("Starting new session...") # Start session with scenario introduction (force restart to replace old conversation) roleplay_engine.start_session(scenario, request.roleplay_data.isAdmin, force_restart=True) # Return the scenario introduction as the response token_counts = get_token_counts_response(use_groq) return RoleplayResponse( session_id=session_id, response=scenario.scenario_intro, token_counts=token_counts ) 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, request.roleplay_data.isAdmin) 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, request.roleplay_data.isAdmin, force_restart=True) # Return the scenario introduction token_counts = get_token_counts_response(use_groq) return RoleplayResponse( session_id=session_id, response=scenario.scenario_intro, token_counts=token_counts ) # Add learner response and get AI reply try: ai_response = roleplay_engine.add_learner_response( session_id, scenario, request.query, request.roleplay_data.isAdmin, use_groq ) 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") token_counts = get_token_counts_response(use_groq) return RoleplayResponse( session_id=session_id, response=ai_response, token_counts=token_counts ) 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, use_groq: bool = True) -> Optional[RoleplayScenario]: """Create scenario from the roleplay request data""" try: print(f"DEBUG create_scenario_from_request: request.session_id='{request.session_id}'") 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": roleplay_data.difficulty_level } # 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( session_id=request.session_id, 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, is_admin=roleplay_data.isAdmin, use_groq=use_groq ) 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, is_admin: int) -> 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, is_admin) 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, scenario_intro=scenario.scenario_intro, 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_slides(scenario: RoleplayScenario) -> List[Dict[str, str]]: """Convert scenario to slide-based array format""" try: slides = [] # Slide 1: Context context = scenario.scenario_setup.get('context', '').strip() if context: slides.append({ "heading": "Context", "content": context }) # Slide 2: Constraints/Policies (only if exists) constraints = scenario.scenario_setup.get('constraints', '').strip() if constraints and constraints.lower() not in ['', 'not specified', 'none']: slides.append({ "heading": "Constraints/Policies", "content": constraints }) return slides except Exception as e: print(f"Error formatting scenario as slides: {e}") # Fallback to minimal slides return [ { "heading": "Context", "content": f"This is a {scenario.category} roleplay scenario where you will practice your skills." } ] def get_token_counts_response(use_groq: bool) -> TokenCounts: """Get token counts from the appropriate AI service""" # Use the global service instances that have been accumulating token counts ai_service = groq_service if use_groq else ollama_service token_data = ai_service.get_token_counts() return TokenCounts( preview=TokenCount( input=token_data['preview']['input'], output=token_data['preview']['output'], total=token_data['preview']['total'] ), conversation=TokenCount( input=token_data['conversation']['input'], output=token_data['conversation']['output'], total=token_data['conversation']['total'] ), assessment=TokenCount( input=token_data['assessment']['input'], output=token_data['assessment']['output'], total=token_data['assessment']['total'] ), service_used="groq" if use_groq else "ollama" ) @app.post("/roleplay_scenario", response_model=ScenarioPreviewResponse) async def get_roleplay_scenario_preview(request: EndSessionRequest): """Generate and return formatted preview data as slide array""" try: use_groq = request.roleplay_data.groqRoleplay == 1 print(f"Processing preview request for session: {request.session_id}") print(f"DEBUG: Using {'Groq' if use_groq else 'Ollama'} for preview generation") # First check if scenario already exists for this session existing_scenario = load_scenario_for_session(request.session_id, request.roleplay_data.isAdmin) if existing_scenario: print("Using existing scenario for preview") slides_response = format_scenario_as_slides(existing_scenario) token_counts = get_token_counts_response(use_groq) return ScenarioPreviewResponse(slides=slides_response, token_counts=token_counts) # Create new scenario from request data print("Creating new scenario for preview") scenario = create_scenario_from_request(request, use_groq) if not scenario: raise HTTPException(status_code=500, detail="Failed to create scenario for preview") # Format and return slide preview data slides_response = format_scenario_as_slides(scenario) token_counts = get_token_counts_response(use_groq) print("Preview slides generated successfully") return ScenarioPreviewResponse(slides=slides_response, token_counts=token_counts) 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", response_model=EndSessionResponse) async def end_session(request: RoleplayRequest): """End a roleplay session and automatically generate skill assessment""" try: use_groq = request.roleplay_data.groqRoleplay == 1 print(f"DEBUG end_session: session_id='{request.session_id}', roleplay_data.isAdmin={request.roleplay_data.isAdmin}") print(f"DEBUG: Using {'Groq' if use_groq else 'Ollama'} for assessment generation") # Load scenario for the session scenario = load_scenario_for_session(request.session_id, request.roleplay_data.isAdmin) if not scenario: raise HTTPException(status_code=404, detail="Session/Scenario not found") # Get conversation turns for assessment conversation_turns = roleplay_engine.get_conversation_turns_for_assessment(request.session_id, request.roleplay_data.isAdmin) if not conversation_turns: raise HTTPException(status_code=404, detail="No conversation history found") # End the session success = roleplay_engine.end_session(request.session_id, request.roleplay_data.isAdmin) if not success: raise HTTPException(status_code=500, detail="Failed to end session") # Generate skill assessment print(f"DEBUG: About to generate assessment for session {request.session_id}") assessment = skill_analyzer.analyze_session(request.session_id, scenario, conversation_turns, request.roleplay_data.isAdmin, use_groq) print(f"DEBUG: Assessment generated: {assessment is not None}") # Get final token counts BEFORE resetting (includes all operations: preview, conversation, assessment) final_token_counts = get_token_counts_response(use_groq) print(f"DEBUG: Final token counts before reset: {final_token_counts}") # Reset token counts to zero after getting the final counts ai_service = groq_service if use_groq else ollama_service ai_service.reset_token_counts() print(f"DEBUG: Token counts reset to zero after end_session") # Delete conversation after successful assessment if assessment: print(f"DEBUG: Deleting conversation after successful assessment") conversation_deleted = roleplay_engine.json_handler.delete_conversation(request.session_id, request.roleplay_data.isAdmin) print(f"DEBUG: Conversation deleted: {conversation_deleted}") response_data = { "message": "Session ended successfully", "session_id": request.session_id, "token_counts": final_token_counts } # Include assessment in response if generated successfully if assessment: response_data["assessment"] = assessment.to_dict() return EndSessionResponse(**response_data) except HTTPException: raise 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, is_admin: int = 0): """Get conversation history for a session""" try: history = roleplay_engine.get_conversation_history(session_id, is_admin) 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)