LangfuseConnector
LangfuseConnector
Langfuse를 Haystack에서 다루는 방법을 배워 볼게요. LangfuseConnector는 Langfuse를 이용해 Haystack 파이프라인에 추적(tracing) 기능을 더해줘요.
본문
개요
LangfuseConnector는 Langfuse를 사용해 Haystack 파이프라인에 추적 기능을 통합해요. API 호출, 컨텍스트 데이터, 프롬프트 같은 파이프라인 실행 정보를 상세히 포착해요. 이 컴포넌트를 써서 할 수 있는 일이에요.
- 토큰 사용량과 비용 같은 모델 성능을 모니터링해요.
- 낮은 품질의 출력을 찾아내고 사용자 피드백을 수집해서 파이프라인을 개선할 영역을 찾아요.
- 파이프라인 실행에서 파인튜닝·테스트용 데이터셋을 만들어요.
통합을 쓰려면 파이프라인에 LangfuseConnector를 추가하고 파이프라인을 실행한 뒤, Langfuse 웹사이트에서 추적 데이터를 보면 돼요. 이 컴포넌트를 다른 컴포넌트에 연결하지 마세요. LangfuseConnector는 파이프라인의 백그라운드에서 그냥 실행되니깐요.
이 컴포넌트를 쓸 때 선택적으로 두 파라미터를 더 정의할 수 있어요.
httpx_client: Langfuse API 호출을 위한 선택적 커스텀httpx.Client인스턴스예요. 커스텀 클라이언트는 YAML에서 파이프라인을 역직렬화할 때 버려져요. HTTPX 클라이언트는 직렬화할 수 없기 때문이에요. 그런 경우 Langfuse가 기본 클라이언트를 만들어요.span_handler: 스팬(span) 처리를 위한 선택적 커스텀 핸들러예요. 제공하지 않으면DefaultSpanHandler를 사용해요. 스팬 핸들러는 스팬을 생성·처리하는 방식을 정의해서, 컴포넌트 타입에 따른 스팬 타입 커스터마이징과 스팬 후처리를 가능하게 해요. 자세한 내용은 아래 고급 사용법에서 다뤄요.
사전 준비
LangfuseConnector를 쓰기 전에 필요한 것들이에요.
- 활성화된 Langfuse 계정이 있어야 해요.
HAYSTACK_CONTENT_TRACING_ENABLED환경 변수를true로 설정해서 파이프라인에서 추적을 켜요.- 계정 프로필에서 찾을 수 있는
LANGFUSE_SECRET_KEY와LANGFUSE_PUBLIC_KEY환경 변수를 설정해요.
설치
LangfuseConnector를 쓰려면 먼저 langfuse-haystack 패키지를 설치해요.
pip install langfuse-haystack
사용 주의: 추적을 제대로 하려면 Haystack 컴포넌트를 import 하기 전에 항상 환경 변수를 설정해야 해요. Haystack이 import 과정에서 내부 추적 컴포넌트를 초기화하기 때문이에요. 아래 예시는 먼저 환경 변수를 설정한 뒤 관련 Haystack 컴포넌트를 import 해요. 더 나은 방법은 스크립트를 실행하기 전에 셸에서 환경 변수를 설정하는 거예요. 설정을 코드와 분리해 두면 여러 환경을 관리하기 쉬워져요.
사용법
아래 예시에서 LangfuseConnector를 파이프라인에 tracer로 추가하고 있어요. 파이프라인을 실행할 때마다 프롬프트, 완성 결과, 메타데이터를 포함한 전체 실행 컨텍스트가 담긴 트레이스(trace) 하나가 생성돼요. 출력에 찍힌 URL 링크를 따라가면 트레이스를 볼 수 있어요.
import os
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"
os.environ["TOKENIZERS_PARALLELISM"] = "false"
os.environ["HAYSTACK_CONTENT_TRACING_ENABLED"] = "true"
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack import Pipeline
from haystack_integrations.components.connectors.langfuse import LangfuseConnector
if __name__ == "__main__":
pipe = Pipeline()
pipe.add_component("tracer", LangfuseConnector("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])
print(response["tracer"]["trace_url"])
Agent와 함께 쓰기
import os
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"
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.langfuse import LangfuseConnector
@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"
else:
result = a / b
else:
return f"Error: Unknown operation '{operation}'"
return f"The result of {a} {operation} {b} is {result}"
if __name__ == "__main__":
# Create components
chat_generator = OpenAIChatGenerator()
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"],
)
langfuse_connector = LangfuseConnector("Agent Example")
# Create and run pipeline
pipe = Pipeline()
pipe.add_component("tracer", langfuse_connector)
pipe.add_component("agent", agent)
response = pipe.run(
data={
"agent": {
"messages": [
ChatMessage.from_user(
"What's the weather in Berlin and calculate 15 + 27?",
),
],
},
"tracer": {"invocation_context": {"test": "agent_with_tools"}},
},
)
print(response["agent"]["last_message"].text)
print(response["tracer"]["trace_url"])
고급 사용법: SpanHandler로 Langfuse 트레이스 커스터마이징
Haystack의 SpanHandler 인터페이스는 Langfuse 트레이스 생성을 위해 스팬을 생성·처리하는 방식을 커스터마이즈할 수 있게 해줘요. 이를 통해 커스텀 메트릭을 기록하거나 태그를 추가하거나 메타데이터를 통합할 수 있어요.
SpanHandler나 그 기본 구현인 DefaultSpanHandler를 확장해서 스팬 처리 로직을 정의하면, Langfuse에 무엇을 기록할지 정밀하게 제어할 수 있어요.
from haystack_integrations.components.connectors.langfuse import LangfuseConnector
from haystack_integrations.tracing.langfuse import DefaultSpanHandler, LangfuseSpan
from typing import Optional
class CustomSpanHandler(DefaultSpanHandler):
def handle(self, span: LangfuseSpan, component_type: Optional[str]) -> None:
# Custom logic to add metadata or modify span
if component_type == "OpenAIChatGenerator":
output = span._data.get("haystack.component.output", {})
if len(output.get("text", "")) < 10:
span._span.update(level="WARNING", status_message="Response too short")
# Add the custom handler to the LangfuseConnector
connector = LangfuseConnector(
"Custom Handler Example", span_handler=CustomSpanHandler()
)