集成 Langfuse 遥测与离线评测

.env.example

@@ -28,3 +28,14 @@ EMBEDDINGS_MODEL=text-embedding-3-small
 # 未配置 SEARCH_API_KEY 时深度研究开关会提示不可用，普通对话不受影响。
 SEARCH_API_KEY=
 SEARCH_BASE_URL=https://api.tavily.com
+
+# === Langfuse（可选，遥测 + 评测） ===
+# 未配置时遥测整体停用（零开销），聊天与点赞/点踩 UI 不受影响（反馈被静默丢弃）。
+# 密钥来自 Langfuse 项目设置（cloud.langfuse.com 免费注册，或 docker compose 自托管）。
+# BASE_URL：EU 区 https://cloud.langfuse.com、US 区 https://us.cloud.langfuse.com、
+# 自托管填实例地址。
+LANGFUSE_PUBLIC_KEY=
+LANGFUSE_SECRET_KEY=
+LANGFUSE_BASE_URL=https://cloud.langfuse.com
+# pnpm eval 用 Langfuse dataset 替代内置样例集时指定 dataset 名（可选）
+# LANGFUSE_EVAL_DATASET=

README.md

@@ -69,3 +69,11 @@ RAG 依赖 Postgres 的 pgvector 扩展（迁移会自动 `CREATE EXTENSION vect
 - 配置 `SEARCH_API_KEY`（默认 [Tavily](https://tavily.com)，`SEARCH_BASE_URL` 可换兼容服务）即可启用；未配置时开关会提示不可用，普通对话不受影响。
 - 编排走 AI SDK v7 多步工具循环（`streamText` + `webSearch`/`readUrl` 工具 + `stopWhen` 步数上限），与现有 chat 链路同源。
 - 设计与调研结论见 `docs/deep-research/设计说明.md`。
+
+## 遥测与评测（Langfuse）
+
+AI SDK v7 原生遥测（OpenTelemetry）+ [Langfuse](https://langfuse.com)：每轮对话一条 trace（含每步 LLM 调用、工具执行、token 用量与耗时），线程聚合为 session；assistant 消息下的 👍/👎 作为 score 回写到对应 trace。
+
+- 配置 `LANGFUSE_PUBLIC_KEY` / `LANGFUSE_SECRET_KEY` / `LANGFUSE_BASE_URL`（云端免费额度或 docker compose 自托管均可）即可启用；未配置时遥测整体停用、零开销，点赞/点踩被静默丢弃。
+- `pnpm eval` 跑离线评测（规则断言 + LLM-as-a-judge），结果作为 experiment run 上报，可在 Langfuse UI 跨 run 对比；设 `LANGFUSE_EVAL_DATASET` 可改用远端 dataset。
+- 调研对比（AI SDK 遥测现状、Langfuse vs LangSmith）与设计取舍见 `docs/observability/`。

app/api/chat/route.ts

@@ -5,14 +5,34 @@ import {
   tool,
   type UIMessage,
 } from "ai"
+import { after } from "next/server"
 import { frontendTools } from "@assistant-ui/react-ai-sdk"
 import type { ToolJSONSchema } from "assistant-stream"
+import {
+  getActiveTraceId,
+  propagateAttributes,
+  startActiveObservation,
+  type LangfuseSpan,
+} from "@langfuse/tracing"
 import { z } from "zod"
 import { resolveAttachmentParts } from "@/lib/chat/resolve-attachments"
 import { minimaxChatModel } from "@/lib/ai/minimax"
 import { researchTools } from "@/lib/chat/research-tools"
 import { isSearchConfigured } from "@/lib/ai/search"
-import { RESEARCH_MAX_STEPS, RESEARCH_SYSTEM_PROMPT } from "@/constants/research"
+import {
+  flushLangfuseSpans,
+  isLangfuseConfigured,
+  isValidTraceId,
+} from "@/lib/observability/langfuse"
+import {
+  CHAT_TRACE_NAME,
+  TELEMETRY_FUNCTION_IDS,
+  TRACE_TAGS,
+} from "@/constants/observability"
+import {
+  RESEARCH_MAX_STEPS,
+  RESEARCH_SYSTEM_PROMPT,
+} from "@/constants/research"
 
 // 深度研究可能多步循环，耗时较长，放宽单次请求时长上限
 export const maxDuration = 120
@@ -24,7 +44,13 @@ const getWeather = tool({
   }),
   execute: async ({ location }) => {
     // Deterministic mock reading (hashed from the city name) - no real weather API/key involved.
-    const conditions = ["Sunny", "Partly Cloudy", "Cloudy", "Light Rain", "Clear"]
+    const conditions = [
+      "Sunny",
+      "Partly Cloudy",
+      "Cloudy",
+      "Light Rain",
+      "Clear",
+    ]
     const seed = [...location].reduce((acc, c) => acc + c.charCodeAt(0), 0)
     return {
       location,
@@ -42,27 +68,47 @@ const compareTable = tool({
   inputSchema: z.object({
     title: z.string(),
     unit: z.string().optional(),
-    columns: z.array(z.string()).describe("Category labels, e.g. country names"),
+    columns: z
+      .array(z.string())
+      .describe("Category labels, e.g. country names"),
     series: z.array(
       z.object({
         name: z.string(),
-        values: z.array(z.number()).describe("One value per column, same order as columns"),
-      }),
+        values: z
+          .array(z.number())
+          .describe("One value per column, same order as columns"),
+      })
     ),
   }),
   execute: async (input) => input,
 })
 
-export async function POST(req: Request) {
-  const {
-    messages,
-    tools,
-    deepResearch,
-  }: {
-    messages: UIMessage[]
-    tools?: Record<string, ToolJSONSchema>
-    deepResearch?: boolean
-  } = await req.json()
+type ChatRequestBody = {
+  messages: UIMessage[]
+  tools?: Record<string, ToolJSONSchema>
+  deepResearch?: boolean
+  /** useChat 的 chat id == assistant-ui threadListItem.id == threads.id */
+  id?: string
+}
+
+/** trace 根观测的 input 记录最后一条用户消息的纯文本（完整 prompt 在 generation 观测里已有） */
+function lastUserText(messages: UIMessage[]): string {
+  const lastUser = messages.findLast((m) => m.role === "user")
+  if (!lastUser) return ""
+  return lastUser.parts
+    .filter(
+      (part): part is Extract<typeof part, { type: "text" }> =>
+        part.type === "text"
+    )
+    .map((part) => part.text)
+    .join("\n")
+}
+
+async function runChat(
+  body: ChatRequestBody,
+  turn?: LangfuseSpan
+): Promise<Response> {
+  const { messages, tools, deepResearch } = body
 
   // 研究模式：加入联网检索/深读工具、放宽步数、注入研究系统提示
   const research = deepResearch === true
@@ -84,14 +130,72 @@ export async function POST(req: Request) {
       : "用户开启了深度研究，但服务端未配置搜索服务（SEARCH_API_KEY），请如实告知该功能暂不可用，并基于已有知识尽力回答。"
     : undefined
 
+  turn?.update({ input: lastUserText(messages) })
+
   const result = streamText({
     model: minimaxChatModel(),
     system,
-    messages: await convertToModelMessages(resolvedMessages, { tools: allTools }),
+    messages: await convertToModelMessages(resolvedMessages, {
+      tools: allTools,
+    }),
     tools: allTools,
     // 研究模式允许更多工具轮次；普通对话维持原来的小步数
     stopWhen: isStepCount(research && searchReady ? RESEARCH_MAX_STEPS : 5),
+    telemetry: { functionId: TELEMETRY_FUNCTION_IDS.chat },
+    // handler 返回后流仍在继续，根观测在流真正结束/出错/中止时才收尾
+    ...(turn && {
+      onEnd: ({ text }) => {
+        turn.update({ output: text }).end()
+      },
+      onError: ({ error }) => {
+        turn
+          .update({
+            level: "ERROR",
+            statusMessage:
+              error instanceof Error ? error.message : String(error),
+          })
+          .end()
+      },
+      onAbort: () => {
+        turn.update({ statusMessage: "aborted by client" }).end()
+      },
+    }),
+  })
+
+  // serverless 下函数在响应后可能立刻冻结，响应结束后冲刷 span 批次
+  if (turn) after(() => flushLangfuseSpans())
+
+  // 服务端把 traceId 下发为 assistant 消息 id：前端点赞/点踩时直接以消息 id 回写 score，
+  // 无需另建 message↔trace 映射。未启用遥测（或拿到无效 traceId）时交回 AI SDK 默认生成。
+  const traceId = getActiveTraceId()
+  return result.toUIMessageStreamResponse({
+    ...(traceId &&
+      isValidTraceId(traceId) && { generateMessageId: () => traceId }),
   })
+}
+
+export async function POST(req: Request) {
+  const body: ChatRequestBody = await req.json()
+
+  if (!isLangfuseConfigured()) return runChat(body)
 
-  return result.toUIMessageStreamResponse()
+  return startActiveObservation(
+    CHAT_TRACE_NAME,
+    (turn) =>
+      propagateAttributes(
+        {
+          traceName: CHAT_TRACE_NAME,
+          // threadId 作为 sessionId，同一线程的多轮对话在 Langfuse 里聚成一个 session
+          sessionId: body.id,
+          tags: [
+            body.deepResearch === true
+              ? TRACE_TAGS.deepResearch
+              : TRACE_TAGS.chat,
+          ],
+        },
+        () => runChat(body, turn)
+      ),
+    // 流式响应在 handler 返回后才结束，span 由 onEnd/onError/onAbort 收尾
+    { endOnExit: false }
+  )
 }

app/api/feedback/route.ts

@@ -0,0 +1,43 @@
+import { z } from "zod"
+import {
+  getLangfuseClient,
+  isLangfuseConfigured,
+  isValidTraceId,
+} from "@/lib/observability/langfuse"
+import { USER_FEEDBACK_SCORE_NAME } from "@/constants/observability"
+
+// 用户对 assistant 消息的点赞/点踩，作为 score 回写到 Langfuse 对应 trace。
+// assistant 消息 id 由 chat route 下发，值即该轮对话的 traceId（见 chat/route.ts）。
+
+const bodySchema = z.object({
+  messageId: z.string(),
+  type: z.enum(["positive", "negative"]),
+  comment: z.string().max(500).optional(),
+})
+
+export async function POST(req: Request) {
+  // 未启用遥测：静默接受，反馈不落任何地方（前端无需感知配置状态）
+  if (!isLangfuseConfigured()) return new Response(null, { status: 204 })
+
+  const parsed = bodySchema.safeParse(await req.json().catch(() => null))
+  if (!parsed.success) return new Response("Bad request", { status: 400 })
+  const { messageId, type, comment } = parsed.data
+
+  // 只有 traceId 格式的消息 id 才可回写；历史消息或遥测未启用期间生成的消息直接忽略
+  if (!isValidTraceId(messageId)) return new Response(null, { status: 204 })
+
+  const langfuse = getLangfuseClient()
+  langfuse.score.create({
+    // 幂等 id：同一条消息改票时覆盖同一个 score，不产生重复计数
+    id: `${USER_FEEDBACK_SCORE_NAME}-${messageId}`,
+    traceId: messageId,
+    name: USER_FEEDBACK_SCORE_NAME,
+    value: type === "positive" ? 1 : 0,
+    dataType: "BOOLEAN",
+    ...(comment && { comment }),
+  })
+  // route handler 生命周期短，立即冲刷而不是等批量间隔
+  await langfuse.score.flush()
+
+  return new Response(null, { status: 204 })
+}

app/page.tsx

@@ -1,23 +1,49 @@
 "use client"
 
 import { useMemo } from "react"
-import { AssistantRuntimeProvider, useRemoteThreadListRuntime } from "@assistant-ui/react"
-import { AssistantChatTransport, useChatRuntime } from "@assistant-ui/react-ai-sdk"
+import {
+  AssistantRuntimeProvider,
+  useRemoteThreadListRuntime,
+  type FeedbackAdapter,
+} from "@assistant-ui/react"
+import {
+  AssistantChatTransport,
+  useChatRuntime,
+} from "@assistant-ui/react-ai-sdk"
 import { Base } from "@/components/examples/base"
 import { AssistantTools } from "@/components/assistant-ui/tools"
 import { postgresThreadListAdapter } from "@/lib/chat/thread-list-adapter"
 import { usePostgresThreadHistoryAdapter } from "@/lib/chat/use-thread-history-adapter"
 import { r2AttachmentAdapter } from "@/lib/chat/attachment-adapter"
 import { useResearchMode } from "@/lib/chat/research-mode"
 
+// 点赞/点踩 → /api/feedback → Langfuse score。assistant 消息 id 即该轮 traceId
+//（chat route 下发），服务端未启用遥测时该请求会被静默吞掉。fire-and-forget，
+// UI 的已提交态由 assistant-ui 本地维护，不依赖请求结果。
+const feedbackAdapter: FeedbackAdapter = {
+  submit({ message, type }) {
+    void fetch("/api/feedback", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ messageId: message.id, type }),
+    }).catch(() => {})
+  },
+}
+
 function useMyChatRuntime() {
   const history = usePostgresThreadHistoryAdapter()
   // 把「深度研究」开关状态随每条消息发给 chat route。用 getState() 而非闭包快照，
   // 保证读到发送时的最新开关值。
   const transport = useMemo(
     () =>
       new AssistantChatTransport({
-        prepareSendMessagesRequest: ({ id, messages, trigger, messageId, body }) => ({
+        prepareSendMessagesRequest: ({
+          id,
+          messages,
+          trigger,
+          messageId,
+          body,
+        }) => ({
           body: {
             ...body,
             id,
@@ -28,9 +54,16 @@ function useMyChatRuntime() {
           },
         }),
       }),
-    [],
+    []
   )
-  return useChatRuntime({ transport, adapters: { history, attachments: r2AttachmentAdapter } })
+  return useChatRuntime({
+    transport,
+    adapters: {
+      history,
+      attachments: r2AttachmentAdapter,
+      feedback: feedbackAdapter,
+    },
+  })
 }
 
 export default function Page() {