""" agents/detect_agent.py ----------------------- Auto-Detect Agent for AutoDevAgent. Makes a fast, cheap LLM call (Llama 3.1 8B) to classify the user's task into a supported language with a confidence score and one-sentence reason. The result pre-selects the language radio button in the UI with a coloured badge so the user can verify or override. Design: - Uses the fast 8B model — classification is simple, 70B is overkill. - Returns a DetectionResult Pydantic model — typed, validated, clean. - If detection fails or returns UNKNOWN, the UI defaults to Python (the most common language for generic tasks). - The LLM is prompted to return JSON only. The _parse_json helper from planning_agent is replicated here to keep the module self-contained and avoid circular imports. - Override is always available and prominent in the UI. Any override is logged to session history. Usage: from agents.detect_agent import DetectAgent from pipeline.state import PipelineState agent = DetectAgent() result = agent.detect("find duplicates in a list") print(result.language) # Language.PYTHON print(result.confidence) # "high" print(result.reason) # "task involves list manipulation in Python" """ import json import logging from langchain_groq import ChatGroq from langchain_core.messages import SystemMessage, HumanMessage from config import settings from pipeline.state import ( DetectionResult, Language, ) logger = logging.getLogger(__name__) # ------------------------------------------------------------------ # # Prompt # # ------------------------------------------------------------------ # DETECT_SYSTEM = """ You are a programming language classifier. Classify the given task into one of these languages: python, sql, unknown. Use "python" for: general coding tasks, algorithms, data manipulation, file processing, APIs, machine learning, scripting. Use "sql" for: database queries, SELECT/INSERT/UPDATE/DELETE statements, aggregations, JOINs, schema design. Use "unknown" for: tasks that are ambiguous or don't clearly fit either. Return ONLY a JSON object in this exact format — no markdown, no explanation: { "language": "python", "confidence": "high", "reason": "one sentence explaining the detection" } confidence must be one of: "high", "medium", "low" """.strip() CODING_TASK_SYSTEM = """ You are a gate classifier for a code assistant that only handles Python and SQL programming tasks. Decide whether the user's input is a programming/coding task or not. A CODING task includes: writing code, debugging, algorithms, SQL queries, data structures, functions, classes, scripts, APIs, database design, anything that requires generating code. A NON-CODING task includes: general knowledge questions, factual questions, opinions, math calculations without code, translations, creative writing, geography, history, or anything that does not require writing Python or SQL code. Return ONLY a JSON object — no markdown, no explanation: { "is_coding": true, "confidence": "high", "reason": "one sentence explaining the decision" } is_coding must be true or false. confidence must be one of: "high", "medium", "low". """.strip() SCOPE_CHECK_SYSTEM = """ You are a scope classifier for a code assistant that generates focused Python or SQL code snippets. The assistant can only produce a single self-contained file of at most ~100 lines. Decide whether the user's task is realistically achievable as a SINGLE code snippet. IN SCOPE (single snippet) examples: - Write a Python function to reverse a string - Sort a list of dictionaries by a key - SQL query to find top 5 customers by revenue - SQL query to find the second highest salary from an employees table - SQL query with date filters: customers active in last 30 days but not in previous 30 days - SQL query to calculate a running total using window functions or subqueries - SQL query to de-duplicate a table keeping the latest row per user - Fetch video titles from YouTube API using Python - Parse a CSV and calculate column averages - Implement binary search in Python - Write a regex to validate email addresses OUT OF SCOPE (full application / multi-file project) examples: - Build a YouTube app / clone - Build a social media platform - Create a full e-commerce website - Build a chat application - Make an Android/iOS app - Build a REST API with authentication, database, deployment - Create a machine learning pipeline with training, evaluation, and deployment Rules: - A task is OUT OF SCOPE only if it requires: a frontend UI framework, multiple files/modules, user authentication flows, deployment infrastructure, or weeks of engineering work. - A task is IN SCOPE if it can be done in a single Python function/class or a single SQL query — even if that query is complex (uses subqueries, CTEs, window functions, date ranges, JOINs). - IMPORTANT: ANY single SQL query — no matter how complex the logic — is always IN SCOPE. A SQL query never "requires a backend server" on its own; it is just a query. - Be STRICT — "build an app" is always out of scope, but "write a query / function" is always in scope. If OUT OF SCOPE, provide a short, friendly suggestion showing how to break the task into a concrete single-snippet version the assistant CAN help with. Return ONLY a JSON object — no markdown, no explanation: { "in_scope": true, "confidence": "high", "reason": "one sentence", "suggestion": "" } If in_scope is false, suggestion must be a non-empty string with a concrete reframed example. confidence must be one of: "high", "medium", "low". """.strip() # ------------------------------------------------------------------ # # Agent # # ------------------------------------------------------------------ # class DetectAgent: """ Classifies a task description into a programming language. Uses a single fast LLM call to return a DetectionResult with the detected language, confidence level, and a brief reason. Attributes: llm: ChatGroq using the fast 8B model. """ def __init__(self) -> None: """Initialise with the fast LLM from config.""" self.llm = ChatGroq( api_key = settings.groq_api_key, model = settings.groq_model_fast, temperature= 0.0, # Fully deterministic — classification task max_tokens = 150, # Short response expected ) def is_coding_task(self, task: str) -> tuple[bool, str]: """ Classify whether the task is a coding task at all. Returns: (is_coding: bool, reason: str) is_coding=False means the task is non-technical and should be rejected before the pipeline runs. """ logger.info("DetectAgent coding-gate check: '%s'", task[:60]) messages = [ SystemMessage(content=CODING_TASK_SYSTEM), HumanMessage(content=f"Task: {task}"), ] try: response = self.llm.invoke(messages) parsed = _parse_json(response.content.strip()) is_coding = bool(parsed.get("is_coding", True)) confidence = parsed.get("confidence", "low").lower().strip() reason = parsed.get("reason", "") logger.info("DetectAgent gate: is_coding=%s (%s) — %s", is_coding, confidence, reason) # Only block when the model is confident it's NOT coding if not is_coding and confidence in ("high", "medium"): return False, reason return True, reason except Exception as e: logger.warning("DetectAgent gate failed: %s — allowing task through", e) return True, "" def is_in_scope(self, task: str) -> tuple[bool, str]: """ Check whether the task is realistic for a single code snippet (~100 lines). Tasks like "build a YouTube app" or "create a social media platform" are full projects — out of scope for this assistant. We catch them here and return a friendly reframing suggestion instead of wasting pipeline retries. Returns: (in_scope: bool, suggestion: str) in_scope=False means the task is too large; suggestion holds a concrete example of how the user could break it down. """ logger.info("DetectAgent scope check: '%s'", task[:80]) messages = [ SystemMessage(content=SCOPE_CHECK_SYSTEM), HumanMessage(content=f"Task: {task}"), ] try: response = self.llm.invoke(messages) parsed = _parse_json(response.content.strip()) in_scope = bool(parsed.get("in_scope", True)) confidence = parsed.get("confidence", "low").lower().strip() reason = parsed.get("reason", "") suggestion = parsed.get("suggestion", "").strip() logger.info("DetectAgent scope: in_scope=%s (%s) — %s", in_scope, confidence, reason) # Block at any confidence level when out-of-scope. # The cost of letting a massive project through (wasted retries, # broken output) outweighs occasionally rejecting a borderline task. # Unlike the non-coding gate (low-confidence = probably fine), # a low-confidence out-of-scope still means "probably too big". if not in_scope: return False, suggestion return True, "" except Exception as e: logger.warning("DetectAgent scope check failed: %s — allowing task through", e) return True, "" def detect(self, task: str) -> DetectionResult: """ Detect the programming language for a given task description. Args: task: The user's task description string. Returns: DetectionResult with language, confidence, and reason. Falls back to Language.UNKNOWN on any failure. """ logger.info("DetectAgent classifying task: '%s'", task[:60]) messages = [ SystemMessage(content=DETECT_SYSTEM), HumanMessage(content=f"Task: {task}"), ] try: response = self.llm.invoke(messages) raw = response.content.strip() logger.debug("DetectAgent raw response: %s", raw) parsed = _parse_json(raw) # Validate and normalise language value lang_str = parsed.get("language", "unknown").lower().strip() try: language = Language(lang_str) except ValueError: language = Language.UNKNOWN confidence = parsed.get("confidence", "low").lower().strip() if confidence not in ("high", "medium", "low"): confidence = "low" reason = parsed.get("reason", "Could not determine reason.") result = DetectionResult( language = language, confidence = confidence, reason = reason, ) logger.info( "DetectAgent: %s (%s confidence) — %s", result.language.value, result.confidence, result.reason, ) return result except Exception as e: logger.warning("DetectAgent failed: %s — defaulting to UNKNOWN", e) return DetectionResult( language = Language.UNKNOWN, confidence = "low", reason = "Auto-detection failed — please select language manually.", ) # ------------------------------------------------------------------ # # Helpers # # ------------------------------------------------------------------ # def _parse_json(raw: str) -> dict: """ Safely parse JSON from LLM response, stripping markdown fences. Args: raw: Raw string content from the LLM. Returns: Parsed dict. Raises: ValueError: If JSON cannot be parsed after cleaning. """ cleaned = raw.strip() if cleaned.startswith("```"): cleaned = cleaned.split("\n", 1)[-1] if cleaned.endswith("```"): cleaned = cleaned.rsplit("```", 1)[0] cleaned = cleaned.strip() try: return json.loads(cleaned) except json.JSONDecodeError as e: raise ValueError(f"DetectAgent returned invalid JSON: {e}") from e