博客

DeerFlow Backend Sandbox 执行环境分析

May 3, 2026

xiwang

全栈开发者

本文档分析了在 deer-flow 后端体系中，代码沙箱执行环境（Sandbox）的架构设计与实现原理。沙箱主要用于让 AI 或用户安全、隔离地执行代码块（Shell 命令、读写文件等）。

1. 核心抽象设计 (Core Abstractions)

后端的沙箱系统被设计为一个标准的面向对象接口体系，具有极高的解耦和可替换性：

1.1 Sandbox (执行层基类)

定义在 packages/harness/deerflow/sandbox/sandbox.py。 它是一个标准的接口层，规定了所有沙箱必须提供以下能力：

• execute_command(command)：执行终端命令。
• read_file(path) / write_file(path, content) / update_file：文本与二进制文件的 I/O 操作。
• list_dir(path, max_depth)：目录遍历。

1.2 SandboxProvider (生命周期、路由与缓存引擎)

定义在 sandbox_provider.py，但在 aio_sandbox_provider.py 中有极其高级的实现。 它是单例管理器，负责沙箱生命周期的调度：

• acquire：获取或新建一个特定线程的沙箱空间。在 AIO Provider 中，它实现了一个极致的三层缓存架构：
  • 第一层 (In-process cache)：极速复用当前进程已挂载的活跃沙箱。
  • 第二层 (Warm pool)：将释放的沙箱置于“保暖池”（容器不销毁，只解除绑定），以便同一个 Thread 再次申请时“零冷启动”秒速复用。
  • 第三层 (Cross-process Discovery)：通过确定性哈希 ID（Deterministic ID）和跨进程文件锁（File Lock），让不同进程甚至不同 Pod 能够发现彼此启动的共享底层沙箱，解决多进程冲突。
• get：根据 ID 提取已挂载的沙箱实例并在后台自动续签闲置过期时间 (Idle Timeout)。
• release：将闲置沙箱退回到保暖池，只有超过系统限定副本数（Replicas）或超时时才会真正执行销毁任务。

---

2. 具体实现机制 (Concrete Implementations)

目前系统中存在两种截然不同的沙箱实现策略：

2.1 本地宿主机伪沙箱 (LocalSandbox)

定义在 sandbox/local/local_sandbox.py。 这是一种非容器化的轻量级实现形式，直接在后端进程宿主机通过原生进程执行命令。

• 执行原理：底层使用了 Python 原生的 subprocess.run(command, shell=True) 来直接执行 Bash/Zsh 脚本，这代表着它并没有真正的物理隔离，执行的是宿主机的直接命令。
• 智能路径映射机制（Path Mapping）：为了让从外部或者习惯了容器化路径的工具能够平滑运行，它包含了一个精妙的正则拦截系统。
  • 它在内存里维护了一套映射表（如把虚拟容器路径 /mnt/skills 映射到真实的宿主物理路径 /Users/...）。
  • 在把用户的命令喂给 subprocess.run 执行之前，它会先通过正则，把命令里出现的虚拟路径全部**替换（Resolve）**成真实的绝对路径。
  • 当真实命令吐出 Output 日志时，它又会把这些真实的、可能泄漏隐私的绝对物理路径，全部**反向替换（Reverse Resolve）**回虚拟容器路径。实现了对大模型视角下的“路径伪装隔离”。

2.2 容器化云原生沙箱 (AioSandbox 与 SandboxBackend)

定义在社区模块 community/aio_sandbox/ 下。这是一套真正的物理级别的容器化隔离方案，被拆分为执行者（AioSandbox）和驱动引擎（SandboxBackend）两个维度：

• 执行器 (AioSandbox) 通信隔离：所有的 execute_command 并不是本地 Subprocess，而是被转化为长超时的 HTTP REST API，发送给内部运行了 agent-infra/sandbox 服务的 Docker 容器。
• 底层驱动引擎 (SandboxBackend)：负责产生并调度这些隔离容器。它支持两种模式：
  • RemoteSandboxBackend：连接远程 K8s / K3s 的 Provisioner，动态生成远端 Pod。
  • LocalContainerBackend：直接按需在物理机上启动新的 Docker 容器，并分配闲置端口。特别地，在 macOS 系统下，它会优先检测并采用苹果官方原生支持的轻量级虚拟化方案 (Apple Container)，仅在找不到时降级回退给 Docker。此设计极大优化了 Mac 上的 I/O 与执行速度。
  • 请求将指令发往由 agent-infra/sandbox 提供的一个运行在大后方的 Docker 容器内部服务（默认为 HTTP API 监听）。
  • 由该远程 Docker 容器内部安全地执行这些指令代码，并将结果 JSON（通常包含 result.data.output）返回给后端的 AioSandbox 包装实例。

3. 架构总结

通过这套抽象实现，当 AI 大模型提出如 在命令行执行 python script.py 的需求时：

• 系统不需重构大模型逻辑，即可通过配置自由决定是采用零成本的本地直连替换（LocalSandbox），还是高规格的容器 HTTP 远程直连（AioSandbox）。
• 两套底层驱动均共享同一个上层执行接口，达到了业务代码的高度复用。

Tags:

• deer-flow
• sandbox

DeerFlow Backend 架构分析：记忆、工具与技能系统

May 2, 2026

xiwang

全栈开发者

本文档分析了 deer-flow 后端（基于 LangGraph 开发，主要代码位于 packages/harness/deerflow 下）的核心三大系统实现机制。

1. 记忆系统 (Memory System)

记忆系统被精心设计为“短期（会话）”与“长期（全局）”两个层级：

1.1 短期记忆（基于 Thread Checkpointer）

通过 LangGraph 原生支持的 checkpointer/async_provider.py，记录特定对话流（Thread ID）的每一次状态流转（状态机快照）。这意味着用户刷新页面或断开连接后，对话上下文依然完整保留且可追溯。

1.2 长期异步记忆（全局档案提取）

核心分布在 agents/memory/storage.py 与 agents/middlewares/memory_middleware.py：

• 无感更新（中间件拦截）：MemoryMiddleware 会拦截所有对话记录。为了节省 Token 并消除噪音，它会精准过滤掉内部大模型思考链和工具调用的中间过程记录，只保留纯人类提问和 AI 最终回复。
• 队列防抖处理：整理后的“干净”对话记录会被推入 queue.py 的内存防抖队列中。这种设计有效防止了高频、高并发聊天引发的后端计算风暴。
• 结构化档案与 LLM 提取：后台异步消费队列，调用 LLM 进行记忆提取，将其存储为持久化的结构化 JSON 档案（保存在本地配置路径内）。保存的档案节点非常精细，包括：
  • workContext：工作上下文
  • personalContext：个人偏好习惯
  • topOfMind：近期关注重点（脑海优先区域）
  • history：包含 recentMonths、earlierContext 等长期和近期时间线归纳
  • facts：用户相关的客观事实规律摘要

这种设计使得 Agent 具备了跨越时间和窗口了解用户的能力。

---

2. 工具系统 (Tool System)

工具系统具有极高的模块化和扩展性，主要分布在 tools/ 和 community/ 等目录下：

2.1 Builtins (内置原生工具)

位于 tools/builtins 目录。这些通常是需要最快响应速度、最底层依赖的基础计算工具等。

2.2 Community (外部社区整合功能)

高度隔离的第三方服务接入层：

• 多模态搜索：集成了 ddg_search (DuckDuckGo)、tavily 高级搜索引擎以及 image_search 图片搜索。
• 安全沙箱执行环境：如 aio_sandbox，允许大模型生成的代码在一个隔离安全的环境（可能支持了本地或远程后端）中真机运行。
• 文档读取工具：诸如读取复杂网页的 firecrawl 爬虫工具等。

2.3 MCP (Model Context Protocol) 扩展协议

目录结构涵盖 mcp/ 和配置文件 config/acp_config.py，标志着应用引入了工业界最新的大模型上下文通信标准协议（MCP）。这使得该后端可以在未来轻松地“挂载”诸如本地数据库读取器、VS Code 本地环境探针等强大的动态独立工具，彻底解耦代码。

---

3. 技能系统 (Skills System)

通过分析 skills/ 目录下的组成部分（loader.py, parser.py, installer.py, validation.py, types.py），可以发现“技能(Skills)”和传统的 Python 代码“工具(Tools)”在架构地位上存在本质差异：

3.1 定义上的差异

技能是一种软编码（Soft-coded）的注入型操作规范（SOP）或声明式模板，而非特定的 Python 执行逻辑。根据 parser.py 的源码实现，技能具有及其严格的物理约束：

• 文件名必须严格为：SKILL.md
• 文件头部必须包含标准的 YAML 前置元数据（Front-matter，需包含 name 与 description），用于大模型技能发现。

3.2 动态加载生命周期

• 解析（Parser）：运行时热解析特定格式的外部文件资源。
• 静态校验（Validation）：由于技能文件具备特定类型（见 types.py），系统会在挂载前通过 Schema 校验其参数与输入输出定义的合法性。
• 注册安装（Installer & Loader）：经验证后的技能会被挂载进大模型（Lead Agent 等）的 Prompts 集合或能力感知池中。

3.3 核心意义

这种极致设计意味着，在不需要后端工程师改动或部署一行 Python 业务代码的情况下，仅通过上传、配置新的技能文件库，就可以瞬间赋能 AI 掌握全新的办事规范（比如：“如何审查 React 代码的规范”或“提交 PR 的标准格式”）。

Tags:

• deer-flow
• backend

claude-mem

Feb 2, 2026

xiwang

全栈开发者

claude-mem 是一个专为 Claude Code 设计的插件，旨在为 AI 助手提供持久化记忆能力。

---

1. claude-mem 是什么？

claude-mem 是一个持久化内存压缩系统。它通过以下方式工作：

• 自动捕获：在编码会话期间，它会自动记录 Claude 的所有操作和工具使用情况。
• AI 压缩：利用 Claude 的 agent-sdk，将这些记录通过 AI 进行语义摘要和压缩。
• 内容注入：在未来的会话中，它会将相关的历史背景重新注入，让 Claude “记起”之前的工作。

2. 它解决了什么问题？

具体来说，它通过以下方式解决了开发者在使用 AI Agent 时常遇到的痛点：

1. 解决“上下文丢失”问题

• 痛点：通常情况下，当你关闭一个 Claude Code 会话或重新连接时，AI 会失去之前所有的操作上下文。
• 解决方案：claude-mem 会自动捕获 Claude 在编码过程中的所有操作和工具使用记录，并在未来的会话中自动注入相关的上下文。 这使得 Claude 能够保持对项目知识的连续性。

2. 提高长周期项目的协作效率

• 痛点：对于跨度数天或数周的大型项目，用户需要不断重复向 AI 解释背景信息，导致浪费大量的 Token 和时间。
• 解决方案：该项目利用 AI（通过 Claude 的 agent-sdk）对过去的会话进行语义总结和压缩， 确保 Claude 即使在新的会话中也“记得”之前的设计决策或已修复的 Bug。

3. 精准的知识检索

• 痛点：在海量的历史操作记录中，很难手动找到特定的代码片段或决策逻辑。
• 解决方案：它构建了一个持久化内存系统，支持基于技能的搜索（Skill-Based Search）。 开发者可以使用自然语言查询项目历史， 其底层的 Chroma 向量数据库支持混合语义检索，能智能地找回最相关的上下文。

4. 优化 Token 使用成本

• 痛点：如果一次性把所有历史记录塞给 AI，会导致 Token 消耗极快且容易混淆。
• 解决方案：它采用了渐进式披露 (Progressive Disclosure) 的策略， 通过 3 层工作流（搜索、时间线、获取详情）分阶段检索，帮助用户在保持上下文的同时，节省约 10 倍的 Token 消耗。

3. 如何安装与使用？

快速安装步骤：

1. 在终端中启动一个新的 Claude Code 会话。
2. 输入以下命令添加插件：

/plugin marketplace add thedotmack/claude-mem

3. 执行安装命令：

/plugin install claude-mem

4. 重启 Claude Code 即可生效。

使用说明：

• 自动化运行：安装后无需手动干预，它会自动捕获观察结果并生成摘要。
• 智能搜索：你可以使用 mem-search 技能，通过自然语言查询项目历史。
• Web 查看器：在浏览器访问 http://localhost:37777，可以实时查看内存流和历史记录。
• 隐私控制：可以使用 <private> 标签来标记不希望被存入内存的敏感内容。

Tags:

• claude
• claude-mem

LLM temerature 和 top_p

Jan 9, 2026

xiwang

全栈开发者

Temperature（温度）和 Top_p（核采样）是控制 LLM 输出随机性和创造力的两个核心参数。

简单来说：Temperature 调整概率分布的形状，而 Top_p 则是切断低概率的选项。

以下是详细的选择指南和最佳实践：

1. 核心概念速览

• Temperature (0.0 - 2.0):

• 作用：控制“大胆程度”。

• 低温 (Low)：模型变得保守、自信，倾向于选择概率最高的词。输出稳定、重复。

• 高温 (High)：模型变得“疯狂”，会给低概率词更多的机会。输出多样、不可预测。

• Top_p (0.0 - 1.0):

• 作用：控制“候选词范围”。

• 低 Top_p：只从前 X% 概率最高的词里选（例如只看前 10% 的词）。

• 高 Top_p：允许从更广泛的词汇库中选择。

---

2. 场景化配置推荐表

这是最实用的速查表，根据你的任务类型进行选择：

任务场景	Temperature 推荐	Top_p 推荐	目标效果
代码生成 / 数学解题	0.0 - 0.2	0.1 - 0.2	精确、确定性。避免模型瞎编乱造，保证逻辑严密。
数据提取 / 格式转换	0.0	0.1	严格。你需要 JSON 就是 JSON，不要多余的废话。
通用问答 / 客服	0.5 - 0.7	0.8 - 0.9	平衡。既要准确，又要像人说话一样自然，不呆板。
文章润色 / 摘要	0.5 - 0.7	0.9	流畅。保留原文意思，但在措辞上有一点灵活性。
创意写作 / 头脑风暴	0.8 - 1.0+	0.9 - 1.0	发散。需要模型给出意想不到的点子，允许偶尔的“胡言乱语”。

---

3. 如何协同调整？（黄金法则）

大多数 API 提供商（如 OpenAI）建议遵循一个原则：通常只调整其中一个，而不是同时调整两个。

策略 A：锁定 Top_p，只调 Temperature（推荐）

这是最常见的做法。

• 将 Top_p 设置为 1.0（即不限制候选池）。
• 通过 Temperature 来控制整体的随机性。
• 写代码？Temp = 0.1
• 写小说？Temp = 0.9

策略 B：锁定 Temperature，只调 Top_p

• 将 Temperature 设置为 1.0（标准分布）。
• 通过 Top_p 来切断长尾词。
• 想要极度稳健？Top_p = 0.1
• 想要多样性？Top_p = 0.95

4. 深入理解：它们到底有什么区别？

为了形象地理解，假设模型在预测下一个词，候选词及其概率如下： [苹果: 50%, 香蕉: 30%, 飞船: 15%, 恐龙: 5%]

• 调节 Temperature (例如调高)：

• 它会“拉平”概率分布。

• 苹果的优势变小（变成 40%），恐龙的机会变大（变成 15%）。

• 结果：所有词都有机会被选中，甚至是很离谱的词。

• 调节 Top_p (例如设为 0.8)：

• 模型只取累计概率达到 80% 的前几个词。

• 50% (苹果) + 30% (香蕉) = 80%。

• 结果：飞船和恐龙直接被切掉了，根本不参与随机。模型只能在苹果和香蕉里选。

总结

• 如果不确定：默认保持 Top_p = 1.0，只调整 Temperature。
• 要精准：Temperature 设为 0。
• 要创意：Temperature 设为 0.8 ~ 1.0。

Tags:

• LLM

LlamaCoder：开源版 Claude Artifacts 的全链路流程解析

Jan 1, 2026

xiwang

全栈开发者

LlamaCoder 是一个由 Nutlope 开发的开源项目，旨在利用 Llama 3.1 等大语言模型实现类似 Claude Artifacts 的代码生成与实时预览功能。本文将深入剖析 LlamaCoder 的核心工作流程，包括会话创建、多模型协作以及代码生成的具体实现逻辑，帮助读者理解其背后的技术架构与设计思路。

功能分析

1. /api/create-chat - 创建聊天会话

功能概述： 初始化一个新的 AI 编码对话会话

核心流程：

用户提交 prompt
     ↓
1. 创建 Chat 记录（数据库）
     ↓
2. 并行执行两个任务：
   ├── fetchTitle() - 生成聊天标题（3-5词）
   └── fetchTopExample() - 匹配相似示例（landing page/blog app等）
     ↓
3. 如果有截图 → 调用视觉模型分析截图
     ↓
4. 如果 quality === "high" → 调用架构模型生成项目计划
     ↓
5. 保存初始消息到数据库：
   ├── System 消息（编码指令）
   └── User 消息（prompt 或项目计划）
     ↓
6. 返回 { chatId, lastMessageId }

使用的模型：

• meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo - 标题生成、示例匹配
• Qwen/Qwen3-VL-32B-Instruct - 截图分析
• Qwen/Qwen3-Next-80B-A3B-Instruct - 项目计划

---

2. /api/get-next-completion-stream-promise - 流式生成代码

功能概述： 根据对话历史流式生成 AI 响应

核心流程：

接收 messageId, model
     ↓
1. 查询消息及历史记录（position <= 当前消息）
     ↓
2. Token 优化：
   ├── optimizeMessagesForTokens() - 移除早期 assistant 消息中的代码块
   └── 消息长度限制 - 超过10条时保留 [前3条 + 最后7条]
     ↓
3. 调用 LLM（stream: true）
     ↓
4. 直接返回流式响应

Token 优化策略：

1. 代码块剥离 - 保留最后 2 条 assistant 消息的完整内容，早期消息删除代码块
2. 消息裁剪 - 最多保留 10 条消息：[0, 1, 2] + 最后7条

工作流程（从前端视角）

[page.tsx:132-167]

用户提交表单
     ↓
POST /api/create-chat
（创建会话、生成标题、项目计划）
     ↓
返回 { chatId, lastMessageId }
     ↓
POST /api/get-next-completion-stream-promise
（核心：这里调用 LLM stream: true）
     ↓
获取流式响应 → 跳转到 /chats/{chatId}
（页面边接收边显示 AI 生成的代码）

简单总结：

接口	作用	返回
/api/create-chat	初始化对话，生成标题、项目计划	chatId, lastMessageId
/api/get-next-completion-stream-promise	流式生成 AI 代码响应	可读流

页面跳转到 /chats/{chatId} 时的完整流程

时间线总览

[首页] 用户提交表单
    ↓
POST /api/create-chat
    → 创建 Chat 记录
    → 生成初始消息（system + user）
    → 返回 { chatId, lastMessageId }
    ↓
POST /api/get-next-completion-stream-promise
    → 获取流式响应 Promise
    ↓
router.push(`/chats/${chatId}`)
    ↓
┌─────────────────────────────────────────────────┐
│  [chats/[id]/page.tsx] 服务端渲染                │
├─────────────────────────────────────────────────┤
│  1. getChatById(id)                              │
│     → 从数据库加载 Chat 和消息                     │
│  2. 生成 HTML + 初始数据                          │
│  3. 返回给客户端                                  │
└─────────────────────────────────────────────────┘
    ↓
┌─────────────────────────────────────────────────┐
│  [page.client.tsx] 客户端激活                     │
├─────────────────────────────────────────────────┤
│  1. 接收服务端传来的 chat 数据                    │
│  2. 从 Context 获取 streamPromise                │
│  3. useEffect 触发流处理                          │
│  4. 开始接收 AI 流式响应                          │
│  5. 逐字显示代码                                  │
│  6. 流结束后保存消息到数据库                       │
└─────────────────────────────────────────────────┘

详细步骤

步骤 1: 服务端渲染 (page.tsx)

// 用户访问 /chats/abc123
export default async function Page({ params }) {
  const id = (await params).id;

  // 从数据库加载聊天数据
  const chat = await getChatById(id);
  // {
  //   id: "abc123",
  //   title: "Todo App",
  //   messages: [
  //     { position: 0, role: "system", content: "..." },
  //     { position: 1, role: "user", content: "Build me a todo app" }
  //   ]
  // }

  return <PageClient chat={chat} />;
}

此时 AI 还没有生成代码，只有初始的 system 和 user 消息。

---

步骤 2: 客户端激活

export default function PageClient({ chat }: { chat: Chat }) {
  const context = use(Context);
  // streamPromise 是从首页传过来的！
  const [streamPromise, setStreamPromise] = useState(
    context.streamPromise  // ← 这是已经开始的流
  );

---

步骤 3: 处理流式响应

useEffect(() => {
  async function f() {
    const stream = await streamPromise;  // 获取可读流

    ChatCompletionStream.fromReadableStream(stream)
      .on("content", (delta, content) => {
        // 每收到一个 token 触发
        setStreamText((text) => text + delta);
        // "im" -> "imp" -> "impor" -> "import" ...

        // 检测到代码时显示代码查看器
        if (content.includes("```")) {
          setIsShowingCodeViewer(true);
        }
      })
      .on("finalContent", async (finalText) => {
        // 流结束后保存到数据库
        const message = await createMessage(
          chat.id,
          finalText,        // 完整的 AI 响应
          "assistant",
          allFiles,         // 提取的文件
        );

        router.refresh();  // 刷新服务端数据
      });
  }

  f();
}, [streamPromise]);

---

步骤 4: UI 实时更新

时间轴：
T+0s:   页面加载，显示 ChatLog（只有初始消息）
        ┌─────────────────────────┐
        │ system: You are expert  │
        │ user: Build me a todo   │
        └─────────────────────────┘

T+1s:   开始接收流
        ┌─────────────────────────┐
        │ system: You are expert  │
        │ user: Build me a todo   │
        │ assistant: Here's a     │ ← 实时显示
        └─────────────────────────┘

T+2s:   继续接收
        │ assistant: Here's a React│

T+5s:   检测到代码块，打开 CodeViewer
        ┌────────────┬─────────────┐
        │ ChatLog    │ CodeViewer  │
        │ ...        │ import React│
        │            │ export ...  │
        └────────────┴─────────────┘

T+30s:  流结束，保存到数据库
        router.refresh() → 服务端数据更新

---

关键点总结

阶段	位置	发生了什么
首页	page.tsx	调用 /api/create-chat → 创建 Chat
首页	page.tsx	调用 /api/get-next-completion-stream-promise → 获取流
跳转	router.push	导航到 /chats/{id}，携带 streamPromise
服务端	[id]/page.tsx	getChatById() → 加载数据库数据（此时无 AI 响应）
客户端	page.client.tsx	useEffect → 处理流，逐字显示
结束	page.client.tsx	createMessage() → 保存 AI 响应到数据库

关键理解： 页面跳转时，AI 已经在生成代码了（流已经开始），页面只是负责接收和显示这个流。

getChatById 功能解析

这是一个服务端数据获取函数，用于加载聊天会话的完整数据。

核心功能

const getChatById = cache(async (id: string) => {
  // 1. 查询 Chat 基本信息
  // 2. 查询消息（分批加载策略）
  // 3. 返回组合数据
});

详细步骤

1. 查询 Chat 记录
   ↓
2. 统计总消息数 (totalMessages)
   ↓
3. 分批加载消息：
   ├── position 0,1 (必须加载：system prompt + 初始用户消息)
   └── position ≥2 的最近 100 条
   ↓
4. 计算版本计数器 (assistantMessagesCountBefore)
   ↓
5. 返回组合数据

消息加载策略

假设数据库有 200 条消息：

position 0: [system] 编码指令
position 1: [user] 初始 prompt
position 2-101: [对话...]
position 102-200: [最近对话...]

加载结果：
✓ position 0 (必须)
✓ position 1 (必须)
✓ position 102-200 (最近 100 条)
✗ position 2-101 (跳过，节省数据传输)

最终按 position 排序返回

为什么要 cache()？

const getChatById = cache(async (id: string) => {

React Server Component 的缓存机制：

同一个请求中，getChatById 可能被调用多次：
├── generateMetadata() 需要 chat.title
└── Page() 需要 chat 数据

使用 cache() 确保只查询一次数据库，
第二次调用直接返回缓存结果

返回数据结构

{
  id: "abc123",
  title: "Budgeting App",
  model: "deepseek-v3-2-251201",
  quality: "high",
  prompt: "Build me a budgeting app",

  messages: [
    { position: 0, role: "system", content: "..." },
    { position: 1, role: "user", content: "..." },
    { position: 102, role: "assistant", content: "..." },
    // ... 最多 103 条消息
  ],

  totalMessages: 200,           // 总消息数（用于显示"加载更多"）
  assistantMessagesCountBefore: 50  // 用于版本控制
}

为什么这样设计？

1. 性能 - 不加载所有历史消息，只加载必要的 + 最近 100 条
2. 成本 - 减少数据库查询和数据传输
3. 用户体验 - 用户主要关心最近的对话内容

流式响应中提取代码到多个文件的完整解析

1. AI 输出格式

AI 被要求以特定的 Markdown 格式输出多个文件：

Here's your todo app:

```tsx{path=src/App.tsx}
import React from 'react';
export default function App() { ... }

body { margin: 0; }

{
  "name": "todo-app"
}

### 2. 代码块提取 (`lib/utils.ts`)

#### `parseReplySegments()` - 实时解析流式内容

```typescript
// 逐行解析，支持流式（未闭合的代码块标记为 partial）
export function parseReplySegments(markdown: string): ReplySegment[] {
  const lines = markdown.split("\n");
  const fenceRegex = /^```([^\n]*)$/;

  let openTag: string | null = null;  // 当前代码块是否打开
  let codeBuffer: string[] = [];       // 代码内容缓冲区

  for (const line of lines) {
    const match = line.match(fenceRegex);
    if (match && !openTag) {
      // 开始代码块: tsx{path=src/App.tsx}
      openTag = match[1] || "";
    } else if (match && openTag) {
      // 结束代码块: ```
      segments.push({
        type: "file",
        code: codeBuffer.join("\n"),
        language: "tsx",
        path: "src/App.tsx",
        isPartial: false,  // ← 已完成
      });
      openTag = null;
    } else if (openTag) {
      // 代码块内的行
      codeBuffer.push(line);
    }
  }

  // 如果流结束时代码块未闭合
  if (openTag) {
    segments.push({
      type: "file",
      code: codeBuffer.join("\n"),
      language: "tsx",
      path: "src/App.tsx",
      isPartial: true,  // ← 正在生成中
    });
  }
}

extractAllCodeBlocks() - 提取所有已完成的代码块

export function extractAllCodeBlocks(input: string) {
  const codeBlockRegex = /```([^\n]*)\n([\s\S]*?)\n```/g;
  const files = [];

  let match;
  while ((match = codeBlockRegex.exec(input)) !== null) {
    const fenceTag = match[1];  // "tsx{path=src/App.tsx}"
    const code = match[2];      // 代码内容

    // 解析标签
    const { language, path } = parseFenceTag(fenceTag);
    // language = "tsx"
    // path = "src/App.tsx"

    files.push({ code, language, path });
  }

  return files;
}

parseFenceTag() - 解析代码块标签

function parseFenceTag(tag: string) {
  // 输入: "tsx{path=src/App.tsx}"

  const langMatch = tag.match(/^([A-Za-z0-9]+)/);
  const language = langMatch ? langMatch[1] : "text";
  // language = "tsx"

  const pathMatch = tag.match(/path\s*=\s*([^}\s]+)/);
  const path = pathMatch ? pathMatch[1] : `file.${getExtensionForLanguage(language)}`;
  // path = "src/App.tsx"

  return { language, path };
}

3. 流式处理流程 (page.client.tsx)

ChatCompletionStream.fromReadableStream(stream)
  .on("content", (delta, content) => {
    // 每收到一个 token
    setStreamText((text) => text + delta);

    // 解析当前内容
    const segments = parseReplySegments(content);
    // [
    //   { type: "text", content: "Here's your app:\n\n" },
    //   { type: "file", code: "import...", language: "tsx", path: "src/App.tsx", isPartial: true }
    // ]

    // 检测到文件时打开代码查看器
    if (segments.some((seg) => seg.type === "file")) {
      setIsShowingCodeViewer(true);
    }

    // 第一个完整文件时显示预览
    if (segments.some((seg) => seg.type === "file" && !seg.isPartial)) {
      setActiveTab("preview");
    }
  })
  .on("finalContent", async (finalText) => {
    // 流结束后提取所有文件
    const currentFiles = extractAllCodeBlocks(finalText);
    // [
    //   { code: "import React...", language: "tsx", path: "src/App.tsx" },
    //   { code: "body { margin: 0 }", language: "css", path: "src/styles.css" },
    //   { code: "{ \"name\": \"todo\" }", language: "json", path: "package.json" }
    // ]

    // 合并之前的文件（同一 path 的文件会被覆盖）
    const fileMap = new Map();
    previousFiles.forEach((f) => fileMap.set(f.path, f));
    currentFiles.forEach((f) => fileMap.set(f.path, f));
    const allFiles = Array.from(fileMap.values());

    // 保存到数据库
    await createMessage(chat.id, finalText, "assistant", allFiles);
  });

4. 实时渲染 (code-viewer.tsx)

// 合并流中的文件
const streamAllFiles = extractAllCodeBlocks(streamText);  // 已完成的
const latestStreamBlock = extractLatestStreamBlock(streamText);  // 正在生成的

// 合并：同一 path 的文件，新的覆盖旧的
let mergedStreamFiles = [...streamAllFiles];
if (latestStreamBlock) {
  const existingIdx = mergedStreamFiles.findIndex(
    (f) => f.path === latestStreamBlock.path,
  );
  if (existingIdx !== -1) {
    mergedStreamFiles[existingIdx] = latestStreamBlock;  // 更新正在生成的文件
  } else {
    mergedStreamFiles.push(latestStreamBlock);  // 添加新文件
  }
}

// 与之前消息的文件合并
const baseFiles = lastMessage ? getFilesFromMessage(lastMessage) : [];
const files = mergeFiles(baseFiles, mergedStreamFiles);

5. 可视化流程

AI 流式输出:
"Here's your app:\n\n```tsx{path=App.tsx}\nimpor" (t+1s)
"Here's your app:\n\n```tsx{path=App.tsx}\nimport" (t+2s)
"Here's your app:\n\n```tsx{path=App.tsx}\nimport React" (t+3s)
...
"Here's your app:\n\n```tsx{path=App.tsx}\nexport default" (t+10s)
"Here's your app:\n\n```tsx{path=App.tsx}\nexport default\n```\n\n" (t+11s - 文件1完成)
"```css{path=styles.css}\nbody" (t+12s - 开始文件2)
...

实时解析:
t+1s:  [{ type: "file", path: "App.tsx", isPartial: true, code: "impor" }]
t+10s: [{ type: "file", path: "App.tsx", isPartial: true, code: "export default" }]
t+11s: [{ type: "file", path: "App.tsx", isPartial: false, code: "export default" }] ← 完成
t+12s: [{ type: "file", path: "App.tsx", isPartial: false },
        { type: "file", path: "styles.css", isPartial: true }]

UI 显示:
┌─────────────────────────────────┐
│ 📁 App.tsx                      │
│ import React;                   │
│ export default function App() { │ ← 实时更新
│   return <div>Hello</div>;      │
│ }                               │
├─────────────────────────────────┤
│ 📁 styles.css (generating...)   │
│ body                            │
└─────────────────────────────────┘

关键点总结

函数	作用	时机
parseReplySegments()	实时解析，包括 partial 文件	每次收到 token
extractAllCodeBlocks()	提取所有已完成的代码块	流结束后/查看历史
parseFenceTag()	解析 tsx{path=src/App.tsx}	解析代码块标签
extractLatestStreamBlock()	获取当前正在生成的代码块	实时显示

StickToBottom 包裹 SyntaxHighlighter 的原因

StickToBottom 是一个组件，用于在内容动态增长时自动滚动到底部。这对于流式代码生成非常重要。

问题场景

AI 正在生成代码（流式输出）：

第1秒:  import React from 'react';
第2秒:  import React from 'react';
        export default function App() {
第3秒:  import React from 'react';
        export default function App() {
          return <div>Hello
第4秒:  import React from 'react';
        export default function App() {
          return <div>Hello World</div>;
        }

如果不自动滚动：

初始视图（可见第1-20行）：
┌─────────────────────────────┐
│ import React from 'react';  │ ← 第1行（可见）
│ ...                         │
│ const data = [              │ ← 第20行（可见）
└─────────────────────────────┘

第5秒后（AI生成了50行代码）：
┌─────────────────────────────┐
│ import React from 'react';  │ ← 第1行（仍然可见）
│ ...                         │
│ const data = [              │ ← 第20行（仍然可见）
│                             │
│ ↑ 用户看不到正在生成的代码！ │
└─────────────────────────────┘
   新增的第21-50行在视野外

使用 StickToBottom 后：

第5秒后（自动滚动到底部）：
┌─────────────────────────────┐
│ ...                         │
│ const data = [              │
│   { id: 1, name: 'A' },     │
│ ];                          │
│                             │
│ export default function App()│ ← 新生成的代码（可见）
│   return <div>Hello         │ ← 用户看到实时生成！
└─────────────────────────────┘
   自动滚动到最后

实现原理

code-viewer.tsx

<StickToBottom
  className="relative grow overflow-hidden *:!h-[inherit]"
  resize="smooth"
  initial={false}
>
  <StickToBottom.Content>
    <SyntaxHighlighter
      files={files.map((f) => ({
        path: f.path,
        content: f.code,
        language: f.language,
      }))}
      activePath={
        streamText
          ? latestStreamBlock?.path || files.at(-1)?.path
          : undefined
      }
      disableSelection={!!streamText}
      isStreaming={!!streamText}
    />
  </StickToBottom.Content>
</StickToBottom>

StickToBottom 的工作机制：

1. 监听内容高度变化
2. 当内容增长时，自动滚动到底部
3. resize="smooth" - 平滑滚动动画
4. initial={false} - 初始不滚动（只有内容变化时才滚动）

配合 SyntaxHighlighter 的自动滚动

// SyntaxHighlighter 内部也有自动滚动逻辑
useEffect(() => {
  if (!isStreaming || !editorRef.current) return;

  const editor = editorRef.current;
  const lineCount = model?.getLineCount?.() || 1;

  // 滚动到最后
  editor.revealLine?.(lineCount);
  editor.setScrollTop?.(scrollHeight);
}, [file?.content, activeFile, isStreaming]);

为什么要两层自动滚动？

层级	作用	范围
StickToBottom	滚动整个代码查看器容器	外层容器滚动
Monaco Editor	滚动编辑器内容到最后一行	编辑器内部滚动

┌─────────────────────────────────────────┐
│ StickToBottom (外层容器)                  │
│ ┌─────────────────────────────────────┐ │
│ │ Monaco Editor (内层编辑器)           │ │
│ │                                     │ │
│ │  import React;                      │ │
│ │  ...                                │ │
│ │  export default... (生成中)         │ │ ← 两者协同确保这里可见
│ │                                     │ │
│ └─────────────────────────────────────┘ │
└─────────────────────────────────────────┘

视觉效果

无 StickToBottom:
[ 代码生成中... ]
│ 第1行  │
│ 第2行  │
│ 第3行  │  ← 用户手动滚动
...
│ 第50行 │  ← 用户需要手动滚动到这里才能看到新代码

有 StickToBottom:
[ 代码生成中... ]
...
│ 第48行 │  ← 自动跟随
│ 第49行 │  ← 自动跟随
│ 第50行 │  ← 自动跟随，用户始终看到最新生成的代码

总结

StickToBottom 确保在 AI 流式生成代码时：

1. 用户始终看到最新代码 - 自动滚动到正在生成的位置
2. 无需手动滚动 - 提供更好的观看体验
3. 平滑动画 - resize="smooth" 提供流畅的视觉效果
4. 只在流式时启用 - isStreaming 控制是否启用

Tags:

• LLM
• Artifacts

Newer posts

Older posts