"""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, )