570 lines
27 KiB
Python
570 lines
27 KiB
Python
"""
|
||
OpenAI兼容服务基类
|
||
为LLM和VLM服务提供统一的OpenAI兼容API处理逻辑
|
||
"""
|
||
|
||
import json
|
||
import time
|
||
import asyncio
|
||
import httpx
|
||
from typing import Optional, Dict, Any, List, Callable
|
||
from .core import BaseAPIService, HTTPClientPool
|
||
from .ollama_utils import wait_before_ollama_unload
|
||
from ..utils.common import (
|
||
format_api_error, ProgressBar, log_complete, log_error,
|
||
PREFIX, PROCESS_PREFIX, WARN_PREFIX, ERROR_PREFIX, format_elapsed_time,
|
||
TASK_IMAGE_CAPTION, TASK_VIDEO_CAPTION
|
||
)
|
||
from .thinking_control import build_thinking_suppression
|
||
|
||
|
||
class OpenAICompatibleService(BaseAPIService):
|
||
"""
|
||
OpenAI兼容API服务基类
|
||
处理所有OpenAI格式的API请求（智谱、硅基流动、302.ai、Ollama等）
|
||
"""
|
||
|
||
# ---已知的API端点路径（用于智能检测）---
|
||
_known_endpoints = ['/chat/completions', '/v1/messages', '/completions']
|
||
|
||
@staticmethod
|
||
def parse_api_url(raw_url: str) -> str:
|
||
"""
|
||
智能解析 base_url，生成最终请求地址
|
||
|
||
规则：
|
||
1. '#' 结尾 → 强制使用完整地址（移除#）
|
||
2. 已包含已知端点路径 → 直接使用，不再拼接
|
||
3. 其他 → 正常拼接 /chat/completions
|
||
|
||
参数:
|
||
raw_url: 用户输入的原始URL
|
||
|
||
返回:
|
||
str: 最终请求地址
|
||
"""
|
||
if not raw_url:
|
||
return ''
|
||
|
||
url = raw_url.strip()
|
||
|
||
# 规则1：井号强制模式 - 用户明确要求使用完整地址
|
||
if url.endswith('#'):
|
||
return url[:-1].rstrip('/')
|
||
|
||
# 规则2：智能检测 - 检查URL中是否已包含已知的API端点
|
||
for endpoint in OpenAICompatibleService._known_endpoints:
|
||
if endpoint in url:
|
||
# 已包含完整端点，直接返回（移除末尾斜杠）
|
||
return url.rstrip('/')
|
||
|
||
if 'api.openai.com' in url and '/v1' not in url:
|
||
url = url.rstrip('/') + '/v1'
|
||
|
||
# 规则3：常规模式 - 需要拼接 /chat/completions
|
||
return url.rstrip('/') + '/chat/completions'
|
||
|
||
# _provider_base_urls 和 _provider_display_names 已移除，相关逻辑改由 config_manager 统一管理
|
||
|
||
@staticmethod
|
||
def _filter_payload(payload: Dict[str, Any], level: int) -> Dict[str, Any]:
|
||
"""
|
||
根据重试级别清洗请求体 (简化的三级降级策略)
|
||
|
||
Level 0: 完整请求 (按用户设置发送)
|
||
Level 1: 移除思维链参数 (thinking, enable_thinking, reasoning_effort, 等)
|
||
Level 2: 最小可用集 (仅 model, messages, stream)
|
||
"""
|
||
if level <= 0:
|
||
return payload.copy()
|
||
|
||
filtered = payload.copy()
|
||
|
||
# Level 1: 移除思维链参数 + response_format
|
||
# 注：response_format 在 Zhipu GLM-4V 等严格服务商上会触发 400，Level-1 同步移除
|
||
thinking_keys = [
|
||
"thinking", "enable_thinking", "reasoning_effort",
|
||
"reasoning", "thinking_level", "think",
|
||
"response_format", # BUG-02 修复：Level-1 一并移除，防止部分服务商拒绝该字段
|
||
]
|
||
for k in thinking_keys:
|
||
filtered.pop(k, None)
|
||
|
||
if level >= 2:
|
||
# Level 2: 最小可用集 - 仅保留必选参数
|
||
core_keys = ["model", "messages", "stream"]
|
||
filtered = {k: filtered[k] for k in core_keys if k in filtered}
|
||
|
||
return filtered
|
||
|
||
@staticmethod
|
||
def _merge_system_prompts(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||
"""
|
||
合并多条System Message为单一条目
|
||
并将System Message置于列表首位
|
||
解决部分服务商不支持多条System Message的问题
|
||
"""
|
||
system_contents = []
|
||
other_messages = []
|
||
|
||
for msg in messages:
|
||
if msg.get('role') == 'system':
|
||
content = msg.get('content', '')
|
||
if content:
|
||
system_contents.append(content)
|
||
else:
|
||
other_messages.append(msg)
|
||
|
||
if not system_contents:
|
||
return messages
|
||
|
||
# 合并内容
|
||
merged_system = "\n\n".join(system_contents)
|
||
|
||
# 构建新列表：System在首位 + 其他消息
|
||
return [{"role": "system", "content": merged_system}] + other_messages
|
||
|
||
@classmethod
|
||
async def _http_request_chat_completions(
|
||
cls,
|
||
base_url: str,
|
||
api_key: str,
|
||
model: str,
|
||
messages: List[Dict[str, Any]],
|
||
temperature: float = 0.7,
|
||
top_p: float = 0.9,
|
||
max_tokens: int = 2000,
|
||
thinking_extra: Optional[Dict[str, Any]] = None,
|
||
enable_advanced_params: bool = False,
|
||
stream_callback: Optional[Callable[[str], None]] = None,
|
||
request_id: Optional[str] = None,
|
||
provider_display_name: str = "未知服务",
|
||
cancel_event: Optional[Any] = None,
|
||
task_type: str = None,
|
||
source: str = None,
|
||
filter_thinking_output: bool = True
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
使用HTTP直连调用/chat/completions接口
|
||
统一处理所有OpenAI兼容的服务商 (支持三级降级重试)
|
||
|
||
参数:
|
||
enable_advanced_params: 是否发送高级参数(temperature/top_p/max_tokens)
|
||
"""
|
||
from ..server import is_streaming_progress_enabled
|
||
|
||
try:
|
||
# 清理输入参数
|
||
base_url = cls._sanitize_config_str(base_url)
|
||
api_key = cls._sanitize_config_str(api_key)
|
||
|
||
# 构建请求URL
|
||
url = cls.parse_api_url(base_url)
|
||
|
||
# 预处理：合并System Prompts (Level 0 默认应用)
|
||
merged_messages = cls._merge_system_prompts(messages)
|
||
|
||
# 构建基础请求体 (仅必选参数)
|
||
initial_payload = {
|
||
"model": model,
|
||
"messages": merged_messages,
|
||
"stream": True
|
||
}
|
||
|
||
# 视觉任务(图像/视频反推)对于某些模型严格要求带有 max_tokens，否则会触发 openai_error，兼容 V1
|
||
is_vision_task = task_type in [TASK_IMAGE_CAPTION, TASK_VIDEO_CAPTION] if task_type else False
|
||
|
||
# 仅在用户开启"启用高级参数"时才发送 temperature、top_p、max_tokens（视觉任务强制发送）
|
||
if enable_advanced_params or is_vision_task:
|
||
initial_payload["temperature"] = temperature
|
||
initial_payload["top_p"] = top_p
|
||
if is_vision_task and not enable_advanced_params:
|
||
# BUG-07 修复：视觉任务强制发送 max_tokens 时，使用保守上限
|
||
# 部分中转站（如 Gemini Flash via 代理）的 max_tokens 上限为 1024-1500
|
||
# 采用 min(max_tokens, 1500) 作为兜底，降低兼容性问题
|
||
initial_payload["max_tokens"] = min(max_tokens, 1500)
|
||
else:
|
||
initial_payload["max_tokens"] = max_tokens
|
||
|
||
|
||
|
||
# 添加思维链控制参数
|
||
if thinking_extra:
|
||
initial_payload.update(thinking_extra)
|
||
|
||
# 构建请求头
|
||
headers = {"Content-Type": "application/json"}
|
||
if api_key and api_key.strip():
|
||
headers["Authorization"] = f"Bearer {api_key}"
|
||
|
||
# 获取HTTP客户端
|
||
request_timeout = 180.0
|
||
if task_type in (TASK_IMAGE_CAPTION, TASK_VIDEO_CAPTION):
|
||
request_timeout = 300.0
|
||
client = HTTPClientPool.get_client(
|
||
provider=provider_display_name,
|
||
base_url=base_url,
|
||
timeout=request_timeout
|
||
)
|
||
|
||
# 前置中断检查：如果 ComfyUI 已经中断了，不启动请求
|
||
from server import PromptServer
|
||
if hasattr(PromptServer.instance, 'execution_interrupted') and PromptServer.instance.execution_interrupted:
|
||
return {"success": False, "error": "任务被中断", "interrupted": True}
|
||
|
||
# 创建统一进度条（自动处理等待→生成→完成的完整生命周期）
|
||
pbar = ProgressBar(
|
||
request_id=request_id,
|
||
service_name=provider_display_name,
|
||
streaming=is_streaming_progress_enabled(),
|
||
task_type=task_type,
|
||
source=source
|
||
)
|
||
|
||
start_time = time.perf_counter()
|
||
last_error_msg = ""
|
||
|
||
# 三级降级重试循环 (Level 0 -> Level 2)
|
||
for retry_level in range(3):
|
||
current_payload = cls._filter_payload(initial_payload, retry_level)
|
||
|
||
# 如果不是Level 0，打印降级重试警告（换行输出）
|
||
if retry_level > 0:
|
||
removed_keys = set(initial_payload.keys()) - set(current_payload.keys())
|
||
removed_str = ", ".join(removed_keys) if removed_keys else "无参数变动"
|
||
print(f"\n{WARN_PREFIX} ⚠️ HTTP 400错误, 触发Level-{retry_level}降级重试 | 服务:{provider_display_name} | 移除参数:[{removed_str}]", flush=True)
|
||
|
||
# 关键修复：停止旧的进度条后再创建新的，防止线程泄漏
|
||
if pbar:
|
||
try:
|
||
pbar.error(f"Retry Level {retry_level}...") # 标记前一个进度条为错误/重试状态
|
||
except:
|
||
pbar._stop_timer()
|
||
|
||
|
||
# 重新创建进度条用于新一轮重试
|
||
pbar = ProgressBar(
|
||
request_id=request_id,
|
||
service_name=provider_display_name,
|
||
extra_info=f"Retry-{retry_level}",
|
||
streaming=is_streaming_progress_enabled(),
|
||
task_type=task_type,
|
||
source=source
|
||
)
|
||
|
||
async def _do_stream_request():
|
||
nonlocal pbar
|
||
|
||
# 定义请求核心逻辑
|
||
async def _request_core():
|
||
async with client.stream('POST', url, headers=headers, json=current_payload, follow_redirects=True) as response:
|
||
if response.status_code != 200:
|
||
error_text = await response.aread()
|
||
try:
|
||
error_data = json.loads(error_text)
|
||
msg = error_data.get('error', {}).get('message', f'HTTP {response.status_code}')
|
||
except:
|
||
msg = f'HTTP {response.status_code}: {error_text.decode("utf-8", errors="ignore")[:200]}'
|
||
|
||
# 智能识别认证错误
|
||
from ..utils.common import _is_auth_error
|
||
if response.status_code == 401 or _is_auth_error(msg.lower()):
|
||
msg = "API Key无效或缺失"
|
||
|
||
return {
|
||
"success": False,
|
||
"error": msg,
|
||
"status_code": response.status_code,
|
||
"should_retry": response.status_code == 400
|
||
}
|
||
|
||
full_content = ""
|
||
reasoning_content = ""
|
||
stream_error = None # 用于捕获流内错误
|
||
|
||
async for line in response.aiter_lines():
|
||
# 此处的循环检查依然保留，作为双重保险
|
||
if cancel_event is not None and cancel_event.is_set():
|
||
raise asyncio.CancelledError()
|
||
|
||
if not line or line == "data: [DONE]" or line == "data:[DONE]": continue
|
||
if line.startswith("data: "): line = line[6:]
|
||
elif line.startswith("data:"): line = line[5:]
|
||
|
||
try:
|
||
chunk = json.loads(line)
|
||
# --- 调试日志 (2级): 输出原始流式数据 ---
|
||
# print(f"[DEBUG-2] Chunk: {line[:200]}...", flush=True)
|
||
|
||
# 关键修复：检测流内错误（部分代理以HTTP 200+SSE方式返回错误）
|
||
if chunk.get('error'):
|
||
err = chunk['error']
|
||
if isinstance(err, dict):
|
||
stream_error = err.get('message', str(err))
|
||
else:
|
||
stream_error = str(err)
|
||
break # 立即中止流读取
|
||
|
||
if chunk.get('choices'):
|
||
delta = chunk['choices'][0].get('delta', {})
|
||
content = delta.get('content', '') or ''
|
||
# 针对不同厂商的推理字段进行广谱捕获
|
||
reasoning = (
|
||
delta.get('reasoning_content', '') or
|
||
delta.get('reasoning', '') or
|
||
delta.get('thinking', '') or
|
||
delta.get('thinking_process', '') or # 备选
|
||
''
|
||
)
|
||
if reasoning:
|
||
reasoning_content += reasoning
|
||
progress_count = len(full_content) + len(reasoning_content)
|
||
pbar.set_generating(progress_count)
|
||
pbar.update(progress_count)
|
||
if content:
|
||
full_content += content
|
||
if stream_callback: stream_callback(content)
|
||
progress_count = len(full_content) + len(reasoning_content)
|
||
pbar.set_generating(progress_count)
|
||
pbar.update(progress_count)
|
||
except json.JSONDecodeError:
|
||
continue # 非 JSON 行（如注释、空行），正常跳过
|
||
except asyncio.CancelledError:
|
||
raise # 必须重新抛出，不能被吐掉
|
||
except Exception as chunk_err:
|
||
# 其他异常：记录警告后继续，避免因单块错误中断整个流
|
||
print(f"\n{WARN_PREFIX} SSE 块解析异常: {chunk_err}", flush=True)
|
||
continue
|
||
|
||
# 如果流中检测到错误，动态判断是否应该触发降级重试
|
||
if stream_error:
|
||
# BUG-05 修复: 部分代理（xFlow/Grok等）以 HTTP 200+SSE error 返回参数错误时
|
||
# 应当触发降级重试，而非直接永久失败
|
||
_RETRYABLE_STREAM_ERROR_KEYWORDS = [
|
||
"unsupported", "invalid parameter", "invalid_request",
|
||
"unknown field", "extra inputs", "not supported",
|
||
"unrecognized", "unexpected", "disallowed",
|
||
"does not support"
|
||
]
|
||
stream_error_lower = stream_error.lower()
|
||
is_retryable_stream_error = any(
|
||
kw in stream_error_lower
|
||
for kw in _RETRYABLE_STREAM_ERROR_KEYWORDS
|
||
)
|
||
return {
|
||
"success": False,
|
||
"error": stream_error,
|
||
"status_code": 200, # HTTP层面成功，业务层面失败
|
||
"should_retry": is_retryable_stream_error # 动态判断是否应该降级重试
|
||
}
|
||
|
||
final_content = full_content
|
||
# 只有当用户没有开启“过滤思维链”时，才手动把 reasoning_content 拼回去
|
||
if reasoning_content and not filter_thinking_output:
|
||
final_content = f"<think>{reasoning_content}</think>\n{full_content}"
|
||
|
||
elapsed_ms = int((time.perf_counter() - start_time) * 1000)
|
||
if not final_content.strip():
|
||
pbar.error("响应内容为空")
|
||
# --- 调试日志 (1级): 警告响应内容为空 ---
|
||
print(f"\n{WARN_PREFIX} [API响应调试] 模型:{model} | 状态:成功 | 但最终内容为空字符串，触发降级重试", flush=True)
|
||
return {
|
||
"success": False,
|
||
"error": "API returned empty content",
|
||
"status_code": 200,
|
||
"should_retry": True
|
||
}
|
||
|
||
pbar.done(char_count=len(final_content), elapsed_ms=elapsed_ms)
|
||
|
||
return {"success": True, "content": final_content}
|
||
|
||
# 定义监视器逻辑：每100ms检查一次中断信号
|
||
async def _monitor_interrupts(target_task):
|
||
while not target_task.done():
|
||
is_interrupted = False
|
||
if cancel_event is not None and cancel_event.is_set():
|
||
is_interrupted = True
|
||
else:
|
||
try:
|
||
from server import PromptServer
|
||
if hasattr(PromptServer.instance, 'execution_interrupted') and PromptServer.instance.execution_interrupted:
|
||
is_interrupted = True
|
||
except:
|
||
pass
|
||
|
||
if is_interrupted:
|
||
target_task.cancel()
|
||
return True
|
||
await asyncio.sleep(0.1)
|
||
return False
|
||
|
||
# 并发运行请求和监视器
|
||
req_task = asyncio.create_task(_request_core())
|
||
monitor_task = asyncio.create_task(_monitor_interrupts(req_task))
|
||
|
||
try:
|
||
result = await req_task
|
||
# 关键修复：API 返回错误时，确保进度条被停止
|
||
if not result.get("success") and not result.get("interrupted"):
|
||
if not getattr(pbar, '_closed', False):
|
||
pbar.error(result.get("error", "API 错误"))
|
||
return result
|
||
except asyncio.CancelledError:
|
||
pbar.cancel(f"{WARN_PREFIX} 任务被中断 | 服务:{provider_display_name}")
|
||
return {"success": False, "error": "中断", "interrupted": True}
|
||
finally:
|
||
if not monitor_task.done():
|
||
monitor_task.cancel()
|
||
|
||
# 执行请求
|
||
try:
|
||
result = await _do_stream_request()
|
||
except Exception as req_err:
|
||
# 网络层面的异常（非HTTP响应）
|
||
|
||
# 特别处理编码错误 (UnicodeEncodeError)
|
||
if isinstance(req_err, UnicodeEncodeError):
|
||
error_detail = "网络请求编码异常: 检测到非法字符 (\u2026 或其他非 ASCII 字符)。请检查服务商配置中的 API Key 或 URL 是否包含多余的省略号、引号或空格。"
|
||
if 'pbar' in locals() and pbar:
|
||
pbar.error(error_detail)
|
||
return {"success": False, "error": error_detail}
|
||
|
||
if 'pbar' in locals() and pbar:
|
||
pbar.error(f"网络请求异常: {req_err}")
|
||
return {"success": False, "error": f"网络请求异常: {req_err}"}
|
||
|
||
# 检查结果
|
||
if result["success"]:
|
||
# Ollama 服务成功后尝试卸载模型
|
||
if provider_display_name.lower().find("ollama") != -1:
|
||
try:
|
||
from ..config_manager import config_manager
|
||
service_config = config_manager.get_service(provider_display_name) or {}
|
||
# 注入当前实际使用的 base_url，防止回退到 localhost 导致清理显存失败
|
||
service_config['base_url'] = base_url
|
||
await cls._unload_ollama_model(model, service_config)
|
||
except:
|
||
pass
|
||
return result
|
||
|
||
if result.get("interrupted"):
|
||
return result
|
||
|
||
last_error_msg = result["error"]
|
||
|
||
# 只有 should_retry 为 True (HTTP 400) 且还有重试机会时，才继续循环
|
||
if not result.get("should_retry"):
|
||
break # 非400错误（如401, 500等），不进行降级重试，直接返回错误
|
||
|
||
# 所有重试耗尽或非可重试错误
|
||
if 'pbar' in locals() and pbar:
|
||
pbar.error(last_error_msg)
|
||
return {"success": False, "error": last_error_msg}
|
||
|
||
# 关键修复：单独捕获 CancelledError，确保进度条被正确停止
|
||
except asyncio.CancelledError:
|
||
if 'pbar' in locals() and pbar:
|
||
pbar.cancel(f"{WARN_PREFIX} 任务被外部取消 | 服务:{provider_display_name}")
|
||
return {"success": False, "error": "任务被取消", "interrupted": True}
|
||
|
||
except Exception as e:
|
||
if 'pbar' in locals() and pbar:
|
||
pbar.error(format_api_error(e, provider_display_name))
|
||
return {"success": False, "error": format_api_error(e, provider_display_name)}
|
||
|
||
@staticmethod
|
||
async def _unload_ollama_model(model: str, provider_config: Dict[str, Any]):
|
||
"""
|
||
卸载Ollama模型以释放显存和内存
|
||
|
||
参数:
|
||
model: 模型名称
|
||
provider_config: 提供商配置字典
|
||
"""
|
||
try:
|
||
# 检查是否启用自动释放
|
||
auto_unload = provider_config.get('auto_unload', True)
|
||
if not auto_unload:
|
||
from ..utils.common import PROCESS_PREFIX
|
||
print(f"{PROCESS_PREFIX} Ollama模型已保留 | 模型:{model}")
|
||
return
|
||
|
||
await wait_before_ollama_unload()
|
||
|
||
# 获取base_url
|
||
base_url = provider_config.get('base_url', 'http://localhost:11434')
|
||
if base_url.endswith('/v1'):
|
||
base_url = base_url[:-3]
|
||
|
||
# 调用Ollama API卸载模型
|
||
url = f"{base_url}/api/generate"
|
||
payload = {
|
||
"model": model,
|
||
"keep_alive": 0
|
||
}
|
||
|
||
# 创建临时客户端（卸载操作不需要复用，禁用代理避免localhost请求被拦截）
|
||
async with httpx.AsyncClient(timeout=5.0, trust_env=False) as client:
|
||
response = await client.post(url, json=payload)
|
||
if response.status_code == 200:
|
||
from ..utils.common import PROCESS_PREFIX
|
||
print(f"{PROCESS_PREFIX} Ollama模型已释放 | 模型:{model}")
|
||
|
||
except Exception as e:
|
||
from ..utils.common import WARN_PREFIX
|
||
print(f"{WARN_PREFIX} Ollama模型释放失败（不影响结果） | 模型:{model} | 错误:{str(e)[:50]}")
|
||
|
||
@classmethod
|
||
def get_provider_display_name(cls, provider: str) -> str:
|
||
"""
|
||
获取提供商显示名称
|
||
优先从config_manager获取服务的真实名称，兜底使用provider key
|
||
"""
|
||
# 优先尝试从config_manager获取服务名称
|
||
try:
|
||
from ..config_manager import config_manager
|
||
service = config_manager.get_service(provider)
|
||
if service and 'name' in service:
|
||
return service['name']
|
||
except Exception:
|
||
pass
|
||
|
||
# 兜底直接返回key
|
||
return provider
|
||
|
||
@classmethod
|
||
def get_provider_base_url(cls, provider: str, config: Optional[Dict[str, Any]] = None) -> Optional[str]:
|
||
"""
|
||
获取提供商的base_url
|
||
仅用于custom provider的逻辑，其他情况应直接从config获取
|
||
"""
|
||
if provider == 'custom' and config:
|
||
base_url = config.get('base_url')
|
||
# 确保base_url不以/chat/completions结尾
|
||
if base_url and base_url.endswith('/chat/completions'):
|
||
base_url = base_url[:-len('/chat/completions')]
|
||
return base_url
|
||
|
||
return None
|
||
|
||
@staticmethod
|
||
def _sanitize_config_str(s: str) -> str:
|
||
"""
|
||
清理配置字符串中的非法字符（如空格、引号、不可见字符等）
|
||
防止因复制粘贴导致的请求失败。
|
||
|
||
关键修复：传入 None 或非字符串时统一返回空字符串 ""，
|
||
避免后续拼接产生 "Bearer None" 等无效认证头。
|
||
"""
|
||
if not s or not isinstance(s, str):
|
||
return "" # 统一返回空字符串，不返回 None
|
||
# 1. 移除两端空格和常见引号
|
||
s = s.strip().strip('"').strip("'")
|
||
# 2. 移除常见的 Unicode 干扰字符（如 \u2026 省略号）
|
||
# 注意：这里我们选择将其替换为空，因为 API Key 中不应包含这些字符
|
||
s = s.replace('\u2026', '')
|
||
# 3. 移除不可见的制表符和换行符
|
||
s = s.replace('\t', '').replace('\n', '').replace('\r', '')
|
||
return s
|