Ablaufverfolgung für LangGraph

LangGraph ist eine Open Source-Bibliothek zum Erstellen zustandsbehafteter, mehrstufiger Anwendungen mit LLMs, die zum Erstellen von Agent- und Multi-Agent-Workflows verwendet werden.

Die MLflow-Ablaufverfolgung bietet automatische Ablaufverfolgungsfunktionen für LangGraph als Erweiterung der LangChain-Integration. Durch das Aktivieren der automatischen Ablaufverfolgung für LangChain durch Aufrufen der Funktion mlflow.langchain.autolog erfasst MLflow automatisch die Graphausführung in einer Ablaufverfolgung und protokolliert sie im aktiven MLflow-Experiment.

import mlflow

mlflow.langchain.autolog()

Voraussetzungen

Um die MLflow-Ablaufverfolgung mit LangGraph zu verwenden, müssen Sie MLflow und die relevanten LangGraph- und LangChain-Pakete (z. B. langgraph, langchain_core, langchain_openai) installieren.

Entwicklung

Installieren Sie für Entwicklungsumgebungen das vollständige MLflow-Paket mit Databricks-Extras und LangGraph/LangChain-Paketen:

pip install --upgrade "mlflow[databricks]>=3.1" langgraph langchain_core langchain_openai
# Add other langchain packages if needed by your graph

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

Produktion

Für Produktionsbereitstellungen installieren Sie mlflow-tracing und die LangGraph-/LangChain-Pakete.

pip install --upgrade mlflow-tracing langgraph langchain_core langchain_openai
# Add other langchain packages if needed by your graph

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

Bevor Sie die 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 Ihre LLM-Anbieter-API-Schlüssel konfiguriert sind. Verwenden Sie in Produktionsumgebungen Mosaic AI Gateway oder Databricks-Secrets anstelle von hartcodierten Werten für eine sichere Verwaltung von API-Schlüsseln.

export OPENAI_API_KEY="your-openai-api-key"
# Add other provider keys as needed

Verwendung

Wenn Sie den folgenden Code ausführen, wird ein Trace für das Diagramm generiert, wie im obigen Videoclip gezeigt.

from typing import Literal
import os
import mlflow

from langchain_core.messages import AIMessage, ToolCall
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

# Ensure your OPENAI_API_KEY (or other LLM provider keys) is set in your environment
# os.environ["OPENAI_API_KEY"] = "your-openai-api-key" # Uncomment and set if not globally configured

# Enabling tracing for LangGraph (LangChain)
mlflow.langchain.autolog()

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


@tool
def get_weather(city: Literal["nyc", "sf"]):
    """Use this to get weather information."""
    if city == "nyc":
        return "It might be cloudy in nyc"
    elif city == "sf":
        return "It's always sunny in sf"


llm = ChatOpenAI(model="gpt-4o-mini")
tools = [get_weather]
graph = create_react_agent(llm, tools)

# Invoke the graph
result = graph.invoke(
    {"messages": [{"role": "user", "content": "what is the weather in sf?"}]}
)

Hinzufügen von Spannen innerhalb eines Knotens oder Tools

Durch die Kombination der automatischen Ablaufverfolgung mit den manuellen Ablaufverfolgungs-APIs können Sie untergeordnete Spannen innerhalb eines Knotens oder Tools hinzufügen, um detailliertere Erkenntnisse für den Schritt zu erhalten.

Sehen wir uns beispielsweise das Lernprogramm des LangGraph-Code-Assistenten an . Der check_code Knoten besteht tatsächlich aus zwei verschiedenen Überprüfungen für den generierten Code. Sie können für jede Validierung eine Spanne hinzufügen, um zu sehen, welche Validierungen ausgeführt wurden. Erstellen Sie dazu einfach manuelle Spannweiten innerhalb der Knotenfunktion.

def code_check(state: GraphState):
    # State
    messages = state["messages"]
    code_solution = state["generation"]
    iterations = state["iterations"]

    # Get solution components
    imports = code_solution.imports
    code = code_solution.code

    # Check imports
    try:
        # Create a child span manually with mlflow.start_span() API
        with mlflow.start_span(name="import_check", span_type=SpanType.TOOL) as span:
            span.set_inputs(imports)
            exec(imports)
            span.set_outputs("ok")
    except Exception as e:
        error_message = [("user", f"Your solution failed the import test: {e}")]
        messages += error_message
        return {
            "generation": code_solution,
            "messages": messages,
            "iterations": iterations,
            "error": "yes",
        }

    # Check execution
    try:
        code = imports + "\n" + code
        with mlflow.start_span(name="execution_check", span_type=SpanType.TOOL) as span:
            span.set_inputs(code)
            exec(code)
            span.set_outputs("ok")
    except Exception as e:
        error_message = [("user", f"Your solution failed the code execution test: {e}")]
        messages += error_message
        return {
            "generation": code_solution,
            "messages": messages,
            "iterations": iterations,
            "error": "yes",
        }

    # No errors
    return {
        "generation": code_solution,
        "messages": messages,
        "iterations": iterations,
        "error": "no",
    }

Auf diese Weise enthält die Spanne für den Knoten check_code untergeordnete Spannen, die aufzeichnen, ob die jeweilige Überprüfung fehlerhaft ist oder nicht, einschließlich ihrer Ausnahmedetails.

Automatische Ablaufverfolgung deaktivieren

Die automatische Ablaufverfolgung für LangGraph kann durch den Aufruf von mlflow.langchain.autolog(disable=True) oder mlflow.autolog(disable=True) global deaktiviert werden.