# -*- coding: utf-8 -*- """ Compatibility Layer - Adapters for legacy to EvoAgent migration. Provides: - LegacyAgentAdapter: Wraps old AnalystAgent to work with new interfaces - Migration utilities for gradual adoption """ from typing import Any, Dict, Optional from agentscope.message import Msg from .agent_core import EvoAgent class LegacyAgentAdapter: """ Adapter to make legacy AnalystAgent compatible with EvoAgent interfaces. This allows gradual migration by wrapping existing agents. """ def __init__(self, legacy_agent: Any): """ Initialize adapter. Args: legacy_agent: Legacy AnalystAgent instance """ self._agent = legacy_agent self.agent_id = getattr(legacy_agent, 'agent_id', getattr(legacy_agent, 'name', 'unknown')) self.analyst_type = getattr(legacy_agent, 'analyst_type_key', None) @property def name(self) -> str: """Get agent name.""" return getattr(self._agent, 'name', self.agent_id) @property def toolkit(self) -> Any: """Get agent toolkit.""" return getattr(self._agent, 'toolkit', None) @property def model(self) -> Any: """Get agent model.""" return getattr(self._agent, 'model', None) @property def memory(self) -> Any: """Get agent memory.""" return getattr(self._agent, 'memory', None) async def reply(self, x: Msg = None) -> Msg: """ Delegate to legacy agent's reply method. Args: x: Input message Returns: Response message """ return await self._agent.reply(x) def reload_runtime_assets(self, active_skill_dirs: Optional[list] = None) -> None: """ Reload runtime assets if supported. Args: active_skill_dirs: Optional list of active skill directories """ if hasattr(self._agent, 'reload_runtime_assets'): self._agent.reload_runtime_assets(active_skill_dirs) def to_evo_agent( self, workspace_manager: Optional[Any] = None, enable_tool_guard: bool = False, ) -> EvoAgent: """ Convert legacy agent to EvoAgent. Args: workspace_manager: Optional workspace manager enable_tool_guard: Whether to enable tool guard Returns: New EvoAgent instance with same configuration """ return EvoAgent( agent_id=self.agent_id, model=self.model, formatter=getattr(self._agent, 'formatter', None), toolkit=self.toolkit, workspace_manager=workspace_manager, config=getattr(self._agent, 'config', {}), long_term_memory=getattr(self._agent, 'long_term_memory', None), enable_tool_guard=enable_tool_guard, sys_prompt=getattr(self._agent, '_sys_prompt', None), ) def __getattr__(self, name: str) -> Any: """Delegate unknown attributes to wrapped agent.""" return getattr(self._agent, name) def is_legacy_agent(agent: Any) -> bool: """ Check if an agent is a legacy agent. Args: agent: Agent instance to check Returns: True if legacy agent """ return hasattr(agent, 'analyst_type_key') and not isinstance(agent, EvoAgent) def adapt_agent(agent: Any) -> Any: """ Wrap agent in adapter if it's a legacy agent. Args: agent: Agent instance Returns: Adapted agent or original if already EvoAgent """ if is_legacy_agent(agent): return LegacyAgentAdapter(agent) return agent def adapt_agents(agents: list) -> list: """ Wrap multiple agents in adapters. Args: agents: List of agent instances Returns: List of adapted agents """ return [adapt_agent(agent) for agent in agents]