Trace Decorators

Hierarchical tracing decorators for automatic instrumentation of workflows, nodes, tools, and LLM calls with comprehensive metadata collection.

Features

• Hierarchical Tracing: Parent-child trace relationships
• Automatic Instrumentation: Decorator-based tracing
• Custom Metrics: Collect performance and business metrics
• LangSmith Integration: Seamless integration with LangSmith
• Async Support: Full async/await support

Quick Start

from packages.observability import trace_workflow, trace_node, trace_tool

@trace_workflow(name="my_workflow")
async def my_workflow():
    result = await my_node()
    return result

@trace_node(name="my_node")
async def my_node():
    return await my_tool()

@trace_tool(name="my_tool")
async def my_tool():
    return "result"

Decorator Types

Workflow Tracing

from packages.observability import trace_workflow

@trace_workflow(
    name="data_processing_workflow",
    tags=["data", "processing"],
    user_id="user123",
    session_id="session456"
)
async def process_data(data):
    # Your workflow logic
    return processed_data

Node Tracing

from packages.observability import trace_node

@trace_node(
    name="data_validation",
    tags=["validation"],
    parent_trace_id="workflow_trace_id"
)
async def validate_data(data):
    # Your node logic
    return validation_result

Tool Tracing

from packages.observability import trace_tool

@trace_tool(
    name="search_tool",
    tags=["search", "retrieval"]
)
async def search_documents(query):
    # Your tool logic
    return search_results

LLM Call Tracing

from packages.observability import trace_llm_call

@trace_llm_call(
    model="gpt-4",
    tags=["llm", "generation"]
)
async def generate_response(prompt):
    # Your LLM call
    return response

Reasoning Tracing

from packages.observability import trace_reasoning

@trace_reasoning(
    strategy="chain_of_thought",
    tags=["reasoning", "cot"]
)
async def reason_about_problem(problem):
    # Your reasoning logic
    return reasoning_result

Advanced Usage

Custom Metrics

from packages.observability import get_tracer

@trace_workflow(name="my_workflow")
async def my_workflow():
    tracer = get_tracer()
    
    # Add custom metrics
    tracer.add_metrics("workflow_trace_id", {
        "custom_metric": 42,
        "processing_time": 1.5,
        "data_size": 1000
    })
    
    return result

Hierarchical Traces

@trace_workflow(name="parent_workflow")
async def parent_workflow():
    # Get current trace ID
    tracer = get_tracer()
    parent_trace_id = tracer.start_trace("parent", "workflow")
    
    # Pass to child
    result = await child_workflow(parent_trace_id)
    
    tracer.end_trace(parent_trace_id)
    return result

@trace_workflow(name="child_workflow")
async def child_workflow(parent_trace_id):
    # Child inherits from parent
    return await process_data()

Trace Hierarchy

# Get trace hierarchy
tracer = get_tracer()
hierarchy = tracer.get_trace_hierarchy("root_trace_id")

print(hierarchy)
# {
#   "trace_id": "root_trace_id",
#   "component": "workflow",
#   "operation": "my_workflow",
#   "children": [
#     {
#       "trace_id": "child_trace_id",
#       "component": "node",
#       "operation": "my_node",
#       "children": []
#     }
#   ]
# }

LangSmith Integration

Automatic Tracing

# LangSmith integration is automatic
@trace_workflow(name="my_workflow")
async def my_workflow():
    # Traces are automatically sent to LangSmith
    return result

Custom Evaluators

from packages.observability import get_tracer

@trace_workflow(name="evaluation_workflow")
async def evaluation_workflow():
    tracer = get_tracer()
    
    # Create evaluation dataset
    dataset_id = tracer.create_evaluation_dataset(
        name="my_dataset",
        description="Test dataset",
        examples=[
            {"inputs": "test input", "outputs": "expected output"}
        ]
    )
    
    # Run evaluation
    results = tracer.run_evaluation(
        dataset_id=dataset_id,
        evaluator_func=my_evaluator
    )
    
    return results

Configuration

Global Tracer Setup

from packages.observability import set_tracer, HierarchicalTracer

# Configure tracer
tracer = HierarchicalTracer(
    project_name="my_project",
    api_key="your-langsmith-key",
    enable_metrics=True,
    enable_hierarchical=True
)

# Set as global tracer
set_tracer(tracer)

Trace Levels

from packages.observability import TraceLevel

@trace_workflow(
    name="my_workflow",
    level=TraceLevel.WORKFLOW
)
async def my_workflow():
    pass

@trace_node(
    name="my_node",
    level=TraceLevel.NODE
)
async def my_node():
    pass

Best Practices

1. Use Appropriate Decorators: Choose the right decorator for your component type
2. Add Meaningful Tags: Use tags for filtering and organization
3. Include User Context: Add user_id and session_id for user-specific traces
4. Monitor Performance: Use metrics to track performance
5. Handle Errors: Ensure traces are completed even on errors

Error Handling

@trace_workflow(name="error_handling_workflow")
async def workflow_with_error_handling():
    tracer = get_tracer()
    trace_id = tracer.start_trace("workflow", "error_handling")
    
    try:
        result = await risky_operation()
        tracer.end_trace(trace_id, status="completed")
        return result
    except Exception as e:
        tracer.end_trace(trace_id, status="failed", error=str(e))
        raise

Performance Considerations

• Minimal Overhead: Decorators add less than 1ms overhead
• Async Support: Full async/await support
• Batch Operations: Traces are batched for efficiency
• Memory Management: Automatic cleanup of old traces

API Reference

Decorators

Decorator	Description	Parameters
@trace_workflow	Trace workflows	name, tags, user_id, session_id
@trace_node	Trace nodes	name, tags, parent_trace_id
@trace_tool	Trace tools	name, tags, parent_trace_id
@trace_llm_call	Trace LLM calls	model, tags, parent_trace_id
@trace_reasoning	Trace reasoning	strategy, tags, parent_trace_id

Tracer Methods

Method	Description
start_trace()	Start a new trace
end_trace()	End a trace
add_metrics()	Add metrics to trace
get_trace_hierarchy()	Get trace hierarchy
create_evaluation_dataset()	Create evaluation dataset
run_evaluation()	Run evaluation