125 lines
4.0 KiB
Python
125 lines
4.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Assemble system prompts from run workspace assets and toolkit context."""
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .agent_workspace import load_agent_workspace_config
|
|
from backend.config.bootstrap_config import get_bootstrap_config_for_run
|
|
from .skills_manager import SkillsManager
|
|
from .workspace_manager import RunWorkspaceManager
|
|
|
|
|
|
def _read_file_if_exists(path: Path) -> str:
|
|
if not path.exists() or not path.is_file():
|
|
return ""
|
|
return path.read_text(encoding="utf-8").strip()
|
|
|
|
|
|
def _append_section(parts: list[str], title: str, content: str) -> None:
|
|
content = content.strip()
|
|
if content:
|
|
parts.append(f"## {title}\n{content}")
|
|
|
|
|
|
def _build_skill_metadata_summary(skills_manager: SkillsManager, config_name: str, agent_id: str) -> str:
|
|
"""Create a compact summary of active skills for prompt routing."""
|
|
metadata_items = skills_manager.list_active_skill_metadata(config_name, agent_id)
|
|
if not metadata_items:
|
|
return ""
|
|
|
|
lines: list[str] = [
|
|
"You can use the following active skills. Prefer the most relevant one, then read its SKILL.md if needed for detailed workflow:",
|
|
]
|
|
for item in metadata_items:
|
|
parts = [f"- `{item.skill_name}`"]
|
|
if item.description:
|
|
parts.append(item.description)
|
|
if item.version:
|
|
parts.append(f"version: {item.version}")
|
|
parts.append(f"path: {item.path}")
|
|
lines.append(" | ".join(parts))
|
|
return "\n".join(lines)
|
|
|
|
|
|
def build_agent_system_prompt(
|
|
agent_id: str,
|
|
config_name: str,
|
|
toolkit: Any,
|
|
) -> str:
|
|
"""Build the final system prompt for an agent.
|
|
|
|
Always reads fresh from disk — no caching.
|
|
"""
|
|
sections: list[str] = []
|
|
|
|
skills_manager = SkillsManager()
|
|
asset_dir = skills_manager.get_agent_asset_dir(config_name, agent_id)
|
|
asset_dir.mkdir(parents=True, exist_ok=True)
|
|
workspace_manager = RunWorkspaceManager(project_root=skills_manager.project_root)
|
|
required_files = ["SOUL.md", "PROFILE.md", "AGENTS.md", "POLICY.md", "MEMORY.md"]
|
|
if not all((asset_dir / filename).exists() for filename in required_files):
|
|
workspace_manager.ensure_agent_assets(config_name=config_name, agent_id=agent_id)
|
|
agent_config = load_agent_workspace_config(asset_dir / "agent.yaml")
|
|
bootstrap_config = get_bootstrap_config_for_run(
|
|
skills_manager.project_root,
|
|
config_name,
|
|
)
|
|
|
|
_append_section(
|
|
sections,
|
|
"Bootstrap",
|
|
bootstrap_config.prompt_body,
|
|
)
|
|
|
|
prompt_files = agent_config.prompt_files or [
|
|
"SOUL.md",
|
|
"PROFILE.md",
|
|
"AGENTS.md",
|
|
"POLICY.md",
|
|
"MEMORY.md",
|
|
]
|
|
included_files = set(prompt_files)
|
|
title_map = {
|
|
"SOUL.md": "Soul",
|
|
"PROFILE.md": "Profile",
|
|
"AGENTS.md": "Agent Guide",
|
|
"POLICY.md": "Policy",
|
|
"MEMORY.md": "Memory",
|
|
}
|
|
for filename in prompt_files:
|
|
_append_section(
|
|
sections,
|
|
title_map.get(filename, filename),
|
|
_read_file_if_exists(asset_dir / filename),
|
|
)
|
|
|
|
if "POLICY.md" not in included_files:
|
|
_append_section(
|
|
sections,
|
|
"Policy",
|
|
_read_file_if_exists(asset_dir / "POLICY.md"),
|
|
)
|
|
|
|
skill_prompt = toolkit.get_agent_skill_prompt()
|
|
if skill_prompt:
|
|
_append_section(sections, "Skills", str(skill_prompt))
|
|
|
|
metadata_summary = _build_skill_metadata_summary(
|
|
skills_manager=skills_manager,
|
|
config_name=config_name,
|
|
agent_id=agent_id,
|
|
)
|
|
if metadata_summary:
|
|
_append_section(sections, "Active Skill Catalog", metadata_summary)
|
|
|
|
activated_notes = toolkit.get_activated_notes()
|
|
if activated_notes:
|
|
_append_section(sections, "Tool Usage Notes", str(activated_notes))
|
|
|
|
return "\n\n".join(section for section in sections if section.strip())
|
|
|
|
|
|
def clear_prompt_factory_cache() -> None:
|
|
"""No-op retained for compatibility with runtime reload hooks."""
|