AI Helper 文档中心

主题

输出流处理 ​

本文引用的源码文件

本文引用的文件

更新摘要 ​

变更内容

  • 新增统一聊天客户端 chat-client.js，提供协议抽象与自动流式/非流式处理
  • 增强错误传播机制，支持 SSE 流内错误检测与多字节字符解码改进
  • 更新流控制器架构，集成新的空闲超时常量与流式响应聚合逻辑
  • 优化多进程间通信，确保流式数据传输的一致性与可靠性

目录 ​

  1. 引言
  2. 项目结构
  3. 核心组件
  4. 架构总览
  5. 详细组件分析
  6. 依赖关系分析
  7. 性能与内存保护
  8. 故障排查指南
  9. 结论

引言 ​

本文面向"输出流处理机制"，围绕以下目标展开：

  • 标准输出捕获：stdout/stderr 分离、实时流式传输、缓冲区管理与内存限制。
  • WebSocket 消息广播：多客户端支持、连接状态监听、断线重连与回放。
  • 输出缓冲策略：5MB 上限截断、增量追加算法、截断标记通知与回放能力。
  • 流式数据传输：数据分片、顺序保证、错误处理与性能优化。
  • 统一聊天客户端：协议抽象、自动流式/非流式处理、增强的错误传播与多字节字符解码。
  • 提供流处理架构图与 WebSocket 消息格式规范，帮助扩展端与 Agent 端开发者对齐协议与实现。

本项目为双进程架构：Chrome 扩展端（Manifest V3，src/）与独立本地代理端 Agent（Node.js，agent/src/server.js）。两端通过 HTTP/WebSocket 通信；Agent 负责命令执行、子进程 stdout/stderr 流式输出与 WebSocket 广播；扩展端 background 负责统一聊天客户端调用、SSE 解析、节流发送、工具调用检测与 Agent 命令流转发。

项目结构 ​

与输出流处理直接相关的代码分布在两个进程中：

  • Agent 端
    • agent/src/server.js：HTTP + WebSocket 服务器，路由命令执行、WebSocket 连接生命周期管理。
    • agent/src/executor.js：命令执行器，封装 child_process 子进程、stdout/stderr 流、超时策略、缓冲区与广播。
  • 扩展端
    • src/shared/chat-client.js：统一聊天补全客户端，提供协议抽象、自动流式/非流式处理、SSE 聚合与错误处理。
    • src/background/react-loop.js：ReAct 推理循环，集成 StreamController 进行流式响应处理。
    • src/background/tool-executor.js：工具执行层，负责将 Agent 命令的 stdout/stderr 转换为扩展侧消息并转发到 UI。
graph TB
subgraph "扩展端 Background"
TC["tool-executor.js<br/>工具执行层"]
RL["react-loop.js<br/>ReAct 推理循环"]
CC["chat-client.js<br/>统一聊天客户端"]
end
subgraph "Agent 端"
SV["server.js<br/>HTTP + WebSocket 服务器"]
EX["executor.js<br/>命令执行器"]
CP["child_process<br/>子进程 stdout/stderr"]
end
UI["Side Panel / Content Script"]
API["外部 LLM 服务"]
UI --> TC
TC --> RL
RL --> CC
CC --> API
TC --> SV
SV --> EX
EX --> CP
EX --> SV
SV --> UI

图表来源

章节来源

核心组件 ​

  • Agent 命令执行器 executor.js
    • 负责 spawn 子进程、绑定 stdout/stderr 事件、维护运行中进程映射、超时策略、输出缓冲区与 WebSocket 广播。
  • Agent WebSocket 服务器 server.js
    • 负责 HTTP 路由、WebSocket 握手、连接加入/断开、命令执行接口、进程控制接口。
  • 统一聊天客户端 chat-client.js
    • 提供统一的聊天补全调用接口，自动处理流式/非流式协议，支持 SSE 聚合、错误传播与多字节字符解码。
  • ReAct 推理循环 react-loop.js
    • 集成 StreamController 进行流式响应处理，支持工具调用检测、上下文压缩与错误恢复。
  • 扩展端工具执行层 tool-executor.js
    • 负责将 Agent 命令的 stdout/stderr 流转换为扩展侧消息，并通过 chrome.runtime.sendMessage 推送给 UI。

章节来源

架构总览 ​

输出流处理贯穿两条主线：

  • LLM 推理流：扩展端通过统一聊天客户端 chat-client.js 或 react-loop.js 中的 StreamController 读取外部 LLM 的 SSE 流，解析 content/tool_calls/usage，节流发送到 Side Panel。
  • Agent 命令流：扩展端通过 tool-executor.js 调用 Agent 的 HTTP/WebSocket 接口，Agent 端 executor.js 启动子进程并将 stdout/stderr 实时广播给所有连接的 WebSocket 客户端。
sequenceDiagram
participant UI as "Side Panel"
participant BG as "Background(tool-executor)"
participant RL as "ReactLoop"
participant CC as "ChatClient"
participant API as "LLM 服务"
participant AG as "Agent(server.js)"
participant EX as "Executor(executor.js)"
participant CP as "子进程(stdout/stderr)"
Note over UI,API : LLM 推理流
UI->>BG : 发起 LLM 请求
BG->>RL : 创建 ReAct 循环
RL->>CC : 统一聊天客户端调用
CC->>API : 建立 SSE 流自动流式/非流式
API-->>CC : data : {choices : [{delta : {content}}]}
CC-->>RL : 聚合后的完整响应
RL-->>UI : STREAM_START / STREAM_CHUNK / STREAM_DONE
Note over UI,CP : Agent 命令流
UI->>BG : 执行 agent_exec(command)
BG->>AG : POST /api/exec (command, timeoutMs)
AG->>EX : executeCommand(...)
EX->>CP : spawn(shell, command)
CP-->>EX : stdout/stderr 数据块
EX-->>AG : broadcast({type : 'stdout'|'stderr', data})
AG-->>BG : WebSocket 推送
BG-->>UI : AGENT_STREAM / AGENT_STREAM_DONE

图表来源

详细组件分析 ​

统一聊天客户端（chat-client.js） ​

新增 统一聊天客户端提供了协议抽象层，自动处理流式与非流式响应，增强了错误传播和多字节字符解码能力。

  • 协议抽象与自动处理
    • callChatCompletion 函数根据 useStream 参数自动选择流式或非流式协议。
    • applyStreamToBody 智能修改请求体，添加 stream/stream_options 字段。
    • isEventStreamResponse 检测响应是否为 SSE 流，支持网关兼容性。
  • SSE 流聚合与解码
    • readChatSSE 函数静默聚合 SSE 流为 OpenAI 非流式形状响应对象。
    • 使用 TextDecoder 进行 UTF-8 解码，支持多字节字符正确处理。
    • 空闲超时保护：DEFAULT_STREAM_IDLE_TIMEOUT_MS = 120000ms。
  • 错误传播增强
    • SSE 流内错误检测：当检测到 parsed.error 时抛出带后端原因的异常。
    • 非 2xx 响应时读取并回传响应正文，避免吞掉真实错误信息。
    • 支持自定义 fetchImpl 以便注入重试逻辑。
flowchart TD
Start(["收到聊天补全请求"]) --> CheckStream{"useStream?"}
CheckStream --> |是| ApplyStream["applyStreamToBody 添加 stream/stream_options"]
CheckStream --> |否| NoStream["显式设置 stream:false"]
ApplyStream --> Fetch["fetch API 调用"]
NoStream --> Fetch
Fetch --> CheckOK{"response.ok?"}
CheckOK --> |否| ErrorText["读取 errorText 并返回"]
CheckOK --> |是| CheckSSE{"isEventStreamResponse?"}
CheckSSE --> |是| ReadSSE["readChatSSE 聚合 SSE 流"]
CheckSSE --> |否| ParseJSON["解析 JSON 响应"]
ReadSSE --> Aggregate["聚合 choices 和 usage"]
ParseJSON --> Return["返回标准化响应"]
Aggregate --> Return
ErrorText --> Return
Return --> End(["完成"])

图表来源

章节来源

Agent 命令执行器（executor.js） ​

  • 子进程与流绑定
    • 使用 child_process.spawn 启动 shell 命令，stdio 配置为 ['ignore', 'pipe', 'pipe']，分别对应 stdin 忽略、stdout/stderr 管道。
    • 对 proc.stdout 与 proc.stderr 分别注册 data 事件，实现 stdout/stderr 分离处理。
  • 输出缓冲区与内存限制
    • 每个运行中的进程 entry 包含 stdoutBuf 与 stderrBuf，始终收集用于回放给延迟连接的 WS 客户端。
    • 定义 MAX_OUTPUT_BUFFER = 5 * 1024 * 1024（5MB），通过 appendWithLimit 实现增量追加与上限截断。
    • 当追加导致超过上限时，设置 truncStdout/truncStderr 标志，并向所有 WS 客户端广播 stdout_truncated 或 stderr_truncated 消息。
  • 超时策略
    • 受管模式：由 decideTimeoutAction 根据 outputInPeriod、extensions、elapsedMs、baseTimeout 决定 extend 或 kill。
    • 旧模式：到期直接 killProcess。
    • outputInPeriod 在每次 stdout/stderr 数据到达时置为 true，作为"心跳"判断是否静默挂起。
  • WebSocket 广播与多客户端
    • 每个 execId 对应一个 wsClients Set，broadcast 遍历 Set 向所有 readyState=1 的客户端发送 JSON 消息。
    • addWsClient 支持新连接加入运行中进程，或在进程已结束的情况下从 completedProcesses 缓存回放全部输出与结束状态。
    • disconnectWsClient 在所有客户端断开后，仅在非受管模式或已 detach 时才取消超时定时器，避免插件端 SW 被 Chrome 终止导致的误判。
  • 进程终止
    • killProcess 跨平台递归终止进程组：Unix 使用 process.kill(-pid)，Windows 使用 taskkill /T /PID，并在 5 秒后强制 SIGKILL。
flowchart TD
Start(["收到 stdout/stderr 数据块"]) --> ToStr["转换为字符串"]
ToStr --> MarkOutput["标记 outputInPeriod=true"]
MarkOutput --> CollectOptional{"collectOutput?"}
CollectOptional --> |是| AppendFull["追加到完整收集变量"]
CollectOptional --> |否| SkipFull["跳过完整收集"]
AppendFull --> TruncCheck["appendWithLimit 检查 5MB 上限"]
SkipFull --> TruncCheck
TruncCheck --> IsTruncated{"是否触发截断?"}
IsTruncated --> |是| SetFlag["设置 truncStdout/truncStderr"]
IsTruncated --> |否| KeepFlag["保持原标志"]
SetFlag --> BroadcastTrunc["广播 stdout_truncated / stderr_truncated"]
KeepFlag --> BroadcastData["广播 stdout / stderr 数据"]
BroadcastTrunc --> BroadcastData
BroadcastData --> End(["完成"])

图表来源

章节来源

Agent WebSocket 服务器（server.js） ​

  • HTTP 路由
    • 提供 /api/exec 等接口，调用 executor.executeCommand 或 executeCommandSync。
    • 提供 /api/exec/stop 与 /api/exec/detach 接口，分别调用 killProcess 与 detachProcess。
  • WebSocket 连接管理
    • 当客户端连接 /ws 时，若携带 execId，则尝试 addWsClient 将其加入对应进程的 wsClients。
    • 若进程已结束，completedProcesses 缓存会回放 stdout/stderr/error/exit 消息。
    • onclose/onerror 时调用 disconnectWsClient，清理 wsClients 并根据规则决定是否取消超时定时器。
  • 语言与国际化
    • 在请求入口设置 setExecutorLang(lang)，使 executor 内部 translate 返回正确语言文案（如 stdoutTruncated、stderrTruncated）。

章节来源

ReAct 推理循环（react-loop.js） ​

  • StreamController 集成
    • 在流式模式下创建 StreamController 实例，处理 SSE 流解析和内容聚合。
    • 支持工具调用检测，自动规范化 tool_calls ID 以确保与工具消息匹配。
    • 集成空闲超时保护，区分用户取消、流式空闲超时和真正的 API 错误。
  • 上下文自愈机制
    • 检测到上下文超限时自动裁剪消息并重试，最多 CONTEXT_HEAL_MAX_ATTEMPTS 次。
    • 按 CONTEXT_HEAL_RATIOS 逐档更激进地裁剪，无法再裁时抛出错误。
  • 流式与非流式统一处理
    • 无论流式与否，最终都构建标准化的 response 对象，包含 choices、usage 等信息。
    • 非流式模式下同样规范化 tool_calls ID，确保前端渲染一致性。

章节来源

扩展端工具执行层（tool-executor.js） ​

  • Agent 命令执行流程
    • 通过 AgentClient 调用 /api/exec，获取 execId。
    • 建立 WebSocket 连接，订阅 stdout/stderr/error/exit 消息。
    • 使用 sendAgentStream/sendAgentStreamDone 将流式数据转发给 Side Panel。
  • 会话级命令跟踪
    • runningAgentCommands 记录 sessionId → {execId, ws, resolve}，用于用户取消任务时关闭 WebSocket，防止旧命令输出污染新任务。
    • cancelledSessions 记录已取消会话，避免 onclose/onerror 重复清理。

章节来源

依赖关系分析 ​

  • executor.js 依赖
    • child_process.spawn：启动子进程。
    • fs.existsSync：shell 路径存在性检查。
    • crypto.randomBytes：生成 execId。
    • os.platform/os.homedir：平台与环境适配。
    • i18n.translate：截断提示文案国际化。
  • server.js 依赖
    • http + ws：HTTP 服务器与 WebSocketServer。
    • executor.js：命令执行、WS 客户端管理、进程控制。
    • security/i18n/config：安全校验、语言、配置加载。
  • chat-client.js 依赖
    • TextDecoder：UTF-8 文本解码，支持多字节字符。
    • fetch：网络请求，支持 AbortSignal 取消。
    • chrome.storage.local：读取流式开关配置。
  • react-loop.js 依赖
    • StreamController：流式响应处理。
    • chat-client.js：统一聊天客户端调用。
    • 各种工具模块：token-counter、logger、i18n 等。
  • tool-executor.js 依赖
    • AgentClient：调用 Agent HTTP/WebSocket 接口。
    • stream-controller.sendAgentStream/sendAgentStreamDone：统一转发 Agent 命令流。
graph LR
EX["executor.js"] --> CP["child_process"]
EX --> I18N["i18n.translate"]
SV["server.js"] --> EX
SV --> SEC["security.js"]
SV --> CFG["config.js"]
CC["chat-client.js"] --> TEXTDEC["TextDecoder"]
CC --> FETCH["fetch"]
CC --> STORAGE["chrome.storage.local"]
RL["react-loop.js"] --> SC["StreamController"]
RL --> CC
TE["tool-executor.js"] --> AC["AgentClient"]
TE --> SC

图表来源

章节来源

性能与内存保护 ​

  • 输出缓冲区上限
    • 5MB 上限通过 appendWithLimit 实现，避免长时间命令导致 Node.js 堆内存无限增长。
    • 超过上限时仅追加剩余空间，并设置 truncated 标志，后续追加不再修改缓冲区。
  • 增量追加算法
    • 每次 stdout/stderr 数据块先 toString，再增量追加到 stdoutBuf/stderrBuf。
    • collectOutput 可选开启完整收集，用于同步执行场景返回完整输出。
  • 流式响应聚合优化
    • chat-client.js 的 readChatSSE 使用 TextDecoder 流式解码，避免大字符串拼接开销。
    • 支持 idleTimeoutMs 配置，默认 120s 空闲超时，防止资源泄漏。
  • 节流与合并
    • react-loop.js 中的 StreamController 使用 _pendingChunk 累积内容，按 streamChunkDelay 节流发送，减少 UI 渲染压力。
  • 空闲超时保护
    • readChatSSE 在持续无数据时触发 CHAT_STREAM_IDLE_TIMEOUT 错误，主动 cancel 底层 reader，释放连接资源。
  • 进程终止与资源回收
    • killProcess 跨平台递归终止进程组，并在 5 秒后强制 SIGKILL，防止孤儿进程。
    • completedProcesses 缓存 30 秒后自动清理，避免长期占用内存。

章节来源

故障排查指南 ​

  • 输出未显示或中断
    • 检查 Agent 端 executor.js 的 stdout/stderr 事件是否触发。
    • 确认 WebSocket 连接状态，server.js 的 addWsClient/disconnectWsClient 是否正确管理 wsClients。
    • 检查扩展端 tool-executor.js 是否成功建立 WebSocket 并转发 AGENT_STREAM/AGENT_STREAM_DONE。
  • 输出被截断
    • 查看 stdout_truncated/stderr_truncated 消息，确认是否触发 5MB 上限。
    • 检查 truncStdout/truncStderr 标志是否正确设置。
  • 命令超时或卡住
    • 检查 decideTimeoutAction 的输出 InPeriod 逻辑，确认是否有 stdout/stderr 心跳。
    • 验证受管模式与非受管模式的超时行为差异。
  • SSE 流空闲超时
    • 检查 chat-client.js 的 readChatSSE 的 idleTimer 与 Promise.race 逻辑，确认是否持续收到数据。
    • 查看 CHAT_STREAM_IDLE_TIMEOUT 错误日志，确认是否需要调整 DEFAULT_STREAM_IDLE_TIMEOUT_MS。
  • 多字节字符乱码
    • 确认 chat-client.js 正确使用 TextDecoder 进行 UTF-8 解码。
    • 检查 decoder.decode(value, { stream: true }) 是否正确处理跨块的多字节字符。
  • 协议不匹配错误
    • 检查 chat-client.js 的 applyStreamToBody 是否正确添加 stream/stream_options 字段。
    • 确认 isEventStreamResponse 能正确识别 SSE 响应类型。

章节来源

结论 ​

本项目的输出流处理机制通过 Agent 端的 executor.js 与 server.js 实现子进程 stdout/stderr 的分离捕获、实时广播与内存保护；通过扩展端的统一聊天客户端 chat-client.js 与 react-loop.js 实现 LLM 协议的抽象化、自动流式/非流式处理与增强的错误传播。整体设计在多客户端支持、断线回放、超时策略与内存限制方面具备较强的健壮性，同时通过统一聊天客户端简化了协议处理复杂度，适合复杂命令执行与长时任务的实时反馈场景。

新增的统一聊天客户端显著提升了系统的可维护性和扩展性，通过协议抽象层屏蔽了底层实现细节，使得上层业务逻辑无需关心具体的流式或非流式处理差异。增强的错误传播机制确保了错误的准确定位和处理，而改进的多字节字符解码则保证了国际化的用户体验。