Nexora 用 Rust 处理 AI 流式回复：FRB 桥接实战记录

为什么要在 Rust 里收流

聊天应用最简单的版本是发 HTTP 请求，等完整文本返回再塞进消息气泡。真用起来问题陆续冒出来：首字等太久、用户想中途停止、页面离开要清理任务、网络断了还得保留已收到的内容。

Nexora 选择把麻烦事放在 Rust 里做。模型服务商配置、网络流解析、取消控制和 SQLite 持久化都交给 Rust，再通过 flutter_rust_bridge（FRB）把生成过程转成 Dart Stream。Flutter 只管界面：三栏工作区、会话列表、消息展示和停止按钮。

这么做有几个直接好处。跨语言接口可以围绕业务动作命名，页面用 createTurn、streamResponse 和 cancelResponse，不用自己拼底层通信。多个服务商的字段和结束标记各不相同，Rust 统一转成增量、完成、暂停、失败四种事件，Flutter 只消费标准事件。生成任务用助手消息的 message_id 做身份，取消、持久化、查记录都用同一个 ID，能防住对同一条消息重复启动流。

旋武社区与 FRB 项目

开放原子旋武开源社区是开放原子开源基金会旗下的 Rust 专项社区，提供工具链、学习资料、在线编码和开源项目。准备用 Flutter + Rust 开发的同学，可以先从这里了解 Rust 基础，再结合具体仓库学跨语言开发。

flutter_rust_bridge 是连接 Flutter/Dart 与 Rust 的绑定工具。开发者在 Rust 定义 API 和数据结构，代码生成器产出绑定后，Dart 就能调用接口、处理错误和流式事件。鸿蒙平台的接入说明和问题反馈，可以从该仓库查阅。

环境准备与版本对齐

基础安装参考两篇教程：Flutter-OH 环境搭建，以及 VS Code 配置 Rust 开发环境。复现 Nexora 时，还要按项目 README 对齐 FRB 版本、OHOS target 与签名配置。

本文代码对应 Nexora 仓库 a453a69 提交，关键要求如下：

• Flutter：使用支持 OHOS 的发行版
• Dart：^3.9.2
• FRB Dart/Rust 依赖与 codegen：固定 2.13.0-beta.6
• Rust target：aarch64-unknown-linux-ohos
• 鸿蒙 SDK：API 22

检查命令：

复制flutter --version
rustc --version
cargo --version
rustup target add aarch64-unknown-linux-ohos
cargo install flutter_rust_bridge_codegen --version 2.13.0-beta.6 --locked

FRB 的 Dart 依赖、Rust 依赖和生成器必须配套。只升级生成器而保留旧依赖，会出现生成失败或运行时异常。

获取项目并生成绑定：

复制git clone https://atomgit.com/nutpi/Nexora.git
cd Nexora
flutter pub get
flutter_rust_bridge_codegen generate

flutter_rust_bridge.yaml 配置：

复制rust_input: crate::api
rust_root: rust/
dart_output: lib/src/rust

生成器读取 Rust 对外 API，产出 Dart 绑定和 Rust 桥接代码。业务修改写在 API、仓库、页面或存储层，再重新生成。

核心实现：从发送到历史恢复

完整功能链路

Nexora 的聊天主线：配置服务商与模型 → 创建会话 → 发送消息 → Rust 接收流 → Flutter 更新正文 → 完成或停止 → SQLite 保存 → 重启读取历史。

创建消息与启动生成分开。assistantMessage.id 既是任务身份，也是取消、保存和错误定位的入口。Rust 先写数据库终态，再通知页面；持久化失败时错误通过 Result 返回，不能把调用结束直接当成保存成功。

仓库接口：页面只调业务方法

应用入口先初始化 Flutter 绑定和 Rust 库，再注入 NativeChatRepository。

聊天仓库三个关键接口：

复制abstract interface class ChatRepository {
  Future<native.ChatTurnSnapshot> createTurn(
    native.CreateChatTurnRequest request,
  );
  Stream<ai.ChatStreamEvent> streamResponse(String messageId);
  Future<bool> cancelResponse(String messageId);
}

NativeChatRepository 直接转发给 FRB：

复制@override
Stream<ai.ChatStreamEvent> streamResponse(String messageId) =>
    ai.streamChatResponse(messageId: messageId);

@override
Future<bool> cancelResponse(String messageId) =>
    ai.cancelChatStream(messageId: messageId);

生产用原生实现，测试可替换仓库，让页面无需连接真实模型就能验证状态。

Rust 流式入口

跨语言事件定义：

复制pub enum ChatStreamEventKind {
    Delta,
    Completed,
    Failed,
    Paused,
}

pub struct ChatStreamEvent {
    pub message_id: String,
    pub kind: ChatStreamEventKind,
    pub delta: String,  // 本次新增
    pub text: String,   // 截至当前完整正文
    pub error_code: Option<String>,
    pub error_message: Option<String>,
}

stream_chat_response 实现要点：

复制pub async fn stream_chat_response(
    message_id: String,
    sink: StreamSink<ChatStreamEvent>,
) -> Result<(), NativeError> {
    let context = load_chat_generation_context(&message_id)?;
    let provider = resolve_chat_provider(&context.model_record_id)?;
    let mut cancellation = register_stream(&message_id)?;
    let _guard = ActiveStreamGuard::new(message_id.clone());
    let mut text = String::new();

    match receive_response(/* ... */).await {
        Ok(ReceiveResult::Completed) => {
            if text.trim().is_empty() {
                return finish_failed(/* empty_response */);
            }
            persist_assistant_stream_state(&message_id, &text, ChatMessageStatus::Success)?;
            let _ = sink.add(ChatStreamEvent {
                message_id,
                kind: ChatStreamEventKind::Completed,
                delta: String::new(),
                text,
                error_code: None,
                error_message: None,
            });
            Ok(())
        }
        Ok(ReceiveResult::Paused) => {
            persist_assistant_stream_state(&message_id, &text, ChatMessageStatus::Paused)?;
            // 发送 Paused 事件
            Ok(())
        }
        Err(error) => finish_failed(&message_id, &sink, text, error),
    }
}

三个关键点：先准备上下文再注册活动流，完成与暂停都先写数据库，空响应与失败有独立处理。ActiveStreamGuard 的 Drop 在函数退出时清理活动表。

协议解析与部分保存

Rust 内部把 OpenAI、Anthropic、Google、Ollama 等协议转成统一 ParsedFrame。

网络 chunk 可能截断在 UTF-8 字节中间。StreamDecoder 先缓存字节，遇到换行再解析整行：

复制fn push(&mut self, bytes: &[u8]) -> Result<Vec<ParsedFrame>, NativeError> {
    self.buffer.extend_from_slice(bytes);
    let mut frames = Vec::new();
    while let Some(newline) = self.buffer.iter().position(|b| *b == b'\n') {
        let line = self.buffer.drain(..=newline).collect::<Vec<_>>();
        if let Some(frame) = parse_stream_line(self.protocol, &line[..line.len() - 1])? {
            frames.push(frame);
        }
    }
    Ok(frames)
}

当前处理常见单行 JSON SSE 和 Ollama JSON 行。若一个 SSE 事件由多条 data: 行组成 JSON，还需补充事件级聚合。

部分保存阈值：

复制const MAX_RESPONSE_CHARS: usize = 200_000;
const PARTIAL_SAVE_BYTES: usize = 512;
const PARTIAL_SAVE_INTERVAL: Duration = Duration::from_millis(500);

新增至少 512 字节，或距上次保存 500ms，就写入 Pending 快照。500ms 是循环内检查阈值，没有独立计时任务，不能承诺异常退出最多只损失 500ms 文本。

停止生成与历史恢复

用户点击停止后，Flutter 调用 cancelResponse(messageId)。Rust 从活动流表取得 Tokio watch 发送端发出信号；接收循环用 tokio::select! 同时等网络数据和取消通知，收到后返回 Paused。

cancelResponse 返回 true 只说明信号发送成功，不代表已落盘。页面要等流退出或重新查询消息确认结果。

流事件	数据库状态	Flutter 处理
Delta	达到阈值时 Pending	用完整 text 更新消息
Completed	Success	显示最终正文，结束生成
Paused	Paused	保留停止前文本，结束生成
Failed	Error	保留部分文本或错误提示，显示失败原因

工作区用 await for 消费事件，按消息 ID 更新气泡；finally 中重新读取当前话题的持久化消息。StreamSubscription.cancel() 只取消 Dart 订阅，不能替代 Rust 取消 API。

重启后从历史读取。恢复已保存消息与恢复中断的网络流是两件事：当前不能将历史恢复理解为自动续传，也不能保证异常退出后遗留的 Pending 已自动修复。

真机运行

构建与安装

DevEco Studio 打开 ohos/，配置签名后构建：

复制flutter build hap --debug --target-platform ohos-arm64 --codesign

签名产物在 build/ohos/hap/entry-default-signed.hap。安装并启动：

复制hdc -t <设备ID> install -r build/ohos/hap/entry-default-signed.hap
hdc -t <设备ID> shell aa start -a EntryAbility -b com.cherryhq.cherry_studio

验证步骤

1. 配置可用服务商和模型，确认设备能访问接口
2. 新建会话，发送 Reply with exactly: DEEPSEEK_OK，核对返回
3. 发送第二条消息，检查新消息正常出现
4. 发起长回复，观察增量更新；中途停止，核对部分正文保留且状态为 Paused
5. 关闭重启应用，检查已保存的会话和消息内容
6. 用无效凭据测试，检查失败提示和状态复位
7. 分别验证页面销毁、网络中断、应用异常退出，记录终态及重启表现

原有记录显示 DeepSeek 首轮回复通过，历史恢复通过；主动取消、多协议联调、异常退出等需专项复测。

问题排查与贡献

提 Issue

先确定层次：Nexora 业务问题进 Nexora Issues；FRB 生成失败、类型映射异常进 FRB Issues。

至少包含：环境信息（主机、Flutter-OH、Dart、Rust、FRB 版本、SDK、设备型号）、复现步骤、预期与实际行为、完整日志、最小复现仓库。

提 PR

1. 阅读仓库贡献指南，在 Issue 说明修复思路
2. Fork 仓库，从目标分支创建修复分支
3. 修改源码，补充回归测试，重新生成绑定
4. 执行格式、静态检查和测试，在原报错设备验证
5. 推送到个人 Fork，在 AtomGit 提 PR
6. 写明触发条件、根因、修复行为、验证结果、关联 Issue

常见问题速查

现象	优先检查	处理
flutter doctor 未识别鸿蒙工具链	Flutter 路径与 SDK 配置	使用 Flutter-OH，配置本机 SDK
FRB 生成后接口不匹配	版本对齐	统一 2.13.0-beta.6，重新生成
找不到 Clang 或 OHOS target	Native SDK 与 Rust target	安装对应 SDK 和 aarch64-unknown-linux-ohos
HAP 安装签名校验失败	包名、证书、Profile、UDID	为当前设备配置匹配签名
启动时原生库加载失败	HAP 架构与动态库	检查 arm64 包内 so 文件与构建集成
HTTP 401/403	凭据与权限	核对设置页凭据，避免重发错误请求
HTTP 429	额度或频率限制	降低频率或稍后重试
空响应	协议类型与输出字段	检查解析分支，空正文转 chat.empty_response
多行 SSE 解析失败	事件边界聚合	补充按 SSE 空行聚合逻辑
首包前停止不够快	取消信号覆盖范围	把请求发送阶段纳入取消等待
停止后底层还在请求	是否调 Rust 取消 API	使用 cancelResponse，核对终态
chat.stream_already_active	重复启动检查	防住前端重复触发和并行流
连接关闭后标为成功	EOF 与结束标记区分	补充截断判定和回归测试
超长回复保存失败	字符上限与失败路径	按字符边界限制，验证错误终态
重启看不到历史	应用身份与数据目录	使用同一包名，区分覆盖安装与卸载重装
异常退出后 Pending 残留	中断状态处理	启动恢复需补充遗留记录处理，不能当成运行中任务

项目入口

• Nexora 源码与构建说明
• FRB 源码与 Issue
• 旋武社区 Rust 资源
• Flutter-OH 环境搭建
• VS Code Rust 环境教程

Nexora 后续可继续完善：请求发送阶段的取消、严格区分协议结束与网络截断、异常退出后的遗留状态处理、长回复的事件传输优化。扩展工具调用、本地模型等能力时，也可复用消息 ID、统一事件和持久化边界。