ai-model-studio / src /api /transform.py
slk1st's picture
Deploy AI Model Studio
8a03d2c verified
Raw
History Blame Contribute Delete
80.2 kB
"""请求和响应转换工具模块
负责:
1. 将 Gemini 格式的 Payload 转换为 Vcore AI 内部格式
2. 将流式响应聚合成完整的非流式响应
"""
import json
import re
import time
from typing import Any, cast
from src.core.errors import (
VcoreError,
InternalError,
parse_error_response,
)
from src.api.model_config import ModelConfigBuilder
from src.utils.logger import get_logger
from src.utils.string_utils import snake_to_camel, camel_to_snake
logger = get_logger(__name__)
_GEMINI_FUNCTION_NAME_RE = re.compile(r"[^A-Za-z0-9_.-]+")
class RequestTransformer:
"""请求参数转换器"""
def __init__(self, model_builder: ModelConfigBuilder):
self.model_builder = model_builder
def build_vcore_payload(
self,
model: str,
gemini_payload: dict[str, Any],
original_body: dict[str, Any],
kwargs: dict[str, Any]
) -> dict[str, Any]:
"""
构建 Vcore AI 请求 Payload
Returns:
new_body
"""
original_vars: Any = original_body.get('variables', {})
new_variables: dict[str, Any]
if hasattr(original_vars, 'model_dump'):
new_variables = cast(dict[str, Any], original_vars.model_dump())
elif isinstance(original_vars, dict):
new_variables = {str(k): v for k, v in cast(dict[Any, Any], original_vars).items()}
else:
new_variables = {}
gemini_payload = self._normalize_gemini_payload(gemini_payload)
gemini_payload = self._normalize_thought_signature_aliases(gemini_payload)
target_model = self.model_builder.parse_model_name(model)
new_variables['model'] = target_model
# 支持的字段列表(统一使用 camelCase 格式)。尽量覆盖 Gemini generateContent
# 可透传到 Vcore AI Studio 匿名 GraphQL variables 的字段。
supported_fields = self._supported_variable_fields()
canonical_payload = self._canonicalize_supported_fields(gemini_payload, supported_fields)
try:
from src.core.types import GeminiPayload
gemini_payload_obj = GeminiPayload.model_validate(canonical_payload)
dumped_payload = gemini_payload_obj.model_dump(by_alias=True, exclude_none=True)
for field in supported_fields:
if field in dumped_payload:
new_variables[field] = dumped_payload[field]
except Exception as e:
logger.debug(f"Pydantic 验证失败,使用基础转换: {e}")
# 尝试直接从 gemini_payload 透传字段,支持 snake_case 和 camelCase
for field in supported_fields:
# 优先使用 camelCase 版本
if field in canonical_payload:
new_variables[field] = canonical_payload[field]
else:
# 尝试 snake_case 版本
snake_field = camel_to_snake(field)
if snake_field in canonical_payload:
new_variables[field] = canonical_payload[snake_field]
# tools/toolConfig 内部存在大量 snake_case、实验字段和空对象默认值,
# 使用原始规范化 payload 再走后续专用转换,避免 Pydantic dump 过早丢失未知/None 字段。
if 'tools' in canonical_payload:
new_variables['tools'] = canonical_payload['tools']
if 'toolConfig' in canonical_payload:
new_variables['toolConfig'] = canonical_payload['toolConfig']
# 处理 systemInstruction:如果没有 user content,则转换为 user message
self._handle_system_instruction(new_variables)
# 特殊处理:contents 格式转换
if 'contents' in new_variables:
converted_contents = self._normalize_contents(new_variables['contents'])
converted_contents = self._handle_inline_data_case(converted_contents)
converted_contents = self._normalize_contents(converted_contents)
converted_contents = self._handle_base64_in_contents(converted_contents)
# 过滤掉空的 parts(Vcore AI 要求每个 content 至少有一个 part)
converted_contents = self._filter_empty_contents(converted_contents)
converted_contents = self._ensure_function_call_thought_signatures(converted_contents)
# 处理 thoughtSignature 字段的 base64 编码
converted_contents = self._handle_thought_signature(converted_contents)
new_variables['contents'] = converted_contents
# 特殊处理:tools 格式转换
if 'tools' in new_variables:
normalized_tools = self._normalize_tools_format(new_variables['tools'])
if normalized_tools:
new_variables['tools'] = normalized_tools
else:
# 如果转换结果为空列表,确保移除 tools 字段,同时移除 toolConfig 避免 API 报错
del new_variables['tools']
if 'toolConfig' in new_variables:
del new_variables['toolConfig']
# 特殊处理:toolConfig 格式转换
if 'toolConfig' in new_variables:
normalized_tool_config = self._normalize_tool_config(new_variables['toolConfig'])
if normalized_tool_config:
new_variables['toolConfig'] = normalized_tool_config
else:
del new_variables['toolConfig']
# 特殊处理 generationConfig (使用 ModelConfigBuilder 进行格式转换)
gen_config = self.model_builder.build_generation_config(
gen_config={},
gemini_payload=gemini_payload,
**kwargs
)
if gen_config:
new_variables['generationConfig'] = gen_config
# 特殊处理 safetySettings (如果未提供,则使用默认的宽松设置)
if 'safetySettings' not in new_variables and 'safety_settings' not in gemini_payload:
new_variables['safetySettings'] = self.model_builder.build_safety_settings()
new_body: dict[str, Any] = {
"querySignature": original_body.get('querySignature'),
"operationName": original_body.get('operationName'),
"variables": new_variables
}
self._sanitize_vcore_variables_in_place(new_variables)
return new_body
def _supported_variable_fields(self) -> list[str]:
"""Gemini 下游请求可透传到上游 variables 的字段。"""
return [
'contents', 'tools', 'toolConfig', 'systemInstruction',
'safetySettings', 'generationConfig', 'cachedContent', 'labels',
# Gemini/Vcore 常见高级字段;上游不支持时会由上游返回明确错误,
# 这里不主动丢弃,以最大化暴露上游能力。
'modelArmorConfig', 'cachedContentName', 'requestOptions',
'session', 'context', 'examples', 'instances', 'parameters',
]
def _canonicalize_supported_fields(self, payload: dict[str, Any], supported_fields: list[str]) -> dict[str, Any]:
"""把下游 Gemini REST/SDK 常见 snake_case 顶层字段并入 camelCase 标准字段。"""
canonical = payload.copy()
for field in supported_fields:
snake_field = camel_to_snake(field)
if field not in canonical and snake_field in canonical:
canonical[field] = canonical[snake_field]
return canonical
def _normalize_gemini_payload(self, payload: dict[str, Any]) -> dict[str, Any]:
"""兼容 REST、SDK 和部分 OpenAI-like 客户端传来的 Gemini 请求形态。"""
normalized = payload.copy()
if 'contents' in normalized:
normalized['contents'] = self._normalize_contents(normalized['contents'])
elif 'prompt' in normalized:
normalized['contents'] = [{"role": "user", "parts": [{"text": str(normalized['prompt'])}]}]
return normalized
def _normalize_thought_signature_aliases(self, data: Any) -> Any:
"""递归兼容 thought_signature / thoughtSignature 两种字段名。"""
if isinstance(data, list):
return [self._normalize_thought_signature_aliases(item) for item in cast(list[Any], data)]
if isinstance(data, dict):
data_dict = cast(dict[str, Any], data)
normalized: dict[str, Any] = {}
for key, value in data_dict.items():
normalized_value = self._normalize_thought_signature_aliases(value) if isinstance(value, (dict, list)) else value
if key == 'thought_signature':
normalized['thoughtSignature'] = normalized_value
elif key == 'thoughtSignature':
normalized['thoughtSignature'] = normalized_value
else:
normalized[key] = normalized_value
return normalized
return data
def _normalize_contents(self, contents: Any) -> Any:
if contents is None:
return []
if isinstance(contents, str):
return [{"role": "user", "parts": [{"text": contents}]}]
if isinstance(contents, dict):
return [self._normalize_content(contents)]
if isinstance(contents, list):
normalized: list[Any] = []
pending_text_parts: list[dict[str, Any]] = []
for item in cast(list[Any], contents):
if isinstance(item, str):
pending_text_parts.append({"text": item})
elif isinstance(item, dict):
if pending_text_parts:
normalized.append({"role": "user", "parts": pending_text_parts})
pending_text_parts = []
normalized.append(self._normalize_content(cast(dict[str, Any], item)))
if pending_text_parts:
normalized.append({"role": "user", "parts": pending_text_parts})
return normalized
return contents
def _normalize_content(self, content: dict[str, Any]) -> dict[str, Any]:
normalized = content.copy()
source_role = normalized.get('role')
if source_role in {'tool', 'function'} and 'parts' not in normalized and ('content' in normalized or 'response' in normalized):
raw_response = normalized.get('response', normalized.get('content', {}))
normalized['parts'] = [{
"functionResponse": {
"name": str(normalized.get('name') or normalized.get('function_name') or ""),
"response": self._coerce_function_response(raw_response),
}
}]
normalized.pop('content', None)
normalized.pop('response', None)
elif 'content' in normalized and 'parts' not in normalized:
normalized['parts'] = self._normalize_parts(normalized.get('content'))
normalized.pop('content', None)
elif 'parts' in normalized:
normalized['parts'] = self._normalize_parts(normalized.get('parts'))
elif 'text' in normalized:
normalized['parts'] = [{"text": str(normalized.pop('text'))}]
else:
normalized.setdefault('parts', [])
if source_role in {'assistant', 'model'}:
extra_parts = self._openai_tool_history_parts(normalized)
if extra_parts:
normalized['parts'] = list(cast(list[Any], normalized.get('parts') or [])) + extra_parts
normalized.pop('tool_calls', None)
normalized.pop('function_call', None)
normalized.pop('functionCall', None)
role = normalized.get('role')
if role == 'assistant':
normalized['role'] = 'model'
elif role == 'tool':
normalized['role'] = 'function'
elif not role:
normalized['role'] = 'user'
return normalized
def _normalize_parts(self, parts: Any) -> list[dict[str, Any]]:
if parts is None:
return []
if isinstance(parts, str):
return [{"text": parts}]
if isinstance(parts, dict):
return [self._normalize_part(parts)]
if isinstance(parts, list):
normalized: list[dict[str, Any]] = []
for part in cast(list[Any], parts):
if isinstance(part, str):
normalized.append({"text": part})
elif isinstance(part, dict):
normalized_part = self._normalize_part(cast(dict[str, Any], part))
if normalized_part:
normalized.append(normalized_part)
return normalized
return [{"text": str(parts)}]
def _openai_tool_history_parts(self, message: dict[str, Any]) -> list[dict[str, Any]]:
"""兼容下游把 OpenAI tool_calls/function_call 混入 Gemini endpoint 的历史。"""
parts: list[dict[str, Any]] = []
tool_calls = message.get('tool_calls') or message.get('toolCalls')
if isinstance(tool_calls, list):
for tool_call in cast(list[Any], tool_calls):
part = self._openai_tool_call_to_part(tool_call)
if part:
parts.append(part)
function_call = message.get('function_call') or message.get('functionCall')
if isinstance(function_call, dict):
part = self._openai_tool_call_to_part({"function": function_call})
if part:
parts.append(part)
return parts
def _openai_tool_call_to_part(self, tool_call: Any) -> dict[str, Any] | None:
if not isinstance(tool_call, dict):
return None
tool_call_dict = cast(dict[str, Any], tool_call)
func = tool_call_dict.get('function')
if isinstance(func, dict):
func_dict = cast(dict[str, Any], func)
name = func_dict.get('name') or tool_call_dict.get('name') or tool_call_dict.get('functionName')
args = func_dict.get('arguments', tool_call_dict.get('arguments', tool_call_dict.get('args', {})))
thought_signature = (
tool_call_dict.get('thoughtSignature')
or tool_call_dict.get('thought_signature')
or func_dict.get('thoughtSignature')
or func_dict.get('thought_signature')
)
else:
name = tool_call_dict.get('name') or tool_call_dict.get('function_name') or tool_call_dict.get('functionName')
args = tool_call_dict.get('arguments', tool_call_dict.get('args', {}))
thought_signature = tool_call_dict.get('thoughtSignature') or tool_call_dict.get('thought_signature')
if not name:
return None
part: dict[str, Any] = {
"functionCall": {
"name": self._sanitize_function_name(str(name)),
"args": self._coerce_function_args(args),
}
}
if thought_signature:
part["thoughtSignature"] = thought_signature
return part
def _coerce_function_args(self, args: Any) -> dict[str, Any]:
if isinstance(args, dict):
return cast(dict[str, Any], args)
if isinstance(args, str):
try:
parsed = json.loads(args)
return parsed if isinstance(parsed, dict) else {"value": parsed}
except json.JSONDecodeError:
return {"raw": args}
if args is None:
return {}
return {"value": args}
def _coerce_function_response(self, response: Any) -> dict[str, Any]:
if isinstance(response, dict):
return cast(dict[str, Any], response)
if isinstance(response, str):
try:
parsed = json.loads(response)
return parsed if isinstance(parsed, dict) else {"result": parsed}
except json.JSONDecodeError:
return {"result": response}
if response is None:
return {}
return {"result": response}
def _normalize_part(self, part: dict[str, Any]) -> dict[str, Any]:
part_type = part.get('type')
if part_type in {'text', 'input_text'}:
return {"text": str(part.get('text', ''))}
if part_type in {'image_url', 'input_image'}:
url_obj = part.get('image_url') or part.get('input_image') or {}
url = url_obj.get('url') if isinstance(url_obj, dict) else url_obj
if isinstance(url, str) and url.startswith('data:'):
mime, data = self._parse_data_uri(url)
if mime and data:
return {"inlineData": {"mimeType": mime, "data": data}}
if isinstance(url, str) and url.startswith(('http://', 'https://', 'gs://')):
return {"fileData": {"mimeType": self._guess_mime_from_uri(url), "fileUri": url}}
if part_type in {'media', 'file', 'file_data'}:
file_uri = part.get('fileUri') or part.get('file_uri') or part.get('uri') or part.get('url')
mime_type = part.get('mimeType') or part.get('mime_type') or self._guess_mime_from_uri(str(file_uri or ''))
if file_uri:
return {"fileData": {"mimeType": str(mime_type), "fileUri": str(file_uri)}}
if part_type in {'inline_data', 'inlineData'}:
inline = part.get('inlineData') or part.get('inline_data') or part
if isinstance(inline, dict):
data = inline.get('data')
mime_type = inline.get('mimeType') or inline.get('mime_type') or part.get('mimeType') or part.get('mime_type')
if data and mime_type:
return {"inlineData": {"mimeType": str(mime_type), "data": str(data)}}
normalized: dict[str, Any] = {}
for k, v in part.items():
if k == 'type':
continue
normalized[snake_to_camel(k)] = v
return normalized
def _parse_data_uri(self, uri: str) -> tuple[str, str]:
try:
header, data = uri.split(',', 1)
mime = header.split(':', 1)[1].split(';', 1)[0]
return mime, data
except (ValueError, IndexError):
return "", ""
def _guess_mime_from_uri(self, uri: str) -> str:
lower = uri.lower().split('?', 1)[0].split('#', 1)[0]
if lower.endswith(('.jpg', '.jpeg')):
return 'image/jpeg'
if lower.endswith('.webp'):
return 'image/webp'
if lower.endswith('.gif'):
return 'image/gif'
if lower.endswith('.png'):
return 'image/png'
if lower.endswith('.mp4'):
return 'video/mp4'
if lower.endswith('.mov'):
return 'video/quicktime'
if lower.endswith('.webm'):
return 'video/webm'
if lower.endswith('.mp3'):
return 'audio/mpeg'
if lower.endswith('.wav'):
return 'audio/wav'
if lower.endswith('.ogg'):
return 'audio/ogg'
if lower.endswith('.pdf'):
return 'application/pdf'
if lower.endswith('.txt'):
return 'text/plain'
return 'image/png'
def _convert_tools_format(self, data: Any) -> Any:
"""专门处理工具格式转换,统一转换为 camelCase"""
if isinstance(data, dict):
new_dict: dict[str, Any] = {}
data_dict: dict[str, Any] = cast(dict[str, Any], data)
for k, v in data_dict.items():
camel_k = snake_to_camel(k) if '_' in k else k
# 转换 function_declarations 为 functionDeclarations
if k in ['function_declarations', 'functionDeclarations']:
new_dict['functionDeclarations'] = self._convert_tools_format(v)
elif k in ['google_search', 'googleSearch']:
new_dict['googleSearch'] = self._convert_tools_format(v) if isinstance(v, (dict, list)) else v
elif k in ['google_search_retrieval', 'googleSearchRetrieval']:
new_dict['googleSearchRetrieval'] = self._convert_tools_format(v) if isinstance(v, (dict, list)) else v
elif k in ['code_execution', 'codeExecution']:
new_dict['codeExecution'] = self._convert_tools_format(v) if isinstance(v, (dict, list)) else v
elif k in ['url_context', 'urlContext']:
new_dict['urlContext'] = self._convert_tools_format(v) if isinstance(v, (dict, list)) else v
elif k in ['function_calling_config', 'functionCallingConfig']:
new_dict['functionCallingConfig'] = self._convert_tools_format(v)
elif k in ['allowed_function_names', 'allowedFunctionNames']:
new_dict['allowedFunctionNames'] = self._convert_tools_format(v) if isinstance(v, (dict, list)) else v
elif camel_k == "parametersJsonSchema" and isinstance(v, dict):
# parametersJsonSchema 需要特殊处理,确保 properties 和 required 字段一致
new_dict['parametersJsonSchema'] = self._convert_parameters_schema(cast(dict[str, Any], v))
elif k == "parameters" and isinstance(v, dict):
# Schema 对象需要特殊处理
converted_v = v.copy() if isinstance(v, dict) else v
new_dict[k] = self._ensure_function_parameters_schema(self._to_native_schema(converted_v))
elif k == "input_schema" and isinstance(v, dict):
new_dict['parameters'] = self._ensure_function_parameters_schema(self._to_native_schema(cast(dict[str, Any], v)))
elif k == "inputSchema" and isinstance(v, dict):
new_dict['parameters'] = self._ensure_function_parameters_schema(self._to_native_schema(cast(dict[str, Any], v)))
elif k == "name" and not v: # Vcore AI Function name cannot be empty
continue
elif k == "name" and v:
new_dict[k] = self._sanitize_function_name(str(v))
else:
# 对于其他字段,转换为 camelCase(除了特殊字段)
new_dict[camel_k] = self._convert_tools_format(v) if isinstance(v, (dict, list)) else v
return new_dict
elif isinstance(data, list):
return [self._convert_tools_format(item) for item in cast(list[Any], data)]
else:
return data
def _convert_parameters_schema(self, schema: dict[str, Any]) -> dict[str, Any]:
"""
转换 parametersJsonSchema,确保 properties 和 required 字段中的参数名一致
统一使用 snake_case 格式,避免 camelCase 和 snake_case 混用导致的不匹配
"""
new_schema: dict[str, Any] = schema.copy()
unsupported_keys = {
'$schema', '$id', '$defs', 'definitions', 'additionalProperties',
'patternProperties', 'unevaluatedProperties', 'dependentSchemas',
'if', 'then', 'else', 'allOf', 'anyOf', 'oneOf', 'not',
'examples', 'default', 'nullable'
}
for key in unsupported_keys:
new_schema.pop(key, None)
if isinstance(new_schema.get('type'), list):
non_null_types = [item for item in cast(list[Any], new_schema['type']) if item != 'null']
new_schema['type'] = non_null_types[0] if non_null_types else 'string'
# 处理 properties 字段:将 camelCase 转换为 snake_case
if 'properties' in new_schema and isinstance(new_schema['properties'], dict):
old_properties: dict[str, Any] = cast(dict[str, Any], new_schema['properties'])
new_properties: dict[str, Any] = {}
for prop_name, prop_def in old_properties.items():
# 将 camelCase 转换为 snake_case
snake_name = camel_to_snake(str(prop_name))
new_properties[snake_name] = prop_def
# 递归处理嵌套的 schema
if isinstance(prop_def, dict):
new_properties[snake_name] = self._convert_parameters_schema(cast(dict[str, Any], prop_def))
new_schema['properties'] = new_properties
# 处理 required 字段:确保使用 snake_case(通常已经是正确的)
if 'required' in new_schema and isinstance(new_schema['required'], list):
# required 字段通常已经是 snake_case,但为了保险起见,也进行转换
new_required: list[Any] = []
for req_name in cast(list[Any], new_schema['required']):
if isinstance(req_name, str):
snake_name = camel_to_snake(req_name)
new_required.append(snake_name)
else:
new_required.append(req_name)
new_schema['required'] = new_required
# 处理其他可能包含 schema 的字段
for key, value in list(new_schema.items()):
if key not in ['properties', 'required'] and isinstance(value, dict):
new_schema[key] = self._convert_parameters_schema(cast(dict[str, Any], value))
return new_schema
def _sanitize_function_name(self, name: str) -> str:
cleaned = _GEMINI_FUNCTION_NAME_RE.sub("_", name.strip())[:64].strip("._-")
if cleaned and not (cleaned[0].isalpha() or cleaned[0] == "_"):
cleaned = f"tool_{cleaned}"[:64]
return cleaned or "tool"
def _ensure_function_parameters_schema(self, schema: Any) -> dict[str, Any]:
if not isinstance(schema, dict):
return {"type": "OBJECT", "properties": []}
ensured = cast(dict[str, Any], schema).copy()
schema_type = ensured.get('type')
if not schema_type:
ensured['type'] = 'OBJECT'
elif isinstance(schema_type, str):
ensured['type'] = schema_type.upper()
if ensured.get('type') == 'OBJECT' and 'properties' not in ensured:
ensured['properties'] = []
return ensured
def _to_native_schema(self, standard_schema: dict[str, Any]) -> dict[str, Any]:
"""
将标准 JSON Schema 转换为 Vcore AI 原生 Map-style Schema
Args:
standard_schema: 标准 JSON Schema 对象
Returns:
Vcore AI 原生 Schema
"""
native_schema = standard_schema.copy()
unsupported_keys = {
'$schema', '$id', '$defs', 'definitions', 'additionalProperties',
'patternProperties', 'unevaluatedProperties', 'dependentSchemas',
'if', 'then', 'else', 'allOf', 'anyOf', 'oneOf', 'not',
'examples', 'default', 'nullable'
}
for key in unsupported_keys:
native_schema.pop(key, None)
# Vcore AI 要求类型必须是大写 (例如: STRING, OBJECT, INTEGER)
if 'type' in native_schema and isinstance(native_schema['type'], list):
non_null_types = [item for item in cast(list[Any], native_schema['type']) if item != 'null']
native_schema['type'] = non_null_types[0] if non_null_types else 'string'
if 'type' in native_schema and isinstance(native_schema['type'], str):
native_schema['type'] = native_schema['type'].upper()
if 'properties' in native_schema and isinstance(native_schema['properties'], dict):
native_props: list[dict[str, str | dict[str, Any]]] = []
props_dict = cast(dict[str, Any], native_schema['properties'])
for key, value in props_dict.items():
# 递归处理嵌套对象
if isinstance(value, dict):
converted_value = self._to_native_schema(cast(dict[str, Any], value))
else:
converted_value = {}
native_props.append({
"key": str(key),
"value": converted_value
})
native_schema['properties'] = native_props
# 处理数组项
if 'items' in native_schema and isinstance(native_schema['items'], dict):
items_dict = cast(dict[str, Any], native_schema['items'])
native_schema['items'] = self._to_native_schema(items_dict)
return native_schema
def _handle_system_instruction(self, new_variables: dict[str, Any]) -> None:
"""处理 systemInstruction:如果没有 user content,则转换为 user message"""
system_instruction_content = new_variables.get('systemInstruction')
if not system_instruction_content:
return
contents = new_variables.get('contents', [])
# 检查是否已有 user 角色
contents_list: list[Any] = cast(list[Any], contents) if isinstance(contents, list) else []
has_user_role = any(
isinstance(content, dict) and cast(dict[str, Any], content).get('role') == 'user'
for content in contents_list
)
if has_user_role:
return
# 提取文本内容
text_from_system = self._extract_text_from_instruction(system_instruction_content)
if not text_from_system:
return
# 转换为 user message
user_message = {
'role': 'user',
'parts': [{'text': text_from_system}]
}
# 显式转换 contents 为 list[Any] 以修复 pylance 报错
new_contents: list[Any] = list(contents_list)
new_contents.insert(0, user_message)
new_variables['contents'] = new_contents
del new_variables['systemInstruction']
def _extract_text_from_instruction(self, instruction: Any) -> str:
"""从 system instruction 中提取文本内容"""
if isinstance(instruction, str):
return instruction
elif isinstance(instruction, dict):
instruction_dict = cast(dict[str, Any], instruction)
parts = instruction_dict.get('parts', [])
if isinstance(parts, list):
text_parts = []
for part in parts:
if isinstance(part, dict) and 'text' in part:
text_parts.append(str(part['text']))
return "".join(text_parts)
return ""
def _normalize_tools_format(self, tools: Any) -> list[dict[str, Any]]:
"""标准化 tools 格式为 Vcore AI 期望的格式 (List[Tool])"""
converted_tools: Any = self._convert_tools_format(tools)
tool_keys = {
'functionDeclarations', 'googleSearch', 'googleSearchRetrieval',
'codeExecution', 'retrieval', 'urlContext'
}
if isinstance(converted_tools, dict):
# 如果是字典,且包含 functionDeclarations,将其包裹在列表中
if any(key in converted_tools for key in tool_keys):
normalized_tool = self._normalize_single_tool(cast(dict[str, Any], converted_tools), tool_keys)
return [normalized_tool] if normalized_tool else []
# 如果是单个 FunctionDeclaration,包裹成 Tool 再包裹在列表中
if 'name' in converted_tools:
decl = self._normalize_function_declaration(cast(dict[str, Any], converted_tools))
return [{"functionDeclarations": [decl]}] if decl else []
return []
if not isinstance(converted_tools, list) or len(cast(list[Any], converted_tools)) == 0:
return []
converted_tools_list: list[Any] = cast(list[Any], converted_tools)
normalized_tools: list[dict[str, Any]] = []
function_decls: list[dict[str, Any]] = []
for item in converted_tools_list:
if not isinstance(item, dict):
continue
item_dict = cast(dict[str, Any], item)
if any(key in item_dict for key in tool_keys):
normalized_tool = self._normalize_single_tool(item_dict, tool_keys)
if normalized_tool:
normalized_tools.append(normalized_tool)
elif item_dict.get('name'):
decl = self._normalize_function_declaration(item_dict)
if decl:
function_decls.append(decl)
if function_decls:
normalized_tools.insert(0, {"functionDeclarations": function_decls})
return normalized_tools
def _normalize_single_tool(self, tool: dict[str, Any], tool_keys: set[str]) -> dict[str, Any] | None:
normalized = {k: v for k, v in tool.items() if k in tool_keys or k.startswith('x')}
func_decls = tool.get('functionDeclarations')
if isinstance(func_decls, list):
declarations: list[dict[str, Any]] = []
for decl in cast(list[Any], func_decls):
if isinstance(decl, dict):
normalized_decl = self._normalize_function_declaration(cast(dict[str, Any], decl))
if normalized_decl:
declarations.append(normalized_decl)
if declarations:
normalized['functionDeclarations'] = declarations
else:
normalized.pop('functionDeclarations', None)
for native_key in ('googleSearch', 'googleSearchRetrieval', 'codeExecution', 'retrieval', 'urlContext'):
if native_key in tool:
native_value = tool.get(native_key)
normalized[native_key] = native_value if isinstance(native_value, dict) else {}
return normalized if any(key in normalized for key in tool_keys) else None
def _normalize_function_declaration(self, declaration: dict[str, Any]) -> dict[str, Any] | None:
raw_name = declaration.get('name') or declaration.get('function') or declaration.get('functionName')
if not raw_name:
return None
normalized = declaration.copy()
normalized['name'] = self._sanitize_function_name(str(raw_name))
if 'parameters' in normalized:
normalized['parameters'] = self._ensure_function_parameters_schema(normalized['parameters'])
elif 'parametersJsonSchema' in normalized and isinstance(normalized['parametersJsonSchema'], dict):
normalized['parametersJsonSchema'] = self._convert_parameters_schema(cast(dict[str, Any], normalized['parametersJsonSchema']))
else:
normalized['parameters'] = {"type": "OBJECT", "properties": []}
return normalized
def _normalize_tool_config(self, tool_config: Any) -> Any:
converted = self._convert_tools_format(tool_config)
if not isinstance(converted, dict):
return converted
config = cast(dict[str, Any], converted)
fcc = config.get('functionCallingConfig')
if isinstance(fcc, dict):
fcc_dict = cast(dict[str, Any], fcc).copy()
mode = fcc_dict.get('mode')
if isinstance(mode, str):
mode_upper = mode.upper()
if mode_upper in {'AUTO', 'ANY', 'NONE', 'MODE_UNSPECIFIED'}:
fcc_dict['mode'] = mode_upper
allowed = fcc_dict.get('allowedFunctionNames')
if isinstance(allowed, list):
fcc_dict['allowedFunctionNames'] = [
self._sanitize_function_name(str(name))
for name in cast(list[Any], allowed)
if name is not None and str(name).strip()
]
if not fcc_dict.get('allowedFunctionNames'):
fcc_dict.pop('allowedFunctionNames', None)
config['functionCallingConfig'] = fcc_dict
if not config.get('functionCallingConfig'):
config.pop('functionCallingConfig', None)
return config
def _handle_inline_data_case(self, contents: Any) -> Any:
"""
递归处理 contents,确保 inlineData/mimeType 字段名正确 (适配各种客户端传参)
"""
if isinstance(contents, list):
return [self._handle_inline_data_case(item) for item in cast(list[Any], contents)]
if isinstance(contents, dict):
new_dict: dict[str, Any] = {}
for k, v in cast(dict[str, Any], contents).items():
camel_k = snake_to_camel(k)
if k == 'thought_signature':
new_dict['thoughtSignature'] = self._handle_inline_data_case(v)
continue
if camel_k == 'inlineData' and isinstance(v, dict):
v_dict = cast(dict[str, Any], v)
new_inline_data = {}
for ik, iv in v_dict.items():
camel_ik = snake_to_camel(ik)
new_inline_data[camel_ik] = iv
new_dict['inlineData'] = new_inline_data
elif camel_k == 'fileData' and isinstance(v, dict):
v_dict = cast(dict[str, Any], v)
new_file_data = {}
for ik, iv in v_dict.items():
camel_ik = snake_to_camel(ik)
new_file_data[camel_ik] = iv
new_dict['fileData'] = new_file_data
elif camel_k == 'functionCall' and isinstance(v, dict):
func_call = self._convert_tools_format(v)
if isinstance(func_call, dict):
new_dict['functionCall'] = func_call
elif camel_k == 'functionResponse' and isinstance(v, dict):
func_response = self._convert_tools_format(v)
if isinstance(func_response, dict):
new_dict['functionResponse'] = func_response
else:
new_dict[camel_k] = self._handle_inline_data_case(v)
return new_dict
return contents
def _handle_base64_in_contents(self, contents: Any) -> Any:
"""
递归处理 contents 中的 base64 数据。
将 URL-safe Base64 转换为标准 Base64 并补全 padding。
"""
try:
if isinstance(contents, list):
res_list: list[Any] = [self._handle_base64_in_contents(item) for item in cast(list[Any], contents)]
return cast(Any, res_list)
if isinstance(contents, dict):
new_dict: dict[str, Any] = {}
for k, v in cast(dict[str, Any], contents).items():
if k == 'inlineData' and isinstance(v, dict):
v_dict = cast(dict[str, Any], v)
if 'data' in v_dict and isinstance(v_dict['data'], str):
try:
b64_data: str = v_dict['data']
b64_data = b64_data.replace('-', '+').replace('_', '/')
padding = len(b64_data) % 4
if padding:
b64_data += '=' * (4 - padding)
new_inline_data = v_dict.copy()
new_inline_data['data'] = b64_data
new_dict[k] = new_inline_data
continue
except Exception:
pass
new_dict[k] = self._handle_base64_in_contents(v)
return cast(Any, new_dict)
return contents
except Exception as e:
logger.warning(f"Base64 内容处理失败: {e}")
return cast(Any, contents)
def _filter_empty_contents(self, contents: Any) -> Any:
"""
过滤掉空的 contents(parts 为空数组的消息)
Vcore AI 要求每个 content 至少包含一个 part
"""
if not isinstance(contents, list):
return contents
filtered_contents: list[Any] = []
contents_list: list[Any] = cast(list[Any], contents)
# 收集所有 functionCall 的名称,用于修复 functionResponse
function_call_names: list[str] = []
for content in contents_list:
if isinstance(content, dict):
content_dict = cast(dict[str, Any], content)
parts = content_dict.get('parts', [])
if isinstance(parts, list):
for part in cast(list[Any], parts):
if isinstance(part, dict):
part_dict = cast(dict[str, Any], part)
func_call = part_dict.get('functionCall')
if not isinstance(func_call, dict):
func_call = part_dict.get('function_call')
if isinstance(func_call, dict):
func_call_dict = cast(dict[str, Any], func_call)
name = func_call_dict.get('name')
if name and isinstance(name, str):
function_call_names.append(self._sanitize_function_name(name))
for content in contents_list:
if isinstance(content, dict):
content_dict: dict[str, Any] = cast(dict[str, Any], content)
parts = content_dict.get('parts', [])
# 只保留有 parts 且 parts 不为空的 content
if isinstance(parts, list) and len(cast(list[Any], parts)) > 0:
parts_list: list[Any] = cast(list[Any], parts)
# 过滤并验证 parts 中的有效内容
valid_parts: list[Any] = []
for part in parts_list:
if isinstance(part, dict):
part_dict = cast(dict[str, Any], part)
# 清理并修复 part
cleaned_part = self._clean_part_metadata(part_dict, function_call_names)
if cleaned_part:
valid_parts.append(cleaned_part)
if valid_parts:
# 更新 content 的 parts
filtered_content = content_dict.copy()
filtered_content['parts'] = valid_parts
filtered_contents.append(filtered_content)
else:
logger.debug(f"过滤掉空的 content: role={content_dict.get('role', 'unknown')}")
else:
logger.debug(f"过滤掉空的 content: role={content_dict.get('role', 'unknown')}")
else:
# 非 Dict 类型的 content,保留
filtered_contents.append(content)
return filtered_contents
def _clean_part_metadata(self, part_dict: dict[str, Any], function_call_names: list[str]) -> dict[str, Any] | None:
"""
清理 part 中的空元数据字段,修复无效的 functionResponse
Args:
part_dict: 原始 part 字典
function_call_names: 可用的函数调用名称列表
Returns:
清理后的 part 字典,如果 part 无效则返回 None
"""
cleaned_part: dict[str, Any] = {}
has_valid_content = False
# 处理文本内容
if 'text' in part_dict:
text_value = part_dict['text']
if text_value is not None and str(text_value) != "":
cleaned_part['text'] = text_value
has_valid_content = True
# 处理思考标记
if 'thought' in part_dict:
cleaned_part['thought'] = part_dict['thought']
# 处理思考签名 (thoughtSignature)
if 'thoughtSignature' in part_dict:
cleaned_part['thoughtSignature'] = part_dict['thoughtSignature']
elif 'thought_signature' in part_dict:
cleaned_part['thoughtSignature'] = part_dict['thought_signature']
# 处理函数调用
if 'functionCall' in part_dict or 'function_call' in part_dict:
func_call = part_dict.get('functionCall') or part_dict.get('function_call')
if isinstance(func_call, dict):
func_call_dict = cast(dict[str, Any], func_call)
if func_call_dict.get('name'): # 只保留有名称的函数调用
fixed_func_call = func_call_dict.copy()
fixed_func_call['name'] = self._sanitize_function_name(str(fixed_func_call['name']))
if 'args' not in fixed_func_call or fixed_func_call.get('args') is None:
fixed_func_call['args'] = {}
if isinstance(fixed_func_call.get('args'), str):
try:
fixed_func_call['args'] = json.loads(cast(str, fixed_func_call['args']))
except json.JSONDecodeError:
fixed_func_call['args'] = {"raw": fixed_func_call['args']}
if not isinstance(fixed_func_call.get('args'), dict):
fixed_func_call['args'] = {"value": fixed_func_call.get('args')}
cleaned_part['functionCall'] = fixed_func_call
has_valid_content = True
# 处理函数响应
if 'functionResponse' in part_dict or 'function_response' in part_dict:
func_response = part_dict.get('functionResponse') or part_dict.get('function_response')
if isinstance(func_response, dict):
func_response_dict = cast(dict[str, Any], func_response)
current_name = (
func_response_dict.get('name')
or func_response_dict.get('functionName')
or func_response_dict.get('function_name')
)
# 如果 name 为空,尝试修复
if not current_name and function_call_names:
inferred_name = function_call_names[-1] # 使用最后一个 functionCall 的名称
logger.warning(f"修复空的 functionResponse.name,推断为: {inferred_name}")
fixed_func_response = func_response_dict.copy()
fixed_func_response['name'] = self._sanitize_function_name(inferred_name)
fixed_func_response['response'] = self._coerce_function_response(fixed_func_response.get('response', {}))
cleaned_part['functionResponse'] = fixed_func_response
has_valid_content = True
elif current_name:
# name 不为空,直接保留
fixed_func_response = func_response_dict.copy()
fixed_func_response['name'] = self._sanitize_function_name(str(current_name))
fixed_func_response['response'] = self._coerce_function_response(fixed_func_response.get('response', {}))
cleaned_part['functionResponse'] = fixed_func_response
has_valid_content = True
# 如果 name 为空且无法推断,则丢弃这个 functionResponse
# 处理内联数据
if 'inlineData' in part_dict:
inline_data = part_dict['inlineData']
if isinstance(inline_data, dict):
inline_data_dict = cast(dict[str, Any], inline_data)
# 只保留有实际数据的 inlineData
if (inline_data_dict.get('data') and
str(inline_data_dict['data']).strip() and
inline_data_dict.get('mimeType') and
str(inline_data_dict['mimeType']).strip()):
cleaned_part['inlineData'] = inline_data
has_valid_content = True
# 处理文件数据
if 'fileData' in part_dict:
file_data = part_dict['fileData']
if isinstance(file_data, dict):
file_data_dict = cast(dict[str, Any], file_data)
# 只保留有实际数据的 fileData
if (file_data_dict.get('fileUri') and
str(file_data_dict['fileUri']).strip() and
file_data_dict.get('mimeType') and
str(file_data_dict['mimeType']).strip()):
cleaned_part['fileData'] = file_data
has_valid_content = True
# 处理代码执行相关 part
for code_key in ('executableCode', 'codeExecutionResult'):
if code_key in part_dict and part_dict[code_key]:
cleaned_part[code_key] = part_dict[code_key]
has_valid_content = True
# 透传 Gemini 新增/实验 part;只在字段非空时保留。
# 核心 part 字段已在上方严格清洗,不能兜底透传;否则会把
# functionResponse.name=null、空 inlineData/fileData/functionCall 等非法字段重新加入上游 payload。
passthrough_part_keys = (
'videoMetadata', 'mediaResolution', 'thought', 'thoughtSignature', 'thought_signature',
)
for key in passthrough_part_keys:
if key in part_dict and key not in cleaned_part and part_dict[key]:
cleaned_key = 'thoughtSignature' if key == 'thought_signature' else key
cleaned_part[cleaned_key] = part_dict[key]
if cleaned_key not in {'thought', 'thoughtSignature', 'videoMetadata', 'mediaResolution'}:
has_valid_content = True
# 保留与有效媒体 part 绑定的元数据
for metadata_key in ('videoMetadata', 'mediaResolution'):
if metadata_key in part_dict and part_dict[metadata_key]:
cleaned_part[metadata_key] = part_dict[metadata_key]
# 只返回有有效内容的 part
if has_valid_content:
return cleaned_part
else:
logger.debug("过滤掉没有有效内容的 part")
return None
def _sanitize_vcore_variables_in_place(self, variables: dict[str, Any]) -> None:
"""最终出口兜底,确保发往匿名 Vcore 的 contents 不含非法核心 part。"""
contents = variables.get('contents')
if not isinstance(contents, list):
return
function_call_names: list[str] = []
for content in cast(list[Any], contents):
if not isinstance(content, dict):
continue
parts = cast(dict[str, Any], content).get('parts')
if not isinstance(parts, list):
continue
for part in cast(list[Any], parts):
if not isinstance(part, dict):
continue
func_call = cast(dict[str, Any], part).get('functionCall') or cast(dict[str, Any], part).get('function_call')
if isinstance(func_call, dict):
name = cast(dict[str, Any], func_call).get('name')
if name and str(name).strip():
function_call_names.append(self._sanitize_function_name(str(name)))
sanitized_contents: list[dict[str, Any]] = []
dropped_function_responses = 0
for content in cast(list[Any], contents):
if not isinstance(content, dict):
continue
content_dict = cast(dict[str, Any], content).copy()
parts = content_dict.get('parts')
if not isinstance(parts, list):
continue
sanitized_parts: list[dict[str, Any]] = []
for part in cast(list[Any], parts):
if not isinstance(part, dict):
continue
part_dict = cast(dict[str, Any], part)
cleaned = self._sanitize_vcore_part_final(part_dict, function_call_names)
if cleaned:
if ('functionResponse' in part_dict or 'function_response' in part_dict) and 'functionResponse' not in cleaned:
dropped_function_responses += 1
sanitized_parts.append(cleaned)
if sanitized_parts:
content_dict['parts'] = sanitized_parts
sanitized_contents.append(content_dict)
variables['contents'] = sanitized_contents
if dropped_function_responses:
logger.warning(f"最终出口丢弃非法 functionResponse part: count={dropped_function_responses}")
def _sanitize_vcore_part_final(self, part: dict[str, Any], function_call_names: list[str]) -> dict[str, Any] | None:
cleaned: dict[str, Any] = {}
has_content = False
text = part.get('text')
if text is not None and str(text) != "":
cleaned['text'] = text
has_content = True
if 'thought' in part:
cleaned['thought'] = part['thought']
signature = part.get('thoughtSignature') or part.get('thought_signature')
if signature:
cleaned['thoughtSignature'] = signature
func_call = part.get('functionCall') or part.get('function_call')
if isinstance(func_call, dict):
call = cast(dict[str, Any], func_call).copy()
name = call.get('name')
if name and str(name).strip():
call['name'] = self._sanitize_function_name(str(name))
call['args'] = self._coerce_function_args(call.get('args', {}))
cleaned['functionCall'] = call
has_content = True
func_response = part.get('functionResponse') or part.get('function_response')
if isinstance(func_response, dict):
response = cast(dict[str, Any], func_response).copy()
name = response.get('name') or response.get('functionName') or response.get('function_name')
if not name and function_call_names:
name = function_call_names[-1]
if name and str(name).strip():
response['name'] = self._sanitize_function_name(str(name))
response['response'] = self._coerce_function_response(response.get('response', {}))
cleaned['functionResponse'] = response
has_content = True
inline_data = part.get('inlineData') or part.get('inline_data')
if isinstance(inline_data, dict):
inline_dict = cast(dict[str, Any], inline_data)
data = inline_dict.get('data')
mime_type = inline_dict.get('mimeType') or inline_dict.get('mime_type')
if data and str(data).strip() and mime_type and str(mime_type).strip():
cleaned['inlineData'] = {**inline_dict, 'mimeType': str(mime_type)}
cleaned['inlineData'].pop('mime_type', None)
has_content = True
file_data = part.get('fileData') or part.get('file_data')
if isinstance(file_data, dict):
file_dict = cast(dict[str, Any], file_data)
file_uri = file_dict.get('fileUri') or file_dict.get('file_uri')
mime_type = file_dict.get('mimeType') or file_dict.get('mime_type')
if file_uri and str(file_uri).strip() and mime_type and str(mime_type).strip():
cleaned['fileData'] = {**file_dict, 'fileUri': str(file_uri), 'mimeType': str(mime_type)}
cleaned['fileData'].pop('file_uri', None)
cleaned['fileData'].pop('mime_type', None)
has_content = True
for code_key in ('executableCode', 'codeExecutionResult'):
if part.get(code_key):
cleaned[code_key] = part[code_key]
has_content = True
for metadata_key in ('videoMetadata', 'mediaResolution'):
if part.get(metadata_key):
cleaned[metadata_key] = part[metadata_key]
return cleaned if has_content else None
def _ensure_function_call_thought_signatures(self, contents: Any) -> Any:
"""
Gemini 3 工具调用要求历史中的 functionCall part 带 thoughtSignature。
如果客户端没有回传签名,使用官方保底哨兵值跳过校验,避免 400 错误。
"""
if not isinstance(contents, list):
return contents
patched_count = 0
patched_names: set[str] = set()
patched_contents: list[Any] = []
for content in cast(list[Any], contents):
if not isinstance(content, dict):
patched_contents.append(content)
continue
content_dict = cast(dict[str, Any], content)
parts = content_dict.get('parts')
if not isinstance(parts, list):
patched_contents.append(content)
continue
patched_parts: list[Any] = []
for part in cast(list[Any], parts):
if isinstance(part, dict):
part_dict = cast(dict[str, Any], part)
function_call = part_dict.get('functionCall')
if not isinstance(function_call, dict):
function_call = part_dict.get('function_call')
if isinstance(function_call, dict):
patched_part = part_dict.copy()
if 'function_call' in patched_part and 'functionCall' not in patched_part:
patched_part['functionCall'] = patched_part.pop('function_call')
if 'thought_signature' in patched_part and 'thoughtSignature' not in patched_part:
patched_part['thoughtSignature'] = patched_part.pop('thought_signature')
if not patched_part.get('thoughtSignature'):
func_call = cast(dict[str, Any], patched_part.get('functionCall') or {})
patched_count += 1
patched_names.add(str(func_call.get('name') or 'unknown'))
patched_part['thoughtSignature'] = 'skip_thought_signature_validator'
patched_parts.append(patched_part)
else:
patched_parts.append(part)
else:
patched_parts.append(part)
patched_content = content_dict.copy()
patched_content['parts'] = patched_parts
patched_contents.append(patched_content)
if patched_count:
logger.debug(
"已为 %d 个缺少 thoughtSignature 的 functionCall 使用兼容兜底,函数: %s",
patched_count,
", ".join(sorted(patched_names))
)
return patched_contents
def _handle_thought_signature(self, contents: Any) -> Any:
"""
处理 thoughtSignature 字段的 base64 编码
确保 thoughtSignature 字段正确编码为 base64 字符串
"""
import base64
if isinstance(contents, list):
return [self._handle_thought_signature(item) for item in cast(list[Any], contents)]
if isinstance(contents, dict):
new_dict: dict[str, Any] = {}
contents_dict: dict[str, Any] = cast(dict[str, Any], contents)
for k, v in contents_dict.items():
if k == 'parts' and isinstance(v, list):
v_list: list[Any] = cast(list[Any], v)
# 处理 parts 数组中的每个 part
new_parts: list[Any] = []
for part in v_list:
if isinstance(part, dict):
new_part: dict[str, Any] = cast(dict[str, Any], part).copy()
# 检查是否有 thoughtSignature 字段
if 'thoughtSignature' in new_part:
signature_value = new_part['thoughtSignature']
if isinstance(signature_value, str):
# 如果是特定的字符串,进行 base64 编码
if signature_value == "skip_thought_signature_validator":
encoded_bytes = base64.b64encode(signature_value.encode('utf-8'))
new_part['thoughtSignature'] = encoded_bytes.decode('utf-8')
# 如果已经是 base64 编码的字符串,保持不变
# 其他情况也保持不变
new_parts.append(new_part)
else:
new_parts.append(part)
new_dict[k] = new_parts
else:
new_dict[k] = self._handle_thought_signature(v) if isinstance(v, (dict, list)) else v
return new_dict
return contents
@staticmethod
def prepare_headers(creds: dict[str, Any]) -> dict[str, str]:
"""准备请求头"""
# 提取原始头信息
headers = RequestTransformer._extract_headers_from_creds(creds)
# 设置必要的头信息
headers['content-type'] = 'application/json'
# 移除可能导致问题的头
problematic_headers = [
'content-length', 'Content-Length', 'host', 'Host',
'connection', 'Connection', 'accept-encoding'
]
for header in problematic_headers:
headers.pop(header, None)
return headers
@staticmethod
def _extract_headers_from_creds(creds: dict[str, Any]) -> dict[str, str]:
"""从凭证中提取头信息"""
if hasattr(creds, 'model_dump') and hasattr(creds, 'headers'):
headers_attr = getattr(creds, 'headers')
if isinstance(headers_attr, dict):
return cast(dict[str, str], headers_attr).copy()
raw_headers = creds.get('headers')
if isinstance(raw_headers, dict):
return cast(dict[str, str], raw_headers).copy()
return {}
class ResponseAggregator:
"""响应聚合器"""
@staticmethod
def log_stream_forward_complete(prefix: str, chunks: int, bytes_total: int, elapsed: float) -> None:
logger.info(
f"{prefix} 流式转发完成: chunks={chunks}, "
f"bytes={bytes_total}, 耗时={elapsed:.1f}s"
)
@staticmethod
async def aggregate_stream(
stream_generator: Any,
_raw_image_response: bool = False,
progress_context: dict[str, str] | None = None,
) -> dict[str, Any]:
"""
聚合流式响应为非流式对象
"""
all_parts: list[dict[str, Any]] = []
candidate_parts_by_index: dict[int, list[dict[str, Any]]] = {}
finish_reason: str | None = None
finish_message: str | None = None
safety_ratings: list[dict[str, Any]] = []
citation_metadata: dict[str, Any] = {}
grounding_metadata: dict[str, Any] = {}
token_count: int | None = None
avg_logprobs: float | None = None
logprobs_result: dict[str, Any] | None = None
candidate_index = 0
usage_metadata: dict[str, Any] = {}
create_time: str | None = None
model_version: str | None = None
prompt_feedback: dict[str, Any] = {}
response_id: str | None = None
model_status: dict[str, Any] | None = None
buffered_chunk_count = 0
buffered_bytes_total = 0
aggregate_started_at = time.monotonic()
aggregate_error: Exception | None = None
try:
async for stream_item in stream_generator:
try:
if isinstance(stream_item, dict):
chunk = stream_item
chunk_bytes = 0
else:
actual_json_str = str(stream_item).strip()
if actual_json_str.startswith("data: "):
actual_json_str = actual_json_str[6:]
if not actual_json_str:
continue
chunk = json.loads(actual_json_str)
chunk_bytes = len(actual_json_str.encode('utf-8'))
buffered_chunk_count += 1
buffered_bytes_total += chunk_bytes
# 检查 chunk 是否包含错误 (使用统一解析逻辑)
parsed_error = parse_error_response(chunk)
if parsed_error:
raise parsed_error
# 提取顶层元数据
create_time = chunk.get('createTime') or create_time
model_version = chunk.get('modelVersion') or model_version
prompt_feedback = chunk.get('promptFeedback', {}) or prompt_feedback
response_id = chunk.get('responseId') or response_id
usage_metadata = chunk.get('usageMetadata', {}) or usage_metadata
model_status = chunk.get('modelStatus') or model_status
candidates = chunk.get('candidates', [])
if candidates:
candidates_to_read = candidates if _raw_image_response else candidates[:1]
for candidate_offset, candidate in enumerate(candidates_to_read):
if not isinstance(candidate, dict):
continue
candidate_result_index = candidate_offset
if candidate.get('index') is not None:
try:
candidate_result_index = int(candidate['index'])
except (TypeError, ValueError):
candidate_result_index = candidate_offset
# 提取 parts。raw 图片响应会读取所有 candidates,支持一次上游响应返回多张图;
# 普通 Gemini 非流式响应仍只聚合第一个 candidate,避免改变文本接口语义。
content_obj = candidate.get('content', {})
raw_parts = content_obj.get('parts', []) if isinstance(content_obj, dict) else []
parts = ResponseAggregator._normalize_response_parts(raw_parts)
if parts:
all_parts.extend(parts)
candidate_parts_by_index.setdefault(candidate_result_index, []).extend(
[cast(dict[str, Any], part) for part in cast(list[Any], parts) if isinstance(part, dict)]
)
# 提取 candidate 元数据;多图 raw 响应不依赖这些字段,只保留首个 candidate 的元数据。
if candidate_offset == 0:
finish_reason = candidate.get('finishReason') or finish_reason
finish_message = candidate.get('finishMessage') or finish_message
safety_ratings = candidate.get('safetyRatings') or safety_ratings
citation_metadata = candidate.get('citationMetadata') or citation_metadata
grounding_metadata = candidate.get('groundingMetadata') or grounding_metadata
if candidate.get('tokenCount') is not None:
token_count = candidate['tokenCount']
if candidate.get('avgLogprobs') is not None:
avg_logprobs = candidate['avgLogprobs']
if candidate.get('logprobsResult') is not None:
logprobs_result = candidate['logprobsResult']
if candidate.get('index') is not None:
candidate_index = candidate['index']
except json.JSONDecodeError as e:
logger.debug(f"JSON 解析失败,跳过此块: {e}")
continue
except VcoreError as e:
aggregate_error = e
raise
except Exception as e:
aggregate_error = e
raise InternalError(message=f"Non-streaming request error: {e}")
finally:
if buffered_chunk_count > 0:
progress_prefix = progress_context.get("prefix") if isinstance(progress_context, dict) else ""
prefix = f"{progress_prefix} " if progress_prefix else ""
state = "异常" if aggregate_error is not None else "完成"
log_fields = [
f"chunks={buffered_chunk_count}",
f"bytes={buffered_bytes_total}",
f"状态={state}",
]
log_fields.append(f"耗时={time.monotonic() - aggregate_started_at:.1f}s")
logger.info(
f"{prefix}非流式接收缓存结束: {', '.join(log_fields)}"
)
# 流式上游通常把一段连续文本拆成多个 text part。非流式响应若原样返回
# 多个相邻 text part,部分 SDK/客户端会在 part 边界自行插入换行或段落,
# 导致最终文本出现类似“当\n\n掉”“</\n\nrelationships>”的伪换行。
# 因此在构造非流式结果前合并相邻、同类型的文本 part,保持文本内容本身不变。
all_parts = ResponseAggregator._merge_adjacent_text_parts(all_parts)
if candidate_parts_by_index:
candidate_parts_by_index = {
idx: ResponseAggregator._merge_adjacent_text_parts(parts)
for idx, parts in candidate_parts_by_index.items()
}
# 处理图片响应特例
image_outputs = ResponseAggregator._extract_image_outputs(all_parts)
is_image_response = bool(image_outputs)
if _raw_image_response and image_outputs:
return ResponseAggregator._raw_image_response_from_images(image_outputs)
if is_image_response and candidate_parts_by_index:
result_candidates: list[dict[str, Any]] = []
for idx, parts in sorted(candidate_parts_by_index.items(), key=lambda item: item[0]):
item: dict[str, Any] = {
"index": idx,
"content": {
"parts": parts,
"role": "model",
},
}
if finish_reason:
item["finishReason"] = finish_reason.upper()
result_candidates.append(item)
result: dict[str, Any] = {"candidates": result_candidates}
optional_result_fields: dict[str, Any] = {
"createTime": create_time,
"modelVersion": model_version,
"promptFeedback": prompt_feedback,
"responseId": response_id,
"usageMetadata": usage_metadata,
"modelStatus": model_status,
}
for key, value in optional_result_fields.items():
if value is not None and value != {} and value != "":
result[key] = value
return result
# 处理图片响应特例
image_outputs = ResponseAggregator._extract_image_outputs(all_parts)
if _raw_image_response and image_outputs:
return ResponseAggregator._raw_image_response_from_images(image_outputs)
# 构建最终响应
if not all_parts:
all_parts = [{"text": " "}]
result_candidate: dict[str, Any] = {
"index": candidate_index
}
if finish_reason:
result_candidate["finishReason"] = finish_reason.upper()
result_candidate["content"] = {
"parts": all_parts,
"role": "model"
}
# 构建候选结果,只添加非空字段
optional_candidate_fields: dict[str, Any] = {
"finishMessage": finish_message,
"safetyRatings": safety_ratings,
"citationMetadata": citation_metadata,
"groundingMetadata": grounding_metadata,
"tokenCount": token_count,
"avgLogprobs": avg_logprobs,
"logprobsResult": logprobs_result
}
for key, value in optional_candidate_fields.items():
if value is not None and value != [] and value != {}:
result_candidate[key] = value
# 构建最终结果,只添加非空字段
result: dict[str, Any] = {"candidates": [result_candidate]}
optional_result_fields: dict[str, Any] = {
"createTime": create_time,
"modelVersion": model_version,
"promptFeedback": prompt_feedback,
"responseId": response_id,
"usageMetadata": usage_metadata,
"modelStatus": model_status
}
for key, value in optional_result_fields.items():
if value is not None and value != {} and value != "":
result[key] = value
return result
@staticmethod
def _extract_image_outputs(parts: list[dict[str, Any]]) -> list[dict[str, str]]:
images: list[dict[str, str]] = []
for part in parts:
inline_data = part.get('inlineData') or part.get('inline_data')
if isinstance(inline_data, dict):
mime_type = inline_data.get('mimeType') or inline_data.get('mime_type') or 'image/png'
data = inline_data.get('data')
if isinstance(data, str) and data.strip():
images.append({"b64_json": data, "mime_type": str(mime_type)})
file_data = part.get('fileData') or part.get('file_data')
if isinstance(file_data, dict):
mime_type = file_data.get('mimeType') or file_data.get('mime_type') or 'image/png'
file_uri = file_data.get('fileUri') or file_data.get('file_uri') or file_data.get('uri') or file_data.get('url')
if isinstance(file_uri, str) and file_uri.strip():
images.append({"url": file_uri.strip(), "mime_type": str(mime_type)})
return images
@staticmethod
def _merge_adjacent_text_parts(parts: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""合并相邻文本 part,避免非流式客户端在 part 边界额外插入换行。"""
merged: list[dict[str, Any]] = []
def mergeable_text_signature(part: dict[str, Any]) -> tuple[Any, Any] | None:
if "text" not in part:
return None
# 只要是纯文本 part 就允许合并。Vcore/Gemini 可能在 text part 上附带
# thought/thoughtSignature 以外的无害元数据;这些元数据如果阻止合并,
# Gemini 非流式客户端仍会在碎片边界插入额外换行。
blocking_keys = {
"inlineData", "inline_data", "fileData", "file_data",
"functionCall", "function_call", "functionResponse", "function_response",
"executableCode", "executable_code", "codeExecutionResult", "code_execution_result",
}
if any(key in part and ResponseAggregator._has_meaningful_part_value(part.get(key)) for key in blocking_keys):
return None
# thoughtSignature 标记了独立上下文,保守起见不跨 signature 合并。
signature = part.get("thoughtSignature") or part.get("thought_signature")
return (bool(part.get("thought")), signature)
for raw_part in parts:
part = dict(raw_part)
sig = mergeable_text_signature(part)
if sig is None or not merged:
merged.append(part)
continue
previous = merged[-1]
previous_sig = mergeable_text_signature(previous)
if previous_sig == sig:
previous["text"] = str(previous.get("text", "")) + str(part.get("text", ""))
else:
merged.append(part)
return merged
@staticmethod
def _normalize_response_parts(raw_parts: Any) -> list[dict[str, Any]]:
"""清洗上游 part 中的空占位字段,避免 Gemini 非流式返回碎片化 parts。"""
if not isinstance(raw_parts, list):
return []
normalized: list[dict[str, Any]] = []
placeholder_keys = {
"inlineData", "inline_data", "fileData", "file_data",
"functionCall", "function_call", "functionResponse", "function_response",
"executableCode", "executable_code", "codeExecutionResult", "code_execution_result",
}
for item in raw_parts:
if not isinstance(item, dict):
continue
part = dict(cast(dict[str, Any], item))
if 'inline_data' in part and 'inlineData' not in part:
inline_data = part['inline_data']
if isinstance(inline_data, dict):
part['inlineData'] = {snake_to_camel(str(k)): v for k, v in cast(dict[Any, Any], inline_data).items()}
else:
part['inlineData'] = inline_data
if 'file_data' in part and 'fileData' not in part:
file_data = part['file_data']
if isinstance(file_data, dict):
part['fileData'] = {snake_to_camel(str(k)): v for k, v in cast(dict[Any, Any], file_data).items()}
else:
part['fileData'] = file_data
for key in list(part.keys()):
if key in placeholder_keys and not ResponseAggregator._has_meaningful_part_value(part.get(key)):
part.pop(key, None)
# 部分上游/客户端会把 part 类型放在 data/type 上;Gemini 响应无需保留
# data='text' 这类非标准占位,否则下游可能把 text part 当成多模态 part。
if part.get("data") == "text":
part.pop("data", None)
if part.get("type") == "text":
part.pop("type", None)
normalized.append(part)
return normalized
@staticmethod
def _has_meaningful_part_value(value: Any) -> bool:
"""判断 part 子字段是否真实有内容;空壳对象不应阻止文本合并。"""
if value is None or value is False:
return False
if isinstance(value, str):
return value != ""
if isinstance(value, dict):
return any(ResponseAggregator._has_meaningful_part_value(v) for v in value.values())
if isinstance(value, (list, tuple, set)):
return any(ResponseAggregator._has_meaningful_part_value(v) for v in value)
return True
@staticmethod
def _raw_image_response_from_images(images: list[dict[str, str]]) -> dict[str, Any]:
return {
"created": int(time.time()),
"data": [
dict(image)
for image in images
],
}