Tracing
Manual Tracing
Full control over tracing with manual span creation and instrumentation
Manual Tracing
Manual tracing gives you complete control over what gets traced. Use it for custom operations, non-LLM code, or when you need fine-grained control.
When to Use Manual Tracing
- Custom business logic: Data processing, validation, transformations
- Non-LLM operations: Database queries, API calls, file operations
- Complex pipelines: Multi-step workflows with custom structure
- Fine-grained control: Specific timing, attributes, or hierarchy
Basic Manual Tracing
Context Manager Pattern (Recommended)
from brokle import Brokle
client = Brokle(api_key="bk_...")
with client.start_as_current_span(name="process_document") as span:
# Your code here
result = analyze_document(document)
# Update span with output
span.update(output=result)
client.flush()import { Brokle } from 'brokle';
const client = new Brokle({ apiKey: 'bk_...' });
await client.startActiveSpan('process_document', async (span) => {
const result = await analyzeDocument(document);
client.updateCurrentSpan({ output: result });
return result;
});
await client.shutdown();Decorator Pattern (Python)
from brokle import observe
@observe(name="analyze_sentiment")
def analyze_sentiment(text: str) -> dict:
# Function is automatically traced
score = sentiment_model.predict(text)
return {"text": text, "score": score}
# Each call creates a trace
result = analyze_sentiment("Great product!")Function Pattern (JavaScript)
import { observe } from 'brokle';
const analyzeSentiment = observe(
{ name: 'analyze_sentiment' },
async (text) => {
const score = await sentimentModel.predict(text);
return { text, score };
}
);
const result = await analyzeSentiment('Great product!');Setting Span Input and Output
with client.start_as_current_span(name="transform_data") as span:
# Set input explicitly
span.update(input={"raw_data": data, "config": config})
# Perform operation
result = transform(data, config)
# Set output
span.update(output=result)await client.startActiveSpan('transform_data', async (span) => {
client.updateCurrentSpan({ input: { rawData: data, config } });
const result = await transform(data, config);
client.updateCurrentSpan({ output: result });
return result;
});Adding Attributes
Add custom key-value pairs for filtering and analysis:
with client.start_as_current_span(name="process_order") as span:
# Set individual attributes
span.set_attribute("order_id", order.id)
span.set_attribute("customer_tier", customer.tier)
span.set_attribute("item_count", len(order.items))
span.set_attribute("total_value", order.total)
# Or batch set via metadata
span.update(metadata={
"payment_method": order.payment_method,
"shipping_speed": order.shipping_speed,
"has_discount": order.discount is not None
})
result = process_order(order)
span.update(output=result)await client.startActiveSpan('process_order', async (span) => {
span.setAttribute('orderId', order.id);
span.setAttribute('customerTier', customer.tier);
span.setAttribute('itemCount', order.items.length);
span.setAttribute('totalValue', order.total);
span.setAttribute('paymentMethod', order.paymentMethod);
span.setAttribute('shippingSpeed', order.shippingSpeed);
span.setAttribute('hasDiscount', order.discount !== null);
const result = await processOrder(order);
client.updateCurrentSpan({ output: result });
return result;
});Nested Spans
Create hierarchical traces with parent-child relationships:
with client.start_as_current_span(name="checkout_flow") as parent:
parent.set_attribute("cart_id", cart.id)
# Child span 1: Validate cart
with client.start_as_current_span(name="validate_cart") as validate_span:
validation_result = validate_cart(cart)
validate_span.update(output=validation_result)
if not validation_result.valid:
validate_span.update(error="Cart validation failed")
return
# Child span 2: Process payment
with client.start_as_current_span(name="process_payment") as payment_span:
payment_span.set_attribute("payment_method", cart.payment_method)
payment_result = charge_customer(cart.total)
payment_span.update(output={"transaction_id": payment_result.id})
# Child span 3: Create order
with client.start_as_current_span(name="create_order") as order_span:
order = create_order(cart, payment_result)
order_span.update(output={"order_id": order.id})
parent.update(output={"order_id": order.id, "status": "completed"})await client.startActiveSpan('checkout_flow', async (parentSpan) => {
parentSpan.setAttribute('cartId', cart.id);
// Child span 1: Validate cart
const validationResult = await client.startActiveSpan('validate_cart', async (span) => {
return await validateCart(cart);
});
if (!validationResult.valid) {
client.updateCurrentSpan({ output: { error: 'Cart validation failed' } });
return;
}
// Child span 2: Process payment
const paymentResult = await client.startActiveSpan('process_payment', async (span) => {
span.setAttribute('paymentMethod', cart.paymentMethod);
return await chargeCustomer(cart.total);
});
// Child span 3: Create order
const order = await client.startActiveSpan('create_order', async (span) => {
return await createOrder(cart, paymentResult);
});
client.updateCurrentSpan({ output: { orderId: order.id, status: 'completed' } });
});Resulting Trace Structure
Trace: checkout_flow (1,850ms)
├── cart_id: cart_123
├── output: {order_id: "ord_456", status: "completed"}
│
├── validate_cart (45ms)
│ └── output: {valid: true, items: 3}
│
├── process_payment (1,200ms)
│ ├── payment_method: credit_card
│ └── output: {transaction_id: "txn_789"}
│
└── create_order (605ms)
└── output: {order_id: "ord_456"}Generation Spans
For LLM calls without using wrappers:
with client.start_as_current_generation(
name="summarize_document",
model="gpt-4",
input={"text": document_text}
) as gen:
response = openai.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "Summarize the following document."},
{"role": "user", "content": document_text}
]
)
gen.update(
output=response.choices[0].message.content,
usage={
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens
}
)await client.startActiveGeneration('summarize_document', 'gpt-4', 'openai', async (span) => {
const response = await openai.chat.completions.create({
model: 'gpt-4',
messages: [
{ role: 'system', content: 'Summarize the following document.' },
{ role: 'user', content: documentText }
]
});
client.updateCurrentSpan({
output: response.choices[0].message.content,
usage: {
inputTokens: response.usage.prompt_tokens,
outputTokens: response.usage.completion_tokens
}
});
return response;
});Retrieval Spans
For vector search and document retrieval:
with client.start_as_current_span(
name="semantic_search",
as_type="retrieval"
) as span:
span.set_attribute("index", "product_catalog")
span.set_attribute("top_k", 10)
span.update(input={"query": user_query})
# Perform search
results = vector_db.search(
query=user_query,
top_k=10
)
span.update(
output={
"count": len(results),
"scores": [r.score for r in results],
"document_ids": [r.id for r in results]
}
)Tool Spans
For function/tool execution:
with client.start_as_current_span(
name="calculator",
as_type="tool"
) as span:
span.set_attribute("tool_name", "calculator")
span.update(input={"expression": expression})
result = evaluate_expression(expression)
span.update(output={"result": result})Error Handling
with client.start_as_current_span(name="api_call") as span:
try:
result = external_api.call(params)
span.update(output=result)
except TimeoutError as e:
span.update(
error=f"Timeout after {timeout}s",
metadata={"error_type": "timeout", "retryable": True}
)
raise
except AuthenticationError as e:
span.update(
error="Authentication failed",
metadata={"error_type": "auth", "retryable": False}
)
raise
except Exception as e:
span.update(error=str(e))
raiseawait client.startActiveSpan('api_call', async (span) => {
// Errors are automatically captured by the SDK
const result = await externalApi.call(params);
client.updateCurrentSpan({ output: result });
return result;
});
// If externalApi.call() throws, the SDK automatically records
// the exception and sets ERROR status on the spanAsync Operations
Python
from brokle import AsyncBrokle
import asyncio
client = AsyncBrokle(api_key="bk_...")
async def process_items(items):
async with client.start_as_current_span(name="batch_process") as span:
span.set_attribute("batch_size", len(items))
# Process in parallel
tasks = [process_single(item) for item in items]
results = await asyncio.gather(*tasks, return_exceptions=True)
successes = [r for r in results if not isinstance(r, Exception)]
failures = [r for r in results if isinstance(r, Exception)]
span.update(output={
"processed": len(successes),
"failed": len(failures)
})
return successes
asyncio.run(process_items(items))JavaScript
async function processItems(items) {
return client.startActiveSpan('batch_process', async (span) => {
span.setAttribute('batchSize', items.length);
const results = await Promise.allSettled(
items.map(item => processSingle(item))
);
const successes = results.filter(r => r.status === 'fulfilled');
const failures = results.filter(r => r.status === 'rejected');
client.updateCurrentSpan({
output: {
processed: successes.length,
failed: failures.length
}
});
return successes.map(r => r.value);
});
}Combining Manual and Automatic Tracing
Use manual spans to add context around automatic LLM traces:
from brokle import Brokle, wrap_openai
import openai
client = Brokle(api_key="bk_...")
openai_client = wrap_openai(openai.OpenAI())
with client.start_as_current_span(name="rag_pipeline") as parent:
parent.update_trace(user_id="user_123", session_id="session_456")
# Manual: Retrieve documents
with client.start_as_current_span(name="retrieve_context") as retrieve:
docs = search_documents(query)
retrieve.update(output={"doc_count": len(docs)})
# Automatic: LLM call traced automatically
response = openai_client.chat.completions.create(
model="gpt-4",
messages=build_prompt(query, docs)
)
# Manual: Post-process
with client.start_as_current_span(name="format_response") as format_span:
formatted = format_response(response.choices[0].message.content)
format_span.update(output=formatted)
parent.update(output=formatted)Best Practices
1. Use Descriptive Names
# Good
with client.start_as_current_span(name="validate_payment_card"):
...
# Bad
with client.start_as_current_span(name="step2"):
...2. Add Relevant Context
with client.start_as_current_span(name="process_request") as span:
span.set_attribute("request_id", request.id)
span.set_attribute("user_type", user.type)
span.set_attribute("endpoint", request.path)3. Keep Spans Focused
# Good: Separate concerns
with client.start_as_current_span(name="order_processing"):
with client.start_as_current_span(name="validate"):
validate()
with client.start_as_current_span(name="charge"):
charge()
with client.start_as_current_span(name="fulfill"):
fulfill()
# Bad: Everything in one span
with client.start_as_current_span(name="do_everything"):
validate()
charge()
fulfill()4. Always Handle Errors
with client.start_as_current_span(name="operation") as span:
try:
result = do_work()
span.update(output=result)
except Exception as e:
span.update(error=str(e))
raise # Re-raise after recordingNext Steps
- Automatic Tracing - Zero-code LLM tracing
- Trace Metadata - Add user and session context
- Working with Spans - Advanced span patterns
- Python SDK - Full API reference