Rastreamento do LangGraph

LangGraph é uma biblioteca de código aberto para desenvolver aplicativos multiatores com gerenciamento de estado usando LLMs, utilizados para criar fluxos de trabalho de agentes e multiagentes.

O MLflow Tracing fornece funcionalidade de rastreamento automático para LangGraph, como uma extensão de sua integração com o LangChain. Ao habilitar o rastreamento automático para LangChain chamando a função mlflow.langchain.autolog, o MLflow capturará automaticamente a execução do grafo em um rastreamento e o registrará no Experimento ativo do MLflow.

import mlflow

mlflow.langchain.autolog()

Pré-requisitos

Para usar o MLflow Tracing com LangGraph, você precisa instalar o MLflow e os pacotes LangGraph e LangChain relevantes (por exemplo, langgraph, , langchain_core, langchain_openai).

Desenvolvimento

Para ambientes de desenvolvimento, instale o pacote completo do MLflow com os extras do Databricks e os pacotes LangGraph/LangChain.

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

O pacote completo mlflow[databricks] inclui todos os recursos para desenvolvimento local e experimentação no Databricks.

Produção

Para implantações de produção, instale mlflow-tracing e os pacotes LangGraph/LangChain:

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

O mlflow-tracing pacote é otimizado para uso em produção.

Antes de executar os exemplos, você precisará configurar seu ambiente:

Para usuários fora dos notebooks do Databricks: defina suas variáveis de ambiente do Databricks:

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

Para usuários dentro de blocos de anotações do Databricks: essas credenciais são definidas automaticamente para você.

Chaves de API: verifique se as chaves de API do provedor LLM estão configuradas. Para ambientes de produção, use os segredos do Gateway de IA do Mosaico ou do Databricks em vez de valores codificados para gerenciamento seguro de chaves de API.

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

Uso

A execução do código a seguir gerará um rastreamento para o grafo, conforme mostrado no clipe de vídeo acima.

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

Adicionando faixas dentro de um nó ou uma ferramenta

Combinando o rastreamento automático com as APIs de rastreamento manual, você pode adicionar intervalos filhos dentro de um nó ou ferramenta para obter insights mais detalhados para a etapa.

Vamos usar o tutorial do Assistente de Código do LangGraph, por exemplo. O nó check_code consiste, na realidade, em duas validações diferentes para o código gerado. Talvez você queira adicionar intervalo para cada validação para ver qual validação foi executada. Para isso, basta criar intervalos manuais dentro da função do nó.

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

Dessa forma, o intervalo do nó check_code terá intervalos filhos, que registram se cada validação falha ou não, com seus detalhes de exceção.

Desabilitar o rastreamento automático

O rastreamento automático do LangGraph pode ser desabilitado globalmente chamando mlflow.langchain.autolog(disable=True) ou mlflow.autolog(disable=True).