DatadogConnector

DatadogConnector

Haystack에서 Datadog을 사용하는 방법을 배워요. DatadogConnector는 Datadog의 추적 라이브러리 ddtrace를 통해 Haystack 파이프라인에 추적(tracing) 기능을 통합합니다.

출처: 문서

본문

Overview

DatadogConnector은 Datadog의 추적 라이브러리 ddtrace를 통해 Datadog을 사용해 Haystack 파이프라인에 추적 기능을 통합해요. API 호출, 컨텍스트 데이터, 프롬프트 등 파이프라인 실행에 대한 상세 정보를 캡처하므로, Datadog에서 파이프라인 실행의 완전한 트레이스를 볼 수 있습니다.

Datadog 추적은 DatadogConnector이 초기화되는 즉시 활성화되므로, 파이프라인에 추가하기만 하면 됩니다. 다른 컴포넌트에 연결하거나 실행할 필요가 없어요.

추적 컴포넌트를 식별하는 name을 선택적으로 전달할 수 있습니다(기본값은 datadog).

Prerequisites

DatadogConnector을 작업하기 전에 필요한 것들입니다:

  • 트레이스를 받을 수단, 예를 들어 실행 중인 Datadog Agent. ddtrace는 기본적으로 localhost:8126의 Datadog Agent에 트레이스를 보냅니다.
  • HAYSTACK_CONTENT_TRACING_ENABLED 환경 변수를 true로 설정 — 이렇게 하면 파이프라인에서 콘텐츠 추적(입력과 출력)이 활성화됩니다.
  • 표준 메커니즘으로 ddtrace를 구성 — 예를 들어 DD_SERVICE, DD_ENV, DD_VERSION 환경 변수나 ddtrace-run 명령으로 애플리케이션을 실행. 자세한 내용은 ddtrace 문서를 참고하세요.

Installation

DatadogConnector을 사용하려면 먼저 datadog-haystack 패키지를 설치하세요:

pip install datadog-haystack

사용 시 주의: 올바른 추적을 위해 항상 Haystack 컴포넌트를 임포트하기 전에 환경 변수를 설정하세요. Haystack은 임포트 중에 내부 추적 컴포넌트를 초기화하기 때문에 이는 매우 중요합니다. 아래 예시에서는 먼저 환경 변수를 설정한 다음 관련 Haystack 컴포넌트를 임포트합니다. 또는 스크립트를 실행하기 전에 셸에서 환경 변수를 설정하는 더 나은 방법이 있습니다. 이는 구성을 코드와 분리해 서로 다른 환경을 더 쉽게 관리하게 해줍니다.

Usage

아래 예시에서는 DatadogConnector을 tracer로 파이프라인에 추가하고 있어요. 각 파이프라인 실행은 프롬프트, 완성(completion), 메타데이터를 포함한 전체 실행 컨텍스트가 담긴 트레이스를 생성합니다. 그런 다음 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())
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])

With an Agent

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, f"Weather information for {city} not available")


@tool
def calculate(
    operation: Annotated[
        str,
        "Mathematical operation: add, subtract, multiply, divide",
    ],
    a: Annotated[float, "First number"],
    b: Annotated[float, "Second number"],
) -> str:
    """Perform basic mathematical calculations."""
    if operation == "add":
        result = a + b
    elif operation == "subtract":
        result = a - b
    elif operation == "multiply":
        result = a * b
    elif operation == "divide":
        if b == 0:
            return "Error: Division by zero"
        result = a / b
    else:
        return f"Error: Unknown operation '{operation}'"

    return f"The result of {a} {operation} {b} is {result}"


# Create the chat generator
chat_generator = OpenAIChatGenerator()

# Create the agent with tools
agent = Agent(
    chat_generator=chat_generator,
    tools=[get_weather, calculate],
    system_prompt="You are a helpful assistant with access to weather and calculator tools. Use them when needed.",
    exit_conditions=["text"],
)

# Create the DatadogConnector for tracing
datadog_connector = DatadogConnector("Agent Example")

# Build the pipeline
pipe = Pipeline()
pipe.add_component("tracer", datadog_connector)
pipe.add_component("agent", agent)

# Run the pipeline
response = pipe.run(
    data={
        "agent": {
            "messages": [
                ChatMessage.from_user(
                    "What's the weather in Berlin and calculate 15 + 27?",
                ),
            ],
        },
        "tracer": {},
    },
)

# Display results
print("Agent Response:")
print(response["agent"]["last_message"].text)

Configuring the tracing backend directly

DatadogConnector을 사용하는 대신, DatadogTracer를 활성화해 Datadog 추적 백엔드를 직접 구성할 수 있어요. Haystack 컴포넌트를 임포트하기 전에 HAYSTACK_CONTENT_TRACING_ENABLED 환경 변수를 설정해야 합니다.

import ddtrace

from haystack import tracing
from haystack_integrations.tracing.datadog import DatadogTracer

tracing.enable_tracing(DatadogTracer(ddtrace.tracer))

더 알아보기 (Learn more)