""" 节点配置解析工具 从节点配置或上下文变量中解析 object / list,避免 dict 被 stringify 后无法反序列化。 """ import ast import json import re from typing import Any, List, Optional from ai_platform.nodes.base import NodeContext _VAR_REF_PATTERN = re.compile(r'^\{\{\s*([^}]+)\s*\}\}$') def _parse_structured_string(value: str) -> Any: text = value.strip() if not text: return None try: return json.loads(text) except (json.JSONDecodeError, TypeError): pass try: return ast.literal_eval(text) except (ValueError, SyntaxError): pass return value def resolve_config_value(context: NodeContext, raw: Any) -> Any: """解析配置值:支持 dict/list 直传、{{var}} 引用、JSON/Python 字面量字符串。""" if raw is None or raw == '': return None if isinstance(raw, (dict, list)): return raw if not isinstance(raw, str): return raw stripped = raw.strip() var_match = _VAR_REF_PATTERN.match(stripped) if var_match: var_name = var_match.group(1).strip() if var_name in context.variables: return context.variables[var_name] resolved = context.resolve_template(raw) if isinstance(resolved, (dict, list)): return resolved if isinstance(resolved, str): return _parse_structured_string(resolved) return resolved def resolve_object_config(context: NodeContext, raw: Any) -> Optional[dict]: value = resolve_config_value(context, raw) return value if isinstance(value, dict) else None def resolve_list_config(context: NodeContext, raw: Any) -> Optional[List[Any]]: value = resolve_config_value(context, raw) return value if isinstance(value, list) else None