From 8f796a1c261ae6ae2dce00f6ae89695e5f7333b0 Mon Sep 17 00:00:00 2001 From: cfdaily Date: Sun, 7 Jun 2026 01:29:27 +0800 Subject: [PATCH] auto-sync: 2026-06-07 01:29:27 --- src/daemon/bootstrap.py | 281 ++++++++++++++++------------------------ 1 file changed, 112 insertions(+), 169 deletions(-) diff --git a/src/daemon/bootstrap.py b/src/daemon/bootstrap.py index 8e91210..e5d5aca 100644 --- a/src/daemon/bootstrap.py +++ b/src/daemon/bootstrap.py @@ -1,25 +1,21 @@ -"""Bootstrap 拼装 — L0-L3 四层上下文构建 +"""Bootstrap 拼装 — L2 引擎注入层 -L0 铁律层（~500 tokens） → Hook 注入，不占 bootstrap -L1 角色层（~2000 tokens） → SOUL.md / IDENTITY.md（Agent 自带） -L2 引擎注入层（~1500 tokens） → prompt_templates 按 role 拼装 -L3 被动参考层（按需加载） → Skill description 四要素 +4 段结构（~800 tokens）： + 1. 任务上下文：title / description / must_haves / status + 2. 前序产出：depends_on 产出摘要 + handoff comment + 3. 角色操作规范全文：通过 ROLE_SKILL_MAP 从 Skill 文件读取 + 4. 硬约束：状态流转约束 -Bootstrap 按 role 精确注入：执行者注入 Guardrail + 审查协议， -审查者注入审查协议，庞统注入审查协议。 +A 类 Skill 由引擎确定性注入全文，不靠 Description 触发。 """ -from __future__ import annotations - -import json import logging -import re +import os from pathlib import Path from typing import Any, Dict, List, Optional logger = logging.getLogger("moziplus-v2.bootstrap") -# 粗略 token 估算：英文 ~4 chars/token，中文 ~2 chars/token CHARS_PER_TOKEN = 3.0 @@ -29,188 +25,135 @@ def estimate_tokens(text: str) -> int: class BootstrapBuilder: - """Bootstrap 四层上下文构建器""" + """L2 引擎注入层构建器（v2.1 四段式）""" - def __init__( - self, - template_dir: Optional[Path] = None, - max_tokens: int = 4096, - ): - """ - Args: - template_dir: prompt_templates 目录路径 - max_tokens: bootstrap 最大 token 数 - """ - self.template_dir = template_dir + ROLE_SKILL_MAP = { + "executor": "blackboard-executor", + "reviewer": "blackboard-reviewer", + "reviewer-simayi": "blackboard-reviewer-simayi", + "reviewer-pangtong": "blackboard-reviewer-pangtong", + "planner": "blackboard-planner", + "claim": "blackboard-claim", + } + + # 默认从环境变量或配置读取，fallback 到默认路径 + SKILL_BASE_PATH = os.environ.get( + "MOZI_SKILL_PATH", + os.path.expanduser("~/.sanguo_projects/sanguo_mozi/skills") + ) + + def __init__(self, max_tokens: int = 4096): self.max_tokens = max_tokens - def build( - self, - role: str, - task_context: Optional[Dict[str, Any]] = None, - project_context: Optional[Dict[str, Any]] = None, - guardrail_rules: Optional[str] = None, - review_protocols: Optional[str] = None, - experiences: Optional[List[Dict[str, Any]]] = None, - skill_descriptions: Optional[List[Dict[str, Any]]] = None, - ) -> str: - """构建 bootstrap 文本 + def build(self, task: dict, role: str) -> str: + """构建 4 段 bootstrap 文本 Args: - role: 角色（executor / reviewer / planner / pangtong） - task_context: 任务上下文（黑板数据） - project_context: 项目背景 - guardrail_rules: Guardrail 规则（仅执行者） - review_protocols: 审查协议（执行者+审查者+庞统） - experiences: 相关经验列表 - skill_descriptions: Skill 描述列表 + task: 任务数据字典，需包含 title/description/must_haves/status 等字段 + role: 角色，对应 ROLE_SKILL_MAP 的 key Returns: 拼装好的 bootstrap 文本 """ - layers: List[str] = [] + sections: List[str] = [] - # L2a: 操作规范 - role_template = self._load_template(f"{role}.md") - if role_template: - layers.append(role_template) + # 段 1: 任务上下文 + sections.append(self._format_task_context(task)) - # L2b: 项目背景 - if project_context: - layers.append(self._format_project_context(project_context)) + # 段 2: 前序产出（有依赖时注入） + if task.get("depends_on_outputs"): + sections.append(self._format_prior_outputs(task["depends_on_outputs"])) - # L2c: 任务上下文 - if task_context: - layers.append(self._format_task_context(task_context)) + # 段 3: 角色操作规范全文（通过 ROLE_SKILL_MAP 从 Skill 文件读取） + skill_name = self.ROLE_SKILL_MAP.get(role) + if skill_name: + skill_content = self._read_skill(skill_name) + if skill_content: + sections.append(skill_content) + elif role not in ("discussion",): + logger.warning("No skill mapping for role: %s", role) - # L2d: 前序信息（depends_on 产出摘要） - if task_context and task_context.get("depends_on_outputs"): - layers.append(self._format_depends_on(task_context["depends_on_outputs"])) + # 段 4: 硬约束 + sections.append(self._format_constraints(role)) - # L2e: Guardrail 规则（仅执行者） - if role == "executor" and guardrail_rules: - layers.append(guardrail_rules) + bootstrap = "\n\n---\n\n".join(sections) - # L2f: 审查协议（执行者+审查者+庞统） - if role in ("executor", "reviewer", "pangtong") and review_protocols: - layers.append(review_protocols) - - # L2g: 经验注入 - if experiences: - layers.append(self._format_experiences(experiences)) - - # L3: Skill descriptions - if skill_descriptions: - layers.append(self._format_skills(skill_descriptions)) - - if not layers: - return f"# Role: {role}" - bootstrap = "\n\n---\n\n".join(layers) - - # 检查 token 限制 - tokens = estimate_tokens(bootstrap) - if tokens > self.max_tokens: - logger.warning("Bootstrap exceeds token limit: %d > %d, truncating", - tokens, self.max_tokens) - bootstrap = self._truncate_to_tokens(bootstrap, self.max_tokens) + # Token 预算检查（warn 但不截断） + tokens = max(1, int(len(bootstrap) / CHARS_PER_TOKEN)) + if tokens > 800: + logger.warning("Bootstrap exceeds 800 token budget: %d tokens (role=%s)", + tokens, role) return bootstrap - def _load_template(self, filename: str) -> Optional[str]: - """加载 prompt template""" - if not self.template_dir: - return None - path = self.template_dir / filename - if path.exists(): - return path.read_text(encoding="utf-8") - return None + def build_for_task(self, task: Any, role: str, **kwargs) -> str: + """从 Task 对象构建 bootstrap（便捷方法，兼容旧调用） - def _format_project_context(self, ctx: Dict[str, Any]) -> str: - """格式化项目背景""" - parts = ["## 项目背景"] - if ctx.get("name"): - parts.append(f"项目: {ctx['name']}") - if ctx.get("description"): - parts.append(f"描述: {ctx['description']}") - if ctx.get("agents"): - parts.append(f"团队: {', '.join(ctx['agents'])}") - return "\n".join(parts) + 忽略 kwargs 中的 project_config/experiences 等旧参数。 + must_haves 包含验收标准（不需要 project_context）。 + """ + task_dict = { + "task_id": task.id, + "title": task.title, + "description": task.description, + "must_haves": task.must_haves, + "status": getattr(task, 'status', ''), + } + # depends_on_outputs 如果有的话 + if hasattr(task, 'depends_on_outputs'): + task_dict["depends_on_outputs"] = task.depends_on_outputs - def _format_task_context(self, ctx: Dict[str, Any]) -> str: - """格式化任务上下文""" + return self.build(task=task_dict, role=role) + + def _read_skill(self, skill_name: str) -> str: + """从 Skill 文件读取全文（带 fallback）""" + path = os.path.join(self.SKILL_BASE_PATH, skill_name, "SKILL.md") + try: + with open(path, encoding="utf-8") as f: + return f.read() + except FileNotFoundError: + logger.error("Skill file not found: %s", path) + return "" + + def _format_task_context(self, task: dict) -> str: + """格式化任务上下文（段 1）""" parts = ["## 任务上下文"] - if ctx.get("task_id"): - parts.append(f"任务ID: {ctx['task_id']}") - if ctx.get("title"): - parts.append(f"标题: {ctx['title']}") - if ctx.get("description"): - parts.append(f"描述: {ctx['description']}") - if ctx.get("task_type"): - parts.append(f"类型: {ctx['task_type']}") - if ctx.get("must_haves"): - parts.append(f"必须完成: {ctx['must_haves']}") - if ctx.get("risk_level"): - parts.append(f"风险级别: {ctx['risk_level']}") + if task.get("task_id"): + parts.append(f"任务ID: {task['task_id']}") + if task.get("title"): + parts.append(f"标题: {task['title']}") + if task.get("description"): + parts.append(f"描述: {task['description']}") + if task.get("must_haves"): + parts.append(f"必须完成: {task['must_haves']}") + if task.get("status"): + parts.append(f"当前状态: {task['status']}") return "\n".join(parts) - def _format_depends_on(self, outputs: List[Dict[str, Any]]) -> str: - """格式化前序产出摘要""" + def _format_prior_outputs(self, outputs: list) -> str: + """格式化前序产出摘要（段 2）""" parts = ["## 前序产出"] for out in outputs: parts.append(f"- [{out.get('task_id', '?')}] {out.get('summary', '无摘要')}") return "\n".join(parts) - def _format_experiences(self, experiences: List[Dict[str, Any]]) -> str: - """格式化经验注入""" - parts = ["## 相关经验"] - for exp in experiences[:5]: # 最多 5 条 - parts.append(f"- [{exp.get('category', '?')}] {exp.get('summary', '')}") - return "\n".join(parts) - - def _format_skills(self, skills: List[Dict[str, Any]]) -> str: - """格式化 Skill 描述""" - parts = ["## 可用技能"] - for skill in skills[:10]: - parts.append(f"- **{skill.get('name', '?')}**: {skill.get('description', '')}") - return "\n".join(parts) - - def _truncate_to_tokens(self, text: str, max_tokens: int) -> str: - """截断到指定 token 数""" - max_chars = int(max_tokens * CHARS_PER_TOKEN) - if len(text) <= max_chars: - return text - trunc_marker = "\n\n[... bootstrap truncated]" - return text[:max_chars - len(trunc_marker)] + trunc_marker - - def build_for_task( - self, - task: Any, - role: str, - project_config: Optional[Dict] = None, - experiences: Optional[List[Dict]] = None, - ) -> str: - """从 Task 对象构建 bootstrap（便捷方法）""" - task_context = { - "task_id": task.id, - "title": task.title, - "description": task.description, - "task_type": task.task_type, - "must_haves": task.must_haves, - "risk_level": task.risk_level, - } - - project_context = None - if project_config: - project_context = { - "name": project_config.get("name"), - "description": project_config.get("description"), - "agents": project_config.get("agents", []), - } - - return self.build( - role=role, - task_context=task_context, - project_context=project_context, - experiences=experiences, - ) - + def _format_constraints(self, role: str) -> str: + """格式化硬约束（段 4）""" + constraints = ["## 硬约束"] + if role == "executor": + constraints.extend([ + "- 完成后必须标 review", + "- 产出物不能为空（系统会验证）", + "- handoff comment ≥ 50 字符", + ]) + elif role.startswith("reviewer"): + constraints.extend([ + "- 审查结果必须明确 pass/fail", + "- 评审意见须附证据（文件:行号）", + ]) + elif role == "planner": + constraints.extend([ + "- 需求不清时提问，不要猜", + "- 子任务必须有明确的终态定义", + ]) + return "\n".join(constraints)