chore(langchain): add ruff rules TC (#31921)

See https://docs.astral.sh/ruff/rules/#flake8-type-checking-tc

Author: cbornet

libs/langchain/langchain/agents/agent.py

@@ -1185,7 +1185,7 @@ def _action_agent(self) -> Union[BaseSingleActionAgent, BaseMultiActionAgent]:
         to reflect the changes made in the root_validator.
         """
         if isinstance(self.agent, Runnable):
-            return cast(RunnableAgentType, self.agent)
+            return cast("RunnableAgentType", self.agent)
         return self.agent
 
     def save(self, file_path: Union[Path, str]) -> None:

libs/langchain/langchain/callbacks/streaming_aiter.py

@@ -73,7 +73,7 @@ async def aiter(self) -> AsyncIterator[str]:
                 other.pop().cancel()
 
             # Extract the value of the first completed task
-            token_or_done = cast(Union[str, Literal[True]], done.pop().result())
+            token_or_done = cast("Union[str, Literal[True]]", done.pop().result())
 
             # If the extracted value is the boolean True, the done event was set
             if token_or_done is True:

libs/langchain/langchain/chains/base.py

@@ -411,7 +411,7 @@ def __call__(
 
         return self.invoke(
             inputs,
-            cast(RunnableConfig, {k: v for k, v in config.items() if v is not None}),
+            cast("RunnableConfig", {k: v for k, v in config.items() if v is not None}),
             return_only_outputs=return_only_outputs,
             include_run_info=include_run_info,
         )
@@ -461,7 +461,7 @@ async def acall(
         }
         return await self.ainvoke(
             inputs,
-            cast(RunnableConfig, {k: v for k, v in config.items() if k is not None}),
+            cast("RunnableConfig", {k: v for k, v in config.items() if k is not None}),
             return_only_outputs=return_only_outputs,
             include_run_info=include_run_info,
         )

libs/langchain/langchain/chains/combine_documents/map_rerank.py

@@ -229,7 +229,7 @@ def _process_results(
         docs: list[Document],
         results: Sequence[Union[str, list[str], dict[str, str]]],
     ) -> tuple[str, dict]:
-        typed_results = cast(list[dict], results)
+        typed_results = cast("list[dict]", results)
         sorted_res = sorted(
             zip(typed_results, docs),
             key=lambda x: -int(x[0][self.rank_key]),

libs/langchain/langchain/chains/llm.py

@@ -145,7 +145,7 @@ def generate(
                 **self.llm_kwargs,
             )
         results = self.llm.bind(stop=stop, **self.llm_kwargs).batch(
-            cast(list, prompts),
+            cast("list", prompts),
             {"callbacks": callbacks},
         )
         generations: list[list[Generation]] = []
@@ -172,7 +172,7 @@ async def agenerate(
                 **self.llm_kwargs,
             )
         results = await self.llm.bind(stop=stop, **self.llm_kwargs).abatch(
-            cast(list, prompts),
+            cast("list", prompts),
             {"callbacks": callbacks},
         )
         generations: list[list[Generation]] = []

libs/langchain/langchain/chains/openai_functions/qa_with_structure.py

@@ -76,11 +76,11 @@ def create_qa_with_structure_chain(
         raise ValueError(msg)
     if isinstance(schema, type) and is_basemodel_subclass(schema):
         if hasattr(schema, "model_json_schema"):
-            schema_dict = cast(dict, schema.model_json_schema())
+            schema_dict = cast("dict", schema.model_json_schema())
         else:
-            schema_dict = cast(dict, schema.schema())
+            schema_dict = cast("dict", schema.schema())
     else:
-        schema_dict = cast(dict, schema)
+        schema_dict = cast("dict", schema)
     function = {
         "name": schema_dict["title"],
         "description": schema_dict["description"],

libs/langchain/langchain/chains/query_constructor/base.py

@@ -91,7 +91,7 @@ def from_components(
 
             def ast_parse(raw_filter: str) -> Optional[FilterDirective]:
                 filter_directive = cast(
-                    Optional[FilterDirective],
+                    "Optional[FilterDirective]",
                     get_parser().parse(raw_filter),
                 )
                 return fix_filter_directive(
@@ -144,7 +144,7 @@ def fix_filter_directive(
             return None
         args = [
             cast(
-                FilterDirective,
+                "FilterDirective",
                 fix_filter_directive(
                     arg,
                     allowed_comparators=allowed_comparators,

libs/langchain/langchain/chains/router/llm_router.py

@@ -137,7 +137,7 @@ def _call(
 
         prediction = self.llm_chain.predict(callbacks=callbacks, **inputs)
         return cast(
-            dict[str, Any],
+            "dict[str, Any]",
             self.llm_chain.prompt.output_parser.parse(prediction),
         )
 
@@ -149,7 +149,7 @@ async def _acall(
         _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
         callbacks = _run_manager.get_child()
         return cast(
-            dict[str, Any],
+            "dict[str, Any]",
             await self.llm_chain.apredict_and_parse(callbacks=callbacks, **inputs),
         )
 

libs/langchain/langchain/chat_models/base.py

@@ -322,7 +322,7 @@ class GetPopulation(BaseModel):
 
     if not configurable_fields:
         return _init_chat_model_helper(
-            cast(str, model),
+            cast("str", model),
             model_provider=model_provider,
             **kwargs,
         )
@@ -632,7 +632,7 @@ def with_config(
         **kwargs: Any,
     ) -> _ConfigurableModel:
         """Bind config to a Runnable, returning a new Runnable."""
-        config = RunnableConfig(**(config or {}), **cast(RunnableConfig, kwargs))
+        config = RunnableConfig(**(config or {}), **cast("RunnableConfig", kwargs))
         model_params = self._model_params(config)
         remaining_config = {k: v for k, v in config.items() if k != "configurable"}
         remaining_config["configurable"] = {
@@ -781,7 +781,7 @@ def batch_as_completed(
         if config is None or isinstance(config, dict) or len(config) <= 1:
             if isinstance(config, list):
                 config = config[0]
-            yield from self._model(cast(RunnableConfig, config)).batch_as_completed(  # type: ignore[call-overload]
+            yield from self._model(cast("RunnableConfig", config)).batch_as_completed(  # type: ignore[call-overload]
                 inputs,
                 config=config,
                 return_exceptions=return_exceptions,
@@ -811,7 +811,7 @@ async def abatch_as_completed(
             if isinstance(config, list):
                 config = config[0]
             async for x in self._model(
-                cast(RunnableConfig, config),
+                cast("RunnableConfig", config),
             ).abatch_as_completed(  # type: ignore[call-overload]
                 inputs,
                 config=config,

libs/langchain/langchain/document_loaders/blob_loaders/schema.py

@@ -1,12 +1,9 @@
-from typing import TYPE_CHECKING, Any
+from typing import Any
 
 from langchain_core.document_loaders import Blob, BlobLoader
 
 from langchain._api import create_importer
 
-if TYPE_CHECKING:
-    pass
-
 # Create a way to dynamically look up deprecated imports.
 # Used to consolidate logic for raising deprecation warnings and
 # handling optional imports.