用户上传的图片、PDF 和 Office 文档都应被视为不可信输入，不能直接传给模型，也不适合在 Web 请求中同步解析。更稳妥的做法是把接收、扫描、解析和模型消费拆成有状态的异步流水线，并为每一步设置明确配额。
先定义边界与状态机
一条实用的处理链路可以设计为：客户端上传到应用服务，服务完成流式大小限制和基础类型校验，将文件写入隔离对象存储，再投递队列；Worker 下载文件，执行病毒扫描、页数或像素检查、文本提取，最后把可信产物写入正式存储。LLM 只读取状态为 READY 的文件或解析结果。
建议为每个文件保存独立记录，而不是只依赖队列状态：
| 状态 | 含义 | 允许的后续操作 |
|---|---|---|
UPLOADING | 正在接收数据 | 超时后清理临时文件 |
QUARANTINED | 已进入隔离区 | 投递扫描任务 |
SCANNING | 正在进行安全检查 | 成功后解析，失败后拒绝 |
PARSING | 正在提取文本或元数据 | 可按错误类型重试 |
READY | 产物可供模型使用 | 生成短期访问地址 |
REJECTED | 类型、病毒或配额不合格 | 删除原文件并记录原因 |
FAILED | 基础设施或解析异常 | 延迟重试或人工处理 |
记录中至少包含 fileId、租户、对象键、声明类型、检测类型、字节数、页数、哈希、状态、错误码和时间戳。状态更新应使用条件更新，例如只允许 QUARANTINED 进入 SCANNING，避免重复消息并发处理同一文件。
上传入口只做必要工作
入口层应同时检查租户配额、请求声明和实际内容。扩展名与 Content-Type 只能用于快速拒绝，最终类型要读取文件特征；ZIP 魔数也不能直接证明文件是安全的 DOCX，因为 Office Open XML 本质上是 ZIP 容器。
下面是一个可运行的最小上传服务。它接收 application/octet-stream，通过 x-file-name 传递文件名，将内容流式写入临时目录，使用 file-type 检测类型，然后上传到 S3 兼容存储的隔离前缀并投递 BullMQ：
// upload.mjs
import http from 'node:http';
import { createReadStream, createWriteStream } from 'node:fs';
import { mkdir, rm } from 'node:fs/promises';
import { pipeline } from 'node:stream/promises';
import { Transform } from 'node:stream';
import { randomUUID } from 'node:crypto';
import { fileTypeFromFile } from 'file-type';
import { S3Client, PutObjectCommand } from '@aws-sdk/client-s3';
import { Queue } from 'bullmq';
const MAX_BYTES = 20 * 1024 * 1024;
const allowed = new Set([
'image/jpeg', 'image/png', 'image/webp', 'application/pdf',
'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
'application/vnd.openxmlformats-officedocument.presentationml.presentation',
'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
]);
const s3 = new S3Client({ region: process.env.AWS_REGION });
const queue = new Queue('file-processing', {
connection: { url: process.env.REDIS_URL }
});
await mkdir('tmp', { recursive: true });
http.createServer(async (req, res) => {
if (req.method !== 'POST' || req.url !== '/files') {
res.writeHead(404).end();
return;
}
const declaredLength = Number(req.headers['content-length'] || 0);
if (declaredLength > MAX_BYTES) {
res.writeHead(413).end('file too large');
return;
}
const fileId = randomUUID();
const path = `tmp/${fileId}`;
let bytes = 0;
const limiter = new Transform({
transform(chunk, encoding, callback) {
bytes += chunk.length;
callback(bytes > MAX_BYTES ? new Error('FILE_TOO_LARGE') : null, chunk);
}
});
try {
await pipeline(req, limiter, createWriteStream(path, { flags: 'wx' }));
const detected = await fileTypeFromFile(path);
if (!detected || !allowed.has(detected.mime)) {
res.writeHead(415).end('unsupported file type');
return;
}
const key = `quarantine/${fileId}`;
await s3.send(new PutObjectCommand({
Bucket: process.env.FILE_BUCKET,
Key: key,
Body: createReadStream(path),
ContentType: detected.mime,
Metadata: { fileid: fileId, bytes: String(bytes) }
}));
await queue.add('scan-and-parse', { fileId, key, mime: detected.mime }, {
jobId: fileId,
attempts: 4,
backoff: { type: 'exponential', delay: 5000 }
});
res.writeHead(202, { 'content-type': 'application/json' });
res.end(JSON.stringify({ fileId, status: 'QUARANTINED' }));
} catch (error) {
const status = error.message === 'FILE_TOO_LARGE' ? 413 : 500;
res.writeHead(status).end(error.message);
} finally {
await rm(path, { force: true });
}
}).listen(3000);运行前安装依赖：npm install @aws-sdk/client-s3 bullmq file-type，并设置 AWS_REGION、FILE_BUCKET 和 REDIS_URL。生产环境还要在写文件前检查租户剩余额度，并对上传接口设置并发数和请求速率限制。
隔离、扫描与内容配额
隔离区应使用独立前缀甚至独立存储桶，禁止公开读取，并配置生命周期规则自动删除滞留对象。不要在病毒扫描前生成可被浏览器或模型访问的签名 URL。
ClamAV 可以通过 clamscan 扫描单文件，也可以部署 clamd 复用常驻病毒库。调用外部程序时必须使用 execFile 一类不经过 shell 的 API，并设置超时。扫描结果应区分三类：退出码 0 表示未发现病毒，1 表示发现病毒，其他错误通常是扫描器故障，后者可以重试，不能误判为文件安全。
安全检查不能止于病毒扫描，还要防止资源消耗型文件：
| 类型 | 建议检查 |
|---|---|
| 图片 | 文件字节数、宽高、总像素、解码超时 |
| 页数、是否加密、解析超时、提取文本上限 | |
| DOCX/PPTX/XLSX | 解压后总大小、条目数量、转换后的页数 |
PDF 可使用 Poppler 的 pdfinfo 获取页数和加密信息；图片可用 sharp().metadata() 读取尺寸；Office 文档可在受限容器中使用 LibreOffice 无界面模式转换为 PDF，再统一检查页数。转换进程应限制 CPU、内存、临时磁盘和运行时间。仅靠 Node.js 的 Promise 超时无法终止失控的子进程，超时时还要发送终止信号，容器层也应设置资源上限。
异步解析、重试与清理
Worker 应从隔离区下载到独立临时目录，依次执行扫描、配额检查和解析。文本提取结果也要限制长度；否则一个合法但超长的文档仍可能制造高额模型调用。图片通常不必转成 Base64 存进数据库，可以在任务真正调用模型时生成分钟级签名 URL。
重试必须按错误分类。Redis、对象存储暂时不可用、扫描服务超时属于可重试错误；病毒命中、页数超限、文件加密或格式不支持属于永久失败。对永久失败继续指数退避只会浪费队列容量。
队列提供的是任务调度，不是完整审计记录。Worker 在每个阶段更新数据库状态并写入结构化事件，例如：
{
"fileId": "...",
"stage": "PARSING",
"attempt": 2,
"durationMs": 1840,
"result": "FAILED",
"errorCode": "PARSER_TIMEOUT"
}不要记录提取出的正文、签名 URL 或用户原始文件名，以免日志成为新的敏感数据副本。指标可以围绕各状态数量、队列等待时间、处理耗时、拒绝原因和重试次数建立。
清理需要覆盖四个位置：Web 节点临时文件、Worker 临时目录、隔离对象和不完整的解析产物。进程内使用 finally 删除临时目录；隔离桶使用生命周期规则兜底；任务成功后采用“先写正式产物，再条件更新数据库，最后删除隔离对象”的顺序。若删除失败，后台清理任务可根据数据库状态再次执行。解析产物应使用确定性对象键，例如 parsed/{fileId}/v1/text.txt，让任务重试时覆盖同一版本，而不是不断制造副本。
还要定期扫描长期停留在 UPLOADING、SCANNING 或 PARSING 的记录。处理进程可能在状态更新后、队列确认前崩溃，仅依靠正常回调无法覆盖这类中断。
总结
可靠的多模态输入链路不是一个上传接口加一次模型调用，而是一组彼此约束的阶段：入口流式限制字节数并检测真实类型；文件先进入不可公开访问的隔离区；Worker 在受限环境中完成病毒扫描、页数和像素检查；解析任务异步执行，并按错误是否可恢复决定重试；数据库状态、结构化事件和指标共同提供追踪能力；finally、生命周期规则和后台对账任务负责失败清理。
只有进入 READY 状态的文件才能交给 LLM。这样即使解析器变慢、扫描服务故障或用户提交恶意文件，Web 请求、队列容量和模型成本仍然处在可控制的边界内。
评论