Firebase Genkit流式返回最后两个JSON块合并问题及优化咨询

使用Firebase Genkit搭配Gemini Pro 1.5实现大模型响应流式返回至客户端时，出现异常：正常情况下每个流式返回块都是可解析的有效JSON（示例：{"index":0,"content":[{"text":"this is a test"}]}），但最后两个块会被合并成一个流，格式为{"index":0,"content":[{"text":"this is a test"}]} {"index":0,"content":[{"text":""}]}，导致JSON解析失败。目前临时方案是尝试解析，失败则通过"}{"分割字符串，但希望找到更优雅的解决方案，同时明确该问题的成因及最佳实践。

相关实现代码：

configureGenkit({
  plugins: [
    firebase(),
    vertexAI({ location: "us-central1" }),
    dotprompt(),
  ],
  logLevel: "debug",
  enableTracingAndMetrics: true,
});

const nonEmptyString = z.string().refine((value) => value.trim().length > 0, {
  message: "String cannot be empty",
});

// Define a schema for a non-empty array of non-empty strings
const nonEmptyArray = z.array(nonEmptyString).refine((arr) => arr.length > 0, {
  message: "Array cannot be empty",
});
        
export const carePlanGenerationFlow = onFlow(
      {
        name: "carePlanGenerationFlow",
        inputSchema: z.object({
          content: z.string(),
          sourceUrls: nonEmptyArray.optional(),
        }),
        outputSchema: z.any(),
        authPolicy: noAuth(),
      },
      async (promptData, streamingCallback) => {
        if (!streamingCallback) {
          throw new Error("this flow only works in streaming mode");
        }
    
        const planPrompt = await prompt("carePlan");
        let response =
          "Something went wrong and I am unable to generate the care plan.  " +
          "Please try again.";
    
        try {
          const llmResponse = await planPrompt.generate({
            input: {
              content: promptData.content,
              sourceUrls: promptData.sourceUrls,
            },
            streamingCallback,
          });
    
          response = llmResponse.text();
        } catch (error) {
          console.log("error generating content: ", error);
          response = "Sorry I encountered an error processing your request.  " +
            " Please try submiting your request again.";
        }
    
        return { carePlanText: response };
      },
    );

• 流式传输底层策略：Gemini Pro 1.5在返回收尾响应时，可能因内部缓冲区合并机制，将空响应块（{"index":0,"content":[{"text":""}]}）与前一个有效块打包发送，导致两个JSON对象被拼接成单一字符串。
• Genkit插件处理逻辑：Firebase Genkit的Vertex AI插件在转发流式响应时，未针对这类合并块做拆分处理，直接将拼接后的字符串传递给streamingCallback。

1. 通用流式JSON解析器 #

实现逐字符处理的流式解析器，识别完整JSON对象的边界，不依赖块的完整性。核心逻辑是维护缓冲区，跟踪JSON嵌套层级，提取完整对象后再解析：

class StreamingJsonParser {
  constructor() {
    this.buffer = '';
    this.depth = 0;
    this.inString = false;
    this.escape = false;
  }

  processChunk(chunk, callback) {
    this.buffer += chunk;
    let i = 0;
    while (i < this.buffer.length) {
      const char = this.buffer[i];
      
      if (this.escape) {
        this.escape = false;
        i++;
        continue;
      }

      if (char === '"') {
        this.inString = !this.inString;
      } else if (!this.inString) {
        if (char === '{') {
          this.depth++;
        } else if (char === '}') {
          this.depth--;
          if (this.depth === 0) {
            const jsonStr = this.buffer.slice(0, i + 1);
            this.buffer = this.buffer.slice(i + 1).trim();
            try {
              const json = JSON.parse(jsonStr);
              callback(null, json);
            } catch (err) {
              callback(err);
            }
            i = 0;
            continue;
          }
        } else if (char === '\\') {
          this.escape = true;
        }
      }
      i++;
    }
  }
}

// 在streamingCallback中使用
const parser = new StreamingJsonParser();
const streamingCallback = (chunk) => {
  parser.processChunk(chunk, (err, json) => {
    if (err) {
      console.error('JSON解析失败:', err);
    } else {
      // 处理有效JSON块
      console.log('解析后的响应:', json);
    }
  });
};

2. 过滤空响应块 #

解析完成后，对JSON对象进行校验，过滤掉content.text为空的收尾块，避免无效数据传递到客户端：

if (json.content?.[0]?.text.trim().length === 0) {
  return; // 跳过空响应块
}
// 继续处理有效内容

3. 检查插件配置项 #

查看Vertex AI插件的官方文档，确认是否有控制流式响应块拆分的配置参数，部分LLM流式API支持强制每个响应块为单一JSON对象，若Genkit提供相关配置可尝试启用。

内容的提问来源于stack exchange，提问作者Anthony D.