Nvidia 集成,用于 Haystack
模块 haystack_integrations.components.embedders.nvidia.document_embedder
NvidiaDocumentEmbedder
一个组件,用于使用 NVIDIA NIMs 提供的嵌入模型来嵌入文档。
使用示例
from haystack_integrations.components.embedders.nvidia import NvidiaDocumentEmbedder
doc = Document(content="I love pizza!")
text_embedder = NvidiaDocumentEmbedder(model="NV-Embed-QA", api_url="https://ai.api.nvidia.com/v1/retrieval/nvidia")
text_embedder.warm_up()
result = document_embedder.run([doc])
print(result["documents"][0].embedding)
NvidiaDocumentEmbedder.__init__
def __init__(model: Optional[str] = None,
api_key: Optional[Secret] = Secret.from_env_var("NVIDIA_API_KEY"),
api_url: str = os.getenv("NVIDIA_API_URL", DEFAULT_API_URL),
prefix: str = "",
suffix: str = "",
batch_size: int = 32,
progress_bar: bool = True,
meta_fields_to_embed: Optional[List[str]] = None,
embedding_separator: str = "\n",
truncate: Optional[Union[EmbeddingTruncateMode, str]] = None,
timeout: Optional[float] = None)
创建一个 NvidiaTextEmbedder 组件。
参数:
model: 要使用的嵌入模型。如果未提供特定的模型以及本地托管的 API URL,系统将默认使用通过 /models API 找到的可用模型。api_key: NVIDIA NIM 的 API 密钥。api_url: NVIDIA NIM 的自定义 API URL。API URL 的格式为 http://host:portprefix: 添加到每个文本开头的字符串。suffix: 添加到每个文本末尾的字符串。batch_size: 一次要编码的文档数量。不能大于 50。progress_bar: Whether to show a progress bar or not.meta_fields_to_embed: 需要与 Document 文本一起进行嵌入的元字段列表。embedding_separator: 用于将元字段连接到 Document 文本的分隔符。truncate: 指定如何截断长于最大令牌长度的输入。如果为 None,行为将取决于模型,有关更多信息,请参阅官方文档。timeout: 请求调用的超时时间,如果未设置,则从NVIDIA_TIMEOUT环境变量推断,或默认为 60。
NvidiaDocumentEmbedder.default_model
def default_model()
在本地 NIM 模式下设置默认模型。
NvidiaDocumentEmbedder.warm_up
def warm_up()
Initializes the component.
NvidiaDocumentEmbedder.to_dict
def to_dict() -> Dict[str, Any]
将组件序列化为字典。
返回值:
包含序列化数据的字典。
NvidiaDocumentEmbedder.available_models
@property
def available_models() -> List[Model]
获取可与 NvidiaDocumentEmbedder 一起使用模型的列表。
NvidiaDocumentEmbedder.from_dict
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "NvidiaDocumentEmbedder"
从字典反序列化组件。
参数:
data: 要反序列化的字典。
返回值:
反序列化后的组件。
NvidiaDocumentEmbedder.run
@component.output_types(documents=List[Document], meta=Dict[str, Any])
def run(
documents: List[Document]
) -> Dict[str, Union[List[Document], Dict[str, Any]]]
嵌入文档列表。
每个 Document 的嵌入存储在Document 的 embedding 字段中。
参数:
documents: 要嵌入的 Document 列表。
引发:
RuntimeError: 如果组件未初始化。TypeError: 如果输入不是字符串。
返回值:
一个包含以下键值对的字典
documents- 包含嵌入的已处理文档的列表。meta- 关于使用统计等元数据。
模块 haystack_integrations.components.embedders.nvidia.text_embedder
NvidiaTextEmbedder
一个组件,用于使用 NVIDIA NIMs 提供的嵌入模型来嵌入字符串。
对于区分查询和文档输入的模型,此组件将输入字符串作为查询进行嵌入。
使用示例
from haystack_integrations.components.embedders.nvidia import NvidiaTextEmbedder
text_to_embed = "I love pizza!"
text_embedder = NvidiaTextEmbedder(model="NV-Embed-QA", api_url="https://ai.api.nvidia.com/v1/retrieval/nvidia")
text_embedder.warm_up()
print(text_embedder.run(text_to_embed))
NvidiaTextEmbedder.__init__
def __init__(model: Optional[str] = None,
api_key: Optional[Secret] = Secret.from_env_var("NVIDIA_API_KEY"),
api_url: str = os.getenv("NVIDIA_API_URL", DEFAULT_API_URL),
prefix: str = "",
suffix: str = "",
truncate: Optional[Union[EmbeddingTruncateMode, str]] = None,
timeout: Optional[float] = None)
创建一个 NvidiaTextEmbedder 组件。
参数:
model: 要使用的嵌入模型。如果未提供特定的模型以及本地托管的 API URL,系统将默认使用通过 /models API 找到的可用模型。api_key: NVIDIA NIM 的 API 密钥。api_url: NVIDIA NIM 的自定义 API URL。API URL 的格式为 http://host:portprefix: 添加到每个文本开头的字符串。suffix: 添加到每个文本末尾的字符串。truncate: 指定如何截断长于最大令牌长度的输入。如果为 None,行为将取决于模型,有关更多信息,请参阅官方文档。timeout: 请求调用的超时时间,如果未设置,则从NVIDIA_TIMEOUT环境变量推断,或默认为 60。
NvidiaTextEmbedder.default_model
def default_model()
在本地 NIM 模式下设置默认模型。
NvidiaTextEmbedder.warm_up
def warm_up()
Initializes the component.
NvidiaTextEmbedder.to_dict
def to_dict() -> Dict[str, Any]
将组件序列化为字典。
返回值:
包含序列化数据的字典。
NvidiaTextEmbedder.available_models
@property
def available_models() -> List[Model]
获取可与 NvidiaTextEmbedder 一起使用模型的列表。
NvidiaTextEmbedder.from_dict
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "NvidiaTextEmbedder"
从字典反序列化组件。
参数:
data: 要反序列化的字典。
返回值:
反序列化后的组件。
NvidiaTextEmbedder.run
@component.output_types(embedding=List[float], meta=Dict[str, Any])
def run(text: str) -> Dict[str, Union[List[float], Dict[str, Any]]]
嵌入一个字符串。
参数:
text: The text to embed.
引发:
RuntimeError: 如果组件未初始化。TypeError: 如果输入不是字符串。
返回值:
一个包含以下键值对的字典
embedding- 文本的嵌入。meta- 关于使用统计等元数据。
模块 haystack_integrations.components.embedders.nvidia.truncate
EmbeddingTruncateMode
指定 NVIDIA 嵌入组件的输入如何被截断。如果为 START,则从开头截断输入。如果为 END,则从结尾截断输入。如果为 NONE,则返回错误(如果输入过长)。
EmbeddingTruncateMode.from_str
@classmethod
def from_str(cls, string: str) -> "EmbeddingTruncateMode"
从字符串创建截断模式。
参数:
string: 要转换的字符串。
返回值:
截断模式。
模块 haystack_integrations.components.generators.nvidia.generator
NvidiaGenerator
使用托管在 NVIDIA NIM 上,并在 NVIDIA API Catalog 中的生成模型生成文本。
使用示例
from haystack_integrations.components.generators.nvidia import NvidiaGenerator
generator = NvidiaGenerator(
model="meta/llama3-70b-instruct",
model_arguments={
"temperature": 0.2,
"top_p": 0.7,
"max_tokens": 1024,
},
)
generator.warm_up()
result = generator.run(prompt="What is the answer?")
print(result["replies"])
print(result["meta"])
print(result["usage"])
此组件需要 NVIDIA API 密钥才能工作。
NvidiaGenerator.__init__
def __init__(model: Optional[str] = None,
api_url: str = os.getenv("NVIDIA_API_URL", DEFAULT_API_URL),
api_key: Optional[Secret] = Secret.from_env_var("NVIDIA_API_KEY"),
model_arguments: Optional[Dict[str, Any]] = None,
timeout: Optional[float] = None)
创建一个 NvidiaGenerator 组件。
参数:
model: 用于文本生成的模型名称。有关支持的模型,请参阅 NVIDIA NIMs。Note: 如果未提供特定的模型以及本地托管的 API URL,系统将默认使用通过 /models API 找到的可用模型。请在 NVIDIA NIM 查看支持的模型。api_key: NVIDIA NIM 的 API 密钥。可以将其设置为NVIDIA_API_KEY环境变量,或在此处传递。api_url: NVIDIA NIM 的自定义 API URL。model_arguments: 要传递给模型提供商的其他参数。这些参数特定于模型。请在 NVIDIA NIM 中搜索您的模型,以查找它接受的参数。timeout: 请求调用的超时时间,如果未设置,则从NVIDIA_TIMEOUT环境变量推断,或默认为 60。
NvidiaGenerator.default_model
def default_model()
在本地 NIM 模式下设置默认模型。
NvidiaGenerator.warm_up
def warm_up()
Initializes the component.
NvidiaGenerator.to_dict
def to_dict() -> Dict[str, Any]
将组件序列化为字典。
返回值:
包含序列化数据的字典。
NvidiaGenerator.available_models
@property
def available_models() -> List[Model]
获取可与 ChatNVIDIA 一起使用模型的列表。
NvidiaGenerator.from_dict
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "NvidiaGenerator"
从字典反序列化组件。
参数:
data: 要反序列化的字典。
返回值:
反序列化后的组件。
NvidiaGenerator.run
@component.output_types(replies=List[str], meta=List[Dict[str, Any]])
def run(prompt: str) -> Dict[str, Union[List[str], List[Dict[str, Any]]]]
使用提供的提示查询模型。
参数:
prompt: 要发送给生成模型的文本。
返回值:
包含以下键的字典
replies- 模型生成的回复。meta- 每个回复的元数据。
模块 haystack_integrations.components.rankers.nvidia.ranker
NvidiaRanker
一个组件,用于使用 NVIDIA NIMs 提供的排序模型来对文档进行排序。
使用示例
from haystack_integrations.components.rankers.nvidia import NvidiaRanker
from haystack import Document
from haystack.utils import Secret
ranker = NvidiaRanker(
model="nvidia/nv-rerankqa-mistral-4b-v3",
api_key=Secret.from_env_var("NVIDIA_API_KEY"),
)
ranker.warm_up()
query = "What is the capital of Germany?"
documents = [
Document(content="Berlin is the capital of Germany."),
Document(content="The capital of Germany is Berlin."),
Document(content="Germany's capital is Berlin."),
]
result = ranker.run(query, documents, top_k=2)
print(result["documents"])
NvidiaRanker.__init__
def __init__(model: Optional[str] = None,
truncate: Optional[Union[RankerTruncateMode, str]] = None,
api_url: str = os.getenv("NVIDIA_API_URL", DEFAULT_API_URL),
api_key: Optional[Secret] = Secret.from_env_var("NVIDIA_API_KEY"),
top_k: int = 5,
query_prefix: str = "",
document_prefix: str = "",
meta_fields_to_embed: Optional[List[str]] = None,
embedding_separator: str = "\n",
timeout: Optional[float] = None)
创建一个 NvidiaRanker 组件。
参数:
model: 要使用的排序模型。truncate: 要使用的截断策略。可以是 "NONE"、"END" 或 RankerTruncateMode。默认为 NIM 的默认值。api_key: NVIDIA NIM 的 API 密钥。api_url: NVIDIA NIM 的自定义 API URL。top_k: 要返回的文档数量。query_prefix: 在排序之前添加到查询文本开头的字符串。可以使用它来在文本前面加上指令,如重排序模型(如bge)所需。.document_prefix: 在排序之前添加到每个文档开头的字符串。可以使用它来在文档前面加上指令,如嵌入模型(如bge)所需。.meta_fields_to_embed: 要与文档一起嵌入的元数据字段列表。embedding_separator: 用于连接元数据字段到文档的分隔符。timeout: 请求调用的超时时间,如果未设置,则从NVIDIA_TIMEOUT环境变量推断,或默认为 60。
NvidiaRanker.to_dict
def to_dict() -> Dict[str, Any]
将排序器序列化为字典。
返回值:
包含排序器属性的字典。
NvidiaRanker.from_dict
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "NvidiaRanker"
从字典反序列化排序器。
参数:
data: 包含排序器属性的字典。
返回值:
反序列化后的排序器。
NvidiaRanker.warm_up
def warm_up()
初始化排序器。
引发:
ValueError: 如果托管的 NVIDIA NIM 需要 API 密钥。
NvidiaRanker.run
@component.output_types(documents=List[Document])
def run(query: str,
documents: List[Document],
top_k: Optional[int] = None) -> Dict[str, List[Document]]
根据给定查询对文档列表进行排序。
参数:
query: 用于对文档进行排序的查询。documents: 要排序的文档列表。top_k: 要返回的文档数量。
引发:
RuntimeError: 如果排序器尚未加载。TypeError: 如果参数类型错误。
返回值:
一个包含排序后文档的字典。
模块 haystack_integrations.components.rankers.nvidia.truncate
RankerTruncateMode
指定 NVIDIA 排序器组件的输入如何被截断。如果为 NONE,则不截断输入并返回错误。如果为 END,则从结尾截断输入。
RankerTruncateMode.from_str
@classmethod
def from_str(cls, string: str) -> "RankerTruncateMode"
从字符串创建截断模式。
参数:
string: 要转换的字符串。
返回值:
截断模式。
