433 lines
13 KiB
Plaintext
433 lines
13 KiB
Plaintext
---
|
|
title: "AutoGen"
|
|
description: "Integrate AgentOps with Microsoft AutoGen for multi-agent workflow tracking"
|
|
---
|
|
|
|
[AutoGen](https://microsoft.github.io/autogen/stable/) 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
|
|
|
|
<CodeGroup>
|
|
```bash pip
|
|
pip install agentops autogen-core python-dotenv
|
|
```
|
|
```bash poetry
|
|
poetry add agentops autogen-core python-dotenv
|
|
```
|
|
```bash uv
|
|
uv pip install agentops autogen-core python-dotenv
|
|
```
|
|
</CodeGroup>
|
|
|
|
## Setting Up API Keys
|
|
|
|
Before using AutoGen with AgentOps, you need to set up your API keys. You can obtain:
|
|
- **OPENAI_API_KEY**: From the [OpenAI Platform](https://platform.openai.com/api-keys)
|
|
- **AGENTOPS_API_KEY**: From your [AgentOps Dashboard](https://app.agentops.ai/)
|
|
|
|
Then to set them up, you can either export them as environment variables or set them in a `.env` file.
|
|
|
|
<CodeGroup>
|
|
```bash Export to CLI
|
|
export OPENAI_API_KEY="your_openai_api_key_here"
|
|
export AGENTOPS_API_KEY="your_agentops_api_key_here"
|
|
```
|
|
```txt Set in .env file
|
|
OPENAI_API_KEY="your_openai_api_key_here"
|
|
AGENTOPS_API_KEY="your_agentops_api_key_here"
|
|
```
|
|
</CodeGroup>
|
|
|
|
Then load the environment variables in your Python code:
|
|
|
|
```python
|
|
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!
|
|
|
|
<CodeGroup>
|
|
```python Countdown
|
|
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())
|
|
```
|
|
|
|
```python Multi-Agent
|
|
import asyncio
|
|
from dataclasses import dataclass
|
|
from typing import List, Dict, Any
|
|
import agentops
|
|
|
|
from autogen_core import (
|
|
DefaultTopicId,
|
|
MessageContext,
|
|
RoutedAgent,
|
|
default_subscription,
|
|
message_handler,
|
|
AgentId,
|
|
SingleThreadedAgentRuntime
|
|
)
|
|
|
|
# Initialize AgentOps
|
|
agentops.init()
|
|
|
|
@dataclass
|
|
class DataMessage:
|
|
"""Message containing data to be processed"""
|
|
data: List[Dict[str, Any]]
|
|
stage: str
|
|
metadata: Dict[str, Any]
|
|
|
|
@default_subscription
|
|
class DataCollectorAgent(RoutedAgent):
|
|
"""Agent responsible for collecting and preparing initial data"""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__("Data collector agent that gathers initial dataset.")
|
|
|
|
@message_handler
|
|
async def handle_message(self, message: DataMessage, ctx: MessageContext) -> None:
|
|
print(f"📊 DataCollector: Collecting data for {message.metadata.get('source', 'unknown')}")
|
|
|
|
# Simulate data collection
|
|
collected_data = [
|
|
{"id": 1, "value": 100, "category": "A"},
|
|
{"id": 2, "value": 200, "category": "B"},
|
|
{"id": 3, "value": 150, "category": "A"},
|
|
{"id": 4, "value": 300, "category": "C"},
|
|
]
|
|
|
|
print(f"✅ DataCollector: Collected {len(collected_data)} records")
|
|
|
|
# Send to processor
|
|
await self.publish_message(
|
|
DataMessage(
|
|
data=collected_data,
|
|
stage="processing",
|
|
metadata={**message.metadata, "collected_count": len(collected_data)}
|
|
),
|
|
DefaultTopicId()
|
|
)
|
|
|
|
@default_subscription
|
|
class DataProcessorAgent(RoutedAgent):
|
|
"""Agent that processes and transforms data"""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__("Data processor agent that transforms collected data.")
|
|
|
|
@message_handler
|
|
async def handle_message(self, message: DataMessage, ctx: MessageContext) -> None:
|
|
if message.stage != "processing":
|
|
return
|
|
|
|
print(f"⚙️ DataProcessor: Processing {len(message.data)} records")
|
|
|
|
# Process data - add calculated fields
|
|
processed_data = []
|
|
for item in message.data:
|
|
processed_item = {
|
|
**item,
|
|
"processed_value": item["value"] * 1.1, # 10% increase
|
|
"status": "processed"
|
|
}
|
|
processed_data.append(processed_item)
|
|
|
|
print(f"✅ DataProcessor: Processed {len(processed_data)} records")
|
|
|
|
# Send to analyzer
|
|
await self.publish_message(
|
|
DataMessage(
|
|
data=processed_data,
|
|
stage="analysis",
|
|
metadata={**message.metadata, "processed_count": len(processed_data)}
|
|
),
|
|
DefaultTopicId()
|
|
)
|
|
|
|
@default_subscription
|
|
class DataAnalyzerAgent(RoutedAgent):
|
|
"""Agent that analyzes processed data and generates insights"""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__("Data analyzer agent that generates insights.")
|
|
|
|
@message_handler
|
|
async def handle_message(self, message: DataMessage, ctx: MessageContext) -> None:
|
|
if message.stage != "analysis":
|
|
return
|
|
|
|
print(f"🧠 DataAnalyzer: Analyzing {len(message.data)} records")
|
|
|
|
# Perform analysis
|
|
total_value = sum(item["processed_value"] for item in message.data)
|
|
avg_value = total_value / len(message.data)
|
|
categories = set(item["category"] for item in message.data)
|
|
|
|
analysis_results = {
|
|
"total_records": len(message.data),
|
|
"total_value": total_value,
|
|
"average_value": avg_value,
|
|
"unique_categories": len(categories),
|
|
"categories": list(categories)
|
|
}
|
|
|
|
print(f"📈 DataAnalyzer: Analysis complete")
|
|
print(f" • Total records: {analysis_results['total_records']}")
|
|
print(f" • Average value: {analysis_results['average_value']:.2f}")
|
|
print(f" • Categories: {', '.join(analysis_results['categories'])}")
|
|
|
|
# Send to reporter
|
|
await self.publish_message(
|
|
DataMessage(
|
|
data=message.data,
|
|
stage="reporting",
|
|
metadata={
|
|
**message.metadata,
|
|
"analysis": analysis_results
|
|
}
|
|
),
|
|
DefaultTopicId()
|
|
)
|
|
|
|
@default_subscription
|
|
class ReportGeneratorAgent(RoutedAgent):
|
|
"""Agent that generates final reports"""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__("Report generator agent that creates final output.")
|
|
|
|
@message_handler
|
|
async def handle_message(self, message: DataMessage, ctx: MessageContext) -> None:
|
|
if message.stage != "reporting":
|
|
return
|
|
|
|
print(f"📝 ReportGenerator: Generating final report")
|
|
|
|
analysis = message.metadata.get("analysis", {})
|
|
|
|
report = f"""
|
|
🎯 DATA PROCESSING REPORT
|
|
========================
|
|
Source: {message.metadata.get('source', 'Unknown')}
|
|
Processing Date: {message.metadata.get('timestamp', 'Unknown')}
|
|
|
|
📊 SUMMARY STATISTICS:
|
|
• Total Records Processed: {analysis.get('total_records', 0)}
|
|
• Total Value: ${analysis.get('total_value', 0):,.2f}
|
|
• Average Value: ${analysis.get('average_value', 0):,.2f}
|
|
• Unique Categories: {analysis.get('unique_categories', 0)}
|
|
• Categories Found: {', '.join(analysis.get('categories', []))}
|
|
|
|
✅ Processing pipeline completed successfully!
|
|
"""
|
|
|
|
print(report)
|
|
print("🎉 Multi-agent data processing workflow completed!")
|
|
|
|
async def run_data_processing_pipeline():
|
|
"""Run a complete data processing pipeline using multiple AutoGen agents"""
|
|
|
|
print("🚀 Starting AutoGen Data Processing Pipeline")
|
|
print("=" * 60)
|
|
|
|
# Create runtime
|
|
runtime = SingleThreadedAgentRuntime()
|
|
|
|
# Register all agents
|
|
await DataCollectorAgent.register(
|
|
runtime,
|
|
"collector",
|
|
lambda: DataCollectorAgent(),
|
|
)
|
|
|
|
await DataProcessorAgent.register(
|
|
runtime,
|
|
"processor",
|
|
lambda: DataProcessorAgent(),
|
|
)
|
|
|
|
await DataAnalyzerAgent.register(
|
|
runtime,
|
|
"analyzer",
|
|
lambda: DataAnalyzerAgent(),
|
|
)
|
|
|
|
await ReportGeneratorAgent.register(
|
|
runtime,
|
|
"reporter",
|
|
lambda: ReportGeneratorAgent(),
|
|
)
|
|
|
|
# Start runtime
|
|
runtime.start()
|
|
print("🤖 AutoGen runtime with 4 agents started")
|
|
|
|
# Trigger the pipeline
|
|
initial_message = DataMessage(
|
|
data=[],
|
|
stage="collection",
|
|
metadata={
|
|
"source": "customer_database",
|
|
"timestamp": "2024-01-15T10:30:00Z",
|
|
"pipeline_id": "data_proc_001"
|
|
}
|
|
)
|
|
|
|
print("📨 Triggering data processing pipeline...")
|
|
await runtime.send_message(
|
|
initial_message,
|
|
AgentId("collector", "default")
|
|
)
|
|
|
|
# Wait for completion
|
|
await runtime.stop_when_idle()
|
|
|
|
print("=" * 60)
|
|
print("✨ Pipeline completed! Check AgentOps dashboard for detailed agent traces.")
|
|
|
|
# Run the pipeline
|
|
if __name__ == "__main__":
|
|
asyncio.run(run_data_processing_pipeline())
|
|
```
|
|
</CodeGroup>
|
|
|
|
## Examples
|
|
|
|
<CardGroup cols={2}>
|
|
<Card title="Agent Chat Example" icon="notebook" href="/v2/examples/autogen">
|
|
Basic multi-agent chat functionality
|
|
</Card>
|
|
<Card title="Math Agent Example" icon="notebook" href="https://github.com/AgentOps-AI/agentops/blob/main/examples/autogen/MathAgent.ipynb" newTab={true}>
|
|
Demonstrates an agent specialized for mathematical problem-solving.
|
|
</Card>
|
|
</CardGroup>
|
|
|
|
Visit your [AgentOps Dashboard](https://app.agentops.ai) to see detailed traces of your AutoGen agent interactions, performance metrics, and workflow analytics.
|
|
|
|
|
|
<script type="module" src="/scripts/github_stars.js"></script>
|
|
<script type="module" src="/scripts/scroll-img-fadein-animation.js"></script>
|
|
<script type="module" src="/scripts/button_heartbeat_animation.js"></script>
|
|
<script type="module" src="/scripts/adjust_api_dynamically.js"></script>
|