Parse a submodule's function declaration from its YAML data.
(
yaml_data: Dict[str, Any],
module_path: str,
declared_name: Optional[str] = None,
)
| 38 | |
| 39 | |
| 40 | def parse_submodule_function_declaration( |
| 41 | yaml_data: Dict[str, Any], |
| 42 | module_path: str, |
| 43 | declared_name: Optional[str] = None, |
| 44 | ) -> SubmoduleFunctionSpec: |
| 45 | """Parse a submodule's function declaration from its YAML data.""" |
| 46 | func_decl = yaml_data.get("function") |
| 47 | if not func_decl or not isinstance(func_decl, dict): |
| 48 | raise ValueError( |
| 49 | f"Submodule {module_path} must define a top-level 'function' block" |
| 50 | ) |
| 51 | |
| 52 | func_name = func_decl.get("name") or declared_name |
| 53 | if not func_name: |
| 54 | raise ValueError(f"Submodule {module_path} function must define a name") |
| 55 | if declared_name and func_name != declared_name: |
| 56 | raise ValueError( |
| 57 | f"Submodule name mismatch: declared '{declared_name}' but function name is '{func_name}'" |
| 58 | ) |
| 59 | |
| 60 | description = func_decl.get("description") or func_decl.get("docstring") or "" |
| 61 | return_var = func_decl.get("return", "prev_output") |
| 62 | |
| 63 | params = func_decl.get("parameters", []) |
| 64 | normalized_params: List[SubmoduleParamSpec] = [] |
| 65 | |
| 66 | if isinstance(params, dict): |
| 67 | |
| 68 | def _normalize_param_list(items: Any, source: str) -> List[Dict[str, Any]]: |
| 69 | if items is None: |
| 70 | return [] |
| 71 | if not isinstance(items, list): |
| 72 | items = [items] |
| 73 | normalized: List[Dict[str, Any]] = [] |
| 74 | for item in items: |
| 75 | if isinstance(item, str): |
| 76 | normalized.append({"name": item, "source": source}) |
| 77 | elif isinstance(item, dict): |
| 78 | entry = dict(item) |
| 79 | entry["source"] = source |
| 80 | normalized.append(entry) |
| 81 | else: |
| 82 | raise ValueError(f"Invalid parameter spec in {module_path}: {item}") |
| 83 | return normalized |
| 84 | |
| 85 | model_params = _normalize_param_list(params.get("model"), "model") |
| 86 | context_params = _normalize_param_list(params.get("context"), "context") |
| 87 | params = model_params + context_params |
| 88 | |
| 89 | if isinstance(params, list): |
| 90 | for p in params: |
| 91 | if isinstance(p, str): |
| 92 | p = {"name": p} |
| 93 | if not isinstance(p, dict): |
| 94 | raise ValueError(f"Invalid parameter spec in {module_path}: {p}") |
| 95 | if not p.get("name"): |
| 96 | raise ValueError( |
| 97 | f"Submodule {module_path} has a parameter without a name" |