Traçage Anthropic

MLflow Tracing fournit une capacité de suivi automatique pour les LLM Anthropic. En activant le traçage automatique pour Anthropic en appelant la fonction mlflow.anthropic.autolog, MLflow capturera les traces imbriquées et les enregistrera dans l'Experimentation MLflow actif lors de l'invocation du SDK Python Anthropic.

Python

import mlflow

mlflow.anthropic.autolog()

La trace MLflow capture automatiquement les informations suivantes concernant les appels Anthropic :

• Prompts et réponses de complétion
• Latences
• Nom du modèle
• Métadonnées supplémentaires telles que temperature, max_tokens, si spécifié.
• Appel de fonction si renvoyé dans la réponse
• Toute exception si levée

remarque

Sur les clusters de compute serverless, l'autologging n'est pas activé automatiquement. Vous devez appeler explicitement mlflow.anthropic.autolog() pour activer le traçage automatique de cette intégration.

remarque

Actuellement, l'intégration MLflow Anthropic ne prend en charge le traçage que pour les appels synchrones pour les interactions textuelles. Les APIs asynchrones ne sont pas tracées, et les entrées complètes ne peuvent pas être enregistrées pour les entrées multimodales.

Prérequis

Pour utiliser MLflow Tracing avec Anthropic, vous devez installer MLflow et le SDK Anthropic.

Development

Pour les environnements de développement, installez le package MLflow complet avec les extras Databricks et anthropic:

Bash

pip install --upgrade "mlflow[databricks]>=3.1" anthropic

Le package mlflow[databricks] complet inclut toutes les fonctionnalités pour le développement local et l’expérimentation sur Databricks.

Production

Pour les déploiements en production, installez mlflow-tracing et anthropic:

Bash

pip install --upgrade mlflow-tracing anthropic

Le mlflow-tracing package est optimisé pour une utilisation en production.

remarque

MLflow 3 est fortement recommandé pour une expérience de traçage optimale avec Anthropic.

Avant d'exécuter les exemples ci-dessous, vous devrez configurer votre environnement :

Pour les utilisateurs en dehors des notebooks Databricks : Définissez vos variables d'environnement Databricks :

Bash

export DATABRICKS_HOST="https://your-workspace.cloud.databricks.com"
export DATABRICKS_TOKEN="your-personal-access-token"

Pour les utilisateurs dans les Notebooks Databricks : ces identifiants sont définis automatiquement pour vous.

Clés API : Assurez-vous que votre clé API Anthropic est configurée. Pour une utilisation en production, nous vous recommandons d’utiliser AI Gateway ou les secrets Databricks plutôt que des variables d’environnement :

Bash

export ANTHROPIC_API_KEY="your-anthropic-api-key"

APIs prises en charge

MLflow prend en charge le traçage automatique pour les APIs Anthropic suivantes :

Achèvement du chat	Appel de fonction	Streaming	Asynchrone	Image	Batch
✅	✅		✅ (*1)

Achèvement du chat	Appel de fonction	Streaming	Asynchrone	Image	Batch
✅	✅		✅ (*1)

(*1) Le support asynchrone a été ajouté dans MLflow 2.21.0.

Pour demander la prise en charge d'APIs supplémentaires, veuillez ouvrir une demande de fonctionnalité sur GitHub.

Exemple de base

Python

import anthropic
import mlflow
import os

# Ensure your ANTHROPIC_API_KEY is set in your environment
# os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-api-key" # Uncomment and set if not globally configured

# Enable auto-tracing for Anthropic
mlflow.anthropic.autolog()

# Set up MLflow tracking to Databricks
mlflow.set_tracking_uri("databricks")
mlflow.set_experiment("/Shared/anthropic-tracing-demo")

# Configure your API key.
client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

# Use the create method to create new message.
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Hello, Claude"},
    ],
)

attention

Pour les environnements de production, utilisez toujours AI Gateway ou les secrets Databricks au lieu de valeurs codées en dur ou de variables d'environnement.

Async

Python

import anthropic
import mlflow
import os

# Ensure your ANTHROPIC_API_KEY is set in your environment
# os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-api-key" # Uncomment and set if not globally configured

# Enable trace logging
mlflow.anthropic.autolog()

# Set up MLflow tracking to Databricks if not already configured
# mlflow.set_tracking_uri("databricks")
# mlflow.set_experiment("/Shared/anthropic-async-demo")

client = anthropic.AsyncAnthropic()

response = await client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Hello, Claude"},
    ],
)

Exemple avancé : Agent d'appel d'outils

MLflow Tracing capture automatiquement la réponse d'appel d'outil des modèles Anthropic. L’instruction de fonction dans la réponse sera mise en surbrillance dans l’interface utilisateur de la trace. De plus, vous pouvez annoter la fonction d’outil avec le décorateur @mlflow.trace pour créer un intervalle pour l’exécution de l’outil.

L'exemple suivant implémente un agent d'appel de fonction simple utilisant Anthropic Tool Calling et MLflow Tracing pour Anthropic. L'exemple utilise en outre le SDK Anthropic asynchrone afin que l'agent puisse gérer les invocations concurrentes sans blocage.

Python

import json
import anthropic
import mlflow
import asyncio
from mlflow.entities import SpanType
import os

# Ensure your ANTHROPIC_API_KEY is set in your environment
# os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-api-key" # Uncomment and set if not globally configured

# Set up MLflow tracking to Databricks if not already configured
# mlflow.set_tracking_uri("databricks")
# mlflow.set_experiment("/Shared/anthropic-tool-agent-demo")

# Assuming autolog is enabled globally or called earlier
# mlflow.anthropic.autolog()

client = anthropic.AsyncAnthropic()
model_name = "claude-3-5-sonnet-20241022"


# Define the tool function. Decorate it with `@mlflow.trace` to create a span for its execution.
@mlflow.trace(span_type=SpanType.TOOL)
async def get_weather(city: str) -> str:
    if city == "Tokyo":
        return "sunny"
    elif city == "Paris":
        return "rainy"
    return "unknown"


tools = [
    {
        "name": "get_weather",
        "description": "Returns the weather condition of a given city.",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    }
]

_tool_functions = {"get_weather": get_weather}


# Define a simple tool calling agent
@mlflow.trace(span_type=SpanType.AGENT)
async def run_tool_agent(question: str):
    messages = [{"role": "user", "content": question}]

    # Invoke the model with the given question and available tools
    ai_msg = await client.messages.create(
        model=model_name,
        messages=messages,
        tools=tools,
        max_tokens=2048,
    )
    messages.append({"role": "assistant", "content": ai_msg.content})

    # If the model requests tool call(s), invoke the function with the specified arguments
    tool_calls = [c for c in ai_msg.content if c.type == "tool_use"]
    for tool_call in tool_calls:
        if tool_func := _tool_functions.get(tool_call.name):
            tool_result = await tool_func(**tool_call.input)
        else:
            raise RuntimeError("An invalid tool is returned from the assistant!")

        messages.append(
            {
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": tool_call.id,
                        "content": tool_result,
                    }
                ],
            }
        )

    # Send the tool results to the model and get a new response
    response = await client.messages.create(
        model=model_name,
        messages=messages,
        max_tokens=2048,
    )

    return response.content[-1].text


# Run the tool calling agent
cities = ["Tokyo", "Paris", "Sydney"]
questions = [f"What's the weather like in {city} today?" for city in cities]
answers = await asyncio.gather(*(run_tool_agent(q) for q in questions))

for city, answer in zip(cities, answers):
    print(f"{city}: {answer}")

Désactiver le suivi automatique

Le traçage automatique pour Anthropic peut être désactivé globalement en appelant mlflow.anthropic.autolog(disable=True) ou mlflow.autolog(disable=True).