Cohere 集成,用于 Haystack
模块 haystack_integrations.components.embedders.cohere.document_embedder
CohereDocumentEmbedder
一个使用 Cohere 模型计算 Document embedding 的组件。
每个 Document 的嵌入存储在Document 的 embedding 字段中。
使用示例
from haystack import Document
from haystack_integrations.components.embedders.cohere import CohereDocumentEmbedder
doc = Document(content="I love pizza!")
document_embedder = CohereDocumentEmbedder()
result = document_embedder.run([doc])
print(result['documents'][0].embedding)
# [-0.453125, 1.2236328, 2.0058594, ...]
CohereDocumentEmbedder.__init__
def __init__(api_key: Secret = Secret.from_env_var(
["COHERE_API_KEY", "CO_API_KEY"]),
model: str = "embed-english-v2.0",
input_type: str = "search_document",
api_base_url: str = "https://api.cohere.com",
truncate: str = "END",
timeout: float = 120.0,
batch_size: int = 32,
progress_bar: bool = True,
meta_fields_to_embed: Optional[List[str]] = None,
embedding_separator: str = "\n",
embedding_type: Optional[EmbeddingTypes] = None)
参数:
api_key:Cohere API 密钥。model:要使用的模型名称。支持的模型包括"embed-english-v3.0","embed-english-light-v3.0","embed-multilingual-v3.0","embed-multilingual-light-v3.0","embed-english-v2.0","embed-english-light-v2.0","embed-multilingual-v2.0"。所有支持的模型列表可以在模型文档中找到。input_type:指定您要提供给模型的输入类型。支持的值有 "search_document", "search_query", "classification" 和 "clustering"。对于旧版本的 embedding 模型(即 v3 以下版本)不是必需的,但对于较新版本(即 v2 以上版本)是必需的。api_base_url:Cohere API 的基本 URL。truncate:截断过长的 embedding,可以从开头或结尾进行截断("NONE"|"START"|"END")。选择 "START" 会丢弃输入的开头部分。"END" 会丢弃输入的结尾部分。在这两种情况下,输入都会被丢弃,直到剩余的输入长度正好是模型的最大输入 token 长度。如果选择 "NONE",当输入超过最大输入 token 长度时,将返回错误。timeout:请求超时(秒)。batch_size:一次编码的 Document 数量。progress_bar:是否显示进度条。在生产环境中禁用进度条可以使日志更干净。meta_fields_to_embed:应与 Document 文本一起进行 embedding 的元字段列表。embedding_separator:用于将元字段连接到 Document 文本的分隔符。embedding_type:返回的 embedding 类型。默认为 float embedding。请注意,int8、uint8、binary 和 ubinary 仅对 v3 模型有效。
CohereDocumentEmbedder.to_dict
def to_dict() -> Dict[str, Any]
将组件序列化为字典。
返回值:
包含序列化数据的字典。
CohereDocumentEmbedder.from_dict
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "CohereDocumentEmbedder"
从字典反序列化组件。
参数:
data: 要反序列化的字典。
返回值:
反序列化后的组件。
CohereDocumentEmbedder.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 列表Documents.
参数:
documents:要嵌入的 Documents。
引发:
TypeError:如果输入不是 Document 列表Documents.
返回值:
包含以下键的字典
documents:带有embedding字段设置的 Documents。meta:关于 embedding 过程的元数据。
CohereDocumentEmbedder.run_async
@component.output_types(documents=List[Document], meta=Dict[str, Any])
async def run_async(
documents: List[Document]
) -> Dict[str, Union[List[Document], Dict[str, Any]]]
嵌入一个 Document 列表异步地嵌入 Documents。
参数:
documents:要嵌入的 Documents。
引发:
TypeError:如果输入不是 Document 列表Documents.
返回值:
包含以下键的字典
documents:带有embedding字段设置的 Documents。meta:关于 embedding 过程的元数据。
模块 haystack_integrations.components.embedders.cohere.document_image_embedder
CohereDocumentImageEmbedder
一个使用 Cohere 模型基于图像计算 Document embedding 的组件。
每个 Document 的嵌入存储在Document 的 embedding 字段中。
使用示例
from haystack import Document
from haystack_integrations.components.embedders.cohere import CohereDocumentImageEmbedder
embedder = CohereDocumentImageEmbedder(model="embed-v4.0")
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 1536),
# ...]
CohereDocumentImageEmbedder.__init__
def __init__(*,
file_path_meta_field: str = "file_path",
root_path: Optional[str] = None,
image_size: Optional[Tuple[int, int]] = None,
api_key: Secret = Secret.from_env_var(
["COHERE_API_KEY", "CO_API_KEY"]),
model: str = "embed-v4.0",
api_base_url: str = "https://api.cohere.com",
timeout: float = 120.0,
embedding_dimension: Optional[int] = None,
embedding_type: EmbeddingTypes = EmbeddingTypes.FLOAT,
progress_bar: bool = True) -> None
创建一个 CohereDocumentImageEmbedder 组件。
参数:
file_path_meta_field: Document 中包含图像或 PDF 文件路径的元数据字段。root_path: 存储文档文件的根目录路径。如果提供,文档元数据中的文件路径将相对于此路径解析。如果为 None,则文件路径被视为绝对路径。image_size:如果提供,则将图像调整到指定尺寸(宽度、高度)以保持纵横比。这可以减小文件大小、内存使用和处理时间,这在处理具有分辨率限制的模型或将图像传输到远程服务时非常有用。api_key:Cohere API 密钥。model:用于计算 embedding 的 Cohere 模型。请阅读Cohere 文档以获取所有支持的模型列表。api_base_url:Cohere API 的基本 URL。timeout:请求超时(秒)。embedding_dimension:要返回的 embedding 的维度。仅对 v4 及更高版本模型有效。请阅读Cohere API 参考以获取可能的取值和支持的模型列表。embedding_type:要返回的 embedding 类型。默认为 float embedding。仅支持 Embed v3.0 及更高版本模型指定非 float 类型。progress_bar: 是否显示进度条。在生产部署中禁用此选项有助于保持日志的整洁。
CohereDocumentImageEmbedder.to_dict
def to_dict() -> dict[str, Any]
将组件序列化为字典。
返回值:
包含序列化数据的字典。
CohereDocumentImageEmbedder.from_dict
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "CohereDocumentImageEmbedder"
从字典反序列化组件。
参数:
data: 要反序列化的字典。
返回值:
反序列化后的组件。
CohereDocumentImageEmbedder.run
@component.output_types(documents=list[Document])
def run(documents: list[Document]) -> dict[str, list[Document]]
嵌入一个图像 Document 列表。
参数:
documents: 要嵌入的 Documents。
返回值:
包含以下键的字典
documents: 带有嵌入的 Documents。
CohereDocumentImageEmbedder.run_async
@component.output_types(documents=list[Document])
async def run_async(documents: list[Document]) -> dict[str, list[Document]]
异步嵌入一个图像 Document 列表。
参数:
documents: 要嵌入的 Documents。
返回值:
包含以下键的字典
documents: 带有嵌入的 Documents。
模块 haystack_integrations.components.embedders.cohere.text_embedder
CohereTextEmbedder
一个使用 Cohere 模型嵌入字符串的组件。
使用示例
from haystack_integrations.components.embedders.cohere import CohereTextEmbedder
text_to_embed = "I love pizza!"
text_embedder = CohereTextEmbedder()
print(text_embedder.run(text_to_embed))
# {'embedding': [-0.453125, 1.2236328, 2.0058594, ...]
# 'meta': {'api_version': {'version': '1'}, 'billed_units': {'input_tokens': 4}}}
CohereTextEmbedder.__init__
def __init__(api_key: Secret = Secret.from_env_var(
["COHERE_API_KEY", "CO_API_KEY"]),
model: str = "embed-english-v2.0",
input_type: str = "search_query",
api_base_url: str = "https://api.cohere.com",
truncate: str = "END",
timeout: float = 120.0,
embedding_type: Optional[EmbeddingTypes] = None)
参数:
api_key:Cohere API 密钥。model:要使用的模型名称。支持的模型包括"embed-english-v3.0","embed-english-light-v3.0","embed-multilingual-v3.0","embed-multilingual-light-v3.0","embed-english-v2.0","embed-english-light-v2.0","embed-multilingual-v2.0"。所有支持的模型列表可以在模型文档中找到。input_type:指定您要提供给模型的输入类型。支持的值有 "search_document", "search_query", "classification" 和 "clustering"。对于旧版本的 embedding 模型(即 v3 以下版本)不是必需的,但对于较新版本(即 v2 以上版本)是必需的。api_base_url:Cohere API 的基本 URL。truncate:截断过长的 embedding,可以从开头或结尾进行截断("NONE"|"START"|"END")。选择 "START" 会丢弃输入的开头部分。"END" 会丢弃输入的结尾部分。在这两种情况下,输入都会被丢弃,直到剩余的输入长度正好是模型的最大输入 token 长度。如果选择 "NONE",当输入超过最大输入 token 长度时,将返回错误。timeout:请求超时(秒)。embedding_type:返回的 embedding 类型。默认为 float embedding。请注意,int8、uint8、binary 和 ubinary 仅对 v3 模型有效。
CohereTextEmbedder.to_dict
def to_dict() -> Dict[str, Any]
将组件序列化为字典。
返回值:
包含序列化数据的字典。
CohereTextEmbedder.from_dict
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "CohereTextEmbedder"
从字典反序列化组件。
参数:
data: 要反序列化的字典。
返回值:
反序列化后的组件。
CohereTextEmbedder.run
@component.output_types(embedding=List[float], meta=Dict[str, Any])
def run(text: str) -> Dict[str, Union[List[float], Dict[str, Any]]]
嵌入文本。
参数:
text:要嵌入的文本。
引发:
TypeError: 如果输入不是字符串。
返回值:
包含以下键的字典
embedding:文本的 embedding。meta:关于请求的元数据。
CohereTextEmbedder.run_async
@component.output_types(embedding=List[float], meta=Dict[str, Any])
async def run_async(
text: str) -> Dict[str, Union[List[float], Dict[str, Any]]]
异步嵌入文本。
这是run 方法。它具有相同的参数和返回值,但可以在异步代码中与await 一起使用。
:param text: 要嵌入的文本。
引发:
TypeError: 如果输入不是字符串。
返回值:
包含以下键的字典
embedding:文本的 embedding。meta:关于请求的元数据。
模块 haystack_integrations.components.embedders.cohere.utils
get_async_response
async def get_async_response(
cohere_async_client: AsyncClientV2,
texts: List[str],
model_name: str,
input_type: str,
truncate: str,
embedding_type: Optional[EmbeddingTypes] = None
) -> Tuple[List[List[float]], Dict[str, Any]]
使用 Cohere API 异步嵌入文本列表。
参数:
cohere_async_client:CohereAsyncClienttexts:要嵌入的文本model_name:要使用的模型名称input_type:"classification", "clustering", "search_document", "search_query" 中的一个。提供给 embed 的输入文本的类型。truncate:"NONE", "START", "END" 中的一个。API 如何处理超过最大 token 长度的文本。embedding_type:要返回的 embedding 类型。默认为 float embedding。
引发:
ValueError:如果在查询 Cohere API 时发生错误。
返回值:
embedding 和元数据的元组。
get_response
def get_response(
cohere_client: ClientV2,
texts: List[str],
model_name: str,
input_type: str,
truncate: str,
batch_size: int = 32,
progress_bar: bool = False,
embedding_type: Optional[EmbeddingTypes] = None
) -> Tuple[List[List[float]], Dict[str, Any]]
使用 Cohere API 嵌入文本列表。
参数:
cohere_client:CohereClienttexts:要嵌入的文本model_name:要使用的模型名称input_type:"classification", "clustering", "search_document", "search_query" 中的一个。提供给 embed 的输入文本的类型。truncate:"NONE", "START", "END" 中的一个。API 如何处理超过最大 token 长度的文本。batch_size:要使用的批处理大小progress_bar:如果True,显示进度条embedding_type:要返回的 embedding 类型。默认为 float embedding。
引发:
ValueError:如果在查询 Cohere API 时发生错误。
返回值:
embedding 和元数据的元组。
模块 haystack_integrations.components.generators.cohere.generator
CohereGenerator
通过 Cohere 的generate 端点使用 Cohere 的模型生成文本。
注意:Cohere 已停用generate API,因此此生成器只是一个包装器,用于CohereChatGenerator,以提供向后兼容。
使用示例
from haystack_integrations.components.generators.cohere import CohereGenerator
generator = CohereGenerator(api_key="test-api-key")
generator.run(prompt="What's the capital of France?")
CohereGenerator.__init__
def __init__(api_key: Secret = Secret.from_env_var(
["COHERE_API_KEY", "CO_API_KEY"]),
model: str = "command-r-08-2024",
streaming_callback: Optional[Callable] = None,
api_base_url: Optional[str] = None,
**kwargs: Any)
实例化一个CohereGenerator 组件。
参数:
api_key:Cohere API 密钥。model:用于生成的 Cohere 模型。streaming_callback:当从流中接收到新 token 时调用的回调函数。回调函数接受 StreamingChunk 作为参数。api_base_url:Cohere 基本 URL。**kwargs:传递给模型的其他参数。这些参数特定于模型。您可以在模型的文档中进行查看。
CohereGenerator.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]]]]
使用提示查询 LLM 以生成回复。
参数:
prompt:要发送到生成模型的提示。
返回值:
包含以下键的字典
replies:模型生成的回复列表。meta:关于请求的信息。
CohereGenerator.run_async
@component.output_types(replies=List[str], meta=List[Dict[str, Any]])
async def run_async(
prompt: str) -> Dict[str, Union[List[str], List[Dict[str, Any]]]]
异步地使用提示查询 LLM 以生成回复。
参数:
prompt:要发送到生成模型的提示。
返回值:
包含以下键的字典
replies:模型生成的回复列表。meta:关于请求的信息。
模块 haystack_integrations.components.generators.cohere.chat.chat_generator
CohereChatGenerator
使用 Cohere 的模型通过 Cohere 的chat 端点完成聊天。
此组件支持使用 Cohere 的视觉模型(如 Command A Vision)进行纯文本和多模态(文本+图像)对话。
支持的图像格式:PNG、JPEG、WEBP、GIF(非动画)。每次请求最多 20 张图像,总限制为 20MB。
您可以通过**generation_kwargs 参数将参数传递给 Cohere API,从而自定义聊天响应的生成方式。您可以在初始化或运行组件时进行设置。任何与cohere.ClientV2.chat 兼容的参数在这里也适用。详情请参阅Cohere API。
以下是使用该组件的示例
简单示例
from haystack.dataclasses import ChatMessage
from haystack.utils import Secret
from haystack_integrations.components.generators.cohere import CohereChatGenerator
client = CohereChatGenerator(model="command-r-08-2024", api_key=Secret.from_env_var("COHERE_API_KEY"))
messages = [ChatMessage.from_user("What's Natural Language Processing?")]
client.run(messages)
# Output: {'replies': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>,
# _content=[TextContent(text='Natural Language Processing (NLP) is an interdisciplinary...
多模态示例
from haystack.dataclasses import ChatMessage, ImageContent
from haystack.utils import Secret
from haystack_integrations.components.generators.cohere import CohereChatGenerator
# Create an image from file path or base64
image_content = ImageContent.from_file_path("path/to/your/image.jpg")
# Create a multimodal message with both text and image
messages = [ChatMessage.from_user(content_parts=["What's in this image?", image_content])]
# Use a multimodal model like Command A Vision
client = CohereChatGenerator(model="command-a-vision-07-2025", api_key=Secret.from_env_var("COHERE_API_KEY"))
response = client.run(messages)
print(response)
高级示例
CohereChatGenerator 可以集成到管道中,并支持 Haystack 的工具架构,从而能够无缝地在各种生成器中调用工具。
from haystack import Pipeline
from haystack.dataclasses import ChatMessage
from haystack.components.tools import ToolInvoker
from haystack.tools import Tool
from haystack_integrations.components.generators.cohere import CohereChatGenerator
# Create a weather tool
def weather(city: str) -> str:
return f"The weather in {city} is sunny and 32°C"
weather_tool = Tool(
name="weather",
description="useful to determine the weather in a given location",
parameters={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city to get weather for, e.g. Paris, London",
}
},
"required": ["city"],
},
function=weather,
)
# Create and set up the pipeline
pipeline = Pipeline()
pipeline.add_component("generator", CohereChatGenerator(model="command-r-08-2024", tools=[weather_tool]))
pipeline.add_component("tool_invoker", ToolInvoker(tools=[weather_tool]))
pipeline.connect("generator", "tool_invoker")
# Run the pipeline with a weather query
results = pipeline.run(
data={"generator": {"messages": [ChatMessage.from_user("What's the weather like in Paris?")]}}
)
# The tool result will be available in the pipeline output
print(results["tool_invoker"]["tool_messages"][0].tool_call_result.result)
# Output: "The weather in Paris is sunny and 32°C"
CohereChatGenerator.__init__
def __init__(api_key: Secret = Secret.from_env_var(
["COHERE_API_KEY", "CO_API_KEY"]),
model: str = "command-r-08-2024",
streaming_callback: Optional[StreamingCallbackT] = None,
api_base_url: Optional[str] = None,
generation_kwargs: Optional[Dict[str, Any]] = None,
tools: Optional[Union[List[Tool], Toolset]] = None,
**kwargs: Any)
初始化 CohereChatGenerator 实例。
参数:
api_key:Cohere API 的 API 密钥。model:要使用的模型名称。您可以使用command系列的模型。streaming_callback:当从流中接收到新 token 时调用的回调函数。回调函数接受 StreamingChunk 作为参数。api_base_url:Cohere API 的基本 URL。generation_kwargs:用于模型在生成过程中的其他参数。有关参数列表,请参阅Cohere Chat 端点。其中一些参数是- 'messages':用户和模型之间的消息列表,用于为模型提供响应用户消息的对话上下文。
- 'system_message':指定后,会在对话开头添加一条系统消息。
- 'citation_quality':默认为
accurate。通过允许用户指定他们想要accurate(精确)还是fast(快速)的结果,来决定在 RAG 流中生成引文的方法。 - 'temperature':一个非负浮点数,用于调整生成过程中的随机性程度。较低的温度意味着较低的随机生成。
tools:模型可以使用的一系列 Tool 对象或 Toolset。每个工具都应有一个唯一的名称。
CohereChatGenerator.to_dict
def to_dict() -> Dict[str, Any]
将组件序列化为字典。
返回值:
包含序列化数据的字典。
CohereChatGenerator.from_dict
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "CohereChatGenerator"
从字典反序列化组件。
参数:
data: 要反序列化的字典。
返回值:
反序列化后的组件。
CohereChatGenerator.run
@component.output_types(replies=List[ChatMessage])
def run(
messages: List[ChatMessage],
generation_kwargs: Optional[Dict[str, Any]] = None,
tools: Optional[Union[List[Tool], Toolset]] = None,
streaming_callback: Optional[StreamingCallbackT] = None
) -> Dict[str, List[ChatMessage]]
根据提供的消息和生成参数调用 chat 端点。
参数:
messages:代表输入消息的ChatMessage实例列表。generation_kwargs:用于 chat 生成的其他关键字参数。这些参数可能会覆盖在 **init** 方法中传递的参数。有关 Cohere API 支持的参数的更多详细信息,请参阅 Cohere 的文档。tools: 一个工具列表或 Toolset,模型可以为此准备调用。如果设置,它将覆盖tools参数。streaming_callback: 当从流中接收到新 token 时调用的回调函数。回调函数接受 StreamingChunk 作为参数。
返回值:
包含以下键的字典
replies:代表生成响应的ChatMessage实例列表。
CohereChatGenerator.run_async
@component.output_types(replies=List[ChatMessage])
async def run_async(
messages: List[ChatMessage],
generation_kwargs: Optional[Dict[str, Any]] = None,
tools: Optional[Union[List[Tool], Toolset]] = None,
streaming_callback: Optional[StreamingCallbackT] = None
) -> Dict[str, List[ChatMessage]]
根据提供的消息和生成参数异步调用 chat 端点。
参数:
messages:代表输入消息的ChatMessage实例列表。generation_kwargs:用于 chat 生成的其他关键字参数。这些参数可能会覆盖在 **init** 方法中传递的参数。有关 Cohere API 支持的参数的更多详细信息,请参阅 Cohere 的文档。tools:模型可以准备调用的工具列表。如果设置,它将覆盖tools参数。streaming_callback: 当从流中接收到新 token 时调用的回调函数。
返回值:
包含以下键的字典
replies:代表生成响应的ChatMessage实例列表。
模块 haystack_integrations.components.rankers.cohere.ranker
CohereRanker
使用 Cohere 模型根据查询的相似度对 Documents 进行排序。
Documents 按与查询的语义相关性从高到低排序。
使用示例
from haystack import Document
from haystack_integrations.components.rankers.cohere import CohereRanker
ranker = CohereRanker(model="rerank-v3.5", top_k=2)
docs = [Document(content="Paris"), Document(content="Berlin")]
query = "What is the capital of germany?"
output = ranker.run(query=query, documents=docs)
docs = output["documents"]
CohereRanker.__init__
def __init__(model: str = "rerank-v3.5",
top_k: int = 10,
api_key: Secret = Secret.from_env_var(
["COHERE_API_KEY", "CO_API_KEY"]),
api_base_url: str = "https://api.cohere.com",
meta_fields_to_embed: Optional[List[str]] = None,
meta_data_separator: str = "\n",
max_tokens_per_doc: int = 4096)
创建 'CohereRanker' 的实例。
参数:
model:Cohere 模型名称。请在Cohere 文档中查看支持的模型列表。top_k: 要返回的最大文档数。api_key:Cohere API 密钥。api_base_url:Cohere API 的基本 URL。meta_fields_to_embed:应与文档内容连接进行重新排序的元字段列表。meta_data_separator:用于将元字段连接到文档内容的the separator。max_tokens_per_doc:每个文档要嵌入的最大 token 数,默认为 4096。
CohereRanker.to_dict
def to_dict() -> Dict[str, Any]
将组件序列化为字典。
返回值:
包含序列化数据的字典。
CohereRanker.from_dict
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "CohereRanker"
从字典反序列化组件。
参数:
data: 要反序列化的字典。
返回值:
反序列化后的组件。
CohereRanker.run
@component.output_types(documents=List[Document])
def run(query: str,
documents: List[Document],
top_k: Optional[int] = None) -> Dict[str, List[Document]]
使用 Cohere Reranker 根据查询对文档列表进行重新排序。
参数:
query: 查询字符串。documents: Document 列表。top_k: 您希望 Ranker 返回的最大 Document 数量。
引发:
ValueError: 如果top_k不大于 0。
返回值:
包含以下键的字典
documents: 按相似度降序排列的、与给定查询最相似的 Document 列表。
