文档API 参考📓 教程🧑‍🍳 食谱🤝 集成💜 Discord🎨 Studio
API 参考

Amazon Bedrock

Haystack 的 Amazon Bedrock 集成

模块 haystack_integrations.common.amazon_bedrock.errors

AmazonBedrockError

Amazon Bedrock 集成产生的任何错误。

此错误透明地封装了其源，可以直接访问其属性：例如，如果原始错误具有message 属性，AmazonBedrockError.message 将存在并具有预期内容。

AWSConfigurationError

AWS 配置不正确时引发的异常

AmazonBedrockConfigurationError

AmazonBedrock 节点配置不正确时引发的异常

AmazonBedrockInferenceError

Bedrock 推理节点中出现问题的异常

模块 haystack_integrations.components.embedders.amazon_bedrock.document_embedder

AmazonBedrockDocumentEmbedder

一个用于使用 Amazon Bedrock 计算文档嵌入的组件。每个文档的嵌入存储在Document 的 embedding 字段中。

使用示例

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(
    model="cohere.embed-english-v3",
    input_type="search_document",
)

doc = Document(content="I love Paris in the winter.", meta={"name": "doc1"})

result = embedder.run([doc])
print(result['documents'][0].embedding)

# [0.002, 0.032, 0.504, ...]

AmazonBedrockDocumentEmbedder.__init__

def __init__(
        model: Literal[
            "amazon.titan-embed-text-v1",
            "cohere.embed-english-v3",
            "cohere.embed-multilingual-v3",
            "amazon.titan-embed-text-v2:0",
            "amazon.titan-embed-image-v1",
        ],
        aws_access_key_id: Optional[Secret] = Secret.from_env_var(
            "AWS_ACCESS_KEY_ID", strict=False),
        aws_secret_access_key: Optional[Secret] = Secret.
    from_env_var(  # noqa: B008
        "AWS_SECRET_ACCESS_KEY", strict=False),
        aws_session_token: Optional[Secret] = Secret.from_env_var(
            "AWS_SESSION_TOKEN", strict=False),
        aws_region_name: Optional[Secret] = Secret.from_env_var(
            "AWS_DEFAULT_REGION", strict=False),
        aws_profile_name: Optional[Secret] = Secret.from_env_var("AWS_PROFILE",
                                                                 strict=False),
        batch_size: int = 32,
        progress_bar: bool = True,
        meta_fields_to_embed: Optional[List[str]] = None,
        embedding_separator: str = "\n",
        boto3_config: Optional[Dict[str, Any]] = None,
        **kwargs: Any) -> None

使用提供的参数初始化 AmazonBedrockDocumentEmbedder。参数传递给

Amazon Bedrock 客户端。

请注意，如果 AWS 环境配置正确，则不需要 AWS 凭证。这些凭证会自动从环境或 AWS 配置文件中加载，无需通过构造函数显式提供。如果 AWS 环境未配置，用户需要通过构造函数提供 AWS 凭证。除了模型之外，三个必需参数是aws_access_key_id, aws_secret_access_key 和aws_region_name.

参数:

  • model：要使用的嵌入模型。模型必须以 Amazon Bedrock 文档中概述的格式指定。
  • aws_access_key_id：AWS 访问密钥 ID。
  • aws_secret_access_key：AWS 秘密访问密钥。
  • aws_session_token：AWS 会话令牌。
  • aws_region_name：AWS 区域名称。
  • aws_profile_name：AWS 配置文件名称。
  • batch_size：一次编码的文档数量。只有 Cohere 模型支持批量推理。此参数对于 Amazon Titan 模型会被忽略。
  • progress_bar: 是否显示进度条。在生产部署中禁用此选项有助于保持日志的整洁。
  • meta_fields_to_embed: 需要与 Document 文本一起进行嵌入的元字段列表。
  • embedding_separator: 用于将元字段连接到 Document 文本的分隔符。
  • boto3_config：boto3 客户端的配置。
  • kwargs：用于模型推理的附加参数。例如，input_type 和truncate 用于 Cohere 模型。

引发:

  • ValueError：如果模型不支持。
  • AmazonBedrockConfigurationError：如果 AWS 环境配置不正确。

AmazonBedrockDocumentEmbedder.run

@component.output_types(documents=List[Document])
def run(documents: List[Document]) -> Dict[str, List[Document]]

使用指定模型嵌入提供的Document。

参数:

  • documents：要嵌入的Document。

引发:

  • AmazonBedrockInferenceError：如果推理失败。

返回值:

包含以下键的字典

  • documents：要嵌入的已填充embedding 字段的 Document。

AmazonBedrockDocumentEmbedder.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

AmazonBedrockDocumentEmbedder.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "AmazonBedrockDocumentEmbedder"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

模块 haystack_integrations.components.embedders.amazon_bedrock.text_embedder

AmazonBedrockTextEmbedder

一个用于使用 Amazon Bedrock 嵌入字符串的组件。

使用示例

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"] = "..."

embedder = AmazonBedrockTextEmbedder(
    model="cohere.embed-english-v3",
    input_type="search_query",
)

print(text_embedder.run("I love Paris in the summer."))

# {'embedding': [0.002, 0.032, 0.504, ...]}

AmazonBedrockTextEmbedder.__init__

def __init__(
        model: Literal[
            "amazon.titan-embed-text-v1",
            "cohere.embed-english-v3",
            "cohere.embed-multilingual-v3",
            "amazon.titan-embed-text-v2:0",
            "amazon.titan-embed-image-v1",
        ],
        aws_access_key_id: Optional[Secret] = Secret.from_env_var(
            "AWS_ACCESS_KEY_ID", strict=False),
        aws_secret_access_key: Optional[Secret] = Secret.
    from_env_var(  # noqa: B008
        "AWS_SECRET_ACCESS_KEY", strict=False),
        aws_session_token: Optional[Secret] = Secret.from_env_var(
            "AWS_SESSION_TOKEN", strict=False),
        aws_region_name: Optional[Secret] = Secret.from_env_var(
            "AWS_DEFAULT_REGION", strict=False),
        aws_profile_name: Optional[Secret] = Secret.from_env_var("AWS_PROFILE",
                                                                 strict=False),
        boto3_config: Optional[Dict[str, Any]] = None,
        **kwargs: Any) -> None

使用提供的参数初始化 AmazonBedrockTextEmbedder。参数传递给

Amazon Bedrock 客户端。

请注意，如果 AWS 环境配置正确，则不需要 AWS 凭证。这些凭证会自动从环境或 AWS 配置文件中加载，无需通过构造函数显式提供。如果 AWS 环境未配置，用户需要通过构造函数提供 AWS 凭证。除了模型之外，三个必需参数是aws_access_key_id, aws_secret_access_key 和aws_region_name.

参数:

  • model：要使用的嵌入模型。模型必须以 Amazon Bedrock 文档中概述的格式指定。
  • aws_access_key_id：AWS 访问密钥 ID。
  • aws_secret_access_key：AWS 秘密访问密钥。
  • aws_session_token：AWS 会话令牌。
  • aws_region_name：AWS 区域名称。
  • aws_profile_name：AWS 配置文件名称。
  • boto3_config：boto3 客户端的配置。
  • kwargs：用于模型推理的附加参数。例如，input_type 和truncate 用于 Cohere 模型。

引发:

  • ValueError：如果模型不支持。
  • AmazonBedrockConfigurationError：如果 AWS 环境配置不正确。

AmazonBedrockTextEmbedder.run

@component.output_types(embedding=List[float])
def run(text: str) -> Dict[str, List[float]]

使用 Amazon Bedrock 模型嵌入输入文本。

参数:

  • text：要嵌入的输入文本。

引发:

  • TypeError：如果输入文本不是字符串。
  • AmazonBedrockInferenceError：如果模型推理失败。

返回值:

包含以下键的字典

  • embedding：输入文本的嵌入。

AmazonBedrockTextEmbedder.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

AmazonBedrockTextEmbedder.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "AmazonBedrockTextEmbedder"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

模块 haystack_integrations.components.embedders.amazon_bedrock.document_image_embedder

AmazonBedrockDocumentImageEmbedder

一个用于使用 Amazon Bedrock 模型计算基于图像的文档嵌入的组件。

每个 Document 的嵌入存储在Document 的 embedding 字段中。

使用示例

from haystack import Document
rom 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"] = "..."

embedder = AmazonBedrockDocumentImageEmbedder(model="amazon.titan-embed-image-v1")

documents = [
    Document(content="A photo of a cat", meta={"file_path": "cat.jpg"}),
    Document(content="A photo of a dog", meta={"file_path": "dog.jpg"}),
]

result = embedder.run(documents=documents)
documents_with_embeddings = result["documents"]
print(documents_with_embeddings)

# [Document(id=...,
#           content='A photo of a cat',
#           meta={'file_path': 'cat.jpg',
#                 'embedding_source': {'type': 'image', 'file_path_meta_field': 'file_path'}},
#           embedding=vector of size 512),
#  ...]

AmazonBedrockDocumentImageEmbedder.__init__

def __init__(
        *,
        model: Literal["amazon.titan-embed-image-v1",
                       "cohere.embed-english-v3",
                       "cohere.embed-multilingual-v3"],
        aws_access_key_id: Optional[Secret] = Secret.from_env_var(
            "AWS_ACCESS_KEY_ID", strict=False),
        aws_secret_access_key: Optional[Secret] = Secret.
    from_env_var(  # noqa: B008
        "AWS_SECRET_ACCESS_KEY", strict=False),
        aws_session_token: Optional[Secret] = Secret.from_env_var(
            "AWS_SESSION_TOKEN", strict=False),
        aws_region_name: Optional[Secret] = Secret.from_env_var(
            "AWS_DEFAULT_REGION", strict=False),
        aws_profile_name: Optional[Secret] = Secret.from_env_var("AWS_PROFILE",
                                                                 strict=False),
        file_path_meta_field: str = "file_path",
        root_path: Optional[str] = None,
        image_size: Optional[Tuple[int, int]] = None,
        progress_bar: bool = True,
        boto3_config: Optional[Dict[str, Any]] = None,
        **kwargs: Any) -> None

创建 AmazonBedrockDocumentImageEmbedder 组件。

参数:

  • model：用于计算嵌入的 Bedrock 模型。传入有效的模型 ID。支持的模型有
  • "amazon.titan-embed-image-v1"
  • "cohere.embed-english-v3"
  • "cohere.embed-multilingual-v3"
  • aws_access_key_id：AWS 访问密钥 ID。
  • aws_secret_access_key：AWS 秘密访问密钥。
  • aws_session_token：AWS 会话令牌。
  • aws_region_name：AWS 区域名称。
  • aws_profile_name：AWS 配置文件名称。
  • file_path_meta_field: Document 中包含图像或 PDF 文件路径的元数据字段。
  • root_path: 存储文档文件的根目录路径。如果提供，文档元数据中的文件路径将相对于此路径解析。如果为 None，则文件路径被视为绝对路径。
  • image_size：如果提供，则将图像大小调整为适合指定尺寸（宽度、高度）并保持纵横比。这会减小文件大小、内存使用和处理时间，这在使用具有分辨率限制的模型或将图像传输到远程服务时很有益。
  • progress_bar: 如果True，在嵌入文档时显示进度条。
  • boto3_config：boto3 客户端的配置。
  • kwargs：用于模型推理的附加参数。例如，用于 Amazon Titan 模型的 embeddingConfig 和用于 Cohere 模型的 embedding_types。

引发:

  • ValueError：如果模型不支持。
  • AmazonBedrockConfigurationError：如果 AWS 环境配置不正确。

AmazonBedrockDocumentImageEmbedder.to_dict

def to_dict() -> dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

AmazonBedrockDocumentImageEmbedder.from_dict

@classmethod
def from_dict(cls, data: dict[str,
                              Any]) -> "AmazonBedrockDocumentImageEmbedder"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

AmazonBedrockDocumentImageEmbedder.run

@component.output_types(documents=list[Document])
def run(documents: list[Document]) -> dict[str, list[Document]]

嵌入图像列表。

参数:

  • documents: 要嵌入的 Documents。

返回值:

包含以下键的字典

  • documents: 带有嵌入的 Documents。

模块 haystack_integrations.components.generators.amazon_bedrock.generator

AmazonBedrockGenerator

使用 Amazon Bedrock 上托管的模型生成文本。

例如，要使用 Anthropic Claude 模型，请在model 参数中传入 'anthropic.claude-v2'。通过本地 AWS 配置文件或直接通过aws_access_key_id, aws_secret_access_key, aws_session_token 和aws_region_name 参数提供 AWS 凭证。

使用示例

from haystack_integrations.components.generators.amazon_bedrock import AmazonBedrockGenerator

generator = AmazonBedrockGenerator(
        model="anthropic.claude-v2",
        max_length=99
)

print(generator.run("Who is the best American actor?"))

AmazonBedrockGenerator 使用 AWS 进行身份验证。您可以使用 AWS CLI 通过 IAM 进行身份验证。有关设置 IAM 基于身份策略的更多信息，请参阅 [Amazon Bedrock 文档] (https://docs.aws.amazon.com/bedrock/latest/userguide/security_iam_id-based-policy-examples.html)。如果 AWS 环境配置正确，则不需要 AWS 凭证，因为它们会自动从环境或 AWS 配置文件中加载。如果 AWS 环境未配置，请将aws_access_key_id, aws_secret_access_key, aws_session_token 和aws_region_name 设置为环境变量或将其作为 Secret 参数传递。确保您设置的区域支持 Amazon Bedrock。

AmazonBedrockGenerator.__init__

def __init__(
        model: str,
        aws_access_key_id: Optional[Secret] = Secret.from_env_var(
            "AWS_ACCESS_KEY_ID", strict=False),
        aws_secret_access_key: Optional[Secret] = Secret.
    from_env_var(  # noqa: B008
        "AWS_SECRET_ACCESS_KEY", strict=False),
        aws_session_token: Optional[Secret] = Secret.from_env_var(
            "AWS_SESSION_TOKEN", strict=False),
        aws_region_name: Optional[Secret] = Secret.from_env_var(
            "AWS_DEFAULT_REGION", strict=False),
        aws_profile_name: Optional[Secret] = Secret.from_env_var("AWS_PROFILE",
                                                                 strict=False),
        max_length: Optional[int] = None,
        truncate: Optional[bool] = None,
        streaming_callback: Optional[Callable[[StreamingChunk], None]] = None,
        boto3_config: Optional[Dict[str, Any]] = None,
        model_family: Optional[MODEL_FAMILIES] = None,
        **kwargs: Any) -> None

创建一个新的AmazonBedrockGenerator 实例。

参数:

  • model: 要使用的模型名称。
  • aws_access_key_id：AWS 访问密钥 ID。
  • aws_secret_access_key：AWS 秘密访问密钥。
  • aws_session_token：AWS 会话令牌。
  • aws_region_name：AWS 区域名称。确保您设置的区域支持 Amazon Bedrock。
  • aws_profile_name：AWS 配置文件名称。
  • max_length：生成文本的最大长度。这也可以通过使用模型特定参数名称在kwargs 参数中设置。
  • truncate：已弃用。此参数不再有任何效果。
  • streaming_callback: 当从流中接收到新 token 时调用的回调函数。回调函数接受 StreamingChunk 作为参数。
  • boto3_config：boto3 客户端的配置。
  • model_family：要使用的模型族。如果未提供，则根据模型名称选择模型适配器。
  • kwargs：要传递给模型的附加关键字参数。您可以在 AWS Bedrock 的 文档中找到模型特定参数。这些参数特定于模型。您可以在模型的文档中找到它们。

引发:

  • ValueError：如果模型名称为空或 None。
  • AmazonBedrockConfigurationError：如果 AWS 环境配置不正确或模型不受支持。

AmazonBedrockGenerator.run

@component.output_types(replies=List[str], meta=Dict[str, Any])
def run(
    prompt: str,
    streaming_callback: Optional[Callable[[StreamingChunk], None]] = None,
    generation_kwargs: Optional[Dict[str, Any]] = None
) -> Dict[str, Union[List[str], Dict[str, Any]]]

根据给定的提示生成字符串响应列表。

参数:

  • prompt：生成响应的提示。
  • streaming_callback: 当从流中接收到新 token 时调用的回调函数。
  • generation_kwargs：传递给生成器的附加关键字参数。

引发:

  • ValueError：如果提示为空或 None。
  • AmazonBedrockInferenceError：如果无法调用模型。

返回值:

包含以下键的字典

  • replies：生成的响应列表。
  • meta：包含响应元数据的字典。

AmazonBedrockGenerator.get_model_adapter

@classmethod
def get_model_adapter(
        cls,
        model: str,
        model_family: Optional[str] = None) -> Type[BedrockModelAdapter]

获取给定模型的模型适配器。

如果如果提供了 model_family，则返回模型族的适配器。如果未提供 model_family，则根据模型名称自动检测适配器。

参数:

  • model：模型名称。
  • model_family：模型族。

引发:

  • AmazonBedrockConfigurationError：如果模型族不受支持或无法自动检测模型。

返回值:

模型适配器类，如果未找到适配器，则为 None。

AmazonBedrockGenerator.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

AmazonBedrockGenerator.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "AmazonBedrockGenerator"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

模块 haystack_integrations.components.generators.amazon_bedrock.adapters

BedrockModelAdapter

Amazon Bedrock 模型适配器的基类。

此类的每个子类都旨在解决其所适配的特定大型语言模型的独特特性，专注于准备请求和从 Amazon Bedrock 托管的大型语言模型中提取响应。

参数:

  • model_kwargs：模型的关键字参数。您可以在 Amazon Bedrock API 文档中找到完整的参数列表。
  • max_length：生成文本的最大长度。这被映射到每个模型的正确参数。如果model_kwargs 中存在相应的参数，则它将被覆盖。

BedrockModelAdapter.prepare_body

@abstractmethod
def prepare_body(prompt: str, **inference_kwargs: Any) -> Dict[str, Any]

准备 Amazon Bedrock 请求的正文。

每个子类都应该实现此方法以准备特定模型的请求正文。

参数:

  • prompt：要发送到模型的提示。
  • inference_kwargs：传递给处理程序的附加关键字参数。

返回值:

包含请求正文的字典。

BedrockModelAdapter.get_responses

def get_responses(response_body: Dict[str, Any]) -> List[str]

从 Amazon Bedrock 响应中提取响应。

参数:

  • response_body：来自 Amazon Bedrock 请求的响应正文。

返回值:

响应列表。

BedrockModelAdapter.get_stream_responses

def get_stream_responses(
        stream: EventStream,
        streaming_callback: SyncStreamingCallbackT) -> List[str]

从 Amazon Bedrock 流式响应中提取响应。

参数:

  • stream：来自 Amazon Bedrock 请求的流式响应。
  • streaming_callback：流式响应的处理程序。

返回值:

字符串响应列表。

AnthropicClaudeAdapter

Anthropic Claude 模型的适配器。

参数:

  • model_kwargs：模型的关键字参数。您可以在 Claude 模型的 Amazon Bedrock API 文档此处找到完整的参数列表。一些示例参数有
  • use_messages_api：是否使用消息 API，默认值：True
  • include_thinking：是否包含思考输出，默认值：True
  • thinking_tag：思考内容的 XML 标签，默认值："thinking"
  • max_length：生成文本的最大长度

AnthropicClaudeAdapter.prepare_body

def prepare_body(prompt: str, **inference_kwargs: Any) -> Dict[str, Any]

准备 Claude 模型的主体

参数:

  • prompt：要发送到模型的提示。
  • inference_kwargs：传递给处理程序的附加关键字参数。

返回值:

包含以下键的字典

  • prompt：要发送到模型的提示。
  • 指定的推理参数。

MistralAdapter

Mistral 模型的适配器。

MistralAdapter.prepare_body

def prepare_body(prompt: str, **inference_kwargs: Any) -> Dict[str, Any]

准备 Mistral 模型的主体

参数:

  • prompt：要发送到模型的提示。
  • inference_kwargs：传递给处理程序的附加关键字参数。

返回值:

包含以下键的字典

  • prompt：要发送到模型的提示。
  • 指定的推理参数。

CohereCommandAdapter

Cohere Command 模型的适配器。

CohereCommandAdapter.prepare_body

def prepare_body(prompt: str, **inference_kwargs: Any) -> Dict[str, Any]

准备 Command 模型的主体

参数:

  • prompt：要发送到模型的提示。
  • inference_kwargs：传递给处理程序的附加关键字参数。

返回值:

包含以下键的字典

  • prompt：要发送到模型的提示。
  • 指定的推理参数。

CohereCommandRAdapter

Cohere Command R 模型的适配器。

CohereCommandRAdapter.prepare_body

def prepare_body(prompt: str, **inference_kwargs: Any) -> Dict[str, Any]

准备 Command 模型的主体

参数:

  • prompt：要发送到模型的提示。
  • inference_kwargs：传递给处理程序的附加关键字参数。

返回值:

包含以下键的字典

  • prompt：要发送到模型的提示。
  • 指定的推理参数。

AI21LabsJurassic2Adapter

AI21 Labs 的 Jurassic 2 模型的模型适配器。

AI21LabsJurassic2Adapter.prepare_body

def prepare_body(prompt: str, **inference_kwargs: Any) -> Dict[str, Any]

准备 Jurassic 2 模型的主体。

参数:

  • prompt：要发送到模型的提示。
  • inference_kwargs：传递给处理程序的附加关键字参数。

返回值:

包含以下键的字典

  • prompt：要发送到模型的提示。
  • 指定的推理参数。

AmazonTitanAdapter

Amazon Titan 模型的适配器。

AmazonTitanAdapter.prepare_body

def prepare_body(prompt: str, **inference_kwargs: Any) -> Dict[str, Any]

准备 Titan 模型的主体

参数:

  • prompt：要发送到模型的提示。
  • inference_kwargs：传递给处理程序的附加关键字参数。

返回值:

包含以下键的字典

  • inputText：要发送到模型的提示。
  • 指定的推理参数。

MetaLlamaAdapter

Meta Llama2 模型的适配器。

MetaLlamaAdapter.prepare_body

def prepare_body(prompt: str, **inference_kwargs: Any) -> Dict[str, Any]

准备 Llama2 模型的主体

参数:

  • prompt：要发送到模型的提示。
  • inference_kwargs：传递给处理程序的附加关键字参数。

返回值:

包含以下键的字典

  • prompt：要发送到模型的提示。
  • 指定的推理参数。

模块 haystack_integrations.common.amazon_bedrock.errors

AmazonBedrockError

Amazon Bedrock 集成产生的任何错误。

此错误透明地封装了其源，可以直接访问其属性：例如，如果原始错误具有message 属性，AmazonBedrockError.message 将存在并具有预期内容。

AWSConfigurationError

AWS 配置不正确时引发的异常

AmazonBedrockConfigurationError

AmazonBedrock 节点配置不正确时引发的异常

AmazonBedrockInferenceError

Bedrock 推理节点中出现问题的异常

模块 haystack_integrations.components.generators.amazon_bedrock.chat.chat_generator

AmazonBedrockChatGenerator

使用通过 Bedrock Converse API 提供的 Amazon Bedrock 托管的 LLM 完成聊天。

例如，要使用 Anthropic Claude 3 Sonnet 模型，请使用 'anthropic.claude-3-5-sonnet-20240620-v1:0' 模型名称初始化此组件。

使用示例

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("What's Natural Language Processing?")]


client = AmazonBedrockChatGenerator(model="anthropic.claude-3-5-sonnet-20240620-v1:0",
                                    streaming_callback=print_streaming_chunk)
client.run(messages, generation_kwargs={"max_tokens": 512})

多模态示例

from haystack.dataclasses import ChatMessage, ImageContent
from haystack_integrations.components.generators.amazon_bedrock import AmazonBedrockChatGenerator

generator = AmazonBedrockChatGenerator(model="anthropic.claude-3-5-sonnet-20240620-v1:0")

image_content = ImageContent.from_file_path(file_path="apple.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

print(response)
> The image shows a red apple.

### Tool usage example
# AmazonBedrockChatGenerator supports Haystack's unified tool architecture, allowing tools to be used
# across different chat generators. The same tool definitions and usage patterns work consistently
# whether using Amazon Bedrock, OpenAI, Ollama, or any other supported LLM providers.

```python
from haystack.dataclasses import ChatMessage
from haystack.tools import Tool
from haystack_integrations.components.generators.amazon_bedrock import AmazonBedrockChatGenerator

def weather(city: str):
    return f'The weather in {city} is sunny and 32°C'

__Define tool parameters__

tool_parameters = {
    "type": "object",
    "properties": {"city": {"type": "string"}},
    "required": ["city"]
}

__Create weather tool__

weather_tool = Tool(
    name="weather",
    description="useful to determine the weather in a given location",
    parameters=tool_parameters,
    function=weather
)

__Initialize generator with tool__

client = AmazonBedrockChatGenerator(
    model="anthropic.claude-3-5-sonnet-20240620-v1:0",
    tools=[weather_tool]
)

__Run initial query__

messages = [ChatMessage.from_user("What's the weather like in Paris?")]
results = client.run(messages=messages)

__Get tool call from response__

tool_message = next(msg for msg in results["replies"] if msg.tool_call)
tool_call = tool_message.tool_call

__Execute tool and send result back__

weather_result = weather(**tool_call.arguments)
new_messages = [
    messages[0],
    tool_message,
    ChatMessage.from_tool(tool_result=weather_result, origin=tool_call)
]

__Get final response__

final_result = client.run(new_messages)
print(final_result["replies"][0].text)

> Based on the information I've received, I can tell you that the weather in Paris is
> currently sunny with a temperature of 32°C (which is about 90°F).

AmazonBedrockChatGenerator 使用 AWS 进行身份验证。您可以使用 AWS CLI 通过 IAM 进行身份验证。有关设置 IAM 基于身份策略的更多信息，请参阅 [Amazon Bedrock 文档] (https://docs.aws.amazon.com/bedrock/latest/userguide/security_iam_id-based-policy-examples.html)。

如果 AWS 环境配置正确，则不需要 AWS 凭证，因为它们会自动从环境或 AWS 配置文件中加载。如果 AWS 环境未配置，请设置aws_access_key_id, aws_secret_access_key 和aws_region_name 设置为环境变量或将其作为 Secret 参数传递。确保您设置的区域支持 Amazon Bedrock。

AmazonBedrockChatGenerator.__init__

def __init__(
        model: str,
        aws_access_key_id: Optional[Secret] = Secret.from_env_var(
            ["AWS_ACCESS_KEY_ID"], strict=False),
        aws_secret_access_key: Optional[Secret] = Secret.
    from_env_var(  # noqa: B008
        ["AWS_SECRET_ACCESS_KEY"], strict=False),
        aws_session_token: Optional[Secret] = Secret.from_env_var(
            ["AWS_SESSION_TOKEN"], strict=False),
        aws_region_name: Optional[Secret] = Secret.from_env_var(
            ["AWS_DEFAULT_REGION"], strict=False),
        aws_profile_name: Optional[Secret] = Secret.from_env_var(
            ["AWS_PROFILE"], strict=False),
        generation_kwargs: Optional[Dict[str, Any]] = None,
        streaming_callback: Optional[StreamingCallbackT] = None,
        boto3_config: Optional[Dict[str, Any]] = None,
        tools: Optional[Union[List[Tool], Toolset]] = None,
        *,
        guardrail_config: Optional[Dict[str, str]] = None) -> None

使用提供的参数初始化AmazonBedrockChatGenerator。参数传递给

Amazon Bedrock 客户端。

请注意，如果 AWS 环境配置正确，则不需要 AWS 凭证。这些凭证会自动从环境或 AWS 配置文件中加载，无需通过构造函数显式提供。如果 AWS 环境未配置，用户需要通过构造函数提供 AWS 凭证。除了模型之外，三个必需参数是aws_access_key_id, aws_secret_access_key 和aws_region_name.

参数:

  • model：用于文本生成的模型。模型必须在 Amazon Bedrock 中可用，并且必须以 Amazon Bedrock 文档中概述的格式指定。
  • aws_access_key_id：AWS 访问密钥 ID。
  • aws_secret_access_key：AWS 秘密访问密钥。
  • aws_session_token：AWS 会话令牌。
  • aws_region_name：AWS 区域名称。确保您设置的区域支持 Amazon Bedrock。
  • aws_profile_name：AWS 配置文件名称。
  • generation_kwargs：发送到模型的关键字参数。这些参数特定于模型。您可以在 AWS Bedrock API 文档中找到模型特定参数。
  • streaming_callback：从流中收到新令牌时调用的回调函数。默认情况下，模型未设置为流式传输。要启用流式传输，请将此参数设置为处理流式传输块的回调函数。回调函数接收一个 StreamingChunk 对象并打开流式传输模式。
  • boto3_config：boto3 客户端的配置。
  • tools：模型可以使用的一系列 Tool 对象或 Toolset。每个工具都应有一个唯一的名称。
  • guardrail_config：在 Amazon Bedrock 中创建的防护的可选配置。这必须以字典形式提供，与 GuardrailConfiguration 匹配，或者在流式传输模式下（当streaming_callback 设置时），匹配 GuardrailStreamConfiguration。如果trace 设置为enabled，则防护跟踪将包含在生成的 ChatMessage 的 meta 属性的 trace 键下。注意：在流式传输模式下启用防护可能会引入额外的延迟。要管理此问题，您可以调整的 ChatMessage 的 meta 属性的 trace 键下。注意：在流式传输模式下启用防护可能会引入额外的延迟。要管理此问题，您可以调整streamProcessingMode 参数。有关更多信息，请参阅 Guardrails 流式传输文档。streamProcessingMode 参数。有关更多信息，请参阅 Guardrails 流式传输文档。

引发:

  • ValueError：如果模型名称为空或 None。
  • AmazonBedrockConfigurationError：如果 AWS 环境配置不正确或模型不受支持。

AmazonBedrockChatGenerator.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

AmazonBedrockChatGenerator.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "AmazonBedrockChatGenerator"

从字典反序列化组件。

参数:

  • data: 包含序列化数据的字典。

返回值:

AmazonBedrockChatGenerator 实例。AmazonBedrockChatGenerator.

AmazonBedrockChatGenerator.run

@component.output_types(replies=List[ChatMessage])
def run(
    messages: List[ChatMessage],
    streaming_callback: Optional[StreamingCallbackT] = None,
    generation_kwargs: Optional[Dict[str, Any]] = None,
    tools: Optional[Union[List[Tool], Toolset]] = None
) -> Dict[str, List[ChatMessage]]

使用 Converse API 对 Amazon Bedrock 模型执行同步推理调用。

根据是否提供流式回调，支持标准响应和流式响应。

参数:

  • messages：一组ChatMessage 对象构成聊天历史记录。
  • streaming_callback：用于处理流式输出的可选回调。
  • generation_kwargs：生成参数的可选字典。一些常见参数是
  • maxTokens：要生成的最大令牌数。
  • stopSequences：停止生成的停止序列列表。
  • temperature：采样温度。
  • topP：核采样参数。
  • tools：模型在执行期间可能调用的可选工具列表。

引发:

  • AmazonBedrockInferenceError：如果 Bedrock 推理 API 调用失败。

返回值:

包含模型生成回复的字典，位于"replies" 键下。

AmazonBedrockChatGenerator.run_async

@component.output_types(replies=List[ChatMessage])
async def run_async(
    messages: List[ChatMessage],
    streaming_callback: Optional[StreamingCallbackT] = None,
    generation_kwargs: Optional[Dict[str, Any]] = None,
    tools: Optional[Union[List[Tool], Toolset]] = None
) -> Dict[str, List[ChatMessage]]

使用 Converse API 对 Amazon Bedrock 模型执行异步推理调用。

专为需要非阻塞或并发执行的用例而设计。

参数:

  • messages：一组ChatMessage 对象构成聊天历史记录。
  • streaming_callback：用于处理流式输出的可选异步兼容回调。
  • generation_kwargs：生成参数的可选字典。一些常见参数是
  • maxTokens：要生成的最大令牌数。
  • stopSequences：停止生成的停止序列列表。
  • temperature：采样温度。
  • topP：核采样参数。
  • tools：模型可用的可选工具对象列表或工具集。

引发:

  • AmazonBedrockInferenceError：如果 Bedrock 推理 API 调用失败。

返回值:

包含模型生成回复的字典，位于"replies" 键下。

模块 haystack_integrations.components.rankers.amazon_bedrock.ranker

AmazonBedrockRanker

使用 Amazon Bedrock 的 Cohere Rerank 模型根据文档与查询的相似性对文档进行排名。

Documents 按与查询的语义相关性从高到低排序。

支持的 Amazon Bedrock 模型

  • 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(
    model="cohere.rerank-v3-5:0",
    top_k=2,
    aws_region_name=Secret.from_token("eu-central-1")
)

docs = [Document(content="Paris"), Document(content="Berlin")]
query = "What is the capital of germany?"
output = ranker.run(query=query, documents=docs)
docs = output["documents"]

AmazonBedrockRanker 使用 AWS 进行身份验证。您可以使用 AWS CLI 通过 IAM 进行身份验证。有关设置 IAM 基于身份策略的更多信息，请参阅 [Amazon Bedrock 文档] (https://docs.aws.amazon.com/bedrock/latest/userguide/security_iam_id-based-policy-examples.html)。

如果 AWS 环境配置正确，则不需要 AWS 凭证，因为它们会自动从环境或 AWS 配置文件中加载。如果 AWS 环境未配置，请设置aws_access_key_id, aws_secret_access_key 和aws_region_name 设置为环境变量或将其作为 Secret 参数传递。确保您设置的区域支持 Amazon Bedrock。

AmazonBedrockRanker.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

AmazonBedrockRanker.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "AmazonBedrockRanker"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

AmazonBedrockRanker.run

@component.output_types(documents=List[Document])
def run(query: str,
        documents: List[Document],
        top_k: Optional[int] = None) -> Dict[str, List[Document]]

使用 Amazon Bedrock Reranker 根据查询对文档列表进行重新排名。

参数:

  • query: 查询字符串。
  • documents: Document 列表。
  • top_k: 您希望 Ranker 返回的最大 Document 数量。

引发:

  • ValueError: 如果top_k 不大于 0。

返回值:

包含以下键的字典

  • documents: 按相似度降序排列的、与给定查询最相似的 Document 列表。

模块 haystack_integrations.components.downloaders.s3.s3_downloader

S3Downloader

一个用于将文件从 AWS S3 存储桶下载到本地文件系统的组件。支持按文件扩展名过滤。

S3Downloader.__init__

def __init__(
    *,
    aws_access_key_id: Optional[Secret] = Secret.from_env_var(
        "AWS_ACCESS_KEY_ID", strict=False),
    aws_secret_access_key: Optional[Secret] = Secret.
    from_env_var(  # noqa: B008
        "AWS_SECRET_ACCESS_KEY", strict=False),
    aws_session_token: Optional[Secret] = Secret.from_env_var(
        "AWS_SESSION_TOKEN", strict=False),
    aws_region_name: Optional[Secret] = Secret.from_env_var(
        "AWS_DEFAULT_REGION", strict=False),
    aws_profile_name: Optional[Secret] = Secret.from_env_var("AWS_PROFILE",
                                                             strict=False),
    boto3_config: Optional[Dict[str, Any]] = None,
    file_root_path: Optional[str] = None,
    file_extensions: Optional[List[str]] = None,
    file_name_meta_key: str = "file_name",
    max_workers: int = 32,
    max_cache_size: int = 100,
    s3_key_generation_function: Optional[Callable[[Document], str]] = None
) -> None

使用提供的参数初始化使用提供的参数初始化 S3Downloader。

请注意，如果 AWS 环境配置正确，则不需要 AWS 凭证。这些凭证会自动从环境或 AWS 配置文件中加载，无需通过构造函数显式提供。如果 AWS 环境未配置，用户需要通过构造函数提供 AWS 凭证。三个必需参数是aws_access_key_id, aws_secret_access_key 和aws_region_name.

参数:

  • aws_access_key_id：AWS 访问密钥 ID。
  • aws_secret_access_key：AWS 秘密访问密钥。
  • aws_session_token：AWS 会话令牌。
  • aws_region_name：AWS 区域名称。
  • aws_profile_name：AWS 配置文件名称。
  • boto3_config：boto3 客户端的配置。
  • file_root_path：文件将下载到的路径。可以通过此参数或FILE_ROOT_PATH 环境变量设置。如果两者都未设置，则会引发ValueError。
  • file_extensions：允许下载的文件扩展名。默认情况下，允许所有文件扩展名。
  • max_workers：用于并发下载的最大工作线程数。
  • max_cache_size：要缓存的最大文件数。
  • file_name_meta_key：包含要下载的文件名的元键的名称。文件名也将用于创建本地文件路径进行下载。默认情况下，使用Document.meta["file_name"]。如果您想使用Document.meta 中的不同键，您可以在此处设置。
  • s3_key_generation_function：一个可选函数，用于生成要下载的文件的 S3 键。如果未提供，默认行为是使用Document.meta[file_name_meta_key]。该函数必须接受一个Document 对象并返回一个字符串。如果设置了环境变量S3_DOWNLOADER_PREFIX，其值将自动作为前缀添加到生成的 S3 键中。

引发:

  • ValueError：如果未通过构造函数或FILE_ROOT_PATH 环境变量设置 file_root_path。FILE_ROOT_PATH 环境变量。

S3Downloader.warm_up

def warm_up() -> None

通过初始化设置和存储来预热组件。

S3Downloader.run

@component.output_types(documents=List[Document])
def run(documents: List[Document]) -> Dict[str, List[Document]]

将文件从 AWS S3 存储桶下载到本地文件系统。

返回包含下载文件路径的增强Document。

参数:

  • documents：在元字段中包含要下载的文件名的文档。

引发:

  • S3Error：如果下载尝试失败或 S3 存储桶中不存在文件。
  • ValueError：如果未设置文件下载路径。

返回值:

一个字典，包含

  • documents：已下载的Document；每个都具有meta['file_path'].

S3Downloader.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

S3Downloader.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "S3Downloader"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

模块 haystack_integrations.common.s3.utils

S3Storage

此类提供了一个用于从 AWS S3 存储桶下载文件的存储类。

S3Storage.__init__

def __init__(s3_bucket: str,
             session: Session,
             s3_prefix: Optional[str] = None,
             endpoint_url: Optional[str] = None,
             config: Optional[Config] = None) -> None

使用提供的参数初始化 S3Storage 对象。

参数:

  • s3_bucket：要下载文件的 S3 存储桶的名称。
  • session：用于 S3 客户端的会话。
  • s3_prefix：S3 存储桶中文件的可选前缀。可用于指定文件夹或命名结构。例如，如果文件位于“folder/subfolder/file.txt”文件夹中，则 s3_prefix 应为“folder/subfolder/”。如果文件位于 S3 存储桶的根目录，则 s3_prefix 应为 None。
  • endpoint_url：要下载文件的 S3 存储桶的端点 URL。
  • config：用于 S3 客户端的配置。

S3Storage.download

def download(key: str, local_file_path: Path) -> None

从 S3 下载文件。

参数:

  • key：要下载的文件的键。
  • local_file_path：文件下载到的文件夹路径。如果不存在，则会创建。文件将下载到与键同名的文件夹中。

引发:

  • S3ConfigurationError：如果无法创建 S3 会话客户端。
  • S3StorageError：如果 S3 存储桶中不存在文件或无法下载文件。

S3Storage.from_env

@classmethod
def from_env(cls, *, session: Session, config: Config) -> "S3Storage"

从环境变量创建 S3Storage 对象。

模块 haystack_integrations.common.s3.errors

S3Error

S3 相关组件中出现问题的异常

S3ConfigurationError

AmazonS3 节点配置不正确时引发的异常

S3StorageError

在与 S3Storage 对象交互时发生错误时引发此异常。