Śledzenie LangGraph

LangGraph to biblioteka typu open source służąca do tworzenia stanowych, wielopodmiotowych aplikacji z LLM, używanych do tworzenia przepływów pracy agenta i wielu agentów.

Śledzenie MLflow zapewnia funkcję automatycznego śledzenia dla języka LangGraph jako rozszerzenie integracji biblioteki LangChain. Po włączeniu automatycznego śledzenia LangChain przez wywołanie funkcji mlflow.langchain.autolog, narzędzie MLflow automatycznie przechwytuje wykonanie grafu w postaci śladu i loguje je w aktywnym eksperymencie MLflow.

import mlflow

mlflow.langchain.autolog()

Wymagania wstępne

Aby użyć śledzenia MLflow w usłudze LangGraph, należy zainstalować bibliotekę MLflow i odpowiednie pakiety LangGraph i LangChain (np. langgraph, langchain_core, langchain_openai).

Rozwój

W przypadku środowisk deweloperskich zainstaluj pełny pakiet MLflow z dodatkiami usługi Databricks i pakietami LangGraph/LangChain:

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

Pełny mlflow[databricks] pakiet zawiera wszystkie funkcje lokalnego programowania i eksperymentowania w usłudze Databricks.

Produkcja

W przypadku wdrożeń produkcyjnych zainstaluj mlflow-tracing oraz pakiety LangGraph/LangChain.

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

Pakiet mlflow-tracing jest zoptymalizowany pod kątem użycia w środowisku produkcyjnym.

Przed uruchomieniem przykładów należy skonfigurować środowisko:

W przypadku użytkowników niekorzystających z notesów Databricks: Ustaw zmienne środowiskowe usługi Databricks:

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

W przypadku użytkowników w notesach usługi Databricks: poświadczenia te są ustawiane automatycznie.

Klucze interfejsu API: upewnij się, że klucze interfejsu API dostawcy LLM są skonfigurowane. W środowiskach produkcyjnych użyj wpisów tajnych usługi Mosaic AI Gateway lub Databricks zamiast zakodowanych na stałe wartości do bezpiecznego zarządzania kluczami interfejsu API.

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

Zastosowanie

Uruchomienie poniższego kodu spowoduje wygenerowanie śladu wykresu, jak pokazano na powyższym klipie wideo.

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

Dodawanie zakresów w węźle lub narzędziu

Łącząc automatyczne śledzenie z interfejsami API śledzenia ręcznego, można dodać podrzędne elementy wewnątrz węzła lub narzędzia, aby uzyskać bardziej szczegółowe informacje o kroku.

Weźmy na przykład samouczek Asystenta kodu firmy LangGraph. Węzeł check_code faktycznie składa się z dwóch różnych walidacji wygenerowanego kodu. Możesz dodać zakres dla każdej walidacji, aby zobaczyć, która walidacja została wykonana. W tym celu wystarczy utworzyć ręczne zakresy w funkcji danego węzła.

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

W ten sposób zakres węzła check_code będzie miał zakresy podrzędne, które rejestrują, czy każda walidacja kończy się niepowodzeniem, wraz ze szczegółami wyjątku.

Wyłączanie automatycznego śledzenia

Automatyczne śledzenie dla języka LangGraph można wyłączyć globalnie, wywołując polecenie mlflow.langchain.autolog(disable=True) lub mlflow.autolog(disable=True).