ai-flow-python/app/service/engine/parameter_resolver.py

import re
from typing import Any

from app.service.engine.execution_context import ExecutionContext

_ARRAY_COMBINE_RE = re.compile(r"^\[(.*)?\]$")


class ParameterResolver:
    def resolve(self, value: Any, context: ExecutionContext) -> Any:
        if value is None:
            return None

        if isinstance(value, str):
            m = _ARRAY_COMBINE_RE.match(value)
            if m:
                content = (m.group(1) or "").strip()
                if not content:
                    return []
                elements = content.split(",")
                result = []
                for elem in elements:
                    elem = elem.strip()
                    if elem:
                        resolved = self.resolve(elem, context)
                        if resolved is not None:
                            result.append(resolved)
                return result

            if value.startswith("$"):
                path = value[1:]
                return context.lookup(path)
            return value

        if isinstance(value, dict):
            return {
                str(k): self.resolve(v, context) for k, v in value.items()
            }

        if isinstance(value, list):
            return [self.resolve(v, context) for v in value]

        return value