在 Haystack 中使用 OpenRouterChatGenerator：统一接入多模型 Chat Completion 的实战指南
发布时间：2026/9/14 16:21:02 作者：尧图编辑部 阅读量：1,286
在 Haystack 中使用 OpenRouterChatGenerator统一接入多模型 Chat Completion 的实战指南【免费下载链接】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导读OpenRouter 是一个统一的多模型网关通过单个 API 即可调用 DeepSeek、Claude、GPT 等大量第三方托管模型。Haystack 提供了OpenRouterChatGenerator组件让开发者可以像使用原生OpenAIChatGenerator一样在 Haystack Pipeline 中无缝接入 OpenRouter 的 Chat Completion 端点并直接获得流式输出、工具调用、结构化输出与推理内容reasoning/thinking等能力。读完本文你将掌握该组件的完整参数体系、同步/异步调用方式、流式与工具调用配置以及它在 Haystack Pipeline 中的集成方法与底层实现原理。本文对应仓库文档OpenRouter 集成 API 参考。一、组件概览基于 OpenAIChatGenerator 的轻量扩展OpenRouterChatGenerator的完整限定名是haystack_integrations.components.generators.openrouter.chat.chat_generator.OpenRouterChatGenerator其基类是 Haystack 核心库中的OpenAIChatGenerator见 openai.py。这意味着它天然继承了 Haystack 对 OpenAI Chat Completion 协议的全部适配逻辑——客户端生命周期管理、消息格式转换、流式分块解析、工具调用序列化等而 OpenRouter 的端点恰好兼容 OpenAI 的请求/响应结构因此这套适配可以开箱即用。组件使用ChatMessage数据结构定义于 chat_message.py组织输入与输出保证在多轮对话、Agent 工作流等场景中上下文连贯。其官方文档列出的核心特性包括主兼容性与 OpenRouter Chat Completion 端点完全兼容流式支持支持从 OpenRouter Chat Completion 端点接收流式响应高可定制性支持 OpenRouter Chat Completion 端点支持的所有参数通过generation_kwargs透传推理内容支持可提取支持推理的模型如 DeepSeek R1、启用扩展思考的 Claude产出的 reasoning/thinking 内容存放在ChatMessage的ReasoningContent字段中。注意推理内容仅在非流式请求下被捕获。需要说明的是该集成组件本体由haystack_integrations集成包提供对应导入路径haystack_integrations.components.generators.openrouter而其依赖的基类、ChatMessage/ReasoningContent/StreamingChunk等数据类均由本仓库的 Haystack 核心库实现这也是本文能够结合仓库源码深入讲解其原理的原因。二、环境准备与 API Key 配置安装 Haystack 与 OpenRouter 集成在项目中安装 Haystack 核心库并安装包含haystack_integrations.components.generators.openrouter模块的对应集成包然后通过文档中的导入路径引入组件from haystack_integrations.components.generators.openrouter import ( OpenRouterChatGenerator, ) from haystack.dataclasses import ChatMessage配置 API Key组件默认从环境变量OPENROUTER_API_KEY读取密钥签名如下api_key: Secret Secret.from_env_var(OPENROUTER_API_KEY)Haystack 的Secret机制详见 secret-management 概念文档支持环境变量、文件、显式字符串等多种来源可避免在代码与配置文件中明文暴露密钥。启动应用前设置环境变量即可export OPENROUTER_API_KEYyour_openrouter_api_key三、快速上手最小可用示例官方文档给出的最小示例同时演示了推理内容的读取from haystack_integrations.components.generators.openrouter import ( OpenRouterChatGenerator, ) from haystack.dataclasses import ChatMessage messages [ChatMessage.from_user(Whats Natural Language Processing?)] client OpenRouterChatGenerator( modeldeepseek/deepseek-r1, generation_kwargs{reasoning: {effort: high}}, ) response client.run(messages) print(response[replies][0].reasoning) # Access reasoning content print(response[replies][0].text) # Access final answer要点拆解model指定 OpenRouter 上的模型 ID格式通常为厂商/模型名如deepseek/deepseek-r1支持的模型清单以 OpenRouter 模型列表 为准generation_kwargs在初始化时传入这里配置了{reasoning: {effort: high}}以让 DeepSeek R1 输出高强度的推理过程run(messages)返回的字典只包含一个键replies值为list[ChatMessage]每个回复ChatMessage上reasoning属性返回ReasoningContent | Nonetext属性返回最终答案文本——两者在 chat_message.py 中分别由reasoningproperty第 408 行起与文本内容 getter 提供ReasoningContent数据类定义于第 169 行起包含reasoning_text与extra字段并支持to_dict/from_dict序列化。四、初始化参数全解析init完整构造签名如下__init__( *, api_key: Secret Secret.from_env_var(OPENROUTER_API_KEY), model: str openai/gpt-5-mini, streaming_callback: StreamingCallbackT | None None, api_base_url: str | None https://openrouter.ai/api/v1, generation_kwargs: dict[str, Any] | None None, tools: ToolsType | None None, timeout: float | None None, extra_headers: dict[str, Any] | None None, max_retries: int | None None, http_client_kwargs: dict[str, Any] | None None ) - None各参数说明参数类型与默认值说明api_keySecret默认读取OPENROUTER_API_KEYOpenRouter API 密钥modelstr默认openai/gpt-5-mini要使用的 OpenRouter Chat Completion 模型名称streaming_callbackStreamingCallbackT \| None默认None流式回调函数每收到一个新 token 时被调用回调参数为StreamingChunkapi_base_urlstr \| None默认https://openrouter.ai/api/v1OpenRouter API 基础地址一般无需修改generation_kwargsdict[str, Any] \| None透传给 OpenRouter 端点的其余生成参数详见下一节toolsToolsType \| None可接受Tool对象列表或单个Toolset实例供模型准备函数调用timeoutfloat \| NoneOpenRouter API 调用的超时时间。若未设置回退到OPENAI_TIMEOUT环境变量再回退到 30 秒依据基类_client_kwargs的实现见 openai.pyextra_headersdict[str, Any] \| None附加到请求上的 HTTP 头。可用于传递站点 URL 或标题等元信息以参与 OpenRouter 平台的排名max_retriesint \| None内部错误时重试联系服务商的最大次数。未设置时回退到OPENAI_MAX_RETRIES环境变量再回退到 5同样见 openai.pyhttp_client_kwargsdict[str, Any] \| None用于配置自定义httpx.Client/httpx.AsyncClient的关键字参数字典适合做代理、TLS 定制等高级场景五、generation_kwargs透传 OpenRouter 的全部生成参数generation_kwargs是发挥该组件定制能力的核心入口——文档明确说明所有参数都会被原样发送到 OpenRouter 端点。常用参数如下参数说明max_tokens输出文本的最大 token 数上限temperature采样温度。越高模型越冒险创意型应用可尝试 0.9答案明确的场景用 0等价 argmax 采样top_p核采样nucleus sampling替代温度模型只考虑累计概率质量达top_p的 token如 0.1 表示只考虑概率质量前 10% 的 tokenstream是否流式返回部分进度。若开启token 以>run( messages: list[ChatMessage] | str, streaming_callback: StreamingCallbackT | None None, generation_kwargs: dict[str, Any] | None None, *, tools: ToolsType | None None, tools_strict: bool | None None ) - dict[str, list[ChatMessage]]参数说明messageslist[ChatMessage]或str。传入字符串时会被自动转换为包含一个 user 角色的ChatMessage列表传入空列表时直接返回空replies基类run中的短路逻辑见 openai.pystreaming_callback流式回调若提供则本次调用切换为流式模式generation_kwargs本次调用的额外生成参数按 key 覆盖初始化值tools若设置覆盖初始化时的toolstools_strict是否启用工具调用的严格 Schema 遵循模式开启后模型将严格按工具定义中的parameters字段生成参数但可能增加延迟。返回值{replies: [ChatMessage, ...]}其中replies为模型生成的回复列表。run_async异步run_async( messages: list[ChatMessage] | str, streaming_callback: StreamingCallbackT | None None, generation_kwargs: dict[str, Any] | None None, *, tools: ToolsType | None None, tools_strict: bool | None None ) - dict[str, list[ChatMessage]]run_async是run的异步版本参数与返回值完全一致可用await在异步代码中调用。唯一的差异点是异步场景下流式回调必须是协程coroutine。二者分别使用AsyncOpenAI客户端与同步OpenAI客户端发起请求对应基类的warm_up/warm_up_async初始化见 openai.py适合在 FastAPI 服务、异步 Agent 循环等场景中避免阻塞事件循环。七、流式输出与流式回调流式能力使组件可以在 token 逐块生成时就推送给用户显著降低首字延迟。配置方式有两种初始化时配置streaming_callbackmy_callback此后所有run调用默认走流式运行时配置在run调用中传入streaming_callback仅本次生效。回调函数签名要求接收一个StreamingChunk参数同步模式下回调为普通函数异步模式下为协程可在每个增量块到达时做增量渲染、日志记录或转发。在底层基类将 OpenAI 协议中的ChatCompletionChunk逐块转换为StreamingChunk转换逻辑见 openai.py聚合完成后由_convert_streaming_chunks_to_chat_message汇合成单个ChatMessage返回。from haystack.dataclasses import StreamingChunk def my_streaming_callback(chunk: StreamingChunk) - None: print(chunk.content, end, flushTrue) client OpenRouterChatGenerator( modeldeepseek/deepseek-r1, streaming_callbackmy_streaming_callback, ) response client.run([ChatMessage.from_user(Tell me a short joke)])需要注意开启流式后response[replies]中仍会返回聚合完成的ChatMessage同时每个StreamingChunk的meta中会携带model、finish_reason、received_at、usage等元信息。八、工具调用让模型准备函数调用组件支持 Haystack 的Tool与Toolset体系。将工具列表传给初始化参数tools或run的运行时参数即可from haystack.dataclasses import ChatMessage from haystack.tools import create_tool_from_function def get_weather(city: str) - str: Get the current weather of a city. return fWeather in {city}: sunny, 25°C tool create_tool_from_function(get_weather) client OpenRouterChatGenerator( modelopenai/gpt-5-mini, tools[tool], ) response client.run([ChatMessage.from_user(Whats the weather in Berlin?)]) reply response[replies][0] print(reply.tool_calls) # 模型产出的 ToolCall 列表相关行为说明模型返回的 tool calls 会被解析为ChatMessage上的ToolCall列表解析逻辑见 openai.py随后你可以在 Pipeline 中将其路由到真正的工具组件执行tools_strictTrue时基类会将工具 JSON Schema 递归改造为 OpenAI strict 模式所有 object 设置additionalProperties: false、所有属性列入required以保证模型输出严格符合 Schema——这通过_make_schema_strict实现见 openai.py初始化时传入的工具会在warm_up阶段被预热warm_up_tools并且组件会校验工具名不重复。九、序列化to_dict / from_dict 与 Pipeline 集成组件提供标准的 Haystack 序列化协议to_dict() - dict[str, Any]将组件序列化为字典包含model、api_base_url、generation_kwargs、api_key、timeout、max_retries、tools等全部初始化参数流式回调函数会被序列化为可反序列化的 callable 名称对应基类实现见 openai.pyfrom_dict(data)从字典还原组件反序列化工具与回调。利用这一协议OpenRouterChatGenerator可以无缝嵌入 Haystack Pipeline并被 YAML 化保存与加载from haystack import Pipeline from haystack.components.builders import ChatPromptBuilder from haystack_integrations.components.generators.openrouter import OpenRouterChatGenerator pipe Pipeline() pipe.add_component(prompt_builder, ChatPromptBuilder(template[ {role: user, content: Explain {{topic}} in simple terms} ])) pipe.add_component(llm, OpenRouterChatGenerator(modelopenai/gpt-5-mini)) pipe.connect(prompt_builder.prompt, llm.messages) result pipe.run({prompt_builder: {topic: quantum computing}}) print(result[llm][replies][0].text)在这个典型 RAG/对话 Pipeline 中ChatPromptBuilder将模板渲染成ChatMessage列表OpenRouterChatGenerator接收messages输入并输出replies。你还可以像使用任何生成器组件一样将其接到检索器、Agent见 agent.mdx等上下游组件上。十、底层原理从参数到 ChatMessage 的完整调用链结合 openai.py 源码OpenRouterChatGenerator.run的实际执行路径如下客户端预热run首先调用warm_up()异步路径调用warm_up_async()若客户端尚未创建则基于api_base_url、timeout、max_retries、extra_headers等构造OpenAI或AsyncOpenAI客户端HTTP 层通过init_http_client(http_client_kwargs)定制消息归一化_normalize_messages将str输入转换为 user 角色ChatMessage空消息列表直接返回空结果参数合并与组装_prepare_api_call合并初始化与运行时的generation_kwargs运行时优先将ChatMessage列表转换为 OpenAI 字典格式to_openai_dict_format并组装model、n、tools、stream、response_format等请求参数端点调用根据是否结构化输出选择chat.completions.create或chat.completions.parse端点通过openai_endpoint标记分发见 openai.py响应转换非流式时每个choice被_convert_chat_completion_to_chat_message转换为ChatMessage其中包含文本、tool calls、以及meta模型名、index、finish_reason、usage、logprobs 等流式时则逐块转换为StreamingChunk并最终聚合结果校验_check_finish_reason检查每个回复的finish_reason若为length输出被截断或content_filter被内容过滤器截断则记录 warning提示相应调参见 openai.py。正是这条调用链使得同一套组件逻辑既支持 OpenAI 原生端点也支持兼容该协议的 OpenRouter 网关——OpenRouterChatGenerator只需把api_base_url指向https://openrouter.ai/api/v1即可复用全部能力这也是理解该组件设计哲学的关键。十一、实战要点与注意事项推理内容只在非流式下捕获若需要读取reply.reasoning如 DeepSeek R1 的思考链请勿同时开启流式回调流式场景下推理 token 不会进入ReasoningContent字段默认模型未指定model时默认使用openai/gpt-5-mini请结合 OpenRouter 平台的实际模型列表确认可用性环境变量回退链timeout/max_retries均可通过OPENAI_TIMEOUT/OPENAI_MAX_RETRIES环境变量覆盖这沿袭自基类设计结构化输出response_format可传 Pydantic 模型或 JSON Schema若与流式同时使用建议使用 JSON Schema 形式运行时可覆盖run中传入的generation_kwargs、tools、tools_strict会覆盖初始化值适合多租户、多任务复用同一个组件实例的场景异步优先在高并发 Web 服务中优先使用run_async并提供协程型流式回调避免阻塞事件循环。相关源码与文档导航集成组件 API 参考openrouter.md基类OpenAIChatGenerator完整实现openai.pyChatMessage/ReasoningContent/StreamingChunk数据类chat_message.py密钥管理机制secret-management.mdxAgent 工作流集成agent.mdx【免费下载链接】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），仅供参考
尧图网站开发 