用 OpenTelemetry 追踪

在 LangSmith 中配置 OpenTelemetry 追踪，包括 LANGSMITH_OTEL_ENABLED，以及用 OpenTelemetry Collector 做 OTEL fanout。

LangSmith 支持基于 OpenTelemetry 的追踪，让你能从任何 OpenTelemetry 兼容的应用发送追踪。本指南涵盖对 LangChain 应用的自动插桩，以及对其他框架的手动插桩。

学习如何用 OpenTelemetry 配 LangSmith 来追踪你的 LLM 应用。

OTel 追踪如何工作

下图展示用 LangSmith 做 OpenTelemetry 追踪的基本流程，包括一条 span 流被路由到多个可观测性后端的 fanout 模式。

%%{init: {"theme": "base", "themeVariables": {"lineColor": "#40668D", "primaryColor": "#E5F4FF", "primaryTextColor": "#030710", "primaryBorderColor": "#006DDD"}}}%%
flowchart TD
    App["Your application\n(LangChain, LangGraph,\nor any OTel-compatible app)"]
    SDK["OpenTelemetry SDK\n(instruments code, creates spans)"]
    Collector["OpenTelemetry Collector\n(receives, processes, and routes spans)"]
    LangSmith["LangSmith\n(traces + runs dashboard)"]
    Other["Other observability backend\n(Datadog, Honeycomb, Grafana, etc.)"]

    App -->|"emits spans"| SDK
    SDK -->|"exports via OTLP"| Collector
    Collector -->|"fanout"| LangSmith
    Collector -->|"fanout"| Other

    classDef process fill:#E5F4FF,stroke:#006DDD,stroke-width:2px,color:#030710
    classDef output fill:#EBD0F0,stroke:#885270,stroke-width:2px,color:#441E33
    classDef neutral fill:#F2FAFF,stroke:#40668D,stroke-width:2px,color:#2F4B68

    class App,SDK process
    class Collector neutral
    class LangSmith,Other output

OpenTelemetry SDK 插桩你的应用代码并发出 span。span 经 OTLP 协议传到 OpenTelemetry Collector，它批量处理并把它们同时路由到一个或多个目的地（fanout）。LangSmith 在其 OTLP 端点接收 span，并在仪表盘上把它们显示为追踪。

NOTE

对下面请求中的自托管安装或区域 SaaS，适当更新 LangSmith URL：GCP EU 用 eu.api.smith.langchain.com；GCP APAC 用 apac.api.smith.langchain.com；AWS US 用 aws.api.smith.langchain.com。

追踪一个 LangChain 应用

如果你在用 LangChain 或 LangGraph，用内置集成来追踪你的应用：

安装带 OpenTelemetry 支持的 LangSmith 包：

pip

bash

pip install "langsmith[otel]"
pip install langchain

INFO

需要 Python SDK 版本 langsmith>=0.3.18。我们推荐 langsmith>=0.4.25 以受益于重要的 OpenTelemetry 修复。

在你的 LangChain/LangGraph App 中，通过设置 LANGSMITH_OTEL_ENABLED 环境变量启用 OpenTelemetry 集成：

Shell

bash

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.

创建一个带追踪的 LangChain 应用。例如：

python

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)

应用运行后，在你的 LangSmith 仪表盘查看追踪（示例）。

追踪一个非 LangChain 应用

对非 LangChain 应用或自定义插桩，你可以用一个标准 OpenTelemetry 客户端在 LangSmith 中追踪你的应用。（我们推荐 langsmith ≥ 0.4.25。）

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

pip

bash

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

为端点设置环境变量，替换成你的具体值：

Shell

bash

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

NOTE

OTEL_EXPORTER_OTLP_ENDPOINT 是一个基础 URL。OTLP/HTTP 导出器自己会附加信号路径（/v1/traces），所以别把它包含进来：把基础端点设为 https://api.smith.langchain.com/otel/v1/traces 会把追踪发到 /otel/v1/traces/v1/traces，返回 404。只在使用逐信号的 OTEL_EXPORTER_OTLP_TRACES_ENDPOINT 变量或 collector 导出器的 traces_endpoint 时（它们按所给原样使用）才包含完整路径。

NOTE

如果你在自托管 LangSmith，把基础端点换成你的 LangSmith api 端点并追加 /api/v1。例如：OTEL_EXPORTER_OTLP_ENDPOINT=https://ai-company.com/api/v1/otel

可选：指定一个不同于 "default" 的自定义项目名：

Shell

bash

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

记录一条追踪。

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

python

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

在你的 LangSmith 仪表盘查看追踪（示例）。

NOTE

如果你的 span 引用了另一服务或进程里的父项，见分布式追踪中的上下文传播了解父子关联如何工作，以及一个 span 何时会被丢弃。

把追踪发送到一个替代提供商

虽然 LangSmith 是 OpenTelemetry 追踪的默认目的地，你也能配置 OpenTelemetry 把追踪发送到其他可观测性平台。

INFO

在 LangSmith Python SDK ≥ 0.4.1 中可用。我们推荐 ≥ 0.4.25 以获得改进 OTEL 导出和混合 fan-out 稳定性的修复。

用环境变量做全局配置

默认情况下，LangSmith OpenTelemetry 导出器会把数据发送到 LangSmith API OTEL 端点，但这可以通过设置标准 OTEL 环境变量来定制：

bash

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")
OTEL_RESOURCE_ATTRIBUTES: Attach custom metadata fields at the process level (see below)

添加进程级资源属性

你能用标准 OpenTelemetry OTEL_RESOURCE_ATTRIBUTES 环境变量，把自定义元数据附加到进程发出的每条追踪。与 span 级属性（在代码里逐 span 设置）不同，资源属性在进程级设置一次并自动传播到所有 span。这使它很适合任何你想给自己的追踪打标的自定义元数据，例如用户名、请求 ID、环境或部署版本。

该值是 key=value 对的逗号分隔列表：

bash

OTEL_RESOURCE_ATTRIBUTES="username=abc,id=1,environment=production"

你可以用任何你喜欢的自定义键，或遵循 OpenTelemetry 资源语义约定来处理像 deployment.environment、service.version 和 cloud.region 这样的标准字段。

这些属性出现在 LangSmith 中追踪的元数据下，你可以用它们跨你的工作区过滤和分组追踪。

LangSmith 默认用 HTTP 追踪导出器。如果你想用自己的追踪提供商，你可以：

如上所示设置 OTEL 环境变量，或
在初始化 LangChain 组件之前设一个全局追踪提供商，LangSmith 会检测并使用它而非创建它自己的。

配置替代 OTLP 端点

要把追踪发送到不同提供商，用你提供商的端点配置 OTLP 导出器：

python

import os

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)

INFO

混合追踪在版本 ≥ 0.4.1 中可用。要只把追踪发送到你的 OTEL 端点，设：

LANGSMITH_OTEL_ONLY="true" （建议：langsmith ≥ 0.4.25。）

受支持的 OpenTelemetry 属性与事件映射

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

核心 LangSmith 属性

OpenTelemetry 属性	LangSmith 字段	备注
`langsmith.trace.name`	Run name	为运行覆盖 span 名
`langsmith.span.kind`	运行类型	值：`llm`、`chain`、`tool`、`retriever`、`embedding`、`prompt`、`parser`
`langsmith.trace.id`	Trace ID	该 span 所属的追踪（根运行）；设置它以附加到一条现有追踪
`langsmith.span.id`	Run ID	此 span 的运行 ID（一个 UUID）；覆盖从 OTLP span ID 派生的 ID
`langsmith.span.parent_id`	Parent run ID	按运行 ID 把该 span 嵌套到一个现有运行之下
`langsmith.span.dotted_order`	Dotted order	在追踪树中的位置：`<parent.dotted_order>.<timestamp><span.id>`。见 `dotted_order`。
`langsmith.trace.session_id`	Session ID	相关追踪的会话标识符
`langsmith.trace.session_name`	Session name	会话的名字
`langsmith.span.tags`	Tags	附加到 span 的自定义标签（逗号分隔）
`langsmith.metadata.{key}`	`metadata.{key}`	带 langsmith 前缀的自定义元数据

GenAI 标准属性

OpenTelemetry 属性	LangSmith 字段	备注
`gen_ai.system`	`metadata.ls_provider`	GenAI 系统（例如 "openai"、"anthropic"）
`gen_ai.operation.name`	Run type	把 "chat"/"completion" 映射到 "llm"，"embedding" 到 "embedding"
`gen_ai.prompt`	`inputs`	发送给模型的输入提示
`gen_ai.completion`	`outputs`	模型生成的输出
`gen_ai.prompt.{n}.role`	`inputs.messages[n].role`	第 n 条输入消息的角色
`gen_ai.prompt.{n}.content`	`inputs.messages[n].content`	第 n 条输入消息的内容
`gen_ai.prompt.{n}.message.role`	`inputs.messages[n].role`	角色的另一种格式
`gen_ai.prompt.{n}.message.content`	`inputs.messages[n].content`	内容的另一种格式
`gen_ai.completion.{n}.role`	`outputs.messages[n].role`	第 n 条输出消息的角色
`gen_ai.completion.{n}.content`	`outputs.messages[n].content`	第 n 条输出消息的内容
`gen_ai.completion.{n}.message.role`	`outputs.messages[n].role`	角色的另一种格式
`gen_ai.completion.{n}.message.content`	`outputs.messages[n].content`	内容的另一种格式
`gen_ai.input.messages`	`inputs.messages`	输入消息数组
`gen_ai.output.messages`	`outputs.messages`	输出消息数组
`gen_ai.tool.name`	`invocation_params.tool_name`	工具名，也把运行类型设为 "tool"

GenAI 请求参数

OpenTelemetry 属性	LangSmith 字段	备注
`gen_ai.request.model`	`invocation_params.model`	请求所用的模型名
`gen_ai.response.model`	`invocation_params.model`	响应中返回的模型名
`gen_ai.request.temperature`	`invocation_params.temperature`	Temperature 设置
`gen_ai.request.top_p`	`invocation_params.top_p`	Top-p 采样设置
`gen_ai.request.max_tokens`	`invocation_params.max_tokens`	最大 token 设置
`gen_ai.request.frequency_penalty`	`invocation_params.frequency_penalty`	Frequency penalty 设置
`gen_ai.request.presence_penalty`	`invocation_params.presence_penalty`	Presence penalty 设置
`gen_ai.request.seed`	`invocation_params.seed`	生成所用的随机种子
`gen_ai.request.stop_sequences`	`invocation_params.stop`	停止生成的序列
`gen_ai.request.top_k`	`invocation_params.top_k`	Top-k 采样参数
`gen_ai.request.encoding_formats`	`invocation_params.encoding_formats`	输出编码格式

GenAI 用量指标

OpenTelemetry 属性	LangSmith 字段	备注
`gen_ai.usage.input_tokens`	`usage_metadata.input_tokens`	所用输入 token 数
`gen_ai.usage.output_tokens`	`usage_metadata.output_tokens`	所用输出 token 数
`gen_ai.usage.total_tokens`	`usage_metadata.total_tokens`	所用总 token 数
`gen_ai.usage.prompt_tokens`	`usage_metadata.input_tokens`	所用输入 token 数（已弃用）
`gen_ai.usage.completion_tokens`	`usage_metadata.output_tokens`	所用输出 token 数（已弃用）
`gen_ai.usage.details.reasoning_tokens`	`usage_metadata.reasoning_tokens`	所用 reasoning token 数

TraceLoop 属性

OpenTelemetry 属性	LangSmith 字段	备注
`traceloop.entity.input`	`inputs`	来自 TraceLoop 的完整输入值
`traceloop.entity.output`	`outputs`	来自 TraceLoop 的完整输出值
`traceloop.entity.name`	Run name	来自 TraceLoop 的实体名
`traceloop.span.kind`	Run type	映射到 LangSmith 运行类型
`traceloop.llm.request.type`	Run type	"embedding" 映射到 "embedding"，其他映射到 "llm"
`traceloop.association.properties.{key}`	`metadata.{key}`	带 traceloop 前缀的自定义元数据

OpenInference 属性

OpenTelemetry 属性	LangSmith 字段	备注
`input.value`	`inputs`	完整输入值，可为字符串或 JSON
`output.value`	`outputs`	完整输出值，可为字符串或 JSON
`openinference.span.kind`	Run type	把各种 kind 映射到 LangSmith 运行类型
`llm.system`	`metadata.ls_provider`	LLM 系统提供商
`llm.model_name`	`metadata.ls_model_name`	来自 OpenInference 的模型名
`tool.name`	Run name	当 span 类型为 "TOOL" 时的工具名
`metadata`	`metadata.*`	要合并的元数据 JSON 字符串

LLM 属性

OpenTelemetry 属性	LangSmith 字段	备注
`llm.input_messages`	`inputs.messages`	输入消息
`llm.output_messages`	`outputs.messages`	输出消息
`llm.token_count.prompt`	`usage_metadata.input_tokens`	提示 token 数
`llm.token_count.completion`	`usage_metadata.output_tokens`	补全 token 数
`llm.token_count.total`	`usage_metadata.total_tokens`	总 token 数
`llm.usage.total_tokens`	`usage_metadata.total_tokens`	另一种总 token 数
`llm.invocation_parameters`	`invocation_params.*`	调用参数的 JSON 字符串
`llm.presence_penalty`	`invocation_params.presence_penalty`	Presence penalty
`llm.frequency_penalty`	`invocation_params.frequency_penalty`	Frequency penalty
`llm.request.functions`	`invocation_params.functions`	函数定义

提示模板属性

OpenTelemetry 属性	LangSmith 字段	备注
`llm.prompt_template.variables`	Run type	把运行类型设为 "prompt"，与 input.value 一起用

检索器属性

OpenTelemetry 属性	LangSmith 字段	备注
`retrieval.documents.{n}.document.content`	`outputs.documents[n].page_content`	第 n 条检索文档的内容
`retrieval.documents.{n}.document.metadata`	`outputs.documents[n].metadata`	第 n 条检索文档的元数据（JSON）

工具属性

OpenTelemetry 属性	LangSmith 字段	备注
`tools`	`invocation_params.tools`	工具定义数组
`tool_arguments`	`invocation_params.tool_arguments`	作为 JSON 或键值对的工具参数

Logfire 属性

OpenTelemetry 属性	LangSmith 字段	备注
`prompt`	`inputs`	Logfire 提示输入
`all_messages_events`	`outputs`	Logfire 消息事件输出
`events`	`inputs`/`outputs`	Logfire 事件数组，拆分 input/choice 事件

OpenTelemetry 事件映射

事件名	LangSmith 字段	备注
`gen_ai.content.prompt`	`inputs`	从事件属性提取提示内容
`gen_ai.content.completion`	`outputs`	从事件属性提取补全内容
`gen_ai.system.message`	`inputs.messages[]`	会话中的系统消息
`gen_ai.user.message`	`inputs.messages[]`	会话中的用户消息
`gen_ai.assistant.message`	`outputs.messages[]`	会话中的助手消息
`gen_ai.tool.message`	`outputs.messages[]`	工具响应消息
`gen_ai.choice`	`outputs`	带结束原因的模型 choice/响应
`exception`	`status`、`error`	把状态设为 "error" 并提取异常消息/堆栈

事件属性提取

对消息事件，提取以下属性：

content → 消息内容
role → 消息角色
id → tool_call_id（对工具消息）
gen_ai.event.content → 完整消息 JSON

对 choice 事件：

finish_reason → choice 结束原因
message.content → choice 消息内容
message.role → choice 消息角色
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 追踪

用 LangSmith SDK 的 OpenTelemetry 助手来配置导出。下面的例子追踪一个 Google ADK agent：

python

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

NOTE

你无需设置 OTEL 环境变量或导出器。configure() 会为 LangSmith 自动接线它们；插桩器（如 GoogleADKInstrumentor）创建 span。

这里是一个示例，展示结果追踪在 LangSmith 中的样子。

把 OpenTelemetry span 关联到一条 LangSmith SDK 追踪

原生 OTLP parentSpanId 无法引用一个现有的 LangSmith 运行：OTLP span ID 是 8 字节，而 LangSmith 运行 ID 是完整 UUID。要把一个 OpenTelemetry span 附加到别处创建的运行（例如一个 LangChain-SDK 运行），用父项完整 UUID 设置 langsmith.* 属性。它们覆盖从原生 OTLP span 派生的 ID，所以该 span 在同一追踪中嵌套到现有运行之下。

python

import uuid
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)

NOTE

langsmith.span.dotted_order 编码 span 在追踪树中的位置。用父项的 dotted_order、一个点，然后这个 span 的时间戳跟它的 langsmith.span.id 来构造它。

给追踪添加一个附件

LangSmith 支持向追踪附加文件。这在构建带多模态输入或输出的 agent 时很有用。用 OpenTelemetry 追踪时也支持附件。

下面的例子追踪一个 Google ADK agent并给追踪添加一个附件。它用一个 LangSmith 的 OtelSpanProcessor 加一个自定义 AttachmentSpanProcessor 的组合，后者用 on_end() 给父 span 添加一个图片附件。

python

import asyncio
import base64
import json
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 collector 做 fan-out

当你需要 OTEL fanout 时用 LANGSMITH_OTEL_ENABLED=true。配置你的应用发出一次 OTEL span，然后用一个 OpenTelemetry Collector 把它们路由到 LangSmith 和任何额外的可观测性后端。

当你在追踪应用并想要多目的路由时用这种方法。如果你在运营 LangSmith 平台基础设施遥测（来自 Kubernetes 上自托管 LangSmith 服务的日志、指标、追踪），请改用为你的 collector 配置 LangSmith 遥测指南。

对更高级的场景，你能用 OpenTelemetry Collector 把你的遥测数据 fan out 到多个目的地。这比在你的应用代码里配置多个导出器是更可扩展的做法。

为你的环境安装 OpenTelemetry Collector。

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

yaml

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:

exporters:
  otlphttp/langsmith:
    traces_endpoint: https://api.smith.langchain.com/otel/v1/traces
    headers:
      x-api-key: ${env:LANGSMITH_API_KEY}
      Langsmith-Project: my_project
  otlphttp/other_provider:
    traces_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]

配置你的应用发送到 collector：

python

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

当你的 LLM 应用跨多个服务或进程时，分布式追踪至关重要。OpenTelemetry 的上下文传播能力确保追踪跨服务边界仍保持连接。

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

在分布式系统中，上下文传播在服务间传递追踪元数据，使相关的 span 关联到同一追踪：

Trace ID：整条追踪的唯一标识符
Span ID：当前 span 的唯一标识符
Sampling Decision：指示这条追踪是否应被采样

WARNING

一个父项从未被发送到 LangSmith 的 span 会被丢弃。

OTel 端点是异步的：它接受一个批次，返回 200，并在后台物化运行。当一个 span 的 parentSpanId 引用了一个 LangSmith 尚未收到的父项时，子项被缓冲并在父项到达时关联。这无论顺序如何都能工作；一个子项可以在父项之前到达、在一个单独请求中，仍能正确关联。

不过，如果那个父 span 从未被导出到 LangSmith，缓冲的子项会过期并永不作为运行出现。LangSmith 不会返回错误，因为 200 在处理之前就发了。这常发生在一整条分布式追踪只有部分到达 LangSmith 时：父项由一个导出到不同后端的服务发出，或被某个采样决策移除。

要避免静默丢失，确保你想进 LangSmith 的每个 span，其祖先也在缓冲窗口内被导出到 LangSmith。在自托管部署上，这个窗口由 REDIS_RUNS_EXPIRY_SECONDS（默认 12 小时）控制。

用 LangChain 设置分布式追踪

要跨多个服务启用分布式追踪：

python

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)

数据集

创建数据集

运行评估

评估技术

定义评估目标

打分方法

实验配置

多模态评估

分析实验结果

微调

教程

常见数据类型

评估器类型

UI

SDK

框架与集成

改进评估器

反馈

追踪设置

集成

手动埋点

配置与故障排除

项目与环境设置

高级追踪技术

数据与隐私

故障排除指南

查看与管理追踪

批量导出追踪数据

轨迹视图

数据类型参考

监控与告警

在线评估器

自动化

开发你的应用

设置依赖

持久化

能力

助手

运行

并发消息处理

如何构建

鉴权与访问控制

自定义鉴权教程

服务器定制

静态加密

替换内置后端

扩展 HTTP 服务器

请求头与日志

部署指南

参考

配置

参考

提示词

创建与管理提示词

连接模型

上下文中枢

教程

全栈 Web 应用

Agent Server API

控制平面 API

计费与用量

参考

按云服务商起步

用 Terraform 部署

AWS

GCP

Azure

安装实例

配置实例

连接外部服务

平台鉴权与访问控制

管理实例

自托管可观测性

混合部署

脚本

参考

组织