Rastreo de LangGraph

LangGraph es una librería de código abierto para la construcción de aplicaciones con estado y multi-actor con LLMs, utilizada para crear flujos de trabajo de agentes y multi-agentes.

Rastrep de MLflow proporciona la capacidad de rastreo automático para LangGraph, como una extensión de su integración LangChain. Al habilitar el trazado automático para LangChain llamando a la función mlflow.langchain.autolog, MLflow capturará automáticamente la ejecución del gráfico en un trazado y lo registrará en el Experimento MLflow activo.

import mlflow

mlflow.langchain.autolog()

Prerrequisitos

Para usar el seguimiento de MLflow con LangGraph, debe instalar MLflow y los paquetes de LangGraph y LangChain pertinentes (por ejemplo, langgraph, , langchain_core). langchain_openai

Desarrollo

Para entornos de desarrollo, instale el paquete MLflow completo con extras de Databricks y paquetes LangGraph/LangChain:

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

El paquete completo mlflow[databricks] incluye todas las características para el desarrollo local y la experimentación en Databricks.

Producción

Para las implementaciones de producción, instale mlflow-tracing y los paquetes LangGraph/LangChain:

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

El mlflow-tracing paquete está optimizado para su uso en producción.

Antes de ejecutar los ejemplos, deberá configurar el entorno:

Para los usuarios que no utilizan los cuadernos de Databricks: establece las variables de entorno de Databricks:

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

Para usuarios dentro de cuadernos Databricks: Estas credenciales se establecen automáticamente para usted.

Claves de API: asegúrese de que las claves de API del proveedor llM están configuradas. Para entornos de producción, use Mosaic AI Gateway o secretos de Databricks en lugar de valores codificados de forma rígida para la administración segura de claves de API.

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

Uso

Al ejecutar el código siguiente, se generará un seguimiento para el gráfico, tal como se muestra en el clip de vídeo anterior.

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

Agregar intervalos dentro de un nodo o una herramienta

Al combinar el seguimiento automático con las API de seguimiento manual, puede agregar intervalos secundarios dentro de un nodo o herramienta para obtener información más detallada sobre el paso.

Tomemos, por ejemplo, el tutorial de Code Assistant de LangGraph. El check_code nodo consta realmente de dos validaciones diferentes para el código generado. Es posible que quiera agregar intervalo para cada validación para ver qué validación se ejecutó. Para ello, basta con crear intervalos manuales dentro de la función de nodo.

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

De este modo, el intervalo del check_code nodo tendrá intervalos secundarios, que registran si se produce un error en cada validación o no, con sus detalles de excepción.

Deshabilitar el seguimiento automático

El seguimiento automático de LangGraph se puede deshabilitar globalmente llamando a mlflow.langchain.autolog(disable=True) o mlflow.autolog(disable=True).