Back to Details

netra-best-practices

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Netra Best Practices

Netra 最佳实践

Use this skill as the default end-to-end guide for integrating, operating, and improving AI systems with Netra.
将本技能作为使用Netra集成、运维和优化AI系统的默认端到端指南。

Scope

适用范围

  • Code-focused integration and operations guidance only.
  • Includes runnable SDK patterns for setup, tracing, metrics, evaluations, and simulations.
  • 仅专注于代码层面的集成与运维指导。
  • 包含可运行的SDK模式，涵盖设置、追踪、指标、评估和模拟。

When To Use

使用场景

  • You want one workflow that replaces separate setup and instrumentation skills.
  • You need traceability from request context to span-level debugging.
  • You need repeatable quality validation with evaluations and simulations.
  • You need production-safe troubleshooting guidance.
  • 你需要一个能替代单独设置和插桩技能的统一工作流。
  • 你需要从请求上下文到跨度级别的可调试追踪能力。
  • 你需要通过评估和模拟实现可重复的质量验证。
  • 你需要生产环境安全的故障排查指导。

Coverage

覆盖内容

This unified skill combines guidance for:
  • SDK setup and initialization.
  • Context tracking (user/session/tenant/custom attributes/events/conversation).
  • Tracing method selection (auto-instrumentation, decorators, manual spans).
  • Custom spans, usage/cost, and action records.
  • OpenTelemetry custom metrics.
  • Integration patterns (FastAPI, Express, multi-agent, batch, streaming).
  • Evaluation setup and regression tracking.
  • Multi-turn simulation setup and analysis.
  • Troubleshooting and production triage.
本统一技能整合了以下方面的指导：
  • SDK设置与初始化。
  • 上下文追踪（用户/会话/租户/自定义属性/事件/对话）。
  • 追踪方式选择（自动插桩、装饰器、手动跨度）。
  • 自定义跨度、使用量/成本和操作记录。
  • OpenTelemetry自定义指标。
  • 集成模式（FastAPI、Express、多Agent、批处理、流式处理）。
  • 评估设置与回归追踪。
  • 多轮对话模拟设置与分析。
  • 故障排查与生产环境问题分流。

End-To-End Workflow

端到端工作流

  1. Install and initialize Netra once at startup.
  2. Enable only required instruments for your framework/providers.
  3. Set request context (
    user_id
    , 
    session_id
    , 
    tenant_id
    ) at request entry.
  4. Choose tracing strategy:
    • auto-instrumentation for fast coverage,
    • decorators for semantic boundaries,
    • manual spans for precise lifecycle control.
  5. Add custom attributes, events, usage, and actions on key operations.
  6. Add business metrics with bounded-cardinality attributes.
  7. Validate trace hierarchy and metadata in Observability.
  8. Build eval datasets and run test suites before releases.
  9. Build multi-turn simulations for critical scenarios/personas.
  10. Apply troubleshooting triage for missing telemetry or SDK errors.
  1. 在启动时一次性安装并初始化Netra。
  2. 仅为你的框架/提供商启用所需的插桩工具。
  3. 在请求入口处设置请求上下文（
    user_id
    、
    session_id
    、
    tenant_id
    ）。
  4. 选择追踪策略：
    • 自动插桩：快速覆盖提供商/框架。
    • 装饰器：明确业务语义边界。
    • 手动跨度：精确控制生命周期。
  5. 在关键操作上添加自定义属性、事件、使用量和操作记录。
  6. 添加带有有限基数属性的业务指标。
  7. 在可观测性平台中验证追踪层级和元数据。
  8. 在发布前构建评估数据集并运行测试套件。
  9. 为关键场景/角色构建多轮对话模拟。
  10. 针对缺失遥测数据或SDK错误应用故障排查分流方案。

Tracing Method Decision Guide

追踪方式决策指南

  • Auto-instrumentation: choose when you need quickest provider/framework visibility.
  • Decorators (
    @workflow
    , 
    @agent
    , 
    @task
    , 
    @span
    ): choose for clear business semantics.
  • Manual spans (
    Netra.start_span
    / 
    Netra.startSpan
    ): choose for custom operation boundaries, retries, side effects, and explicit async lifecycle control.
  • 自动插桩：当你需要最快获取提供商/框架可见性时选择。
  • 装饰器（
    @workflow
    、
    @agent
    、
    @task
    、
    @span
    ）：为清晰的业务语义选择。
  • 手动跨度（
    Netra.start_span
    / 
    Netra.startSpan
    ）：为自定义操作边界、重试、副作用和显式异步生命周期控制选择。

Setup Baseline

设置基准

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

Netra.init(
    app_name="my-ai-app",
    instruments={InstrumentSet.OPENAI, InstrumentSet.FASTAPI},
)
typescript
import { Netra, NetraInstruments } from "netra-sdk";

await Netra.init({
    appName: "my-ai-app",
    headers: `x-api-key=${process.env.NETRA_API_KEY}`,
    instruments: new Set([NetraInstruments.OPENAI, NetraInstruments.EXPRESS]),
});
python
from netra import Netra
from netra.instrumentation.instruments import InstrumentSet

Netra.init(
    app_name="my-ai-app",
    instruments={InstrumentSet.OPENAI, InstrumentSet.FASTAPI},
)
typescript
import { Netra, NetraInstruments } from "netra-sdk";

await Netra.init({
    appName: "my-ai-app",
    headers: `x-api-key=${process.env.NETRA_API_KEY}`,
    instruments: new Set([NetraInstruments.OPENAI, NetraInstruments.EXPRESS]),
});

Context Tracking Pattern

上下文追踪模式

python
from netra import Netra
from netra.session_manager import ConversationType


def handle_request(request):
    Netra.set_user_id(request.user_id)
    Netra.set_session_id(request.session_id)
    Netra.set_tenant_id(request.tenant_id)

    Netra.set_custom_attributes("customer_tier", request.customer_tier)
    Netra.set_custom_attributes("region", request.region)

    Netra.add_conversation(
        conversation_type=ConversationType.INPUT,
        role="user",
        content=request.message,
    )

    response = run_agent(request.message)

    Netra.add_conversation(
        conversation_type=ConversationType.OUTPUT,
        role="assistant",
        content=response,
    )

    Netra.set_custom_event(
        event_name="request_completed",
        attributes={"status": "success"},
    )

    return response
python
from netra import Netra
from netra.session_manager import ConversationType


def handle_request(request):
    Netra.set_user_id(request.user_id)
    Netra.set_session_id(request.session_id)
    Netra.set_tenant_id(request.tenant_id)

    Netra.set_custom_attributes("customer_tier", request.customer_tier)
    Netra.set_custom_attributes("region", request.region)

    Netra.add_conversation(
        conversation_type=ConversationType.INPUT,
        role="user",
        content=request.message,
    )

    response = run_agent(request.message)

    Netra.add_conversation(
        conversation_type=ConversationType.OUTPUT,
        role="assistant",
        content=response,
    )

    Netra.set_custom_event(
        event_name="request_completed",
        attributes={"status": "success"},
    )

    return response

Decorator + Manual Span Pattern

装饰器 + 手动跨度模式

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


@workflow(name="order-fulfillment")
def fulfill_order(order: dict):
    return OrderAgent().orchestrate(order)


@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")

    @span(name="shipping-quote", as_type=SpanType.TOOL)
    def dispatch(self, order: dict):
        with Netra.start_span("persist-order") as custom:
            custom.set_usage([
                UsageModel(
                    model="gpt-4o",
                    usage_type="generation",
                    units_used=1,
                    cost_in_usd=0.01,
                )
            ])
            custom.set_action([
                ActionModel(
                    action="DB",
                    action_type="INSERT",
                    metadata={"table": "orders"},
                    success=True,
                )
            ])
            custom.add_event("order.persisted", {"order_id": order.get("id")})
        return {"status": "queued", "order_id": order.get("id")}

    def orchestrate(self, order: dict):
        self.validate(order)
        return self.dispatch(order)
python
from netra import Netra, SpanType, UsageModel, ActionModel
from netra.decorators import workflow, agent, task, span


@workflow(name="order-fulfillment")
def fulfill_order(order: dict):
    return OrderAgent().orchestrate(order)


@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")

    @span(name="shipping-quote", as_type=SpanType.TOOL)
    def dispatch(self, order: dict):
        with Netra.start_span("persist-order") as custom:
            custom.set_usage([
                UsageModel(
                    model="gpt-4o",
                    usage_type="generation",
                    units_used=1,
                    cost_in_usd=0.01,
                )
            ])
            custom.set_action([
                ActionModel(
                    action="DB",
                    action_type="INSERT",
                    metadata={"table": "orders"},
                    success=True,
                )
            ])
            custom.add_event("order.persisted", {"order_id": order.get("id")})
        return {"status": "queued", "order_id": order.get("id")}

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

Custom Metrics Pattern

自定义指标模式

python
from netra import Netra

Netra.init(app_name="my-service", enable_metrics=True, metrics_export_interval_ms=30000)
meter = Netra.get_meter("my-service")

requests = meter.create_counter("llm.requests.total", unit="1")
errors = meter.create_counter("llm.errors.total", unit="1")
latency = meter.create_histogram("llm.latency_ms", unit="ms")
active = meter.create_up_down_counter("llm.active_requests", unit="1")
python
from netra import Netra

Netra.init(app_name="my-service", enable_metrics=True, metrics_export_interval_ms=30000)
meter = Netra.get_meter("my-service")

requests = meter.create_counter("llm.requests.total", unit="1")
errors = meter.create_counter("llm.errors.total", unit="1")
latency = meter.create_histogram("llm.latency_ms", unit="ms")
active = meter.create_up_down_counter("llm.active_requests", unit="1")

Integration Pattern Checks

集成模式检查

  • FastAPI/Express: initialize once and add request context middleware.
  • Multi-agent: set 
    @workflow
    root and 
    @agent
    boundaries.
  • Batch: create per-item spans and capture failure metadata.
  • Streaming: ensure spans close only after stream completion.
  • FastAPI/Express：初始化一次并添加请求上下文中间件。
  • 多Agent：设置
    @workflow
    根节点和
    @agent
    边界。
  • 批处理：为每个项目创建跨度并捕获失败元数据。
  • 流式处理：确保仅在流完成后关闭跨度。

Evaluation Code Sample (Python)

评估代码示例（Python）

python
from netra import Netra

Netra.init(app_name="my-app")


def task_fn(input_data):
    # Replace with your app invocation and return output text.
    return my_agent_response(input_data)


dataset = Netra.evaluation.get_dataset(dataset_id="YOUR_DATASET_ID")
result = Netra.evaluation.run_test_suite(
    name="baseline-eval",
    data=dataset,
    task=task_fn,
)

print(result)
python
from netra import Netra

Netra.init(app_name="my-app")


def task_fn(input_data):
    # 替换为你的应用调用逻辑并返回输出文本。
    return my_agent_response(input_data)


dataset = Netra.evaluation.get_dataset(dataset_id="YOUR_DATASET_ID")
result = Netra.evaluation.run_test_suite(
    name="baseline-eval",
    data=dataset,
    task=task_fn,
)

print(result)

Evaluation Code Sample (TypeScript)

评估代码示例（TypeScript）

typescript
import { Netra } from "netra-sdk-js";
import OpenAI from "openai";

const client = new Netra({ apiKey: process.env.NETRA_API_KEY! });
const openai = new OpenAI();

async function taskFn(inputData: any): Promise<string> {
  const response = await openai.chat.completions.create({
        model: "gpt-4o-mini",
        messages: [{ role: "user", content: String(inputData) }],
  });
  return response.choices[0].message.content || "";
}

const dataset = await client.evaluation.getDataset("dataset-123");

if (dataset) {
  const result = await client.evaluation.runTestSuite(
     "baseline-eval",
     dataset,
     taskFn,
     ["correctness", "relevance"],
     10
  );
  console.log(result?.runId);
}
typescript
import { Netra } from "netra-sdk-js";
import OpenAI from "openai";

const client = new Netra({ apiKey: process.env.NETRA_API_KEY! });
const openai = new OpenAI();

async function taskFn(inputData: any): Promise<string> {
  const response = await openai.chat.completions.create({
        model: "gpt-4o-mini",
        messages: [{ role: "user", content: String(inputData) }],
  });
  return response.choices[0].message.content || "";
}

const dataset = await client.evaluation.getDataset("dataset-123");

if (dataset) {
  const result = await client.evaluation.runTestSuite(
     "baseline-eval",
     dataset,
     taskFn,
     ["correctness", "relevance"],
     10
  );
  console.log(result?.runId);
}

Simulation Code Sample (TypeScript)

模拟代码示例（TypeScript）

typescript
import { Netra } from "netra-sdk-js";
import { BaseTask, TaskResult } from "netra-sdk-js/simulation";

const client = new Netra({ apiKey: process.env.NETRA_API_KEY! });

class MyAgentTask extends BaseTask {
    constructor(private readonly agent: any) {
        super();
    }

    async run(message: string, sessionId?: string | null): Promise<TaskResult> {
        const response = await this.agent.chat(message, { sessionId });
        return {
            message: response.text,
            sessionId: sessionId || "default",
        };
    }
}

const result = await client.simulation.runSimulation({
    name: "Customer Support Simulation",
    datasetId: "dataset-123",
    task: new MyAgentTask(myAgent),
    context: { environment: "staging" },
    maxConcurrency: 5,
});

console.log(result?.completed.length, result?.failed.length);
typescript
import { Netra } from "netra-sdk-js";
import { BaseTask, TaskResult } from "netra-sdk-js/simulation";

const client = new Netra({ apiKey: process.env.NETRA_API_KEY! });

class MyAgentTask extends BaseTask {
    constructor(private readonly agent: any) {
        super();
    }

    async run(message: string, sessionId?: string | null): Promise<TaskResult> {
        const response = await this.agent.chat(message, { sessionId });
        return {
            message: response.text,
            sessionId: sessionId || "default",
        };
    }
}

const result = await client.simulation.runSimulation({
    name: "Customer Support Simulation",
    datasetId: "dataset-123",
    task: new MyAgentTask(myAgent),
    context: { environment: "staging" },
    maxConcurrency: 5,
});

console.log(result?.completed.length, result?.failed.length);

Troubleshooting Triage

故障排查分流

  1. Confirm package installation and runtime compatibility.
  2. Confirm single 
    Netra.init()
    call before critical imports.
  3. Verify API key headers and endpoint configuration.
  4. Verify enabled instruments match runtime dependencies.
  5. Verify network egress and shutdown flush.
  6. For missing traces/metrics, check environment flags and export intervals.
  1. 确认包安装和运行时兼容性。
  2. 确认在关键导入前仅调用一次
    Netra.init()
    。
  3. 验证API密钥头和端点配置。
  4. 验证启用的插桩工具与运行时依赖匹配。
  5. 验证网络出口和关闭时的刷新操作。
  6. 对于缺失的追踪/指标，检查环境标志和导出间隔。

Production Checklist

生产环境检查清单

  • Initialization occurs once per process.
  • IDs are set at every request boundary and not leaked across users.
  • Span naming and boundaries reflect business intent.
  • No sensitive payloads are exported unintentionally.
  • Metrics dimensions are bounded and low cardinality.
  • Evaluations and simulations run before release promotion.
  • 初始化在每个进程中仅执行一次。
  • 在每个请求边界设置ID，避免跨用户泄露。
  • 跨度命名和边界反映业务意图。
  • 不会意外导出敏感负载。
  • 指标维度为有限基数。
  • 在发布升级前运行评估和模拟。

References

参考资料