追踪 LangGraph🦜🕸️

LangGraph 是一个用于使用 LLM 构建有状态、多智能体应用的开源库，用于创建智能体和多智能体工作流。

MLflow Tracing（追踪）作为其 LangChain 集成的扩展，为 LangGraph 提供了自动追踪功能。通过调用 mlflow.langchain.autolog() 函数启用 LangChain 的自动追踪后，MLflow 会自动将图执行捕获为追踪并记录到活动的 MLflow Experiment 中。

import mlflow

mlflow.langchain.autolog()

提示

MLflow LangGraph 集成不仅限于追踪。MLflow 为 LangGraph 提供了完整的追踪体验，包括模型追踪、依赖管理和评估。请查阅 MLflow LangChain Flavor 了解更多信息！

使用方法

运行以下代码将为图生成一个追踪，如上面的视频片段所示。

from typing import Literal

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

# Enabling tracing for LangGraph (LangChain)
mlflow.langchain.autolog()

# Optional: Set a tracking URI and an experiment
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("LangGraph")


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

在节点或工具中添加 Span

通过将自动追踪与手动追踪 API 结合使用，您可以在节点或工具内部添加子 Span，以获取更详细的步骤洞察。

例如，以 LangGraph 的代码助手教程为例。check_code 节点实际上包含对生成代码的两个不同验证。您可能想为每个验证添加 Span，以查看执行了哪些验证。为此，只需在节点函数内部创建手动 Span 即可。

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 节点的 Span 将包含子 Span，记录每个验证是否失败以及其异常详情。

禁用自动追踪

可以通过调用 mlflow.langchain.autolog(disable=True) 或 mlflow.autolog(disable=True) 全局禁用 LangGraph 的自动追踪。