Then instrument your agent code. startActiveObservation opens a trace-level span; updateActiveTrace and updateActiveObservation attach input, output, and metadata.
The SDK is built on OpenTelemetry, so trace context propagates automatically through any code running inside an active observation. Nested observations are linked to their parent without manual ID plumbing:
typescript
await startActiveObservation('span', { name: 'main-request' }, async () => {
// Any observation started here is automatically a child of 'main-request'.const span = startObservation('span', { name: 'downstream-operation' })
// ... do work ...
span.end()
})
For propagation across separate services or processes, use the standard W3C
traceparent header from @opentelemetry/api so the receiving service continues
the same OpenTelemetry trace.
Correlation
Group related traces under a single session by setting a session ID on each trace. Filter and replay full sessions from the dashboard at https://app.agenticants.ai:
Filtering and querying traces by tag is a dashboard feature. Browse and filter
traces by tag at https://app.agenticants.ai.
Sampling Strategies
Head-Based Sampling
Because the SDK is built on OpenTelemetry, head-based sampling is configured with a standard OTel sampler on the NodeTracerProvider. Use TraceIdRatioBasedSampler to keep a fixed fraction of traces:
typescript
const provider = new NodeTracerProvider({
sampler: new TraceIdRatioBasedSampler(0.1), // Sample 10% of traces
spanProcessors: [
new AntsPlatformSpanProcessor({
publicKey: process.env.ANTS_PLATFORM_PUBLIC_KEY,
secretKey: process.env.ANTS_PLATFORM_SECRET_KEY,
baseUrl: 'https://api.agenticants.ai',
}),
],
})
provider.register()
setAntsPlatformTracerProvider(provider)
For custom per-request logic, gate whether you open an observation at all:
Tail-based sampling decides whether to keep a trace after it completes, based on
its outcome (errors, latency, customer tier). In the Python SDK you implement this
by deciding whether to flush a finished trace. A common pattern is to mark traces
you want to keep with a tag, then apply your own keep/drop logic before
client.flush():
// Automatic timing: duration is the wall-clock time between start and endconst span = startObservation('generation', { name: 'llm-call' })
const result = await llm.generate(prompt)
span.end() // Duration calculated automatically// You can also record your own measurement as metadataconst start = Date.now()
const opResult = await operation()
const opSpan = startObservation('span', { name: 'operation' })
opSpan.update({ metadata: { durationMs: Date.now() - start } }).end()
Token Tracking
The simplest way to capture tokens and cost is the OpenAI drop-in, which records
usage automatically. Just swap the import:
python
# Replace `import openai` with the instrumented drop-infrom ants_platform.openai import openai
response = openai.chat.completions.create(
model='gpt-4',
messages=messages,
)
# Prompt/completion tokens and cost are captured automatically.
If you manage the generation span yourself, pass usage_details:
python
from ants_platform import get_client
client = get_client()
with client.start_generation(name='llm-inference', model='gpt-4') as span:
response = openai.chat.completions.create(model='gpt-4', messages=messages)
span.update(
output=response.choices[0].message.content,
usage_details={
'input': response.usage.prompt_tokens,
'output': response.usage.completion_tokens,
'total': response.usage.total_tokens,
},
)
# Classify errors and record them on the current span.# The SDK level is one of DEBUG / DEFAULT / WARNING / ERROR.if isinstance(error, (ValidationError, RateLimitError)):
level = 'WARNING'else:
level = 'ERROR'
client = get_client()
client.update_current_span(
level=level,
status_message=str(error),
metadata={'recoverable': isinstance(error, RetryableError)},
)