Haystack 与 OpenTelemetry 集成指南：用 OpenTelemetryConnector 与 OpenTelemetryTracer 实现管线级可观测性-平芜编程栈

Haystack 与 OpenTelemetry 集成指南：用 OpenTelemetryConnector 与 OpenTelemetryTracer 实现管线级可观测性

【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack

Haystack 是用于构建生产级 LLM 应用的 AI 编排框架，本文聚焦于它在 OpenTelemetry 上的官方集成，讲解如何通过OpenTelemetryConnector组件与OpenTelemetryTracer追踪器，把管线（Pipeline）中各组件间的操作与数据流以 Span 的形式发送到任意兼容 OpenTelemetry 的后端。读完本文，你将掌握从零配置 TracerProvider、启用内容追踪（Content Tracing）、在管线上挂载追踪组件，以及结合 Jaeger 本地可视化追踪链路的完整实战方案。

什么是 OpenTelemetry 集成

OpenTelemetry 是一套开源的可观测性框架，用于统一采集 traces（追踪）、metrics（指标）与 logs（日志）。Haystack 通过opentelemetry-haystack集成包与 OpenTelemetry 对接，将每次管线运行产生的完整执行上下文（包括输入提示词、生成结果与元数据）发送到任何兼容 OpenTelemetry 的后端，从而支持生产环境下的性能监控与故障排查。

在本仓库中，该集成对应两个核心概念：

OpenTelemetryConnector：一个 Haystack 组件，把它加入管线即可自动开启追踪；
OpenTelemetryTracer：实现 HaystackTracer抽象接口的追踪器，用于把 Span 写入 OpenTelemetry。

此外，Haystack 核心还内置了一套与具体后端无关的追踪基础设施（tracer.py），包括Tracer、Span抽象基类、ProxyTracer代理容器以及enable_tracing/disable_tracing/is_tracing_enabled等全局开关，OpenTelemetryTracer正是这套基础设施在 OpenTelemetry 上的具体落地。

前提条件

在使用本集成之前，需要先完成以下两件事：

安装集成包：opentelemetry-haystack（包含OpenTelemetryConnector与OpenTelemetryTracer）。
配置 OpenTelemetryTracerProvider：至少需要提供一个带导出器（Exporter）的 TracerProvider，例如使用 OTLP 导出器把 Span 发送到 Collector 或后端。TracerProvider 必须在启用追踪器之前完成配置。

如果希望进一步下钻到管线更深层次的调用（如 HTTP 请求、OpenAI 调用），官方还推荐安装额外的 OpenTelemetry 生态工具，例如：

urllib3instrumentation：追踪管线中的 HTTP 请求；
OpenAI instrumentation：追踪 OpenAI API 请求。

这类工具并非必需，但能让追踪数据更细粒度。

快速开始：在管线中启用 OpenTelemetry 追踪

方式一：使用 OpenTelemetryConnector 组件

OpenTelemetryConnector连接 Haystack 与 OpenTelemetry，用于启用管线组件内部操作及数据流的追踪。使用方式非常特殊：把它加入管线即可，但不要连接到任何其他组件——它作为一个"独立节点"存在于管线图中，负责在追踪开启后自动标记所有管线操作。

在初始化连接器之前，必须确保 OpenTelemetry 的TracerProvider（例如带导出器的 Provider）已经配置完成。同时，如果需要追踪管线组件的输入与输出内容，还需把环境变量HAYSTACK_CONTENT_TRACING_ENABLED设为"true"。

完整示例：

import os os.environ["HAYSTACK_CONTENT_TRACING_ENABLED"] = "true" from opentelemetry import trace from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.semconv.resource import ResourceAttributes # 配置 OpenTelemetry SDK。大多数后端都要求提供服务名。 resource = Resource(attributes={ResourceAttributes.SERVICE_NAME: "haystack"}) tracer_provider = TracerProvider(resource=resource) processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")) tracer_provider.add_span_processor(processor) trace.set_tracer_provider(tracer_provider) 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.opentelemetry import OpenTelemetryConnector pipe = Pipeline() pipe.add_component("tracer", OpenTelemetryConnector()) pipe.add_component("prompt_builder", ChatPromptBuilder()) pipe.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini")) pipe.connect("prompt_builder.prompt", "llm.messages") 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])

要点说明：

HAYSTACK_CONTENT_TRACING_ENABLED必须在导入任何 Haystack 组件之前设置，因为该开关在模块导入阶段即被读取（见下文"内容追踪"一节）；
OpenTelemetryConnector与ChatPromptBuilder、OpenAIChatGenerator之间无需connect，它独立工作；
管线运行时，prompt_builder生成的提示词会被传入llm的messages输入槽，这些数据流经管线时即可被追踪。

方式二：直接启用 OpenTelemetryTracer

如果不想往管线里加组件，也可以直接启用OpenTelemetryTracer。它同样会追踪任何 Haystack 管线的运行，无需修改管线定义：

import os os.environ["HAYSTACK_CONTENT_TRACING_ENABLED"] = "true" from opentelemetry import trace from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.semconv.resource import ResourceAttributes # 配置 OpenTelemetry SDK。服务名对大多数后端都是必需的。 resource = Resource(attributes={ResourceAttributes.SERVICE_NAME: "haystack"}) tracer_provider = TracerProvider(resource=resource) tracer_provider.add_span_processor( BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")), ) trace.set_tracer_provider(tracer_provider) from haystack import tracing from haystack_integrations.tracing.opentelemetry import OpenTelemetryTracer # 启用 OpenTelemetry 追踪器 tracing.enable_tracing(OpenTelemetryTracer(trace.get_tracer("my_application")))

之后每次pipeline.run(...)都会产生一条包含完整执行上下文的 trace，可在兼容 OpenTelemetry 的后端中查看。

两种方式的取舍

维度	OpenTelemetryConnector	OpenTelemetryTracer
启用方式	作为组件加入管线，初始化后自动开启	调用`tracing.enable_tracing()`全局启用
是否需要连接其他组件	否，独立节点即可	不适用
管线定义侵入性	需要修改管线	无需修改管线
适用场景	希望把追踪作为管线配置的一部分管理	希望全局统一开启追踪

环境变量与配置项

HAYSTACK_CONTENT_TRACING_ENABLED

作用：控制是否追踪管线组件的内容（输入与输出），例如查询内容、文档内容、回答内容等；
取值：必须设置为字符串"true"才会启用；默认关闭；
安全性：默认关闭是为了防止敏感用户信息被发送到追踪后端；
解析位置：在 tracer.py 中，ProxyTracer.__init__通过os.getenv(HAYSTACK_CONTENT_TRACING_ENABLED_ENV_VAR, "false").lower() == "true"解析该变量，因此在导入 Haystack 之前设置即可生效。

也可以在 Python 中动态开启：

from haystack import tracing tracing.tracer.is_content_tracing_enabled = True

HAYSTACK_AUTO_TRACE_ENABLED

用于控制 Haystack 是否自动探测并启用追踪后端。在 2.x 时代，Haystack 会在满足以下条件时自动开启追踪：

已安装并配置opentelemetry-sdk（OpenTelemetry）；
已安装ddtrace（Datadog）。

如果希望关闭这种自动行为，可以：

运行应用时设置环境变量HAYSTACK_AUTO_TRACE_ENABLED=false；
或在 Python 中调用：

from haystack.tracing import disable_tracing disable_tracing()

:::note 版本差异自 Haystack 3.0 起，OpenTelemetry 追踪不再在安装opentelemetry-sdk时自动启用，需要显式安装opentelemetry-haystack集成并启用OpenTelemetryTracer或添加OpenTelemetryConnector。本文 API 参考对应 2.22 版本，使用时请留意版本差异。 :::

组件 API 详解

OpenTelemetryConnector

OpenTelemetryConnector连接 Haystack 与 OpenTelemetry，以启用管线组件内操作与数据流的追踪。它位于集成包模块haystack_integrations.components.connectors.opentelemetry.opentelemetry_connector。

`init`

__init__(name: str = 'opentelemetry') -> None

初始化 OpenTelemetryConnector 组件。

参数：

name（str）——用于标识该追踪组件的名称，会由run方法返回，可用于标记该连接器产生的 trace。默认值为"opentelemetry"。

`run`

run() -> dict[str, str]

运行 OpenTelemetryConnector 组件。

返回值：

dict[str, str]——包含以下键的字典：
- name：追踪组件的名称。

`to_dict`

to_dict() -> dict[str, Any]

将组件序列化为字典。

返回值：

dict[str, Any]——序列化后的组件字典（用于 YAML/JSON 形式的管线配置与反序列化）。

`from_dict`

from_dict(data: dict[str, Any]) -> OpenTelemetryConnector

从字典反序列化组件。

参数：

data（dict[str, Any]）——组件的字典表示。

返回值：

OpenTelemetryConnector——反序列化后的组件实例。

OpenTelemetrySpan

OpenTelemetrySpan包装了 OpenTelemetry 原生 Span，使其符合 Haystack 的Span抽象接口。它位于haystack_integrations.tracing.opentelemetry.tracer。

`init`

__init__(span: opentelemetry.trace.Span) -> None

创建 OpenTelemetrySpan 实例，包装一个 OpenTelemetry 原生 Span 对象。

`set_tag`

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

在 Span 上设置单个标签。

参数：

key（str）——标签名称；
value（Any）——标签值。

注意：值会被序列化后再写入，因此最好使用字符串、数字、布尔值等简单类型。具体实现中，标签值会先经过coerce_tag_value的强制转换（见下文"标签值序列化"）。

`raw_span`

raw_span() -> Any

提供对追踪器底层 Span 对象的完整访问权限。

返回值：

Any——底层 Span 对象。

`get_correlation_data_for_logs`

get_correlation_data_for_logs() -> dict[str, Any]

返回包含日志关联数据的字典，可用于将日志与 trace 关联起来（Log-Trace Correlation）。

OpenTelemetryTracer

OpenTelemetryTracer是 HaystackTracer抽象基类的 OpenTelemetry 实现，位于haystack_integrations.tracing.opentelemetry.tracer。

`init`

__init__(tracer: opentelemetry.trace.Tracer) -> None

创建 OpenTelemetryTracer 实例，包装一个 OpenTelemetry 原生Tracer对象。

`trace`

trace( operation_name: str, tags: dict[str, Any] | None = None, parent_span: Span | None = None, ) -> Iterator[Span]

激活并返回一个新的 Span，该 Span 继承自当前活动的 Span。

参数：

operation_name（str）——被追踪操作的名称；
tags（dict[str, Any] | None）——应用到新建 Span 上的标签；
parent_span（Span | None）——新建 Span 的父 Span；若为None，新 Span 将成为根 Span。

该方法以上下文管理器（Context Manager）形式使用，配合with语句即可包裹一段代码的执行。

`current_span`

current_span() -> Span | None

返回当前活动的 Span。

返回值：

Span | None——当前活动的 Span；如果没有活动 Span，则返回None。

深入底层：Haystack 追踪基础设施

OpenTelemetryTracer之所以能无缝工作，是因为 Haystack 核心提供了一套与后端无关的追踪抽象（tracer.py）。理解这套机制有助于排查问题或编写自定义追踪器。

Tracer 与 Span 抽象接口

Span：被插桩操作的接口，核心方法是set_tag(key, value)（设置单个标签）、set_tags(tags)（批量设置标签）、raw_span()（获取底层 Span）、set_content_tag(key, value)（设置包含敏感内容的标签，默认受is_content_tracing_enabled门控）、get_correlation_data_for_logs()（返回日志关联数据）；
Tracer：通过trace(operation_name, tags, parent_span)上下文管理器创建并提交 Span，并提供current_span()获取当前活动 Span。

ProxyTracer 与全局开关

tracer是一个全局ProxyTracer实例，内部持有actual_tracer（默认是NullTracer空实现）。这种代理模式让你可以在不改变全局引用的情况下随时替换底层追踪器：

enable_tracing(provided_tracer)：把全局追踪器替换为传入的Tracer实现；
disable_tracing()：恢复为NullTracer（no-op）；
is_tracing_enabled()：判断追踪是否开启。

ProxyTracer同时负责在初始化时读取HAYSTACK_CONTENT_TRACING_ENABLED环境变量，把结果存入is_content_tracing_enabled属性，供Span.set_content_tag判断是否写入内容标签。

标签值序列化

大多数追踪后端不支持复杂类型作为标签值，因此 Haystack 在 utils.py 中提供了coerce_tag_value：

对bool、str、int、float等原始类型，直接透传；
对None，转为空字符串""；
对其他对象，尝试序列化为 JSON 字符串；序列化失败时退化为str(value)。

其中_serializable_value会递归处理list/dict，并在对象带有_to_trace_dict方法时使用其追踪表示、在对象带有to_dict方法时使用其字典表示，从而把文档、消息等 Haystack 数据类安全地转成可观测的标签内容。

内容追踪：逐步审查管线输入输出

Haystack 允许追踪管线组件的输入与输出值，用于逐步骤调查管线执行情况。由于默认关闭（避免敏感信息外泄），开启方式有两种：

运行应用时设置环境变量：

HAYSTACK_CONTENT_TRACING_ENABLED=true <command to run your Haystack pipeline>

在 Python 中显式开启：

from haystack import tracing tracing.tracer.is_content_tracing_enabled = True

从源码看，内容追踪标签在多个位置被写入，例如 pipeline.py 中_run_component使用span.set_content_tag("haystack.component.input", inputs)记录组件输入、span.set_content_tag("haystack.component.output", component_output)记录组件输出；在 Agent 场景中，agent.py 记录haystack.agent.input/haystack.agent.output，tool_calling.py 记录haystack.agent.step.tool.input/haystack.agent.step.tool.output（含错误信息）。这些标签在 OpenTelemetry 后端中呈现为 Span 上的属性，可以精确还原每一步的输入与输出。

需要提醒的是：这些内容标签可能包含提示词、检索文档、模型回答等敏感数据，生产环境中请根据合规要求谨慎开启。

本地可视化：用 Jaeger 调试追踪链路

Jaeger 是轻量级的开源分布式追踪后端，适合本地开发时可视化 Haystack 管线的执行顺序与耗时分布。整个链路如下：

启动 Jaeger 容器（同时提供后端与 UI）：

docker run --rm -d --name jaeger \ -e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \ -p 6831:6831/udp \ -p 6832:6832/udp \ -p 5778:5778 \ -p 16686:16686 \ -p 4317:4317 \ -p 4318:4318 \ -p 14250:14250 \ -p 14268:14268 \ -p 14269:14269 \ -p 9411:9411 \ jaegertracing/all-in-one:latest

其中16686为 Jaeger UI 端口，4318为 OTLP HTTP 追踪接收端口。

安装依赖：

pip install opentelemetry-sdk pip install opentelemetry-exporter-otlp

配置 OpenTelemetry 指向 Jaeger：

from opentelemetry import trace from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.semconv.resource import ResourceAttributes # 服务名对大多数后端都是必需的 resource = Resource(attributes={ResourceAttributes.SERVICE_NAME: "haystack"}) tracer_provider = TracerProvider(resource=resource) processor = BatchSpanProcessor( OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces") ) tracer_provider.add_span_processor(processor) trace.set_tracer_provider(tracer_provider)

启用追踪并运行管线：

import haystack.tracing haystack.tracing.auto_enable_tracing() # 2.x 自动探测方式 # 或：tracing.enable_tracing(OpenTelemetryTracer(trace.get_tracer("my_application"))) ... pipeline.run(...) ...

查看结果：在浏览器打开 Jaeger UI（http://localhost:16686），即可按服务名、操作名检索并查看 Span 的时间线、层级关系与标签。

自定义追踪后端：实现自己的 Tracer

如果 OpenTelemetry 或 Datadog 都不满足需求，可以基于 Haystack 的Tracer抽象快速实现自己的追踪器。以下示例以 OpenTelemetry 包为例，展示实现要点：

import contextlib from typing import Optional, Dict, Any, Iterator from opentelemetry import trace from opentelemetry.trace import NonRecordingSpan from haystack.tracing import Tracer, Span from haystack.tracing import utils as tracing_utils import opentelemetry.trace class OpenTelemetrySpan(Span): def __init__(self, span: opentelemetry.trace.Span) -> None: self._span = span def set_tag(self, key: str, value: Any) -> None: # 追踪后端通常不支持任意标签值， # coerce_tag_value 会强制把值转换为 Python 原始类型（int/float/bool/str）， # 或尝试以字符串形式转储。 coerced_value = tracing_utils.coerce_tag_value(value) self._span.set_attribute(key, coerced_value) class OpenTelemetryTracer(Tracer): def __init__(self, tracer: opentelemetry.trace.Tracer) -> None: self._tracer = tracer @contextlib.contextmanager def trace( self, operation_name: str, tags: Optional[Dict[str, Any]] = None ) -> Iterator[Span]: with self._tracer.start_as_current_span(operation_name) as span: span = OpenTelemetrySpan(span) if tags: span.set_tags(tags) yield span def current_span(self) -> Optional[Span]: current_span = trace.get_current_span() if isinstance(current_span, NonRecordingSpan): return None return OpenTelemetrySpan(current_span)

然后让 Haystack 使用该自定义追踪器：

from haystack import tracing haystack_tracer = OpenTelemetryTracer(tracer) tracing.enable_tracing(haystack_tracer)

实现Tracer.trace时，务必以上下文管理器返回 Span，并正确处理current_span（例如区分非录制 Span 与真实 Span），这样管线内所有组件都会自动使用你的追踪器产生嵌套 Span。

与其他追踪后端的横向定位

除了 OpenTelemetry，Haystack 2.22 时代还提供以下开箱即用的追踪方案，便于你在选型时横向对比：

Datadog：安装ddtrace后，可用ddtrace <command>运行应用，或在代码中tracing.enable_tracing(DatadogTracer(ddtrace.tracer))；
Langfuse：通过LangfuseConnector组件（pip install langfuse-haystack）在 Langfuse UI 中可视化管线 trace；
Weights & Biases Weave：通过WeaveConnector组件（pip install weave-haystack）在 W&B 框架中追踪与监控管线组件；
自定义后端：实现Tracer接口即可接入任意可观测性平台。