尧图网页设计 
Haystack Anthropic 集成完全指南AnthropicChatGenerator、Foundry/Vertex 变体与 TokenCounter 实战【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本文是 Haystack 官方 Anthropic 集成anthropic-haystack的技术参考围绕AnthropicChatGenerator、AnthropicFoundryChatGenerator、AnthropicVertexChatGenerator与AnthropicTokenCounter四个组件系统讲解初始化参数、generation_kwargs支持范围、多模态输入、函数调用、流式输出、Prompt Caching、序列化与 Pipeline 集成方式。读完本文你将能直接使用 Claude 系列模型构建可生产化的对话、Agent 与 RAG 应用并在 Azure Foundry 与 Google Vertex AI 之间按需切换后端。集成全景一个包四条接入路径Anthropic 集成并不在 Haystack 核心代码库中而是通过独立的anthropic-haystack包提供本文所依据的 API 参考位于 docs-website/reference/integrations-api/anthropic.md。安装方式统一为pip install anthropic-haystack该包当前提供四个核心类覆盖三种模型托管形态与一种精确计数能力组件模块路径用途AnthropicChatGeneratorhaystack_integrations.components.generators.anthropic直连 Anthropic Messages API 的对话补全AnthropicFoundryChatGeneratorhaystack_integrations.components.generators.anthropic通过 Azure Foundry 调用 ClaudeAnthropicChatGenerator子类AnthropicVertexChatGeneratorhaystack_integrations.components.generators.anthropic通过 Anthropic Vertex AI API 调用 ClaudeAnthropicChatGenerator子类AnthropicTokenCounterhaystack_integrations.token_counters.anthropic调用POST /v1/messages/count_tokens精确统计输入 token其中 Foundry 与 Vertex 两个生成器都直接继承自AnthropicChatGenerator因此继承并重用了后者的工具调用、流式与序列化能力差异仅在于客户端目标端点与认证方式。AnthropicChatGenerator直连 Claude 的对话生成器AnthropicChatGenerator使用 Anthropic 的大语言模型完成对话补全输入输出均遵循 Haystack 统一的 ChatMessage 数据格式user、system、assistant、tool四种角色由 ChatRole 枚举定义并支持包含文本与图像的多模态输入。最小可用示例from haystack_integrations.components.generators.anthropic import ( AnthropicChatGenerator, ) from haystack.dataclasses import ChatMessage generator AnthropicChatGenerator( generation_kwargs{ max_tokens: 1000, temperature: 0.7, }, ) messages [ ChatMessage.from_system( You are a helpful, respectful and honest assistant ), ChatMessage.from_user(Whats Natural Language Processing?), ] print(generator.run(messagesmessages))该组件在 Pipeline 中最常见的位置是接在ChatPromptBuilder之后运行时的必填输入为messagesChatMessage列表输出键为replies模型回复的ChatMessage列表。支持的模型清单SUPPORTED_MODELS组件内置了一份非穷举的模型白名单用于在初始化时做基础校验SUPPORTED_MODELS: list[str] [ claude-opus-4-6, claude-sonnet-4-6, claude-haiku-4-5-20251001, claude-sonnet-4-5-20250929, claude-opus-4-5-20251101, claude-opus-4-1-20250805, claude-sonnet-4-20250514, claude-opus-4-20250514, claude-3-haiku-20240307, ]默认模型为claude-sonnet-4-5。清单之外仍可能有可用模型完整列表请以 Anthropic 官方模型总览为准从当前列表可以看出该集成覆盖了 Opus / Sonnet / Haiku 三代模型家族。初始化参数详解__init__( api_key: Secret Secret.from_env_var(ANTHROPIC_API_KEY), model: str claude-sonnet-4-5, streaming_callback: StreamingCallbackT | None None, generation_kwargs: dict[str, Any] | None None, ignore_tools_thinking_messages: bool True, tools: ToolsType | None None, anthropic_server_tools: list[dict[str, Any]] | None None, *, timeout: float | None None, max_retries: int | None None ) - Noneapi_keySecretAnthropic API 密钥默认从ANTHROPIC_API_KEY环境变量读取。Haystack 的 Secret 机制支持Secret.from_env_var(...)与Secret.from_token(...)两种注入方式推荐环境变量方案。modelstr要使用的模型名称。streaming_callbackStreamingCallbackT | None流式回调收到新 token 时被调用回调函数接收StreamingChunk作为参数。generation_kwargsdict[str, Any] | None发送给 Anthropic 端点的其他生成参数全部透传。ignore_tools_thinking_messagesboolAnthropic 在工具函数调用解析时会先产生一段思维链消息再返回真正的函数名与参数。设为True默认时组件会在检测到工具调用后丢弃这类 thinking 消息。toolsToolsType | None模型可用的 Tool 与/或 Toolset 对象列表也可传单个 Toolset每个工具必须有唯一名称。anthropic_server_toolslist[dict[str, Any]] | None直接传给 API 的 Anthropic 服务端原生工具列表例如网页搜索{type: web_search_20250305}、代码执行工具等由服务商托管的能力。timeoutfloat | NoneAnthropic 客户端调用超时不设置则使用 Anthropic 客户端默认值。max_retriesint | None失败请求的最大重试次数不设置则使用 Anthropic 客户端默认值。generation_kwargs 支持的参数范围所有与anthropic.Message.create兼容的参数均可通过generation_kwargs传入官方文档列出的受支持键包括参数含义system传给模型的系统消息max_tokens最大生成 token 数metadata传给模型的元数据字典stop_sequences模型应停止生成的字符串列表temperature采样温度top_p核采样nucleus sampling的 top_p 值top_ktop-k 采样的 k 值extra_headers传给模型的额外请求头例如用于 beta 功能开关thinking扩展思维参数字典其中budget_tokens必须小于max_tokensoutput_config传给模型的输出配置选项字典其中thinking与output_config是面向 Claude 4 系列推理与结构化输出能力的较新键extra_headers则是开启 Prompt Caching 等 beta 功能的关键入口见下文。多模态输入图文混合组件同时支持文本与图像输入。图像通过 ImageContent 数据类封装可从本地文件或 URL 加载from haystack.dataclasses import ChatMessage, ImageContent image_content ImageContent.from_file_path(path/to/image.jpg) messages [ ChatMessage.from_user( content_parts[Whats in this image?, image_content] ) ] generator AnthropicChatGenerator() result generator.run(messages)从 image_content.py 的源码可见ImageContent.from_file_path底层实际调用了ImageFileToImageContent转换器完成文件读取与 base64 编码此外还提供from_url下载后转 base64内部使用LinkContentFetcher以及可选的size缩放参数以控制传输体积。run 与 run_asyncrun( messages: list[ChatMessage] | str, streaming_callback: StreamingCallbackT | None None, generation_kwargs: dict[str, Any] | None None, tools: ToolsType | None None, ) - dict[str, list[ChatMessage]]messages输入消息列表如果直接传入字符串会被自动包装成一条user角色的ChatMessage这也是AnthropicGenerator被废弃后推荐迁移到本组件的原因之一。streaming_callback本次调用覆盖初始化的流式回调。generation_kwargs与初始化时传入的generation_kwargs按 key 合并——本次调用传入的键优先仅在初始化时设置的键会被保留。tools本次调用覆盖初始化时的工具配置每个工具名称必须唯一。返回值字典键replies对应模型回复的ChatMessage列表。run_async是run的异步版本签名与合并语义完全一致适用于异步 Pipeline 或 Web 服务处理器中。工具函数调用支持tools参数接受三种灵活配置形态Tool 对象列表逐个传入独立工具单个 Toolset整体传入一个工具集混合形态在同一列表中混排多个 Toolset 与独立 Tool。from haystack.tools import Tool, Toolset from haystack_integrations.components.generators.anthropic import AnthropicChatGenerator # 创建独立工具 weather_tool Tool( nameweather, descriptionGet weather info, parameters..., function... ) news_tool Tool( namenews, descriptionGet latest news, parameters..., function... ) # 将相关工具归组为一个 toolset math_toolset Toolset([add_tool, subtract_tool, multiply_tool]) # 混合传递 Toolset 与 Tool generator AnthropicChatGenerator( tools[math_toolset, weather_tool, news_tool] )这样既能将相关工具组织成逻辑分组也能按需附带独立工具。Haystack 侧的Tool/Toolset定义可参考 haystack/tools/tool.py 与 haystack/tools/toolset.py。工具调用在ChatMessage中以ToolCall模型发起的调用见 chat_message.py与ToolCallResult工具执行结果见 chat_message.py两种内容部件流转二者通过origin字段关联。流式输出Streaming向streaming_callback传入回调即可逐 token 接收输出。官方推荐直接使用内置的print_streaming_chunk打印文本 token 与工具事件工具调用与工具结果from haystack.components.generators.utils import print_streaming_chunk # 任何 Generator 或 ChatGenerator 均可配置流式回调 component SomeGeneratorOrChatGenerator(streaming_callbackprint_streaming_chunk) # ChatGenerator 传入消息列表 # from haystack.dataclasses import ChatMessage # component.run([ChatMessage.from_user(Your question here)]) # 非 chat 的 Generator 传入 prompt # component.run({prompt: Your prompt here})需要注意流式只支持单个响应若提供方支持多候选需显式设置n1。除非你需要特定传输通道如 SSE/WebSocket或自定义 UI 格式否则优先使用print_streaming_chunk。Prompt Caching长上下文降本提速Prompt Caching 允许你把大段文本如代码库上下文、长文档只发送一次后续请求直接引用缓存从而降低成本并缩短响应时间特别适合编码助手类应用。启用方式是在generation_kwargs中携带 beta 头并在系统消息的meta中标记缓存控制from haystack_integrations.components.generators.anthropic import AnthropicChatGenerator from haystack.dataclasses import ChatMessage from haystack.utils import Secret generation_kwargs {extra_headers: {anthropic-beta: prompt-caching-2024-07-31}} claude_llm AnthropicChatGenerator( api_keySecret.from_env_var(ANTHROPIC_API_KEY), generation_kwargsgeneration_kwargs, ) system_message ChatMessage.from_system( Replace with some long text documents, code or instructions ) system_message.meta[cache_control] {type: ephemeral} messages [ system_message, ChatMessage.from_user(A query about the long text for example), ] result claude_llm.run(messages) # 后续再次调用复用同一 system_message 即可命中缓存 messages [ system_message, ChatMessage.from_user(Another query about the long text etc), ] result claude_llm.run(messages)这里的关键是system_message.meta[cache_control] {type: ephemeral}Haystack 的ChatMessage自带meta元数据字典序列化时该标记会随请求一起传给 Anthropic从而完成缓存标记。AnthropicFoundryChatGenerator走 Azure Foundry 的 ClaudeAnthropicFoundryChatGenerator是AnthropicChatGenerator的轻量子类——请求与响应结构仍符合 Anthropic Messages API但流量不再流向api.anthropic.com而是经由你的 Azure Foundry 资源。适合组织统一采用 Azure 托管模型计费、网络、合规但仍希望使用 Claude 的场景无此需求时直接使用AnthropicChatGenerator即可。使用前提拥有启用 Foundry 的 Azure 订阅并在 Foundry 资源中部署了所需的 Anthropic 模型。初始化参数与认证方式__init__( *, api_key: Secret | None Secret.from_env_var(ANTHROPIC_FOUNDRY_API_KEY, strictTrue), resource: str | None None, endpoint: str | None None, model: str claude-sonnet-4-5, streaming_callback: Callable[[StreamingChunk], None] | None None, generation_kwargs: dict[str, Any] | None None, ignore_tools_thinking_messages: bool True, tools: ToolsType | None None, anthropic_server_tools: list[dict[str, Any]] | None None, timeout: float | None None, max_retries: int | None None, azure_ad_token_provider: Callable[[], str] | None None ) - None凭据三选一ANTHROPIC_FOUNDRY_API_KEY环境变量推荐api_key初始化参数配合 Haystack Secret APISecret.from_token(your-api-key-here)azure_ad_token_provider可调用对象按需返回新的 Azure AD token适用于 Entra ID / 托管标识这类不适合静态密钥的场景此时api_key可传None。端点二选一必须提供其一resource或ANTHROPIC_FOUNDRY_RESOURCE环境变量Foundry 资源短名称用于推导 URLendpoint完整端点 URL例如https://your-resource.openai.azure.com/anthropic适合自定义域名或非标准路由。其余参数model、streaming_callback、generation_kwargs、ignore_tools_thinking_messages、tools、anthropic_server_tools、timeout、max_retries语义与AnthropicChatGenerator一致generation_kwargs同样支持system、max_tokens、temperature、top_p、top_k、stop_sequences、metadata、extra_headers等键。Foundry 变体当前仅支持文本输入。使用示例与返回结构from haystack_integrations.components.generators.anthropic import AnthropicFoundryChatGenerator from haystack.dataclasses import ChatMessage from haystack.utils import Secret messages [ChatMessage.from_user(Whats Natural Language Processing?)] client AnthropicFoundryChatGenerator( modelclaude-sonnet-4-5, api_keySecret.from_env_var(ANTHROPIC_FOUNDRY_API_KEY), resourcemy-resource, ) response client.run(messages) print(response)官方文档给出的实际返回结构如下注意replies中的_meta字段包含模型名、索引、结束原因与 token 用量 {replies: [ChatMessage(_roleChatRole.ASSISTANT: assistant, _content[TextContent(text Natural Language Processing (NLP) is a field of artificial intelligence that focuses on enabling computers to understand, interpret, and generate human language. It involves developing techniques and algorithms to analyze and process text or speech data, allowing machines to comprehend and communicate in natural languages like English, Spanish, or Chinese.)], _nameNone, _meta{model: claude-sonnet-4-5, index: 0, finish_reason: end_turn, usage: {input_tokens: 15, output_tokens: 64}})]}可用模型取决于 Azure 部署SUPPORTED_MODELS: list[str] [ claude-opus-4-6, claude-sonnet-4-6, claude-sonnet-4-5, claude-opus-4-5, claude-opus-4-1, claude-haiku-4-5, ]与直连版本不同Foundry 变体的清单不含日期后缀且实际可用模型完全取决于你的 Azure Foundry 资源中部署了哪些模型。model参数对应 Foundry 中的部署名deployment name。工具、流式与异步工具与AnthropicChatGenerator相同的tools参数Tool 列表 / 单个 Toolset / 混合run()时传入的工具会覆盖初始化配置。流式streaming_callbackprint_streaming_chunk即可逐 token 打印。异步run_async自动可用适合异步 Pipeline 或 Web handlerimport asyncio from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.anthropic import ( AnthropicFoundryChatGenerator, ) async def main(): generator AnthropicFoundryChatGenerator(resourcemy-resource) result await generator.run_async([ChatMessage.from_user(Hello!)]) print(result[replies][0].text) asyncio.run(main())AnthropicVertexChatGenerator走 GCP Vertex AI 的 ClaudeAnthropicVertexChatGenerator同样是AnthropicChatGenerator的子类通过 Anthropic Vertex AI API 端点调用 ClaudeOpus、Sonnet、Haiku 等模型家族。使用前提拥有启用了 Vertex AI 的 GCP 项目并在 Vertex AI Model Garden 中激活目标 Anthropic 模型发起请求前可能需要先执行gcloud auth login完成 GCP 认证。初始化参数__init__( region: str, project_id: str, model: str claude-sonnet-420250514, streaming_callback: Callable[[StreamingChunk], None] | None None, generation_kwargs: dict[str, Any] | None None, ignore_tools_thinking_messages: bool True, tools: ToolsType | None None, anthropic_server_tools: list[dict[str, Any]] | None None, *, timeout: float | None None, max_retries: int | None None ) - Noneregionstr模型部署区域默认us-central1。project_idstr模型所在的 GCP 项目 ID。文档同时指出若二者均传None组件会回退到PROJECT_ID与REGION环境变量见 anthropicvertexchatgenerator.mdx。认证走 GCP 应用默认凭据ADC不接收api_key参数这与 Foundry 变体显著不同。generation_kwargs支持的键与直连版本一致system、max_tokens、metadata、stop_sequences、temperature、top_p、top_k、extra_headers。服务端工具限制在 Vertex AI 上仅支持基础网页搜索工具{type: web_search_20250305}带动态过滤的网页搜索、网页抓取与代码执行工具均不受支持。当前仅支持文本输入模态。使用示例Vertex 变体的模型名带版本日期后缀from haystack_integrations.components.generators.anthropic import AnthropicVertexChatGenerator from haystack.dataclasses import ChatMessage messages [ChatMessage.from_user(Whats Natural Language Processing?)] client AnthropicVertexChatGenerator( modelclaude-sonnet-420250514, project_idyour-project-id, regionyour-region ) response client.run(messages) print(response)返回结构同样为{replies: [ChatMessage(...)]}_meta中携带model、index、finish_reason与usage输入/输出 token 数。可用模型清单SUPPORTED_MODELS: list[str] [ claude-opus-4-6, claude-sonnet-4-6, claude-sonnet-4-520250929, claude-sonnet-420250514, claude-opus-4-520251101, claude-opus-4-120250805, claude-opus-420250514, claude-haiku-4-520251001, ]同样为非穷举清单完整可用模型 ID 需以 Claude on Vertex AI 的模型可用性页面为准。Prompt Caching 用法与直连版本完全一致extra_headers带anthropic-beta: prompt-caching-2024-07-31头 system_message.meta[cache_control]标记。AnthropicTokenCounter精确、模型相关的 token 计数AnthropicTokenCounter调用 Anthropic 的POST /v1/messages/count_tokens端点对指定 Claude 模型的ChatMessage与可选工具 schema 统计输入 token。该端点返回精确计数且不生成响应因此不产生生成费用。由于每次计数都发起远程 API 调用它需要 API 密钥且会引入网络延迟适合需要精确、按模型计数的场景如 Agent 上下文压缩。需要本地估算时可改用ApproximateTokenCounter或TiktokenCounter参考 docs-website/docs/token-counters。初始化与用法__init__( model: str, *, api_key: Secret Secret.from_env_var(ANTHROPIC_API_KEY), timeout: float | None None, max_retries: int | None None ) - Nonemodelstr必填用于分词的目标模型。token 计数是模型相关的必须针对实际将用于生成的模型计数。api_keySecret默认读取ANTHROPIC_API_KEY环境变量也可显式传入 Secret。timeout/max_retries底层 Anthropic 客户端的 HTTP 超时与失败重试次数。from haystack.dataclasses import ChatMessage from haystack_integrations.token_counters.anthropic import AnthropicTokenCounter counter AnthropicTokenCounter(modelclaude-sonnet-4-5) messages [ ChatMessage.from_system(You are a helpful assistant.), ChatMessage.from_user(How many tokens is this?), ] token_count counter.count(messages) print(token_count)显式配置密钥与客户端参数from haystack.utils import Secret counter AnthropicTokenCounter( modelclaude-sonnet-4-5, api_keySecret.from_env_var(MY_ANTHROPIC_API_KEY), timeout30.0, max_retries3, )count 方法与工具 schema 计数count(messages: list[ChatMessage], tools: ToolsType | None None) - intmessages要计数的ChatMessage列表。tools可选的 Tool 列表其 schema 也会被计入 token 消耗无内容可计时应返回0。token_count counter.count(messages, tools[search_tool])生命周期管理计数器会在首次调用count()时创建 API 客户端如需在应用启动阶段提前创建可显式调用warm_up()。使用完毕后调用close()释放客户端底层 HTTP 资源counter.warm_up() ... counter.close()非文本内容与压缩场景Anthropic 将图片与 PDF 文件也计入请求 token因此计数器对它们执行精确测量而非估算支持与AnthropicChatGenerator相同的内容类型JPEG、PNG、GIF、WebP 图片及application/pdf文件其他 MIME 类型会直接报错而不是估算。在 Agent 场景中可将计数器传给CompactionHook用 Claude 自身的分词器来评估 Agent 对话是否达到上下文窗口上限from haystack.hooks.compaction import CompactionHook, SlidingWindowCompactor compaction_hook CompactionHook( compactorSlidingWindowCompactor(), context_window200_000, token_counterAnthropicTokenCounter(modelclaude-sonnet-4-5), )注意该 Hook 会在 Agent 每一步都执行计数因此每次压缩检查都会产生一次 API 往返开销。组件生命周期与序列化三个生成器与计数器都实现了统一的 Haystack 组件协议方法行为warm_up()创建同步客户端生成器/初始化客户端计数器warm_up_async()创建异步客户端仅生成器close()关闭同步客户端并释放 HTTP 资源close_async()关闭异步客户端仅生成器to_dict()将组件序列化为字典便于保存与恢复from_dict(data)从字典反序列化恢复组件实例to_dict/from_dict使组件可以嵌入 Haystack 的 YAML/JSON 序列化体系配合 haystack/marshal/yaml.py 可实现完整的 Pipeline 定义持久化。组装进 Pipeline完整示例三个生成器在 Pipeline 中的用法完全一致——接在ChatPromptBuilder之后通过连接传递模板化消息。以直连版本为例from haystack import Pipeline from haystack.components.builders import ChatPromptBuilder from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.anthropic import AnthropicChatGenerator from haystack.utils import Secret pipe Pipeline() pipe.add_component(prompt_builder, ChatPromptBuilder()) pipe.add_component( llm, AnthropicChatGenerator(Secret.from_env_var(ANTHROPIC_API_KEY)), ) pipe.connect(prompt_builder, llm) country Germany system_message ChatMessage.from_system( You are an assistant giving out valuable information to language learners., ) messages [ system_message, ChatMessage.from_user(Whats the official language of {{ country }}?), ] res pipe.run( data{ prompt_builder: { template_variables: {country: country}, template: messages, }, }, ) print(res)Foundry 与 Vertex 变体只需把组件替换为AnthropicFoundryChatGenerator(resourcemy-resource)或AnthropicVertexChatGenerator(project_idtest_id, regionus-central1)即可其余 Pipeline 结构保持不变完整示例分别见 anthropicfoundrychatgenerator.mdx 与 anthropicvertexchatgenerator.mdx。迁移提示从 AnthropicGenerator 迁移AnthropicGenerator基于字符串 prompt 的文本补全版本已被标记为废弃并将在未来版本移除官方推荐迁移到AnthropicChatGenerator——后者同样接受纯字符串输入run会自动包装为user角色消息因此迁移成本极低。旧版行为与参数可参考 anthropicgenerator.mdx。延伸阅读完整 API 参考本文依据docs-website/reference/integrations-api/anthropic.md组件使用指南anthropicchatgenerator.mdx、anthropicfoundrychatgenerator.mdx、anthropicvertexchatgenerator.mdxToken 计数指南anthropictokencounter.mdx消息数据模型haystack/dataclasses/chat_message.py含ChatRole、TextContent、ToolCall、ToolCallResult多模态内容模型haystack/dataclasses/image_content.py工具抽象haystack/tools/tool.py、haystack/tools/toolset.py【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Haystack Anthropic 集成完全指南：AnthropicChatGenerator、Foundry/Vertex 变体与 TokenCounter 实战
Haystack Anthropic 集成完全指南AnthropicChatGenerator、Foundry/Vertex 变体与 TokenCounter 实战【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本文是 Haystack 官方 Anthropic 集成anthropic-haystack的技术参考围绕AnthropicChatGenerator、AnthropicFoundryChatGenerator、AnthropicVertexChatGenerator与AnthropicTokenCounter四个组件系统讲解初始化参数、generation_kwargs支持范围、多模态输入、函数调用、流式输出、Prompt Caching、序列化与 Pipeline 集成方式。读完本文你将能直接使用 Claude 系列模型构建可生产化的对话、Agent 与 RAG 应用并在 Azure Foundry 与 Google Vertex AI 之间按需切换后端。集成全景一个包四条接入路径Anthropic 集成并不在 Haystack 核心代码库中而是通过独立的anthropic-haystack包提供本文所依据的 API 参考位于 docs-website/reference/integrations-api/anthropic.md。安装方式统一为pip install anthropic-haystack该包当前提供四个核心类覆盖三种模型托管形态与一种精确计数能力组件模块路径用途AnthropicChatGeneratorhaystack_integrations.components.generators.anthropic直连 Anthropic Messages API 的对话补全AnthropicFoundryChatGeneratorhaystack_integrations.components.generators.anthropic通过 Azure Foundry 调用 ClaudeAnthropicChatGenerator子类AnthropicVertexChatGeneratorhaystack_integrations.components.generators.anthropic通过 Anthropic Vertex AI API 调用 ClaudeAnthropicChatGenerator子类AnthropicTokenCounterhaystack_integrations.token_counters.anthropic调用POST /v1/messages/count_tokens精确统计输入 token其中 Foundry 与 Vertex 两个生成器都直接继承自AnthropicChatGenerator因此继承并重用了后者的工具调用、流式与序列化能力差异仅在于客户端目标端点与认证方式。AnthropicChatGenerator直连 Claude 的对话生成器AnthropicChatGenerator使用 Anthropic 的大语言模型完成对话补全输入输出均遵循 Haystack 统一的 ChatMessage 数据格式user、system、assistant、tool四种角色由 ChatRole 枚举定义并支持包含文本与图像的多模态输入。最小可用示例from haystack_integrations.components.generators.anthropic import ( AnthropicChatGenerator, ) from haystack.dataclasses import ChatMessage generator AnthropicChatGenerator( generation_kwargs{ max_tokens: 1000, temperature: 0.7, }, ) messages [ ChatMessage.from_system( You are a helpful, respectful and honest assistant ), ChatMessage.from_user(Whats Natural Language Processing?), ] print(generator.run(messagesmessages))该组件在 Pipeline 中最常见的位置是接在ChatPromptBuilder之后运行时的必填输入为messagesChatMessage列表输出键为replies模型回复的ChatMessage列表。支持的模型清单SUPPORTED_MODELS组件内置了一份非穷举的模型白名单用于在初始化时做基础校验SUPPORTED_MODELS: list[str] [ claude-opus-4-6, claude-sonnet-4-6, claude-haiku-4-5-20251001, claude-sonnet-4-5-20250929, claude-opus-4-5-20251101, claude-opus-4-1-20250805, claude-sonnet-4-20250514, claude-opus-4-20250514, claude-3-haiku-20240307, ]默认模型为claude-sonnet-4-5。清单之外仍可能有可用模型完整列表请以 Anthropic 官方模型总览为准从当前列表可以看出该集成覆盖了 Opus / Sonnet / Haiku 三代模型家族。初始化参数详解__init__( api_key: Secret Secret.from_env_var(ANTHROPIC_API_KEY), model: str claude-sonnet-4-5, streaming_callback: StreamingCallbackT | None None, generation_kwargs: dict[str, Any] | None None, ignore_tools_thinking_messages: bool True, tools: ToolsType | None None, anthropic_server_tools: list[dict[str, Any]] | None None, *, timeout: float | None None, max_retries: int | None None ) - Noneapi_keySecretAnthropic API 密钥默认从ANTHROPIC_API_KEY环境变量读取。Haystack 的 Secret 机制支持Secret.from_env_var(...)与Secret.from_token(...)两种注入方式推荐环境变量方案。modelstr要使用的模型名称。streaming_callbackStreamingCallbackT | None流式回调收到新 token 时被调用回调函数接收StreamingChunk作为参数。generation_kwargsdict[str, Any] | None发送给 Anthropic 端点的其他生成参数全部透传。ignore_tools_thinking_messagesboolAnthropic 在工具函数调用解析时会先产生一段思维链消息再返回真正的函数名与参数。设为True默认时组件会在检测到工具调用后丢弃这类 thinking 消息。toolsToolsType | None模型可用的 Tool 与/或 Toolset 对象列表也可传单个 Toolset每个工具必须有唯一名称。anthropic_server_toolslist[dict[str, Any]] | None直接传给 API 的 Anthropic 服务端原生工具列表例如网页搜索{type: web_search_20250305}、代码执行工具等由服务商托管的能力。timeoutfloat | NoneAnthropic 客户端调用超时不设置则使用 Anthropic 客户端默认值。max_retriesint | None失败请求的最大重试次数不设置则使用 Anthropic 客户端默认值。generation_kwargs 支持的参数范围所有与anthropic.Message.create兼容的参数均可通过generation_kwargs传入官方文档列出的受支持键包括参数含义system传给模型的系统消息max_tokens最大生成 token 数metadata传给模型的元数据字典stop_sequences模型应停止生成的字符串列表temperature采样温度top_p核采样nucleus sampling的 top_p 值top_ktop-k 采样的 k 值extra_headers传给模型的额外请求头例如用于 beta 功能开关thinking扩展思维参数字典其中budget_tokens必须小于max_tokensoutput_config传给模型的输出配置选项字典其中thinking与output_config是面向 Claude 4 系列推理与结构化输出能力的较新键extra_headers则是开启 Prompt Caching 等 beta 功能的关键入口见下文。多模态输入图文混合组件同时支持文本与图像输入。图像通过 ImageContent 数据类封装可从本地文件或 URL 加载from haystack.dataclasses import ChatMessage, ImageContent image_content ImageContent.from_file_path(path/to/image.jpg) messages [ ChatMessage.from_user( content_parts[Whats in this image?, image_content] ) ] generator AnthropicChatGenerator() result generator.run(messages)从 image_content.py 的源码可见ImageContent.from_file_path底层实际调用了ImageFileToImageContent转换器完成文件读取与 base64 编码此外还提供from_url下载后转 base64内部使用LinkContentFetcher以及可选的size缩放参数以控制传输体积。run 与 run_asyncrun( messages: list[ChatMessage] | str, streaming_callback: StreamingCallbackT | None None, generation_kwargs: dict[str, Any] | None None, tools: ToolsType | None None, ) - dict[str, list[ChatMessage]]messages输入消息列表如果直接传入字符串会被自动包装成一条user角色的ChatMessage这也是AnthropicGenerator被废弃后推荐迁移到本组件的原因之一。streaming_callback本次调用覆盖初始化的流式回调。generation_kwargs与初始化时传入的generation_kwargs按 key 合并——本次调用传入的键优先仅在初始化时设置的键会被保留。tools本次调用覆盖初始化时的工具配置每个工具名称必须唯一。返回值字典键replies对应模型回复的ChatMessage列表。run_async是run的异步版本签名与合并语义完全一致适用于异步 Pipeline 或 Web 服务处理器中。工具函数调用支持tools参数接受三种灵活配置形态Tool 对象列表逐个传入独立工具单个 Toolset整体传入一个工具集混合形态在同一列表中混排多个 Toolset 与独立 Tool。from haystack.tools import Tool, Toolset from haystack_integrations.components.generators.anthropic import AnthropicChatGenerator # 创建独立工具 weather_tool Tool( nameweather, descriptionGet weather info, parameters..., function... ) news_tool Tool( namenews, descriptionGet latest news, parameters..., function... ) # 将相关工具归组为一个 toolset math_toolset Toolset([add_tool, subtract_tool, multiply_tool]) # 混合传递 Toolset 与 Tool generator AnthropicChatGenerator( tools[math_toolset, weather_tool, news_tool] )这样既能将相关工具组织成逻辑分组也能按需附带独立工具。Haystack 侧的Tool/Toolset定义可参考 haystack/tools/tool.py 与 haystack/tools/toolset.py。工具调用在ChatMessage中以ToolCall模型发起的调用见 chat_message.py与ToolCallResult工具执行结果见 chat_message.py两种内容部件流转二者通过origin字段关联。流式输出Streaming向streaming_callback传入回调即可逐 token 接收输出。官方推荐直接使用内置的print_streaming_chunk打印文本 token 与工具事件工具调用与工具结果from haystack.components.generators.utils import print_streaming_chunk # 任何 Generator 或 ChatGenerator 均可配置流式回调 component SomeGeneratorOrChatGenerator(streaming_callbackprint_streaming_chunk) # ChatGenerator 传入消息列表 # from haystack.dataclasses import ChatMessage # component.run([ChatMessage.from_user(Your question here)]) # 非 chat 的 Generator 传入 prompt # component.run({prompt: Your prompt here})需要注意流式只支持单个响应若提供方支持多候选需显式设置n1。除非你需要特定传输通道如 SSE/WebSocket或自定义 UI 格式否则优先使用print_streaming_chunk。Prompt Caching长上下文降本提速Prompt Caching 允许你把大段文本如代码库上下文、长文档只发送一次后续请求直接引用缓存从而降低成本并缩短响应时间特别适合编码助手类应用。启用方式是在generation_kwargs中携带 beta 头并在系统消息的meta中标记缓存控制from haystack_integrations.components.generators.anthropic import AnthropicChatGenerator from haystack.dataclasses import ChatMessage from haystack.utils import Secret generation_kwargs {extra_headers: {anthropic-beta: prompt-caching-2024-07-31}} claude_llm AnthropicChatGenerator( api_keySecret.from_env_var(ANTHROPIC_API_KEY), generation_kwargsgeneration_kwargs, ) system_message ChatMessage.from_system( Replace with some long text documents, code or instructions ) system_message.meta[cache_control] {type: ephemeral} messages [ system_message, ChatMessage.from_user(A query about the long text for example), ] result claude_llm.run(messages) # 后续再次调用复用同一 system_message 即可命中缓存 messages [ system_message, ChatMessage.from_user(Another query about the long text etc), ] result claude_llm.run(messages)这里的关键是system_message.meta[cache_control] {type: ephemeral}Haystack 的ChatMessage自带meta元数据字典序列化时该标记会随请求一起传给 Anthropic从而完成缓存标记。AnthropicFoundryChatGenerator走 Azure Foundry 的 ClaudeAnthropicFoundryChatGenerator是AnthropicChatGenerator的轻量子类——请求与响应结构仍符合 Anthropic Messages API但流量不再流向api.anthropic.com而是经由你的 Azure Foundry 资源。适合组织统一采用 Azure 托管模型计费、网络、合规但仍希望使用 Claude 的场景无此需求时直接使用AnthropicChatGenerator即可。使用前提拥有启用 Foundry 的 Azure 订阅并在 Foundry 资源中部署了所需的 Anthropic 模型。初始化参数与认证方式__init__( *, api_key: Secret | None Secret.from_env_var(ANTHROPIC_FOUNDRY_API_KEY, strictTrue), resource: str | None None, endpoint: str | None None, model: str claude-sonnet-4-5, streaming_callback: Callable[[StreamingChunk], None] | None None, generation_kwargs: dict[str, Any] | None None, ignore_tools_thinking_messages: bool True, tools: ToolsType | None None, anthropic_server_tools: list[dict[str, Any]] | None None, timeout: float | None None, max_retries: int | None None, azure_ad_token_provider: Callable[[], str] | None None ) - None凭据三选一ANTHROPIC_FOUNDRY_API_KEY环境变量推荐api_key初始化参数配合 Haystack Secret APISecret.from_token(your-api-key-here)azure_ad_token_provider可调用对象按需返回新的 Azure AD token适用于 Entra ID / 托管标识这类不适合静态密钥的场景此时api_key可传None。端点二选一必须提供其一resource或ANTHROPIC_FOUNDRY_RESOURCE环境变量Foundry 资源短名称用于推导 URLendpoint完整端点 URL例如https://your-resource.openai.azure.com/anthropic适合自定义域名或非标准路由。其余参数model、streaming_callback、generation_kwargs、ignore_tools_thinking_messages、tools、anthropic_server_tools、timeout、max_retries语义与AnthropicChatGenerator一致generation_kwargs同样支持system、max_tokens、temperature、top_p、top_k、stop_sequences、metadata、extra_headers等键。Foundry 变体当前仅支持文本输入。使用示例与返回结构from haystack_integrations.components.generators.anthropic import AnthropicFoundryChatGenerator from haystack.dataclasses import ChatMessage from haystack.utils import Secret messages [ChatMessage.from_user(Whats Natural Language Processing?)] client AnthropicFoundryChatGenerator( modelclaude-sonnet-4-5, api_keySecret.from_env_var(ANTHROPIC_FOUNDRY_API_KEY), resourcemy-resource, ) response client.run(messages) print(response)官方文档给出的实际返回结构如下注意replies中的_meta字段包含模型名、索引、结束原因与 token 用量 {replies: [ChatMessage(_roleChatRole.ASSISTANT: assistant, _content[TextContent(text Natural Language Processing (NLP) is a field of artificial intelligence that focuses on enabling computers to understand, interpret, and generate human language. It involves developing techniques and algorithms to analyze and process text or speech data, allowing machines to comprehend and communicate in natural languages like English, Spanish, or Chinese.)], _nameNone, _meta{model: claude-sonnet-4-5, index: 0, finish_reason: end_turn, usage: {input_tokens: 15, output_tokens: 64}})]}可用模型取决于 Azure 部署SUPPORTED_MODELS: list[str] [ claude-opus-4-6, claude-sonnet-4-6, claude-sonnet-4-5, claude-opus-4-5, claude-opus-4-1, claude-haiku-4-5, ]与直连版本不同Foundry 变体的清单不含日期后缀且实际可用模型完全取决于你的 Azure Foundry 资源中部署了哪些模型。model参数对应 Foundry 中的部署名deployment name。工具、流式与异步工具与AnthropicChatGenerator相同的tools参数Tool 列表 / 单个 Toolset / 混合run()时传入的工具会覆盖初始化配置。流式streaming_callbackprint_streaming_chunk即可逐 token 打印。异步run_async自动可用适合异步 Pipeline 或 Web handlerimport asyncio from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.anthropic import ( AnthropicFoundryChatGenerator, ) async def main(): generator AnthropicFoundryChatGenerator(resourcemy-resource) result await generator.run_async([ChatMessage.from_user(Hello!)]) print(result[replies][0].text) asyncio.run(main())AnthropicVertexChatGenerator走 GCP Vertex AI 的 ClaudeAnthropicVertexChatGenerator同样是AnthropicChatGenerator的子类通过 Anthropic Vertex AI API 端点调用 ClaudeOpus、Sonnet、Haiku 等模型家族。使用前提拥有启用了 Vertex AI 的 GCP 项目并在 Vertex AI Model Garden 中激活目标 Anthropic 模型发起请求前可能需要先执行gcloud auth login完成 GCP 认证。初始化参数__init__( region: str, project_id: str, model: str claude-sonnet-420250514, streaming_callback: Callable[[StreamingChunk], None] | None None, generation_kwargs: dict[str, Any] | None None, ignore_tools_thinking_messages: bool True, tools: ToolsType | None None, anthropic_server_tools: list[dict[str, Any]] | None None, *, timeout: float | None None, max_retries: int | None None ) - Noneregionstr模型部署区域默认us-central1。project_idstr模型所在的 GCP 项目 ID。文档同时指出若二者均传None组件会回退到PROJECT_ID与REGION环境变量见 anthropicvertexchatgenerator.mdx。认证走 GCP 应用默认凭据ADC不接收api_key参数这与 Foundry 变体显著不同。generation_kwargs支持的键与直连版本一致system、max_tokens、metadata、stop_sequences、temperature、top_p、top_k、extra_headers。服务端工具限制在 Vertex AI 上仅支持基础网页搜索工具{type: web_search_20250305}带动态过滤的网页搜索、网页抓取与代码执行工具均不受支持。当前仅支持文本输入模态。使用示例Vertex 变体的模型名带版本日期后缀from haystack_integrations.components.generators.anthropic import AnthropicVertexChatGenerator from haystack.dataclasses import ChatMessage messages [ChatMessage.from_user(Whats Natural Language Processing?)] client AnthropicVertexChatGenerator( modelclaude-sonnet-420250514, project_idyour-project-id, regionyour-region ) response client.run(messages) print(response)返回结构同样为{replies: [ChatMessage(...)]}_meta中携带model、index、finish_reason与usage输入/输出 token 数。可用模型清单SUPPORTED_MODELS: list[str] [ claude-opus-4-6, claude-sonnet-4-6, claude-sonnet-4-520250929, claude-sonnet-420250514, claude-opus-4-520251101, claude-opus-4-120250805, claude-opus-420250514, claude-haiku-4-520251001, ]同样为非穷举清单完整可用模型 ID 需以 Claude on Vertex AI 的模型可用性页面为准。Prompt Caching 用法与直连版本完全一致extra_headers带anthropic-beta: prompt-caching-2024-07-31头 system_message.meta[cache_control]标记。AnthropicTokenCounter精确、模型相关的 token 计数AnthropicTokenCounter调用 Anthropic 的POST /v1/messages/count_tokens端点对指定 Claude 模型的ChatMessage与可选工具 schema 统计输入 token。该端点返回精确计数且不生成响应因此不产生生成费用。由于每次计数都发起远程 API 调用它需要 API 密钥且会引入网络延迟适合需要精确、按模型计数的场景如 Agent 上下文压缩。需要本地估算时可改用ApproximateTokenCounter或TiktokenCounter参考 docs-website/docs/token-counters。初始化与用法__init__( model: str, *, api_key: Secret Secret.from_env_var(ANTHROPIC_API_KEY), timeout: float | None None, max_retries: int | None None ) - Nonemodelstr必填用于分词的目标模型。token 计数是模型相关的必须针对实际将用于生成的模型计数。api_keySecret默认读取ANTHROPIC_API_KEY环境变量也可显式传入 Secret。timeout/max_retries底层 Anthropic 客户端的 HTTP 超时与失败重试次数。from haystack.dataclasses import ChatMessage from haystack_integrations.token_counters.anthropic import AnthropicTokenCounter counter AnthropicTokenCounter(modelclaude-sonnet-4-5) messages [ ChatMessage.from_system(You are a helpful assistant.), ChatMessage.from_user(How many tokens is this?), ] token_count counter.count(messages) print(token_count)显式配置密钥与客户端参数from haystack.utils import Secret counter AnthropicTokenCounter( modelclaude-sonnet-4-5, api_keySecret.from_env_var(MY_ANTHROPIC_API_KEY), timeout30.0, max_retries3, )count 方法与工具 schema 计数count(messages: list[ChatMessage], tools: ToolsType | None None) - intmessages要计数的ChatMessage列表。tools可选的 Tool 列表其 schema 也会被计入 token 消耗无内容可计时应返回0。token_count counter.count(messages, tools[search_tool])生命周期管理计数器会在首次调用count()时创建 API 客户端如需在应用启动阶段提前创建可显式调用warm_up()。使用完毕后调用close()释放客户端底层 HTTP 资源counter.warm_up() ... counter.close()非文本内容与压缩场景Anthropic 将图片与 PDF 文件也计入请求 token因此计数器对它们执行精确测量而非估算支持与AnthropicChatGenerator相同的内容类型JPEG、PNG、GIF、WebP 图片及application/pdf文件其他 MIME 类型会直接报错而不是估算。在 Agent 场景中可将计数器传给CompactionHook用 Claude 自身的分词器来评估 Agent 对话是否达到上下文窗口上限from haystack.hooks.compaction import CompactionHook, SlidingWindowCompactor compaction_hook CompactionHook( compactorSlidingWindowCompactor(), context_window200_000, token_counterAnthropicTokenCounter(modelclaude-sonnet-4-5), )注意该 Hook 会在 Agent 每一步都执行计数因此每次压缩检查都会产生一次 API 往返开销。组件生命周期与序列化三个生成器与计数器都实现了统一的 Haystack 组件协议方法行为warm_up()创建同步客户端生成器/初始化客户端计数器warm_up_async()创建异步客户端仅生成器close()关闭同步客户端并释放 HTTP 资源close_async()关闭异步客户端仅生成器to_dict()将组件序列化为字典便于保存与恢复from_dict(data)从字典反序列化恢复组件实例to_dict/from_dict使组件可以嵌入 Haystack 的 YAML/JSON 序列化体系配合 haystack/marshal/yaml.py 可实现完整的 Pipeline 定义持久化。组装进 Pipeline完整示例三个生成器在 Pipeline 中的用法完全一致——接在ChatPromptBuilder之后通过连接传递模板化消息。以直连版本为例from haystack import Pipeline from haystack.components.builders import ChatPromptBuilder from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.anthropic import AnthropicChatGenerator from haystack.utils import Secret pipe Pipeline() pipe.add_component(prompt_builder, ChatPromptBuilder()) pipe.add_component( llm, AnthropicChatGenerator(Secret.from_env_var(ANTHROPIC_API_KEY)), ) pipe.connect(prompt_builder, llm) country Germany system_message ChatMessage.from_system( You are an assistant giving out valuable information to language learners., ) messages [ system_message, ChatMessage.from_user(Whats the official language of {{ country }}?), ] res pipe.run( data{ prompt_builder: { template_variables: {country: country}, template: messages, }, }, ) print(res)Foundry 与 Vertex 变体只需把组件替换为AnthropicFoundryChatGenerator(resourcemy-resource)或AnthropicVertexChatGenerator(project_idtest_id, regionus-central1)即可其余 Pipeline 结构保持不变完整示例分别见 anthropicfoundrychatgenerator.mdx 与 anthropicvertexchatgenerator.mdx。迁移提示从 AnthropicGenerator 迁移AnthropicGenerator基于字符串 prompt 的文本补全版本已被标记为废弃并将在未来版本移除官方推荐迁移到AnthropicChatGenerator——后者同样接受纯字符串输入run会自动包装为user角色消息因此迁移成本极低。旧版行为与参数可参考 anthropicgenerator.mdx。延伸阅读完整 API 参考本文依据docs-website/reference/integrations-api/anthropic.md组件使用指南anthropicchatgenerator.mdx、anthropicfoundrychatgenerator.mdx、anthropicvertexchatgenerator.mdxToken 计数指南anthropictokencounter.mdx消息数据模型haystack/dataclasses/chat_message.py含ChatRole、TextContent、ToolCall、ToolCallResult多模态内容模型haystack/dataclasses/image_content.py工具抽象haystack/tools/tool.py、haystack/tools/toolset.py【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考