Haystack 与 Amazon Bedrock 深度集成指南从 S3 下载、向量化到 Agent 式 RAG 的完整实践【免费下载链接】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本指南以当前仓库docs-website/reference_versioned_docs/version-2.20/integrations-api/amazon_bedrock.md的 API 参考为骨架系统梳理 Haystack 的amazon-bedrock-haystack集成包在版本 2.20 中的全部能力包括异常体系、S3 文件下载、文本/文档/图像嵌入、基于 Converse API 的对话生成含流式、工具调用、提示缓存与结构化输出、Cohere 重排序、托管知识库检索以及服务端 Token 计数。读完本文你将掌握如何把 Amazon Bedrock 上的 Claude、Titan、Cohere 等模型无缝接入 Haystack 流水线并能在索引管线与查询管线中正确组合这些组件。安装与 AWS 认证amazon-bedrock-haystack是一个独立的集成包通过 pip 安装pip install amazon-bedrock-haystack所有 Bedrock 相关组件都通过 AWS 认证。推荐优先使用 AWS CLI 配置 IAM 凭证boto3会自动从环境或 AWS 配置文件加载这样初始化组件时无需显式传入任何密钥export AWS_ACCESS_KEY_IDyour-access-key-id export AWS_SECRET_ACCESS_KEYyour-secret-access-key export AWS_DEFAULT_REGIONyour-region如果 AWS 环境未配置则需要通过构造函数或环境变量提供三项必填凭证aws_access_key_id、aws_secret_access_key与aws_region_name也可使用 Haystack 的 Secret 机制安全传入见 secret-management 概念文档from haystack.utils import Secret ranker AmazonBedrockRanker( modelcohere.rerank-v3-5:0, aws_region_nameSecret.from_token(eu-central-1), )注意所设置的区域必须支持 Amazon Bedrock 服务。统一的异常体系整个集成把错误收敛到少量可预测的异常类型源码定义位于haystack_integrations.common.amazon_bedrock.errors与haystack_integrations.common.s3.errors模块异常类基类触发场景AmazonBedrockErrorExceptionBedrock 集成产生的任意错误对源异常透明包装可直接访问原始错误属性如AmazonBedrockError.messageAWSConfigurationErrorAmazonBedrockErrorAWS 环境配置不正确AmazonBedrockConfigurationErrorAmazonBedrockErrorBedrock 组件自身配置不正确AmazonBedrockInferenceErrorAmazonBedrockErrorBedrock 推理调用失败S3ErrorExceptionS3 相关组件产生的错误S3ConfigurationErrorS3ErrorAmazonS3 组件配置不正确S3StorageErrorS3Error与S3Storage交互出错如文件不存在、下载失败AmazonBedrockError的设计亮点是透明包装当底层 SDK 抛出带message属性的异常时AmazonBedrockError实例也会具有同名属性并保留原值方便上层直接透传诊断信息。在流水线中建议对AmazonBedrockInferenceError做重试或降级处理对AmazonBedrockConfigurationError做启动期配置校验。从 S3 下载文件S3Storage 与 S3DownloaderS3Storage底层存储封装haystack_integrations.common.s3.utils.S3Storage提供从 AWS S3 桶下载文件的底层存储类__init__( s3_bucket: str, session: Session, s3_prefix: str | None None, endpoint_url: str | None None, config: Config | None None, ) - Nones3_bucket要下载文件的桶名session创建 S3 客户端所用的 boto3Sessions3_prefix可选键前缀用于限定文件夹或命名结构。例如文件位于folder/subfolder/file.txt时前缀应设为folder/subfolder/文件位于桶根目录时设为Noneendpoint_url自定义端点 URL可用于 MinIO、LocalStack 等 S3 兼容服务config应用于 S3 客户端的 botocoreConfig。核心方法download(key: str, local_file_path: Path) - None将文件下载到本地。目标目录不存在会自动创建下载失败或文件不存在时抛S3StorageError客户端创建失败抛S3ConfigurationErrorclose() - None关闭本实例持有的 S3 客户端from_env(*, session, config, s3_bucket_name_envS3_DOWNLOADER_BUCKET) - S3Storage从环境变量构造实例读取S3_DOWNLOADER_BUCKET必填未设置抛ValueError、S3_DOWNLOADER_PREFIX可选前缀与AWS_ENDPOINT_URL可选自定义端点适合 S3 兼容服务。S3Downloader流水线组件S3Downloaderhaystack_integrations.components.downloaders.s3.s3_downloader负责把 S3 文件并行下载到本地文件系统并把本地路径写回 Document通常放在需要本地文件路径的 Converter 或 Router 之前更多场景见 s3downloader 使用文档。构造函数签名__init__( *, aws_access_key_id: Secret | None Secret.from_env_var(AWS_ACCESS_KEY_ID, strictFalse), aws_secret_access_key: Secret | None Secret.from_env_var(AWS_SECRET_ACCESS_KEY, strictFalse), aws_session_token: Secret | None Secret.from_env_var(AWS_SESSION_TOKEN, strictFalse), aws_region_name: Secret | str | None Secret.from_env_var(AWS_DEFAULT_REGION, strictFalse), aws_profile_name: Secret | None Secret.from_env_var(AWS_PROFILE, strictFalse), boto3_config: dict[str, Any] | None None, file_root_path: str | None None, file_extensions: list[str] | None None, file_name_meta_key: str file_name, max_workers: int 32, max_cache_size: int 100, s3_key_generation_function: Callable[[Document], str] | None None, s3_bucket_name_env: str S3_DOWNLOADER_BUCKET, ) - None关键参数说明file_root_path文件下载的根目录可通过构造函数或FILE_ROOT_PATH环境变量设置二者均未设置时抛ValueError。出于安全考虑下载被严格限定在该目录内文档的文件名若解析到目录之外如绝对路径或含..会被记录日志并跳过写入而不是落盘file_extensions允许下载的扩展名白名单默认全部允许。例如[.pdf, .txt]只下载这两类文件max_workers并发下载线程数默认 32适合大批量文档加速max_cache_size本地缓存文件数上限默认 100。超出后按最近最少使用LRU策略淘汰已下载过的文件会刷新访问时间而不重复下载file_name_meta_key存放待下载文件名的 meta 键默认file_name对应Document.meta[file_name]s3_key_generation_function自定义 S3 键生成函数接收Document返回字符串未提供时默认使用Document.meta[file_name_meta_key]。若设置了S3_DOWNLOADER_PREFIX环境变量其值会自动拼接到生成的键前s3_bucket_name_env指定桶名所在的环境变量名默认S3_DOWNLOADER_BUCKET。生命周期方法warm_up()初始化设置与存储桶环境变量缺失抛ValueError客户端创建失败抛S3ConfigurationErrorclose()关闭持有的 S3 客户端run(documents)接收Document列表返回带meta[file_path]的文档列表下载失败抛S3Errorto_dict()/from_dict()支持 YAML/JSON 序列化。基本用法from haystack.dataclasses import Document from haystack_integrations.components.downloaders.s3 import S3Downloader documents [ Document(meta{file_name: report.pdf}), Document(meta{file_name: data.txt}), ] downloader S3Downloader(file_root_path/tmp/s3_downloads) result downloader.run(documentsdocuments) for doc in result[documents]: print(fFile downloaded to: {doc.meta[file_path]})文本与文档向量化两个 EmbedderBedrock 集成提供两个互补的文本嵌入组件均支持 Amazon Titan 与 Cohere 系列模型例如amazon.titan-embed-text-v1、amazon.titan-embed-text-v2:0、amazon.titan-embed-image-v1、cohere.embed-english-v3、cohere.embed-multilingual-v3、cohere.embed-v4:0完整清单需在 Bedrock 文档中筛选 embedding 类别。当前仓库的 amazonbedrockdocumentembedder 文档 与 amazonbedrocktextembedder 文档 给出了对应的流水线集成示例。AmazonBedrockTextEmbedder查询向量化用于把单个字符串典型场景是查询编码为向量import os from haystack_integrations.components.embedders.amazon_bedrock import AmazonBedrockTextEmbedder os.environ[AWS_ACCESS_KEY_ID] ... os.environ[AWS_SECRET_ACCESS_KEY_ID] ... os.environ[AWS_DEFAULT_REGION] us-east-1 embedder AmazonBedrockTextEmbedder( modelcohere.embed-english-v3, input_typesearch_query, # 模型特定参数通过 kwargs 透传 truncateLEFT, ) print(embedder.run(I love Paris in the summer.)) # {embedding: [0.002, 0.032, 0.504, ...]}构造函数除模型与 AWS 凭证外还支持boto3_config底层 Boto3 客户端配置用于调节重试、超时与连接管理以及任意**kwargs透传模型推理参数Cohere 模型常用input_type如search_query/search_document与truncateAmazon Titan Text Embeddings V2 常用dimensions与normalize。run(text)返回键为embedding的向量非字符串输入抛TypeError推理失败抛AmazonBedrockInferenceError。AmazonBedrockDocumentEmbedder文档批量向量化用于为整批Document计算向量并写回embedding字段。关键参数model嵌入模型batch_size单次编码的文档数默认 32。注意只有 Cohere 模型支持批量推理该参数对 Titan 模型无效progress_bar是否显示进度条生产环境建议关闭以保持日志干净meta_fields_to_embed需要与文档正文一起参与嵌入的 meta 字段列表embedding_separator拼接 meta 字段与正文的分隔符默认\n其余 AWS 凭证、boto3_config、**kwargs与 TextEmbedder 一致。用法示例import os from haystack.dataclasses import Document from haystack_integrations.components.embedders.amazon_bedrock import AmazonBedrockDocumentEmbedder os.environ[AWS_ACCESS_KEY_ID] ... os.environ[AWS_SECRET_ACCESS_KEY_ID] ... os.environ[AWS_DEFAULT_REGION] ... embedder AmazonBedrockDocumentEmbedder( modelcohere.embed-english-v3, input_typesearch_document, ) doc Document(contentI love Paris in the winter., meta{name: doc1}) result embedder.run([doc]) print(result[documents][0].embedding) # [0.002, 0.032, 0.504, ...]若文档带有语义明确的元数据如标题可通过meta_fields_to_embed一并嵌入以提升检索效果doc Document(contentsome text, meta{title: relevant title, page number: 18}) embedder AmazonBedrockDocumentEmbedder( modelcohere.embed-english-v3, meta_fields_to_embed[title], ) docs_w_embeddings embedder.run(documents[doc])[documents]多模态图像嵌入AmazonBedrockDocumentImageEmbedder该组件haystack_integrations.components.embedders.amazon_bedrock.document_image_embedder读取 Document meta 字段中的图片或 PDF 路径用多模态模型Titan 与 Cohere 多模态系列如amazon.titan-embed-image-v1、cohere.embed-english-v3、cohere.embed-multilingual-v3、cohere.embed-v4:0计算嵌入并写回embedding字段。详细说明见 amazonbedrockdocumentimageembedder 文档。构造函数关键参数file_path_meta_field存放图片/PDF 路径的 meta 字段名默认file_pathroot_path文档文件的根目录。提供后 meta 中的路径将相对于它解析为None时按绝对路径处理image_size(width, height)元组。提供后图片会在保持宽高比的前提下缩放以适配该尺寸减少文件体积、内存占用与传输时间适合有分辨率限制的模型progress_bar是否显示进度条**kwargs模型特定参数如 Titan 的embeddingConfig、Cohere 的embedding_types。import os from haystack import Document from haystack_integrations.components.embedders.amazon_bedrock import AmazonBedrockDocumentImageEmbedder os.environ[AWS_ACCESS_KEY_ID] ... os.environ[AWS_SECRET_ACCESS_KEY_ID] ... os.environ[AWS_DEFAULT_REGION] us-east-1 embedder AmazonBedrockDocumentImageEmbedder( modelamazon.titan-embed-image-v1, image_size(1024, 1024), # 可选等比缩放 ) documents [ Document(contentA photo of a cat, meta{file_path: cat.jpg}), Document(contentA photo of a dog, meta{file_path: dog.jpg}), ] result embedder.run(documentsdocuments) documents_with_embeddings result[documents]运行后每个文档的meta中会写入embedding_source字段记录嵌入来源如{type: image, file_path_meta_field: file_path}。检索阶段需用同一模型配合AmazonBedrockTextEmbedder编码查询文本再交给向量检索器。需要注意Cohere 模型的embedding_types参数仅支持传入单个值传入多个值会报错。对话生成AmazonBedrockChatGeneratorAmazonBedrockChatGeneratorhaystack_integrations.components.generators.amazon_bedrock.chat.chat_generator通过 BedrockConverse API完成多轮对话一个组件即可调用 Amazon、Anthropic、Cohere、Meta、Mistral 等多家聊天模型示例模型名如global.anthropic.claude-sonnet-4-6。它通常放置在ChatPromptBuilder之后详细用法见 amazonbedrockchatgenerator 文档。基本用法from haystack_integrations.components.generators.amazon_bedrock import AmazonBedrockChatGenerator from haystack.dataclasses import ChatMessage from haystack.components.generators.utils import print_streaming_chunk messages [ ChatMessage.from_system( \nYou are a helpful, respectful and honest assistant, answer in German only ), ChatMessage.from_user(Whats Natural Language Processing?), ] client AmazonBedrockChatGenerator( modelglobal.anthropic.claude-sonnet-4-6, streaming_callbackprint_streaming_chunk, ) client.run(messages, generation_kwargs{max_tokens: 512})构造函数参数详解__init__( model: str, aws_access_key_id: Secret | None ..., aws_secret_access_key: Secret | None ..., aws_session_token: Secret | None ..., aws_region_name: Secret | str | None ..., aws_profile_name: Secret | None ..., generation_kwargs: dict[str, Any] | None None, streaming_callback: StreamingCallbackT | None None, boto3_config: dict[str, Any] | None None, tools: ToolsType | None None, *, guardrail_config: dict[str, str] | None None, tools_cachepoint_config: dict[str, str] | None None, system_cachepoint_config: dict[str, str] | None None, ) - Nonemodel模型名必须符合 Bedrock 的 model-id / ARN 格式且在该区域可用generation_kwargs生成参数字典常用键包括maxTokens最大生成 token 数、stopSequences停止序列列表、temperature采样温度、topP核采样参数以及response_format见下文结构化输出streaming_callback收到新 token 时回调的函数。默认不启用流式设置后自动切换为流式模式回调接收StreamingChunk对象toolsTool与/或Toolset对象列表或单个Toolset供模型生成调用每个工具名必须唯一guardrail_config在 Bedrock 中已创建的护栏配置字典须匹配GuardrailConfiguration流式模式为GuardrailStreamConfiguration。若设置trace: enabled护栏追踪信息会写入结果ChatMessage的meta[trace]。注意流式模式启用护栏可能引入额外延迟可通过streamProcessingMode调节tools_cachepoint_config/system_cachepoint_config分别为工具定义与系统消息启用提示缓存字典须匹配 Bedrock 的CachePointBlock结构例如{type: default, ttl: 5m}。结构化 JSON 输出通过generation_kwargs[response_format]请求受 JSON Schema 校验的结构化输出generation_kwargs { response_format: { name: person, schema: { type: object, properties: { name: {type: string}, age: {type: integer}, }, required: [name, age], additionalProperties: False, }, } }其中schema为必填的 JSON Schema 字典name可选默认response_schemadescription可选。开启后解析出的 JSON 对象存放在reply.meta[structured_output]中。多模态输入Converse API 支持图片内容可直接与文本混合作为用户消息from haystack.dataclasses import ChatMessage, ImageContent from haystack_integrations.components.generators.amazon_bedrock import AmazonBedrockChatGenerator generator AmazonBedrockChatGenerator(modelglobal.anthropic.claude-sonnet-4-6) image_content ImageContent.from_file_path(file_pathapple.jpg) message ChatMessage.from_user(content_parts[Describe the image using 10 words at most., image_content]) response generator.run(messages[message])[replies][0].text # The image shows a red apple.工具调用AmazonBedrockChatGenerator支持 Haystack 统一工具架构相同的Tool定义与调用模式在 Bedrock、OpenAI、Ollama 等提供商之间保持一致也支持把多个Toolset与独立Tool混合传入。完整的多轮工具调用流程如下from haystack.dataclasses import ChatMessage from haystack.tools import Tool from haystack_integrations.components.generators.amazon_bedrock import AmazonBedrockChatGenerator def weather(city: str): return fThe weather in {city} is sunny and 32°C tool_parameters { type: object, properties: {city: {type: string}}, required: [city], } weather_tool Tool( nameweather, descriptionuseful to determine the weather in a given location, parameterstool_parameters, functionweather, ) client AmazonBedrockChatGenerator( modelglobal.anthropic.claude-sonnet-4-6, tools[weather_tool], ) messages [ChatMessage.from_user(Whats the weather like in Paris?)] results client.run(messagesmessages) # 从回复中提取工具调用 tool_message next(msg for msg in results[replies] if msg.tool_call) tool_call tool_message.tool_call # 执行工具并把结果回传给模型 weather_result weather(**tool_call.arguments) new_messages [ messages[0], tool_message, ChatMessage.from_tool(tool_resultweather_result, origintool_call), ] final_result client.run(new_messages) print(final_result[replies][0].text)提示缓存组件支持提示缓存以降低延迟与输入 token 成本。缓存消息时在ChatMessage.meta中写入cachePoint键每次请求最多可设四个缓存点且输入需达到模型特定的最低 token 阈值msg ChatMessage.from_user( Long message..., meta{cachePoint: {type: default, ttl: 5m}} ) generator AmazonBedrockChatGenerator(modelglobal.anthropic.claude-sonnet-4-6) result generator.run(messages[msg]) # 缓存写入成功后可在回复中读取缓存命中的输入 token 数 result[replies][0].meta[usage][cache_write_input_tokens]缓存工具定义则通过初始化参数tools_cachepoint_config完成设置后所有发给模型的工具在满足阈值时都会参与缓存。缓存系统消息同理使用system_cachepoint_config。流式与异步流式输出只需传入streaming_callback如内置的print_streaming_chunk。run_async提供异步推理入口签名与run一致messages、streaming_callback、generation_kwargs、tools适合非阻塞或并发场景异步流式时优先使用异步回调。run与run_async的messages参数都接受list[ChatMessage]或直接传入字符串会自动包装为 user 角色消息。run级别的generation_kwargs会按 key 与初始化时的参数合并运行期传入的 key 优先。推理失败抛AmazonBedrockInferenceError。重排序AmazonBedrockRankerAmazonBedrockRankerhaystack_integrations.components.rankers.amazon_bedrock.ranker基于查询与文档的语义相关度对文档重排返回从最相关到最不相关的排序结果常用于检索器之后、PromptBuilder 之前。当前支持cohere.rerank-v3-5:0默认与amazon.rerank-v1:0两个模型。from haystack import Document from haystack.utils import Secret from haystack_integrations.components.rankers.amazon_bedrock import AmazonBedrockRanker ranker AmazonBedrockRanker( modelcohere.rerank-v3-5:0, top_k2, aws_region_nameSecret.from_token(eu-central-1), ) docs [Document(contentParis), Document(contentBerlin)] query What is the capital of germany? output ranker.run(queryquery, documentsdocs) docs output[documents]构造函数要点model默认cohere.rerank-v3-5:0top_k最多返回的文档数默认 10run时也可覆盖传入max_chunks_per_doc当文档超过 512 token 时可拆分的最大块数默认 10。文档说明该参数当前尚未在实现中使用仅为未来兼容保留meta_fields_to_embed与meta_data_separator默认\n把指定的 meta 字段与文档内容拼接后参与重排model为空或top_k不大于 0 时抛ValueError。托管知识库检索AmazonBedrockKnowledgeBaseRetrieverAmazonBedrockKnowledgeBaseRetrieverhaystack_integrations.components.retrievers.amazon_bedrock.knowledge_base_retriever从 Amazon Bedrock托管知识库检索文档。与多数 Retriever 不同它不需要 Haystack Document Store 或 Embedder——索引与向量化完全由 AWS 托管组件只接收文本查询。详细说明见 amazonbedrockknowledgebaseretriever 文档。from haystack.utils import Secret from haystack_integrations.components.retrievers.amazon_bedrock import AmazonBedrockKnowledgeBaseRetriever retriever AmazonBedrockKnowledgeBaseRetriever( knowledge_base_idABCDEFGHIJ, aws_region_nameSecret.from_token(eu-central-1), ) result retriever.run(queryWhat are the benefits of managed knowledge bases?) for doc in result[documents]: print(doc.content) print(doc.meta[source]) # S3 / web / Confluence / Salesforce / SharePoint 等来源 print(doc.score)关键配置knowledge_base_id知识库 ID未提供时回退到AWS_KNOWLEDGE_BASE_ID环境变量number_of_results默认返回结果数上限默认 5run的top_k参数可覆盖use_agentic_retrieval为True时优先尝试 Agentic Retrieve API不可用时回退到标准 Retrieve API。默认读取USE_AGENTIC_RETRIEVAL环境变量未设置则为True。每个返回的 Document 都带有score以及来源相关的元数据source底层内容的 S3、web、Confluence、Salesforce、SharePoint 或自定义位置、knowledge_base_id、knowledge_base_type。使用时需要先在 AWS 侧创建知识库并完成文档摄取。检索失败抛AmazonBedrockInferenceError。服务端 Token 计数AmazonBedrockTokenCounterAmazonBedrockTokenCounterhaystack_integrations.token_counters.amazon_bedrock.token_counter实现 Haystack 的TokenCounter协议通过 Bedrock 的CountTokens API计数。与本地分词器不同它会将输入发送到服务端返回模型精确的分词结果——包括 Bedrock 对消息、系统提示与工具 schema 应用的格式化因此计数与等价的 Converse 请求完全一致。from haystack.dataclasses import ChatMessage from haystack_integrations.token_counters.amazon_bedrock import AmazonBedrockTokenCounter counter AmazonBedrockTokenCounter(modelanthropic.claude-3-5-sonnet-20240620-v1:0) messages [ChatMessage.from_user(Hello, how are you?)] token_count counter.count(messages) print(fToken count: {token_count})要点与限制count(messages, toolsNone)返回给定消息与工具将消耗的输入 token 数tools传入时其 schema 也参与计数为None则只统计消息无内容可计时返回0由于依赖服务端 APImessages必须构成完整、合法的会话需以 user 消息开头工具结果必须与产生它的工具调用成对出现——Bedrock 会像 Converse 推理 API 一样校验输入。因此它适合在发送整个请求前评估请求体量而无法对孤立片段如单条工具结果消息计数需要在 compactor 等场景对单条消息做片段级计数时应改用本地计数器如ApproximateTokenCountermodel为空抛ValueErrorCountTokens 请求失败抛AmazonBedrockInferenceError。to_dict()/from_dict()支持序列化。实战组装一个端到端 RAG 流水线把上述组件组合起来可以得到一个完整的下载 → 向量化 → 检索 → 重排 → 生成链路。索引管线用S3Downloader拉取 S3 文件用AmazonBedrockDocumentEmbedder编码后写入InMemoryDocumentStore查询管线用AmazonBedrockTextEmbedder编码查询、InMemoryEmbeddingRetriever召回、AmazonBedrockRanker精排最后由AmazonBedrockChatGenerator生成答案from haystack import Document, Pipeline from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.components.writers import DocumentWriter from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever from haystack_integrations.components.downloaders.s3 import S3Downloader from haystack_integrations.components.embedders.amazon_bedrock import ( AmazonBedrockDocumentEmbedder, AmazonBedrockTextEmbedder, ) from haystack_integrations.components.rankers.amazon_bedrock import AmazonBedrockRanker from haystack_integrations.components.generators.amazon_bedrock import AmazonBedrockChatGenerator document_store InMemoryDocumentStore(embedding_similarity_functioncosine) # —— 索引管线S3 下载 → 文档向量化 → 写入存储 —— indexing Pipeline() indexing.add_component(downloader, S3Downloader(file_root_path/tmp/s3_downloads)) indexing.add_component( embedder, AmazonBedrockDocumentEmbedder(modelcohere.embed-english-v3, input_typesearch_document), ) indexing.add_component(writer, DocumentWriter(document_storedocument_store)) indexing.connect(downloader.documents, embedder.documents) indexing.connect(embedder, writer) indexing.run({ downloader: { documents: [ Document(meta{file_name: report.pdf}), Document(meta{file_name: data.txt}), ] } }) # —— 查询管线查询向量化 → 召回 → 重排 → 生成 —— query_pipeline Pipeline() query_pipeline.add_component( text_embedder, AmazonBedrockTextEmbedder(modelcohere.embed-english-v3, input_typesearch_query), ) query_pipeline.add_component( retriever, InMemoryEmbeddingRetriever(document_storedocument_store), ) query_pipeline.add_component(ranker, AmazonBedrockRanker(modelcohere.rerank-v3-5:0, top_k3)) query_pipeline.add_component( llm, AmazonBedrockChatGenerator(modelglobal.anthropic.claude-sonnet-4-6), ) query_pipeline.connect(text_embedder.embedding, retriever.query_embedding) query_pipeline.connect(retriever.documents, ranker.documents) query_pipeline.connect(ranker.documents, llm.documents) result query_pipeline.run({text_embedder: {text: Who lives in Berlin?}}) print(result[llm][replies][0].text)在流水线内使用组件时各组件均通过to_dict()/from_dict()支持 YAML/JSON 序列化与反序列化便于用 Haystack 的Pipeline.dumps()/loads()持久化整个拓扑。若后续需要把多条消息整体送交模型前先估算 token可在生成器之前接入AmazonBedrockTokenCounter做预算控制。小结amazon-bedrock-haystack集成覆盖了构建生产级 RAG 与 Agent 应用所需的完整环节从S3Downloader拉取原始文件到AmazonBedrockTextEmbedder/AmazonBedrockDocumentEmbedder/AmazonBedrockDocumentImageEmbedder完成文本与多模态向量化再到AmazonBedrockRanker精排、AmazonBedrockKnowledgeBaseRetriever直连托管知识库以及基于 Converse API 的AmazonBedrockChatGenerator流式、工具调用、提示缓存、护栏与结构化输出和AmazonBedrockTokenCounter做请求预检。所有组件共享同一套 AWS 凭证加载机制与异常体系配合 Haystack 的 Pipeline 编排能力可以在不引入额外基础设施的前提下把 Bedrock 上的各家基础模型统一纳入模块化流水线。更多组件级细节可继续查阅 当前仓库的 Amazon Bedrock API 参考 及 version-2.20 版本的对应参考文档。【免费下载链接】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），仅供参考