メインコンテンツまでスキップ

LangGraph のトレース

オートログによるLangChainトレース

LangGraph は、LLMを使用してステートフルなマルチアクターアプリケーションを構築するためのオープンソースライブラリであり、エージェントおよびマルチエージェントワークフローの作成に使用されます。

MLflow Tracing は、LangChain統合の拡張として、LangGraph の自動トレース機能を提供します。mlflow.langchain.autolog 関数を呼び出して LangChain の自動トレースを有効にすると、MLflow グラフの実行を自動的にトレースにキャプチャし、アクティブなエクスペリメントMLflowに記録します。

Python
import mlflow

mlflow.langchain.autolog()
注記

サーバレス コンピュート クラスターでは、自動ログは自動的に有効になりません。 この統合の自動トレースを有効にするには、明示的に mlflow.langchain.autolog() を呼び出す必要があります。

前提 条件​

LangGraphで MLflow Tracing を使用するには、 MLflow と関連するLangGraphおよび LangChain パッケージ( langgraph、 langchain_core、 langchain_openaiなど)をインストールする必要があります。

開発環境の場合は、Databricks エクストラと LangGraph/LangChain パッケージを含む完全な MLflow パッケージをインストールします。

Bash
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 3 は、LangChain 自動ログ統合に依存しているため、LangGraph で最適なトレース エクスペリエンスを実現することを強くお勧めします。

例を実行する前に、環境を構成する必要があります。

Databricks ノートブックの外部ユーザーの場合 : Databricks 環境変数を設定します。

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

Databricks ノートブック内のユーザーの場合 : これらの資格情報は自動的に設定されます。

API キー : LLM プロバイダーの API キーが構成されていることを確認します。本番運用環境では、安全な キー管理のために、ハードコードされた値の代わりにMosaic AI Gateway またはDatabricksシークレットAPI を使用します。

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

使い方​

次のコードを実行すると、上のビデオ クリップに示すように、グラフのトレースが生成されます。

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

ノードまたはツール内でのスパンの追加​

自動トレースと 手動トレース APIsを組み合わせることで、ノードやツール内に子スパンを追加して、ステップのより詳細な知見を得ることができます。

LangGraphの コードアシスタント チュートリアルを例にとってみましょう。check_codeノードは、実際には、生成されたコードに対する 2 つの異なる検証で構成されています。各検証にスパンを追加して、どの検証が実行されたかを確認することもできます。これを行うには、ノード関数内に手動スパンを作成するだけです。

Python
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 子スパン

警告

本番運用環境では、安全な キー管理のために、ハードコードされた値の代わりにMosaic AI Gateway またはDatabricksシークレットAPI を使用します。

自動トレースを無効にする​

LangGraph の自動トレースは、 mlflow.langchain.autolog(disable=True) または mlflow.autolog(disable=True)を呼び出すことでグローバルに無効にできます。

に最終更新