手搓 Agent 透明大脑：流式推理过程可视化与毫秒级 Token 粒度统计

一、为什么市面上的 Agent 都不愿“透明”？

1.1 一个被全行业刻意忽视的体验黑洞

打开任何一个 AI Agent 产品——无论是国外还是国内主流大模型应用——你会发现一个惊人的共同点：你永远看不到 AI 是怎么想的。

用户输入一个问题，等了几秒，Agent 返回了一串“调用工具”、“搜索中”、“正在思考”的提示，然后给出了答案。中间发生了什么？模型在决定调用工具之前做了什么判断？为什么选了工具 A 而不是工具 B？每一步消耗了多少 Token？花了多少钱？

这些问题，用户永远不知道。不是因为技术做不到，而是因为没人愿意做。

我们来剖析一下背后的原因：

（1）产品理念冲突：魔法感 vs. 可信赖

大厂的产品经理们喜欢“魔法感”。用户说完话，AI 直接给答案，好像一切都理所当然。如果让用户看到 AI 其实是一步步推出来的——“我得先查一下当前时间”、“我需要搜索最新新闻”、“我还得验证一下这个地址是否安全”——这种“祛魅”会削弱产品的神秘感。

但对于一个个人 AI 管家来说，它需要的恰恰是可信赖，而不是魔法感。管家是可以被主人质疑的——“你刚才为什么要做这个操作？”——如果没有推理过程的记录，就无法追溯决策依据。

（2）商业利益冲突：成本透明化会劝退用户

Token 消耗明细这个东西，技术上就是解析流式响应中 usage 字段的几行代码。但为什么 ChatGPT 不告诉你每次对话花了多少钱？因为一旦用户知道了精确成本，就会开始计算 ROI，从而减少使用频率。对商业公司来说，模糊成本、制造黑盒，更符合商业利益。

（3）技术成本与商业回报不匹配

这两个功能确实有一定的工作量。深度思考折叠区涉及流式解析、前后端通信、动态 DOM 渲染、历史消息回显、交互优化（折叠/复制/滚动控制）等一系列问题。Token 明细虽然代码量不大，但需要理解 SSE 协议、设计估算算法、处理好前后端展示。做出来之后，公司怎么赚钱？不直接。所以大公司不会优先做这类“不赚钱但费工程师”的功能。

1.2 本文要做什么

本文以我个人开发的“Web3多链监控交易系统”中的Agent功能板块为蓝本，详细拆解两个核心功能的完整实现：

功能 核心价值 技术难点

深度思考折叠区 实时流式展示 AI 的完整推理过程 SSE 流解析、缓冲区设计、前后端通信、工具调用后的递归处理、历史消息回显

Token 消耗明细 细粒度展示系统提示词、对话历史、工具定义各部分的 Token 消耗 Token 估算算法、usage 字段解析、明细可视化

文章会贴出所有关键技术环节的代码片段（仅保留完整方法实现，但核心算法和数据结构全部公开），以及设计原理的深入分析。读完本文，你将完全理解如何为自己的 Agent 系统实现类似功能。

1.3 技术栈与架构 预览

本文涉及的系统采用以下技术栈：

后端：Python + PyWebView（桌面端框架）+ DeepSeek API

前端：原生 JavaScript（无框架），通过 window.evaluate_js 与 Python 通信

数据流：SSE（Server-Sent Events）流式响应

数据库：SQLite（存储聊天历史与推理内容）

整体数据流如下图所示：

用户消息 → agent_process_input → _build_system_prompt → _call_model_with_tools

        ↓

    _stream_deepseek_reasoner（核心函数）

        ↓

  ┌───────────┼───────────┐

  │      │      │

 解析 SSE  提取 usage 检测 tool_calls

  │      │      │

  ↓      ↓      ↓

 推理内容  Token 明细  执行工具 → 清洗消息 → 递归调用

  ↓      ↓

 _push_thinking 拼接明细

  ↓      ↓

  └───────────┼───────────┘

        ↓

  window.evaluate_js → 前端实时渲染

在后续章节中，我们将沿着这条数据流，逐一拆解每个环节的实现细节。

二、整体技术架构设计

在深入具体实现之前，必须先把整个系统的架构和数据流讲清楚。深度思考可视化不是单一模块的功能，而是跨越多层架构的全链路设计——从最底层的 HTTP 流式请求，到中间的 Python 业务逻辑，到最上层的前端 DOM 渲染，每一层都有需要解决的技术问题。

2.1 核心模块与职责划分

整个系统由四个核心模块构成，它们各自承担明确的职责：

模块 文件 核心职责

AI 调用层 ai_calls.py 发起流式请求、解析 SSE 响应、提取推理内容与 Token 信息、处理工具调用的递归逻辑

消息推送层 base.py 提供 _push_thinking 方法，将推理片段安全地推送到前端，处理 JSON 转义与异常兜底

前端渲染层 agent.js 接收推理片段，动态创建/更新折叠区 DOM，处理用户交互（折叠、复制、滚动控制）

系统提示词层 ai_calls.py 构建完整的系统提示词，注入主人信息、反思笔记、关键词记忆检索结果等

设计原则：

单向数据流：推理内容从后端 → 前端是单向推送，前端不需要轮询或请求。

职责分离：AI 调用层只负责“获取数据”，消息推送层只负责“传递数据”，前端渲染层只负责“展示数据”。

降级容错：每一层都有独立的错误处理，单层失败不影响其他层的正常运行。

2.2 全链路数据流详解

下图展示了从用户发送消息到前端展示推理过程的完整数据流：

┌─────────────────────────────────────────────────────────────┐

│            用户发送消息             │

└────────────────────────┬────────────────────────────────────┘

             ▼

┌─────────────────────────────────────────────────────────────┐

│ agent_process_input (api_agent.py)             │

│ ├── 提取平台、模型、用户输入等参数              │

│ ├── 生成 batch_id（用于关联工具调用日志）          │

│ ├── 调用 _save_chat_message 保存用户消息到数据库       │

│ ├── 调用 _build_system_prompt 构建系统提示词         │

│ │  ├── 注入主人信息（偏好链、兴趣领域等）          │

│ │  ├── 注入 Agent 自我认知（角色、性格等）         │

│ │  ├── 注入反思笔记（最近一次深度观察）           │

│ │  ├── 注入关键词匹配记忆检索结果              │

│ │  └── 拼接工具列表描述                  │

│ ├── 调用 _load_recent_history 加载最近对话历史        │

│ └── 调用 _call_model_with_tools(messages, "thinking")    │

└────────────────────────┬────────────────────────────────────┘

             ▼

┌─────────────────────────────────────────────────────────────┐

│ _call_model_with_tools (ai_calls.py)            │

│ ├── 根据 model_type 路由到对应方法             │

│ │  ├── "thinking" → _stream_deepseek_reasoner       │

│ │  ├── "deepseek" → _call_deepseek_with_tools       │

│ │  ├── "local"  → _call_local_model           │

│ │  └── 其他    → 检查自定义模型配置           │

│ └── 返回最终回复文本或 None                 │

└────────────────────────┬────────────────────────────────────┘

             ▼

┌─────────────────────────────────────────────────────────────┐

│ _stream_deepseek_reasoner (ai_calls.py)          │

│ ├── 清洗消息（_sanitize_messages）             │

│ ├── 构建流式请求 payload（含 reasoning_effort='max'）    │

│ ├── 发起 HTTP 流式 POST 请求                │

│ │                             │

│ ├── 【循环解析 SSE 流】                   │

│ │  ├── 提取 reasoning_content → 缓冲区累积         │

│ │  │  └── 达到阈值 → _push_thinking → 前端实时显示    │

│ │  ├── 提取 content → 累积最终回复文本           │

│ │  ├── 提取 tool_calls → 累积工具调用指令         │

│ │  └── 提取 usage → 保存 Token 消耗信息          │

│ │                             │

│ ├── 【流结束后处理】                    │

│ │  ├── 推送剩余缓冲区内容                 │

│ │  ├── 拼接 Token 消耗明细到推理文本            │

│ │  └── 保存完整推理文本到 self._last_reasoning       │

│ │                             │

│ └── 【分支判断】                      │

│   ├── 有 tool_calls → 执行工具 → 清洗消息 → 递归调用自身  │

│   └── 无 tool_calls → 返回累积的最终回复         │

└────────────────────────┬────────────────────────────────────┘

             ▼

┌─────────────────────────────────────────────────────────────┐

│ _push_thinking (base.py)                  │

│ ├── 将推理片段包装为 JSON 安全字符串             │

│ ├── 调用 window.evaluate_js 执行前端接收函数         │

│ └── 异常兜底：推送失败不影响主流程              │

└────────────────────────┬────────────────────────────────────┘

             ▼

┌─────────────────────────────────────────────────────────────┐

│ window.receiveAgentThinking (agent.js)           │

│ ├── 【首次调用】动态创建折叠区 DOM 结构            │

│ │  ├── 创建 think-block 容器                │

│ │  ├── 创建折叠切换栏（标题 + 复制按钮 + 展开/折叠箭头）   │

│ │  └── 创建 think-steps 内容容器              │

│ ├── 【每次调用】追加推理步骤到 think-steps          │

│ ├── 自动滚动到底部（悬停时暂停）               │

│ └── 折叠切换 / 复制 / 悬停暂停等交互处理           │

└─────────────────────────────────────────────────────────────┘

2.3 关键设计决策分析

在整个架构中，有几项关键设计决策值得展开讨论：

决策一：为什么在 Python 层解析 SSE 流，而非让前端直接连接 API？

这个问题在系统设计初期就遇到了。如果前端直接连接 DeepSeek API，可以省去中间的推送层，架构更简单。但这样做会引入几个问题：

API Key 暴露风险：Key 必须存在前端，任何能看到网页源代码的人都能获取。

工具调用无法执行：深度思考模型经常需要调用工具（搜索、读文件等），这些工具只能在 Python 后端执行。如果前端直连 API，工具调用链路会被打断。

消息清洗逻辑无处安放：工具调用后的消息清洗、上下文管理必须在服务端统一处理。

因此，集中式架构 是必然选择——Python 后端作为唯一入口，统一处理鉴权、工具执行、消息清洗，前端只负责渲染。

决策二：缓冲区阈值的确定

在流式推送推理内容时，不可能每收到一个 token 就推送一次——那会产生数百次 DOM 操作。但推送间隔太长，用户又会感觉卡顿。

我选择的方案是多重条件触发：

80 字符阈值：经过实测，80 个中文字符在界面上约占 2-3 行，是一个自然的视觉块。

语义边界：遇到换行符 \n 或句号 。 立即推送，因为这些是自然阅读的停顿点。

兜底机制：流结束后，推送缓冲区剩余的所有内容。

这种设计在“实时性”和“渲染性能”之间取得了平衡。

决策三：工具调用后为什么需要递归而非降级？

深度思考模型在调用工具后，必须再次进入深度思考模式来生成最终回复。如果降级到普通 DeepSeek 模型，虽然能成功返回结果，但会失去两个关键能力：

推理连贯性：降级后的回复是“断档”的，与之前的推理过程完全脱节。

工具链支持：如果工具执行后发现需要进一步操作（如学习文档后又需要搜索相关内容），降级模式无法再次调用工具。

但递归调用面临一个核心问题：reasoning_content 污染。上一轮推理的内容如果继续携带，会导致消息体急剧膨胀，最终超过 API 的上下文限制。因此必须在每次递归前，遍历 messages 列表，将所有消息中的 reasoning_content 字段删除。这个清洗逻辑是保证递归能持续进行的关键。

2.4 前后端通信协议

后端向前端推送数据，通过 window.evaluate_js 执行 JS 代码。这不是通常意义上的“协议”，但因为涉及 JSON 序列化和字符转义，必须谨慎处理：

推理内容：通过 _push_thinking 推送，前端用 window.receiveAgentThinking 接收。

系统消息：通过 _push_message 推送，前端用 window.receiveAgentMessage 接收。

状态更新：通过 _push_status 推送，前端用 window.setAgentStatus 接收。

三者的底层实现原理一致，都是 window.evaluate_js 的封装。这里有一个容易踩的坑：JSON 字符串中的特殊字符——如换行、引号、反斜杠——必须先通过 json.dumps 序列化，否则在 JS 层面会被解析错误。

三、后端深度思考可视化的核心实现

3.1 流式请求的发起与参数配置

深度思考模式的关键在于向 DeepSeek API 发起一个携带特定参数的流式请求。与普通聊天请求不同，这里需要额外设置两个参数：

"stream": True — 启用 SSE（Server-Sent Events）流式响应

"reasoning_effort": "max" — 指定推理强度为最高级别      

# 流式请求的核心参数构建
payload = {
  'model': model,
  'messages': messages,
  'max_tokens': 8192,
  'stream': True,      # 启用 SSE 流式响应
  'tools': tools,      # 工具定义列表
  'reasoning_effort': 'max' # 深度思考强度
}
 
headers = {
  'Authorization': f"Bearer {api_key}",
  'Content-Type': 'application/json',
  'Accept': 'text/event-stream' # 告知服务器期望 SSE 格式
}
 
resp = requests.post(url, json=payload, headers=headers,
           timeout=config['timeout'] * 3, stream=True)

设计要点：

timeout 设置为常规的 3 倍，因为深度思考模式的响应时间可能长达数十秒甚至数分钟。

Accept: text/event-stream 并非所有 API 都要求，但显式声明可以避免某些代理服务器缓存响应。

stream=True 告诉 requests 库不要立即下载整个响应体，而是返回一个可迭代的流对象。

3.2 SSE 协议与数据帧解析

SSE（Server-Sent Events）是一种基于 HTTP 的单向流协议。服务端持续发送 data: 前缀的文本行，客户端逐行解析。DeepSeek API 的 SSE 响应格式如下：

data: {"choices":[{"delta":{"reasoning_content":"你好"},"index":0}]}
 
data: {"choices":[{"delta":{"reasoning_content":"，我"},"index":0}]}
 
data: {"choices":[{"delta":{"content":"你好！"},"index":0}]}
 
data: [DONE]

每一行 data: 后面是一个完整的 JSON 对象。[DONE] 是流结束的标志。

解析这种格式的代码并不复杂，但需要注意几个细节：

空行需要跳过（SSE 协议用空行分隔事件）。

data: 前缀的长度固定为 6 个字符（包括空格）。

某些代理或网络环境可能将单个 chunk 拆分成多行，需要按行迭代而非按 chunk 解析。

# SSE 响应解析的核心循环
for line in resp.iter_lines(decode_unicode=True):
    if not line:                        # 跳过空行
        continue
    if not line.startswith('data: '):   # 只处理 data 行
        continue
 
    data_str = line[6:]                 # 去掉 "data: " 前缀
    if data_str.strip() == '[DONE]':    # 流结束标志
        break
 
    chunk = json.loads(data_str)
    # 后续解析 delta...

3.3 缓冲区设计：在实时性与渲染性能间取得平衡

从 SSE 流中提取到的 reasoning_content 是逐 token 返回的——每个 chunk 可能只有 1-3 个字符。如果每收到一个 token 就推送到前端，会产生极其密集的 DOM 操作，导致前端渲染卡顿。

解决方法是引入一个字符缓冲区。推理内容先在缓冲区累积，只有满足特定条件时才推送到前端：

self._reasoning_buffer = ""
 
# 在流式解析循环中
reason_content = delta.get('reasoning_content', '')
if reason_content:
    self._reasoning_buffer += reason_content
 
    # 多重条件触发推送
    should_push = (
        len(self._reasoning_buffer) >= 80 or          # 长度阈值
        '\n' in self._reasoning_buffer or             # 换行符
        '。' in self._reasoning_buffer                # 句号（语义边界）
    )
    if should_push and self._reasoning_buffer.strip():
        self._push_thinking(self._reasoning_buffer)
        self._reasoning_buffer = ""
 
# 流结束后推送剩余内容
if self._reasoning_buffer.strip():
    self._push_thinking(self._reasoning_buffer)

为什么是 80 字符？

基于实际测试，80 个中文字符在界面上约占据 2-3 行的空间。这是一个自然的视觉块——用户能够在一次扫视中读完，同时也不会因为推送过于频繁而产生视觉疲劳。

为什么要检测换行和句号？

这是语义层面的优化。换行和句号是自然语言中的天然停顿点。在这些位置推送内容，不会打断用户的阅读节奏。想象一下，如果在一个词中间突然断掉，用户体验会很差。

为什么流结束后还要推送一次？

流结束时，缓冲区里可能还有不足 80 字符的剩余内容。如果不推送，这部分内容就会丢失，用户永远看不到推理的最后一段。

3.4 _push_thinking 的安全实现

_push_thinking 是将推理内容从 Python 后端传递到前端 JS 的桥梁。它的本质是调用 window.evaluate_js，但需要处理 JSON 序列化带来的安全问题。

def _push_thinking(self, step: str):
    if self.window:
        safe_step = json.dumps(step)   # 关键：安全序列化
        try:
            self.window.evaluate_js(
                f"if(window.receiveAgentThinking){{window.receiveAgentThinking({safe_step})}}"
            )
        except:
            pass  # 推送失败不影响主流程

为什么必须用 json.dumps 而不是直接拼接字符串？

推理内容中可能包含各种特殊字符——换行 \n、双引号 "、反斜杠 \、甚至 Unicode 转义序列。如果直接用 f-string 拼接到 JS 代码中，任何未转义的特殊字符都会导致 JS 语法错误，整个推送链路中断。

json.dumps 会自动处理所有这些转义，输出符合 JSON 规范的字符串。在 JS 侧，这个字符串会被自动解析为合法的 JavaScript 字符串字面量。

错误处理的考量：try-except 确保即使前端页面已关闭、或者 receiveAgentThinking 函数未定义，推送失败也不会影响后端的流式解析和最终回复生成。

3.5 流式解析的完整骨架

将上述各环节组装起来，得到流式解析的完整骨架代码：

# 深度思考流式解析的核心骨架（不含工具调用和错误处理细节）
reasoning_parts = []
final_reply = ""
self._reasoning_buffer = ""
 
for line in resp.iter_lines(decode_unicode=True):
    if not line or not line.startswith('data: '):
        continue
    data_str = line[6:]
    if data_str.strip() == '[DONE]':
        break
 
    chunk = json.loads(data_str)
 
    # 提取 usage（Token 消耗信息，详见第五章）
    if 'usage' in chunk and chunk['usage']:
        self._last_usage = chunk['usage']
 
    delta = chunk.get('choices', [{}])[0].get('delta', {})
 
    # 提取并缓冲推理内容
    reason = delta.get('reasoning_content', '')
    if reason and self.thinking_enabled:
        self._reasoning_buffer += reason
        if len(self._reasoning_buffer) >= 80 or '\n' in self._reasoning_buffer or '。' in self._reasoning_buffer:
            if self._reasoning_buffer.strip() and self._reasoning_buffer not in reasoning_parts:
                self._push_thinking(self._reasoning_buffer)
                reasoning_parts.append(self._reasoning_buffer)
            self._reasoning_buffer = ""
 
    # 提取正式回复内容
    content = delta.get('content', '')
    if content:
        final_reply += content
 
    # 提取工具调用（处理详见第四章）
    if 'tool_calls' in delta:
        for tc in delta['tool_calls']:
            tool_calls.append(tc)
 
# 推送剩余缓冲内容
if self._reasoning_buffer and self._reasoning_buffer.strip():
    self._push_thinking(self._reasoning_buffer)
    reasoning_parts.append(self._reasoning_buffer)
 
# 拼接完整推理文本（包含 Token 明细）
if reasoning_parts:
    reasoning_text = ''.join(reasoning_parts)
    # Token 明细拼接逻辑（详见第五章）
    self._last_reasoning = reasoning_text

几个值得关注的细节：

reasoning_parts 列表的去重：这里用 not in reasoning_parts 判断是否已经推送过相同内容。由于缓冲区可能因为多种条件同时触发推送，有概率出现重复。去重保证了前端不会显示重复的推理段落。

self.thinking_enabled 标志位：只在用户手动开启“显示深度思考”时才推送推理内容。这个标志位由前端开关控制，传递给后端的 agent_process_input。关闭时，推理内容仍然会被流式接收，只是不推送到前端，最终只显示正式回复。

self._last_reasoning 的保存：完整的推理文本被保存在实例变量中，供后续保存到数据库时使用（聊天记录的 reasoning_content 字段）。

四、工具调用后的递归处理与消息清洗

这是整个深度思考可视化系统中最棘手的技术挑战。如果处理不当，会导致两种后果：要么深度思考模式被迫降级为普通模式，要么因上下文过长直接被 API 拒绝。

4.1 问题的本质：工具调用改变了消息结构

在普通对话中，messages 列表的结构很简单——系统提示词、用户消息、AI 回复交替排列。但一旦模型决定调用工具，消息结构会发生根本性变化：

# 工具调用前的 messages
[
  {"role": "system", "content": "系统提示词..."},
  {"role": "user", "content": "帮我分析这个代币"},
  {"role": "assistant", "content": null, "tool_calls": [...], "reasoning_content": "用户想要..."}
]
 
# 工具执行后追加的消息
[
  ...上述消息...,
  {"role": "tool", "tool_call_id": "call_xxx", "name": "check_token_security", "content": "{...结果...}"}
]

此时，对话尚未结束。模型需要基于工具返回的结果，再次进行推理，生成最终回复。这意味着必须发起第二次 API 请求。

4.2 常见错误方案：降级到普通模型

最简单的做法是：工具调用后，改用普通 DeepSeek 模型（非深度思考）发起第二次请求。

# 错误做法：降级到普通模型
final_reply = self._call_deepseek_with_tools(messages)  # 丢失 reasoning_effort

这种做法能成功返回结果，但引入了两个严重问题：

问题一：推理链断裂。用户在折叠区看到的是第一轮推理（“我需要调用工具 X”），然后第二轮的推理完全消失。最终回复看起来像是“从天而降”，缺乏连贯性。

问题二：无法再次调用工具。普通 DeepSeek 模型也可能决定调用工具。如果第二次请求又触发了工具调用，而第三次请求还是普通模型……这样的循环中，深度思考能力被完全放弃。

4.3 正确方案：清洗消息后递归调用

正确做法是：工具调用后，继续以深度思考模式发起请求，但发起前必须清洗掉上一轮的 reasoning_content。

为什么需要清洗？因为第一轮推理的内容已经完成了它的使命——决定了要调用什么工具。如果继续携带这段推理内容，会导致：

上下文浪费：推理内容通常长达数百甚至数千字符，占据宝贵的上下文窗口。

逻辑混淆：模型可能被上一轮的推理内容干扰，产生“我在重复思考”的错觉。

API 限制：当 messages 中包含 reasoning_content 字段时，API 要求后续请求也必须携带 reasoning_effort 参数。清洗后重新发起，参数保持一致。

清洗逻辑的实现如下：

# 工具调用后的消息清洗与递归调用（核心片段）
if tool_calls:
    valid_calls = [tc for tc in tool_calls if tc.get('function', {}).get('name')]
 
    if valid_calls:
        # 保存本轮推理内容，供前端展示
        saved_reasoning = self._last_reasoning if self._last_reasoning else ""
 
        # 构建 assistant 消息（含工具调用指令）
        assistant_msg = {
            "role": "assistant",
            "tool_calls": valid_calls
        }
        if saved_reasoning:
            assistant_msg["reasoning_content"] = saved_reasoning
        messages.append(assistant_msg)
 
        # 执行每个工具，追加 tool 消息
        for tc in valid_calls:
            func_name = tc['function']['name']
            arguments = json.loads(tc['function'].get('arguments', '{}'))
            tool_result = self._execute_tool(func_name, arguments)
            if tool_result is None:
                tool_result = ""
            messages.append({
                "role": "tool",
                "tool_call_id": tc.get('id', ''),
                "name": func_name,
                "content": tool_result
            })
 
        # 关键：清洗所有消息中的 reasoning_content
        clean_messages = []
        for m in messages:
            clean_m = {k: v for k, v in m.items() if k != 'reasoning_content'}
            clean_messages.append(clean_m)
 
        # 递归调用，保持深度思考模式
        final_reply = self._call_model_with_tools(clean_messages, "thinking")
        return final_reply

4.4 清洗逻辑的深层考量

逐条消息遍历并重建，初看似乎低效，但实际上有几层深意：

第一层：避免浅拷贝陷阱。Python 的字典是可变对象。如果直接用 del msg['reasoning_content'] 修改原消息，会污染调用方的 messages 列表。函数外部的代码可能还需要用到完整的消息（比如保存到数据库）。因此必须创建新字典。

第二层：选择性保留。并不是所有字段都需要传递。reasoning_content 是最需要清除的，但如果有其他内部标记字段（如调试信息），也应该在此处过滤。

第三层：可扩展性。如果未来 DeepSeek API 增加了其他“一次性使用”的字段，可以集中在这个清洗循环中处理，不需要修改调用链的其他部分。

4.5 递归终止条件

递归调用不会无限进行。有两个自然终止条件：

模型不再返回 tool_calls：这说明模型认为已有足够信息生成最终回复，此时 _stream_deepseek_reasoner 返回累积的 final_reply，递归链结束。

递归深度达到上限：_call_deepseek_with_tools 中有 5 轮迭代的上限（for iteration in range(5)），_stream_deepseek_reasoner 每次递归都会经过这个限制，防止无限循环。

实际使用中，绝大多数场景在 1-2 轮递归后即可完成。一次性调用 11 个工具的极端情况，递归轮次等于工具调用批次数（通常 2-3 批），不会超过限制。

4.6 推理内容的连续性处理

递归调用中，每一轮都会产生新的推理内容（self._last_reasoning）。如何向用户展示完整的推理链？我采用的策略是拼接：

new_reasoning = self._last_reasoning if self._last_reasoning else ""
if saved_reasoning and new_reasoning:
    if new_reasoning.strip().startswith(saved_reasoning.strip()[:50]):
        # 新推理包含旧推理的延续，直接用新的
        self._last_reasoning = new_reasoning
    else:
        # 两段推理独立，拼接
        self._last_reasoning = saved_reasoning + "\n\n" + new_reasoning
elif saved_reasoning:
    self._last_reasoning = saved_reasoning

这样，无论递归了多少轮，用户在前端折叠区看到的都是一段连贯的推理文本，不会出现断层。

五、Token 消耗明细的统计与可视化

Token 消耗明细看似只是展示几个数字，但要真正做到细粒度可视化，需要解决三个问题：数据从哪来、如何拆分各部分消耗、如何实时展示。

5.1 数据来源：流式响应中的 usage 字段

在 SSE 流式响应中，usage 信息并非在每个 chunk 中都存在。实际测试中发现：

在流式传输过程中，大部分 chunk 不包含 usage 字段

通常在流的最后一个有效 chunk 中，会附带完整的 usage 信息

某些情况下，usage 可能单独作为一个 chunk 出现（内容仅含 usage，无 choices）

因此需要在解析循环中持续检测，一旦出现就保存下来：

# 在流式解析循环中检测并保存 usage
chunk = json.loads(data_str)
 
# usage 可能在任意一个 chunk 中出现
if 'usage' in chunk and chunk['usage']:
    self._last_usage = chunk['usage']
    # 典型结构：{"prompt_tokens": 2456, "completion_tokens": 512, "total_tokens": 2968}

_last_usage 被保存在实例变量中，供后续的明细计算使用。使用实例变量而非局部变量的原因是：流式解析和最终回复生成可能跨越多个函数调用，需要一个跨函数的状态存储。

5.2 为什么需要估算？

API 返回的 usage 只给出了三个总数：

字段 含义

prompt_tokens 输入的总 Token 数（系统提示词 + 对话历史 + 工具定义 + 用户消息）

completion_tokens 输出的总 Token 数（正式回复 + 推理内容）

total_tokens 两者之和

但作为开发者，我们想知道的是：系统提示词占了多少？对话历史占了多少？工具定义又占了多少？ 这些信息 API 不提供，必须自行估算。

5.3 Token 估算算法设计

估算的核心思想是：基于字符类型区分，中文和英文字符对应的 Token 数不同。

def _estimate_tokens(text: str) -> int:
    """基于中英文字符比例估算 Token 数"""
    import re
    # 匹配中文字符范围（包括中文标点）
    chinese = len(re.findall(r'[\u4e00-\u9fff\u3000-\u303f\uff00-\uffef]', text))
    others = len(text) - chinese
    # 中文约 0.6 token/字，英文约 0.25 token/字
    return int(chinese * 0.6 + others * 0.25)

参数来源说明：

0.6：基于中文常见 tokenizer（如 DeepSeek、GPT 系列）的统计。一个中文字符经过 BPE 分词后，平均对应 0.5-0.7 个 token。取 0.6 是实践中验证的稳定值。

0.25：英文单词经过分词后，平均每个字符对应 0.2-0.3 个 token。取 0.25 是经验中值。

为什么不直接用 API 返回的分词结果？

因为 DeepSeek 的 Embedding API 我们暂时用不了（详见之前文章），而聊天 API 并不返回分词细节。估算虽然存在 ±10% 的误差，但对于成本分析和优化提示词来说，已经完全够用。

5.4 各部分 Token 数的实际计算

有了估算函数后，可以分别计算系统提示词、对话历史、工具定义的 Token 数：

# 各部分 Token 估算 核心计算逻辑
system_prompt = messages[0].get("content", "") if messages else ""
history_msgs = messages[1:] if len(messages) > 1 else []
tools_json = json.dumps(tools, ensure_ascii=False) if tools else ""
 
# 对话历史序列化为 JSON 字符串来估算
history_text = json.dumps(history_msgs, ensure_ascii=False)
 
# 各部分估算
system_estimate = _estimate_tokens(system_prompt)
history_estimate = _estimate_tokens(history_text)
tools_estimate = _estimate_tokens(tools_json)
 
# 输入总数减去已估算的各部分，得到“其他”部分（思考指令、格式化开销等）
other_estimate = prompt_tokens - system_estimate - history_estimate - tools_estimate

设计要点：

对话历史不能直接拼接 content 字段来估算，因为每条消息还有 role 等元数据。用 json.dumps 序列化后估算更接近 API 实际处理的字节量。

工具定义的估算同理，API 接收的是 JSON 序列化后的工具列表。

other_estimate 是倒推出来的：从 API 返回的 prompt_tokens 中减去可估算的部分，剩下的就是“思考指令”、“格式化开销”等不可细分的内容。

5.5 树形明细的拼接

计算完成后，将这些数字拼接成用户可读的树形结构：

# Token 消耗明细的拼接
token_detail = (
    f"\n📊 Token 消耗明细"
    f"\n├ 系统提示词 ≈ {system_estimate} tokens ({len(system_prompt)} 字符)"
    f"\n├ 对话历史 ≈ {history_estimate} tokens ({len(history_text)} 字符)"
    f"\n├ 工具定义 ≈ {tools_estimate} tokens ({len(tools_json)} 字符)"
    f"\n├ 其他(思考指令等) ≈ {other_estimate} tokens"
    f"\n├ 输入合计 {prompt_tokens} Token + 输出 {completion_tokens} Token"
    f"\n├ 总计 {total_tokens} Token"
    f"\n└ 模型: {model} | 推理强度: max"
)

这个树形结构使用 Unicode 制表符 ├ 和 └ 绘制，在任何等宽字体下都能正确对齐。

5.6 明细的推送时机与位置

Token 明细拼接完成后，被追加到推理文本的末尾：

if reasoning_parts:
    reasoning_text = ''.join(reasoning_parts)
    reasoning_text += token_detail
    self._last_reasoning = reasoning_text

这样做的好处是：

不额外占用推送通道：明细作为推理内容的一部分，随推理文本一起推送，不需要额外的通信机制。

自然关联：用户展开折叠区，看到的最后一段就是 Token 消耗明细，逻辑位置自然。

历史可查：self._last_reasoning 包含了完整推理文本和 Token 明细，一起保存到数据库的 reasoning_content 字段中，切换面板再切回来时仍然可见。

5.7 前端展示的小技巧

由于 Token 明细以特殊格式（📊 开头，含 Token 关键字）嵌入推理文本，前端在流式渲染时可以对这些行做特殊样式处理：

// 前端对 Token 统计行的特殊样式
if (step.indexOf('📊') !== -1 && step.indexOf('Token') !== -1) {
    // 使用金色高亮和等宽字体
    const tokenLine = document.createElement('div');
    tokenLine.className = 'token-stats-line';
    tokenLine.style.cssText = 'color:#fbbf24; font-family:monospace; white-space:pre-wrap;';
    tokenLine.textContent = step;
    stepsDiv.appendChild(tokenLine);
} else {
    // 普通推理步骤
    const stepEl = document.createElement('div');
    stepEl.textContent = '• ' + step;
    stepsDiv.appendChild(stepEl);
}

这样，用户在折叠区中看到的是：

普通推理内容：白色文本，带 • 前缀

Token 消耗明细：金色高亮、等宽字体、保留树形结构

5.8 实践价值

这个明细功能上线后，我很快就发现了几个优化点：

系统提示词占比过高：某次对话中，系统提示词消耗了 1800+ Token，占总输入的 70% 以上。原因是提示词中包含了冗长的反思笔记和子 Agent 调度说明。优化后精简了提示词中不必要的部分，单次交互成本降低了约 30%。

工具定义膨胀：当 MCP 服务器配置较多时，工具定义可能达到 3000+ Token。解决方案是让 MCP 工具按需动态注入，而非一次性全量加载。

对话历史累积：长对话中，历史消息的 Token 占比逐渐增大。这促使我优化了历史加载逻辑——不是加载固定条数，而是加载固定 Token 总量的历史。

没有这个明细，以上优化点都无从发现。

六、前端折叠区的交互设计与性能优化

后端推送推理内容的机制已经建立，但用户的直观体验最终取决于前端。这一节我们深入到 DOM 操作的细节，剖析一个高性能、交互友好的折叠区是如何构建的。

6.1 动态创建折叠区：初始化与状态管理

前端在接收推理内容时，面临两种完全不同的场景：

场景一：流式推送中。推理内容正在从后端实时推送过来，折叠区需要从无到有动态创建，并且默认展开——用户正在等待回答，自然希望看到 AI 当前的思考状态。

场景二：历史消息回显。用户切换面板或重新打开 Agent 时，从数据库加载的历史消息中可能包含 reasoning_content。这些推理内容应该默认折叠——它们是“过去完成时”的思考，不应该抢占视觉空间。

两种场景用一个函数处理，通过标志位区分：

// 流式场景：动态创建折叠区并实时追加
window.receiveAgentThinking = function(step) {
    let container = document.getElementById('agentMessages');
    if (!container) return;
 
    // 尝试找到正在进行的流式折叠区
    let thinkBlock = container.querySelector('.think-block.think-stream');
 
    if (!thinkBlock) {
        // 首次创建——流式场景
        thinkBlock = document.createElement('div');
        thinkBlock.className = 'think-block think-stream';
        var uniqueId = 'think-stream-' + Date.now();
        thinkBlock.innerHTML = `
            <div class="think-panel" style="background:#0f172a;border-radius:8px;padding:8px 12px;border:1px solid #334155;">
                <div class="think-toggle" style="cursor:pointer;display:flex;justify-content:space-between;align-items:center;">
                    <span style="color:#fbbf24;font-size:12px;">🧠 深度思考（点击查看）</span>
                    <div style="display:flex;align-items:center;gap:8px;">
                        <button class="think-copy-btn" data-target="${uniqueId}">📋 复制</button>
                        <span class="think-arrow">▲</span>
                    </div>
                </div>
                <div id="${uniqueId}" class="think-steps" style="display:block;font-size:12px;color:#94a3b8;padding:8px 0 0 0;white-space:pre-wrap;line-height:1.6;"></div>
            </div>`;
        container.appendChild(thinkBlock);
        // 绑定折叠、复制等事件（详见 6.3）
    }
 
    // 追加思考步骤
    const stepsDiv = thinkBlock.querySelector('.think-steps');
    // Token 明细行特殊样式处理（详见 6.4）
    if (step.indexOf('📊') !== -1 && step.indexOf('Token') !== -1) {
        // 移除旧的 token 统计行，只保留最新的
        var oldLine = stepsDiv.querySelector('.token-stats-line');
        if (oldLine) oldLine.remove();
        var tokenLine = document.createElement('div');
        tokenLine.className = 'token-stats-line';
        tokenLine.style.cssText = 'color:#fbbf24;margin-top:8px;padding-top:6px;border-top:1px solid #334155;white-space:pre-wrap;font-family:monospace;font-size:11px;';
        tokenLine.textContent = step;
        stepsDiv.appendChild(tokenLine);
    } else {
        var stepEl = document.createElement('div');
        stepEl.textContent = '• ' + step;
        stepsDiv.appendChild(stepEl);
    }
};

而对于历史消息回显，处理方式类似，但有两个区别：折叠区使用 think-block 类名（不带 think-stream 标志），默认设置 display:none。

6.2 滚动控制：自动到底部与悬停暂停

流式推送过程中，折叠区的内容在不断增长。如果聊天面板不能自动滚动到底部，用户需要手动拖动滚动条才能看到最新的推理步骤——这在每次交互中都是极其糟糕的体验。

自动滚动逻辑：

// 自动滚动到底部（仅在用户未手动上滑时）
const isNearBottom = container.scrollHeight - container.clientHeight - container.scrollTop < 80;
if (isNearBottom) {
    container.scrollTop = container.scrollHeight;
}

这里的 80 像素容差是一个关键设计。如果用户手动向上滚动了超过 80 像素（比如正在看之前的推理步骤），自动滚动会暂停，不会强制打断用户的阅读。只有当用户滚回底部附近时，自动滚动才会恢复。

鼠标悬停暂停滚动：

这是从实际使用中总结出的交互痛点：用户在阅读推理内容时，如果鼠标悬停在折叠区上，说明正在仔细阅读。此时强制滚动到底部会打断阅读。解决方法是给折叠区绑定两个事件：

// 悬停时暂停自动滚动
thinkBlock.addEventListener('mouseenter', function() {
    thinkBlock._thinkingPaused = true;
});
thinkBlock.addEventListener('mouseleave', function() {
    thinkBlock._thinkingPaused = false;
});
 
// 在滚动逻辑中检查
if (!thinkBlock._thinkingPaused) {
    container.scrollTop = container.scrollHeight;
}

6.3 折叠切换与复制交互

折叠切换是最基本的交互，但设计上有讲究：用户点击折叠区的标题栏（而非内容区域）来切换展开/折叠状态。

// 折叠切换
var toggleDiv = thinkBlock.querySelector('.think-toggle');
var contentDiv = thinkBlock.querySelector('.think-steps');
var arrowSpan = thinkBlock.querySelector('.think-arrow');
 
toggleDiv.addEventListener('click', function() {
    if (contentDiv.style.display === 'none') {
        contentDiv.style.display = 'block';
        arrowSpan.textContent = '▲';
    } else {
        contentDiv.style.display = 'none';
        arrowSpan.textContent = '▼';
    }
});

复制功能需要处理一个技术细节：navigator.clipboard.writeText 在现代浏览器中需要 HTTPS 或 localhost 环境。在 PyWebView 中通常能满足，但如果遇到权限问题，需要降级到传统的 document.execCommand('copy') 方案：

var copyBtn = thinkBlock.querySelector('.think-copy-btn');
copyBtn.addEventListener('click', function(e) {
    e.stopPropagation();  // 阻止冒泡，防止触发折叠切换
    var text = contentDiv.innerText;
    navigator.clipboard.writeText(text).then(function() {
        window.showToast && window.showToast('✅ 已复制思考过程', 1500);
    });
});

6.4 Token 明细行的特殊渲染

Token 明细包含了树形制表符 ├ 和 └，必须使用等宽字体才能正确对齐。前端通过检测 📊 和 Token 关键字来识别明细行，并应用特殊样式（详见 6.1 中的条件分支）。

另外，在流式推送过程中，明细数据可能会更新。因此前端在渲染新的明细行时，会先移除旧的 token-stats-line，保证始终只显示最新的一条明细。

6.5 流式结束后的状态收尾

当后端流式推送完成时，会调用 window.finishAgentThinking 来通知前端。前端的处理是：

window.finishAgentThinking = function() {
    const container = document.getElementById('agentMessages');
    if (!container) return;
    const thinkBlock = container.querySelector('.think-block.think-stream');
    if (thinkBlock) {
        // 移除流式标志，使其变为“历史消息”状态
        thinkBlock.classList.remove('think-stream');
        // 更新标题，去掉“实时”意味
        const titleSpan = thinkBlock.querySelector('.think-toggle span');
        if (titleSpan) {
            titleSpan.textContent = '🧠 深度思考（点击查看）';
        }
    }
};

注意 think-block.think-stream 变成了 think-block（移除了 think-stream 类）。这是为了和“历史消息”的折叠区区分开来。在后续切换面板再切回时，系统会根据这个类名判断哪些是已经完成的折叠区，不应该再被流式推送逻辑修改。

七、总结与展望

7.1 回顾：我们做了什么

本文从零开始，完整拆解了一个 Agent 系统中最具差异化的两个功能——流式推理可视化和 Token 粒度统计——的设计原理与实现细节。

我们沿着数据流的全链路，逐一剖析了每个技术环节：

章节 技术环节 核心难点

第三章 SSE 流式解析与缓冲区设计 如何在实时性与渲染性能间取得平衡；多重条件触发推送的调优

第四章 工具调用后的递归处理 消息结构的动态变化；reasoning_content 的清洗逻辑；递归终止条件的保障

第五章 Token 消耗明细的统计 估算算法的原理与误差控制；树形明细的拼接格式；各部分 Token 的倒推计算

第六章 前端折叠区的交互设计 流式渲染与历史回显的状态区分；自动滚动与悬停暂停的配合；复制功能的降级方案

7.2 设计哲学：透明优于魔法

如果要用一句话总结本文的技术哲学，那就是：透明优于魔法。

市面上的 AI 产品普遍追求“魔法感”——用户说话，AI 回答，中间的思考过程被刻意隐藏。这种设计有其商业逻辑，但对一个个人管家来说，可信赖比魔法感重要得多。

可信赖建立在两个基石之上：

可追溯：当 Agent 做出某个决策时，用户可以回溯完整的推理路径，理解“为什么这样选择”。

可度量：用户知道每次交互消耗了多少资源，可以据此优化自己的使用方式，也可以对 Agent 的成本结构有清晰的认知。

这两个功能，就是这两块基石的具体实现。

7.3 实际价值：从调试效率到成本控制

这两个功能上线后，带来了一系列具体的改善：

调试效率大幅提升。之前，当 Agent 调用了一个非预期的工具时，我只能根据最终结果猜测原因。现在，我可以直接展开折叠区，看到模型在决定调用工具前的完整分析过程。问题定位时间从“数分钟猜测”缩短到“几秒钟定位”。

成本结构清晰可见。Token 明细上线后，我发现了之前完全忽视的几个成本黑洞：系统提示词中的反思笔记在某些情况下超过 2000 Token；工具定义的膨胀在配置了多个 MCP 服务器后达到了 3000+ Token。基于这些数据，我对提示词和工具加载策略进行了针对性优化，单次交互成本降低约 30%。

用户信任感增强。当用户看到 AI 的完整思考过程时，不再觉得它是一个“神秘的黑盒”，而是一个“可以理解的工具”。这种信任感是任何商业产品都无法用营销手段建立的。

7.4 展望：可扩展的方向

目前的功能是“单次交互透明”，未来可以沿着以下方向继续演进：

累计统计：跨对话的 Token 消耗累计，按天/周/月维度呈现，帮助用户管理长期 AI 使用成本。

思考过程挖掘：将多轮对话的推理内容进行聚类分析，发现用户的隐性偏好模式，用于优化系统提示词和记忆检索策略。

多模型对比：在同一场景下使用不同模型（DeepSeek、Qwen、Claude 等），对比它们的推理路径和 Token 效率，为用户提供模型选择的依据。

这些方向都建立在现有的折叠区和 Token 明细基础之上，不需要重构核心架构，可以逐步迭代。

7.5 结语

本文展示的代码片段涵盖了从后端到前端的所有关键技术环节，但保留了完整的工程实现细节。如果你在阅读后感到“原理清楚了，但还有一些细节需要自己探索”，那说明本文达到了它的目的——展示思路，启发实践。

市面上有无数 AI 产品，但愿意让 AI 变得透明的，凤毛麟角。希望这篇文章能给那些同样追求“可信赖 AI”的开发者一些启发和勇气——做那些大厂不愿意做的事，恰恰是独立开发者最大的优势。