Pydantic AI 接入 Amazon Bedrock 完全指南：Converse 与 Mantle 双路由实战

Pydantic AI 接入 Amazon Bedrock 完全指南：Converse 与 Mantle 双路由实战

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai

Amazon Bedrock 以单一入口聚合了 Anthropic、Amazon、Cohere、Meta、Mistral、DeepSeek、Qwen 以及 OpenAI 等众多基础模型。Pydantic AI 通过两条独立的 AWS API 路由接入 Bedrock：覆盖绝大多数模型的Bedrock Converse（bedrock:前缀），以及专门服务 GPT-5.x / GPT-OSS 的Bedrock Mantle（bedrock-mantle:前缀）。读完本文，你将掌握两条路由的选型、安装与认证配置、模型设置定制、提示缓存（Prompt Caching）、服务层级、推理配置文件、重试策略与自定义 HTTP 头等完整实战方案，并能通过源码与测试理解其底层实现机制。

双路由总览：按模型前缀选择入口

Bedrock 上托管着不同厂商、不同年代的模型，Pydantic AI 依据模型前缀决定走哪条 AWS API：

两条路由共享同一套 AWS 凭证体系。bedrock:前缀始终走 Converse API；若用它请求前沿 OpenAI 模型（如 GPT-5.4 及更新版本），会在模型构造阶段直接抛出错误并引导你改用bedrock-mantle:——从源码看，BedrockProvider.model_profile中对非gpt-oss开头的 OpenAI 模型返回bedrock_supported_on_converse=False，而BedrockConverseModel.__init__检测到该标记后会抛出带 Mantle 提示的UserError，避免用户在请求阶段才收到晦涩的 Converse 报错。

Bedrock Converse

BedrockConverseModel对接 Bedrock Runtime 的 Converse / ConverseStream / CountTokens 三组 API，覆盖 Bedrock 上最广泛的模型目录。模型名由BedrockModelName类型约束，它在源码中显式枚举了各厂商当前最新模型（Anthropic、Amazon Nova、Cohere、Meta Llama、Mistral、DeepSeek、Qwen、Google Gemma、MiniMax、NVIDIA Nemotron、Writer Palmyra、Z.AI GLM、Moonshot Kimi 等），同时由于 Bedrock 模型名常带日期戳且列表频繁变动，类型上也允许任意字符串。

安装

使用BedrockConverseModel需安装pydantic-ai，或以bedrock可选组安装pydantic-ai-slim：

pip/uv-add "pydantic-ai-slim[bedrock]"

该可选组会引入boto3。若缺少依赖，源码会在导入阶段抛出明确提示（见 bedrock.py 与 bedrock provider）。

配置与前置条件

使用 AWS Bedrock 需要一个已启用 Bedrock 服务的 AWS 账号及相应凭证。凭证来源有两种：直接提供 AWS 凭证，或传入一个预配置好的 boto3 client。BedrockModelName中列出了可用的 Bedrock 模型清单，涵盖 Anthropic、Amazon、Cohere、Meta 与 Mistral 等厂商的模型。

环境变量

可通过环境变量设置 AWS 凭证（boto3 还支持配置文件等其他方式）：

export AWS_BEARER_TOKEN_BEDROCK='your-api-key' # 或者： export AWS_ACCESS_KEY_ID='your-access-key' export AWS_SECRET_ACCESS_KEY='your-secret-key' export AWS_DEFAULT_REGION='us-east-1' # 或你偏好的区域

从 BedrockProvider 源码 可以看到完整的凭证解析顺序：AWS_BEARER_TOKEN_BEDROCK存在时走 bearer token 认证（签名版本为bearer），否则回退到AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY/AWS_SESSION_TOKEN的 SigV4 认证；区域未指定时读取AWS_DEFAULT_REGION；此外还支持AWS_READ_TIMEOUT（默认 300 秒）与AWS_CONNECT_TIMEOUT（默认 60 秒）两个超时环境变量。未提供区域且无 boto3 client 时会抛出UserError。

然后即可按模型名直接使用：

from pydantic_ai import Agent agent = Agent('bedrock:anthropic.claude-sonnet-4-5-20250929-v1:0') ...

也可以直接用模型名初始化模型对象：

from pydantic_ai import Agent from pydantic_ai.models.bedrock import BedrockConverseModel model = BedrockConverseModel('anthropic.claude-sonnet-4-5-20250929-v1:0') agent = Agent(model) ...

定制 Bedrock Runtime API 调用

可通过BedrockModelSettings为 Converse 请求追加额外参数，例如护栏（guardrail）配置与性能配置。参数会按名称透传到 boto3 请求体：bedrock_guardrail_config映射为guardrailConfig、bedrock_performance_configuration映射为performanceConfig（见_messages_create）。

from pydantic_ai import Agent from pydantic_ai.models.bedrock import BedrockConverseModel, BedrockModelSettings # 定义带护栏与性能配置的 Bedrock 模型设置 bedrock_model_settings = BedrockModelSettings( bedrock_guardrail_config={ 'guardrailIdentifier': 'v1', 'guardrailVersion': 'v1', 'trace': 'enabled' }, bedrock_performance_configuration={ 'latency': 'optimized' } ) model = BedrockConverseModel(model_name='us.amazon.nova-pro-v1:0') agent = Agent(model=model, model_settings=bedrock_model_settings)

当护栏配置中trace设为'enabled'（如上例）时，Bedrock 返回的护栏评估结果会原样存入ModelResponse.provider_details的'trace'键，例如result.all_messages()[-1].provider_details['trace']。源码中_process_response会把响应中的trace与finish_reason一并写入provider_details（见 bedrock.py）。

BedrockModelSettings还支持以下透传参数（定义见 bedrock.py）：

• bedrock_request_metadata：附加到请求的元数据（映射为requestMetadata）；
• bedrock_additional_model_response_fields_paths：从模型响应中提取额外字段的 JSON 路径（映射为additionalModelResponseFieldPaths）；
• bedrock_prompt_variables：提示模板变量（映射为promptVariables）；
• bedrock_additional_model_requests_fields：模型专属参数的兜底逃生舱（映射为additionalModelRequestFields）。

自定义 HTTP 头

使用ModelSettings.extra_headers为Converse、ConverseStream、CountTokens三类请求附加 HTTP 头，适用于需要经过 API 网关或代理的场景：

from pydantic_ai import Agent from pydantic_ai.models.bedrock import BedrockModelSettings agent = Agent( 'bedrock:us.amazon.nova-micro-v1:0', model_settings=BedrockModelSettings( extra_headers={'X-Tenant-ID': 'example-tenant'}, ), )

实现上，Pydantic AI 通过 boto3 的事件钩子在每次请求时注入这些头：_register_extra_headers为三个操作注册provide-client-params与before-call事件处理器，_inject_extra_headers在签名前写入（见 bedrock.py）。由于注入发生在签名之前，SigV4 认证下这些头会被纳入请求签名。

注意：不要用extra_headers覆盖 boto3 自身管理的头（如Authorization、User-Agent、X-Amz-Date、Host、Content-Length），这些值可能被忽略或导致请求失败——botocore 会在注入后重新计算并覆盖它们。

服务层级（Service Tier）

Bedrock 支持通过推理配置文件控制服务层级来管理吞吐与成本。可以使用统一的service_tier字段，也可以使用 Bedrock 专有的bedrock_service_tier字段；两者同时设置时bedrock_service_tier优先（源码中先检查bedrock_service_tier，命中则直接使用，否则才回退到统一的service_tier，见 bedrock.py）。

统一字段在 Bedrock 上的映射关系：

• 'auto'：请求中省略serviceTier字段，由 AWS 应用服务端默认值（Standard 层级）；
• 'default'：显式发送{'type': 'default'}——显式退出未来服务端自动升级到 premium 层级的机制；
• 'flex'：发送{'type': 'flex'}；
• 'priority'：发送{'type': 'priority'}。

若要请求 Bedrock 的'reserved'层级（需要预先购买的容量预留），必须直接设置bedrock_service_tier——统一字段无法表达该值。

提示缓存（Prompt Caching）

Bedrock 支持在 Anthropic 模型上启用提示缓存，让昂贵的上下文可以在多次请求间复用。Pydantic AI 提供四种缓存方式：

1. 用CachePoint缓存用户消息：插入一个CachePoint标记，缓存它之前的所有内容。如果CachePoint位于用户提示内容块的开头（该消息内其前无内容），则改为缓存到上一条用户消息的结尾。传入CachePoint(ttl='1h')可启用延长缓存时长。
2. 缓存系统指令：将BedrockModelSettings.bedrock_cache_instructions设为True（默认使用 5m TTL），或直接指定'5m'/'1h'。同时存在静态与动态指令时，缓存点放在最后一条静态指令之后，这样动态指令的变化不会使静态缓存失效。
3. 缓存工具定义：将BedrockModelSettings.bedrock_cache_tool_definitions设为True（默认 5m TTL），或直接指定'5m'/'1h'。
4. 缓存全部消息：将BedrockModelSettings.bedrock_cache_messages设为True（默认 5m TTL），或直接指定'5m'/'1h'，自动缓存最后一条用户消息。

缓存点由_get_cache_point生成：True产生{'type': 'default'}，字符串值则额外带上ttl。设置项仅在模型 profile 声明支持时生效——bedrock_cache_instructions/bedrock_cache_messages依赖bedrock_supports_prompt_caching，bedrock_cache_tool_definitions依赖bedrock_supports_tool_caching。

最低 token 阈值：AWS 只有在分段超过各厂商特定的最低 token 阈值后才会提供缓存命中（参见 Bedrock 提示缓存文档）。过短的提示或工具定义低于阈值会绕过缓存，因此小负载不要指望省钱。

示例 1：自动消息缓存

用bedrock_cache_messages自动缓存最后一条用户消息：

from pydantic_ai import Agent from pydantic_ai.models.bedrock import BedrockModelSettings agent = Agent( 'bedrock:us.anthropic.claude-sonnet-4-5-20250929-v1:0', system_prompt='You are a helpful assistant.', model_settings=BedrockModelSettings( bedrock_cache_messages=True, # 自动缓存最后一条消息 ), ) # 最后一条消息被自动缓存——无需手动 CachePoint result1 = agent.run_sync('What is the capital of France?') # 后续相似的对话受益于缓存 result2 = agent.run_sync('What is the capital of Germany?') print(f'Cache write: {result1.usage.cache_write_tokens}') print(f'Cache read: {result2.usage.cache_read_tokens}')

示例 2：组合式缓存策略

组合多个缓存设置以获得最大收益：

from pydantic_ai import Agent, RunContext from pydantic_ai.models.bedrock import BedrockConverseModel, BedrockModelSettings model = BedrockConverseModel('us.anthropic.claude-sonnet-4-5-20250929-v1:0') agent = Agent( model, system_prompt='Detailed instructions...', model_settings=BedrockModelSettings( bedrock_cache_instructions=True, # 缓存系统指令 bedrock_cache_tool_definitions='1h', # 以 1h TTL 缓存工具定义 bedrock_cache_messages=True, # 同时缓存最后一条消息 ), ) @agent.tool def search_docs(ctx: RunContext, query: str) -> str: """Search documentation.""" return f'Results for {query}' result = agent.run_sync('Search for Python best practices') print(result.output)

示例 3：用 CachePoint 精细控制

手动插入CachePoint标记以精确控制缓存位置：

from pydantic_ai import Agent, CachePoint agent = Agent( 'bedrock:us.anthropic.claude-sonnet-4-5-20250929-v1:0', system_prompt='Instructions...', ) # 为特定内容块手动控制缓存点 result = agent.run_sync([ 'Long context from documentation...', CachePoint(), # 缓存到此为止的所有内容 'First question' ]) print(result.output)

读取缓存用量统计

通过RequestUsage读取缓存用量统计：

from pydantic_ai import Agent, CachePoint agent = Agent('bedrock:us.anthropic.claude-sonnet-4-5-20250929-v1:0') async def main(): result = await agent.run( [ 'Reference material...', CachePoint(), 'What changed since last time?', ] ) usage = result.usage print(f'Cache writes: {usage.cache_write_tokens}') print(f'Cache reads: {usage.cache_read_tokens}')

底层由_map_usage从响应中的cacheReadInputTokens/cacheWriteInputTokens等字段提取，并剥离区域前缀后映射到RequestUsage。

缓存点数量限制

Bedrock 强制每个请求最多4 个缓存点。Pydantic AI 自动管理该上限，确保请求始终合规。

缓存点的分配方式

缓存点可放置在三个位置：

1. 系统提示：通过bedrock_cache_instructions设置（在最后一个系统提示块后加缓存点）；
2. 工具定义：通过bedrock_cache_tool_definitions设置（在最后一个工具定义后加缓存点）；
3. 消息：通过CachePoint标记或bedrock_cache_messages设置（在消息内容中加缓存点）。

每个设置最多占用 1 个缓存点，但可以组合使用。

自动限流机制

当所有来源（设置 +CachePoint标记）的缓存点超过 4 个时，Pydantic AI 会自动从较旧的消息内容中移除多余的缓存点（保留最新的）。实现见_limit_cache_points：先统计系统提示与工具中的缓存点数，若两者之和已超 4 则抛出UserError（无法自动修复的配置错误），否则按"从新到旧"遍历消息，在剩余预算内保留最近的缓存点并移除更旧的。

from pydantic_ai import Agent, CachePoint from pydantic_ai.models.bedrock import BedrockModelSettings agent = Agent( 'bedrock:us.anthropic.claude-sonnet-4-5-20250929-v1:0', system_prompt='Instructions...', model_settings=BedrockModelSettings( bedrock_cache_instructions=True, # 1 个缓存点 bedrock_cache_tool_definitions=True, # 1 个缓存点 ), ) @agent.tool_plain def search() -> str: return 'data' # 已使用 2 个缓存点（指令 + 工具） # 还能再添加 2 个 CachePoint 标记（总计上限 4 个） result = agent.run_sync([ 'Context 1', CachePoint(), # 最旧——将被移除 'Context 2', CachePoint(), # 保留（第 3 个点） 'Context 3', CachePoint(), # 保留（第 4 个点） 'Question' ]) # 最终缓存点：instructions + tools + Context 2 + Context 3 = 4 print(result.output)

要点总结：

• 系统与工具缓存点始终保留；
• bedrock_cache_messages创建的缓存点始终保留（它是最新的消息缓存点）；
• 消息中额外的CachePoint标记在超限时按从旧到新的顺序被移除；
• 这保证了关键缓存（指令/工具）得以维持，同时仍能受益于消息级缓存。

另外，源码还会处理一个 AWS 的边界限制：缓存点不能紧跟在 document/video 内容块之后。_insert_cache_point_before_trailing_documents（bedrock.py）会自动把缓存点插到尾部文档/视频组之前；若消息只含文档/视频而无文本，则会抛出提示添加文本的UserError。

provider参数

通过provider参数传入自定义BedrockProvider，适用于直接指定凭证或使用自定义 boto3 client 的场景：

from pydantic_ai import Agent from pydantic_ai.models.bedrock import BedrockConverseModel from pydantic_ai.providers.bedrock import BedrockProvider # 直接使用 AWS 凭证 model = BedrockConverseModel( 'anthropic.claude-sonnet-4-5-20250929-v1:0', provider=BedrockProvider( region_name='us-east-1', aws_access_key_id='your-access-key', aws_secret_access_key='your-secret-key', ), ) agent = Agent(model) ...

也可以传入预配置的 boto3 client：

import boto3 from pydantic_ai import Agent from pydantic_ai.models.bedrock import BedrockConverseModel from pydantic_ai.providers.bedrock import BedrockProvider # 使用预配置的 boto3 client bedrock_client = boto3.client('bedrock-runtime', region_name='us-east-1') model = BedrockConverseModel( 'anthropic.claude-sonnet-4-5-20250929-v1:0', provider=BedrockProvider(bedrock_client=bedrock_client), ) agent = Agent(model) ...

从源码看，BedrockProvider还支持aws_session_token、profile_name、aws_read_timeout、aws_connect_timeout、base_url（自定义 endpoint）等参数；bedrock_client一旦提供，其余参数全部忽略。BedrockProvider.client支持运行时替换——这对长驻服务中轮换短期凭证（如 STS 临时凭证）非常有用，替换后所有使用该 provider 的模型都会自动生效（见 providers/bedrock.py）。

使用 AWS 应用推理配置文件（Application Inference Profiles）

AWS Bedrock 支持自定义应用推理配置文件用于成本追踪与资源管理。设置bedrock_inference_profile可通过推理配置文件路由请求，同时保留基础模型名用于能力检测：

from pydantic_ai import Agent from pydantic_ai.models.bedrock import BedrockConverseModel from pydantic_ai.providers.bedrock import BedrockProvider provider = BedrockProvider(region_name='us-east-2') model = BedrockConverseModel( 'us.anthropic.claude-opus-4-5-20251101-v1:0', provider=provider, settings={ 'bedrock_inference_profile': 'arn:aws:bedrock:us-east-2:123456789012:application-inference-profile/my-profile', }, ) agent = Agent(model)

实现上，_messages_create会用bedrock_inference_profile的值作为modelId发送给 Converse / ConverseStream，而model_name仍保留基础模型名用于能力探测与 token 计数（count_tokens使用remove_bedrock_geo_prefix(self.model_name)）。

配置重试

Bedrock 使用 boto3 内置的重试机制，可通过传入带重试设置的自定义 boto3 client 来配置：

import boto3 from botocore.config import Config from pydantic_ai import Agent from pydantic_ai.models.bedrock import BedrockConverseModel from pydantic_ai.providers.bedrock import BedrockProvider # 配置重试设置 config = Config( retries={ 'max_attempts': 5, 'mode': 'adaptive' # 推荐用于限流场景 } ) bedrock_client = boto3.client( 'bedrock-runtime', region_name='us-east-1', config=config ) model = BedrockConverseModel( 'us.amazon.nova-micro-v1:0', provider=BedrockProvider(bedrock_client=bedrock_client), ) agent = Agent(model)

注意 boto3 的计数口径：boto3 从另一侧计数，Config(retries={'max_attempts': N})允许总共1 + N次尝试。同时，boto3 之下没有httpx2传输层，因此这是 Agent 重试预算与网络之间唯一的重试层。各层的叠加关系参见重试乘法。

重试模式

• 'legacy'（默认）：5 次尝试，基础重试行为；
• 'standard'：3 次尝试，错误覆盖更全面；
• 'adaptive'：3 次尝试并带客户端限流（推荐用于处理ThrottlingException）。

注意：与其他使用 httpx 发起 HTTP 请求的 provider 不同，Bedrock 使用 boto3 原生重试机制。传输层重试中描述的重试策略不适用于 Bedrock。

Bedrock Mantle

Amazon Bedrock Mantle 通过 OpenAI 兼容 API 服务 OpenAI 模型（GPT-5.x 与 GPT-OSS）。使用bedrock-mantle:前缀：

from pydantic_ai import Agent agent = Agent('bedrock-mantle:openai.gpt-5.6-luna')

需要安装bedrock-mantle可选组：

pip/uv-add "pydantic-ai-slim[bedrock-mantle]"

BedrockMantleProvider与 Converse 路由共享同一套 AWS 凭证——通过AWS_BEARER_TOKEN_BEDROCK的 bearer token，或通过 SigV4 的 AWS access keys / profile——并根据region_name（或AWS_DEFAULT_REGION/AWS_REGION环境变量）推导 endpoint。从源码看，未指定区域且未传base_url时会抛出UserError，提示需要设置区域或 base_url（bedrock_mantle.py）。

模型名决定 endpoint 家族（接口路由由bedrock_mantle_model_profile依据模型名前缀解析）：

源码中两种客户端共用同一套 transport 与认证：_client服务/openai/v1（GPT-5.x），_v1_client服务/v1（GPT-OSS 的 chat 与 responses），两者由同一个 base client 通过with_options派生（bedrock_mantle.py）。

如需使用自定义 Mantle origin（例如代理），可向BedrockMantleProvider传入base_url；其 origin（剥离任何/openai/v1或/v1后缀后）与region_name一样用于在两个 endpoint 家族之间按模型路由：

from pydantic_ai import Agent from pydantic_ai.models.bedrock_mantle import BedrockMantleResponsesModel from pydantic_ai.providers.bedrock_mantle import BedrockMantleProvider provider = BedrockMantleProvider(base_url='https://bedrock-mantle.us-east-1.api.aws/openai/v1') model = BedrockMantleResponsesModel('openai.gpt-5.6-luna', provider=provider) agent = Agent(model)

_mantle_origin（bedrock_mantle.py）会剥离去掉尾部/openai/v1或/v1后缀，从单一 origin 派生出两个兄弟 endpoint。

功能支持

Mantle 模型由 Pydantic AI 的 OpenAI 模型类提供服务——BedrockMantleResponsesModel与BedrockMantleChatModel分别继承OpenAIResponsesModel与OpenAIChatModel——因此它们接受与直接 OpenAI 模型相同的设置（OpenAIResponsesModelSettings与OpenAIChatModelSettings）。构造时若把 Responses 模型（如 GPT-5.x）误用 Chat 模型类（或反之），会在构造阶段抛出带指引的UserError。

上面提到的 Converse 路由特性——提示缓存、服务层级与应用推理配置文件——是 Converse API 特有的，不适用于 Mantle 路由。特别是bedrock_service_tier是 Converse 设置；Mantle 模型由 OpenAI 模型类服务，会把统一的service_tier作为同名 OpenAI 参数转发。

另外，源码对 Mantle 模型做了一些重要的能力裁剪：关闭了图片输出（supports_image_output=False，Bedrock Mantle 不像直连 OpenAI API 那样提供图片输出），并禁用了 OpenAI 原生工具（supported_native_tools=frozenset()，AWS 的服务端工具集成走 Lambda 或 MCP 而非 OpenAI 原生工具）。从 test_bedrock_mantle.py 的测试可以看到：Mantle 的/openai/v1Responses 端点在跨请求时会重置工具调用 ID 为call_0，Pydantic AI 会将其限定为response_id:call_0以保证历史全局唯一（openai_responses_tool_call_ids_are_response_scoped标记，见 test_reused_tool_call_ids 与 providers/bedrock_mantle.py）；GPT-OSS 的/v1/responses则天然返回全局唯一 ID，不受影响。

结语

本文以官方文档为骨架、以仓库源码为佐证，系统梳理了 Pydantic AI 接入 Amazon Bedrock 的两条完整路径。对于绝大多数 Bedrock 模型，使用bedrock:前缀的 Converse 路由即可，配合BedrockModelSettings实现护栏、性能、服务层级、提示缓存与推理配置文件的精细控制；对于 OpenAI GPT-5.x 与 GPT-OSS 前沿模型，则切换到bedrock-mantle:前缀的 OpenAI 兼容路由。两条路由共享 AWS 凭证体系，在模型构造阶段即可完成能力校验，把大量配置错误提前暴露。建议读者结合实际模型名（BedrockModelName枚举可见于 bedrock.py）与测试用例（tests/models/test_bedrock.py、tests/models/test_bedrock_mantle.py）动手验证，让 Agent 应用在 Bedrock 上稳定、经济地运行。

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai