Dataiku + carbon-llm : tracker chaque LLM call dans vos recipes Python

Variables, onglet Settings

# Variable name        Variable value
CARBON_LLM_API_KEY     live_xxxxxxxxxxxxxxxxxxxx
CARBON_LLM_BASE_URL    https://carbon-llm.com
CARBON_LLM_DEFAULT_TENANT_ID    ${dataiku.project_key}

Project library → carbon_llm.py

"""
carbon-llm helper, appelable depuis n'importe quel Python recipe ou notebook DSS.
Stockez ce fichier dans le project library : Settings → Code → Libraries → Edit lib.

Usage :
    from carbon_llm import track_llm_call
    track_llm_call(model="gpt-4o", prompt_tokens=1240, completion_tokens=580)
"""

import dataiku
import requests
from typing import Optional

# Lire les variables Dataiku, niveau project en priorité, puis instance
_client = dataiku.api_client()
_proj = dataiku.Project()
_vars = {**_client.get_general_settings().get_raw().get("studioGlobalSettings", {}).get("variables", {}),
         **_proj.get_variables().get("standard", {})}

API_KEY = _vars.get("CARBON_LLM_API_KEY")
BASE_URL = _vars.get("CARBON_LLM_BASE_URL", "https://carbon-llm.com")
DEFAULT_TENANT = _vars.get("CARBON_LLM_DEFAULT_TENANT_ID", _proj.project_key)


def track_llm_call(
    model: str,
    prompt_tokens: int,
    completion_tokens: int,
    tenant_id: Optional[str] = None,
    ingestion_source: str = "dataiku",
) -> None:
    """Fire-and-forget POST vers carbon-llm. Ne lève jamais d'exception,
    les events ratés sont loggés stderr uniquement."""
    if not API_KEY:
        # Pas configuré → no-op silencieux (utile en dev local sans clé)
        return

    try:
        requests.post(
            f"{BASE_URL}/api/v1/track",
            headers={
                "Authorization": f"Bearer {API_KEY}",
                "Content-Type": "application/json",
            },
            json={
                "model": model,
                "prompt_tokens": prompt_tokens,
                "completion_tokens": completion_tokens,
                "tenant_id": tenant_id or DEFAULT_TENANT,
                "ingestion_source": ingestion_source,
            },
            timeout=5.0,
        )
    except Exception as e:
        # Best-effort. Pas de raise, un échec carbon-llm ne casse pas votre flow data.
        print(f"[carbon-llm] track failed: {e}")

Recipe Python utilisant le LLM Mesh

import dataiku
from carbon_llm import track_llm_call

# Récupérer une connexion LLM Mesh configurée dans Administration
client = dataiku.api_client()
project = dataiku.Project()
llm = project.get_llm("openai:my-openai-connection:gpt-4o")

# Appel standard LLM Mesh
response = llm.new_completion()\
    .with_message("Résume ce texte en 3 lignes : ...")\
    .execute()

# Track, l'objet usage est exposé par LLMResponse
track_llm_call(
    model="gpt-4o",
    prompt_tokens=response.usage.prompt_tokens,
    completion_tokens=response.usage.completion_tokens,
    tenant_id=f"{project.project_key}:{dataiku.api_client().get_auth_info()['authIdentifier']}",
)

print(response.text)

Notebook DSS, client OpenAI direct

import os
from openai import OpenAI
from carbon_llm import track_llm_call
import dataiku

# Récupérer la clé OpenAI depuis les variables Dataiku
client = OpenAI(api_key=dataiku.get_custom_variables()["OPENAI_API_KEY"])

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Résume ce CR de réunion : ..."}],
)

# Track immédiatement après la réponse
track_llm_call(
    model="gpt-4o",
    prompt_tokens=response.usage.prompt_tokens,
    completion_tokens=response.usage.completion_tokens,
    tenant_id=dataiku.Project().project_key,
)

print(response.choices[0].message.content)

Scenario step Python

import dataiku
from carbon_llm import track_llm_call

# Si vos recipes ont logué les usages dans une dataset commune (recommandé pour batch) :
client = dataiku.api_client()
project = dataiku.Project()
usage_ds = dataiku.Dataset("llm_usage_log")  # une dataset DSS qui collecte les rows

df = usage_ds.get_dataframe()
for _, row in df.iterrows():
    track_llm_call(
        model=row["model"],
        prompt_tokens=int(row["prompt_tokens"]),
        completion_tokens=int(row["completion_tokens"]),
        tenant_id=f"{project.project_key}:scenario:{row['scenario_run_id']}",
        ingestion_source="dataiku_scenario",
    )

# Vide la dataset après tracking, évite les double-compte au prochain scenario
usage_ds.write_with_schema(df.iloc[0:0])