Xshsama/dev

ai_collab/__init__.py

@@ -1,6 +1,6 @@
 """AI Collaboration System - Multi-AI orchestration and workflow management."""
 
-__version__ = "0.1.7"
+__version__ = "0.1.8"
 __author__ = "Sky Hua"
 __email__ = "dev@sky-hua.xyz"
 

ai_collab/core/detector.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import re
 from typing import Dict, List, Optional, Tuple
 
 from pydantic import BaseModel, Field
@@ -97,6 +98,24 @@
     "docs-writing": "document-first",
 }
 
+LOW_SIGNAL_TASK_LITERALS = {
+    "1",
+    "1.",
+    "1。",  # noqa: RUF001
+    "hello",
+    "hi",
+    "hey",
+    "ok",
+    "okay",
+    "test",
+    "sb",
+    "?",
+    "？",  # noqa: RUF001
+    ".",
+    "..",
+    "...",
+}
+
 
 class CollaborationResult(BaseModel):
     """Result of collaboration detection."""
@@ -143,6 +162,8 @@ def detect(self, task: str, current_provider: str) -> CollaborationResult:
         auto_cfg = self.config.auto_collaboration or {}
         if not self._is_auto_collaboration_enabled(auto_cfg):
             return CollaborationResult(need_collaboration=False)
+        if self._is_low_signal_task(task):
+            return CollaborationResult(need_collaboration=False)
 
         profile = ProjectProfiler().detect()
         categories = list(profile.categories)
@@ -393,6 +414,28 @@ def _is_auto_collaboration_enabled(self, auto_cfg: Dict) -> bool:
             return bool(auto_cfg.get("auto_orchestration_enabled"))
         return True
 
+    def _is_low_signal_task(self, task: str) -> bool:
+        normalized = re.sub(r"\s+", " ", str(task or "").strip())
+        if not normalized:
+            return False
+
+        lower = normalized.lower()
+        if lower in LOW_SIGNAL_TASK_LITERALS:
+            return True
+        if re.fullmatch(r"[\W_]+", lower):
+            return True
+        if re.fullmatch(r"\d+(?:[.,]\d+)?", lower):
+            return True
+
+        ascii_tokens = re.findall(r"[a-z0-9]+", lower)
+        cjk_chars = re.findall(r"[\u4e00-\u9fff]", normalized)
+        token_count = len(ascii_tokens) + len(cjk_chars)
+        if token_count > 1:
+            return False
+
+        token = ascii_tokens[0] if ascii_tokens else "".join(cjk_chars)
+        return bool(token) and len(token) <= 3
+
     def _is_planner_multi_agent(self, plan: Dict[str, object]) -> bool:
         mode = str(plan.get("mode", "single-agent"))
         if mode == "multi-agent":

ai_collab/core/orchestrator.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+import re
 from typing import Dict, List, Optional, Set
 
 from ai_collab.core.config import Config, resolve_collaboration_role_leads
@@ -56,6 +57,29 @@
     "deliver": "quality-review",
 }
 
+SMALL_BOUNDED_TASK_CUES = (
+    "hello",
+    "demo",
+    "example",
+    "sample",
+    "print",
+    "typo",
+    "rename",
+    "format",
+    "lint",
+    "comment",
+    "小改",
+    "微调",
+    "一行",
+    "一句",
+    "示例",
+    "错别字",
+    "改名",
+    "重命名",
+    "格式",
+    "注释",
+)
+
 
 class OrchestrationPlanner:
     """Builds controller-visible agent roster and role assignments."""
@@ -176,6 +200,7 @@ def _plan_roles(
         available_agents: List[Dict[str, str]],
     ) -> List[str]:
         roles: List[str] = []
+        small_bounded_task = self._is_small_bounded_task(task)
         enabled_agents = {item.get("agent", "") for item in available_agents if item.get("agent")}
         configured_defaults = self._configured_default_roles(enabled_agents=enabled_agents)
 
@@ -198,20 +223,21 @@ def _plan_roles(
             if intent == "architecture":
                 roles.append("tech-selection")
             roles.append("implementation")
-            if intent in {"implementation", "debug", "testing", "security"}:
+            if intent in {"implementation", "debug", "testing", "security"} and not small_bounded_task:
                 roles.append("quality-review")
 
         if not roles:
             roles.extend(configured_defaults)
 
         if not roles:
             roles.append("implementation")
-            if len(enabled_agents) > 1:
+            if len(enabled_agents) > 1 and not small_bounded_task:
                 roles.append("quality-review")
         elif (
             "implementation" in roles
             and "quality-review" not in roles
             and len(enabled_agents) > 1
+            and not small_bounded_task
         ):
             roles.append("quality-review")
 
@@ -224,6 +250,24 @@ def _plan_roles(
             seen.add(role)
         return deduped
 
+    def _is_small_bounded_task(self, task: str) -> bool:
+        normalized = re.sub(r"\s+", " ", str(task or "").strip())
+        if not normalized:
+            return False
+
+        lower = normalized.lower()
+        if re.fullmatch(r"[\W_]+", lower):
+            return True
+        if re.fullmatch(r"\d+(?:[.,]\d+)?", lower):
+            return True
+        if any(cue in lower for cue in SMALL_BOUNDED_TASK_CUES):
+            return True
+
+        ascii_tokens = re.findall(r"[a-z0-9]+", lower)
+        if ascii_tokens and len(ascii_tokens) <= 2 and len(lower) <= 12:
+            return True
+        return False
+
     def _configured_default_roles(self, *, enabled_agents: Set[str]) -> List[str]:
         auto_cfg = self.config.auto_collaboration or {}
         assignment_map = self._assignment_map()

ai_collab/launcher_service.py

@@ -7,6 +7,7 @@
 from typing import Any, Callable, Optional, Sequence
 
 from ai_collab.core.config import Config
+from ai_collab.core.detector import CollaborationDetector, CollaborationResult
 from ai_collab.core.selector import ModelSelector
 from ai_collab.ux_lab_v3 import (
     DEFAULT_AGENTS,
@@ -181,6 +182,77 @@ def hydrate_controller_plan_models(*, config: Config, task: str, controller_plan
     return controller_plan
 
 
+def _single_agent_step_goal(*, task: str, lang: str) -> str:
+    normalized = _normalize_text(task)
+    lower = normalized.lower()
+    if lower in {"1", "1.", "1。", "hello", "hi", "hey", "ok", "okay", "test", "sb"}:
+        return "直接响应用户输入" if lang == "zh-CN" else "Respond directly to the user's prompt"
+    return normalized or ("完成当前任务" if lang == "zh-CN" else "Complete the task")
+
+
+def _single_agent_done_when(*, task: str, lang: str) -> str:
+    normalized = _normalize_text(task)
+    if lang == "zh-CN":
+        if normalized:
+            return f"直接完成“{normalized}”并给出可检查结果。"
+        return "直接完成当前任务并给出可检查结果。"
+    if normalized:
+        return f"Directly complete '{normalized}' and provide a checkable result."
+    return "Directly complete the task and provide a checkable result."
+
+
+def _single_agent_approval_question(*, task: str, lang: str) -> str:
+    normalized = _normalize_text(task)
+    if lang == "zh-CN":
+        return f"是否按单 Agent 方式直接执行“{normalized or '当前任务'}”？"
+    return f"Execute '{normalized or 'this task'}' as a single-agent task?"
+
+
+def _single_agent_controller_plan(
+    *,
+    config: Config,
+    controller: str,
+    task: str,
+    lang: str,
+    detection: Optional[CollaborationResult] = None,
+) -> dict[str, Any]:
+    selector = ModelSelector(config)
+    model = "unknown"
+    try:
+        model = str(selector.select_model(controller, task, "default").model or "unknown")
+    except Exception:  # noqa: BLE001
+        pass
+
+    return {
+        "plan_version": "1.0",
+        "controller": controller,
+        "requires_multi_agent": False,
+        "workflow_engine": str(getattr(detection, "workflow_engine", "") or "v2"),
+        "session_preset": str(getattr(detection, "session_preset", "") or "auto"),
+        "workflow_blueprint": str(getattr(detection, "workflow_blueprint", "") or "delivery-loop"),
+        "agents": [
+            {
+                "name": controller,
+                "model": model,
+                "persona": "implementation-engineer",
+                "why": "single-agent task",
+            }
+        ],
+        "steps": [
+            {
+                "id": "S1",
+                "owner": controller,
+                "goal": _single_agent_step_goal(task=task, lang=lang),
+                "input": _normalize_text(task),
+                "output": "direct-response",
+                "done_when": _single_agent_done_when(task=task, lang=lang),
+                "eta_minutes": 8,
+            }
+        ],
+        "approval_question": _single_agent_approval_question(task=task, lang=lang),
+    }
+
+
 def resolve_task_text(*, task: Optional[str], task_file: Optional[Path]) -> str:
     if task_file is not None:
         return Path(task_file).expanduser().read_text(encoding="utf-8").strip()
@@ -351,8 +423,23 @@ def run_launcher_flow(
     available = enabled_agents(config)
     resolved_controller = resolve_controller(controller, config.current_controller, available)
     mode = planner_mode if planner_mode in {"live", "mock"} else "live"
-
-    if mode == "mock":
+    detection: Optional[CollaborationResult] = None
+    try:
+        detection = CollaborationDetector(config).detect(resolved_task, resolved_controller)
+    except Exception:  # noqa: BLE001
+        detection = None
+
+    if detection is not None and not detection.need_collaboration:
+        controller_plan = _single_agent_controller_plan(
+            config=config,
+            controller=resolved_controller,
+            task=resolved_task,
+            lang=lang,
+            detection=detection,
+        )
+        plan = map_controller_plan_to_items(controller_plan, lang=lang)
+        error = None
+    elif mode == "mock":
         plan = build_mock_plan_v3(resolved_task, resolved_controller, lang, available_agents=available)
         controller_plan = None
         error = None

ai_collab/tui/launcher_service.py

@@ -7,6 +7,7 @@
 from typing import Any, Callable, Optional, Sequence
 
 from ai_collab.core.config import Config
+from ai_collab.core.detector import CollaborationDetector, CollaborationResult
 from ai_collab.core.selector import ModelSelector
 from ai_collab.ux_lab_v3 import (
     DEFAULT_AGENTS,
@@ -181,6 +182,77 @@ def hydrate_controller_plan_models(*, config: Config, task: str, controller_plan
     return controller_plan
 
 
+def _single_agent_step_goal(*, task: str, lang: str) -> str:
+    normalized = _normalize_text(task)
+    lower = normalized.lower()
+    if lower in {"1", "1.", "1。", "hello", "hi", "hey", "ok", "okay", "test", "sb"}:
+        return "直接响应用户输入" if lang == "zh-CN" else "Respond directly to the user's prompt"
+    return normalized or ("完成当前任务" if lang == "zh-CN" else "Complete the task")
+
+
+def _single_agent_done_when(*, task: str, lang: str) -> str:
+    normalized = _normalize_text(task)
+    if lang == "zh-CN":
+        if normalized:
+            return f"直接完成“{normalized}”并给出可检查结果。"
+        return "直接完成当前任务并给出可检查结果。"
+    if normalized:
+        return f"Directly complete '{normalized}' and provide a checkable result."
+    return "Directly complete the task and provide a checkable result."
+
+
+def _single_agent_approval_question(*, task: str, lang: str) -> str:
+    normalized = _normalize_text(task)
+    if lang == "zh-CN":
+        return f"是否按单 Agent 方式直接执行“{normalized or '当前任务'}”？"
+    return f"Execute '{normalized or 'this task'}' as a single-agent task?"
+
+
+def _single_agent_controller_plan(
+    *,
+    config: Config,
+    controller: str,
+    task: str,
+    lang: str,
+    detection: Optional[CollaborationResult] = None,
+) -> dict[str, Any]:
+    selector = ModelSelector(config)
+    model = "unknown"
+    try:
+        model = str(selector.select_model(controller, task, "default").model or "unknown")
+    except Exception:  # noqa: BLE001
+        pass
+
+    return {
+        "plan_version": "1.0",
+        "controller": controller,
+        "requires_multi_agent": False,
+        "workflow_engine": str(getattr(detection, "workflow_engine", "") or "v2"),
+        "session_preset": str(getattr(detection, "session_preset", "") or "auto"),
+        "workflow_blueprint": str(getattr(detection, "workflow_blueprint", "") or "delivery-loop"),
+        "agents": [
+            {
+                "name": controller,
+                "model": model,
+                "persona": "implementation-engineer",
+                "why": "single-agent task",
+            }
+        ],
+        "steps": [
+            {
+                "id": "S1",
+                "owner": controller,
+                "goal": _single_agent_step_goal(task=task, lang=lang),
+                "input": _normalize_text(task),
+                "output": "direct-response",
+                "done_when": _single_agent_done_when(task=task, lang=lang),
+                "eta_minutes": 8,
+            }
+        ],
+        "approval_question": _single_agent_approval_question(task=task, lang=lang),
+    }
+
+
 def resolve_task_text(*, task: Optional[str], task_file: Optional[Path]) -> str:
     if task_file is not None:
         return Path(task_file).expanduser().read_text(encoding="utf-8").strip()
@@ -351,8 +423,23 @@ def run_launcher_flow(
     available = enabled_agents(config)
     resolved_controller = resolve_controller(controller, config.current_controller, available)
     mode = planner_mode if planner_mode in {"live", "mock"} else "live"
-
-    if mode == "mock":
+    detection: Optional[CollaborationResult] = None
+    try:
+        detection = CollaborationDetector(config).detect(resolved_task, resolved_controller)
+    except Exception:  # noqa: BLE001
+        detection = None
+
+    if detection is not None and not detection.need_collaboration:
+        controller_plan = _single_agent_controller_plan(
+            config=config,
+            controller=resolved_controller,
+            task=resolved_task,
+            lang=lang,
+            detection=detection,
+        )
+        plan = map_controller_plan_to_items(controller_plan, lang=lang)
+        error = None
+    elif mode == "mock":
         plan = build_mock_plan_v3(resolved_task, resolved_controller, lang, available_agents=available)
         controller_plan = None
         error = None