113 lines
3.2 KiB
Python
113 lines
3.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Assemble system prompts from base prompts, run assets, and toolkit context."""
|
|
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
from backend.config.bootstrap_config import get_bootstrap_config_for_run
|
|
from .prompt_loader import PromptLoader
|
|
from .skills_manager import SkillsManager
|
|
|
|
_prompt_loader = PromptLoader()
|
|
|
|
|
|
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_agent_system_prompt(
|
|
agent_id: str,
|
|
config_name: str,
|
|
toolkit: Any,
|
|
analyst_type: Optional[str] = None,
|
|
) -> str:
|
|
"""Build the final system prompt for an agent."""
|
|
sections: list[str] = []
|
|
|
|
if analyst_type:
|
|
personas_config = _prompt_loader.load_yaml_config(
|
|
"analyst",
|
|
"personas",
|
|
)
|
|
persona = personas_config.get(analyst_type, {})
|
|
focus_text = "\n".join(
|
|
f"- {item}" for item in persona.get("focus", [])
|
|
)
|
|
description = persona.get("description", "").strip()
|
|
base_prompt = _prompt_loader.load_prompt(
|
|
"analyst",
|
|
"system",
|
|
variables={
|
|
"analyst_type": persona.get("name", analyst_type),
|
|
"focus": focus_text,
|
|
"description": description,
|
|
},
|
|
)
|
|
elif agent_id == "portfolio_manager":
|
|
base_prompt = _prompt_loader.load_prompt(
|
|
"portfolio_manager",
|
|
"system",
|
|
)
|
|
elif agent_id == "risk_manager":
|
|
base_prompt = _prompt_loader.load_prompt(
|
|
"risk_manager",
|
|
"system",
|
|
)
|
|
else:
|
|
raise ValueError(f"Unsupported agent prompt build for: {agent_id}")
|
|
|
|
sections.append(base_prompt.strip())
|
|
|
|
skills_manager = SkillsManager()
|
|
asset_dir = skills_manager.get_agent_asset_dir(config_name, agent_id)
|
|
asset_dir.mkdir(parents=True, exist_ok=True)
|
|
bootstrap_config = get_bootstrap_config_for_run(
|
|
skills_manager.project_root,
|
|
config_name,
|
|
)
|
|
|
|
_append_section(
|
|
sections,
|
|
"Bootstrap",
|
|
bootstrap_config.prompt_body,
|
|
)
|
|
|
|
_append_section(
|
|
sections,
|
|
"Role",
|
|
_read_file_if_exists(asset_dir / "ROLE.md"),
|
|
)
|
|
_append_section(
|
|
sections,
|
|
"Style",
|
|
_read_file_if_exists(asset_dir / "STYLE.md"),
|
|
)
|
|
_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))
|
|
|
|
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:
|
|
"""Clear cached prompt and YAML templates before hot reload."""
|
|
_prompt_loader.clear_cache()
|