From 86524a65f6fa64bf9afc1ec9d248879eaab2ffe5 Mon Sep 17 00:00:00 2001
From: Henry Li <henry1943@163.com>
Date: Wed, 14 Jan 2026 07:16:07 +0800
Subject: [PATCH] feat: add reflection modules

---
 backend/src/reflection/__init__.py  |  3 ++
 backend/src/reflection/resolvers.py | 81 +++++++++++++++++++++++++++++
 2 files changed, 84 insertions(+)
 create mode 100644 backend/src/reflection/__init__.py
 create mode 100644 backend/src/reflection/resolvers.py

diff --git a/backend/src/reflection/__init__.py b/backend/src/reflection/__init__.py
new file mode 100644
index 0000000..8098439
--- /dev/null
+++ b/backend/src/reflection/__init__.py
@@ -0,0 +1,3 @@
+from .resolvers import resolve_class, resolve_variable
+
+__all__ = ["resolve_class", "resolve_variable"]
diff --git a/backend/src/reflection/resolvers.py b/backend/src/reflection/resolvers.py
new file mode 100644
index 0000000..3b73a10
--- /dev/null
+++ b/backend/src/reflection/resolvers.py
@@ -0,0 +1,81 @@
+from importlib import import_module
+from typing import Type, TypeVar
+
+T = TypeVar("T")
+
+
+def resolve_variable[T](
+    variable_path: str,
+    expected_type: Type[T] | tuple[Type, ...] | None = None,
+) -> T:
+    """Resolve a variable from a path.
+
+    Args:
+        variable_path: The path to the variable (e.g. "parent_package_name.sub_package_name.module_name:variable_name").
+        expected_type: Optional type or tuple of types to validate the resolved variable against.
+            If provided, uses isinstance() to check if the variable is an instance of the expected type(s).
+
+    Returns:
+        The resolved variable.
+
+    Raises:
+        ImportError: If the module path is invalid or the attribute doesn't exist.
+        ValueError: If the resolved variable doesn't pass the validation checks.
+    """
+    try:
+        module_path, variable_name = variable_path.rsplit(":", 1)
+    except ValueError as err:
+        raise ImportError(
+            f"{variable_path} doesn't look like a variable path. Example: parent_package_name.sub_package_name.module_name:variable_name"
+        ) from err
+
+    try:
+        module = import_module(module_path)
+    except ImportError as err:
+        raise ImportError(f"Could not import module {module_path}") from err
+
+    try:
+        variable = getattr(module, variable_name)
+    except AttributeError as err:
+        raise ImportError(
+            f"Module {module_path} does not define a {variable_name} attribute/class"
+        ) from err
+
+    # Type validation
+    if expected_type is not None:
+        if not isinstance(variable, expected_type):
+            type_name = (
+                expected_type.__name__
+                if isinstance(expected_type, type)
+                else " or ".join(t.__name__ for t in expected_type)
+            )
+            raise ValueError(
+                f"{variable_path} is not an instance of {type_name}, got {type(variable).__name__}"
+            )
+
+    return variable
+
+
+def resolve_class(class_path: str, base_class: Type[T] | None = None) -> Type[T]:
+    """Resolve a class from a module path and class name.
+
+    Args:
+        class_path: The path to the class (e.g. "langchain_openai:ChatOpenAI").
+        base_class: The base class to check if the resolved class is a subclass of.
+
+    Returns:
+        The resolved class.
+
+    Raises:
+        ImportError: If the module path is invalid or the attribute doesn't exist.
+        ValueError: If the resolved object is not a class or not a subclass of base_class.
+    """
+    model_class = resolve_variable(class_path, expected_type=type)
+
+    if not isinstance(model_class, type):
+        raise ValueError(f"{class_path} is not a valid class")
+
+    if base_class is not None and not issubclass(model_class, base_class):
+        raise ValueError(f"{class_path} is not a subclass of {base_class.__name__}")
+
+    return model_class