socweb/doc/plan/cot-trace-implementation-plan.md

COT 可视化（结构化执行轨迹）生产落地计划

背景与方案评估

关于"Gemini CoT"的定位澄清

Gemini 的原生 CoT 是指模型在输出最终答案前的推理 token（类似 Claude Extended Thinking）。Azure OpenAI / GPT-4o 不暴露模型级别的推理 scratchpad，因此本方案实现的是 Agent Activity Trace（结构化执行轨迹），本质是：

拦截 LangGraph ReAct 图的运行事件 → 结构化摘要 → SSE 推送 → 前端时间线渲染

这是比暴露原始 CoT 更合理的选择，也是 Gemini/ChatGPT Pro 实际采用的方式。

现状与差距

层	现状	生产缺口
后端 SSE	只有 token/tool_start/tool_end/done，无摘要字段	需补全 6 类事件 + 摘要 + 耗时 + 错误检测
前端 SSE 消费	tool_start/tool_end 完全忽略	新增全部事件处理分支
前端数据模型	Message 无 traceItems 字段	扩展接口
前端 UI	无 Trace 组件	新建 TracePanel，复用已有 Collapsible/Spinner
GeminiMessage	不接收 model prop	需透传 selectedModel
数据持久化	Message 表无 metadata 字段	Trace 为会话内存态，不持久化（历史消息无 trace，合理）

---

生产级事件协议

后端完整事件集（6 类）

// 1. 状态事件 — 阶段感知
{"type": "status", "stage": "分析问题", "message": "正在理解您的问题..."}

// 2. 工具调用开始
{
  "type": "tool_start",
  "tool": "kb_search",
  "title": "检索知识库",
  "input_summary": "查询：产品规划路线图",
  "ts": 1712620800000
}

// 3. 工具调用成功结束
{
  "type": "tool_end",
  "tool": "kb_search",
  "title": "检索知识库",
  "output_summary": "命中 3 条知识库记录",
  "status": "success",
  "duration_ms": 842,
  "ts": 1712620800842
}

// 4. 工具调用失败
{
  "type": "tool_error",
  "tool": "web_search",
  "title": "外部搜索",
  "error_summary": "请求超时，已跳过",
  "duration_ms": 8000,
  "ts": 1712620808000
}

// 5. token（现有，不变）
{"type": "token", "content": "根据知识库..."}

// 6. done（现有，不变）
{"type": "done"}

---

后端实现（backend/app/api/chat.py）

全部改动

新增导入：

import time

在 stream_response 生成器函数中：

async def generate():
    full_content: list[str] = []
    tool_start_ts: dict[str, int] = {}   # 记录各工具的起始时间戳
    has_tool_activity = False              # 是否有过工具调用
    final_status_emitted = False          # "整理答案"状态是否已发出

    try:
        # ① 在 graph 开始前发出初始状态
        yield _sse({"type": "status", "stage": "分析问题", "message": "正在理解您的问题..."})

        async for event in graph.astream_events(input_data, config=config, version="v2"):
            kind = event.get("event", "")

            # ② token 事件：在首个 token 前，若有工具调用则发"整理答案"状态
            if kind == "on_chat_model_stream":
                chunk = event.get("data", {}).get("chunk")
                if chunk and hasattr(chunk, "content") and chunk.content:
                    if isinstance(chunk.content, str):
                        # 若工具调用已完成，在首 token 前插入"整理答案"状态
                        if has_tool_activity and not final_status_emitted:
                            yield _sse({"type": "status", "stage": "整理答案",
                                        "message": "正在结合检索结果生成回复..."})
                            final_status_emitted = True
                        full_content.append(chunk.content)
                        yield _sse({"type": "token", "content": chunk.content})

            # ③ 工具开始
            elif kind == "on_tool_start":
                tool_name = event.get("name", "unknown")
                tool_input = event.get("data", {}).get("input", {})
                ts = int(time.time() * 1000)
                tool_start_ts[tool_name] = ts
                has_tool_activity = True
                yield _sse({
                    "type": "tool_start",
                    "tool": tool_name,
                    "title": TOOL_TITLES.get(tool_name, tool_name),
                    "input_summary": _summarize_input(tool_name, tool_input),
                    "ts": ts,
                })

            # ④ 工具结束（含错误检测）
            elif kind == "on_tool_end":
                tool_name = event.get("name", "unknown")
                output = event.get("data", {}).get("output", "")
                output_str = output if isinstance(output, str) else str(output)
                ts = int(time.time() * 1000)
                duration_ms = ts - tool_start_ts.pop(tool_name, ts)
                is_error = _is_tool_error(output_str)
                if is_error:
                    yield _sse({
                        "type": "tool_error",
                        "tool": tool_name,
                        "title": TOOL_TITLES.get(tool_name, tool_name),
                        "error_summary": _extract_error_summary(output_str),
                        "duration_ms": duration_ms,
                        "ts": ts,
                    })
                else:
                    yield _sse({
                        "type": "tool_end",
                        "tool": tool_name,
                        "title": TOOL_TITLES.get(tool_name, tool_name),
                        "output_summary": _summarize_output(tool_name, output_str),
                        "status": "success",
                        "duration_ms": duration_ms,
                        "ts": ts,
                    })

    except Exception as exc:
        # 现有错误处理逻辑保持不变
        ...
    finally:
        ai_content = "".join(full_content)
        if ai_content:
            await _persist_ai_message(request.conversation_id, ai_content)
        yield _sse({"type": "done"})

新增辅助函数（同文件尾部）：

def _sse(data: dict) -> bytes:
    return f"data: {json.dumps(data, ensure_ascii=False)}\n\n".encode("utf-8")


TOOL_TITLES: dict[str, str] = {
    "kb_search":          "检索知识库",
    "ticket_list":        "查询工单列表",
    "ticket_detail":      "查询工单详情",
    "web_search":         "外部搜索",
    "generate_document":  "生成文档",
    "sandbox_run":        "执行沙盒代码",
}

# 工具输出中的错误关键词（工具均返回字符串而非 raise）
_ERROR_KEYWORDS = (
    "出错", "失败", "超时", "error", "failed", "timeout", "not available",
    "no download link", "检索出错", "检索超时", "execution failed",
)

def _is_tool_error(output: str) -> bool:
    lo = output.lower()
    return any(kw in lo for kw in _ERROR_KEYWORDS)

def _extract_error_summary(output: str) -> str:
    # 取首行，截断到 60 字符
    first_line = output.split("\n")[0].strip()
    return first_line[:60] if first_line else "工具调用失败"

def _summarize_input(tool_name: str, inp: dict | str) -> str:
    if isinstance(inp, str):
        return inp[:60]
    match tool_name:
        case "kb_search":
            return f"查询：{str(inp.get('query', ''))[:50]}"
        case "ticket_list":
            return f"第 {inp.get('page', 1)} 页，每页 {inp.get('page_size', 20)} 条"
        case "ticket_detail":
            return f"工单 ID：{inp.get('ticket_id', '')}"
        case "web_search":
            return f"搜索：{str(inp.get('query', ''))[:50]}"
        case "generate_document":
            return str(inp.get('prompt', ''))[:60]
        case "sandbox_run":
            lang = inp.get('language', 'python')
            lines = len(str(inp.get('code', '')).splitlines())
            return f"{lang} 代码（{lines} 行）"
        case _:
            return str(inp)[:60]

def _summarize_output(tool_name: str, output: str) -> str:
    if not output or output.strip() == "(no output)":
        return "无结果"
    match tool_name:
        case "kb_search":
            count = output.count("---") + 1 if "---" in output else 1
            return f"命中 {count} 条知识库记录"
        case "ticket_list":
            import re
            m = re.search(r"Found (\d+) tickets", output)
            return f"返回 {m.group(1)} 条工单" if m else "工单列表已获取"
        case "ticket_detail":
            return "工单详情已获取"
        case "web_search":
            count = output.count("##")
            return f"找到 {max(count, 1)} 条搜索结果"
        case "generate_document":
            if "Download:" in output:
                doc_type = "文档"
                if "[PPT]" in output:
                    doc_type = "PPT"
                elif "[Excel]" in output or "[Table]" in output:
                    doc_type = "表格"
                elif "[Word]" in output:
                    doc_type = "Word 文档"
                return f"{doc_type}已生成，可下载"
            return "文档生成完成"
        case "sandbox_run":
            lines = len(output.splitlines())
            exit_match = output.startswith("[Exit code:")
            suffix = "（含错误）" if exit_match else ""
            return f"执行完成，输出 {lines} 行{suffix}"
        case _:
            return output[:60]

---

前端实现

文件 1：lib/api.ts — 类型扩展

// 扩展 ChatStreamEvent（完整字段）
export interface ChatStreamEvent {
  type: "token" | "status" | "tool_start" | "tool_end" | "tool_error" | "done" | "error";
  // token
  content?: string;
  // status
  stage?: string;
  message?: string;
  // tool_start / tool_end / tool_error
  tool?: string;
  title?: string;
  input_summary?: string;
  output_summary?: string;
  error_summary?: string;
  status?: "success" | "error";
  duration_ms?: number;
  ts?: number;
}

// 前端 Trace 条目（统一结构）
export interface TraceItem {
  id: string;                                              // 唯一 id
  type: "status" | "tool_start" | "tool_end" | "tool_error";
  tool?: string;                                           // 工具名（tool_* 类型）
  title: string;                                           // 展示标题
  message?: string;                                        // status 的描述文本
  inputSummary?: string;
  outputSummary?: string;
  errorSummary?: string;
  itemStatus: "running" | "success" | "error" | "info";  // UI 状态
  durationMs?: number;
  startTs: number;                                         // 毫秒时间戳
}

streamChat 函数签名不变，只需更新 ChatStreamEvent 类型定义即可。

---

文件 2：components/gemini/GeminiMessage.tsx — 接口扩展与 TracePanel 集成

扩展 Message 接口：

import type { TraceItem } from "@/lib/api";

export interface Message {
  id: string;
  role: "user" | "assistant";
  content: string;
  timestamp?: Date;
  attachments?: AttachmentData[];
  traceItems?: TraceItem[];         // 新增：执行轨迹（流式构建，不持久化）
}

扩展 GeminiMessageProps：

interface GeminiMessageProps {
  message: Message;
  model?: "flash" | "auto" | "pro";   // 新增：用于决定 TracePanel 展示层级
  onRegenerate?: (id: string) => void;
}

Assistant 消息 JSX — 在 content 上方插入 TracePanel：

// 在 assistant 分支内，<div className="space-y-0.5"> 之前：
{message.traceItems && message.traceItems.length > 0 && (
  <TracePanel
    items={message.traceItems}
    model={model ?? "auto"}
    className="mb-3"
  />
)}
<div className="space-y-0.5">{renderContent(message.content)}</div>

---

文件 3：components/gemini/GeminiChat.tsx — 事件处理与 model 透传

handleSend 的 onEvent 回调 — 完整替换：

(event) => {
  if (event.type === "token" && event.content) {
    // 现有 token 逻辑，不变
    setConversations((prev) => prev.map((c) => {
      if (c.id !== streamConvId) return c;
      const exists = c.messages.some((m) => m.id === aiMsgId);
      if (!exists) {
        return { ...c, messages: [...c.messages, {
          id: aiMsgId, role: "assistant" as const,
          content: event.content!, timestamp: new Date(),
          traceItems: [],  // 初始化 traceItems
        }]};
      }
      return { ...c, messages: c.messages.map((m) =>
        m.id === aiMsgId ? { ...m, content: m.content + event.content } : m
      )};
    }));

  } else if (event.type === "status") {
    // status 事件：追加 info 条目（分析问题 / 整理答案）
    const item: TraceItem = {
      id: `status-${event.ts ?? Date.now()}`,
      type: "status",
      title: event.stage ?? "处理中",
      message: event.message,
      itemStatus: "info",
      startTs: event.ts ?? Date.now(),
    };
    _appendTraceItem(streamConvId, aiMsgId, item, setConversations);

  } else if (event.type === "tool_start" && event.tool) {
    // 工具开始：状态为 running
    const item: TraceItem = {
      id: `${event.tool}-${event.ts ?? Date.now()}`,
      type: "tool_start",
      tool: event.tool,
      title: event.title ?? event.tool,
      inputSummary: event.input_summary,
      itemStatus: "running",
      startTs: event.ts ?? Date.now(),
    };
    _appendTraceItem(streamConvId, aiMsgId, item, setConversations);

  } else if (event.type === "tool_end" && event.tool) {
    // 工具结束：更新 running → success
    _updateTraceItem(streamConvId, aiMsgId, event.tool, {
      type: "tool_end",
      outputSummary: event.output_summary,
      itemStatus: "success",
      durationMs: event.duration_ms,
    }, setConversations);

  } else if (event.type === "tool_error" && event.tool) {
    // 工具错误：更新 running → error
    _updateTraceItem(streamConvId, aiMsgId, event.tool, {
      type: "tool_error",
      errorSummary: event.error_summary,
      itemStatus: "error",
      durationMs: event.duration_ms,
    }, setConversations);
  }
}

新增辅助函数（文件顶层，组件外部）：

function _appendTraceItem(
  convId: string, msgId: string, item: TraceItem,
  setConversations: React.Dispatch<React.SetStateAction<Conversation[]>>
) {
  setConversations((prev) => prev.map((c) => {
    if (c.id !== convId) return c;
    return { ...c, messages: c.messages.map((m) =>
      m.id === msgId ? { ...m, traceItems: [...(m.traceItems ?? []), item] } : m
    )};
  }));
}

function _updateTraceItem(
  convId: string, msgId: string, tool: string,
  updates: Partial<TraceItem>,
  setConversations: React.Dispatch<React.SetStateAction<Conversation[]>>
) {
  setConversations((prev) => prev.map((c) => {
    if (c.id !== convId) return c;
    return { ...c, messages: c.messages.map((m) => {
      if (m.id !== msgId) return m;
      // 找到最后一个同名 running 条目并更新
      const items = [...(m.traceItems ?? [])];
      for (let i = items.length - 1; i >= 0; i--) {
        if (items[i].tool === tool && items[i].itemStatus === "running") {
          items[i] = { ...items[i], ...updates };
          break;
        }
      }
      return { ...m, traceItems: items };
    })};
  }));
}

透传 selectedModel 给 GeminiMessage：

找到 GeminiMessage 的渲染位置，新增 model={selectedModel} prop。

---

文件 4：components/gemini/TracePanel.tsx（新建）

完整组件，使用已有的 Collapsible（components/ui/collapsible.tsx）和 Spinner（components/ui/spinner.tsx）。

"use client";

import { useState } from "react";
import { CheckCircle2, XCircle, ChevronDown, ChevronRight, Loader2, Zap, Brain } from "lucide-react";
import { cn } from "@/lib/utils";
import { Collapsible, CollapsibleContent, CollapsibleTrigger } from "@/components/ui/collapsible";
import type { TraceItem } from "@/lib/api";

interface TracePanelProps {
  items: TraceItem[];
  model: "flash" | "auto" | "pro";
  className?: string;
}

// ─── 工具图标映射 ─────────────────────────────────────────
const TOOL_ICONS: Record<string, string> = {
  kb_search: "🗂️",
  web_search: "🌐",
  ticket_list: "🎫",
  ticket_detail: "🎫",
  generate_document: "📄",
  sandbox_run: "⚙️",
};

// ─── 状态图标 ─────────────────────────────────────────────
function StatusIcon({ status }: { status: TraceItem["itemStatus"] }) {
  switch (status) {
    case "running":
      return <Loader2 size={13} className="animate-spin text-[var(--gem-accent)]" />;
    case "success":
      return <CheckCircle2 size={13} className="text-emerald-400" />;
    case "error":
      return <XCircle size={13} className="text-red-400" />;
    case "info":
      return <Brain size={13} className="text-[var(--gem-text-muted)]" />;
  }
}

// ─── 耗时格式化 ───────────────────────────────────────────
function formatDuration(ms?: number): string {
  if (!ms) return "";
  if (ms < 1000) return `${ms}ms`;
  return `${(ms / 1000).toFixed(1)}s`;
}

// ─── 单条 Trace 条目 ──────────────────────────────────────
function TraceItemRow({ item, expanded }: { item: TraceItem; expanded: boolean }) {
  const icon = item.tool ? TOOL_ICONS[item.tool] ?? "🔧" : null;
  const isRunning = item.itemStatus === "running";

  return (
    <div
      className={cn(
        "flex items-start gap-2 py-1.5 px-2 rounded-lg text-xs transition-colors",
        isRunning && "bg-[var(--gem-surface-2)]",
      )}
    >
      {/* 状态图标 */}
      <div className="mt-0.5 shrink-0">
        <StatusIcon status={item.itemStatus} />
      </div>

      {/* 内容 */}
      <div className="flex-1 min-w-0">
        <div className="flex items-center gap-1.5">
          {icon && <span className="text-xs">{icon}</span>}
          <span
            className={cn(
              "font-medium",
              item.itemStatus === "error"
                ? "text-red-400"
                : "text-[var(--gem-text)]",
            )}
          >
            {item.title}
          </span>
          {item.durationMs !== undefined && (
            <span className="text-[var(--gem-text-muted)] ml-auto shrink-0">
              {formatDuration(item.durationMs)}
            </span>
          )}
        </div>

        {/* 详情（Pro 模式或展开状态下显示） */}
        {expanded && (
          <div className="mt-0.5 space-y-0.5">
            {item.inputSummary && (
              <p className="text-[var(--gem-text-muted)] truncate">{item.inputSummary}</p>
            )}
            {item.outputSummary && (
              <p className="text-[var(--gem-text-secondary)] truncate">{item.outputSummary}</p>
            )}
            {item.errorSummary && (
              <p className="text-red-400 truncate">{item.errorSummary}</p>
            )}
            {item.message && item.type === "status" && (
              <p className="text-[var(--gem-text-muted)]">{item.message}</p>
            )}
          </div>
        )}
      </div>
    </div>
  );
}

// ─── 主组件 ───────────────────────────────────────────────
export function TracePanel({ items, model, className }: TracePanelProps) {
  const isPro = model === "pro";
  const [open, setOpen] = useState(isPro);  // Pro 默认展开

  // 生成单行摘要（Auto 模式折叠时显示）
  const summaryText = (() => {
    const running = items.filter((i) => i.itemStatus === "running");
    if (running.length > 0) return `正在 ${running[running.length - 1].title}...`;
    const tools = items.filter((i) => i.type === "tool_end");
    const errors = items.filter((i) => i.type === "tool_error");
    if (errors.length > 0) return `已完成（${errors.length} 个工具调用失败）`;
    if (tools.length > 0) {
      const names = tools.map((t) => t.title).join("、");
      return `已完成：${names}`;
    }
    return "正在分析...";
  })();

  const hasRunning = items.some((i) => i.itemStatus === "running");

  return (
    <Collapsible open={open} onOpenChange={setOpen} className={cn("w-full", className)}>
      {/* 触发行（始终可见）*/}
      <CollapsibleTrigger asChild>
        <button
          className={cn(
            "flex w-full items-center gap-2 rounded-lg px-2 py-1.5 text-xs",
            "text-[var(--gem-text-muted)] hover:text-[var(--gem-text)]",
            "hover:bg-[var(--gem-surface-2)] transition-colors",
          )}
        >
          <Zap
            size={12}
            className={cn(
              "shrink-0",
              hasRunning ? "text-[var(--gem-accent)] animate-pulse" : "text-[var(--gem-text-muted)]",
            )}
          />
          <span className="flex-1 text-left truncate">{summaryText}</span>
          {open
            ? <ChevronDown size={12} className="shrink-0" />
            : <ChevronRight size={12} className="shrink-0" />
          }
        </button>
      </CollapsibleTrigger>

      {/* 展开内容 */}
      <CollapsibleContent>
        <div
          className={cn(
            "mt-1 ml-2 border-l border-[var(--gem-border)] pl-3 space-y-0.5",
          )}
        >
          {items.map((item) => (
            <TraceItemRow
              key={item.id}
              item={item}
              expanded={isPro}  // Pro 模式显示摘要详情
            />
          ))}
        </div>
      </CollapsibleContent>
    </Collapsible>
  );
}

---

关键文件清单

文件	改动类型	核心内容
backend/app/api/chat.py	修改	6 类 SSE 事件、摘要函数、错误检测、耗时计算
frontend/lib/api.ts	修改	ChatStreamEvent 扩展、新增 TraceItem 类型
frontend/components/gemini/GeminiMessage.tsx	修改	Message 加 traceItems、GeminiMessageProps 加 model、集成 TracePanel
frontend/components/gemini/GeminiChat.tsx	修改	onEvent 补全所有事件分支、_appendTraceItem/_updateTraceItem 辅助函数、透传 model
frontend/components/gemini/TracePanel.tsx	新建	Auto/Pro 双模式时间线，使用已有 Collapsible + Spinner

---

约束说明

• 前端文件为 read-only（CLAUDE.md 限制），需用户显式授权后执行
• TracePanel 仅使用现有 CSS 变量（--gem-*）和已有 UI 组件，不引入新依赖
• Trace 数据为会话内存态，历史消息加载时 traceItems 为空（符合预期）
• 工具错误通过输出字符串检测（因工具层均 return string 不 raise），关键词见 _ERROR_KEYWORDS

---

验证方式

1. 后端事件格式验证

   curl -N -X POST http://localhost:8000/api/chat/stream \
     -H "Content-Type: application/json" \
     -d '{"message":"帮我搜索产品规划","conversation_id":"test-1","tools":["knowledge"],"model":"flash"}'
   

   期望看到：status → tool_start（含 input_summary）→ tool_end（含 output_summary + duration_ms）→ token... → done

2. 工具错误验证：断开 KB Agent，发送知识库查询，期望看到 tool_error 事件（含 error_summary）

3. Auto 模式：选 Auto + 知识库工具，助手消息上方出现单行折叠状态栏，点击展开显示时间线

4. Pro 模式：切换 Pro，状态栏默认展开，每条工具调用显示标题 + 摘要 + 耗时

5. 无工具时：不选任何工具，TracePanel 不出现（traceItems 为空）

6. 多工具顺序调用：同时开启 knowledge + tickets，验证时间线条目顺序正确，各自 duration 准确