From bfe3cb7ae66d8cbec938f63748661a4ccc020bb2 Mon Sep 17 00:00:00 2001 From: cfdaily Date: Sun, 17 May 2026 06:03:05 +0800 Subject: [PATCH] auto-sync: 2026-05-17 06:03:05 --- src/daemon/bootstrap.py | 212 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 212 insertions(+) create mode 100644 src/daemon/bootstrap.py diff --git a/src/daemon/bootstrap.py b/src/daemon/bootstrap.py new file mode 100644 index 0000000..839ee9c --- /dev/null +++ b/src/daemon/bootstrap.py @@ -0,0 +1,212 @@ +"""Bootstrap 拼装 — L0-L3 四层上下文构建 + +L0 铁律层（~500 tokens） → Hook 注入，不占 bootstrap +L1 角色层（~2000 tokens） → SOUL.md / IDENTITY.md（Agent 自带） +L2 引擎注入层（~1500 tokens） → prompt_templates 按 role 拼装 +L3 被动参考层（按需加载） → Skill description 四要素 + +Bootstrap 按 role 精确注入：执行者注入 Guardrail + 审查协议， +审查者注入审查协议，庞统注入审查协议。 +""" + +from __future__ import annotations + +import json +import logging +import re +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 + + +def estimate_tokens(text: str) -> int: + """估算文本 token 数""" + return max(1, int(len(text) / CHARS_PER_TOKEN)) + + +class BootstrapBuilder: + """Bootstrap 四层上下文构建器""" + + 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 + 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 文本 + + Args: + role: 角色（executor / reviewer / planner / pangtong） + task_context: 任务上下文（黑板数据） + project_context: 项目背景 + guardrail_rules: Guardrail 规则（仅执行者） + review_protocols: 审查协议（执行者+审查者+庞统） + experiences: 相关经验列表 + skill_descriptions: Skill 描述列表 + + Returns: + 拼装好的 bootstrap 文本 + """ + layers: List[str] = [] + + # L2a: 操作规范 + role_template = self._load_template(f"{role}.md") + if role_template: + layers.append(role_template) + + # L2b: 项目背景 + if project_context: + layers.append(self._format_project_context(project_context)) + + # L2c: 任务上下文 + if task_context: + layers.append(self._format_task_context(task_context)) + + # L2d: 前序信息（depends_on 产出摘要） + if task_context and task_context.get("depends_on_outputs"): + layers.append(self._format_depends_on(task_context["depends_on_outputs"])) + + # L2e: Guardrail 规则（仅执行者） + if role == "executor" and guardrail_rules: + layers.append(guardrail_rules) + + # 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)) + + 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) + + 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 _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) + + def _format_task_context(self, ctx: Dict[str, Any]) -> str: + """格式化任务上下文""" + 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']}") + return "\n".join(parts) + + def _format_depends_on(self, outputs: List[Dict[str, Any]]) -> str: + """格式化前序产出摘要""" + 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 + return text[:max_chars] + "\n\n[... bootstrap truncated]" + + 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, + )