企业官网建设流程全解析

参数说明：

• messages：list[ChatMessage]或str。传入字符串时会被自动转换为包含一个 user 角色的ChatMessage列表；传入空列表时直接返回空replies（基类run中的短路逻辑，见 openai.py）；
• streaming_callback：流式回调，若提供则本次调用切换为流式模式；
• generation_kwargs：本次调用的额外生成参数，按 key 覆盖初始化值；
• tools：若设置，覆盖初始化时的tools；
• tools_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 逐块生成时就推送给用户，显著降低首字延迟。配置方式有两种：

1. 初始化时配置：streaming_callback=my_callback，此后所有run调用默认走流式；
2. 运行时配置：在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="", flush=True) client = OpenRouterChatGenerator( model="deepseek/deepseek-r1", streaming_callback=my_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 f"Weather in {city}: sunny, 25°C" tool = create_tool_from_function(get_weather) client = OpenRouterChatGenerator( model="openai/gpt-5-mini", tools=[tool], ) response = client.run([ChatMessage.from_user("What's the weather in Berlin?")]) reply = response["replies"][0] print(reply.tool_calls) # 模型产出的 ToolCall 列表

相关行为说明：

• 模型返回的 tool calls 会被解析为ChatMessage上的ToolCall列表（解析逻辑见 openai.py），随后你可以在 Pipeline 中将其路由到真正的工具组件执行；
• tools_strict=True时，基类会将工具 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.py）；
• from_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(model="openai/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的实际执行路径如下：

1. 客户端预热：run首先调用warm_up()（异步路径调用warm_up_async()），若客户端尚未创建，则基于api_base_url、timeout、max_retries、extra_headers等构造OpenAI（或AsyncOpenAI）客户端，HTTP 层通过init_http_client(http_client_kwargs)定制；
2. 消息归一化：_normalize_messages将str输入转换为 user 角色ChatMessage；空消息列表直接返回空结果；
3. 参数合并与组装：_prepare_api_call合并初始化与运行时的generation_kwargs（运行时优先），将ChatMessage列表转换为 OpenAI 字典格式（to_openai_dict_format），并组装model、n、tools、stream、response_format等请求参数；
4. 端点调用：根据是否结构化输出选择chat.completions.create或chat.completions.parse端点（通过openai_endpoint标记分发，见 openai.py）；
5. 响应转换：非流式时，每个choice被_convert_chat_completion_to_chat_message转换为ChatMessage，其中包含文本、tool calls、以及meta（模型名、index、finish_reason、usage、logprobs 等）；流式时则逐块转换为StreamingChunk并最终聚合；
6. 结果校验：_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.py
• ChatMessage/ReasoningContent/StreamingChunk数据类：chat_message.py
• 密钥管理机制：secret-management.mdx
• Agent 工作流集成：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