​
Basic Manual Trace Control

​
Starting and Ending Traces

The most basic form of manual trace control involves starting a trace, executing your code, and then ending the trace with a specific state:

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

You can provide meaningful names and tags when starting traces:

# 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

For batch processing scenarios, you can selectively end traces based on processing results:

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

You can update metadata on running traces at any point during execution using the update_trace_metadata function. This is useful for adding context, tracking progress, or storing intermediate results.

​
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

The function automatically maps user-friendly keys to semantic conventions when possible:

# 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

You can specify a custom prefix for your metadata attributes:

# 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

Here’s how to use metadata updates to track progress through a complex workflow:

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

The update_trace_metadata function supports various data types:

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

Manual trace control works seamlessly with AgentOps decorators:

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

Here’s a comprehensive example showing manual trace control in a customer service application:

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")

This example demonstrates:

  • Individual trace management for each customer request
  • Integration with decorated agents and tools
  • Different end states based on business logic
  • Proper error handling with appropriate trace states
  • Use of tags for categorization