LLM 应用可观测性：基于 Langfuse 与 OpenTelemetry 的 AI SDK 链路追踪实践

Langfuse 是基于 OpenTelemetry 构建的开源 LLM 可观测性平台。本文以 Next.js 博客助手为例，结合 Langfuse TypeScript SDK 与 AI SDK 遥测接口，对对话轮次、标题生成、向量化与语义重排进行链路追踪。

方案架构与核心组件

生产环境可观测性的作用

离线评估（Evals）依赖人工构造的数据集，覆盖范围受限于编写者的预判。生产环境的链路数据可以补充三类信息：

真实使用模式：用户的实际提问方式与意图分布，可用于校准提示词与工具描述。
未覆盖的边界场景：线上出现的异常回答、工具误调用或超时中断，可直接沉淀为新的评估用例。
成本与 Token 消耗：按调用类型统计输入输出 Token 与延迟，用于定位高消耗的提示词或冗余的工具输出。

Langfuse 提供云端托管服务（cloud.langfuse.com），也支持通过 Docker 自托管部署。

Trace 与 Observation 数据模型

Langfuse 沿用 OpenTelemetry 的追踪模型：

Trace：一次完整交互的容器，例如用户发送一条消息到助手回复结束的全过程。Trace 上可附加 sessionId、userId、tags 与 metadata。
Observation：Trace 内的工作单元，对应 OpenTelemetry 中的 Span。Langfuse 将其细分为 SPAN（通用步骤）、AGENT（智能体调用入口）、GENERATION（模型生成，附带 Token 用量与首字延迟 ttft）、TOOL（工具调用）、EMBEDDING（向量化调用）等类型。
Session：以相同 sessionId 聚合的多条 Trace，对应聊天场景中的一个会话。

一次对话请求在 Langfuse 控制台中的实际树形结构：

Text

Session (sessionId = 会话 ID)
└── Trace: blog-agent-turn
    ├── SPAN: blog-agent-turn                        手动创建的单轮根 observation
    ├── AGENT: invoke_agent <chat-model>             streamText（gen_ai.agent.name = blog-agent-turn）
    │   ├── SPAN: step 1
    │   │   ├── GENERATION: chat <chat-model>
    │   │   └── TOOL: searchPosts
    │   │       ├── EMBEDDING: embeddings <emb-model>  查询向量化（gen_ai.agent.name = embed-query）
    │   │       └── SPAN: semantic-rerank              手动埋点
    │   └── SPAN: step 2
    │       └── GENERATION: chat <chat-model>
    ├── AGENT: invoke_agent <aux-model>              首轮回复后的标题生成（gen_ai.agent.name = session-title）
    │   └── SPAN: step 1
    │       └── GENERATION: chat <aux-model>
    └── AGENT: invoke_agent <aux-model>              达到水位时的滚动压缩（gen_ai.agent.name = compaction-summary）

手动创建的 blog-agent-turn（SPAN）与 AI SDK 自动生成的 invoke_agent（AGENT）在 Trace 顶层并列：因为 startObservation 仅创建 Span 而不切换当前活动的 OTel 上下文，后续的 streamText 与异步标题生成均直接挂在 Trace 根下。

项目中的 Trace 划分

博客助手按入口划分为三类 Trace，前台对话与后台任务互不嵌套：

traceName	触发入口	主要 observation	附加属性
`blog-agent-turn`	对话路由 `streamAgentChat`	单轮根 Span、主模型调用、工具、向量化、重排、标题生成、滚动压缩	`sessionId`、`userId`
`post-index-sync`	文章写操作后的单篇索引同步	`embed-post-chunks`	`metadata.diaryId`
`index-reconciliation`	管理端手动对账 / 首次检索时的启动对账	`embed-post-chunks`	`userId`、`metadata.action`

核心模块实现

安装依赖与环境变量

安装 Langfuse 与 OpenTelemetry 依赖：

Bash

npm install @langfuse/otel @langfuse/tracing @langfuse/vercel-ai-sdk @opentelemetry/api @opentelemetry/sdk-node

@langfuse/otel：提供 LangfuseSpanProcessor，负责将 OTel Span 转换并批量发送到 Langfuse。
@langfuse/tracing：提供 startObservation、propagateAttributes 等手动埋点与属性透传 API。
@langfuse/vercel-ai-sdk：AI SDK 遥测集成，将 generateText、streamText、embed 等调用映射为 Langfuse observation。
@opentelemetry/sdk-node：Node.js 端 OTel SDK，管理 Span Processor 与上下文管理器。

Langfuse TypeScript SDK 要求 Node.js 22 及以上版本，在 package.json 中声明运行时约束：

JSON

{
  "engines": {
    "node": ">=22"
  }
}

配置环境变量（.env），密钥可在 Langfuse 项目设置的 API Keys 页面获取：

Bash

# 未配置密钥时不初始化遥测
LANGFUSE_PUBLIC_KEY=pk-lf-xxx
LANGFUSE_SECRET_KEY=sk-lf-xxx
LANGFUSE_BASE_URL=https://cloud.langfuse.com

相关文档：Langfuse TypeScript SDK、Vercel AI SDK 集成。

进程级初始化与关闭

遥测模块集中在 telemetry.ts，对外暴露初始化、关闭与埋点函数。初始化遵循两个约束：未配置密钥时直接返回，不启动 OTel SDK；重复调用保持幂等。

实现初始化与关闭逻辑：

TypeScript

import { LangfuseSpanProcessor } from '@langfuse/otel';
import { LangfuseVercelAiSdkIntegration } from '@langfuse/vercel-ai-sdk';
import { NodeSDK } from '@opentelemetry/sdk-node';
import { registerTelemetry } from 'ai';

let otelNodeSdk: NodeSDK | null = null;
let isInitialized = false;

/**
 * 判断是否具备 Langfuse 遥测运行条件
 */
export function isTelemetryConfigured(): boolean {
  const publicKey = process.env.LANGFUSE_PUBLIC_KEY?.trim();
  const secretKey = process.env.LANGFUSE_SECRET_KEY?.trim();
  return Boolean(publicKey && secretKey);
}

/**
 * 进程级初始化 Langfuse 遥测（仅在 Node 运行时且已配置密钥时启动）
 */
export async function initTelemetry(): Promise<boolean> {
  if (!isTelemetryConfigured()) {
    return false;
  }

  if (isInitialized) {
    return true;
  }

  const processor = new LangfuseSpanProcessor({
    publicKey: process.env.LANGFUSE_PUBLIC_KEY,
    secretKey: process.env.LANGFUSE_SECRET_KEY,
    baseUrl: process.env.LANGFUSE_BASE_URL || undefined,
    environment: process.env.NODE_ENV,
  });

  otelNodeSdk = new NodeSDK({
    spanProcessors: [processor],
  });

  otelNodeSdk.start();
  registerTelemetry(new LangfuseVercelAiSdkIntegration());
  isInitialized = true;

  const onShutdown = async () => {
    await shutdownTelemetry();
  };

  process.once('SIGTERM', onShutdown);
  process.once('SIGINT', onShutdown);

  return true;
}

/**
 * 关闭遥测 SDK 资源并刷写数据
 */
export async function shutdownTelemetry(): Promise<void> {
  if (otelNodeSdk) {
    try {
      await otelNodeSdk.shutdown();
    } catch {
      // 捕获并忽略关闭异常
    } finally {
      otelNodeSdk = null;
      isInitialized = false;
    }
  }
}

environment：写入每条 Trace 的环境标签，便于在控制台区分 development 与 production 数据。
registerTelemetry：AI SDK v7 的全局遥测注册入口，注册后所有声明了 telemetry 配置的调用均会产出 observation。
NodeSDK.shutdown()：在进程收到 SIGTERM / SIGINT 时刷写 Span Processor 中尚未导出的批次数据。

通过 Next.js instrumentation 挂载

Next.js 在服务端启动时调用根目录 instrumentation.ts 中导出的 register 函数（参考 Next.js Instrumentation）。OTel Node SDK 依赖 Node.js API，需要按 NEXT_RUNTIME 区分运行时并动态导入，防止被打包进 Edge 运行时。

编辑 src/instrumentation.ts：

TypeScript

/**
 * Next.js 服务端启动钩子（仅在 Node.js 运行时执行）
 */
export async function register(): Promise<void> {
  if (process.env.NEXT_RUNTIME === 'nodejs') {
    const { initTelemetry } = await import('@/services/blog-agent/telemetry');
    await initTelemetry();
  }
}

为 AI SDK 调用声明 functionId

AI SDK 内置遥测支持（参考 AI SDK Telemetry）。在 generateText、streamText、embed、embedMany 中传入 telemetry 配置后，集成层会按 OpenTelemetry gen_ai 语义规范生成 invoke_agent <model>、step N、chat <model>、embeddings <model> 等节点，并将 functionId 写入节点的 attributes.gen_ai.agent.name 元数据属性，用于区分同一模型在不同业务场景下的调用。

为主对话模型调用声明 functionId：

TypeScript

const result = streamText({
  model,
  // ...
  stopWhen: [isStepCount(MAX_AGENT_STEPS), hasAdminApprovalRequestEmitted()],
  maxOutputTokens: CHAT_MAX_OUTPUT_TOKENS,
  abortSignal,
  telemetry: {
    functionId: 'blog-agent-turn',
  },
});

为辅助模型调用（会话标题、滚动压缩摘要）声明 functionId：

TypeScript

const { text } = await generateText({
  model: getAuxiliaryModel(),
  prompt,
  maxOutputTokens: SUMMARY_MAX_OUTPUT_TOKENS,
  telemetry: {
    functionId: 'compaction-summary',
  },
});

向量化调用的输入输出体积较大，通过 recordInputs 与 recordOutputs 控制是否记录原文：

TypeScript

// 文章分块批量向量化：输入为文章全文分块，输出为高维向量，均不记录
const { embeddings } = await embedMany({
  model: getEmbeddingModel(),
  values,
  telemetry: {
    functionId: 'embed-post-chunks',
    recordInputs: false,
    recordOutputs: false,
  },
});

// 查询向量化：保留查询文本便于排查检索问题，不记录输出向量
const { embedding } = await embed({
  model: getEmbeddingModel(),
  value,
  telemetry: {
    functionId: 'embed-query',
    recordOutputs: false,
  },
});

项目中声明的 functionId 列表：

functionId	调用类型	所在模块	输入输出记录策略
`blog-agent-turn`	`streamText`	单轮编排	全量记录
`session-title`	`generateText`	会话管理	全量记录
`compaction-summary`	`generateText`	滚动压缩	全量记录
`embed-query`	`embed`	检索	仅记录输入
`embed-post-chunks`	`embedMany`	索引同步	均不记录

透传 Trace 属性

propagateAttributes 基于 OTel 异步上下文，将 traceName、sessionId、userId、metadata 附加到回调函数内创建的全部 observation 上，调用链中的子函数无需感知 Trace 对象。

封装带降级的属性透传函数：

TypeScript

import { propagateAttributes, type PropagateAttributesParams } from '@langfuse/tracing';

/**
 * 在 trace 上下文中透传 traceName、sessionId、userId 及元数据
 */
export async function withTraceContext<T>(
  params: PropagateAttributesParams,
  fn: () => T | Promise<T>,
): Promise<T> {
  if (!isTelemetryConfigured()) {
    return fn();
  }

  return propagateAttributes(params, fn);
}

在管理端 API 路由中为索引对账任务附加操作者与动作类型：

TypeScript

const user = await requireAdmin();
const { action } = parsed.data;

const result = await withTraceContext(
  {
    traceName: 'index-reconciliation',
    userId: String(user.id),
    metadata: { action },
  },
  () => (action === 'rebuild' ? rebuildPostIndex({ force: true }) : reconcilePostIndex()),
);

单轮根 Observation

AI SDK 自动生成的 observation 只覆盖模型调用本身。为在 Trace 顶层记录整轮的最终输出（经过链接净化后的助手文本）、结束原因与中断状态，手动创建一个 blog-agent-turn 根 Span，并返回幂等的结束句柄。

实现 startTurnObservation：

TypeScript

import { startObservation, type LangfuseSpan } from '@langfuse/tracing';

export interface TurnObservationEndParams {
  output?: unknown;
  finishReason?: string;
  isAborted?: boolean;
  metadata?: Record<string, unknown>;
}

export interface TurnObservationHandle {
  end(params?: TurnObservationEndParams): void;
}

/**
 * 开启单轮助手的根 observation（blog-agent-turn）
 */
export function startTurnObservation(params: { input: string }): TurnObservationHandle {
  if (!isTelemetryConfigured()) {
    return { end: () => {} };
  }

  let observation: LangfuseSpan | null = null;
  try {
    observation = startObservation(
      'blog-agent-turn',
      { input: params.input },
      { asType: 'span' },
    );
  } catch {
    observation = null;
  }

  let hasEnded = false;

  return {
    end: (endParams?: TurnObservationEndParams) => {
      if (hasEnded) return;
      hasEnded = true;

      if (!observation) return;

      try {
        observation.update({
          output: endParams?.output,
          metadata: {
            finishReason: endParams?.finishReason ?? 'unknown',
            isAborted: Boolean(endParams?.isAborted),
            ...endParams?.metadata,
          },
        });
        observation.end();
      } catch {
        // 忽略遥测异常，保障主业务流程
      }
    },
  };
}

空句柄降级：未配置密钥时返回空实现，调用方无需判断遥测状态。
幂等结束：流式响应存在正常结束、客户端中断、异常捕获等多条收尾路径，hasEnded 标记防止重复调用 end()。
异常隔离：startObservation 与 update 均包裹在 try/catch 中，遥测故障不影响对话主流程。

非 AI SDK 调用的手动埋点

语义重排使用 DashScope 的 HTTP 接口直接调用，不经过 AI SDK，需要手动创建 observation。为控制上报体积，输入仅记录查询词与候选数量，不记录候选文档全文。

实现 traceRerank 包装函数：

TypeScript

export interface TraceRerankParams {
  query: string;
  candidateCount: number;
  model: string;
  threshold?: number;
}

export interface RerankExecutionResult<R> {
  results: R;
  usage?: Record<string, unknown>;
}

/**
 * 手动追踪语义重排过程（输入仅记录查询词与候选数量，不记录候选文档全文）
 */
export async function traceRerank<R>(
  params: TraceRerankParams,
  execute: () => Promise<RerankExecutionResult<R>>,
): Promise<RerankExecutionResult<R>> {
  if (!isTelemetryConfigured()) {
    return execute();
  }

  let observation: LangfuseSpan | null = null;
  try {
    observation = startObservation(
      'semantic-rerank',
      {
        input: {
          query: params.query,
          candidateCount: params.candidateCount,
        },
      },
      { asType: 'span' },
    );
  } catch {
    observation = null;
  }

  try {
    const output = await execute();
    if (observation) {
      observation.update({
        output: {
          resultCount: Array.isArray(output.results) ? output.results.length : undefined,
          results: output.results,
        },
        metadata: {
          model: params.model,
          threshold: params.threshold,
          usage: output.usage,
        },
      });
      observation.end();
    }
    return output;
  } catch (error) {
    if (observation) {
      observation.update({
        level: 'ERROR',
        statusMessage: error instanceof Error ? error.message : String(error),
      });
      observation.end();
    }
    throw error;
  }
}

在重排器中使用（execute 返回结果与接口用量，供 observation 记录）：

TypeScript

const execution = await traceRerank(
  {
    query: trimmedQuery,
    candidateCount: documents.length,
    model,
    threshold: options?.minScore,
  },
  async () => {
    const response = await fetch(getDashScopeRerankUrl(), {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        Authorization: `Bearer ${apiKey}`,
      },
      body: JSON.stringify(requestBody),
      signal: AbortSignal.timeout(timeoutMs),
    });

    if (!response.ok) {
      const errorText = await response.text();
      throw new Error(`DashScope rerank request failed with status ${response.status}: ${errorText}`);
    }

    const data = (await response.json()) as any;
    // parseRerankResults 为示意函数：过滤越界索引并映射为 { index, relevanceScore }
    return {
      results: parseRerankResults(data, documents.length),
      usage: data?.usage,
    };
  },
);

return execution.results;

level: 'ERROR'：异常分支将 observation 标记为错误级别，可在控制台按级别筛选失败的重排调用。
statusMessage：记录异常信息，异常仍向上抛出，由调用方执行降级（回退至 RRF 融合排序）。

完整流程演练与场景验证

对话轮次的端到端链路

对话服务 streamAgentChat 在最外层使用 withTraceContext 包裹整轮逻辑，内部依次创建根 observation、执行模型调用、在流结束回调中结束 observation：

TypeScript

export async function streamAgentChat(
  input: BlogAgentChatRequest,
  requestSignal?: AbortSignal,
): Promise<Response> {
  const user = await requireUser();
  const { sessionId, trigger, message } = input;
  const userPromptText = extractTextMessage(
    (message?.parts || []) as Array<Record<string, unknown>>,
  );

  return withTraceContext(
    {
      traceName: 'blog-agent-turn',
      sessionId,
      userId: String(user.id),
    },
    async () => {
      const turnObservation = startTurnObservation({ input: userPromptText });

      // 并发流锁、审批守卫、会话事务、历史读取等前置步骤省略

      const uiStream = createUIMessageStream({
        execute: async ({ writer }) => {
          const streamResult = await runAgentTurn({
            /* ... */
          });
          // 合并模型流到 writer
        },
        onFinish: async ({ responseMessage, finishReason }) => {
          try {
            // 净化未落地链接、保存助手消息、调度滚动压缩
            await saveAssistantMessageOnEnd({ sessionId, assistantSeq, responseMessage });

            turnObservation.end({
              output: extractTextMessage(finalParts),
              finishReason: finishReason ?? 'unknown',
              isAborted: isStreamInterrupted,
              metadata: {
                durationMs: Date.now() - startTime,
                stepCount: steps.length,
              },
            });
          } catch (error) {
            turnObservation.end({
              finishReason: 'error',
              isAborted: true,
            });
            logAgentError({ phase: 'chat_stream', userId: user.id, sessionId, error });
          } finally {
            safeRelease();
          }
        },
      });

      // 返回流式响应
    },
  );
}

执行流说明：

属性透传：withTraceContext 建立的上下文贯穿整轮请求，sessionId 使同一会话的多轮对话在 Langfuse Sessions 视图中按时间排列。
自动关联：runAgentTurn 内的 streamText、工具中的 embed-query 与 semantic-rerank 均在该上下文中执行，自动归入同一 Trace。
后台任务继承上下文：标题生成与滚动压缩通过 Promise.resolve().then(task) 调度，异步上下文随 Promise 链传播，因此 session-title 与 compaction-summary 也归入当前轮次的 Trace，便于核对单轮全部模型开销。
结束时机：根 observation 在 onFinish 中结束，此时已完成链接净化与落库，记录的 output 与用户最终看到的内容一致。

后台任务的上下文隔离

部分后台任务由前台请求触发，但不属于该轮对话。例如首次检索时触发的启动对账、文章保存后的单篇索引同步。如果直接在当前上下文中调度，其 embed-post-chunks 调用会被归入触发它的对话 Trace，干扰单轮耗时与成本统计。

使用 ROOT_CONTEXT 切断与父上下文的关联：

TypeScript

import { context, ROOT_CONTEXT } from '@opentelemetry/api';

/**
 * 在独立的根上下文中执行任务（用于解耦索引对账等后台任务与前台轮次 Trace）
 */
export async function withIsolatedRootContext<T>(fn: () => T | Promise<T>): Promise<T> {
  return context.with(ROOT_CONTEXT, fn);
}

在单篇索引同步调度中组合使用，先隔离上下文，再开启新的命名 Trace：

TypeScript

export function schedulePostIndexSync(diaryId: number): void {
  runAgentBackgroundTask('index_sync', { diaryId }, async () => {
    await withIsolatedRootContext(() =>
      withTraceContext(
        {
          traceName: 'post-index-sync',
          metadata: { diaryId },
        },
        () => syncPostIndex(diaryId),
      ),
    );
  });
}

启动对账采用相同的组合方式：

TypeScript

void withIsolatedRootContext(() =>
  withTraceContext({ traceName: 'index-reconciliation' }, () => reconcilePostIndex()),
).catch(() => {});

上下文归属规则对比：

后台任务	是否隔离上下文	归属 Trace	原因
会话标题生成	否	`blog-agent-turn`	属于首轮对话的衍生开销
滚动压缩摘要	否	`blog-agent-turn`	由当前轮次的消息量触发
单篇索引同步	是	`post-index-sync`	由文章写操作触发，与对话无关
启动对账	是	`index-reconciliation`	进程级一次性任务，耗时与文章总量相关

遥测降级与单元测试

遥测模块的所有导出函数在未配置密钥时均退化为直接执行回调，测试环境默认不初始化 OTel SDK。

验证降级行为与幂等性：

TypeScript

import { describe, it, expect, beforeEach, afterEach } from 'vitest';
import {
  isTelemetryConfigured,
  initTelemetry,
  shutdownTelemetry,
  withTraceContext,
  startTurnObservation,
  traceRerank,
} from '../telemetry';

describe('blog-agent telemetry', () => {
  const originalEnv = process.env;

  beforeEach(() => {
    process.env = { ...originalEnv };
    delete process.env.LANGFUSE_PUBLIC_KEY;
    delete process.env.LANGFUSE_SECRET_KEY;
  });

  afterEach(async () => {
    await shutdownTelemetry();
    process.env = originalEnv;
  });

  it('在仅配置 PUBLIC_KEY 时返回 false', () => {
    process.env.LANGFUSE_PUBLIC_KEY = 'pk-lf-test';
    expect(isTelemetryConfigured()).toBe(false);
  });

  it('在配置密钥时成功初始化并具有幂等性', async () => {
    process.env.LANGFUSE_PUBLIC_KEY = 'pk-lf-test';
    process.env.LANGFUSE_SECRET_KEY = 'sk-lf-test';
    expect(await initTelemetry()).toBe(true);
    expect(await initTelemetry()).toBe(true);
  });

  it('多次调用 end() 具有幂等性', () => {
    process.env.LANGFUSE_PUBLIC_KEY = 'pk-lf-test';
    process.env.LANGFUSE_SECRET_KEY = 'sk-lf-test';
    const handle = startTurnObservation({ input: '测试输入' });
    expect(() => {
      handle.end({ output: '首次结束', finishReason: 'stop' });
      handle.end({ output: '二次结束', finishReason: 'stop' });
    }).not.toThrow();
  });

  it('未启用遥测时直接透传执行闭包', async () => {
    const result = await withTraceContext(
      { traceName: 'test-trace', sessionId: 'sess-1' },
      async () => 'ok',
    );
    expect(result).toBe('ok');
  });

  it('在未启用遥测时直接执行 rerank 闭包并返回结果', async () => {
    const mockResult = [{ index: 0, relevanceScore: 0.85 }];
    const result = await traceRerank(
      { query: 'Next.js 缓存', candidateCount: 10, model: 'qwen3-rerank', threshold: 0.5 },
      async () => ({ results: mockResult, usage: { total_tokens: 120 } }),
    );
    expect(result.results).toEqual(mockResult);
  });
});

对话服务的测试通过 vi.mock('../telemetry') 替换埋点函数，并使用 MockLanguageModelV3 构造模型流，断言 startTurnObservation 以用户输入调用、end() 在流结束后携带最终输出与 finishReason 调用：

TypeScript

const telemetryMock = vi.hoisted(() => {
  const observationEnd = vi.fn();
  const startTurnObservation = vi.fn(() => ({ end: observationEnd }));
  const withTraceContext = vi.fn(async (_params: any, fn: () => any) => fn());
  return { observationEnd, startTurnObservation, withTraceContext };
});

vi.mock('../telemetry', () => ({
  startTurnObservation: telemetryMock.startTurnObservation,
  withTraceContext: telemetryMock.withTraceContext,
}));

部署与生产注意事项

运行时与环境配置

Node.js 版本：Langfuse TypeScript SDK 依赖 Node.js 22 及以上版本，部署镜像与 CI 需同步升级。
运行时限定：OTel Node SDK 仅在 NEXT_RUNTIME === 'nodejs' 时初始化，使用该能力的 API 路由需声明 export const runtime = 'nodejs'。
按需启用：以是否同时配置 LANGFUSE_PUBLIC_KEY 与 LANGFUSE_SECRET_KEY 作为开关，本地开发与测试环境可不配置。
环境区分：environment 取自 NODE_ENV，在控制台通过环境筛选隔离开发与生产数据。

数据上报时机

LangfuseSpanProcessor 以批量方式异步导出 Span，数据上报时机与部署形态相关：

常驻进程（自托管 Node 服务、容器）：批次按时间间隔与数量阈值自动导出，进程退出时由 SIGTERM 处理函数调用 shutdown() 刷写剩余数据。
Serverless 环境：函数实例在响应返回后可能被冻结，需在请求结束前显式刷写：导出 LangfuseSpanProcessor 实例并调用 forceFlush()，在 Next.js 中可结合 after() 在响应发送后执行，参考 Langfuse Serverless 说明。

数据体积与敏感信息

向量数据：embed / embedMany 的输出为高维浮点数组，设置 recordOutputs: false 避免占用存储；文章分块全文同时关闭 recordInputs。
手动埋点裁剪：重排 observation 仅记录查询词、候选数量与重排结果索引，不记录候选文档原文。
用户标识：userId 使用数据库内部 ID 的字符串形式，不上报邮箱或用户名等个人信息。
自托管选项：对话内容涉及隐私数据时，可使用 Docker 自托管 Langfuse，数据不出内网。

排查与数据回流

按会话回溯：在 Sessions 视图中按 sessionId 查看多轮对话的完整上下文与每轮工具调用序列。
Token、缓存与延迟分析：按节点类型或 attributes.gen_ai.agent.name（即 functionId）筛选调用，结合 input_tokens、cache_read.input_tokens（提示词缓存命中量）与 ttft（首字延迟）定位开销集中的环节。对于 DashScope 等未内置单价的模型，控制台默认显示 no cost，可在 Langfuse 项目设置的 Models 页面按模型名配置输入输出单价以启用成本换算。
异常定位：通过根 observation 的 metadata.isAborted 与 finishReason 筛选中断或异常结束的轮次，结合 level: 'ERROR' 的重排记录排查检索降级。
评估用例沉淀：将线上表现不佳的 Trace 输入与输出整理为评估数据集，加入离线 Evals 回归测试，形成“生产观测、用例补充、离线评估”的迭代闭环。

LLM 应用可观测性：基于 Langfuse 与 OpenTelemetry 的 AI SDK 链路追踪实践

方案架构与核心组件

生产环境可观测性的作用

Trace 与 Observation 数据模型

项目中的 Trace 划分

核心模块实现

安装依赖与环境变量

进程级初始化与关闭

通过 Next.js instrumentation 挂载

为 AI SDK 调用声明 functionId

透传 Trace 属性

单轮根 Observation

非 AI SDK 调用的手动埋点

完整流程演练与场景验证

对话轮次的端到端链路

后台任务的上下文隔离

遥测降级与单元测试

部署与生产注意事项

运行时与环境配置

数据上报时机

数据体积与敏感信息

排查与数据回流

本文收录于专栏

AI 应用开发笔记