Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion openrag/api/routers/admin/partitions.py
Original file line number Diff line number Diff line change
Expand Up @@ -325,7 +325,7 @@ async def create_partition(
- `indexation_preset`
- `retrieval_preset`
- `chat_history_depth`
- `chat_llm`
- `chat_llm` (must name a registered LLM endpoint — 422 otherwise; explicit `null` resets to the default LLM)

**Permissions:**
- Requires partition owner role
Expand Down
17 changes: 17 additions & 0 deletions openrag/api/schemas/admin/partition_schemas.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,14 @@ def validate_non_empty_name(cls, value: str) -> str:
"""Normalize non-null partition and preset names."""
return _normalize_name(value)

@field_validator("chat_llm")
@classmethod
def normalize_chat_llm(cls, value: str | None) -> str | None:
"""Blank and null both mean "use the default LLM" (stored as NULL)."""
if value is None:
return None
return value.strip() or None


class UpdatePartitionRequest(BaseModel):
"""Request body for updating partition preset assignments."""
Expand Down Expand Up @@ -73,6 +81,15 @@ def reject_null_chat_history_depth(cls, value: int | None) -> int:
"""Reject explicit null for chat history depth updates."""
return _reject_explicit_null("chat_history_depth", value)

@field_validator("chat_llm")
@classmethod
def normalize_chat_llm(cls, value: str | None) -> str | None:
"""Normalize ``chat_llm`` — unlike the other fields, explicit null is
allowed: it (or a blank string) resets the partition to the default LLM."""
if value is None:
return None
return value.strip() or None

@model_validator(mode="after")
def require_at_least_one_update(self) -> UpdatePartitionRequest:
"""Reject empty update payloads."""
Expand Down
8 changes: 6 additions & 2 deletions openrag/di/container.py
Original file line number Diff line number Diff line change
Expand Up @@ -531,8 +531,11 @@ def query_service(self) -> QueryService:
"""QueryService — lazily built, cached for the container's lifetime.

Shares the same core LLM construction as ``retrieval_service``
(built from ``settings.llm``); the web-search service comes from
the ``WebSearchFactory`` (provider is ``None`` when
(built from ``settings.llm``); the named ``llm_factory`` lets the
service honor a partition's ``chat_llm`` model-endpoint preset for
query generation and answer generation. The web-search service
comes from the
``WebSearchFactory`` (provider is ``None`` when
``WEBSEARCH_API_TOKEN`` is unset — web search silently disabled).
"""
if self._query_service is None:
Expand All @@ -554,6 +557,7 @@ def query_service(self) -> QueryService:
config=settings,
web_search_service=WebSearchFactory.create_service(settings),
workspace_service=self.workspace_service,
llm_factory=self.llm_factory,
)
return self._query_service

Expand Down
1 change: 1 addition & 0 deletions openrag/services/inference/ollama_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ def __init__(
timeout=timeout,
headers={"Content-Type": "application/json"},
)
logger.bind(model=self._model, endpoint=self._endpoint, timeout=timeout).debug("OllamaClient ready")

@with_circuit_breaker("llm")
@with_retry(max_attempts=3)
Expand Down
9 changes: 9 additions & 0 deletions openrag/services/inference/vllm_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,15 @@ def __init__(
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
self._client = httpx.AsyncClient(timeout=timeout, headers=headers)
# Same construction breadcrumb as VLLMEmbedder: the component factories
# cache instances per endpoint name, so this fires once per configured
# endpoint and shows which base URL/model a preset name resolved to.
logger.bind(
model=self._model,
endpoint=self._endpoint,
timeout=timeout,
enable_thinking=self._enable_thinking,
).debug(f"{type(self).__name__} ready")

def _resolve_overrides(self, kwargs: dict) -> tuple[str, str, dict[str, str] | None]:
"""Read ``metadata.llm_override`` from *kwargs* without mutating caller data.
Expand Down
34 changes: 31 additions & 3 deletions openrag/services/orchestrators/partition_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,11 @@
# ``list_existant_partitions``. Matched case-insensitively.
_RESERVED_PARTITION_NAMES = frozenset({"all"})

# Columns where an explicit ``None`` in a PATCH is a real value (SQL NULL =
# "reset to default"), not the omitted-field sentinel that the None-filter
# in ``update_partition`` gives every other column.
_NULLABLE_COLUMNS = frozenset({"chat_llm"})


def _validate_limit(limit: int | None) -> None:
"""Reject negative ``limit`` values before they reach ``rows[:limit]``.
Expand Down Expand Up @@ -223,6 +228,8 @@ async def create_partition(
# Validate the preset references before touching the DB.
if self._config is not None:
self._validate_preset_refs({"partition": partition, **config_fields})
if chat_llm:
self._validate_chat_llm_ref(chat_llm)

await self._partition_repo.create_partition(name=partition, user_id=user_id, max_owned=max_owned)

Expand Down Expand Up @@ -271,18 +278,26 @@ async def delete_partition(self, partition: str) -> None:
async def update_partition(self, partition: str, **fields: object) -> dict | None:
"""Update a partition's config columns and re-resolve the cache.

``None`` values are ignored (so partial PATCH semantics work). When a
``None`` values are ignored (so partial PATCH semantics work) —
except for the ``_NULLABLE_COLUMNS`` (``chat_llm``), where an
explicit ``None`` clears the column back to its default. When a
preset reference changes, the merged row is validated *before* the
write so an unknown preset name fails fast.
write so an unknown preset name fails fast; likewise an assigned
``chat_llm`` must name a catalogued LLM endpoint.
"""
await self._ensure_partition(partition)
updates = {k: v for k, v in fields.items() if v is not None}
updates = {k: v for k, v in fields.items() if v is not None or k in _NULLABLE_COLUMNS}

if self._config is not None and updates:
current = await self._partition_repo.get_partition_row(partition)
if current is None:
raise PartitionNotFoundError(f"Partition '{partition}' does not exist.")
self._validate_preset_refs({**current, **updates})
# Only the incoming value is checked — a *stored* name that went
# stale (endpoint deleted later) must not block unrelated PATCHes;
# QueryService falls back to the default LLM for those at runtime.
if updates.get("chat_llm"):
self._validate_chat_llm_ref(updates["chat_llm"])

result = await self._partition_repo.update_partition(partition, **updates)

Expand Down Expand Up @@ -324,6 +339,19 @@ def _validate_preset_refs(self, row: dict) -> None:
except ConfigError as exc:
raise ValidationError(exc.message, code="PRESET_NOT_FOUND") from exc

def _validate_chat_llm_ref(self, chat_llm: str) -> None:
"""Assignment-time check: ``chat_llm`` must name a catalogued LLM endpoint.

Only guards assignment — a stored name can go stale afterwards (the
endpoint may be renamed or deleted), which QueryService tolerates by
falling back to the default LLM at request time.
"""
if chat_llm not in self._require_config().models.llm:
raise ValidationError(
f"LLM endpoint '{chat_llm}' referenced by chat_llm not found.",
code="MODEL_ENDPOINT_NOT_FOUND",
)

def _partition_detail(self, row: dict, cfg: PartitionConfig) -> dict:
"""Shape a resolved row into the ``PartitionDetailResponse`` payload."""
return {
Expand Down
79 changes: 67 additions & 12 deletions openrag/services/orchestrators/query_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -107,9 +107,11 @@ def __init__(
config: Settings,
web_search_service: Any | None,
workspace_service: WorkspaceService,
llm_factory: Callable[[str], LLM] | None = None,
) -> None:
self._retrieval = retrieval_service
self._llm = llm
self._llm_factory = llm_factory
self._web = web_search_service
self._workspace = workspace_service

Expand Down Expand Up @@ -156,11 +158,58 @@ def _resolve_chat_history_depth(self, partition: list[str] | None) -> int:
return max(explicit)
return self._default_chat_history_depth

def _resolve_llm(self, partition: list[str] | None) -> LLM:
"""Effective LLM for this request — query generation and answering.

Honors a partition's configured ``chat_llm`` model-endpoint preset
(set via the admin API) over the default LLM. Resolved once per
request (``chat`` / ``chat_stream`` / ``complete``) and used for
both the query-contextualization call and the final answer;
map-reduce stays on the default LLM until its post-release
refactor. Same partition semantics as
``_resolve_chat_history_depth``: no partition and the ``"all"``
sentinel use the default; a multi-partition request uses the
preset only when every partition that sets one names the same
endpoint — with conflicting presets there is no single owning
partition, so the default applies.

``chat_llm`` is not validated against the endpoint catalog when it
Comment thread
Ahmath-Gadji marked this conversation as resolved.
Resolved
is assigned, so an unknown name (e.g. an endpoint deleted after
assignment) must not fail the chat request — it falls back to the
default LLM with a warning.
"""
if self._llm_factory is None or not partition or "all" in partition:
logger.bind(partitions=partition).debug("Answering with the default LLM (no partition-scoped preset)")
return self._llm
names = {
cfg.chat_llm
for name in partition
if (cfg := self._config.partitions.get(name)) is not None and cfg.chat_llm
}
if len(names) != 1:
logger.bind(partitions=partition, chat_llm_presets=sorted(names)).debug(
"Answering with the default LLM (no single chat_llm preset among the partitions)"
)
return self._llm
(chat_llm,) = names
try:
llm = self._llm_factory(chat_llm)
except KeyError:
logger.warning(
"Partition chat_llm preset not found in the model-endpoint catalog — using the default LLM",
chat_llm=chat_llm,
partitions=partition,
)
return self._llm
logger.bind(chat_llm=chat_llm, partitions=partition).debug("Answering with the partition's chat_llm preset")
return llm

# ------------------------------------------------------------------
# Query generation (was RagPipeline.generate_query — no LangChain)
# ------------------------------------------------------------------

async def generate_query(self, messages: list[dict]) -> SearchQueries:
async def generate_query(self, messages: list[dict], llm: LLM | None = None) -> SearchQueries:
llm = llm or self._llm
last_user = messages[-1]["content"]
if RAGMODE(self._rag_mode) is RAGMODE.SIMPLERAG:
return SearchQueries(query_list=[Query(query=last_user)])
Expand All @@ -180,7 +229,7 @@ async def generate_query(self, messages: list[dict]) -> SearchQueries:
}
for attempt in (1, 2):
try:
resp = await self._llm.chat(llm_messages, **params)
resp = await llm.chat(llm_messages, **params)
content = resp["choices"][0]["message"]["content"]
return SearchQueries.model_validate_json(_json_slice(content))
except Exception as exc:
Expand All @@ -198,6 +247,9 @@ async def generate_query(self, messages: list[dict]) -> SearchQueries:
# ------------------------------------------------------------------

async def _infer_relevancy(self, query: str, doc) -> tuple[bool, str]:
# Deliberately pinned to the default LLM: map-reduce is slated for a
# full post-release refactor, and routing it through the partition
# chat_llm preset is part of that work.
async with get_llm_semaphore():
try:
resp = await self._llm.chat(
Expand Down Expand Up @@ -246,9 +298,9 @@ async def _batch(chunks: list, summaries: list) -> bool:
# Preparation (was RagPipeline._prepare_for_chat_completion)
# ------------------------------------------------------------------

async def _prepare_chat(self, partition: list[str] | None, payload: dict):
async def _prepare_chat(self, partition: list[str] | None, payload: dict, llm: LLM | None = None):
messages = payload["messages"][-self._resolve_chat_history_depth(partition) :]
queries = await self.generate_query(messages)
queries = await self.generate_query(messages, llm=llm)

metadata = payload.get("metadata") or {}
use_map_reduce = metadata.get("use_map_reduce", False)
Expand Down Expand Up @@ -334,9 +386,9 @@ async def _gather_rag_and_web(self, query_list, partition, top_k, filter_params)
doc_lists, web_lists = await asyncio.gather(rag, web)
return doc_lists, web_lists

async def _prepare_completions(self, partition: list[str], payload: dict):
async def _prepare_completions(self, partition: list[str], payload: dict, llm: LLM | None = None):
prompt = payload["prompt"]
queries = await self.generate_query([{"role": "user", "content": prompt}])
queries = await self.generate_query([{"role": "user", "content": prompt}], llm=llm)
chunks = await self._retrieval.retrieve_multi(partitions=partition, search_queries=queries)
docs = [c.to_langchain() for c in chunks]
context, included = format_context(
Expand Down Expand Up @@ -416,14 +468,15 @@ async def chat(
) -> dict:
"""Non-streaming chat completion → finalized OpenAI dict."""
metadata = payload.get("metadata") or {}
llm = self._resolve_llm(partitions)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One thing still worries me here: the HTTP layer runs the availability and token checks against the default LLM before we reach this selected chat_llm. So a partition with a valid custom chat model can still be blocked if the default LLM is down, and the request can be checked against the wrong context size. I think those preflight checks should use the same LLM that will actually answer the request.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Confirmed and agreed this is a real gap: the check_llm_model_availability dependency validates only config.llm, and check_tokens_limit uses get_max_model_tokens(), a single global primed once at startup from the default model. So a partition with a healthy custom chat_llm can be rejected when the default endpoint is down, and its request is sized against the wrong context window.

I'm scoping this as a follow-up rather than folding it into this PR, for three reasons:

  1. It's a pre-existing limitation of the HTTP preflight layer, independent of this PR's diff — this PR resolves the answer LLM in the service, which is what makes the gap reachable, but the fix lives entirely in api/routers/user/chat.py + api/dependencies/llm.py.
  2. A correct fix has to resolve the partition→chat_llm mapping before the Depends, then run availability against that endpoint and cache/fetch max_model_len per endpoint (today it's one startup-primed global) — a non-trivial change to a dependency that runs on every chat request, deserving its own tests.
  3. Keeping it separate avoids expanding this PR's blast radius right before merge.

Behaviorally, this PR is a strict improvement over main (the preset now actually answers); the preflight mismatch only degrades the default-down / non-default-context edge case, which existed implicitly before. I'll open a tracking issue for "run chat preflight checks against the resolved per-partition LLM" and link it here. Shout if you'd rather I pull it into this PR instead.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Tracking issue opened: #639.

if partitions is None and not metadata.get("websearch", False):
docs, web_results = [], []
else:
payload, docs, web_results = await self._prepare_chat(partitions, payload)
payload, docs, web_results = await self._prepare_chat(partitions, payload, llm)
sources = prepare_sources(docs, web_results)

payload["messages"] = self._sanitize_messages(payload["messages"])
chunk = await self._llm.chat(payload["messages"], **_sampling(payload))
chunk = await llm.chat(payload["messages"], **_sampling(payload))
chunk["model"] = model_name
content = chunk.get("choices", [{}])[0].get("message", {}).get("content", "") or ""
clean, citations = extract_and_strip_sources_block(content)
Expand All @@ -441,14 +494,15 @@ async def chat_stream(
) -> AsyncIterator[str]:
"""Streaming chat completion → SSE strings with filtered sources."""
metadata = payload.get("metadata") or {}
llm = self._resolve_llm(partitions)
if partitions is None and not metadata.get("websearch", False):
docs, web_results = [], []
else:
payload, docs, web_results = await self._prepare_chat(partitions, payload)
payload, docs, web_results = await self._prepare_chat(partitions, payload, llm)
sources = prepare_sources(docs, web_results)

payload["messages"] = self._sanitize_messages(payload["messages"])
llm_stream = self._llm.stream_chat(payload["messages"], **_sampling(payload))
llm_stream = llm.stream_chat(payload["messages"], **_sampling(payload))
async for sse_line in stream_with_source_filtering(llm_stream, sources, model_name):
yield sse_line

Expand All @@ -460,13 +514,14 @@ async def complete(
prepare_sources: PrepareSources,
) -> dict:
"""Non-streaming text completion → finalized OpenAI dict."""
llm = self._resolve_llm(partitions)
if partitions is None:
docs = []
else:
payload, docs = await self._prepare_completions(partitions, payload)
payload, docs = await self._prepare_completions(partitions, payload, llm)
sources = prepare_sources(docs, [])

resp = await self._llm.generate(payload["prompt"], **_sampling(payload, key="prompt"))
resp = await llm.generate(payload["prompt"], **_sampling(payload, key="prompt"))
text = resp.get("choices", [{}])[0].get("text", "") or ""
clean, citations = extract_and_strip_sources_block(text)
resp["choices"][0]["text"] = clean
Expand Down
Loading
Loading