refactor(nodes): optimize OpenAI compatible adapter and translator

Motivation:
LLM API calls frequently failed with HTTP 429 and 50x errors, crashing the workflow. Code redundancy existed between different LLM nodes. Non-standard payload fields caused 400 errors with some providers.

Implementation:
1. Extracted api_client.py for centralized HTTP handling with smart retries (Retry-After, backoff, jitter) and structured error logging.
2. Implemented graceful degradation to return error text instead of raising exceptions.
3. Cleaned up non-standard OpenAI payload fields.
4. Added input validation to fail fast on empty prompts or API keys.
5. Refactored both openai_compatible.py and llm_translator.py to reuse LLMClient.
6. Handled dual import registration in __init__.py for robust dynamic loading.
This commit is contained in:
科林_Cyril Pilgrim
2026-02-28 16:31:06 +08:00
parent 739324c58b
commit 1a15c8d129
6 changed files with 634 additions and 581 deletions
+3
View File
@@ -33,5 +33,8 @@ ENV/
.DS_Store
Thumbs.db
# Internal docs (private)
CONTRIBUTING.md
# Tests
tests/
+2
View File
@@ -25,6 +25,8 @@ if nodes_dir.exists():
spec = importlib.util.spec_from_file_location(module_name, py_file)
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
# Also register under simple name for inter-module imports
sys.modules[py_file.stem] = module
spec.loader.exec_module(module)
# Register node mappings
+308
View File
@@ -0,0 +1,308 @@
"""
LLMClient — Robust HTTP client for OpenAI-compatible LLM APIs.
Single source of truth for all API calls in ComfyUI-LLMs-Toolkit.
Features:
- Smart retry with Retry-After header respect + exponential backoff + jitter
- Structured error diagnostics with actionable hints
- SSL compatibility for Chinese API providers
- Request size logging
"""
import time
import random
import json
import ssl
import re
import urllib.request
import urllib.error
from typing import Dict, Any, Tuple, Optional
from dataclasses import dataclass, field
# ─── Error Classification ────────────────────────────────────────────────────
@dataclass
class APIError:
"""Structured API error with diagnostics"""
error_type: str
cause: str
hint: str
details: list = field(default_factory=list)
api_message: str = ""
def user_message(self) -> str:
"""Format a user-friendly error message (for graceful degradation)"""
parts = [f"[{self.error_type}] {self.cause}"]
if self.api_message:
parts.append(f"API: {self.api_message[:120]}")
parts.append(f"建议: {self.hint}")
return "\n".join(parts)
def classify_error(
error: Exception,
provider_name: str,
model: str,
request_size_mb: float = 0.0,
elapsed_ms: int = 0
) -> APIError:
"""Classify an exception into a structured APIError with diagnostics."""
error_str = str(error)
err = APIError(
error_type="UNKNOWN",
cause=error_str[:200],
hint="检查终端完整日志获取更多信息"
)
if "HTTP 401" in error_str or "invalid_api_key" in error_str or "Unauthorized" in error_str:
err.error_type = "AUTH"
err.cause = "API Key 无效或已过期"
err.hint = f"请检查 {provider_name} 的 API Key 是否正确配置"
err.details.append("验证失败,请确认 API Key 有效且未过期")
elif "HTTP 429" in error_str or "rate_limit" in error_str:
err.error_type = "RATE_LIMIT"
err.cause = "请求频率超限 (已自动重试后仍失败)"
err.hint = "请等待片刻后重试,或升级 API 套餐"
err.details.append("触发了速率限制")
elif "HTTP 413" in error_str or "Payload Too Large" in error_str:
err.error_type = "PAYLOAD_TOO_LARGE"
err.cause = f"请求体过大 ({request_size_mb:.2f}MB)"
err.hint = "请压缩图片或减少输入内容"
err.details.append("建议: 压缩图片到 1024x1024 以下,或降低图片质量")
elif "HTTP 400" in error_str or "invalid_request" in error_str:
err.error_type = "BAD_REQUEST"
err.cause = "请求参数有误"
err.hint = f"请检查模型名称 '{model}' 是否正确"
if request_size_mb > 10:
err.details.append(f"请求体较大 ({request_size_mb:.2f}MB),可能超过服务器限制")
elif "HTTP 404" in error_str or "model_not_found" in error_str:
err.error_type = "MODEL"
err.cause = f"模型 '{model}' 不存在"
err.hint = f"请确认 {provider_name} 支持该模型名称"
elif "HTTP 5" in error_str:
err.error_type = "SERVER"
err.cause = "API 服务端错误 (已自动重试后仍失败)"
err.hint = f"{provider_name} 服务可能暂时不可用,请稍后重试"
elif "Broken pipe" in error_str:
err.error_type = "BROKEN_PIPE"
err.cause = "连接在传输数据时被服务器关闭"
err.hint = "可能是请求体过大或服务器超时"
if request_size_mb > 5:
err.details.append(f"请求体很大 ({request_size_mb:.2f}MB),建议压缩图片")
elif "timed out" in error_str.lower() or "timeout" in error_str.lower():
err.error_type = "TIMEOUT"
err.cause = f"请求超时 ({elapsed_ms / 1000:.1f}秒)"
err.hint = "网络连接缓慢或 API 响应时间过长"
elif "connection" in error_str.lower() or "URLError" in type(error).__name__:
err.error_type = "NETWORK"
err.cause = "无法连接到 API 服务器"
err.hint = "请检查网络连接和代理设置"
err.details.append("可能原因: 网络断开 / 防火墙阻止 / 代理配置错误")
elif "missing 'choices'" in error_str:
err.error_type = "RESPONSE"
err.cause = "API 返回了非标准格式"
err.hint = "API 可能返回了错误信息而非正常结果"
# Extract API error message from JSON
msg_match = re.search(r'"message"\s*:\s*"([^"]+)"', error_str)
if msg_match:
err.api_message = msg_match.group(1)
return err
def log_error(err: APIError, provider_name: str, model: str,
request_size_mb: float, elapsed_ms: int) -> None:
"""Print structured error block to terminal."""
ts = time.strftime('%H:%M:%S')
print(f"")
print(f"[LLMs_Toolkit] {ts} ✗ {err.error_type} ERROR ({elapsed_ms}ms)")
print(f" ┌──────────────────────────────────────────────")
print(f" │ 服务商 {provider_name}")
print(f" │ 模型 {model or 'default'}")
if request_size_mb > 0:
print(f" │ 请求大小 {request_size_mb:.2f} MB")
print(f" │ 原因 {err.cause}")
if err.api_message:
print(f" │ API消息 {err.api_message[:120]}")
print(f" │ 建议 {err.hint}")
if err.details:
print(f" │")
for i, detail in enumerate(err.details, 1):
print(f" │ {i}. {detail}")
print(f" └──────────────────────────────────────────────")
print(f"")
# ─── HTTP Client ─────────────────────────────────────────────────────────────
# Shared SSL context (created once, reused across all calls)
_ssl_context = None
def _get_ssl_context() -> ssl.SSLContext:
"""Get or create a lenient SSL context for Chinese API providers."""
global _ssl_context
if _ssl_context is None:
_ssl_context = ssl.create_default_context()
_ssl_context.check_hostname = False
_ssl_context.verify_mode = ssl.CERT_NONE
_ssl_context.set_ciphers("DEFAULT")
_ssl_context.options |= getattr(ssl, "OP_LEGACY_SERVER_CONNECT", 0)
return _ssl_context
def _normalize_url(base_url: str, path: str = "/chat/completions") -> str:
"""Normalize base URL and append path, avoiding double slashes."""
return base_url.rstrip("/") + path
def _parse_retry_after(headers) -> Optional[float]:
"""Extract Retry-After value from response headers (seconds)."""
retry_after = headers.get("Retry-After")
if retry_after:
try:
return float(retry_after)
except ValueError:
pass
return None
def _is_retryable(code: int) -> bool:
"""Check if an HTTP status code warrants a retry."""
return code == 429 or (500 <= code < 600)
class LLMClient:
"""
Robust HTTP client for OpenAI-compatible chat completions API.
Usage:
client = LLMClient(base_url, api_key)
content, data = client.chat(payload)
"""
TAG = "[LLMs_Toolkit]"
def __init__(
self,
base_url: str,
api_key: str,
max_retries: int = 3,
timeout: int = 180,
):
self.url = _normalize_url(base_url)
self.api_key = api_key
self.max_retries = max_retries
self.timeout = timeout
self._ctx = _get_ssl_context()
def chat(self, payload: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]:
"""
Send a chat completion request.
Returns:
Tuple of (response_content, full_response_data)
Raises:
Exception with structured error message if all retries fail.
"""
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}",
}
data_bytes = json.dumps(payload).encode("utf-8")
data_size_mb = len(data_bytes) / (1024 * 1024)
if data_size_mb > 1:
print(f"{self.TAG} 请求体大小: {data_size_mb:.2f} MB")
last_error = None
for attempt in range(self.max_retries + 1):
try:
req = urllib.request.Request(
self.url, data=data_bytes, headers=headers, method="POST"
)
with urllib.request.urlopen(
req, timeout=self.timeout, context=self._ctx
) as resp:
body = resp.read().decode("utf-8")
data = json.loads(body)
if "choices" not in data or len(data["choices"]) == 0:
raise ValueError(
f"API response missing 'choices'. Response: {json.dumps(data)[:300]}"
)
content = data["choices"][0]["message"]["content"]
return content, data
except urllib.error.HTTPError as e:
error_body = e.read().decode("utf-8", errors="replace")
if _is_retryable(e.code) and attempt < self.max_retries:
# Respect Retry-After header if present
wait = _parse_retry_after(e.headers) or self._backoff(attempt, is_rate_limit=(e.code == 429))
print(
f"{self.TAG} HTTP {e.code} 错误, "
f"重试 {attempt + 1}/{self.max_retries} "
f"(等待 {wait:.1f}s)..."
)
time.sleep(wait)
continue
raise Exception(f"HTTP {e.code} | {error_body}")
except urllib.error.URLError as e:
last_error = e
error_msg = str(e.reason)
if self._is_connection_error(error_msg) and attempt < self.max_retries:
wait = self._backoff(attempt)
print(f"{self.TAG} 连接错误, 重试 {attempt + 1}/{self.max_retries} (等待 {wait:.1f}s)...")
time.sleep(wait)
continue
raise Exception(f"URLError | {error_msg}")
except (TimeoutError, OSError) as e:
last_error = e
if attempt < self.max_retries:
wait = self._backoff(attempt)
print(f"{self.TAG} 超时/IO错误, 重试 {attempt + 1}/{self.max_retries} (等待 {wait:.1f}s)...")
time.sleep(wait)
continue
raise Exception(f"TimeoutError | 请求超过 {self.timeout} 秒未响应")
# All retries exhausted
if last_error:
raise Exception(
f"URLError | {str(getattr(last_error, 'reason', last_error))} "
f"(重试 {self.max_retries} 次后仍失败)"
)
raise Exception("Unknown error occurred")
@staticmethod
def _backoff(attempt: int, is_rate_limit: bool = False) -> float:
"""Calculate wait time with exponential backoff + jitter."""
base = 2 ** attempt
if is_rate_limit:
base = max(5, base * 3) # Much longer for rate limits
jitter = random.uniform(0, base * 0.3)
return base + jitter
@staticmethod
def _is_connection_error(msg: str) -> bool:
"""Check if error message indicates a retryable connection issue."""
keywords = ("Broken pipe", "Connection", "EOF", "reset by peer", "ECONNRESET")
return any(kw.lower() in msg.lower() for kw in keywords)
+40 -79
View File
@@ -1,9 +1,19 @@
"""
LLM Translator — Unix-philosophy translation node.
Do One Thing and Do It Well: translate text via LLM API.
Reuses the shared LLMClient for robustness.
"""
import time
import json
import urllib.request
import urllib.error
from typing import Dict, Any, Tuple
try:
from .api_client import LLMClient
except ImportError:
from api_client import LLMClient
class LLMTranslator:
"""
@@ -53,69 +63,6 @@ class LLMTranslator:
FUNCTION = "translate"
CATEGORY = "🚦ComfyUI_LLMs_Toolkit/Utility"
def __init__(self):
pass
def _call_api(
self,
llm_config: Dict[str, Any],
messages: list,
temperature: float = 0.3
) -> str:
"""调用LLM API - 核心通信逻辑"""
import ssl
base_url = llm_config.get("base_url", "").rstrip("/")
model = llm_config.get("model", "")
api_key = llm_config.get("api_key", "")
url = f"{base_url}/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": 4096
}
ctx = ssl.create_default_context()
last_error = None
for attempt in range(3):
try:
req = urllib.request.Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers=headers,
method="POST"
)
with urllib.request.urlopen(req, timeout=60, context=ctx) as response:
data = json.loads(response.read().decode("utf-8"))
if "choices" not in data or not data["choices"]:
raise ValueError("Empty response from API")
return data["choices"][0]["message"]["content"].strip()
except urllib.error.HTTPError as e:
error_body = e.read().decode("utf-8", errors="ignore")
raise Exception(f"HTTP Error {e.code}: {error_body}")
except (ssl.SSLError, urllib.error.URLError, ConnectionError, TimeoutError) as e:
last_error = e
print(f"[LLMs_Toolkit] Translator attempt {attempt + 1}/3 failed: {e}")
import time
time.sleep(1)
continue
except Exception as e:
raise Exception(f"Translation failed: {str(e)}")
raise Exception(f"Translation failed after 3 attempts: {last_error}")
def translate(
self,
llm_config: Dict[str, Any],
@@ -123,39 +70,53 @@ class LLMTranslator:
target_language: str,
glossary: str = ""
) -> Tuple[str]:
"""
执行翻译任务
Unix Philosophy: Clean input, clean output.
"""
"""Execute translation. Returns error text on failure instead of crashing."""
if not text.strip():
return ("",)
start_time = time.time()
# 1. 构建系统指令 (System Prompt)
# Build system instruction
system_instruction = (
f"You are a professional translator. Translate the following text into {target_language}. "
"Maintain the original tone, style, and formatting. "
"Output ONLY the translated text, no explanations."
)
if glossary.strip():
system_instruction += f"\n\nGlossary (Strictly follow):\n{glossary}"
# 2. 构建消息体
# Build payload
messages = [
{"role": "system", "content": system_instruction},
{"role": "user", "content": text}
]
payload = {
"model": llm_config.get("model", ""),
"messages": messages,
"temperature": 0.3,
"max_tokens": 4096
}
# 3. 调用API (Opinionated defaults: temperature=0.3 for stability)
translated_text = self._call_api(llm_config, messages, temperature=0.3)
# Call API via shared client
try:
client = LLMClient(
base_url=llm_config.get("base_url", ""),
api_key=llm_config.get("api_key", ""),
max_retries=3,
timeout=60,
)
translated_text, _ = client.chat(payload)
# 4. 简单日志
elapsed = int((time.time() - start_time) * 1000)
print(f"[LLM Translator] {len(text)} chars -> {target_language} ({elapsed}ms)")
elapsed = int((time.time() - start_time) * 1000)
print(f"[LLM Translator] {len(text)} chars -> {target_language} ({elapsed}ms)")
return (translated_text.strip(),)
except Exception as e:
elapsed = int((time.time() - start_time) * 1000)
print(f"[LLM Translator] ✗ Translation failed ({elapsed}ms): {e}")
return (f"[Translation Error] {str(e)[:200]}",)
return (translated_text,)
# ComfyUI Node Registration
NODE_CLASS_MAPPINGS = {"LLMTranslator": LLMTranslator}
+180 -502
View File
@@ -1,190 +1,90 @@
"""
OpenAI-compatible LLM adapter for ComfyUI.
Unified interface for multiple LLM providers with graceful degradation.
API calls are handled by the shared LLMClient (api_client.py).
"""
import time
import random
import json as json_lib
import ssl
import urllib.request
import urllib.error
from typing import Optional, List, Dict, Tuple, Any, Union
from dataclasses import dataclass
try:
from .api_client import LLMClient, classify_error, log_error
except ImportError:
from api_client import LLMClient, classify_error, log_error
@dataclass
class ProviderConfig:
"""LLM provider endpoint configuration"""
name: str
base_url: str
requires_string_content: bool = False
# ─── Message Building ────────────────────────────────────────────────────────
# Providers that require plain string content instead of structured arrays
_STRING_CONTENT_PROVIDERS = frozenset({
"Spark/星火", "Baichuan/百川", "SenseChat/日日新"
})
class ProviderRegistry:
"""Centralized registry for LLM API providers"""
PROVIDERS = {
"Qwen/通义千问": ProviderConfig(
"Qwen",
"https://dashscope.aliyuncs.com/compatible-mode/v1"
),
"DeepSeek/深度求索": ProviderConfig(
"DeepSeek",
"https://api.deepseek.com/v1/"
),
"DouBao/豆包": ProviderConfig(
"DouBao",
"https://ark.cn-beijing.volces.com/api/v3/"
),
"Spark/星火": ProviderConfig(
"Spark",
"https://spark-api-open.xf-yun.com/v1/",
requires_string_content=True
),
"GLM/智谱清言": ProviderConfig(
"GLM",
"https://open.bigmodel.cn/api/paas/v4/"
),
"Moonshot/月之暗面": ProviderConfig(
"Moonshot",
"https://api.moonshot.cn/v1"
),
"Baichuan/百川": ProviderConfig(
"Baichuan",
"https://api.baichuan-ai.com/v1/",
requires_string_content=True
),
"MiniMax/MiniMax": ProviderConfig(
"MiniMax",
"https://api.minimax.chat/v1/"
),
"StepFun/阶跃星辰": ProviderConfig(
"StepFun",
"https://api.stepfun.com/v1/"
),
"SenseChat/日日新": ProviderConfig(
"SenseChat",
"https://api.sensenova.cn/compatible-mode/v1",
requires_string_content=True
),
}
@classmethod
def get_provider(cls, provider_key: str) -> ProviderConfig:
"""Get provider config or treat key as custom URL"""
return cls.PROVIDERS.get(provider_key.strip(),
ProviderConfig("Custom", provider_key.strip()))
def _build_content(
prompt: str,
image_url: Optional[Union[str, List[str]]] = None
) -> Union[str, List[Dict[str, Any]]]:
"""Build user message content (multimodal or text-only)."""
# Fast path: no image → just return the string
if not image_url:
return prompt
content: List[Dict[str, Any]] = []
class MessageBuilder:
"""Builds chat messages with support for multimodal content"""
@staticmethod
def build_content(prompt: str, image_url: Optional[Union[str, List[str]]] = None) -> List[Dict[str, Any]]:
"""Build content array for user message"""
content = []
if image_url:
if isinstance(image_url, list):
for url in image_url:
if isinstance(url, str) and url.startswith("data:image"):
content.append({
"type": "image_url",
"image_url": {"url": url}
})
elif isinstance(image_url, str):
if not image_url.startswith("data:image"):
raise ValueError("Image must be a valid base64 data URL")
content.append({
"type": "image_url",
"image_url": {"url": image_url}
})
if prompt.strip():
# Normalize to list
urls = image_url if isinstance(image_url, list) else [image_url]
for url in urls:
if isinstance(url, str) and url.startswith("data:image"):
content.append({
"type": "text",
"text": prompt
"type": "image_url",
"image_url": {"url": url}
})
if not content:
raise ValueError("Content cannot be empty")
return content
@staticmethod
def build_messages(
content: List[Dict[str, Any]],
system_prompt: Optional[str] = None
) -> List[Dict[str, Any]]:
"""Build complete message array"""
messages = []
if system_prompt:
messages.append({
"role": "system",
"content": system_prompt
})
messages.append({
"role": "user",
"content": content
})
return messages
@staticmethod
def adapt_for_provider(
messages: List[Dict[str, Any]],
provider: ProviderConfig,
prompt: str
) -> List[Dict[str, Any]]:
"""Adapt messages for provider-specific requirements"""
if provider.requires_string_content:
# Some providers require string content, not structured arrays
adapted = []
for msg in messages:
if msg["role"] == "system":
adapted.append(msg)
elif msg["role"] == "user":
adapted.append({
"role": "user",
"content": prompt if isinstance(prompt, str) else prompt
})
return adapted
return messages
if prompt.strip():
content.append({"type": "text", "text": prompt})
return content if content else prompt
class TokenEstimator:
"""Estimates token usage for billing/logging purposes"""
@staticmethod
def estimate_input_tokens(
prompt: str,
system_prompt: Optional[str] = None,
image_url: Optional[Union[str, List[str]]] = None
) -> int:
"""Simple character-based token estimation"""
tokens = len(prompt)
if system_prompt:
tokens += len(system_prompt)
if image_url:
if isinstance(image_url, list):
for url in image_url:
if isinstance(url, str):
tokens += len(url) // 1000
elif isinstance(image_url, str):
# Rough estimate: base64 image ~= length/1000 tokens
tokens += len(image_url) // 1000
return tokens
def _build_messages(
content: Union[str, List[Dict[str, Any]]],
system_prompt: Optional[str] = None,
provider: str = ""
) -> List[Dict[str, Any]]:
"""Build complete message array, adapting for provider quirks."""
messages: List[Dict[str, Any]] = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
# Some providers require plain string content
if provider in _STRING_CONTENT_PROVIDERS and isinstance(content, list):
# Extract text from structured content
text_parts = [c["text"] for c in content if c.get("type") == "text"]
user_content = " ".join(text_parts) if text_parts else ""
else:
user_content = content
messages.append({"role": "user", "content": user_content})
return messages
# ─── ComfyUI Node ────────────────────────────────────────────────────────────
class OpenAICompatibleLoader:
"""
OpenAI-compatible API adapter for ComfyUI
OpenAI-compatible API adapter for ComfyUI.
Supports multiple LLM providers with unified interface.
Designed for extensibility and clean architecture.
Features graceful degradation — errors return readable text
instead of crashing the workflow.
"""
TAG = "[LLMs_Toolkit]"
@classmethod
def INPUT_TYPES(cls):
return {
@@ -219,304 +119,91 @@ class OpenAICompatibleLoader:
})
}
}
RETURN_TYPES = ("STRING", "INT", "INT")
RETURN_NAMES = ("text", "input_tokens", "output_tokens")
FUNCTION = "generate"
CATEGORY = "🚦ComfyUI_LLMs_Toolkit/Generate"
OUTPUT_NODE = True
def __init__(self):
self._conversation_history: List[Dict[str, Any]] = []
self._request_start_time: float = 0.0
def _log_request_start(
self,
provider: ProviderConfig,
model: str,
system_prompt: Optional[str],
prompt: str,
image_input: Optional[Union[str, List[str]]]
) -> None:
"""Geek-style request logging with timing"""
self._request_start_time = time.time()
# ── Logging ──────────────────────────────────────────────────────────
@staticmethod
def _log_start(provider_name: str, model: str,
system_prompt: Optional[str], prompt: str,
image_input: Optional[Union[str, List[str]]]) -> float:
"""Log request start, return start time."""
start = time.time()
ts = time.strftime('%H:%M:%S')
# Build context flags
flags = []
if system_prompt:
flags.append('sys')
img_count = 0
if image_input:
if isinstance(image_input, list):
img_count = len(image_input)
else:
img_count = 1
flags.append(f'img(x{img_count})')
n = len(image_input) if isinstance(image_input, list) else 1
flags.append(f'img(x{n})')
flag_str = f"[{' '.join(flags)}]" if flags else ""
# Compact header
print(f"[LLMs_Toolkit] {ts} → {provider.name}/{model or 'default'} {flag_str}")
# Content preview (indented)
print(f"[LLMs_Toolkit] {ts} → {provider_name}/{model or 'default'} {flag_str}")
if system_prompt:
sys_preview = system_prompt[:60].replace('\n', ' ')
print(f" sys│ {sys_preview}...")
prompt_preview = prompt[:80].replace('\n', ' ')
print(f" in │ {prompt_preview}...")
def _log_request_complete(
self,
response: str,
input_tokens: int,
output_tokens: int
) -> None:
"""Geek-style completion logging with metrics"""
# Calculate elapsed time
elapsed_ms = int((time.time() - self._request_start_time) * 1000)
# Format token counts (k suffix for thousands)
in_tok = f"{input_tokens/1000:.1f}k" if input_tokens >= 1000 else str(input_tokens)
out_tok = f"{output_tokens/1000:.1f}k" if output_tokens >= 1000 else str(output_tokens)
# Response preview
preview = response[:80].replace('\n', ' ')
print(f" out│ {preview}...")
print(f"[LLMs_Toolkit] {time.strftime('%H:%M:%S')} ← {in_tok}/{out_tok}t ({elapsed_ms}ms)\n")
def _log_error(
self,
error: Exception,
provider: 'ProviderConfig',
model: str,
request_size_mb: float = 0.0
) -> None:
"""Structured error logging with diagnostics and actionable hints"""
elapsed_ms = int((time.time() - self._request_start_time) * 1000)
ts = time.strftime('%H:%M:%S')
print(f" sys│ {system_prompt[:60].replace(chr(10), ' ')}...")
print(f" in │ {prompt[:80].replace(chr(10), ' ')}...")
# Classify error and extract details
error_str = str(error)
error_type = "UNKNOWN"
cause = error_str
hint = "检查终端完整日志获取更多信息"
details = [] # Additional diagnostic details
if "HTTP 401" in error_str or "invalid_api_key" in error_str or "Unauthorized" in error_str:
error_type = "AUTH"
cause = "API Key 无效或已过期"
hint = f"请检查 {provider.name} 的 API Key 是否正确配置"
details.append("验证失败,请确认 API Key 有效且未过期")
return start
elif "HTTP 429" in error_str or "rate_limit" in error_str:
error_type = "RATE_LIMIT"
cause = "请求频率超限"
hint = "请等待片刻后重试,或升级 API 套餐"
details.append("触发了速率限制,建议降低请求频率")
@staticmethod
def _log_done(response: str, in_tok: int, out_tok: int, start: float) -> None:
"""Log completed request."""
ms = int((time.time() - start) * 1000)
fmt = lambda t: f"{t/1000:.1f}k" if t >= 1000 else str(t)
print(f" out│ {response[:80].replace(chr(10), ' ')}...")
print(f"[LLMs_Toolkit] {time.strftime('%H:%M:%S')} ← {fmt(in_tok)}/{fmt(out_tok)}t ({ms}ms)\n")
elif "HTTP 413" in error_str or "Payload Too Large" in error_str:
error_type = "PAYLOAD_TOO_LARGE"
cause = f"请求体过大 ({request_size_mb:.2f}MB)"
hint = "请压缩图片或减少输入内容"
details.append("服务器拒绝接收过大的请求")
details.append("建议: 压缩图片到 1024x1024 以下,或降低图片质量")
# ── Memory ───────────────────────────────────────────────────────────
elif "HTTP 400" in error_str or "invalid_request" in error_str:
error_type = "BAD_REQUEST"
cause = "请求参数有误"
hint = f"请检查模型名称 '{model}' 是否正确"
if request_size_mb > 10:
details.append(f"请求体较大 ({request_size_mb:.2f}MB),可能超过服务器限制")
details.append("检查所有参数是否符合 API 要求")
elif "HTTP 404" in error_str or "model_not_found" in error_str:
error_type = "MODEL"
cause = f"模型 '{model}' 不存在"
hint = f"请确认 {provider.name} 支持该模型名称"
details.append("模型名称可能拼写错误或该服务商不支持此模型")
elif "HTTP 5" in error_str:
error_type = "SERVER"
cause = "API 服务端错误"
hint = f"{provider.name} 服务可能暂时不可用,请稍后重试"
details.append("这是服务器端的问题,不是客户端配置错误")
elif "Broken pipe" in error_str:
error_type = "BROKEN_PIPE"
cause = "连接在传输数据时被服务器关闭"
hint = "可能是请求体过大或服务器超时"
if request_size_mb > 5:
details.append(f"请求体很大 ({request_size_mb:.2f}MB),建议压缩图片")
if elapsed_ms > 60000:
details.append(f"传输时间过长 ({elapsed_ms/1000:.1f}秒),可能触发服务器超时")
details.append("建议: 1) 压缩图片 2) 检查网络上传速度 3) 使用更快的网络")
elif "TimeoutError" in type(error).__name__ or "timed out" in error_str.lower():
error_type = "TIMEOUT"
cause = f"请求超时 ({elapsed_ms/1000:.1f}秒)"
hint = "网络连接缓慢或 API 响应时间过长"
if request_size_mb > 5:
details.append(f"请求体较大 ({request_size_mb:.2f}MB),传输需要更长时间")
details.append("建议: 检查网络连接速度")
elif "URLError" in type(error).__name__ or "connection" in error_str.lower():
error_type = "NETWORK"
cause = "无法连接到 API 服务器"
hint = "请检查网络连接和代理设置"
details.append("可能原因: 1) 网络断开 2) 防火墙阻止 3) 代理配置错误")
elif "missing 'choices'" in error_str:
error_type = "RESPONSE"
cause = "API 返回了非标准格式"
hint = "API 可能返回了错误信息而非正常结果"
details.append("服务器返回的响应格式不符合预期")
# Parse API error message if present
api_message = ""
if '"message"' in error_str:
try:
# Try to extract JSON error message
import re
msg_match = re.search(r'"message"\s*:\s*"([^"]+)"', error_str)
if msg_match:
api_message = msg_match.group(1)
except Exception:
pass
# Print structured error block
print(f"")
print(f"[LLMs_Toolkit] {ts} ✗ {error_type} ERROR ({elapsed_ms}ms)")
print(f" ┌──────────────────────────────────────────────")
print(f" │ 服务商 {provider.name}")
print(f" │ 模型 {model or 'default'}")
if request_size_mb > 0:
print(f" │ 请求大小 {request_size_mb:.2f} MB")
print(f" │ 原因 {cause}")
if api_message:
print(f" │ API消息 {api_message[:120]}")
print(f" │ 建议 {hint}")
# Print additional diagnostic details
if details:
print(f" │")
print(f" │ 详细诊断:")
for i, detail in enumerate(details, 1):
print(f" │ {i}. {detail}")
print(f" └──────────────────────────────────────────────")
print(f"")
def _sync_api_call(
self,
endpoint: str,
payload: Dict[str, Any],
api_key: str,
max_retries: int = 2
) -> Tuple[str, Dict[str, Any]]:
"""Synchronous API call using urllib (avoids asyncio event loop conflicts)"""
url = f"{endpoint}/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
}
data_bytes = json_lib.dumps(payload).encode("utf-8")
# Log request size for debugging
data_size_mb = len(data_bytes) / (1024 * 1024)
if data_size_mb > 1:
print(f"[LLMs_Toolkit] 图片请求体大小: {data_size_mb:.2f} MB")
last_error = None
# Build a lenient SSL context to handle TLS compatibility issues
# (e.g. UNEXPECTED_EOF_WHILE_READING from some Chinese API providers)
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
ssl_context.set_ciphers("DEFAULT")
ssl_context.options |= getattr(ssl, "OP_LEGACY_SERVER_CONNECT", 0)
for attempt in range(max_retries + 1):
try:
req = urllib.request.Request(url, data=data_bytes, headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=180, context=ssl_context) as response:
response_body = response.read().decode("utf-8")
data = json_lib.loads(response_body)
if "choices" not in data or len(data["choices"]) == 0:
raise ValueError(f"API response missing 'choices'. Full response: {data}")
content = data["choices"][0]["message"]["content"]
return content, data
except urllib.error.HTTPError as e:
error_body = e.read().decode("utf-8", errors="replace")
raise Exception(f"HTTP {e.code} | {error_body}")
except urllib.error.URLError as e:
last_error = e
error_msg = str(e.reason)
# Check if it's a broken pipe or connection error
if "Broken pipe" in error_msg or "Connection" in error_msg:
if attempt < max_retries:
print(f"[LLMs_Toolkit] 连接错误,重试 {attempt + 1}/{max_retries}...")
time.sleep(2 ** attempt) # Exponential backoff
continue
raise Exception(f"URLError | {error_msg}")
except TimeoutError:
raise Exception(f"TimeoutError | 请求超过 180 秒未响应")
# If all retries failed
if last_error:
raise Exception(f"URLError | {str(last_error.reason)} (重试 {max_retries} 次后仍失败)")
raise Exception("Unknown error occurred")
def _build_payload(
self,
model: str,
messages: List[Dict[str, Any]],
temperature: float,
max_tokens: int
) -> Dict[str, Any]:
"""Build API request payload"""
return {
"model": model or "glm-4",
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"request_id": f"req-{int(time.time() * 1000)}-{random.randint(1000, 9999)}",
"timestamp": int(time.time() * 1000),
"seed": random.randint(1, 1000000)
}
def _manage_conversation_history(
self,
messages: List[Dict[str, Any]],
enable_memory: bool
def _apply_memory(
self, messages: List[Dict[str, Any]], enable: bool
) -> List[Dict[str, Any]]:
"""Manage conversation history with memory support"""
if not enable_memory:
"""Manage conversation history."""
if not enable:
self._conversation_history = []
return messages
# Append to history and return full context
for msg in messages:
if not any(
h["role"] == msg["role"] and h["content"] == msg["content"]
for h in self._conversation_history
):
self._conversation_history.append(msg)
return self._conversation_history
return list(self._conversation_history)
# ── Result helpers ───────────────────────────────────────────────────
@staticmethod
def _success(text: str, in_tok: int, out_tok: int) -> dict:
return {
"ui": {
"text": [f"Token Usage:\nInput: {in_tok}\nOutput: {out_tok}"]
},
"result": (text, in_tok, out_tok)
}
@staticmethod
def _error_result(error_msg: str) -> dict:
"""Return a graceful error result instead of crashing."""
return {
"ui": {
"text": [f"⚠ Error:\n{error_msg}"]
},
"result": (f"[Error] {error_msg}", 0, 0)
}
# ── Main entry ───────────────────────────────────────────────────────
def generate(
self,
llm_config: Dict[str, Any],
@@ -527,91 +214,82 @@ class OpenAICompatibleLoader:
prep_img: Optional[str] = None,
enable_memory: bool = False,
seed: int = 0
) -> Tuple[str, int, int]:
"""Main generation entry point"""
) -> dict:
"""Main generation entry point with graceful degradation."""
# Extract config parameters
# ── Extract config ───────────────────────────────────────────
base_url = llm_config.get("base_url", "")
model = llm_config.get("model", "")
api_key = llm_config.get("api_key", "")
provider_name = llm_config.get("provider", "Custom")
# Log using config
print(f"[LLMs_Toolkit] 使用配置: {provider_name} / {model}")
# ── Input validation (fail fast, don't waste API quota) ──────
if not prompt or not prompt.strip():
return self._error_result("输入的 Prompt 为空,请填写内容后重试")
if not api_key or not api_key.strip():
return self._error_result("API Key 为空,请在 LLMs Loader 节点中配置")
if not base_url or not base_url.strip():
return self._error_result("Base URL 为空,请检查服务商配置")
if not model or not model.strip():
return self._error_result("模型名称为空,请在 LLMs Loader 节点中填写")
# Handle prep_img: parse JSON list if applicable
# ── Parse image input ────────────────────────────────────────
image_input = None
if prep_img:
if prep_img.strip().startswith("["):
if prep_img and prep_img.strip():
stripped = prep_img.strip()
if stripped.startswith("["):
try:
image_input = json_lib.loads(prep_img)
image_input = json_lib.loads(stripped)
except Exception:
# Fallback: treat as single string
image_input = prep_img
image_input = stripped
else:
image_input = prep_img
# Get provider configuration
provider = ProviderRegistry.get_provider(provider_name)
# Log request details
self._log_request_start(
provider=provider,
model=model,
system_prompt=system_prompt,
prompt=prompt,
image_input=image_input
)
# Build message content
content = MessageBuilder.build_content(prompt, image_input)
# Build messages with system prompt
messages = MessageBuilder.build_messages(content, system_prompt)
# Apply provider-specific adaptations
messages = MessageBuilder.adapt_for_provider(messages, provider, prompt)
# Manage conversation history
messages = self._manage_conversation_history(messages, enable_memory)
# Build request payload
payload = self._build_payload(model, messages, temperature, max_tokens)
image_input = stripped
# Calculate request size for error logging
# ── Logging ──────────────────────────────────────────────────
print(f"{self.TAG} 使用配置: {provider_name} / {model}")
start = self._log_start(
provider_name, model, system_prompt, prompt, image_input
)
# ── Build messages ───────────────────────────────────────────
content = _build_content(prompt, image_input)
messages = _build_messages(content, system_prompt, provider_name)
messages = self._apply_memory(messages, enable_memory)
# ── Build payload (clean, no non-standard fields) ────────────
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
}
# Calculate request size for diagnostics
payload_bytes = json_lib.dumps(payload).encode("utf-8")
request_size_mb = len(payload_bytes) / (1024 * 1024)
# Estimate input tokens
input_tokens = TokenEstimator.estimate_input_tokens(
prompt, system_prompt, image_input
)
# ── Make API call ────────────────────────────────────────────
try:
# Synchronous call to avoid asyncio event loop conflicts
response_content, data = self._sync_api_call(
base_url, payload, api_key
)
client = LLMClient(base_url, api_key)
response_content, data = client.chat(payload)
# Extract actual token usage from response
output_tokens = data.get("usage", {}).get("completion_tokens", 0)
# Extract real token usage from API response (prefer actual over estimate)
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0) or len(prompt)
output_tokens = usage.get("completion_tokens", 0)
# Log completion
self._log_request_complete(response_content, input_tokens, output_tokens)
return {
"ui": {
"text": [f"Token Usage:\nInput: {int(input_tokens)}\nOutput: {int(output_tokens)}"]
},
"result": (response_content, int(input_tokens), int(output_tokens))
}
self._log_done(response_content, input_tokens, output_tokens, start)
return self._success(response_content, input_tokens, output_tokens)
except Exception as e:
self._log_error(e, provider, model, request_size_mb)
raise
elapsed_ms = int((time.time() - start) * 1000)
err = classify_error(e, provider_name, model, request_size_mb, elapsed_ms)
log_error(err, provider_name, model, request_size_mb, elapsed_ms)
# Graceful degradation: return error text instead of crashing
return self._error_result(err.user_message())
# ComfyUI node registration
NODE_CLASS_MAPPINGS = {"OpenAICompatibleLoader": OpenAICompatibleLoader}
NODE_DISPLAY_NAME_MAPPINGS = {"OpenAICompatibleLoader": "OpenAI Compatible Adapter"}
+101
View File
@@ -0,0 +1,101 @@
import json
import urllib.request
import urllib.parse
import sys
import time
# 测试用的一个简单 ComfyUI 工作流,只包含我们修改的加载器和处理器
# 我们需要向本地模型发起请求(因为可能没有 API Key),由于没有Key,它可能会返回 Error,
# 但为了验证 UI 格式,只要看 error 结构或正常响应是否包含 `ui` 参数。
# 注意我们手动传入一些测试参数
workflow = {
"1": {
"inputs": {
"provider": "Custom/自定义",
"model": "foo",
"api_key": "YOUR_API_KEY", # 替换为您想要测试的真实 KEY 或者随便乱填以产生报错
"custom_base_url": "http://example.com"
},
"class_type": "LLM_Loader"
},
"2": {
"inputs": {
"system_prompt": "你是一个AI大模型",
"prompt": "你好",
"prep_img": "",
"temperature": 0.7,
"seed": 0,
"llm_config": [
"1",
0
]
},
"class_type": "OpenAICompatibleLoader"
},
"3": {
"inputs": {
"filename_prefix": "ComfyUI",
"images": [
"4",
0
]
},
"class_type": "SaveImage"
},
"4": {
"inputs": {
"image": "example.png"
},
"class_type": "LoadImage"
}
}
def queue_prompt(prompt_workflow):
p = {"prompt": prompt_workflow}
data = json.dumps(p).encode('utf-8')
req = urllib.request.Request("http://127.0.0.1:8188/prompt", data=data)
return json.loads(urllib.request.urlopen(req).read())
def get_history(prompt_id):
with urllib.request.urlopen("http://127.0.0.1:8188/history/{}".format(prompt_id)) as response:
return json.loads(response.read())
def test():
try:
print("Submitting workflow to ComfyUI...")
response = queue_prompt(workflow)
prompt_id = response['prompt_id']
print(f"Submitted! Prompt ID: {prompt_id}")
while True:
history = get_history(prompt_id)
if prompt_id in history:
print("\nExecution finished!")
result = history[prompt_id]
# Check for UI data in node '2'
outputs = result.get('outputs', {})
if '2' in outputs:
node_2_output = outputs['2']
if 'ui' in node_2_output:
print("✅ Success! 'ui' attribute found in response:")
print(json.dumps(node_2_output['ui'], indent=2, ensure_ascii=False))
else:
print("❌ Failed! No 'ui' attribute found in response:")
print(json.dumps(node_2_output, indent=2, ensure_ascii=False))
break
time.sleep(1)
except urllib.error.HTTPError as e:
error_body = e.read().decode('utf-8', errors='replace')
print(f"HTTP Error {e.code}:\n{error_body}")
except Exception as e:
print(f"Test failed (is ComfyUI running?): {e}")
if __name__ == "__main__":
test()