feat: wire model_capability into pipeline — tier-adaptive prompts + ctx auto-detect

Model tier detection now active (was orphaned code):
- orchestrator.__init__: detect tier from generator.llm.ollama_model
- ctx.model_tier written at run() start, flows to Generator
- Generator._build_system(): SMALL gets strict format constraints
  (1 function, ≤10 lines, preserve indent, no explanation, ≤15 words between tools)
  LARGE gets minimal constraint; MEDIUM uses base rules as-is
- _get_max_retries() respects model_strategy.max_retries

Context window auto-detection (get_effective_ctx):
- 4-layer probe: llama.cpp /props → vLLM /v1/models → Ollama modelinfo.llama.context_length → tier default
- Passed to LLMBackend n_ctx in build_pipeline
- SMALL=16K, MEDIUM=32K, LARGE=64K defaults; native ctx capped at 64K×0.8

501 tests green, 0 regression.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

kaiwu/cli/commands/task.py

@@ -71,10 +71,15 @@ def build_pipeline(model_path, ollama_url, ollama_model, project_root, verbose):
     _cfg = _load_cfg().get("default", {})
     _api_key = _cfg.get("api_key", "")
 
+    # Ctx自适应：检测模型可用上下文窗口
+    from kaiwu.core.model_capability import get_effective_ctx
+    effective_ctx = get_effective_ctx(ollama_model, ollama_url)
+
     llm = LLMBackend(
         model_path=model_path,
         ollama_url=ollama_url,
         ollama_model=ollama_model,
+        n_ctx=effective_ctx,
         verbose=verbose,
         api_key=_api_key,
     )

kaiwu/core/context.py

@@ -82,3 +82,6 @@ class TaskContext:
 
     # AdaptThink: think模式配置（orchestrator根据expert_type×difficulty设置）
     think_config: dict = field(default_factory=lambda: {"think": False, "budget": 0})
+
+    # 模型能力等级（orchestrator检测后写入，Generator按此调整约束）
+    model_tier: str = ""  # "small"/"medium"/"large"

kaiwu/core/model_capability.py

@@ -150,6 +150,61 @@ def _detect_from_name(model_name: str) -> ModelTier:
     return ModelTier.MEDIUM
 
 
+def get_effective_ctx(model_name: str,
+                      ollama_url: str = "http://localhost:11434") -> int:
+    """
+    获取当前模型实际可用的ctx大小。
+    查询链：llama.cpp /props → vLLM /v1/models → Ollama modelinfo → 按tier默认值。
+    失败全部静默，返回保守默认值。
+    """
+    import httpx
+
+    # 1. llama.cpp /props → 运行时真实值，最准
+    try:
+        r = httpx.get("http://localhost:8080/props", timeout=2)
+        if r.status_code == 200:
+            n_ctx = r.json().get("n_ctx")
+            if n_ctx and n_ctx > 0:
+                return int(n_ctx * 0.8)
+    except Exception:
+        pass
+
+    # 2. vLLM /v1/models → max_model_len
+    try:
+        vllm_url = ollama_url.replace("11434", "8000")
+        r = httpx.get(f"{vllm_url}/v1/models", timeout=2)
+        if r.status_code == 200:
+            data = r.json().get("data", [])
+            if data and "max_model_len" in data[0]:
+                return int(data[0]["max_model_len"] * 0.8)
+    except Exception:
+        pass
+
+    # 3. Ollama /api/show → modelinfo.llama.context_length（模型原生上限）
+    try:
+        r = httpx.post(
+            f"{ollama_url}/api/show",
+            json={"name": model_name},
+            timeout=3,
+        )
+        if r.status_code == 200:
+            data = r.json()
+            native_ctx = data.get("modelinfo", {}).get("llama.context_length", 0)
+            if native_ctx > 0:
+                # 原生上限取80%，且不超过65536（避免本地推理速度崩）
+                return min(int(native_ctx * 0.8), 65536)
+    except Exception:
+        pass
+
+    # 4. 按tier给保守默认值
+    tier = detect_model_tier(model_name, ollama_url)
+    return {
+        ModelTier.SMALL: 16384,
+        ModelTier.MEDIUM: 32768,
+        ModelTier.LARGE: 65536,
+    }[tier]
+
+
 def get_strategy(tier: ModelTier) -> ModelStrategy:
     """Get execution strategy for a given tier."""
     return STRATEGIES[tier]

kaiwu/core/orchestrator.py

@@ -41,6 +41,7 @@ from kaiwu.flywheel.strategy_stats import StrategyStats
 from kaiwu.flywheel.user_pattern_memory import UserPatternMemory
 from kaiwu.telemetry.client import TelemetryClient
 from kaiwu.audit.logger import AuditLogger
+from kaiwu.core.model_capability import detect_model_tier, STRATEGIES, ModelTier
 
 logger = logging.getLogger(__name__)
 
@@ -135,6 +136,15 @@ class PipelineOrchestrator:
         self._user_patterns = UserPatternMemory()
         self._telemetry = TelemetryClient()
         self._audit = AuditLogger()
+        # 模型能力检测（从LLM后端取模型名，失败默认MEDIUM）
+        try:
+            model_name = getattr(self.generator.llm, 'ollama_model', '') or ''
+            ollama_url = getattr(self.generator.llm, 'ollama_url', 'http://localhost:11434')
+            self._model_tier = detect_model_tier(model_name, ollama_url)
+            self._model_strategy = STRATEGIES[self._model_tier]
+        except Exception:
+            self._model_tier = ModelTier.MEDIUM
+            self._model_strategy = STRATEGIES[ModelTier.MEDIUM]
         self.bus = bus or EventBus()
         self._wink = WinkMonitor()
         self._cognitive_gate = CognitiveGate()
@@ -182,6 +192,9 @@ class PipelineOrchestrator:
             expert_system_prompt=gate_result.get("system_prompt", ""),
         )
 
+        # 模型能力等级注入ctx
+        ctx.model_tier = self._model_tier.value
+
         # 用户错误模式提示注入
         warning = self._user_patterns.get_warning_hint()
         if warning:
@@ -838,9 +851,12 @@ class PipelineOrchestrator:
         self._audit.log(stage, detail)
 
     def _get_max_retries(self, gate_result: dict) -> int:
-        """Dynamic retry budget based on task difficulty."""
+        """Dynamic retry budget based on task difficulty and model strategy."""
         difficulty = gate_result.get("difficulty", "easy")
-        return self._RETRY_BY_DIFFICULTY.get(difficulty, self.MAX_RETRIES)
+        base = self._RETRY_BY_DIFFICULTY.get(difficulty, self.MAX_RETRIES)
+        # 模型策略可以覆盖（小模型限制更严）
+        strategy_max = self._model_strategy.max_retries
+        return min(base, strategy_max)
 
     @staticmethod
     def _needs_realtime_data(user_input: str) -> bool:

kaiwu/experts/generator.py

@@ -306,13 +306,32 @@ class GeneratorExpert:
 
     def _build_system(self, ctx: TaskContext, base_system: str = "") -> str:
         """Combine expert_system_prompt (from registry) with base system prompt.
-        Appends WEB_DESIGN_RULES when the task involves web/HTML generation."""
+        Appends tier-specific constraints and WEB_DESIGN_RULES when applicable."""
         expert_prompt = ctx.expert_system_prompt or ""
         base = base_system or GENERATOR_BASE_SYSTEM
         if expert_prompt:
             system = f"{expert_prompt}\n\n{base}"
         else:
             system = base
+
+        # 模型能力自适应：按tier注入不同强度的格式约束
+        tier = getattr(ctx, 'model_tier', '')
+        if tier == "small":
+            system += (
+                "\n\n## 格式约束（小模型严格执行）\n"
+                "- 每次只修改1个函数，修改行数≤10行\n"
+                "- class内方法必须保持原有缩进（通常4空格）\n"
+                "- 直接输出代码，禁止任何解释文字\n"
+                "- 工具调用之间不超过15个词\n"
+                "- 禁止输出markdown代码块标记\n"
+            )
+        elif tier == "large":
+            system += (
+                "\n\n## 格式约束\n"
+                "- 保持代码风格一致，缩进与原文件匹配\n"
+            )
+        # medium: 用GENERATOR_BASE_SYSTEM已有的约束即可
+
         # Append web design rules for HTML/CSS/web tasks
         if self._is_web_task(ctx.user_input):
             system = f"{system}\n\n{WEB_DESIGN_RULES}"