跳到主要内容

Tracing LangGraph🦜🕸️

LangGraph 是一个开源库，用于使用 LLM 构建有状态、多参与者的应用程序，可用于创建代理和多代理工作流。

MLflow Tracing 为 LangGraph 提供了自动跟踪功能，作为其 LangChain 集成的扩展。通过调用 mlflow.langchain.autolog() 函数为 LangChain 启用自动跟踪，MLflow 将自动捕获图执行到跟踪中，并将其记录到活动的 MLflow 实验中。在 TypeScript 中，您可以将 MLflow LangChain 回调传递给 callbacks 选项。

python
import mlflow

mlflow.langchain.autolog()

开始使用​

MLflow 在 Python 和 TypeScript/JavaScript 中都支持 LangGraph 的跟踪。请选择下面的相应选项卡开始。

1. 启动 MLflow​

如果您还没有 MLflow 服务器，请按照自托管指南启动 MLflow 服务器。

2. 安装依赖项​

bash
pip install langgraph langchain-openai 'mlflow[genai]'

3. 启用跟踪​

python
import mlflow

# Calling autolog for LangChain will enable trace logging.
mlflow.langchain.autolog()

# Optional: Set a tracking URI and an experiment
mlflow.set_experiment("LangChain")
mlflow.set_tracking_uri("https://:5000")

4. 定义 LangGraph 代理并调用它​

python
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("https://: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?"}]}
)

5. 在 MLflow UI 中查看跟踪​

访问 https://:5000（或您的自定义 MLflow 跟踪服务器 URL）以在 MLflow UI 中查看跟踪。

令牌使用跟踪​

MLflow >= 3.1.0 支持 LangGraph 的令牌使用情况跟踪。图调用期间每次 LLM 调用的令牌使用情况将记录在 mlflow.chat.tokenUsage span 属性中，整个跟踪的总使用情况将记录在 mlflow.trace.tokenUsage 元数据字段中。

python
import json
import mlflow

mlflow.langchain.autolog()

# Execute the agent graph defined in the previous example
graph.invoke({"messages": [{"role": "user", "content": "what is the weather in sf?"}]})

# Get the trace object just created
last_trace_id = mlflow.get_last_active_trace_id()
trace = mlflow.get_trace(trace_id=last_trace_id)

# Print the token usage
total_usage = trace.info.token_usage
print("== Total token usage: ==")
print(f"  Input tokens: {total_usage['input_tokens']}")
print(f"  Output tokens: {total_usage['output_tokens']}")
print(f"  Total tokens: {total_usage['total_tokens']}")

# Print the token usage for each LLM call
print("\n== Token usage for each LLM call: ==")
for span in trace.data.spans:
    if usage := span.get_attribute("mlflow.chat.tokenUsage"):
        print(f"{span.name}:")
        print(f"  Input tokens: {usage['input_tokens']}")
        print(f"  Output tokens: {usage['output_tokens']}")
        print(f"  Total tokens: {usage['total_tokens']}")
bash
== Total token usage: ==
  Input tokens: 149
  Output tokens: 135
  Total tokens: 284

== Token usage for each LLM call: ==
ChatOpenAI_1:
  Input tokens: 58
  Output tokens: 87
  Total tokens: 145
ChatOpenAI_2:
  Input tokens: 91
  Output tokens: 48
  Total tokens: 139

在节点或工具中添加 span​

通过将自动跟踪与手动跟踪 API 结合使用，您可以在节点或工具内部添加子 span，以获得有关该步骤的更详细的见解。

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

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 节点的 span 将包含子 span，这些子 span 会记录每次验证是否失败，以及它们的异常详细信息。

LangGraph Child Span

异步上下文传播

当在 LangGraph 节点或工具中的手动 @mlflow.trace 装饰器中使用 ainvoke() 等异步方法时，请启用内联跟踪器执行以确保正确的上下文传播

python
mlflow.langchain.autolog(run_tracer_inline=True)

这确保手动跟踪的 span 正确嵌套在自动日志跟踪层次结构中。如果没有此设置，在异步场景中，手动 span 可能显示为单独的跟踪。

警告

当启用 run_tracer_inline=True 时，避免在同一异步函数内连续调用多个图调用，因为这可能会导致跟踪意外合并。如果需要进行多次顺序调用，请执行以下任一操作：

  • 将每次调用包装在单独的异步任务中
  • 如果您不需要手动跟踪集成，请使用默认的 run_tracer_inline=False

线程 ID 跟踪​

从 MLflow 3.6 开始，MLflow 将自动记录跟踪的线程（会话）ID，并允许您在 UI 中将一组跟踪视为一个会话。要启用此功能，您需要在调用图时在配置中传递 thread_id。

python
graph.invoke(inputs, {"configurable": {"thread_id": "1"}})

线程 ID 将记录在跟踪元数据中，并在 MLflow 跟踪 UI 中显示。

LangGraph Thread ID

通过导航到侧边栏的“会话”选项卡，您可以查看会话中的所有跟踪。

LangGraph Session Page

禁用自动跟踪​

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