模块 haystack_integrations.components.connectors.weave.weave_connector

WeaveConnector

收集您管道中的跟踪信息，并将它们发送到 Weights & Biases。

将此组件添加到您的管道中，以与 Weights & Biases Weave 框架集成，用于跟踪和监控您的管道组件。

请注意，您需要设置WANDB_API_KEY 环境变量为您的 Weights & Biases API 密钥。

注意：如果您没有 Weights & Biases 帐户，它将交互式地要求您设置一个帐户，然后您的输入将存储在 ~/.netrc 中

此外，您还需要设置HAYSTACK_CONTENT_TRACING_ENABLED 环境变量为true 以启用 Haystack 在您的管道中进行跟踪。

要使用此连接器，只需将其添加到您的管道中，无需任何连接，它将自动开始向 Weights & Biases 发送跟踪信息。

示例:

import os

from haystack import Pipeline
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage

from haystack_integrations.components.connectors import WeaveConnector

os.environ["HAYSTACK_CONTENT_TRACING_ENABLED"] = "true"

pipe = Pipeline()
pipe.add_component("prompt_builder", ChatPromptBuilder())
pipe.add_component("llm", OpenAIChatGenerator(model="gpt-3.5-turbo"))
pipe.connect("prompt_builder.prompt", "llm.messages")

connector = WeaveConnector(pipeline_name="test_pipeline")
pipe.add_component("weave", connector)

messages = [
    ChatMessage.from_system(
        "Always respond in German even if some input data is in other languages."
    ),
    ChatMessage.from_user("Tell me about {{location}}"),
]

response = pipe.run(
    data={
        "prompt_builder": {
            "template_variables": {"location": "Berlin"},
            "template": messages,
        }
    }
)
print(response["llm"]["replies"][0])

然后，您应该前往https://wandb.ai/<user_name>/projects，在您创建WeaveConnector

WeaveConnector.init

def __init__(pipeline_name: str,
             weave_init_kwargs: Optional[dict[str, Any]] = None) -> None

初始化 WeaveConnector。

参数:

pipeline_name：您要跟踪的管道的名称。
weave_init_kwargs：要传递给 WeaveTracer 客户端的附加参数。

WeaveConnector.warm_up

def warm_up() -> None

初始化 WeaveTracer。

WeaveConnector.to_dict

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

将组件序列化为字典。

返回值:

包含重建此组件所需的所有信息的字典。

WeaveConnector.from_dict

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

从字典反序列化组件。

参数:

data: 要反序列化的字典。

返回值:

反序列化后的组件。

模块 haystack_integrations.tracing.weave.tracer

WeaveSpan

Haystack 的 Span 接口与 Weave 的 Call 对象之间的桥梁。

存储组件执行及其输入和输出的元数据，并管理描述操作的属性/标签。

WeaveSpan.set_tag

def set_tag(key: str, value: Any) -> None

通过将其添加到调用的输入中来设置标签。

参数:

key：标签键。
value：标签值。

WeaveSpan.raw_span

def raw_span() -> Any

访问底层的 Weave Call 对象。

WeaveSpan.get_correlation_data_for_logs

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

用于日志记录的相关数据。

WeaveTracer

实现了 Haystack 的 Tracer，以提供与 Weights and Bias Weave 的接口。

它负责创建和管理 Weave 调用，并将 Haystack spans 转换为 Weave spans。它为每个 Haystack 组件的运行创建 spans。

WeaveTracer.init

def __init__(project_name: str, **weave_init_kwargs: Any) -> None

初始化 WeaveTracer。

参数:

project_name：要跟踪的项目名称，这将是 Weave 项目中显示的名称。
weave_init_kwargs：要传递给 Weave 客户端的附加参数。

WeaveTracer.current_span

def current_span() -> Optional[Span]

获取当前活动 span。

WeaveTracer.trace

@contextlib.contextmanager
def trace(operation_name: str,
          tags: Optional[dict[str, Any]] = None,
          parent_span: Optional[WeaveSpan] = None) -> Iterator[WeaveSpan]

一个上下文管理器，用于创建和管理 spans，以便在 Weights & Biases Weave 中跟踪操作。

它有两个主要工作流程：

A) 对于常规操作 (operation_name != "haystack.component.run")：立即创建 Weave Call 创建具有此调用的 WeaveSpan 设置提供的任何标签在 with 块中使用 span 结束后，使用管道输出数据更新调用

B) 对于组件运行 (operation_name == "haystack.component.run")：最初创建 WeaveSpan 而不创建 Call (延迟创建) 设置提供的任何标签在 with 块中使用 span 在所有组件信息可用后，仅在最后创建实际的 Weave Call 使用组件输出数据更新调用

这种区分很重要，因为 Weave 的调用在创建后无法更新，而内容标签仅在稍后阶段设置在 Span 上。为了在创建调用时获取输入，我们需要在 yield span 之后创建调用。