Spaces:
Running
Running
| """Query Complexity Classifier | |
| Determines whether a query needs full debate or can be answered directly. | |
| This prevents over-activation: simple factual questions get direct answers, | |
| while complex/ambiguous questions trigger full multi-agent reasoning. | |
| """ | |
| import re | |
| from enum import Enum | |
| class QueryComplexity(Enum): | |
| """Query complexity levels""" | |
| SIMPLE = "simple" # Direct factual answer, no debate needed | |
| MEDIUM = "medium" # Limited debate (2-3 agents) | |
| COMPLEX = "complex" # Full debate with all relevant agents | |
| class QueryClassifier: | |
| """Classify query complexity to determine reasoning depth.""" | |
| # Factual keywords (SIMPLE queries) | |
| FACTUAL_PATTERNS = [ | |
| r"what is .*\?", # "What is the speed of light?" | |
| r"define ", # "Define entropy" | |
| r"what (year|date|time) ", # "What year did..." | |
| r"how fast is", # "How fast is..." | |
| r"how high is", | |
| r"how long is", | |
| r"what (color|size|shape)", | |
| r"who is .*\?$", # "Who is Einstein?" | |
| r"where (is|are)", # "Where is the capital?" | |
| r"list of ", # "List of elements" | |
| r"formula for", # "Formula for..." | |
| r"calculate ", # "Calculate..." | |
| ] | |
| # Ambiguous keywords (COMPLEX queries) | |
| AMBIGUOUS_PATTERNS = [ | |
| r"could|might|may|possibly", # Uncertainty | |
| r"what does .* mean", # Interpretation | |
| r"why", # Explanation (often multi-faceted) | |
| r"how (do|does|should)", # Process/methodology | |
| r"discuss", | |
| r"compare", | |
| r"contrast", | |
| r"relationship between", | |
| r"difference between", | |
| ] | |
| # Ethics/Philosophy keywords (COMPLEX queries) | |
| ETHICS_PATTERNS = [ | |
| r"should (we |i )", | |
| r"is it (right|wrong|ethical|moral)", | |
| r"is it (good|bad|fair)", | |
| r"ought", | |
| r"morally?", | |
| r"ethics?", | |
| r"value of", | |
| r"meaning of", | |
| r"purpose of", | |
| r"implications of", | |
| ] | |
| # Multi-domain keywords (COMPLEX queries) | |
| MULTIDOMAIN_PATTERNS = [ | |
| r"connect .* to", | |
| r"relate .* to", | |
| r"how does .* affect", | |
| r"impact (of|on)", | |
| r"relationship .*between", | |
| r"interaction .*between", | |
| ] | |
| # Subjective/opinion keywords (COMPLEX queries) | |
| SUBJECTIVE_PATTERNS = [ | |
| r"think", | |
| r"opinion", | |
| r"perspective", | |
| r"view(point)?", | |
| r"argue(ment)?", | |
| r"debate", | |
| r"controversy", | |
| r"controversial", | |
| ] | |
| def classify(self, query: str) -> QueryComplexity: | |
| """Classify query complexity. | |
| Args: | |
| query: The user query | |
| Returns: | |
| QueryComplexity level (SIMPLE, MEDIUM, or COMPLEX) | |
| """ | |
| query_lower = query.lower().strip() | |
| # SIMPLE: Pure factual queries | |
| if self._is_factual(query_lower): | |
| # But check if it has complexity markers too | |
| if self._has_ambiguity(query_lower) or self._has_ethics(query_lower): | |
| return QueryComplexity.COMPLEX | |
| return QueryComplexity.SIMPLE | |
| # COMPLEX: Ethics, philosophy, interpretation, multi-domain | |
| if self._has_ethics(query_lower): | |
| return QueryComplexity.COMPLEX | |
| if self._has_ambiguity(query_lower): | |
| return QueryComplexity.COMPLEX | |
| if self._has_multidomain(query_lower): | |
| return QueryComplexity.COMPLEX | |
| if self._has_subjective(query_lower): | |
| return QueryComplexity.COMPLEX | |
| # MEDIUM: Everything else | |
| return QueryComplexity.MEDIUM | |
| def _is_factual(self, query: str) -> bool: | |
| """Check if query is direct factual question.""" | |
| return any(re.search(pattern, query) for pattern in self.FACTUAL_PATTERNS) | |
| def _has_ambiguity(self, query: str) -> bool: | |
| """Check if query has ambiguity markers.""" | |
| return any(re.search(pattern, query) for pattern in self.AMBIGUOUS_PATTERNS) | |
| def _has_ethics(self, query: str) -> bool: | |
| """Check if query involves ethics/philosophy.""" | |
| return any(re.search(pattern, query) for pattern in self.ETHICS_PATTERNS) | |
| def _has_multidomain(self, query: str) -> bool: | |
| """Check if query spans multiple domains.""" | |
| return any(re.search(pattern, query) for pattern in self.MULTIDOMAIN_PATTERNS) | |
| def _has_subjective(self, query: str) -> bool: | |
| """Check if query invites subjective reasoning.""" | |
| return any(re.search(pattern, query) for pattern in self.SUBJECTIVE_PATTERNS) | |
| def select_agents( | |
| self, complexity: QueryComplexity, domain: str | |
| ) -> dict[str, float]: | |
| """Select agents and their weights based on complexity and domain. | |
| Args: | |
| complexity: Query complexity level | |
| domain: Detected query domain | |
| Returns: | |
| Dict mapping agent names to activation weights (0-1) | |
| """ | |
| # All available agents with their domains | |
| all_agents = { | |
| "Newton": ["physics", "mathematics", "systems"], | |
| "Quantum": ["physics", "uncertainty", "systems"], | |
| "Philosophy": ["philosophy", "meaning", "consciousness"], | |
| "DaVinci": ["creativity", "systems", "innovation"], | |
| "Empathy": ["ethics", "consciousness", "meaning"], | |
| "Ethics": ["ethics", "consciousness", "meaning"], | |
| } | |
| domain_agents = all_agents | |
| if complexity == QueryComplexity.SIMPLE: | |
| # Simple queries: just the primary agent for the domain | |
| # Activate only 1 agent at full strength | |
| primary = self._get_primary_agent(domain) | |
| return {primary: 1.0} | |
| elif complexity == QueryComplexity.MEDIUM: | |
| # Medium queries: primary + 1-2 secondary agents | |
| # Soft gating with weighted influence | |
| primary = self._get_primary_agent(domain) | |
| secondaries = self._get_secondary_agents(domain, count=1) | |
| weights = {primary: 1.0} | |
| for secondary in secondaries: | |
| weights[secondary] = 0.6 | |
| return weights | |
| else: # COMPLEX | |
| # Complex queries: all relevant agents for domain + cross-domain | |
| # Full soft gating | |
| primary = self._get_primary_agent(domain) | |
| secondaries = self._get_secondary_agents(domain, count=2) | |
| cross_domain = self._get_cross_domain_agents(domain, count=1) | |
| weights = {primary: 1.0} | |
| for secondary in secondaries: | |
| weights[secondary] = 0.7 | |
| for cross in cross_domain: | |
| weights[cross] = 0.4 | |
| return weights | |
| def _get_primary_agent(self, domain: str) -> str: | |
| """Get the primary agent for a domain.""" | |
| domain_map = { | |
| "physics": "Newton", | |
| "mathematics": "Newton", | |
| "creativity": "DaVinci", | |
| "ethics": "Ethics", | |
| "philosophy": "Philosophy", | |
| "meaning": "Philosophy", | |
| "consciousness": "Empathy", | |
| "uncertainty": "Quantum", | |
| "systems": "Newton", | |
| } | |
| return domain_map.get(domain, "Newton") | |
| def _get_secondary_agents(self, domain: str, count: int = 1) -> list[str]: | |
| """Get secondary agents for a domain.""" | |
| domain_map = { | |
| "physics": ["Quantum", "DaVinci"], | |
| "mathematics": ["Quantum", "Philosophy"], | |
| "creativity": ["Quantum", "Empathy"], | |
| "ethics": ["Philosophy", "Empathy"], | |
| "philosophy": ["Empathy", "Ethics"], | |
| "meaning": ["Quantum", "DaVinci"], | |
| "consciousness": ["Philosophy", "Quantum"], | |
| "uncertainty": ["Philosophy", "DaVinci"], | |
| "systems": ["DaVinci", "Philosophy"], | |
| } | |
| candidates = domain_map.get(domain, ["Philosophy", "DaVinci"]) | |
| return candidates[:count] | |
| def _get_cross_domain_agents(self, domain: str, count: int = 1) -> list[str]: | |
| """Get cross-domain agents (useful for all domains).""" | |
| # Philosophy and Empathy are useful everywhere | |
| candidates = ["Philosophy", "Empathy", "DaVinci"] | |
| return candidates[:count] | |