From 1a15c8d1299c49aa72a8a020fdfd826354046bfd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A7=91=E6=9E=97=5FCyril=20Pilgrim?= Date: Sat, 28 Feb 2026 16:31:06 +0800 Subject: [PATCH] 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. --- .gitignore | 3 + __init__.py | 2 + nodes/api_client.py | 308 +++++++++++++++++ nodes/llm_translator.py | 119 +++---- nodes/openai_compatible.py | 682 ++++++++++--------------------------- test_api.py | 101 ++++++ 6 files changed, 634 insertions(+), 581 deletions(-) create mode 100644 nodes/api_client.py create mode 100644 test_api.py diff --git a/.gitignore b/.gitignore index 1d0d50f..d7395a0 100644 --- a/.gitignore +++ b/.gitignore @@ -33,5 +33,8 @@ ENV/ .DS_Store Thumbs.db +# Internal docs (private) +CONTRIBUTING.md + # Tests tests/ diff --git a/__init__.py b/__init__.py index 39b03a4..aaba82d 100644 --- a/__init__.py +++ b/__init__.py @@ -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 diff --git a/nodes/api_client.py b/nodes/api_client.py new file mode 100644 index 0000000..879d786 --- /dev/null +++ b/nodes/api_client.py @@ -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) diff --git a/nodes/llm_translator.py b/nodes/llm_translator.py index e177882..a27c556 100644 --- a/nodes/llm_translator.py +++ b/nodes/llm_translator.py @@ -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} diff --git a/nodes/openai_compatible.py b/nodes/openai_compatible.py index 360cfe4..3d9c348 100644 --- a/nodes/openai_compatible.py +++ b/nodes/openai_compatible.py @@ -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"} - diff --git a/test_api.py b/test_api.py new file mode 100644 index 0000000..142164a --- /dev/null +++ b/test_api.py @@ -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()