em-agent-framework

A production-ready AI agent framework with support for Gemini and Anthropic (Claude) models via Vertex AI.

Features

• Multi-Model Support: Seamless integration with Gemini and Anthropic models
• Automatic Fallback: Cascading fallback across multiple models on failure
• Parallel Tool Execution: Execute independent function calls concurrently
• Recursive Agent Calls: Agents can spawn sub-agents for parallel task decomposition
• Context Injection: Pass large data/secrets to tools without sending through LLM
• Group Chat: Multi-agent conversations with flexible routing strategies
• Metrics & Observability: Built-in tracking for performance and usage

Installation

pip install em-agent-framework

Requirements

• Python 3.8+
• Google Cloud Project with Vertex AI enabled
• Vertex AI API credentials configured

Quick Start

import asyncio
from typing import Annotated
from em_agent_framework.core.agent import Agent
from em_agent_framework.config.settings import ModelConfig, AgentConfig

# Define a tool
def get_weather(city: Annotated[str, "City name"]) -> str:
    """Get weather for a city."""
    return f"Weather in {city}: Sunny, 72°F"

async def main():
    # Configure models with fallback
    model_configs = [
        ModelConfig(name="gemini-2.0-flash-exp", provider="gemini"),
        ModelConfig(name="claude-3-5-sonnet-v2@20241022", provider="anthropic"),
    ]

    # Create agent
    agent = Agent(
        name="assistant",
        system_instruction="You are a helpful assistant.",
        tools=[get_weather],
        model_configs=model_configs,
        agent_config=AgentConfig(verbose=True)
    )

    # Send message
    response = await agent.send_message("What's the weather in Tokyo?")
    print(response)

asyncio.run(main())

Key Features

Model Fallback

Automatically falls back to alternative models if the primary model fails:

model_configs = [
    ModelConfig(name="gemini-2.0-flash-exp", provider="gemini"),  # Try first
    ModelConfig(name="claude-3-5-sonnet-v2@20241022", provider="anthropic"),  # Fallback
]

Parallel Tool Execution

Execute multiple independent function calls concurrently:

agent_config = AgentConfig(
    enable_parallel_tools=True,
    max_parallel_tools=5
)

Context Injection

Pass data to tools without sending it through the LLM (ideal for large datasets, API keys, or user info):

def analyze_data(metric: Annotated[str, "Metric to analyze"], context: dict) -> str:
    """Analyze user data."""
    df = context.get('dataframe')  # Not sent to LLM
    userid = context.get('userid')  # Not sent to LLM
    return f"User {userid}: Analysis complete"

agent = Agent(
    name="analyst",
    tools=[analyze_data],
    context={
        'dataframe': large_df,      # Never sent to LLM
        'userid': 'USER_12345',     # Never sent to LLM
    },
    model_configs=model_configs,
    agent_config=agent_config
)

Benefits:

• Reduce token costs: Large data stays local
• Security: Sensitive data (API keys, credentials) stays private
• Authentication: Pass user info for validation

Dynamic Tool Loading

Load tools on-demand instead of all at once:

# Define tools
def basic_search(query: Annotated[str, "Search query"]) -> str:
    return f"Results for: {query}"

def advanced_analysis(data: Annotated[str, "Data to analyze"]) -> str:
    return f"Analysis of: {data}"

# Create agent with complementary tools
agent = Agent(
    name="assistant",
    tools=[basic_search],  # Loaded immediately
    complementary_tools=[advanced_analysis],  # Available via search_tool
    model_configs=model_configs,
    agent_config=agent_config
)

# Agent can dynamically load tools when needed
# Just mention the tool name and the agent will use search_tool to find and load it
response = await agent.send_message("Use advanced_analysis to analyze this data")

Dynamic Instructions

Load specialized instructions on-demand:

# Create instructions.json
{
    "instructions": [
        {
            "id": "code_review",
            "description": "Code review guidelines",
            "instruction": "Review code for: correctness, efficiency, security, readability"
        },
        {
            "id": "api_design",
            "description": "API design principles",
            "instruction": "Design RESTful APIs following best practices"
        }
    ]
}

# Create agent with instructions file
agent = Agent(
    name="assistant",
    system_instruction="You are a code assistant.",
    instructions_file="instructions.json",
    model_configs=model_configs,
    agent_config=agent_config
)

# Agent can load instructions dynamically
response = await agent.send_message("Load code_review instructions and review this code: ...")

Recursive Agent Calls

Agents can spawn sub-agents to handle subtasks in parallel, enabling complex task decomposition:

# Create agent with recursion support
agent_config = AgentConfig(
    enable_recursive_agents=True,  # Enable recursive_agent_call tool
    max_recursion_depth=3,  # Allow up to 3 levels of nesting
    verbose=True
)

agent = Agent(
    name="math_coordinator",
    system_instruction=(
        "You are a math assistant. When you receive complex problems with "
        "multiple independent calculations, use recursive_agent_call to spawn "
        "sub-agents for each piece. This enables parallel execution."
    ),
    tools=[add, multiply, subtract, divide],
    model_configs=model_configs,
    agent_config=agent_config
)

# Agent automatically decomposes task into parallel subtasks
response = await agent.send_message(
    "Calculate (125 + 75) * (20 - 8). Use recursive_agent_call to "
    "handle (125 + 75) and (20 - 8) as separate subtasks."
)

Features:

• Agent Hierarchy: Each agent has a unique ID and tracks its parent_agent_id
• Depth Control: Configurable max_recursion_depth prevents infinite recursion
• Metadata Tracking: Sub-agent responses include agent_id, parent_agent_id, and recursion_depth
• UI Bundling: Metadata enables grouping of intermediate responses in the UI
• Parallel Execution: Sub-agents run independently for better performance

Multi-Agent Group Chat

from em_agent_framework.core.group_chat.manager import GroupChatManager

# Create specialized agents
researcher = Agent(name="researcher", system_instruction="Research expert", ...)
developer = Agent(name="developer", system_instruction="Python developer", ...)

# Create group chat with skill-based routing
manager = GroupChatManager(
    agents=[researcher, developer],
    strategy="skill_based",  # Routes based on agent descriptions
    max_total_turns=10
)

# Start conversation
await manager.initiate_conversation(
    query="Research and build a JSON parser"
)

Configuration

Agent Configuration

AgentConfig(
    max_turns=100,                  # Max conversation turns
    max_retries_per_model=3,        # Retries before fallback
    verbose=True,                   # Print debug logs
    enable_parallel_tools=True,     # Enable parallel execution
    max_parallel_tools=5,           # Max concurrent tools
    enable_recursive_agents=False,  # Enable recursive_agent_call tool
    max_recursion_depth=2           # Max depth for recursive agent calls
)

Model Configuration

ModelConfig(
    name="gemini-2.0-flash-exp",
    provider="gemini",          # "gemini" or "anthropic"
    temperature=0.7,            # optional; omit (None, the default) to let the
                                # provider use its own tuned default
    max_output_tokens=8192,
    timeout=10.0               # Request timeout (seconds)
)

temperature defaults to None (omit) as of 1.3.3. When unset, the parameter is left out of the request so each provider applies its own reasoning-tuned default. Pass an explicit float to pin it (e.g. temperature=0.1 to restore the pre-1.3.3 behavior). See CHANGELOG.md.

License

Apache-2.0 License - see LICENSE file for details