LangSmith 支持基于 OpenTelemetry 的追踪，允许您从任何 OpenTelemetry 兼容的应用程序发送追踪。本指南涵盖 LangChain 应用程序的自动插桩和其他框架的手动插桩。

了解如何使用 OpenTelemetry 在 LangSmith 中追踪您的 LLM 应用程序。

追踪 LangChain 应用程序

如果您使用 LangChain 或 LangGraph，请使用内置集成来追踪您的应用程序：

1. 安装带有 OpenTelemetry 支持的 LangSmith 包：

   pip install "langsmith[otel]"
   pip install langchain
   

1. In your LangChain/LangGraph App, enable the OpenTelemetry integration by setting the LANGSMITH_OTEL_ENABLED 环境变量：

   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.
   

1. 创建一个带有追踪功能的 LangChain 应用程序。例如：

   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)
   

1. 在您的 LangSmith 仪表板中查看追踪（示例），一旦您的应用程序运行。

追踪非 LangChain 应用程序

对于非 LangChain 应用程序或自定义插桩，您可以使用标准 OpenTelemetry 客户端在 LangSmith 中追踪您的应用程序。（我们推荐 **langsmith ≥ 0.4.25**.)

1. 安装 OpenTelemetry SDK、OpenTelemetry 导出器包以及 OpenAI 包：

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

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

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

可选：指定除“default”以外的自定义项目名称：

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

1. 记录一条追踪。

此代码设置了一个 OTEL 追踪器和导出器，它将把追踪发送到 LangSmith。然后它调用 OpenAI 并发送所需的 OpenTelemetry 属性。

   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-5.4-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()
   

1. 在您的 LangSmith 仪表板中查看追踪（示例).

将追踪发送到备用提供商

虽然 LangSmith 是 OpenTelemetry 追踪的默认目标，您也可以将 OpenTelemetry 配置为向其他可观测性平台发送追踪。

使用环境变量进行全局配置

默认情况下，LangSmith OpenTelemetry 导出器会向 LangSmith API 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 导出器：

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

# 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 属性和事件映射

通过 OpenTelemetry 将追踪发送到 LangSmith 时，以下属性会映射到 LangSmith 字段：

核心 LangSmith 属性

GenAI 标准属性

GenAI 请求参数

GenAI 使用指标

TraceLoop 属性

OpenInference 属性

LLM 属性

提示词模板属性

检索器属性

工具属性

Logfire 属性

OpenTelemetry 事件映射

事件属性提取

对于 message 事件，提取以下属性：

• * content → message content
• * role → message role
• * id → tool\_call\_id（用于 tool messages）
• * gen_ai.event.content → 完整 message JSON

对于 choice 事件：

• * 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 → 错误消息
• * exception.stacktrace → 错误堆栈跟踪（附加到消息）

实现示例

使用 LangSmith SDK 进行追踪

使用 LangSmith SDK 的 OpenTelemetry 辅助工具配置导出。以下示例 追踪一个 Google ADK 代理:

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

以下是 示例 展示了最终追踪在 LangSmith 中的样子。

将 OpenTelemetry 跨度链接到 LangSmith SDK 追踪

原生 OTLP parentSpanId 无法引用现有的 LangSmith 运行：OTLP 跨度 ID 为 8 字节，而 LangSmith 运行 ID 是完整的 UUID。若要将 OpenTelemetry 跨度附加到在其他地方创建的运行（例如 LangChain-SDK 运行），请设置 langsmith.* 属性使用父级的完整 UUID。这些属性会覆盖从原生 OTLP 跨度派生的 ID，因此跨度嵌套在同一追踪中的现有运行下。

from datetime import datetime, timezone

from langsmith import get_current_run_tree, traceable
from opentelemetry import trace

tracer = trace.get_tracer("my-harness")


def emit_otel_child(parent):
    """Emit an OTel span that nests under an existing LangSmith run."""
    child_id = uuid.uuid4()
    start = datetime.now(timezone.utc)
    # dotted_order = parent's dotted_order, a dot, then this span's timestamp + id
    dotted = f"{parent.dotted_order}.{start.strftime('%Y%m%dT%H%M%S%fZ')}{child_id}"

    with tracer.start_as_current_span("otel-child") as span:
        span.set_attribute("langsmith.trace.id", str(parent.trace_id))           # same trace
        span.set_attribute("langsmith.span.parent_id", str(parent.id))           # nest under parent
        span.set_attribute("langsmith.span.id", str(child_id))                   # this span's run id
        span.set_attribute("langsmith.span.dotted_order", dotted)                # position in the tree
        span.set_attribute("langsmith.trace.session_name", parent.session_name)  # same project
        # ... your instrumented work here ...


@traceable
def my_agent():
    parent = get_current_run_tree()  # the LangSmith run to nest under
    emit_otel_child(parent)

向追踪添加附件

LangSmith 支持 向追踪附加文件。这在构建具有多模态输入或输出的代理时非常有用。使用 OpenTelemetry 进行追踪时也支持附件。

以下示例 追踪一个 Google ADK 代理 并向追踪添加附件。它结合使用了 LangSmith 的 OtelSpanProcessor 和一个自定义 AttachmentSpanProcessor ，使用 on_end() 将图像附件添加到父跨度。

from pathlib import Path
from dotenv import load_dotenv
from google.adk import Runner
from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService
from google.genai import types
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider, SpanProcessor
from langsmith.integrations.otel import OtelSpanProcessor


class AttachmentSpanProcessor(SpanProcessor):
    """Custom SpanProcessor to add attachments to invocation spans."""

    def __init__(self):
        self.attachment_data = None

    def set_attachment(self, attachment_data):
        self.attachment_data = attachment_data

    def on_end(self, span):
        if span.name == "invocation" and self.attachment_data:
            attachments_json = json.dumps([self.attachment_data])
            span._attributes["langsmith.attachments"] = attachments_json

load_dotenv()

# Set up TracerProvider manually
provider = TracerProvider()
trace.set_tracer_provider(provider)

# Add attachment processor FIRST (runs before LangSmith processor)
attachment_processor = AttachmentSpanProcessor()
provider.add_span_processor(attachment_processor)

# Add LangSmith processor SECOND (receives already-modified spans)
langsmith_processor = OtelSpanProcessor(project="travel-assistant")
provider.add_span_processor(langsmith_processor)

def get_flight_info(destination: str, departure_date: str) -> dict:
    """Get flight information for a destination."""
    return {
        "destination": destination,
        "departure_date": departure_date,
        "price": "$450",
        "duration": "5h 30m",
        "airline": "Example Airways"
    }

def get_hotel_recommendations(city: str, check_in: str) -> dict:
    """Get hotel recommendations for a city."""
    return {
        "city": city,
        "check_in": check_in,
        "hotels": [
            {"name": "Grand Plaza Hotel", "rating": 4.5, "price": "$120/night"},
            {"name": "City Center Inn", "rating": 4.2, "price": "$95/night"}
        ]
    }

async def main():
    # Prepare the attachment
    receipt_path = Path("receipt-template-example.png")
    with open(receipt_path, "rb") as img_file:
        image_bytes = img_file.read()
        image_base64 = base64.b64encode(image_bytes).decode("ascii")

    attachment_data = {
        "name": "receipt-template-example",
        "content": image_base64,
        "mime_type": "image/jpeg",
    }

    attachment_processor.set_attachment(attachment_data)

    # Create ADK agent
    agent = LlmAgent(
        name="travel_assistant",
        tools=[get_flight_info, get_hotel_recommendations],
        model="gemini-2.0-flash-exp",
        instruction="You are a helpful travel assistant that can help with flights and hotels.",
    )

    # Set up session and runner
    session_service = InMemorySessionService()
    runner = Runner(
        app_name="travel_app",
        agent=agent,
        session_service=session_service
    )

    await session_service.create_session(
        app_name="travel_app",
        user_id="traveler_456",
        session_id="session_789"
    )

    # Send a message to the agent
    new_message = types.Content(
        parts=[types.Part(text="I need to book a flight to Paris for March 15th and find a good hotel.")],
        role="user",
    )

    # Run the agent and process events
    events = runner.run(
        user_id="traveler_456",
        session_id="session_789",
        new_message=new_message,
    )

    for event in events:
        print(event)

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

以下是 示例 展示了最终追踪在 LangSmith 中的样子。

高级配置

使用 OpenTelemetry 收集器进行扇出

使用 LANGSMITH_OTEL_ENABLED=true 当需要 OTEL 扇出时。配置应用程序一次发出 OTEL 跨度，然后使用 OpenTelemetry 收集器将其路由到 LangSmith 和其他可观测性后端。

当追踪应用程序并需要多目标路由时使用此方法。如果运营 LangSmith 平台基础设施遥测（来自 Kubernetes 上自托管 LangSmith 服务的日志、指标、追踪），请使用 为 LangSmith 遥测配置您的收集器 指南。

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

1. 安装 OpenTelemetry 收集器 用于您的环境。

1. 创建配置文件（例如， 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. 配置您的应用程序发送到收集器：

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

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

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

在分布式系统中，上下文传播在服务之间传递追踪元数据，以便相关跨度链接到同一追踪：

• * **追踪 ID**：整个追踪的唯一标识符
• * **跨度 ID**：当前跨度的唯一标识符
• * **采样决策**：指示是否应对此追踪进行采样

使用 LangChain 设置分布式追踪

要在多个服务之间启用分布式追踪：

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

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)