cfdaily
160 lines
5.7 KiB
Python
160 lines
5.7 KiB
Python
"""Bootstrap 拼装 — L2 引擎注入层
|
||
|
||
4 段结构（~800 tokens）：
|
||
1. 任务上下文：title / description / must_haves / status
|
||
2. 前序产出：depends_on 产出摘要 + handoff comment
|
||
3. 角色操作规范全文：通过 ROLE_SKILL_MAP 从 Skill 文件读取
|
||
4. 硬约束：状态流转约束
|
||
|
||
A 类 Skill 由引擎确定性注入全文，不靠 Description 触发。
|
||
"""
|
||
|
||
import logging
|
||
import os
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
logger = logging.getLogger("moziplus-v2.bootstrap")
|
||
|
||
CHARS_PER_TOKEN = 3.0
|
||
|
||
|
||
def estimate_tokens(text: str) -> int:
|
||
"""估算文本 token 数"""
|
||
return max(1, int(len(text) / CHARS_PER_TOKEN))
|
||
|
||
|
||
class BootstrapBuilder:
|
||
"""L2 引擎注入层构建器（v2.1 四段式）"""
|
||
|
||
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, task: dict, role: str) -> str:
|
||
"""构建 4 段 bootstrap 文本
|
||
|
||
Args:
|
||
task: 任务数据字典，需包含 title/description/must_haves/status 等字段
|
||
role: 角色，对应 ROLE_SKILL_MAP 的 key
|
||
|
||
Returns:
|
||
拼装好的 bootstrap 文本
|
||
"""
|
||
sections: List[str] = []
|
||
|
||
# 段 1: 任务上下文
|
||
sections.append(self._format_task_context(task))
|
||
|
||
# 段 2: 前序产出（有依赖时注入）
|
||
if task.get("depends_on_outputs"):
|
||
sections.append(self._format_prior_outputs(task["depends_on_outputs"]))
|
||
|
||
# 段 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)
|
||
|
||
# 段 4: 硬约束
|
||
sections.append(self._format_constraints(role))
|
||
|
||
bootstrap = "\n\n---\n\n".join(sections)
|
||
|
||
# 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 build_for_task(self, task: Any, role: str, **kwargs) -> str:
|
||
"""从 Task 对象构建 bootstrap（便捷方法，兼容旧调用）
|
||
|
||
忽略 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
|
||
|
||
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 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_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_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)
|