使用 Temporal 进行追踪

Temporal 是一个持久化执行平台，使开发者能够构建弹性的分布式应用程序。本指南将向您展示如何使用 OpenTelemetry 在 LangSmith 中追踪 Temporal 工作流和活动。 LangSmith 支持 OpenTelemetry (OTEL) 追踪摄取，这与 Temporal 原生的 OpenTelemetry 拦截器无缝集成。这使得您可以对工作流执行、活动以及其中的任何 LLM 调用进行完整的分布式追踪。

前提条件

一个 LangSmith 账户和 API 密钥
正在运行的 Temporal 服务器（本地或云端）
您所用语言的 OpenTelemetry SDK

环境变量

为所有实现设置以下环境变量：

变量名	必需	描述
`LANGSMITH_API_KEY`	是	您从设置中获取的 LangSmith API 密钥。
`LANGSMITH_PROJECT`	否	项目名称（默认为 `"default"`）。

对于欧盟区域或自托管的 LangSmith 安装，还需将 LANGCHAIN_BASE_URL 设置为您的 LangSmith 实例 URL。

设置追踪

Go
Python
TypeScript / JavaScript

Go 使用 langsmith-go SDK 与 Temporal 的 OpenTelemetry 拦截器来自动追踪工作流和活动。

安装

安装 LangSmith Go SDK、Temporal SDK 和 OpenTelemetry 拦截器：

go get github.com/langchain-ai/langsmith-go@v0.1.0-alpha.7
go get go.temporal.io/sdk
go get go.temporal.io/sdk/contrib/opentelemetry

初始化追踪器

初始化 LangSmith 追踪器，创建 Temporal 的 OpenTelemetry 拦截器，并将其注册到 Temporal 客户端和 worker：

package main

import (
	"context"
	"log"

	"github.com/langchain-ai/langsmith-go"
	"go.temporal.io/sdk/client"
	"go.temporal.io/sdk/contrib/opentelemetry"
	"go.temporal.io/sdk/interceptor"
	"go.temporal.io/sdk/worker"
)

func main() {
	ctx := context.Background()

	// 初始化 LangSmith 追踪器（读取 LANGSMITH_API_KEY 和 LANGSMITH_PROJECT）
	ls, err := langsmith.NewTracer(
		langsmith.WithServiceName("temporal-worker"),
	)
	if err != nil {
		log.Fatal("Failed to initialize LangSmith tracer:", err)
	}
	defer ls.Shutdown(ctx)

	// 创建 Temporal 追踪拦截器
	tracer := ls.Tracer("temporal-app")
	tracingInterceptor, err := opentelemetry.NewTracingInterceptor(
		opentelemetry.TracerOptions{Tracer: tracer},
	)
	if err != nil {
		log.Fatal("Failed to create tracing interceptor:", err)
	}

	// 创建带有追踪功能的 Temporal 客户端
	c, err := client.Dial(client.Options{
		Interceptors: []interceptor.ClientInterceptor{tracingInterceptor},
	})
	if err != nil {
		log.Fatal("Failed to create Temporal client:", err)
	}
	defer c.Close()

	// 创建带有追踪功能的 worker（使用相同的客户端）
	w := worker.New(c, "my-task-queue", worker.Options{})
	w.RegisterWorkflow(MyWorkflow)
	w.RegisterActivity(MyActivity)

	// 启动 worker
	if err := w.Run(worker.InterruptCh()); err != nil {
		log.Fatal("Worker failed:", err)
	}
}

定义工作流和活动

定义一个执行活动的工作流。该活动演示了如何为 LangSmith 可见性添加自定义跨度属性：

package main

import (
	"context"
	"fmt"
	"time"

	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	"go.temporal.io/sdk/activity"
	"go.temporal.io/sdk/workflow"
)

// MyWorkflow 执行一个活动
func MyWorkflow(ctx workflow.Context, input string) (string, error) {
	ao := workflow.ActivityOptions{
		StartToCloseTimeout: 10 * time.Second,
	}
	ctx = workflow.WithActivityOptions(ctx, ao)

	var result string
	err := workflow.ExecuteActivity(ctx, MyActivity, input).Get(ctx, &result)
	return result, err
}

// MyActivity 使用自定义跨度属性处理输入
func MyActivity(ctx context.Context, input string) (string, error) {
	logger := activity.GetLogger(ctx)
	logger.Info("Processing", "input", input)

	// 获取由 Temporal 拦截器创建的跨度
	span := trace.SpanFromContext(ctx)

	// 添加 Gen AI 属性以提高 LangSmith 可见性
	span.SetAttributes(
		attribute.String("gen_ai.prompt", input),
		attribute.String("gen_ai.operation.name", "chat"),
	)

	result := fmt.Sprintf("Processed: %s", input)

	// 设置完成属性
	span.SetAttributes(
		attribute.String("gen_ai.completion", result),
	)

	return result, nil
}

执行工作流

在单独的客户端应用程序中，初始化追踪器并执行工作流：

client.go

// 在单独的函数或客户端应用程序中
func executeWorkflow() {
	ctx := context.Background()

	// 为客户端初始化追踪器
	ls, err := langsmith.NewTracer(
		langsmith.WithServiceName("temporal-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer ls.Shutdown(ctx)

	// 创建带有追踪功能的客户端
	tracer := ls.Tracer("temporal-app")
	tracingInterceptor, err := opentelemetry.NewTracingInterceptor(
		opentelemetry.TracerOptions{Tracer: tracer},
	)
	if err != nil {
		log.Fatal(err)
	}

	c, err := client.Dial(client.Options{
		Interceptors: []interceptor.ClientInterceptor{tracingInterceptor},
	})
	if err != nil {
		log.Fatal(err)
	}
	defer c.Close()

	// 执行工作流
	workflowOptions := client.StartWorkflowOptions{
		ID:        "my-workflow-1",
		TaskQueue: "my-task-queue",
	}

	we, err := c.ExecuteWorkflow(ctx, workflowOptions, MyWorkflow, "Hello World")
	if err != nil {
		log.Fatal(err)
	}

	var result string
	if err := we.Get(ctx, &result); err != nil {
		log.Fatal(err)
	}

	log.Printf("Workflow result: %s", result)
}

Python 使用 temporalio SDK 与 OpenTelemetry 拦截器，通过 OTLP 将追踪导出到 LangSmith。

安装

安装 Temporal SDK、LangSmith SDK 和 OpenTelemetry 包：

pip install temporalio
pip install langsmith
pip install opentelemetry-sdk
pip install opentelemetry-exporter-otlp-proto-http

初始化追踪器

创建一个配置了 OTLP 导出器的 OpenTelemetry TracerProvider，用于将追踪发送到 LangSmith：

import asyncio
import os
from datetime import timedelta

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource, SERVICE_NAME
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.contrib.opentelemetry import TracingInterceptor
from temporalio.worker import Worker


def init_tracer_provider() -> TracerProvider:
    """使用 LangSmith 导出器初始化 OpenTelemetry。"""

    # 为 LangSmith 创建 OTLP 导出器
    exporter = OTLPSpanExporter(
        endpoint="https://api.smith.langchain.com/otel/v1/traces",
        headers={
            "x-api-key": os.environ.get("LANGSMITH_API_KEY", ""),
            "Langsmith-Project": os.environ.get("LANGSMITH_PROJECT", "default"),
        },
    )

    # 创建带有资源属性的 TracerProvider
    resource = Resource.create({
        SERVICE_NAME: "temporal-worker",
    })

    provider = TracerProvider(resource=resource)
    provider.add_span_processor(BatchSpanProcessor(exporter))

    # 设置为全局提供者
    trace.set_tracer_provider(provider)

    return provider

定义工作流和活动

定义一个工作流类和活动函数。该活动演示了如何为 LangSmith 可见性添加自定义跨度属性：

@activity.defn
async def process_activity(input: str) -> str:
    """使用自定义跨度属性处理输入的活动。"""
    activity.logger.info(f"Processing: {input}")

    # 获取当前跨度并添加 Gen AI 属性
    span = trace.get_current_span()
    span.set_attribute("gen_ai.prompt", input)
    span.set_attribute("gen_ai.operation.name", "chat")

    result = f"Processed: {input}"

    span.set_attribute("gen_ai.completion", result)

    return result


@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self, input: str) -> str:
        return await workflow.execute_activity(
            process_activity,
            input,
            start_to_close_timeout=timedelta(seconds=10),
        )

运行 worker

创建带有 TracingInterceptor 的 Temporal 客户端并启动 worker：

async def main():
    # 初始化追踪
    provider = init_tracer_provider()

    try:
        # 创建带有追踪拦截器的 Temporal 客户端
        client = await Client.connect(
            "localhost:7233",
            interceptors=[TracingInterceptor()],
        )

        # 运行 worker
        worker = Worker(
            client,
            task_queue="my-task-queue",
            workflows=[MyWorkflow],
            activities=[process_activity],
        )

        print("Starting worker...")
        await worker.run()

    finally:
        # 关闭追踪器提供者以刷新追踪
        provider.shutdown()


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

执行工作流

在单独的脚本中，使用追踪拦截器连接到 Temporal 并执行工作流：

client.py

import asyncio
from temporalio.client import Client
from temporalio.contrib.opentelemetry import TracingInterceptor

# 导入相同的追踪器设置
from worker import init_tracer_provider


async def main():
    provider = init_tracer_provider()

    try:
        client = await Client.connect(
            "localhost:7233",
            interceptors=[TracingInterceptor()],
        )

        # 执行工作流
        result = await client.execute_workflow(
            MyWorkflow.run,
            "Hello World",
            id="my-workflow-1",
            task_queue="my-task-queue",
        )
        print(f"Workflow result: {result}")

    finally:
        provider.shutdown()


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

TypeScript 使用 @temporalio/sdk 与 OpenTelemetry 拦截器将追踪发送到 LangSmith。

安装

安装 Temporal SDK、OpenTelemetry 拦截器和追踪包：

npm install @temporalio/client @temporalio/worker @temporalio/activity @temporalio/workflow
npm install @temporalio/interceptors-opentelemetry
npm install @opentelemetry/sdk-node @opentelemetry/sdk-trace-node
npm install @opentelemetry/exporter-trace-otlp-http
npm install @opentelemetry/resources @opentelemetry/semantic-conventions

初始化追踪器

创建一个配置了 OTLP 导出器的 NodeTracerProvider，用于将追踪发送到 LangSmith：

tracer.ts

import { Resource } from '@opentelemetry/resources';
import { ATTR_SERVICE_NAME } from '@opentelemetry/semantic-conventions';
import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-base';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';

export function initTracerProvider(): NodeTracerProvider {
  // 为 LangSmith 创建 OTLP 导出器
  const exporter = new OTLPTraceExporter({
    url: 'https://api.smith.langchain.com/otel/v1/traces',
    headers: {
      'x-api-key': process.env.LANGSMITH_API_KEY || '',
      'Langsmith-Project': process.env.LANGSMITH_PROJECT || 'default',
    },
  });

  // 创建 TracerProvider
  const provider = new NodeTracerProvider({
    resource: new Resource({
      [ATTR_SERVICE_NAME]: 'temporal-worker',
    }),
  });

  provider.addSpanProcessor(new BatchSpanProcessor(exporter));
  provider.register();

  return provider;
}

定义工作流

定义一个使用超时配置代理活动的工作流：

workflows.ts

import { proxyActivities } from '@temporalio/workflow';
import type * as activities from './activities';

const { processActivity } = proxyActivities<typeof activities>({
  startToCloseTimeout: '10 seconds',
});

export async function myWorkflow(input: string): Promise<string> {
  return await processActivity(input);
}

定义活动

定义一个活动，演示如何为 LangSmith 可见性添加自定义跨度属性：

activities.ts

import { log } from '@temporalio/activity';
import { trace } from '@opentelemetry/api';

export async function processActivity(input: string): Promise<string> {
  log.info('Processing', { input });

  // 获取当前跨度并添加 Gen AI 属性
  const span = trace.getActiveSpan();
  span?.setAttribute('gen_ai.prompt', input);
  span?.setAttribute('gen_ai.operation.name', 'chat');

  const result = `Processed: ${input}`;

  span?.setAttribute('gen_ai.completion', result);

  return result;
}

运行 worker

创建一个带有 OpenTelemetry 拦截器的 worker 用于活动，以及一个工作流导出器用于工作流跨度：

worker.ts

import { Worker, NativeConnection } from '@temporalio/worker';
import { Resource } from '@opentelemetry/resources';
import { ATTR_SERVICE_NAME } from '@opentelemetry/semantic-conventions';
import {
  makeWorkflowExporter,
  OpenTelemetryActivityInboundInterceptor,
} from '@temporalio/interceptors-opentelemetry';
import { trace } from '@opentelemetry/api';

import * as activities from './activities';
import { initTracerProvider } from './tracer';

async function run() {
  const provider = initTracerProvider();

  try {
    const connection = await NativeConnection.connect({
      address: 'localhost:7233',
    });

    const worker = await Worker.create({
      connection,
      namespace: 'default',
      taskQueue: 'my-task-queue',
      workflowsPath: require.resolve('./workflows'),
      activities,
      sinks: {
        exporter: makeWorkflowExporter(
          trace.getTracer('temporal-app'),
          new Resource({ [ATTR_SERVICE_NAME]: 'temporal-worker' })
        ),
      },
      interceptors: {
        activity: [() => ({ inbound: new OpenTelemetryActivityInboundInterceptor() })],
      },
    });

    console.log('Starting worker...');
    await worker.run();
  } finally {
    await provider.shutdown();
  }
}

run().catch((err) => {
  console.error(err);
  process.exit(1);
});

执行工作流

在单独的客户端文件中，连接到 Temporal 并执行工作流：

client.ts

import { Client, Connection } from '@temporalio/client';
import { initTracerProvider } from './tracer';

async function run() {
  // 初始化追踪
  const provider = initTracerProvider();

  try {
    const connection = await Connection.connect({ address: 'localhost:7233' });
    const client = new Client({ connection });

    const result = await client.workflow.execute('myWorkflow', {
      taskQueue: 'my-task-queue',
      workflowId: 'my-workflow-1',
      args: ['Hello World'],
    });

    console.log('Workflow result:', result);
  } finally {
    await provider.shutdown();
  }
}

run().catch(console.error);

在 LangSmith 中查看追踪

配置完成后，追踪将出现在您的 LangSmith 项目中：

导航到您的 LangSmith 实例。
选择您的项目。
在追踪选项卡中查看追踪。
点击单个追踪以查看完整的跨度层次结构。

配置选项

设置自定义服务名称

设置自定义服务名称以区分不同的 Temporal worker 或服务：

ls, err := langsmith.NewTracer(
    langsmith.WithServiceName("my-temporal-worker"),
)

resource = Resource.create({
    SERVICE_NAME: "my-temporal-worker",
})

const provider = new NodeTracerProvider({
  resource: new Resource({
    [ATTR_SERVICE_NAME]: 'my-temporal-worker',
  }),
});

添加自定义跨度属性

添加自定义属性以丰富您的追踪：

import "go.opentelemetry.io/otel/attribute"

span := trace.SpanFromContext(ctx)
span.SetAttributes(
    attribute.String("user.id", userID),
    attribute.String("workflow.version", "v2"),
)

from opentelemetry import trace

span = trace.get_current_span()
span.set_attribute("user.id", user_id)
span.set_attribute("workflow.version", "v2")

import { trace } from '@opentelemetry/api';

const span = trace.getActiveSpan();
span?.setAttribute('user.id', userId);
span?.setAttribute('workflow.version', 'v2');

配置采样

对于高吞吐量的工作流，配置采样以减少追踪量：

// 注意：langsmith.NewTracer() 使用默认采样
// 对于自定义采样，直接使用 TracerProvider
tp := sdktrace.NewTracerProvider(
    sdktrace.WithBatcher(exporter),
    sdktrace.WithSampler(sdktrace.TraceIDRatioBased(0.1)), // 10% 采样率
)

from opentelemetry.sdk.trace.sampling import TraceIdRatioBased

provider = TracerProvider(
    resource=resource,
    sampler=TraceIdRatioBased(0.1),  # 10% 采样率
)

import { TraceIdRatioBasedSampler } from '@opentelemetry/sdk-trace-base';

const provider = new NodeTracerProvider({
  resource: resource,
  sampler: new TraceIdRatioBasedSampler(0.1), // 10% 采样率
});

故障排除

追踪未出现

验证 API 密钥：确保 LANGSMITH_API_KEY 设置正确
检查端点：确认您使用的是 https://api.smith.langchain.com/otel/v1/traces
关闭时刷新：在应用程序退出前调用 provider.shutdown() 以刷新待处理的跨度
检查项目：验证追踪是否发送到正确的项目（默认为 "default"）

缺少活动跨度

确保在客户端和 worker 上都配置了追踪拦截器：

客户端：需要拦截器来启动工作流
Worker：需要拦截器来执行活动

上下文传播问题

验证传播器配置是否正确：

Go：langsmith.NewTracer() 自动配置传播器
Python/TypeScript：确保 OpenTelemetry SDK 使用追踪传播器正确初始化

Worker 关闭挂起

如果追踪未刷新，请确保使用适当的超时调用关闭方法：

defer ls.Shutdown(context.Background())

finally:
    provider.shutdown()

finally {
  await provider.shutdown();
}

后续步骤

附加资源

通过 MCP 将这些文档连接到 Claude、VSCode 等，以获取实时答案。

在 GitHub 上编辑此页面或提交问题。

Tracing setup

Configuration & troubleshooting

Viewing & managing traces

Automations

Feedback & evaluation

Monitoring & alerting

Data type reference

前提条件

环境变量

设置追踪

在 LangSmith 中查看追踪

配置选项

设置自定义服务名称

添加自定义跨度属性

配置采样

故障排除

追踪未出现

缺少活动跨度

上下文传播问题

Worker 关闭挂起

后续步骤

附加资源

​前提条件

​环境变量

​设置追踪

​在 LangSmith 中查看追踪

​配置选项

​设置自定义服务名称

​添加自定义跨度属性

​配置采样

​故障排除

​追踪未出现

​缺少活动跨度

​上下文传播问题

​Worker 关闭挂起

​后续步骤

​附加资源

前提条件

环境变量

设置追踪

在 LangSmith 中查看追踪

配置选项

设置自定义服务名称

添加自定义跨度属性

配置采样

故障排除

追踪未出现

缺少活动跨度

上下文传播问题

Worker 关闭挂起

后续步骤

附加资源