refactor: integrate retry decorator settings into DependencyContainer and LangchainSummarizer

Author: a-klos

libs/admin-api-lib/src/admin_api_lib/dependency_container.py

@@ -64,6 +64,7 @@
 from rag_core_lib.impl.settings.langfuse_settings import LangfuseSettings
 from rag_core_lib.impl.settings.ollama_llm_settings import OllamaSettings
 from rag_core_lib.impl.settings.rag_class_types_settings import RAGClassTypeSettings
+from rag_core_lib.impl.settings.retry_decorator_settings import RetryDecoratorSettings
 from rag_core_lib.impl.settings.stackit_vllm_settings import StackitVllmSettings
 from rag_core_lib.impl.tracers.langfuse_traced_chain import LangfuseTracedGraph
 from rag_core_lib.impl.utils.async_threadsafe_semaphore import AsyncThreadsafeSemaphore
@@ -86,6 +87,7 @@ class DependencyContainer(DeclarativeContainer):
     key_value_store_settings = KeyValueSettings()
     summarizer_settings = SummarizerSettings()
     source_uploader_settings = SourceUploaderSettings()
+    retry_decorator_settings = RetryDecoratorSettings()
 
     key_value_store = Singleton(FileStatusKeyValueStore, key_value_store_settings)
     file_service = Singleton(S3Service, s3_settings=s3_settings)
@@ -136,7 +138,9 @@ class DependencyContainer(DeclarativeContainer):
         LangchainSummarizer,
         langfuse_manager=langfuse_manager,
         chunker=summary_text_splitter,
-        semaphore=Singleton(AsyncThreadsafeSemaphore, summarizer_settings.maximum_concurrreny),
+        semaphore=Singleton(AsyncThreadsafeSemaphore, summarizer_settings.maximum_concurrency),
+        summarizer_settings=summarizer_settings,
+        retry_decorator_settings=retry_decorator_settings
     )
 
     summary_enhancer = List(

libs/admin-api-lib/src/admin_api_lib/impl/settings/summarizer_settings.py

@@ -1,6 +1,7 @@
 """Contains settings for summarizer."""
 
-from pydantic import Field
+from typing import Optional
+from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
 
@@ -12,8 +13,22 @@ class SummarizerSettings(BaseSettings):
     ----------
     maximum_input_size : int
         The maximum size of the input that the summarizer can handle. Default is 8000.
-    maximum_concurrreny : int
+    maximum_concurrency : int
         The maximum number of concurrent summarization processes. Default is 10.
+    max_retries: Optional[PositiveInt]
+        Total retries, not counting the initial attempt.
+    retry_base_delay: Optional[float]
+        Base delay in seconds for the first retry.
+    retry_max_delay: Optional[float]
+        Maximum delay cap in seconds for any single wait.
+    backoff_factor: Optional[float]
+        Exponential backoff factor (>= 1).
+    attempt_cap: Optional[int]
+        Cap for exponent growth (backoff_factor ** attempt_cap).
+    jitter_min: Optional[float]
+        Minimum jitter in seconds.
+    jitter_max: Optional[float]
+        Maximum jitter in seconds.
     """
 
     class Config:
@@ -23,4 +38,45 @@ class Config:
         case_sensitive = False
 
     maximum_input_size: int = Field(default=8000)
-    maximum_concurrreny: int = Field(default=10)
+    maximum_concurrency: int = Field(default=10)
+    max_retries: Optional[PositiveInt] = Field(
+        default=None,
+        title="Max Retries",
+        description="Total retries, not counting the initial attempt.",
+    )
+    retry_base_delay: Optional[float] = Field(
+        default=None,
+        ge=0,
+        title="Retry Base Delay",
+        description="Base delay in seconds for the first retry.",
+    )
+    retry_max_delay: Optional[float] = Field(
+        default=None,
+        gt=0,
+        title="Retry Max Delay",
+        description="Maximum delay cap in seconds for any single wait.",
+    )
+    backoff_factor: Optional[float] = Field(
+        default=None,
+        ge=1.0,
+        title="Backoff Factor",
+        description="Exponential backoff factor (>= 1).",
+    )
+    attempt_cap: Optional[int] = Field(
+        default=None,
+        ge=0,
+        title="Attempt Cap",
+        description="Cap for exponent growth (backoff_factor ** attempt_cap).",
+    )
+    jitter_min: Optional[float] = Field(
+        default=None,
+        ge=0.0,
+        title="Jitter Min (s)",
+        description="Minimum jitter in seconds.",
+    )
+    jitter_max: Optional[float] = Field(
+        default=None,
+        ge=0.0,
+        title="Jitter Max (s)",
+        description="Maximum jitter in seconds.",
+    )

libs/admin-api-lib/src/admin_api_lib/impl/summarizer/langchain_summarizer.py

@@ -1,20 +1,24 @@
 """Module for the LangchainSummarizer class."""
 
+import asyncio
 import logging
-import traceback
 from typing import Optional
 
 from langchain.text_splitter import RecursiveCharacterTextSplitter
 from langchain_core.documents import Document
 from langchain_core.runnables import Runnable, RunnableConfig, ensure_config
+from openai import APIConnectionError, APIError, APITimeoutError, RateLimitError
 
+from admin_api_lib.impl.settings.summarizer_settings import SummarizerSettings
 from admin_api_lib.summarizer.summarizer import (
     Summarizer,
     SummarizerInput,
     SummarizerOutput,
 )
 from rag_core_lib.impl.langfuse_manager.langfuse_manager import LangfuseManager
+from rag_core_lib.impl.settings.retry_decorator_settings import RetryDecoratorSettings
 from rag_core_lib.impl.utils.async_threadsafe_semaphore import AsyncThreadsafeSemaphore
+from rag_core_lib.impl.utils.retry_decorator import retry_with_backoff
 
 logger = logging.getLogger(__name__)
 
@@ -32,10 +36,15 @@ def __init__(
         langfuse_manager: LangfuseManager,
         chunker: RecursiveCharacterTextSplitter,
         semaphore: AsyncThreadsafeSemaphore,
+        summarizer_settings: SummarizerSettings,
+        retry_decorator_settings: RetryDecoratorSettings,
     ):
         self._chunker = chunker
         self._langfuse_manager = langfuse_manager
         self._semaphore = semaphore
+        self._retry_decorator_settings = self._create_retry_decorator_settings(
+            summarizer_settings, retry_decorator_settings
+        )
 
     async def ainvoke(self, query: SummarizerInput, config: Optional[RunnableConfig] = None) -> SummarizerOutput:
         """
@@ -65,40 +74,88 @@ async def ainvoke(self, query: SummarizerInput, config: Optional[RunnableConfig]
         """
         assert query, "Query is empty: %s" % query  # noqa S101
         config = ensure_config(config)
-        tries_remaining = config.get("configurable", {}).get("tries_remaining", 3)
-        logger.debug("Tries remaining %d" % tries_remaining)
 
-        if tries_remaining < 0:
-            raise Exception("Summary creation failed.")
         document = Document(page_content=query)
         langchain_documents = self._chunker.split_documents([document])
+        logger.debug("Summarizing %d chunk(s)...", len(langchain_documents))
 
-        outputs = []
-        for langchain_document in langchain_documents:
-            async with self._semaphore:
-                try:
-                    result = await self._create_chain().ainvoke({"text": langchain_document.page_content}, config)
-                    # Extract content from AIMessage if it's not already a string
-                    content = result.content if hasattr(result, "content") else str(result)
-                    outputs.append(content)
-                except Exception as e:
-                    logger.error("Error in summarizing langchain doc: %s %s", e, traceback.format_exc())
-                    config["tries_remaining"] = tries_remaining - 1
-                    result = await self._create_chain().ainvoke({"text": langchain_document.page_content}, config)
-                    # Extract content from AIMessage if it's not already a string
-                    content = result.content if hasattr(result, "content") else str(result)
-                    outputs.append(content)
+        # Fan out with concurrency, bounded by your semaphore inside _summarize_chunk
+        tasks = [asyncio.create_task(self._summarize_chunk(doc.page_content, config)) for doc in langchain_documents]
+        outputs = await asyncio.gather(*tasks)
 
         if len(outputs) == 1:
             return outputs[0]
-        summary = " ".join(outputs)
+
+        # Optional single reduce pass (no recursion)
+        merged = " ".join(outputs)
         logger.debug(
-            "Reduced number of chars from %d to %d"
-            % (len("".join([x.page_content for x in langchain_documents])), len(summary))
+            "Reduced number of chars from %d to %d",
+            len("".join([x.page_content for x in langchain_documents])),
+            len(merged),
+        )
+        return await self._summarize_chunk(merged, config)
+
+    def _create_retry_decorator_settings(
+        self, summarizer_settings: SummarizerSettings, retry_decorator_settings: RetryDecoratorSettings
+    ):
+        return RetryDecoratorSettings(
+            max_retries=(
+                summarizer_settings.max_retries
+                if summarizer_settings.max_retries is not None
+                else retry_decorator_settings.max_retries
+            ),
+            retry_base_delay=(
+                summarizer_settings.retry_base_delay
+                if summarizer_settings.retry_base_delay is not None
+                else retry_decorator_settings.retry_base_delay
+            ),
+            retry_max_delay=(
+                summarizer_settings.retry_max_delay
+                if summarizer_settings.retry_max_delay is not None
+                else retry_decorator_settings.retry_max_delay
+            ),
+            backoff_factor=(
+                summarizer_settings.backoff_factor
+                if summarizer_settings.backoff_factor is not None
+                else retry_decorator_settings.backoff_factor
+            ),
+            attempt_cap=(
+                summarizer_settings.attempt_cap
+                if summarizer_settings.attempt_cap is not None
+                else retry_decorator_settings.attempt_cap
+            ),
+            jitter_min=(
+                summarizer_settings.jitter_min
+                if summarizer_settings.jitter_min is not None
+                else retry_decorator_settings.jitter_min
+            ),
+            jitter_max=(
+                summarizer_settings.jitter_max
+                if summarizer_settings.jitter_max is not None
+                else retry_decorator_settings.jitter_max
+            ),
         )
-        return await self.ainvoke(summary, config)
 
     def _create_chain(self) -> Runnable:
         return self._langfuse_manager.get_base_prompt(self.__class__.__name__) | self._langfuse_manager.get_base_llm(
             self.__class__.__name__
         )
+
+    def _retry_with_backoff_wrapper(self):
+        # Prefer summarizer-specific overrides; fall back to global retry settings
+        return retry_with_backoff(
+            settings=self._retry_decorator_settings,
+            exceptions=(APIError, RateLimitError, APITimeoutError, APIConnectionError),
+            rate_limit_exceptions=(RateLimitError,),
+            logger=logger,
+        )
+
+    async def _summarize_chunk(self, text: str, config: Optional[RunnableConfig]) -> SummarizerOutput:
+        @self._retry_with_backoff_wrapper()
+        async def _call(text: str, config: Optional[RunnableConfig]) -> SummarizerOutput:
+            response = await self._create_chain().ainvoke({"text": text}, config)
+            return response.content if hasattr(response, "content") else str(response)
+
+        # Hold the semaphore for the entire retry lifecycle
+        async with self._semaphore:
+            return await _call(text, config)