跳到主要内容

追踪 Anthropic

MLflow 追踪为 Anthropic LLM 提供自动追踪功能。通过调用 mlflow.anthropic.autolog() 函数为 Anthropic 启用自动追踪，MLflow 将捕获嵌套追踪并在调用 Anthropic Python SDK 时将它们记录到活动的 MLflow 实验中。

Anthropic Tracing via autolog

MLflow 追踪会自动捕获有关 Anthropic 调用的以下信息

  • 提示和完成响应
  • 延迟
  • 模型名称
  • 如果指定，还包括额外的元数据，如 temperature、max_tokens。
  • 如果响应中返回函数调用
  • Token 使用情况信息
  • 如果抛出任何异常
  • 以及更多...

开始使用​

1

安装依赖项

bash
pip install 'mlflow[genai]' anthropic
2

启动 MLflow 服务器

如果您有本地 Python 环境 >= 3.10，您可以使用 mlflow CLI 命令在本地启动 MLflow 服务器。

bash
mlflow server
3

启用跟踪并进行 API 调用

使用 mlflow.anthropic.autolog() 启用追踪，然后像平常一样进行 API 调用。

python
import anthropic
import mlflow

# Enable auto-tracing for Anthropic
mlflow.anthropic.autolog()

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

# Invoke the Anthropic model as usual.
# Make sure your API key is set via the ANTHROPIC_API_KEY environment variable.
client = anthropic.Anthropic()

message = client.messages.create(
    model="claude-sonnet-4-5-2025092",
    max_tokens=512,
    messages=[
        {"role": "user", "content": "Hello, Claude"},
    ],
)
4

在 MLflow UI 中查看跟踪

浏览 MLflow UI，地址为 https://:5000（或您的 MLflow 服务器 URL），您应该会看到 Anthropic API 调用的追踪记录。

Anthropic Tracing

支持的 API​

MLflow 支持以下 Anthropic API 的自动追踪

聊天补全函数调用流式传输异步图像批量
✅✅-✅ (*1)--

(*1) 异步支持已在 MLflow 2.21.0 中添加。

如需支持更多 API，请在 GitHub 上提交 功能请求。

异步​

自 MLflow 2.21.0 起，MLflow 追踪支持 Anthropic SDK 的异步 API。其用法与同步 API 相同。

python
import anthropic

# Enable trace logging
mlflow.anthropic.autolog()

client = anthropic.AsyncAnthropic()

response = await client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Hello, Claude"},
    ],
)

高级示例：工具调用代理​

MLflow 追踪会自动捕获 Anthropic 模型返回的工具调用响应。响应中的函数指令将在追踪 UI 中突出显示。此外，您可以使用 @mlflow.trace 装饰器注释工具函数，为工具执行创建跨度。

以下示例使用 Anthropic 工具调用和 MLflow 对 Anthropic 的追踪来实现一个简单的函数调用代理。该示例进一步使用了异步 Anthropic SDK，以便代理可以在不阻塞的情况下处理并发调用。

python
import json
import anthropic
import mlflow
import asyncio
from mlflow.entities import SpanType

client = anthropic.AsyncAnthropic()
model_name = "claude-sonnet-4-5-20250929"


# Define the tool function. Decorate it with `@mlflow.trace` to create a span for its execution.
@mlflow.trace(span_type=SpanType.TOOL)
async def get_weather(city: str) -> str:
    if city == "Tokyo":
        return "sunny"
    elif city == "Paris":
        return "rainy"
    return "unknown"


tools = [
    {
        "name": "get_weather",
        "description": "Returns the weather condition of a given city.",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    }
]

_tool_functions = {"get_weather": get_weather}


# Define a simple tool calling agent
@mlflow.trace(span_type=SpanType.AGENT)
async def run_tool_agent(question: str):
    messages = [{"role": "user", "content": question}]

    # Invoke the model with the given question and available tools
    ai_msg = await client.messages.create(
        model=model_name,
        messages=messages,
        tools=tools,
        max_tokens=2048,
    )
    messages.append({"role": "assistant", "content": ai_msg.content})

    # If the model requests tool call(s), invoke the function with the specified arguments
    tool_calls = [c for c in ai_msg.content if c.type == "tool_use"]
    for tool_call in tool_calls:
        if tool_func := _tool_functions.get(tool_call.name):
            tool_result = await tool_func(**tool_call.input)
        else:
            raise RuntimeError("An invalid tool is returned from the assistant!")

        messages.append(
            {
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": tool_call.id,
                        "content": tool_result,
                    }
                ],
            }
        )

    # Send the tool results to the model and get a new response
    response = await client.messages.create(
        model=model_name,
        messages=messages,
        max_tokens=2048,
    )

    return response.content[-1].text


# Run the tool calling agent
cities = ["Tokyo", "Paris", "Sydney"]
questions = [f"What's the weather like in {city} today?" for city in cities]
answers = await asyncio.gather(*(run_tool_agent(q) for q in questions))

for city, answer in zip(cities, answers):
    print(f"{city}: {answer}")

Token 用量​

MLflow >= 3.2.0 支持 Anthropic 的 Token 使用情况跟踪。每次 LLM 调用的 Token 使用情况将记录在 mlflow.chat.tokenUsage 属性中。整个追踪的总 Token 使用情况可在追踪信息对象的 token_usage 字段中找到。

python
import json
import anthropic
import mlflow

mlflow.anthropic.autolog()

client = anthropic.Anthropic()
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
)

# 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== Detailed 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: 8
  Output tokens: 12
  Total tokens: 20

== Detailed usage for each LLM call: ==
Messages.create:
  Input tokens: 8
  Output tokens: 12
  Total tokens: 20

支持的 API：​

以下 Anthropic API 支持 Token 使用情况跟踪

聊天补全函数调用流式传输异步图像批量
✅✅-✅ (*1)--

(*1) 异步支持已在 MLflow 2.21.0 中添加。

禁用自动跟踪​

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

后续步骤​

追踪用户反馈

记录用户对追踪的反馈，以跟踪用户满意度。

了解反馈 →

管理提示词

了解如何使用 MLflow 的提示词注册表来管理提示词。

管理提示词 →

评估追踪

使用 LLM 裁判评估追踪，以了解和改进您的 AI 应用程序的行为。

评估追踪 →