使用 OpenTelemetry 跟踪

​

跟踪 LangChain 应用程序

1. 安装支持 OpenTelemetry 的 LangSmith 包：
   pip

   Copy

   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

   Copy

   LANGSMITH_OTEL_ENABLED=true
   LANGSMITH_TRACING=true
   LANGSMITH_ENDPOINT=https://api.smith.langchain.com
   LANGSMITH_API_KEY=<your_langsmith_api_key>
   # 对于链接到多个工作区的 LangSmith API 密钥，设置 LANGSMITH_WORKSPACE_ID 环境变量以指定要使用的工作区。
   

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

   Copy

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

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

   Copy

   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

   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. 记录跟踪。 此代码设置一个 OTEL 跟踪器和导出器，它将跟踪发送到 LangSmith。然后它调用 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: 覆盖端点 URL
OTEL_EXPORTER_OTLP_HEADERS: 添加自定义标头（自动添加 LangSmith API 密钥和项目）
OTEL_SERVICE_NAME: 设置自定义服务名称（默认为 "langsmith"）

1. 如上所示设置 OTEL 环境变量，或
2. Set a global trace provider before initializing LangChain components, which LangSmith will detect and use instead of creating its own.

​

Configure alternate OTLP endpoints

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)

​

Supported OpenTelemetry attribute and event mapping

​

Core LangSmith attributes

​

GenAI standard attributes

​

GenAI request parameters

​

GenAI usage metrics

​

TraceLoop attributes

​

OpenInference attributes

​

LLM attributes

​

Prompt template attributes

​

Retriever attributes

​

Tool attributes

​

Logfire attributes

​

OpenTelemetry event mapping

​

Event attribute extraction

• content → message content
• role → message role
• id → tool_call_id (for tool messages)
• gen_ai.event.content → full message JSON

• finish_reason → choice finish reason
• message.content → choice message content
• message.role → choice message role
• tool_calls.{n}.id → tool call ID
• tool_calls.{n}.function.name → tool function name
• tool_calls.{n}.function.arguments → tool function arguments
• tool_calls.{n}.type → tool call type

• exception.message → error message
• exception.stacktrace → error stacktrace (appended to message)

​

Implementation examples

​

Trace using the 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. View the trace in your LangSmith dashboard (example).

​

Advanced configuration

​

Use OpenTelemetry Collector for fan-out

1. Install the OpenTelemetry Collector for your environment.
2. Create a configuration file (e.g., otel-collector-config.yaml) that exports to multiple destinations:
   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. Configure your application to send to the 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)
   

• Centralized configuration for all your telemetry destinations
• Reduced overhead in your application code
• Better scalability and resilience
• Ability to add or remove destinations without changing application code

​

Distributed tracing with LangChain and OpenTelemetry

​

Context propagation in distributed tracing

• Trace ID: A unique identifier for the entire trace
• Span ID: A unique identifier for the current span
• Sampling Decision: Indicates whether this trace should be sampled

​

Set up distributed tracing with 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)