跳过主内容

使用 OpenTelemetry 跟踪 LangChain

LangSmith Python SDK 开箱即用地支持 OpenTelemetry 集成，让您可以跟踪 LangChain 和 LangGraph 应用程序，并将数据导出到 LangSmith、任何 OTLP 兼容的后端，或同时导出到两者。

本指南涵盖三个核心主题

  • 通过 OTel 将跟踪发送到 LangSmith

  • 导出到 OTel 兼容提供商 – 通过 OpenTelemetry Collector 将跟踪发送到第三方平台。

  • 分布式跟踪 – 跨服务传播上下文以实现端到端可见性。

信息

自 langsmith ≥ 0.4.1 版本起，设置 LANGSMITH_OTEL_ENABLED=true 将默认把跟踪发送到 LangSmith 和您的 OTEL 端点（如果您已初始化全局跟踪提供程序）。无需额外代码即可实现扇出。

通过 OTel 将跟踪发送到 LangSmith​

1. 安装​

信息

需要 Python SDK 版本 langsmith>=0.3.18。

安装支持 OpenTelemetry 的 LangSmith 包

pip install "langsmith[otel]"
pip install langchain

2. 启用 OpenTelemetry 集成​

在您的 LangChain/LangGraph 应用中，通过设置 LANGSMITH_OTEL_ENABLED 环境变量来启用 OpenTelemetry 集成

LANGSMITH_OTEL_ENABLED=true
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT=https://api.smith.langchain.com
LANGSMITH_API_KEY=<your_langsmith_api_key>

3. 创建一个带跟踪功能的 LangChain 应用程序​

这是一个简单的示例，展示了如何将 OpenTelemetry 集成与 LangChain 结合使用

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 中查看跟踪​

应用程序运行后，您将在 LangSmith 仪表板中看到跟踪，示例如此。

将跟踪发送到其他提供商​

虽然 LangSmith 是 OpenTelemetry 跟踪的默认目的地，但您也可以配置 OpenTelemetry 将跟踪发送到其他可观测性平台。

使用环境变量或全局配置​

默认情况下，LangSmith OpenTelemetry 导出器会将数据发送到 LangSmith API OTEL 端点（如果您配置了全局 TracerProvider，也会发送到 OTEL 端点），但这可以通过设置标准 OTEL 环境变量进行自定义

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")

LangSmith 默认使用 HTTP 跟踪导出器。如果您想使用自己的跟踪提供商，您可以选择

  1. 如上所示设置 OTEL 环境变量，或者
  2. 在初始化 LangChain 组件之前设置一个全局跟踪提供程序，LangSmith 将检测并使用它，而不是创建自己的。

配置备用 OTLP 端点​

要将跟踪发送到不同的提供商，请使用您的提供商的端点配置 OTLP 导出器

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)
信息

要在 0.4.1 之后禁用混合行为，仅将跟踪发送到您的 OTEL 端点并排除发送到 LangSmith，请添加一个额外的环境变量

LANGSMITH_OTEL_ONLY = "true"

使用 OpenTelemetry Collector 进行扇出​

对于更高级的场景，您可以使用 OpenTelemetry Collector 将遥测数据扇出到多个目的地。这比在应用程序代码中配置多个导出器是一种更具可伸缩性的方法。

  1. 安装 OpenTelemetry Collector：请按照 官方安装说明 为您的环境进行操作。

  2. 配置 Collector：创建一个配置文件（例如，otel-collector-config.yaml），该文件可导出到多个目的地

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]
  1. 配置您的应用程序以发送到收集器:
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="https://: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 进行分布式跟踪​

当您的 LLM 应用程序跨越多个服务或进程时，分布式跟踪至关重要。OpenTelemetry 的上下文传播能力可确保跟踪在服务边界之间保持连接。

分布式跟踪中的上下文传播​

在分布式系统中，上下文传播在服务之间传递跟踪元数据，从而使相关的 span 链接到同一跟踪

  • 跟踪 ID：整个跟踪的唯一标识符
  • Span ID：当前 span 的唯一标识符
  • 采样决策：指示是否应对此跟踪进行采样

使用 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)

此页面有帮助吗？


您可以留下详细的反馈 在 GitHub 上.