from typing import Literal from pydantic import BaseModel from app.services.llm import LLMService class IntentResponse(BaseModel): category: Literal["SEARCH_REQUIRED", "CHAT_ONLY", "CODING_TASK", "DATA_ANALYSIS"] reasoning: str risk_level: Literal["LOW", "MEDIUM", "HIGH"] class IntentService: def __init__(self, llm_service: LLMService): self.llm = llm_service self.system_prompt = """ You are the 'Intent Analyzer' for an AI Operating System. Your job is to route the user's request to the correct module. Analyze the USER QUERY and return a JSON object. CATEGORIES: - SEARCH_REQUIRED: Query asks for current events, news, specific facts not in general knowledge, or research. (e.g., "Stock price of Apple", "Latest AI papers") - CHAT_ONLY: General greetings, philosophical questions, summaries of previous context, or logic puzzles. (e.g., "Hi", "Explain Stoicism") - CODING_TASK: Requests to write, debug, or explain code. - DATA_ANALYSIS: Requests involving CSVs, charts, or math aggregations. RISK LEVELS: - HIGH: Asking for dangerous/illegal content, PII, or financial advice. - MEDIUM: Ambiguous queries or potential controversies. - LOW: Safe, standard queries. Output format: {"category": "...", "reasoning": "...", "risk_level": "..."} """ async def analyze(self, query: str) -> IntentResponse: # For efficiency, we can use a faster/smaller model or just the main one with strictly low temp prompt = f"{self.system_prompt}\n\nUSER QUERY: {query}" # We'll rely on the main LLM to parse this for now. # ideally this uses a 'router' model (cheap/fast) response_text = await self.llm._generate( messages=[{"role": "user", "content": prompt}], temperature=0.0 ) # Simple parsing logic (robustness would require structured output mode or regex) # Assuming LLM behaves well with JSON instructions. import json import re try: # Clean markdown code blocks if present clean_text = re.sub(r"```json|```", "", response_text).strip() data = json.loads(clean_text) return IntentResponse(**data) except Exception as e: # Fallback to Safe Default print(f"Intent Parsing Failed: {e}. Defaulting to SEARCH.") return IntentResponse( category="SEARCH_REQUIRED", reasoning="Parsing error, defaulting to search.", risk_level="low" ) intent_service = None # initialized in main/deps