OpenTelemetry로 추적하기

​

LangChain 애플리케이션 추적하기

1. OpenTelemetry 지원이 포함된 LangSmith 패키지를 설치합니다:
   pip

   Copy

   pip install "langsmith[otel]"
   pip install langchain
   

   Python SDK 버전 langsmith>=0.3.18이 필요합니다. 중요한 OpenTelemetry 수정 사항을 활용하려면 langsmith>=0.4.25를 권장합니다.
2. LangChain/LangGraph 앱에서 LANGSMITH_OTEL_ENABLED 환경 변수를 설정하여 OpenTelemetry 통합을 활성화합니다:
   Shell

   Copy

   LANGSMITH_OTEL_ENABLED=true
   LANGSMITH_TRACING=true
   LANGSMITH_ENDPOINT=https://api.smith.langchain.com
   LANGSMITH_API_KEY=<your_langsmith_api_key>
   # For LangSmith API keys linked to multiple workspaces, set the LANGSMITH_WORKSPACE_ID environment variable to specify which workspace to use.
   

3. 추적 기능이 있는 LangChain 애플리케이션을 생성합니다. 예를 들어:
   Copy

   import os
   from langchain_openai import ChatOpenAI
   from langchain_core.prompts import ChatPromptTemplate
   
   # Create a chain
   prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
   model = ChatOpenAI()
   chain = prompt | model
   
   # Run the chain
   result = chain.invoke({"topic": "programming"})
   print(result.content)
   

4. 애플리케이션이 실행되면 LangSmith 대시보드(예시)에서 추적 데이터를 확인합니다.

​

LangChain이 아닌 애플리케이션 추적하기

1. OpenTelemetry SDK, OpenTelemetry exporter 패키지 및 OpenAI 패키지를 설치합니다:
   pip

   Copy

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

2. endpoint에 대한 환경 변수를 설정하고, 특정 값으로 대체합니다:
   Shell

   Copy

   OTEL_EXPORTER_OTLP_ENDPOINT=https://api.smith.langchain.com/otel
   OTEL_EXPORTER_OTLP_HEADERS="x-api-key=<your langsmith api key>"
   

   otel exporter가 구성된 방식에 따라 추적만 전송하는 경우 endpoint에 /v1/traces를 추가해야 할 수 있습니다.
   LangSmith를 자체 호스팅하는 경우, 기본 endpoint를 LangSmith api endpoint로 교체하고 /api/v1을 추가하세요. 예: OTEL_EXPORTER_OTLP_ENDPOINT=https://ai-company.com/api/v1/otel
   선택 사항: “default” 이외의 사용자 정의 프로젝트 이름을 지정합니다:
   Shell

   Copy

   OTEL_EXPORTER_OTLP_ENDPOINT=https://api.smith.langchain.com/otel
   OTEL_EXPORTER_OTLP_HEADERS="x-api-key=<your langsmith api key>,Langsmith-Project=<project name>"
   

3. 추적을 기록합니다. 이 코드는 LangSmith로 추적 데이터를 전송할 OTEL tracer와 exporter를 설정합니다. 그런 다음 OpenAI를 호출하고 필요한 OpenTelemetry 속성을 전송합니다.
   Copy

   from openai import OpenAI
   from opentelemetry import trace
   from opentelemetry.sdk.trace import TracerProvider
   from opentelemetry.sdk.trace.export import (
       BatchSpanProcessor,
   )
   from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
   
   client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
   
   otlp_exporter = OTLPSpanExporter(
       timeout=10,
   )
   
   trace.set_tracer_provider(TracerProvider())
   trace.get_tracer_provider().add_span_processor(
       BatchSpanProcessor(otlp_exporter)
   )
   
   tracer = trace.get_tracer(__name__)
   
   def call_openai():
       model = "gpt-4o-mini"
       with tracer.start_as_current_span("call_open_ai") as span:
           span.set_attribute("langsmith.span.kind", "LLM")
           span.set_attribute("langsmith.metadata.user_id", "user_123")
           span.set_attribute("gen_ai.system", "OpenAI")
           span.set_attribute("gen_ai.request.model", model)
           span.set_attribute("llm.request.type", "chat")
   
           messages = [
               {"role": "system", "content": "You are a helpful assistant."},
               {
                   "role": "user",
                   "content": "Write a haiku about recursion in programming."
               }
           ]
   
           for i, message in enumerate(messages):
               span.set_attribute(f"gen_ai.prompt.{i}.content", str(message["content"]))
               span.set_attribute(f"gen_ai.prompt.{i}.role", str(message["role"]))
   
           completion = client.chat.completions.create(
               model=model,
               messages=messages
           )
   
           span.set_attribute("gen_ai.response.model", completion.model)
           span.set_attribute("gen_ai.completion.0.content", str(completion.choices[0].message.content))
           span.set_attribute("gen_ai.completion.0.role", "assistant")
           span.set_attribute("gen_ai.usage.prompt_tokens", completion.usage.prompt_tokens)
           span.set_attribute("gen_ai.usage.completion_tokens", completion.usage.completion_tokens)
           span.set_attribute("gen_ai.usage.total_tokens", completion.usage.total_tokens)
   
           return completion.choices[0].message
   
   if __name__ == "__main__":
       call_openai()
   

4. LangSmith 대시보드(예시)에서 추적 데이터를 확인합니다.

​

다른 제공자에게 추적 데이터 전송하기

​

전역 구성을 위한 환경 변수 사용

OTEL_EXPORTER_OTLP_ENDPOINT: Override the endpoint URL
OTEL_EXPORTER_OTLP_HEADERS: Add custom headers (LangSmith API keys and Project are added automatically)
OTEL_SERVICE_NAME: Set a custom service name (defaults to "langsmith")

1. 위에 표시된 대로 OTEL 환경 변수를 설정하거나,
2. LangChain 컴포넌트를 초기화하기 전에 전역 trace provider를 설정하면, LangSmith가 이를 감지하고 자체 provider를 생성하는 대신 사용합니다.

​

대체 OTLP endpoint 구성

import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Set environment variables for LangChain
os.environ["LANGSMITH_OTEL_ENABLED"] = "true"
os.environ["LANGSMITH_TRACING"] = "true"

# Configure the OTLP exporter for your custom endpoint
provider = TracerProvider()
otlp_exporter = OTLPSpanExporter(
    # Change to your provider's endpoint
    endpoint="https://otel.your-provider.com/v1/traces",
    # Add any required headers for authentication
    headers={"api-key": "your-api-key"}
)
processor = BatchSpanProcessor(otlp_exporter)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

# Create and run a LangChain application
prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
model = ChatOpenAI()
chain = prompt | model
result = chain.invoke({"topic": "programming"})
print(result.content)

​

지원되는 OpenTelemetry 속성 및 이벤트 매핑

​

핵심 LangSmith 속성

​

GenAI 표준 속성

​

GenAI 요청 매개변수

​

GenAI 사용량 메트릭

​

TraceLoop 속성

​

OpenInference 속성

​

LLM 속성

​

프롬프트 템플릿 속성

​

Retriever 속성

​

도구 속성

​

Logfire 속성

​

OpenTelemetry 이벤트 매핑

​

이벤트 속성 추출

• content → 메시지 내용
• role → 메시지 역할
• id → tool_call_id (도구 메시지의 경우)
• gen_ai.event.content → 전체 메시지 JSON

• finish_reason → 선택 완료 이유
• message.content → 선택 메시지 내용
• message.role → 선택 메시지 역할
• tool_calls.{n}.id → 도구 호출 ID
• tool_calls.{n}.function.name → 도구 함수 이름
• tool_calls.{n}.function.arguments → 도구 함수 인수
• tool_calls.{n}.type → 도구 호출 유형

• exception.message → 오류 메시지
• exception.stacktrace → 오류 스택트레이스 (메시지에 추가됨)

​

구현 예제

​

LangSmith SDK를 사용한 추적

import asyncio
from langsmith.integrations.otel import configure
from google.adk import Runner
from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Configure LangSmith OpenTelemetry export (no OTEL env vars or headers needed)
configure(project_name="adk-otel-demo")


async def main():
    agent = LlmAgent(
        name="travel_assistant",
        model="gemini-2.5-flash-lite",
        instruction="You are a helpful travel assistant.",
    )

    session_service = InMemorySessionService()
    runner = Runner(app_name="travel_app", agent=agent, session_service=session_service)

    user_id = "user_123"
    session_id = "session_abc"
    await session_service.create_session(app_name="travel_app", user_id=user_id, session_id=session_id)

    new_message = types.Content(parts=[types.Part(text="Hi! Recommend a weekend trip to Paris.")], role="user")

    for event in runner.run(user_id=user_id, session_id=session_id, new_message=new_message):
        print(event)


if __name__ == "__main__":
    asyncio.run(main())

5. LangSmith 대시보드(예시)에서 추적 데이터를 확인합니다.

​

고급 구성

​

fan-out을 위한 OpenTelemetry Collector 사용

1. 환경에 맞는 OpenTelemetry Collector를 설치합니다.
2. 여러 대상으로 export하는 구성 파일(예: otel-collector-config.yaml)을 생성합니다:
   Copy

   receivers:
     otlp:
       protocols:
         grpc:
           endpoint: 0.0.0.0:4317
         http:
           endpoint: 0.0.0.0:4318
   
   processors:
     batch:
   
   exporters:
     otlphttp/langsmith:
       endpoint: https://api.smith.langchain.com/otel/v1/traces
       headers:
         x-api-key: ${env:LANGSMITH_API_KEY}
         Langsmith-Project: my_project
     otlphttp/other_provider:
       endpoint: https://otel.your-provider.com/v1/traces
       headers:
         api-key: ${env:OTHER_PROVIDER_API_KEY}
   
   service:
     pipelines:
       traces:
         receivers: [otlp]
         processors: [batch]
         exporters: [otlphttp/langsmith, otlphttp/other_provider]
   

3. collector로 전송하도록 애플리케이션을 구성합니다:
   Copy

   import os
   from opentelemetry import trace
   from opentelemetry.sdk.trace import TracerProvider
   from opentelemetry.sdk.trace.export import BatchSpanProcessor
   from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
   from langchain_openai import ChatOpenAI
   from langchain_core.prompts import ChatPromptTemplate
   
   # Point to your local OpenTelemetry Collector
   otlp_exporter = OTLPSpanExporter(
       endpoint="http://localhost:4318/v1/traces"
   )
   provider = TracerProvider()
   processor = BatchSpanProcessor(otlp_exporter)
   provider.add_span_processor(processor)
   trace.set_tracer_provider(provider)
   
   # Set environment variables for LangChain
   os.environ["LANGSMITH_OTEL_ENABLED"] = "true"
   os.environ["LANGSMITH_TRACING"] = "true"
   
   # Create and run a LangChain application
   prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
   model = ChatOpenAI()
   chain = prompt | model
   result = chain.invoke({"topic": "programming"})
   print(result.content)
   

• 모든 텔레메트리 대상에 대한 중앙 집중식 구성
• 애플리케이션 코드의 오버헤드 감소
• 더 나은 확장성과 복원력
• 애플리케이션 코드를 변경하지 않고 대상을 추가하거나 제거할 수 있는 기능

​

LangChain과 OpenTelemetry를 사용한 분산 추적

​

분산 추적의 컨텍스트 전파

• Trace ID: 전체 추적에 대한 고유 식별자
• Span ID: 현재 span에 대한 고유 식별자
• Sampling Decision: 이 추적을 샘플링해야 하는지 여부를 나타냄

​

LangChain으로 분산 추적 설정

import os
from opentelemetry import trace
from opentelemetry.propagate import inject, extract
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
import requests
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Set up OpenTelemetry trace provider
provider = TracerProvider()
otlp_exporter = OTLPSpanExporter(
    endpoint="https://api.smith.langchain.com/otel/v1/traces",
    headers={"x-api-key": os.getenv("LANGSMITH_API_KEY"), "Langsmith-Project": "my_project"}
)
processor = BatchSpanProcessor(otlp_exporter)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)

# Service A: Create a span and propagate context to Service B
def service_a():
    with tracer.start_as_current_span("service_a_operation") as span:
        # Create a chain
        prompt = ChatPromptTemplate.from_template("Summarize: {text}")
        model = ChatOpenAI()
        chain = prompt | model

        # Run the chain
        result = chain.invoke({"text": "OpenTelemetry is an observability framework"})

        # Propagate context to Service B
        headers = {}
        inject(headers)  # Inject trace context into headers

        # Call Service B with the trace context
        response = requests.post(
            "http://service-b.example.com/process",
            headers=headers,
            json={"summary": result.content}
        )
        return response.json()

# Service B: Extract the context and continue the trace
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/process", methods=["POST"])
def service_b_endpoint():
    # Extract the trace context from the request headers
    context = extract(request.headers)
    with tracer.start_as_current_span("service_b_operation", context=context) as span:
        data = request.json
        summary = data.get("summary", "")

        # Process the summary with another LLM chain
        prompt = ChatPromptTemplate.from_template("Analyze the sentiment of: {text}")
        model = ChatOpenAI()
        chain = prompt | model
        result = chain.invoke({"text": summary})

        return jsonify({"analysis": result.content})

if __name__ == "__main__":
    app.run(port=5000)