AutoGen
AutoGen
AgentOps를 Microsoft AutoGen과 통합해 멀티 에이전트 워크플로우를 추적하는 방법을 소개해요. 설치, API 키 설정, 그리고 카운트다운·데이터 처리 예제까지 알아볼게요.
출처: 문서
본문
AutoGen은 멀티 에이전트 대화형 AI 시스템을 구축하는 Microsoft의 프레임워크예요. AgentOps는 AutoGen과의 원활한 통합을 제공하여 멀티 에이전트 워크플로우를 추적하고 모니터링할 수 있게 해줍니다.
설치 (Installation)
poetry add agentops autogen-core python-dotenv
uv pip install agentops autogen-core python-dotenv
API 키 설정 (Setting Up API Keys)
AutoGen을 AgentOps와 함께 사용하기 전에 API 키를 설정해야 해요. 다음을 얻을 수 있습니다.
- OPENAI_API_KEY: OpenAI Platform에서
- AGENTOPS_API_KEY: AgentOps Dashboard에서
그런 다음 환경 변수로 내보내거나 .env 파일에 설정할 수 있어요.
OPENAI_API_KEY="your_openai_api_key_here"
AGENTOPS_API_KEY="your_agentops_api_key_here"
그리고 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는 AutoGen 에이전트를 자동으로 계측하고 그 상호작용을 추적합니다. AutoGen 에이전트를 만들기 전에 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 theme={null}
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())
예제 (Examples)
AgentOps Dashboard를 방문하면 AutoGen 에이전트 상호작용의 상세 트레이스, 성능 메트릭, 워크플로우 분석을 볼 수 있어요.