feat: 增强 Gemini 流式响应兼容性,支持 JSON-array 格式解析

- 添加 Gemini JSON-array/chunks 格式的解析支持(兼容非 SSE 返回)
- 在上游请求中添加 alt=sse 参数,优先使用 SSE 格式
- 为 OpenAI 客户端统一补齐 [DONE] 流结束标记
- 更新 Gemini finishReason 判断逻辑,采用更宽松的结束判定
This commit is contained in:
fawney19
2026-01-23 23:34:38 +08:00
parent c3be83bf94
commit 76ed7e9fcb
5 changed files with 154 additions and 115 deletions

View File

@@ -2005,27 +2005,39 @@ class CliMessageHandlerBase(BaseMessageHandler):
if not line or line.strip() == "" or line == "data: [DONE]": if not line or line.strip() == "" or line == "data: [DONE]":
return [line] if line else [] return [line] if line else []
# 如果不是 data 行,直接透传 provider_format = (ctx.provider_api_format or "").upper()
client_format = (ctx.client_api_format or "").upper()
# 兼容 Gemini 上游 JSON-array/chunks可能是“裸 JSON 行”而非 `data: {...}`
if not line.startswith("data:"): if not line.startswith("data:"):
return [line] if provider_format != "GEMINI":
return [line]
# 提取 data 内容 stripped = line.strip()
data_content = line[5:].strip() # 去掉 "data:" 前缀 if stripped in ("", "[", "]", ","):
return []
# 尝试解析 JSON candidate = stripped.lstrip(",").rstrip(",").strip()
try: try:
data_obj = json.loads(data_content) data_obj = json.loads(candidate)
except json.JSONDecodeError: except json.JSONDecodeError:
# 无法解析,直接透传 # 不是可解析的 JSON 对象:不透传(避免把 Gemini 原始分块泄漏到目标 SSE
return [line] logger.debug(f"Gemini JSON-array line skip: {stripped[:50]}")
return []
else:
# 提取 data 内容
data_content = line[5:].strip() # 去掉 "data:" 前缀
# 类型断言:当 needs_conversion=True 调用此方法时,格式字段必定有值 # 尝试解析 JSON
provider_format = ctx.provider_api_format or "" try:
client_format = ctx.client_api_format or "" data_obj = json.loads(data_content)
except json.JSONDecodeError:
# 无法解析,直接透传
return [line]
# 初始化流式转换状态(首次调用时,根据 Provider 格式选择状态类) # 初始化流式转换状态(首次调用时,根据 Provider 格式选择状态类)
if ctx.stream_conversion_state is None: if ctx.stream_conversion_state is None:
if provider_format.upper() == "GEMINI": if provider_format == "GEMINI":
ctx.stream_conversion_state = GeminiStreamConversionState( ctx.stream_conversion_state = GeminiStreamConversionState(
model=ctx.mapped_model or ctx.model or "", model=ctx.mapped_model or ctx.model or "",
message_id=ctx.response_id or ctx.request_id or "", message_id=ctx.response_id or ctx.request_id or "",

View File

@@ -484,86 +484,102 @@ class StreamProcessor:
message_id=ctx.response_id or ctx.request_id or "", message_id=ctx.response_id or ctx.request_id or "",
) )
skip_next_blank_line = False skip_next_blank_line = False
empty_yield_count = 0 # 空转计数(防护异常情况) empty_yield_count = 0 # 空转计数(防护异常情况)
openai_done_sent = (
False # 统一为 OpenAI 客户端补齐 [DONE](避免不同 Provider 行为差异)
)
def _emit_converted_line(normalized_line: str) -> list[bytes]: def _emit_converted_line(normalized_line: str) -> list[bytes]:
nonlocal skip_next_blank_line nonlocal skip_next_blank_line, openai_done_sent
# 空行:事件分隔符(避免重复输出) # 空行:事件分隔符(避免重复输出)
if normalized_line == "": if normalized_line == "":
if skip_next_blank_line: if skip_next_blank_line:
skip_next_blank_line = False skip_next_blank_line = False
return []
return [b"\n"]
# 丢弃 Provider 的 event 行,避免泄漏/污染目标格式
if normalized_line.startswith("event:"):
return [] return []
return [b"\n"]
# 丢弃 Provider 的 event 行,避免泄漏/污染目标格式 # OpenAI done 信号(仅用于 OpenAI 客户端)
if normalized_line.startswith("event:"): if (
return [] normalized_line.startswith("data:")
and normalized_line[5:].strip() == "[DONE]"
):
skip_next_blank_line = True
if client_format.startswith("OPENAI"):
openai_done_sent = True
return [b"data: [DONE]\n\n"]
return []
# 默认只处理 SSE 的 data 行;但 Gemini 上游可能返回 JSON-array/chunks无 data 前缀)
is_data_line = normalized_line.startswith("data:")
if not is_data_line:
if provider_format != "GEMINI":
return []
data_content = normalized_line.strip()
else:
data_content = normalized_line[5:].strip()
# Gemini 可能包含 JSON 数组包装符,直接忽略
if data_content in ("", "[", "]", ","):
return []
# JSON-array/chunks 可能带前后逗号(对象分隔符),做一次保守清理
data_content = data_content.lstrip(",").rstrip(",").strip()
if data_content in ("", "[", "]", ","):
return []
try:
data_obj = json.loads(data_content)
except json.JSONDecodeError:
# 跨格式转换时JSON 解析失败应跳过而不是透传(避免泄漏 Provider 格式)
logger.warning(
f"[{self.request_id}] JSON 解析失败,跳过该行: {data_content[:100]}"
)
return []
if not isinstance(data_obj, dict):
return []
try:
converted_events = converter_registry.convert_stream_chunk_strict(
data_obj,
provider_format,
client_format,
state=ctx.stream_conversion_state,
)
except Exception as conv_err:
# 首字节后无法 failover输出目标格式错误事件并终止流
# 使用 502 表示上游返回了非预期格式Bad Gateway
ctx.status_code = 502
ctx.error_message = "format_conversion_failed"
# 日志记录完整错误(内部排查),客户端只返回脱敏消息
logger.warning(f"[{self.request_id}] 流式格式转换失败: {conv_err}")
payload = _build_stream_error_payload("响应格式转换失败,请稍后重试")
error_bytes = (
f"data: {json.dumps(payload, ensure_ascii=False)}\n\n".encode(
"utf-8"
)
)
done_bytes = (
b"data: [DONE]\n\n" if client_format.startswith("OPENAI") else b""
)
if done_bytes:
openai_done_sent = True
return [error_bytes, done_bytes]
# OpenAI done 信号
if (
normalized_line.startswith("data:")
and normalized_line[5:].strip() == "[DONE]"
):
skip_next_blank_line = True skip_next_blank_line = True
if client_format.startswith("OPENAI"): out: list[bytes] = []
return [b"data: [DONE]\n\n"] for evt in converted_events:
return [] # 统一使用 SSE 格式输出Gemini streamGenerateContent 也使用 SSE
# 参考: https://ai.google.dev/api/generate-content
# 非 data 行:在跨格式场景下统一丢弃(避免泄露 Provider 格式细节) out.append(
if not normalized_line.startswith("data:"): f"data: {json.dumps(evt, ensure_ascii=False)}\n\n".encode("utf-8")
return [] )
return out
data_content = normalized_line[5:].strip()
# Gemini 可能包含 JSON 数组包装符,直接忽略
if data_content in ("", "[", "]", ","):
return []
try:
data_obj = json.loads(data_content)
except json.JSONDecodeError:
# 跨格式转换时JSON 解析失败应跳过而不是透传(避免泄漏 Provider 格式)
logger.warning(
f"[{self.request_id}] JSON 解析失败,跳过该行: {data_content[:100]}"
)
return []
if not isinstance(data_obj, dict):
return []
try:
converted_events = converter_registry.convert_stream_chunk_strict(
data_obj,
provider_format,
client_format,
state=ctx.stream_conversion_state,
)
except Exception as conv_err:
# 首字节后无法 failover输出目标格式错误事件并终止流
# 使用 502 表示上游返回了非预期格式Bad Gateway
ctx.status_code = 502
ctx.error_message = "format_conversion_failed"
# 日志记录完整错误(内部排查),客户端只返回脱敏消息
logger.warning(f"[{self.request_id}] 流式格式转换失败: {conv_err}")
payload = _build_stream_error_payload("响应格式转换失败,请稍后重试")
error_bytes = f"data: {json.dumps(payload, ensure_ascii=False)}\n\n".encode(
"utf-8"
)
done_bytes = (
b"data: [DONE]\n\n" if client_format.startswith("OPENAI") else b""
)
return [error_bytes, done_bytes]
skip_next_blank_line = True
out: list[bytes] = []
for evt in converted_events:
# 统一使用 SSE 格式输出Gemini streamGenerateContent 也使用 SSE
# 参考: https://ai.google.dev/api/generate-content
out.append(
f"data: {json.dumps(evt, ensure_ascii=False)}\n\n".encode("utf-8")
)
return out
# 统一处理 prefetched + iterator # 统一处理 prefetched + iterator
if prefetched_chunks: if prefetched_chunks:
@@ -653,6 +669,11 @@ class StreamProcessor:
if ctx.error_message == "format_conversion_failed": if ctx.error_message == "format_conversion_failed":
return return
# Provider 流结束后,为 OpenAI 客户端补齐 [DONE](许多上游不发送该哨兵)
if client_format.startswith("OPENAI") and not openai_done_sent:
_mark_stream_started()
yield b"data: [DONE]\n\n"
else: else:
if prefetched_chunks: if prefetched_chunks:
for chunk in prefetched_chunks: for chunk in prefetched_chunks:

View File

@@ -1,12 +1,17 @@
""" """
Gemini SSE/JSON 流解析器 Gemini 流解析器SSE + JSON-array 兼容)
Gemini API 的流式响应格式与 Claude/OpenAI 不同: Gemini streamGenerateContent 常见两种返回(与上游/代理实现有关):
- 使用 JSON 数组格式 (不是 SSE) 1) `?alt=sse`SSE`data: {GenerateContentResponse}`
- 每个块是一个完整的 JSON 对象 2) 默认JSON-array / JSON-chunks`[{...},{...},...]`,可能跨 chunk/跨行)
- 响应以 [ 开始,以 ] 结束,块之间用 , 分隔
参考: https://ai.google.dev/api/generate-content#method:-models.streamgeneratecontent 本解析器提供:
- parse_line(): 适用于 SSE data 行或逐行 JSON 对象
- parse_chunk(): 适用于 JSON-array/chunks可跨 chunk 拼接)
参考:
- https://ai.google.dev/gemini-api/docs/text-generation?lang=python#generate-a-text-stream
- https://generativelanguage.googleapis.com/$discovery/rest?version=v1beta
""" """
import json import json
@@ -20,17 +25,12 @@ class GeminiStreamParser:
解析 Gemini streamGenerateContent API 的响应流。 解析 Gemini streamGenerateContent API 的响应流。
Gemini 流式响应特点: Gemini 流式响应特点:
- 返回 JSON 数组格式: [{chunk1}, {chunk2}, ...] - 每个事件块本质上都是一个 GenerateContentResponse JSON 对象(包含 candidates、usageMetadata 等)
- 每个 chunk 包含 candidates、usageMetadata 等字段 - 结束判定以 `candidates[].finishReason` 为准(存在且不为 FINISH_REASON_UNSPECIFIED
- finish_reason 可能值: STOP, MAX_TOKENS, SAFETY, RECITATION, OTHER
""" """
# 停止原因 # finishReason官方枚举值很多见 discovery这里仅保留一个明确的“未结束”哨兵
FINISH_REASON_STOP = "STOP" FINISH_REASON_UNSPECIFIED = "FINISH_REASON_UNSPECIFIED"
FINISH_REASON_MAX_TOKENS = "MAX_TOKENS"
FINISH_REASON_SAFETY = "SAFETY"
FINISH_REASON_RECITATION = "RECITATION"
FINISH_REASON_OTHER = "OTHER"
def __init__(self) -> None: def __init__(self) -> None:
self._buffer = "" self._buffer = ""
@@ -129,17 +129,18 @@ class GeminiStreamParser:
True 如果是结束事件 True 如果是结束事件
""" """
candidates = event.get("candidates", []) candidates = event.get("candidates", [])
if candidates: if not candidates:
for candidate in candidates: return False
finish_reason = candidate.get("finishReason")
if finish_reason in ( for candidate in candidates:
self.FINISH_REASON_STOP, finish_reason = candidate.get("finishReason")
self.FINISH_REASON_MAX_TOKENS, if not finish_reason:
self.FINISH_REASON_SAFETY, continue
self.FINISH_REASON_RECITATION, # 只要出现非 UNSPECIFIED 的 finishReason通常表示该 candidate 已结束。
self.FINISH_REASON_OTHER, # 例如STOP/MAX_TOKENS/SAFETY/RECITATION/MALFORMED_FUNCTION_CALL/...(枚举持续演进)
): if str(finish_reason) != self.FINISH_REASON_UNSPECIFIED:
return True return True
return False return False
def is_error_event(self, event: Dict[str, Any]) -> bool: def is_error_event(self, event: Dict[str, Any]) -> bool:

View File

@@ -128,6 +128,10 @@ def build_provider_url(
# 上游认证始终使用 header 方式,不使用 URL 参数 # 上游认证始终使用 header 方式,不使用 URL 参数
if resolved_format in (APIFormat.GEMINI, APIFormat.GEMINI_CLI): if resolved_format in (APIFormat.GEMINI, APIFormat.GEMINI_CLI):
effective_query_params.pop("key", None) effective_query_params.pop("key", None)
# Gemini streamGenerateContent 官方支持 `?alt=sse` 返回 SSEdata: {...})。
# 网关侧统一使用 SSE 输出,优先向上游请求 SSE 以减少解析分支;同时保留 JSON-array 兜底解析。
if is_stream:
effective_query_params.setdefault("alt", "sse")
# 添加查询参数 # 添加查询参数
if effective_query_params: if effective_query_params:

View File

@@ -91,6 +91,8 @@ async def test_create_response_stream_converts_claude_to_openai() -> None:
events = [] events = []
for line in text.splitlines(): for line in text.splitlines():
if line.startswith("data: "): if line.startswith("data: "):
if line == "data: [DONE]":
continue
events.append(json.loads(line[6:])) events.append(json.loads(line[6:]))
assert len(events) >= 2 assert len(events) >= 2
@@ -100,4 +102,3 @@ async def test_create_response_stream_converts_claude_to_openai() -> None:
for e in events for e in events
if isinstance(e, dict) if isinstance(e, dict)
) )