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

秘密管理

本页面重点介绍 Haystack 组件中的秘密管理,并介绍Secret 类型用于结构化秘密处理。它解释了在代码中硬编码秘密的缺点,并建议使用环境变量代替。

许多 Haystack 组件与第三方框架和供应商(如 Azure、Google Vertex AI 和 OpenAI)进行交互。它们的库通常要求用户进行身份验证,以确保他们能够访问底层产品。身份验证过程通常使用一个充当第三方后端不透明标识符的秘密值。

本页面介绍了两种主要类型的秘密:基于令牌的秘密和基于环境变量的秘密,以及在使用 Haystack 时如何处理它们。

您可以在我们的Secret 类中找到更多详细信息,请参阅 API 参考

用例 - 问题陈述

问题陈述

让我们考虑一个 RAG 管道示例,该管道嵌入查询、使用 Retriever 组件查找与查询相关的文档,然后利用 LLM 根据检索到的文档生成答案。

OpenAIGenerator 组件在下面的管道中使用,它需要一个 API 密钥来向 OpenAI 服务器进行身份验证并执行生成。假设该组件接受一个str 值用于它

generator = OpenAIGenerator(model="gpt-4", api_key="sk-xxxxxxxxxxxxxxxxxx")
pipeline.add_component("generator", generator)

这样应急是可行的,但这是一种不良做法——我们不应该在代码库中硬编码这些秘密。另一种方法是将密钥存储在外部环境变量中,在 Python 中读取它,然后将其传递给组件。

import os

api_key = os.environ.get("OPENAI_API_KEY")
generator = OpenAIGenerator(model="gpt-4", api_key=api_key)
pipeline.add_component("generator", generator)

这样就好多了——管道按预期工作,而且我们没有在代码中硬编码任何秘密。

请记住,管道是可序列化的。由于 API 密钥是秘密,我们肯定应该避免将其保存到磁盘。让我们修改组件的to_dict 方法以排除密钥。

def to_dict(self) -> Dict[str, Any]:
	# Do not pass the `api_key` init parameter.
	return default_to_dict(self, model=self.model)

但是当管道从磁盘加载时会发生什么?在最好的情况下,组件的后端会自动尝试从硬编码的环境变量中读取密钥,而该密钥与在序列化组件之前传递给组件的密钥相同。但在最坏的情况下,后端不会在硬编码的环境变量中查找密钥,并在调用时失败pipeline.run() 调用。

导入

要在代码中使用 Haystack 秘密,首先使用以下命令导入:

from haystack.utils import Secret

基于令牌的秘密

您可以使用以下命令直接将令牌粘贴为字符串from_token 方法

llm = OpenAIGenerator(api_key=Secret.from_token("sk-randomAPIkeyasdsa32ekasd32e"))

请注意,此类型的代码无法序列化,这意味着您无法将上述组件转换为字典或将包含它的管道保存到 YAML 文件。这是一项安全功能,用于防止敏感数据意外泄露。

基于环境变量的秘密

基于环境变量的秘密更具灵活性。它们允许您指定一个或多个可能包含您的秘密的环境变量。

需要 API 密钥的现有 Haystack 组件(如 OpenAIGenerator)具有默认值Secret.from_env_var (在此例中为OPENAI_API_KEY)。这意味着OpenAIGenerator 将查找环境变量的值OPENAI_API_KEY (如果存在)并将其用于身份验证。并且,当管道被序列化为 YAML 时,只有环境变量的名称会被保存在 YAML 文件中。通过这样做,此方法可确保不会出现安全泄露,因此强烈推荐。

# First, export an environment variable name `OPENAI_API_KEY` with its value
export OPENAI_API_KEY=sk-randomAPIkeyasdsa32ekasd32e

# or alternatively, using Python
# import os
# os.environ[”OPENAI_API_KEY”]=sk-randomAPIkeyasdsa32ekasd32e
llm_generator = OpenAIGenerator() # Uses the default value from the env var for the component

或者,在需要 Secret 的组件中,您可以自定义从中读取 API 密钥的环境变量的名称。

# Export an environment variable with custom name and its value
llm_generator = OpenAIGenerator(api_key=Secret.from_env_var("YOUR_ENV_VAR"))

OpenAIGenerator 在管道中序列化时,YAML 代码将如下所示,使用自定义变量名

components:
  llm:
    init_parameters:
      api_base_url: null
      api_key:
        env_vars:
        - YOUR_ENV_VAR
        strict: true
        type: env_var
      generation_kwargs: {}
      model: gpt-4o-mini
      organization: null
      streaming_callback: null
      system_prompt: null
    type: haystack.components.generators.openai.OpenAIGenerator
    ...

序列化

虽然基于令牌的秘密无法序列化,但基于环境变量的秘密可以转换为字典并从中转换回来。

# Convert to dictionary
env_secret_dict = env_secret.to_dict()

# Create from dictionary
new_env_secret = Secret.from_dict(env_secret_dict)

解析秘密

可以使用以下方法解析这两种类型的秘密以获取其实际值:resolve_value 方法。此方法返回令牌或环境变量的值。


# Resolve the token-based secret
token_value = api_key_secret.resolve_value()

# Resolve the environment variable-based secret
env_value = env_secret.resolve_value()

自定义组件示例

这是一个完整的示例,展示了如何创建使用Secret 类(在 Haystack 中)的组件,重点介绍了基于令牌的身份验证和基于环境变量的身份验证之间的区别,并表明基于令牌的秘密无法序列化。

from haystack.utils import Secret, deserialize_secrets_inplace

@component
class MyComponent:
  def __init__(self, api_key: Optional[Secret] = None, **kwargs):
    self.api_key = api_key
    self.backend = None

  def warm_up(self):
    # Call resolve_value to yield a single result. The semantics of the result is policy-dependent.
    # Currently, all supported policies will return a single string token.
    self.backend = SomeBackend(api_key=self.api_key.resolve_value() if self.api_key else None, ...)

  def to_dict(self):
    # Serialize the policy like any other (custom) data. If the policy is token-based, it will
    # raise an error.
    return default_to_dict(self, api_key=self.api_key.to_dict() if self.api_key else None, ...)

  @classmethod
  def from_dict(cls, data):
    # Deserialize the policy data before passing it to the generic from_dict function.
    api_key_data = data["init_parameters"]["api_key"]
    api_key = Secret.from_dict(api_key_data) if api_key_data is not None else None
    data["init_parameters"]["api_key"] = api_key
		# Alternatively, use the helper function.
		# deserialize_secrets_inplace(data["init_parameters"], keys=["api_key"])
    return default_from_dict(cls, data)

# No authentication.
component = MyComponent(api_key=None)

# Token based authentication
component = MyComponent(api_key=Secret.from_token("sk-randomAPIkeyasdsa32ekasd32e"))
component.to_dict() # Error! Can't serialize authentication tokens

# Environment variable based authentication
component = MyComponent(api_key=Secret.from_env_var("OPENAI_API_KEY"))
component.to_dict() # This is fine