​
Basic Usage

​
Simple Trace Creation

The @trace decorator automatically creates a trace span that encompasses the entire function execution. You can optionally specify custom names and tags to better organize and categorize your traces:

from agentops.sdk.decorators import trace
import agentops

# Initialize AgentOps
agentops.init("your-api-key", auto_start_session=False)

@trace(name="customer-workflow", tags=["production", "customer-service"])
def my_workflow():
    """A simple workflow wrapped in a trace"""
    print("🚀 Starting customer workflow...")
    print("📋 Processing customer request...")
    # Your application logic here
    print("✅ Customer workflow completed successfully!")
    return "Workflow completed"

# Run the function - this creates and manages the trace automatically
print("🎬 Running traced workflow...")
result = my_workflow()
print(f"📊 Result: {result}")

Both name and tags parameters are optional. If no name is provided, the function name will be used as the trace name.

​
Custom Trace Names

You can specify custom names for your traces:

@trace(name="customer-onboarding-flow")
def onboard_customer(customer_data):
    """Customer onboarding process"""
    print(f"👋 Onboarding customer: {customer_data['name']}")
    print("📝 Creating customer profile...")
    print("📧 Sending welcome email...")
    print("✅ Customer onboarding complete!")
    return f"Onboarded customer: {customer_data['name']}"

@trace(name="data-processing-pipeline")
def process_data(input_data):
    """Data processing workflow"""
    print(f"📊 Processing {len(input_data)} data items...")
    print("🔄 Applying transformations...")
    print("✅ Data processing complete!")
    return f"Processed {len(input_data)} items"

# Usage examples
customer = {"name": "Alice Johnson", "email": "alice@example.com"}
result1 = onboard_customer(customer)
print(f"📋 Onboarding result: {result1}")

data_items = ["item1", "item2", "item3", "item4", "item5"]
result2 = process_data(data_items)
print(f"📋 Processing result: {result2}")

​
Adding Tags to Traces

Tags help categorize and filter traces in your dashboard:

@trace(tags=["production", "high-priority"])
def critical_workflow():
    """Critical production workflow"""
    print("🚨 Executing critical production workflow...")
    print("⚡ High priority processing...")
    print("✅ Critical task completed successfully!")
    return "Critical task completed"

@trace(name="user-analysis", tags=["analytics", "user-behavior"])
def analyze_user_behavior(user_id):
    """Analyze user behavior patterns"""
    print(f"🔍 Analyzing behavior for user: {user_id}")
    print("📈 Gathering user interaction data...")
    print("🧠 Running behavior analysis algorithms...")
    print("✅ User behavior analysis complete!")
    return f"Analysis complete for user {user_id}"

# Usage examples
print("🎬 Running critical workflow...")
result1 = critical_workflow()
print(f"📊 Critical workflow result: {result1}")

print("\n🎬 Running user analysis...")
result2 = analyze_user_behavior("user_12345")
print(f"📊 Analysis result: {result2}")

​
Integration with Other Decorators

​
Combining with Agent and Operation Decorators

The @trace decorator works seamlessly with other AgentOps decorators:

import agentops
from agentops.sdk.decorators import trace, agent, operation, tool

# Initialize AgentOps without auto-starting session since we use @trace
agentops.init("your-api-key", auto_start_session=False)

@agent
class DataAnalysisAgent:
    def __init__(self):
        print("🤖 DataAnalysisAgent initialized")
    
    @operation
    def collect_data(self, source):
        print(f"📊 Collecting data from {source}...")
        data = f"Data collected from {source}"
        print(f"✅ Data collection complete: {data}")
        return data
    
    @tool(cost=0.05)
    def analyze_data(self, data):
        print(f"🧠 Analyzing data: {data}")
        analysis = f"Analysis of {data}"
        print(f"✅ Analysis complete: {analysis}")
        return analysis
    
    @operation
    def generate_report(self, analysis):
        print(f"📝 Generating report from: {analysis}")
        report = f"Report: {analysis}"
        print(f"✅ Report generated: {report}")
        return report

@trace(name="complete-analysis-workflow")
def run_analysis_workflow(data_source):
    """Complete data analysis workflow"""
    print(f"🚀 Starting analysis workflow for: {data_source}")
    print("=" * 50)
    
    agent = DataAnalysisAgent()
    
    # Collect data
    print("\n📋 Step 1: Data Collection")
    data = agent.collect_data(data_source)
    
    # Analyze data
    print("\n📋 Step 2: Data Analysis")
    analysis = agent.analyze_data(data)
    
    # Generate report
    print("\n📋 Step 3: Report Generation")
    report = agent.generate_report(analysis)
    
    print("\n🎉 Workflow completed successfully!")
    print("=" * 50)
    
    return {
        "source": data_source,
        "report": report
    }

# Usage
print("🎬 Running complete analysis workflow...")
result = run_analysis_workflow("customer_database")
print(f"\n📊 Final Result:")
print(f"   Source: {result['source']}")
print(f"   Report: {result['report']}")

​
Async Function Support

The @trace decorator fully supports async functions:

import asyncio
import agentops
from agentops.sdk.decorators import trace, operation

# Initialize AgentOps without auto-starting session since we use @trace
agentops.init("your-api-key", auto_start_session=False)

@operation
async def fetch_user_data(user_id):
    """Simulate async data fetching"""
    print(f"🌐 Fetching data for user: {user_id}")
    await asyncio.sleep(1)  # Simulate API call
    data = f"User data for {user_id}"
    print(f"✅ Data fetched: {data}")
    return data

@operation
async def process_user_data(user_data):
    """Simulate async data processing"""
    print(f"⚙️ Processing user data: {user_data}")
    await asyncio.sleep(0.5)  # Simulate processing
    processed = f"Processed: {user_data}"
    print(f"✅ Processing complete: {processed}")
    return processed

@trace(name="async-user-workflow")
async def async_user_workflow(user_id):
    """Async workflow for user processing"""
    print(f"🚀 Starting async workflow for user: {user_id}")
    print("=" * 45)
    
    print("\n📋 Step 1: Fetching user data")
    user_data = await fetch_user_data(user_id)
    
    print("\n📋 Step 2: Processing user data")
    processed_data = await process_user_data(user_data)
    
    print("\n🎉 Async workflow completed!")
    print("=" * 45)
    
    return processed_data

# Usage
async def main():
    print("🎬 Running async user workflow...")
    result = await async_user_workflow("user_123")
    print(f"\n📊 Final Result: {result}")
    print("✨ Check your AgentOps dashboard to see the traced async workflow!")

# Run the async workflow
print("🔄 Starting async demo...")
asyncio.run(main())

​
Error Handling and Trace States

​
Automatic Error Handling

The @trace decorator automatically handles exceptions and sets appropriate trace states:

import agentops
from agentops.sdk.decorators import trace

# Initialize AgentOps without auto-starting session since we use @trace
agentops.init("your-api-key", auto_start_session=False)

@trace(name="error-prone-workflow")
def risky_operation():
    """Operation that might fail"""
    import random
    
    print("🎲 Running risky operation...")
    print("⚠️ This operation has a 50% chance of failure")
    
    if random.random() < 0.5:
        print("❌ Operation failed!")
        raise ValueError("Random failure occurred")
    
    print("✅ Operation succeeded!")
    return "Operation succeeded"

# The trace will automatically be marked with failure state if an exception occurs
print("🎬 Testing automatic error handling...")
for i in range(3):
    print(f"\n🔄 Attempt {i+1}:")
    try:
        result = risky_operation()
        print(f"📊 Success: {result}")
        break
    except ValueError as e:
        print(f"📊 Operation failed: {e}")
        print("🔍 Trace automatically ended with error state")

​
Custom Error Handling

You can implement custom error handling within traced functions:

@trace(name="robust-workflow")
def robust_operation(data):
    """Operation with custom error handling"""
    print(f"🚀 Starting robust operation with data: {data}")
    
    try:
        # Risky operation
        if not data:
            print("⚠️ No data provided!")
            raise ValueError("No data provided")
        
        # Process data
        print("⚙️ Processing data...")
        result = f"Processed: {data}"
        print(f"✅ Processing successful: {result}")
        return {"success": True, "result": result}
        
    except ValueError as e:
        # Handle specific errors
        print(f"❌ Validation error: {e}")
        return {"success": False, "error": str(e)}
    except Exception as e:
        # Handle unexpected errors
        print(f"💥 Unexpected error: {e}")
        return {"success": False, "error": f"Unexpected error: {str(e)}"}

# Usage examples
print("\n🎬 Testing custom error handling...")

print("\n📋 Test 1: Valid data")
result1 = robust_operation("valid_data")
print(f"📊 Result: {result1}")

print("\n📋 Test 2: Empty data")
result2 = robust_operation("")
print(f"📊 Result: {result2}")

print("\n📋 Test 3: None data")
result3 = robust_operation(None)
print(f"📊 Result: {result3}")

​
Real-World Examples

​
E-commerce Order Processing

from agentops.sdk.decorators import trace, agent, operation, tool
from openai import OpenAI
import agentops

agentops.init("your-api-key", auto_start_session=False)

@agent
class OrderProcessor:
    def __init__(self):
        print("🛒 OrderProcessor initialized")
    
    @tool(cost=0.01)
    def validate_payment(self, payment_info):
        """Payment validation service"""
        print(f"💳 Validating payment: {payment_info['card']}")
        result = {"valid": True, "transaction_id": "txn_123"}
        print(f"✅ Payment validation successful: {result['transaction_id']}")
        return result
    
    @tool(cost=0.02)
    def check_inventory(self, product_id, quantity):
        """Inventory check service"""
        print(f"📦 Checking inventory for {product_id} (qty: {quantity})")
        result = {"available": True, "reserved": quantity}
        print(f"✅ Inventory check complete: {quantity} units available")
        return result
    
    @operation
    def calculate_shipping(self, address, items):
        """Calculate shipping costs"""
        print(f"🚚 Calculating shipping to {address['city']}, {address['state']}")
        result = {"cost": 9.99, "method": "standard"}
        print(f"✅ Shipping calculated: ${result['cost']} ({result['method']})")
        return result
    
    @tool(cost=0.005)
    def send_confirmation_email(self, email, order_details):
        """Email service"""
        print(f"📧 Sending confirmation email to {email}")
        result = f"Confirmation sent to {email}"
        print(f"✅ Email sent successfully")
        return result

@trace(name="order-processing", tags=["ecommerce", "orders"])
def process_order(order_data):
    """Complete order processing workflow"""
    print(f"🚀 Starting order processing for {order_data['customer_email']}")
    print("=" * 60)
    
    processor = OrderProcessor()
    
    try:
        # Validate payment
        print("\n📋 Step 1: Payment Validation")
        payment_result = processor.validate_payment(order_data["payment"])
        if not payment_result["valid"]:
            print("❌ Payment validation failed!")
            return {"success": False, "error": "Payment validation failed"}
        
        # Check inventory for all items
        print("\n📋 Step 2: Inventory Check")
        for item in order_data["items"]:
            inventory_result = processor.check_inventory(
                item["product_id"], 
                item["quantity"]
            )
            if not inventory_result["available"]:
                print(f"❌ Item {item['product_id']} not available!")
                return {"success": False, "error": f"Item {item['product_id']} not available"}
        
        # Calculate shipping
        print("\n📋 Step 3: Shipping Calculation")
        shipping = processor.calculate_shipping(
            order_data["shipping_address"], 
            order_data["items"]
        )
        
        # Send confirmation
        print("\n📋 Step 4: Confirmation Email")
        confirmation = processor.send_confirmation_email(
            order_data["customer_email"],
            {
                "items": order_data["items"],
                "shipping": shipping,
                "payment": payment_result
            }
        )
        
        print("\n🎉 Order processing completed successfully!")
        print("=" * 60)
        
        return {
            "success": True,
            "order_id": "ORD_12345",
            "payment": payment_result,
            "shipping": shipping,
            "confirmation": confirmation
        }
        
    except Exception as e:
        print(f"💥 Order processing failed: {e}")
        return {"success": False, "error": str(e)}

# Usage
print("🎬 Running e-commerce order processing demo...")

order = {
    "customer_email": "customer@example.com",
    "payment": {"card": "****1234", "amount": 99.99},
    "items": [{"product_id": "PROD_001", "quantity": 2}],
    "shipping_address": {"city": "New York", "state": "NY"}
}

result = process_order(order)

print(f"\n📊 ORDER PROCESSING RESULT:")
print(f"   Success: {result['success']}")
if result['success']:
    print(f"   Order ID: {result['order_id']}")
    print(f"   Transaction: {result['payment']['transaction_id']}")
    print(f"   Shipping: ${result['shipping']['cost']}")
else:
    print(f"   Error: {result['error']}")

​
Data Analysis Workflow

from agentops.sdk.decorators import trace, agent, operation, tool
from openai import OpenAI
import agentops

agentops.init("your-api-key", auto_start_session=False)

@agent
class DataAnalysisAgent:
    def __init__(self):
        self.client = OpenAI()
        print("🤖 DataAnalysisAgent initialized")
    
    @operation
    def collect_data(self, source):
        """Simulate data collection"""
        print(f"📊 Collecting data from {source}...")
        data = f"Raw data collected from {source}: [sample_data_1, sample_data_2, sample_data_3]"
        print(f"✅ Data collection complete: {len(data)} characters collected")
        return data
    
    @operation
    def analyze_data_with_llm(self, data):
        """Use LLM to analyze the collected data"""
        print("🧠 Analyzing data with LLM...")
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "You are a data analyst. Analyze the provided data and give insights."},
                {"role": "user", "content": f"Please analyze this data: {data}"}
            ]
        )
        analysis = response.choices[0].message.content
        print(f"✅ LLM analysis complete: {len(analysis)} characters generated")
        return analysis
    
    @tool(cost=0.05)
    def generate_visualization(self, analysis):
        """Generate data visualization"""
        print("📈 Generating visualization...")
        visualization = f"Chart generated for: {analysis[:50]}..."
        print(f"✅ Visualization generated: {visualization}")
        return visualization
    
    @operation
    def generate_report(self, analysis, visualization):
        """Generate final report using LLM"""
        print("📝 Generating final report with LLM...")
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "You are a report writer. Create a professional data analysis report."},
                {"role": "user", "content": f"Create a report based on this analysis: {analysis} and visualization: {visualization}"}
            ]
        )
        report = response.choices[0].message.content
        print(f"✅ Final report generated: {len(report)} characters")
        return report

@trace(name="data-analysis-workflow", tags=["analytics", "reporting"])
def run_data_analysis(data_source):
    """Complete data analysis workflow with LLM integration"""
    print(f"🚀 Starting data analysis workflow for: {data_source}")
    print("=" * 60)
    
    agent = DataAnalysisAgent()
    
    # Collect data
    print("\n📋 Step 1: Data Collection")
    raw_data = agent.collect_data(data_source)
    
    # Analyze data using LLM
    print("\n📋 Step 2: LLM Analysis")
    analysis = agent.analyze_data_with_llm(raw_data)
    
    # Generate visualization
    print("\n📋 Step 3: Visualization Generation")
    visualization = agent.generate_visualization(analysis)
    
    # Generate final report using LLM
    print("\n📋 Step 4: Report Generation")
    report = agent.generate_report(analysis, visualization)
    
    print("\n🎉 Workflow completed successfully!")
    print("=" * 60)
    
    return {
        "source": data_source,
        "raw_data": raw_data,
        "analysis": analysis,
        "visualization": visualization,
        "final_report": report
    }

# Usage
print("🎬 Running data analysis workflow demo...")

result = run_data_analysis("customer_database")

print(f"\n📊 ANALYSIS RESULTS:")
print(f"   Data Source: {result['source']}")
print(f"   Raw Data: {result['raw_data'][:80]}...")
print(f"   Analysis Preview: {result['analysis'][:100]}...")
print(f"   Visualization: {result['visualization']}")
print(f"   Final Report Preview: {result['final_report'][:150]}...")

print(f"\n✨ Analysis complete! Check your AgentOps dashboard to see the traced workflow.")

​
Best Practices

​
1. Use Meaningful Names

Choose descriptive names that clearly indicate what the trace represents:

# Good
@trace(name="user-authentication-flow")
def authenticate_user(credentials):
    pass

@trace(name="payment-processing-pipeline")
def process_payment(payment_data):
    pass

# Less descriptive
@trace(name="trace1")
def some_function():
    pass

​
2. Add Relevant Tags

Use tags to categorize traces for easier filtering and analysis:

@trace(name="order-fulfillment", tags=["ecommerce", "fulfillment", "high-priority"])
def fulfill_order(order_id):
    pass

@trace(name="data-sync", tags=["background-job", "data-processing"])
def sync_data():
    pass

​
3. Keep Traces Focused

Each trace should represent a logical unit of work:

# Good - focused on a single workflow
@trace(name="customer-onboarding")
def onboard_customer(customer_data):
    validate_customer(customer_data)
    create_account(customer_data)
    send_welcome_email(customer_data)

# Less focused - mixing different concerns
@trace(name="mixed-operations")
def do_everything():
    onboard_customer(data1)
    process_orders(data2)
    generate_reports(data3)

​
4. Handle Errors Appropriately

Implement proper error handling within traced functions:

@trace(name="data-processing")
def process_data(data):
    try:
        # Main processing logic
        result = complex_processing(data)
        return {"success": True, "result": result}
    except ValidationError as e:
        # Expected errors
        return {"success": False, "error": "validation_failed", "details": str(e)}
    except Exception as e:
        # Unexpected errors
        logger.error(f"Unexpected error in data processing: {e}")
        return {"success": False, "error": "processing_failed"}

The @trace decorator provides a powerful and flexible way to organize your application’s telemetry data. By creating logical groupings of operations, you can better understand your application’s behavior and performance characteristics in the AgentOps dashboard.