HetznerChatGenerator 接入指南：在 Haystack 中调用 Hetzner Inference API 的开源大模型

HetznerChatGenerator 是 Haystack 生态中连接 Hetzner Inference API 的聊天生成组件，让开发者可以直接在 Haystack 的 Pipeline 与 Agent 工作流中调用运行于欧洲数据中心的 Qwen 等开源权重模型。本文基于仓库中的 API 参考文档（docs-website/reference_versioned_docs/version-2.23/integrations-api/hetzner.md）与配套使用指南（docs-website/docs/pipeline-components/generators/hetznerchatgenerator.mdx），完整讲解该组件的模型支持、全部初始化参数、文本/多模态/流式/工具调用用法与底层实现原理，读完即可独立接入并投入实战。

组件概览：一条 OpenAI 兼容的接入通道

HetznerChatGenerator 的基类是 OpenAIChatGenerator，这是理解其行为的关键：Hetzner Inference API 提供了与 OpenAI 兼容的接口，因此该组件继承了 OpenAI 系组件的全部调用机制——输入输出统一使用 ChatMessage 数据结构，生成参数透传到上游 API，并支持流式回调、工具调用与结构化输出。

该组件的典型定位在 Pipeline 中位于 ChatPromptBuilder 之后，必备初始化参数是 api_key（可通过 HETZNER_API_KEY 环境变量设置），必备运行参数是 messages（ChatMessage 对象列表），输出变量为 replies（ChatMessage 对象列表）。安装包名为 hetzner-haystack。

需要说明的是，Hetzner Inference API 目前处于实验（experimental）阶段，模型列表会随时间变化，使用前建议以 API 的 /v1/models 端点返回为准。

安装与 API Key 配置

通过 pip 安装集成包：

pip install hetzner-haystack

推荐用环境变量配置密钥，组件默认从 HETZNER_API_KEY 读取：

export HETZNER_API_KEY="your-hetzner-inference-token"

也可以不依赖环境变量，而是显式传入 Secret。例如把密钥放在另一个环境变量 MY_HETZNER_TOKEN 中：

from haystack.utils import Secret
from haystack_integrations.components.generators.hetzner import HetznerChatGenerator

client = HetznerChatGenerator(api_key=Secret.from_env_var("MY_HETZNER_TOKEN"))

这利用了 Haystack 的密钥管理（secret management）机制，密钥不会明文出现在代码或序列化结果中。

支持的模型与 SUPPORTED_MODELS

组件通过类属性 SUPPORTED_MODELS 声明当前实验阶段支持的模型：

SUPPORTED_MODELS: list[str] = ['Qwen/Qwen3.6-35B-A3B-FP8', 'Qwen3.8-27B']

模型	说明
Qwen/Qwen3.6-35B-A3B-FP8	默认模型
Qwen3.8-27B	备选模型

两个模型均运行于 Hetzner 欧洲数据中心，均支持 262,144 token 的超长上下文窗口（该数据来自 使用指南），并且都支持文本与图像混合输入。

几点需要注意：

• 模型列表并非固定不变，SUPPORTED_MODELS 只是组件开发时的快照，权威清单以 API 的 /v1/models 端点返回为准；
• 不在列表中的模型并不会被组件拒绝，会被原样透传给 API——只要 Hetzner 侧支持，你依然可以传入列表之外的模型名使用；
• 通过 model 参数指定要使用的模型，例如 HetznerChatGenerator(model="Qwen3.8-27B")。

初始化参数详解

组件构造函数签名如下（来自 API 参考文档）：

__init__(
    *,
    api_key: Secret = Secret.from_env_var("HETZNER_API_KEY"),
    model: str = "Qwen/Qwen3.6-35B-A3B-FP8",
    streaming_callback: StreamingCallbackT | None = None,
    api_base_url: str | None = "https://inference.hetzner.com/api/v1",
    generation_kwargs: dict[str, Any] | None = None,
    tools: ToolsType | None = None,
    timeout: float | None = None,
    max_retries: int | None = None,
    http_client_kwargs: dict[str, Any] | None = None
) -> None

各参数含义如下：

参数	类型	默认值	说明
api_key	Secret	HETZNER_API_KEY 环境变量	Hetzner Inference API 令牌
model	str	"Qwen/Qwen3.6-35B-A3B-FP8"	使用的模型名，见 SUPPORTED_MODELS
streaming_callback	StreamingCallbackT | None	None	收到新流式 token 时调用的回调函数，参数为 StreamingChunk
api_base_url	str | None	"https://inference.hetzner.com/api/v1"	Hetzner Inference API 基础地址
generation_kwargs	dict[str, Any] | None	None	其余生成参数，全部直接透传给 Hetzner 端点
tools	ToolsType | None	None	Tool / Toolset 对象列表（或单个 Toolset），供模型准备函数调用；每个工具名必须唯一
timeout	float | None	None	Hetzner API 调用的超时时间
max_retries	int | None	None	内部错误后的最大重试次数；未设置时回退到 OPENAI_MAX_RETRIES 环境变量，再否则为 5
http_client_kwargs	dict[str, Any] | None	None	用于配置自定义 httpx.Client / httpx.AsyncClient 的关键字参数

从基类 OpenAIChatGenerator 源码 可以印证两个默认值细节：timeout 未设置时回退到 OPENAI_TIMEOUT 环境变量（默认 30 秒），max_retries 未设置时回退到 OPENAI_MAX_RETRIES 环境变量（默认 5 次）——因为该组件直接复用 OpenAI 客户端的构造逻辑，仅替换了 base_url 与 api_key。

generation_kwargs：透传的生成参数

generation_kwargs 中所有参数都会被原样发送到 Hetzner 端点（接口与 OpenAI 兼容，因此 OpenAI 聊天补全的通用参数同样适用），文档明确列出的常用参数包括：

• max_tokens：输出文本的最大 token 数上限；
• temperature：采样温度。越高模型越敢"冒险"，创意型任务可尝试 0.9；答案明确的任务建议用 0（即 argmax 采样）；
• top_p：核采样（nucleus sampling），模型只考虑概率质量占比 top_p 的 token。例如 0.1 表示只考虑概率质量前 10% 的 token；
• stream：是否流式返回部分结果。开启后 token 以纯数据 Server-Sent Events 形式到达，并以 data: [DONE] 结束；
• response_format：JSON Schema 或 Pydantic 模型，用于强制模型输出的结构，输出会始终按该格式校验（除非模型返回工具调用）。注意：流式场景下 response_format 必须传 JSON Schema 而不能是 Pydantic 模型。

generation_kwargs 既可以在 __init__ 传入，也可以在 run 方法运行时传入。从基类源码 openai.py 的 _prepare_api_call 可以看到两者的合并规则：{**self.generation_kwargs, **(generation_kwargs or {})}——运行时的参数按 key 覆盖初始化时的参数，仅在初始化时设置的 key 会被保留。

示例：让模型以更低的随机性生成 JSON 结构输出：

from haystack.dataclasses import ChatMessage
from haystack_integrations.components.generators.hetzner import HetznerChatGenerator

client = HetznerChatGenerator(
    generation_kwargs={
        "temperature": 0.2,
        "max_tokens": 200,
        "response_format": {"type": "json_object"},
    }
)
response = client.run([ChatMessage.from_user("List 3 famous German cities as JSON.")])
print(response["replies"][0].text)

基本用法：独立运行

最简调用方式如下：

from haystack_integrations.components.generators.hetzner import HetznerChatGenerator
from haystack.dataclasses import ChatMessage

messages = [ChatMessage.from_user("What's Natural Language Processing?")]

client = HetznerChatGenerator()
response = client.run(messages)
print(response)

API 参考文档给出了一个实际输出样例，返回结构为 {'replies': [ChatMessage(...)]}：

>>{'replies': [ChatMessage(_content='Natural Language Processing (NLP) is a branch of artificial intelligence
>>that focuses on enabling computers to understand, interpret, and generate human language in a way that is
>>meaningful and useful.', _role=<ChatRole.ASSISTANT: 'assistant'>, _name=None,
>>_meta={'model': 'Qwen/Qwen3.6-35B-A3B-FP8', 'index': 0, 'finish_reason': 'stop',
>>'usage': {'prompt_tokens': 15, 'completion_tokens': 36, 'total_tokens': 51}})]}

注意输出要点：

• 返回字典只有 replies 一个键，值是 ChatMessage 列表；
• 每个消息的 _meta 中携带 model（实际使用的模型名）、index、finish_reason、usage（提示词/补全/总 token 数）等元信息；
• 读取正文用 response["replies"][0].text。

run 方法内部流程（可对照 openai.py 源码）为：先 warm_up() 初始化同步 OpenAI 客户端，把 messages 归一化为 ChatMessage 列表，随后构造 API 参数并发起 chat completions 请求，最后把响应转换为 ChatMessage 列表返回。若 finish_reason 为 length（被截断）或 content_filter（命中内容过滤），组件会打印相应警告日志。

多模态输入：文本与图像混合

Hetzner 服务端模型同时接受图像与文本，因此可以把 ImageContent 作为内容块放进传给 run 的 ChatMessage 中。ImageContent 数据结构包含 base64_image（图像的 base64 字符串）、mime_type、detail、meta 等字段，并提供了 from_file_path 与 from_url 两个便捷构造方法。

使用指南给出了图像 URL 直接转 ImageContent 的完整示例：

from haystack.dataclasses import ChatMessage, ImageContent
from haystack_integrations.components.generators.hetzner import HetznerChatGenerator

image = ImageContent.from_url(
    "https://cdn.hetzner.de/cdn/public/Uploads/Finnland_Luftaufnahme-v2.jpg"
)

client = HetznerChatGenerator()
response = client.run(
    [
        ChatMessage.from_user(
            content_parts=["Describe this image in one sentence.", image]
        )
    ]
)
print(response["replies"][0].text)

从 image_content.py 源码 可以看到 from_url 的底层行为：它先用 LinkContentFetcher 拉取 URL 内容并校验 MIME 类型（非图片会抛错，PDF 也会被拒绝），再通过 ImageFileToImageContent 转换器把图像转为 base64。它还支持 size 参数（等比缩放，减小传输体积）、retry_attempts 与 timeout 参数。如果图像来自本地文件，则用 ImageContent.from_file_path(...)。

流式输出

给 streaming_callback 传入回调即可启用流式输出。Haystack 内置了 print_streaming_chunk 工具函数（位于 haystack/components/generators/utils.py），它会把文本 token、工具调用元数据（函数名与参数增量）以及工具结果实时打印到 stdout：

from haystack.components.generators.utils import print_streaming_chunk
from haystack.dataclasses import ChatMessage
from haystack_integrations.components.generators.hetzner import HetznerChatGenerator

client = HetznerChatGenerator(streaming_callback=print_streaming_chunk)
client.run([ChatMessage.from_user("What are Agentic Pipelines? Be brief.")])

流式回调节点收到的是 StreamingChunk 对象，回调签名即 def callback(chunk: StreamingChunk) -> None。底层实现中，流式与非流式走不同的响应处理分支（源码见 openai.py）：流式场景逐块累积增量并通过 _convert_streaming_chunks_to_chat_message 组装成最终的 ChatMessage。你也可以自己编写回调，例如把增量写入前端 WebSocket 或日志。

工具调用（Function Calling）

将 Tool 对象、Toolset，或二者的混合列表传给 tools 参数，模型即可为这些工具准备调用参数。规则与 OpenAI 系组件一致：

• 每个工具必须有唯一名称（基类初始化时会调用 _check_duplicate_tool_names 校验）；
• 工具调用以 ToolCall 形式出现在 assistant 消息中；
• 可通过 tool 角色的 ChatMessage 把工具执行结果回传给模型，形成多轮工具调用循环。

from haystack.dataclasses import ChatMessage
from haystack.tools import create_tool
from haystack_integrations.components.generators.hetzner import HetznerChatGenerator

def get_current_weather(city: str) -> str:
    """Get the current weather of a city."""
    return f"The weather in {city} is sunny, 22°C."

weather_tool = create_tool(get_current_weather)

client = HetznerChatGenerator(tools=[weather_tool])
response = client.run([ChatMessage.from_user("What's the weather in Berlin?")])
print(response["replies"][0].tool_calls)

配合流式回调时，print_streaming_chunk 也会实时打印工具调用的参数增量，方便调试 Agent 行为。

在 Pipeline 中使用

HetznerChatGenerator 是标准 Haystack 组件，可以无缝接入 Pipeline。使用指南给出了与 ChatPromptBuilder 配合的完整示例：

from haystack import Pipeline
from haystack.components.builders import ChatPromptBuilder
from haystack.dataclasses import ChatMessage
from haystack_integrations.components.generators.hetzner import HetznerChatGenerator

prompt_builder = ChatPromptBuilder()
llm = HetznerChatGenerator()

pipe = Pipeline()
pipe.add_component("builder", prompt_builder)
pipe.add_component("llm", llm)
pipe.connect("builder.prompt", "llm.messages")

messages = [
    ChatMessage.from_system("Give brief answers."),
    ChatMessage.from_user("Tell me about {{city}}"),
]

response = pipe.run(
    data={
        "builder": {"template": messages, "template_variables": {"city": "Nuremberg"}}
    },
)
print(response["llm"]["replies"][0].text)

要点：

• ChatPromptBuilder 输出 prompt，通过 pipe.connect("builder.prompt", "llm.messages") 接到生成器的 messages 输入；
• ChatMessage 支持 system / user / assistant / tool 四种角色（见 chat_message.py 源码），系统提示词与用户模板可以放在同一个模板消息列表中；
• 模板中的 {{city}} 由 template_variables 在运行时填充，管道输出通过 response["llm"]["replies"] 取回。

在 Agent 工作流中，该组件同样可以作为 LLM 内核使用：配合 tools 参数与 Haystack 的 Tool / Toolset 体系，即可构建带工具调用能力的 Agent。

序列化：to_dict 与 Pipeline YAML

组件实现了 to_dict() 方法，用于把组件序列化为字典，从而可以嵌入 Pipeline 的 YAML 配置被持久化与反序列化：

to_dict() -> dict[str, Any]

序列化字典包含 model、streaming_callback、api_base_url、generation_kwargs、api_key、timeout、max_retries、tools、http_client_kwargs 等初始化参数（逻辑继承自基类的 default_to_dict，见 openai.py）。其中：

• api_key 以 Secret 形式序列化，不会泄露明文；
• 可调用对象（如流式回调）会序列化为其引用路径，反序列化时再还原；
• generation_kwargs 中的 Pydantic 模型形式的 response_format 会被转换成严格的 JSON Schema 后保存。

因此你可以放心地把它写进 Pipeline YAML，例如：

components:
  llm:
    type: haystack_integrations.components.generators.hetzner.HetznerChatGenerator
    init_parameters:
      model: "Qwen/Qwen3.6-35B-A3B-FP8"
      generation_kwargs:
        temperature: 0.7

底层实现原理：复用 OpenAI 客户端的继承设计

理解 HetznerChatGenerator 的最佳方式是把目光投向它的基类 OpenAIChatGenerator，它本质上是"把 OpenAI 客户端指向 Hetzner 的 base URL"：

1. 客户端构造：warm_up() 时用 api_key、base_url（即 api_base_url）、timeout、max_retries 构造 OpenAI 同步客户端，其中 timeout/max_retries 的默认值取自 OPENAI_TIMEOUT / OPENAI_MAX_RETRIES 环境变量；
2. 消息转换：ChatMessage 通过 to_openai_dict_format() 转换为 OpenAI API 兼容格式（见 chat_message.py），多模态图像内容也会被编码进消息结构；
3. 参数组装：_prepare_api_call 合并 init 与 run 两个层级的 generation_kwargs，处理 n > 1 与流式互斥、response_format 与 parse 端点选择、工具定义的扁平化与去重；
4. 响应转换：非流式响应逐 choice 转成 ChatMessage（含 tool_calls 解析与 JSON 容错）；流式响应把增量块累积成 StreamingChunk 再聚合成完整消息；
5. 元信息透传：model、finish_reason、usage 等 API 元数据写入 ChatMessage.meta，方便后续观测与 token 计量。

这种继承设计意味着：只要你的下游逻辑按 OpenAI 系组件的接口约定编写，切换到 Hetzner 托管的开源模型几乎零成本，只需替换组件类型与 API Key。从当前仓库源码看，基类还提供了 run_async 异步调用与 close/close_async 资源释放方法，可作为扩展点参考。

实战要点小结

• 密钥安全：优先用 HETZNER_API_KEY 环境变量，避免在代码或配置中硬编码令牌；
• 模型确认：实验阶段模型列表会变化，上线前用 /v1/models 端点核对可用模型；列表外的模型名不会被拒绝，但可用性以 API 为准；
• 超长上下文：两个模型都支持 262,144 token 的上下文窗口，适合长文档 RAG 与长对话场景；
• 多模态能力：Hetzner 模型天然支持图像，ImageContent.from_url / from_file_path 提供了最便捷的图片接入方式；
• 结构化输出：需要 JSON 输出时使用 response_format，流式场景务必传 JSON Schema 而非 Pydantic 模型；
• 可观测性：输出 meta 中的 usage 字段可直接用于 token 计量，finish_reason 为 length 时记得调大 max_tokens。