141 lines
5.1 KiB
Python
141 lines
5.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Initialize run-scoped agent workspace assets."""
|
|
|
|
from pathlib import Path
|
|
from typing import Dict, Iterable, Optional
|
|
|
|
from .skills_manager import SkillsManager
|
|
|
|
|
|
class WorkspaceManager:
|
|
"""Create and maintain run-level prompt asset files for each agent."""
|
|
|
|
def __init__(self, project_root: Optional[Path] = None):
|
|
self.skills_manager = SkillsManager(project_root=project_root)
|
|
self.project_root = self.skills_manager.project_root
|
|
|
|
def get_run_dir(self, config_name: str) -> Path:
|
|
return self.project_root / "runs" / config_name
|
|
|
|
def ensure_run_workspace(self, config_name: str) -> Path:
|
|
run_dir = self.get_run_dir(config_name)
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
self.skills_manager.ensure_activation_manifest(config_name)
|
|
bootstrap_path = run_dir / "BOOTSTRAP.md"
|
|
if not bootstrap_path.exists():
|
|
bootstrap_path.write_text(
|
|
"---\n"
|
|
"tickers:\n"
|
|
" - AAPL\n"
|
|
" - MSFT\n"
|
|
"initial_cash: 100000\n"
|
|
"margin_requirement: 0.0\n"
|
|
"enable_memory: false\n"
|
|
"max_comm_cycles: 2\n"
|
|
"agent_overrides: {}\n"
|
|
"---\n\n"
|
|
"# Bootstrap\n\n"
|
|
"Use this file to describe run-specific setup notes, preferred tickers,\n"
|
|
"risk bounds, or strategy constraints before the first execution.\n\n"
|
|
"The YAML front matter above is machine-readable runtime configuration.\n"
|
|
"The markdown body below is injected into agent prompts as run context.\n",
|
|
encoding="utf-8",
|
|
)
|
|
return run_dir
|
|
|
|
def bootstrap_path(self, config_name: str) -> Path:
|
|
return self.get_run_dir(config_name) / "BOOTSTRAP.md"
|
|
|
|
def ensure_agent_assets(
|
|
self,
|
|
config_name: str,
|
|
agent_id: str,
|
|
role_seed: str = "",
|
|
style_seed: str = "",
|
|
policy_seed: str = "",
|
|
) -> Path:
|
|
asset_dir = self.skills_manager.get_agent_asset_dir(
|
|
config_name,
|
|
agent_id,
|
|
)
|
|
asset_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
self._ensure_file(
|
|
asset_dir / "ROLE.md",
|
|
"# Role\n\n"
|
|
"Optional run-scoped role override.\n\n"
|
|
f"{role_seed}".strip()
|
|
+ "\n",
|
|
)
|
|
self._ensure_file(
|
|
asset_dir / "STYLE.md",
|
|
"# Style\n\n"
|
|
"Optional run-scoped communication or reasoning style.\n\n"
|
|
f"{style_seed}".strip()
|
|
+ "\n",
|
|
)
|
|
self._ensure_file(
|
|
asset_dir / "POLICY.md",
|
|
"# Policy\n\n"
|
|
"Optional run-scoped constraints, limits, or strategy policy.\n\n"
|
|
f"{policy_seed}".strip()
|
|
+ "\n",
|
|
)
|
|
return asset_dir
|
|
|
|
def initialize_default_assets(
|
|
self,
|
|
config_name: str,
|
|
agent_ids: Iterable[str],
|
|
analyst_personas: Optional[Dict[str, Dict]] = None,
|
|
) -> None:
|
|
self.ensure_run_workspace(config_name)
|
|
analyst_personas = analyst_personas or {}
|
|
|
|
for agent_id in agent_ids:
|
|
if agent_id.endswith("_analyst"):
|
|
persona = analyst_personas.get(agent_id, {})
|
|
role_seed = persona.get("description", "").strip()
|
|
focus_items = persona.get("focus", [])
|
|
style_seed = "\n".join(f"- {item}" for item in focus_items)
|
|
policy_seed = (
|
|
"State a clear signal, confidence, and the conditions that would invalidate the thesis."
|
|
)
|
|
elif agent_id == "portfolio_manager":
|
|
role_seed = (
|
|
"Synthesize analyst and risk inputs into explicit portfolio decisions."
|
|
)
|
|
style_seed = (
|
|
"Be concise, capital-aware, and explicit about sizing rationale."
|
|
)
|
|
policy_seed = (
|
|
"Respect cash, margin, and portfolio concentration constraints before recording decisions."
|
|
)
|
|
elif agent_id == "risk_manager":
|
|
role_seed = (
|
|
"Quantify concentration, leverage, liquidity, and volatility risk before trade execution."
|
|
)
|
|
style_seed = (
|
|
"Prioritize the highest-severity risk first and state concrete limits."
|
|
)
|
|
policy_seed = (
|
|
"Use available risk tools before issuing the final risk memo."
|
|
)
|
|
else:
|
|
role_seed = ""
|
|
style_seed = ""
|
|
policy_seed = ""
|
|
|
|
self.ensure_agent_assets(
|
|
config_name=config_name,
|
|
agent_id=agent_id,
|
|
role_seed=role_seed,
|
|
style_seed=style_seed,
|
|
policy_seed=policy_seed,
|
|
)
|
|
|
|
@staticmethod
|
|
def _ensure_file(path: Path, content: str) -> None:
|
|
if not path.exists():
|
|
path.write_text(content, encoding="utf-8")
|