From ba4c2335094a38f44927cc61ba2a8eb6bd57f947 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 11 Apr 2026 00:55:39 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20COS=E8=AE=AD=E7=BB=83Agent=E5=A4=A7?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E5=B4=A9=E6=BA=83=20=E2=80=94=20=E6=B7=BB?= =?UTF-8?q?=E5=8A=A0Buffer=E5=AE=89=E5=85=A8=E5=A4=84=E7=90=86=E3=80=81HEA?= =?UTF-8?q?D=E9=A2=84=E6=A3=80=E3=80=81=E5=88=86=E5=9D=97=E6=8F=90?= =?UTF-8?q?=E5=8F=96=E7=AD=96=E7=95=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修复两个训练失败根因: 1. conversations.json (665MB) 超过Node.js字符串限制导致ERR_STRING_TOO_LONG崩溃 2. 单文件崩溃导致整个训练管线中断，后续文件无法处理 - cos.js: 添加readBuffer/head/readPartial方法，cosRequest增加try/catch防止未捕获异常 - corpus-extractor-ops.js: 超大文件(>200MB)采用分块采样策略，ZIP文件提前检测避免无效读取 - cos-training-trigger.js: 增加文件大小预警日志，分类计数(成功/跳过/失败) Agent-Logs-Url: https://github.com/qinfendebingshuo/guanghulab/sessions/f10f1b02-ed4e-4ee8-ac27-0580c1cf8a7e Co-authored-by: qinfendebingshuo <207279273+qinfendebingshuo@users.noreply.github.com> --- scripts/cos-training-trigger.js | 39 ++++- server/age-os/mcp-server/cos.js | 165 +++++++++++++++++- .../mcp-server/tools/corpus-extractor-ops.js | 113 ++++++++++-- 3 files changed, 289 insertions(+), 28 deletions(-) diff --git a/scripts/cos-training-trigger.js b/scripts/cos-training-trigger.js index b583a8ff..f7764b82 100644 --- a/scripts/cos-training-trigger.js +++ b/scripts/cos-training-trigger.js @@ -50,6 +50,7 @@ const DEFAULT_PERSONA = 'zhuyuan'; const PROCESSED_PREFIX = 'tcs-structured/'; const MAX_EXTRACT_PER_RUN = 20; // 每次最多处理文件数 const MAX_TRAIN_PER_RUN = 5; // 每次最多训练文件数 +const MAX_EXTRACT_FILE_SIZE = 200 * 1024 * 1024; // 200MB — 超过此阈值的文件使用分块策略 // ─── 排除路径（不视为语料的目录/文件） ─── const EXCLUDED_PREFIXES = [ @@ -221,20 +222,44 @@ async function cmdExtract(bucket) { let extracted = 0; let errors = 0; + let skipped = 0; const results = []; for (const file of toProcess) { try { - console.log(` 📦 处理: ${file.key}...`); + // 大文件预警 + const sizeMB = (file.size_bytes / 1024 / 1024).toFixed(1); + if (file.size_bytes > MAX_EXTRACT_FILE_SIZE) { + console.log(` 📦 处理: ${file.key} (${sizeMB}MB · 超大文件，使用分块策略)...`); + } else { + console.log(` 📦 处理: ${file.key} (${sizeMB}MB)...`); + } + const result = await extractor.cosExtractCorpus({ bucket: bucketName, key: file.key, output_bucket: bucketName, output_prefix: PROCESSED_PREFIX }); - extracted++; - results.push({ key: file.key, status: 'success', output: result.output?.key }); - console.log(` ✅ 完成: ${result.output?.key || '已处理'} (${result.entries || 0} 条目)`); + + // 根据返回状态分类计数 + if (result.status === 'zip_detected') { + skipped++; + results.push({ key: file.key, status: 'skipped', reason: 'zip_needs_special_tool' }); + console.log(` ⏭️ 跳过: ${file.key} — ZIP文件需要专用工具处理`); + } else if (result.status === 'skipped_too_large') { + skipped++; + results.push({ key: file.key, status: 'skipped', reason: 'too_large', size_mb: result.size_mb }); + console.log(` ⏭️ 跳过: ${file.key} — ${result.message}`); + } else if (result.status === 'partial_extract') { + extracted++; + results.push({ key: file.key, status: 'partial', output: result.output?.key, message: result.message }); + console.log(` 🔶 部分提取: ${result.message}`); + } else { + extracted++; + results.push({ key: file.key, status: 'success', output: result.output?.key }); + console.log(` ✅ 完成: ${result.output?.key || '已处理'} (${result.entries || 0} 条目)`); + } } catch (err) { errors++; results.push({ key: file.key, status: 'error', error: err.message }); @@ -244,16 +269,18 @@ async function cmdExtract(bucket) { console.log(`\n═══ 提取完毕 ═══`); console.log(`✅ 成功: ${extracted}`); + console.log(`⏭️ 跳过: ${skipped}`); console.log(`❌ 失败: ${errors}`); console.log(`⏳ 剩余: ${pending.length - toProcess.length}`); writeGitHubOutput( `extracted=${extracted}`, + `extract_skipped=${skipped}`, `extract_errors=${errors}`, `extract_status=${errors > 0 ? 'partial' : 'success'}` ); - return { extracted, errors, results }; + return { extracted, errors, skipped, results }; } // ═══════════════════════════════════════════ @@ -379,7 +406,7 @@ async function cmdFull(bucket, personaId) { console.log('\n╔═══════════════════════════════════════════╗'); console.log('║ 完整训练管线 · 运行完毕 ║'); console.log('╚═══════════════════════════════════════════╝'); - console.log(` 提取: ${extractResult.extracted} 成功 / ${extractResult.errors} 失败`); + console.log(` 提取: ${extractResult.extracted} 成功 / ${extractResult.skipped || 0} 跳过 / ${extractResult.errors} 失败`); console.log(` 训练: ${trainResult.trained} 成功 / ${trainResult.errors} 失败`); console.log(` 耗时: ${(duration / 1000).toFixed(1)}s`); diff --git a/server/age-os/mcp-server/cos.js b/server/age-os/mcp-server/cos.js index 68bda704..a3dcf5a4 100644 --- a/server/age-os/mcp-server/cos.js +++ b/server/age-os/mcp-server/cos.js @@ -115,14 +115,24 @@ async function personaList(personaId, subPrefix, limit) { return list('team', prefix, limit); } +// ─── 大文件安全阈值 ─── +// Node.js字符串最大长度约 512MB (0x1fffffe8 字符) +// 为安全起见，超过此阈值的文件使用 Buffer 模式读取 +const MAX_STRING_SAFE_BYTES = 400 * 1024 * 1024; // 400MB + /** * 发起 COS HTTP 请求 * * 修复: 签名时必须将URI路径和查询参数分开处理。 * 腾讯云COS签名规范要求 HttpURI 不包含查询字符串(query string), * 否则签名哈希不匹配导致 SignatureDoesNotMatch 错误。 + * + * @param {object} options - 可选配置 + * @param {number} options.timeout - 超时时间(毫秒)，默认30000 + * @param {boolean} options.rawBuffer - 是否返回原始 Buffer 而非字符串 */ -function cosRequest(bucketName, objectKey, method, body, contentType) { +function cosRequest(bucketName, objectKey, method, body, contentType, options) { + const { timeout = 30000, rawBuffer = false } = options || {}; return new Promise((resolve, reject) => { const host = getBucketHost(bucketName); const fullPath = '/' + objectKey; @@ -141,16 +151,45 @@ function cosRequest(bucketName, objectKey, method, body, contentType) { headers['Authorization'] = generateSignature(method, signPathname, host); const req = https.request({ - hostname: host, port: 443, path: fullPath, method, headers, timeout: 30000 + hostname: host, port: 443, path: fullPath, method, headers, timeout }, (res) => { + // HEAD 请求无响应体 + if (method === 'HEAD') { + if (res.statusCode >= 200 && res.statusCode < 300) { + resolve({ statusCode: res.statusCode, body: '', headers: res.headers }); + } else { + reject(new Error(`COS ${method} ${fullPath}: ${res.statusCode}`)); + } + return; + } + const chunks = []; res.on('data', c => chunks.push(c)); res.on('end', () => { - const responseBody = Buffer.concat(chunks).toString(); - if (res.statusCode >= 200 && res.statusCode < 300) { - resolve({ statusCode: res.statusCode, body: responseBody, headers: res.headers }); - } else { - reject(new Error(`COS ${method} ${fullPath}: ${res.statusCode} - ${responseBody.substring(0, 200)}`)); + try { + const buffer = Buffer.concat(chunks); + + if (res.statusCode >= 200 && res.statusCode < 300) { + if (rawBuffer) { + resolve({ statusCode: res.statusCode, body: buffer, headers: res.headers }); + } else if (buffer.length > MAX_STRING_SAFE_BYTES) { + reject(new Error( + `文件过大 (${(buffer.length / 1024 / 1024).toFixed(1)}MB)，` + + `超过安全字符串转换阈值 (${MAX_STRING_SAFE_BYTES / 1024 / 1024}MB)。` + + `请使用 readBuffer() 方法读取大文件。` + )); + } else { + const responseBody = buffer.toString('utf8'); + resolve({ statusCode: res.statusCode, body: responseBody, headers: res.headers }); + } + } else { + const errorBody = buffer.length > MAX_STRING_SAFE_BYTES + ? buffer.slice(0, 500).toString('utf8') + : buffer.toString('utf8'); + reject(new Error(`COS ${method} ${fullPath}: ${res.statusCode} - ${errorBody.substring(0, 200)}`)); + } + } catch (err) { + reject(new Error(`COS响应处理失败: ${err.message}`)); } }); }); @@ -182,6 +221,114 @@ async function read(bucket, key) { }; } +/** + * 读取文件为 Buffer（不做字符串转换，支持任意大小） + */ +async function readBuffer(bucket, key) { + const bucketName = resolveBucketName(bucket); + const result = await cosRequest(bucketName, key, 'GET', null, null, { + rawBuffer: true, + timeout: 120000 // 大文件给2分钟超时 + }); + return { + buffer: result.body, + size_bytes: result.body.length, + last_modified: result.headers['last-modified'] || null + }; +} + +/** + * HEAD请求 — 获取文件元数据（大小/类型等），不下载内容 + */ +async function head(bucket, key) { + const bucketName = resolveBucketName(bucket); + const result = await cosRequest(bucketName, key, 'HEAD'); + return { + size_bytes: parseInt(result.headers['content-length'] || '0', 10), + content_type: result.headers['content-type'] || null, + last_modified: result.headers['last-modified'] || null, + etag: result.headers['etag'] || null + }; +} + +/** + * 分块读取大文件 — 仅读取前 N 字节用于预览/采样 + * 适用于超大语料文件的类型检测和内容预览 + */ +async function readPartial(bucket, key, maxBytes) { + const limit = maxBytes || 1024 * 1024; // 默认1MB + const bucketName = resolveBucketName(bucket); + const host = getBucketHost(bucketName); + const fullPath = '/' + key; + + const qIdx = fullPath.indexOf('?'); + const signPathname = qIdx >= 0 ? fullPath.substring(0, qIdx) : fullPath; + + return new Promise((resolve, reject) => { + const headers = { + Host: host, + Range: `bytes=0-${limit - 1}`, + Authorization: generateSignature('GET', signPathname, host) + }; + + const req = https.request({ + hostname: host, port: 443, path: fullPath, method: 'GET', headers, timeout: 30000 + }, (res) => { + const chunks = []; + let received = 0; + res.on('data', c => { + chunks.push(c); + received += c.length; + if (received >= limit) { + res.destroy(); // 已收到足够数据，停止接收 + } + }); + res.on('end', () => { + try { + const buffer = Buffer.concat(chunks); + const totalSize = parseInt(res.headers['content-range']?.split('/')[1] || '0', 10) + || parseInt(res.headers['content-length'] || '0', 10); + resolve({ + content: buffer.toString('utf8'), + size_bytes: buffer.length, + total_size_bytes: totalSize, + is_partial: totalSize > buffer.length + }); + } catch (err) { + reject(new Error(`COS partial read failed: ${err.message}`)); + } + }); + res.on('close', () => { + // 如果因 destroy() 而关闭，也处理已收到的数据 + if (chunks.length > 0) { + try { + const buffer = Buffer.concat(chunks); + const totalSize = parseInt(res.headers['content-range']?.split('/')[1] || '0', 10) + || parseInt(res.headers['content-length'] || '0', 10); + resolve({ + content: buffer.toString('utf8'), + size_bytes: buffer.length, + total_size_bytes: totalSize, + is_partial: true + }); + } catch (err) { + reject(new Error(`COS partial read failed: ${err.message}`)); + } + } + }); + }); + req.on('error', (err) => { + if (err.code === 'ERR_STREAM_PREMATURE_CLOSE') { + // 预期行为：主动关闭连接 + return; + } + reject(err); + }); + req.on('timeout', () => { req.destroy(); reject(new Error('COS request timeout')); }); + req.end(); + }); +} + async function del(bucket, key) { const bucketName = resolveBucketName(bucket); await cosRequest(bucketName, key, 'DELETE'); @@ -239,8 +386,8 @@ async function checkConnection() { } module.exports = { - write, read, del, list, archive, checkConnection, + write, read, readBuffer, head, readPartial, del, list, archive, checkConnection, personaWrite, personaRead, personaList, validatePersonaCosPath, resolveBucketName, - COS_CONFIG + COS_CONFIG, MAX_STRING_SAFE_BYTES }; diff --git a/server/age-os/mcp-server/tools/corpus-extractor-ops.js b/server/age-os/mcp-server/tools/corpus-extractor-ops.js index 9764111f..a00386a8 100644 --- a/server/age-os/mcp-server/tools/corpus-extractor-ops.js +++ b/server/age-os/mcp-server/tools/corpus-extractor-ops.js @@ -28,6 +28,12 @@ const cos = require('../cos'); const gunzip = promisify(zlib.gunzip); const inflate = promisify(zlib.inflate); +// ─── 大文件处理阈值 ─── +// 超过此阈值的文件使用分块策略（读取部分内容做类型检测和采样） +const MAX_EXTRACT_FILE_SIZE = 200 * 1024 * 1024; // 200MB +// 分块采样时读取的最大字节数 +const PARTIAL_READ_SIZE = 2 * 1024 * 1024; // 2MB + // ─── 支持的压缩格式 ─── const COMPRESSED_EXTENSIONS = ['.zip', '.gz', '.tar.gz', '.tgz', '.json.gz']; @@ -161,7 +167,33 @@ async function cosExtractCorpus(input) { const outputBucket = output_bucket || bucket; const outputBase = output_prefix || 'tcs-structured/'; - // 读取源文件 + // ── 大文件预检: 先用 HEAD 检查文件大小，避免 OOM ── + let fileSize = 0; + try { + const meta = await cos.head(bucket, key); + fileSize = meta.size_bytes; + } catch { + // HEAD 不支持时跳过，由后续读取处理 + } + + // ZIP文件始终返回提示（无论大小），因为需要二进制解析 + if (key.toLowerCase().endsWith('.zip')) { + return { + status: 'zip_detected', + key, + size_bytes: fileSize, + message: 'ZIP文件已检测到。ZIP解析需要完整二进制流处理，建议使用cosParseGitRepoArchive或cosParseNotionExport专用工具处理。', + corpus_type: detectCorpusType(key), + recommendation: '请使用专用解析工具处理ZIP文件' + }; + } + + // ── 超大文件分块处理策略 ── + if (fileSize > MAX_EXTRACT_FILE_SIZE) { + return await extractLargeFile(bucket, key, outputBucket, outputBase, fileSize); + } + + // ── 常规文件: 完整读取 ── const raw = await cos.read(bucket, key); let content = raw.content; let decompressed = false; @@ -178,18 +210,6 @@ async function cosExtractCorpus(input) { } } - // ZIP文件需要特殊处理（ZIP解析需要外部库，这里提取目录列表） - if (key.toLowerCase().endsWith('.zip')) { - return { - status: 'zip_detected', - key, - size_bytes: raw.size_bytes, - message: 'ZIP文件已检测到。ZIP解析需要完整二进制流处理，建议使用cosParseGitRepoArchive或cosParseNotionExport专用工具处理。', - corpus_type: detectCorpusType(key), - recommendation: '请使用专用解析工具处理ZIP文件' - }; - } - // 自动检测语料类型 const corpusType = detectCorpusType(key, content); @@ -224,6 +244,73 @@ async function cosExtractCorpus(input) { }; } +/** + * 超大文件分块提取策略 + * + * 对于超过 MAX_EXTRACT_FILE_SIZE 的文件: + * 1. 使用 readPartial 读取前 2MB 做类型检测 + * 2. 根据类型采样提取部分内容生成TCS + * 3. 标记为 partial_extract 状态 + */ +async function extractLargeFile(bucket, key, outputBucket, outputBase, fileSize) { + const sizeMB = (fileSize / 1024 / 1024).toFixed(1); + + // 读取前 2MB 用于类型检测和内容采样 + let partial; + try { + partial = await cos.readPartial(bucket, key, PARTIAL_READ_SIZE); + } catch (err) { + return { + status: 'skipped_too_large', + key, + size_bytes: fileSize, + size_mb: sizeMB, + message: `文件过大 (${sizeMB}MB)，超过提取阈值 (${MAX_EXTRACT_FILE_SIZE / 1024 / 1024}MB)，且分块读取失败: ${err.message}`, + recommendation: '建议将大文件拆分后重新上传到COS桶' + }; + } + + const corpusType = detectCorpusType(key, partial.content); + + // 用采样内容生成TCS（标记为部分提取） + let tcsData; + switch (corpusType) { + case 'gpt': + tcsData = transformGPTToTCS(partial.content, key); + break; + case 'notion': + tcsData = transformNotionToTCS(partial.content, key); + break; + case 'git-repo': + tcsData = transformGitRepoToTCS(partial.content, key); + break; + default: + tcsData = transformGenericToTCS(partial.content, key); + } + + // 标记为部分提取 + tcsData.metadata = tcsData.metadata || {}; + tcsData.metadata.partial_extract = true; + tcsData.metadata.total_file_size_bytes = fileSize; + tcsData.metadata.sampled_bytes = partial.size_bytes; + tcsData.metadata.extraction_note = `文件过大 (${sizeMB}MB)，仅采样前 ${(partial.size_bytes / 1024 / 1024).toFixed(1)}MB 生成TCS。完整提取需要流式处理。`; + + const outputKey = `${outputBase}${corpusType}/${extractFileName(key)}.partial.tcs.json`; + await cos.write(outputBucket, outputKey, JSON.stringify(tcsData, null, 2), 'application/json'); + + return { + status: 'partial_extract', + source: { bucket, key, size_bytes: fileSize }, + output: { bucket: outputBucket, key: outputKey }, + corpus_type: corpusType, + tcs_version: TCS_CORPUS_VERSION, + entries: tcsData.entries?.length || 0, + size_mb: sizeMB, + sampled_mb: (partial.size_bytes / 1024 / 1024).toFixed(1), + message: `大文件部分提取完成 (采样 ${(partial.size_bytes / 1024 / 1024).toFixed(1)}MB / 共 ${sizeMB}MB)` + }; +} + /** * cosParseGitRepoArchive — 解析代码仓库压缩文件 *