Отслеживание LangGraph

LangGraph — это библиотека с открытым кодом для создания приложений с отслеживанием состояния и многофакторных приложений с помощью LLM, используемых для создания рабочих процессов агента и нескольких агентов.

MLflow Tracing предоставляет возможность автоматического отслеживания для LangGraph, как расширение интеграции с LangChain. Включив автоматическую трассировку для LangChain путем вызова функции mlflow.langchain.autolog, MLflow автоматически записывает выполнение графа в трассировку и сохраняет их в активном эксперименте MLflow.

import mlflow

mlflow.langchain.autolog()

Предпосылки

Чтобы использовать трассировку MLflow с LangGraph, необходимо установить MLflow и соответствующие пакеты LangGraph и LangChain (например, langgraph, langchain_core, langchain_openai).

Развитие

Для сред разработки установите полный пакет MLflow с дополнительными компонентами Databricks и пакетами LangGraph/LangChain:

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

Полный mlflow[databricks] пакет включает все функции для локальной разработки и экспериментирования в Databricks.

Производство

Для рабочих развертываний установите mlflow-tracing и пакеты LangGraph/LangChain:

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

Пакет mlflow-tracing оптимизирован для использования в рабочей среде.

Перед выполнением примеров необходимо настроить среду:

Для пользователей вне ноутбуков Databricks: задайте переменные среды Databricks:

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

Для пользователей записных книжек Databricks: эти учетные данные устанавливаются автоматически.

Ключи API. Убедитесь, что ключи API поставщика LLM настроены. Для рабочих сред используйте шлюз ИИ или секреты Databricks вместо жестко закодированных значений для безопасного управления ключами API.

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

Использование

При выполнении следующего кода будет создана трассировка для графа, как показано в приведенном выше видеоролике.

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

Добавление диапазонов в узле или инструменте

Объединяя автоматическую трассировку с API трассировки вручную, можно добавить дочерние диапазоны внутри узла или средства, чтобы получить более подробные сведения о шаге.

Например, давайте рассмотрим учебник помощника по коду LangGraph. Узел check_code фактически состоит из двух разных проверок для созданного кода. Вы можете добавить диапазон для каждой проверки, чтобы узнать, какая проверка была выполнена. Для этого внутри функции node вручную создайте спаны.

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

Таким образом, интервал для узла check_code будет иметь дочерние интервалы, которые фиксируют, завершается ли каждая проверка успешно или сбоем, с указанием деталей исключения.

Отключение автоматической трассировки

Автоматическая трассировка для LangGraph может быть отключена глобально путем вызова mlflow.langchain.autolog(disable=True) или mlflow.autolog(disable=True).