# -*- coding: utf-8 -*-
"""EvoAgent - Core agent implementation for 大时代.
This module provides the main EvoAgent class built on AgentScope's ReActAgent,
with integrated tools, skills, and memory management based on CoPaw design.
Key features:
- Workspace-driven configuration from Markdown files
- Dynamic skill loading from skills/active directories
- Tool-guard security interception
- Hook system for extensibility
- Runtime skill and prompt reloading
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional, Type, TYPE_CHECKING
from agentscope.agent import ReActAgent
from agentscope.memory import InMemoryMemory
from agentscope.message import Msg
from agentscope.tool import Toolkit
from .tool_guard import ToolGuardMixin
from .hooks import (
HookManager,
BootstrapHook,
MemoryCompactionHook,
WorkspaceWatchHook,
HOOK_PRE_REASONING,
)
from ..prompts.builder import (
build_system_prompt_from_workspace,
)
from ..agent_workspace import load_agent_workspace_config
from ..skills_manager import SkillsManager
# Team infrastructure imports (graceful import - may not exist yet)
try:
from backend.agents.team.messenger import AgentMessenger
from backend.agents.team.task_delegator import TaskDelegator
TEAM_INFRA_AVAILABLE = True
except ImportError:
TEAM_INFRA_AVAILABLE = False
AgentMessenger = None
TaskDelegator = None
if TYPE_CHECKING:
from agentscope.formatter import FormatterBase
from agentscope.model import ModelWrapperBase
logger = logging.getLogger(__name__)
class EvoAgent(ToolGuardMixin, ReActAgent):
"""EvoAgent with integrated tools, skills, and memory management.
This agent extends ReActAgent with:
- Workspace-driven configuration from AGENTS.md/SOUL.md/PROFILE.md/etc.
- Dynamic skill loading from skills/active directories
- Tool-guard security interception (via ToolGuardMixin)
- Hook system for extensibility (bootstrap, memory compaction)
- Runtime skill and prompt reloading
MRO note
~~~~~~~~
``ToolGuardMixin`` overrides ``_acting`` and ``_reasoning`` via
Python's MRO: EvoAgent → ToolGuardMixin → ReActAgent.
Example:
agent = EvoAgent(
agent_id="fundamentals_analyst",
config_name="smoke_fullstack",
workspace_dir=Path("runs/smoke_fullstack/agents/fundamentals_analyst"),
model=model_instance,
formatter=formatter_instance,
)
"""
def __init__(
self,
agent_id: str,
config_name: str,
workspace_dir: Path,
model: "ModelWrapperBase",
formatter: "FormatterBase",
skills_manager: Optional[SkillsManager] = None,
sys_prompt: Optional[str] = None,
max_iters: int = 10,
memory: Optional[Any] = None,
long_term_memory: Optional[Any] = None,
long_term_memory_mode: str = "static_control",
enable_tool_guard: bool = True,
enable_bootstrap_hook: bool = True,
enable_memory_compaction: bool = False,
memory_manager: Optional[Any] = None,
memory_compact_threshold: Optional[int] = None,
env_context: Optional[str] = None,
prompt_files: Optional[List[str]] = None,
# Portfolio manager specific parameters
initial_cash: Optional[float] = None,
margin_requirement: Optional[float] = None,
):
"""Initialize EvoAgent.
Args:
agent_id: Unique identifier for this agent
config_name: Run configuration name (e.g., "smoke_fullstack")
workspace_dir: Agent workspace directory containing markdown files
model: LLM model instance
formatter: Message formatter instance
skills_manager: Optional SkillsManager instance
sys_prompt: Optional override for system prompt
max_iters: Maximum reasoning-acting iterations
memory: Optional memory instance (defaults to InMemoryMemory)
enable_tool_guard: Enable tool-guard security interception
enable_bootstrap_hook: Enable bootstrap guidance on first interaction
enable_memory_compaction: Enable automatic memory compaction
memory_manager: Optional memory manager for compaction
memory_compact_threshold: Token threshold for memory compaction
env_context: Optional environment context to prepend to system prompt
prompt_files: List of markdown files to load (defaults to standard set)
"""
self.agent_id = agent_id
self.config_name = config_name
self.workspace_dir = Path(workspace_dir)
self.workspace_id = config_name
self.config = {"config_name": config_name}
self._skills_manager = skills_manager or SkillsManager()
self._env_context = env_context
self._prompt_files = prompt_files
# Initialize tool guard
if enable_tool_guard:
self._init_tool_guard()
# Load agent configuration from workspace
self._agent_config = self._load_agent_config()
# Build or use provided system prompt
if sys_prompt is not None:
self._sys_prompt = sys_prompt
else:
self._sys_prompt = self._build_system_prompt()
# Create toolkit with skills
toolkit = self._create_toolkit()
# Initialize hook manager
self._hook_manager = HookManager()
# Build kwargs for parent ReActAgent
kwargs = {
"name": agent_id,
"model": model,
"sys_prompt": self._sys_prompt,
"toolkit": toolkit,
"memory": memory or InMemoryMemory(),
"formatter": formatter,
"max_iters": max_iters,
}
# Add long-term memory if provided
if long_term_memory:
kwargs["long_term_memory"] = long_term_memory
kwargs["long_term_memory_mode"] = long_term_memory_mode
# Initialize parent ReActAgent
super().__init__(**kwargs)
# Register hooks
self._register_hooks(
enable_bootstrap=enable_bootstrap_hook,
enable_memory_compaction=enable_memory_compaction,
memory_manager=memory_manager,
memory_compact_threshold=memory_compact_threshold,
)
# Initialize team infrastructure if available
self._messenger: Optional["AgentMessenger"] = None
self._task_delegator: Optional["TaskDelegator"] = None
if TEAM_INFRA_AVAILABLE:
self._init_team_infrastructure()
logger.info(
"EvoAgent initialized: %s (workspace: %s)",
agent_id,
workspace_dir,
)
def _load_agent_config(self) -> Dict[str, Any]:
"""Load agent configuration from workspace.
Returns:
Agent configuration dictionary
"""
config_path = self.workspace_dir / "agent.yaml"
if config_path.exists():
loaded = load_agent_workspace_config(config_path)
return dict(loaded.values)
return {}
def _build_system_prompt(self) -> str:
"""Build system prompt from workspace markdown files.
Uses PromptBuilder to load and combine AGENTS.md, SOUL.md,
PROFILE.md, and other configured files.
Returns:
Complete system prompt string
"""
prompt = build_system_prompt_from_workspace(
workspace_dir=self.workspace_dir,
enabled_files=self._prompt_files,
agent_id=self.agent_id,
extra_context=self._env_context,
)
return prompt
def _create_toolkit(self) -> Toolkit:
"""Create and populate toolkit with agent skills.
Loads skills from the agent's active skills directory and
registers them with the toolkit.
Returns:
Configured Toolkit instance
"""
toolkit = Toolkit(
agent_skill_instruction=(
"You have access to specialized skills. "
"Each skill lives in a directory and is described by SKILL.md. "
"Follow the skill instructions when they are relevant to the current task."
""
),
agent_skill_template="- {name} (dir: {dir}): {description}",
)
# Register skills from active directory
active_skills_dir = self._skills_manager.get_agent_active_root(
self.config_name,
self.agent_id,
)
if active_skills_dir.exists():
for skill_dir in sorted(active_skills_dir.iterdir()):
if skill_dir.is_dir() and (skill_dir / "SKILL.md").exists():
try:
toolkit.register_agent_skill(str(skill_dir))
logger.debug("Registered skill: %s", skill_dir.name)
except Exception as e:
logger.error(
"Failed to register skill '%s': %s",
skill_dir.name,
e,
)
return toolkit
def _register_hooks(
self,
enable_bootstrap: bool,
enable_memory_compaction: bool,
memory_manager: Optional[Any],
memory_compact_threshold: Optional[int],
) -> None:
"""Register agent hooks.
Args:
enable_bootstrap: Enable bootstrap hook
enable_memory_compaction: Enable memory compaction hook
memory_manager: Memory manager instance
memory_compact_threshold: Token threshold for compaction
"""
# Bootstrap hook - checks BOOTSTRAP.md on first interaction
if enable_bootstrap:
bootstrap_hook = BootstrapHook(
workspace_dir=self.workspace_dir,
language="zh",
)
self._hook_manager.register(
hook_type=HOOK_PRE_REASONING,
hook_name="bootstrap",
hook=bootstrap_hook,
)
logger.debug("Registered bootstrap hook")
# Memory compaction hook
if enable_memory_compaction and memory_manager is not None:
compaction_hook = MemoryCompactionHook(
memory_manager=memory_manager,
memory_compact_threshold=memory_compact_threshold,
)
self._hook_manager.register(
hook_type=HOOK_PRE_REASONING,
hook_name="memory_compaction",
hook=compaction_hook,
)
logger.debug("Registered memory compaction hook")
# Workspace watch hook - auto-reload markdown files on change
workspace_watch_hook = WorkspaceWatchHook(
workspace_dir=self.workspace_dir,
)
self._hook_manager.register(
hook_type=HOOK_PRE_REASONING,
hook_name="workspace_watch",
hook=workspace_watch_hook,
)
logger.debug("Registered workspace watch hook")
async def _reasoning(self, tool_choice: Optional[str] = None, **kwargs) -> Msg:
"""Override reasoning to execute pre-reasoning hooks.
Args:
tool_choice: Optional tool choice for structured output
**kwargs: Additional arguments for reasoning
Returns:
Response message
"""
# Execute pre-reasoning hooks
kwargs = await self._hook_manager.execute(
hook_type=HOOK_PRE_REASONING,
agent=self,
kwargs=kwargs,
)
# Call parent (which may be ToolGuardMixin's _reasoning)
return await super()._reasoning(tool_choice=tool_choice, **kwargs)
def reload_runtime_assets(self, active_skill_dirs: Optional[List[Path]] = None) -> None:
"""Reload toolkit and system prompt from current run assets.
Refreshes prompt files from workspace config and rebuilds the toolkit.
"""
# Rebuild system prompt (also refreshes _agent_config and _prompt_files)
self.rebuild_sys_prompt()
# Reload skills/toolkit
self.reload_skills(active_skill_dirs=active_skill_dirs)
def reload_skills(self, active_skill_dirs: Optional[List[Path]] = None) -> None:
"""Reload skills at runtime.
Rebuilds the toolkit with current skills from the active directory.
Args:
active_skill_dirs: Optional list of specific skill directories to load
"""
logger.info("Reloading skills for agent: %s", self.agent_id)
# Create new toolkit
new_toolkit = Toolkit(
agent_skill_instruction=(
"You have access to specialized skills. "
"Each skill lives in a directory and is described by SKILL.md. "
"Follow the skill instructions when they are relevant to the current task."
""
),
agent_skill_template="- {name} (dir: {dir}): {description}",
)
# Register skills
if active_skill_dirs is None:
active_skills_dir = self._skills_manager.get_agent_active_root(
self.config_name,
self.agent_id,
)
if active_skills_dir.exists():
active_skill_dirs = [
d for d in active_skills_dir.iterdir()
if d.is_dir() and (d / "SKILL.md").exists()
]
else:
active_skill_dirs = []
for skill_dir in active_skill_dirs:
if skill_dir.exists() and (skill_dir / "SKILL.md").exists():
try:
new_toolkit.register_agent_skill(str(skill_dir))
logger.debug("Reloaded skill: %s", skill_dir.name)
except Exception as e:
logger.error(
"Failed to reload skill '%s': %s",
skill_dir.name,
e,
)
# Replace toolkit
self.toolkit = new_toolkit
logger.info("Skills reloaded for agent: %s", self.agent_id)
def _make_decision(
self,
ticker: str,
action: str,
quantity: int,
confidence: int = 50,
reasoning: str = "",
) -> "ToolResponse":
"""Record a trading decision for a ticker (PM agent compatibility).
Args:
ticker: Stock ticker symbol (e.g., "AAPL")
action: Decision - "long", "short" or "hold"
quantity: Number of shares to trade (0 for hold)
confidence: Confidence level 0-100
reasoning: Explanation for this decision
Returns:
ToolResponse confirming decision recorded
"""
from agentscope.message import TextBlock
from agentscope.tool import ToolResponse
if action not in ["long", "short", "hold"]:
return ToolResponse(
content=[
TextBlock(
type="text",
text=f"Invalid action: {action}. Must be 'long', 'short', or 'hold'.",
),
],
)
# Store decision in metadata for retrieval
if not hasattr(self, "_decisions"):
self._decisions = {}
self._decisions[ticker] = {
"action": action,
"quantity": quantity if action != "hold" else 0,
"confidence": confidence,
"reasoning": reasoning,
}
return ToolResponse(
content=[
TextBlock(
type="text",
text=f"Decision recorded: {action} {quantity} shares of {ticker} "
f"(confidence: {confidence}%)",
),
],
)
def get_decisions(self) -> Dict[str, Dict]:
"""Get decisions from current cycle (PM compatibility)."""
return getattr(self, "_decisions", {}).copy()
def get_portfolio_state(self) -> Dict[str, Any]:
"""Get current portfolio state (PM compatibility)."""
return getattr(self, "_portfolio", {}).copy()
def load_portfolio_state(self, portfolio: Dict[str, Any]) -> None:
"""Load portfolio state (PM compatibility).
Args:
portfolio: Portfolio state dict with cash, positions, margin_used
"""
if not portfolio:
return
if not hasattr(self, "_portfolio"):
self._portfolio = {
"cash": 100000.0,
"positions": {},
"margin_used": 0.0,
"margin_requirement": 0.25,
}
self._portfolio = {
"cash": portfolio.get("cash", self._portfolio["cash"]),
"positions": portfolio.get("positions", {}).copy(),
"margin_used": portfolio.get("margin_used", 0.0),
"margin_requirement": portfolio.get(
"margin_requirement",
self._portfolio["margin_requirement"],
),
}
def update_portfolio(self, portfolio: Dict[str, Any]) -> None:
"""Update portfolio after external execution (PM compatibility).
Args:
portfolio: Portfolio updates to apply
"""
if not hasattr(self, "_portfolio"):
self._portfolio = {
"cash": 100000.0,
"positions": {},
"margin_used": 0.0,
"margin_requirement": 0.25,
}
self._portfolio.update(portfolio)
def rebuild_sys_prompt(self) -> None:
"""Rebuild and replace the system prompt at runtime.
Useful after updating AGENTS.md, SOUL.md, PROFILE.md, etc.
to ensure the prompt reflects the latest configuration.
Updates both self._sys_prompt and the first system-role
message stored in self.memory.content.
"""
logger.info("Rebuilding system prompt for agent: %s", self.agent_id)
# Reload agent config in case it changed
self._agent_config = self._load_agent_config()
# Refresh prompt_files from updated config
if "prompt_files" in self._agent_config:
self._prompt_files = list(self._agent_config["prompt_files"])
# Rebuild prompt
self._sys_prompt = self._build_system_prompt()
# Update memory if system message exists
if hasattr(self, "memory") and self.memory.content:
for msg, _marks in self.memory.content:
if getattr(msg, "role", None) == "system":
msg.content = self._sys_prompt
logger.debug("Updated system message in memory")
break
logger.info("System prompt rebuilt for agent: %s", self.agent_id)
async def reply(
self,
msg: Msg | List[Msg] | None = None,
structured_model: Optional[Type[Any]] = None,
) -> Msg:
"""Process a message and return a response.
Args:
msg: Input message(s) from user
structured_model: Optional pydantic model for structured output
Returns:
Response message
"""
# Handle list of messages
if isinstance(msg, list):
# Process each message in sequence
for m in msg[:-1]:
await self.memory.add(m)
msg = msg[-1] if msg else None
return await super().reply(msg=msg, structured_model=structured_model)
def get_agent_info(self) -> Dict[str, Any]:
"""Get agent information.
Returns:
Dictionary with agent metadata
"""
return {
"agent_id": self.agent_id,
"config_name": self.config_name,
"workspace_dir": str(self.workspace_dir),
"skills_count": len([
s for s in self._skills_manager.list_active_skill_metadata(
self.config_name,
self.agent_id,
)
]),
"registered_hooks": self._hook_manager.list_hooks(),
"team_infra_available": TEAM_INFRA_AVAILABLE,
}
def _init_team_infrastructure(self) -> None:
"""Initialize team infrastructure components (messenger and task delegator).
This method initializes the AgentMessenger for inter-agent communication
and the TaskDelegator for subagent delegation.
"""
if not TEAM_INFRA_AVAILABLE:
return
try:
self._messenger = AgentMessenger()
self._task_delegator = TaskDelegator(agent=self)
logger.debug(
"Team infrastructure initialized for agent: %s",
self.agent_id,
)
except Exception as e:
logger.warning(
"Failed to initialize team infrastructure for %s: %s",
self.agent_id,
e,
)
self._messenger = None
self._task_delegator = None
@property
def messenger(self) -> Optional["AgentMessenger"]:
"""Get the agent's messenger for inter-agent communication.
Returns:
AgentMessenger instance if available, None otherwise
"""
return self._messenger
async def delegate_task(
self,
task_type: str,
task_data: Dict[str, Any],
target_agent: Optional[str] = None,
) -> Dict[str, Any]:
"""Delegate a task to a subagent using the TaskDelegator.
Args:
task_type: Type of task to delegate
task_data: Data/payload for the task
target_agent: Optional specific agent ID to delegate to
Returns:
Dict containing the delegation result
"""
if not TEAM_INFRA_AVAILABLE or self._task_delegator is None:
return {
"success": False,
"error": "Team infrastructure not available",
}
try:
return await self._task_delegator.delegate_task(
task_type=task_type,
task_data=task_data,
target_agent=target_agent,
)
except Exception as e:
logger.error(
"Task delegation failed for %s: %s",
self.agent_id,
e,
)
return {"success": False, "error": str(e)}
__all__ = ["EvoAgent"]