cfdaily
12f6ac3b1f
改动： - toolchain_routes.py: issue_assigned/review_result/ci_failure steps 加文档同步 step - toolchain_handler.py: ToolchainConstraintsSection 加 §6 文档同步约束 + Red Flag - bootstrap.py: reviewer 硬约束加需求-设计-编码一致性检查 - ticker.py: 庞统 round review 三问第 2 问加 docs/design 同步确认 - prompt_composer.py: DeliveryChecklistSection 从弱提醒改为强制步骤 Closes #105
202 lines
7.3 KiB
Python
202 lines
7.3 KiB
Python
"""
|
||
prompt_composer.py — PromptSection Protocol + PromptContext + PromptComposer
|
||
|
||
拼装器：有序管理 prompt 段落，按优先级排序后合并为最终 prompt。
|
||
"""
|
||
|
||
import logging
|
||
from dataclasses import dataclass, field
|
||
from typing import Dict, List, Optional, Protocol, runtime_checkable
|
||
|
||
logger = logging.getLogger("moziplus-v2.prompt_composer")
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Section 优先级范围约定
|
||
# ---------------------------------------------------------------------------
|
||
PRIORITY_CONTEXT = 10 # 任务上下文
|
||
PRIORITY_PRIOR = 20 # 前序信息
|
||
PRIORITY_ROLE = 30 # 角色规范
|
||
PRIORITY_API = 40 # API 操作指令
|
||
PRIORITY_CONSTRAINTS = 50 # 硬约束
|
||
PRIORITY_EXTENSION = 60 # 扩展段
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# PromptSection Protocol
|
||
# ---------------------------------------------------------------------------
|
||
@runtime_checkable
|
||
class PromptSection(Protocol):
|
||
"""一个 prompt 段"""
|
||
|
||
name: str # 段名（去重用，同名覆盖）
|
||
priority: int # 排序优先级（小数字=靠前）
|
||
|
||
def render(self, context: "PromptContext") -> str:
|
||
"""渲染此段的文本内容。返回空字符串表示不注入。"""
|
||
...
|
||
|
||
def should_include(self, context: "PromptContext") -> bool:
|
||
"""是否注入此段（默认 True，条件段可覆盖）。"""
|
||
...
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# PromptContext 数据对象
|
||
# ---------------------------------------------------------------------------
|
||
@dataclass
|
||
class PromptContext:
|
||
"""Prompt 渲染的统一上下文"""
|
||
|
||
task_id: str
|
||
title: str
|
||
description: str
|
||
must_haves: str
|
||
project_id: str
|
||
agent_id: str
|
||
|
||
task: Optional[Dict] = None
|
||
role: str = "executor"
|
||
spawn_type: str = "executor"
|
||
|
||
# mail 专用
|
||
from_agent: str = ""
|
||
mail_type: str = "" # inform / request
|
||
|
||
# toolchain 专用
|
||
event_type: str = "" # ci_failure / review_request / ...
|
||
event_data: Dict = field(default_factory=dict)
|
||
action_type: str = "" # 动作分类（review_result / ci_failure / ...）
|
||
action_steps: list = field(default_factory=list) # 结构化编号步骤列表
|
||
|
||
# 前序产出
|
||
depends_on_outputs: Optional[List] = None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# PromptComposer 拼装器
|
||
# ---------------------------------------------------------------------------
|
||
class PromptComposer:
|
||
"""有序拼装 prompt sections"""
|
||
|
||
SEPARATOR = "\n\n---\n\n"
|
||
TOKEN_BUDGET_WARN = 800 # token 预算警告阈值
|
||
CHARS_PER_TOKEN = 3.5 # 估算比率
|
||
|
||
def __init__(self) -> None:
|
||
self._sections: List[PromptSection] = []
|
||
|
||
def add(self, section: PromptSection) -> None:
|
||
"""添加一个 section（同名覆盖）"""
|
||
self._sections = [s for s in self._sections if s.name != section.name]
|
||
self._sections.append(section)
|
||
|
||
def add_many(self, sections: List[PromptSection]) -> None:
|
||
"""批量添加"""
|
||
for s in sections:
|
||
self.add(s)
|
||
|
||
def compose(self, context: PromptContext) -> str:
|
||
"""拼装最终 prompt
|
||
|
||
1. 过滤 should_include=False 的段
|
||
2. 按 priority 排序
|
||
3. 逐段 render
|
||
4. 过滤空段
|
||
5. 用分隔符连接
|
||
6. Token 预算警告（不截断）
|
||
"""
|
||
active = [s for s in self._sections if s.should_include(context)]
|
||
active.sort(key=lambda s: s.priority)
|
||
|
||
parts = [s.render(context) for s in active]
|
||
parts = [p for p in parts if p.strip()]
|
||
|
||
result = self.SEPARATOR.join(parts)
|
||
|
||
# Token 估算
|
||
tokens = max(1, int(len(result) / self.CHARS_PER_TOKEN))
|
||
logger.debug(
|
||
"Composed prompt from %d sections, %d tokens",
|
||
len(parts), tokens,
|
||
)
|
||
|
||
if tokens > self.TOKEN_BUDGET_WARN:
|
||
logger.warning(
|
||
"Prompt exceeds %d token budget: %d tokens (task_id=%s)",
|
||
self.TOKEN_BUDGET_WARN, tokens, context.task_id,
|
||
)
|
||
|
||
return result
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
class GiteaConventionSection:
|
||
"""Gitea 标题规范引导段 — 提醒 Agent 创建 Issue/PR 时遵循标题格式。"""
|
||
|
||
name: str = "gitea_convention"
|
||
priority: int = 55 # CONSTRAINTS(50) 和 EXTENSION(60) 之间
|
||
|
||
CONVENTION_TEXT = (
|
||
"## Gitea 标题规范\n"
|
||
"创建 Issue/PR 时，标题**必须**包含项目代号前缀：\n"
|
||
"- Issue: `[代号] type: 简述`，如 `[moz] bug: Mail API 500`\n"
|
||
"- PR: `[代号] type(scope): 简述`，如 `[moz] impl(daemon): WikiGuideSection 注入`\n"
|
||
"代号：moz=moziplus_v2, quant=quant_live, vnpy=vnpy\n"
|
||
"type: bug/feat/impl/fix/docs/test/ci/refactor/chore"
|
||
)
|
||
|
||
def render(self, context: "PromptContext") -> str:
|
||
return self.CONVENTION_TEXT
|
||
|
||
def should_include(self, context: "PromptContext") -> bool:
|
||
return True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# WikiGuideSection — 知识查询引导段
|
||
# ---------------------------------------------------------------------------
|
||
class WikiGuideSection:
|
||
"""知识查询引导段 — 引导 Agent 在关键决策点查 wiki-vault。"""
|
||
|
||
name: str = "wiki_guide"
|
||
priority: int = 60 # PRIORITY_EXTENSION
|
||
|
||
WIKI_GUIDE = (
|
||
"## 知识查询引导\n"
|
||
"涉及方案设计、编码实现、故障排查时，先查 wiki-vault 相关实践：\n"
|
||
"- 路径：/Volumes/KnowledgeBase/wiki-vault/\n"
|
||
"- 速查：index.md → grep 关键词 → summary 字段 → 按需读全文\n"
|
||
"- 查不到：在 _meta/knowledge-gaps.md 记录"
|
||
)
|
||
|
||
def render(self, context: "PromptContext") -> str:
|
||
return self.WIKI_GUIDE
|
||
|
||
def should_include(self, context: "PromptContext") -> bool:
|
||
return True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# DeliveryChecklistSection — 交付检查清单
|
||
# ---------------------------------------------------------------------------
|
||
class DeliveryChecklistSection:
|
||
"""交付检查清单 — 提醒 Agent 完成前同步关联成果物。"""
|
||
|
||
name: str = "delivery_checklist"
|
||
priority: int = 55 # CONSTRAINTS(50) 和 EXTENSION(60) 之间
|
||
|
||
CHECKLIST_TEXT = (
|
||
"## 交付检查（强制）\n"
|
||
"⚠️ 这是必须执行的步骤，不是提醒。代码改动完成后，以下检查每一项都要有明确结论：\n"
|
||
"- 改了实现 → docs/design/ 对应设计文档是否需要更新？需要则在同一 PR 中更新\n"
|
||
"- 改了实现 → tests/ 是否有对应测试脚本需要更新？需要则在同一 PR 中更新\n"
|
||
"- action report 中必须逐项说明上述检查结果（如「文档无需更新」「测试已补充」）\n"
|
||
"- 所有成果物变更通过 PR 流程：PR review 把关设计合理性，CI 把关代码质量，CD 把关部署正确性\n"
|
||
)
|
||
|
||
def render(self, context: "PromptContext") -> str:
|
||
return self.CHECKLIST_TEXT
|
||
|
||
def should_include(self, context: "PromptContext") -> bool:
|
||
return True
|