수동 트레이스 제어
수동 트레이스 제어 (Manual Trace Control)
start_trace와 end_trace 메서드로 트레이스를 고급 관리하는 방법을 소개해요. 기본 제어부터 실행 중 메타데이터 업데이트, 데코레이터 통합까지 알아볼게요.
출처: 문서
본문
기본 수동 트레이스 제어 (Basic Manual Trace Control)
트레이스 시작과 종료 (Starting and Ending Traces)
수동 트레이스 제어의 가장 기본적인 형태는 트레이스를 시작하고, 코드를 실행하고, 특정 상태로 트레이스를 종료하는 것입니다.
import agentops
# Initialize without automatic session creation
agentops.init("your-api-key", auto_start_session=False)
# Start a trace manually
trace = agentops.start_trace("my-workflow")
try:
# Your application logic here
result = perform_some_operation()
# End the trace successfully
agentops.end_trace(trace, "Success")
except Exception as e:
# End the trace with failure state
agentops.end_trace(trace, "Indeterminate")
트레이스 이름과 태그 (Trace Names and Tags)
트레이스를 시작할 때 의미 있는 이름과 태그를 제공할 수 있어요.
# Start a trace with custom name and tags
trace = agentops.start_trace(
trace_name="customer-service-workflow",
tags=["customer-123", "priority-high", "support"]
)
선택적 트레이스 종료 배치 처리 (Batch Processing with Selective Trace Ending)
배치 처리 시나리오에서는 처리 결과에 따라 트레이스를 선택적으로 종료할 수 있습니다.
import agentops
# Initialize AgentOps
agentops.init("your-api-key", auto_start_session=False)
# Sample batch items to process
batch_items = [
{"id": 1, "data": "item_1_data", "valid": True},
{"id": 2, "data": "item_2_data", "valid": False},
{"id": 3, "data": "item_3_data", "valid": True},
]
@agentops.operation(name="process_item")
def process_item(item):
"""Simulate processing an item"""
if not item.get("valid", False):
raise ValueError(f"Invalid item: {item['id']}")
return {"processed": True, "result": f"Processed {item['data']}"}
# Start traces for batch items
for i, item in enumerate(batch_items):
trace = agentops.start_trace(f"batch_item_{i+1}")
try:
result = process_item(item)
if result.get("processed"):
agentops.end_trace(trace, "Success")
else:
agentops.end_trace(trace, "Indeterminate")
except Exception as e:
agentops.end_trace(trace, "Error")
실행 중 트레이스 메타데이터 업데이트 (Updating Trace Metadata During Execution)
update_trace_metadata 함수를 사용하면 실행 중인 트레이스의 메타데이터를 언제든지 업데이트할 수 있어요. 컨텍스트 추가, 진행률 추적, 중간 결과 저장에 유용합니다.
기본 메타데이터 업데이트 (Basic Metadata Updates)
import agentops
# Initialize AgentOps
agentops.init("your-api-key", auto_start_session=False)
# Start a trace with initial tags
trace = agentops.start_trace("ai-agent-workflow", tags=["startup", "initialization"])
# Your AI agent code runs here...
process_user_request()
# Update metadata with results
agentops.update_trace_metadata({
"operation_name": "AI Agent Processing Complete",
"stage": "completed",
"response_quality": "high",
"tags": ["ai-agent", "completed", "success"] # Tags show current status
})
# End the trace
agentops.end_trace(trace, "Success")
시맨틱 컨벤션 지원 (Semantic Convention Support)
이 함수는 가능할 때 사용자 친화적인 키를 시맨틱 컨벤션으로 자동 매핑합니다.
# These keys will be mapped to semantic conventions
agentops.update_trace_metadata({
"operation_name": "AI Agent Data Processing",
"tags": ["production", "batch-job", "gpt-4"], # Maps to core.tags
"agent_name": "DataProcessorAgent", # Maps to agent.name
"workflow_name": "Intelligent ETL Pipeline", # Maps to workflow.name
})
커스텀 프리픽스가 있는 고급 메타데이터 (Advanced Metadata with Custom Prefix)
메타데이터 속성에 커스텀 프리픽스를 지정할 수 있습니다.
# Use a custom prefix for business-specific metadata
agentops.update_trace_metadata({
"customer_id": "CUST_456",
"order_value": 99.99,
"payment_method": "credit_card",
"agent_interaction": "customer_support"
}, prefix="business")
# Results in:
# business.customer_id = "CUST_456"
# business.order_value = 99.99
# business.payment_method = "credit_card"
실제 예제: 진행률 추적 (Real-World Example: Progress Tracking)
메타데이터 업데이트로 복잡한 워크플로우의 진행률을 추적하는 방법입니다.
import agentops
from agentops.sdk.decorators import operation
agentops.init(auto_start_session=False)
@operation
def process_batch(batch_data):
# Simulate batch processing
return f"Processed {len(batch_data)} items"
def run_etl_pipeline(data_batches):
"""ETL pipeline with progress tracking via metadata"""
trace = agentops.start_trace("etl-pipeline", tags=["data-processing"])
total_batches = len(data_batches)
processed_records = 0
# Initial metadata
agentops.update_trace_metadata({
"operation_name": "ETL Pipeline Execution",
"pipeline_stage": "starting",
"total_batches": total_batches,
"processed_batches": 0,
"processed_records": 0,
"estimated_completion": "calculating...",
"tags": ["etl", "data-processing", "async-operation"]
})
try:
for i, batch in enumerate(data_batches):
# Update progress
agentops.update_trace_metadata({
"pipeline_stage": "processing",
"current_batch": i + 1,
"processed_batches": i,
"progress_percentage": round((i / total_batches) * 100, 2)
})
# Process the batch
result = process_batch(batch)
processed_records += len(batch)
# Update running totals
agentops.update_trace_metadata({
"processed_records": processed_records,
"last_batch_result": result
})
# Final metadata update
agentops.update_trace_metadata({
"operation_name": "ETL Pipeline Completed",
"pipeline_stage": "completed",
"processed_batches": total_batches,
"progress_percentage": 100.0,
"completion_status": "success",
"total_execution_time": "calculated_automatically",
"tags": ["etl", "completed", "success"]
})
agentops.end_trace(trace, "Success")
except Exception as e:
# Error metadata
agentops.update_trace_metadata({
"operation_name": "ETL Pipeline Failed",
"pipeline_stage": "failed",
"error_message": str(e),
"completion_status": "error",
"failed_at_batch": i + 1 if 'i' in locals() else 0,
"tags": ["etl", "failed", "error"]
})
agentops.end_trace(trace, "Error")
raise
# Example usage
data_batches = [
["record1", "record2", "record3"],
["record4", "record5"],
["record6", "record7", "record8", "record9"]
]
run_etl_pipeline(data_batches)
지원되는 데이터 타입 (Supported Data Types)
update_trace_metadata 함수는 다양한 데이터 타입을 지원합니다.
agentops.update_trace_metadata({
"operation_name": "Multi-type Data Example",
"successful_operation": True,
"tags": ["example", "demo", "multi-agent"],
"processing_steps": ["validation", "transformation", "output"]
})
# Note: Lists are automatically converted to JSON strings for OpenTelemetry compatibility
데코레이터와의 통합 (Integration with Decorators)
수동 트레이스 제어는 AgentOps 데코레이터와 원활하게 동작합니다.
import agentops
from agentops.sdk.decorators import agent, operation, tool
agentops.init("your-api-key", auto_start_session=False)
@agent
class CustomerServiceAgent:
@operation
def analyze_request(self, request):
return f"Analyzed: {request}"
@tool(cost=0.02)
def lookup_customer(self, customer_id):
return f"Customer data for {customer_id}"
# Manual trace with decorated components
trace = agentops.start_trace("customer-service")
try:
agent = CustomerServiceAgent()
customer_data = agent.lookup_customer("CUST_123")
analysis = agent.analyze_request("billing issue")
agentops.end_trace(trace, "Success")
except Exception as e:
agentops.end_trace(trace, "Error")
실제 예제 (Real-World Example)
고객 서비스 애플리케이션에서 수동 트레이스 제어를 보여주는 포괄적인 예제입니다.
import agentops
from agentops.sdk.decorators import agent, operation, tool
from openai import OpenAI
agentops.init(auto_start_session=False)
client = OpenAI()
@operation
def analyze_sentiment(text):
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": f"Analyze sentiment: {text}"}]
)
return response.choices[0].message.content.strip()
@tool(cost=0.01)
def lookup_order(order_id):
return f"Order {order_id} details"
def process_customer_requests(requests):
"""Process multiple customer requests with individual trace tracking"""
results = []
for i, request in enumerate(requests):
trace = agentops.start_trace(
f"customer_request_{i+1}",
tags=["customer-service", request.get("priority", "normal")]
)
try:
sentiment = analyze_sentiment(request["message"])
if "order" in request:
order_info = lookup_order(request["order"])
if "positive" in sentiment.lower() or "neutral" in sentiment.lower():
agentops.end_trace(trace, "Success")
results.append({"status": "resolved", "sentiment": sentiment})
else:
agentops.end_trace(trace, "Escalation_Required")
results.append({"status": "escalated", "sentiment": sentiment})
except Exception as e:
agentops.end_trace(trace, "Error")
results.append({"status": "error", "error": str(e)})
return results
customer_requests = [
{"message": "I love this product!", "priority": "low"},
{"message": "My order is completely wrong!", "order": "12345", "priority": "high"},
{"message": "When will my package arrive?", "order": "67890", "priority": "normal"}
]
results = process_customer_requests(customer_requests)
print(f"Processed {len(results)} customer requests")
이 예제는 다음을 보여줍니다:
- 각 고객 요청에 대한 개별 트레이스 관리
- 데코레이터된 에이전트와 도구와의 통합
- 비즈니스 로직에 기반한 서로 다른 종료 상태
- 적절한 트레이스 상태를 사용한 올바른 오류 처리
- 분류를 위한 태그 사용