Haystack × Datadog：用 DatadogConnector 与 DatadogTracer 实现 Pipeline 全链路可观测

Haystack × Datadog用 DatadogConnector 与 DatadogTracer 实现 Pipeline 全链路可观测【免费下载链接】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 管道Pipeline的每一次运行接入 Datadog 分布式追踪体系通过datadog-haystack集成提供的DatadogConnector组件与DatadogTracer追踪器捕获组件级输入输出、Prompt、Completion 等上下文数据并在 Datadog Dashboard 上查看完整的执行链路。读完本文你将掌握两种启用方式组件方式与直接配置方式、环境变量配置要点、Agent 场景的接入范例以及 Haystack 底层追踪抽象Span/Tracer/ProxyTracer的工作原理。一、为什么要在 Haystack Pipeline 上接入 DatadogHaystack 以模块化 Pipeline 组织 LLM 应用一条链路里往往串联了 Prompt 构建、检索、生成等多个组件排查慢在哪一步、Prompt 最终长什么样、模型返回了什么并不直观。Datadog 集成解决的就是这个问题——它把 Datadog 的可观测能力与 Haystack 的管道执行模型打通让每个组件、每次调用都能形成可检索、可关联的 trace。正如 API 参考文档docs-website/reference_versioned_docs/version-2.23/integrations-api/datadog.md所述DatadogConnector将 Haystack 连接到 Datadog以启用对Pipeline 各组件内部操作与数据流的追踪。将其加入管道无需与其他任何组件连接开启追踪后它会自动记录所有管道操作包括 API 调用、上下文数据、提示词与模型响应。版本说明本仓库当前版本为 3.x见 VERSION.txt。自 v3.0 起DatadogTracer已从 Haystack 主包迁移至独立的datadog-haystack集成包详见下文迁移注意API 参考文档所对应的集成类路径为haystack_integrations.*。二、前置条件与安装使用 Datadog 追踪前需要准备好以下三项一个能接收 trace 的后端例如正在运行的 Datadog Agent。ddtrace默认将 trace 发送到localhost:8126。开启内容追踪设置环境变量HAYSTACK_CONTENT_TRACING_ENABLEDtrue用于记录各管道组件的输入与输出。配置ddtrace通过标准机制进行例如DD_SERVICE、DD_ENV、DD_VERSION环境变量或使用ddtrace-run命令启动应用。安装集成包pip install datadog-haystack重要时序约束HAYSTACK_CONTENT_TRACING_ENABLED必须在导入任何 Haystack 组件之前设置。这是因为 Haystack 在 import 阶段就会初始化内部的追踪设施——具体来说全局tracer对象在 haystack/tracing/tracer.py 处被实例化为ProxyTracer其构造时读取该环境变量并缓存到is_content_tracing_enabled属性见 haystack/tracing/tracer.py。最稳妥的做法是在 Shell 中先导出变量再运行脚本把配置与代码分离export HAYSTACK_CONTENT_TRACING_ENABLEDtrue export DD_SERVICEmy-haystack-app export DD_ENVproduction export DD_VERSION1.0.0 python my_app.py三、启用方式一Pipeline 组件方式DatadogConnector3.1 基本用法将DatadogConnector作为管道中的一个组件加入即可不需要连接到任何其他组件。它一旦被初始化就会启用 Datadog 追踪甚至无需运行。文档中的标准示例import os os.environ[HAYSTACK_CONTENT_TRACING_ENABLED] true 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.datadog import DatadogConnector pipe Pipeline() pipe.add_component(tracer, DatadogConnector(Chat example)) pipe.add_component(prompt_builder, ChatPromptBuilder()) pipe.add_component(llm, OpenAIChatGenerator(modelgpt-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])要点说明DatadogConnector(Chat example)中的name参数用于标识该追踪组件默认值为datadog它会作为run方法的返回值之一可用于标记本连接器产生的 trace。run时向tracer组件传入空字典即可如tracer: {}因为它不接收任何输入数据。每次pipe.run(...)都会产生一条完整的 trace包含整个执行上下文提示词、补全结果、元数据可在 Datadog Dashboard 中查看。组件方式的一个突出优点是追踪配置随 Pipeline 一起序列化例如序列化为 YAML便于把可观测性作为管道定义的一部分进行版本管理。3.2 在 Agent 场景中使用DatadogConnector同样适用于 Agent 工作流。在 docs-website/docs/pipeline-components/connectors/datadogconnector.mdx 中给出了带工具调用的完整示例核心结构如下import os os.environ[HAYSTACK_CONTENT_TRACING_ENABLED] true from typing import Annotated from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.tools import tool from haystack import Pipeline from haystack_integrations.components.connectors.datadog import DatadogConnector tool def get_weather(city: Annotated[str, The city to get weather for]) - str: Get current weather information for a city. weather_data { Berlin: 18°C, partly cloudy, New York: 22°C, sunny, Tokyo: 25°C, clear skies, } return weather_data.get(city, fWeather information for {city} not available) chat_generator OpenAIChatGenerator() agent Agent( chat_generatorchat_generator, tools[get_weather, calculate], system_promptYou are a helpful assistant with access to weather and calculator tools. Use them when needed., exit_conditions[text], ) pipe Pipeline() pipe.add_component(tracer, DatadogConnector(Agent Example)) pipe.add_component(agent, agent) response pipe.run( data{ agent: { messages: [ChatMessage.from_user(Whats the weather in Berlin and calculate 15 27?)], }, tracer: {}, }, ) print(Agent Response:) print(response[agent][last_message].text)这样一来Agent 的每一轮推理、工具选择与工具返回结果都会落入 Datadog 的 trace 中便于观察多步 Agent 的执行路径。四、启用方式二直接配置追踪后端DatadogTracer如果不想把追踪器作为管道组件管理也可以直接启用DatadogTracer作为 Haystack 的全局追踪后端。这种方式同样要求在任何 Haystack 组件导入前设置好环境变量import ddtrace from haystack import tracing from haystack_integrations.tracing.datadog import DatadogTracer tracing.enable_tracing(DatadogTracer(ddtrace.tracer))调用tracing.enable_tracing(...)后全局ProxyTracer内部的actual_tracer会从默认的NullTracer空实现不做任何事切换为 Datadog 实现见 haystack/tracing/tracer.py。此后任意 Pipeline 的运行都会被自动追踪无需向管道添加任何组件。对比两种方式对比项DatadogConnector组件方式DatadogTracer直接方式启用时机组件初始化时调用enable_tracing时是否修改 Pipeline 定义是作为组件加入否全局生效是否可随管道序列化如 YAML是否典型场景想用 Pipeline 定义管理可观测性想以最少侵入启用追踪五、API 参考DatadogConnector__init__(name: str datadog) - None初始化DatadogConnector组件。namestr用于标识该追踪组件的名称。它由run方法返回可用于标记该连接器产生的 trace默认值为datadog。run() - dict[str, str]运行DatadogConnector组件。由于追踪由初始化触发run本身不做追踪工作仅返回包含以下键的字典name该追踪组件的名称。to_dict() - dict[str, Any]将组件序列化为字典供 Pipeline 序列化使用。from_dict(data: dict[str, Any]) - DatadogConnector从字典反序列化组件实例。datadict[str, Any]组件的字典表示。返回反序列化后的DatadogConnector实例。六、API 参考DatadogSpan 与 DatadogTracerDatadogSpan继承自SpanDatadogSpan是对 Datadog 原生 span 的一层包装实现了 Haystack 的Span抽象接口接口定义见 haystack/tracing/tracer.py。__init__(span: ddSpan) - None包装一个 Datadog 底层 span 对象创建DatadogSpan实例。set_tag(key: str, value: Any) - None在 span 上设置单个标签。keystr标签名。valueAny标签值。注意值会被序列化为字符串因此最好使用字符串、数字、布尔值等简单类型。raw_span() - Any返回底层 tracer 的 span 对象便于在需要完整访问底层 span 能力时使用。get_correlation_data_for_logs() - dict[str, Any]返回用于日志与 trace 关联的字典trace ID、span ID 等配合 Datadog 的日志追踪关联功能使用。DatadogTracer继承自TracerDatadogTracer是 HaystackTracer抽象haystack/tracing/tracer.py的 Datadog 实现负责创建并提交 span。__init__(tracer: ddTracer) - None包装一个 Datadog 原生 tracer即ddtrace.tracer创建实例。trace(operation_name: str, tags: dict[str, Any] | None None, parent_span: Span | None None) - Iterator[Span]激活并返回一个新的 span该 span 继承当前活动 span 作为父 span。operation_name被追踪操作的名称。tags应用于新创建 span 的标签。parent_span新 span 的父 span若为None则新 span 成为根 span。current_span() - Span | None返回当前活动的 span若无活动 span 则返回None。七、底层原理Haystack 的追踪抽象与内容追踪开关理解这两层 API 之前有必要看一下 Haystack 核心的追踪基础设施位于 haystack/tracing/tracer.pySpan/Tracer抽象基类Tracer.trace是带contextmanager装饰的上下文管理器方法Pipeline 执行组件时进入with块创建 span退出时提交 spancurrent_span返回当前活动 span。ProxyTracer代理模式全局tracer是一个ProxyTracer内部持有真正的actual_tracer。enable_tracing(provided_tracer)替换actual_tracerdisable_tracing()将其还原为NullTraceris_tracing_enabled()则判断当前是否处于开启状态见 haystack/tracing/tracer.py。这种代理设计的好处是用户直接import tracer的对象引用无需被 monkey-patch切换实现即可全局生效。内容追踪开关Span.set_content_tag(key, value)用于写入内容类标签如查询内容、文档内容、答案内容默认被禁用。启用途径有二设置环境变量HAYSTACK_CONTENT_TRACING_ENABLEDtrue或在自定义 tracer 实现中重写set_content_tag方法见 haystack/tracing/tracer.py。启用后组件输入输出中的敏感内容才会进入 Datadog这正是文档要求显式开启该变量的原因。标签值类型规整大多数追踪后端不接受复杂类型因此 haystack/tracing/utils.py 中的coerce_tag_value会把非原始类型的值如列表、字典、文档对象序列化为 JSON 字符串后再打上 span无法序列化时兜底转为字符串。日志与 trace 关联Haystack 的日志处理器会在启用追踪时自动向日志事件注入当前 span 的关联数据correlate_logs_with_traces见 haystack/logging.py其数据来源正是各Span实现的get_correlation_data_for_logs()。这意味着在 Datadog 中Haystack 的结构化日志可以与对应 trace 自动关联无需手工拼接 trace ID。这些机制共同保证了只要一个 tracer 被启用Pipeline 里所有组件的运行都会自动产生 span无需在每个组件中手动埋点。八、迁移注意从 v2.x 到 v3.0 的变化根据 MIGRATION.mdv3.0 起 Datadog 追踪经历了如下调整DatadogTracer移出 Haystack 主包from haystack.tracing.datadog import DatadogTracer改为from haystack_integrations.tracing.datadog import DatadogTracer对应的集成包为datadog-haystack。不再自动启用v2.x 中只要安装了ddtraceDatadog 追踪就可能被自动开启v3.0 起必须显式启用——要么向 Pipeline 添加DatadogConnector组件要么手动调用tracing.enable_tracing(DatadogTracer(ddtrace.tracer))。这使追踪的启用变得明确、可预期。从 v2.x 迁移的对照# v2.x旧自动或手动启用 # from haystack.tracing.datadog import DatadogTracer # tracing.enable_tracing(DatadogTracer(ddtrace.tracer)) # v3.0新通过组件启用 from haystack import Pipeline from haystack_integrations.components.connectors.datadog import DatadogConnector pipe Pipeline() pipe.add_component(tracer, DatadogConnector())九、关联资源在仓库中继续深入可以参考以下文件API 参考文档Datadog 集成本文依据的核心文档含完整 API 签名。Datadog 开发指南集成概览、安装与DatadogTracer用法。DatadogConnector 组件文档组件属性表、Agent 示例与完整代码。haystack/tracing/tracer.pySpan、Tracer、ProxyTracer、NullTracer及enable_tracing/disable_tracing的实现。haystack/tracing/utils.pyspan 标签值的类型规整与序列化逻辑。haystack/logging.py日志与 trace 关联的处理器实现。MIGRATION.mdDatadogTracer迁移说明及 v2.x/v3.0 代码对照。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考