chore(core): add ruff rule PLR2004 (#33706)

Co-authored-by: Mason Daugherty <[email protected]>

Author: cbornet

libs/core/langchain_core/language_models/chat_models.py

@@ -1763,9 +1763,12 @@ def _gen_info_and_msg_metadata(
     }
 
 
+_MAX_CLEANUP_DEPTH = 100
+
+
 def _cleanup_llm_representation(serialized: Any, depth: int) -> None:
     """Remove non-serializable objects from a serialized object."""
-    if depth > 100:  # Don't cooperate for pathological cases
+    if depth > _MAX_CLEANUP_DEPTH:  # Don't cooperate for pathological cases
         return
 
     if not isinstance(serialized, dict):

libs/core/langchain_core/messages/utils.py

@@ -328,12 +328,16 @@ def _convert_to_message(message: MessageLikeRepresentation) -> BaseMessage:
     """
     if isinstance(message, BaseMessage):
         message_ = message
-    elif isinstance(message, str):
-        message_ = _create_message_from_message_type("human", message)
-    elif isinstance(message, Sequence) and len(message) == 2:
-        # mypy doesn't realise this can't be a string given the previous branch
-        message_type_str, template = message  # type: ignore[misc]
-        message_ = _create_message_from_message_type(message_type_str, template)
+    elif isinstance(message, Sequence):
+        if isinstance(message, str):
+            message_ = _create_message_from_message_type("human", message)
+        else:
+            try:
+                message_type_str, template = message
+            except ValueError as e:
+                msg = "Message as a sequence must be (role string, template)"
+                raise NotImplementedError(msg) from e
+            message_ = _create_message_from_message_type(message_type_str, template)
     elif isinstance(message, dict):
         msg_kwargs = message.copy()
         try:

libs/core/langchain_core/prompts/chat.py

@@ -1343,11 +1343,25 @@ def _create_template_from_message_type(
                 raise ValueError(msg)
             var_name = template[1:-1]
             message = MessagesPlaceholder(variable_name=var_name, optional=True)
-        elif len(template) == 2 and isinstance(template[1], bool):
-            var_name_wrapped, is_optional = template
+        else:
+            try:
+                var_name_wrapped, is_optional = template
+            except ValueError as e:
+                msg = (
+                    "Unexpected arguments for placeholder message type."
+                    " Expected either a single string variable name"
+                    " or a list of [variable_name: str, is_optional: bool]."
+                    f" Got: {template}"
+                )
+                raise ValueError(msg) from e
+
+            if not isinstance(is_optional, bool):
+                msg = f"Expected is_optional to be a boolean. Got: {is_optional}"
+                raise ValueError(msg)  # noqa: TRY004
+
             if not isinstance(var_name_wrapped, str):
                 msg = f"Expected variable name to be a string. Got: {var_name_wrapped}"
-                raise ValueError(msg)  # noqa:TRY004
+                raise ValueError(msg)  # noqa: TRY004
             if var_name_wrapped[0] != "{" or var_name_wrapped[-1] != "}":
                 msg = (
                     f"Invalid placeholder template: {var_name_wrapped}."
@@ -1357,14 +1371,6 @@ def _create_template_from_message_type(
             var_name = var_name_wrapped[1:-1]
 
             message = MessagesPlaceholder(variable_name=var_name, optional=is_optional)
-        else:
-            msg = (
-                "Unexpected arguments for placeholder message type."
-                " Expected either a single string variable name"
-                " or a list of [variable_name: str, is_optional: bool]."
-                f" Got: {template}"
-            )
-            raise ValueError(msg)
     else:
         msg = (
             f"Unexpected message type: {message_type}. Use one of 'human',"
@@ -1418,10 +1424,11 @@ def _convert_to_message_template(
                 )
                 raise ValueError(msg)
             message = (message["role"], message["content"])
-        if len(message) != 2:
+        try:
+            message_type_str, template = message
+        except ValueError as e:
             msg = f"Expected 2-tuple of (role, template), got {message}"
-            raise ValueError(msg)
-        message_type_str, template = message
+            raise ValueError(msg) from e
         if isinstance(message_type_str, str):
             message_ = _create_template_from_message_type(
                 message_type_str, template, template_format=template_format

libs/core/langchain_core/runnables/base.py

@@ -118,6 +118,8 @@
 
 Other = TypeVar("Other")
 
+_RUNNABLE_GENERIC_NUM_ARGS = 2  # Input and Output
+
 
 class Runnable(ABC, Generic[Input, Output]):
     """A unit of work that can be invoked, batched, streamed, transformed and composed.
@@ -309,15 +311,18 @@ def InputType(self) -> type[Input]:  # noqa: N802
         for base in self.__class__.mro():
             if hasattr(base, "__pydantic_generic_metadata__"):
                 metadata = base.__pydantic_generic_metadata__
-                if "args" in metadata and len(metadata["args"]) == 2:
+                if (
+                    "args" in metadata
+                    and len(metadata["args"]) == _RUNNABLE_GENERIC_NUM_ARGS
+                ):
                     return metadata["args"][0]
 
         # If we didn't find a Pydantic model in the parent classes,
         # then loop through __orig_bases__. This corresponds to
         # Runnables that are not pydantic models.
         for cls in self.__class__.__orig_bases__:  # type: ignore[attr-defined]
             type_args = get_args(cls)
-            if type_args and len(type_args) == 2:
+            if type_args and len(type_args) == _RUNNABLE_GENERIC_NUM_ARGS:
                 return type_args[0]
 
         msg = (
@@ -340,12 +345,15 @@ def OutputType(self) -> type[Output]:  # noqa: N802
         for base in self.__class__.mro():
             if hasattr(base, "__pydantic_generic_metadata__"):
                 metadata = base.__pydantic_generic_metadata__
-                if "args" in metadata and len(metadata["args"]) == 2:
+                if (
+                    "args" in metadata
+                    and len(metadata["args"]) == _RUNNABLE_GENERIC_NUM_ARGS
+                ):
                     return metadata["args"][1]
 
         for cls in self.__class__.__orig_bases__:  # type: ignore[attr-defined]
             type_args = get_args(cls)
-            if type_args and len(type_args) == 2:
+            if type_args and len(type_args) == _RUNNABLE_GENERIC_NUM_ARGS:
                 return type_args[1]
 
         msg = (
@@ -2750,6 +2758,9 @@ def _seq_output_schema(
     return last.get_output_schema(config)
 
 
+_RUNNABLE_SEQUENCE_MIN_STEPS = 2
+
+
 class RunnableSequence(RunnableSerializable[Input, Output]):
     """Sequence of `Runnable` objects, where the output of one is the input of the next.
 
@@ -2872,8 +2883,11 @@ def __init__(
                 steps_flat.extend(step.steps)
             else:
                 steps_flat.append(coerce_to_runnable(step))
-        if len(steps_flat) < 2:
-            msg = f"RunnableSequence must have at least 2 steps, got {len(steps_flat)}"
+        if len(steps_flat) < _RUNNABLE_SEQUENCE_MIN_STEPS:
+            msg = (
+                f"RunnableSequence must have at least {_RUNNABLE_SEQUENCE_MIN_STEPS} "
+                f"steps, got {len(steps_flat)}"
+            )
             raise ValueError(msg)
         super().__init__(
             first=steps_flat[0],
@@ -4477,7 +4491,7 @@ def get_input_schema(self, config: RunnableConfig | None = None) -> type[BaseMod
             # on itemgetter objects, so we have to parse the repr
             items = str(func).replace("operator.itemgetter(", "")[:-1].split(", ")
             if all(
-                item[0] == "'" and item[-1] == "'" and len(item) > 2 for item in items
+                item[0] == "'" and item[-1] == "'" and item != "''" for item in items
             ):
                 fields = {item[1:-1]: (Any, ...) for item in items}
                 # It's a dict, lol

libs/core/langchain_core/runnables/branch.py

@@ -36,6 +36,8 @@
     get_unique_config_specs,
 )
 
+_MIN_BRANCHES = 2
+
 
 class RunnableBranch(RunnableSerializable[Input, Output]):
     """`Runnable` that selects which branch to run based on a condition.
@@ -91,7 +93,7 @@ def __init__(
             TypeError: If a branch is not a `tuple` or `list`.
             ValueError: If a branch is not of length `2`.
         """
-        if len(branches) < 2:
+        if len(branches) < _MIN_BRANCHES:
             msg = "RunnableBranch requires at least two branches"
             raise ValueError(msg)
 
@@ -118,7 +120,7 @@ def __init__(
                 )
                 raise TypeError(msg)
 
-            if len(branch) != 2:
+            if len(branch) != _MIN_BRANCHES:
                 msg = (
                     f"RunnableBranch branches must be "
                     f"tuples or lists of length 2, not {len(branch)}"

libs/core/langchain_core/runnables/graph_mermaid.py

@@ -454,7 +454,10 @@ def _render_mermaid_using_api(
                 return img_bytes
 
             # If we get a server error (5xx), retry
-            if 500 <= response.status_code < 600 and attempt < max_retries:
+            if (
+                requests.codes.internal_server_error <= response.status_code
+                and attempt < max_retries
+            ):
                 # Exponential backoff with jitter
                 sleep_time = retry_delay * (2**attempt) * (0.5 + 0.5 * random.random())  # noqa: S311 not used for crypto
                 time.sleep(sleep_time)

libs/core/langchain_core/tools/base.py

@@ -872,16 +872,19 @@ def run(
                     tool_kwargs |= {config_param: config}
                 response = context.run(self._run, *tool_args, **tool_kwargs)
             if self.response_format == "content_and_artifact":
-                if not isinstance(response, tuple) or len(response) != 2:
-                    msg = (
-                        "Since response_format='content_and_artifact' "
-                        "a two-tuple of the message content and raw tool output is "
-                        f"expected. Instead generated response of type: "
-                        f"{type(response)}."
-                    )
+                msg = (
+                    "Since response_format='content_and_artifact' "
+                    "a two-tuple of the message content and raw tool output is "
+                    f"expected. Instead, generated response is of type: "
+                    f"{type(response)}."
+                )
+                if not isinstance(response, tuple):
                     error_to_raise = ValueError(msg)
                 else:
-                    content, artifact = response
+                    try:
+                        content, artifact = response
+                    except ValueError:
+                        error_to_raise = ValueError(msg)
             else:
                 content = response
         except (ValidationError, ValidationErrorV1) as e:
@@ -998,16 +1001,19 @@ async def arun(
                 coro = self._arun(*tool_args, **tool_kwargs)
                 response = await coro_with_context(coro, context)
             if self.response_format == "content_and_artifact":
-                if not isinstance(response, tuple) or len(response) != 2:
-                    msg = (
-                        "Since response_format='content_and_artifact' "
-                        "a two-tuple of the message content and raw tool output is "
-                        f"expected. Instead generated response of type: "
-                        f"{type(response)}."
-                    )
+                msg = (
+                    "Since response_format='content_and_artifact' "
+                    "a two-tuple of the message content and raw tool output is "
+                    f"expected. Instead, generated response is of type: "
+                    f"{type(response)}."
+                )
+                if not isinstance(response, tuple):
                     error_to_raise = ValueError(msg)
                 else:
-                    content, artifact = response
+                    try:
+                        content, artifact = response
+                    except ValueError:
+                        error_to_raise = ValueError(msg)
             else:
                 content = response
         except ValidationError as e:

libs/core/langchain_core/utils/function_calling.py

@@ -653,6 +653,9 @@ class Person(BaseModel):
     return messages
 
 
+_MIN_DOCSTRING_BLOCKS = 2
+
+
 def _parse_google_docstring(
     docstring: str | None,
     args: list[str],
@@ -671,7 +674,7 @@ def _parse_google_docstring(
                 arg for arg in args if arg not in {"run_manager", "callbacks", "return"}
             }
             if filtered_annotations and (
-                len(docstring_blocks) < 2
+                len(docstring_blocks) < _MIN_DOCSTRING_BLOCKS
                 or not any(block.startswith("Args:") for block in docstring_blocks[1:])
             ):
                 msg = "Found invalid Google-Style docstring."

libs/core/langchain_core/utils/pydantic.py

@@ -65,8 +65,8 @@ def get_pydantic_major_version() -> int:
 PYDANTIC_MAJOR_VERSION = PYDANTIC_VERSION.major
 PYDANTIC_MINOR_VERSION = PYDANTIC_VERSION.minor
 
-IS_PYDANTIC_V1 = PYDANTIC_VERSION.major == 1
-IS_PYDANTIC_V2 = PYDANTIC_VERSION.major == 2
+IS_PYDANTIC_V1 = False
+IS_PYDANTIC_V2 = True
 
 PydanticBaseModel = BaseModel
 TypeBaseModel = type[BaseModel]

libs/core/pyproject.toml

@@ -104,7 +104,6 @@ ignore = [
     "ANN401",  # No Any types
     "BLE",     # Blind exceptions
     "ERA",     # No commented-out code
-    "PLR2004", # Comparison to magic number
 ]
 unfixable = [
     "B028",    # People should intentionally tune the stacklevel
@@ -125,7 +124,7 @@ ignore-var-parameters = true  # ignore missing documentation for *args and **kwa
 "langchain_core/utils/mustache.py" = [ "PLW0603",]
 "langchain_core/sys_info.py" = [ "T201",]
 "tests/unit_tests/test_tools.py" = [ "ARG",]
-"tests/**" = [ "D1", "S", "SLF",]
+"tests/**" = [ "D1", "PLR2004", "S", "SLF",]
 "scripts/**" = [ "INP", "S",]
 
 [tool.coverage.run]