Rastreamento LangGraph

LangGraph é uma biblioteca de código aberto para a criação de aplicativos com estado e com múltiplos atores com LLMs, usados para criar fluxos de trabalho de agentes e multiagentes.

MLflow Tracing fornece capacidade de rastreamento automático para LangGraph, como extensão da integração com LangChain. Ao habilitar o rastreamento automático para LangChain através da função mlflow.langchain.autolog, o MLflow capturará automaticamente a execução do gráfico em um traço e registrará no experimento ativo do MLflow.

import mlflow

mlflow.langchain.autolog()

Pré-requisitos

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

Desenvolvimento

Para ambientes de desenvolvimento, instale o pacote MLflow completo 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 utilizadores fora dos notebooks Databricks: defina as suas variáveis de ambiente Databricks:

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

Para utilizadores dentro de blocos de notas Databricks: estas credenciais são definidas automaticamente para si.

Chaves de API: verifique se as chaves de API do provedor LLM estão configuradas. Para ambientes de produção, use segredos do Mosaic AI Gateway ou 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

Utilização

A execução do código a seguir gerará um rastreamento para o gráfico, 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 extensões dentro de um nó ou de uma ferramenta

Ao combinar o rastreamento automático com as APIs de rastreamento manual, pode-se adicionar subspans num nó ou numa ferramenta, para obter informações mais detalhadas para a etapa.

Vamos pegar o tutorial do Assistente de Código do LangGraph, por exemplo. Na verdade, o nó check_code consiste em duas validações diferentes para o código gerado. Você pode adicionar span para cada validação para ver qual validação foi executada. Para fazer isso, simplesmente criar intervalos manuais dentro da função de um 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 span para o check_code nó terá spans filho, que registram se cada validação falha ou não, com seus detalhes de exceção.

Desativar o rastreamento automático

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