WeaveConnector

WeaveConnector

Weights & Biases의 Weave 프레임워크를 사용해 파이프라인 컴포넌트를 추적(tracing)하고 모니터링하는 방법을 담은 컴포넌트예요. 파이프라인 실행을 Weave 대시보드에 시각화해 줘요.

파이프라인에서 가장 흔한 위치: 어디든 좋아요. 다른 컴포넌트와 연결되지 않아요. 필수 init 변수: pipeline_name — 파이프라인 이름. Weave 대시보드에도 이 이름으로 표시돼요. 출력 변수: pipeline_name — 방금 실행된 파이프라인의 이름 API 레퍼런스: Weave GitHub 링크: https://github.com/deepset-ai/haystack-core-integrations/tree/main/integrations/weave 패키지 이름: weave-haystack

출처: 문서

본문

개요 (Overview)

이 통합은 Weights & Biases에서 파이프라인 실행을 추적하고 시각화할 수 있게 해 줘요.

Haystack 추적 도구가 포착한 정보(API 호출, 컨텍스트 데이터, 프롬프트 등)가 Weights & Biases로 전송되고, 여기서 파이프라인 실행의 전체 트레이스를 볼 수 있어요.

사전 준비 (Prerequisites)

이 기능을 쓰려면 Weave 계정이 필요해요. Weights & Biases 웹사이트에서 무료로 가입할 수 있어요.

그런 다음 Weights & Biases API 키와 함께 WANDB_API_KEY 환경 변수를 설정해야 해요. 로그인하면 홈 페이지에서 API 키를 찾을 수 있어요.

그 후 https://wandb.ai/<user_name>/projects로 가면, WeaveConnector를 만들 때 지정한 파이프라인 이름 아래에서 파이프라인의 전체 트레이스를 볼 수 있어요.

또한 HAYSTACK_CONTENT_TRACING_ENABLED 환경 변수를 true로 설정해야 해요.

사용법 (Usage)

먼저 이 connector를 쓰려면 weave-haystack 패키지를 설치해요:

pip install weave-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.weave import WeaveConnector

pipe = Pipeline()
pipe.add_component("prompt_builder", ChatPromptBuilder())
pipe.add_component("llm", OpenAIChatGenerator())
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,
        },
    },
)

그런 다음 https://wandb.ai/<user_name>/projects에서 WeaveConnector를 만들 때 지정한 파이프라인 이름 아래에서 파이프라인의 전체 트레이스를 볼 수 있어요.

Agent와 함께 사용하기 (With an Agent)

import os
# Enable Haystack content tracing
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.weave import WeaveConnector

@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 WeaveConnector for tracing
weave_connector = WeaveConnector(pipeline_name="Agent Example")

# Build the pipeline
pipe = Pipeline()
pipe.add_component("tracer", weave_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)
print(f"\nPipeline Name: {response['tracer']['pipeline_name']}")
print(
    "\nCheck your Weights & Biases dashboard at https://wandb.ai/<user_name>/projects to see the traces!",
)

더 알아보기 (Learn more)

• Weave API 레퍼런스
• Weights & Biases Web사이트