242 lines
8.4 KiB
Python
242 lines
8.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Initialize run-scoped agent workspace assets."""
|
|
|
|
from pathlib import Path
|
|
from typing import Dict, Iterable, Optional
|
|
|
|
import yaml
|
|
|
|
from .skills_manager import SkillsManager
|
|
from .team_pipeline_config import ensure_team_pipeline_config
|
|
|
|
|
|
class RunWorkspaceManager:
|
|
"""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)
|
|
ensure_team_pipeline_config(
|
|
project_root=self.project_root,
|
|
config_name=config_name,
|
|
default_analysts=[
|
|
"fundamentals_analyst",
|
|
"technical_analyst",
|
|
"sentiment_analyst",
|
|
"valuation_analyst",
|
|
],
|
|
)
|
|
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)
|
|
(asset_dir / "skills" / "installed").mkdir(parents=True, exist_ok=True)
|
|
(asset_dir / "skills" / "active").mkdir(parents=True, exist_ok=True)
|
|
(asset_dir / "skills" / "disabled").mkdir(parents=True, exist_ok=True)
|
|
(asset_dir / "skills" / "local").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",
|
|
)
|
|
self._ensure_file(
|
|
asset_dir / "SOUL.md",
|
|
"# Soul\n\n"
|
|
"Describe the agent's temperament, reasoning posture, and voice.\n\n",
|
|
)
|
|
self._ensure_file(
|
|
asset_dir / "PROFILE.md",
|
|
"# Profile\n\n"
|
|
"Track this agent's long-lived investment style, preferences, and strengths.\n\n",
|
|
)
|
|
self._ensure_file(
|
|
asset_dir / "AGENTS.md",
|
|
"# Agent Guide\n\n"
|
|
"Document how this agent should work, collaborate, and choose tools or skills.\n\n",
|
|
)
|
|
self._ensure_file(
|
|
asset_dir / "MEMORY.md",
|
|
"# Memory\n\n"
|
|
"Store durable lessons, heuristics, and reminders for this agent.\n\n",
|
|
)
|
|
self._ensure_file(
|
|
asset_dir / "HEARTBEAT.md",
|
|
"# Heartbeat\n\n"
|
|
"Optional checklist for periodic review or self-reflection.\n\n",
|
|
)
|
|
self._ensure_agent_yaml(
|
|
asset_dir / "agent.yaml",
|
|
agent_id=agent_id,
|
|
)
|
|
return asset_dir
|
|
|
|
def load_agent_file(
|
|
self,
|
|
*,
|
|
config_name: str,
|
|
agent_id: str,
|
|
filename: str,
|
|
) -> str:
|
|
"""Load one run-scoped agent workspace file."""
|
|
path = self.get_agent_asset_dir(config_name, agent_id) / filename
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"File not found: {filename}")
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
def update_agent_file(
|
|
self,
|
|
*,
|
|
config_name: str,
|
|
agent_id: str,
|
|
filename: str,
|
|
content: str,
|
|
) -> None:
|
|
"""Write one run-scoped agent workspace file."""
|
|
asset_dir = self.get_agent_asset_dir(config_name, agent_id)
|
|
asset_dir.mkdir(parents=True, exist_ok=True)
|
|
path = asset_dir / filename
|
|
path.write_text(content, encoding="utf-8")
|
|
|
|
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")
|
|
|
|
@staticmethod
|
|
def _ensure_agent_yaml(path: Path, agent_id: str) -> None:
|
|
if path.exists():
|
|
return
|
|
|
|
payload = {
|
|
"agent_id": agent_id,
|
|
"prompt_files": [
|
|
"SOUL.md",
|
|
"PROFILE.md",
|
|
"AGENTS.md",
|
|
"POLICY.md",
|
|
"MEMORY.md",
|
|
],
|
|
"enabled_skills": [],
|
|
"disabled_skills": [],
|
|
"active_tool_groups": [],
|
|
"disabled_tool_groups": [],
|
|
}
|
|
path.write_text(
|
|
yaml.safe_dump(payload, allow_unicode=True, sort_keys=False),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
# Backward-compatible alias: code importing WorkspaceManager from this module should continue to work.
|
|
WorkspaceManager = RunWorkspaceManager
|