feat: Add file config resource abastraction (#271)

.changeset/sixty-clouds-run.md

@@ -0,0 +1,5 @@
+---
+"llama-index-workflows": patch
+---
+
+Add ResourceConfig for resource-level configuration injection

packages/llama-index-workflows/src/workflows/resource.py

@@ -3,14 +3,22 @@
 
 from __future__ import annotations
 
+import functools
 import inspect
+import json
+from pathlib import Path
 from typing import (
+    Annotated,
     Any,
     Awaitable,
     Callable,
     Generic,
+    Optional,
+    Type,
     TypeVar,
     cast,
+    get_args,
+    get_origin,
 )
 
 from pydantic import (
@@ -19,6 +27,7 @@ from pydantic import (
 )
 
 T = TypeVar("T")
+B = TypeVar("B", bound=BaseModel)
 
 
 class _Resource(Generic[T]):
@@ -33,16 +42,118 @@ class _Resource(Generic[T]):
         self._is_async = inspect.iscoroutinefunction(factory)
         self.name = getattr(factory, "__qualname__", type(factory).__name__)
         self.cache = cache
+        self.resource_configs: Optional[dict[str, BaseModel]] = None  # noqa: UP045
+
+    def prepare_resource_configs(self) -> None:
+        if self.resource_configs is None:
+            params = inspect.signature(self._factory).parameters
+            resource_configs: dict[str, BaseModel] = {}
+            if len(params) > 0:
+                for param in params.values():
+                    if get_origin(param.annotation) is Annotated:
+                        args = get_args(param.annotation)
+                        if len(args) == 2 and isinstance(args[1], _ResourceConfig):
+                            resource_config = args[1]
+                            resource_config.cls_factory = args[0]
+                            value = resource_config.call()
+                            resource_configs.update({param.name: value})
+            self.resource_configs = resource_configs
+        return None
 
     async def call(self) -> T:
         """Invoke the underlying factory, awaiting if necessary."""
+        self.prepare_resource_configs()
+        args = cast(dict[str, BaseModel], self.resource_configs)
         if self._is_async:
-            result = await cast(Callable[..., Awaitable[T]], self._factory)()
+            result = await cast(Callable[..., Awaitable[T]], self._factory)(**args)
         else:
-            result = cast(Callable[..., T], self._factory)()
+            result = cast(Callable[..., T], self._factory)(**args)
         return result
 
 
+@functools.lru_cache(maxsize=1)
+def _get_resource_config_data(
+    config_file: str,
+    path_selector: str | None,
+) -> dict[str, Any]:
+    with open(config_file, "r") as f:
+        data = json.load(f)
+    if path_selector is not None:
+        keys = path_selector.split(".")
+        val: dict[str, Any] = data
+        cumulative_path = ""
+        for key in keys:
+            cumulative_path += key + "."
+            got = cast(Optional[dict[str, Any]], val.get(key))  # noqa: UP045
+            if not isinstance(got, dict):
+                raise ValueError(
+                    f"Expected dictionary for configuration from {config_file} at path {cumulative_path.strip('.')}, got: {type(got)}"
+                )
+            val = got
+        return val
+    return data
+
+
+class _ResourceConfig(Generic[B]):
+    """
+    Internal wrapper for a pydantic-based resource whose configuration can be read from a JSON file.
+    """
+
+    def __init__(
+        self,
+        config_file: str,
+        path_selector: str | None,
+        cls_factory: Type[B] | None = None,
+    ) -> None:
+        if not Path(config_file).is_file():
+            raise FileNotFoundError(f"No such file: {config_file}")
+        if Path(config_file).suffix != ".json":
+            raise ValueError(
+                "Only JSON files can be used to load Pydantic-based resources."
+            )
+        self.config_file = config_file
+        self.path_selector = path_selector
+        self.cls_factory = cls_factory
+
+    @property
+    def name(self) -> str:
+        if self.path_selector is not None:
+            return self.config_file + "." + self.path_selector
+        return self.config_file
+
+    # make async for compatibility with _Resource
+    def call(self) -> B:
+        sel_data = _get_resource_config_data(
+            config_file=self.config_file, path_selector=self.path_selector
+        )
+        # let validation error bubble up
+        if self.cls_factory is not None:
+            return self.cls_factory.model_validate(sel_data)
+        else:
+            raise ValueError(
+                "Class factory should be set to a BaseModel subclass before calling"
+            )
+
+
+def ResourceConfig(
+    config_file: str,
+    path_selector: str | None = None,
+) -> _ResourceConfig:
+    """
+    Wrapper for a _ResourceConfig.
+
+    Attributes:
+        config_file (str): JSON file where the configuration is stored
+        path_selector (str | None): Path selector to retrieve a specific value from the JSON map
+        cache (bool): Cache the resource's value to avoid re-computation.
+
+    Returns:
+        _ResourceConfig: A configured resource representation
+    """
+
+    return _ResourceConfig(config_file=config_file, path_selector=path_selector)
+
+
 class ResourceDefinition(BaseModel):
     """Definition for a resource injection requested by a step signature.
 
@@ -58,11 +169,14 @@ class ResourceDefinition(BaseModel):
     type_annotation: Any = None
 
 
-def Resource(factory: Callable[..., T], cache: bool = True) -> _Resource[T]:
+def Resource(
+    factory: Callable[..., T],
+    cache: bool = True,
+) -> _Resource:
     """Declare a resource to inject into step functions.
 
     Args:
-        factory (Callable[..., T]): Function returning the resource instance. May be async.
+        factory (Callable[..., T] | None): Function returning the resource instance. May be async.
         cache (bool): If True, reuse the produced resource across steps. Defaults to True.
 
     Returns:

packages/llama-index-workflows/tests/test_resources.py

@@ -1,6 +1,8 @@
 # SPDX-License-Identifier: MIT
 # Copyright (c) 2026 LlamaIndex Inc.
 
+import json
+from pathlib import Path
 from typing import Annotated, Optional
 from unittest import mock
 
@@ -8,7 +10,13 @@ import pytest
 from pydantic import BaseModel, Field
 from workflows.decorators import step
 from workflows.events import Event, StartEvent, StopEvent
-from workflows.resource import Resource, ResourceManager
+from workflows.resource import (
+    Resource,
+    ResourceConfig,
+    ResourceManager,
+    _Resource,
+    _ResourceConfig,
+)
 from workflows.workflow import Workflow
 
 # Global counters used in resource workflow tests
@@ -41,6 +49,157 @@ class MessageStopEvent(StopEvent):
     llm_response: Optional[str] = Field(default=None)
 
 
+class FileData(BaseModel):
+    file: str
+    permission_mode: str
+
+
+class FileOperator:
+    def __init__(self, data: FileData) -> None:
+        self.file = data.file
+        self.permission_mode = data.permission_mode
+
+    def operate(self) -> str:
+        if self.permission_mode == "r":
+            with open(self.file, self.permission_mode) as f:
+                return f.read()
+        elif self.permission_mode == "w":
+            with open(self.file, self.permission_mode) as f:
+                f.write("hello world!")
+                return "hello world!"
+        else:
+            raise ValueError(f"Unsupported operation: {self.permission_mode}")
+
+
+class ChatMessages(BaseModel):
+    messages: list[str]
+
+
+class Fs(BaseModel):
+    files: list[str]
+    dirs: list[str]
+
+
+@pytest.mark.asyncio
+async def test_function_resource_init() -> None:
+    def get_string() -> str:
+        return "string"
+
+    retval = Resource(get_string)
+    assert isinstance(retval, _Resource)
+    assert "get_string" in retval.name
+    assert retval.cache
+    assert not retval._is_async
+
+    result = await retval.call()
+    assert result == "string"
+
+
+def test_resource_config_init(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.chdir(tmp_path)
+
+    data = {"messages": ["hello"]}
+
+    with open("config.json", "w") as f:
+        json.dump(data, f)
+
+    retval = ResourceConfig(config_file="config.json")
+    assert isinstance(retval, _ResourceConfig)
+    assert retval.path_selector is None
+    assert retval.config_file == "config.json"
+    assert retval.cls_factory is None
+    assert retval.name == "config.json"
+
+    # modify path selector, modify name
+    retval.path_selector = "hello.world"
+    assert retval.name == "config.json.hello.world"
+
+    retval.path_selector = None
+
+    with pytest.raises(
+        ValueError,
+        match="Class factory should be set to a BaseModel subclass before calling",
+    ):
+        retval.call()
+
+    # define a cls_factory for the resource to be called
+    retval.cls_factory = ChatMessages
+
+    result = retval.call()
+    assert isinstance(result, ChatMessages)
+    assert result.messages == ["hello"]
+
+
+def test_resource_config_path_selector(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    data = {
+        "memory": {"messages": ["hello"]},
+        "core": {"fs": {"files": ["hello.py"], "dirs": ["hello/"]}},
+    }
+
+    with open("config.json", "w") as f:
+        json.dump(data, f)
+
+    resource = ResourceConfig(config_file="config.json", path_selector="memory")
+    assert resource.name == "config.json.memory"
+    resource.cls_factory = ChatMessages
+    value = resource.call()
+    assert isinstance(value, ChatMessages)
+    assert value.messages == ["hello"]
+    resource.path_selector = "core.fs"
+    assert resource.name == "config.json.core.fs"
+    resource.cls_factory = Fs
+    value = resource.call()
+    assert isinstance(value, Fs)
+    assert value.files == ["hello.py"]
+    assert value.dirs == ["hello/"]
+
+
+def test_resource_config_path_selector_error(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    data = {
+        "core": {"fs": {"files": ["hello.py"], "dirs": ["hello/"]}},
+    }
+
+    with open("config.json", "w") as f:
+        json.dump(data, f)
+
+    # path selector does not return a dict
+    resource = ResourceConfig(config_file="config.json", path_selector="core.fs.files")
+    resource.cls_factory = Fs
+    with pytest.raises(
+        ValueError,
+        match=r"Expected dictionary for configuration from config.json at path core.fs.files, got: .*",
+    ):
+        resource.call()
+
+    # path selector does not exist
+    resource.path_selector = "core.filesystem"
+    with pytest.raises(
+        ValueError,
+        match=r"Expected dictionary for configuration from config.json at path core.filesystem, got: .*",
+    ):
+        resource.call()
+
+    # error occurs before reaching the end of the path_selector
+    # (tests the not the full path_selector is shown, but only up to the item with the error)
+    resource.path_selector = "core.filesystem.fs"
+    with pytest.raises(
+        ValueError,
+        match=r"Expected dictionary for configuration from config.json at path core.filesystem, got: .*",
+    ):
+        resource.call()
+
+
 @pytest.mark.asyncio
 async def test_resource() -> None:
     m = Memory.from_defaults("user_id_123", token_limit=60000)
@@ -93,6 +252,47 @@ async def test_resource_async() -> None:
     m.put.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_resource_config(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.chdir(tmp_path)
+
+    data = {"file": "hello.py", "permission_mode": "r"}
+
+    with open("config.json", "w") as f:
+        json.dump(data, f)
+
+    with open("hello.py", "w") as f:
+        f.write("print('hello')")
+
+    def get_file_operator(
+        config: Annotated[FileData, ResourceConfig(config_file="config.json")],
+    ) -> FileOperator:
+        return FileOperator(data=config)
+
+    class TestWorkflow(Workflow):
+        @step
+        def start_step(self, ev: StartEvent) -> SecondEvent:
+            print("Start step is done", flush=True)
+            return SecondEvent(msg="Hello")
+
+        @step
+        def f1(
+            self,
+            ev: SecondEvent,
+            file_operator: Annotated[FileOperator, Resource(get_file_operator)],
+        ) -> StopEvent:
+            assert file_operator.file == "hello.py"
+            assert file_operator.permission_mode == "r"
+            assert file_operator.operate() == "print('hello')"
+            return StopEvent(result=None)
+
+    wf = TestWorkflow(disable_validation=True)
+    await wf.run()
+
+
 @pytest.mark.asyncio
 async def test_caching_behavior() -> None:
     class CounterThing:
@@ -132,7 +332,7 @@ async def test_caching_behavior() -> None:
     await wf_1.run()
     assert (
         cc == 2  # type: ignore
-    )  # this is expected to be 2, as it is a cached resource shared by test_step and test_step_2, which means at test_step it counter_thing.counter goes from 0 to 1 and at test_step_2 goes from 1 to 2
+    )  # this is expected to be 2, as it is a cached resource shared by test_step and test_step_2, which means at test_step counter_thing.counter goes from 0 to 1 and at test_step_2 goes from 1 to 2
 
     wf_2 = TestWorkflow(disable_validation=True)
     await wf_2.run()
@@ -141,6 +341,57 @@ async def test_caching_behavior() -> None:
     )  # the cache is workflow-specific, so since wf_2 is different from wf_1, we expect no interference between the two
 
 
+@pytest.mark.asyncio
+async def test_caching_behavior_resource_configs(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.chdir(tmp_path)
+
+    data = {"file": "hello.py", "permission_mode": "r"}
+    data_1 = {"file": "bye.py", "permission_mode": "w"}
+
+    with open("config.json", "w") as f:
+        json.dump(data, f)
+
+    def get_file_operator(
+        config: Annotated[FileData, ResourceConfig(config_file="config.json")],
+    ) -> FileOperator:
+        return FileOperator(data=config)
+
+    class TestWorkflow(Workflow):
+        @step
+        def start_step(
+            self,
+            ev: StartEvent,
+            file: Annotated[FileOperator, Resource(get_file_operator)],
+        ) -> SecondEvent:
+            print("first step")
+            assert file.file == "hello.py"
+            assert file.permission_mode == "r"
+            # change config.json: the underlying resource has been cached, so will be unaffected
+            with open("config.json", "w") as f:
+                json.dump(data_1, f)
+            return SecondEvent(msg="Hello")
+
+        @step
+        def f1(
+            self,
+            ev: SecondEvent,
+            file: Annotated[FileOperator, Resource(get_file_operator)],
+        ) -> StopEvent:
+            print("second step")
+            # this resource has been cached,
+            # so even if config.json has changed,
+            # the resource remains the same
+            assert file.file == "hello.py"
+            assert file.permission_mode == "r"
+            return StopEvent()
+
+    wf = TestWorkflow()
+    await wf.run()
+
+
 @pytest.mark.asyncio
 async def test_non_caching_behavior() -> None:
     class CounterThing: