restore typehints in rag

Author: eaidova

notebooks/llm-rag-langchain/ov_langchain_helper.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import queue
-from typing import Any, Iterator, Optional, Sequence
+from typing import Any, Dict, Iterator, List, Optional, Sequence
 
 from langchain_core.callbacks.manager import CallbackManagerForLLMRun
 from langchain_core.language_models.llms import LLM
@@ -102,7 +102,7 @@ def from_model_path(
     def _call(
         self,
         prompt: str,
-        stop: Optional[list[str]] = None,
+        stop: Optional[List[str]] = None,
         run_manager: Optional[CallbackManagerForLLMRun] = None,
         **kwargs: Any,
     ) -> str:
@@ -128,7 +128,7 @@ def _call(
     def _stream(
         self,
         prompt: str,
-        stop: Optional[list[str]] = None,
+        stop: Optional[List[str]] = None,
         run_manager: Optional[CallbackManagerForLLMRun] = None,
         **kwargs: Any,
     ) -> Iterator[GenerationChunk]:
@@ -239,8 +239,8 @@ def __init__(self, **kwargs: Any):
 
     def _generate(
         self,
-        messages: list[BaseMessage],
-        stop: Optional[list[str]] = None,
+        messages: List[BaseMessage],
+        stop: Optional[List[str]] = None,
         run_manager: Optional[CallbackManagerForLLMRun] = None,
         **kwargs: Any,
     ) -> ChatResult:
@@ -250,8 +250,8 @@ def _generate(
 
     def _stream(
         self,
-        messages: list[BaseMessage],
-        stop: Optional[list[str]] = None,
+        messages: List[BaseMessage],
+        stop: Optional[List[str]] = None,
         run_manager: Optional[CallbackManagerForLLMRun] = None,
         **kwargs: Any,
     ) -> Iterator[ChatGenerationChunk]:
@@ -266,7 +266,7 @@ def _stream(
 
     def _to_chat_prompt(
         self,
-        messages: list[BaseMessage],
+        messages: List[BaseMessage],
     ) -> str:
         """Convert a list of messages into a prompt format expected by wrapped LLM."""
         try:
@@ -472,22 +472,22 @@ def run_mean_pooling(model_output: Any, attention_mask: Any) -> Any:
 
     model_config = ConfigDict(extra="forbid", protected_namespaces=())
 
-    def embed_documents(self, texts: list[str]) -> list[list[float]]:
+    def embed_documents(self, texts: List[str]) -> List[List[float]]:
         """Compute doc embeddings using a HuggingFace transformer model.
 
         Args:
             texts: The list of texts to embed.
 
         Returns:
-            list of embeddings, one for each text.
+            List of embeddings, one for each text.
         """
 
         texts = list(map(lambda x: x.replace("\n", " "), texts))
         embeddings = self.encode(texts, show_progress_bar=self.show_progress, **self.encode_kwargs)
 
         return embeddings
 
-    def embed_query(self, text: str) -> list[float]:
+    def embed_query(self, text: str) -> List[float]:
         """Compute query embeddings using a HuggingFace transformer model.
 
         Args:
@@ -529,20 +529,20 @@ def __init__(self, **kwargs: Any):
         if "-zh" in self.model_path:
             self.query_instruction = DEFAULT_QUERY_BGE_INSTRUCTION_ZH
 
-    def embed_documents(self, texts: list[str]) -> list[list[float]]:
+    def embed_documents(self, texts: List[str]) -> List[List[float]]:
         """Compute doc embeddings using a HuggingFace transformer model.
 
         Args:
             texts: The list of texts to embed.
 
         Returns:
-            list of embeddings, one for each text.
+            List of embeddings, one for each text.
         """
         texts = [self.embed_instruction + t.replace("\n", " ") for t in texts]
         embeddings = self.encode(texts, **self.encode_kwargs)
         return embeddings
 
-    def embed_query(self, text: str) -> list[float]:
+    def embed_query(self, text: str) -> List[float]:
         """Compute query embeddings using a HuggingFace transformer model.
 
         Args: