Decorators provide a clean, declarative way to instrument your code. They automatically create spans with semantic meaning, making your traces easier to understand and navigate in the Netra dashboard.
Decorators: Overview
Netra provides four decorators, each designed for a specific type of operation:
| Decorator | Purpose | Span Type |
|---|
@workflow | High-level business transactions | Span |
@agent | AI agents or orchestrators | Agent |
@task | Individual units of work | Tool |
@span | Generic operations with customizable type | Configurable |
@workflow
Use @workflow to mark high-level business transactions or processes. Workflows typically represent complete user-facing operations that may involve multiple steps.
from netra import workflow
@workflow
def process_customer_order(order_id: str):
order = fetch_order(order_id)
inventory = check_inventory(order.items)
payment = process_payment(order)
shipment = create_shipment(order)
return {"order": order, "shipment": shipment}
# With custom name
@workflow(name="order-fulfillment")
def fulfill_order(order_id: str):
# ...
pass
@workflow:
- User-initiated actions (e.g., “submit order”, “generate report”)
- End-to-end processes that span multiple operations
- Top-level entry points in your application
@agent
Use @agent to mark AI agents or autonomous components that make decisions. Agent spans help you track reasoning steps and decision-making processes.
from netra import agent
@agent
class CustomerSupportAgent:
def handle_ticket(self, ticket: dict):
analysis = self.analyze_ticket(ticket)
response = self.generate_response(analysis)
return response
def analyze_ticket(self, ticket: dict):
# Analysis logic
pass
def generate_response(self, analysis: dict):
# Response generation
pass
# Function-based agent
@agent
def research_agent(query: str):
sources = search_sources(query)
synthesis = synthesize_information(sources)
return synthesis
@agent:
- Autonomous AI components that make decisions
- Multi-step reasoning processes
- Components that orchestrate other tools or services
@task
Use @task to mark individual units of work. Tasks are typically discrete operations that perform a specific function within a larger workflow.
from netra import task
@task
def fetch_user_profile(user_id: str):
response = requests.get(f"/api/users/{user_id}")
return response.json()
@task
def send_notification(user_id: str, message: str):
notification_service.send(user_id, message)
@task(name="validate-input")
def validate_order_input(order: dict):
if not order.get("items"):
raise ValueError("Order must contain at least one item")
return True
@task:
- Individual operations within a workflow
- Tool calls or function executions
- Database operations, API calls, or computations
@span
Use @span for generic tracing with full control over the span type. This is the most flexible decorator, allowing you to specify the exact span type.
from netra import span
from netra import SpanType
# Default span type
@span
def process_data(data: list):
return [transform(item) for item in data]
# With custom span type
@span(as_type=SpanType.GENERATION)
def generate_embedding(text: str):
embedding = embedding_model.embed(text)
return embedding
@span(name="vector-search", as_type=SpanType.TOOL)
def search_vector_db(query: str, top_k: int):
return vector_store.search(query, top_k)
Available Span Types
| Span Type | Use Case |
|---|
SpanType.SPAN | Generic operations (default) |
SpanType.GENERATION | LLM text generation |
SpanType.EMBEDDING | Vector embedding operations |
SpanType.TOOL | Tool or function calls |
SpanType.AGENT | AI agent operations |
Decorating Classes
When you apply a decorator to a class, all public methods of that class are automatically instrumented.
from netra import agent, task
@agent
class OrderProcessor:
# All public methods are traced as part of the agent
def process_order(self, order: dict):
self.validate_order(order)
self.charge_payment(order)
self.fulfill_order(order)
@task # Override with specific decorator
def validate_order(self, order: dict):
# Validation logic
pass
def charge_payment(self, order: dict):
# Payment logic
pass
def fulfill_order(self, order: dict):
# Fulfillment logic
pass
Async and Generator Support
Decorators work seamlessly with async functions and generators:
from netra import workflow, task
# Async functions
@workflow
async def async_workflow():
result = await some_async_operation()
return result
# Async generators
@task
async def stream_results(query: str):
stream = await openai.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": query}],
stream=True,
)
async for chunk in stream:
yield chunk.choices[0].delta.content or ""
# Sync generators
@task
def generate_items(count: int):
for i in range(count):
yield process_item(i)
For streaming responses, the span remains open until the stream is fully consumed. This ensures accurate latency measurements for streaming operations.
Automatic Parameter Capture
Decorators automatically capture function parameters as span attributes, making it easy to understand what inputs were provided:
from netra import task
@task
def search_products(query: str, category: str, limit: int = 10):
# Parameters are automatically captured:
# - query: "laptop"
# - category: "electronics"
# - limit: 10
return product_service.search(query, category, limit)
# Call the function
search_products("laptop", "electronics")
Exception Handling
Decorators automatically capture exceptions and mark spans with error status:
from netra import task
@task
def risky_operation(data: dict):
if not data.get("valid"):
raise ValueError("Invalid data provided")
# Span is automatically marked as ERROR
# Exception details are recorded
return process_data(data)
Combining Decorators with Manual Tracing
You can combine decorators with manual span operations for additional context:
from netra import workflow, Netra
@workflow
def process_order(order: dict):
# Add custom attributes to the current span
current_span = Netra.get_current_span()
if current_span:
current_span.set_attribute("order.id", order["id"])
current_span.set_attribute("order.total", order["total"])
# Add custom events
current_span.add_event("order-validated", {
"item_count": len(order["items"]),
})
result = fulfill_order(order)
if current_span:
current_span.add_event("order-fulfilled")
return result
Best Practices
-
Use semantic decorators - Choose the decorator that best describes the operation’s purpose (
@workflow for processes, @agent for AI components, @task for individual operations).
-
Name spans meaningfully - Use the
name parameter when the function name isn’t descriptive enough.
-
Don’t over-instrument - Focus on high-value operations. Not every function needs a decorator.
-
Combine with auto-instrumentation - Let auto-instrumentation handle LLM calls and database operations while using decorators for your application logic.
-
Use class decoration sparingly - Decorating an entire class instruments all methods, which may create noise. Consider decorating individual methods instead.
Learn More
