Monitoring en ligne - CoreWeave Docs

Ceci est un notebook interactif. Vous pouvez l’exécuter localement ou utiliser les liens ci-dessous :

Intégrer avec Weave : tableau de bord de production

Ce notebook montre comment utiliser les API et les fonctions de Weave pour créer un tableau de bord personnalisé de monitoring en production, en complément de la vue Traces de Weave. Ce guide s’adresse aux développeurs et aux ingénieurs ML qui exécutent des applications LLM en production et souhaitent une visibilité sur mesure sur les performances, les coûts et les feedbacks des utilisateurs, au-delà de ce que propose la vue Traces par défaut. Ce notebook aborde les points suivants :

Récupérer depuis Weave les traces, les coûts, les feedbacks et d’autres métriques.
Créer des vues agrégées des feedbacks des utilisateurs et de la répartition des coûts.
Créer des visualisations de l’évolution de l’utilisation des jetons et de la latence dans le temps.

À la fin de ce notebook, vous disposerez d’un tableau de bord personnalisé fonctionnel qui regroupe dans une vue unique les données de trace, les coûts et les feedbacks de votre projet Weave. Vous pouvez essayer ce tableau de bord avec votre propre projet Weave en installant streamlit et en exécutant le script du tableau de bord de production. Exemple de tableau de bord de production avec Weave

Exemple de tableau de bord de production avec Weave

Configuration

Pour commencer, installez les paquets suivants :

!pip install streamlit pandas plotly weave

Implémentation

Les sections suivantes expliquent comment initialiser le client Weave, récupérer les données des appels et générer les visualisations du tableau de bord.

Initialiser le client Weave et définir les coûts

Commencez par configurer une fonction qui initialise le client Weave et ajoute les coûts de chaque modèle. Cette étape est requise pour que les requêtes de coûts en aval puissent associer une tarification par jeton à chaque appel. W&B inclut les coûts standard de nombreux modèles courants et vous permet également d’ajouter vos propres coûts et modèles personnalisés. L’exemple suivant montre comment ajouter des coûts personnalisés pour quelques modèles et utiliser les coûts standard pour les autres. Les coûts sont calculés à partir des jetons suivis pour chaque appel dans Weave. Pour de nombreuses bibliothèques de fournisseurs de LLM, Weave suit automatiquement l’utilisation des jetons, mais vous pouvez également renvoyer vos propres nombres de jetons pour n’importe quel appel. Pour en savoir plus sur la définition du nombre de jetons et le calcul des coûts pour un modèle personnalisé, consultez le cookbook sur les coûts personnalisés.

PROJECT_NAME = "wandb-smle/weave-cookboook-demo"
python
import weave

MODEL_NAMES = [
    # nom du modèle, coût du prompt, coût de la complétion
    ("gpt-4o-2024-05-13", 0.03, 0.06),
    ("gpt-4o-mini-2024-07-18", 0.03, 0.06),
    ("gemini/gemini-1.5-flash", 0.00025, 0.0005),
    ("gpt-4o-mini", 0.03, 0.06),
    ("gpt-4-turbo", 0.03, 0.06),
    ("claude-3-haiku-20240307", 0.01, 0.03),
    ("gpt-4o", 0.03, 0.06),
]

def init_weave_client(project_name):
    try:
        client = weave.init(project_name)
        for model, prompt_cost, completion_cost in MODEL_NAMES:
            client.add_cost(
                llm_id=model,
                prompt_token_cost=prompt_cost,
                completion_token_cost=completion_cost,
            )
    except Exception as e:
        print(f"Failed to initialize Weave client for project '{project_name}': {e}")
        return None
    else:
        return client

client = init_weave_client(PROJECT_NAME)

Récupérer les données des appels depuis Weave

Une fois le client initialisé et les coûts configurés, l’étape suivante consiste à récupérer les données des appels depuis Weave. Deux options s’offrent à vous pour récupérer ces données :

Récupérer les données appel par appel.
Utiliser des API de haut niveau.

Les sections suivantes décrivent chacune de ces options.

Récupérer les données appel par appel

La première méthode pour accéder aux données de Weave consiste à récupérer une liste d’appels filtrés, puis à extraire les données souhaitées appel par appel. Pour ce faire, utilisez l’API calls_query_stream afin de récupérer les données des appels depuis Weave :

API calls_query_stream : cette API récupère les données des appels depuis Weave.
Dictionnaire filter : ce dictionnaire contient les paramètres de filtrage utilisés pour récupérer les données des appels. Pour plus de détails, voir la référence CallSchema.
Liste expand_columns : cette liste contient les colonnes à développer dans les données des appels.
Liste sort_by : cette liste contient les paramètres de tri des données des appels.
Booléen include_costs : ce booléen indique si les coûts doivent être inclus dans les données des appels.
Booléen include_feedback : ce booléen indique si le feedback doit être inclus dans les données des appels.

import itertools
from datetime import datetime, timedelta

import pandas as pd

def fetch_calls(client, project_id, start_time, trace_roots_only, limit):
    filter_params = {
        "project_id": project_id,
        "filter": {"started_at": start_time, "trace_roots_only": trace_roots_only},
        "expand_columns": ["inputs.example", "inputs.model"],
        "sort_by": [{"field": "started_at", "direction": "desc"}],
        "include_costs": True,
        "include_feedback": True,
    }
    try:
        calls_stream = client.server.calls_query_stream(filter_params)
        calls = list(
            itertools.islice(calls_stream, limit)
        )  # limiter le nombre d'appels récupérés s'ils sont trop nombreux
        print(f"Fetched {len(calls)} calls.")
    except Exception as e:
        print(f"Error fetching calls: {e}")
        return []
    else:
        return calls

calls = fetch_calls(client, PROJECT_NAME, datetime.now() - timedelta(days=1), True, 100)
python
# les données brutes se présentent sous forme de liste d'objets Call
pd.DataFrame([call.dict() for call in calls]).head(3)

Traitez les appels à l’aide des données renvoyées par Weave. Extrayez les informations pertinentes et stockez-les dans une liste de dictionnaires. Convertissez ensuite cette liste de dictionnaires en DataFrame pandas, puis renvoyez-le.

import json
from datetime import datetime

import pandas as pd

def process_calls(calls):
    records = []
    for call in calls:
        feedback = call.summary.get("weave", {}).get("feedback", [])
        thumbs_up = sum(
            1
            for item in feedback
            if isinstance(item, dict) and item.get("payload", {}).get("emoji") == "👍"
        )
        thumbs_down = sum(
            1
            for item in feedback
            if isinstance(item, dict) and item.get("payload", {}).get("emoji") == "👎"
        )
        latency = call.summary.get("weave", {}).get("latency_ms", 0)

        records.append(
            {
                "Call ID": call.id,
                "Trace ID": call.trace_id,  # ID unique de la trace, utilisable pour la récupérer
                "Display Name": call.display_name,  # nom facultatif, que vous pouvez définir dans l’interface utilisateur ou par programmation
                "Latency (ms)": latency,
                "Thumbs Up": thumbs_up,
                "Thumbs Down": thumbs_down,
                "Started At": pd.to_datetime(getattr(call, "started_at", datetime.min)),
                "Inputs": json.dumps(call.inputs, default=str),
                "Outputs": json.dumps(call.output, default=str),
            }
        )
    return pd.DataFrame(records)
python
df_calls = process_calls(calls)
df_calls.head(3)

Utiliser les API de haut niveau

Plutôt que de parcourir chaque appel un par un, Weave fournit également des API de haut niveau qui permettent d’accéder directement aux coûts des modèles, au feedback et à d’autres métriques. Par exemple, pour les coûts, utilisez l’API query_costs pour récupérer les coûts de tous les LLM utilisés dans le projet :

# Utiliser l'API des coûts pour obtenir les coûts
costs = client.query_costs()
df_costs = pd.DataFrame([cost.dict() for cost in costs])
df_costs["total_cost"] = (
    df_costs["prompt_token_cost"] + df_costs["completion_token_cost"]
)

# afficher uniquement la première ligne de chaque llm_id unique
df_costs

Collecter les entrées et générer les visualisations

Une fois les données d’appels et les coûts disponibles sous forme de DataFrames, vous pouvez générer les visualisations avec plotly. Voici un tableau de bord de base, que vous pouvez personnaliser à votre guise. Pour un exemple plus avancé, consultez l’exemple Streamlit du dépôt knowledge-worker-weave.

import plotly.express as px
import plotly.graph_objects as go

def plot_feedback_pie_chart(thumbs_up, thumbs_down):
    fig = go.Figure(
        data=[
            go.Pie(
                labels=["Thumbs Up", "Thumbs Down"],
                values=[thumbs_up, thumbs_down],
                marker={"colors": ["#66b3ff", "#ff9999"]},
                hole=0.3,
            )
        ]
    )
    fig.update_traces(textinfo="percent+label", hoverinfo="label+percent")
    fig.update_layout(showlegend=False, title="Feedback Summary")
    return fig

def plot_model_cost_distribution(df):
    fig = px.bar(
        df,
        x="llm_id",
        y="total_cost",
        color="llm_id",
        title="Cost Distribution by Model",
    )
    fig.update_layout(xaxis_title="Model", yaxis_title="Cost (USD)")
    return fig

# Consultez le code source pour voir l’ensemble des graphiques
python
plot_feedback_pie_chart(df_calls["Thumbs Up"].sum(), df_calls["Thumbs Down"].sum())
python
plot_model_cost_distribution(df_costs)

Conclusion

Ce cookbook a montré comment créer un tableau de bord personnalisé de monitoring en production à l’aide des API et des fonctions de Weave. Weave mise sur des intégrations rapides, qui simplifient aussi bien l’ingestion des données que leur extraction pour des traitements personnalisés.

Entrée des données :
- Traçage indépendant du framework avec le décorateur @weave-op() et possibilité d’importer des appels depuis un fichier CSV (voir le cookbook d’importation associé).
- Points de terminaison de l’API de service pour journaliser vers Weave depuis différents frameworks et langages de programmation. Pour plus de détails, consultez la référence de l’API de service.
Sortie des données :
- Téléchargez les données aux formats CSV, TSV, JSONL ou JSON. Pour plus de détails, consultez la référence de l’API de service.
- Exportez les données par programmation. Consultez la section « Use Python » du panneau d’exportation, comme décrit dans ce cookbook. Pour plus de détails, consultez Interroger et exporter des appels.

Ce tableau de bord personnalisé complète la vue Traces native de Weave pour offrir un monitoring sur mesure des applications LLM en production. Pour découvrir un tableau de bord plus élaboré, consultez l’exemple Streamlit dans le dépôt agent-dev-collection, dans lequel vous pouvez renseigner l’URL de votre propre projet Weave.

​Intégrer avec Weave : tableau de bord de production

​Configuration

​Implémentation

​Initialiser le client Weave et définir les coûts

​Récupérer les données des appels depuis Weave

​Récupérer les données appel par appel

​Utiliser les API de haut niveau

​Collecter les entrées et générer les visualisations

​Conclusion

Intégrer avec Weave : tableau de bord de production

Configuration

Implémentation

Initialiser le client Weave et définir les coûts

Récupérer les données des appels depuis Weave

Récupérer les données appel par appel

Utiliser les API de haut niveau

Collecter les entrées et générer les visualisations

Conclusion