在 Haystack 中集成 Eden AI：统一接入多供应商 Embedding 与 Chat 生成模型

在 Haystack 中集成 Eden AI统一接入多供应商 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/haystackEden AI 是一个统一、兼容 OpenAI 协议的 AI 聚合 API通过单个 API Key 即可路由到 OpenAI、Mistral、Cohere、Google、Anthropic 等多家供应商的数百个模型并内置供应商故障切换provider fallback与欧盟数据驻留EU data residency。Haystack 的edenai-haystack集成提供三个核心组件EdenAIDocumentEmbedder文档批量向量化、EdenAITextEmbedder查询字符串向量化和EdenAIChatGenerator对话生成让你在 Haystack 流水线中一键切换到 Eden AI 网关构建语义检索、RAG 与对话式应用。读完本文你将掌握这三个组件的完整参数语义、单独使用方式、流水线接入方法及其底层继承自 OpenAI 组件的实现原理。本文以 version-2.18 的 Eden AI API 参考文档为主体骨架结合仓库中 EdenAIDocumentEmbedder 组件指南、EdenAITextEmbedder 组件指南、EdenAIChatGenerator 组件指南以及 Haystack 核心库中对应 OpenAI 基类的源码展开讲解。一、集成背景为什么在 Haystack 中使用 Eden AIHaystack 是面向生产级 LLM 应用的开源 AI 编排框架通过组件化流水线对检索retrieval、路由routing、记忆memory与生成generation进行显式控制。Eden AI 集成的作用在于在不改动流水线结构的前提下把 Embedding 与 Chat 生成调用统一转发到 Eden AI 网关。其核心设计思想体现在 API 参考文档对三个组件的“Bases”声明上EdenAIDocumentEmbedder继承自OpenAIDocumentEmbedderEdenAITextEmbedder继承自OpenAITextEmbedderEdenAIChatGenerator继承自OpenAIChatGenerator。也就是说Eden AI 组件本质上复用了 Haystack 官方 OpenAI 组件的能力仅将api_base_url指向 Eden AI 的 OpenAI 兼容端点https://api.edenai.run/v3从而获得单一 API Key 多供应商一次接入即覆盖 OpenAI、Mistral、Cohere、Google、Jina、Anthropic 等供应商的模型provider/model命名约定模型名形如openai/text-embedding-3-small、mistral/mistral-embed、openai/gpt-4o-mini、anthropic/claude-sonnet-4-5欧盟数据驻留适用于对数据主权sovereignty有要求的场景内置供应商回退在某个供应商故障时自动切换便于构建高可用的生产系统。从底层实现看Haystack 核心库中的 OpenAITextEmbedder 与 OpenAIDocumentEmbedder 使用 OpenAI 官方 Python SDK 创建OpenAI/AsyncOpenAI客户端并通过base_url指向自定义端点这正是 Eden AI 集成能够换基座不换代码的机制基础。二、安装与 API Key 配置安装集成包来自独立的haystack-core-integrations仓库发布名为edenai-haystackpip install edenai-haystackEdenAIDocumentEmbedder、EdenAITextEmbedder与EdenAIChatGenerator三个组件默认都从EDENAI_API_KEY环境变量读取密钥例如export EDENAI_API_KEYyour-edenai-api-key也可以显式传入 API Key。推荐使用 Haystack 的Secret工具类避免密钥以明文形式进入流水线序列化结果from haystack.utils import Secret from haystack_integrations.components.embedders.edenai import EdenAIDocumentEmbedder embedder EdenAIDocumentEmbedder( api_keySecret.from_token(your-api-key), modelopenai/text-embedding-3-small, )Secret.from_env_var(EDENAI_API_KEY)是三个组件的默认参数值Secret.from_token(...)则适合在测试脚本或临时环境中直接注入密钥。三、EdenAIDocumentEmbedder批量文档向量化EdenAIDocumentEmbedder用于为一组 Document 计算 Embedding并将结果写入每个 Document 的embedding字段。它应在索引流水线indexing pipeline中、DocumentWriter之前使用若只需对单个字符串如用户查询编码则应使用EdenAITextEmbedder。3.1 支持的模型SUPPORTED_MODELSAPI 参考文档列出的非穷举模型清单如下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, ]该清单只是组件内置提示并非硬性白名单完整可用模型请查阅 Eden AI 模型目录文档中引用自 Eden AI 官方 models catalog。3.2 构造参数详解EdenAIDocumentEmbedder.__init__的完整签名如下__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各参数语义与取值建议参数类型默认值说明modelstropenai/text-embedding-3-smallEden AI 嵌入模型名必须采用provider/model格式如mistral/mistral-embed、cohere/embed-english-v3.0api_keySecretEDENAI_API_KEY环境变量Eden AI API 密钥api_base_urlstr \| Nonehttps://api.edenai.run/v3Eden AI 的 OpenAI 兼容端点基址一般无需修改prefixstr附加到每段待编码文本开头的字符串suffixstr附加到每段待编码文本末尾的字符串batch_sizeint32每次 API 调用同时编码的 Document 数量控制吞吐与请求频率progress_barboolTrue是否显示进度条生产部署建议关闭以保持日志干净meta_fields_to_embedlist[str] \| NoneNone需要连同正文一起编码的元数据字段名列表embedding_separatorstr\n元数据字段与正文拼接时使用的分隔符timeoutfloat \| NoneNoneAPI 调用超时秒数未设置时回退到OPENAI_TIMEOUT环境变量再回退到 30 秒max_retriesint \| NoneNone遇到内部错误后重试 Eden AI 的最大次数未设置时回退到OPENAI_MAX_RETRIES环境变量再回退到 5 次http_client_kwargsdict[str, Any] \| NoneNone用于定制底层httpx.Client/httpx.AsyncClient的关键字参数字典如自定义代理、证书等3.3 单独使用from haystack.dataclasses import Document from haystack_integrations.components.embedders.edenai import EdenAIDocumentEmbedder doc Document(contentI love pizza!) document_embedder EdenAIDocumentEmbedder(modelmistral/mistral-embed) result document_embedder.run([doc]) print(result[documents][0].embedding) # [0.017020374536514282, -0.023255806416273117, ...]run方法的输入为documents: list[Document]输出字典包含两个键documents携带embedding的新 Document 列表与meta模型名与 token 用量等元信息。3.4 在索引流水线中使用将文档嵌入步骤接入标准索引流水线前面接文件转换器、后面接DocumentWriter写入文档存储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(modelopenai/text-embedding-3-small) ) indexing_pipeline.add_component(writer, DocumentWriter(document_storedocument_store)) indexing_pipeline.connect(converter, embedder) indexing_pipeline.connect(embedder, writer) indexing_pipeline.run({converter: {sources: [./my_document.txt]}})3.5 底层实现原理从源码看EdenAIDocumentEmbedder的基类 OpenAIDocumentEmbedder 在run方法中依次完成类型校验仅接受list[Document]→warm_up()惰性创建 OpenAI 客户端 →_prepare_texts_to_embed()组装待编码文本 →_embed_batch()分批调用client.embeddings.create()→ 用dataclasses.replace生成携带 embedding 的新 Document。其中两个值得留意的实现细节元数据拼接逻辑_prepare_texts_to_embed对于meta_fields_to_embed中指定的字段取doc.meta[key]的非空值按embedding_separator与正文拼接最终文本结构为prefix 元数据(以分隔符连接) 正文 suffix。这使文档的类别、作者等元信息可以参与语义编码提升检索效果。分批与进度_embed_batch默认以batch_size32分批调用进度条由tqdm实现异步路径使用tqdm.asyncio若某批次因APIError失败默认记录日志后跳过该批继续处理通过raise_on_failure可改为直接抛出异常。超时与重试_client_kwargs()中timeout未显式设置时读取OPENAI_TIMEOUT默认30.0秒max_retries未显式设置时读取OPENAI_MAX_RETRIES默认5。Eden AI 组件完整继承了这套行为。四、EdenAITextEmbedder查询字符串向量化EdenAITextEmbedder用于把单个字符串典型场景是检索阶段的用户查询编码为向量输出可直接连到 Embedding Retriever 的query_embedding输入端。它应位于查询/RAG 流水线中 Retriever 之前。4.1 构造参数详解__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与EdenAIDocumentEmbedder相比它不包含batch_size、progress_bar、meta_fields_to_embed、embedding_separator等批量处理参数单条文本无需分片其余参数语义完全一致。SUPPORTED_MODELS清单与 Document 版本相同。4.2 单独使用from haystack.utils import Secret from haystack_integrations.components.embedders.edenai import EdenAITextEmbedder embedder EdenAITextEmbedder( api_keySecret.from_token(your-api-key), modelopenai/text-embedding-3-small, ) result embedder.run(textHow can I use the Eden AI embedding models with Haystack?) print(result[embedding]) # [-0.0015687942504882812, 0.052154541015625, 0.037109375...]run方法输入text: str输出字典包含embedding: list[float]与meta。若传入非字符串基类 OpenAITextEmbedder 会抛出TypeError并提示应改用 Document 版组件编码时先拼接prefix text suffix再调用client.embeddings.create()。4.3 在语义搜索流水线中使用结合EdenAIDocumentEmbedder索引文档、EdenAITextEmbedder编码查询并用InMemoryEmbeddingRetriever完成相似度检索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_functioncosine) documents [ Document(contentMy name is Wolfgang and I live in Berlin), Document(contentI saw a black horse running), Document(contentGermany has many big cities), ] document_embedder EdenAIDocumentEmbedder(modelopenai/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(modelopenai/text-embedding-3-small) ) query_pipeline.add_component( retriever, InMemoryEmbeddingRetriever(document_storedocument_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])这里体现了两类组件在流水线中的典型分工Document Embedder 负责离线索引侧的批量向量化Text Embedder 负责在线查询侧的实时编码两侧需使用同一模型以保证向量空间一致。五、EdenAIChatGenerator多供应商对话生成EdenAIChatGenerator通过 Eden AI 的 OpenAI 兼容端点生成聊天回复适用于需要访问 500 模型OpenAI、Anthropic、Mistral、Google、Cohere 等的对话与 RAG 场景。它继承OpenAIChatGenerator保留其全部标准配置仅在api_base_url上指向 Eden AI。在流水线中通常位于ChatPromptBuilder之后接收list[ChatMessage]输出replies。5.1 构造参数详解__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参数类型默认值说明api_keySecretEDENAI_API_KEY环境变量Eden AI API 密钥modelstropenai/gpt-4o-miniprovider/model格式模型名如anthropic/claude-sonnet-4-5、mistral/mistral-large-latest、google/gemini-2.5-flashstreaming_callbackStreamingCallbackT \| NoneNone流式响应时逐 chunk 回调的可调用对象generation_kwargsdict[str, Any] \| NoneNone透传给底层生成 API 的参数如max_tokens、temperature、top_pEden AI 专属参数例如 fallback 模型原样转发timeoutint \| NoneNone等待 API 响应的最大秒数max_retriesint \| NoneNone失败请求的最大重试次数toolsToolsType \| NoneNone供模型进行函数调用function calling的工具列表或单个Toolsettools_strictboolFalse为True时强制工具调用遵循严格 schemahttp_client_kwargsdict[str, Any] \| NoneNone透传给底层 HTTP 客户端的关键字参数5.2 单独使用from haystack_integrations.components.generators.edenai import EdenAIChatGenerator from haystack.dataclasses import ChatMessage messages [ChatMessage.from_user(Whats Natural Language Processing?)] client EdenAIChatGenerator(modelmistral/mistral-large-latest) response client.run(messages) print(response[replies][0].text)结合流式回调与密钥注入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_keySecret.from_env_var(EDENAI_API_KEY), modelmistral/mistral-large-latest, streaming_callbackprint_streaming_chunk, ) message ChatMessage.from_user(Whats Natural Language Processing? Be brief.) print(generator.run([message]))5.3 工具调用与流式输出组件指南明确说明其工具调用能力tools参数接受Tool对象列表、单个Toolset或二者混用便于把相关工具组织为逻辑分组、同时保留独立工具tools_strictTrue可启用工具调用的严格 schema 约束。流式输出则通过streaming_callback启用可复用 Haystack 提供的print_streaming_chunk等工具函数直接把 token 逐块输出。5.4 在 RAG 流水线中使用下面是一个完整 RAG 示例抓取网页内容、转换为 Document、经ChatPromptBuilder组装提示再由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(modelmistral/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(namefetcher, instancefetcher) rag_pipeline.add_component(nameconverter, instanceconverter) 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)该示例演示了EdenAIChatGenerator在流水线中的标准接法prompt_builder.prompt连向llm.messages结果从llm.replies读取。六、序列化支持三个组件都实现了to_dict() - dict[str, Any]用于把组件序列化为字典例如保存为 YAML 流水线定义并配套from_dict反序列化能力。序列化输出包含type与全部初始化参数api_key以Secret形式安全存储不会泄露明文。这意味着包含 Eden AI 组件的流水线可以像普通 Haystack 流水线一样通过 marshal/yaml 等机制保存、加载与版本化参考文档中对该方法返回值的描述为Dictionary with serialized data。七、总结与进一步阅读Eden AI 集成让 Haystack 用户以极低成本获得多供应商模型接入能力Embedding 侧通过EdenAIDocumentEmbedder与EdenAITextEmbedder覆盖索引与查询两个环节生成侧通过EdenAIChatGenerator统一对接 500 模型并支持工具调用、流式输出与供应商回退。由于三者分别继承 Haystack 官方 OpenAI 组件其超时、重试、分批、元数据拼接等行为与 OpenAI 组件保持一致迁移成本极低。继续深入阅读完整 API 参考version-2.18 Eden AI API 文档组件使用指南EdenAIDocumentEmbedder、EdenAITextEmbedder、EdenAIChatGenerator基类实现OpenAIDocumentEmbedder 源码、OpenAITextEmbedder 源码相关组件总览Embedders 组件索引、Generators 组件索引【免费下载链接】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），仅供参考