fix: COS训练Agent大文件崩溃 — 添加Buffer安全处理、HEAD预检、分块提取策略
修复两个训练失败根因: 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>
This commit is contained in:
copilot-swe-agent[bot]
GitHub
parent
0ace2a3039
commit
ba4c233509
|
|
@ -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`);
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
};
|
||||
|
|
|
|||
|
|
@ -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 — 解析代码仓库压缩文件
|
||||
*
|
||||
|
|
|
|||
Loading…
Reference in New Issue