【Bug已解决】openclaw stream interrupted / SSE connection dropped — OpenClaw 流式传输中断解决方案 【Bug已解决】openclaw: stream interrupted / SSE connection dropped — OpenClaw 流式传输中断解决方案1. 问题描述在使用 OpenClaw 的流式输出SSE/Server-Sent Events功能时系统报出流中断或连接断开错误# 流式传输中断 $ openclaw --stream 生成长文档 Error: stream interrupted SSE connection dropped at chunk 145/200 Partial output: 7250 characters received Error: ECONNRESET at chunk 146 # SSE 连接超时 $ openclaw --stream 分析大型代码 Error: SSE timeout No data received for 30s Last activity: 2024-07-07T10:05:30Z Connection assumed dead # 流式响应不完整 $ openclaw --stream 生成报告 Warning: Stream ended prematurely Expected: [DONE] marker Received: close event without [DONE] Output may be incomplete # 断流后无法恢复 $ openclaw --stream --resume Error: Cannot resume stream Session ID: abc-123 expired Stream state lost, cannot reconnect这个问题在以下场景中特别常见生成大量内容时10K 字符网络不稳定或延迟高中间层中断长连接服务器端超时设置过短负载均衡器断开空闲连接企业安全网关拦截 SSE 流量2. 原因分析OpenClaw建立SSE连接 ↓ 服务器流式推送数据 ←──── 持续推送chunk ↓ 网络中断/超时 ←──── 连接断开 ↓ ECONNRESET/超时 → 流中断原因分类具体表现占比网络不稳定ECONNRESET约 30%服务器超时空闲断开约 25%中间层中断长连接断开约 20%负载均衡空闲超时约 10%安全网关SSE拦截约 8%客户端超时请求超时约 7%深层原理SSEServer-Sent Events是一种基于 HTTP 的流式传输协议服务器通过持久的 HTTP 连接持续推送数据块chunk。与 WebSocket 不同SSE 是单向的服务器到客户端使用Content-Type: text/event-stream。流式传输依赖于长连接保持打开状态但网络中的多个组件可能中断长连接中间层服务器如 Nginx默认超时为 60 秒负载均衡器如 AWS ALB默认空闲超时为 60 秒企业安全网关可能拦截长时间无数据的连接。当连接中断时客户端收到ECONNRESETTCP 连接被重置或超时错误导致部分输出丢失。3. 解决方案方案一实现流式重连和恢复最推荐// 创建 SSE 流式重连管理器 class SSEResumeManager { SSE 流式重连管理器 constructor(options {}) { this.maxRetries options.maxRetries || 3; this.retryDelay options.retryDelay || 2000; this.sessionId null; this.receivedChunks 0; this.receivedContent ; this.lastEventId null; } async streamRequest(url, body, onChunk) { let attempt 0; while (attempt this.maxRetries) { try { const response await this.startStream(url, body, onChunk); return response; } catch (error) { attempt; if (attempt this.maxRetries) { throw new Error( 流式传输失败已重试 ${this.maxRetries} 次。 已接收 ${this.receivedChunks} 个数据块 ${this.receivedContent.length} 字符。 错误: ${error.message} ); } console.warn( 流中断 (${attempt}/${this.maxRetries}): ${error.message} ${this.retryDelay}ms 后重连... ); await this.sleep(this.retryDelay); this.retryDelay * 2; // 指数退避 } } } async startStream(url, body, onChunk) { const headers { Content-Type: application/json, Accept: text/event-stream }; // 如果有 session ID添加恢复头 if (this.sessionId) { headers[X-Session-ID] this.sessionId; headers[X-Resume-From] this.lastEventId || 0; headers[X-Received-Chunks] String(this.receivedChunks); } const response await fetch(url, { method: POST, headers, body: JSON.stringify(body) }); // 获取 session ID this.sessionId response.headers.get(X-Session-ID); if (!response.ok) { throw new Error(HTTP ${response.status}: ${response.statusText}); } const reader response.body.getReader(); const decoder new TextDecoder(); let buffer ; try { while (true) { const { done, value } await reader.read(); if (done) { // 检查是否正常结束 if (!buffer.includes([DONE])) { throw new Error(流未正常结束缺少 [DONE] 标记); } return this.receivedContent; } buffer decoder.decode(value, { stream: true }); // 解析 SSE 事件 const events this.parseSSEEvents(buffer); buffer events.remaining; for (const event of events.parsed) { if (event.id) { this.lastEventId event.id; } if (event.data [DONE]) { return this.receivedContent; } if (event.data) { this.receivedChunks; this.receivedContent event.data; onChunk(event.data, this.receivedChunks); } } } } catch (error) { // 连接中断准备重连 throw error; } } parseSSEEvents(buffer) { const events []; const lines buffer.split(\n); let remaining ; let currentEvent {}; for (let i 0; i lines.length; i) { const line lines[i]; if (line.startsWith(id:)) { currentEvent.id line.slice(3).trim(); } else if (line.startsWith(data:)) { currentEvent.data (currentEvent.data || ) line.slice(5).trim(); } else if (line Object.keys(currentEvent).length 0) { events.push(currentEvent); currentEvent {}; } else if (i lines.length - 1) { remaining line; } } if (Object.keys(currentEvent).length 0) { events.push(currentEvent); } return { parsed: events, remaining }; } sleep(ms) { return new Promise(resolve setTimeout(resolve, ms)); } getPartialOutput() { return { content: this.receivedContent, chunks: this.receivedChunks, complete: false, sessionId: this.sessionId }; } } // 使用示例 const manager new SSEResumeManager({ maxRetries: 3, retryDelay: 2000 }); manager.streamRequest( https://api.openclaw.com/v1/chat/stream, { prompt: 生成长文档, max_tokens: 4000 }, (chunk, chunkNum) { process.stdout.write(chunk); if (chunkNum % 50 0) { console.error(\n[已接收 ${chunkNum} 个数据块]); } } ).then(fullContent { console.log(\n\n✅ 流式传输完成); }).catch(error { console.error(\n\n❌, error.message); const partial manager.getPartialOutput(); console.log(部分输出: ${partial.content.length} 字符); });方案二配置心跳和超时# 配置 SSE 心跳和超时 python3 -c import json with open(.openclaw/config.json, r) as f: config json.load(f) config[streaming] { heartbeat: { enabled: True, # 启用心跳 interval: 15000, # 15秒发送心跳 timeout: 45000 # 45秒无响应判定断开 }, timeout: { connect: 10000, # 连接超时10秒 read: 60000, # 读取超时60秒 idle: 30000, # 空闲超时30秒 total: 300000 # 总超时5分钟 }, retry: { enabled: True, # 启用重试 maxRetries: 3, # 最大重试3次 baseDelay: 1000, # 基础延迟1秒 maxDelay: 10000, # 最大延迟10秒 backoff: exponential, # 指数退避 jitter: 0.3 # 30%抖动 }, resume: { enabled: True, # 启用恢复 sessionIdHeader: X-Session-ID, resumeFromHeader: X-Resume-From, maxResumeAttempts: 3 }, buffer: { size: 65536, # 缓冲区64KB flushInterval: 100 # 100ms刷新 } } with open(.openclaw/config.json, w) as f: json.dump(config, f, indent2) print(流式配置: 15秒心跳60秒超时3次重试恢复) # 配置服务器端 SSE 支持 python3 -c import json with open(.openclaw/config.json, r) as f: config json.load(f) config[server] { sse: { keepAlive: True, # 保持连接 keepAliveInterval: 15000, # 15秒心跳 compression: True, # 压缩传输 cors: { enabled: True, origins: [*] } }, timeout: { requestTimeout: 0, # 不超时流式 headersTimeout: 60000, # 头部超时60秒 keepAliveTimeout: 65000 # 保持连接65秒 } } with open(.openclaw/config.json, w) as f: json.dump(config, f, indent2) print(服务器SSE: 心跳15秒不超时压缩) 方案三配置客户端连接超时和 keepalive# 客户端超时配置 cat ~/.openclaw/config.json EOF { streamTimeout: 300000, requestTimeout: 300000, keepalive: true, keepaliveInterval: 30, reconnect: true, maxReconnectAttempts: 5, reconnectDelay: 2000 } EOF # 环境变量设置 export OPENCLAW_STREAM_TIMEOUT300000 export OPENCLAW_REQUEST_TIMEOUT300000 export OPENCLAW_KEEPALIVEtrue export OPENCLAW_KEEPALIVE_INTERVAL30 # 永久设置 cat ~/.zshrc EOF export OPENCLAW_STREAM_TIMEOUT300000 export OPENCLAW_REQUEST_TIMEOUT300000 export OPENCLAW_KEEPALIVEtrue export OPENCLAW_KEEPALIVE_INTERVAL30 EOF source ~/.zshrc # 验证 openclaw --stream 测试任务方案四实现客户端缓冲和续传# 创建流式客户端缓冲管理器 import os import json import time import hashlib class StreamBufferManager: 流式输出缓冲管理器 def __init__(self, cache_dir.openclaw/stream_cache): self.cache_dir cache_dir os.makedirs(cache_dir, exist_okTrue) self.session_id None self.buffer [] self.checksum def start_session(self, prompt, params): 开始流式会话 self.session_id hashlib.md5( f{prompt}{time.time()}.encode() ).hexdigest()[:12] session_data { session_id: self.session_id, prompt: prompt, params: params, started_at: time.time(), status: streaming, received_chunks: 0, total_chars: 0, chunks: [] } self._save_session(session_data) return self.session_id def add_chunk(self, chunk): 添加数据块 self.buffer.append(chunk) # 定期保存 if len(self.buffer) 10: self._flush_buffer() def _flush_buffer(self): 刷新缓冲到磁盘 if not self.buffer or not self.session_id: return session self._load_session() for chunk in self.buffer: session[chunks].append({ data: chunk, timestamp: time.time(), index: session[received_chunks] }) session[received_chunks] 1 session[total_chars] len(chunk) self._save_session(session) self.buffer [] def complete(self): 标记完成 self._flush_buffer() session self._load_session() session[status] completed session[completed_at] time.time() self._save_session(session) def get_partial(self, session_idNone): 获取部分输出 sid session_id or self.session_id if not sid: return None session self._load_session(sid) if not session: return None content .join(c[data] for c in session[chunks]) return { session_id: sid, content: content, chunks: session[received_chunks], total_chars: session[total_chars], status: session[status], started_at: session[started_at] } def resume_info(self): 获取恢复信息 self._flush_buffer() session self._load_session() return { session_id: self.session_id, received_chunks: session[received_chunks], last_chunk_index: session[received_chunks] - 1, total_chars: session[total_chars] } def _session_file(self, session_idNone): sid session_id or self.session_id return os.path.join(self.cache_dir, fsession_{sid}.json) def _save_session(self, data): with open(self._session_file(), w) as f: json.dump(data, f, ensure_asciiFalse) def _load_session(self, session_idNone): filepath self._session_file(session_id) if os.path.exists(filepath): with open(filepath, r) as f: return json.load(f) return None def cleanup_old_sessions(self, max_age_hours24): 清理旧会话 now time.time() for f in os.listdir(self.cache_dir): if not f.startswith(session_): continue filepath os.path.join(self.cache_dir, f) mtime os.path.getmtime(filepath) if now - mtime max_age_hours * 3600: os.remove(filepath) print(f 清理: {f}) # 使用示例 if __name__ __main__: manager StreamBufferManager() # 开始会话 session_id manager.start_session(生成报告, {max_tokens: 4000}) print(f会话: {session_id}) # 模拟接收数据 for i in range(100): manager.add_chunk(f这是第{i1}块数据。) # 获取部分输出 partial manager.get_partial() print(f已接收: {partial[chunks]} 块, {partial[total_chars]} 字符) # 获取恢复信息 resume manager.resume_info() print(f恢复信息: {resume}) # 清理 manager.cleanup_old_sessions()方案五降级为非流式模式# 当流式不可靠时切换为非流式模式 openclaw --no-stream 任务 # 使用非流式 # 配置自动降级 python3 -c import json with open(.openclaw/config.json, r) as f: config json.load(f) config[streaming][fallback] { enabled: True, # 启用降级 onFailCount: 2, # 2次失败后降级 fallbackMode: non-stream, # 非流式 fallbackTimeout: 120000, # 非流式超时120秒 retryStreamLater: True, # 稍后重试流式 notifyUser: True # 通知用户 } with open(.openclaw/config.json, w) as f: json.dump(config, f, indent2) print(降级策略: 2次失败后切换非流式) # 使用轮询模式替代 SSE openclaw --poll-mode 任务 # 轮询模式: # 1. 发送请求 # 2. 定期轮询结果 # 3. 拼接完整输出 # 轮询模式实现 cat .openclaw/poll_mode.js JEOF // 轮询模式替代 SSE class PollModeClient { constructor(apiUrl, pollInterval 2000) { this.apiUrl apiUrl; this.pollInterval pollInterval; } async request(prompt, onProgress) { // 1. 发送请求 const startResponse await fetch(${this.apiUrl}/start, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ prompt }) }); const { taskId } await startResponse.json(); // 2. 轮询结果 let lastChunkIndex 0; let complete false; let fullContent ; while (!complete) { await new Promise(r setTimeout(r, this.pollInterval)); const pollResponse await fetch( ${this.apiUrl}/status/${taskId}?from${lastChunkIndex} ); const status await pollResponse.json(); if (status.chunks) { for (const chunk of status.chunks) { fullContent chunk.data; lastChunkIndex chunk.index 1; if (onProgress) onProgress(chunk.data, fullContent.length); } } complete status.complete; if (status.error) { throw new Error(status.error); } } return fullContent; } } module.exports PollModeClient; JEOF方案六监控和诊断# 创建流式传输监控工具 import time import json import os class StreamMonitor: 流式传输监控 def __init__(self): self.sessions [] self.stats { total_sessions: 0, completed: 0, interrupted: 0, avg_duration: 0, avg_chunks: 0, interruption_rate: 0 } def start_session(self, session_id, prompt_size): session { id: session_id, start_time: time.time(), prompt_size: prompt_size, chunks_received: 0, bytes_received: 0, interruptions: 0, retries: 0, status: streaming, events: [] } self.sessions.append(session) self.stats[total_sessions] 1 return session def record_chunk(self, session_id, chunk_size): session self._find_session(session_id) if session: session[chunks_received] 1 session[bytes_received] chunk_size def record_interruption(self, session_id, reason): session self._find_session(session_id) if session: session[interruptions] 1 session[events].append({ type: interruption, reason: reason, time: time.time(), chunks_before: session[chunks_received] }) def record_retry(self, session_id): session self._find_session(session_id) if session: session[retries] 1 def complete_session(self, session_id, success): session self._find_session(session_id) if session: session[end_time] time.time() session[duration] session[end_time] - session[start_time] session[status] completed if success else failed if success: self.stats[completed] 1 else: self.stats[interrupted] 1 self._update_stats() def _update_stats(self): total self.stats[total_sessions] if total 0: self.stats[interruption_rate] self.stats[interrupted] / total completed [s for s in self.sessions if s.get(duration)] if completed: self.stats[avg_duration] sum(s[duration] for s in completed) / len(completed) self.stats[avg_chunks] sum(s[chunks_received] for s in completed) / len(completed) def _find_session(self, session_id): for s in self.sessions: if s[id] session_id: return s return None def report(self): print( 流式传输监控报告 ) print(f总会话数: {self.stats[total_sessions]}) print(f成功完成: {self.stats[completed]}) print(f中断失败: {self.stats[interrupted]}) print(f中断率: {self.stats[interruption_rate]:.1%}) print(f平均时长: {self.stats[avg_duration]:.1f}s) print(f平均数据块: {self.stats[avg_chunks]:.0f}) # 最近的会话 print(\n--- 最近5个会话 ---) for s in self.sessions[-5:]: status ✅ if s[status] completed else ❌ duration s.get(duration, time.time() - s[start_time]) print(f {status} {s[id][:8]} | {s[chunks_received]}块 | f{duration:.1f}s | 中断:{s[interruptions]} | 重试:{s[retries]}) # 保存报告 report_file .openclaw/logs/stream_monitor.json os.makedirs(os.path.dirname(report_file), exist_okTrue) with open(report_file, w) as f: json.dump({ stats: self.stats, sessions: self.sessions[-20:] # 保留最近20个 }, f, indent2, ensure_asciiFalse, defaultstr) if __name__ __main__: monitor StreamMonitor() # 模拟监控 sid monitor.start_session(abc-123, 5000) for i in range(50): monitor.record_chunk(abc-123, 150) monitor.record_interruption(abc-123, ECONNRESET) monitor.record_retry(abc-123) for i in range(50, 100): monitor.record_chunk(abc-123, 150) monitor.complete_session(abc-123, True) monitor.report()4. 各方案对比总结方案适用场景推荐指数方案一重连恢复通用方案⭐⭐⭐⭐⭐方案二心跳超时长连接保活⭐⭐⭐⭐⭐方案三客户端超时超时配置⭐⭐⭐⭐⭐方案四缓冲续传大内容⭐⭐⭐⭐方案五降级模式不可靠环境⭐⭐⭐⭐方案六监控诊断运维⭐⭐⭐⭐5. 常见问题 FAQ5.1 Windows 上 SSE 兼容性Windows 上 curl/PowerShell 的 SSE 支持有限# PowerShell SSE 客户端 $env:OPENCLAW_STREAM_MODE powershell # 使用 .NET HttpClient python3 -c import json with open(.openclaw/config.json, r) as f: config json.load(f) config[streaming][windowsCompat] { useHttpClient: True, # 使用 HttpClient usePolling: False, # 不使用轮询 bufferSize: 8192, # 8KB缓冲 noDelay: True # 无延迟 } with open(.openclaw/config.json, w) as f: json.dump(config, f, indent2) print(Windows SSE 兼容配置) # 测试 SSE curl -N -H Accept: text/event-stream http://localhost:3000/api/stream5.2 Docker 中 SSE 超时Docker 网络配置可能影响 SSE# Docker 网络超时 docker run --network host openclaw --stream 任务 # Docker Compose 配置 services: openclaw: environment: - STREAM_TIMEOUT300000 # 5分钟 networks: - openclaw_net # 如果使用 Docker 负载均衡 # 配置 docker-compose.yml lb: image: nginx volumes: - ./nginx.conf:/etc/nginx/nginx.conf # 配置中设置 streamTimeout: 3000005.3 CI/CD 中流式输出CI 环境可能不支持 SSE# CI 中使用非流式 env: OPENCLAW_NO_STREAM: true steps: - name: Run non-stream run: openclaw --no-stream 任务 - name: Run with timeout run: | timeout 120 openclaw --stream 任务 || \ openclaw --no-stream 任务 # 降级5.4 企业安全网关拦截 SSE企业安全网关可能拦截 text/event-stream# 检测 SSE 是否被拦截 curl -v -H Accept: text/event-stream https://api.openclaw.com/test-stream 21 | head -20 # 如果被拦截使用 WebSocket 替代 python3 -c import json with open(.openclaw/config.json, r) as f: config json.load(f) config[streaming][transport] websocket # websocket | sse | polling config[streaming][fallbackChain] [sse, websocket, polling] with open(.openclaw/config.json, w) as f: json.dump(config, f, indent2) print(传输降级链: SSE - WebSocket - 轮询) # 或联系网络管理员将 API 端点加入白名单 # 确保安全网关允许 text/event-stream Content-Type 通过5.5 移动网络不稳定移动网络频繁切换导致断流# 配置移动网络优化 python3 -c import json with open(.openclaw/config.json, r) as f: config json.load(f) config[streaming][mobile] { shorterChunks: True, # 更短的数据块 maxChunkSize: 500, # 最大500字符/块 frequentHeartbeat: True, # 频繁心跳 heartbeatInterval: 5000, # 5秒心跳 aggressiveRetry: True, # 激进重试 maxRetries: 5, # 最大5次重试 quickReconnect: True, # 快速重连 reconnectDelay: 500 # 500ms重连 } with open(.openclaw/config.json, w) as f: json.dump(config, f, indent2) print(移动网络优化: 短块5秒心跳5次重试) 5.6 生成内容过长导致超时生成大量内容时总时间过长# 分段生成 openclaw --stream --max-tokens 2000 生成第一部分 openclaw --stream --max-tokens 2000 继续生成第二部分 --context previous openclaw --stream --max-tokens 2000 继续生成第三部分 --context previous # 配置自动分段 python3 -c import json with open(.openclaw/config.json, r) as f: config json.load(f) config[streaming][segmentation] { enabled: True, # 启用分段 maxTokensPerSegment: 2000, # 每段最多2000 Token continueMarker: 继续, # 继续标记 autoContinue: True, # 自动继续 maxSegments: 10, # 最大10段 mergeOutput: True # 合并输出 } with open(.openclaw/config.json, w) as f: json.dump(config, f, indent2) print(分段生成: 每段2000Token自动继续合并) 5.7 SSE 事件解析错误数据块格式不正确# 健壮的 SSE 解析器 class RobustSSEParser: 健壮的 SSE 事件解析器 def __init__(self): self.buffer self.events [] def feed(self, data): 输入数据 self.buffer data while True: # 查找完整事件以双换行结尾 event_end self._find_event_end() if event_end -1: break event_data self.buffer[:event_end] self.buffer self.buffer[event_end 2:] # 跳过双换行 event self._parse_event(event_data) if event: self.events.append(event) def _find_event_end(self): 查找事件结束位置 for i in range(len(self.buffer) - 1): if self.buffer[i] \n and self.buffer[i1] \n: return i return -1 def _parse_event(self, raw): 解析单个事件 event {} for line in raw.split(\n): line line.strip() if not line or line.startswith(:): continue # 注释行 if : in line: field, value line.split(:, 1) value value.lstrip( ) # 移除前导空格 if field id: event[id] value elif field event: event[event] value elif field data: if data in event: event[data] \n value else: event[data] value elif field retry: event[retry] int(value) return event if event else None def get_events(self): 获取已解析的事件 events self.events self.events [] return events def get_remaining(self): 获取未解析的缓冲 return self.buffer5.8 多客户端共享流多个客户端订阅同一流# 配置流式广播 python3 -c import json with open(.openclaw/config.json, r) as f: config json.load(f) config[streaming][broadcast] { enabled: True, # 启用广播 maxClients: 10, # 最大10个客户端 clientTimeout: 30000, # 客户端超时30秒 bufferSize: 100, # 缓冲100个块 lateJoinBehavior: from_start, # from_start | from_now compression: True # 压缩 } with open(.openclaw/config.json, w) as f: json.dump(config, f, indent2) print(流式广播: 最多10客户端缓冲100块) 排查清单速查表□ 1. 启用心跳: streaming.heartbeat.interval15000 □ 2. 配置重试: maxRetries3, 指数退避 □ 3. 实现流恢复: X-Session-ID X-Resume-From □ 4. streamTimeout: 300000 keepalive □ 5. AWS ALB: idle_timeout300 □ 6. 配置降级: SSE → WebSocket → 轮询 □ 7. 缓冲持久化: 防止部分输出丢失 □ 8. 监控中断率和重试次数 □ 9. 移动网络: 短块频繁心跳 □ 10. 长内容: 分段生成6. 总结最常见原因网络不稳定导致 ECONNRESET30%和服务器/中间层超时断开25%重连恢复实现 SSE 重连管理器使用 Session ID 和 Last-Event-ID 恢复中断的流心跳保活配置 15 秒心跳间隔防止连接被判定为空闲45 秒无响应判定断开客户端超时设置streamTimeout: 300000和keepalive: true确保长连接不被中断最佳实践建议实现流式缓冲持久化保存部分输出配置传输降级链SSE → WebSocket → 轮询长内容分段生成每段 2000 Token部署流式监控跟踪中断率和重试次数故障排查流程图flowchart TD A[流式中断] -- B[检查错误类型] B -- C{ECONNRESET?} C --|是| D[网络问题] C --|否| E{超时?} D -- F[启用重连恢复] E --|是| G[检查超时配置] E --|否| H{缺少DONE?} F -- I[Session-IDResume-From] G -- J[检查连接超时] H -- K[检查服务器端] I -- L[配置指数退避] J -- M[streamTimeout: 300000] K -- N[检查心跳配置] L -- O[启用心跳15秒] M -- O N -- O O -- P[openclaw --stream测试] P -- Q{成功?} Q --|是| R[✅ 问题解决] Q --|否| S[降级非流式] S -- T[openclaw --no-stream] T -- U[轮询模式替代] U -- R R -- V[部署缓冲持久化] V -- W[配置监控] W -- X[✅ 长期稳定]