Instrumentation LLM
Les appels LLM ont des signaux spécifiques (tokens, coût, latence au premier token, rate-limits). Le SDK fournit des helpers pour les capturer sans plomberie manuelle.
Capture en un appel — trace_llm_call
Passez la réponse brute du vendor : le helper normalise les tokens, lit les headers de rate-limit et émet la complétion (et le raisonnement, pour les blocs « extended thinking » d'Anthropic).
from niaone_sdk import trace_llm_callwith step(flow, asset=llm_asset, flow=Flow(name="support_bot_main", phase=FlowPhase.ANALYSIS, step="answer")) as s: resp = client_openai.chat.completions.create(model="gpt-4o", messages=msgs) trace_llm_call(s, resp) # tokens + rate-limit + payload de complétion| Provider | Reconnu via | Capturé |
|---|---|---|
| OpenAI | usage, choices[].message.content | tokens, complétion |
| Anthropic | usage, content[].text, blocs thinking | tokens (+ cache/reasoning), complétion, raisonnement |
| Gemini / Vertex | usage_metadata, candidates[] | tokens, complétion |
| LangChain JS | llmOutput.tokenUsage | tokens |
Confidentialité
capture_content=False (Py) / captureContent: false (TS) émet uniquement les métriques, pas le
texte. Le niveau content_sensitivity / contentSensitivity taggue les payloads (défaut
INTERNAL). Voir Fiabilité & confidentialité.
Tokens & rate-limits à la main
Si vous ne voulez pas tout capturer, normalisez l'usage et/ou les headers séparément.
from niaone_sdk import normalize_llm_usage, capture_rate_limit_headerss.add_metrics(normalize_llm_usage(resp.usage)) # -> input_tokens / output_tokens / ...capture_rate_limit_headers(s, resp.headers) # -> ratelimit_remaining_* / retry_afterStreaming & temps au premier token (TTFT)
Enveloppez le flux de chunks : le SDK mesure time_to_first_token_ms, le nombre de chunks, le débit
(tokens_per_second) et marque stream_aborted si l'appelant coupe avant la fin.
from niaone_sdk import track_stream # variante async: track_async_streamstream = client_openai.chat.completions.create(model="gpt-4o", messages=msgs, stream=True)text = ""for chunk in track_stream(s, stream, token_extractor=lambda c: c.choices[0].delta.content or ""): text += chunk.choices[0].delta.content or ""Provenance des prompts
Enregistrez vos prompts pour tracer quelle version a produit quelle sortie. Trois modes de capture :
HASH_ONLY (défaut, hash seul), FULL (contenu), REDACTED (variables {{var}} masquées).
from niaone_sdk import register_prompt, attach_promptsfrom niaone_sdk.enums import PromptRole, PromptCaptureMode, TemplateSyntaxsystem_ref = register_prompt(flow, name="support-system", content=SYSTEM_PROMPT, role=PromptRole.SYSTEM, version="v1.2.0", capture_mode=PromptCaptureMode.FULL)user_ref = register_prompt(flow, name="support-user", content="Analyse: {{question}}", role=PromptRole.USER, template_syntax=TemplateSyntax.JINJA2, variables=["question"], capture_mode=PromptCaptureMode.REDACTED)with step(flow, asset=llm_asset, flow=Flow(name="support_bot_main")) as s: attach_prompts(s, [system_ref, user_ref])