使用 Instructor 从 Anthropic Claude 提取结构化输出：完整实战指南-开发者社区

使用 Instructor 从 Anthropic Claude 提取结构化输出：完整实战指南

【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor

本文是基于开源仓库 instructor 的 Anthropic 集成实战指南。全文围绕docs/integrations/anthropic.md展开，讲解如何用 Instructor 的from_provider一行代码接入 Claude 模型，完成从基础工具调用、异步、并行工具、多模态、流式输出到 Prompt Caching 与 Extended Thinking 的完整结构化输出管线。读完本文，你将掌握 Claude + Pydantic 类型安全输出的全部核心用法，并了解其底层模式处理器（Mode Handler）的实现原理。

快速开始：安装并创建 Claude 客户端

使用 Claude 进行结构化输出的前提是安装带 Anthropic 扩展的 Instructor 包：

pip install "instructor[anthropic]"

安装完成后，通过from_provider方法即可快速创建已打补丁的客户端。传入"anthropic/claude-sonnet-5"这种provider/model格式的模型字符串，Instructor 会根据 provider 前缀自动完成客户端初始化与模式选择（instructor/v2/auto_client.py中维护了ALIAS_TO_PROVIDER别名表与supported_providers列表）。其底层工厂函数是from_anthropic，它接受anthropic.Anthropic、AsyncAnthropic、AnthropicBedrock、AnthropicVertex等同步/异步客户端，并做了模式规范化（ANTHROPIC_TOOLS → TOOLS）与模式注册校验。

# Standard library imports from typing import List # Third-party imports import instructor from pydantic import BaseModel, Field # Set up environment (typically handled before script execution) # os.environ["ANTHROPIC_API_KEY"] = "your-api-key" # Uncomment and replace with your API key if not set # Define your models with proper type annotations class Properties(BaseModel): """Model representing a key-value property.""" name: str = Field(description="The name of the property") value: str = Field(description="The value of the property") class User(BaseModel): """Model representing a user with properties.""" name: str = Field(description="The user's full name") age: int = Field(description="The user's age in years") properties: List[Properties] = Field(description="List of user properties") client = instructor.from_provider( "anthropic/claude-sonnet-5", mode=instructor.Mode.TOOLS ) try: # Extract structured data user_response = client.create( max_tokens=1024, messages=[ { "role": "system", "content": "Extract structured information based on the user's request.", }, { "role": "user", "content": "Create a user for a model with a name, age, and properties.", }, ], response_model=User, ) # Print the result as formatted JSON print(user_response.model_dump_json(indent=2)) # Expected output: # { # "name": "John Doe", # "age": 35, # "properties": [ # { # "name": "City", # "value": "New York" # }, # { # "name": "Occupation", # "value": "Software Engineer" # } # ] # } except instructor.exceptions.InstructorError as e: print(f"Validation error: {e}") except Exception as e: print(f"Unexpected error: {e}")

从源码结构看，Mode.TOOLS对应的AnthropicToolsHandler在prepare_request阶段会做三件事：提取并合并 system 消息（extract_system_messages/combine_system_messages）、将 Pydantic 模型转为 Anthropic 工具 schema（tools=[...]）、根据场景设置tool_choice。其中 schema 生成由generate_anthropic_schema完成，输出name、description、input_schema三个字段，并使用functools.lru_cache(maxsize=256)缓存，避免重复计算相同模型。

异步调用

Instructor 对异步场景开箱即用，只需在from_provider中传入async_client=True：

import asyncio async_client = instructor.from_provider( "anthropic/claude-sonnet-5", async_client=True, mode=instructor.Mode.TOOLS, ) async def extract_user(): return await async_client.create( messages=[{"role": "user", "content": "Extract: Jason is 25 years old"}], response_model=User, ) user = asyncio.run(extract_user()) print(user)

from_anthropic会依据传入客户端的类型自动返回同步Instructor或AsyncInstructor实例（见 instructor/v2/providers/anthropic/client.py），因此无需手工区分 API。

并行工具调用：Iterable[Union[...]] 自动检测

当你的响应模型是Iterable[Union[Model1, Model2, ...]]时，Instructor 会自动切换到并行工具模式——无需显式指定Mode.PARALLEL_TOOLS，只要使用Mode.TOOLS（或保持默认）即可：

自动将tool_choice设为"auto"（并行调用的必要条件）；
为联合类型的每一个成员生成工具 schema；
返回一个生成器，逐一产出每个工具调用的结果；
每个产出项都会用对应的 Pydantic 模型做校验。

from typing import Iterable, Literal from pydantic import BaseModel import instructor class Weather(BaseModel): location: str units: Literal["imperial", "metric"] class GoogleSearch(BaseModel): query: str # No need to specify Mode.PARALLEL_TOOLS - it's auto-detected! client = instructor.from_provider( "anthropic/claude-sonnet-5", mode=instructor.Mode.TOOLS, # or just omit and use default ) results = client.create( messages=[ {"role": "system", "content": "You must always use tools"}, { "role": "user", "content": "What is the weather in toronto and dallas and who won the super bowl?", }, ], response_model=Iterable[Weather | GoogleSearch], # Auto-detects parallel mode ) for item in results: print(item)

源码中，AnthropicToolsHandler.prepare_request通过get_origin(response_model) is typing.Iterable判断并行场景（注意：流式模式下Iterable[T]会被当作流式而非并行工具，见 handlers.py）。随后handle_parallel_model为联合类型每个成员生成 Anthropic 工具 schema。响应解析时，handler 遍历消息中的tool_use块，按工具名查注册表，并用model_validate_json校验（handlers.py）。

多模态：图片与 PDF

Instructor 提供统一的、与提供商无关的多模态接口，支持从 URL、本地文件或 base64 字符串加载媒体，并自动完成各提供商特有的格式转换，保证代码整洁且面向未来兼容。仓库中的示例素材包括一张蓝莓植株图片 tests/assets/image.jpg 和一份包含假发票的 PDF tests/assets/invoice.pdf。

分析图片

下面的示例用Image.from_url加载图片；同时也支持from_path（本地文件）和from_base64（base64 字符串），以及自动嗅探来源的autodetect方法：

from instructor.processing.multimodal import Image from pydantic import BaseModel, Field import instructor class ImageDescription(BaseModel): objects: list[str] = Field(..., description="The objects in the image") scene: str = Field(..., description="The scene of the image") colors: list[str] = Field(..., description="The colors in the image") client = instructor.from_provider("anthropic/claude-sonnet-5") # Multiple ways to load an image: response = client.create( response_model=ImageDescription, max_tokens=1000, messages=[ { "role": "user", "content": [ "What is in this image?", # Option 1: Local file (samples included in this repo) Image.from_path("tests/assets/image.jpg"), # Option 2: Direct URL # Image.from_url("https://.../image.jpg") # Option 3: Base64 string # Image.from_base64("base64_encoded_string_here") # Option 4: Autodetect # Image.autodetect(<url|path|base64>) ], }, ], ) print(response) # Example output: # ImageDescription( # objects=['blueberries', 'leaves'], # scene='A blueberry bush with clusters of ripe blueberries and some unripe ones against a cloudy sky', # colors=['green', 'blue', 'purple', 'white'] # )

Image类定义在 instructor/v2/core/multimodal.py，source字段同时接受 URL、路径、base64 与原始字节。instructor.processing.multimodal作为兼容导出层，把 v2 的实现重新导出（multimodal.py）。

提取 PDF 内容

PDF 的用法与图片完全对称，同样是from_url/from_path/from_base64/autodetect四选一：

from instructor.processing.multimodal import PDF from pydantic import BaseModel import instructor class Receipt(BaseModel): total: int items: list[str] client = instructor.from_provider("anthropic/claude-sonnet-5") # Multiple ways to load an PDF: response = client.create( response_model=Receipt, max_tokens=1000, messages=[ { "role": "user", "content": [ "Extract out the total and line items from the invoice", # Option 1: Local file (samples included in this repo) PDF.from_path("tests/assets/invoice.pdf"), # Option 2: Direct URL # PDF.from_url("https://.../invoice.pdf"), # Option 3: Base64 string # PDF.from_base64("base64_encoded_string_here") # Option 4: Autodetect # PDF.autodetect(<url|path|base64>) ], }, ], ) print(response) #> Receipt(total=220, items=['English Tea', 'Tofu'])

如果你想在多次请求间复用（缓存）同一份 PDF，可以使用带缓存控制的PDFWithCacheControl类（对应源码中的 PDFWithCacheControl），结合create_with_completion获取原始 completion 以核对缓存命中情况：

from instructor.processing.multimodal import PDFWithCacheControl from pydantic import BaseModel import instructor class Receipt(BaseModel): total: int items: list[str] client = instructor.from_provider("anthropic/claude-sonnet-5") response, completion = client.create_with_completion( response_model=Receipt, max_tokens=1000, messages=[ { "role": "user", "content": [ "Extract out the total and line items from the invoice", PDFWithCacheControl.from_path("tests/assets/invoice.pdf"), ], }, ], ) assert ( completion.usage.cache_creation_input_tokens > 0 or completion.usage.cache_read_input_tokens > 0 ) print(response) #> Receipt(total=220, items=['English Tea', 'Tofu'])

PDFWithCacheControl底层会通过pdf_with_cache_control_to_anthropic（位于 instructor/v2/providers/anthropic/multimodal.py）把 PDF 转成带cache_control的 Anthropic 消息块。多模态的完整讲解见 docs/concepts/multimodal.md。

流式输出

Instructor 提供两种流式手段：

Iterables：适合流式返回同类型对象的列表（例如一次提取多个用户）；
Partial Streaming：适合流式返回单个对象，边生成边处理。

部分流式（Partials）

使用create_partial流式产出单个对象。注意：流式时不要在响应模型中声明 validator，否则会破坏流式过程。

# Third-party imports import instructor from pydantic import BaseModel, Field # Set up environment (typically handled before script execution) # os.environ["ANTHROPIC_API_KEY"] = "your-api-key" # Uncomment and replace with your API key if not set # Initialize client with explicit mode client = instructor.from_provider( "anthropic/claude-sonnet-5", mode=instructor.Mode.TOOLS, ) # Define your model with proper annotations class User(BaseModel): """Model representing a user profile.""" name: str = Field(description="The user's full name") age: int = Field(description="The user's age in years") bio: str = Field(description="A biographical description of the user") try: # Stream partial objects as they're generated for partial_user in client.create_partial( messages=[ { "role": "system", "content": "Create a detailed user profile based on the information provided.", }, {"role": "user", "content": "Create a user profile for Jason, age 25"}, ], response_model=User, max_tokens=4096, ): print(f"Current state: {partial_user}") # Expected output: #> Current state: name='Jason' age=None bio=None #> Current state: name='Jason' age=25 bio='Jason is a 25-year-old with an adventurous spirit and a love for technology. He is' #> Current state: name='Jason' age=25 bio='Jason is a 25-year-old with an adventurous spirit and a love for technology. He is always on the lookout for new challenges and opportunities to grow both personally and professionally.' except Exception as e: print(f"Error during streaming: {e}")

流式解析依赖AnthropicHandlerBase中的extract_streaming_json：在TOOLS/PARALLEL_TOOLS模式下读取chunk.delta.partial_json，在JSON/JSON_SCHEMA模式下读取chunk.delta.text（handlers.py）。

Iterable 流式

create_iterable用于从单次提示中提取多个同类型对象：

# Third-party imports from instructor import from_provider from pydantic import BaseModel, Field # Set up environment (typically handled before script execution) # os.environ["ANTHROPIC_API_KEY"] = "your-api-key" # Uncomment and replace with your API key if not set # Initialize client with explicit mode client = from_provider(mode=instructor.Mode.TOOLS) # Define your model with proper annotations class User(BaseModel): """Model representing a basic user.""" name: str = Field(description="The user's full name") age: int = Field(description="The user's age in years") try: # Create an iterable of user objects users = client.create_iterable( messages=[ { "role": "system", "content": "Extract all users from the provided text into structured format.", }, { "role": "user", "content": """ Extract users: 1. Jason is 25 years old 2. Sarah is 30 years old 3. Mike is 28 years old """, }, ], max_tokens=4096, response_model=User, ) # Process each user as it's extracted for user in users: print(user) # Expected output: #> name='Jason' age=25 #> name='Sarah' age=30 #> name='Mike' age=28 except Exception as e: print(f"Error during iteration: {e}")

在create_iterable（非流式参数下）中，Iterable[T]同样会被识别为并行工具并逐个校验；在流式场景则退化为StreamingModelState驱动的增量解析。两者完整语义见 docs/concepts/iterable.md 与 docs/concepts/partial.md。

Instructor Modes：模式选择与自动检测

针对 Anthropic 支持的不同响应方式，Instructor 提供多种模式：

instructor.Mode.JSON：使用 Anthropic 的文本补全能力，从纯文本回复中提取并解析目标响应模型；
instructor.Mode.TOOLS：使用 Anthropic 的工具调用 API 返回结构化输出，且能从Iterable[Union[...]]响应模型自动检测并行工具；
instructor.Mode.PARALLEL_TOOLS：已废弃——请改用Mode.TOOLS+Iterable[Union[Model1, Model2, ...]]，并行模式会自动被检测。

从源码看，Mode.JSON对应的AnthropicJSONHandler会把 Pydantic 模型的model_json_schema()以 JSON 形式注入 system 消息，提示模型"返回符合该 schema 的 JSON 实例，而不是 schema 本身"，随后用extract_json_from_codeblock从回复中抽取 JSON 并校验。

Mode 自动检测

Mode.TOOLS会根据响应模型与参数智能调整行为：

Response Model	Parameters	Behavior
`Model`	Regular	Single tool (forced)
`Model`	`thinking={...}`	Single tool with extended thinking (auto)
`Iterable[Union[Model1, Model2]]`	Regular	Parallel tools (auto)
`Iterable[Union[Model1, Model2]]`	`thinking={...}`	Parallel with thinking

建议一律使用Mode.TOOLS，因为它自动覆盖上述所有场景，是保证输出 schema 最稳妥的方式。源码中的判定逻辑位于AnthropicToolsHandler.prepare_request（handlers.py）：普通模型走强制工具调用tool_choice={"type": "tool", "name": ..., "disable_parallel_tool_use": True}（防止模型对同一工具发出多个tool_use块导致解析失败）；开启 thinking 或并行时则改为tool_choice={"type": "auto"}。

原生结构化输出（JSON_SCHEMA）

此外，仓库还实现了AnthropicStructuredOutputsHandler（Mode.JSON_SCHEMA），它走 Claude 的原生结构化输出强制，即通过output_format={"type": "json_schema", "schema": ...}参数下发 schema，并自动追加structured-outputs-2025-11-13beta 头。注意其适用前提：需要 Anthropic SDK 支持output_format参数（建议anthropic>=0.71.0），否则会发出警告并回退到 JSON 模式指令。

缓存（Prompt Caching）

Instructor 支持对文本输入和图片做 Anthropic Prompt Caching。仓库还提供了基于缓存实现 Anthropic Contextual Retrieval 的完整走读文章 docs/blog/posts/anthropic-prompt-caching.md。

文本输入缓存

假设你有一个很大的book.txt需要反复携带在上下文中，可以在消息内容块上标记cache_control：

# Third-party imports import instructor from pydantic import BaseModel, Field # Set up environment (typically handled before script execution) # os.environ["ANTHROPIC_API_KEY"] = "your-api-key" # Uncomment and replace with your API key if not set # Define your Pydantic model with proper annotations class Character(BaseModel): """Model representing a character extracted from text.""" name: str = Field(description="The character's full name") description: str = Field(description="A description of the character") # Initialize client with explicit mode and prompt caching client = instructor.from_provider( "anthropic/claude-sonnet-5", mode=instructor.Mode.TOOLS, ) try: # Load your large context with open("./book.txt") as f: book = f.read() # Make multiple calls using the cached context for _ in range(2): # The first time processes the large text, subsequent calls use the cache resp, completion = client.create_with_completion( messages=[ { "role": "system", "content": "Extract character information from the provided text.", }, { "role": "user", "content": [ { "type": "text", "text": "<book>" + book + "</book>", "cache_control": {"type": "ephemeral"}, # Mark for caching }, { "type": "text", "text": "Extract a character from the text given above", }, ], }, ], response_model=Character, max_tokens=1000, ) # Process the result print(f"Character: {resp.name}") print(f"Description: {resp.description}") # The completion contains the raw response print(f"Raw completion length: {len(completion)}") # Note: Second iteration should be faster due to cache hit except Exception as e: print(f"Error: {e}")

注意：create_with_completion返回(响应模型实例, 原始 completion)二元组，completion.usage中的cache_creation_input_tokens/cache_read_input_tokens字段可用于验证缓存是否生效（前面PDFWithCacheControl示例正是用该字段做断言）。

图片缓存

图片同样支持缓存，对反复使用同一批图片的场景能显著降低成本。可以给type: "image"内容块附加cache_control，并配合autodetect_images=True自动处理图片内容：

# Third-party imports import instructor from pydantic import BaseModel, Field # Set up environment (typically handled before script execution) # os.environ["ANTHROPIC_API_KEY"] = "your-api-key" # Uncomment and replace with your API key if not set # Define your model for image analysis class ImageAnalyzer(BaseModel): """Model for analyzing image content.""" content_description: str = Field( description="Description of what appears in the images" ) objects: list[str] = Field(description="List of objects visible in the images") scene_type: str = Field( description="Type of scene shown in the images (indoor, outdoor, etc.)" ) # Initialize client with explicit mode and image caching enabled client = instructor.from_provider( "anthropic/claude-sonnet-5", mode=instructor.Mode.TOOLS, ) try: # Configure cache control for images cache_control = {"type": "ephemeral"} # Make a request with cached images response = client.create( response_model=ImageAnalyzer, messages=[ { "role": "system", "content": "Analyze the content of the provided images in detail.", }, { "role": "user", "content": [ "What is in these two images?", # Remote image with caching { "type": "image", "source": "https://example.com/image.jpg", "cache_control": cache_control, }, # Local image with caching { "type": "image", "source": "path/to/image.jpg", "cache_control": cache_control, }, ], }, ], autodetect_images=True, # Automatically handle image content ) # Process the results print(f"Description: {response.content_description}") print(f"Objects: {', '.join(response.objects)}") print(f"Scene type: {response.scene_type}") # Subsequent identical requests will use cached images except Exception as e: print(f"Error during image analysis: {e}")

缓存的通用机制与更多实践可参考 docs/concepts/caching.md。

扩展思考（Extended Thinking）

Anthropic 的 Claude 系列支持扩展思考，让模型在处理复杂问题前先进行推理，再给出结构化输出。在 Instructor 中使用Mode.TOOLS并传入thinking参数即可开启。

在 TOOLS 模式下使用扩展思考

import instructor from pydantic import BaseModel class Answer(BaseModel): answer: float client = instructor.from_provider("anthropic/claude-sonnet-5") response = client.create( response_model=Answer, messages=[ { "role": "user", "content": "Which is larger, 9.11 or 9.8?", }, ], max_tokens=2000, thinking={"type": "adaptive"}, tool_choice={"type": "auto"}, ) # Response is a validated Answer object assert isinstance(response, Answer) assert response.answer == 9.8

工作原理

当对 Sonnet 5 传入thinking={"type": "adaptive"}时：

工具选择：使用自适应思考时必须显式传tool_choice={"type": "auto"}——思考模式不支持强制工具选择；
模型推理：Claude 自行决定本次请求所需的推理量，自适应思考不使用budget_tokens；
结构化输出：推理结束后，模型返回符合响应模型的合法工具调用；
校验：返回内容自动按 Pydantic 模型完成校验。

源码层面，AnthropicToolsHandler.prepare_request检测到thinking.type == "enabled"时会强制tool_choice={"type": "auto"}，并向 system 追加"只返回工具调用、不要多余文本"的指令（handlers.py），确保思考内容不会污染结构化输出。

废弃说明

Mode.ANTHROPIC_REASONING_TOOLS已废弃，请改用Mode.TOOLS+thinking参数。两种模式目前都支持思考，但标准TOOLS模式更受推荐且更灵活。

错误处理与自动重试

结构化输出失败通常分两类：模型输出不合法，或输出在 Pydantic 校验阶段未通过。文档示例中统一捕获instructor.exceptions.InstructorError（校验错误）与兜底Exception。从源码看，Anthropic handler 还会抛出更细粒度的异常：

IncompleteOutputException：当stop_reason == "max_tokens"（或 OpenAI 风格的finish_reason == "length"）时抛出，携带last_completion便于续写（handlers.py）；
ResponseParsingError：JSON 模式下响应中没有可解析文本时抛出。

校验失败时会触发 reask 机制：AnthropicToolsHandler.handle_reask会把上一次回复中的每个tool_use块都收集起来，并为每一个工具调用生成对应的tool_result（is_error=True）回传，避免并行工具场景因遗漏 tool_result 而收到 400 错误（对应 issue #2485，见 handlers.py）。