Netra Best Practices

Netra Best Practices

Use this skill as the default end-to-end guide for integrating, operating, and improving AI systems with Netra. Also you can create custom metrics using this skill.

Installation

Detect the project's package manager before installing netra-sdk. Check the project root in priority order:

Do NOT run multiple install commands or install globally.

Use-Case specific references

• Instrumenting an LLM application: references/instrumentation.md
• custom metrics for the agent (counters, histograms, gauges): references/custom-metrics.md
• Single-turn evaluations (datasets, test suites, custom evaluators): references/single-turn-evaluations.md

Feedback

If the user is unhappy with the results, ask them to open an issue at https://github.com/KeyValueSoftwareSystems/netra-skills/issues/new.

Netra Custom Metrics

Emit and export application-specific metrics using the OpenTelemetry-backed Meter API exposed by the Netra SDK.

Workflow

• Ensure Netra.init() is called with enable_metrics=True before creating any instruments or calling Netra.get_meter().
• Identify which instrument type best fits each signal (counter, histogram, up-down counter, or observable gauge).
• Define instruments once at module/service level; record measurements inside request or event handlers.
• Always call Netra.shutdown() at the end of the application lifecycle to ensure metrics are flushed to the backend.
• Verify exports are reaching the backend using the Netra dashboard or OTLP endpoint logs.

Initialization

Enable the metrics pipeline in Netra.init and obtain a Meter scoped to your service or module.

NETRA_API_KEY=
NETRA_OTLP_ENDPOINT=

import os
from netra import Netra

Netra.init(
    app_name="my-ai-app",
    environment="production",
    headers=f"x-api-key={os.getenv('NETRA_API_KEY')}",
    enable_metrics=True,
    metrics_export_interval_ms=10000,  # export every 10 s
)

# Recommended: use your service or module name
meter = Netra.get_meter("my_service")

[!IMPORTANT]

Initialization Order Matters: You must call Netra.init() before calling Netra.get_meter(). If you define instruments at the module level (standard practice), ensure the module is imported after Netra.init() has been executed in your application's entry point (e.g., main.py).

get_meter accepts a name (instrumentation scope, defaults to "netra") and an optional version string. If enable_metrics is False or no OTLP endpoint is configured, a no-op MeterProvider is installed — your code can still call get_meter and record metrics safely, they will simply be discarded.

Counter

Monotonically increasing value — use for request counts, completed jobs, or any value that only goes up.

request_counter = meter.create_counter(
    name="http.requests",
    description="Number of HTTP requests processed",
    unit="1",
)

request_counter.add(1, attributes={"route": "/api/health", "status": "ok"})

UpDownCounter

Value that can increase or decrease — use for active connections, queue depth, or in-flight requests.

active_connections = meter.create_up_down_counter(
    name="connections.active",
    description="Number of active client connections",
    unit="1",
)

active_connections.add(1,  attributes={"region": "us-east-1"})  # connection opened
active_connections.add(-1, attributes={"region": "us-east-1"})  # connection closed

Histogram

Distribution of measurements — use for latency, payload sizes, token counts, or any value where the distribution matters.

latency = meter.create_histogram(
    name="db.query.latency_ms",
    description="Database query latency",
    unit="ms",
)

latency.record(15.3, attributes={"operation": "read",  "table": "users"})
latency.record(30.7, attributes={"operation": "write", "table": "orders"})

Observable Instruments

Use observable (pull-based) instruments for system-level or slowly-changing values. Netra invokes registered callbacks periodically on each export cycle.

import psutil
from opentelemetry.metrics import Observation

def _cpu_callback(options):
    yield Observation(psutil.cpu_percent(), attributes={"resource": "cpu"})

def _memory_callback(options):
    mem = psutil.virtual_memory()
    yield Observation(mem.used, attributes={"unit": "bytes"})

meter.create_observable_gauge(
    name="system.cpu.utilization",
    description="CPU utilization percentage",
    callbacks=[_cpu_callback],
)

meter.create_observable_gauge(
    name="system.memory.used_bytes",
    description="Used memory in bytes",
    callbacks=[_memory_callback],
)

NOTE: Callbacks must be fast and non-blocking. Heavy I/O inside a callback will stall the export cycle.

[!WARNING]

Flush Your Metrics: Always call Netra.shutdown() on graceful termination. Metrics are buffered and exported in batches (defined by metrics_export_interval_ms). If the application exits without a shutdown call, the last batch of metrics will likely be lost.

Recommended instrument strategy

1. Use Counter for events that only accumulate (requests, errors, retries). 2. Use Histogram for any measurement whose distribution matters (latency, token count, payload size). 3. Use UpDownCounter for values that rise and fall (connections, queue depth). 4. Use Observable Gauge for asynchronous system-level signals (CPU, memory) polled by the SDK.

Validation checklist

1. enable_metrics=True is passed to Netra.init(). 2. Netra.get_meter() and instrument creation happen strictly after Netra.init(). 3. Instruments are defined once at module/service level, not inside hot loops. 4. If instruments are in a separate module, that module is imported after Netra.init() in the entry point. 5. attributes dicts use only low-cardinality keys — avoid user IDs or request IDs as metric labels. 6. Netra.shutdown() is called on graceful termination (e.g., in a finally block or FastAPI lifespan) to flush the last export batch. 7. Metrics appear in the Netra dashboard within one export interval.

References

• https://docs.getnetra.ai/sdk-reference/custom-metric
• https://docs.getnetra.ai/Observability/Traces/configuration/initialization
• https://opentelemetry.io/docs/specs/otel/metrics/api/

Netra Observability

Instrument LLM applications/AI agents with Netra, following best practices and tailored to your use case.

Workflow

• Assess the current environment - Ensure netra-sdk is installed and is the latest version, and figure out which libraries are installed for the LLM application.
• Determine the kind of instrumentation necessary for the application.
• Instrument the application and instruct the user on steps they need to take to set up their environment.

Auto Instrumentation

Auto-instrumentation is the fastest way to start. Netra patches supported libraries and automatically captures spans for LLM calls, frameworks, vector DBs, HTTP, and more.

Initialize Netra once at application startup, before using libraries you want instrumented.

NETRA_API_KEY=
NETRA_OTLP_ENDPOINT=

The user may create their api key by visiting the dashboard and going to Settings -> Project -> API Keys.

import os
from netra import Netra
from netra.instrumentation.instruments import InstrumentSet

# Initialize before importing providers/frameworks for best coverage
Netra.init(
    app_name="my-ai-app",
    environment="production",
    headers=f"x-api-key={os.getenv('NETRA_API_KEY')}",
    trace_content=True,
    instruments={InstrumentSet.OPENAI, InstrumentSet.LANGCHAIN},
)

# Optional context for trace grouping
Netra.set_user_id("user-123")
Netra.set_session_id("session-abc")

import { Netra, NetraInstruments } from "netra-sdk";

// Always await init in TypeScript so instrumentations are ready
await Netra.init({
	appName: "my-ai-app",
	environment: "production",
	headers: `x-api-key=${process.env.NETRA_API_KEY}`,
	traceContent: true,
	instruments: new Set([NetraInstruments.OPENAI, NetraInstruments.LANGCHAIN]),
});

// Optional context for trace grouping
Netra.setUserId("user-123");
Netra.setSessionId("session-abc");

Instrumentation through decorators

Use decorators when you want semantic application spans with very little code overhead.

• @workflow: top-level business process.
• @agent: agent/orchestrator behavior.
• @task: discrete unit of work/tool call.
• @span: generic/custom span type.

NOTE: Always ensure netra decorators are placed immediately above the function definition.

Python decorators

from netra import Netra, SpanType
from netra.decorators import workflow, agent, task, span

@workflow(name="order-fulfillment")
def fulfill_order(order: dict):
    current = Netra.get_current_span()
    if current:
        current.set_attribute("order.id", order.get("id"))
        current.add_event("order.received", {"item_count": len(order.get("items", []))})

    result = OrderAgent().orchestrate(order)

    if current:
        current.add_event("order.completed")
    return result

@agent
class OrderAgent:
    @task(name="validate-order")
    def validate(self, order: dict):
        if not order.get("items"):
            raise ValueError("Order must contain at least one item")
        return True

    @span(name="shipping-quote", as_type=SpanType.TOOL)
    def dispatch(self, order: dict):
        return {"status": "queued", "order_id": order.get("id")}

    def orchestrate(self, order: dict):
        self.validate(order)
        return self.dispatch(order)

@tool #Langchain tools if the app uses langchain
@task(name="Tool Call")
def tool_call():
	return "Tool output"

TypeScript decorators

TypeScript decorators require "experimentalDecorators": true in tsconfig.json.

import { SpanType } from "netra-sdk";
import { workflow, agent, task, span } from "netra-sdk/decorators";

@workflow({ name: "order-fulfillment" })
async function fulfillOrder(order: { id: string; items: unknown[] }) {
	const result = await new OrderAgent().orchestrate(order);
	return result;
}

@agent({ name: "order-agent" })
class OrderAgent {
	@task({ name: "validate-order" })
	async validate(order: { items: unknown[] }) {
		if (!order.items?.length) {
			throw new Error("Order must contain at least one item");
		}
	}

	@span({ name: "shipping-quote", asType: SpanType.TOOL })
	async dispatch(order: { id: string }) {
		return { status: "queued", orderId: order.id };
	}

	async orchestrate(order: { id: string; items: unknown[] }) {
		await this.validate(order);
		return this.dispatch(order);
	}
}

@task({ name: "Tool Call" })
async function toolCall() {
	return "Tool output";
}

Decorators automatically capture parameters and exceptions. Keep instrumentation focused on high-value workflow boundaries to avoid noisy traces.

Manual Instrumentation

Use manual tracing when you need full lifecycle and metadata control, advanced nesting, or custom span boundaries.

Python manual tracing

from netra import Netra, SpanType, UsageModel

def chat_with_ai(user_message: str) -> str:
    with Netra.start_span(
        "chat-completion",
        as_type=SpanType.GENERATION,
        attributes={"entrypoint": "chat_with_ai"},
        module_name="chat",
    ) as span:
        span.set_prompt(user_message)
        span.set_model("gpt-4")
        span.set_llm_system("openai")
        span.add_event("generation.started")

        try:
            response_text = "Hello from model"

            span.set_usage([
                UsageModel(
                    model="gpt-4",
                    cost_in_usd=0.001,
                    usage_type="chat",
                    units_used=1,
                )
            ])
            span.set_success()
            return response_text
        except Exception as exc:
            span.set_error(str(exc))
            raise

TypeScript manual tracing

In TypeScript, always call end() (preferably in finally).

import { Netra, SpanType } from "netra-sdk";

async function chatWithAI(userMessage: string): Promise<string> {
	const span = Netra.startSpan(
		"chat-completion",
		{
			asType: SpanType.GENERATION,
			moduleName: "chat",
			attributes: { entrypoint: "chatWithAI" },
		}
	);

	span.setPrompt(userMessage);
	span.setModel("gpt-4");
	span.setLlmSystem("openai");
	span.addEvent("generation.started");

	try {
		const responseText = "Hello from model";

		span.setUsage([
			{
				model: "gpt-4",
				costInUsd: 0.001,
				usageType: "chat",
				unitsUsed: 1,
			},
		]);
		span.setSuccess();
		return responseText;
	} catch (error: any) {
		span.setError(error?.message || "unknown error");
		throw error;
	} finally {
		span.end();
	}
}

Recommended rollout strategy

1. Start with auto-instrumentation for broad, immediate coverage. 2. Add decorators for business semantics (workflow -> agent -> task). 3. Use manual spans only for operations requiring precise boundaries or custom metadata.

Validation checklist

1. Netra.init() / await Netra.init() is called once at startup. 2. Initialization happens before instrumented library usage. 3. High-level operations appear as workflow spans. 4. TypeScript manual spans always call span.end(). 5. shutdown() is called on graceful app termination.

References

• https://docs.getnetra.ai/Observability/Traces/auto-instrumentation
• https://docs.getnetra.ai/Observability/Traces/decorators
• https://docs.getnetra.ai/Observability/Traces/manual-tracing
• https://docs.getnetra.ai/Observability/Traces/configuration/initialization
• https://docs.getnetra.ai/sdk-reference/sdk/python
• https://docs.getnetra.ai/sdk-reference/sdk/typescript

Netra Single-Turn Evaluations

Evaluate LLM or agent outputs on a per-input basis using datasets, automated test suites, and pluggable evaluators through the Netra SDK.

Workflow

• Ensure Netra.init() is called before using any evaluation APIs.
• Prepare a dataset — either inline in code or managed via the Netra API.
• Define a task function that takes an input and returns an output (the code under test).
• Optionally implement custom local evaluators to score each output.
• Run Netra.evaluation.run_test_suite() to execute the task against every dataset item, run evaluators, and report results.
• Review results on the Netra dashboard or inspect the returned summary dict.

Initialization

Enable evaluation by initializing Netra at application startup. The evaluation client (Netra.evaluation) is created automatically.

NETRA_API_KEY=
NETRA_OTLP_ENDPOINT=

import os
from netra import Netra

Netra.init(
    app_name="my-ai-app",
    environment="production",
    headers=f"x-api-key={os.getenv('NETRA_API_KEY')}",
)

[!IMPORTANT]

Netra.evaluation is None if Netra.init() has not been called or if the OTLP endpoint is missing. Always init first.

Dataset Preparation

A dataset is a list of items, each with an input and optional expected_output, metadata, and tags.

Inline dataset (no API call)

Use Dataset and DatasetItem to define items directly in code. Best for quick experiments or CI pipelines.

from netra.evaluation import DatasetItem, Dataset

dataset = Dataset(items=[
    DatasetItem(
        input="What is the capital of France?",
        expected_output="Paris",
    ),
    DatasetItem(
        input="Summarize this article in one sentence.",
        expected_output="The article discusses recent advances in renewable energy.",
        metadata={"category": "summarization"},
    ),
])

API-managed dataset

Create a persistent dataset on the Netra platform, add items via API, then fetch them for test runs. Useful when datasets are shared across runs or managed from the dashboard.

response = Netra.evaluation.create_dataset(
    name="QA Golden Set",
    tags=["qa", "v1"],
)
dataset_id = response.id

Netra.evaluation.add_dataset_item(
    dataset_id=dataset_id,
    item=DatasetItem(
        input="What is the capital of France?",
        expected_output="Paris",
        metadata={"difficulty": "easy"},
        tags=["geography"],
    ),
)

fetched = Netra.evaluation.get_dataset(dataset_id)
dataset = Dataset(items=fetched.items)

Task Function

The task is a callable that receives a single dataset item's input and returns the output to be evaluated. It can be sync or async.

from openai import OpenAI

client = OpenAI()

def my_task(input):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": input}],
    )
    return response.choices[0].message.content

async def my_async_task(input):
    response = await async_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": input}],
    )
    return response.choices[0].message.content

Each task invocation is automatically wrapped in a TestRun.{name} span, so the call is traced end-to-end without extra instrumentation.

Running a Test Suite

run_test_suite is the main entry point. It creates a test run, executes the task for every item, runs local evaluators, submits results, and marks the run as completed.

result = Netra.evaluation.run_test_suite(
    name="QA Agent v2",
    data=dataset,
    task=my_task,
    evaluators=[my_evaluator],   # optional, see Custom Evaluators below
    max_concurrency=50,          # default 50
)

Parameters:

Return value on success:

{
    "runId": "uuid-of-the-run",
    "items": [
        {
            "index": 0,
            "status": "completed",
            "traceId": "abc123...",
            "spanId": "def456...",
            "testRunItemId": "ghi789...",
        },
        # ...
    ],
}

Returns None if validation fails or the run could not be created.

Custom Evaluators

Write local evaluators that run client-side before results are sent to the platform. Subclass BaseEvaluator and implement evaluate().

from netra.evaluation import BaseEvaluator, EvaluatorConfig, EvaluatorContext, EvaluatorOutput, ScoreType

class ExactMatchEvaluator(BaseEvaluator):
    def evaluate(self, context: EvaluatorContext) -> EvaluatorOutput:
        is_match = str(context.task_output).strip().lower() == str(context.expected_output).strip().lower()
        return EvaluatorOutput(
            evaluator_name=self.config.name,
            result=is_match,
            is_passed=is_match,
            reason="Exact match" if is_match else "Output did not match expected",
        )

exact_match = ExactMatchEvaluator(
    EvaluatorConfig(
        name="exact_match",
        label="Exact Match",
        score_type=ScoreType.BOOLEAN,
    )
)

`EvaluatorConfig` fields:

`EvaluatorContext` fields (passed to `evaluate`):

`EvaluatorOutput` fields (returned from `evaluate`):

evaluate() can be sync or async — the framework awaits coroutines automatically.

Numerical evaluator example

class RelevanceScoreEvaluator(BaseEvaluator):
    def evaluate(self, context: EvaluatorContext) -> EvaluatorOutput:
        score = compute_relevance(context.input, context.task_output)  # 0.0–1.0
        return EvaluatorOutput(
            evaluator_name=self.config.name,
            result=score,
            is_passed=score >= 0.7,
            reason=f"Relevance score: {score:.2f}",
        )

relevance = RelevanceScoreEvaluator(
    EvaluatorConfig(
        name="relevance",
        label="Relevance Score",
        score_type=ScoreType.NUMERICAL,
    )
)

Platform Evaluators

In addition to local evaluators, Netra provides built-in platform evaluators that run server-side after trace ingestion. These are configured via the Netra dashboard and attached to datasets. Available types:

Platform evaluators with evalType: TURN run automatically for each single-turn test run item after the trace is ingested. No client-side code is needed — attach them to your dataset from the dashboard.

Full Example

import os
from netra import Netra
from netra.evaluation import (
    BaseEvaluator,
    DatasetItem,
    Dataset,
    EvaluatorConfig,
    EvaluatorContext,
    EvaluatorOutput,
    ScoreType,
)
from openai import OpenAI

Netra.init(
    app_name="my-ai-app",
    environment="staging",
    headers=f"x-api-key={os.getenv('NETRA_API_KEY')}",
)

client = OpenAI()

def qa_task(input):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": input}],
    )
    return response.choices[0].message.content

class ContainsExpectedEvaluator(BaseEvaluator):
    def evaluate(self, context: EvaluatorContext) -> EvaluatorOutput:
        output = str(context.task_output).lower()
        expected = str(context.expected_output).lower()
        contains = expected in output
        return EvaluatorOutput(
            evaluator_name=self.config.name,
            result=contains,
            is_passed=contains,
            reason="Output contains expected answer" if contains else "Expected answer not found in output",
        )

dataset = Dataset(items=[
    DatasetItem(input="What is the capital of France?", expected_output="Paris"),
    DatasetItem(input="What is 2 + 2?", expected_output="4"),
    DatasetItem(input="Who wrote Hamlet?", expected_output="Shakespeare"),
])

result = Netra.evaluation.run_test_suite(
    name="QA Bot Regression v1",
    data=dataset,
    task=qa_task,
    evaluators=[
        ContainsExpectedEvaluator(
            EvaluatorConfig(
                name="contains_expected",
                label="Contains Expected Answer",
                score_type=ScoreType.BOOLEAN,
            )
        )
    ],
)

Netra.shutdown()

Validation checklist

1. Netra.init() is called before accessing Netra.evaluation. 2. NETRA_API_KEY and NETRA_OTLP_ENDPOINT environment variables are set. 3. Every DatasetItem has a non-empty input. 4. The task function accepts a single argument (the item input) and returns the output. 5. Custom evaluator evaluate() returns an EvaluatorOutput with evaluator_name matching config.name. 6. ScoreType matches the type of result in EvaluatorOutput (bool for BOOLEAN, number for NUMERICAL, string for CATEGORICAL). 7. Netra.shutdown() is called on graceful termination to flush pending traces and evaluation data. 8. Test run results appear on the Netra dashboard after the run completes.

References

• https://docs.getnetra.ai/Evaluations/overview
• https://docs.getnetra.ai/sdk-reference/evaluation/python
• https://docs.getnetra.ai/Observability/Traces/configuration/initialization
• https://docs.getnetra.ai/sdk-reference/sdk/python