Verdict - CoreWeave Docs

Weave est conçu pour suivre et consigner automatiquement tous les appels effectués via la bibliothèque Python Verdict. Lorsque vous travaillez avec des pipelines d’évaluation d’IA, le débogage est essentiel. Qu’une étape du pipeline échoue, que les sorties soient inattendues ou que des opérations imbriquées compliquent la lecture, il peut être difficile de localiser précisément les problèmes. Les applications Verdict comportent souvent plusieurs étapes de pipeline, juges et transformations ; il est donc utile de bien comprendre le fonctionnement interne de vos flux de travail d’évaluation. Weave simplifie ce processus en capturant automatiquement les traces de vos applications Verdict. Vous pouvez ainsi surveiller et analyser les performances de votre pipeline pour déboguer et optimiser vos flux de travail d’évaluation d’IA.

Premiers pas

Pour activer le traçage Weave dans vos pipelines Verdict, appelez weave.init(project=...) au début de votre script. Utilisez l’argument project pour journaliser vers une équipe W&B précise avec team-name/project-name, ou transmettez project-name pour journaliser vers votre équipe ou votre entity par défaut.

import weave
from verdict import Pipeline
from verdict.common.judge import JudgeUnit
from verdict.schema import Schema

# Initialiser Weave avec le nom de votre projet
weave.init("verdict_demo")

# Créer un pipeline d’évaluation simple
pipeline = Pipeline()
pipeline = pipeline >> JudgeUnit().prompt("Rate the quality of this text: {source.text}")

# Créer des exemples de données
data = Schema.of(text="This is a sample text for evaluation.")

# Exécuter le pipeline : Weave le trace automatiquement
output = pipeline.run(data)

print(output)

Suivi des métadonnées des appels

Pour joindre des métadonnées personnalisées aux appels de votre pipeline Verdict, utilisez le gestionnaire de contexte weave.attributes. Ce gestionnaire de contexte vous permet d’étiqueter un bloc de code précis, comme l’exécution d’un pipeline ou un lot d’évaluation, afin de pouvoir ensuite filtrer et regrouper les traces associées dans l’interface de Weights & Biases.

import weave
from verdict import Pipeline
from verdict.common.judge import JudgeUnit
from verdict.schema import Schema

# Initialiser Weave avec le nom de votre projet
weave.init("verdict_demo")

pipeline = Pipeline()
pipeline = pipeline >> JudgeUnit().prompt("Evaluate sentiment: {source.text}")

data = Schema.of(text="I love this product!")

with weave.attributes({"evaluation_type": "sentiment", "batch_id": "batch_001"}):
    output = pipeline.run(data)

print(output)

Weave suit automatiquement les métadonnées associées à la trace de l’appel du pipeline Verdict. Vous pouvez consulter ces métadonnées dans l’interface web de Weave.

Traces

Stocker les traces de vos pipelines d’évaluation d’IA dans une base de données centralisée est utile aussi bien en développement qu’en production. Ces traces facilitent le débogage et l’amélioration de vos flux de travail d’évaluation, tout en constituant un jeu de données précieux. Weave capture automatiquement les traces de vos applications Verdict. Il suit et consigne tous les appels effectués via la bibliothèque Verdict, notamment :

Les étapes d’exécution de Pipeline.
Les évaluations JudgeUnit.
Les transformations Layer.
Les opérations de pooling.
Les unités et transformations personnalisées.

Vous pouvez consulter les traces dans l’interface web de Weave, qui présente la structure hiérarchique de l’exécution de votre pipeline.

Exemple de traçage de pipeline

L’exemple suivant montre comment Weave trace les opérations de pipeline imbriquées, ce qui vous permet de voir comment chaque étape d’un pipeline Verdict à plusieurs phases est capturée :

import weave
from verdict import Pipeline, Layer
from verdict.common.judge import JudgeUnit
from verdict.transform import MeanPoolUnit
from verdict.schema import Schema

# Initialiser Weave avec le nom de votre projet
weave.init("verdict_demo")

# Créer un pipeline en plusieurs étapes
pipeline = Pipeline()
pipeline = pipeline >> Layer([
    JudgeUnit().prompt("Rate coherence: {source.text}"),
    JudgeUnit().prompt("Rate relevance: {source.text}"),
    JudgeUnit().prompt("Rate accuracy: {source.text}")
], 3)
pipeline = pipeline >> MeanPoolUnit()

# Données d’exemple
data = Schema.of(text="This is an evaluation of text quality across multiple dimensions.")

# Exécuter le pipeline : Weave trace toutes les opérations
result = pipeline.run(data)

print(f"Average score: {result}")

Vous obtenez ainsi une trace détaillée qui présente :

L’exécution principale du Pipeline.
Chaque évaluation JudgeUnit au sein de la Layer.
L’étape d’agrégation MeanPoolUnit.
Les informations de durée pour chaque opération.

Configuration

Lorsque vous appelez weave.init(), Weave active automatiquement le traçage des pipelines Verdict. L’intégration applique un patching à la méthode Pipeline.__init__() afin d’injecter un VerdictTracer qui transmet toutes les données de trace vers Weave. Aucune configuration supplémentaire n’est nécessaire. Weave se charge automatiquement des opérations suivantes :

Capture de toutes les opérations du pipeline.
Suivi de la durée d’exécution.
Journalisation des entrées et des sorties.
Conservation de la hiérarchie des traces.
Gestion de l’exécution simultanée des pipelines.

Traceurs personnalisés et Weave

Si vous utilisez déjà des traceurs Verdict personnalisés dans votre application, le VerdictTracer de Weave peut s’exécuter en parallèle, ce qui vous évite d’avoir à choisir entre les intégrations :

import weave
from verdict import Pipeline
from verdict.common.judge import JudgeUnit
from verdict.util.tracing import ConsoleTracer
from verdict.schema import Schema

# Initialiser Weave avec le nom de votre projet
weave.init("verdict_demo")

# Vous pouvez continuer à utiliser les traceurs intégrés de Verdict
console_tracer = ConsoleTracer()

# Créer un pipeline avec à la fois le traçage Weave (automatique) et le traçage console
pipeline = Pipeline(tracer=[console_tracer])  # Le traceur Weave est ajouté automatiquement
pipeline = pipeline >> JudgeUnit().prompt("Evaluate: {source.text}")

data = Schema.of(text="Sample evaluation text")

# Les traces sont envoyées à la fois vers Weave et vers la console
result = pipeline.run(data)

Modèles et évaluations

Organiser et évaluer des systèmes d’IA comportant plusieurs composants de pipeline peut s’avérer complexe. Avec weave.Model, vous pouvez capturer et organiser les détails de vos expériences, comme les prompts, les configurations de pipeline et les paramètres d’évaluation, ce qui facilite la comparaison de différentes itérations. L’exemple suivant montre comment encapsuler un pipeline Verdict dans un weave.Model :

import asyncio
import weave
from verdict import Pipeline
from verdict.common.judge import JudgeUnit
from verdict.schema import Schema

# Initialisez Weave avec le nom de votre projet
weave.init("verdict_demo")

class TextQualityEvaluator(weave.Model):
    judge_prompt: str
    pipeline_name: str

    @weave.op()
    async def predict(self, text: str) -> dict:
        pipeline = Pipeline(name=self.pipeline_name)
        pipeline = pipeline >> JudgeUnit().prompt(self.judge_prompt)
        
        data = Schema.of(text=text)
        result = pipeline.run(data)
        
        return {
            "text": text,
            "quality_score": result.score if hasattr(result, 'score') else result,
            "evaluation_prompt": self.judge_prompt
        }

model = TextQualityEvaluator(
    judge_prompt="Rate the quality of this text on a scale of 1-10: {source.text}",
    pipeline_name="text_quality_evaluator"
)

text = "This is a well-written and informative piece of content that provides clear value to readers."

prediction = asyncio.run(model.predict(text))

# si vous utilisez un Jupyter Notebook, exécutez :
# prediction = await model.predict(text)

print(prediction)

Ce code crée un modèle que vous pouvez visualiser dans l’interface de Weights & Biases, avec à la fois la structure du pipeline et les résultats de l’évaluation.

Évaluations

Les évaluations vous permettent de mesurer les performances de vos pipelines d’évaluation eux-mêmes. La classe weave.Evaluation vous permet de capturer l’efficacité de vos pipelines Verdict sur des tâches ou des jeux de données spécifiques :

import asyncio
import weave
from verdict import Pipeline
from verdict.common.judge import JudgeUnit
from verdict.schema import Schema

# Initialiser Weave
weave.init("verdict_demo")

# Créer le modèle d'évaluation
class SentimentEvaluator(weave.Model):
    @weave.op()
    async def predict(self, text: str) -> dict:
        pipeline = Pipeline()
        pipeline = pipeline >> JudgeUnit().prompt(
            "Classify sentiment as positive, negative, or neutral: {source.text}"
        )
        
        data = Schema.of(text=text)
        result = pipeline.run(data)
        
        return {"sentiment": result}

# Données de test
texts = [
    "I love this product, it's amazing!",
    "This is terrible, worst purchase ever.",
    "The weather is okay today."
]
labels = ["positive", "negative", "neutral"]

examples = [
    {"id": str(i), "text": texts[i], "target": labels[i]}
    for i in range(len(texts))
]

# Fonction de score
@weave.op()
def sentiment_accuracy(target: str, output: dict) -> dict:
    predicted = output.get("sentiment", "").lower()
    return {"correct": target.lower() in predicted}

model = SentimentEvaluator()

evaluation = weave.Evaluation(
    dataset=examples,
    scorers=[sentiment_accuracy],
)

scores = asyncio.run(evaluation.evaluate(model))
# dans un Jupyter Notebook, exécutez plutôt :
# scores = await evaluation.evaluate(model)

print(scores)

Vous obtenez ainsi une trace d’évaluation qui montre les performances de votre pipeline Verdict sur différents cas de test.

Bonnes pratiques

Les sections suivantes décrivent les bonnes pratiques de surveillance des performances et de gestion des erreurs lorsque vous utilisez Weave avec des pipelines Verdict.

Surveillance des performances

Weave capture automatiquement les informations de durée de toutes les opérations du pipeline, qui vous permettent d’identifier les goulots d’étranglement de performances d’un run à l’autre :

import weave
from verdict import Pipeline, Layer
from verdict.common.judge import JudgeUnit
from verdict.schema import Schema

weave.init("verdict_demo")

# Créer un pipeline dont les performances peuvent varier
pipeline = Pipeline()
pipeline = pipeline >> Layer([
    JudgeUnit().prompt("Quick evaluation: {source.text}"),
    JudgeUnit().prompt("Detailed analysis: {source.text}"),  # Celui-ci peut être plus lent
], 2)

data = Schema.of(text="Sample text for performance testing")

# Exécuter plusieurs fois pour observer l’évolution des temps d’exécution
for i in range(3):
    with weave.attributes({"run_number": i}):
        result = pipeline.run(data)

Gestion des erreurs

Weave capture automatiquement les exceptions levées pendant l’exécution du pipeline. Les échecs sont ainsi enregistrés dans la trace, même lorsque votre application gère l’exception :

import weave
from verdict import Pipeline
from verdict.common.judge import JudgeUnit
from verdict.schema import Schema

weave.init("verdict_demo")

pipeline = Pipeline()
pipeline = pipeline >> JudgeUnit().prompt("Process: {source.invalid_field}")  # Cette ligne provoque une erreur

data = Schema.of(text="Sample text")

try:
    result = pipeline.run(data)
except Exception as e:
    print(f"Pipeline failed: {e}")
    # Les détails de l’erreur sont capturés dans la trace Weave

L’intégration de Weave à Verdict vous offre une visibilité sur vos pipelines d’évaluation d’IA, ce qui facilite le débogage, l’optimisation et la compréhension de vos flux de travail d’évaluation.

​Premiers pas

​Suivi des métadonnées des appels

​Traces

​Exemple de traçage de pipeline

​Configuration

​Traceurs personnalisés et Weave

​Modèles et évaluations

​Évaluations

​Bonnes pratiques

​Surveillance des performances

​Gestion des erreurs