Verfolgung der Anthropik

MLflow Tracing bietet eine automatische Ablaufverfolgungsfunktion für Anthropic-LLMs. Durch Aktivieren der automatischen Ablaufverfolgung für Anthropic durch Aufrufen der mlflow.anthropic.autolog Funktion erfasst MLflow geschachtelte Ablaufverfolgungen und protokolliert sie beim aktiven MLflow-Experiment bei Aufruf des Anthropic Python SDK.

import mlflow

mlflow.anthropic.autolog()

Die MLflow-Nachverfolgung erfasst automatisch die folgenden Informationen zu Anthropic-Anfragen:

• Eingabeaufforderungen und Vervollständigungsantworten
• Wartezeiten
• Modellname
• Zusätzliche Metadaten wie temperature, max_tokens, falls angegeben.
• Funktionsaufrufe, wenn sie in der Antwort enthalten sind
• Jede ausgelöste Ausnahme

Hinweis

Derzeit unterstützt MLflow-Anthropic-Integration nur die Ablaufverfolgung von synchronen Aufrufen bei Textinteraktionen. Asynchrone APIs sind nicht nachverfolgbar, und vollständige Eingaben können für multimodale Eingaben nicht aufgezeichnet werden.

Voraussetzungen

Um die MLflow-Ablaufverfolgung mit Anthropic zu verwenden, müssen Sie MLflow und das Anthropic SDK installieren.

Entwicklung

Für Entwicklungsumgebungen installieren Sie das vollständige MLflow-Paket mit Databricks-Zusatzkomponenten und anthropic:

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

Das vollständige mlflow[databricks] Paket enthält alle Features für die lokale Entwicklung und Das Experimentieren mit Databricks.

Produktion

Für die Produktionsimplementierung installieren Sie mlflow-tracing und anthropic.

pip install --upgrade mlflow-tracing anthropic

Das Paket ist für den mlflow-tracing Produktionseinsatz optimiert.

Hinweis

MLflow 3 wird dringend empfohlen, um die beste Erfahrung beim Tracking mit Anthropic zu erzielen.

Bevor Sie die folgenden Beispiele ausführen, müssen Sie Ihre Umgebung konfigurieren:

Für Benutzer außerhalb von Databricks-Notizbüchern: Legen Sie Ihre Databricks-Umgebungsvariablen fest:

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

Für Benutzer innerhalb von Databricks-Notizbüchern: Diese Anmeldeinformationen werden automatisch für Sie festgelegt.

API-Schlüssel: Stellen Sie sicher, dass Ihr anthropischer API-Schlüssel konfiguriert ist:

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

Unterstützte APIs

MLflow unterstützt die automatische Ablaufverfolgung für die folgenden anthropischen APIs:

Chatvervollständigung	Funktionsaufruf	Streamen	Asynchron	Abbildung	Stapel
✅	✅		✅ (*1)

(*1) Asynchrone Unterstützung wurde in MLflow 2.21.0 hinzugefügt.

Um Support für zusätzliche APIs anzufordern, öffnen Sie eine Featureanfrage auf GitHub.

Einfaches Beispiel

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"},
    ],
)

Asynchron

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"},
    ],
)

Erweitertes Beispiel: Tool-Aufruf-Agent

MLflow Tracing erfasst automatisch die Tool-Aufrufantwort von Anthropic-Modellen. Die Funktionsanweisung in der Antwort wird in der Ablaufverfolgungsschnittstelle hervorgehoben. Darüber hinaus können Sie die Toolfunktion mit dem @mlflow.trace Dekorator annotieren, um einen Bereich für die Toolausführung zu erstellen.

Im folgenden Beispiel wird ein einfacher Funktionsaufruf-Agent mithilfe von Anthropic Tool Calling und MLflow Tracing für Anthropic implementiert. Im Beispiel wird außerdem das asynchrone Anthropic SDK verwendet, sodass der Agent gleichzeitige Aufrufe ohne Blockieren verarbeiten kann.

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}")

Automatische Ablaufverfolgung deaktivieren

Die automatische Ablaufverfolgung für Anthropic kann global durch Aufrufen von mlflow.anthropic.autolog(disable=True) oder mlflow.autolog(disable=True) deaktiviert werden.