Claude Agent SDK 结构化输出 - JSON Schema / Zod / Pydantic 验证
使用 JSON Schema、Zod 或 Pydantic 从代理工作流返回经过验证的 JSON,在多轮工具使用后获得类型安全的结构化数据。(原文为英文,已翻译)
结构化输出允许你定义希望从代理返回的数据的精确形状。代理可以使用完成任务所需的任何工具,你最终仍然会获得与你的 schema 匹配的验证 JSON。为你需要的结构定义 JSON Schema,SDK 会根据该 schema 验证输出,并在不匹配时重新提示。如果在重试限制内验证未成功,结果会是错误而不是结构化数据;参见错误处理。
要获得完整的类型安全,请使用 Zod(TypeScript)或 Pydantic(Python)来定义你的 schema,并获得强类型的对象。
为什么使用结构化输出?
代理默认返回自由格式的文本,这对于聊天来说有效,但当你需要以编程方式使用输出时则不然。结构化输出为你提供可以直接传递给应用程序逻辑、数据库或 UI 组件的类型化数据。
考虑一个食谱应用程序,其中代理在网上搜索并带回食谱。如果没有结构化输出,你将获得需要自己解析的自由格式文本。使用结构化输出后,你定义所需的形状并获得可以直接在你的应用中使用的类型化数据。
没有结构化输出
Here's a classic chocolate chip cookie recipe!
**Chocolate Chip Cookies**
Prep time: 15 minutes | Cook time: 10 minutes
Ingredients:
- 2 1/4 cups all-purpose flour
- 1 cup butter, softened
...
要在你的应用中使用它,你需要解析出标题、将 “15 minutes” 转换为数字、将原料与说明分离,并处理跨响应的不一致格式。
使用结构化输出
{
"name": "Chocolate Chip Cookies",
"prep_time_minutes": 15,
"cook_time_minutes": 10,
"ingredients": [
{ "item": "all-purpose flour", "amount": 2.25, "unit": "cups" },
{ "item": "butter, softened", "amount": 1, "unit": "cup" }
// ...
],
"steps": ["Preheat oven to 375°F", "Cream butter and sugar" /* ... */]
}
可以直接在你的 UI 中使用的类型化数据。
快速开始
要使用结构化输出,定义一个描述你希望数据形状的 JSON Schema,然后通过 outputFormat 选项(TypeScript)或 output_format 选项(Python)将其传递给 query()。当代理完成时,结果消息包含一个 structured_output 字段,其中包含与你的 schema 匹配的验证数据。
下面的示例要求代理研究 Anthropic 并以结构化输出返回公司名称、成立年份和总部。
TypeScript
import { query } from "@anthropic-ai/claude-agent-sdk";
// Define the shape of data you want back
const schema = {
type: "object",
properties: {
company_name: { type: "string" },
founded_year: { type: "number" },
headquarters: { type: "string" }
},
required: ["company_name"]
};
for await (const message of query({
prompt: "Research Anthropic and provide key company information",
options: {
outputFormat: {
type: "json_schema",
schema: schema
}
}
})) {
// The result message contains structured_output with validated data
if (message.type === "result" && message.subtype === "success" && message.structured_output) {
console.log(message.structured_output);
// { company_name: "Anthropic", founded_year: 2021, headquarters: "San Francisco, CA" }
}
}
Python
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
# Define the shape of data you want back
schema = {
"type": "object",
"properties": {
"company_name": {"type": "string"},
"founded_year": {"type": "number"},
"headquarters": {"type": "string"},
},
"required": ["company_name"],
}
async def main():
async for message in query(
prompt="Research Anthropic and provide key company information",
options=ClaudeAgentOptions(
output_format={"type": "json_schema", "schema": schema}
),
):
# The result message contains structured_output with validated data
if isinstance(message, ResultMessage) and message.structured_output:
print(message.structured_output)
# {'company_name': 'Anthropic', 'founded_year': 2021, 'headquarters': 'San Francisco, CA'}
asyncio.run(main())
使用 Zod 和 Pydantic 的类型安全 schema
你可以使用 Zod(TypeScript)或 Pydantic(Python)来定义你的 schema,而不是手写 JSON Schema。这些库为你生成 JSON Schema,并让你将响应解析为完全类型化的对象,可以在整个代码库中使用,具有自动完成和类型检查。
下面的示例定义了一个功能实现计划的 schema,包含摘要、步骤列表(每个步骤都有复杂性级别)和潜在风险。代理规划该功能并返回一个类型化的 FeaturePlan 对象。然后你可以访问 plan.summary 等属性,并使用完整的类型安全迭代 plan.steps。
TypeScript
import { z } from "zod";
import { query } from "@anthropic-ai/claude-agent-sdk";
// Define schema with Zod
const FeaturePlan = z.object({
feature_name: z.string(),
summary: z.string(),
steps: z.array(
z.object({
step_number: z.number(),
description: z.string(),
estimated_complexity: z.enum(["low", "medium", "high"])
})
),
risks: z.array(z.string())
});
type FeaturePlan = z.infer<typeof FeaturePlan>;
// Convert to JSON Schema
const schema = z.toJSONSchema(FeaturePlan);
// Use in query
for await (const message of query({
prompt:
"Plan how to add dark mode support to a React app. Break it into implementation steps.",
options: {
outputFormat: {
type: "json_schema",
schema: schema
}
}
})) {
if (message.type === "result" && message.subtype === "success" && message.structured_output) {
// Validate and get fully typed result
const parsed = FeaturePlan.safeParse(message.structured_output);
if (parsed.success) {
const plan: FeaturePlan = parsed.data;
console.log(`Feature: ${plan.feature_name}`);
console.log(`Summary: ${plan.summary}`);
plan.steps.forEach((step) => {
console.log(`${step.step_number}. [${step.estimated_complexity}] ${step.description}`);
});
}
}
}
Python
import asyncio
from pydantic import BaseModel
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
class Step(BaseModel):
step_number: int
description: str
estimated_complexity: str # 'low', 'medium', 'high'
class FeaturePlan(BaseModel):
feature_name: str
summary: str
steps: list[Step]
risks: list[str]
async def main():
async for message in query(
prompt="Plan how to add dark mode support to a React app. Break it into implementation steps.",
options=ClaudeAgentOptions(
output_format={
"type": "json_schema",
"schema": FeaturePlan.model_json_schema(),
}
),
):
if isinstance(message, ResultMessage) and message.structured_output:
# Validate and get fully typed result
plan = FeaturePlan.model_validate(message.structured_output)
print(f"Feature: {plan.feature_name}")
print(f"Summary: {plan.summary}")
for step in plan.steps:
print(
f"{step.step_number}. [{step.estimated_complexity}] {step.description}"
)
asyncio.run(main())
好处:
- 完整的类型推断(TypeScript)和类型提示(Python)
- 使用
safeParse()或model_validate()进行运行时验证 - 更好的错误消息
- 可组合、可重用的 schema
输出格式配置
outputFormat(TypeScript)或 output_format(Python)选项接受一个具有以下内容的对象:
type:设置为"json_schema"以使用结构化输出schema:定义你输出结构的 JSON Schema 对象。你可以使用z.toJSONSchema()从 Zod schema 生成,或使用.model_json_schema()从 Pydantic 模型生成
SDK 支持标准 JSON Schema 功能,包括所有基本类型(object、array、string、number、boolean、null)、enum、const、required、嵌套对象和 $ref 定义。有关支持的功能和限制的完整列表,请参阅 JSON Schema 限制。
示例:TODO 跟踪代理
此示例演示了结构化输出如何与多步工具使用一起工作。代理需要在代码库中查找 TODO 注释,然后查找每个注释的 git blame 信息。它自主决定使用哪些工具(Grep 进行搜索、Bash 运行 git 命令)并将结果合并到单个结构化响应中。
该 schema 包括可选字段(author 和 date),因为 git blame 信息可能并非对所有文件都可用。代理填写它可以找到的内容并省略其余内容。
TypeScript
import { query } from "@anthropic-ai/claude-agent-sdk";
// Define structure for TODO extraction
const todoSchema = {
type: "object",
properties: {
todos: {
type: "array",
items: {
type: "object",
properties: {
text: { type: "string" },
file: { type: "string" },
line: { type: "number" },
author: { type: "string" },
date: { type: "string" }
},
required: ["text", "file", "line"]
}
},
total_count: { type: "number" }
},
required: ["todos", "total_count"]
};
// Agent uses Grep to find TODOs, Bash to get git blame info
for await (const message of query({
prompt: "Find all TODO comments in this codebase and identify who added them",
options: {
outputFormat: {
type: "json_schema",
schema: todoSchema
}
}
})) {
if (message.type === "result" && message.subtype === "success" && message.structured_output) {
const data = message.structured_output as { total_count: number; todos: Array<{ file: string; line: number; text: string; author?: string; date?: string }> };
console.log(`Found ${data.total_count} TODOs`);
data.todos.forEach((todo) => {
console.log(`${todo.file}:${todo.line} - ${todo.text}`);
if (todo.author) {
console.log(` Added by ${todo.author} on ${todo.date}`);
}
});
}
}
Python
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
# Define structure for TODO extraction
todo_schema = {
"type": "object",
"properties": {
"todos": {
"type": "array",
"items": {
"type": "object",
"properties": {
"text": {"type": "string"},
"file": {"type": "string"},
"line": {"type": "number"},
"author": {"type": "string"},
"date": {"type": "string"},
},
"required": ["text", "file", "line"],
},
},
"total_count": {"type": "number"},
},
"required": ["todos", "total_count"],
}
async def main():
# Agent uses Grep to find TODOs, Bash to get git blame info
async for message in query(
prompt="Find all TODO comments in this codebase and identify who added them",
options=ClaudeAgentOptions(
output_format={"type": "json_schema", "schema": todo_schema}
),
):
if isinstance(message, ResultMessage) and message.structured_output:
data = message.structured_output
print(f"Found {data['total_count']} TODOs")
for todo in data["todos"]:
print(f"{todo['file']}:{todo['line']} - {todo['text']}")
if "author" in todo:
print(f" Added by {todo['author']} on {todo['date']}")
asyncio.run(main())
错误处理
当代理无法生成与你的 schema 匹配的有效 JSON 时,结构化输出生成可能失败。这通常发生在 schema 对任务来说太复杂、任务本身模糊不清,或者代理在尝试修复验证错误时达到重试限制时。
发生错误时,结果消息有一个 subtype 指示出错原因:
| Subtype | 含义 |
|---|---|
success | 输出已成功生成并验证 |
error_max_structured_output_retries | 多次尝试后代理无法生成有效输出 |
下面的示例检查 subtype 字段以确定输出是成功生成还是需要处理失败:
TypeScript
for await (const msg of query({
prompt: "Extract contact info from the document",
options: {
outputFormat: {
type: "json_schema",
schema: contactSchema
}
}
})) {
if (msg.type === "result") {
if (msg.subtype === "success" && msg.structured_output) {
// Use the validated output
console.log(msg.structured_output);
} else if (msg.subtype === "error_max_structured_output_retries") {
// Handle the failure - retry with simpler prompt, fall back to unstructured, etc.
console.error("Could not produce valid output");
}
}
}
Python
async for message in query(
prompt="Extract contact info from the document",
options=ClaudeAgentOptions(
output_format={"type": "json_schema", "schema": contact_schema}
),
):
if isinstance(message, ResultMessage):
if message.subtype == "success" and message.structured_output:
# Use the validated output
print(message.structured_output)
elif message.subtype == "error_max_structured_output_retries":
# Handle the failure
print("Could not produce valid output")
避免错误的提示:
- 保持 schema 简洁。 具有许多必填字段的深度嵌套 schema 更难满足。从简单开始,根据需要增加复杂性。
- 将 schema 与任务匹配。 如果任务可能没有你的 schema 所需的所有信息,请将这些字段设为可选。
- 使用清晰的提示。 模糊的提示使代理更难知道要生成什么输出。
相关资源
- JSON Schema 文档:学习 JSON Schema 语法,用于定义具有嵌套对象、数组、枚举和验证约束的复杂 schema
- API 结构化输出:在没有工具使用的单轮请求中直接使用 Claude API 的结构化输出
- 自定义工具:在返回结构化输出之前为你的代理提供执行期间要调用的自定义工具
本文翻译自 Anthropic Claude Code 官方文档,最近一次同步:2025-05-01。