ComfyUI 설정 파일 개발
ComfyUI Custom Node의 Config들을 저장 파일을 개발해줘. config 파일을 사람이 수정하기 쉬운 파일 형식으로 수정 해줘. 파일을 저장하지 않고 설정을 읽기만 하는 readonly 모드로 동작하면 돼. 여러 설정을 저장할거지만 주 용도는 LLM 엔드포인트 정보를 담을거야. 각 LLM Model 별로 (디스플레이용 모델명, 실제 모델명(request시), base url, api key)를 저장하면 돼. Config 정보에 관한 요청이 들어오면(e.g. api 날릴 때 등) 해당하는 값을 리턴하면 돼. 이 때 config 파일이 초기 read했을 때와 바뀌었을 수도 있어. 조금 반영이 늦더라도 1분 이내에는 변경된 값을 리턴해야해.
좋아요! “사람이 직접 수정하기 쉬운 설정 파일” + “읽기 전용(파일 저장 X)” + “최대 1분 내 최신값 반영” 조건을 만족하는 ComfyUI 커스텀 노드 패키지를 설계/구현해 드릴게요.
아래 그대로 복사해 넣으면 바로 동작합니다.
1) 파일 구조
ComfyUI/
└─ custom_nodes/
└─ comfyui_llm_config_reader/
├─ __init__.py
├─ llm_config_reader.py # 커스텀 노드 + 설정 로더
└─ llm_config.toml # 사람이 읽고 수정하기 쉬운 TOML 설정 파일
기본 경로는
llm_config.toml이고,LLM_CONFIG_PATH환경변수로 바꿀 수 있어요.
2) 설정 파일 예시 (TOML)
사람이 손으로 편집하기 쉽고, 주석도 넣기 좋은 TOML을 사용합니다.
(API Key는 깃에 올리지 않도록 환경변수 치환도 지원해요.)
llm_config.toml
# 버전 정보(선택)
version = 1
[settings]
# 파일 변경 반영 주기(초). 60초 이내 최신값 요구이므로 기본 30초로 여유 있게.
refresh_interval_seconds = 30
[llm_models."gpt-4o-mini"]
display_name = "GPT-4o mini"
request_model = "gpt-4o-mini"
base_url = "https://api.openai.com/v1"
api_key = "${OPENAI_API_KEY}" # 환경변수로 안전하게 주입
[llm_models."claude-3-5-sonnet"]
display_name = "Claude 3.5 Sonnet"
request_model = "claude-3-5-sonnet"
base_url = "https://api.anthropic.com"
api_key = "${ANTHROPIC_API_KEY}"
# 원하는 만큼 모델을 계속 추가
# [llm_models."문자열키"] ...
-
키 선택 규칙:
llm_models."<키>"의<키>는 자유로운 문자열입니다.
조회 시<키>,display_name,request_model셋 중 어느 걸로 입력해도 매칭되도록 했어요. -
${ENV_NAME}형태는 환경변수로 치환됩니다.
3) 커스텀 노드/로더 코드
comfyui_llm_config_reader/llm_config_reader.py
import os
import time
import threading
from typing import Dict, Any, Optional, Tuple
# Python 3.11+ tomllib, 3.10 이하면 tomli 사용
try:
import tomllib # type: ignore
_TOML_LOADS = tomllib.loads
except Exception:
import tomli # type: ignore
_TOML_LOADS = tomli.loads
def _expand_env_vars(value: Any) -> Any:
"""문자열 내 ${ENV} 패턴을 환경변수로 치환. 그 외 타입은 그대로 반환."""
if isinstance(value, str):
out = ""
i = 0
while i < len(value):
if value[i:i+2] == "${":
j = value.find("}", i+2)
if j != -1:
env_name = value[i+2:j]
out += os.environ.get(env_name, "")
i = j + 1
else:
out += value[i]
i += 1
else:
out += value[i]
i += 1
return out
if isinstance(value, dict):
return {k: _expand_env_vars(v) for k, v in value.items()}
if isinstance(value, list):
return [_expand_env_vars(v) for v in value]
return value
def _normalize_key(s: str) -> str:
"""매칭 편의를 위한 정규화(소문자, 앞뒤 공백 제거)."""
return s.strip().lower()
class ReadOnlyConfig:
"""
- TOML 설정을 읽어서 메모리에 캐시
- 파일이 바뀌었거나 refresh_interval이 지났으면 재로드
- 쓰기는 절대 하지 않음 (read-only)
"""
def __init__(self, path: Optional[str] = None):
self.path = path or os.environ.get(
"LLM_CONFIG_PATH",
os.path.join(os.path.dirname(__file__), "llm_config.toml"),
)
self._lock = threading.Lock()
self._loaded_at: float = 0.0
self._last_mtime: float = -1.0
self._refresh_interval: int = 30 # 기본 30초(<= 60초 요구 충족)
self._raw: Dict[str, Any] = {}
self._index: Dict[str, Dict[str, str]] = {} # 정규화 키 -> 모델 dict
def _read_text(self) -> str:
with open(self.path, "rb") as f:
return f.read().decode("utf-8")
def _build_index(self, data: Dict[str, Any]) -> None:
self._index.clear()
models = data.get("llm_models", {}) or {}
for model_key, cfg in models.items():
if not isinstance(cfg, dict):
continue
# 환경변수 치환
cfg = _expand_env_vars(cfg)
# 필수 필드 체크
required = ["display_name", "request_model", "base_url", "api_key"]
missing = [k for k in required if k not in cfg or not str(cfg[k]).strip()]
if missing:
# 불완전한 엔트리는 스킵 (명확한 로그가 필요하면 print 가능)
continue
# 모델 접근을 위한 다양한 키들 인덱싱
candidates = {
str(model_key),
str(cfg.get("display_name", "")),
str(cfg.get("request_model", "")),
}
for c in filter(None, candidates):
self._index[_normalize_key(c)] = {
"display_name": str(cfg["display_name"]),
"request_model": str(cfg["request_model"]),
"base_url": str(cfg["base_url"]),
"api_key": str(cfg["api_key"]),
}
def _reload_locked(self) -> None:
txt = self._read_text()
data = _TOML_LOADS(txt)
# settings.refresh_interval_seconds 반영 (최소 5초, 최대 60초로 가드)
refresh = int(
(data.get("settings") or {}).get("refresh_interval_seconds", 30)
)
refresh = max(5, min(60, refresh))
self._refresh_interval = int(
os.environ.get("LLM_CONFIG_REFRESH_SECONDS", refresh)
)
self._raw = _expand_env_vars(data)
self._build_index(self._raw)
self._loaded_at = time.time()
self._last_mtime = os.path.getmtime(self.path)
def _stale_or_changed(self) -> bool:
try:
mtime = os.path.getmtime(self.path)
except FileNotFoundError:
return True
now = time.time()
return (mtime != self._last_mtime) or (now - self._loaded_at >= self._refresh_interval)
def _ensure_loaded(self) -> None:
with self._lock:
if self._loaded_at == 0 or self._stale_or_changed():
self._reload_locked()
def get_model(self, key: str) -> Dict[str, str]:
"""key는 모델 키/표시명/요청용 모델명 중 아무거나 가능."""
if not key or not str(key).strip():
raise KeyError("model key is empty")
self._ensure_loaded()
hit = self._index.get(_normalize_key(key))
if not hit:
# 최신 반영 지연 가능성을 고려해 한 번 더 강제 리로드 후 재시도
with self._lock:
self._reload_locked()
hit = self._index.get(_normalize_key(key))
if not hit:
raise KeyError(f"model not found: {key}")
return hit
def list_models(self) -> Dict[str, Dict[str, str]]:
"""디버그/인스펙션용: (표시명, 요청모델, base_url만 노출하는 안전 리스트)."""
self._ensure_loaded()
out = {}
# 역으로 display_name 기준 집계
for k, v in self._index.items():
dn = v["display_name"]
out[dn] = {
"display_name": v["display_name"],
"request_model": v["request_model"],
"base_url": v["base_url"],
# api_key는 보안상 미노출
}
return out
# 전역 싱글톤 로더 (ComfyUI 노드들이 공유)
_CONFIG = ReadOnlyConfig()
# ===== ComfyUI Custom Nodes =====
class LLMConfigLookup:
"""
입력: model_key (문자열) — 예: "gpt-4o-mini" 또는 "GPT-4o mini" 또는 "gpt-4o-mini"(request_model)
출력: display_name, request_model, base_url, api_key (모두 STRING)
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"model_key": ("STRING", {"default": "", "multiline": False}),
},
}
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING")
RETURN_NAMES = ("display_name", "request_model", "base_url", "api_key")
FUNCTION = "lookup"
CATEGORY = "LLM/Config"
def lookup(self, model_key: str):
try:
m = _CONFIG.get_model(model_key)
except Exception as e:
# ComfyUI에서 에러가 보이도록 예외 전달
raise RuntimeError(f"[LLMConfigLookup] {e}")
return (m["display_name"], m["request_model"], m["base_url"], m["api_key"])
class LLMConfigListModels:
"""
디버그용: 사용 가능한 모델들의 (표시명, 요청모델, base_url) JSON을 반환.
"""
@classmethod
def INPUT_TYPES(cls):
return {"required": {}}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("models_json",)
FUNCTION = "list_models"
CATEGORY = "LLM/Config"
def list_models(self):
import json
data = _CONFIG.list_models()
return (json.dumps(data, ensure_ascii=False, indent=2),)
# ComfyUI 등록용 매핑
NODE_CLASS_MAPPINGS = {
"LLMConfigLookup": LLMConfigLookup,
"LLMConfigListModels": LLMConfigListModels,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"LLMConfigLookup": "LLM: Config Lookup (ReadOnly)",
"LLMConfigListModels": "LLM: List Models (ReadOnly)",
}
comfyui_llm_config_reader/__init__.py
# 빈 파일이면 충분합니다. (패키지 인식용)
4) 동작 보장(요구사항 매핑)
-
사람이 수정하기 쉬운 포맷 : TOML 사용 + 주석 가능 + 간결한 키/값 구조.
-
읽기 전용 : 코드 어디에서도 파일을 쓰지 않음. 오직 읽기/파싱만 수행.
-
다중 설정(특히 LLM 엔드포인트) :
llm_models."<키>"마다display_name,request_model,base_url,api_key저장.
-
요청 시 해당 값 리턴 :
LLMConfigLookup노드가 입력model_key로 검색하여 4개 값을 출력.<키>/display_name/request_model중 무엇으로 넣어도 매칭됨.
-
변경 반영(최대 1분) :
-
파일
mtime변화 감지 +refresh_interval_seconds(기본 30초) 또는LLM_CONFIG_REFRESH_SECONDS(env) 기준 TTL 캐시. -
호출 시마다 stale 또는 변경 이면 즉시 재로드 → 최대 60초 이내 최신값 보장.
-
신규 항목이 추가된 직후에도, 다음 호출 때 최대 TTL 뒤에는 반영.
-
-
보안 고려 :
-
${ENV}치환 지원으로 API Key를 환경변수에서 주입 가능. -
ListModels노드에서는 API Key 미노출.
-
5) 사용 방법 (간단)
-
위 파일 구조대로 복사 → ComfyUI 재실행.
-
llm_config.toml에 모델들을 추가/수정. -
워크플로우에서 LLM: Config Lookup (ReadOnly) 노드 배치 →
model_key입력(예:gpt-4o-mini). -
노드 출력 4개(표시명/요청모델/base_url/api_key)를 원하는 LLM 호출 노드에 연결.
-
설정 파일을 수정하면, 최대 30초(기본) 내 새 값이 자동 반영됩니다. (환경변수로 5~60초 사이 튜닝 가능)
6) 커스터마이징 팁
-
YAML 선호 시 :
tomllib대신yaml.safe_load로 바꾸고 파일 확장자만.yaml로 바꾸면 됩니다. -
모델 키 자동 드롭다운 : ComfyUI의 동적 입력 UI가 필요하면,
INPUT_TYPES를 주기적으로 재구성하는 커스텀 프론트엔드가 필요합니다. 기본 구현은 문자열 입력으로 단순/견고하게 처리했습니다. -
필드 확장 : per-model 타임아웃, 헤더 등 추가 필드가 필요하면 TOML에 키를 늘리고
_build_index에서 그대로 통과시켜 출력을 늘리면 됩니다.
원하시면 YAML 버전 또는 특정 LLM 호출 노드(예: OpenAI/Anthropic 커넥터)까지 바로 연결되는 노드도 추가로 만들어 드릴게요.
좋은데 LLM 말고도 다른 설정들도 저장할 수 있는 범용적 config 파일이라서 파일명을 바꿔주면 좋겠어
좋은 포인트예요! 범용 설정 파일로 쓰기 좋게 파일명을 config.toml로 바꾸고, 코드/환경변수도 일반화 했습니다. (이전 llm_config.toml·LLM_CONFIG_PATH도 자동 하위 호환)
아래 그대로 교체/추가하면 됩니다.
변경 요약 (step-by-step)
-
설정 파일 기본명을
config.toml로 변경 -
경로 환경변수도 일반화:
COMFY_CONFIG_PATH(구LLM_CONFIG_PATH도 인식) -
새로고침 주기 환경변수:
COMFY_CONFIG_REFRESH_SECONDS(구LLM_CONFIG_REFRESH_SECONDS도 인식) -
코드가
config.toml을 우선 사용, 없으면 기존llm_config.toml자동 폴백 -
[llm_models]외에 임의 섹션([feature_flags],[storage]등) 자유 확장 가능
디렉토리 구조
ComfyUI/
└─ custom_nodes/
└─ comfyui_llm_config_reader/
├─ __init__.py
├─ llm_config_reader.py # (교체본) 범용 config 로더 + LLM 조회 노드
└─ config.toml # ← 새 기본 설정 파일명 (범용)
# └─ llm_config.toml # (선택) 기존 파일명도 계속 동작(폴백)
예시 config.toml (범용)
version = 1
[settings]
# 파일 변경 반영 주기(초). 1분 이내 요구 → 기본 30초.
refresh_interval_seconds = 30
# ── LLM 엔드포인트들 ──────────────────────────────────────────────
[llm_models."gpt-4o-mini"]
display_name = "GPT-4o mini"
request_model = "gpt-4o-mini"
base_url = "https://api.openai.com/v1"
api_key = "${OPENAI_API_KEY}"
[llm_models."claude-3-5-sonnet"]
display_name = "Claude 3.5 Sonnet"
request_model = "claude-3-5-sonnet"
base_url = "https://api.anthropic.com"
api_key = "${ANTHROPIC_API_KEY}"
# ── (예) 다른 범용 설정 섹션들 ───────────────────────────────────
[feature_flags]
use_cache = true
show_beta = false
[storage.s3]
bucket = "my-bucket"
region = "ap-northeast-2"
access_key = "${S3_ACCESS_KEY}"
secret_key = "${S3_SECRET_KEY}"
-
임의 섹션은 그대로 저장 되며, LLM 노드는
[llm_models]만 사용합니다. -
${ENV}표기값은 환경변수로 치환됩니다.
교체본 코드: llm_config_reader.py
import os
import time
import threading
from typing import Dict, Any, Optional
# Python 3.11+ tomllib, 3.10 이하면 tomli 사용
try:
import tomllib # type: ignore
_TOML_LOADS = tomllib.loads
except Exception:
import tomli # type: ignore
_TOML_LOADS = tomli.loads
def _expand_env_vars(value: Any) -> Any:
"""문자열 내 ${ENV} 패턴을 환경변수로 치환. 그 외 타입은 그대로."""
if isinstance(value, str):
out, i = "", 0
while i < len(value):
if value[i:i+2] == "${":
j = value.find("}", i+2)
if j != -1:
env_name = value[i+2:j]
out += os.environ.get(env_name, "")
i = j + 1
else:
out += value[i]
i += 1
else:
out += value[i]
i += 1
return out
if isinstance(value, dict):
return {k: _expand_env_vars(v) for k, v in value.items()}
if isinstance(value, list):
return [_expand_env_vars(v) for v in value]
return value
def _normalize_key(s: str) -> str:
return str(s).strip().lower()
class ReadOnlyConfig:
"""
- TOML 설정을 읽어 메모리 캐시
- 파일 mtime/TTL(최소 5s, 최대 60s) 기반 자동 재로드
- 절대 쓰기 없음 (read-only)
- 파일명 일반화:
* 우선순위: COMFY_CONFIG_PATH > LLM_CONFIG_PATH(하위호환) >
./config.toml > ./llm_config.toml(하위호환)
"""
def __init__(self, path: Optional[str] = None):
# 1) 명시적 인자
explicit = path
# 2) 환경변수 (일반화 + 하위호환)
env_path = os.environ.get("COMFY_CONFIG_PATH") or os.environ.get("LLM_CONFIG_PATH")
# 3) 기본 경로 탐색 (우선 config.toml, 없으면 llm_config.toml)
here = os.path.dirname(__file__)
default_primary = os.path.join(here, "config.toml")
default_legacy = os.path.join(here, "llm_config.toml")
if explicit:
chosen = explicit
elif env_path:
chosen = env_path
elif os.path.exists(default_primary):
chosen = default_primary
elif os.path.exists(default_legacy):
chosen = default_legacy
else:
# 파일이 아직 없을 수 있으니 새 표준명으로 고정
chosen = default_primary
self.path = chosen
self._lock = threading.Lock()
self._loaded_at: float = 0.0
self._last_mtime: float = -1.0
self._refresh_interval: int = 30 # 기본 30초 (<=60초 요구 충족)
self._raw: Dict[str, Any] = {}
self._index: Dict[str, Dict[str, str]] = {} # 정규화 키 -> LLM 모델 dict
def _read_text(self) -> str:
with open(self.path, "rb") as f:
return f.read().decode("utf-8")
def _build_index_llm(self, data: Dict[str, Any]) -> None:
"""[llm_models] 섹션만 인덱싱. 다른 섹션은 그대로 self._raw에 보존."""
self._index.clear()
models = (data.get("llm_models") or {})
if not isinstance(models, dict):
return
for model_key, cfg in models.items():
if not isinstance(cfg, dict):
continue
# 환경변수 치환
cfg = _expand_env_vars(cfg)
# 필수 필드 체크
required = ["display_name", "request_model", "base_url", "api_key"]
missing = [k for k in required if k not in cfg or not str(cfg[k]).strip()]
if missing:
# 불완전 항목은 스킵
continue
# 접근 키(모델키/표시명/요청모델) 인덱싱
candidates = {
str(model_key),
str(cfg.get("display_name", "")),
str(cfg.get("request_model", "")),
}
for c in filter(None, candidates):
self._index[_normalize_key(c)] = {
"display_name": str(cfg["display_name"]),
"request_model": str(cfg["request_model"]),
"base_url": str(cfg["base_url"]),
"api_key": str(cfg["api_key"]),
}
def _reload_locked(self) -> None:
txt = self._read_text()
data = _TOML_LOADS(txt)
# refresh_interval_seconds (최소 5, 최대 60) + env override(일반/하위호환)
conf_val = int((data.get("settings") or {}).get("refresh_interval_seconds", 30))
conf_val = max(5, min(60, conf_val))
env_override = (
os.environ.get("COMFY_CONFIG_REFRESH_SECONDS")
or os.environ.get("LLM_CONFIG_REFRESH_SECONDS")
)
if env_override:
try:
conf_val = max(5, min(60, int(env_override)))
except ValueError:
pass
self._refresh_interval = conf_val
self._raw = _expand_env_vars(data)
self._build_index_llm(self._raw)
self._loaded_at = time.time()
try:
self._last_mtime = os.path.getmtime(self.path)
except FileNotFoundError:
self._last_mtime = -1.0
def _stale_or_changed(self) -> bool:
try:
mtime = os.path.getmtime(self.path)
except FileNotFoundError:
return True
now = time.time()
return (mtime != self._last_mtime) or (now - self._loaded_at >= self._refresh_interval)
def _ensure_loaded(self) -> None:
with self._lock:
if self._loaded_at == 0 or self._stale_or_changed():
self._reload_locked()
# ── LLM 전용 헬퍼 (기존 API 유지) ─────────────────────────────
def get_llm_model(self, key: str) -> Dict[str, str]:
if not key or not str(key).strip():
raise KeyError("model key is empty")
self._ensure_loaded()
hit = self._index.get(_normalize_key(key))
if not hit:
# 갱신 지연 가능성 → 강제 리로드 1회 재시도
with self._lock:
self._reload_locked()
hit = self._index.get(_normalize_key(key))
if not hit:
raise KeyError(f"model not found: {key}")
return hit
def list_llm_models(self) -> Dict[str, Dict[str, str]]:
self._ensure_loaded()
out: Dict[str, Dict[str, str]] = {}
for _, v in self._index.items():
dn = v["display_name"]
out[dn] = {
"display_name": v["display_name"],
"request_model": v["request_model"],
"base_url": v["base_url"],
# api_key는 보안상 미노출
}
return out
# ── 범용 접근(옵션): 원본 트리에서 임의 섹션 읽기 ─────────────
def get_raw_config(self) -> Dict[str, Any]:
"""전체 설정 트리를 그대로(ENV 확장 후) 반환. 읽기 전용."""
self._ensure_loaded()
return self._raw
# 전역 싱글톤
_CONFIG = ReadOnlyConfig()
# ===== ComfyUI Custom Nodes =====
class LLMConfigLookup:
"""
입력: model_key (문자열) — "gpt-4o-mini" / "GPT-4o mini" / "gpt-4o-mini"(request_model)
출력: display_name, request_model, base_url, api_key
"""
@classmethod
def INPUT_TYPES(cls):
return {"required": {"model_key": ("STRING", {"default": "", "multiline": False})}}
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING")
RETURN_NAMES = ("display_name", "request_model", "base_url", "api_key")
FUNCTION = "lookup"
CATEGORY = "LLM/Config"
def lookup(self, model_key: str):
try:
m = _CONFIG.get_llm_model(model_key)
except Exception as e:
raise RuntimeError(f"[LLMConfigLookup] {e}")
return (m["display_name"], m["request_model"], m["base_url"], m["api_key"])
class LLMConfigListModels:
"""디버그용: 사용 가능한 LLM 모델 목록(JSON) 반환 (API Key 미포함)."""
@classmethod
def INPUT_TYPES(cls):
return {"required": {}}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("models_json",)
FUNCTION = "list_models"
CATEGORY = "LLM/Config"
def list_models(self):
import json
data = _CONFIG.list_llm_models()
return (json.dumps(data, ensure_ascii=False, indent=2),)
# (선택) 범용 조회 노드: 임의 경로를 JSONPath 유사 키로 조회하고 STRING 반환
class ConfigGetRawJSON:
"""
예) path="feature_flags.use_cache" → "true"
path="storage.s3.region" → "ap-northeast-2"
"""
@classmethod
def INPUT_TYPES(cls):
return {"required": {"path": ("STRING", {"default": "", "multiline": False})}}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("value",)
FUNCTION = "get_raw"
CATEGORY = "LLM/Config"
def get_raw(self, path: str):
import json
def get_in(d: Any, dotted: str):
cur = d
for part in filter(None, dotted.split(".")):
if isinstance(cur, dict) and part in cur:
cur = cur[part]
else:
raise KeyError(f"config path not found: {dotted}")
return cur
try:
raw = _CONFIG.get_raw_config()
val = get_in(raw, path) if path else raw
# 스칼라면 문자열화, 그 외엔 JSON 문자열로 반환
if isinstance(val, (str, int, float, bool)) or val is None:
return (str(val),)
return (json.dumps(val, ensure_ascii=False, indent=2),)
except Exception as e:
raise RuntimeError(f"[ConfigGetRawJSON] {e}")
NODE_CLASS_MAPPINGS = {
"LLMConfigLookup": LLMConfigLookup,
"LLMConfigListModels": LLMConfigListModels,
"ConfigGetRawJSON": ConfigGetRawJSON, # ← 범용 읽기 노드(선택)
}
NODE_DISPLAY_NAME_MAPPINGS = {
"LLMConfigLookup": "LLM: Config Lookup (ReadOnly)",
"LLMConfigListModels": "LLM: List Models (ReadOnly)",
"ConfigGetRawJSON": "Config: Get Raw JSON (ReadOnly)",
}
노드/패키지명은 그대로 두고, 파일명/환경변수/로더 만 일반화했습니다. 원하시면 패키지/노드 이름도
comfyui_config_reader로 리네임해 드릴 수 있어요.
사용 방법 (요약)
-
config.toml로 설정 저장(범용 섹션 자유 추가 가능). -
필요 시 경로 오버라이드:
COMFY_CONFIG_PATH=/abs/path/to/config.toml -
새로고침 주기 조정:
COMFY_CONFIG_REFRESH_SECONDS=10(5~60 유효) -
ComfyUI에서
-
LLM: Config Lookup → LLM 엔드포인트 4종 값 출력
-
Config: Get Raw JSON → 임의 설정 값 읽기 (예:
feature_flags.use_cache)
-
-
파일을 수정하면 최대 1분 내(기본 30초) 반영됩니다.
필요하면 폴더/노드 이름까지 “LLM”을 빼고 전면 리네이밍한 버전도 만들어 드릴게요.
응 다 빼줘
좋아—이제 “LLM” 흔적 전부 뺐어요. 패키지/노드/환경변수/카테고리/필드 이름까지 전부 범용(config/endpoint)으로 바꾼 최종본을 드립니다. 그대로 넣으면 작동합니다.
0) 한눈에 보기 (step-by-step)
-
폴더명 변경 :
comfyui_config_reader(이전 폴더 삭제해도 됨) -
설정 파일명 :
config.toml -
환경변수 :
-
경로:
COMFY_CONFIG_PATH -
새로고침:
COMFY_CONFIG_REFRESH_SECONDS(5~60초)
-
-
섹션 이름 :
[endpoints."<키>"](필드:display_name,request_name,base_url,api_key) -
노드 이름 :
-
ConfigEndpointLookup(엔드포인트 4종 값 반환) -
ConfigListEndpoints(엔드포인트 목록 JSON) -
ConfigGetValue(임의 경로 값 가져오기)
-
1) 디렉토리 구조
ComfyUI/
└─ custom_nodes/
└─ comfyui_config_reader/
├─ __init__.py
├─ config_reader.py # 커스텀 노드 + 읽기 전용 로더
└─ config.toml # 사람이 편집하기 쉬운 TOML
2) 예시 config.toml (범용)
version = 1
[settings]
# 파일 변경 반영 주기(초). "1분 내 반영" 요건 → 기본 30초.
refresh_interval_seconds = 30
# ── 엔드포인트(예: 텍스트 모델, 임베딩 서비스, 사내 API 등) ─────────────
[endpoints."text-model-a"]
display_name = "Text Model A"
request_name = "gpt-4o-mini" # 실제 요청에 쓸 식별자/모델명/버전 등
base_url = "https://api.openai.com/v1"
api_key = "${OPENAI_API_KEY}" # 환경변수 치환 지원
[endpoints."text-model-b"]
display_name = "Text Model B"
request_name = "claude-3-5-sonnet"
base_url = "https://api.anthropic.com"
api_key = "${ANTHROPIC_API_KEY}"
# ── 다른 범용 설정들 ────────────────────────────────────────────────
[feature_flags]
use_cache = true
show_beta = false
[storage.s3]
bucket = "my-bucket"
region = "ap-northeast-2"
access_key = "${S3_ACCESS_KEY}"
secret_key = "${S3_SECRET_KEY}"
-
${ENV}표기는 환경변수로 자동 치환됩니다. -
endpoints."<키>"의<키>로,display_name·request_name까지 모두 조회 키 로 사용할 수 있어요.
3) 코드: config_reader.py
import os
import time
import threading
from typing import Dict, Any, Optional
# Python 3.11+ tomllib / 3.10-: tomli
try:
import tomllib # type: ignore
_TOML_LOADS = tomllib.loads
except Exception:
import tomli # type: ignore
_TOML_LOADS = tomli.loads
def _expand_env_vars(value: Any) -> Any:
"""문자열 내 ${ENV} 패턴을 환경변수로 치환. 그 외 타입은 그대로."""
if isinstance(value, str):
out, i = "", 0
while i < len(value):
if value[i:i+2] == "${":
j = value.find("}", i+2)
if j != -1:
env_name = value[i+2:j]
out += os.environ.get(env_name, "")
i = j + 1
else:
out += value[i]
i += 1
else:
out += value[i]
i += 1
return out
if isinstance(value, dict):
return {k: _expand_env_vars(v) for k, v in value.items()}
if isinstance(value, list):
return [_expand_env_vars(v) for v in value]
return value
def _norm(s: str) -> str:
return str(s).strip().lower()
class ReadOnlyConfig:
"""
- TOML을 읽어 메모리에 캐시
- mtime/TTL(최소 5s, 최대 60s)로 자동 재로드
- 파일 쓰기 없음(read-only)
- 파일 결정 우선순위: (1) 인자 path > (2) COMFY_CONFIG_PATH > (3) ./config.toml
"""
def __init__(self, path: Optional[str] = None):
here = os.path.dirname(__file__)
chosen = (
path
or os.environ.get("COMFY_CONFIG_PATH")
or os.path.join(here, "config.toml")
)
self.path = chosen
self._lock = threading.Lock()
self._loaded_at: float = 0.0
self._last_mtime: float = -1.0
self._refresh_interval: int = 30 # 기본 30초(<=60s 요건 충족)
self._raw: Dict[str, Any] = {}
self._endpoint_index: Dict[str, Dict[str, str]] = {} # 정규화 키 -> endpoint dict
def _read_text(self) -> str:
with open(self.path, "rb") as f:
return f.read().decode("utf-8")
def _build_index_endpoints(self, data: Dict[str, Any]) -> None:
self._endpoint_index.clear()
eps = (data.get("endpoints") or {})
if not isinstance(eps, dict):
return
for ep_key, cfg in eps.items():
if not isinstance(cfg, dict):
continue
cfg = _expand_env_vars(cfg)
# 필수 필드
required = ["display_name", "request_name", "base_url", "api_key"]
missing = [k for k in required if k not in cfg or not str(cfg[k]).strip()]
if missing:
# 불완전 항목은 스킵
continue
# 접근 키: 정의 키 / display_name / request_name
candidates = {
str(ep_key),
str(cfg.get("display_name", "")),
str(cfg.get("request_name", "")),
}
for c in filter(None, candidates):
self._endpoint_index[_norm(c)] = {
"display_name": str(cfg["display_name"]),
"request_name": str(cfg["request_name"]),
"base_url": str(cfg["base_url"]),
"api_key": str(cfg["api_key"]),
}
def _reload_locked(self) -> None:
txt = self._read_text()
data = _TOML_LOADS(txt)
# refresh_interval_seconds (5~60) + env override
conf_val = int((data.get("settings") or {}).get("refresh_interval_seconds", 30))
conf_val = max(5, min(60, conf_val))
env_override = os.environ.get("COMFY_CONFIG_REFRESH_SECONDS")
if env_override:
try:
conf_val = max(5, min(60, int(env_override)))
except ValueError:
pass
self._refresh_interval = conf_val
self._raw = _expand_env_vars(data)
self._build_index_endpoints(self._raw)
self._loaded_at = time.time()
try:
self._last_mtime = os.path.getmtime(self.path)
except FileNotFoundError:
self._last_mtime = -1.0
def _stale_or_changed(self) -> bool:
try:
mtime = os.path.getmtime(self.path)
except FileNotFoundError:
return True
now = time.time()
return (mtime != self._last_mtime) or (now - self._loaded_at >= self._refresh_interval)
def _ensure_loaded(self) -> None:
with self._lock:
if self._loaded_at == 0 or self._stale_or_changed():
self._reload_locked()
# ── Endpoints 전용 API ─────────────────────────────────────────
def get_endpoint(self, key: str) -> Dict[str, str]:
if not key or not str(key).strip():
raise KeyError("endpoint key is empty")
self._ensure_loaded()
hit = self._endpoint_index.get(_norm(key))
if not hit:
# 갱신 지연 가능성 → 강제 리로드 1회 재시도
with self._lock:
self._reload_locked()
hit = self._endpoint_index.get(_norm(key))
if not hit:
raise KeyError(f"endpoint not found: {key}")
return hit
def list_endpoints(self) -> Dict[str, Dict[str, str]]:
self._ensure_loaded()
out: Dict[str, Dict[str, str]] = {}
for _, v in self._endpoint_index.items():
dn = v["display_name"]
out[dn] = {
"display_name": v["display_name"],
"request_name": v["request_name"],
"base_url": v["base_url"],
# api_key는 보안상 미노출
}
return out
# ── 범용 읽기 ──────────────────────────────────────────────────
def get_raw(self) -> Dict[str, Any]:
self._ensure_loaded()
return self._raw
# 전역 싱글톤
_CONFIG = ReadOnlyConfig()
# ===== ComfyUI Custom Nodes =====
class ConfigEndpointLookup:
"""
입력: key — "text-model-a" / "Text Model A" / "gpt-4o-mini"(request_name)
출력: display_name, request_name, base_url, api_key (모두 STRING)
"""
@classmethod
def INPUT_TYPES(cls):
return {"required": {"key": ("STRING", {"default": "", "multiline": False})}}
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING")
RETURN_NAMES = ("display_name", "request_name", "base_url", "api_key")
FUNCTION = "lookup"
CATEGORY = "Config/Endpoints (ReadOnly)"
def lookup(self, key: str):
try:
ep = _CONFIG.get_endpoint(key)
except Exception as e:
raise RuntimeError(f"[ConfigEndpointLookup] {e}")
return (ep["display_name"], ep["request_name"], ep["base_url"], ep["api_key"])
class ConfigListEndpoints:
"""디버그/인스펙션용: 엔드포인트 목록(JSON, api_key 미포함)."""
@classmethod
def INPUT_TYPES(cls):
return {"required": {}}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("endpoints_json",)
FUNCTION = "list_endpoints"
CATEGORY = "Config/Endpoints (ReadOnly)"
def list_endpoints(self):
import json
data = _CONFIG.list_endpoints()
return (json.dumps(data, ensure_ascii=False, indent=2),)
class ConfigGetValue:
"""
임의 경로 값 읽기 ('.' 구분)
예) path="feature_flags.use_cache" → "true"
path="storage.s3.region" → "ap-northeast-2"
path="endpoints.text-model-a" → 해당 엔드포인트 JSON
"""
@classmethod
def INPUT_TYPES(cls):
return {"required": {"path": ("STRING", {"default": "", "multiline": False})}}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("value",)
FUNCTION = "get_value"
CATEGORY = "Config/General (ReadOnly)"
def get_value(self, path: str):
import json
def get_in(d: Any, dotted: str):
cur = d
for part in filter(None, dotted.split(".")):
if isinstance(cur, dict) and part in cur:
cur = cur[part]
else:
raise KeyError(f"config path not found: {dotted}")
return cur
try:
raw = _CONFIG.get_raw()
val = get_in(raw, path) if path else raw
if isinstance(val, (str, int, float, bool)) or val is None:
return (str(val),)
return (json.dumps(val, ensure_ascii=False, indent=2),)
except Exception as e:
raise RuntimeError(f"[ConfigGetValue] {e}")
NODE_CLASS_MAPPINGS = {
"ConfigEndpointLookup": ConfigEndpointLookup,
"ConfigListEndpoints": ConfigListEndpoints,
"ConfigGetValue": ConfigGetValue,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"ConfigEndpointLookup": "Config: Endpoint Lookup",
"ConfigListEndpoints": "Config: List Endpoints",
"ConfigGetValue": "Config: Get Value",
}
__init__.py
# 패키지 인식용 (비워두면 됩니다)
4) 요구사항 대응 체크
-
사람이 편집하기 쉬움: TOML + 주석 가능
-
읽기 전용: 파일 쓰기 없음
-
주요 용도(모델/서비스 엔드포인트):
endpoints섹션 표준화 -
필요한 값 반환:
ConfigEndpointLookup가 display_name, request_name, base_url, api_key 출력 -
변경 반영: mtime + TTL(기본 30초, 5~60초 조정) → 최대 1분 내 반영 보장
-
보안: 환경변수 치환 , 목록 노드에서 api_key 미노출
5) 마이그레이션 가이드
-
예전 파일/이름을 쓰셨다면:
-
폴더:
custom_nodes/comfyui_llm_config_reader→ 삭제 또는 보관 -
설정:
[llm_models]→[endpoints]로 바꾸고request_model→request_name -
노드: 기존 워크플로우에선 새 노드로 교체
-
LLMConfigLookup→ConfigEndpointLookup -
LLMConfigListModels→ConfigListEndpoints -
ConfigGetRawJSON→ConfigGetValue
-
-
필요하면, 기존 워크플로우 자동 변환 스크립트(YAML/JSON 경로 교체)도 만들어 드릴게요.
다른 노드 “파일들”에서 설정을 읽어야하는데 이건 어떻게 해? 싱글톤 인터페이스가 좀 어색해보여
좋은 피드백! “여러 노드/파일에서 자연스럽게 읽는” 패턴으로 바꿔줄게요.
싱글톤 대신 경로별 Multiton 레지스트리 + 얇은 함수형 API 를 제공합니다.
즉, 어디서든 from comfyui_config_reader.config import endpoint, get만 임포트해 쓰면 됩니다.
아래 세 파일만 추가/교체하면 끝!
1) 구조
ComfyUI/
└─ custom_nodes/
└─ comfyui_config_reader/
├─ __init__.py
├─ config_core.py # 핵심: ReadOnlyStore (경로별 캐시/자동 리로드)
├─ config.py # 공용 API: endpoint()/get()/get_store()
└─ config_nodes.py # (옵션) 데모 노드들: Lookup/List/Get
2) 핵심 스토어 (경로별 Multiton 아님 — “인스턴스형”, thread-safe)
config_core.py
import os
import time
import threading
from typing import Dict, Any, Optional
# Python 3.11+: tomllib / 이하: tomli
try:
import tomllib # type: ignore
_TOML_LOADS = tomllib.loads
except Exception:
import tomli # type: ignore
_TOML_LOADS = tomli.loads
def _expand_env_vars(value: Any) -> Any:
if isinstance(value, str):
out, i = "", 0
while i < len(value):
if value[i:i+2] == "${":
j = value.find("}", i+2)
if j != -1:
env_name = value[i+2:j]
out += os.environ.get(env_name, "")
i = j + 1
else:
out += value[i]; i += 1
else:
out += value[i]; i += 1
return out
if isinstance(value, dict):
return {k: _expand_env_vars(v) for k, v in value.items()}
if isinstance(value, list):
return [_expand_env_vars(v) for v in value]
return value
def _norm(s: str) -> str:
return str(s).strip().lower()
class ReadOnlyStore:
"""
- TOML 읽기 전용 스토어
- mtime/TTL(5~60s) 기반 자동 재로드 → '1분 내 반영' 보장
- 엔드포인트 인덱싱 + 임의 경로 접근
- 파일 쓰기 없음
"""
def __init__(self, path: str):
self.path = path
self._lock = threading.Lock()
self._loaded_at: float = 0.0
self._last_mtime: float = -1.0
self._refresh_interval: int = 30 # 기본 30s
self._raw: Dict[str, Any] = {}
self._endpoint_index: Dict[str, Dict[str, str]] = {}
self._version: int = 0 # 리로드마다 증가
# ---------- 내부 로딩 ----------
def _read_text(self) -> str:
with open(self.path, "rb") as f:
return f.read().decode("utf-8")
def _build_index_endpoints(self, data: Dict[str, Any]) -> None:
self._endpoint_index.clear()
eps = (data.get("endpoints") or {})
if not isinstance(eps, dict):
return
for ep_key, cfg in eps.items():
if not isinstance(cfg, dict):
continue
cfg = _expand_env_vars(cfg)
required = ["display_name", "request_name", "base_url", "api_key"]
if any(k not in cfg or not str(cfg[k]).strip() for k in required):
continue
for c in {str(ep_key), str(cfg.get("display_name","")), str(cfg.get("request_name",""))}:
if c:
self._endpoint_index[_norm(c)] = {
"display_name": str(cfg["display_name"]),
"request_name": str(cfg["request_name"]),
"base_url": str(cfg["base_url"]),
"api_key": str(cfg["api_key"]),
}
def _reload_locked(self) -> None:
txt = self._read_text()
data = _TOML_LOADS(txt)
# refresh_interval_seconds (5~60) + env override
conf_val = int((data.get("settings") or {}).get("refresh_interval_seconds", 30))
conf_val = max(5, min(60, conf_val))
env_override = os.environ.get("COMFY_CONFIG_REFRESH_SECONDS")
if env_override:
try:
conf_val = max(5, min(60, int(env_override)))
except ValueError:
pass
self._refresh_interval = conf_val
self._raw = _expand_env_vars(data)
self._build_index_endpoints(self._raw)
self._loaded_at = time.time()
try:
self._last_mtime = os.path.getmtime(self.path)
except FileNotFoundError:
self._last_mtime = -1.0
self._version += 1
def _stale_or_changed(self) -> bool:
try:
mtime = os.path.getmtime(self.path)
except FileNotFoundError:
return True
now = time.time()
return (mtime != self._last_mtime) or (now - self._loaded_at >= self._refresh_interval)
def _ensure_loaded(self) -> None:
with self._lock:
if self._loaded_at == 0 or self._stale_or_changed():
self._reload_locked()
# ---------- 퍼블릭 API ----------
@property
def version(self) -> int:
self._ensure_loaded()
return self._version
def snapshot(self) -> Dict[str, Any]:
"""ENV 치환 후 전체 트리의 '일관된 스냅샷'을 반환(얕은 복사)."""
self._ensure_loaded()
return dict(self._raw)
def get_value(self, dotted_path: str) -> Any:
self._ensure_loaded()
cur = self._raw
for part in filter(None, dotted_path.split(".")):
if isinstance(cur, dict) and part in cur:
cur = cur[part]
else:
raise KeyError(f"config path not found: {dotted_path}")
return cur
def get_endpoint(self, key: str) -> Dict[str, str]:
if not key or not str(key).strip():
raise KeyError("endpoint key is empty")
self._ensure_loaded()
hit = self._endpoint_index.get(_norm(key))
if not hit:
# 갱신 지연 가능성 → 강제 리로드 1회 재시도
with self._lock:
self._reload_locked()
hit = self._endpoint_index.get(_norm(key))
if not hit:
raise KeyError(f"endpoint not found: {key}")
return hit
def list_endpoints(self) -> Dict[str, Dict[str, str]]:
self._ensure_loaded()
out: Dict[str, Dict[str, str]] = {}
for _, v in self._endpoint_index.items():
dn = v["display_name"]
out[dn] = {
"display_name": v["display_name"],
"request_name": v["request_name"],
"base_url": v["base_url"],
# api_key는 미노출
}
return out
3) 공용 API (어디서든 임포트해 쓰는 “함수형” 인터페이스)
config.py
import os
from typing import Optional, Dict, Any
from .config_core import ReadOnlyStore
# 경로별 Multiton 레지스트리
_REGISTRY: Dict[str, ReadOnlyStore] = {}
def _default_path() -> str:
here = os.path.dirname(__file__)
return os.environ.get("COMFY_CONFIG_PATH", os.path.join(here, "config.toml"))
def get_store(path: Optional[str] = None) -> ReadOnlyStore:
"""
- 경로별로 ReadOnlyStore 인스턴스를 재사용(멀티톤)
- 명시 path 없으면 COMFY_CONFIG_PATH 또는 패키지 기본 config.toml
"""
p = os.path.abspath(path or _default_path())
if p not in _REGISTRY:
_REGISTRY[p] = ReadOnlyStore(p)
return _REGISTRY[p]
# ---- 얇은 함수형 헬퍼들 (권장) ----
def endpoint(key: str, *, path: Optional[str] = None) -> Dict[str, str]:
"""엔드포인트 dict 반환: display_name, request_name, base_url, api_key"""
return get_store(path).get_endpoint(key)
def endpoints(*, path: Optional[str] = None) -> Dict[str, Dict[str, str]]:
"""엔드포인트 목록(JSON용)"""
return get_store(path).list_endpoints()
def get(dotted_path: str, *, path: Optional[str] = None) -> Any:
"""임의 설정값 반환: 'section.sub.key'"""
return get_store(path).get_value(dotted_path)
def snapshot(*, path: Optional[str] = None) -> Dict[str, Any]:
"""전체 트리 스냅샷"""
return get_store(path).snapshot()
포인트: 다른 노드 파일 에서는
config.endpoint(),config.get()같은 함수형 API만 쓰면 됩니다.
내부적으로는 경로별 인스턴스를 재사용하지만, 노드 코드에서는 싱글톤에 의존하지 않습니다.
4) (옵션) 데모 노드 — 기존 노드를 이 API로 얇게 래핑
config_nodes.py
from .config import endpoint as _endpoint, endpoints as _endpoints, get as _get
import json
class ConfigEndpointLookup:
@classmethod
def INPUT_TYPES(cls):
return {"required": {"key": ("STRING", {"default": "", "multiline": False}),
"config_path": ("STRING", {"default": "", "multiline": False})}}
RETURN_TYPES = ("STRING","STRING","STRING","STRING")
RETURN_NAMES = ("display_name","request_name","base_url","api_key")
FUNCTION = "lookup"
CATEGORY = "Config/Endpoints (ReadOnly)"
def lookup(self, key: str, config_path: str):
p = config_path or None
ep = _endpoint(key, path=p)
return (ep["display_name"], ep["request_name"], ep["base_url"], ep["api_key"])
class ConfigListEndpoints:
@classmethod
def INPUT_TYPES(cls):
return {"required": {"config_path": ("STRING", {"default": "", "multiline": False})}}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("endpoints_json",)
FUNCTION = "list_endpoints"
CATEGORY = "Config/Endpoints (ReadOnly)"
def list_endpoints(self, config_path: str):
p = config_path or None
data = _endpoints(path=p)
return (json.dumps(data, ensure_ascii=False, indent=2),)
class ConfigGetValue:
@classmethod
def INPUT_TYPES(cls):
return {"required": {"path": ("STRING", {"default": "", "multiline": False}),
"config_path": ("STRING", {"default": "", "multiline": False})}}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("value",)
FUNCTION = "get_value"
CATEGORY = "Config/General (ReadOnly)"
def get_value(self, path: str, config_path: str):
p = config_path or None
val = _get(path, path=p)
if isinstance(val, (str, int, float, bool)) or val is None:
return (str(val),)
return (json.dumps(val, ensure_ascii=False, indent=2),)
NODE_CLASS_MAPPINGS = {
"ConfigEndpointLookup": ConfigEndpointLookup,
"ConfigListEndpoints": ConfigListEndpoints,
"ConfigGetValue": ConfigGetValue,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"ConfigEndpointLookup": "Config: Endpoint Lookup",
"ConfigListEndpoints": "Config: List Endpoints",
"ConfigGetValue": "Config: Get Value",
}
5) 다른 노드 “파일들”에서 쓰는 법 (가장 깔끔한 패턴)
다른 플러그인/노드 파일 어디에서든:
# 예: OpenAI 호출 노드 파일 내부
from comfyui_config_reader.config import endpoint, get
def _prepare_client():
ep = endpoint("text-model-a") # 또는 display_name / request_name로 조회
api_key = ep["api_key"]
base_url = ep["base_url"]
# ... 클라이언트 생성
return api_key, base_url
def _s3_settings():
region = get("storage.s3.region") # 범용 설정 접근
bucket = get("storage.s3.bucket")
return region, bucket
-
기본 경로는
COMFY_CONFIG_PATH또는comfyui_config_reader/config.toml. -
특정 파일을 쓰고 싶으면
endpoint("text-model-a", path="/abs/path/to/config.toml")처럼 함수 인자 로 주면 됩니다. -
내부는 경로별 레지스트리(Multiton)라 같은 경로는 인스턴스를 공유(성능↑)하지만, 호출부는 함수형 이라 싱글톤에 종속되지 않아요.
6) 요건 체크
-
여러 노드/파일에서 공통 사용: 함수형 API 로 간단 임포트
-
싱글톤 어색함 제거: 경로별 Multiton + helper 함수 → 명시적 DI도 가능(
get_store(path)) -
읽기 전용 + 1분 내 반영: mtime + 5~60s TTL(기본 30s)
-
사람이 편집 쉬움: TOML + ENV 치환
-
엔드포인트 4필드:
display_name, request_name, base_url, api_key
필요하면 get_store(path)로 명시적 의존성 주입(DI)도 가능해요. 예컨대 프레임워크 초기화 시:
from comfyui_config_reader.config import get_store
STORE = get_store("/path/to/config.toml")
# 이후 어디서든:
ep = STORE.get_endpoint("text-model-a")
이렇게 하면 아예 전역 함수도 안 쓰고, “내가 주입한 스토어”만 참조하는 형태로 완전히 분리할 수 있습니다.
아니 싱글톤은 괜찮은데 외부에서 접근이 쉽지 않아보여서. 싱글톤이 낫지 않아?
좋아, 싱글톤으로 단순·직관 접근 가능하게 정리해줄게.
외부 노드 “파일들” 어디서든 한 줄 임포트로 바로 쓰는 형태야:
-
함수형:
from comfyui_config_reader import endpoint, get -
객체형:
from comfyui_config_reader import CONFIG
둘 다 같은 싱글톤 인스턴스를 뒤에서 사용해. (thread-safe, read-only, 1분 내 반영)
1) 파일 구조 (심플)
ComfyUI/
└─ custom_nodes/
└─ comfyui_config_reader/
├─ __init__.py
├─ config_singleton.py # 싱글톤 본체
└─ config.toml
2) config.toml 예시 (범용)
version = 1
[settings]
refresh_interval_seconds = 30 # 5~60 사이 권장
[endpoints."text-model-a"]
display_name = "Text Model A"
request_name = "gpt-4o-mini"
base_url = "https://api.openai.com/v1"
api_key = "${OPENAI_API_KEY}"
[feature_flags]
use_cache = true
${ENV}는 환경변수 치환됨.
3) 싱글톤 구현: config_singleton.py
import os
import time
import threading
from typing import Dict, Any, Optional
# Python 3.11+: tomllib / 이하: tomli
try:
import tomllib # type: ignore
_TOML_LOADS = tomllib.loads
except Exception:
import tomli # type: ignore
_TOML_LOADS = tomli.loads
def _expand_env(v: Any) -> Any:
if isinstance(v, str):
out, i = "", 0
while i < len(v):
if v[i:i+2] == "${":
j = v.find("}", i+2)
if j != -1:
out += os.environ.get(v[i+2:j], "")
i = j + 1
else:
out += v[i]; i += 1
else:
out += v[i]; i += 1
return out
if isinstance(v, dict):
return {k: _expand_env(x) for k, x in v.items()}
if isinstance(v, list):
return [_expand_env(x) for x in v]
return v
def _norm(s: str) -> str:
return str(s).strip().lower()
class _ReadOnlyConfig:
"""
- TOML 파일 읽기 전용
- mtime 변경 + TTL(5~60s, 기본 30s) 자동 재로드
- endpoints 4필드 인덱싱(display_name, request_name, base_url, api_key)
- 임의 경로 접근 가능 ('.' 구분)
"""
def __init__(self, path: str):
self.path = path
self._lock = threading.Lock()
self._loaded_at = 0.0
self._last_mtime = -1.0
self._refresh = 30
self._raw: Dict[str, Any] = {}
self._ep_index: Dict[str, Dict[str, str]] = {}
# -------- 내부 로딩 --------
def _read_text(self) -> str:
with open(self.path, "rb") as f:
return f.read().decode("utf-8")
def _build_index(self, data: Dict[str, Any]) -> None:
self._ep_index.clear()
eps = (data.get("endpoints") or {})
if not isinstance(eps, dict): return
for key, cfg in eps.items():
if not isinstance(cfg, dict): continue
cfg = _expand_env(cfg)
req = ["display_name", "request_name", "base_url", "api_key"]
if any(k not in cfg or not str(cfg[k]).strip() for k in req):
continue
for c in {str(key), str(cfg.get("display_name","")), str(cfg.get("request_name",""))}:
if c:
self._ep_index[_norm(c)] = {
"display_name": str(cfg["display_name"]),
"request_name": str(cfg["request_name"]),
"base_url": str(cfg["base_url"]),
"api_key": str(cfg["api_key"]),
}
def _reload_locked(self) -> None:
txt = self._read_text()
data = _TOML_LOADS(txt)
# refresh_interval_seconds (5~60) + env override
r = int((data.get("settings") or {}).get("refresh_interval_seconds", 30))
r = max(5, min(60, r))
env_r = os.environ.get("COMFY_CONFIG_REFRESH_SECONDS")
if env_r:
try: r = max(5, min(60, int(env_r)))
except ValueError: pass
self._refresh = r
self._raw = _expand_env(data)
self._build_index(self._raw)
self._loaded_at = time.time()
try:
self._last_mtime = os.path.getmtime(self.path)
except FileNotFoundError:
self._last_mtime = -1.0
def _stale_or_changed(self) -> bool:
try:
m = os.path.getmtime(self.path)
except FileNotFoundError:
return True
now = time.time()
return (m != self._last_mtime) or (now - self._loaded_at >= self._refresh)
def _ensure_loaded(self) -> None:
with self._lock:
if self._loaded_at == 0 or self._stale_or_changed():
self._reload_locked()
# -------- 퍼블릭 API --------
def endpoint(self, key: str) -> Dict[str, str]:
if not key or not str(key).strip():
raise KeyError("endpoint key is empty")
self._ensure_loaded()
hit = self._ep_index.get(_norm(key))
if not hit:
# 즉시 재시도(갱신 지연 보정)
with self._lock:
self._reload_locked()
hit = self._ep_index.get(_norm(key))
if not hit:
raise KeyError(f"endpoint not found: {key}")
return hit
def endpoints(self) -> Dict[str, Dict[str, str]]:
self._ensure_loaded()
out: Dict[str, Dict[str, str]] = {}
for _, v in self._ep_index.items():
out[v["display_name"]] = {
"display_name": v["display_name"],
"request_name": v["request_name"],
"base_url": v["base_url"],
}
return out # api_key 미노출
def get(self, dotted: str) -> Any:
self._ensure_loaded()
cur = self._raw
for part in filter(None, dotted.split(".")):
if isinstance(cur, dict) and part in cur:
cur = cur[part]
else:
raise KeyError(f"config path not found: {dotted}")
return cur
def snapshot(self) -> Dict[str, Any]:
self._ensure_loaded()
return dict(self._raw)
# ===== 모듈 전역 싱글톤 =====
_lock = threading.Lock()
_instance: Optional[_ReadOnlyConfig] = None
_default_path = None # configure()로 지정 가능
def _resolve_default_path() -> str:
here = os.path.dirname(__file__)
return _default_path or os.environ.get("COMFY_CONFIG_PATH", os.path.join(here, "config.toml"))
def _get_instance() -> _ReadOnlyConfig:
global _instance
if _instance is None:
with _lock:
if _instance is None:
_instance = _ReadOnlyConfig(_resolve_default_path())
return _instance
def configure(*, path: Optional[str] = None, refresh_seconds: Optional[int] = None) -> None:
"""
싱글톤 설정(선택): 경로/TTL 오버라이드.
- 가능한 한 '초기화 시 1회'만 호출 추천.
- 이후에도 호출 가능하며, 새 인스턴스로 교체됨( thread-safe ).
"""
global _instance, _default_path
if path:
_default_path = os.path.abspath(path)
if refresh_seconds is not None:
os.environ["COMFY_CONFIG_REFRESH_SECONDS"] = str(refresh_seconds)
with _lock:
_instance = _ReadOnlyConfig(_resolve_default_path())
# ---- 외부에 노출할 얇은 API (가장 쉬운 접근) ----
def endpoint(key: str) -> Dict[str, str]:
return _get_instance().endpoint(key)
def endpoints() -> Dict[str, Dict[str, str]]:
return _get_instance().endpoints()
def get(dotted: str) -> Any:
return _get_instance().get(dotted)
def snapshot() -> Dict[str, Any]:
return _get_instance().snapshot()
# ---- 객체형 접근을 위한 프록시(CONFIG) ----
class _ConfigProxy:
def __getattr__(self, name: str):
# CONFIG.endpoint(...), CONFIG.get(...) 형태 지원
return getattr(_get_instance(), name)
CONFIG = _ConfigProxy()
4) __init__.py (한 줄 임포트 지원)
# 외부에서 아주 간단히 접근 가능하게 re-export
from .config_singleton import (
CONFIG, # 객체형: CONFIG.endpoint(...), CONFIG.get(...)
configure, # 선택: 초기화 시 경로/TTL 지정
endpoint, # 함수형: endpoint("text-model-a")
endpoints,
get,
snapshot,
)
5) 외부 노드 “파일들”에서 쓰는 법
가장 간단 (함수형)
from comfyui_config_reader import endpoint, get
def build_client():
ep = endpoint("text-model-a") # 키 / display_name / request_name 중 아무거나 OK
base_url = ep["base_url"]
api_key = ep["api_key"]
# ... 클라이언트 생성
객체형이 익숙하다면
from comfyui_config_reader import CONFIG
def build_client():
ep = CONFIG.endpoint("text-model-a")
# CONFIG.get("storage.s3.region") 등도 동일
시작 시 경로/TTL 지정(선택)
# 예: __init__ 혹은 ComfyUI 부팅 초기에 한 번
from comfyui_config_reader import configure
configure(path="/abs/path/to/config.toml", refresh_seconds=20) # 5~60 사이
- 이후 모든 모듈에서 같은 싱글톤을 사용하고, 파일 변경은 최대 1분 내 반영돼요(기본 30초, env/설정으로 조절).
6) 요건 재확인
-
✅ 외부 접근 쉬움:
from comfyui_config_reader import endpoint또는CONFIG.endpoint -
✅ 싱글톤: 모듈 전역 한 개 인스턴스, thread-safe lazy init
-
✅ 읽기 전용: 쓰기 없음
-
✅ 1분 내 반영: mtime + TTL(기본 30s, 5~60s)
-
✅ 사람이 편집 쉬움: TOML +
${ENV}치환 -
✅ 엔드포인트 4필드:
display_name,request_name,base_url,api_key
필요하면, 지금 구조 위에 “엔드포인트 호출용 노드”도 바로 얹어줄게.
endpoint 말고 좀 더 좋은 용어 선정해줘. 그리고 model configuration (parameter 등도 설정에 같이 포함시키면 좋겠어)
좋아! “endpoint” 대신 더 자연스러운 용어로 profile(프로필) 을 쓰고, 여기에 model configuration(파라미터 등) 까지 함께 담는 방향으로 정리했어.
외부 노드/파일에서는 싱글톤 을 아주 간단히 쓰면 되고, 기존 endpoint 명칭은 하위 호환 별칭 도 제공해둘게.
1) 용어 선정 (step-by-step)
-
profile (프로필) — 사람이 이해하기 쉬움. “연결 정보 + 호출 규칙 + 모델 파라미터 묶음”을 한 세트로 보기에 적합.
-
대안(참고): provider, backend, connector, target.
- 기술적으로는 backend 도 괜찮지만, “설정 묶음” 뉘앙스는 profile 이 가장 직관적.
-
결론: profiles 섹션으로 표준화하고, endpoints 는 자동 인식(하위 호환) 유지.
2) config.toml 스키마 (모델 파라미터 포함)
version = 1
[settings]
# 파일 변경 감지 주기(초). 5~60 권장. "1분 내 반영" 요건 만족.
refresh_interval_seconds = 30
# 모든 프로필에 공통 적용될 기본값(선택)
[defaults.params]
temperature = 0.3
max_tokens = 2048
[defaults.http]
timeout_seconds = 30
[defaults.http.headers]
User-Agent = "ComfyUI-ConfigReader/1.0"
[defaults.http.query]
# 공통 쿼리 파라미터가 필요하면 여기에
# ── 프로필들 ─────────────────────────────────────────────
[profiles."text-model-a"]
display_name = "Text Model A"
request_name = "gpt-4o-mini"
base_url = "https://api.openai.com/v1"
api_key = "${OPENAI_API_KEY}"
# 이 프로필의 모델 파라미터(기본값 위에 덮어쓰기)
[profiles."text-model-a".params]
temperature = 0.2
top_p = 0.9
stop = ["\n\nHuman:", "\n\nAssistant:"]
# HTTP 관련(선택): 헤더, 타임아웃, 쿼리 등
[profiles."text-model-a".http]
timeout_seconds = 20
[profiles."text-model-a".http.headers]
X-Org-Id = "${ORG_ID}"
[profiles."text-model-a".http.query]
apiVersion = "2024-06-01"
[profiles."text-model-b"]
display_name = "Text Model B"
request_name = "claude-3-5-sonnet"
base_url = "https://api.anthropic.com"
api_key = "${ANTHROPIC_API_KEY}"
# params/http 섹션이 없으면 defaults만 적용
-
${ENV}는 환경변수 치환됨. -
params : temperature/top_p/max_tokens/stop/seed 등 원하는 파라미터 자유 확장.
-
http : timeout_seconds / headers / query 등 요청 관련 옵션.
-
상단의 [defaults] 는 모든 프로필에 병합(merge) 되어 기본값 역할.
3) 싱글톤 API (drop-in): profile + params + http
외부 어디서든 한 줄로 접근:
from comfyui_config_reader import profile, params, http, get
# 또는 객체형:
from comfyui_config_reader import CONFIG
-
profile(key)→ 연결정보 + 모델파라미터 + http 설정까지 한 번에 -
params(key)→ 해당 프로필의 모델 파라미터 dict -
http(key)→ 해당 프로필의 HTTP 옵션 dict -
get("storage.s3.region")처럼 범용 키 접근도 유지 -
하위 호환:
endpoint,endpoints별칭도 남겨둠(기존 코드 안깨짐)
4) 교체본 코드 (config_singleton.py)
아래 파일 하나로 기존 싱글톤 구현을 대체하면 돼. (기능: profiles + defaults 병합 + params/http 지원 + legacy endpoints 인식)
import os
import time
import threading
from typing import Dict, Any, Optional
# Python 3.11+: tomllib / 이하: tomli
try:
import tomllib # type: ignore
_TOML_LOADS = tomllib.loads
except Exception:
import tomli # type: ignore
_TOML_LOADS = tomli.loads
def _expand_env(v: Any) -> Any:
if isinstance(v, str):
out, i = "", 0
while i < len(v):
if v[i:i+2] == "${":
j = v.find("}", i+2)
if j != -1:
out += os.environ.get(v[i+2:j], "")
i = j + 1
else:
out += v[i]; i += 1
else:
out += v[i]; i += 1
return out
if isinstance(v, dict):
return {k: _expand_env(x) for k, x in v.items()}
if isinstance(v, list):
return [_expand_env(x) for x in v]
return v
def _norm(s: str) -> str:
return str(s).strip().lower()
def _deep_merge(base: Dict[str, Any], over: Dict[str, Any]) -> Dict[str, Any]:
"""얕은 dict update가 아닌, 중첩 dict까지 병합."""
out = dict(base)
for k, v in (over or {}).items():
if isinstance(v, dict) and isinstance(out.get(k), dict):
out[k] = _deep_merge(out[k], v)
else:
out[k] = v
return out
class _ReadOnlyConfig:
"""
- 읽기 전용 TOML
- mtime + TTL(5~60s, 기본 30s) 자동 재로드 → 1분 내 변경 반영
- 'profiles' 표준 + 'endpoints' 레거시도 자동 인식
- 각 profile에:
* display_name, request_name, base_url, api_key (필수)
* params(dict) : 모델 파라미터
* http(dict) : timeout_seconds, headers{}, query{}
"""
def __init__(self, path: str):
self.path = path
self._lock = threading.Lock()
self._loaded_at = 0.0
self._last_mtime = -1.0
self._refresh = 30
self._raw: Dict[str, Any] = {}
self._idx: Dict[str, Dict[str, Any]] = {} # 정규화 키 -> profile dict
# -------- 내부 로딩 --------
def _read_text(self) -> str:
with open(self.path, "rb") as f:
return f.read().decode("utf-8")
def _extract_profiles_table(self, data: Dict[str, Any]) -> Dict[str, Any]:
# 표준 'profiles' 우선, 없으면 'endpoints'(하위 호환)
if isinstance(data.get("profiles"), dict):
return data["profiles"]
if isinstance(data.get("endpoints"), dict):
return data["endpoints"]
return {}
def _build_index(self, data: Dict[str, Any]) -> None:
self._idx.clear()
# defaults (ENV 확장 포함)
defaults = _expand_env(data.get("defaults") or {})
defaults_params = defaults.get("params") or {}
defaults_http = defaults.get("http") or {}
if "headers" not in defaults_http: defaults_http["headers"] = {}
if "query" not in defaults_http: defaults_http["query"] = {}
profiles_tbl = self._extract_profiles_table(data)
if not isinstance(profiles_tbl, dict):
return
for key, raw_cfg in profiles_tbl.items():
if not isinstance(raw_cfg, dict):
continue
cfg = _expand_env(raw_cfg) # ENV 치환
# 필수 필드
req = ["display_name", "request_name", "base_url", "api_key"]
if any(k not in cfg or not str(cfg[k]).strip() for k in req):
continue
# params/http 병합 (defaults → profile override)
p_params = cfg.get("params") or {}
p_http = cfg.get("http") or {}
if "headers" not in p_http: p_http["headers"] = {}
if "query" not in p_http: p_http["query"] = {}
merged_params = {**defaults_params, **p_params}
merged_http = _deep_merge(defaults_http, p_http)
entry = {
"display_name": str(cfg["display_name"]),
"request_name": str(cfg["request_name"]),
"base_url": str(cfg["base_url"]),
"api_key": str(cfg["api_key"]),
"params": merged_params,
"http": merged_http,
}
# 접근 키: 정의 키 / display_name / request_name
for c in {str(key), str(cfg.get("display_name","")), str(cfg.get("request_name",""))}:
if c:
self._idx[_norm(c)] = entry
def _reload_locked(self) -> None:
txt = self._read_text()
data = _TOML_LOADS(txt)
# refresh_interval_seconds (5~60) + env override
r = int((data.get("settings") or {}).get("refresh_interval_seconds", 30))
r = max(5, min(60, r))
env_r = os.environ.get("COMFY_CONFIG_REFRESH_SECONDS")
if env_r:
try: r = max(5, min(60, int(env_r)))
except ValueError: pass
self._refresh = r
self._raw = _expand_env(data)
self._build_index(self._raw)
self._loaded_at = time.time()
try:
self._last_mtime = os.path.getmtime(self.path)
except FileNotFoundError:
self._last_mtime = -1.0
def _stale_or_changed(self) -> bool:
try:
m = os.path.getmtime(self.path)
except FileNotFoundError:
return True
now = time.time()
return (m != self._last_mtime) or (now - self._loaded_at >= self._refresh)
def _ensure_loaded(self) -> None:
with self._lock:
if self._loaded_at == 0 or self._stale_or_changed():
self._reload_locked()
# -------- 퍼블릭 API --------
def profile(self, key: str) -> Dict[str, Any]:
if not key or not str(key).strip():
raise KeyError("profile key is empty")
self._ensure_loaded()
hit = self._idx.get(_norm(key))
if not hit:
# 즉시 재시도(갱신 지연 보정)
with self._lock:
self._reload_locked()
hit = self._idx.get(_norm(key))
if not hit:
raise KeyError(f"profile not found: {key}")
return hit
def profiles(self) -> Dict[str, Dict[str, Any]]:
"""api_key는 숨기고, 나머지 메타/params/http는 노출."""
self._ensure_loaded()
out: Dict[str, Dict[str, Any]] = {}
seen = set()
for _, v in self._idx.items():
dn = v["display_name"]
if dn in seen: # display_name 중복 키 방지
continue
seen.add(dn)
out[dn] = {
"display_name": v["display_name"],
"request_name": v["request_name"],
"base_url": v["base_url"],
"params": v.get("params", {}),
"http": v.get("http", {}),
}
return out
def params(self, key: str) -> Dict[str, Any]:
return dict(self.profile(key).get("params", {}))
def http(self, key: str) -> Dict[str, Any]:
return dict(self.profile(key).get("http", {}))
def get(self, dotted: str) -> Any:
self._ensure_loaded()
cur = self._raw
for part in filter(None, dotted.split(".")):
if isinstance(cur, dict) and part in cur:
cur = cur[part]
else:
raise KeyError(f"config path not found: {dotted}")
return cur
def snapshot(self) -> Dict[str, Any]:
self._ensure_loaded()
return dict(self._raw)
# ===== 모듈 전역 싱글톤 =====
_lock = threading.Lock()
_instance: Optional[_ReadOnlyConfig] = None
_default_path = None # configure()로 지정 가능
def _resolve_default_path() -> str:
here = os.path.dirname(__file__)
return _default_path or os.environ.get("COMFY_CONFIG_PATH", os.path.join(here, "config.toml"))
def _get_instance() -> _ReadOnlyConfig:
global _instance
if _instance is None:
with _lock:
if _instance is None:
_instance = _ReadOnlyConfig(_resolve_default_path())
return _instance
def configure(*, path: Optional[str] = None, refresh_seconds: Optional[int] = None) -> None:
"""선택: 경로/TTL 오버라이드(초기화 시 1회 권장)."""
global _instance, _default_path
if path:
_default_path = os.path.abspath(path)
if refresh_seconds is not None:
os.environ["COMFY_CONFIG_REFRESH_SECONDS"] = str(refresh_seconds)
with _lock:
_instance = _ReadOnlyConfig(_resolve_default_path())
# ---- 외부에 노출할 간단 API ----
def profile(key: str) -> Dict[str, Any]:
return _get_instance().profile(key)
def profiles() -> Dict[str, Dict[str, Any]]:
return _get_instance().profiles()
def params(key: str) -> Dict[str, Any]:
return _get_instance().params(key)
def http(key: str) -> Dict[str, Any]:
return _get_instance().http(key)
def get(dotted: str) -> Any:
return _get_instance().get(dotted)
def snapshot() -> Dict[str, Any]:
return _get_instance().snapshot()
# ---- 하위 호환 별칭 (endpoint*) ----
def endpoint(key: str) -> Dict[str, Any]:
return profile(key)
def endpoints() -> Dict[str, Dict[str, Any]]:
return profiles()
# ---- 객체형 접근 프록시(CONFIG) ----
class _ConfigProxy:
def __getattr__(self, name: str):
return getattr(_get_instance(), name)
CONFIG = _ConfigProxy()
__init__.py (re-export)
from .config_singleton import (
CONFIG,
configure,
profile, profiles,
params, http,
get, snapshot,
# legacy alias
endpoint, endpoints,
)
5) 외부에서 쓰는 법 (예시)
from comfyui_config_reader import profile, params, http
# (1) 한 번에 전부 받기
cfg = profile("Text Model A") # 또는 "text-model-a" / "gpt-4o-mini"
api_key = cfg["api_key"]
base_url = cfg["base_url"]
req_name = cfg["request_name"]
p = cfg["params"] # dict (temperature/top_p/...)
h = cfg["http"] # dict (timeout_seconds/headers/query)
# (2) 필요만 따로
p = params("text-model-a")
h = http("text-model-a")
6) 포인트 정리
-
“endpoint” → “profile” 로 명확화 (설정 묶음)
-
model configuration(params) 을 프로필 내부에 포함, defaults 로 공통값 지원
-
싱글톤 그대로 유지 + 외부 접근 간단화(함수/객체형 모두)
-
하위 호환:
endpoint(s)별칭 제공 -
파일 변경은 최대 1분 내 반영(mtime + TTL 5~60s, 기본 30s)
원하면, ComfyUI 노드(시각적)도 params/http를 JSON으로 같이 내보내는 ConfigProfileLookup 노드 버전도 바로 만들어 줄게!
profile은 좀 너무 범용적인데
좋아, “profile”이 너무 범용적으로 느껴진다면 더 목적에 딱 맞는 preset(프리셋) 으로 용어를 좁히자고 제안할게. “프리셋”은 모델 연결 정보 + 요청 식별자 + 모델 파라미터 + HTTP 옵션 을 한 세트로 담는 의미가 분명하고, UI/워크플로우에서도 흔히 쓰이는 표현이라 직관적이야.
아래는 용어/스키마/코드 전부 ‘preset’으로 정리한 최종본 이야. (기존 profile(s)/endpoint(s)는 하위 호환 별칭 유지)
1) 이름 · 스키마 결정 (step-by-step)
-
섹션명:
[presets."키"] -
필수 필드:
display_name,request_name,base_url,api_key -
선택 필드:
params(모델 파라미터),http(timeout/headers/query 등) -
공통 기본값 병합:
defaults.params,defaults.http를 각 프리셋에 머지
예시 config.toml
version = 1
[settings]
refresh_interval_seconds = 30 # (5~60) 1분 내 반영 요건 충족
# 모든 프리셋 공통 기본값(선택)
[defaults.params]
temperature = 0.3
max_tokens = 2048
[defaults.http]
timeout_seconds = 30
[defaults.http.headers]
User-Agent = "ComfyUI-ConfigReader/1.0"
# ── 프리셋들 ─────────────────────────────────────────────────────
[presets."text-model-a"]
display_name = "Text Model A"
request_name = "gpt-4o-mini"
base_url = "https://api.openai.com/v1"
api_key = "${OPENAI_API_KEY}"
[presets."text-model-a".params]
temperature = 0.2
top_p = 0.9
[presets."text-model-a".http]
timeout_seconds = 20
[presets."text-model-a".http.headers]
X-Org-Id = "${ORG_ID}"
[presets."text-model-b"]
display_name = "Text Model B"
request_name = "claude-3-5-sonnet"
base_url = "https://api.anthropic.com"
api_key = "${ANTHROPIC_API_KEY}"
# ── 다른 일반 설정 섹션들도 자유롭게 ─────────────────────────────
[feature_flags]
use_cache = true
${ENV}표기는 환경변수로 치환돼.
하위 호환 :[profiles]나[endpoints]를 써도 자동 인식(내부에서presets로 취급).
2) 외부 접근 API (싱글톤, 한 줄로 끝)
-
함수형:
from comfyui_config_reader import preset, presets, params, http, get -
객체형(원하면):
from comfyui_config_reader import CONFIG→CONFIG.preset("...")
사용 예
from comfyui_config_reader import preset, params, http, get
cfg = preset("text-model-a") # "키" / "display_name" / "request_name" 아무거나 OK
api_key = cfg["api_key"]
base_url = cfg["base_url"]
req_name = cfg["request_name"]
p = cfg["params"] # 모델 파라미터 dict
h = cfg["http"] # HTTP 옵션 dict
temperature = params("text-model-a").get("temperature", 0.3)
timeout = http("text-model-a").get("timeout_seconds", 30)
region = get("feature_flags.use_cache") # 범용 키 접근
3) 드롭인 교체 코드
아래 두 파일만 교체하면 끝이야.
custom_nodes/comfyui_config_reader/__init__.py
from .config_singleton import (
CONFIG,
configure,
preset, presets,
params, http,
get, snapshot,
# legacy alias (하위 호환)
profile as profile, profiles as profiles,
endpoint as endpoint, endpoints as endpoints,
)
custom_nodes/comfyui_config_reader/config_singleton.py
import os, time, threading
from typing import Dict, Any, Optional
try:
import tomllib # py3.11+
_TOML_LOADS = tomllib.loads
except Exception:
import tomli # py3.10-
_TOML_LOADS = tomli.loads
def _expand_env(v: Any) -> Any:
if isinstance(v, str):
out, i = "", 0
while i < len(v):
if v[i:i+2] == "${":
j = v.find("}", i+2)
if j != -1:
out += os.environ.get(v[i+2:j], ""); i = j + 1
else:
out += v[i]; i += 1
else:
out += v[i]; i += 1
return out
if isinstance(v, dict): return {k: _expand_env(x) for k, x in v.items()}
if isinstance(v, list): return [_expand_env(x) for x in v]
return v
def _norm(s: str) -> str:
return str(s).strip().lower()
def _deep_merge(a: Dict[str, Any], b: Dict[str, Any]) -> Dict[str, Any]:
out = dict(a)
for k, v in (b or {}).items():
if isinstance(v, dict) and isinstance(out.get(k), dict):
out[k] = _deep_merge(out[k], v)
else:
out[k] = v
return out
class _ReadOnlyConfig:
"""
- 읽기 전용 TOML (파일 쓰기 없음)
- mtime + TTL(5~60s, 기본 30s) 자동 재로드 → 1분 내 변경 반영
- 표준: [presets]; 레거시: [profiles], [endpoints] 자동 인식
- preset 필수 필드: display_name, request_name, base_url, api_key
+ params(dict), http(dict: timeout_seconds, headers{}, query{})
"""
def __init__(self, path: str):
self.path = path
self._lock = threading.Lock()
self._loaded_at = 0.0
self._last_mtime = -1.0
self._refresh = 30
self._raw: Dict[str, Any] = {}
self._idx: Dict[str, Dict[str, Any]] = {}
# --- 내부 로딩 ---
def _read_text(self) -> str:
with open(self.path, "rb") as f:
return f.read().decode("utf-8")
def _extract_presets_table(self, data: Dict[str, Any]) -> Dict[str, Any]:
if isinstance(data.get("presets"), dict): return data["presets"]
if isinstance(data.get("profiles"), dict): return data["profiles"] # legacy
if isinstance(data.get("endpoints"), dict): return data["endpoints"] # legacy
return {}
def _build_index(self, data: Dict[str, Any]) -> None:
self._idx.clear()
# defaults
defaults = _expand_env(data.get("defaults") or {})
d_params = defaults.get("params") or {}
d_http = defaults.get("http") or {}
if "headers" not in d_http: d_http["headers"] = {}
if "query" not in d_http: d_http["query"] = {}
tbl = self._extract_presets_table(data)
if not isinstance(tbl, dict): return
for key, raw in tbl.items():
if not isinstance(raw, dict): continue
cfg = _expand_env(raw)
req = ["display_name", "request_name", "base_url", "api_key"]
if any(k not in cfg or not str(cfg[k]).strip() for k in req):
continue
p_params = cfg.get("params") or {}
p_http = cfg.get("http") or {}
if "headers" not in p_http: p_http["headers"] = {}
if "query" not in p_http: p_http["query"] = {}
merged_params = {**d_params, **p_params}
merged_http = _deep_merge(d_http, p_http)
entry = {
"display_name": str(cfg["display_name"]),
"request_name": str(cfg["request_name"]),
"base_url": str(cfg["base_url"]),
"api_key": str(cfg["api_key"]),
"params": merged_params,
"http": merged_http,
}
for c in {str(key), str(cfg.get("display_name","")), str(cfg.get("request_name",""))}:
if c:
self._idx[_norm(c)] = entry
def _reload_locked(self) -> None:
txt = self._read_text()
data = _TOML_LOADS(txt)
r = int((data.get("settings") or {}).get("refresh_interval_seconds", 30))
r = max(5, min(60, r))
env_r = os.environ.get("COMFY_CONFIG_REFRESH_SECONDS")
if env_r:
try: r = max(5, min(60, int(env_r)))
except ValueError: pass
self._refresh = r
self._raw = _expand_env(data)
self._build_index(self._raw)
self._loaded_at = time.time()
try:
self._last_mtime = os.path.getmtime(self.path)
except FileNotFoundError:
self._last_mtime = -1.0
def _stale_or_changed(self) -> bool:
try:
m = os.path.getmtime(self.path)
except FileNotFoundError:
return True
now = time.time()
return (m != self._last_mtime) or (now - self._loaded_at >= self._refresh)
def _ensure_loaded(self) -> None:
with self._lock:
if self._loaded_at == 0 or self._stale_or_changed():
self._reload_locked()
# --- 퍼블릭 API ---
def preset(self, key: str) -> Dict[str, Any]:
if not key or not str(key).strip():
raise KeyError("preset key is empty")
self._ensure_loaded()
hit = self._idx.get(_norm(key))
if not hit:
with self._lock:
self._reload_locked()
hit = self._idx.get(_norm(key))
if not hit:
raise KeyError(f"preset not found: {key}")
return hit
def presets(self) -> Dict[str, Dict[str, Any]]:
self._ensure_loaded()
out: Dict[str, Dict[str, Any]] = {}
seen = set()
for _, v in self._idx.items():
dn = v["display_name"]
if dn in seen: continue
seen.add(dn)
out[dn] = {
"display_name": v["display_name"],
"request_name": v["request_name"],
"base_url": v["base_url"],
"params": v.get("params", {}),
"http": v.get("http", {}),
}
return out
def params(self, key: str) -> Dict[str, Any]:
return dict(self.preset(key).get("params", {}))
def http(self, key: str) -> Dict[str, Any]:
return dict(self.preset(key).get("http", {}))
def get(self, dotted: str) -> Any:
self._ensure_loaded()
cur = self._raw
for part in filter(None, dotted.split(".")):
if isinstance(cur, dict) and part in cur:
cur = cur[part]
else:
raise KeyError(f"config path not found: {dotted}")
return cur
def snapshot(self) -> Dict[str, Any]:
self._ensure_loaded()
return dict(self._raw)
# ----- 싱글톤 -----
_lock = threading.Lock()
_instance: Optional[_ReadOnlyConfig] = None
_default_path = None
def _resolve_default_path() -> str:
here = os.path.dirname(__file__)
return _default_path or os.environ.get("COMFY_CONFIG_PATH", os.path.join(here, "config.toml"))
def _get_instance() -> _ReadOnlyConfig:
global _instance
if _instance is None:
with _lock:
if _instance is None:
_instance = _ReadOnlyConfig(_resolve_default_path())
return _instance
def configure(*, path: Optional[str] = None, refresh_seconds: Optional[int] = None) -> None:
global _instance, _default_path
if path:
_default_path = os.path.abspath(path)
if refresh_seconds is not None:
os.environ["COMFY_CONFIG_REFRESH_SECONDS"] = str(refresh_seconds)
with _lock:
_instance = _ReadOnlyConfig(_resolve_default_path())
# ---- 외부 API ----
def preset(key: str) -> Dict[str, Any]: return _get_instance().preset(key)
def presets() -> Dict[str, Dict[str, Any]]: return _get_instance().presets()
def params(key: str) -> Dict[str, Any]: return _get_instance().params(key)
def http(key: str) -> Dict[str, Any]: return _get_instance().http(key)
def get(dotted: str) -> Any: return _get_instance().get(dotted)
def snapshot() -> Dict[str, Any]: return _get_instance().snapshot()
# ---- 하위 호환 별칭 ----
def profile(key: str) -> Dict[str, Any]: return preset(key)
def profiles() -> Dict[str, Dict[str, Any]]:return presets()
def endpoint(key: str) -> Dict[str, Any]: return preset(key)
def endpoints() -> Dict[str, Dict[str, Any]]: return presets()
# ---- 객체형 프록시 ----
class _ConfigProxy:
def __getattr__(self, name: str):
return getattr(_get_instance(), name)
CONFIG = _ConfigProxy()
4) (원하면) ComfyUI 노드도 용어 맞춰 변경
-
ConfigPresetLookup:display_name,request_name,base_url,api_key,params(json),http(json)출력 -
ConfigListPresets: 모든 프리셋 JSON
원하시면 바로 생성해 줄게.
5) 마이그레이션 정리
-
profile(s)/endpoint(s)호출 코드는 그대로 동작(별칭 제공). -
설정 파일에서
[profiles]/[endpoints]유지해도 인식하지만, 앞으로는[presets]권장. -
모델 파라미터/HTTP 옵션은 프리셋 내부(
params/http)에 넣으면 되고, 공통값은defaults에 두면 자동 병합돼.
필요하면 “preset” 대신 deployment , model_preset 같은 더 구체적인 대안도 줄 수 있어. 하지만 모델 호출용 세팅 묶음 이라는 관점에서 사용자/오퍼레이터 모두에게 가장 익숙하고 짧은 용어는 “preset”이라 이걸 기본 권장으로 뒀어.
아니 llm endpoint 라는 느낌이 전혀 안들어서
오케이—“LLM 엔드포인트” 느낌을 확실히 주도록 용어를 llm_endpoints로 표준화했어요.
(이전 presets/profiles/endpoints는 자동 하위 호환 으로 계속 읽습니다.)
아래 두 파일만 교체하면 끝! 외부에서는 llm_endpoint(...), llm_params(...) 같은 이름으로 바로 씁니다.
1) 스키마 (config.toml)
version = 1
[settings]
# 파일 변경 감지 주기(초) — 5~60 권장. "1분 이내 반영" 요건 충족.
refresh_interval_seconds = 30
# 공통 기본값(선택): 모든 LLM 엔드포인트에 병합됨
[defaults.params]
temperature = 0.3
max_tokens = 2048
[defaults.http]
timeout_seconds = 30
[defaults.http.headers]
User-Agent = "ComfyUI-ConfigReader/1.0"
# ── LLM 엔드포인트들 ─────────────────────────────────────────
[llm_endpoints."gpt-4o-mini-a"]
display_name = "OpenAI GPT-4o mini A"
model = "gpt-4o-mini" # (= request_name 별칭)
base_url = "https://api.openai.com/v1"
api_key = "${OPENAI_API_KEY}"
[llm_endpoints."gpt-4o-mini-a".params]
temperature = 0.2
top_p = 0.9
[llm_endpoints."gpt-4o-mini-a".http]
timeout_seconds = 20
[llm_endpoints."gpt-4o-mini-a".http.headers]
X-Org-Id = "${ORG_ID}"
[llm_endpoints."claude-sonnet"]
display_name = "Anthropic Claude 3.5 Sonnet"
model = "claude-3-5-sonnet"
base_url = "https://api.anthropic.com"
api_key = "${ANTHROPIC_API_KEY}"
# ── 다른 일반 설정도 자유롭게 ────────────────────────────────
[feature_flags]
use_cache = true
-
필수:
display_name,model(=request_name),base_url,api_key -
선택:
params(모델 파라미터),http(timeout/headers/query 등) -
병합 규칙 :
defaults.params/defaults.http→ 각 엔드포인트의params/http에 덮어쓰기 병합
하위 호환:
[presets],[profiles],[endpoints]도 자동 인식합니다.
필드도request_name/request_model/model중 아무거나 써도 OK(내부에서model로 통일).
2) 외부 접근 API (싱글톤, 직관적)
from comfyui_config_reader import (
llm_endpoint, llm_endpoints, # 메타 + 연결 정보
llm_params, llm_http, # 모델 파라미터 / HTTP 옵션만
get, CONFIG, # 범용 키 접근 / 객체형 접근
)
사용 예:
cfg = llm_endpoint("gpt-4o-mini-a") # 키 / display_name / model 아무거나
api_key = cfg["api_key"]
base_url = cfg["base_url"]
model = cfg["model"]
params = llm_params("gpt-4o-mini-a") # {"temperature": 0.2, "top_p": 0.9, ...}
http = llm_http("gpt-4o-mini-a") # {"timeout_seconds": ..., "headers": {...}, ...}
# 범용 설정
use_cache = get("feature_flags.use_cache")
3) 드롭인 교체 코드
custom_nodes/comfyui_config_reader/__init__.py
from .config_singleton import (
CONFIG,
configure,
llm_endpoint, llm_endpoints,
llm_params, llm_http,
get, snapshot,
# legacy alias (하위 호환)
preset as preset, presets as presets,
profile as profile, profiles as profiles,
endpoint as endpoint, endpoints as endpoints,
params as params, http as http,
)
custom_nodes/comfyui_config_reader/config_singleton.py
import os, time, threading
from typing import Dict, Any, Optional
try:
import tomllib # py3.11+
_TOML_LOADS = tomllib.loads
except Exception:
import tomli # py3.10-
_TOML_LOADS = tomli.loads
def _expand_env(v: Any) -> Any:
if isinstance(v, str):
out, i = "", 0
while i < len(v):
if v[i:i+2] == "${":
j = v.find("}", i+2)
if j != -1:
out += os.environ.get(v[i+2:j], ""); i = j + 1
else:
out += v[i]; i += 1
else:
out += v[i]; i += 1
return out
if isinstance(v, dict): return {k: _expand_env(x) for k, x in v.items()}
if isinstance(v, list): return [_expand_env(x) for x in v]
return v
def _norm(s: str) -> str:
return str(s).strip().lower()
def _deep_merge(a: Dict[str, Any], b: Dict[str, Any]) -> Dict[str, Any]:
out = dict(a)
for k, v in (b or {}).items():
if isinstance(v, dict) and isinstance(out.get(k), dict):
out[k] = _deep_merge(out[k], v)
else:
out[k] = v
return out
# --- 필드 별칭 정규화 (model / request_name / request_model) ---
def _normalize_model_field(cfg: Dict[str, Any]) -> str:
for k in ("model", "request_name", "request_model"):
if k in cfg and str(cfg[k]).strip():
return str(cfg[k]).strip()
return ""
class _ReadOnlyConfig:
"""
- 읽기 전용 TOML
- mtime + TTL(5~60s, 기본 30s) 자동 재로드 → 1분 내 변경 반영
- 표준 섹션: [llm_endpoints]
(하위 호환: [presets], [profiles], [endpoints] 도 자동 인식)
- 각 엔드포인트:
* display_name, model(=request_name/… 별칭), base_url, api_key
* params(dict), http(dict: timeout_seconds, headers{}, query{})
"""
def __init__(self, path: str):
self.path = path
self._lock = threading.Lock()
self._loaded_at = 0.0
self._last_mtime = -1.0
self._refresh = 30
self._raw: Dict[str, Any] = {}
self._idx: Dict[str, Dict[str, Any]] = {}
# --- 내부 로딩 ---
def _read_text(self) -> str:
with open(self.path, "rb") as f:
return f.read().decode("utf-8")
def _extract_llm_table(self, data: Dict[str, Any]) -> Dict[str, Any]:
# 1순위 표준
if isinstance(data.get("llm_endpoints"), dict): return data["llm_endpoints"]
# 하위 호환
for key in ("presets", "profiles", "endpoints"):
if isinstance(data.get(key), dict): return data[key]
return {}
def _build_index(self, data: Dict[str, Any]) -> None:
self._idx.clear()
# defaults
defaults = _expand_env(data.get("defaults") or {})
d_params = defaults.get("params") or {}
d_http = defaults.get("http") or {}
if "headers" not in d_http: d_http["headers"] = {}
if "query" not in d_http: d_http["query"] = {}
tbl = self._extract_llm_table(data)
if not isinstance(tbl, dict): return
for key, raw in tbl.items():
if not isinstance(raw, dict): continue
cfg = _expand_env(raw)
display = str(cfg.get("display_name", "")).strip()
model = _normalize_model_field(cfg)
base = str(cfg.get("base_url", "")).strip()
api = str(cfg.get("api_key", "")).strip()
if not (display and model and base and api):
continue # 필수 누락은 스킵
p_params = cfg.get("params") or {}
p_http = cfg.get("http") or {}
if "headers" not in p_http: p_http["headers"] = {}
if "query" not in p_http: p_http["query"] = {}
merged_params = {**d_params, **p_params}
merged_http = _deep_merge(d_http, p_http)
entry = {
"display_name": display,
"model": model,
"base_url": base,
"api_key": api,
"params": merged_params,
"http": merged_http,
}
# 접근 키: 정의 키 / display_name / model
for c in {str(key), display, model}:
if c:
self._idx[_norm(c)] = entry
def _reload_locked(self) -> None:
txt = self._read_text()
data = _TOML_LOADS(txt)
r = int((data.get("settings") or {}).get("refresh_interval_seconds", 30))
r = max(5, min(60, r))
env_r = os.environ.get("COMFY_CONFIG_REFRESH_SECONDS")
if env_r:
try: r = max(5, min(60, int(env_r)))
except ValueError: pass
self._refresh = r
self._raw = _expand_env(data)
self._build_index(self._raw)
self._loaded_at = time.time()
try:
self._last_mtime = os.path.getmtime(self.path)
except FileNotFoundError:
self._last_mtime = -1.0
def _stale_or_changed(self) -> bool:
try:
m = os.path.getmtime(self.path)
except FileNotFoundError:
return True
now = time.time()
return (m != self._last_mtime) or (now - self._loaded_at >= self._refresh)
def _ensure_loaded(self) -> None:
with self._lock:
if self._loaded_at == 0 or self._stale_or_changed():
self._reload_locked()
# --- 퍼블릭 API ---
def llm_endpoint(self, key: str) -> Dict[str, Any]:
if not key or not str(key).strip():
raise KeyError("llm_endpoint key is empty")
self._ensure_loaded()
hit = self._idx.get(_norm(key))
if not hit:
with self._lock:
self._reload_locked()
hit = self._idx.get(_norm(key))
if not hit:
raise KeyError(f"llm_endpoint not found: {key}")
return hit
def llm_endpoints(self) -> Dict[str, Dict[str, Any]]:
self._ensure_loaded()
out: Dict[str, Dict[str, Any]] = {}
seen = set()
for _, v in self._idx.items():
dn = v["display_name"]
if dn in seen: continue
seen.add(dn)
out[dn] = {
"display_name": v["display_name"],
"model": v["model"],
"base_url": v["base_url"],
"params": v.get("params", {}),
"http": v.get("http", {}),
}
return out
def llm_params(self, key: str) -> Dict[str, Any]:
return dict(self.llm_endpoint(key).get("params", {}))
def llm_http(self, key: str) -> Dict[str, Any]:
return dict(self.llm_endpoint(key).get("http", {}))
def get(self, dotted: str) -> Any:
self._ensure_loaded()
cur = self._raw
for part in filter(None, dotted.split(".")):
if isinstance(cur, dict) and part in cur:
cur = cur[part]
else:
raise KeyError(f"config path not found: {dotted}")
return cur
def snapshot(self) -> Dict[str, Any]:
self._ensure_loaded()
return dict(self._raw)
# ----- 싱글톤 -----
_lock = threading.Lock()
_instance: Optional[_ReadOnlyConfig] = None
_default_path = None
def _resolve_default_path() -> str:
here = os.path.dirname(__file__)
return _default_path or os.environ.get("COMFY_CONFIG_PATH", os.path.join(here, "config.toml"))
def _get_instance() -> _ReadOnlyConfig:
global _instance
if _instance is None:
with _lock:
if _instance is None:
_instance = _ReadOnlyConfig(_resolve_default_path())
return _instance
def configure(*, path: Optional[str] = None, refresh_seconds: Optional[int] = None) -> None:
global _instance, _default_path
if path:
_default_path = os.path.abspath(path)
if refresh_seconds is not None:
os.environ["COMFY_CONFIG_REFRESH_SECONDS"] = str(refresh_seconds)
with _lock:
_instance = _ReadOnlyConfig(_resolve_default_path())
# ---- 외부 API ----
def llm_endpoint(key: str) -> Dict[str, Any]: return _get_instance().llm_endpoint(key)
def llm_endpoints() -> Dict[str, Dict[str, Any]]: return _get_instance().llm_endpoints()
def llm_params(key: str) -> Dict[str, Any]: return _get_instance().llm_params(key)
def llm_http(key: str) -> Dict[str, Any]: return _get_instance().llm_http(key)
def get(dotted: str) -> Any: return _get_instance().get(dotted)
def snapshot() -> Dict[str, Any]: return _get_instance().snapshot()
# ---- 하위 호환 별칭 (필요 시 기존 코드 그대로 동작) ----
def preset(key: str) -> Dict[str, Any]: return llm_endpoint(key)
def presets() -> Dict[str, Dict[str, Any]]: return llm_endpoints()
def profile(key: str) -> Dict[str, Any]: return llm_endpoint(key)
def profiles() -> Dict[str, Dict[str, Any]]: return llm_endpoints()
def endpoint(key: str) -> Dict[str, Any]: return llm_endpoint(key)
def endpoints() -> Dict[str, Dict[str, Any]]: return llm_endpoints()
def params(key: str) -> Dict[str, Any]: return llm_params(key)
def http(key: str) -> Dict[str, Any]: return llm_http(key)
# ---- 객체형 프록시 ----
class _ConfigProxy:
def __getattr__(self, name: str):
return getattr(_get_instance(), name)
CONFIG = _ConfigProxy()
4) 왜 이게 좋은가 (짧게)
-
이름부터 LLM 엔드포인트 임이 분명 (
llm_endpoints) -
모델 파라미터/HTTP 옵션을 같은 묶음 에서 관리 (
params,http) -
기존 설정/코드 안 깨짐(자동 호환)
-
읽기 전용 + mtime/TTL(5~60s)로 최대 1분 내 변경 반영
-
외부 사용은 한 줄 임포트 로 끝:
llm_endpoint("...")
원하면 ComfyUI 노드 이름도 Config: LLM Endpoint Lookup, Config: List LLM Endpoints로 맞춰서 출력(파라미터/HTTP JSON 포함) 버전까지 바로 만들어줄게!
api_providers display_name, model_id, base_url, api_key 이걸로 해줘.
좋아! 이름은api_providers, 필드는 display_name, model_id, base_url, api_key로 딱 맞춰서 정리했어.
(읽기 전용 · 최대 1분 내 변경 반영 · 다른 노드/파일에서 한 줄로 접근)
아래 코드/예시 그대로 교체하면 동작해.
1) config.toml 스키마
version = 1
[settings]
# 파일 변경 감지 주기(초) — 5~60 권장. "1분 이내 반영" 요건 충족.
refresh_interval_seconds = 30
# ── API Providers ─────────────────────────────────────
[api_providers."openai-mini"]
display_name = "OpenAI GPT-4o mini"
model_id = "gpt-4o-mini"
base_url = "https://api.openai.com/v1"
api_key = "${OPENAI_API_KEY}"
[api_providers."anthropic-sonnet"]
display_name = "Anthropic Claude 3.5 Sonnet"
model_id = "claude-3-5-sonnet"
base_url = "https://api.anthropic.com"
api_key = "${ANTHROPIC_API_KEY}"
# ── 다른 범용 설정도 자유롭게 ─────────────────────────
[feature_flags]
use_cache = true
-
${ENV}표기는 환경변수로 치환. -
조회 키는 정의 키 /
display_name/model_id중 아무거나 사용 가능.
2) 외부에서 쓰는 법 (싱글톤, 한 줄)
from comfyui_config_reader import api_provider, api_providers, get
cfg = api_provider("openai-mini") # 또는 "OpenAI GPT-4o mini" / "gpt-4o-mini"
api_key = cfg["api_key"]
base_url = cfg["base_url"]
model_id = cfg["model_id"]
listing = api_providers() # {display_name: {display_name, model_id, base_url}}
use_cache = get("feature_flags.use_cache")
선택(초기화 시 한 번):
from comfyui_config_reader import configure
configure(path="/abs/path/to/config.toml", refresh_seconds=20) # 5~60
3) 드롭인 코드
custom_nodes/comfyui_config_reader/__init__.py
from .config_singleton import (
CONFIG,
configure,
api_provider, api_providers,
get, snapshot,
)
custom_nodes/comfyui_config_reader/config_singleton.py
import os, time, threading
from typing import Dict, Any, Optional
try:
import tomllib # py3.11+
_TOML_LOADS = tomllib.loads
except Exception:
import tomli # py3.10-
_TOML_LOADS = tomli.loads
def _expand_env(v: Any) -> Any:
if isinstance(v, str):
out, i = "", 0
while i < len(v):
if v[i:i+2] == "${":
j = v.find("}", i+2)
if j != -1:
out += os.environ.get(v[i+2:j], ""); i = j + 1
else:
out += v[i]; i += 1
else:
out += v[i]; i += 1
return out
if isinstance(v, dict): return {k: _expand_env(x) for k, x in v.items()}
if isinstance(v, list): return [_expand_env(x) for x in v]
return v
def _norm(s: str) -> str:
return str(s).strip().lower()
# 레거시 필드 호환: model_id 대신 model/request_name/request_model 이 들어와도 인식
def _normalize_model_id(cfg: Dict[str, Any]) -> str:
for k in ("model_id", "model", "request_name", "request_model"):
if k in cfg and str(cfg[k]).strip():
return str(cfg[k]).strip()
return ""
class _ReadOnlyConfig:
"""
- 읽기 전용 TOML 로더 (파일 쓰기 없음)
- mtime + TTL(5~60s, 기본 30s) 자동 재로드 → 1분 내 변경 반영
- 표준 섹션: [api_providers]
(참고: 원하면 하위 호환 섹션을 추가로 인식하도록 확장 가능)
- 각 provider 필수 필드: display_name, model_id, base_url, api_key
"""
def __init__(self, path: str):
self.path = path
self._lock = threading.Lock()
self._loaded_at = 0.0
self._last_mtime = -1.0
self._refresh = 30
self._raw: Dict[str, Any] = {}
self._idx: Dict[str, Dict[str, str]] = {}
# --- 내부 로딩 ---
def _read_text(self) -> str:
with open(self.path, "rb") as f:
return f.read().decode("utf-8")
def _providers_table(self, data: Dict[str, Any]) -> Dict[str, Any]:
tbl = data.get("api_providers")
return tbl if isinstance(tbl, dict) else {}
def _build_index(self, data: Dict[str, Any]) -> None:
self._idx.clear()
tbl = self._providers_table(data)
if not isinstance(tbl, dict): return
for key, raw in tbl.items():
if not isinstance(raw, dict): continue
cfg = _expand_env(raw)
display = str(cfg.get("display_name", "")).strip()
model = _normalize_model_id(cfg)
base = str(cfg.get("base_url", "")).strip()
api = str(cfg.get("api_key", "")).strip()
if not (display and model and base and api):
continue # 필수 누락은 스킵
entry = {
"display_name": display,
"model_id": model,
"base_url": base,
"api_key": api,
}
# 접근 키: 정의 키 / display_name / model_id
for c in {str(key), display, model}:
if c:
self._idx[_norm(c)] = entry
def _reload_locked(self) -> None:
txt = self._read_text()
data = _TOML_LOADS(txt)
r = int((data.get("settings") or {}).get("refresh_interval_seconds", 30))
r = max(5, min(60, r))
env_r = os.environ.get("COMFY_CONFIG_REFRESH_SECONDS")
if env_r:
try: r = max(5, min(60, int(env_r)))
except ValueError: pass
self._refresh = r
self._raw = _expand_env(data)
self._build_index(self._raw)
self._loaded_at = time.time()
try:
self._last_mtime = os.path.getmtime(self.path)
except FileNotFoundError:
self._last_mtime = -1.0
def _stale_or_changed(self) -> bool:
try:
m = os.path.getmtime(self.path)
except FileNotFoundError:
return True
now = time.time()
return (m != self._last_mtime) or (now - self._loaded_at >= self._refresh)
def _ensure_loaded(self) -> None:
with self._lock:
if self._loaded_at == 0 or self._stale_or_changed():
self._reload_locked()
# --- 퍼블릭 API ---
def api_provider(self, key: str) -> Dict[str, str]:
if not key or not str(key).strip():
raise KeyError("api_provider key is empty")
self._ensure_loaded()
hit = self._idx.get(_norm(key))
if not hit:
with self._lock:
self._reload_locked()
hit = self._idx.get(_norm(key))
if not hit:
raise KeyError(f"api_provider not found: {key}")
return hit
def api_providers(self) -> Dict[str, Dict[str, str]]:
"""API Key는 목록에선 숨김."""
self._ensure_loaded()
out: Dict[str, Dict[str, str]] = {}
seen = set()
for _, v in self._idx.items():
dn = v["display_name"]
if dn in seen: continue
seen.add(dn)
out[dn] = {
"display_name": v["display_name"],
"model_id": v["model_id"],
"base_url": v["base_url"],
}
return out
def get(self, dotted: str) -> Any:
self._ensure_loaded()
cur = self._raw
for part in filter(None, dotted.split(".")):
if isinstance(cur, dict) and part in cur:
cur = cur[part]
else:
raise KeyError(f"config path not found: {dotted}")
return cur
def snapshot(self) -> Dict[str, Any]:
self._ensure_loaded()
return dict(self._raw)
# ----- 싱글톤 -----
_lock = threading.Lock()
_instance: Optional[_ReadOnlyConfig] = None
_default_path = None
def _resolve_default_path() -> str:
here = os.path.dirname(__file__)
return _default_path or os.environ.get("COMFY_CONFIG_PATH", os.path.join(here, "config.toml"))
def _get_instance() -> _ReadOnlyConfig:
global _instance
if _instance is None:
with _lock:
if _instance is None:
_instance = _ReadOnlyConfig(_resolve_default_path())
return _instance
def configure(*, path: Optional[str] = None, refresh_seconds: Optional[int] = None) -> None:
global _instance, _default_path
if path:
_default_path = os.path.abspath(path)
if refresh_seconds is not None:
os.environ["COMFY_CONFIG_REFRESH_SECONDS"] = str(refresh_seconds)
with _lock:
_instance = _ReadOnlyConfig(_resolve_default_path())
# ---- 외부 API ----
def api_provider(key: str) -> Dict[str, str]: return _get_instance().api_provider(key)
def api_providers() -> Dict[str, Dict[str, str]]: return _get_instance().api_providers()
def get(dotted: str) -> Any: return _get_instance().get(dotted)
def snapshot() -> Dict[str, Any]: return _get_instance().snapshot()
# ---- 객체형 프록시 ----
class _ConfigProxy:
def __getattr__(self, name: str):
return getattr(_get_instance(), name)
CONFIG = _ConfigProxy()
4) (선택) ComfyUI 노드도 용어 맞춰 제공
원한다면 custom_nodes/comfyui_config_reader/config_nodes.py에 아래처럼:
from .config_singleton import api_provider, api_providers
import json
class ConfigAPIProviderLookup:
@classmethod
def INPUT_TYPES(cls):
return {"required": {"key": ("STRING", {"default": "", "multiline": False})}}
RETURN_TYPES = ("STRING","STRING","STRING","STRING")
RETURN_NAMES = ("display_name","model_id","base_url","api_key")
FUNCTION = "lookup"
CATEGORY = "Config/API Providers (ReadOnly)"
def lookup(self, key: str):
p = api_provider(key)
return (p["display_name"], p["model_id"], p["base_url"], p["api_key"])
class ConfigListAPIProviders:
@classmethod
def INPUT_TYPES(cls):
return {"required": {}}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("providers_json",)
FUNCTION = "list_providers"
CATEGORY = "Config/API Providers (ReadOnly)"
def list_providers(self):
return (json.dumps(api_providers(), ensure_ascii=False, indent=2),)
NODE_CLASS_MAPPINGS = {
"ConfigAPIProviderLookup": ConfigAPIProviderLookup,
"ConfigListAPIProviders": ConfigListAPIProviders,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"ConfigAPIProviderLookup": "Config: API Provider Lookup",
"ConfigListAPIProviders": "Config: List API Providers",
}
5) 체크리스트
-
✅ 섹션/필드명:
api_providers/ (display_name, model_id, base_url, api_key) -
✅ 읽기 전용 + 파일 변경 최대 1분 내 반영 (mtime + TTL 5~60, 기본 30)
-
✅ 어디서든 간단 접근:
api_provider("키|표시명|model_id") -
✅ 사람이 편집 쉬운 TOML +
${ENV}치환 -
✅ 목록 노드는 api_key 비노출
필요하면, 기존 워크플로우(노드 그래프)에서 자동으로 이 이름들로 바꿔주는 스크립트도 만들어줄게.
아 params도 포함해줘. 그런데 [api_providers."anthropic-sonnet"] 이 안에 parameters를 dictionary로 넣을 수 없나? 따로 항목을 빼니까 보기가 불편한데 이런건 yaml에서만 가능?
좋아, TOML에서도 딕셔너리(맵)를 한 섹션 안에 넣을 수 있어.
YAML만 되는 게 아니고, TOML은 두 가지 방법을 지원해:
-
인라인 테이블 : 한 줄
{ ... }로 딕셔너리 값을 넣기 -
점 표기(dotted keys) :
parameters.temperature = 0.7처럼 같은 섹션 안에서 키를 점으로 이어 쓰기
아래에 둘 다 예시 + 코드를 싱글톤 API에 반영해줄게. api_providers 항목 안에 parameters를 그대로 넣고, 기본값 병합도 지원해.
1) TOML 작성법 (step-by-step)
A. 인라인 테이블(깔끔, 한 줄)
[api_providers."anthropic-sonnet"]
display_name = "Anthropic Claude 3.5 Sonnet"
model_id = "claude-3-5-sonnet"
base_url = "https://api.anthropic.com"
api_key = "${ANTHROPIC_API_KEY}"
parameters = { temperature = 0.2, top_p = 0.9, stop = ["\n\nHuman:", "\n\nAssistant:"] }
-
장점: 한눈에 보기 좋음.
-
유의: TOML 1.0 스펙상 인라인 테이블은 한 줄 이어야 해. (줄바꿈/주석 넣기 어려움)
B. 점 표기(dotted keys, 같은 섹션 내부 다줄)
[api_providers."anthropic-sonnet"]
display_name = "Anthropic Claude 3.5 Sonnet"
model_id = "claude-3-5-sonnet"
base_url = "https://api.anthropic.com"
api_key = "${ANTHROPIC_API_KEY}"
parameters.temperature = 0.2
parameters.top_p = 0.9
parameters.stop = ["\n\nHuman:", "\n\nAssistant:"] # ← 주석 OK
-
장점: 여러 줄로 읽기 편하고 주석 가능.
-
내부적으로는 딕셔너리로 파싱됨. (YAML 필요 없음)
(선택) 공통 기본값 병합
[defaults.parameters]
temperature = 0.3
max_tokens = 2048
- 각 provider의
parameters위에 덮어쓰기 병합 돼서 최종값이 만들어져.
2) 전체 예시 config.toml
version = 1
[settings]
refresh_interval_seconds = 30 # 5~60(요구: 1분 내 반영)
[defaults.parameters]
temperature = 0.3
max_tokens = 2048
# ── OpenAI (인라인 테이블) ───────────────────────────
[api_providers."openai-mini"]
display_name = "OpenAI GPT-4o mini"
model_id = "gpt-4o-mini"
base_url = "https://api.openai.com/v1"
api_key = "${OPENAI_API_KEY}"
parameters = { temperature = 0.25, top_p = 0.95 }
# ── Anthropic (점 표기) ───────────────────────────────
[api_providers."anthropic-sonnet"]
display_name = "Anthropic Claude 3.5 Sonnet"
model_id = "claude-3-5-sonnet"
base_url = "https://api.anthropic.com"
api_key = "${ANTHROPIC_API_KEY}"
parameters.temperature = 0.2
parameters.top_p = 0.9
parameters.stop = ["\n\nHuman:", "\n\nAssistant:"]
3) 외부 사용 방법 (그대로, + parameters 지원)
from comfyui_config_reader import api_provider, api_providers, api_parameters, get
cfg = api_provider("anthropic-sonnet") # 키 / display_name / model_id 아무거나
print(cfg["model_id"], cfg["base_url"], cfg["api_key"])
print(cfg["parameters"]) # ← 최종 병합된 dict
# 파라미터만 바로
params = api_parameters("openai-mini") # {"temperature": 0.25, "top_p": 0.95, "max_tokens": 2048}
use_cache = get("feature_flags.use_cache")
4) 드롭인 교체 코드
custom_nodes/comfyui_config_reader/__init__.py
from .config_singleton import (
CONFIG,
configure,
api_provider, api_providers,
api_parameters, # ← 추가
get, snapshot,
)
custom_nodes/comfyui_config_reader/config_singleton.py
import os, time, threading
from typing import Dict, Any, Optional
try:
import tomllib # py3.11+
_TOML_LOADS = tomllib.loads
except Exception:
import tomli # py3.10-
_TOML_LOADS = tomli.loads
def _expand_env(v: Any) -> Any:
if isinstance(v, str):
out, i = "", 0
while i < len(v):
if v[i:i+2] == "${":
j = v.find("}", i+2)
if j != -1:
out += os.environ.get(v[i+2:j], ""); i = j + 1
else:
out += v[i]; i += 1
else:
out += v[i]; i += 1
return out
if isinstance(v, dict): return {k: _expand_env(x) for k, x in v.items()}
if isinstance(v, list): return [_expand_env(x) for x in v]
return v
def _norm(s: str) -> str:
return str(s).strip().lower()
def _normalize_model_id(cfg: Dict[str, Any]) -> str:
for k in ("model_id", "model", "request_name", "request_model"):
if k in cfg and str(cfg[k]).strip():
return str(cfg[k]).strip()
return ""
def _deep_merge(a: Dict[str, Any], b: Dict[str, Any]) -> Dict[str, Any]:
out = dict(a)
for k, v in (b or {}).items():
if isinstance(v, dict) and isinstance(out.get(k), dict):
out[k] = _deep_merge(out[k], v)
else:
out[k] = v
return out
class _ReadOnlyConfig:
"""
- 읽기 전용 TOML
- mtime + TTL(5~60s, 기본 30s) 자동 재로드 → 1분 내 변경 반영
- 표준 섹션: [api_providers]
- 각 provider 필수: display_name, model_id, base_url, api_key
+ 선택: parameters(dict) # 인라인 테이블 또는 dotted keys 모두 지원
- [defaults.parameters] 병합 지원(개별 provider.parameters가 우선)
"""
def __init__(self, path: str):
self.path = path
self._lock = threading.Lock()
self._loaded_at = 0.0
self._last_mtime = -1.0
self._refresh = 30
self._raw: Dict[str, Any] = {}
self._idx: Dict[str, Dict[str, Any]] = {}
# --- 내부 로딩 ---
def _read_text(self) -> str:
with open(self.path, "rb") as f:
return f.read().decode("utf-8")
def _providers_table(self, data: Dict[str, Any]) -> Dict[str, Any]:
tbl = data.get("api_providers")
return tbl if isinstance(tbl, dict) else {}
def _build_index(self, data: Dict[str, Any]) -> None:
self._idx.clear()
# defaults.parameters (alias: defaults.params도 지원)
defaults = _expand_env(data.get("defaults") or {})
d_params = defaults.get("parameters") or defaults.get("params") or {}
tbl = self._providers_table(data)
if not isinstance(tbl, dict): return
for key, raw in tbl.items():
if not isinstance(raw, dict): continue
cfg = _expand_env(raw)
display = str(cfg.get("display_name", "")).strip()
model = _normalize_model_id(cfg)
base = str(cfg.get("base_url", "")).strip()
api = str(cfg.get("api_key", "")).strip()
if not (display and model and base and api):
continue # 필수 누락 스킵
# provider.parameters (alias: params) 추출 + defaults 병합
p_params = cfg.get("parameters")
if p_params is None:
p_params = cfg.get("params") # 과거 명칭 호환
if not isinstance(p_params, dict):
p_params = {}
merged_params = {**d_params, **p_params}
entry = {
"display_name": display,
"model_id": model,
"base_url": base,
"api_key": api,
"parameters": merged_params, # ← 포함
}
# 접근 키: 정의 키 / display_name / model_id
for c in {str(key), display, model}:
if c:
self._idx[_norm(c)] = entry
def _reload_locked(self) -> None:
txt = self._read_text()
data = _TOML_LOADS(txt)
r = int((data.get("settings") or {}).get("refresh_interval_seconds", 30))
r = max(5, min(60, r))
env_r = os.environ.get("COMFY_CONFIG_REFRESH_SECONDS")
if env_r:
try: r = max(5, min(60, int(env_r)))
except ValueError: pass
self._refresh = r
self._raw = _expand_env(data)
self._build_index(self._raw)
self._loaded_at = time.time()
try:
self._last_mtime = os.path.getmtime(self.path)
except FileNotFoundError:
self._last_mtime = -1.0
def _stale_or_changed(self) -> bool:
try:
m = os.path.getmtime(self.path)
except FileNotFoundError:
return True
now = time.time()
return (m != self._last_mtime) or (now - self._loaded_at >= self._refresh)
def _ensure_loaded(self) -> None:
with self._lock:
if self._loaded_at == 0 or self._stale_or_changed():
self._reload_locked()
# --- 퍼블릭 API ---
def api_provider(self, key: str) -> Dict[str, Any]:
if not key or not str(key).strip():
raise KeyError("api_provider key is empty")
self._ensure_loaded()
hit = self._idx.get(_norm(key))
if not hit:
with self._lock:
self._reload_locked()
hit = self._idx.get(_norm(key))
if not hit:
raise KeyError(f"api_provider not found: {key}")
return hit
def api_providers(self) -> Dict[str, Dict[str, Any]]:
"""목록에서는 api_key 숨김, parameters는 노출."""
self._ensure_loaded()
out: Dict[str, Dict[str, Any]] = {}
seen = set()
for _, v in self._idx.items():
dn = v["display_name"]
if dn in seen: continue
seen.add(dn)
out[dn] = {
"display_name": v["display_name"],
"model_id": v["model_id"],
"base_url": v["base_url"],
"parameters": v.get("parameters", {}),
}
return out
def api_parameters(self, key: str) -> Dict[str, Any]:
return dict(self.api_provider(key).get("parameters", {}))
def get(self, dotted: str) -> Any:
self._ensure_loaded()
cur = self._raw
for part in filter(None, dotted.split(".")):
if isinstance(cur, dict) and part in cur:
cur = cur[part]
else:
raise KeyError(f"config path not found: {dotted}")
return cur
def snapshot(self) -> Dict[str, Any]:
self._ensure_loaded()
return dict(self._raw)
# ----- 싱글톤 -----
_lock = threading.Lock()
_instance: Optional[_ReadOnlyConfig] = None
_default_path = None
def _resolve_default_path() -> str:
here = os.path.dirname(__file__)
return _default_path or os.environ.get("COMFY_CONFIG_PATH", os.path.join(here, "config.toml"))
def _get_instance() -> _ReadOnlyConfig:
global _instance
if _instance is None:
with _lock:
if _instance is None:
_instance = _ReadOnlyConfig(_resolve_default_path())
return _instance
def configure(*, path: Optional[str] = None, refresh_seconds: Optional[int] = None) -> None:
global _instance, _default_path
if path:
_default_path = os.path.abspath(path)
if refresh_seconds is not None:
os.environ["COMFY_CONFIG_REFRESH_SECONDS"] = str(refresh_seconds)
with _lock:
_instance = _ReadOnlyConfig(_resolve_default_path())
# ---- 외부 API ----
def api_provider(key: str) -> Dict[str, Any]: return _get_instance().api_provider(key)
def api_providers() -> Dict[str, Dict[str, Any]]: return _get_instance().api_providers()
def api_parameters(key: str) -> Dict[str, Any]: return _get_instance().api_parameters(key)
def get(dotted: str) -> Any: return _get_instance().get(dotted)
def snapshot() -> Dict[str, Any]: return _get_instance().snapshot()
# ---- 객체형 프록시 ----
class _ConfigProxy:
def __getattr__(self, name: str):
return getattr(_get_instance(), name)
CONFIG = _ConfigProxy()
5) 정리
-
✅ TOML에서도 딕셔너리 가능 : 인라인 테이블
{...}또는 점 표기parameters.foo = -
✅
api_providers각 항목에parameters포함 지원 (인라인/점표기 모두 OK) -
✅
defaults.parameters로 공통값 병합 -
✅ 외부 접근:
api_provider("키")["parameters"]또는api_parameters("키") -
✅ 읽기 전용 + 파일 변경 최대 1분 내 반영
원하면 parameters 외에 http 같은 섹션도 같은 방식으로 붙여서 병합 지원해줄 수 있어!
Pydantic 이용해서 api provider 클래스를 만들고 이걸 리스트로 리턴하는 함수가 있으면 좋겠어. 리스트는 api_provider_id 를 사용해서 정렬된 상태로 리턴하고 싶어. 예시에서 "anthropic-sonnet" 이걸 id로 할거야.
좋다—Pydantic 모델 로 ApiProvider 클래스를 정의하고, 이를 api_provider_id(예: "anthropic-sonnet")로 정렬해서 리턴하는 함수까지 깔끔하게 붙였어.
기존 싱글톤/api_providers 스키마와 100% 호환되며, parameters(dict)도 동일하게 지원/병합돼.
아래 3개만 넣으면 끝:
1) config.toml (예시)
version = 1
[settings]
refresh_interval_seconds = 30
[defaults.parameters]
temperature = 0.3
max_tokens = 2048
[api_providers."openai-mini"]
display_name = "OpenAI GPT-4o mini"
model_id = "gpt-4o-mini"
base_url = "https://api.openai.com/v1"
api_key = "${OPENAI_API_KEY}"
parameters = { temperature = 0.25, top_p = 0.95 }
[api_providers."anthropic-sonnet"]
display_name = "Anthropic Claude 3.5 Sonnet"
model_id = "claude-3-5-sonnet"
base_url = "https://api.anthropic.com"
api_key = "${ANTHROPIC_API_KEY}"
# TOML dotted-keys로 dict 채우기
parameters.temperature = 0.2
parameters.top_p = 0.9
parameters.stop = ["\n\nHuman:", "\n\nAssistant:"]
2) custom_nodes/comfyui_config_reader/models.py
from typing import Any, Dict
from pydantic import BaseModel, Field
class ApiProvider(BaseModel):
"""
단일 API Provider 엔트리 (정렬 키: api_provider_id)
"""
api_provider_id: str = Field(..., description='예: "anthropic-sonnet"')
display_name: str
model_id: str
base_url: str
api_key: str
parameters: Dict[str, Any] = Field(default_factory=dict)
3) custom_nodes/comfyui_config_reader/config_singleton.py (드롭인 교체본)
import os, time, threading
from typing import Dict, Any, Optional, List
try:
import tomllib # py3.11+
_TOML_LOADS = tomllib.loads
except Exception:
import tomli # py3.10-
_TOML_LOADS = tomli.loads
from .models import ApiProvider # ← Pydantic 모델
def _expand_env(v: Any) -> Any:
if isinstance(v, str):
out, i = "", 0
while i < len(v):
if v[i:i+2] == "${":
j = v.find("}", i+2)
if j != -1:
out += os.environ.get(v[i+2:j], ""); i = j + 1
else:
out += v[i]; i += 1
else:
out += v[i]; i += 1
return out
if isinstance(v, dict): return {k: _expand_env(x) for k, x in v.items()}
if isinstance(v, list): return [_expand_env(x) for x in v]
return v
def _norm(s: str) -> str:
return str(s).strip().lower()
def _normalize_model_id(cfg: Dict[str, Any]) -> str:
for k in ("model_id", "model", "request_name", "request_model"):
if k in cfg and str(cfg[k]).strip():
return str(cfg[k]).strip()
return ""
def _deep_merge(a: Dict[str, Any], b: Dict[str, Any]) -> Dict[str, Any]:
out = dict(a)
for k, v in (b or {}).items():
if isinstance(v, dict) and isinstance(out.get(k), dict):
out[k] = _deep_merge(out[k], v)
else:
out[k] = v
return out
class _ReadOnlyConfig:
"""
- 읽기 전용 TOML
- mtime + TTL(5~60s, 기본 30s) 자동 재로드 → 1분 내 변경 반영
- 표준 섹션: [api_providers]
- 각 provider 필수: display_name, model_id, base_url, api_key
+ 선택: parameters(dict) — 인라인 테이블/점 표기 모두 지원
- [defaults.parameters] 병합 지원 (provider.parameters가 우선)
- Pydantic 모델(ApiProvider) 변환/정렬 지원
"""
def __init__(self, path: str):
self.path = path
self._lock = threading.Lock()
self._loaded_at = 0.0
self._last_mtime = -1.0
self._refresh = 30
self._raw: Dict[str, Any] = {}
# 키(조회 식별자: id/display_name/model_id) → provider entry
self._idx: Dict[str, Dict[str, Any]] = {}
# 고유 id → provider entry (중복 제거/정렬 용)
self._by_id: Dict[str, Dict[str, Any]] = {}
# --- 내부 로딩 ---
def _read_text(self) -> str:
with open(self.path, "rb") as f:
return f.read().decode("utf-8")
def _providers_table(self, data: Dict[str, Any]) -> Dict[str, Any]:
tbl = data.get("api_providers")
return tbl if isinstance(tbl, dict) else {}
def _build_index(self, data: Dict[str, Any]) -> None:
self._idx.clear()
self._by_id.clear()
# defaults.parameters (alias: defaults.params)
defaults = _expand_env(data.get("defaults") or {})
d_params = defaults.get("parameters") or defaults.get("params") or {}
tbl = self._providers_table(data)
if not isinstance(tbl, dict): return
for provider_id, raw in tbl.items():
if not isinstance(raw, dict): continue
cfg = _expand_env(raw)
display = str(cfg.get("display_name", "")).strip()
model = _normalize_model_id(cfg)
base = str(cfg.get("base_url", "")).strip()
api = str(cfg.get("api_key", "")).strip()
if not (display and model and base and api):
continue # 필수 누락 스킵
# provider.parameters (alias: params) + defaults 병합
p_params = cfg.get("parameters")
if p_params is None:
p_params = cfg.get("params")
if not isinstance(p_params, dict):
p_params = {}
merged_parameters = {**d_params, **p_params}
entry = {
"api_provider_id": str(provider_id), # ← 원본 TOML 키
"display_name": display,
"model_id": model,
"base_url": base,
"api_key": api,
"parameters": merged_parameters,
}
# 고유 id 사전
self._by_id[str(provider_id)] = entry
# 조회 인덱스(여러 키로 접근 가능: id / display_name / model_id)
for c in {str(provider_id), display, model}:
if c:
self._idx[_norm(c)] = entry
def _reload_locked(self) -> None:
txt = self._read_text()
data = _TOML_LOADS(txt)
r = int((data.get("settings") or {}).get("refresh_interval_seconds", 30))
r = max(5, min(60, r))
env_r = os.environ.get("COMFY_CONFIG_REFRESH_SECONDS")
if env_r:
try: r = max(5, min(60, int(env_r)))
except ValueError: pass
self._refresh = r
self._raw = _expand_env(data)
self._build_index(self._raw)
self._loaded_at = time.time()
try:
self._last_mtime = os.path.getmtime(self.path)
except FileNotFoundError:
self._last_mtime = -1.0
def _stale_or_changed(self) -> bool:
try:
m = os.path.getmtime(self.path)
except FileNotFoundError:
return True
now = time.time()
return (m != self._last_mtime) or (now - self._loaded_at >= self._refresh)
def _ensure_loaded(self) -> None:
with self._lock:
if self._loaded_at == 0 or self._stale_or_changed():
self._reload_locked()
# --- 퍼블릭 키-값 API ---
def api_provider(self, key: str) -> Dict[str, Any]:
if not key or not str(key).strip():
raise KeyError("api_provider key is empty")
self._ensure_loaded()
hit = self._idx.get(_norm(key))
if not hit:
with self._lock:
self._reload_locked()
hit = self._idx.get(_norm(key))
if not hit:
raise KeyError(f"api_provider not found: {key}")
return hit
def api_providers(self) -> Dict[str, Dict[str, Any]]:
"""목록(요약): api_key는 숨기고 parameters는 노출."""
self._ensure_loaded()
out: Dict[str, Dict[str, Any]] = {}
for pid, v in sorted(self._by_id.items(), key=lambda kv: kv[0]):
out[v["display_name"]] = {
"display_name": v["display_name"],
"model_id": v["model_id"],
"base_url": v["base_url"],
"parameters": v.get("parameters", {}),
}
return out
# --- Pydantic 모델 API ---
def api_provider_models(self) -> List[ApiProvider]:
"""ApiProvider 리스트(정렬: api_provider_id). api_key 포함 — 로깅 유의!"""
self._ensure_loaded()
models: List[ApiProvider] = [
ApiProvider(**v) for _, v in sorted(self._by_id.items(), key=lambda kv: kv[0])
]
return models
def api_provider_model(self, api_provider_id: str) -> ApiProvider:
self._ensure_loaded()
entry = self._by_id.get(api_provider_id)
if not entry:
# 다른 키로 들어왔을 수도 있으니 인덱스도 조회
entry = self._idx.get(_norm(api_provider_id))
if not entry:
raise KeyError(f"api_provider not found: {api_provider_id}")
return ApiProvider(**entry)
# --- 범용 ---
def api_parameters(self, key: str) -> Dict[str, Any]:
return dict(self.api_provider(key).get("parameters", {}))
def get(self, dotted: str) -> Any:
self._ensure_loaded()
cur = self._raw
for part in filter(None, dotted.split(".")):
if isinstance(cur, dict) and part in cur:
cur = cur[part]
else:
raise KeyError(f"config path not found: {dotted}")
return cur
def snapshot(self) -> Dict[str, Any]:
self._ensure_loaded()
return dict(self._raw)
# ----- 싱글톤 -----
_lock = threading.Lock()
_instance: Optional[_ReadOnlyConfig] = None
_default_path = None
def _resolve_default_path() -> str:
here = os.path.dirname(__file__)
return _default_path or os.environ.get("COMFY_CONFIG_PATH", os.path.join(here, "config.toml"))
def _get_instance() -> _ReadOnlyConfig:
global _instance
if _instance is None:
with _lock:
if _instance is None:
_instance = _ReadOnlyConfig(_resolve_default_path())
return _instance
def configure(*, path: Optional[str] = None, refresh_seconds: Optional[int] = None) -> None:
global _instance, _default_path
if path:
_default_path = os.path.abspath(path)
if refresh_seconds is not None:
os.environ["COMFY_CONFIG_REFRESH_SECONDS"] = str(refresh_seconds)
with _lock:
_instance = _ReadOnlyConfig(_resolve_default_path())
# ---- 외부 API (함수형) ----
def api_provider(key: str) -> Dict[str, Any]: return _get_instance().api_provider(key)
def api_providers() -> Dict[str, Dict[str, Any]]: return _get_instance().api_providers()
def api_parameters(key: str) -> Dict[str, Any]: return _get_instance().api_parameters(key)
def api_provider_models() -> List[ApiProvider]: return _get_instance().api_provider_models()
def api_provider_model(api_provider_id: str) -> ApiProvider: return _get_instance().api_provider_model(api_provider_id)
def get(dotted: str) -> Any: return _get_instance().get(dotted)
def snapshot() -> Dict[str, Any]: return _get_instance().snapshot()
# ---- 객체형 프록시 ----
class _ConfigProxy:
def __getattr__(self, name: str):
return getattr(_get_instance(), name)
CONFIG = _ConfigProxy()
4) custom_nodes/comfyui_config_reader/__init__.py (re-export)
from .config_singleton import (
CONFIG,
configure,
api_provider, api_providers, api_parameters,
api_provider_model, api_provider_models,
get, snapshot,
)
사용 예 (간단)
from comfyui_config_reader import (
api_provider_models, api_provider_model,
api_provider, api_parameters,
)
# 전체 Pydantic 리스트 (정렬: api_provider_id)
providers = api_provider_models()
# ex) ["anthropic-sonnet", "openai-mini", ...] 순서
for p in providers:
print(p.api_provider_id, p.model_id, p.parameters.get("temperature"))
# 단건 조회(Pydantic)
sonnet = api_provider_model("anthropic-sonnet")
print(sonnet.model_dump()) # pydantic v2 / v1은 .dict()
# 키/표시명/모델ID 아무거나로 접근(딕셔너리)
cfg = api_provider("OpenAI GPT-4o mini")
print(cfg["base_url"], cfg["api_key"])
# 파라미터만
params = api_parameters("gpt-4o-mini")
비고
-
보안 주의 :
api_provider_models()는 객체에api_key가 포함돼요. 로깅/디버깅 시 마스킹하세요. -
TOML 파라미터 입력 : 인라인 테이블(
{...}) 또는 점 표기(parameters.x = ...) 둘 다 OK. -
정렬 보장 : 리스트는 항상
api_provider_id(TOML 키) 기준 오름차순. -
성능/일관성 : mtime + TTL(기본 30s) 캐시 → “최대 1분 내 변경 반영” 요건 충족.
필요하면 같은 방식으로 http 섹션(헤더/쿼리/타임아웃)도 병합해서 모델에 넣을 수 있게 확장해줄게!