使用TransformStream处理LangChain.js流时遇响应体锁定错误求助

问题背景 #

使用langchain.js的streamEvents向前端流式传输事件时，自定义TransformStream剥离敏感数据（删除data.input、data.output.context和data.output.question），但偶尔会抛出TypeError: Response body object should not be disturbed or locked错误。

错误原因 #

这个错误通常由以下情况导致：

• 原响应体流被其他逻辑提前锁定或消费，TransformStream处理时触发冲突。
• 异常场景下直接将原始chunk入队，导致二进制数据与处理后的文本数据混合，引发流状态紊乱。

修复方案 #

1. 确保流唯一处理 #

转发响应前，保证原响应体流未被其他操作读取或锁定，获取原流后立即通过TransformStream处理，避免中间添加其他消费流的逻辑。

2. 优化TransformStream逻辑 #

复用解码/编码器实例，改进错误处理逻辑，避免流数据格式混乱：

// 复用解码器和编码器，避免重复实例化
const decoder = new TextDecoder();
const encoder = new TextEncoder();

const transformStream = new TransformStream({
    transform(chunk, controller) {
        try {
            // 开启stream模式处理分片数据
            const text = decoder.decode(chunk, { stream: true });
            const lines = text.split('\n');
            
            const processedLines = lines.map(line => {
                if (line.startsWith('data:')) {
                    try {
                        const data = JSON.parse(line.slice(5));
                        delete data.data.input;
                        if (typeof data.data.output === 'object') {
                            delete data.data.output.context;
                            delete data.data.output.question;
                        }
                        return `data: ${JSON.stringify(data)}`;
                    } catch (parseErr) {
                        // JSON解析失败时保留原始行
                        return line;
                    }
                }
                return line;
            });
            
            controller.enqueue(encoder.encode(processedLines.join('\n')));
        } catch (err) {
            // 全局异常时统一转为文本格式入队，避免格式冲突
            controller.enqueue(encoder.encode(decoder.decode(chunk)));
        }
    },
    flush(controller) {
        // 处理解码器中剩余的未解码数据
        controller.enqueue(encoder.encode(decoder.decode()));
    }
});

3. 规范响应转发流程 #

返回前端时正确使用处理后的流，确保响应头配置正确：

// 假设原响应为originalResponse
const processedStream = originalResponse.body.pipeThrough(transformStream);
return new Response(processedStream, {
    headers: {
        ...originalResponse.headers,
        'Content-Type': 'text/event-stream', // 确保SSE类型正确
    },
});

关键注意事项 #

• 处理SSE流时，必须保证每个data:行格式合规，避免JSON序列化错误导致前端解析失败。
• 禁止在同一个响应体流上多次调用pipeThrough或pipeTo，这会直接锁定流导致报错。

内容的提问来源于stack exchange，提问作者Kunal