001
import { BaseCallbackHandler } from '@langchain/core/callbacks/base'
002
import type { Serialized } from '@langchain/core/load/serializable'
003

004
// 复用前面定义的 Span 结构
005
interface Span {
006
  spanId: string
007
  name: string
008
  startTime: number
009
  duration?: number
010
  status: 'ok' | 'error' | 'degraded'
011
  input: Record<string, any>
012
  output: Record<string, any>
013
}
014

015
class TraceCallbackHandler extends BaseCallbackHandler {
016
  name = 'TraceCallbackHandler'
017

018
  // 和前面的 TraceContext 一样，收集所有 Span
019
  private spans: Span[] = []
020
  // runId → Span 的映射，用于在 end 事件里找到对应的 start
021
  private activeSpans = new Map<string, Span>()
022

023
  // --- 模型事件 ---
024

025
  handleChatModelStart(
026
    llm: Serialized,
027
    messages: any[][],
028
    runId: string
029
  ) {
030
    // 模型开始推理 → 创建一个新 Span
031
    const span: Span = {
032
      spanId: runId,
033
      name: `llm:${llm.id?.at(-1) ?? 'unknown'}`,
034
      startTime: Date.now(),
035
      status: 'ok',
036
      input: { messages },
037
      output: {},
038
    }
039
    this.activeSpans.set(runId, span)
040
  }
041

042
  handleLLMEnd(output: any, runId: string) {
043
    // 模型推理完成 → 填充 duration 和 output，关闭 Span
044
    const span = this.activeSpans.get(runId)
045
    if (span) {
046
      span.duration = Date.now() - span.startTime
047
      span.output = {
048
        text: output.generations?.[0]?.[0]?.text?.slice(0, 200),
049
        tokenUsage: output.llmOutput?.tokenUsage,
050
      }
051
      this.spans.push(span)
052
      this.activeSpans.delete(runId)
053
    }
054
  }
055

056
  handleLLMError(error: Error, runId: string) {
057
    const span = this.activeSpans.get(runId)
058
    if (span) {
059
      span.duration = Date.now() - span.startTime
060
      span.status = 'error'
061
      span.output = { error: error.message }
062
      this.spans.push(span)
063
      this.activeSpans.delete(runId)
064
    }
065
  }
066

067
  // --- 工具事件 ---
068

069
  handleToolStart(tool: Serialized, input: string, runId: string) {
070
    const span: Span = {
071
      spanId: runId,
072
      name: `tool:${tool.id?.at(-1) ?? 'unknown'}`,
073
      startTime: Date.now(),
074
      status: 'ok',
075
      input: { toolInput: input },
076
      output: {},
077
    }
078
    this.activeSpans.set(runId, span)
079
  }
080

081
  handleToolEnd(output: string, runId: string) {
082
    const span = this.activeSpans.get(runId)
083
    if (span) {
084
      span.duration = Date.now() - span.startTime
085
      span.output = { result: output.slice(0, 200) }
086
      this.spans.push(span)
087
      this.activeSpans.delete(runId)
088
    }
089
  }
090

091
  handleToolError(error: Error, runId: string) {
092
    const span = this.activeSpans.get(runId)
093
    if (span) {
094
      span.duration = Date.now() - span.startTime
095
      span.status = 'error'
096
      span.output = { error: error.message }
097
      this.spans.push(span)
098
      this.activeSpans.delete(runId)
099
    }
100
  }
101

102
  // --- 检索器事件 ---
103

104
  handleRetrieverStart(
105
    retriever: Serialized,
106
    query: string,
107
    runId: string
108
  ) {
109
    const span: Span = {
110
      spanId: runId,
111
      name: `retriever:${retriever.id?.at(-1) ?? 'unknown'}`,
112
      startTime: Date.now(),
113
      status: 'ok',
114
      input: { query },
115
      output: {},
116
    }
117
    this.activeSpans.set(runId, span)
118
  }
119

120
  handleRetrieverEnd(documents: any[], runId: string) {
121
    const span = this.activeSpans.get(runId)
122
    if (span) {
123
      span.duration = Date.now() - span.startTime
124
      span.output = {
125
        documentCount: documents.length,
126
        documents: documents.map(d => d.pageContent?.slice(0, 100)),
127
      }
128
      // 检索到了但结果为空 → 标记为 degraded
129
      if (documents.length === 0) {
130
        span.status = 'degraded'
131
      }
132
      this.spans.push(span)
133
      this.activeSpans.delete(runId)
134
    }
135
  }
136

137
  handleRetrieverError(error: Error, runId: string) {
138
    const span = this.activeSpans.get(runId)
139
    if (span) {
140
      span.duration = Date.now() - span.startTime
141
      span.status = 'error'
142
      span.output = { error: error.message }
143
      this.spans.push(span)
144
      this.activeSpans.delete(runId)
145
    }
146
  }
147

148
  // --- 对外暴露的查询方法 ---
149

150
  getSpans(): Span[] {
151
    return this.spans
152
  }
153

154
  hasError(): boolean {
155
    return this.spans.some(s => s.status === 'error')
156
  }
157

158
  hasDegraded(): boolean {
159
    return this.spans.some(s => s.status === 'degraded')
160
  }
161

162
  toJSON() {
163
    return {
164
      spanCount: this.spans.length,
165
      spans: this.spans,
166
    }
167
  }
168
}

1
import { ChatOpenAI } from '@langchain/openai'
2

3
const handler = new TraceCallbackHandler()
4

5
// handler 会收到这个 model 实例以及它内部所有子组件的事件
6
const model = new ChatOpenAI({
7
  model: 'gpt-4.1-mini',
8
  callbacks: [handler],
9
})

01
const handler = new TraceCallbackHandler()
02

03
const result = await agent.invoke(
04
  {
05
    messages: [{ role: 'user', content: '明天上海天气怎么样？' }],
06
  },
07
  {
08
    callbacks: [handler],
09
  },
10
)
11

12
// 这次调用结束后，handler 里就有了完整的 Span 列表
13
console.log(handler.toJSON())

01
import { createAgent, tool } from 'langchain'
02
import * as z from 'zod'
03

04
const getWeather = tool(
05
  async ({ city }) => `${city}：明天小雨，17-22 度`,
06
  {
07
    name: 'get_weather',
08
    description: '查询某个城市未来的天气情况',
09
    schema: z.object({
10
      city: z.string().describe('要查询天气的城市名'),
11
    }),
12
  },
13
)
14

15
const createReminder = tool(
16
  async ({ content, time }) => `提醒已创建：${time} - ${content}`,
17
  {
18
    name: 'create_reminder',
19
    description: '帮用户创建一个提醒事项',
20
    schema: z.object({
21
      content: z.string().describe('提醒内容'),
22
      time: z.string().describe('提醒时间'),
23
    }),
24
  },
25
)
26

27
const sendMessage = tool(
28
  async ({ text }) => `已发送给用户：${text}`,
29
  {
30
    name: 'send_message',
31
    description: '把一段结果发送给用户',
32
    schema: z.object({
33
      text: z.string().describe('要发送的内容'),
34
    }),
35
  },
36
)
37

38
const agent = createAgent({
39
  model: 'openai:gpt-4.1-mini',
40
  tools: [getWeather, createReminder, sendMessage],
41
  systemPrompt: '你是用户的生活助理，会在需要时连续调用工具。',
42
})
43

44
// 每次请求创建一个新的 handler 实例（等价于每次请求创建一个新的 TraceContext）
45
const handler = new TraceCallbackHandler()
46

47
const result = await agent.invoke(
48
  {
49
    messages: [
50
      {
51
        role: 'user',
52
        content: '帮我看看明天上海天气。如果下雨就提醒我带伞，再把结果顺手发给我。',
53
      },
54
    ],
55
  },
56
  {
57
    callbacks: [handler],
58
  },
59
)
60

61
// 调用结束后，handler 里已经自动收集了所有 Span
62
const trace = handler.toJSON()
63
console.log(`共 ${trace.spanCount} 个 Span：`)
64
trace.spans.forEach(span => {
65
  console.log(
66
    `  [${span.status}] ${span.name} - ${span.duration}ms`
67
  )
68
})

01
const result = await agent.invoke(
02
  {
03
    messages: [{ role: 'user', content: '明天上海天气怎么样？' }],
04
  },
05
  {
06
    tags: ['single-agent', 'weather-flow', 'staging'],
07
    metadata: {
08
      user_id: 'user_2048',
09
      session_id: 'session_20260331_01',
10
    },
11
  },
12
)

01
import { traceable } from 'langsmith/traceable'
02

03
// 非 LangChain 组件，需要手动包装才能纳入 Trace 链路
04
const customBusinessLogic = traceable(
05
  async (input: string) => {
06
    // 你自己的业务逻辑...
07
    return result
08
  },
09
  { name: 'custom_business_logic', run_type: 'chain' }
10
)

01
import { CallbackHandler } from '@langfuse/langchain'
02

03
const langfuseHandler = new CallbackHandler({
04
  sessionId: 'session_20260331_01',
05
  userId: 'user_2048',
06
  tags: ['single-agent', 'weather-flow'],
07
})
08

09
const result = await agent.invoke(
10
  {
11
    messages: [{ role: 'user', content: '明天上海天气怎么样？' }],
12
  },
13
  {
14
    callbacks: [langfuseHandler],
15
  },
16
)
17

18
// 确保 trace 数据发送完毕（LangFuse 默认异步批量发送）
19
await langfuseHandler.flushAsync()

1
// 前面的用法：直接用 Langfuse 原生 SDK，手动创建 trace 和 span
2
const langfuse = new Langfuse({ publicKey: '...', secretKey: '...' })
3
const trace = langfuse.trace({ name: 'ai-companion-chat' })
4
const span = trace.span({ name: 'memory_retrieval', input: { query } })
5
// ... 手动结束 span ...
6
span.end({ output: { memories } })

1
// LangChain 里的用法：传入 CallbackHandler，框架自动调用
2
const handler = new CallbackHandler({ userId: '...' })
3
await agent.invoke(input, { callbacks: [handler] })

01
import { CallbackHandler as LangFuseHandler } from '@langfuse/langchain'
02

03
const traceHandler = new TraceCallbackHandler()  // 自己的追踪器
04
const langfuseHandler = new LangFuseHandler()    // 发到 LangFuse
05

06
const result = await agent.invoke(
07
  { messages: [{ role: 'user', content: '明天上海天气怎么样？' }] },
08
  {
09
    callbacks: [
10
      traceHandler,      // 本地记录，用于采样决策
11
      langfuseHandler,   // 发到 LangFuse 平台
12
      // LangSmith 不用手动加，环境变量开了就自动生效
13
    ],
14
  },
15
)
16

17
// 用本地 handler 做采样决策
18
if (traceHandler.hasError() || traceHandler.hasDegraded()) {
19
  // 完整写入 D1...
20
}

01
const handler = new TraceCallbackHandler()
02

03
const result = await agent.invoke(
04
  { messages: [{ role: 'user', content: userMessage }] },
05
  { callbacks: [handler] },
06
)
07

08
// 调用结束后，handler 里已经有了所有 Span
09
// 用和前面一样的采样逻辑做决策
10
const totalDuration = handler.getSpans().reduce(
11
  (sum, s) => sum + (s.duration ?? 0), 0
12
)
13

14
if (handler.hasError()) {
15
  // 有错误 → 完整写入 D1
16
  await storeFullTrace(env, handler.toJSON(), userId, sessionId)
17
} else if (handler.hasDegraded()) {
18
  // 有降级 → 完整写入 D1
19
  await storeFullTrace(env, handler.toJSON(), userId, sessionId)
20
} else if (totalDuration > 5000) {
21
  // 异常慢 → 完整写入 D1
22
  await storeFullTrace(env, handler.toJSON(), userId, sessionId)
23
} else if (Math.random() < 0.05) {
24
  // 正常请求 5% 采样
25
  await storeFullTrace(env, handler.toJSON(), userId, sessionId)
26
} else {
27
  // 其余只存摘要
28
  await storeSummary(env, handler.toJSON())
29
}

01
handleRetrieverEnd(documents: any[], runId: string) {
02
  const span = this.activeSpans.get(runId)
03
  if (span) {
04
    span.duration = Date.now() - span.startTime
05
    span.output = { documentCount: documents.length }
06

07
    // 检索结果为空 → 自动标记为 degraded
08
    if (documents.length === 0) {
09
      span.status = 'degraded'
10
    }
11

12
    this.spans.push(span)
13
    this.activeSpans.delete(runId)
14
  }
15
}