[SOT] Add collections.namedtuple support (#72495)

SigureMo · Copilot · web-flow · commit b5f449e832b6 · 2025-04-26T13:55:41.000+08:00
* [SOT] Add namedtuple support

* Update python/paddle/jit/sot/opcode_translator/executor/variables/container.py

Co-authored-by: Copilot &lt;175728472+Copilot@users.noreply.github.com&gt;

* refine type

---------

Co-authored-by: Copilot &lt;175728472+Copilot@users.noreply.github.com&gt;
diff --git a/python/paddle/jit/sot/opcode_translator/executor/variables/base.py b/python/paddle/jit/sot/opcode_translator/executor/variables/base.py
@@ -16,6 +16,7 @@
 
 import inspect
 import operator
+from contextlib import contextmanager
 from functools import cached_property
 from queue import Queue
 from typing import TYPE_CHECKING, Any, Callable, Optional
@@ -654,3 +655,45 @@ def from_value(
         if isinstance(value, VariableBase):
             return value
         return None
+
+
+@contextmanager
+def signature_clear_guard(fn, name):
+    if not hasattr(fn, name):
+        yield
+    else:
+        saved_attr = getattr(fn, name)
+        delattr(fn, name)
+        yield
+        setattr(fn, name, saved_attr)
+
+
+def fn_bind_inputs(
+    fn: Callable[..., Any],
+    graph: FunctionGraph,
+    *args: Any,
+    **kwargs: Any,
+):
+    # temparay clear the fn.__signature__ to avoid signature check error
+    with signature_clear_guard(fn, "__signature__"), signature_clear_guard(
+        fn, "__wrapped__"
+    ):
+        sig = inspect.signature(fn)
+        bound_args = sig.bind(*args, **kwargs)
+    bound_args.apply_defaults()
+    parameters = {}
+    for name, value in bound_args.arguments.items():
+        assert name in sig.parameters
+        # Convert varargs and kwargs to Variable
+        if sig.parameters[name].kind == inspect.Parameter.VAR_POSITIONAL:
+            tracker = DummyTracker(value)
+        elif sig.parameters[name].kind == inspect.Parameter.VAR_KEYWORD:
+            tracker = DummyTracker(list(value.values()))
+        # Convert default args to Variable
+        elif not isinstance(value, VariableBase):
+            tracker = ConstTracker(value)
+        else:
+            tracker = value.tracker
+        value = VariableFactory.from_value(value, graph, tracker)
+        parameters[name] = value
+    return parameters
diff --git a/python/paddle/jit/sot/opcode_translator/executor/variables/callable.py b/python/paddle/jit/sot/opcode_translator/executor/variables/callable.py
@@ -41,6 +41,7 @@
     is_break_graph_tensor_methods,
     is_builtin_fn,
     is_directly_run_api,
+    is_namedtuple_class,
     is_not_supported_paddle_layer,
     is_paddle_api,
     log,
@@ -86,6 +87,7 @@
 from .base import (
     VariableBase,
     VariableFactory,
+    fn_bind_inputs,
 )
 from .basic import (
     ConstantVariable,
@@ -1077,3 +1079,33 @@ def from_value(value: Any, graph: FunctionGraph, tracker: Tracker):
         if inspect.isclass(value) and value in PD_PURE_CLASSES:
             return PureClassVariable(value, graph, tracker)
         return None
+
+
+class NamedTupleClassVariable(ClassVariable):
+    def __init__(
+        self, class_: type[Any], graph: FunctionGraph, tracker: Tracker
+    ):
+        super().__init__(class_, graph, tracker)
+
+    def call_function(self, /, *args, **kwargs):
+        from .container import NamedTupleVariable
+
+        parameters = fn_bind_inputs(self.value, self.graph, *args, **kwargs)
+        fields = self.get_py_value()._fields
+        assert all(
+            field in parameters for field in fields
+        ), f"All fields of namedtuple should be in parameters, but got parameter {parameters} and fields {fields}"
+
+        parameters_tuple = tuple(parameters[field] for field in fields)
+        return NamedTupleVariable(
+            parameters_tuple,
+            self.get_py_value(),
+            self.graph,
+            DummyTracker([self, *args, *kwargs.values()]),
+        )
+
+    @VariableFactory.register_from_value(successor="ClassVariable")
+    def from_value(value: Any, graph: FunctionGraph, tracker: Tracker):
+        if is_namedtuple_class(value):
+            return NamedTupleClassVariable(value, graph, tracker)
+        return None
diff --git a/python/paddle/jit/sot/opcode_translator/executor/variables/container.py b/python/paddle/jit/sot/opcode_translator/executor/variables/container.py
@@ -21,7 +21,7 @@
 
 import paddle
 
-from ....utils import ConstTypes
+from ....utils import ConstTypes, is_namedtuple_class
 from ....utils.exceptions import FallbackError, InnerError
 from ..dispatcher import Dispatcher
 from ..guard import (
@@ -683,6 +683,49 @@ def from_value(value: Any, graph: FunctionGraph, tracker: Tracker):
         return None
 
 
+class NamedTupleVariable(TupleVariable):
+    def __init__(
+        self,
+        val_tuple: tuple[VariableBase, ...],
+        cls: type[Any],
+        graph: FunctionGraph,
+        tracker: Tracker,
+    ):
+        super().__init__(val_tuple, graph, tracker)
+        self.cls = cls
+        self.fields = cls._fields
+
+    def getattr(self, name: str, default=None):
+        from .callable import BuiltinVariable
+
+        if default is not None:
+            raise FallbackError(
+                "default argument for getattr is not implemented"
+            )
+
+        if name == "_fields":
+            return VariableFactory.from_value(
+                self.fields, self.graph, DummyTracker([self])
+            )
+
+        if name in self.fields:
+            idx = self.fields.index(name)
+            idx_var = ConstantVariable(idx, self.graph, DummyTracker([self]))
+            return BuiltinVariable(
+                operator.getitem, self.graph, DanglingTracker()
+            ).bind_dangling_fn(self, name)(idx_var)
+        return super().getattr(name, default)
+
+    def get_py_type(self):
+        return self.cls
+
+    @VariableFactory.register_from_value()
+    def from_value(value: Any, graph: FunctionGraph, tracker: Tracker):
+        if is_namedtuple_class(type(value)):
+            return NamedTupleVariable(value, type(value), graph, tracker)
+        return None
+
+
 class RangeVariable(ContainerVariable):
     """
     RangeVariable is a wrapper for range.
diff --git a/python/paddle/jit/sot/opcode_translator/executor/virtual_frame.py b/python/paddle/jit/sot/opcode_translator/executor/virtual_frame.py
@@ -15,9 +15,7 @@
 from __future__ import annotations
 
 import builtins
-import inspect
 import re
-from contextlib import contextmanager
 from typing import TYPE_CHECKING, Any, NamedTuple
 
 import paddle
@@ -29,12 +27,11 @@
     CellTracker,
     ConstTracker,
     DanglingTracker,
-    DummyTracker,
     LocalTracker,
     Tracker,
 )
 from .variable_stack import VariableStack
-from .variables.base import VariableBase, VariableFactory
+from .variables.base import VariableBase, VariableFactory, fn_bind_inputs
 from .variables.basic import (
     CellVariable,
     FunctionGlobalVariable,
@@ -116,17 +113,6 @@ def __repr__(self) -> str:
         return f"FunctionClosureTracker(fn={self.fn}, idx={self.idx})"
 
 
-@contextmanager
-def signature_clear_guard(fn, name):
-    if not hasattr(fn, name):
-        yield
-    else:
-        saved_attr = getattr(fn, name)
-        delattr(fn, name)
-        yield
-        setattr(fn, name, saved_attr)
-
-
 def validate_value(value):
     assert isinstance(
         value, VariableBase
@@ -238,27 +224,9 @@ def from_inline_call(
             )
 
         # convert locals
-        # temparay clear the fn.__signature__ to avoid signature check error
-        with signature_clear_guard(
-            fn_value, "__signature__"
-        ), signature_clear_guard(fn_value, "__wrapped__"):
-            sig = inspect.signature(fn_value)
-            bound_args = sig.bind(*call_args, **call_kwargs)
-        bound_args.apply_defaults()
-        for name, value in bound_args.arguments.items():
-            assert name in sig.parameters
-            # Convert varargs and kwargs to Variable
-            if sig.parameters[name].kind == inspect.Parameter.VAR_POSITIONAL:
-                tracker = DummyTracker(value)
-            elif sig.parameters[name].kind == inspect.Parameter.VAR_KEYWORD:
-                tracker = DummyTracker(list(value.values()))
-            # Convert default args to Variable
-            elif not isinstance(value, VariableBase):
-                tracker = ConstTracker(value)
-            else:
-                tracker = value.tracker
-            value = VariableFactory.from_value(value, graph, tracker)
-            vframe.locals[name] = value
+        vframe.locals.update(
+            fn_bind_inputs(fn_value, graph, *call_args, **call_kwargs)
+        )
 
         log(
             5,
diff --git a/python/paddle/jit/sot/utils/__init__.py b/python/paddle/jit/sot/utils/__init__.py
@@ -93,6 +93,7 @@
     is_break_graph_api,
     is_builtin_fn,
     is_comprehensive_name,
+    is_namedtuple_class,
     is_paddle_api,
     is_strict_mode,
     list_contain_by_id,
diff --git a/python/paddle/jit/sot/utils/utils.py b/python/paddle/jit/sot/utils/utils.py
@@ -231,6 +231,17 @@ def is_break_graph_api(func):
     return func in break_graph_set
 
 
+def is_namedtuple_class(cls):
+    if not inspect.isclass(cls):
+        return False
+    if not issubclass(cls, tuple):
+        return False
+    # The signature created by nametuple function
+    namedtuple_attrs = {"_make", "_asdict", "_fields", "_replace"}
+    cls_attrs = set(dir(cls))
+    return namedtuple_attrs.issubset(cls_attrs)
+
+
 def map_if(
     *structures: NestedStructure[T1],
     pred: Callable[[T1], bool],
diff --git a/test/sot/test_collections_namedtuple.py b/test/sot/test_collections_namedtuple.py
@@ -0,0 +1,97 @@
+# Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import unittest
+from collections import namedtuple
+
+from test_case_base import TestCaseBase
+
+import paddle
+from paddle.jit.sot import psdb
+
+Singleton = namedtuple("Singleton", ["a"])
+Pair = namedtuple("Pair", ["a", "b"])
+Triplet = namedtuple("Triple", ["a", "b", "c"])
+Quartet = namedtuple("Quartet", ["a", "b", "c", "d"])
+QuartetWithDefault = namedtuple(
+    "QuartetWithDefault", ["a", "b", "c", "d"], defaults=[1, 2, 3]
+)
+
+
+@psdb.check_no_breakgraph
+def create_namedtuple():
+    x = Singleton(1)
+    y = Singleton(2)
+    z = Singleton(3)
+    return x.a + y.a + z.a
+
+
+@psdb.check_no_breakgraph
+def create_namedtuple_with_tensor(x):
+    y = Pair(x + 1, x + 2)
+    z = Pair(x + 3, x + 4)
+    return y.a + y.b + z.a + z.b
+
+
+@psdb.check_no_breakgraph
+def load_namedtuple(x, y):
+    return x.a + y.a + y.b
+
+
+@psdb.check_no_breakgraph
+def create_namedtuple_with_default():
+    u = QuartetWithDefault(1, 2, 3, 4)
+    v = QuartetWithDefault(5, 6, 7)
+    w = QuartetWithDefault(8, 9)
+    x = QuartetWithDefault(10)
+    return (
+        (u.a + u.b + u.c + u.d)
+        + (v.a + v.b + v.c + v.d)
+        + (w.a + w.b + w.c + w.d)
+        + (x.a + x.b + x.c + x.d)
+    )
+
+
+@psdb.check_no_breakgraph
+def get_fields():
+    x = Singleton(1)
+    y = Pair(2, 3)
+    z = Triplet(4, 5, 6)
+    return x._fields + y._fields + z._fields
+
+
+class TestNamedTuple(TestCaseBase):
+    def test_create_namedtuple(self):
+        self.assert_results(create_namedtuple)
+
+    def test_create_namedtuple_with_tensor(self):
+        x = paddle.to_tensor(1)
+        self.assert_results(create_namedtuple_with_tensor, x)
+
+    def test_load_namedtuple(self):
+        x = Singleton(paddle.to_tensor(1))
+        y = Pair(paddle.to_tensor(2), paddle.to_tensor(3))
+        self.assert_results(load_namedtuple, x, y)
+
+    def test_create_namedtuple_with_default(self):
+        self.assert_results(create_namedtuple_with_default)
+
+    def test_get_fields(self):
+        self.assert_results(get_fields)
+
+
+if __name__ == "__main__":
+    unittest.main()
