diff --git a/src/agentman/frameworks/__init__.py b/src/agentman/frameworks/__init__.py index ba30373..5bbd8d8 100644 --- a/src/agentman/frameworks/__init__.py +++ b/src/agentman/frameworks/__init__.py @@ -1,7 +1,7 @@ """Framework support for AgentMan.""" -from .base import BaseFramework from .agno import AgnoFramework +from .base import BaseFramework from .fast_agent import FastAgentFramework __all__ = ["BaseFramework", "AgnoFramework", "FastAgentFramework"] diff --git a/src/agentman/frameworks/agno.py b/src/agentman/frameworks/agno.py index 5cad9cb..69dd97e 100644 --- a/src/agentman/frameworks/agno.py +++ b/src/agentman/frameworks/agno.py @@ -2,6 +2,7 @@ from typing import List +from .agno_builder import AgnoAgentConfig, AgnoCodeGenerator, AgnoConfigBuilder, AgnoFrameworkConfig, AgnoTeamConfig from .base import BaseFramework @@ -10,358 +11,70 @@ class AgnoFramework(BaseFramework): def build_agent_content(self) -> str: """Build the Python agent file content for Agno framework.""" - lines = [] - # Determine if we need advanced features - has_multiple_agents = len(self.config.agents) > 1 - has_servers = bool(self.config.servers) + # Create configuration builder + builder = AgnoConfigBuilder() - # Enhanced imports based on features needed - imports = [ - "import os", - "from agno.agent import Agent", - ] + # Create main framework configuration + framework_config = AgnoFrameworkConfig(has_prompt_file=self.has_prompt_file) - # Add dotenv import for loading .env files - imports.append("from dotenv import load_dotenv") - imports.append("") - imports.append("# Load environment variables from .env file") - imports.append("load_dotenv()") - imports.append("") - - # Model imports - default_model = self.config.default_model or "" - if "anthropic" in default_model.lower() or "claude" in default_model.lower(): - imports.append("from agno.models.anthropic import Claude") - elif "openai" in default_model.lower() or "gpt" in default_model.lower(): - imports.append("from agno.models.openai import OpenAILike") - elif "/" in default_model: - # Custom model with provider prefix (e.g., "ollama/llama3", "groq/mixtral") - imports.append("from agno.models.openai import OpenAILike") - - # Check agent models to determine what imports we need - for agent in self.config.agents.values(): - agent_model = agent.model or default_model - if agent_model: - if "anthropic" in agent_model.lower() or "claude" in agent_model.lower(): - if "from agno.models.anthropic import Claude" not in imports: - imports.append("from agno.models.anthropic import Claude") - elif "openai" in agent_model.lower() or "gpt" in agent_model.lower(): - if "from agno.models.openai import OpenAILike" not in imports: - imports.append("from agno.models.openai import OpenAILike") - elif "/" in agent_model: - # Custom model with provider prefix - if "from agno.models.openai import OpenAILike" not in imports: - imports.append("from agno.models.openai import OpenAILike") - - if not any("anthropic" in imp or "openai" in imp for imp in imports): - # Default to both if model is not specified or unclear - imports.extend([ - "from agno.models.openai import OpenAILike", - "from agno.models.anthropic import Claude", - ]) - - # Tool imports based on servers - tool_imports = [] - if has_servers: - # Map server types to appropriate tools - for server_name, server in self.config.servers.items(): - if server_name in ["web_search", "search", "browser"]: - tool_imports.append("from agno.tools.duckduckgo import DuckDuckGoTools") - elif server_name in ["finance", "yfinance", "stock"]: - tool_imports.append("from agno.tools.yfinance import YFinanceTools") - elif server_name in ["file", "filesystem"]: - tool_imports.append("from agno.tools.file import FileTools") - elif server_name in ["shell", "terminal"]: - tool_imports.append("from agno.tools.shell import ShellTools") - elif server_name in ["python", "code"]: - tool_imports.append("from agno.tools.python import PythonTools") - - # Remove duplicates and add to imports - for tool_import in sorted(set(tool_imports)): - imports.append(tool_import) - - # Team imports if multiple agents - if has_multiple_agents: - imports.append("from agno.team.team import Team") - - # Advanced feature imports (always include for better examples) - imports.extend([ - "from agno.tools.reasoning import ReasoningTools", - "# Optional: Uncomment for advanced features", - "# from agno.storage.sqlite import SqliteStorage", - "# from agno.memory.v2.db.sqlite import SqliteMemoryDb", - "# from agno.memory.v2.memory import Memory", - "# from agno.knowledge.url import UrlKnowledge", - "# from agno.vectordb.lancedb import LanceDb", - ]) - - lines.extend(imports + [""]) - - # Generate agents with enhanced capabilities + # Build agent configurations agent_vars = [] for agent in self.config.agents.values(): agent_var = f"{agent.name.lower().replace('-', '_')}_agent" - agent_vars.append((agent_var, agent)) + agent_vars.append(agent_var) + + # Build model configuration + model = agent.model or self.config.default_model + model_config = builder.build_model_config(model) + + # Build tools + tools = builder.build_tools_for_servers(agent.servers) - lines.extend([ - f"# Agent: {agent.name}", - f"{agent_var} = Agent(", - f' name="{agent.name}",', - f' instructions="""{agent.instruction}""",', - ]) + # Add tool imports + framework_config.tool_imports.update(builder.get_tool_imports(tools)) - # Add role if we have multiple agents - if has_multiple_agents: + # Create role if multiple agents + role = None + if len(self.config.agents) > 1: role = f"Handle {agent.name.lower().replace('-', ' ')} requests" - lines.append(f' role="{role}",') - # Add model - model = agent.model or self.config.default_model - if model: - model_code = self._generate_model_code(model) - lines.append(f' {model_code}') - - # Enhanced tools based on servers - tools = [] - if agent.servers: - for server_name in agent.servers: - if server_name in ["web_search", "search", "browser"]: - tools.append("DuckDuckGoTools()") - elif server_name in ["finance", "yfinance", "stock"]: - tools.append("YFinanceTools(stock_price=True, analyst_recommendations=True)") - elif server_name in ["file", "filesystem"]: - tools.append("FileTools()") - elif server_name in ["shell", "terminal"]: - tools.append("ShellTools()") - elif server_name in ["python", "code"]: - tools.append("PythonTools()") - - # Always add reasoning tools for better performance - tools.append("ReasoningTools(add_instructions=True)") - - if tools: - tools_str = ", ".join(tools) - lines.append(f' tools=[{tools_str}],') - - # Add other properties - if not agent.use_history: - lines.append(" add_history_to_messages=False,") - else: - lines.append(" add_history_to_messages=True,") - - if agent.human_input: - lines.append(" human_input=True,") - - # Enhanced agent properties - lines.extend([ - " markdown=True,", - " add_datetime_to_instructions=True,", - " # Optional: Enable advanced features", - " # storage=SqliteStorage(table_name='agent_sessions', db_file='tmp/agent.db'),", - " # memory=Memory(model=Claude(id='claude-sonnet-4-20250514'), db=SqliteMemoryDb()),", - " # enable_agentic_memory=True,", - ")", - "" - ]) - - # Team creation for multi-agent scenarios - if has_multiple_agents: - team_name = "AgentTeam" - lines.extend([ - "# Multi-Agent Team", - f"{team_name.lower()} = Team(", - f' name="{team_name}",', - " mode='coordinate', # or 'sequential' for ordered execution", - ]) - - # Use the first agent's model for team coordination - if agent_vars: - first_model = agent_vars[0][1].model or self.config.default_model - model_code = self._generate_model_code(first_model) - lines.append(f' {model_code}') - - # Add all agents as team members - member_vars = [var for var, _ in agent_vars] - members_str = ", ".join(member_vars) - lines.append(f' members=[{members_str}],') - - lines.extend([ - " tools=[ReasoningTools(add_instructions=True)],", - " instructions=[", - " 'Collaborate to provide comprehensive responses',", - " 'Consider multiple perspectives and expertise areas',", - " 'Present findings in a structured, easy-to-follow format',", - " 'Only output the final consolidated response',", - " ],", - " markdown=True,", - " show_members_responses=True,", - " enable_agentic_context=True,", - " add_datetime_to_instructions=True,", - " success_criteria='The team has provided a complete and accurate response.',", - ")", - "" - ]) - - # Main function and execution logic - lines.extend(self._generate_main_function(has_multiple_agents, agent_vars)) - - lines.extend([ - "", - 'if __name__ == "__main__":', - " main()", - ]) - - return "\n".join(lines) - - def _generate_model_code(self, model: str) -> str: - """Generate the appropriate model instantiation code for Agno framework.""" - if not model: - return 'model=Claude(id="anthropic/claude-3-sonnet-20241022"),' - - model_lower = model.lower() - - # Anthropic models - if "anthropic" in model_lower or "claude" in model_lower: - return f'model=Claude(id="{model}"),' - - # OpenAI models - elif "openai" in model_lower or "gpt" in model_lower: - model_code = 'model=OpenAILike(\n' - model_code += f' id="{model}",\n' - model_code += ' api_key=os.getenv("OPENAI_API_KEY"),\n' - model_code += ' base_url=os.getenv("OPENAI_BASE_URL"),\n' - model_code += ' ),' - return model_code - - # Custom OpenAI-like models (with provider prefix) - elif "/" in model: - provider, model_name = model.split("/", 1) - provider_upper = provider.upper() - - # Generate OpenAILike model with custom configuration - model_code = 'model=OpenAILike(\n' - model_code += f' id="{model}",\n' - model_code += f' api_key=os.getenv("{provider_upper}_API_KEY"),\n' - model_code += f' base_url=os.getenv("{provider_upper}_BASE_URL"),\n' - model_code += ' ),' - return model_code - - # Default to OpenAILike for unrecognized patterns - else: - # Check if we have OpenAI-like environment variables configured - has_openai_config = any( - (isinstance(secret, str) and secret in ["OPENAI_API_KEY", "OPENAI_BASE_URL"]) - or (hasattr(secret, 'name') and secret.name in ["OPENAI_API_KEY", "OPENAI_BASE_URL"]) - for secret in self.config.secrets + # Create agent configuration + agent_config = AgnoAgentConfig( + name=agent.name, + variable_name=agent_var, + instruction=agent.instruction, + role=role, + model_config=model_config, + tools=tools, + use_history=agent.use_history, + human_input=agent.human_input, ) - if has_openai_config: - # Use OpenAI environment variables for custom models - model_code = 'model=OpenAILike(\n' - model_code += f' id="{model}",\n' - model_code += ' api_key=os.getenv("OPENAI_API_KEY"),\n' - model_code += ' base_url=os.getenv("OPENAI_BASE_URL"),\n' - model_code += ' ),' - return model_code - else: - return f'model=OpenAILike(id="{model}"),' - - def _generate_main_function(self, has_multiple_agents: bool, agent_vars: list) -> List[str]: - """Generate the main function and execution logic.""" - lines = ["def main() -> None:"] - - # Handle prompt file loading - if self.has_prompt_file: - lines.extend([ - " # Check if prompt.txt exists and load its content", - " import os", - " prompt_file = 'prompt.txt'", - " if os.path.exists(prompt_file):", - " with open(prompt_file, 'r', encoding='utf-8') as f:", - " prompt_content = f.read().strip()", - ]) - - # Enhanced execution logic - if has_multiple_agents: - # Use team for multi-agent scenarios - team_name = "AgentTeam" - if self.has_prompt_file: - lines.extend([ - " if prompt_content:", - f" {team_name.lower()}.print_response(", - " prompt_content,", - " stream=True,", - " show_full_reasoning=True,", - " stream_intermediate_steps=True,", - " )", - " else:", - f" {team_name.lower()}.print_response(", - " 'Hello! How can our team help you today?',", - " stream=True,", - " show_full_reasoning=True,", - " stream_intermediate_steps=True,", - " )", - " else:", - f" {team_name.lower()}.print_response(", - " 'Hello! How can our team help you today?',", - " stream=True,", - " show_full_reasoning=True,", - " stream_intermediate_steps=True,", - " )", - ]) - else: - lines.extend([ - f" {team_name.lower()}.print_response(", - " 'Hello! How can our team help you today?',", - " stream=True,", - " show_full_reasoning=True,", - " stream_intermediate_steps=True,", - " )", - ]) - - elif agent_vars: - # Single agent scenario with enhanced features - primary_agent_var, primary_agent = agent_vars[0] - if self.has_prompt_file: - lines.extend([ - " if prompt_content:", - f" {primary_agent_var}.print_response(", - " prompt_content,", - " stream=True,", - " show_full_reasoning=True,", - " stream_intermediate_steps=True,", - " )", - " else:", - f" {primary_agent_var}.print_response(", - " 'Hello! How can I help you today?',", - " stream=True,", - " show_full_reasoning=True,", - " stream_intermediate_steps=True,", - " )", - " else:", - f" {primary_agent_var}.print_response(", - " 'Hello! How can I help you today?',", - " stream=True,", - " show_full_reasoning=True,", - " stream_intermediate_steps=True,", - " )", - ]) - else: - lines.extend([ - f" {primary_agent_var}.print_response(", - " 'Hello! How can I help you today?',", - " stream=True,", - " show_full_reasoning=True,", - " stream_intermediate_steps=True,", - " )", - ]) - else: - lines.extend([ - " print('No agents defined')", - ]) - - return lines + framework_config.agents.append(agent_config) + + # Create team if multiple agents + if len(self.config.agents) > 1: + team_name = "agentteam" + + # Use first agent's model for team coordination + team_model = None + if framework_config.agents: + team_model = framework_config.agents[0].model_config + + team_config = AgnoTeamConfig( + name="AgentTeam", + variable_name=team_name, + mode="coordinate", + agent_variables=agent_vars, + model_config=team_model, + ) + + framework_config.team = team_config + + # Generate code using structured builder + generator = AgnoCodeGenerator(framework_config) + return generator.generate_complete_code() def get_requirements(self) -> List[str]: """Get requirements for Agno framework with enhanced tool support.""" @@ -445,22 +158,26 @@ def get_requirements(self) -> List[str]: requirements.extend(server_reqs) # Always include core advanced features - requirements.extend([ - # Core MCP support - "mcp", - # Environment file support - "python-dotenv", - # Optional but commonly used packages - "sqlalchemy", # For storage and memory - "lancedb", # For knowledge and vector databases - "tantivy", # For hybrid search - ]) + requirements.extend( + [ + # Core MCP support + "mcp", + # Environment file support + "python-dotenv", + # Optional but commonly used packages + "sqlalchemy", # For storage and memory + "lancedb", # For knowledge and vector databases + "tantivy", # For hybrid search + ] + ) # Multi-agent scenarios get additional dependencies if len(self.config.agents) > 1: - requirements.extend([ - "asyncio", # Usually built-in but explicit for clarity - ]) + requirements.extend( + [ + "asyncio", # Usually built-in but explicit for clarity + ] + ) return requirements @@ -484,10 +201,12 @@ def _generate_env_file(self): env_lines.extend(["", "# Custom Model Provider Configuration"]) for provider in sorted(custom_providers): provider_upper = provider.upper() - env_lines.extend([ - f"# {provider_upper}_API_KEY=your-{provider}-api-key", - f"# {provider_upper}_BASE_URL=your-{provider}-base-url", - ]) + env_lines.extend( + [ + f"# {provider_upper}_API_KEY=your-{provider}-api-key", + f"# {provider_upper}_BASE_URL=your-{provider}-base-url", + ] + ) # Process secrets to generate environment variables for secret in self.config.secrets: diff --git a/src/agentman/frameworks/agno_builder.py b/src/agentman/frameworks/agno_builder.py new file mode 100644 index 0000000..6f4f9de --- /dev/null +++ b/src/agentman/frameworks/agno_builder.py @@ -0,0 +1,479 @@ +"""Agno framework builder using structured configuration.""" + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Set + + +@dataclass +class AgnoModelConfig: + """Configuration for Agno model instances.""" + + model_type: str # "claude", "openai", "custom" + model_id: str + provider: Optional[str] = None + api_key_env: Optional[str] = None + base_url_env: Optional[str] = None + + +@dataclass +class AgnoToolConfig: + """Configuration for Agno tool instances.""" + + tool_class: str + import_path: str + params: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class AgnoAgentConfig: + """Configuration for Agno agent instances.""" + + name: str + variable_name: str + instruction: str + role: Optional[str] = None + model_config: Optional[AgnoModelConfig] = None + tools: List[AgnoToolConfig] = field(default_factory=list) + use_history: bool = True + human_input: bool = False + enhanced_properties: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class AgnoTeamConfig: + """Configuration for Agno team instances.""" + + name: str + variable_name: str + mode: str = "coordinate" + agent_variables: List[str] = field(default_factory=list) + model_config: Optional[AgnoModelConfig] = None + tools: List[AgnoToolConfig] = field(default_factory=list) + instructions: List[str] = field(default_factory=list) + enhanced_properties: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class AgnoFrameworkConfig: + """Structured configuration for Agno framework.""" + + agents: List[AgnoAgentConfig] = field(default_factory=list) + team: Optional[AgnoTeamConfig] = None + has_prompt_file: bool = False + required_imports: Set[str] = field(default_factory=set) + tool_imports: Set[str] = field(default_factory=set) + + +class AgnoCodeGenerator: + """Generate Agno code using structured configuration.""" + + def __init__(self, config: AgnoFrameworkConfig): + self.config = config + + def generate_imports(self) -> List[str]: + """Generate import statements based on configuration using improved import management.""" + # Core imports that are always needed + imports_set = { + "import os", + "from agno.agent import Agent", + "from dotenv import load_dotenv", + "from agno.tools.reasoning import ReasoningTools", # Always include + } + + # Add model imports + for agent in self.config.agents: + if agent.model_config: + if agent.model_config.model_type == "claude": + imports_set.add("from agno.models.anthropic import Claude") + elif agent.model_config.model_type in ["openai", "custom"]: + imports_set.add("from agno.models.openai import OpenAILike") + + if self.config.team and self.config.team.model_config: + if self.config.team.model_config.model_type == "claude": + imports_set.add("from agno.models.anthropic import Claude") + elif self.config.team.model_config.model_type in ["openai", "custom"]: + imports_set.add("from agno.models.openai import OpenAILike") + + # Add tool imports + if self.config.tool_imports: + imports_set.update(self.config.tool_imports) + + # Add team import if needed + if self.config.team: + imports_set.add("from agno.team.team import Team") + + # Convert to sorted list and add structure + imports = sorted(list(imports_set)) + + # Add environment loading and optional imports + imports.extend( + [ + "", + "# Load environment variables from .env file", + "load_dotenv()", + "", + "# Optional: Uncomment for advanced features", + "# from agno.storage.sqlite import SqliteStorage", + "# from agno.memory.v2.db.sqlite import SqliteMemoryDb", + "# from agno.memory.v2.memory import Memory", + "# from agno.knowledge.url import UrlKnowledge", + "# from agno.vectordb.lancedb import LanceDb", + "", + ] + ) + + return imports + + def generate_agent_definitions(self) -> List[str]: + """Generate agent definition code.""" + lines = [] + + for agent in self.config.agents: + lines.extend(self._generate_single_agent(agent)) + lines.append("") + + return lines + + def generate_team_definition(self) -> List[str]: + """Generate team definition if needed.""" + if not self.config.team: + return [] + + lines = [ + "# Multi-Agent Team", + f"{self.config.team.variable_name} = Team(", + f' name="{self.config.team.name}",', + f" mode='{self.config.team.mode}', # or 'sequential' for ordered execution", + ] + + # Add model configuration + if self.config.team.model_config: + model_code = self._generate_model_instantiation(self.config.team.model_config) + lines.append(f" {model_code}") + + # Add team members + if self.config.team.agent_variables: + members_str = ", ".join(self.config.team.agent_variables) + lines.append(f" members=[{members_str}],") + + # Add tools + if self.config.team.tools: + tools_list = [self._generate_tool_instantiation(tool) for tool in self.config.team.tools] + tools_str = ", ".join(tools_list) + lines.append(f" tools=[{tools_str}],") + else: + lines.append(" tools=[ReasoningTools(add_instructions=True)],") + + # Add instructions + if self.config.team.instructions: + lines.append(" instructions=[") + lines.extend(f' "{instruction}",' for instruction in self.config.team.instructions) + lines.append(" ],") + else: + lines.extend( + [ + " instructions=[", + " 'Collaborate to provide comprehensive responses',", + " 'Consider multiple perspectives and expertise areas',", + " 'Present findings in a structured, easy-to-follow format',", + " 'Only output the final consolidated response',", + " ],", + ] + ) + + # Add enhanced properties + for key, value in self.config.team.enhanced_properties.items(): + if isinstance(value, str): + lines.append(f' {key}="{value}",') + else: + lines.append(f" {key}={value},") + + # Add default enhanced properties + lines.extend( + [ + " markdown=True,", + " show_members_responses=True,", + " enable_agentic_context=True,", + " add_datetime_to_instructions=True,", + " success_criteria='The team has provided a complete and accurate response.',", + ")", + "", + ] + ) + + return lines + + def generate_main_function(self) -> List[str]: + """Generate main function and execution logic.""" + lines = ["def main() -> None:"] + + # Handle prompt file loading + if self.config.has_prompt_file: + lines.extend( + [ + " # Check if prompt.txt exists and load its content", + " import os", + " prompt_file = 'prompt.txt'", + " if os.path.exists(prompt_file):", + " with open(prompt_file, 'r', encoding='utf-8') as f:", + " prompt_content = f.read().strip()", + ] + ) + + # Determine execution target + if self.config.team: + target_var = self.config.team.variable_name + elif self.config.agents: + target_var = self.config.agents[0].variable_name + else: + target_var = None + + if target_var: + if self.config.has_prompt_file: + lines.extend( + [ + " if prompt_content:", + f" {target_var}.print_response(", + " prompt_content,", + " stream=True,", + " show_full_reasoning=True,", + " stream_intermediate_steps=True,", + " )", + " else:", + f" {target_var}.print_response(", + " 'Hello! How can I help you today?',", + " stream=True,", + " show_full_reasoning=True,", + " stream_intermediate_steps=True,", + " )", + " else:", + f" {target_var}.print_response(", + " 'Hello! How can I help you today?',", + " stream=True,", + " show_full_reasoning=True,", + " stream_intermediate_steps=True,", + " )", + ] + ) + else: + greeting = ( + "'Hello! How can our team help you today?'" + if self.config.team + else "'Hello! How can I help you today?'" + ) + lines.extend( + [ + f" {target_var}.print_response(", + f" {greeting},", + " stream=True,", + " show_full_reasoning=True,", + " stream_intermediate_steps=True,", + " )", + ] + ) + else: + lines.append(" print('No agents defined')") + + return lines + + def generate_entry_point(self) -> List[str]: + """Generate script entry point.""" + return [ + "", + 'if __name__ == "__main__":', + " main()", + ] + + def generate_complete_code(self) -> str: + """Generate the complete Agno Python code.""" + lines = [] + + lines.extend(self.generate_imports()) + lines.extend(self.generate_agent_definitions()) + lines.extend(self.generate_team_definition()) + lines.extend(self.generate_main_function()) + lines.extend(self.generate_entry_point()) + + return "\n".join(lines) + + def _generate_single_agent(self, agent: AgnoAgentConfig) -> List[str]: + """Generate code for a single agent.""" + lines = [ + f"# Agent: {agent.name}", + f"{agent.variable_name} = Agent(", + f' name="{agent.name}",', + f' instructions="""{agent.instruction}""",', + ] + + # Add role if specified + if agent.role: + lines.append(f' role="{agent.role}",') + + # Add model configuration + if agent.model_config: + model_code = self._generate_model_instantiation(agent.model_config) + lines.append(f" {model_code}") + + # Add tools + if agent.tools: + tools_list = [self._generate_tool_instantiation(tool) for tool in agent.tools] + # Always add reasoning tools + tools_list.append("ReasoningTools(add_instructions=True)") + tools_str = ", ".join(tools_list) + lines.append(f" tools=[{tools_str}],") + else: + lines.append(" tools=[ReasoningTools(add_instructions=True)],") + + # Add history setting + lines.append(f" add_history_to_messages={str(agent.use_history)},") + + # Add human input if enabled + if agent.human_input: + lines.append(" human_input=True,") + + # Add enhanced properties + for key, value in agent.enhanced_properties.items(): + if isinstance(value, str): + lines.append(f' {key}="{value}",') + else: + lines.append(f" {key}={value},") + + # Add default enhanced properties + lines.extend( + [ + " markdown=True,", + " add_datetime_to_instructions=True,", + " # Optional: Enable advanced features", + " # storage=SqliteStorage(table_name='agent_sessions', db_file='tmp/agent.db'),", + " # memory=Memory(model=Claude(id='claude-sonnet-4-20250514'), db=SqliteMemoryDb()),", + " # enable_agentic_memory=True,", + ")", + ] + ) + + return lines + + def _generate_model_instantiation(self, model_config: AgnoModelConfig) -> str: + """Generate model instantiation code.""" + if model_config.model_type == "claude": + return f'model=Claude(id="{model_config.model_id}"),' + if model_config.model_type == "openai": + return ( + 'model=OpenAILike(\n' + f' id="{model_config.model_id}",\n' + f' api_key=os.getenv("{model_config.api_key_env or "OPENAI_API_KEY"}"),\n' + f' base_url=os.getenv("{model_config.base_url_env or "OPENAI_BASE_URL"}"),\n' + ' ),' + ) + if model_config.model_type == "custom": + api_key_env = model_config.api_key_env or f"{model_config.provider.upper()}_API_KEY" + base_url_env = model_config.base_url_env or f"{model_config.provider.upper()}_BASE_URL" + return ( + 'model=OpenAILike(\n' + f' id="{model_config.model_id}",\n' + f' api_key=os.getenv("{api_key_env}"),\n' + f' base_url=os.getenv("{base_url_env}"),\n' + ' ),' + ) + return f'model=OpenAILike(id="{model_config.model_id}"),' + + def _generate_tool_instantiation(self, tool_config: AgnoToolConfig) -> str: + """Generate tool instantiation code.""" + if tool_config.params: + params_str = ", ".join(f"{k}={v}" for k, v in tool_config.params.items()) + return f"{tool_config.tool_class}({params_str})" + return f"{tool_config.tool_class}()" + + +class AgnoConfigBuilder: + """Builder for creating AgnoFrameworkConfig from agentfile configuration.""" + + def __init__(self): + # Server to tool mapping using cleaner approach inspired by PR #4 + self.server_tool_mapping = { + # Search and web tools + "web_search": ("agno.tools.duckduckgo.DuckDuckGoTools", "DuckDuckGoTools()"), + "search": ("agno.tools.duckduckgo.DuckDuckGoTools", "DuckDuckGoTools()"), + "browser": ("agno.tools.duckduckgo.DuckDuckGoTools", "DuckDuckGoTools()"), + # Finance tools + "finance": ( + "agno.tools.yfinance.YFinanceTools", + "YFinanceTools(stock_price=True, analyst_recommendations=True)", + ), + "yfinance": ( + "agno.tools.yfinance.YFinanceTools", + "YFinanceTools(stock_price=True, analyst_recommendations=True)", + ), + "stock": ( + "agno.tools.yfinance.YFinanceTools", + "YFinanceTools(stock_price=True, analyst_recommendations=True)", + ), + # File and system tools + "file": ("agno.tools.file.FileTools", "FileTools()"), + "filesystem": ("agno.tools.file.FileTools", "FileTools()"), + "shell": ("agno.tools.shell.ShellTools", "ShellTools()"), + "terminal": ("agno.tools.shell.ShellTools", "ShellTools()"), + "python": ("agno.tools.python.PythonTools", "PythonTools()"), + "code": ("agno.tools.python.PythonTools", "PythonTools()"), + } + + def build_model_config(self, model: str) -> AgnoModelConfig: + """Build model configuration from model string using improved logic.""" + if not model: + # Provide sensible default as seen in PR #4 + return AgnoModelConfig("claude", "anthropic/claude-3-sonnet-20241022") + + model_lower = model.lower() + + # Anthropic models + if "anthropic" in model_lower or "claude" in model_lower: + return AgnoModelConfig("claude", model) + + # OpenAI models + if "openai" in model_lower or "gpt" in model_lower: + return AgnoModelConfig("openai", model, api_key_env="OPENAI_API_KEY", base_url_env="OPENAI_BASE_URL") + + # Custom models with provider prefix (e.g., "ollama/llama3", "groq/mixtral") + if "/" in model: + provider, _ = model.split("/", 1) + return AgnoModelConfig( + "custom", + model, + provider=provider, + api_key_env=f"{provider.upper()}_API_KEY", + base_url_env=f"{provider.upper()}_BASE_URL", + ) + + # Default fallback to OpenAI-like + return AgnoModelConfig("openai", model, api_key_env="OPENAI_API_KEY", base_url_env="OPENAI_BASE_URL") + + def build_tools_for_servers(self, servers: List[str]) -> List[AgnoToolConfig]: + """Build tool configurations for given servers, ensuring no duplicates.""" + seen_tools = set() + tools = [] + for server in servers: + if server in self.server_tool_mapping: + import_path, init_str = self.server_tool_mapping[server] + tool_class = import_path.split('.')[-1] + + # Use tool_class as the unique identifier to avoid duplicates + if tool_class not in seen_tools: + # Parse params from init_str for backward compatibility + params = {} + if "(" in init_str and init_str != f"{tool_class}()": + # Extract parameters from init string if present + if "stock_price=True" in init_str: + params = {"stock_price": True, "analyst_recommendations": True} + + tool = AgnoToolConfig( + tool_class=tool_class, + import_path=f"from {'.'.join(import_path.split('.')[:-1])} import {tool_class}", + params=params, + ) + tools.append(tool) + seen_tools.add(tool_class) + return tools + + def get_tool_imports(self, tools: List[AgnoToolConfig]) -> Set[str]: + """Get import statements for tools.""" + return {tool.import_path for tool in tools} diff --git a/src/agentman/frameworks/base.py b/src/agentman/frameworks/base.py index 3186674..c4f013a 100644 --- a/src/agentman/frameworks/base.py +++ b/src/agentman/frameworks/base.py @@ -1,8 +1,8 @@ """Base framework interface for AgentMan.""" from abc import ABC, abstractmethod -from typing import List from pathlib import Path +from typing import List from agentman.agentfile_parser import AgentfileConfig diff --git a/src/agentman/frameworks/fast_agent.py b/src/agentman/frameworks/fast_agent.py index 17661f2..7afbfd6 100644 --- a/src/agentman/frameworks/fast_agent.py +++ b/src/agentman/frameworks/fast_agent.py @@ -1,9 +1,11 @@ """Fast-Agent framework implementation for AgentMan.""" from typing import List + import yaml from .base import BaseFramework +from .fast_agent_builder import FastAgentCodeGenerator, FastAgentConfig class FastAgentFramework(BaseFramework): @@ -11,67 +13,59 @@ class FastAgentFramework(BaseFramework): def build_agent_content(self) -> str: """Build the Python agent file content for Fast-Agent framework.""" - lines = [] - - # Imports - lines.extend([ - "import asyncio", - "from mcp_agent.core.fastagent import FastAgent", - "", - "# Create the application", - 'fast = FastAgent("Generated by Agentman")', - "", - ]) - - # Agent definitions + + # Create structured configuration + config = FastAgentConfig(name="Generated by Agentman", has_prompt_file=self.has_prompt_file) + + # Add agents from config for agent in self.config.agents.values(): - lines.append(agent.to_decorator_string(self.config.default_model)) + agent_config = { + "name": agent.name, + "instruction": agent.instruction, + "servers": agent.servers, + "model": agent.model or self.config.default_model, + "use_history": agent.use_history, + "human_input": agent.human_input, + "default": agent.default, + } + config.agents.append(agent_config) - # Router definitions + # Add routers from config for router in self.config.routers.values(): - lines.append(router.to_decorator_string(self.config.default_model)) + router_config = { + "name": router.name, + "agents": router.agents, + "model": router.model or self.config.default_model, + "instruction": router.instruction, + "default": router.default, + } + config.routers.append(router_config) - # Chain definitions + # Add chains from config for chain in self.config.chains.values(): - lines.append(chain.to_decorator_string()) + chain_config = { + "name": chain.name, + "sequence": chain.sequence, + "instruction": chain.instruction, + "cumulative": chain.cumulative, + "default": chain.default, + } + config.chains.append(chain_config) - # Orchestrator definitions + # Add orchestrators from config for orchestrator in self.config.orchestrators.values(): - lines.append(orchestrator.to_decorator_string(self.config.default_model)) - - # Main function - lines.extend([ - "async def main() -> None:", - " async with fast.run() as agent:", - ]) - - # Check if prompt.txt exists and add prompt loading - if self.has_prompt_file: - lines.extend([ - " # Check if prompt.txt exists and load its content", - " import os", - " prompt_file = 'prompt.txt'", - " if os.path.exists(prompt_file):", - " with open(prompt_file, 'r', encoding='utf-8') as f:", - " prompt_content = f.read().strip()", - " if prompt_content:", - " await agent(prompt_content)", - " else:", - " await agent()", - " else:", - " await agent()", - ]) - else: - lines.extend([" await agent()"]) - - lines.extend([ - "", - "", - 'if __name__ == "__main__":', - " asyncio.run(main())", - ]) + orchestrator_config = { + "name": orchestrator.name, + "agents": orchestrator.agents, + "model": orchestrator.model or self.config.default_model, + "instruction": orchestrator.instruction, + "default": orchestrator.default, + } + config.orchestrators.append(orchestrator_config) - return "\n".join(lines) + # Generate code using structured builder + generator = FastAgentCodeGenerator(config) + return generator.generate_complete_code() def get_requirements(self) -> List[str]: """Get requirements for Fast-Agent framework.""" diff --git a/src/agentman/frameworks/fast_agent_builder.py b/src/agentman/frameworks/fast_agent_builder.py new file mode 100644 index 0000000..8148c1d --- /dev/null +++ b/src/agentman/frameworks/fast_agent_builder.py @@ -0,0 +1,186 @@ +"""FastAgent framework builder using structured configuration.""" + +from dataclasses import dataclass, field +from typing import Any, Dict, List + + +@dataclass +class FastAgentConfig: + """Structured configuration for FastAgent.""" + + name: str = "Generated by Agentman" + agents: List[Dict[str, Any]] = field(default_factory=list) + routers: List[Dict[str, Any]] = field(default_factory=list) + chains: List[Dict[str, Any]] = field(default_factory=list) + orchestrators: List[Dict[str, Any]] = field(default_factory=list) + has_prompt_file: bool = False + + +class FastAgentCodeGenerator: + """Generate FastAgent code using structured configuration.""" + + def __init__(self, config: FastAgentConfig): + self.config = config + + def generate_imports(self) -> List[str]: + """Generate import statements.""" + return [ + "import asyncio", + "from mcp_agent.core.fastagent import FastAgent", + "", + ] + + def generate_app_creation(self) -> List[str]: + """Generate FastAgent app creation.""" + return [ + "# Create the application", + f'fast = FastAgent("{self.config.name}")', + "", + ] + + def generate_agent_decorators(self) -> List[str]: + """Generate agent decorator definitions.""" + return [self._build_agent_decorator(agent_config) for agent_config in self.config.agents] + + def generate_router_decorators(self) -> List[str]: + """Generate router decorator definitions.""" + return [self._build_router_decorator(router_config) for router_config in self.config.routers] + + def generate_chain_decorators(self) -> List[str]: + """Generate chain decorator definitions.""" + return [self._build_chain_decorator(chain_config) for chain_config in self.config.chains] + + def generate_orchestrator_decorators(self) -> List[str]: + """Generate orchestrator decorator definitions.""" + return [ + self._build_orchestrator_decorator(orchestrator_config) for orchestrator_config in self.config.orchestrators + ] + + def generate_main_function(self) -> List[str]: + """Generate main function and execution logic.""" + lines = ["async def main() -> None:", " async with fast.run() as agent:"] + + if self.config.has_prompt_file: + lines.extend( + [ + " # Check if prompt.txt exists and load its content", + " import os", + " prompt_file = 'prompt.txt'", + " if os.path.exists(prompt_file):", + " with open(prompt_file, 'r', encoding='utf-8') as f:", + " prompt_content = f.read().strip()", + " if prompt_content:", + " await agent(prompt_content)", + " else:", + " await agent()", + " else:", + " await agent()", + ] + ) + else: + lines.extend([" await agent()"]) + + return lines + + def generate_entry_point(self) -> List[str]: + """Generate script entry point.""" + return [ + "", + "", + 'if __name__ == "__main__":', + " asyncio.run(main())", + ] + + def generate_complete_code(self) -> str: + """Generate the complete FastAgent Python code.""" + lines = [] + + lines.extend(self.generate_imports()) + lines.extend(self.generate_app_creation()) + lines.extend(self.generate_agent_decorators()) + lines.extend(self.generate_router_decorators()) + lines.extend(self.generate_chain_decorators()) + lines.extend(self.generate_orchestrator_decorators()) + lines.extend(self.generate_main_function()) + lines.extend(self.generate_entry_point()) + + return "\n".join(lines) + + def _build_agent_decorator(self, agent_config: Dict[str, Any]) -> str: + """Build a @fast.agent decorator from configuration.""" + params = [f'name="{agent_config["name"]}"', f'instruction="""{agent_config["instruction"]}"""'] + + if agent_config.get("servers"): + servers_str = "[" + ", ".join(f'"{s}"' for s in agent_config["servers"]) + "]" + params.append(f"servers={servers_str}") + + if agent_config.get("model"): + params.append(f'model="{agent_config["model"]}"') + + if not agent_config.get("use_history", True): + params.append("use_history=False") + + if agent_config.get("human_input"): + params.append("human_input=True") + + if agent_config.get("default"): + params.append("default=True") + + return "@fast.agent(\n " + ",\n ".join(params) + "\n)" + + def _build_router_decorator(self, router_config: Dict[str, Any]) -> str: + """Build a @fast.router decorator from configuration.""" + params = [f'name="{router_config["name"]}"'] + + if router_config.get("agents"): + agents_str = "[" + ", ".join(f'"{a}"' for a in router_config["agents"]) + "]" + params.append(f"agents={agents_str}") + + if router_config.get("model"): + params.append(f'model="{router_config["model"]}"') + + if router_config.get("instruction"): + params.append(f'instruction="""{router_config["instruction"]}"""') + + if router_config.get("default"): + params.append("default=True") + + return "@fast.router(\n " + ",\n ".join(params) + "\n)" + + def _build_chain_decorator(self, chain_config: Dict[str, Any]) -> str: + """Build a @fast.chain decorator from configuration.""" + params = [f'name="{chain_config["name"]}"'] + + if chain_config.get("sequence"): + sequence_str = "[" + ", ".join(f'"{s}"' for s in chain_config["sequence"]) + "]" + params.append(f"sequence={sequence_str}") + + if chain_config.get("instruction"): + params.append(f'instruction="""{chain_config["instruction"]}"""') + + if chain_config.get("cumulative"): + params.append("cumulative=True") + + if chain_config.get("default"): + params.append("default=True") + + return "@fast.chain(\n " + ",\n ".join(params) + "\n)" + + def _build_orchestrator_decorator(self, orchestrator_config: Dict[str, Any]) -> str: + """Build a @fast.orchestrator decorator from configuration.""" + params = [f'name="{orchestrator_config["name"]}"'] + + if orchestrator_config.get("agents"): + agents_str = "[" + ", ".join(f'"{a}"' for a in orchestrator_config["agents"]) + "]" + params.append(f"agents={agents_str}") + + if orchestrator_config.get("model"): + params.append(f'model="{orchestrator_config["model"]}"') + + if orchestrator_config.get("instruction"): + params.append(f'instruction="""{orchestrator_config["instruction"]}"""') + + if orchestrator_config.get("default"): + params.append("default=True") + + return "@fast.orchestrator(\n " + ",\n ".join(params) + "\n)" diff --git a/tests/test_refactored_framework_builders.py b/tests/test_refactored_framework_builders.py new file mode 100644 index 0000000..80685d6 --- /dev/null +++ b/tests/test_refactored_framework_builders.py @@ -0,0 +1,208 @@ +"""Tests for the refactored framework builders using structured configuration.""" + +import pytest +from src.agentman.agentfile_parser import AgentfileParser +from src.agentman.agent_builder import AgentBuilder +from src.agentman.frameworks.fast_agent_builder import FastAgentConfig, FastAgentCodeGenerator +from src.agentman.frameworks.agno_builder import AgnoFrameworkConfig, AgnoCodeGenerator, AgnoConfigBuilder +import tempfile +from pathlib import Path + + +class TestRefactoredFrameworkBuilders: + """Test the new structured framework builders.""" + + def test_fast_agent_structured_config_creation(self): + """Test that FastAgent uses structured configuration objects.""" + content = """ +FROM yeahdongcn/agentman-base:latest +FRAMEWORK fast-agent +MODEL anthropic/claude-3-sonnet-20241022 +AGENT helper +INSTRUCTION You are a helpful assistant +SERVERS web_search +""" + parser = AgentfileParser() + config = parser.parse_content(content) + + with tempfile.TemporaryDirectory() as temp_dir: + builder = AgentBuilder(config, temp_dir) + code = builder.framework.build_agent_content() + + # Verify structured approach produces expected output + assert "FastAgent(" in code + assert "@fast.agent(" in code + assert 'name="helper"' in code + assert 'instruction="""You are a helpful assistant"""' in code + assert 'servers=["web_search"]' in code + assert "asyncio.run(main())" in code + + # Verify no string concatenation artifacts + assert not code.startswith('["') + assert not "\\n" in code.replace("\\n", "") + + def test_agno_structured_config_creation(self): + """Test that Agno uses structured configuration objects.""" + content = """ +FROM yeahdongcn/agentman-base:latest +FRAMEWORK agno +MODEL anthropic/claude-3-sonnet-20241022 +AGENT researcher +INSTRUCTION Research specialist +SERVERS web_search finance +""" + parser = AgentfileParser() + config = parser.parse_content(content) + + with tempfile.TemporaryDirectory() as temp_dir: + builder = AgentBuilder(config, temp_dir) + code = builder.framework.build_agent_content() + + # Verify structured approach produces expected output + assert "from agno.agent import Agent" in code + assert "from agno.models.anthropic import Claude" in code + assert "from agno.tools.duckduckgo import DuckDuckGoTools" in code + assert "from agno.tools.yfinance import YFinanceTools" in code + assert "researcher_agent = Agent(" in code + assert "Claude(id=" in code + assert "DuckDuckGoTools()" in code + assert "YFinanceTools(stock_price=True, analyst_recommendations=True)" in code + + def test_fast_agent_structured_builder_directly(self): + """Test FastAgent structured builder directly.""" + config = FastAgentConfig( + name="Test App", + agents=[ + { + "name": "test_agent", + "instruction": "Test instruction", + "servers": ["web_search"], + "model": "gpt-4", + "use_history": True, + "human_input": False, + "default": False, + } + ], + has_prompt_file=False, + ) + + generator = FastAgentCodeGenerator(config) + code = generator.generate_complete_code() + + assert 'FastAgent("Test App")' in code + assert '@fast.agent(' in code + assert 'name="test_agent"' in code + assert 'instruction="""Test instruction"""' in code + assert 'servers=["web_search"]' in code + assert 'model="gpt-4"' in code + + def test_agno_config_builder_directly(self): + """Test Agno config builder directly.""" + builder = AgnoConfigBuilder() + + # Test model config building + claude_config = builder.build_model_config("anthropic/claude-3-sonnet") + assert claude_config.model_type == "claude" + assert claude_config.model_id == "anthropic/claude-3-sonnet" + + openai_config = builder.build_model_config("openai/gpt-4") + assert openai_config.model_type == "openai" + assert openai_config.model_id == "openai/gpt-4" + + custom_config = builder.build_model_config("groq/mixtral-8x7b") + assert custom_config.model_type == "custom" + assert custom_config.model_id == "groq/mixtral-8x7b" + assert custom_config.provider == "groq" + + # Test tool building + tools = builder.build_tools_for_servers(["web_search", "finance"]) + assert len(tools) == 2 + tool_classes = [tool.tool_class for tool in tools] + assert "DuckDuckGoTools" in tool_classes + assert "YFinanceTools" in tool_classes + + def test_code_quality_improvements(self): + """Test that refactored code has better structure and quality.""" + content = """ +FROM yeahdongcn/agentman-base:latest +FRAMEWORK agno +MODEL anthropic/claude-3-sonnet-20241022 +AGENT researcher +INSTRUCTION Research specialist +SERVERS web_search +AGENT analyst +INSTRUCTION Data analyst +SERVERS finance +""" + parser = AgentfileParser() + config = parser.parse_content(content) + + with tempfile.TemporaryDirectory() as temp_dir: + builder = AgentBuilder(config, temp_dir) + code = builder.framework.build_agent_content() + + # Verify clean structure + lines = code.split('\n') + + # Check for proper imports organization + import_section_found = False + agent_section_found = False + team_section_found = False + main_section_found = False + + for line in lines: + if line.startswith('import ') or line.startswith('from '): + import_section_found = True + elif '= Agent(' in line: + agent_section_found = True + elif '= Team(' in line: + team_section_found = True + elif 'def main(' in line: + main_section_found = True + + assert import_section_found + assert agent_section_found + assert team_section_found # Multiple agents should create team + assert main_section_found + + # Verify no manual string concatenation artifacts + assert "lines.extend([" not in code + assert "\", \"" not in code # No manual quote handling + + # Verify proper Python syntax + try: + compile(code, '', 'exec') + except SyntaxError: + pytest.fail("Generated code has syntax errors") + + def test_backward_compatibility(self): + """Test that refactored builders maintain backward compatibility.""" + # Test complex configuration that worked before + content = """ +FROM yeahdongcn/agentman-base:latest +FRAMEWORK fast-agent +MODEL anthropic/claude-3-sonnet-20241022 + +AGENT router_agent +INSTRUCTION Route requests to appropriate agents +MODEL openai/gpt-4 +SERVERS web_search +DEFAULT true + +ROUTER main_router +AGENTS router_agent +INSTRUCTION Route user requests +""" + parser = AgentfileParser() + config = parser.parse_content(content) + + with tempfile.TemporaryDirectory() as temp_dir: + builder = AgentBuilder(config, temp_dir) + code = builder.framework.build_agent_content() + + # Should generate both agent and router + assert "@fast.agent(" in code + assert "@fast.router(" in code + assert "default=True" in code + assert 'model="openai/gpt-4"' in code + assert 'agents=["router_agent"]' in code