⚡️ Speed up method ResultData.serialize_results by 22% in PR #6028 (PlaygroundPage)

[codeflash-ai]

⚡️ This pull request contains optimizations for PR #6028

If you approve this dependent PR, these changes will be merged into the original PR branch PlaygroundPage.

This PR will be automatically closed if the original PR is merged.

---

📄 22% (0.22x) speedup for ResultData.serialize_results in src/backend/base/langflow/graph/schema.py

⏱️ Runtime : 25.3 microseconds → 20.8 microseconds (best of 258 runs)

📝 Explanation and details

To optimize the provided Python code for faster execution, we should focus on reducing redundant checks, minimizing the calls to serialization functions, and utilizing efficient data structures and algorithms. The core parts of the code revolve around the serialize function and the ResultData model serialization.

Here is a revised version of the code with optimizations.

1. Optimize serialize function.

   • Directly return primitive types early to avoid unnecessary function calls.
   • Use specific type checks before attempting to dispatch through the _serialize_dispatcher.
   • Minimize instance checks only for types that need special handling.

2. Optimize the serialize_results method.

   • Use list comprehension where possible for faster iteration.

Changes made.

• Direct Primitive Return: Directly return primitive types early in the serialize function.
• Simplified Type Handling: Reduced unnecessary checks and utilized more direct type expressions.
• List Comprehension Optimization: Leveraged dictionary comprehension for better performance.

These optimizations will enhance performance by reducing the overhead of redundant operations and leveraging the efficiency of Python's built-in types and comprehensions.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 16 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	undefined

🌀 Generated Regression Tests Details

from typing import Any, List

# imports
import pytest  # used for our unit tests
from langflow.graph.schema import ResultData
# function to test
from langflow.serialization import serialize
from langflow.serialization.constants import MAX_ITEMS_LENGTH, MAX_TEXT_LENGTH
from loguru import logger
from pydantic import BaseModel, field_serializer
from pydantic.v1 import BaseModel as BaseModelV1


class SimpleModel(BaseModel):
    field: str

class NestedModel(BaseModel):
    field: SimpleModel

MyList = List[int]

class MyClass:
    pass
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

from typing import Any

# imports
import pytest  # used for our unit tests
from langflow.graph.schema import ResultData
from langflow.serialization.constants import MAX_ITEMS_LENGTH, MAX_TEXT_LENGTH
from loguru import logger
from pydantic import BaseModel, field_serializer
from pydantic.v1 import BaseModel as BaseModelV1


# Complex Data Types
class CustomClass:
    def __repr__(self):
        return "CustomClass()"

def test_serialize_custom_class():
    obj = CustomClass()

class PydanticModel(BaseModel):
    field: str

def test_serialize_pydantic_model():
    obj = PydanticModel(field="value")

def test_serialize_type_alias():
    from typing import List

def test_serialize_generic_type():
    from typing import Optional

def test_serialize_long_string():
    long_string = "a" * 10000

def test_serialize_large_list():
    large_list = list(range(1000))

class BadRepr:
    def __repr__(self):
        raise ValueError("Bad repr")

def test_serialize_bad_repr():
    obj = BadRepr()

# Recursive Structures
def test_serialize_self_referencing_list():
    a = []
    a.append(a)

def test_serialize_self_referencing_dict():
    b = {}
    b["self"] = b

# Performance and Scalability
def test_serialize_large_nested_dict():
    large_nested_dict = {"level1": {"level2": {"level3": "value"}}}

def test_serialize_list_of_dicts():
    list_of_dicts = [{"key": "value"}] * 1000

# Custom Serialization Logic
def test_serialize_result_data_simple_dict():
    class TestResultData(ResultData):
        results: dict

    obj = TestResultData(results={"key": "value"})
    codeflash_output = obj.serialize_results(obj.results)

def test_serialize_result_data_nested_dict():
    class TestResultData(ResultData):
        results: dict

    obj = TestResultData(results={"outer_key": {"inner_key": "inner_value"}})
    codeflash_output = obj.serialize_results(obj.results)

def test_serialize_result_data_list():
    class TestResultData(ResultData):
        results: list

    obj = TestResultData(results=[1, 2, 3])
    codeflash_output = obj.serialize_results(obj.results)

# Optional Parameters
def test_serialize_truncate_long_string():
    long_string = "a" * 10000

def test_serialize_truncate_large_list():
    large_list = list(range(1000))

# Logging and Debugging