From 4fbbdb5290744488ebcb5222dd1c32861ea21c28 Mon Sep 17 00:00:00 2001
From: tzt <14718231+flying-travel@user.noreply.gitee.com>
Date: Sun, 30 Aug 2026 21:08:45 +0800
Subject: [PATCH] =?UTF-8?q?feat(v2):=20T5=20WorkerLoop=20+=20=E6=8E=A5?=
 =?UTF-8?q?=E5=9C=B0=E9=AA=8C=E8=AF=81=EF=BC=88=E4=BB=A3=E7=A0=81=E6=B2=99?=
 =?UTF-8?q?=E7=AE=B1/facts=E5=AF=B9=E7=85=A7/=E7=BB=93=E6=9E=84=E6=A3=80?=
 =?UTF-8?q?=E6=9F=A5=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 router_system/verifier.py | 135 ++++++++++++++++++++++++++++
 router_system/worker.py   | 180 ++++++++++++++++++++++++++++++++++++++
 tests/conftest.py         |   8 +-
 tests/test_worker.py      | 179 +++++++++++++++++++++++++++++++++++++
 任务拆解与执行计划.md     |   2 +-
 5 files changed, 502 insertions(+), 2 deletions(-)
 create mode 100644 router_system/verifier.py
 create mode 100644 router_system/worker.py
 create mode 100644 tests/test_worker.py

diff --git a/router_system/verifier.py b/router_system/verifier.py
new file mode 100644
index 0000000..6402c02
--- /dev/null
+++ b/router_system/verifier.py
@@ -0,0 +1,135 @@
+"""接地验证器（D4）—— Worker 自验证的"接地"来源。
+
+验证分层优先级（D4）：
+  可执行验证（跑代码/跑测试） > facts 对照 > 结构检查 > 模型自由判断（最后手段）
+
+- 可执行验证：code 域 .py 工件在临时目录子进程沙箱运行（timeout、-I 隔离、捕获输出）。
+- facts 对照：把回答/工件与知识库 facts（statement/keywords）比对覆盖度。
+- 结构检查：工件非空、长度达标、不含纯占位。
+
+说明：沙箱目前做"临时目录 + 超时 + 解释器隔离"，Windows 下真正禁网需系统级工具，
+此处以超时与隔离为主要护栏（文档如实记录）。验证器为纯逻辑，可注入 runner 便于单测。
+"""
+from __future__ import annotations
+
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+# 判定为"含代码"的启发标记
+_CODE_HINTS = ("def ", "class ", "import ", "return ", "if __name__", "print(")
+
+
+def detect_artifact_language(name: str) -> str:
+    """按文件名推断工件语言：python / json / text。"""
+    suffix = Path(name).suffix.lower()
+    if suffix in (".py", ".pyw"):
+        return "python"
+    if suffix in (".json",):
+        return "json"
+    return "text"
+
+
+def extract_code_block(text: str) -> str:
+    """从模型输出提取 python 代码块（剥除 markdown 围栏），无则返回原文本。"""
+    t = text.strip()
+    fence = chr(96) * 3  # 三个反引号
+    marker = fence + "python"
+    start = t.find(marker)
+    if start == -1:
+        return t
+    body_start = start + len(marker)
+    end = t.find(fence, body_start)
+    if end == -1:
+        return t[body_start:].strip()
+    return t[body_start:end].strip()
+
+
+def run_code_sandbox(code: str, timeout_s: float = 10.0,
+                     runner: Optional[Callable[[List[str], Dict[str, str], float], Tuple[int, str, str]]] = None
+                     ) -> Tuple[int, str, str]:
+    """在临时目录子进程运行 python 代码。返回 (returncode, stdout, stderr)。
+
+    隔离措施：临时工作目录、-I 隔离模式、timeout 超时强杀、捕获输出。
+    runner 可注入（测试用假执行器，避免真跑任意代码）。
+    """
+    if runner is not None:
+        return runner([sys.executable, "-I"], {}, timeout_s)
+    with tempfile.TemporaryDirectory(prefix="v2_sandbox_") as tmp:
+        script = Path(tmp) / "main.py"
+        script.write_text(code, encoding="utf-8")
+        try:
+            proc = subprocess.run(
+                [sys.executable, "-I", str(script)],
+                capture_output=True, text=True, timeout=timeout_s,
+                cwd=tmp,
+                creationflags=subprocess.CREATE_NO_WINDOW,
+            )
+            return proc.returncode, proc.stdout or "", proc.stderr or ""
+        except subprocess.TimeoutExpired:
+            return -1, "", "timeout exceeded"
+
+
+class Verifier:
+    """按 D4 分层执行接地验证。"""
+
+    def __init__(self, code_timeout_s: float = 10.0,
+                 sandbox_runner: Optional[Callable[..., Tuple[int, str, str]]] = None):
+        self.code_timeout_s = code_timeout_s
+        self._sandbox_runner = sandbox_runner
+
+    def verify(self, domain: str, artifact_name: str, artifact_text: str,
+               query: str, kb: Any = None) -> Tuple[bool, List[str]]:
+        """返回 (passed, details)。kb 为 KnowledgeBase（可 None，跳过 facts 层）。"""
+        lang = detect_artifact_language(artifact_name)
+        details: List[str] = []
+
+        # 1) 可执行验证（D4 最高优先级）：code + python 工件且含代码
+        if lang == "python" and any(h in artifact_text for h in _CODE_HINTS):
+            rc, out, err = run_code_sandbox(artifact_text, self.code_timeout_s,
+                                            runner=self._sandbox_runner)
+            if rc == 0:
+                details.append("代码沙箱运行通过 (rc=0)")
+                return True, details
+            # 可执行验证失败 -> 直接拒绝（不落回结构检查，避免"假通过"）
+            details.append(f"代码沙箱运行失败 rc={rc}: {(err or out)[:120]}")
+            return False, details
+        elif lang == "python":
+            details.append("工件不含可执行代码（跳过沙箱，进入 facts/结构检查）")
+
+        # 2) facts 对照
+        if kb is not None:
+            fact_hits = self._check_facts(domain, artifact_text, kb)
+            if fact_hits:
+                details.append(f"facts 对照命中 {fact_hits}")
+                return True, details
+
+        # 3) 结构检查
+        text = artifact_text.strip()
+        if len(text) < 20:
+            details.append(f"工件过短（{len(text)} 字符）")
+            return False, details
+        if text.lower() in ("pass", "none", "todo", "待实现", "略"):
+            details.append("工件为占位内容")
+            return False, details
+        details.append("结构检查通过（非空、长度达标）")
+        return True, details
+
+    def _check_facts(self, domain: str, text: str, kb: Any) -> int:
+        """统计工件文本命中知识库事实的数量。"""
+        facts = kb.facts(domain) or []
+        if not facts:
+            return 0
+        hits = 0
+        for f in facts:
+            kw = f.get("keywords") or []
+            if any(str(k) in text for k in kw):
+                hits += 1
+        return hits
+
+
+def build_verifier(cfg: Dict[str, Any]) -> Verifier:
+    """cfg 为 config.worker 段。"""
+    return Verifier(code_timeout_s=float(cfg.get("code_timeout_s", 10)))
diff --git a/router_system/worker.py b/router_system/worker.py
new file mode 100644
index 0000000..98aed8c
--- /dev/null
+++ b/router_system/worker.py
@@ -0,0 +1,180 @@
+"""WorkerLoop —— 小模型（本地 llama.cpp）的"实现/自验证"循环（端云协同的执行者）。
+
+流程（对齐《实现方案_v2》5.1 T5 / 4.4）：
+  读 brief+当前步 -> 模型生成工件 -> 接地验证（D4 分层）
+    -> 通过：写 progress(done)
+    -> 失败：自修 <= max_fix_attempts 次（把验证错误回喂重新生成）
+    -> 仍失败：写 issue（增量、带锚点）
+
+- generate 为可注入的文本生成器（真实为 llama-server 端点；测试用假实现）。
+- 工件落盘：runs/<request_id>/artifacts/<step>.py（由 pipeline 负责写盘，本模块只产出文本）。
+"""
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from typing import Any, Awaitable, Callable, Dict, List, Optional
+
+from .verifier import Verifier, detect_artifact_language, extract_code_block
+from .workspace import Workspace, build_anchor
+
+# 按领域推断默认工件扩展名
+_DOMAIN_EXT = {
+    "code": ".py",
+    "math": ".md",
+    "legal": ".md",
+    "medical": ".md",
+    "finance": ".md",
+    "life": ".md",
+    "education": ".md",
+    "general": ".md",
+}
+
+
+def artifact_name_for(step_id: str, domain: str) -> str:
+    """为 step 生成工件文件名。"""
+    ext = _DOMAIN_EXT.get(domain, ".md")
+    return f"{step_id}{ext}"
+
+
+@dataclass
+class StepOutcome:
+    """单步执行结果。"""
+    step_id: str
+    status: str                    # done | issue
+    summary: str = ""
+    model_used: str = "local"
+    attempts: int = 0
+    issue_id: Optional[str] = None
+    artifact_name: Optional[str] = None
+    artifact_text: str = ""
+    details: List[str] = field(default_factory=list)
+
+
+class WorkerLoop:
+    """小模型 Worker：实现 -> 验证 -> 自修 -> issue。"""
+
+    def __init__(
+        self,
+        generate: Callable[[str], Awaitable[str]],
+        verifier: Optional[Verifier] = None,
+        kb: Any = None,
+        max_fix_attempts: int = 2,
+        model_used: str = "local-llama",
+    ):
+        self.generate = generate
+        self.verifier = verifier or Verifier()
+        self.kb = kb
+        self.max_fix_attempts = max_fix_attempts
+        self.model_used = model_used
+
+    def _domain_from(self, ws: Workspace) -> str:
+        tags = (ws.get("brief") or {}).get("tags") or []
+        for t in tags:
+            if t != "safety":
+                return t
+        return "general"
+
+    async def run_step(self, ws: Workspace, step_id: str,
+                       existing_artifact: str = "") -> StepOutcome:
+        """执行单个 step。existing_artifact 为该步当前已有工件全文（若有）。"""
+        domain = self._domain_from(ws)
+        brief = ws.get("brief") or {}
+        plan = brief.get("plan") or []
+        step_def = next((p for p in plan if p.get("id") == step_id), {})
+        done_criteria = step_def.get("done_criteria", "")
+
+        artifact_name = artifact_name_for(step_id, domain)
+        current = existing_artifact
+        details: List[str] = []
+
+        for attempt in range(1, self.max_fix_attempts + 1):
+            prompt = self._build_prompt(ws, step_id, current, attempt, done_criteria)
+            out = await self.generate(prompt)
+            if domain == "code":
+                candidate = extract_code_block(out)
+            else:
+                candidate = out.strip()
+            details.append(f"attempt{attempt}: 生成 {len(candidate)} 字符")
+
+            passed, v_details = self.verifier.verify(
+                domain, artifact_name, candidate, ws["query"], kb=self.kb)
+            details.extend(f"  - {d}" for d in v_details)
+            if passed:
+                # 写回交流文本：progress(done) + 摘要
+                ws.add_progress(step_id, "done", f"步骤完成（{attempt} 次尝试）",
+                                artifact=build_anchor(artifact_name, 1))
+                return StepOutcome(
+                    step_id=step_id, status="done",
+                    summary=f"步骤完成（{attempt} 次尝试）",
+                    model_used=self.model_used, attempts=attempt,
+                    artifact_name=artifact_name, artifact_text=candidate,
+                    details=details,
+                )
+            # 未通过：带错误反馈重新生成（自修）
+            current = candidate
+            feedback = "；".join(v_details)
+            details.append(f"attempt{attempt} 未通过，进入自修")
+
+        # 全部尝试失败 -> 写 issue
+        anchor = build_anchor(artifact_name, 1, 30)
+        iid = ws.add_issue(
+            step=step_id,
+            anchor=anchor,
+            observed=f"验证未通过：{'；'.join(d for d in details if d.startswith('  - ')) or '未知'}",
+            expected=done_criteria or "满足该步 done_criteria",
+            tried=f"已自修 {self.max_fix_attempts} 次",
+            ask="请裁决该步的实现方向或提供兜底实现",
+        )
+        return StepOutcome(
+            step_id=step_id, status="issue", summary="未能通过验证，已上报 issue",
+            model_used=self.model_used, attempts=self.max_fix_attempts,
+            issue_id=iid, artifact_name=artifact_name, artifact_text=current,
+            details=details,
+        )
+
+    def _build_prompt(self, ws: Workspace, step_id: str, current: str,
+                      attempt: int, done_criteria: str) -> str:
+        base = ws.render_for_worker(step_id, artifact_text=current or None)
+        if attempt > 1:
+            base += (
+                "\n\n[注意] 上次生成的工件未通过接地验证。请修正以下问题后重新输出"
+                f"完整工件。本次为第 {attempt} 次尝试。"
+            )
+        return base
+
+
+def build_worker(cfg: Dict[str, Any], kb: Any = None,
+                 generate: Optional[Callable[[str], Awaitable[str]]] = None) -> WorkerLoop:
+    """cfg 为 config.worker 段。generate 缺省时用 llama-server 端点客户端（惰性）。"""
+    if generate is None:
+        generate = _make_llama_generate(cfg)
+    verifier = Verifier(code_timeout_s=float(cfg.get("code_timeout_s", 10)))
+    return WorkerLoop(
+        generate=generate,
+        verifier=verifier,
+        kb=kb,
+        max_fix_attempts=int(cfg.get("max_fix_attempts", 2)),
+        model_used=cfg.get("backend", "llama_server"),
+    )
+
+
+def _make_llama_generate(cfg: Dict[str, Any]) -> Callable[[str], Awaitable[str]]:
+    """返回调用本地 llama-server（OpenAI 兼容 /v1/chat/completions）的生成器。"""
+    base_url = cfg.get("base_url", f"http://127.0.0.1:{cfg.get('port', 8901)}/v1")
+    model = cfg.get("model", "local")
+    temperature = float(cfg.get("temperature", 0.3))
+    timeout_s = float(cfg.get("per_step_timeout_s", 300))
+
+    async def _gen(prompt: str) -> str:
+        import httpx
+        async with httpx.AsyncClient(timeout=timeout_s) as client:
+            resp = await client.post(
+                f"{base_url}/chat/completions",
+                json={"model": model, "messages": [{"role": "user", "content": prompt}],
+                      "temperature": temperature, "max_tokens": 4096},
+            )
+            resp.raise_for_status()
+            return resp.json()["choices"][0]["message"]["content"]
+
+    return _gen
diff --git a/tests/conftest.py b/tests/conftest.py
index 05e476e..bdd5d29 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -10,5 +10,11 @@ from router_system.router import build_router
 
 @pytest.fixture()
 def router():
-    """????????? mock ?????????????"""
+    """共享的 mock 全链路路由实例（零依赖、离线可跑）"""
     return build_router()
+
+@pytest.fixture()
+def kb():
+    """v2 验证接地用的知识库实例（facts 对照）。"""
+    from router_system.knowledge import KnowledgeBase
+    return KnowledgeBase()
diff --git a/tests/test_worker.py b/tests/test_worker.py
new file mode 100644
index 0000000..e331c24
--- /dev/null
+++ b/tests/test_worker.py
@@ -0,0 +1,179 @@
+"""T5 WorkerLoop + 接地验证单测（封闭，注入假 generate / 假沙箱 runner）。"""
+import pytest
+
+from router_system.knowledge import KnowledgeBase
+from router_system.verifier import (
+    Verifier,
+    detect_artifact_language,
+    extract_code_block,
+    run_code_sandbox,
+)
+from router_system.worker import WorkerLoop, artifact_name_for
+from router_system.workspace import Workspace
+
+
+def _brief(domain="code"):
+    return {
+        "goal": "实现快排",
+        "constraints": ["标准库"],
+        "tags": [domain],
+        "acceptance": [{"id": "a1", "check": "可运行", "machine_checkable": True}],
+        "plan": [
+            {"id": "s1", "task": "实现快排", "deps": [], "done_criteria": "函数可运行"},
+            {"id": "s2", "task": "写测试", "deps": ["s1"], "done_criteria": "3/3 通过"},
+        ],
+    }
+
+
+def _ws(domain="code"):
+    ws = Workspace.new(request_id="abc123def456", query="写个快排",
+                       api_token_cap=8000, rounds_cap=6)
+    ws.apply_brief(_brief(domain))
+    return ws
+
+
+def _fake_gen(texts):
+    """顺序返回预设生成文本的假 generate。"""
+    it = iter(texts)
+
+    async def _g(prompt):
+        try:
+            return next(it)
+        except StopIteration:
+            return ""
+
+    return _g
+
+
+# ---------- verifier: 代码沙箱 ----------
+def test_verifier_code_pass(tmp_path):
+    v = Verifier(sandbox_runner=lambda cmd, env, t: (0, "ok", ""))
+    passed, det = v.verify("code", "s1.py", "def f():\n    return 1\n", "q")
+    assert passed is True
+    assert any("沙箱运行通过" in d for d in det)
+
+
+def test_verifier_code_fail_then_structure():
+    # 沙箱失败 + 无 facts + 文本过短 -> False
+    v = Verifier(sandbox_runner=lambda cmd, env, t: (1, "", "boom"))
+    passed, det = v.verify("code", "s1.py", "def f(): pass", "q")
+    assert passed is False
+
+
+def test_verifier_facts_hit(kb):
+    v = Verifier()
+    # medical 域：回答命中"高血压"事实关键词
+    passed, det = v.verify("medical", "s1.md",
+                           "高血压患者应低盐低脂饮食、控制体重、规律运动", "高血压饮食", kb=kb)
+    assert passed is True
+    assert any("facts" in d for d in det)
+
+
+def test_verifier_structure_pass():
+    v = Verifier()
+    passed, _ = v.verify("general", "s1.md", "这是一段足够长的、非占位的正常回答内容。", "q")
+    assert passed is True
+
+
+def test_verifier_placeholder_fail():
+    v = Verifier()
+    passed, det = v.verify("general", "s1.md", "待实现", "q")
+    assert passed is False
+
+
+# ---------- run_code_sandbox ----------
+def test_run_code_sandbox_good_code():
+    rc, out, err = run_code_sandbox("print(1 + 1)")
+    assert rc == 0
+    assert out.strip() == "2"
+
+
+def test_run_code_sandbox_bad_code():
+    rc, out, err = run_code_sandbox("raise ValueError('x')")
+    assert rc != 0
+    assert "ValueError" in (err or "")
+
+
+# ---------- extract_code_block ----------
+def test_extract_code_block():
+    bt = chr(96) * 3
+    text = f"前言\n{bt}python\ndef f(): return 1\n{bt}\n后记"
+    assert extract_code_block(text) == "def f(): return 1"
+
+
+def test_extract_code_block_no_fence():
+    assert extract_code_block("print(1)") == "print(1)"
+
+
+# ---------- detect_artifact_language ----------
+def test_detect_language():
+    assert detect_artifact_language("s1.py") == "python"
+    assert detect_artifact_language("s1.json") == "json"
+    assert detect_artifact_language("s1.md") == "text"
+
+
+# ---------- artifact_name ----------
+def test_artifact_name():
+    assert artifact_name_for("s1", "code") == "s1.py"
+    assert artifact_name_for("s1", "legal") == "s1.md"
+
+
+# ---------- WorkerLoop 三路径 ----------
+def test_worker_one_pass():
+    code = "def quicksort(arr):\n    return sorted(arr)"
+    ws = _ws()
+    w = WorkerLoop(_fake_gen([code]),
+                   verifier=Verifier(sandbox_runner=lambda c, e, t: (0, "", "")),
+                   max_fix_attempts=2)
+    out = asyncio_run(w.run_step(ws, "s1"))
+    assert out.status == "done"
+    assert out.attempts == 1
+    assert out.artifact_text == code
+    assert len(ws["progress"]) == 1
+    assert ws["progress"][0]["status"] == "done"
+
+
+def test_worker_self_fix_success():
+    bad = "def quicksort(arr):\n    return arr[0]  # 错"
+    good = "def quicksort(arr):\n    return sorted(arr)"
+    # 第一次沙箱失败，第二次通过
+    results = iter([(1, "", "IndexError"), (0, "", "")])
+
+    def fake_run(cmd, env, t):
+        return next(results)
+
+    ws = _ws()
+    w = WorkerLoop(_fake_gen([bad, good]),
+                   verifier=Verifier(sandbox_runner=fake_run),
+                   max_fix_attempts=2)
+    out = asyncio_run(w.run_step(ws, "s1"))
+    assert out.status == "done"
+    assert out.attempts == 2
+    assert len(ws["progress"]) == 1
+
+
+def test_worker_issue_after_exhaust():
+    bad = "def quicksort(arr):\n    return arr[0]"
+    ws = _ws()
+    w = WorkerLoop(_fake_gen([bad, bad]),
+                   verifier=Verifier(sandbox_runner=lambda c, e, t: (1, "", "Err")),
+                   max_fix_attempts=2)
+    out = asyncio_run(w.run_step(ws, "s1"))
+    assert out.status == "issue"
+    assert out.issue_id is not None
+    assert len(ws["issues"]) == 1
+    assert ws["issues"][0]["step"] == "s1"
+
+
+def test_worker_medical_facts_pass(kb):
+    # 医疗域：模型给出含事实关键词的回答 -> facts 验证通过
+    answer = "高血压患者应低盐低脂饮食、控制体重、规律运动、戒烟限酒"
+    ws = _ws(domain="medical")
+    w = WorkerLoop(_fake_gen([answer]), kb=kb, max_fix_attempts=2)
+    out = asyncio_run(w.run_step(ws, "s1"))
+    assert out.status == "done"
+
+
+def asyncio_run(coro):
+    import asyncio
+    return asyncio.run(coro)
diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md
index 1b8da30..58b313b 100644
--- a/任务拆解与执行计划.md
+++ b/任务拆解与执行计划.md
@@ -71,7 +71,7 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯
 | T2 | 运维层：hw_profile + llama_server 进程管理 | ✅ 完成 | T2 |
 | T3 | ArchitectClient（DeepSeek API，JSON 约束） | ✅ 完成 | T3 |
 | T4 | Workspace（交流文本 schema/校验/渲染/rollup） | ✅ 完成 | T4 |
-| T5 | WorkerLoop + 接地验证 | ⬜ | |
+| T5 | WorkerLoop + 接地验证 | ✅ 完成 | T5 |
 | T6 | CollaborativePipeline 编排 | ⬜ | |
 | T7 | 网关扩展（/chat 切 v2，/chat/legacy） | ⬜ | |
 | T8 | 人工检验队列 ReviewQueue | ⬜ | |