使用 OpenTelemetry 进行追踪

​

追踪 LangChain 应用程序

1. 安装支持 OpenTelemetry 的 LangSmith 包
   pip

   复制

   向 AI 提问

   pip install "langsmith[otel]"
   pip install langchain
   

   需要 Python SDK 版本 langsmith>=0.3.18。我们推荐 langsmith>=0.4.25 以受益于重要的 OpenTelemetry 修复。
2. 在您的 LangChain/LangGraph 应用程序中，通过设置 LANGSMITH_OTEL_ENABLED 环境变量来启用 OpenTelemetry 集成
   Shell

   复制

   向 AI 提问

   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 应用程序。例如
   复制

   向 AI 提问

   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 导出器包以及 OpenAI 包
   pip

   复制

   向 AI 提问

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

2. 设置端点环境变量，替换为您的特定值
   Shell

   复制

   向 AI 提问

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

   根据您的 otel 导出器配置方式，如果您只发送追踪，您可能需要在端点后附加 /v1/traces。
   如果您是自托管 LangSmith，请将基本端点替换为您的 LangSmith api 端点并附加 /api/v1。例如：OTEL_EXPORTER_OTLP_ENDPOINT=https://ai-company.com/api/v1/otel
   可选：指定一个自定义项目名称而不是“default”
   Shell

   复制

   向 AI 提问

   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. 记录追踪。 此代码设置了一个 OTEL 追踪器和导出器，它将向 LangSmith 发送追踪。然后它调用 OpenAI 并发送所需的 OpenTelemetry 属性。
   复制

   向 AI 提问

   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 组件之前设置一个全局追踪提供商，LangSmith 将检测并使用它而不是创建自己的。

​

配置备用 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)

​

支持的 OpenTelemetry 属性和事件映射

​

核心 LangSmith 属性

​

GenAI 标准属性

​

GenAI 请求参数

​

GenAI 使用指标

​

TraceLoop 属性

​

OpenInference 属性

​

LLM 属性

​

提示模板属性

​

检索器属性

​

工具属性

​

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 仪表板中查看追踪（示例）。

​

高级配置

​

使用 OpenTelemetry Collector 进行扇出

1. 为您的环境安装 OpenTelemetry Collector。
2. 创建一个配置文件（例如，otel-collector-config.yaml），该文件导出到多个目的地
   复制

   向 AI 提问

   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
   复制

   向 AI 提问

   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 的分布式追踪

​

分布式追踪中的上下文传播

• 追踪 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)