AutoGen is Microsoft’s framework for building multi-agent conversational AI systems. AgentOps provides seamless integration with AutoGen to track and monitor your multi-agent workflows.

​
Installation

pip install agentops autogen-core python-dotenv

​
Setting Up API Keys

Before using AutoGen with AgentOps, you need to set up your API keys. You can obtain:

Then to set them up, you can either export them as environment variables or set them in a .env file.

export OPENAI_API_KEY="your_openai_api_key_here"
export AGENTOPS_API_KEY="your_agentops_api_key_here"

Then load the environment variables in your Python code:

from dotenv import load_dotenv
import os

# Load environment variables from .env file
load_dotenv()

# Set up environment variables with fallback values
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")
os.environ["AGENTOPS_API_KEY"] = os.getenv("AGENTOPS_API_KEY")

​
Usage

AgentOps automatically instruments AutoGen agents and tracks their interactions. Simply initialize AgentOps before creating your AutoGen agents!

import asyncio
from dataclasses import dataclass
from typing import Callable
import agentops

from autogen_core import (
    DefaultTopicId, 
    MessageContext, 
    RoutedAgent, 
    default_subscription, 
    message_handler,
    AgentId,
    SingleThreadedAgentRuntime
)

# Initialize AgentOps
agentops.init()

@dataclass
class CountdownMessage:
    """Message containing a number for countdown operations"""
    content: int

@default_subscription
class ModifierAgent(RoutedAgent):
    """Agent that modifies numbers by applying a transformation function"""
    
    def __init__(self, modify_val: Callable[[int], int]) -> None:
        super().__init__("A modifier agent that transforms numbers.")
        self._modify_val = modify_val

    @message_handler
    async def handle_message(self, message: CountdownMessage, ctx: MessageContext) -> None:
        """Handle incoming messages and apply modification"""
        original_val = message.content
        modified_val = self._modify_val(original_val)
        
        print(f"🔧 ModifierAgent: Transformed {original_val} → {modified_val}")
        
        # Publish the modified value to continue the workflow
        await self.publish_message(
            CountdownMessage(content=modified_val), 
            DefaultTopicId()
        )

@default_subscription  
class CheckerAgent(RoutedAgent):
    """Agent that checks if a condition is met and decides whether to continue"""
    
    def __init__(self, stop_condition: Callable[[int], bool]) -> None:
        super().__init__("A checker agent that validates conditions.")
        self._stop_condition = stop_condition

    @message_handler
    async def handle_message(self, message: CountdownMessage, ctx: MessageContext) -> None:
        """Handle incoming messages and check stopping condition"""
        value = message.content
        
        if not self._stop_condition(value):
            print(f"✅ CheckerAgent: {value} passed validation, continuing workflow")
            # Continue the workflow by publishing the message
            await self.publish_message(
                CountdownMessage(content=value), 
                DefaultTopicId()
            )
        else:
            print(f"🛑 CheckerAgent: {value} failed validation, stopping workflow")
            print("🎉 Countdown completed successfully!")

async def run_countdown_workflow():
    """Run a countdown workflow from 10 to 1 using AutoGen agents"""
    
    print("🚀 Starting AutoGen Countdown Workflow")
    print("=" * 50)
    
    # Create the AutoGen runtime
    runtime = SingleThreadedAgentRuntime()
    
    # Register the modifier agent (subtracts 1 from each number)
    await ModifierAgent.register(
        runtime,
        "modifier",
        lambda: ModifierAgent(modify_val=lambda x: x - 1),
    )
    
    # Register the checker agent (stops when value <= 1)
    await CheckerAgent.register(
        runtime,
        "checker", 
        lambda: CheckerAgent(stop_condition=lambda x: x <= 1),
    )
    
    # Start the runtime
    runtime.start()
    print("🤖 AutoGen runtime started")
    print("📨 Sending initial message with value: 10")
    
    # Send initial message to start the countdown
    await runtime.send_message(
        CountdownMessage(10), 
        AgentId("checker", "default")
    )
    
    # Wait for the workflow to complete
    await runtime.stop_when_idle()
    
    print("=" * 50)
    print("✨ Workflow completed! Check your AgentOps dashboard for detailed traces.")

# Run the workflow
if __name__ == "__main__":
    asyncio.run(run_countdown_workflow())

​
Examples

Agent Chat Example

Basic multi-agent chat functionality

Math Agent Example

Demonstrates an agent specialized for mathematical problem-solving.

Visit your AgentOps Dashboard to see detailed traces of your AutoGen agent interactions, performance metrics, and workflow analytics.