butler/orchestrator/prompt_assembler.py
Rama main · réplica de lectura del repositorio original.
"""System prompt assembly for ButlerOrchestrator (ENG-12)."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, cast
from butler.model_resolve import resolve_effective_model
from butler.workflows.loader import format_workflows_for_prompt
from butler.orchestrator.prompt_assembler_ops import import_system_reminder_safe
from butler.core.prompt_renderer import render_orchestrator_turn
from butler.orchestrator.prompt_assembler_ops import static_system_reminder_enabled_safe
from butler.agent_profiles import get_profile
from butler.project.meta import lifecycle_operating_hint
from butler.agent_profiles import get_model_aware_prompt_extra
if TYPE_CHECKING:
from butler.orchestrator import ButlerOrchestrator
from butler.orchestrator import memory_bridge
from butler.orchestrator.templates import (
format_project_list,
format_skill_summaries,
lead_template_path,
read_template_cached,
template_path,
)
logger = logging.getLogger(__name__)
_MEMORY_OFFLINE_APPENDIX = (
"\n\n## 记忆子系统离线 (memory offline)\n"
"本次会话无法 recall 历史上下文,所有"
" recall 类工具将返回空结果。"
"请主动告知用户\"本次对话无历史记忆\","
"避免产生已记住对方偏好的错觉。"
)
def system_prompt_placeholders(
orch: ButlerOrchestrator,
*,
for_role: str = "default",
) -> dict[str, str]:
template = template_path()
body = read_template_cached(template)
if not body:
if not template.exists():
logger.warning("Butler system template missing at %s", template)
current = orch.project_manager.current_project or "(未选择)"
project_list = format_project_list(orch._settings)
proj = orch.project_manager.get_current()
em = resolve_effective_model("butler", project=proj, settings=orch._settings)
mc = em.config
prov = mc.provider or orch._settings.default_provider or ""
parts_model = [
f"- Provider: `{prov or 'unset'}`",
f"- Model: `{mc.model or 'unset'}`",
f"- Sources: `{', '.join(em.sources)}`",
]
if mc.temperature is not None:
parts_model.append(f"- Temperature: `{mc.temperature}`")
if mc.max_tokens is not None:
parts_model.append(f"- Max tokens: `{mc.max_tokens}`")
model_block = "\n".join(parts_model)
skills_block = "(技能管理器不可用)"
if orch._skill_manager is not None:
skills_block = format_skill_summaries(orch._skill_manager.list_skills())
memory_ctx = memory_bridge.build_memory_context(orch, for_role=for_role)
workflows_block = format_workflows_for_prompt(orch.project_manager.get_current())
return {
"template_body": body,
"butler_name": orch._settings.butler_name,
"owner_name": orch._settings.owner_name,
"current_project": current,
"project_list": project_list,
"memory_context": memory_ctx,
"butler_model": model_block,
"skill_summaries": skills_block,
"workflows_block": workflows_block,
"model_block": model_block,
}
def build_dynamic_system_reminder(
orch: ButlerOrchestrator,
*,
for_role: str = "default",
) -> str:
"""Dynamic context injected into the user turn when static system mode is on."""
wrap_system_reminder = import_system_reminder_safe()
if wrap_system_reminder is None:
return ""
ph = system_prompt_placeholders(orch, for_role=for_role)
chunks = [
"## 记忆与上下文\n" + (ph.get("memory_context") or "(无)"),
"## 可用技能概要\n" + ph.get("skill_summaries", ""),
"## 项目工作流\n" + ph.get("workflows_block", ""),
]
if orch.memory_offline:
chunks.append(_MEMORY_OFFLINE_APPENDIX.lstrip("\n"))
body = "\n\n".join(c for c in chunks if c and str(c).strip())
return cast(str, wrap_system_reminder(body))
def build_static_system_prompt(orch: ButlerOrchestrator) -> str:
"""System prompt without dynamic memory (paired with build_dynamic_system_reminder)."""
ph = system_prompt_placeholders(orch, for_role="default")
body = ph.get("template_body") or ""
placeholders = {
"butler_name": ph["butler_name"],
"owner_name": ph["owner_name"],
"current_project": ph["current_project"],
"project_list": ph["project_list"],
"memory_context": "(见本轮 system-reminder)",
"butler_model": ph["butler_model"],
"skill_summaries": "(见本轮 system-reminder)",
}
rendered = body
for k, v in placeholders.items():
rendered = rendered.replace("{" + k + "}", v)
appendix = (
"\n\n## Butler 模型\n"
f"{ph['model_block']}\n\n"
"## 可用技能概要\n"
"(见本轮 system-reminder)\n\n"
"## 项目工作流\n"
"(见本轮 system-reminder)"
)
return rendered.rstrip() + appendix
def resolve_system_prompt(
orch: ButlerOrchestrator,
*,
role: str = "butler",
session_key: str = "",
) -> tuple[str, str | None]:
"""Return (system_prompt, optional user-side reminder)."""
if role == "lead":
return build_lead_system_prompt(orch, session_key=session_key), None
return cast(tuple[str, str | None], render_orchestrator_turn(orch, for_role=role))
def assemble_default_system_prompt(
orch: ButlerOrchestrator,
*,
for_role: str = "default",
) -> str:
ph = system_prompt_placeholders(orch, for_role=for_role)
body = ph.get("template_body") or ""
placeholders = {
"butler_name": ph["butler_name"],
"owner_name": ph["owner_name"],
"current_project": ph["current_project"],
"project_list": ph["project_list"],
"memory_context": ph["memory_context"],
"butler_model": ph["butler_model"],
"skill_summaries": ph["skill_summaries"],
}
rendered = body
for k, v in placeholders.items():
rendered = rendered.replace("{" + k + "}", v)
offline_appendix = _MEMORY_OFFLINE_APPENDIX if orch.memory_offline else ""
appendix = (
"\n\n## Butler 模型\n"
f"{ph['model_block']}\n\n"
"## 可用技能概要\n"
f"{ph['skill_summaries']}\n\n"
"## 项目工作流\n"
f"{ph['workflows_block']}"
) + offline_appendix
return rendered.rstrip() + appendix
def build_system_prompt(orch: ButlerOrchestrator) -> str:
if static_system_reminder_enabled_safe():
return build_static_system_prompt(orch)
return assemble_default_system_prompt(orch, for_role="default")
def build_lead_system_prompt(
orch: ButlerOrchestrator,
*,
session_key: str = "",
) -> str:
"""System prompt for project Lead gateway loop (phase 2)."""
path = lead_template_path()
body = read_template_cached(path)
if not body:
if not path.exists():
logger.warning("Lead system template missing at %s", path)
prof = get_profile("lead")
body = prof.system_prompt if prof else ""
current = orch.project_manager.resolve_active_project_name(
session_key=session_key,
) or orch.project_manager.current_project or "(未选择)"
memory_ctx = memory_bridge.build_memory_context(orch, for_role="lead")
workflows_block = format_workflows_for_prompt(
orch.project_manager.get_current(session_key=session_key or "")
)
proj = orch.project_manager.get_current(session_key=session_key or "")
lifecycle_block = lifecycle_operating_hint(proj)
placeholders = {
"current_project": current,
"memory_context": memory_ctx or "(无)",
"workflows_block": workflows_block or "(无)",
"lifecycle_block": lifecycle_block or "",
}
rendered = body
for k, v in placeholders.items():
rendered = rendered.replace("{" + k + "}", v)
if lifecycle_block and "{lifecycle_block}" not in body:
rendered = rendered.rstrip() + "\n\n" + lifecycle_block
profile = get_profile("lead")
if profile:
mc = orch._model_credentials("butler")
prov = mc.get("provider") or ""
extra = get_model_aware_prompt_extra(str(prov))
if extra:
rendered += "\n\n" + extra
return cast(str, rendered.rstrip())
__all__ = [
"assemble_default_system_prompt",
"build_dynamic_system_reminder",
"build_lead_system_prompt",
"build_static_system_prompt",
"build_system_prompt",
"resolve_system_prompt",
"system_prompt_placeholders",
]