193 lines
6.8 KiB
Python
193 lines
6.8 KiB
Python
import re
|
|
from typing import List, Tuple
|
|
from app.models.schemas import TaskType, PriorityLevel, RequestAnalysis
|
|
|
|
# Regex rules and indicator vocabularies
|
|
DEBUG_PATTERNS = [
|
|
r"\b(debug|bug|stacktrace|exception|error|fix this error|traceback|segfault|syntaxerror|nullpointer)\b",
|
|
r"\b(crash|panicked|failed test|reproduce|root cause|undefined is not a function)\b",
|
|
]
|
|
|
|
CODING_PATTERNS = [
|
|
r"\b(write a python|write code|implement|function|class |def |const |async def|typescript|javascript|golang|rust|react|sql query|endpoint|api client)\b",
|
|
r"\b(refactor|algorithm|regex|unit test|dockerfile|kubernetes yaml|prisma|fastapi)\b",
|
|
r"```[a-zA-Z0-9_-]*\n",
|
|
]
|
|
|
|
REASONING_PATTERNS = [
|
|
r"\b(explain why|reasoning|step by step|proof|derive|why did|compare and contrast|architectural decision|tradeoffs|trade-offs|consequences of)\b",
|
|
r"\b(game theory|logical fallacy|syllogism|first principles|hypothesize)\b",
|
|
]
|
|
|
|
MATH_PATTERNS = [
|
|
r"\b(calculate|integral|derivative|matrix|eigenvalue|probability|equation|differential equation|solve for x|combinatorics|standard deviation)\b",
|
|
r"[\$\\\^\{\}\[\]\=]{3,}",
|
|
]
|
|
|
|
SUMMARIZATION_PATTERNS = [
|
|
r"\b(summarize|summary|tldr|tl;dr|key points|bullet points of this|condense|digest|brief overview)\b",
|
|
]
|
|
|
|
EXTRACTION_PATTERNS = [
|
|
r"\b(extract|extract json|parse this table|pull all emails|convert to json|output as csv|structured data from)\b",
|
|
]
|
|
|
|
TRANSLATION_PATTERNS = [
|
|
r"\b(translate to|translate into|in french|in spanish|in german|in japanese|in chinese|in hindi|translate from)\b",
|
|
]
|
|
|
|
CREATIVE_PATTERNS = [
|
|
r"\b(write a poem|write a story|screenplay|brainstorm catchy|creative writing|dialogue between|fiction|metaphor)\b",
|
|
]
|
|
|
|
ANALYSIS_PATTERNS = [
|
|
r"\b(analyze|breakdown|evaluate|critique|audit|financial report|trend analysis|pros and cons)\b",
|
|
]
|
|
|
|
|
|
def estimate_tokens(text: str) -> int:
|
|
"""Fast, reliable token estimation (roughly 4 characters per token)."""
|
|
return max(1, len(text) // 4)
|
|
|
|
|
|
def analyze_request_heuristics(prompt: str) -> RequestAnalysis:
|
|
text_lower = prompt.lower()
|
|
tokens = estimate_tokens(prompt)
|
|
|
|
detected_keywords: List[str] = []
|
|
task_scores: dict[TaskType, float] = {t: 0.0 for t in TaskType}
|
|
|
|
# 1. Evaluate patterns
|
|
for p in DEBUG_PATTERNS:
|
|
matches = re.findall(p, text_lower)
|
|
if matches:
|
|
task_scores[TaskType.DEBUGGING] += len(matches) * 2.5
|
|
detected_keywords.extend(matches)
|
|
|
|
for p in CODING_PATTERNS:
|
|
matches = re.findall(p, text_lower)
|
|
if matches:
|
|
task_scores[TaskType.CODING] += len(matches) * 2.0
|
|
detected_keywords.extend(matches)
|
|
|
|
for p in REASONING_PATTERNS:
|
|
matches = re.findall(p, text_lower)
|
|
if matches:
|
|
task_scores[TaskType.REASONING] += len(matches) * 2.0
|
|
detected_keywords.extend(matches)
|
|
|
|
for p in MATH_PATTERNS:
|
|
matches = re.findall(p, text_lower)
|
|
if matches:
|
|
task_scores[TaskType.MATH] += len(matches) * 2.2
|
|
detected_keywords.extend(matches)
|
|
|
|
for p in SUMMARIZATION_PATTERNS:
|
|
matches = re.findall(p, text_lower)
|
|
if matches:
|
|
task_scores[TaskType.SUMMARIZATION] += len(matches) * 2.5
|
|
detected_keywords.extend(matches)
|
|
|
|
for p in EXTRACTION_PATTERNS:
|
|
matches = re.findall(p, text_lower)
|
|
if matches:
|
|
task_scores[TaskType.EXTRACTION] += len(matches) * 2.0
|
|
detected_keywords.extend(matches)
|
|
|
|
for p in TRANSLATION_PATTERNS:
|
|
matches = re.findall(p, text_lower)
|
|
if matches:
|
|
task_scores[TaskType.TRANSLATION] += len(matches) * 3.0
|
|
detected_keywords.extend(matches)
|
|
|
|
for p in CREATIVE_PATTERNS:
|
|
matches = re.findall(p, text_lower)
|
|
if matches:
|
|
task_scores[TaskType.CREATIVE] += len(matches) * 2.0
|
|
detected_keywords.extend(matches)
|
|
|
|
for p in ANALYSIS_PATTERNS:
|
|
matches = re.findall(p, text_lower)
|
|
if matches:
|
|
task_scores[TaskType.ANALYSIS] += len(matches) * 1.8
|
|
detected_keywords.extend(matches)
|
|
|
|
# Long context check
|
|
if tokens > 4000:
|
|
task_scores[TaskType.LONG_CONTEXT] += 3.0
|
|
|
|
# 2. Determine Primary Task
|
|
best_task = TaskType.GENERAL_QA
|
|
max_score = 0.0
|
|
for task, score in task_scores.items():
|
|
if score > max_score:
|
|
max_score = score
|
|
best_task = task
|
|
|
|
# 3. Derive Complexity (0.0 to 1.0)
|
|
complexity = 0.35 # Base complexity
|
|
|
|
# Token length factor
|
|
if tokens > 10000:
|
|
complexity += 0.30
|
|
elif tokens > 3000:
|
|
complexity += 0.20
|
|
elif tokens > 800:
|
|
complexity += 0.10
|
|
elif tokens < 30:
|
|
complexity -= 0.10
|
|
|
|
# Task inherent complexity
|
|
if best_task in (TaskType.DEBUGGING, TaskType.MATH, TaskType.REASONING):
|
|
complexity += 0.30
|
|
elif best_task in (TaskType.CODING, TaskType.ANALYSIS, TaskType.LONG_CONTEXT):
|
|
complexity += 0.20
|
|
elif best_task in (TaskType.TRANSLATION, TaskType.EXTRACTION, TaskType.SUMMARIZATION):
|
|
complexity += 0.05
|
|
elif best_task == TaskType.GENERAL_QA and tokens < 50:
|
|
complexity -= 0.15
|
|
|
|
# Multi-step keyword boosts
|
|
if any(k in text_lower for k in ["distributed", "async", "concurrency", "deadlock", "microservices", "architecture"]):
|
|
complexity += 0.25
|
|
if any(k in text_lower for k in ["simple", "quick", "one liner", "easy", "hello world"]):
|
|
complexity -= 0.20
|
|
|
|
complexity = max(0.05, min(0.99, round(complexity, 2)))
|
|
|
|
# 4. Priority and Capabilities
|
|
coding_required = best_task in (TaskType.CODING, TaskType.DEBUGGING) or task_scores[TaskType.CODING] > 0
|
|
reasoning_required = best_task in (TaskType.REASONING, TaskType.DEBUGGING, TaskType.MATH) or complexity >= 0.70
|
|
|
|
if complexity >= 0.75:
|
|
complexity_label = PriorityLevel.HIGH
|
|
quality_req = PriorityLevel.HIGH
|
|
latency_pri = PriorityLevel.LOW
|
|
cost_sens = PriorityLevel.LOW
|
|
elif complexity >= 0.45:
|
|
complexity_label = PriorityLevel.MEDIUM
|
|
quality_req = PriorityLevel.MEDIUM
|
|
latency_pri = PriorityLevel.MEDIUM
|
|
cost_sens = PriorityLevel.MEDIUM
|
|
else:
|
|
complexity_label = PriorityLevel.LOW
|
|
quality_req = PriorityLevel.LOW
|
|
latency_pri = PriorityLevel.HIGH
|
|
cost_sens = PriorityLevel.HIGH
|
|
|
|
return RequestAnalysis(
|
|
task_type=best_task,
|
|
complexity=complexity,
|
|
complexity_label=complexity_label,
|
|
reasoning_required=reasoning_required,
|
|
coding_required=coding_required,
|
|
vision_required=False,
|
|
tools_required=False,
|
|
context_size=tokens,
|
|
latency_priority=latency_pri,
|
|
cost_sensitivity=cost_sens,
|
|
quality_requirement=quality_req,
|
|
keywords_detected=list(set(detected_keywords))[:10],
|
|
analyzer_used="rules",
|
|
)
|