使用 Claude Agent SDK 和 MLflow 快速原型化和评估代理
构建智能体应用很复杂，但不必花费数周或数月的时间。在这篇博客中，我们将向您展示如何使用 Claude Agent SDK 原型设计一个智能体，然后使用 MLflow 对其进行检测和评估，从而在几小时内而不是几周或几个月内将想法转化为可衡量的结果。
为什么快速构建智能体很困难
从头开始构建一个智能体意味着在您可以测试核心想法之前需要大量的基础设施工作。除了编写智能体逻辑外，您还需要构建工作流编排、工具连接器和数据管道。这会延迟回答一个关键问题：您的智能体是否真正解决了您的核心业务问题？
最近，许多公司发布了自己的编码智能体，如 Claude Code、Codex 和 Gemini CLI。虽然这些主要是开发人员工具，为调试和重构等任务提供自动化体验，但它们包含了强大智能体框架的构建块。Anthropic 的 Claude Agent SDK 更进一步，以生产就绪的 SDK 形式提供了与 Claude Code 相同的底层框架，您可以直接将其集成到您的应用程序中。
为什么可观察性和评估很重要
无论您选择哪个框架，可观察性和评估对于确保您的智能体可用于生产仍然至关重要。如果没有跟踪，您的智能体就是一个难以调试和推理的黑匣子。如果没有系统的评估，您就无法衡量改进、发现回归或了解您的智能体在哪里成功和失败。这些功能必须从第一天起就内置进来。
在这篇博客中，我们将使用 Claude Agent SDK 原型设计一个旅行智能体，并使用 MLflow 的自动跟踪和评估套件来理解和改进其行为。
要求
claude-agent-sdk >= 0.1.0mlflow >= 3.5.0
智能体创建
想象一下您正在经营一家旅行社，您的团队花费数小时为客户搜索航班选项和旅行文件。一个 AI 智能体可以自动化这项研究，但您不想在验证想法之前花费数月时间来构建基础设施。让我们看看如何快速原型设计一个旅行智能体。
这是我们原型旅行智能体的结构
travel-agent-prototype/
├── run_agent.py # Driver code for running the agent
└── tools/ # CLI wrappers for your APIs
├── __init__.py
├── docs.py # Travel document search tool
├── flights.py # Flight search tool
└── server.py # MCP server for the tools
让我们逐一了解这些文件中的内容。
运行您的智能体
在接下来的部分中，我们定义智能体的驱动代码 (run_agent.py)。首先，我们将定义 ClaudeCodeOptions 对象，它指定了各种有用的选项，包括允许和不允许的工具、最大轮数、环境变量等等。
from pathlib import Path
from claude_agent_sdk import ClaudeAgentOptions
from tools import travel_mcp_server
TRAVEL_AGENT_OPTIONS = ClaudeAgentOptions(
mcp_servers={"travel": travel_mcp_server},
allowed_tools=[
"mcp__travel__search_flights",
"mcp__travel__search_travel_docs",
],
system_prompt=TRAVEL_AGENT_SYSTEM_PROMPT,
cwd=str(Path.cwd()),
max_turns=10,
)
您可能会注意到我们还没有定义系统提示。您的系统提示是您定义智能体行为以及它有权访问哪些工具以及智能体应如何使用它们的地方。您可以让 Claude 编写此内容，但您应该谨慎对待其中的内容，因为它决定了您智能体的行为。
TRAVEL_AGENT_SYSTEM_PROMPT = """
# Travel Agent Assistant
You are a travel agent assistant with access to flight search and travel documentation tools.
## Execution Flow
1. Determine user intent: flight search or travel information lookup
2. For flight searches, use the `search_flights` tool with origin, destination, dates, and preferences
3. For travel information, use the `search_travel_docs` tool to extract relevant information about destinations, activities, logistics
4. Combine both capabilities when planning complete trips
## Success Criteria
Present the user with clear, relevant, specific, and actionable advice. Cover all aspects of their travel (e.g., logistics, activities, local insights, etc.) and present them with tradeoffs of any choices they need to make.
"""
设置好选项后，我们可以使用以下内容运行我们的智能体
from claude_agent_sdk import ClaudeSDKClient
async def run_travel_agent(query: str) -> str:
messages = []
async with ClaudeSDKClient(options=TRAVEL_AGENT_OPTIONS) as client:
await client.query(query)
async for message in client.receive_response():
messages.append(message)
return messages[-1].result # Return the final output
我应该如何实现我的工具？
在上面的示例中，我们用 Python 实现了工具并创建了一个进程内 MCP 服务器。以下是 tools/server.py 的代码
from claude_agent_sdk import create_sdk_mcp_server
from .docs import search_travel_docs
from .flights import search_flights
travel_mcp_server = create_sdk_mcp_server(
name="travel",
version="1.0.0",
tools=[search_flights, search_travel_docs],
)
作为示例，以下是位于 tools/docs.py 中的 search_travel_docs 的签名
from claude_agent_sdk import tool
@tool(
"search_travel_docs",
"Search travel documentation and guides",
{
"query": str,
"num_results": int,
},
)
async def search_travel_docs(args):
pass
您可以在这些函数中放置任何内容！在此实例中，我们的搜索航班工具连接到第三方航班搜索 API，旅行文档工具利用远程向量存储。
您也不局限于使用 MCP 服务器——在 CLI 中实现工具是限制智能体需要处理的上下文量的好方法，直到它需要该工具为止。
跟踪智能体
现在我们有了一个强大的智能体，我们可能想了解幕后发生了什么。MLflow 可以自动跟踪 Claude Code，记录 Claude Code 所采取的每一步的输入、输出和元数据。这对于查明错误和意外行为的来源至关重要。
我们可以通过一个简单的单行代码启用自动跟踪
import mlflow
@mlflow.anthropic.autolog()
# Define ClaudeSDKClient after this...
设置自动跟踪后，对 claude 的调用应导致新的跟踪流向您的实验。您可以使用 mlflow.set_experiment 和 mlflow.set_tracking_uri 分别配置您的实验或跟踪存储。
在我们查看一些跟踪时，我们注意到我们的智能体倾向于进行冗余的检索步骤。
在下一节中，我们将讨论如何使用 MLflow 评估来自信地迭代我们的智能体以解决这些问题。
运行评估
构建一个智能体系统只是第一步。要确保您的旅行智能体在生产中可靠运行，您需要系统评估。没有结构化测试，您就无法衡量改进、发现回归或了解您的智能体在哪里成功和失败。现在我们已经为 Claude Agent SDK 启用了跟踪，使用 mlflow.genai.evaluate 运行评估非常简单
mlflow.genai.evaluate(
predict_fn=...,
data=...,
scorers=...,
)
我们将在接下来的部分中讨论这些参数中包含的内容。
predict_fn 中包含什么？
predict 函数是运行您的智能体并生成要评估的跟踪的代码。以下是我们旅行智能体工作示例的 predict_fn
def run_travel_agent_with_timeout(query: str, timeout: int = 300) -> str:
async def run_with_timeout():
return await asyncio.wait_for(run_travel_agent(query), timeout=timeout)
return asyncio.run(run_with_timeout())
data 中包含什么？
data 参数接受用于评估的输入集和可选的期望值。将这些视为单元测试——该数据集中的每一行都是您希望在智能体的每次迭代中测试的案例。在我们的例子中，它可能看起来像
data = [
{
"inputs": {
"query": "What are some essential winter travel tips for New York City?"
},
"expectations": {
"expected_topics": "The response should cover topics like clothing and transportation"
}
},
{
"inputs": {
"query": "Where is a better place to visit in the summer, Tokyo or Paris?"
},
"expectations": {
"expected_topics": "The response should cover topics like weather, activities, and vibes"
}
},
{
"inputs": {
"query": "I only have a day in Athens, what should I do?"
},
"expectations": {
"expected_topics": "The response should cover topics like food, attractions, and activities."
}
},
]
我们经常希望持久化我们的测试用例，这可以通过 MLflow 评估数据集完成
from mlflow.genai.datasets import create_dataset
dataset = create_dataset(
name="travel_agent_test_cases",
experiment_id=["YOUR_EXPERIMENT_ID"],
tags={"complexity": "basic", "priority": "critical"},
)
dataset.merge_records(data)
scorers 中包含什么？
Scorers 是定义智能体评估标准的统一接口。您应该花大部分时间思考如何定义这些，因为它们定义了您如何看待智能体的质量。对于我们上面的场景，我们可能想定义一个检索冗余判断器，它可以查看跟踪并确定是否进行了冗余的检索调用
from mlflow.genai.judges import make_judge
redundancy_judge = make_judge(
name="retrieval_redundancy",
model="openai:/gpt-4o",
instructions=(
"Analyze {{ trace }} to check if there are redundant retrieval calls. "
"Look at the source IDs returned from the retrieval tool. "
"If multiple retrieval calls have the same source IDs, there is likely a redundancy. "
"Return 'pass' if there are no redundant calls, 'fail' if there are redundant calls."
),
)
我们还可以定义操作指标，例如智能体使用的检索调用次数
from mlflow.genai import scorer
@scorer
def num_retrieval_calls(trace):
return sum(
[
1 if span.span_type == "TOOL" and "search_travel_docs" in span.name else 0
for span in trace.data.spans
]
)
最后，随着我们的迭代，我们想确保智能体的质量不会下降。我们将引入一个通用的综合性评分器来检查这一点
from mlflow.genai.judges import make_judge
comprehensive_judge = make_judge(
name="comprehensive",
model="openai:/gpt-4o-mini",
instructions="""
Evaluate if the outputs comprehensively covers all relevant aspects for the query in the inputs, including the expected topics. Does the response address the full scope of what a traveler would need to know? Return 'pass' if the output is comprehensive or 'fail' if not.
Outputs: {{outputs}}
Expected Topics: {{expectations}}
"""
)
整合所有内容
现在我们有了评估的所有组件，我们可以将它们整合在一起，如下所示
from mlflow.genai import evaluate
evaluate(
data=dataset,
predict_fn=run_travel_agent_with_timeout,
scorers=[
redundancy_judge,
num_retrieval_calls,
comprehensive_judge,
],
)
这将对您的跟踪生成评估结果，以及一个新运行以将这些结果存储在一个地方
回到我们关于冗余检索调用的最初场景——我们可以看到我们的检索冗余判断器在许多测试用例中都失败了。为解决此问题，我们在系统提示中添加了一行以强调检索调用的效率
TRAVEL_AGENT_SYSTEM_PROMPT = """
...
3. For travel information, use the `search_travel_docs` tool to extract relevant information about destinations, activities, logistics
- When using this tool, please ensure you only query about a topic once. All queries should be completely distinct!
...
"""
重新运行我们的评估会产生以下结果
现在我们可以看到检索冗余指标在所有示例中都通过了，并且在我们的更改后，平均检索调用次数从 3.2 次下降到 1.8 次！正如综合性评分器所示，我们也保持了智能体的质量！
结论
原型设计智能体不需要数月的基础设施工作。通过利用 Claude Agent SDK 和 MLflow 的可观察性工具，我们以传统方法所需时间的一小部分构建并验证了一个旅行智能体原型。
关键要点
- 快速原型设计：使用 Claude Agent SDK 等现有智能体框架快速验证您的想法
- 自动跟踪：MLflow 的
@mlflow.anthropic.autolog()以零检测捕获智能体执行的每一个操作 - 自信迭代：使用
mlflow.genai.evaluate()通过自定义评分器和判断器客观地衡量质量改进
从第一天起就内置可观察性和评估，您可以自信地从原型转向生产就绪的智能体，确信您的系统能够可靠地运行。后续步骤
- 探索 Claude Agent SDK 文档 以了解高级智能体功能
- 了解 Claude Code 自动跟踪 以自动跟踪 Claude Agent SDK 和 Claude Code CLI
- 阅读有关 MLflow 的评估和监控框架 的信息，了解完整的评估生态系统