refactor(phase 6): inference adapters

REFACTORING_DECISION_LOG.md

@@ -548,6 +548,102 @@ producing `AttributeError: 'TranscriberConfig' object has no attribute
 
 ---
 
+## Phase 6B — vLLM inference clients + legacy shims (2026-05-07)
+
+**1. `VLLMVision(VLLMClient, VLM)` — multiple inheritance kept for nominal typing.**
+`VLLMClient` provides the full implementation (httpx pool, retry,
+circuit breaker, `aclose()`). `VLM` is a pure abstract mixin with no
+conflicting methods, so the MRO is linear and clean. Adds only
+`_max_tokens`, `caption_image()`, and `caption_images_batch()`.
+- Why: VLM and LLM talk to the same vLLM OpenAI-compatible
+  chat/completions endpoint, so `VLLMClient` is the right concrete
+  base. Keeping `VLM` in the bases preserves nominal typing —
+  `isinstance(vision, VLM)` works, and any future code that type-checks
+  against the VLM ABC will accept `VLLMVision` without a cast.
+- Alternative considered: single inheritance `VLLMVision(VLLMClient)`
+  only, relying on structural/duck typing for registry lookup. Rejected
+  — the registry is currently structurally typed, but explicit ABC
+  conformance is cheap here (no diamond, no conflicting methods) and
+  makes the intent clear to readers.
+
+**2. `LLM.generate()` and `LLM.chat()` return `dict` (full OpenAI-compatible response body), not `str`.**
+The original ABC typed both methods as `→ str`, which forced callers to
+re-construct the surrounding OpenAI envelope when building RAG answers
+(losing `model`, `usage`, `finish_reason`, etc.). The concrete vLLM
+implementation already returned the full `httpx` JSON body; the `str`
+annotation was aspirational, not real.
+- Why: RAG answers are ultimately forwarded to the client in OpenAI format.
+  Stripping to plain text at the LLM boundary means the pipeline has to
+  re-wrap the content into `{"choices": [{"message": {"content": …}}]}`
+  further up — metadata (token counts, model id, stop reason) is lost in
+  the process. Returning `dict` preserves the full payload and keeps
+  back-ends interchangeable without wrapping shims. Using bare `dict` (not
+  a `TypedDict`) is a deliberate first step: it is backward-compatible with
+  all current callers and eases compat-shim re-exports while the
+  refactoring is still ongoing.
+- Alternative considered: introduce typed response models (`ChatCompletion`,
+  `CompletionResponse`, `ChatCompletionChunk`) immediately. Rejected as
+  premature — Phase 6 adds the concrete client; Phase 10 (API layer
+  clean-up) is the right time to freeze the contract with typed models.
+  The `dict` annotation signals intent without coupling every caller to a
+  model definition that will evolve.
+- `stream_chat` stays `AsyncIterator[str]` yielding raw SSE lines
+  (`data: {…}` strings). Parsing SSE chunks into typed dicts is Phase 10+
+  work; the current shape keeps the streaming path consistent with
+  OpenAI's SDK behaviour.
+- Future normalisation — when the typed models land, callers will migrate
+  to this pattern (TypedDict shown; Pydantic models are equally valid and
+  would expose `chat_content` / `completion_text` as properties instead):
+
+```python
+from typing import TypedDict
+
+class _Message(TypedDict):
+    role: str
+    content: str
+
+class _Choice(TypedDict):
+    index: int
+    message: _Message        # chat completions
+    finish_reason: str | None
+
+class _CompletionChoice(TypedDict):
+    index: int
+    text: str                # text completions
+    finish_reason: str | None
+
+class _Usage(TypedDict):
+    prompt_tokens: int
+    completion_tokens: int
+    total_tokens: int
+
+class ChatCompletion(TypedDict):
+    id: str
+    object: str              # "chat.completion"
+    model: str
+    choices: list[_Choice]
+    usage: _Usage
+
+class Completion(TypedDict):
+    id: str
+    object: str              # "text_completion"
+    model: str
+    choices: list[_CompletionChoice]
+    usage: _Usage
+
+# Convenience extractors at the pipeline boundary:
+def chat_content(resp: ChatCompletion) -> str:
+    return resp["choices"][0]["message"]["content"]
+
+def completion_text(resp: Completion) -> str:
+    return resp["choices"][0]["text"]
+```
+
+  Until then, callers that need the text can use
+  `resp["choices"][0]["message"]["content"]` directly.
+
+---
+
 ## Template for future entries
 
 ```

openrag/components/indexer/chunker/chunker.py

@@ -6,7 +6,7 @@
 Scheduled for removal in Phase 12.
 """
 
-from typing import ClassVar, Literal
+from typing import Any, ClassVar, Literal
 
 # Side-effect import: pre-loads the indexer-utils submodule so the legacy
 # circular import between `components.utils` and `components.indexer.utils.files`
@@ -52,6 +52,9 @@ async def chat(self, messages: list[dict[str, str]], **kwargs) -> str:
         out = await self._llm.ainvoke(lc_msgs)
         return out.content
 
+    async def stream_chat(self, messages: list[dict[str, str]], **kwargs) -> Any:
+        pass  # Not implemented since the contextualizer never streams.
+
 
 def _chunks_to_documents(chunks: list, base_metadata: dict) -> list[Document]:
     """Convert a list of core domain Chunks into legacy LangChain Documents.

openrag/components/indexer/embeddings/__init__.py

@@ -1,5 +1,7 @@
+from services.inference.vllm_client import VLLMEmbedder  # noqa: F401
+
 from .base import BaseEmbedding
-from .openai import OpenAIEmbedding
+from .openai import _ShimOpenAIEmbedding as OpenAIEmbedding
 
 EMBEDDER_MAPPING = {
     "openai": OpenAIEmbedding,

openrag/components/indexer/embeddings/openai.py

@@ -1,6 +1,17 @@
+"""Backward-compatibility shim — delegates to services.inference.vllm_client.
+
+All new code should import directly from ``services.inference.vllm_client``.
+"""
+
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+
+import httpx
 import openai
+from core.config.endpoints import EmbedderConfig
 from langchain_core.documents.base import Document
 from openai import OpenAI
+from services.inference.vllm_client import VLLMEmbedder  # noqa: F401
 from utils.exceptions.embeddings import *
 from utils.logger import get_logger
 
@@ -9,7 +20,75 @@
 logger = get_logger()
 
 
+_SYNC_POOL = ThreadPoolExecutor(max_workers=1)
+
+
+def _run_sync(coro):
+    """Run an async coroutine from sync code, safe inside a running event loop (e.g. Ray)."""
+    return _SYNC_POOL.submit(asyncio.run, coro).result()
+
+
+def _normalize_texts(texts: list[str | Document]) -> list[str]:
+    return [item.page_content if isinstance(item, Document) else item for item in texts]
+
+
+class _ShimOpenAIEmbedding(BaseEmbedding):
+    """Legacy shim — delegates to ``VLLMEmbedder`` for actual HTTP transport.
+
+    Preserves the sync ``embed_documents``/``embed_query`` contract expected by
+    ``vectordb.py`` (via LangChain's ``aembed_documents`` thread wrapper) while
+    using VLLMEmbedder's long-lived async httpx pool under the hood.
+    """
+
+    def __init__(self, embeddings_config: EmbedderConfig):
+        self._delegate = VLLMEmbedder(
+            endpoint=embeddings_config.base_url,
+            model_name=embeddings_config.model_name,
+            max_model_len=embeddings_config.max_model_len,
+            api_key=embeddings_config.api_key,
+        )
+
+    @property
+    def embedding_dimension(self) -> int:
+        # Probe once if unknown — legacy callers (e.g. MilvusDB schema creation) read
+        # this before any embed() call, but VLLMEmbedder only learns its dimension
+        # from a real response. The probe must run on a one-off sync httpx.Client:
+        # asyncio.run() here would tear down the loop and leave the delegate's
+        # long-lived AsyncClient pool with stale connections, breaking the next
+        # real async call with "Event loop is closed".
+        try:
+            return self._delegate.dimension
+        except RuntimeError:
+            pass
+        body: dict = {"model": self._delegate._model, "input": ["dim-probe"]}
+        if self._delegate._max_model_len is not None:
+            body["truncate_prompt_tokens"] = self._delegate._max_model_len
+        with httpx.Client(timeout=30.0, headers=dict(self._delegate._client.headers)) as client:
+            resp = client.post(f"{self._delegate._endpoint}/embeddings", json=body)
+            resp.raise_for_status()
+        self._delegate._dimension = len(resp.json()["data"][0]["embedding"])
+        return self._delegate._dimension
+
+    def embed_documents(self, texts: list[str | Document]) -> list[list[float]]:
+        if not texts:
+            return []
+        return _run_sync(self._delegate.embed(_normalize_texts(texts)))
+
+    async def aembed_documents(self, texts: list[str | Document]) -> list[list[float]]:
+        if not texts:
+            return []
+        return await self._delegate.embed(_normalize_texts(texts))
+
+    def embed_query(self, text: str) -> list[float]:
+        return _run_sync(self._delegate.embed_single(text))
+
+    async def aembed_query(self, text: str) -> list[float]:
+        return await self._delegate.embed_single(text)
+
+
 class OpenAIEmbedding(BaseEmbedding):
+    """Legacy OpenAI embedding wrapper. New code should use VLLMEmbedder (via DI)."""
+
     def __init__(self, embeddings_config):
         self.embedding_model = embeddings_config.model_name
         self.base_url = embeddings_config.base_url
@@ -20,16 +99,12 @@ def __init__(self, embeddings_config):
     @property
     def embedding_dimension(self) -> int:
         try:
-            # Test call to get embedding dimension
             output = self.embed_documents([Document(page_content="test")])
             return len(output[0])
         except Exception:
             raise
 
     def embed_documents(self, texts: list[str | Document]) -> list[list[float]]:
-        """
-        Embed documents using the configured embedder.
-        """
         if isinstance(texts[0], Document):
             texts = [doc.page_content for doc in texts]
 
@@ -69,9 +144,6 @@ def embed_documents(self, texts: list[str | Document]) -> list[list[float]]:
             )
 
     def embed_query(self, text: str) -> list[float]:
-        """
-        Embed a query using the configured embedder.
-        """
         try:
             output = self.embed_documents([Document(page_content=text)])
             return output[0]

openrag/components/llm.py

@@ -1,13 +1,59 @@
+"""Backward-compatibility shim — delegates to services.inference.vllm_client.
+
+All new code should import directly from ``services.inference.vllm_client``.
+"""
+
 import copy
 import json
+import warnings
 
 import httpx
+from config.models import LLMConfig
+from services.inference.vllm_client import VLLMClient  # noqa: F401
 from utils.logger import get_logger
 
 logger = get_logger()
 
 
+class _LLMShim:
+    """Legacy shim — delegates to ``VLLMClient`` for retry, circuit breaker,
+    and connection pooling while preserving the generator-based interface."""
+
+    def __init__(self, llm_config: LLMConfig, logger=None):
+        warnings.warn(
+            "components.llm.LLM is deprecated — use services.inference.vllm_client.VLLMClient",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        self.logger = logger
+        config_kwargs = {k: v for k, v in llm_config.model_dump().items() if k not in ("api_key", "base_url", "model")}
+        self._delegate = VLLMClient(
+            endpoint=llm_config.base_url,
+            model_name=llm_config.model,
+            api_key=llm_config.api_key,
+            **config_kwargs,
+        )
+
+    async def completions(self, request: dict):
+        prompt = request.pop("prompt")
+        response = await self._delegate.generate(prompt, **request)
+        yield response
+
+    async def chat_completion(self, request: dict):
+        messages = request.pop("messages")
+        stream = request.pop("stream", False)
+
+        if stream:
+            async for line in self._delegate.stream_chat(messages, **request):
+                yield line
+        else:
+            resp_dict = await self._delegate.chat(messages, **request)
+            yield resp_dict
+
+
 class LLM:
+    """Legacy LLM wrapper. New code should use VLLMClient (via DI) instead."""
+
     def __init__(self, llm_config, logger=None):
         self.logger = logger
         default_llm_config = llm_config.model_dump()
@@ -21,7 +67,6 @@ def __init__(self, llm_config, logger=None):
         }
 
     def _extract_llm_overrides(self, request: dict):
-        """Extract and apply LLM overrides from metadata.llm_override."""
         metadata = request.get("metadata") or {}
         llm_override = metadata.pop("llm_override", None) or {}
 
@@ -87,7 +132,7 @@ async def chat_completion(self, request: dict):
                     logger.error(f"Error while streaming chat completion: {str(e)}")
                     raise
 
-            else:  # Handle non-streaming response
+            else:
                 try:
                     response = await client.post(
                         url=f"{base_url}/chat/completions",

openrag/components/pipeline.py

@@ -27,7 +27,7 @@
 from pydantic import ValidationError
 from utils.logger import get_logger
 
-from .llm import LLM
+from .llm import _LLMShim as LLM
 from .map_reduce import RAGMapReduce
 from .reranker import BaseReranker, RerankerFactory
 from .retriever import BaseRetriever, RetrieverFactory
@@ -90,7 +90,7 @@ def __init__(self) -> None:
         self.allow_filterless_fallback = config.retriever.allow_filterless_fallback
 
         self.reranker_enabled = config.reranker.enabled
-        self.reranker: BaseReranker = RerankerFactory.get_reranker(config)
+        self.reranker: BaseReranker = RerankerFactory.get_reranker(config.reranker)
         logger.debug("Reranker", enabled=self.reranker_enabled, provider=config.reranker.provider)
         self.reranker_top_k = config.reranker.top_k