企业官网建设流程全解析

Haystack 集成 Anthropic Claude 全指南：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 官方 API 参考文档（docs-website/reference_versioned_docs/version-2.18/integrations-api/anthropic.md）整理而成，完整讲解anthropic-haystack集成包中的四个核心类：AnthropicChatGenerator、AnthropicFoundryChatGenerator、AnthropicVertexChatGenerator与AnthropicTokenCounter。读者将掌握如何在 Haystack 中配置 Anthropic API Key 与模型、传递generation_kwargs控制生成参数、实现多模态（文本+图像）对话、接入工具调用（Tool/Toolset/Anthropic 服务端原生工具）、启用流式输出与提示词缓存，以及用官方 token 计数接口精确估算 Agent 上下文开销。

一、集成包概览：anthropic-haystack

AnthropicChatGenerator是anthropic-haystack集成包的核心组件，用于调用 Anthropic 的 Claude 大语言模型完成对话补全（chat completion）。它严格遵循 Haystack 的ChatMessage数据格式作为输入和输出（ChatMessage定义见 haystack/dataclasses/chat_message.py，支持user、system、assistant、tool四种角色，并提供from_user、from_system、from_assistant等工厂方法），同时支持多模态输入——文本与图片可以混合出现在同一条消息中。

在同一集成包中还提供了两个面向云厂商的变体：

• AnthropicFoundryChatGenerator：通过 Azure Foundry 部署调用 Claude 模型；
• AnthropicVertexChatGenerator：通过 Google Cloud Vertex AI 的 Anthropic API 端点调用 Claude 模型。

两者都继承自AnthropicChatGenerator，请求与响应结构与 Anthropic Messages API 保持一致，只是流量经各自的云资源转发。

安装集成包：

pip install anthropic-haystack

二、AnthropicChatGenerator：基础用法与核心 API

2.1 最小可用示例

AnthropicChatGenerator默认从ANTHROPIC_API_KEY环境变量读取密钥，默认模型为claude-sonnet-4-5。最简单的调用方式：

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("What's Natural Language Processing?"), ] print(generator.run(messages=messages))

run()的返回值为dict[str, list[ChatMessage]]，唯一键replies包含模型的回复消息列表。

2.2 多模态输入（文本 + 图片）

AnthropicChatGenerator支持图片输入，通过haystack.dataclasses中的ImageContent与FileContent表示非文本内容（ImageContent定义见 haystack/dataclasses/image_content.py，内置格式到 MIME 类型的映射，覆盖 JPEG、PNG、GIF、WebP、PDF 等）。示例：

from haystack.dataclasses import ChatMessage, ImageContent image_content = ImageContent.from_file_path("path/to/image.jpg") messages = [ ChatMessage.from_user( content_parts=["What's in this image?", image_content] ) ] generator = AnthropicChatGenerator() result = generator.run(messages)

注意：run()既支持list[ChatMessage]，也支持直接传str——字符串会被自动转换为一条user角色的ChatMessage。

2.3 支持的模型列表（SUPPORTED_MODELS）

组件内置一份非穷尽的模型清单（完整清单以 Anthropic 官方模型总览为准），当前版本包含：

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", ]

从源码结构看，SUPPORTED_MODELS作为类级常量存在，用于校验与提示模型名称拼写；实际生成时仍需使用与你 API 账户/云资源中可用的模型 ID。

2.4 构造函数与全部参数解析

AnthropicChatGenerator.__init__的完整签名如下：

__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 ) -> None

各参数含义：

2.5 generation_kwargs：支持的生成参数

generation_kwargs中支持的参数会全部发送给 Anthropic 的 Message API。支持的键包括：

• system：传给模型的系统消息（与ChatMessage.from_system等价但更直接）；
• max_tokens：生成的最大 token 数；
• metadata：传给模型的元数据字典；
• stop_sequences：模型停止生成的字符串列表；
• temperature：采样温度；
• top_p：核采样（nucleus sampling）的 top_p 值；
• top_k：top-k 采样的 k 值；
• extra_headers：额外请求头字典（例如用于启用 beta 功能）；
• thinking：扩展思考（Extended Thinking）参数字典，其中budget_tokens必须小于max_tokens；
• output_config：传给模型的输出配置选项字典。

generation_kwargs既可以放在__init__中，也可以在run()调用时传入。run()中的generation_kwargs与初始化时的配置按键级合并：run()传入的键优先，初始化时设置的键在未被覆盖时保留。

2.6 run 与 run_async

run()与run_async()签名一致：

run( 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角色消息；
• streaming_callback：覆盖初始化时设置的回调；
• generation_kwargs：与初始化配置按键级合并，此处优先；
• tools：若设置，将覆盖初始化时传入的tools参数。

run_async是run的异步版本，适合在异步 Pipeline 或 Web 处理器（如 FastAPI 路由）中使用，用法与run一致，通过await调用。

2.7 生命周期方法：warm_up / close 及其异步版本

• warm_up()：创建同步 Anthropic 客户端（预初始化，避免首次调用时延迟）；
• warm_up_async()：创建异步 Anthropic 客户端；
• close()：关闭同步 Anthropic 客户端，释放底层 HTTP 资源；
• close_async()：关闭异步 Anthropic 客户端。

这四个方法构成了组件完整的生命周期管理能力，推荐在应用启动时warm_up()、结束时close()，避免每次请求重建连接。

2.8 序列化：to_dict / from_dict

• to_dict() -> dict[str, Any]：将组件序列化为字典，用于 Pipeline 的 YAML/JSON 持久化；
• from_dict(data: dict[str, Any]) -> AnthropicChatGenerator：类方法，从字典反序列化出组件实例。

这两个方法保证组件可以被 Haystack 的序列化体系（见 haystack/core/serialization.py）完整保存与恢复。

三、在 Pipeline 中集成 AnthropicChatGenerator

AnthropicChatGenerator最常见的 Pipeline 位置是接在ChatPromptBuilder之后：先由模板构建ChatMessage列表，再交给生成器补全。示例：

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("What's the official language of {{ country }}?"), ] res = pipe.run( data={ "prompt_builder": { "template_variables": {"country": country}, "template": messages, }, }, ) print(res)

四、工具调用（Function Calling）与流式输出

4.1 灵活的工具配置

tools参数支持三种组织方式（相关类型定义见 haystack/tools/）：

• Tool 对象列表：[weather_tool, news_tool]；
• 单个 Toolset：直接传入一个Toolset；
• 混合列表：一个列表中同时包含多个Toolset与独立Tool。

from haystack.tools import Tool, Toolset from haystack_integrations.components.generators.anthropic import AnthropicChatGenerator weather_tool = Tool( name="weather", description="Get weather info", parameters=..., function=... ) news_tool = Tool( name="news", description="Get latest news", parameters=..., function=... ) math_toolset = Toolset([add_tool, subtract_tool, multiply_tool]) generator = AnthropicChatGenerator( tools=[math_toolset, weather_tool, news_tool] # Toolset 与 Tool 混合 )

工具在run()中传入时覆盖初始化配置。AnthropicServerTools则用于 Anthropic 托管/服务端原生工具，例如网页搜索：{"type": "web_search_20250305"}。

4.2 ignore_tools_thinking_messages

Anthropic 的工具（函数调用）解析机制会在返回真正的函数名与参数之前产生一段 "chain of thought" 思考消息。当ignore_tools_thinking_messages=True（默认值）时，组件在检测到工具调用后会丢弃这些思考消息，避免它们混入上下文或污染消息历史。

4.3 流式输出

通过streaming_callback参数启用流式输出，回调接收StreamingChunk参数。Haystack 内置print_streaming_chunk可以打印文本 token 与工具事件（工具调用与工具结果）：

from haystack.components.generators.utils import print_streaming_chunk generator = AnthropicChatGenerator( streaming_callback=print_streaming_chunk, )

流式模式只支持单一响应（若供应商支持多候选需设置n=1）。默认优先使用print_streaming_chunk，仅当需要特定传输通道（如 SSE/WebSocket）或自定义 UI 格式化时才编写自定义回调。

五、提示词缓存（Prompt Caching）

Anthropic 的提示词缓存允许将大段文本块（如代码库上下文、长文档）发送一次后，在后续请求中复用，从而降低成本和响应延迟。启用方式：通过generation_kwargs的extra_headers传入 beta 头，并在需要缓存的ChatMessage的meta中设置cache_control：

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_key=Secret.from_env_var("ANTHROPIC_API_KEY"), generation_kwargs=generation_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)

组件直接运行时或置于 Pipeline 中均适用，特别适合需要完整代码库上下文的编码助手与长文档处理场景。

六、AnthropicFoundryChatGenerator：通过 Azure Foundry 调用 Claude

6.1 适用场景与前置条件

AnthropicFoundryChatGenerator是AnthropicChatGenerator的薄封装子类，让你通过 Azure Foundry 部署调用 Claude 模型（Opus、Sonnet、Haiku 等）。当组织在 Azure 上统一模型托管（计费、网络、合规）但仍需使用 Claude 时适用。使用前需具备：启用 Foundry 的 Azure 订阅，以及在 Foundry 资源中部署好所需的 Anthropic 模型。

6.2 构造参数

__init__( *, api_key: Secret | None = Secret.from_env_var( "ANTHROPIC_FOUNDRY_API_KEY", strict=True ), 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参数配合Secret；或azure_ad_token_provider可调用对象（按需返回新鲜 Azure AD token，适合 Entra ID / 托管身份场景，此时api_key可传None）；
• 端点二选一：resource参数（或ANTHROPIC_FOUNDRY_RESOURCE环境变量，短资源名，用于推导 URL）；或endpoint参数（完整 URL，如"https://your-resource.openai.azure.com/anthropic"，适合自定义域名）。resource与endpoint必须提供其一；
• model：Foundry 中的部署名称，默认claude-sonnet-4-5；
• 其余参数（streaming_callback、generation_kwargs、ignore_tools_thinking_messages、tools、anthropic_server_tools、timeout、max_retries）语义与AnthropicChatGenerator一致。

6.3 支持的模型

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", ]

此列表非穷尽，实际可用性取决于你的 Azure Foundry 资源配置。该变体当前仅支持文本输入模态。

6.4 使用示例

from haystack_integrations.components.generators.anthropic import AnthropicFoundryChatGenerator from haystack.dataclasses import ChatMessage from haystack.utils import Secret messages = [ChatMessage.from_user("What's Natural Language Processing?")] client = AnthropicFoundryChatGenerator( model="claude-sonnet-4-5", api_key=Secret.from_env_var("ANTHROPIC_FOUNDRY_API_KEY"), resource="my-resource", ) response = client.run(messages) print(response)

输出示例（replies中为ChatMessage，_meta携带模型名、索引、结束原因与 token 用量）：

{'replies': [ChatMessage(_role=<ChatRole.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.")], _name=None, _meta={'model': 'claude-sonnet-4-5', 'index': 0, 'finish_reason': 'end_turn', 'usage': {'input_tokens': 15, 'output_tokens': 64}})]}

AnthropicFoundryChatGenerator同样提供warm_up/warm_up_async（分别创建同步/异步 Foundry 客户端）、to_dict/from_dict序列化方法，以及自动接通的run_async（见其 mdx 使用文档 anthropicfoundrychatgenerator.mdx）。在 Pipeline 中的接法与AnthropicChatGenerator相同，只需将AnthropicChatGenerator(...)替换为AnthropicFoundryChatGenerator(resource="my-resource")并连接ChatPromptBuilder即可。

七、AnthropicVertexChatGenerator：通过 Vertex AI 调用 Claude

7.1 前置条件

AnthropicVertexChatGenerator继承自AnthropicChatGenerator，通过 Anthropic Vertex AI API 端点调用 Claude 模型。使用前需要：启用 Vertex AI 的 GCP 项目；在 Vertex AI Model Garden 中激活目标 Anthropic 模型；请求前可能需要通过gcloud auth login完成 GCP 认证。模型 ID 需使用带@日期后缀的格式，如claude-sonnet-4@20250514。

7.2 构造参数

__init__( region: str, project_id: str, model: str = "claude-sonnet-4@20250514", 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 ) -> None

• region（必填）：模型部署的区域，文档默认us-central1；
• project_id（必填）：GCP 项目 ID；
• model：模型名称，默认claude-sonnet-4@20250514；
• 其余参数语义同基类。generation_kwargs支持的键（system、max_tokens、metadata、stop_sequences、temperature、top_p、top_k、extra_headers）与AnthropicChatGenerator完全一致；
• 服务端工具限制：在 Vertex AI 上仅支持基础网页搜索工具{"type": "web_search_20250305"}，不支持动态过滤网页搜索、网页抓取（web fetch）与代码执行工具。

7.3 支持的模型

SUPPORTED_MODELS: list[str] = [ "claude-opus-4-6", "claude-sonnet-4-6", "claude-sonnet-4-5@20250929", "claude-sonnet-4@20250514", "claude-opus-4-5@20251101", "claude-opus-4-1@20250805", "claude-opus-4@20250514", "claude-haiku-4-5@20251001", ]

7.4 使用示例

from haystack_integrations.components.generators.anthropic import AnthropicVertexChatGenerator from haystack.dataclasses import ChatMessage messages = [ChatMessage.from_user("What's Natural Language Processing?")] client = AnthropicVertexChatGenerator( model="claude-sonnet-4@20250514", project_id="your-project-id", region="your-region", ) response = client.run(messages) print(response)

返回结构与 Foundry 变体一致：replies键下为ChatMessage列表，_meta中带有model、index、finish_reason与usage（input/output tokens）。该变体当前仅支持文本输入，同样提供warm_up/warm_up_async/to_dict/from_dict，并支持提示词缓存（通过extra_headers传入anthropic-beta: prompt-caching-2024-07-31并对 system 消息设置meta["cache_control"] = {"type": "ephemeral"}），Pipeline 集成方式与其他两个生成器完全一致。

八、AnthropicTokenCounter：精确计数 Claude 输入 token

8.1 设计原理

AnthropicTokenCounter实现 Haystack 的TokenCounter协议（见 haystack/token_counters/types/protocol.py，协议要求实现count(messages, tools)、to_dict、from_dict），通过 Anthropic 的POST /v1/messages/count_tokens端点对指定 Claude 模型的ChatMessage列表与可选工具 schema 进行精确计数——该端点返回精确数值而不会触发生成，因此不产生生成费用。

因为每次计数都是一次远程 API 调用，需要 API Key 并带来网络延迟。当需要针对 Claude 模型的精确、按模型计数的数值时使用它；如需本地估算，可改用 Haystack 的ApproximateTokenCounter或TiktokenCounter。

8.2 构造参数与用法

__init__( model: str, *, api_key: Secret = Secret.from_env_var("ANTHROPIC_API_KEY"), timeout: float | None = None, max_retries: int | None = None ) -> None

• model（必填）：用于 token 化的 Anthropic 模型。token 计数是模型相关的，必须对打算实际使用的模型进行计数；
• api_key：默认ANTHROPIC_API_KEY环境变量；
• timeout/max_retries：底层 Anthropic 客户端的 HTTP 超时与失败重试次数。

用法示例：

from haystack.dataclasses import ChatMessage from haystack_integrations.token_counters.anthropic import AnthropicTokenCounter counter = AnthropicTokenCounter(model="claude-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)

count()完整签名：count(messages: list[ChatMessage], tools: ToolsType | None = None) -> int，当无可测量内容时返回0。tools参数用于把随消息一起发送的工具 schema 计入 token 消耗。

8.3 客户端生命周期与 HTTP 行为

计数器在首次调用count()时才创建 API 客户端；若希望应用启动时就绪，可显式调用warm_up()（初始化 Anthropic 客户端）；结束使用后调用close()释放底层 HTTP 资源。

8.4 非文本内容的计数策略

与本地估算器（对图片按固定值估算）不同，Anthropic 将图片与 PDF 文件作为请求的一部分计数，因此AnthropicTokenCounter对它们进行精确测量。支持与AnthropicChatGenerator相同的内容类型：JPEG、PNG、GIF、WebP 图片与application/pdf文件；其他 MIME 类型会抛出错误而非被估算。

8.5 与 Agent 压缩（Compaction）结合

将计数器传给CompactionHook（见 haystack/hooks/compaction/），即可用 Claude 同款 tokenizer 来衡量 Agent 的对话上下文窗口：

from haystack.hooks.compaction import CompactionHook, SlidingWindowCompactor compaction_hook = CompactionHook( compactor=SlidingWindowCompactor(), context_window=200_000, token_counter=AnthropicTokenCounter(model="claude-sonnet-4-5"), )

注意：CompactionHook会在 Agent 每一步都执行计数检查，因此每次压缩检查都会产生一次 API 往返开销，需在成本与上下文管理之间权衡。

8.6 序列化

AnthropicTokenCounter同样提供to_dict()（序列化为字典）与from_dict(data)（类方法反序列化），使其配置可随 Pipeline 一起持久化与恢复。

九、三个生成器变体对比与选型建议

选型原则：没有云托管约束时优先AnthropicChatGenerator；组织标准化在 Azure 时选 Foundry 变体；已深度使用 GCP 生态时选 Vertex 变体。

十、延伸阅读

• 组件详细使用指南（含安装、Pipeline 示例、流式与缓存）：anthropicchatgenerator.mdx、anthropicfoundrychatgenerator.mdx、anthropicvertexchatgenerator.mdx；
• TokenCounter 专项文档：anthropictokencounter.mdx 与 token-counters.mdx；
• 当前版本（非 2.18 版本化）的 API 参考：docs-website/reference/integrations-api/anthropic.md；
• 消息数据类与图片内容实现：haystack/dataclasses/chat_message.py、haystack/dataclasses/image_content.py；
• 工具与 TokenCounter 类型体系：haystack/tools/tool.py、haystack/tools/toolset.py、haystack/token_counters/types/protocol.py；
• 密钥管理与序列化基础：haystack/utils/auth.py、haystack/core/serialization.py。

以上内容均可在本仓库对应路径找到原始依据，模型可用性、云资源配额等以你的 Anthropic / Azure / GCP 账户实际配置为准。

【免费下载链接】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