Haystack 集成 Eden AI：通过统一 OpenAI 兼容 API 实现多供应商 Embedding 与 Chat 生成

Haystack 集成 Eden AI：通过统一 OpenAI 兼容 API 实现多供应商 Embedding 与 Chat 生成

【免费下载链接】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 生态中的 Eden AI 集成组件（edenai-haystack），系统讲解EdenAIDocumentEmbedder、EdenAITextEmbedder与EdenAIChatGenerator三个组件的初始化参数、运行契约、序列化方式及流水线集成方案。读完本文，你将掌握如何用一把 Eden AI API Key 在 Haystack 中路由 OpenAI、Mistral、Cohere、Google、Anthropic 等多家供应商的嵌入与对话模型，并搭建出带欧盟数据驻留（EU data residency）特性的 RAG 流水线。

一、集成背景：Eden AI 是什么，为什么要接入 Haystack

Eden AI 是一个统一的多供应商 AI API 网关，通过单个 API Key 即可访问来自 OpenAI、Mistral、Cohere、Google、Jina、Anthropic 等 500+ 模型，并且内置供应商回退（provider fallback）与欧盟数据驻留能力。对于 Haystack 用户而言，它的核心价值在于：

• 一套代码、多模型可选：模型使用provider/model命名约定，例如"openai/text-embedding-3-small"、"mistral/mistral-embed"、"openai/gpt-4o-mini"、"anthropic/claude-sonnet-4-5"，切换供应商只需改一个字符串。
• OpenAI 兼容协议：Eden AI 提供 OpenAI-compatible 端点，因此集成组件可以复用 Haystack 中成熟的 OpenAI 系列组件（OpenAIDocumentEmbedder、OpenAITextEmbedder、OpenAIChatGenerator）的全部配置与调用链路，仅替换api_base_url。
• 主权友好（sovereignty-friendly）：数据默认停留在欧盟区域处理，适合对数据驻留有合规要求的场景。

从源码结构看，三个 Eden 组件都继承自对应的 OpenAI 组件（见 haystack/components/embedders/openai_document_embedder.py、haystack/components/embedders/openai_text_embedder.py、haystack/components/generators/chat/openai.py），因此底层复用 OpenAI Python SDK 的客户端管理与 HTTP 调用机制，本仓库内的 API 参考文档收录于 docs-website/reference_versioned_docs/version-2.19/integrations-api/edenai.md。

二、安装与密钥配置

2.1 安装集成包

Eden AI 集成作为独立包发布，需额外安装：

pip install edenai-haystack

安装后即可从haystack_integrations命名空间导入三个组件：

from haystack_integrations.components.embedders.edenai import EdenAIDocumentEmbedder, EdenAITextEmbedder from haystack_integrations.components.generators.edenai import EdenAIChatGenerator

2.2 API Key 的两种注入方式

三个组件都通过api_key: Secret参数接收密钥，默认从EDENAI_API_KEY环境变量读取。推荐方式与直接注入方式的对比：

# 方式一（推荐）：环境变量，组件默认行为 # 先执行 export EDENAI_API_KEY=<your-key>，然后： embedder = EdenAIDocumentEmbedder(model="openai/text-embedding-3-small") # 方式二：初始化时用 Secret 显式传入 from haystack.utils import Secret embedder = EdenAITextEmbedder( api_key=Secret.from_token("<your-api-key>"), model="openai/text-embedding-3-small", )

使用Secret而非裸字符串的好处在于：序列化（to_dict）时不会把明文密钥写入磁盘，符合 Haystack 的密钥管理最佳实践，详见 docs-website/docs/concepts/secret-management.mdx。

三、EdenAIDocumentEmbedder：批量文档向量化

3.1 组件定位

EdenAIDocumentEmbedder用于为一组Document批量计算 Embedding，并把结果写入每个Document的embedding字段。在索引流水线中，它通常位于 DocumentWriter 之前，即「转换器 → 向量化 → 写入文档库」的标准位置。

3.2 初始化参数全解

来自 docs-website/reference_versioned_docs/version-2.19/integrations-api/edenai.md 的完整签名：

__init__( *, model: str = "openai/text-embedding-3-small", api_key: Secret = Secret.from_env_var("EDENAI_API_KEY"), api_base_url: str | None = "https://api.edenai.run/v3", prefix: str = "", suffix: str = "", batch_size: int = 32, progress_bar: bool = True, meta_fields_to_embed: list[str] | None = None, embedding_separator: str = "\n", timeout: float | None = None, max_retries: int | None = None, http_client_kwargs: dict[str, Any] | None = None ) -> None

参数背后的实现细节：在 haystack/components/embedders/openai_document_embedder.py 的_client_kwargs()中，timeout与max_retries的取值逻辑是「显式传参优先，其次读OPENAI_TIMEOUT/OPENAI_MAX_RETRIES环境变量，最后落到默认值」，Eden 组件继承该机制；文本构造逻辑见同文件_prepare_texts_to_embed()（第 227-241 行），它将prefix + embedding_separator.join(meta_values + [content]) + suffix拼成最终待嵌入文本，meta_fields_to_embed中缺失或为None的字段会被自动跳过。

3.3 支持的模型列表

组件暴露了非穷举的SUPPORTED_MODELS常量，完整清单以 Eden AI 官方模型目录为准：

SUPPORTED_MODELS: list[str] = [ "openai/text-embedding-3-small", "openai/text-embedding-3-large", "mistral/mistral-embed", "cohere/embed-english-v3.0", "google/text-embedding-004", ]

3.4 单独使用与索引流水线

单独使用：

from haystack import Document from haystack_integrations.components.embedders.edenai import EdenAIDocumentEmbedder doc = Document(content="I love pizza!") document_embedder = EdenAIDocumentEmbedder(model="mistral/mistral-embed") result = document_embedder.run([doc]) print(result["documents"][0].embedding) # [0.017020374536514282, -0.023255806416273117, ...]

run方法接收documents: list[Document]，返回documents（已填充embedding的文档列表）与meta（包含模型名与 token 用量统计）。将向量化嵌入索引流水线：

from haystack import Pipeline from haystack.components.converters import TextFileToDocument from haystack.components.writers import DocumentWriter from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack_integrations.components.embedders.edenai import EdenAIDocumentEmbedder document_store = InMemoryDocumentStore() indexing_pipeline = Pipeline() indexing_pipeline.add_component("converter", TextFileToDocument()) indexing_pipeline.add_component( "embedder", EdenAIDocumentEmbedder(model="openai/text-embedding-3-small") ) indexing_pipeline.add_component("writer", DocumentWriter(document_store=document_store)) indexing_pipeline.connect("converter", "embedder") indexing_pipeline.connect("embedder", "writer") indexing_pipeline.run({"converter": {"sources": ["./my_document.txt"]}})

四、EdenAITextEmbedder：查询字符串向量化

4.1 组件定位

EdenAITextEmbedder用于把单个字符串（典型场景是用户查询）转成向量，在查询/RAG 流水线中通常位于 Embedding Retriever 之前。它与EdenAIDocumentEmbedder的分工是：后者处理文档列表并回写Document.embedding，前者处理单条文本并直接返回embedding与meta。

4.2 初始化参数

签名（相对 Document 版本少了batch_size、progress_bar、meta_fields_to_embed、embedding_separator四个批量相关参数）：

__init__( *, model: str = "openai/text-embedding-3-small", api_key: Secret = Secret.from_env_var("EDENAI_API_KEY"), api_base_url: str | None = "https://api.edenai.run/v3", prefix: str = "", suffix: str = "", timeout: float | None = None, max_retries: int | None = None, http_client_kwargs: dict[str, Any] | None = None ) -> None

各参数含义与 Document 版一致：prefix/suffix拼接进待嵌入文本；timeout与max_retries同样遵循「显式参数 →OPENAI_TIMEOUT/OPENAI_MAX_RETRIES环境变量 → 30 秒/5 次」的取值链路（见 haystack/components/embedders/openai_text_embedder.py 的_client_kwargs()）。SUPPORTED_MODELS与 Document 版完全相同。

4.3 单独使用与查询流水线

单独使用：

from haystack.utils import Secret from haystack_integrations.components.embedders.edenai import EdenAITextEmbedder embedder = EdenAITextEmbedder( api_key=Secret.from_token("<your-api-key>"), model="openai/text-embedding-3-small", ) result = embedder.run(text="How can I use the Eden AI embedding models with Haystack?") print(result["embedding"]) # [-0.0015687942504882812, 0.052154541015625, 0.037109375...]

run方法接收text: str（非字符串输入会抛出TypeError），返回{"embedding": [...], "meta": {"model": ..., "usage": ...}}。

将其接入一个端到端语义检索流水线，与EdenAIDocumentEmbedder形成「索引 + 查询」闭环：

from haystack import Pipeline from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.dataclasses import Document from haystack_integrations.components.embedders.edenai import ( EdenAIDocumentEmbedder, EdenAITextEmbedder, ) document_store = InMemoryDocumentStore(embedding_similarity_function="cosine") documents = [ Document(content="My name is Wolfgang and I live in Berlin"), Document(content="I saw a black horse running"), Document(content="Germany has many big cities"), ] document_embedder = EdenAIDocumentEmbedder(model="openai/text-embedding-3-small") documents_with_embeddings = document_embedder.run(documents)["documents"] document_store.write_documents(documents_with_embeddings) query_pipeline = Pipeline() query_pipeline.add_component( "text_embedder", EdenAITextEmbedder(model="openai/text-embedding-3-small") ) query_pipeline.add_component( "retriever", InMemoryEmbeddingRetriever(document_store=document_store) ) query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding") result = query_pipeline.run({"text_embedder": {"text": "Who lives in Berlin?"}}) print(result["retriever"]["documents"][0])

注意：上述示例中索引与查询两侧必须使用同一嵌入模型，否则向量空间不一致会直接导致检索失效。

五、EdenAIChatGenerator：多供应商 Chat 生成

5.1 组件定位

EdenAIChatGenerator通过 Eden AI 的 OpenAI 兼容端点执行对话补全。它继承 Haystack 的OpenAIChatGenerator，把api_base_url指向 Eden AI，同时保留 OpenAI 版本的全部标准配置能力。它输入/输出均为ChatMessage列表，在对话流水线中通常位于 ChatPromptBuilder 之后。

5.2 初始化参数全解

__init__( *, api_key: Secret = Secret.from_env_var("EDENAI_API_KEY"), model: str = "openai/gpt-4o-mini", streaming_callback: StreamingCallbackT | None = None, generation_kwargs: dict[str, Any] | None = None, timeout: int | None = None, max_retries: int | None = None, tools: ToolsType | None = None, tools_strict: bool = False, http_client_kwargs: dict[str, Any] | None = None ) -> None

5.3 单独使用与流式输出

基础对话：

from haystack_integrations.components.generators.edenai import EdenAIChatGenerator from haystack.dataclasses import ChatMessage messages = [ChatMessage.from_user("What's Natural Language Processing?")] client = EdenAIChatGenerator(model="mistral/mistral-large-latest") response = client.run(messages) print(response["replies"][0].text)

启用流式输出，只需传入一个流式回调（Haystack 内置print_streaming_chunk，可即时打印每个 token）：

from haystack_integrations.components.generators.edenai import EdenAIChatGenerator from haystack.components.generators.utils import print_streaming_chunk from haystack.dataclasses import ChatMessage from haystack.utils import Secret generator = EdenAIChatGenerator( api_key=Secret.from_env_var("EDENAI_API_KEY"), model="mistral/mistral-large-latest", streaming_callback=print_streaming_chunk, ) message = ChatMessage.from_user("What's Natural Language Processing? Be brief.") print(generator.run([message]))

5.4 基于网页内容的 RAG 流水线

下面是一个完整的 RAG 示例：抓取 URL 内容 → 转成文档 → 拼进提示词 → 用EdenAIChatGenerator生成回答，完整覆盖了「抓取—转换—提示—生成」四条连接：

from haystack import Pipeline from haystack.components.builders import ChatPromptBuilder from haystack.components.fetchers import LinkContentFetcher from haystack.components.converters import HTMLToDocument from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.edenai import EdenAIChatGenerator fetcher = LinkContentFetcher() converter = HTMLToDocument() prompt_builder = ChatPromptBuilder(variables=["documents"]) llm = EdenAIChatGenerator(model="mistral/mistral-large-latest") message_template = """Answer the following question based on the contents of the article: {{query}}\n Article: {{documents[0].content}} \n """ messages = [ChatMessage.from_user(message_template)] rag_pipeline = Pipeline() rag_pipeline.add_component(name="fetcher", instance=fetcher) rag_pipeline.add_component(name="converter", instance=converter) rag_pipeline.add_component("prompt_builder", prompt_builder) rag_pipeline.add_component("llm", llm) rag_pipeline.connect("fetcher.streams", "converter.sources") rag_pipeline.connect("converter.documents", "prompt_builder.documents") rag_pipeline.connect("prompt_builder.prompt", "llm.messages") question = "What is Eden AI?" result = rag_pipeline.run( { "fetcher": {"urls": ["https://www.edenai.co/"]}, "prompt_builder": { "template_variables": {"query": question}, "template": messages, }, }, ) print(result["llm"]["replies"][0].text)

六、序列化与组件生命周期

三个 Eden 组件均实现to_dict()，用于把组件序列化为字典（例如用于保存/加载流水线定义）。to_dict返回包含初始化参数的字典，其中api_key以Secret形式序列化而非明文，避免密钥落盘。

从基类源码可以确认更多生命周期细节（见 haystack/components/embedders/openai_document_embedder.py）：

• warm_up()/warm_up_async()：惰性创建同步/异步 OpenAI 客户端，调用发生在首次run之前；
• close()/close_async()：释放客户端连接，供流水线结束或组件回收时调用；
• _embed_batch()内部按batch_size分批请求，失败时默认记录日志并跳过该批（raise_on_failure=True时改为抛异常），并累加usage.prompt_tokens/usage.total_tokens到meta。

Eden 组件继承上述全部机制，因此以EdenAIDocumentEmbedder等组件构建的流水线同样支持to_dict序列化、warm_up预热与close释放，行为与原生 OpenAI 组件保持一致。

七、生产实践建议

• 密钥安全：始终通过EDENAI_API_KEY环境变量注入密钥，不要硬编码；如需在初始化时传入，使用Secret.from_token(...)而非裸字符串。
• 超时与重试：无需逐组件传参时，可统一设置OPENAI_TIMEOUT与OPENAI_MAX_RETRIES环境变量，所有 Eden 组件都会自动读取（默认分别为 30 秒与 5 次）。
• 进度条开关：文档 Embedder 的progress_bar在 CI/生产日志环境中建议置为False，保持日志干净。
• 模型一致性：同一 RAG 系统中 Document Embedder 与 Text Embedder 必须使用同一嵌入模型；切换模型后需重新索引全部文档。
• 多供应商容灾：借助generation_kwargs透传 Eden AI 的回退模型参数，可为 Chat 生成配置供应商级 fallback，提升流水线可用性。
• 完整模型清单：SUPPORTED_MODELS仅为常用子集，最新可用模型请查阅 Eden AI 官方模型目录。

通过以上三个组件，Haystack 开发者可以在不修改流水线结构的前提下，自由切换嵌入与生成模型的底层供应商，兼顾模型选型的灵活性、数据驻留的合规性以及单一密钥的运维简洁性。

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