Instrumenter : flows & steps
Trois façons d'instrumenter, de la plus sûre à la plus avancée. Toutes produisent les mêmes records.
Cycle de vie manuel
Ouvrez un flow, ouvrez des steps, clôturez-les. end_step/end_flow prennent un statut et des
métriques, pas les entrées/sorties (qui passent par payload).
flow = start_flow(client, flow_definition_id="my_sia", name="my_sia_main")step = start_step(flow, asset=asset, flow=Flow(name="my_sia_main", phase=FlowPhase.ANALYSIS, step="answer"))# ... travail ...end_step(step, "success", metrics={"input_tokens": 50})end_flow(flow, "success")Pattern recommandé — context manager
Le plus sûr : tout est encapsulé, la clôture (succès ou exception) est automatique, et le SDK
propage flow_id, parent_span_id et la cohérence du flow.name. Les steps imbriqués se relient
automatiquement à leur parent.
from niaone_sdk import flow, stepwith flow(client, flow_definition_id="my_sia", name="my_sia_main") as f: with step(f, asset=preprocess_asset, flow=Flow(name="my_sia_main", phase=FlowPhase.PREPROCESS, step="snapshot")) as s1: ... # parent commun with step(f, asset=llm_asset, flow=Flow(name="my_sia_main", phase=FlowPhase.ANALYSIS, step="answer")) as s2: payload(s2, "output", Data(role=DataRole.OUTPUT, kind=DataKind.COMPLETION, format=DataFormat.TEXT), result)# end_step / end_flow appelés automatiquementPourquoi ce pattern règle 90 % des problèmes
Tous vos top-level steps partagent un parent commun (le span racine du flow) → le Flow Diagram ancre le DAG (graphe d'exécution) sans ambiguïté, et vous évitez le rendu « cascade » des SIAs qui dispatchent. C'est le correctif du check hierarchy_resolvable.
Décorateur / fonction d'ordre supérieur
Pour envelopper une fonction sans boilerplate. Il faut un flow actif sur le contexte
(start_flow / flow() / withFlow).
from niaone_sdk import instrumentfrom niaone_sdk.enums import SpanKind@instrument( Asset(kind=AssetKind.LLM, role=AssetRole.ANALYSIS, name="gpt-4o"), Flow(name="my_sia_main", phase=FlowPhase.ANALYSIS, step="answer"), kind=SpanKind.LLM,)def analyze(user_input: str) -> str: return llm_call(user_input)Le décorateur s'appelle « instrument »
Il n'existe pas de @flow_step. Le bon symbole est instrument(asset, flow, …) (décorateur en
Python, HOF en TypeScript).
Auto-instrumentation FastAPI / Express
L'auto-instrumentation OTel crée souvent un span racine « http request » sans sémantique métier.
S'il devient racine du trace, NIA One le filtre (pas de flow.name) et tous vos steps deviennent
orphelins. Fix : ouvrez un flow NIA One dès l'entrée du handler, avant tout traitement.
@app.post("/predict")async def predict(req: PredictRequest): with flow(client, flow_definition_id="prediction_api", name="prediction_main") as f: return await run_prediction(f, req)