Merge branch 'main' into anydoc-support

Conflicts, and how each was taken:

- application/core/settings.py — ours. The renamed OCR_ENABLED /
  OCR_ATTACHMENTS_ENABLED / OCR_MIN_CHARS_PER_PAGE accept main's
  DOCLING_OCR_* spellings as AliasChoices, so nothing is dropped.
- application/Dockerfile — both. Main's install layers plus the
  INSTALL_DOCLING build arg.
- application/parser/file/constants.py — both imports.
- deployment/docker-compose.yaml — both. The INSTALL_DOCLING /
  INSTALL_TESSERACT build args on backend and worker, and main's
  -Q docsgpt,parsing,embeddings, which query embedding needs.
- tests/conftest.py — theirs. Both sides fixed the same pytest-postgresql
  9.0.0 autocommit= breakage; main's spelling is the one already on main.
- application/requirements.txt — the comments claimed different reasons
  torch is in core. Main's is the true one now: it removed
  sentence-transformers, so docling is torch's only remaining consumer.

Two things the merge broke without conflicting:

- onnxruntime. This branch moved it out of core into the docling extra;
  main meanwhile made it the runtime local embeddings execute on
  (fastembed). Git took the deletion, leaving fastembed with no pinned
  runtime in a repo that pins everything. Restored to core, and no longer
  pinned twice from the extra.
- The frontend copy of ATTACHMENT_PARSER_EXTENSIONS. The backend list is
  derived and picked up the anydoc suffixes; the hand-kept frontend mirror
  did not, so the composer would refuse files the API accepts.
  tests/parser/file/test_constants.py is what caught it.

ruff, pytest (9897 passed), frontend build and docs build all pass. The
image build is unverified: no Docker daemon on this machine.
This commit is contained in:
Alex committed 2026-09-04 16:42:48 +01:00
commit 3947c66cda
97 files changed
+8661 -1049

No files matched your search

+20 -6
View File
@@ -17,11 +17,6 @@ RUN if [ -f /usr/bin/python3.12 ]; then \
echo "Python 3.12 not found"; exit 1; \
fi
# Download and unzip the model
RUN wget https://d3dg1063dc54p9.cloudfront.net/models/embeddings/mpnet-base-v2.zip && \
unzip mpnet-base-v2.zip -d models && \
rm mpnet-base-v2.zip
# Install Rust
RUN wget -q -O - https://sh.rustup.rs | sh -s -- -y
@@ -91,7 +86,26 @@ RUN groupadd -r appuser && \
# Copy the virtual environment and model from the builder stage
COPY --from=builder /venv /venv
COPY --from=builder /models /app/models
# Pre-fetch the embedding models into FastEmbed's cache so a fresh container
# does not download on first ingest and an air-gapped install works at all.
# Both defaults are baked: an upgraded deployment keeps using mpnet until it
# runs the re-embed script, while a new one starts on granite.
# The prefetch writes hub-layout snapshots (including tokenizer.json) here, so
# HF_HUB_CACHE has to point at the same directory: chunking loads the tokenizer
# through ``tokenizers``, which reads the hub cache and would otherwise fetch
# over the network on first ingest -- and fall back to cl100k when offline.
ENV EMBEDDINGS_CACHE_DIR=/app/models \
HF_HUB_CACHE=/app/models
# Only the modules the prefetch imports are copied first. It reaches nothing
# beyond model_registry, which is stdlib-only, so keeping the full source copy
# below this layer stops an unrelated edit from re-downloading ~780 MB of model
# artifacts on every build.
COPY __init__.py /app/application/__init__.py
COPY scripts/__init__.py scripts/prefetch_models.py /app/application/scripts/
COPY vectorstore/__init__.py vectorstore/model_registry.py /app/application/vectorstore/
ARG EMBEDDINGS_PREFETCH=""
RUN PYTHONPATH=/app /venv/bin/python -m application.scripts.prefetch_models ${EMBEDDINGS_PREFETCH}
# Copy your application code
COPY . /app/application
@@ -0,0 +1,35 @@
"""0031 token_usage cache tokens — persist the prompt-cache breakdown.
Providers report how much of each prompt was served from the prompt cache
(``cached_tokens``) and, on newer OpenAI-family models, how much was written
to it (``cache_write_tokens``). The LLM clients already parsed both and the
usage layer discarded them, so ``token_usage`` could not show a cache hit
rate. These two nullable columns carry the breakdown; NULL means the
provider reported nothing (distinct from 0). ``prompt_tokens`` keeps its
meaning as the provider's total. Idempotent both ways.
Revision ID: 0031_token_usage_cache_tokens
Revises: 0030_superseded_messages
"""
from typing import Sequence, Union
from alembic import op
revision: str = "0031_token_usage_cache_tokens"
down_revision: Union[str, None] = "0030_superseded_messages"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.execute("ALTER TABLE token_usage ADD COLUMN IF NOT EXISTS cached_tokens integer;")
op.execute(
"ALTER TABLE token_usage ADD COLUMN IF NOT EXISTS cache_write_tokens integer;"
)
def downgrade() -> None:
op.execute("ALTER TABLE token_usage DROP COLUMN IF EXISTS cache_write_tokens;")
op.execute("ALTER TABLE token_usage DROP COLUMN IF EXISTS cached_tokens;")
+44 -2
View File
@@ -39,6 +39,10 @@ from application.stt.stt_creator import STTCreator
from application.tts.tts_creator import TTSCreator
from application.upload_limits import (
copy_upload_to_path,
enforce_parseable_attachment,
is_unsupported_upload_message,
UnsupportedUploadTypeError,
unsupported_upload_message,
upload_limit_message,
UploadTooLargeError,
)
@@ -125,11 +129,27 @@ def _enforce_uploaded_audio_size_limit(file, filename: str) -> None:
enforce_audio_file_size_limit(size_bytes)
def _get_store_attachment_user_error(exc: Exception) -> str:
def _get_store_attachment_user_error(exc: Exception, filename: str | None = None) -> str:
"""Map an upload failure to a fixed client-facing message.
Every branch returns text this module composes itself. Nothing is read off
the exception, so no exception state — message or traceback — can reach a
response body (CodeQL py/stack-trace-exposure).
Args:
exc: The exception raised while processing one uploaded file.
filename: That file's name, which supplies the extension for the
unsupported-type message.
Returns:
The message to report for this failure.
"""
if isinstance(exc, AudioFileTooLargeError):
return build_stt_file_size_limit_message()
if isinstance(exc, UploadTooLargeError):
return upload_limit_message()
if isinstance(exc, UnsupportedUploadTypeError):
return unsupported_upload_message(filename)
return "Failed to process file"
@@ -214,6 +234,10 @@ class StoreAttachment(Resource):
with tempfile.TemporaryDirectory() as temp_dir:
staged_path = os.path.join(temp_dir, original_filename)
copy_upload_to_path(file, staged_path)
# Refuse here — nothing is stored or queued yet — so a
# video the picker let through never reaches the
# worker's reader, which would open it as text.
enforce_parseable_attachment(staged_path, original_filename)
with open(staged_path, "rb") as staged_stream:
staged_upload = FileStorage(
stream=staged_stream,
@@ -242,7 +266,9 @@ class StoreAttachment(Resource):
errors.append({
"upload_index": idx,
"filename": file.filename,
"error": _get_store_attachment_user_error(file_err),
"error": _get_store_attachment_user_error(
file_err, file.filename
),
})
if not tasks:
@@ -263,6 +289,22 @@ class StoreAttachment(Resource):
),
413,
)
if errors and all(
is_unsupported_upload_message(error.get("error")) for error in errors
):
# Tell the user which type was refused rather than the
# generic copy — this is the branch a phone picker that
# ignores ``accept`` lands in.
return make_response(
jsonify(
{
"success": False,
"message": errors[0]["error"],
"errors": errors,
}
),
400,
)
return make_response(
jsonify({"status": "error", "message": "No valid files to upload"}),
400,
+17 -3
View File
@@ -11,6 +11,7 @@ from application.api.answer.services.conversation_service import (
TERMINATED_RESPONSE_PLACEHOLDER,
)
from application.storage.db.base_repository import looks_like_uuid, row_to_dict
from application.storage.db.repositories.agents import AgentsRepository
from application.storage.db.repositories.attachments import AttachmentsRepository
from application.storage.db.repositories.conversations import ConversationsRepository
from application.storage.db.repositories.message_events import MessageEventsRepository
@@ -357,10 +358,20 @@ class SubmitFeedback(Resource):
description="Submit feedback for a conversation",
)
def post(self):
data = request.get_json() or {}
decoded_token = request.decoded_token
# api_key callers (widgets) carry no JWT; resolve the key to its owner.
scoped_api_key = None
if not decoded_token:
return make_response(jsonify({"success": False}), 401)
data = request.get_json()
api_key = data.get("api_key")
agent = None
if api_key:
with db_readonly() as conn:
agent = AgentsRepository(conn).find_by_key(api_key)
if not agent:
return make_response(jsonify({"success": False}), 401)
decoded_token = {"sub": agent.get("user_id")}
scoped_api_key = api_key
required_fields = ["feedback", "conversation_id", "question_index"]
missing_fields = check_required_fields(data, required_fields)
if missing_fields:
@@ -387,7 +398,10 @@ class SubmitFeedback(Resource):
with db_session() as conn:
repo = ConversationsRepository(conn)
conv = repo.get_any(data["conversation_id"], user_id)
if conv is None:
# A key may only rate conversations it created.
if conv is None or (
scoped_api_key and conv.get("api_key") != scoped_api_key
):
return make_response(
jsonify({"success": False, "message": "Not found"}), 404
)
+6
View File
@@ -800,6 +800,12 @@ class CreateWikiSource(Resource):
config={"kind": "wiki"},
directory_structure={},
tokens=0,
# Wiki pages are embedded like any other source, so record
# which model did it. Left unset, the column reads as NULL,
# which the boot mismatch check takes to mean "pre-dates the
# column, therefore the legacy model" -- and reports a
# correctly-embedded source as stale on every startup.
model=settings.EMBEDDINGS_NAME,
)
if initial_content:
WikiPagesRepository(conn).upsert(
+10
View File
@@ -35,6 +35,10 @@ from application.storage.db.bootstrap import ( # noqa: E402
ensure_database_ready,
ensure_vector_schema,
)
from application.storage.db.embeddings_pin import ( # noqa: E402
resolve_embeddings_pin,
warn_on_source_model_mismatch,
)
from application.stt.upload_limits import ( # noqa: E402
build_stt_file_size_limit_message,
should_reject_stt_request,
@@ -62,6 +66,12 @@ ensure_database_ready(
logger=logging.getLogger("application.app"),
)
# Which embedding model this installation uses is a property of its index, not of
# the release. Resolve it before the vector schema hook below, which sizes the
# table from EMBEDDINGS_NAME, and before anything embeds.
resolve_embeddings_pin(logging.getLogger("application.app"))
warn_on_source_model_mismatch(logging.getLogger("application.app"))
# Own the vector DB's schema here too, so the retrieval hot path is pure reads
# instead of re-running DDL for every source of every request.
if settings.AUTO_VECTOR_SCHEMA:
+17 -2
View File
@@ -115,9 +115,24 @@ def _trim_native_heap() -> None:
pass
# Tasks that allocate almost nothing and run on a latency-sensitive path, so
# the reclaim below costs far more than it recovers. Query embedding is one:
# measured at ~86 ms for the collect against ~8 ms for the embed itself on a
# worker holding the ONNX model, i.e. a 9x slowdown of the whole round trip.
_NO_RECLAIM_TASKS = frozenset({"application.vectorstore.embeddings_tasks.embed_texts"})
@task_postrun.connect
def _reclaim_memory_after_task(*args, **kwargs):
"""Drop per-task allocations so the prefork child's RSS doesn't ratchet."""
def _reclaim_memory_after_task(task=None, **kwargs):
"""Drop per-task allocations so the prefork child's RSS doesn't ratchet.
Skipped for the tasks in :data:`_NO_RECLAIM_TASKS`. This exists for the
large transient allocations docling/torch parsing makes; running a full
generational collect after a task that allocated a few kilobytes just
charges the next task for walking the whole heap.
"""
if getattr(task, "name", None) in _NO_RECLAIM_TASKS:
return
gc.collect()
torch = sys.modules.get("torch")
if torch is not None:
+13 -2
View File
@@ -13,7 +13,10 @@ result_serializer = 'json'
accept_content = ['json']
# Autodiscover tasks
imports = ('application.api.user.tasks',)
imports = (
'application.api.user.tasks',
'application.vectorstore.embeddings_tasks',
)
# Project-scoped queue so a stray sibling worker on the same broker
# (other repo, same default ``celery`` queue) can't grab DocsGPT tasks.
@@ -25,8 +28,13 @@ task_default_routing_key = "docsgpt"
# Celery worker (headless/scheduled agent) is served by a separate parsing worker
# and never self-deadlocks the awaiting worker. The tool also passes the queue at
# apply_async time, so this routing is the default for any other enqueuer.
# Query embedding gets its own queue for the same reason parsing does: a query
# waiting behind a multi-minute ingest is a query that has timed out. A bare
# worker still consumes it, but its concurrency is shared -- run a separate
# ``-Q embeddings`` worker to actually isolate query latency from ingest.
task_routes = {
"application.api.user.tasks.parse_document": {"queue": settings.DOCUMENT_PARSE_QUEUE},
"application.vectorstore.embeddings_tasks.embed_texts": {"queue": settings.EMBEDDINGS_QUEUE},
}
# Declare every queue so a bare ``celery worker`` (no -Q) consumes ALL of them —
@@ -34,7 +42,10 @@ task_routes = {
# heavy OCR isolated run one worker with ``-Q docsgpt`` and another with
# ``-Q parsing``. (dict.fromkeys dedupes if DOCUMENT_PARSE_QUEUE == "docsgpt".)
task_queues = tuple(
Queue(name) for name in dict.fromkeys(["docsgpt", settings.DOCUMENT_PARSE_QUEUE])
Queue(name)
for name in dict.fromkeys(
["docsgpt", settings.DOCUMENT_PARSE_QUEUE, settings.EMBEDDINGS_QUEUE]
)
)
beat_scheduler = "redbeat.RedBeatScheduler"
+109 -164
View File
@@ -33,10 +33,8 @@ class Settings(BaseSettings):
OIDC_GROUPS_CLAIM: str = "groups" # ID-token/userinfo claim carrying group membership
OIDC_ADMIN_GROUPS: Optional[str] = None # comma-separated groups granted admin; unset = no OIDC admin mapping
# RBAC (admin/user roles). Persisted admin grants live in the user_roles
# table and apply only under AUTH_TYPE=oidc. LOCAL_MODE_ADMIN is the only
# non-DB admin path and applies only to AUTH_TYPE=None (no-auth self-host).
# It MUST stay False in any networked deployment.
# RBAC: persisted admin grants live in user_roles (AUTH_TYPE=oidc only). This is the
# only non-DB admin path, for AUTH_TYPE=None self-host. MUST stay False if networked.
LOCAL_MODE_ADMIN: bool = False
# SCIM 2.0 provisioning (IdP-driven user create/deactivate at /scim/v2)
@@ -45,41 +43,56 @@ class Settings(BaseSettings):
LLM_PROVIDER: str = "docsgpt"
LLM_NAME: Optional[str] = None # if LLM_PROVIDER is openai, LLM_NAME can be gpt-4 or gpt-3.5-turbo
# Legacy model on purpose: an install that never pinned this has vectors from it, and
# granite is the same width so a swap would fail silently. New installs get granite from
# .env-template; existing ones switch by setting this and running application.scripts.reembed.
EMBEDDINGS_NAME: str = "huggingface_sentence-transformers/all-mpnet-base-v2"
EMBEDDINGS_BASE_URL: Optional[str] = None # Remote embeddings API URL (OpenAI-compatible)
EMBEDDINGS_KEY: Optional[str] = None # api key for embeddings (if using openai, just copy API_KEY)
EMBEDDINGS_MAX_INPUT_TOKENS: Optional[int] = None # truncate each remote embed input to N tokens (overflow lost)
EMBEDDINGS_BATCH_SIZE: int = 32 # chunks per embed request during ingest (1 = legacy per-chunk behaviour)
EMBEDDINGS_BATCH_SIZE: int = 32 # chunks per store transaction / remote embed request
# Documents per local ONNX forward pass. Each pass pads to its longest input, and that
# waste grows with the square of chunk length: at 1250 tokens, 32 peaked at 6.6 GB, 1 at 2.9 GB.
EMBEDDINGS_MODEL_BATCH_SIZE: int = 1
# Intra-op threads for the local ONNX runner; None = every core. It scales sub-linearly,
# so several single-threaded workers beat one many-threaded process on the same cores.
EMBEDDINGS_THREADS: Optional[int] = None
EMBEDDINGS_CACHE_DIR: Optional[str] = None # where FastEmbed caches model artifacts
# Pooling ("cls"/"mean") and L2 normalisation. Read from the model's own repository;
# set these only for a repository that declares neither, or to override what it declares.
EMBEDDINGS_POOLING: Optional[str] = None
EMBEDDINGS_NORMALIZE: Optional[bool] = None
# Embed on the worker so the API holds no model (~890 MB), at one broker round trip per
# query. Ignored when EMBEDDINGS_BASE_URL is set, which is the better answer for production.
EMBEDDINGS_DELEGATE_TO_WORKER: bool = True
EMBEDDINGS_QUEUE: str = "embeddings" # queue the embed task is routed to
EMBEDDINGS_DELEGATE_TIMEOUT: int = 60 # seconds to wait for the worker
GITHUB_INGEST_MAX_FILE_BYTES: int = 1048576 # skip repo blobs larger than this (0 = no cap)
GITHUB_INGEST_MAX_WORKERS: int = 8 # parallel file fetches per GitHub repo ingest
# Optional directory of operator-supplied model YAMLs, loaded after the
# built-in catalog under application/core/models/. Later wins on
# Operator-supplied model YAMLs, loaded after the built-in catalog; later wins on
# duplicate model id. See application/core/models/README.md.
MODELS_CONFIG_DIR: Optional[str] = None
CELERY_BROKER_URL: str = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND: str = "redis://localhost:6379/1"
# Prefetch=1 caps SIGKILL loss to one task. Visibility timeout must exceed
# the longest legitimate task runtime (ingest, agent webhook) but stay
# short enough that SIGKILLed tasks redeliver promptly. 1h matches Onyx
# and Dify defaults; long ingests can override via env.
# Prefetch=1 caps SIGKILL loss to one task. Visibility timeout must exceed the longest
# legitimate task runtime but stay short enough that SIGKILLed tasks redeliver promptly.
CELERY_WORKER_PREFETCH_MULTIPLIER: int = 1
CELERY_VISIBILITY_TIMEOUT: int = 3600
# Recycle the prefork worker child once its resident size crosses this many
# kilobytes — backstops native-heap growth from docling/torch parsing. 0 disables.
# Recycle a prefork child past this resident size in KB; backstops docling/torch heap growth.
# Checked between tasks, so it does not bound the peak within one. 0 disables.
CELERY_WORKER_MAX_MEMORY_PER_CHILD: int = 4194304
# Recycle the child after this many tasks; 0 disables (memory cap is the primary knob).
CELERY_WORKER_MAX_TASKS_PER_CHILD: int = 0
CELERY_WORKER_MAX_TASKS_PER_CHILD: int = 0 # recycle after N tasks; 0 disables
# Only consulted when VECTOR_STORE=mongodb or when running scripts/db/backfill.py; user data lives in Postgres.
MONGO_URI: Optional[str] = None
# User-data Postgres DB.
POSTGRES_URI: Optional[str] = None
# On app startup, apply pending Alembic migrations. Default ON for dev; disable in prod if you manage schema out-of-band.
# On startup, apply pending Alembic migrations. Disable if you manage schema out-of-band.
AUTO_MIGRATE: bool = True
# On app startup, create the target Postgres database if it's missing (requires CREATEDB privilege). Dev-friendly default.
# On startup, create the target Postgres database if missing (needs CREATEDB privilege).
AUTO_CREATE_DB: bool = True
# On app startup, create the pgvector/graph tables and verify the embedding dimension. Set False to manage them
# out-of-band — there is no Alembic migration for the vector DB because it may be a separate cluster.
# On startup, create the pgvector/graph tables and verify the embedding dimension. No Alembic
# migration covers the vector DB (it may be a separate cluster); set False to manage it yourself.
AUTO_VECTOR_SCHEMA: bool = True
LLM_PATH: str = os.path.join(current_dir, "models/docsgpt-7b-f16.gguf")
DEFAULT_MAX_HISTORY: int = 150
@@ -94,8 +107,7 @@ class Settings(BaseSettings):
"request_limit": 500,
}
UPLOAD_FOLDER: str = "inputs"
# Public upload request cap is applied by Flask before multipart parsing.
# The per-file cap is also enforced while copying each controlled stream.
# Request cap is applied by Flask before multipart parsing; the per-file cap also while copying.
UPLOAD_MAX_REQUEST_BYTES: int = Field(default=256 * 1024 * 1024, gt=0)
UPLOAD_MAX_FILE_BYTES: int = Field(default=100 * 1024 * 1024, gt=0)
PARSE_SPEC_MAX_BYTES: int = Field(default=10 * 1024 * 1024, gt=0)
@@ -207,30 +219,22 @@ class Settings(BaseSettings):
# unaffected and always uses docling, because chunking and retrieval do
# depend on that structure.
ATTACHMENT_PDF_TEXT_FAST_PATH: bool = True
# Median characters per sampled page below which a PDF is treated as a scan
# and handed to docling. Measured separation on real uploads: scans at
# 0-17 chars/page, text-layer documents at 433-6834.
# Median chars per sampled page below which a PDF is treated as a scan and handed to docling.
# Measured on real uploads: scans at 0-17 chars/page, text-layer documents at 433-6834.
ATTACHMENT_PDF_TEXT_MIN_MEDIAN_CHARS: int = 32
ATTACHMENT_TEXT_MAX_BYTES: int = 5_000_000
AGENT_IMAGE_MAX_BYTES: int = 5_000_000
AGENT_IMAGE_MAX_PIXELS: int = 16_777_216
VECTOR_STORE: str = "faiss" # "faiss" or "elasticsearch" or "qdrant" or "milvus" or "lancedb" or "pgvector"
# Allow-list of retriever keys an agent may use. Values must match the
# ``RetrieverCreator.retrievers`` registry keys (``classic`` / ``default``),
# Retriever keys an agent may use; must match RetrieverCreator.retrievers registry keys,
# NOT the legacy ``classic_rag`` label which never matched the registry.
RETRIEVERS_ENABLED: list = ["classic", "default"]
# Concurrent per-source vector searches within one retrieval (multi-source chats);
# the query is embedded once and shared across sources.
# Concurrent per-source searches in one retrieval; the query is embedded once and shared.
RETRIEVAL_MAX_PARALLEL_SOURCES: int = 4
# Kill-switch for per-source retrieval dispatch. When False the retrieval
# path collapses to today's single-retriever behavior (consumed by the
# Dispatcher in a later change; defined here so the flag exists up front).
# Kill-switch for per-source retrieval dispatch; False collapses to a single retriever.
PER_SOURCE_RETRIEVAL_ENABLED: bool = True
# Flagship GraphRAG flag. Reserved and unused for now; gates graph-aware
# ingestion/retrieval when that feature lands.
GRAPHRAG_ENABLED: bool = False
# Model for ingest-time graph extraction; None reuses the instance default
# model (LLM_PROVIDER/LLM_NAME). Operator-overridable (e.g. a cheaper model).
GRAPHRAG_ENABLED: bool = False # gates graph-aware ingestion/retrieval
# Model for ingest-time graph extraction; None reuses LLM_PROVIDER/LLM_NAME.
GRAPHRAG_EXTRACTION_MODEL: Optional[str] = None
# Hard cap on chunks extracted per source (cost control).
GRAPHRAG_MAX_CHUNKS_FOR_EXTRACTION: int = 2000
@@ -293,9 +297,8 @@ class Settings(BaseSettings):
ELASTIC_URL: Optional[str] = None # url for elasticsearch
ELASTIC_INDEX: Optional[str] = "docsgpt" # index name for elasticsearch
# Legacy AWS credentials from the retired SageMaker LLM provider.
# Still read as a deprecated fallback by S3 storage (see the S3_*
# block below); do not use for new deployments.
# Legacy AWS credentials from the retired SageMaker provider. Still read as a deprecated
# fallback by S3 storage; do not use for new deployments.
SAGEMAKER_REGION: Optional[str] = None
SAGEMAKER_ACCESS_KEY: Optional[str] = None
SAGEMAKER_SECRET_KEY: Optional[str] = None
@@ -315,16 +318,11 @@ class Settings(BaseSettings):
QDRANT_PATH: Optional[str] = None
QDRANT_DISTANCE_FUNC: str = "Cosine"
# PGVector vectorstore config. Write the URI in whichever form you
# prefer — ``postgres://``, ``postgresql://``, or even the SQLAlchemy
# dialect form (``postgresql+psycopg://``) are all accepted and
# normalized internally for ``psycopg.connect()``.
# PGVector config. postgres://, postgresql:// and postgresql+psycopg:// are all accepted
# and normalized internally for psycopg.connect().
PGVECTOR_CONNECTION_STRING: Optional[str] = None
# Per-process psycopg connection pool for the vector store; 0 = one direct
# connection per store instance, the legacy behaviour.
PGVECTOR_POOL_MAX_SIZE: int = 8
# IVFFlat probes for vector search. ``None`` derives sqrt(lists) from the
# index itself; set an integer to pin it. Higher = better recall, more scan.
PGVECTOR_POOL_MAX_SIZE: int = 8 # per-process pool; 0 = one direct connection per store
# IVFFlat probes; None derives sqrt(lists) from the index. Higher = better recall, more scan.
PGVECTOR_IVFFLAT_PROBES: Optional[int] = None
# Milvus vectorstore config
MILVUS_COLLECTION_NAME: Optional[str] = "docsgpt"
@@ -338,11 +336,8 @@ class Settings(BaseSettings):
FLASK_DEBUG_MODE: bool = False
STORAGE_TYPE: str = "local" # local or s3
# S3-compatible object storage (used when STORAGE_TYPE=s3). Works with AWS
# S3 and any S3-compatible service (MinIO, Cloudflare R2, Backblaze B2,
# DigitalOcean Spaces, ...). For non-AWS services, set S3_ENDPOINT_URL and
# usually S3_PATH_STYLE=true. The SAGEMAKER_* credentials are still read as
# a deprecated fallback for backward compatibility.
# S3-compatible object storage (STORAGE_TYPE=s3): AWS S3, MinIO, R2, B2, Spaces, ...
# For non-AWS, set S3_ENDPOINT_URL and usually S3_PATH_STYLE=true.
S3_BUCKET_NAME: str = "docsgpt-test-bucket"
S3_ENDPOINT_URL: Optional[str] = None # custom endpoint for S3-compatible services; omit for AWS
S3_ACCESS_KEY_ID: Optional[str] = None
@@ -371,34 +366,21 @@ class Settings(BaseSettings):
# Tool pre-fetch settings
ENABLE_TOOL_PREFETCH: bool = True
# When True, OpenAI Responses API calls are persisted server-side
# (store=true) so a previous_response_id can chain turns. When False
# (the default) Responses calls are stateless (store=false) and any
# reasoning is carried across the in-turn tool loop via encrypted
# reasoning items instead.
# True persists Responses API calls server-side so previous_response_id can chain turns.
# False keeps them stateless, carrying reasoning across the tool loop as encrypted items.
OPENAI_RESPONSES_STORE: bool = False
OPENAI_REASONING_SUMMARY: str = "auto"
# OpenAI-compatible clients can identify a logical chat with session
# headers even though chat-completions itself has no conversation field.
# Lets OpenAI-compatible clients identify a logical chat by session header, which
# chat-completions itself has no field for.
V1_SESSION_TTL_SECONDS: int = 24 * 60 * 60
# Optional cheaper model for first-party conversation titles. When unset,
# listed conversations use their answer model, but title work is still
# dispatched off the response path.
# Optional cheaper model for conversation titles; unset reuses the answer model.
TITLE_MODEL_ID: Optional[str] = None
# Config-free tools on by default in agentless chats. ``scheduler`` is
# dual-registered (also in ``BUILTIN_AGENT_TOOLS``) so the same synthetic id
# resolves whether reached via defaults or the agent picker.
#
# ``code_executor`` and ``artifact_generator`` belong here too on any
# deployment that runs a sandbox (see SANDBOX_BACKEND / SANDBOX_GATEWAY_URL):
# ``artifact_generator`` renders the .docx/.pdf/.xlsx/.pptx files users ask
# chat for. They are left out of the shipped default because both execute
# through the sandbox runner and would fail on every call without one — add
# them explicitly once a runner is configured:
# DEFAULT_CHAT_TOOLS = [..., "code_executor", "artifact_generator"]
# Config-free tools on by default in agentless chats. ``scheduler`` is dual-registered in
# BUILTIN_AGENT_TOOLS so one synthetic id resolves via defaults or the agent picker.
# Add "code_executor" and "artifact_generator" once a sandbox runner is configured — both
# execute through it and would fail on every call without one.
DEFAULT_CHAT_TOOLS: list = [
"memory",
"read_webpage",
@@ -411,81 +393,62 @@ class Settings(BaseSettings):
COMPRESSION_MODEL_OVERRIDE: Optional[str] = None # Use different model for compression
COMPRESSION_PROMPT_VERSION: str = "v1.0" # Track prompt iterations
COMPRESSION_MAX_HISTORY_POINTS: int = 3 # Keep only last N compression points to prevent DB bloat
COMPRESSION_RECENT_FIELD_MAX_TOKENS: int = 8000 # Per-field cap on the verbatim tail kept after a compression point (0 disables)
TOOL_RESULT_MAX_TOKENS: int = 20000 # Cap on a single tool result entering the LLM context (0 disables); journal/DB keep the full result
# Per-field cap on the verbatim tail kept after a compression point (0 disables).
COMPRESSION_RECENT_FIELD_MAX_TOKENS: int = 8000
# Cap on one tool result entering the LLM context (0 disables); journal/DB keep it whole.
TOOL_RESULT_MAX_TOKENS: int = 20000
# Agent Guardrails
# Master switch. When False, no guardrail stage runs regardless of what an
# agent's config says.
GUARDRAILS_ENABLED: bool = True
# Registry-key allowlist; values must match GuardrailCreator.checks keys.
# Empty means "every registered check".
GUARDRAILS_ENABLED: bool = True # master switch; False disables every stage
# Allowlist of GuardrailCreator.checks keys; empty means every registered check.
GUARDRAILS_CHECKS_ENABLED: list = []
# Instance floor: a GuardrailsConfig fragment every agent inherits and
# cannot weaken. Agents may add controls or make an action stricter, never
# looser. "enabled" is required — without it the floor parses but applies
# to nothing. Example:
# A GuardrailsConfig fragment every agent inherits and cannot weaken; agents may add
# controls or make an action stricter, never looser. "enabled" is required — without it
# the floor parses but applies to nothing. Example:
# {"enabled": true, "mode": "scan_all",
# "controls": [{"check": "secrets", "stage": "output",
# "action": "redact"}]}
# "controls": [{"check": "secrets", "stage": "output", "action": "redact"}]}
GUARDRAILS_FLOOR: dict = {}
# Judge model for the topic/policy checks. None reuses the request's model.
# Judge model for the topic/policy checks; None reuses the request's model.
GUARDRAILS_JUDGE_MODEL: Optional[str] = None
# Persist scanned text alongside guardrail_events. Off by default: the
# pre-redaction text is exactly the sensitive material a PII control exists
# to keep out of storage.
# Persist scanned text alongside guardrail_events. Off by default: pre-redaction text is
# exactly the material a PII control exists to keep out of storage.
GUARDRAILS_STORE_SCANNED_TEXT: bool = False
GUARDRAILS_EVENTS_RETENTION_DAYS: int = Field(default=30, ge=1)
# Internal SSE push channel (notifications + durable replay journal)
# Master switch — when False, /api/events emits a "push_disabled" comment
# and returns; clients fall back to polling. Publisher becomes a no-op.
# Internal SSE push channel (notifications + durable replay journal).
# False makes /api/events emit "push_disabled" and return; clients fall back to polling.
ENABLE_SSE_PUSH: bool = True
# Per-user durable backlog cap (~entries). At typical event rates this
# gives ~24h of replay; tune up for verbose feeds, down for memory.
# Per-user durable backlog cap in entries; ~24h of replay at typical rates.
EVENTS_STREAM_MAXLEN: int = 1000
# Bounds uvicorn's graceful-shutdown drain (uvicorn_worker doesn't forward
# --graceful-timeout). Keep below the gunicorn --timeout (180) watchdog.
# Used by gunicorn_worker.BoundedDrainUvicornWorker.
# Bounds uvicorn's shutdown drain (uvicorn_worker doesn't forward --graceful-timeout).
# Keep below the gunicorn --timeout (180) watchdog. Used by BoundedDrainUvicornWorker.
GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS: int = 30
WSGI_THREADPOOL_WORKERS: int = 96
SSE_KEEPALIVE_SECONDS: int = Field(default=15, ge=1)
# Cap on simultaneous SSE connections per user. Each connection holds
# one WSGI thread (32 per gunicorn worker) and one Redis pub/sub
# connection. 8 covers normal multi-tab use without letting one user
# starve the pool. Set to 0 to disable the cap.
# Simultaneous SSE connections per user; each holds a WSGI thread and a Redis pub/sub
# connection. 8 covers multi-tab use without one user starving the pool. 0 disables.
SSE_MAX_CONCURRENT_PER_USER: int = 8
# Per-request cap on the number of backlog entries XRANGE returns
# for ``/api/events`` snapshots. Bounds the bytes a single replay
# can move from Redis to the wire — a malicious client looping
# ``Last-Event-ID=<oldest>`` reconnects can only enumerate this
# many entries per round-trip. Combined with the per-user
# connection cap above and the windowed budget below, total
# enumeration throughput is bounded.
# Backlog entries XRANGE returns per /api/events snapshot. Bounds what one replay moves
# from Redis to the wire: a client looping Last-Event-ID reconnects enumerates at most
# this many per round-trip, and the budget below bounds total throughput.
EVENTS_REPLAY_MAX_PER_REQUEST: int = 200
EVENTS_REPLAY_MAX_AGE_HOURS: int = 48
# Sliding-window cap on snapshot replays per user. Once the budget
# is exhausted the route returns HTTP 429 with the cursor pinned;
# the client backs off and retries after the window rolls over.
# Sliding-window cap on snapshot replays per user; exhausting it returns 429 with the
# cursor pinned so the client backs off until the window rolls over.
EVENTS_REPLAY_BUDGET_REQUESTS_PER_WINDOW: int = 30
EVENTS_REPLAY_BUDGET_WINDOW_SECONDS: int = 60
# Retention for the ``message_events`` journal. The ``cleanup_message_events``
# beat task deletes rows older than this. Reconnect-replay only
# needs the journal for streams a client could still be tailing,
# so 14 days is a generous default that covers paused/tool-action
# flows without unbounded table growth.
# Retention for the message_events journal, enforced by the cleanup_message_events beat
# task. Replay only needs streams a client could still be tailing.
MESSAGE_EVENTS_RETENTION_DAYS: int = 14
# Remote Device feature.
REMOTE_DEVICE_SESSION_IDLE_SECONDS: int = 60
REMOTE_DEVICE_REQUIRE_SIGNATURE: bool = False
REMOTE_DEVICE_PAIRING_TTL_SECONDS: int = 600
# Redis-backed broker tunables (route invocations cross-process so a
# scheduled/Celery run reaches the web-held device session). The command
# queue TTL must exceed the max command drain deadline (the tool caps
# timeout_ms at 600s, drained with a +5s margin = 605s) so a queued command
# for a briefly-offline device isn't evicted before its own drain gives up.
# Redis broker tunables, routing invocations cross-process so a scheduled run reaches the
# web-held device session. The queue TTL must exceed the max drain deadline (605s) so a
# command for a briefly-offline device isn't evicted before its own drain gives up.
REMOTE_DEVICE_CMD_QUEUE_TTL_SECONDS: int = 900
REMOTE_DEVICE_INVOCATION_TTL_SECONDS: int = 900
REMOTE_DEVICE_OUTPUT_STREAM_MAXLEN: int = 10_000
@@ -500,23 +463,19 @@ class Settings(BaseSettings):
SCHEDULE_ONCE_MAX_HORIZON: int = 31_536_000
SCHEDULE_RUN_OUTPUT_RETENTION_DAYS: int = 90
# Code-execution sandbox (see artifacts-code-execution-spec.md §4 C2).
# The app is a CLIENT of an always-on runner; defaults are safe so app
# import never fails when the sandbox is unconfigured.
# Code-execution sandbox. The app is a CLIENT of an always-on runner; defaults are safe so
# app import never fails when the sandbox is unconfigured.
SANDBOX_BACKEND: str = "jupyter" # "jupyter" (self-host) | "daytona" (Daytona Cloud)
# URL of the Jupyter Kernel Gateway runner (the docsgpt-sandbox service).
SANDBOX_GATEWAY_URL: str = "http://localhost:8888"
SANDBOX_GATEWAY_AUTH_TOKEN: Optional[str] = None # gateway auth token, if set
# Kernelspec launched per session. Defaults to the env-scrubbing "docsgpt-python"
# spec (shipped by the docsgpt-sandbox runner) so kernel code cannot read the
# gateway auth token or operator secrets from os.environ. The stock "python3"
# spec inherits the gateway env verbatim and must not be used with untrusted code.
# Kernelspec per session. The env-scrubbing "docsgpt-python" spec keeps kernel code from
# reading the gateway token or operator secrets from os.environ; the stock "python3" spec
# inherits the gateway env verbatim and must not be used with untrusted code.
SANDBOX_KERNEL_NAME: str = "docsgpt-python"
SANDBOX_MAX_TTL: int = 1200 # hard cap (s) on agent-selectable keep-alive TTL
# Per-process/worker cap on concurrent live sandbox sessions. Backend-agnostic
# (complements DAYTONA_MAX_SANDBOXES); when reached, an LRU-idle session is
# evicted to make room. This bound is local to each app/worker process.
# 0 (or any non-positive value) disables the cap (unlimited sessions).
# Concurrent live sessions per process, backend-agnostic; at the cap an LRU-idle session is
# evicted. 0 or negative disables the cap.
SANDBOX_MAX_SESSIONS: int = 32
SANDBOX_EXEC_TIMEOUT: int = 60 # default wall-clock cap (s) per exec call
SANDBOX_HTTP_TIMEOUT: int = 10 # fixed cap (s) for REST control calls (create/delete/alive/interrupt)
@@ -526,44 +485,31 @@ class Settings(BaseSettings):
# ``read_document`` parsing on a dedicated Celery ``parsing`` queue (backend parser).
DOCUMENT_PARSE_QUEUE: str = "parsing" # queue the parse_document task is routed to
DOCUMENT_PARSE_TIMEOUT: int = 120 # seconds the tool awaits the enqueued parse before degrading
# The base timeout is a FLOOR: the awaited window (and the task's per-call Celery
# time limits) grow with the document's size, because OCR cost scales with pages
# -- a 30-page scan needs ~60s, a 100-page scan several minutes. Without this a
# large scan is silently dropped from the node/tool at the base window.
# The base timeout is a FLOOR: the window grows with document size, because OCR cost scales
# with pages. Without this a large scan is silently dropped at the base window.
DOCUMENT_PARSE_TIMEOUT_PER_MB: int = 60 # extra seconds of parse window per MiB of input
DOCUMENT_PARSE_TIMEOUT_MAX: int = 900 # absolute ceiling on the size-scaled parse window
DOCUMENT_PARSE_MAX_BYTES: int = 0 # cap on a parsed document's bytes (0 = reuse SANDBOX_MAX_INPUT_BYTES)
DOCUMENT_MAX_DECOMPRESSED_BYTES: int = 300 * 1024 * 1024
DOCUMENT_MAX_ARCHIVE_ENTRIES: int = 10000
# Per-agent-node cap on files passed natively to the node's LLM (vision/doc
# inputs). Files past the cap are extracted to text or dropped, not attached
# natively, to bound context/cost. Re-uses SANDBOX_MAX_INPUT_BYTES per file.
# Files per node passed natively to the LLM; past the cap they are extracted to text or
# dropped, to bound context and cost. Re-uses SANDBOX_MAX_INPUT_BYTES per file.
WORKFLOW_NODE_NATIVE_MAX_FILES: int = 5
# Per-agent-node cap on documents extracted to text via the parsing worker.
# Each non-native, non-text document issues a separate blocking parse, so a
# node referencing many documents (e.g. the ``*`` token) is bounded here to
# avoid serializing dozens of parses; documents past the cap are skipped with
# a truncation note instead of extracted.
# Documents per node extracted via the parsing worker. Each issues a separate blocking
# parse; past the cap they are skipped with a truncation note.
WORKFLOW_NODE_EXTRACT_MAX_FILES: int = 5
# Total wall clock one node may spend on blocking document parses, shared
# across all of them. The per-document window scales with size (up to
# DOCUMENT_PARSE_TIMEOUT_MAX), so without a shared budget a node could
# serialize WORKFLOW_NODE_EXTRACT_MAX_FILES full windows and hold a web
# threadpool slot for that whole time.
# Wall clock one node may spend on blocking parses, shared across all of them. Without it a
# node could serialize WORKFLOW_NODE_EXTRACT_MAX_FILES full windows on a web threadpool slot.
WORKFLOW_NODE_EXTRACT_BUDGET_SECONDS: int = 900
# A workflow run row is pre-created as ``running`` and finalized when its
# generator completes; a client disconnect or worker crash can strand it in
# ``running`` forever. The beat reaper fails runs still ``running`` past this
# many seconds. Generous so a legitimately long run is never cut off.
# A run row is pre-created as ``running``; a disconnect or crash can strand it there. The
# beat reaper fails runs still ``running`` past this. Generous so a long run is never cut off.
WORKFLOW_RUN_STALE_SECONDS: int = 3600
# Runner container resource caps — consumed by the docsgpt-sandbox compose
# service (deployment/sandbox), not by the app client. cgroup CPU/mem caps
# are part of the untrusted-code security boundary.
# Runner container caps, consumed by the docsgpt-sandbox compose service, not the app.
# These cgroup limits are part of the untrusted-code security boundary.
SANDBOX_MEMORY: str = "1g" # docker mem_limit for the runner container
SANDBOX_CPUS: str = "1.0" # docker cpu quota for the runner container
# Daytona Cloud managed backend (used only when SANDBOX_BACKEND="daytona").
# The app is a REST client of Daytona Cloud authenticated by DAYTONA_API_KEY;
# all knobs are optional so app import never fails when the backend is unused.
# Daytona Cloud backend (SANDBOX_BACKEND="daytona"). All knobs are optional so app import
# never fails when the backend is unused.
DAYTONA_API_KEY: Optional[str] = None # Daytona Cloud API key (secret)
DAYTONA_API_URL: Optional[str] = None # override Daytona API base URL, if self-targeting
DAYTONA_TARGET: Optional[str] = None # Daytona region/target, e.g. "us"
@@ -572,8 +518,7 @@ class Settings(BaseSettings):
DAYTONA_AUTO_STOP_INTERVAL: int = 15 # minutes idle before Daytona auto-stops a sandbox (0 disables)
DAYTONA_AUTO_DELETE_INTERVAL: int = 60 # minutes after stop before Daytona auto-deletes (-1 disables)
DAYTONA_MAX_SANDBOXES: int = 50 # cap on concurrent live Daytona sandboxes (cost-DoS guard)
# Per-user artifact quotas (generous defaults; enforced at persistence time).
# For all three, 0 (or any non-positive value) disables that quota (unlimited).
# Per-user artifact quotas, enforced at persistence time. 0 or negative disables a quota.
ARTIFACT_MAX_BYTES: int = 50 * 1024 * 1024 # cap on a single stored artifact version's bytes
ARTIFACT_MAX_COUNT_PER_USER: int = 5000 # cap on artifacts a user may own
ARTIFACT_MAX_TOTAL_BYTES_PER_USER: int = 5 * 1024 * 1024 * 1024 # cap on a user's total stored bytes
+4 -2
View File
@@ -157,12 +157,14 @@ class GraphStore:
"""Dimension of the configured embeddings model, matching ``PGVectorStore``.
Falls back to ``DEFAULT_NAME_EMBEDDING_DIM`` so the graph table and the
pgvector ``documents`` table always agree on the configured model.
pgvector ``documents`` table always agree on the configured model. A
model outside the registry reports ``None`` rather than no attribute,
so the fallback cannot be left to ``getattr``.
"""
from application.vectorstore.base import get_embeddings
embedding = get_embeddings()
return getattr(embedding, "dimension", DEFAULT_NAME_EMBEDDING_DIM)
return getattr(embedding, "dimension", None) or DEFAULT_NAME_EMBEDDING_DIM
@staticmethod
def create_schema(conn, *, dimension: int = DEFAULT_NAME_EMBEDDING_DIM) -> None:
+10 -6
View File
@@ -6,7 +6,7 @@ from typing import Any, Dict, Generator, List, Optional, Tuple
from anthropic import Anthropic
from application.core.settings import settings
from application.llm.base import BaseLLM
from application.llm.base import BaseLLM, optional_int
from application.storage.storage_creator import StorageCreator
logger = logging.getLogger(__name__)
@@ -387,8 +387,12 @@ class AnthropicLLM(BaseLLM):
# message_start are reused below instead.
if not output_only:
base_input = int(getattr(usage, "input_tokens", 0) or 0)
created = int(getattr(usage, "cache_creation_input_tokens", 0) or 0)
cached = int(getattr(usage, "cache_read_input_tokens", 0) or 0)
# None (unreported) and 0 (reported, no caching) must stay
# distinguishable — see ``optional_int``.
created = optional_int(
getattr(usage, "cache_creation_input_tokens", None)
)
cached = optional_int(getattr(usage, "cache_read_input_tokens", None))
except (TypeError, ValueError):
return
@@ -398,11 +402,11 @@ class AnthropicLLM(BaseLLM):
prompt = int(previous.get("prompt_tokens", 0) or 0)
details = previous.get("prompt_tokens_details")
else:
prompt = base_input + created + cached
prompt = base_input + (created or 0) + (cached or 0)
details = {}
if cached:
if cached is not None:
details["cached_tokens"] = cached
if created:
if created is not None:
details["cache_creation_tokens"] = created
details = details or None
+29
View File
@@ -33,6 +33,29 @@ _STREAM_RETRYABLE_TRANSPORT_ERRORS = (
)
def optional_int(value) -> Optional[int]:
"""Coerce a provider-reported count to int, keeping "unreported" as None.
Cache bins persist as NULL when a provider says nothing and as 0 when it
reports zero, so the two must stay distinguishable all the way from the
usage object: providers report ``cached_tokens: 0`` on every uncached
request, and folding that into "unknown" would file the ordinary case as
no-data.
Args:
value: The raw attribute off a provider usage object.
Returns:
The integer value, or None when absent or not a number.
"""
if value is None:
return None
try:
return int(value)
except (TypeError, ValueError):
return None
class BaseLLM(ABC):
# Stamped onto the ``llm_stream_start`` event so dashboards can group
# calls by vendor. Subclasses override.
@@ -710,6 +733,7 @@ class BaseLLM(ABC):
completion_tokens,
latency_ms,
cached_tokens=None,
cache_write_tokens=None,
error=None,
):
# Non-streaming counterpart to ``_emit_stream_finished_log``. Paired
@@ -730,6 +754,8 @@ class BaseLLM(ABC):
}
if cached_tokens is not None:
extra["cached_tokens"] = int(cached_tokens)
if cache_write_tokens is not None:
extra["cache_write_tokens"] = int(cache_write_tokens)
if error is not None:
extra["error_class"] = type(error).__name__
logging.info("llm_gen_finished", extra=extra)
@@ -757,6 +783,7 @@ class BaseLLM(ABC):
completion_tokens,
latency_ms,
cached_tokens=None,
cache_write_tokens=None,
error=None,
):
# Paired with ``llm_stream_start`` so cost dashboards can sum tokens
@@ -774,6 +801,8 @@ class BaseLLM(ABC):
}
if cached_tokens is not None:
extra["cached_tokens"] = int(cached_tokens)
if cache_write_tokens is not None:
extra["cache_write_tokens"] = int(cache_write_tokens)
if error is not None:
extra["error_class"] = type(error).__name__
logging.info("llm_stream_finished", extra=extra)
+38 -13
View File
@@ -9,7 +9,7 @@ from typing import Any, Callable
from openai import BadRequestError, OpenAI
from application.core.settings import settings
from application.llm.base import BaseLLM
from application.llm.base import BaseLLM, optional_int
from application.storage.storage_creator import StorageCreator
# Placeholder sent to OpenAI-compatible backends that require no credentials.
@@ -1305,19 +1305,38 @@ class OpenAILLM(BaseLLM):
}
input_details = getattr(usage, "prompt_tokens_details", None)
output_details = getattr(usage, "completion_tokens_details", None)
try:
cached = int(getattr(input_details, "cached_tokens", 0) or 0)
reasoning = int(getattr(output_details, "reasoning_tokens", 0) or 0)
except (TypeError, ValueError):
cached = 0
reasoning = 0
if cached:
result["prompt_tokens_details"] = {"cached_tokens": cached}
cached = optional_int(getattr(input_details, "cached_tokens", None))
written = optional_int(getattr(input_details, "cache_write_tokens", None))
reasoning = optional_int(
getattr(output_details, "reasoning_tokens", None)
)
details = self._prompt_cache_details(cached, written)
if details:
result["prompt_tokens_details"] = details
if reasoning:
result["completion_tokens_details"] = {"reasoning_tokens": reasoning}
self._last_usage = result
self._last_usage_claimed = False
@staticmethod
def _prompt_cache_details(cached: int | None, written: int | None) -> dict:
"""Build the ``prompt_tokens_details`` breakdown from the two cache bins.
A bin is included whenever the provider reported it, zero included:
downstream, an absent bin persists as NULL ("we don't know") and a
reported zero as 0 ("no cache hits"), and OpenAI reports
``cached_tokens: 0`` on every uncached request. Collapsing the two
would file every ordinary request as unknown. ``cache_write_tokens``
is reported by newer OpenAI-family deployments that charge for cache
writes.
"""
details = {}
if cached is not None:
details["cached_tokens"] = cached
if written is not None:
details["cache_write_tokens"] = written
return details
@staticmethod
def _function_call_ids(response):
"""call_ids of every ``function_call`` item in a Responses output.
@@ -1356,10 +1375,16 @@ class OpenAILLM(BaseLLM):
"completion_tokens": completion,
"total_tokens": int(getattr(usage, "total_tokens", 0) or prompt + completion),
}
cached = int(getattr(input_details, "cached_tokens", 0) or 0)
reasoning = int(getattr(output_details, "reasoning_tokens", 0) or 0)
if cached:
result["prompt_tokens_details"] = {"cached_tokens": cached}
cached = optional_int(getattr(input_details, "cached_tokens", None))
written = optional_int(
getattr(input_details, "cache_write_tokens", None)
)
reasoning = optional_int(
getattr(output_details, "reasoning_tokens", None)
)
details = self._prompt_cache_details(cached, written)
if details:
result["prompt_tokens_details"] = details
if reasoning:
result["completion_tokens_details"] = {"reasoning_tokens": reasoning}
self._last_usage = result
+81 -25
View File
@@ -3,10 +3,17 @@ from typing import List, Tuple
import logging
from application.parser.chunking_creator import ChunkerCreator
from application.parser.schema.base import Document
from application.utils import get_encoding
from application.parser.tokenization import get_token_counter
logger = logging.getLogger(__name__)
# Smallest share of ``max_tokens`` a chunk must keep for body text when the
# header is duplicated onto every chunk. A header that leaves less than this
# makes each chunk mostly repeated header and multiplies the chunk count -- at
# a budget of 32 out of 1250 a document splits into 39x more chunks than it
# needs -- so duplication is dropped rather than honoured.
_MIN_BODY_BUDGET_RATIO = 0.25
class Chunker:
"""Classic token-window chunker (registered as ``classic_chunk``).
@@ -24,10 +31,15 @@ class Chunker:
duplicate_headers: bool = False,
):
self.chunking_strategy = chunking_strategy
self.max_tokens = max_tokens
# A budget below 1 would ask for a chunk per token; the strategy
# chunkers clamp the same way.
self.max_tokens = max(1, int(max_tokens))
self.min_tokens = min_tokens
self.duplicate_headers = duplicate_headers
self.encoding = get_encoding()
# Counted in the embedding model's tokenizer, not cl100k: ``max_tokens``
# is compared against a limit the embedding server enforces in its own
# units, so counting in any other unit is a guess.
self.counter = get_token_counter()
def separate_header_and_body(self, text: str) -> Tuple[str, str]:
header_pattern = r"^(.*?\n){3}"
@@ -42,28 +54,73 @@ class Chunker:
def split_document(self, doc: Document) -> List[Document]:
split_docs = []
header, body = self.separate_header_and_body(doc.text)
header_tokens = self.encoding.encode_ordinary(header) if header else []
body_tokens = self.encoding.encode_ordinary(body)
"""Split one oversized document into ``max_tokens``-sized chunks.
current_position = 0
part_index = 0
while current_position < len(body_tokens):
end_position = current_position + self.max_tokens - len(header_tokens)
chunk_tokens = (header_tokens + body_tokens[current_position:end_position]
if self.duplicate_headers or part_index == 0 else body_tokens[current_position:end_position])
chunk_text = self.encoding.decode(chunk_tokens)
new_doc = Document(
text=chunk_text,
doc_id=f"{doc.doc_id}-{part_index}",
embedding=doc.embedding,
extra_info={**(doc.extra_info or {}), "token_count": len(chunk_tokens)}
Pieces are sliced out of the original text rather than decoded back
from token ids. WordPiece tokenizers normalise as they decode --
all-mpnet-base-v2 lowercases -- so a decode round-trip would rewrite
every stored document.
"""
header, body = self.separate_header_and_body(doc.text)
header_tokens = self.counter.count(header) if header else 0
if header and header_tokens >= self.max_tokens:
# The header alone fills the budget, so no cut of the body can keep
# a chunk within it and duplicating it would leave a one-token body
# budget -- a chunk per body token. It is only the first three
# lines, not something worth preserving at that cost, so it goes
# back to being ordinary text and the document splits evenly.
logger.warning(
"Header of %s is %d token(s), at or over the %d-token chunk "
"budget; treating it as body text.",
doc.doc_id,
header_tokens,
self.max_tokens,
)
body = f"{header}{body}"
header, header_tokens = "", 0
# A chunk carrying the header has that much less room for body text.
with_header_budget = max(1, self.max_tokens - header_tokens)
duplicate_headers = self.duplicate_headers
if duplicate_headers and with_header_budget < self.max_tokens * _MIN_BODY_BUDGET_RATIO:
logger.warning(
"Header of %s leaves only %d of %d tokens for body text; "
"carrying it on the first chunk only.",
doc.doc_id,
with_header_budget,
self.max_tokens,
)
duplicate_headers = False
if duplicate_headers:
body_pieces = self.counter.split(body, with_header_budget)
else:
body_pieces = self.counter.split(
body, self.max_tokens, first_max_tokens=with_header_budget
)
if not body_pieces and header:
# Nothing but a header: the loop below only ever emits the header
# attached to a body piece, so without this the document is dropped
# from the index entirely.
body_pieces = [""]
split_docs = []
for part_index, piece in enumerate(body_pieces):
include_header = bool(header) and (duplicate_headers or part_index == 0)
chunk_text = f"{header}{piece}" if include_header else piece
split_docs.append(
Document(
text=chunk_text,
doc_id=f"{doc.doc_id}-{part_index}",
embedding=doc.embedding,
extra_info={
**(doc.extra_info or {}),
"token_count": self.counter.count(chunk_text),
},
)
)
split_docs.append(new_doc)
current_position = end_position
part_index += 1
header_tokens = []
return split_docs
def classic_chunk(self, documents: List[Document]) -> List[Document]:
@@ -71,8 +128,7 @@ class Chunker:
i = 0
while i < len(documents):
doc = documents[i]
tokens = self.encoding.encode_ordinary(doc.text)
token_count = len(tokens)
token_count = self.counter.count(doc.text)
if self.min_tokens <= token_count <= self.max_tokens:
doc.extra_info = doc.extra_info or {}
+9 -17
View File
@@ -16,7 +16,7 @@ from typing import List
from application.parser.chunking import Chunker
from application.parser.chunking_creator import ChunkerCreator
from application.parser.schema.base import Document
from application.utils import get_encoding
from application.parser.tokenization import get_token_counter
logger = logging.getLogger(__name__)
@@ -41,19 +41,16 @@ class _BaseStrategyChunker:
self.max_tokens = max(1, int(max_tokens))
self.min_tokens = max(0, int(min_tokens))
self.duplicate_headers = duplicate_headers
self.encoding = get_encoding()
# Same unit as the embedding server counts in; see
# ``application.parser.tokenization``.
self.counter = get_token_counter()
def _token_count(self, text: str) -> int:
return len(self.encoding.encode_ordinary(text))
return self.counter.count(text)
def _split_by_tokens(self, text: str) -> List[str]:
"""Split ``text`` into pieces no larger than ``max_tokens`` tokens."""
tokens = self.encoding.encode_ordinary(text)
pieces = []
for start in range(0, len(tokens), self.max_tokens):
chunk_tokens = tokens[start:start + self.max_tokens]
pieces.append(self.encoding.decode(chunk_tokens))
return pieces
return self.counter.split(text, self.max_tokens)
def _emit(self, base: Document, part_index: int, text: str) -> Document:
"""Build a child Document carrying token_count and inherited info."""
@@ -188,16 +185,11 @@ class ParentChildChunker(_BaseStrategyChunker):
processed: List[Document] = []
child_size = self._child_size()
for doc in documents:
tokens = self.encoding.encode_ordinary(doc.text)
part_index = 0
for p_start in range(0, len(tokens), self.max_tokens):
parent_tokens = tokens[p_start:p_start + self.max_tokens]
parent_text = self.encoding.decode(parent_tokens)
for parent_text in self.counter.split(doc.text, self.max_tokens):
if not parent_text.strip():
continue
for c_start in range(0, len(parent_tokens), child_size):
child_tokens = parent_tokens[c_start:c_start + child_size]
child_text = self.encoding.decode(child_tokens)
for child_text in self.counter.split(parent_text, child_size):
if not child_text.strip():
continue
child = Document(
@@ -208,7 +200,7 @@ class ParentChildChunker(_BaseStrategyChunker):
embedding=doc.embedding,
extra_info={
**(doc.extra_info or {}),
"token_count": len(child_tokens),
"token_count": self.counter.count(child_text),
"parent_text": parent_text,
},
)
+68
View File
@@ -1,5 +1,7 @@
"""Shared file-extension constants for parsing and ingestion flows."""
import os
from application.parser.file.anydoc_parser import ANYDOC_GAINED_SUFFIXES
from application.stt.constants import SUPPORTED_AUDIO_EXTENSIONS
@@ -32,3 +34,69 @@ SUPPORTED_SOURCE_EXTENSIONS = (
*SUPPORTED_SOURCE_IMAGE_EXTENSIONS,
*SUPPORTED_AUDIO_EXTENSIONS,
)
# Suffixes the attachment path has a dedicated parser for — exactly the keys
# of ``get_default_file_extractor()``. Kept as a literal (importing ``bulk``
# here would drag docling into the API process), with
# ``tests/parser/file/test_constants.py`` asserting the two agree.
#
# This is *not* the whole attachment allow-list: a suffix with no parser is
# read by ``SimpleDirectoryReader``'s plain-text fallthrough, which is right
# for a .py or a .log and catastrophic for a video — the reader "extracts"
# megabytes of binary garbage, truncates it, and stores it with
# ``extraction.status == "ok"``. So unparsed suffixes are admitted on
# content instead (``upload_limits.enforce_parseable_attachment``): text
# passes, binary is refused. Zip is deliberately absent — source ingestion
# extracts archives, the attachment path does not, and a zip fails the
# content check like any other binary.
#
# Mirrored in ``frontend/src/constants/fileUpload.ts``; update both together.
ATTACHMENT_PARSER_EXTENSIONS = frozenset(
{
*SUPPORTED_SOURCE_EXTENSIONS,
".xhtml",
".adoc",
".asciidoc",
".tiff",
".tif",
".bmp",
".webp",
".vtt",
".xml",
}
# .txt has no parser of its own — it *is* the plain-text fallthrough. It
# must be sniffed like any other unparsed suffix, or renaming a video to
# notes.txt walks straight back into the bug this gate exists for.
- {".txt"}
)
def attachment_extension(filename: str | None) -> str:
"""Return the lower-cased extension of ``filename`` including the dot, or ``""``.
Args:
filename: A bare filename or path; ``None`` and empty strings yield ``""``.
Returns:
The last suffix in lower case (``".pdf"``), or ``""`` when there is none.
"""
if not filename:
return ""
return os.path.splitext(os.path.basename(str(filename)))[1].lower()
def has_attachment_parser(filename: str | None) -> bool:
"""Return whether an attachment's suffix has a dedicated parser.
A False result does not mean the file is refused: it means nothing but the
plain-text fallthrough will read it, so it has to earn its place on
content. The decision is by extension only, never by the mime type a
browser reports (mobile pickers ignore ``accept`` and lie).
Args:
filename: The upload's filename.
Returns:
True when the suffix is in ``ATTACHMENT_PARSER_EXTENSIONS``.
"""
return attachment_extension(filename) in ATTACHMENT_PARSER_EXTENSIONS
+275
View File
@@ -0,0 +1,275 @@
"""Counting and splitting text in the embedding model's own tokenizer.
Chunk sizes only mean something in the units the embedding server counts. The
chunker used to count cl100k (tiktoken) while the server counted whatever its
model used, so ``max_tokens`` was a value in one unit compared against a limit
in another. Every recalibration of that number was really an attempt to guess
the conversion factor.
This module removes the conversion. :func:`get_token_counter` returns a counter
backed by the configured embedding model's tokenizer, falling back to cl100k
when that tokenizer cannot be loaded -- an offline install, or a model the
registry does not describe.
Splitting never round-trips through ``decode``. Byte-level BPE decodes
losslessly, but WordPiece tokenizers do not: all-mpnet-base-v2 lowercases, so
decoding ``"Hello World"`` yields ``"hello world"`` and would silently rewrite
every stored document. Counters therefore cut the *original* string at
character offsets the tokenizer reports.
"""
from __future__ import annotations
import logging
import threading
from typing import Iterator, List, Optional, Tuple
from application.core.settings import settings
from application.utils import get_encoding
from application.vectorstore.model_registry import resolve
logger = logging.getLogger(__name__)
_cache: dict = {}
_cache_lock = threading.Lock()
def _windows(total: int, first: int, rest: int) -> Iterator[Tuple[int, int]]:
"""Yield ``(start, end)`` token windows, the first sized independently.
A header consumes part of the first chunk's budget but none of the rest,
so the first window is often smaller than those that follow.
"""
start = 0
budget = max(1, first)
while start < total:
end = min(start + budget, total)
yield start, end
start = end
budget = max(1, rest)
# A token spanning more characters than this collapsed a run the tokenizer
# could not break up: WordPiece emits a single ``[UNK]`` for any word longer
# than ``max_input_chars_per_word``. Charging that once makes a base64 blob or
# a minified bundle look tiny, so nothing splits it and an oversized chunk
# reaches the embedding server. Real tokens are a few characters, so prose
# never reaches this bound.
_MAX_CHARS_PER_TOKEN = 16
def _token_weight(start: int, end: int) -> int:
"""Tokens a span costs, charging a collapsed run by its length."""
span = end - start
return max(1, -(-span // _MAX_CHARS_PER_TOKEN))
def _cap_piece_chars(pieces: List[str], first: int, rest: int) -> List[str]:
"""Cut any piece holding more characters than its budget can cover."""
capped: List[str] = []
budget = first
for piece in pieces:
limit = max(1, budget * _MAX_CHARS_PER_TOKEN)
while len(piece) > limit:
capped.append(piece[:limit])
piece = piece[limit:]
budget = rest
limit = max(1, budget * _MAX_CHARS_PER_TOKEN)
capped.append(piece)
budget = rest
return capped
class TokenCounter:
"""Counts and splits text in one tokenizer's units.
Attributes:
name: Human-readable identifier of the underlying tokenizer, for logs.
"""
name = "unknown"
def count(self, text: str) -> int:
"""Number of tokens ``text`` occupies."""
raise NotImplementedError
def split(
self, text: str, max_tokens: int, first_max_tokens: Optional[int] = None
) -> List[str]:
"""Cut ``text`` into consecutive pieces of at most ``max_tokens``.
The concatenation of the returned pieces equals ``text`` exactly.
Args:
text: Source text.
max_tokens: Token budget per piece; values below 1 are treated as 1.
first_max_tokens: Budget for the first piece only, when it must
leave room for a header. Defaults to ``max_tokens``.
Returns:
The pieces, in order. An empty ``text`` yields an empty list.
"""
raise NotImplementedError
class TiktokenCounter(TokenCounter):
"""cl100k counter -- the historical behaviour, and the fallback."""
name = "cl100k_base"
def __init__(self) -> None:
self._encoding = get_encoding()
def count(self, text: str) -> int:
if not text:
return 0
return len(self._encoding.encode_ordinary(text))
def split(
self, text: str, max_tokens: int, first_max_tokens: Optional[int] = None
) -> List[str]:
if not text:
return []
rest = max(1, max_tokens)
first = max(1, first_max_tokens if first_max_tokens is not None else rest)
tokens = self._encoding.encode_ordinary(text)
if len(tokens) <= first:
return [text]
# Cut the original string at the character offsets the tokenizer
# reports, the way :class:`HuggingFaceCounter` does. Decoding each
# window on its own instead splits any multi-byte character that
# straddles a boundary across two byte sequences, and each half decodes
# to U+FFFD -- roughly one boundary in five on CJK text, silently
# destroying a character per cut.
offsets = self._encoding.decode_with_offsets(tokens)[1]
end_of_text = len(text)
pieces: List[str] = []
cursor = 0
for _, end in _windows(len(tokens), first, rest):
# Offsets index into the decoded string. ``encode_ordinary``
# round-trips for anything cl100k can represent, so it is ``text``;
# clamping keeps the cuts in range if it ever is not, and slicing
# ``text`` monotonically keeps the pieces reassembling exactly
# either way.
end_char = end_of_text if end >= len(offsets) else min(offsets[end], end_of_text)
if end_char <= cursor:
continue
pieces.append(text[cursor:end_char])
cursor = end_char
if cursor < end_of_text:
if pieces:
pieces[-1] = pieces[-1] + text[cursor:]
else:
pieces.append(text[cursor:])
return pieces
class HuggingFaceCounter(TokenCounter):
"""Counts in a Hugging Face tokenizer, slicing by character offsets."""
def __init__(self, tokenizer, name: str) -> None:
self._tokenizer = tokenizer
self.name = name
def _encode(self, text: str):
return self._tokenizer.encode(text, add_special_tokens=False)
def count(self, text: str) -> int:
if not text:
return 0
encoding = self._encode(text)
if not encoding.offsets:
return len(encoding.ids)
return sum(_token_weight(start, end) for start, end in encoding.offsets)
def split(
self, text: str, max_tokens: int, first_max_tokens: Optional[int] = None
) -> List[str]:
if not text:
return []
rest = max(1, max_tokens)
first = max(1, first_max_tokens if first_max_tokens is not None else rest)
offsets = self._encode(text).offsets
if self.count(text) <= first:
return [text]
pieces: List[str] = []
cursor = 0
for start_token, end_token in _windows(len(offsets), first, rest):
window = offsets[start_token:end_token]
# Some tokenizers emit (0, 0) for specials or normalised-away
# characters; those carry no span to cut on.
spans = [end for _, end in window if end > cursor]
if not spans:
continue
end_char = max(spans)
pieces.append(text[cursor:end_char])
cursor = end_char
if cursor < len(text):
# Trailing characters the tokenizer dropped (e.g. whitespace) belong
# to the final piece, so no input is lost.
if pieces:
pieces[-1] = pieces[-1] + text[cursor:]
else:
pieces.append(text[cursor:])
return _cap_piece_chars(pieces, first, rest)
def _load_hf_counter(repo: str) -> Optional[HuggingFaceCounter]:
"""Load ``repo``'s tokenizer, or ``None`` if it is not reachable."""
try:
from tokenizers import Tokenizer
tokenizer = Tokenizer.from_pretrained(repo)
# Repos ship padding and truncation defaults meant for inference
# batches. Left on, every count returns the padded width (128 for
# mpnet) and the offsets carry (0, 0) entries for the padding, so both
# counting and slicing are wrong.
tokenizer.no_padding()
tokenizer.no_truncation()
return HuggingFaceCounter(tokenizer, repo)
except Exception as exc: # noqa: BLE001 -- chunking must never hard-fail here
logger.warning(
"Could not load the tokenizer for %s (%s); counting chunk sizes in "
"cl100k instead. Chunk sizes will be approximate for this model.",
repo,
exc,
)
return None
def get_token_counter(embeddings_name: Optional[str] = None) -> TokenCounter:
"""Return the counter for ``embeddings_name``, cached per process.
Args:
embeddings_name: Model name; defaults to ``settings.EMBEDDINGS_NAME``.
Returns:
A :class:`HuggingFaceCounter` for a model whose tokenizer could be
loaded, else a :class:`TiktokenCounter`.
"""
name = embeddings_name or getattr(settings, "EMBEDDINGS_NAME", None)
key = name or "__default__"
with _cache_lock:
if key in _cache:
return _cache[key]
spec = resolve(name)
repo = spec.repo if spec else name
counter: TokenCounter
if repo and (spec is None or spec.provider == "fastembed"):
counter = _load_hf_counter(repo) or TiktokenCounter()
else:
counter = TiktokenCounter()
with _cache_lock:
_cache[key] = counter
logger.info("Chunking will count tokens with %s", counter.name)
return counter
def reset_cache() -> None:
"""Drop cached counters. For tests and for a settings change at runtime."""
with _cache_lock:
_cache.clear()
+2 -3
View File
@@ -7,10 +7,9 @@
# not either: OCR_ENABLED=true with the tesseract binary (or a DeepSeek-OCR
# endpoint) runs through application/parser/file/ocr_parser.py.
#
# Install ON TOP of requirements.txt (torch/transformers stay pinned there,
# shared with sentence-transformers):
# Install ON TOP of requirements.txt, which pins torch, transformers and
# onnxruntime (rapidocr runs on the same onnxruntime fastembed uses):
# pip install -r requirements.txt -r requirements-docling.txt
# Docker: build with --build-arg INSTALL_DOCLING=true
docling==2.119.0
rapidocr==3.9.2
onnxruntime==1.28.0
+10 -7
View File
@@ -21,6 +21,10 @@ ddgs>=8.0.0
fast-ebook
elevenlabs==2.62.0
Flask==3.1.3
fastembed==0.8.0
# fastembed's runtime: local embeddings execute on it, so it stays pinned in
# core even though the docling extra's rapidocr runs on the same package.
onnxruntime==1.28.0
faiss-cpu==1.15.0
fastmcp==3.4.6
flask-restx==1.3.2
@@ -91,19 +95,18 @@ referencing>=0.28.0,<0.38.0
regex==2026.7.19
requests==2.34.2
retry==0.9.2
sentence-transformers==5.7.0
sqlalchemy>=2.0,<3
starlette>=1.0,<2
tiktoken==0.13.0
tokenizers==0.22.2
# torch stays in core: sentence-transformers (local embeddings) needs it even
# with the docling extra uninstalled.
# torch and transformers serve docling's model stack alone -- embeddings moved
# to fastembed/onnxruntime. Still in core rather than requirements-docling.txt,
# so installing the optional extra never re-resolves the shared pins; both can
# move out with docling.
torch==2.11.0
tqdm==4.67.3
# Kept in core for sentence-transformers, so installing the optional docling
# extra never changes the shared model stack. Capped <5.9.0 to match
# docling-core's own darwin pin: 5.9+ breaks docling's PDF layout model on
# Apple Silicon ("Cannot convert a MPS Tensor to float64").
# Capped <5.9.0 to match docling-core's own darwin pin: 5.9+ breaks docling's
# PDF layout model on Apple Silicon ("Cannot convert a MPS Tensor to float64").
transformers==5.8.1
typing-extensions==4.16.0
typing-inspect==0.9.0
View File
Whitespace-only changes.
+89
View File
@@ -0,0 +1,89 @@
"""Download embedding model artifacts into FastEmbed's cache.
Run at image build time so a fresh container does not download a model on its
first ingest, and an air-gapped install works at all. Both the legacy and the
current default are baked: an upgraded deployment keeps using mpnet until it
runs ``reembed``, while a new one starts on granite.
Usage::
python -m application.scripts.prefetch_models # the defaults
python -m application.scripts.prefetch_models granite-311m # a subset
"""
from __future__ import annotations
import logging
import sys
from typing import List, Optional, Sequence
from application.vectorstore.model_registry import (
DEFAULT_LEGACY,
DEFAULT_NEW_INSTALL,
known_names,
resolve,
)
logger = logging.getLogger("prefetch_models")
#: Fetched when no names are given.
DEFAULT_MODELS = (DEFAULT_LEGACY, DEFAULT_NEW_INSTALL)
def prefetch(names: Sequence[str], cache_dir: Optional[str] = None) -> List[str]:
"""Fetch each named model's artifacts.
Args:
names: Registry names or aliases.
cache_dir: FastEmbed cache directory; its default when omitted.
Returns:
The repositories actually fetched.
Raises:
SystemExit: If a name is not in the registry, since a silent skip at
build time becomes a download at run time on an offline host.
"""
from fastembed import TextEmbedding
from fastembed.common.model_description import ModelSource, PoolingType
pooling_types = {"cls": PoolingType.CLS, "mean": PoolingType.MEAN}
fetched: List[str] = []
for name in names:
spec = resolve(name)
if spec is None:
raise SystemExit(
f"Unknown embedding model {name!r}. Known: {', '.join(known_names())}"
)
if spec.provider != "fastembed":
logger.info("Skipping %s: served remotely, nothing to cache.", spec.name)
continue
logger.info("Fetching %s", spec.repo)
TextEmbedding.add_custom_model(
model=spec.repo,
pooling=pooling_types[spec.pooling],
normalization=spec.normalize,
sources=ModelSource(hf=spec.repo),
dim=spec.dimension,
model_file=spec.onnx_file,
)
kwargs = {"model_name": spec.repo}
if cache_dir:
kwargs["cache_dir"] = cache_dir
TextEmbedding(**kwargs)
fetched.append(spec.repo)
return fetched
def main(argv: Optional[Sequence[str]] = None) -> int:
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
import os
names = list(argv) if argv else list(DEFAULT_MODELS)
fetched = prefetch(names, os.environ.get("EMBEDDINGS_CACHE_DIR"))
logger.info("Cached %d model(s): %s", len(fetched), ", ".join(fetched))
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))
+542
View File
@@ -0,0 +1,542 @@
"""Re-embed an existing index with the currently configured embedding model.
Changing ``EMBEDDINGS_NAME`` does not change vectors already stored. Because
several models share a width -- granite-311m and all-mpnet-base-v2 are both
768-dimensional -- a swapped model does not fail any dimension check. It simply
retrieves against vectors that mean something else, which shows up as bad
answers rather than as an error.
This script rebuilds the vectors from the chunk text already held in the store,
so it never re-downloads, re-parses or re-chunks anything: no source files, no
crawl, no docling. It works against pgvector and FAISS. With ``GRAPHRAG_ENABLED``
it also rewrites ``graph_nodes.name_embedding``, which seeds graph traversal and
would otherwise be left behind in the previous model's space.
Usage::
python -m application.scripts.reembed --dry-run # report, change nothing
python -m application.scripts.reembed # re-embed everything
python -m application.scripts.reembed --sources a,b # only these sources
"""
from __future__ import annotations
import argparse
import logging
import sys
import time
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
from psycopg import sql
from application.core.settings import settings
from application.vectorstore.model_registry import resolve
from application.vectorstore.vector_creator import VectorCreator
logger = logging.getLogger("reembed")
#: Stores this script knows how to rewrite in place.
SUPPORTED_STORES = ("pgvector", "faiss")
#: Chunks embedded (and, for pgvector, written) per transaction.
DEFAULT_BATCH_SIZE = 64
@dataclass
class _RebuildDoc:
"""Minimal document shape ``FaissStore(docs_init=...)`` accepts."""
page_content: str
metadata: Dict[str, Any]
class ReembedError(RuntimeError):
"""Raised for a misconfiguration the user needs to fix before running."""
def _log_setup(verbose: bool) -> None:
logging.basicConfig(
level=logging.DEBUG if verbose else logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
def _batched(items: Sequence[Any], size: int) -> Iterable[Sequence[Any]]:
for start in range(0, len(items), size):
yield items[start : start + size]
def list_source_ids(store_type: str) -> List[str]:
"""Every source id that has vectors in the configured store.
Args:
store_type: ``"pgvector"`` or ``"faiss"``.
Returns:
Source ids, sorted for a stable and resumable ordering.
"""
if store_type == "pgvector":
return _pgvector_source_ids()
return _faiss_source_ids()
def _pgvector_source_ids() -> List[str]:
store = VectorCreator.create_vectorstore("pgvector", source_id="")
# ``_get_connection`` hands back a pooled connection, not a context
# manager: closing it via ``with`` would cost the pool a slot.
conn = store._get_connection()
cursor = conn.cursor()
try:
# Table and column names cannot be bound as parameters, so they go
# through ``sql.Identifier``, which quotes them. They are internal
# constants rather than user input, but composing SQL by f-string is
# the habit worth not having.
cursor.execute(
sql.SQL(
"SELECT DISTINCT source_id FROM {table} "
"WHERE source_id IS NOT NULL ORDER BY source_id"
).format(table=sql.Identifier(store._table_name))
)
return [row[0] for row in cursor.fetchall()]
finally:
cursor.close()
store.close()
def _faiss_source_ids() -> List[str]:
"""Directories under the FAISS root that hold an index."""
from application.storage.storage_creator import StorageCreator
storage = StorageCreator.get_storage()
root = "indexes"
ids: List[str] = []
try:
for path in storage.list_files(root):
# ``indexes/<source_id>/index.faiss``
parts = str(path).split("/")
if parts and parts[-1].endswith(".faiss") and len(parts) >= 2:
ids.append(parts[-2])
except Exception as exc: # noqa: BLE001 -- surfaced to the user below
raise ReembedError(
f"Could not list FAISS indexes under {root!r}: {exc}"
) from exc
return sorted(set(ids))
def _graph_nodes_exist(conn) -> bool:
"""True when the GraphRAG node table is present in this database."""
cursor = conn.cursor()
try:
cursor.execute("SELECT to_regclass('public.graph_nodes')")
row = cursor.fetchone()
return bool(row and row[0])
finally:
cursor.close()
def reembed_graph_nodes(store, conn, source_id: str, batch_size: int, dry_run: bool) -> int:
"""Rewrite one source's GraphRAG node name vectors.
``graph_nodes.name_embedding`` is what every graph traversal starts from:
the retriever embeds the query and takes the nearest node names as seeds.
It is written once at extraction time and never revisited, so re-embedding
only the chunk table leaves the graph seeding against the previous model.
Nothing errors -- mpnet and granite-311m are both 768-dimensional, so the
column accepts the mismatch -- the graph just walks from the wrong nodes.
The embedded text is ``graph_nodes.name``, which is already stored, so this
needs no LLM re-extraction.
Args:
store: The pgvector store, for its embeddings client.
conn: Open connection to the same database.
source_id: Source whose nodes to re-embed.
batch_size: Names per embed call and per transaction.
dry_run: When true, count the work and change nothing.
Returns:
Number of node rows seen (dry run) or rewritten.
"""
if not _graph_nodes_exist(conn):
return 0
cursor = conn.cursor()
try:
cursor.execute(
"SELECT id, name FROM graph_nodes "
"WHERE source_id = %s AND name_embedding IS NOT NULL ORDER BY id",
(source_id,),
)
rows = cursor.fetchall()
finally:
cursor.close()
if dry_run or not rows:
return len(rows)
written = 0
for batch in _batched(rows, batch_size):
vectors = store._embedding.embed_documents([row[1] or "" for row in batch])
cursor = conn.cursor()
try:
cursor.executemany(
"UPDATE graph_nodes SET name_embedding = %s::vector WHERE id = %s",
[
(str(list(vector)), row[0])
for vector, row in zip(vectors, batch)
],
)
conn.commit()
except Exception:
conn.rollback()
raise
finally:
cursor.close()
written += len(batch)
return written
def _count_source_chunks(conn, table: str, source_id: str) -> int:
"""How many chunk rows one source holds."""
cursor = conn.cursor()
try:
cursor.execute(
sql.SQL("SELECT count(*) FROM {table} WHERE source_id = %s").format(
table=sql.Identifier(table)
),
(source_id,),
)
return int(cursor.fetchone()[0])
finally:
cursor.close()
def _fetch_chunk_page(
conn, table: str, source_id: str, after_id: Optional[Any], limit: int
) -> List[Tuple[Any, str]]:
"""One page of ``(id, text)`` ordered by id, starting after ``after_id``.
Keyed off ``id`` rather than ``OFFSET`` so each page costs an index seek,
and the page boundaries stay stable across the updates the caller makes
between pages -- only the vector column changes, never ``id``.
"""
cursor = conn.cursor()
try:
if after_id is None:
cursor.execute(
sql.SQL(
"SELECT id, text FROM {table} WHERE source_id = %s "
"ORDER BY id LIMIT %s"
).format(table=sql.Identifier(table)),
(source_id, limit),
)
else:
cursor.execute(
sql.SQL(
"SELECT id, text FROM {table} WHERE source_id = %s AND id > %s "
"ORDER BY id LIMIT %s"
).format(table=sql.Identifier(table)),
(source_id, after_id, limit),
)
return cursor.fetchall()
finally:
cursor.close()
def _write_chunk_page(conn, table: str, vector_column: str, store, page) -> None:
"""Embed one page and commit its vectors."""
vectors = store._embedding.embed_documents([row[1] or "" for row in page])
cursor = conn.cursor()
try:
cursor.executemany(
sql.SQL("UPDATE {table} SET {column} = %s::vector WHERE id = %s").format(
table=sql.Identifier(table),
column=sql.Identifier(vector_column),
),
[(str(list(vector)), row[0]) for vector, row in zip(vectors, page)],
)
conn.commit()
except Exception:
conn.rollback()
raise
finally:
cursor.close()
def reembed_pgvector(source_id: str, batch_size: int, dry_run: bool) -> Tuple[int, int]:
"""Rewrite one source's vectors in place.
Rows are updated, never deleted and re-inserted, so an interrupted run
leaves every chunk present and simply re-does its batch next time.
Chunks are read a page at a time rather than all at once. A source's text is
the whole corpus: at the default chunk size 200k chunks materialise to about
1.6 GB of Python strings, several times that for non-Latin scripts, and the
result set is held alongside them until the cursor closes. That is enough to
be OOM-killed inside a 4 GiB container while also holding the model -- and a
kill here is what a torn index costs the most.
Args:
source_id: Source whose chunks to re-embed.
batch_size: Chunks per page, per embed call and per transaction.
dry_run: When true, count the work and change nothing.
Returns:
``(chunks_seen, chunks_written)``.
"""
store = VectorCreator.create_vectorstore("pgvector", source_id=source_id)
table, vector_column = store._table_name, store._vector_column
conn = store._get_connection()
seen = written = 0
try:
if dry_run:
seen = _count_source_chunks(conn, table, source_id)
else:
after_id: Optional[Any] = None
while True:
page = _fetch_chunk_page(
conn, table, source_id, after_id, batch_size
)
if not page:
break
_write_chunk_page(conn, table, vector_column, store, page)
after_id = page[-1][0]
seen += len(page)
written += len(page)
# The graph seeds every traversal from its own vectors, so leaving them
# in the old model's space is the same silent mismatch this script
# exists to remove -- and at equal widths nothing would report it.
if getattr(settings, "GRAPHRAG_ENABLED", False):
nodes = reembed_graph_nodes(store, conn, source_id, batch_size, dry_run)
if nodes:
logger.info(
" %s: %d graph node name(s)%s",
source_id,
nodes,
"" if dry_run else " re-embedded",
)
finally:
store.close()
return seen, written
def reembed_faiss(source_id: str, batch_size: int, dry_run: bool) -> Tuple[int, int]:
"""Rebuild one FAISS index from the chunk text in its sidecar.
A FAISS index is a flat array, so this rebuilds rather than updates. Every
new vector is computed *before* the old index is touched, so an interrupt
during embedding leaves the existing index intact -- and ``save_local``
writes each file to a temporary path and moves it into place, so an
interrupt during the write does too.
Args:
source_id: Source whose index to rebuild.
batch_size: Chunks per embed call.
dry_run: When true, count the work and change nothing.
Returns:
``(chunks_seen, chunks_written)``.
"""
# A width change is the main reason to run this, and it is exactly what
# ``assert_embedding_dimensions`` refuses to open. The chunk text lives in
# the sidecar rather than the index, so reading it needs no matching width,
# and the index this reads is replaced below.
store = VectorCreator.create_vectorstore(
"faiss",
source_id=source_id,
embeddings_key=settings.EMBEDDINGS_KEY,
skip_dimension_check=True,
)
chunks: List[Dict[str, Any]] = store.get_chunks() or []
if dry_run or not chunks:
return len(chunks), 0
docs = [
_RebuildDoc(chunk.get("text") or "", chunk.get("metadata") or {})
for chunk in chunks
]
# Constructing with ``docs_init`` embeds everything into a fresh in-memory
# index and touches nothing on disk. The existing index is only replaced by
# ``save_local`` below, so a failure while embedding leaves it intact.
# Keep the existing ids: GraphRAG's ``graph_node_chunks`` rows and any id a
# client already holds point at them.
ids = [chunk.get("doc_id") for chunk in chunks]
rebuilt = VectorCreator.create_vectorstore(
"faiss",
source_id=source_id,
embeddings_key=settings.EMBEDDINGS_KEY,
docs_init=docs,
ids=ids if all(ids) else None,
batch_size=batch_size,
)
rebuilt.save_local()
return len(chunks), len(docs)
def record_source_model(source_id: str) -> None:
"""Stamp ``sources.model`` with the model these vectors were just built by.
``sources`` lives in the user-data database while the vectors may not, so
this takes its own session. Left unwritten, the column keeps naming the old
model and every consumer that trusts it -- the boot mismatch check above
all -- reports a source as stale immediately after it was migrated.
"""
from sqlalchemy import text
from application.storage.db.session import db_session
try:
with db_session() as conn:
conn.execute(
text("UPDATE sources SET model = :model WHERE id = :id"),
{"model": settings.EMBEDDINGS_NAME, "id": source_id},
)
except Exception as exc: # noqa: BLE001 — the vectors are already rewritten
logger.warning(
" %s: re-embedded, but could not update sources.model (%s). Retrieval "
"is correct; the mismatch warning may persist until it is.",
source_id,
exc,
)
def run(
store_type: str,
source_ids: Optional[Sequence[str]],
batch_size: int,
dry_run: bool,
) -> int:
"""Re-embed the requested sources. Returns a process exit code."""
model = settings.EMBEDDINGS_NAME
spec = resolve(model)
logger.info(
"Re-embedding %s with %s (%s)",
store_type,
model,
f"{spec.dimension} dims" if spec else "width determined at runtime",
)
if dry_run:
logger.info("Dry run: nothing will be written.")
ids = list(source_ids) if source_ids else list_source_ids(store_type)
if not ids:
logger.warning("No sources found; nothing to do.")
return 0
logger.info("%d source(s) to process", len(ids))
handler = reembed_pgvector if store_type == "pgvector" else reembed_faiss
total_seen = total_written = 0
failures: List[str] = []
started = time.monotonic()
for position, source_id in enumerate(ids, start=1):
try:
seen, written = handler(source_id, batch_size, dry_run)
if written and not dry_run:
record_source_model(source_id)
total_seen += seen
total_written += written
logger.info(
"[%d/%d] %s: %d chunk(s)%s",
position,
len(ids),
source_id,
seen,
"" if dry_run else f", {written} re-embedded",
)
except Exception as exc: # noqa: BLE001 -- one bad source must not stop the run
failures.append(source_id)
logger.error("[%d/%d] %s FAILED: %s", position, len(ids), source_id, exc)
elapsed = time.monotonic() - started
logger.info(
"Done in %.1fs: %d chunk(s) seen, %d re-embedded, %d source(s) failed",
elapsed,
total_seen,
total_written,
len(failures),
)
if failures:
logger.error("Failed sources: %s", ", ".join(failures))
logger.error("Re-run with --sources %s to retry just those.", ",".join(failures))
return 1
if dry_run:
logger.info("Re-run without --dry-run to apply.")
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="python -m application.scripts.reembed",
description=__doc__.split("\n\n")[0],
)
parser.add_argument(
"--dry-run", action="store_true", help="report what would change, write nothing"
)
parser.add_argument(
"--sources", help="comma-separated source ids; default is every source"
)
parser.add_argument(
"--batch-size",
type=int,
default=DEFAULT_BATCH_SIZE,
help=f"chunks per embed call (default {DEFAULT_BATCH_SIZE})",
)
parser.add_argument("-v", "--verbose", action="store_true", help="debug logging")
return parser
def main(argv: Optional[Sequence[str]] = None) -> int:
args = build_parser().parse_args(argv)
_log_setup(args.verbose)
# Which model this installation uses may live in ``app_metadata`` rather
# than the environment -- ``application.app`` resolves it at boot, and this
# script never imports that. Without this, an install pinned to granite
# with no EMBEDDINGS_NAME set (every stock Kubernetes deployment: the
# manifests carry no embedding config at all) would re-embed its whole
# index with the *legacy* code default and stamp ``sources.model`` to
# match -- the silent cross-model index this script exists to repair.
from application.storage.db.embeddings_pin import resolve_embeddings_pin
resolve_embeddings_pin(logger)
# Embed in this process. ``EMBEDDINGS_DELEGATE_TO_WORKER`` exists to keep a
# model out of the API, which serves one query at a time and holds the
# model for nothing in between. This is the opposite case: a batch job that
# embeds every chunk in the index, where a broker round trip per batch adds
# latency and a dependency on a worker running. Loading the model here also
# means the script reports a real failure for a model it cannot load,
# instead of timing out against an empty queue.
if getattr(settings, "EMBEDDINGS_DELEGATE_TO_WORKER", False):
logger.info("Embedding in-process; worker delegation does not apply here.")
settings.EMBEDDINGS_DELEGATE_TO_WORKER = False
store_type = (settings.VECTOR_STORE or "").lower()
if store_type not in SUPPORTED_STORES:
logger.error(
"VECTOR_STORE is %r; this script supports %s. For other stores, "
"delete and re-ingest the affected sources.",
settings.VECTOR_STORE,
" and ".join(SUPPORTED_STORES),
)
return 2
sources = (
[s.strip() for s in args.sources.split(",") if s.strip()]
if args.sources
else None
)
try:
return run(store_type, sources, max(1, args.batch_size), args.dry_run)
except ReembedError as exc:
logger.error("%s", exc)
return 2
if __name__ == "__main__":
sys.exit(main())
+83 -20
View File
@@ -76,6 +76,43 @@ def ensure_database_ready(
_run_migrations(log)
def _release_boot_only_embeddings(log: logging.Logger) -> None:
"""Drop a model this hook loaded that the process will never use again.
``ensure_vector_schema`` has to run the model to learn the width of a model
the registry does not describe. ``EmbeddingsSingleton`` then caches it for
the life of the process -- correct when this process embeds, pure waste in
an API that delegates every embed to the worker, where it costs ~400 MB for
a small model and ~800 MB for a granite-sized one that is never called.
Only the local-ONNX case is dropped. A ``RemoteEmbeddings`` is cheap and is
exactly what the process goes on to use, and with delegation off the model
would only be rebuilt on the first query.
Bounds retention, not the transient peak: the load still happens, and the
ONNX Runtime arena may not return every page to the OS.
"""
from application.core.settings import settings
if settings.EMBEDDINGS_BASE_URL:
return
if getattr(settings, "EMBEDDINGS_DELEGATE_TO_WORKER", False) is not True:
return
import gc
from application.vectorstore.base import EmbeddingsSingleton
if EmbeddingsSingleton._instances.pop(settings.EMBEDDINGS_NAME, None) is None:
return
gc.collect()
log.info(
"ensure_vector_schema: released the embeddings model loaded to read its "
"width; this process delegates embedding to the worker and would never "
"have used it."
)
def ensure_vector_schema(*, logger: Optional[logging.Logger] = None) -> None:
"""Create the pgvector schema once at boot and verify its dimension.
@@ -127,27 +164,13 @@ def ensure_vector_schema(*, logger: Optional[logging.Logger] = None) -> None:
PGVectorStore,
)
# Loading the model here is deliberate: the process loads it on first
# retrieval anyway, and EmbeddingsSingleton caches it.
dim: Optional[int] = None
try:
from application.vectorstore.base import get_embeddings
# All this needs is an integer, and for a model the registry describes that
# is a lookup. It used to construct the embeddings instance, which loaded
# ~800 MB of ONNX into every API and worker process at import purely to
# read ``.dimension`` off it.
from application.vectorstore.model_registry import dimension_for
dim = getattr(get_embeddings(), "dimension", None)
except Exception as exc: # noqa: BLE001 — never block boot on the model
log.warning(
"ensure_vector_schema: could not load the embeddings model (%s); "
"creating the table with %d dimensions and skipping the dimension "
"check.",
exc,
DEFAULT_EMBEDDING_DIM,
)
if dim is None:
log.warning(
"ensure_vector_schema: the embeddings model exposes no dimension; "
"using %d and skipping the dimension check.",
DEFAULT_EMBEDDING_DIM,
)
dim: Optional[int] = dimension_for(settings.EMBEDDINGS_NAME)
graph_enabled = bool(getattr(settings, "GRAPHRAG_ENABLED", False))
started = time.monotonic()
@@ -166,6 +189,46 @@ def ensure_vector_schema(*, logger: Optional[logging.Logger] = None) -> None:
finally:
cursor.close()
if dim is None:
# An unregistered model only reports its width once something has
# run it. Build it in-process rather than through
# ``get_embeddings``: at boot there is no Celery task in flight, so
# a delegating client would dispatch to a worker that may not be up
# yet. The width must come from the model and not from the existing
# table -- reading the table would make the check below compare a
# value against itself, which is how a model of a different width
# silently inherits a table it does not fit.
try:
from application.vectorstore.base import build_local_embeddings
embedding = build_local_embeddings()
dim = getattr(embedding, "dimension", None)
if not dim:
# A remote client knows nothing about its server until it
# has called it, so ask once. Without this the table is
# sized at the default and the check below is skipped --
# which is how a remote model of any other width silently
# got a vector(768) column, the exact failure this hook
# exists to catch. Milvus and Qdrant probe the same way.
dim = len(embedding.embed_query("dimension probe"))
except Exception as exc: # noqa: BLE001 — never block boot on the model
log.warning(
"ensure_vector_schema: could not determine the embedding width "
"(%s); creating the table with %d dimensions and skipping the "
"dimension check.",
exc,
DEFAULT_EMBEDDING_DIM,
)
finally:
_release_boot_only_embeddings(log)
if dim is None:
log.warning(
"ensure_vector_schema: the embeddings model exposes no dimension; "
"using %d and skipping the dimension check.",
DEFAULT_EMBEDDING_DIM,
)
PGVectorStore.create_schema(conn, dimension=dim or DEFAULT_EMBEDDING_DIM)
if graph_enabled:
from application.graphrag.store import (
+150
View File
@@ -0,0 +1,150 @@
"""The embedding model an installation is pinned to.
``EMBEDDINGS_NAME`` has a code-level default, and moving that default would
re-point an existing index at a different vector space without anything
noticing: mpnet and granite are both 768-dimensional, so no width check fires
and retrieval simply gets worse. Which model an index was built with is a
property of the installation, not of the release it happens to be running.
So it is resolved once at boot and stored in ``app_metadata``. A fresh install
is pinned to the current recommendation; an install that already has sources is
pinned to the legacy model it has been using all along, and told how to move.
An explicit ``EMBEDDINGS_NAME`` in the environment always wins over both.
"""
from __future__ import annotations
import logging
from typing import Optional
from sqlalchemy import text
from application.core.settings import settings
from application.storage.db.repositories.app_metadata import AppMetadataRepository
from application.storage.db.session import db_session
from application.vectorstore.model_registry import (
DEFAULT_LEGACY,
DEFAULT_NEW_INSTALL,
resolve,
)
logger = logging.getLogger(__name__)
PIN_KEY = "embeddings_name"
NOTICE_KEY = "embeddings_legacy_notice_shown"
def _has_sources(conn) -> bool:
"""True when this installation already has an index to protect.
Counts ``sources`` rather than vector rows so the answer is the same for
every vector store, including the FAISS ones whose vectors are not in this
database at all.
"""
if conn.execute(text("SELECT to_regclass('public.sources')")).scalar() is None:
return False
return bool(conn.execute(text("SELECT EXISTS (SELECT 1 FROM sources)")).scalar())
def _legacy_notice(model: str) -> str:
return (
f"Embeddings: this installation is using {model}, the model its index was "
"built with, and will keep using it.\n"
"To move to granite (multilingual, a 32k-token context, same 768 dimensions):\n"
" 1. EMBEDDINGS_NAME=ibm-granite/granite-embedding-311m-multilingual-r2\n"
" 2. python -m application.scripts.reembed\n"
"Changing the model without step 2 leaves queries searching a different "
"vector space than the stored vectors, which fails silently."
)
def resolve_embeddings_pin(log: Optional[logging.Logger] = None) -> None:
"""Point ``settings.EMBEDDINGS_NAME`` at this installation's pinned model.
Called once at boot, before anything embeds and before the vector schema
hook reads the width. Consumers read ``settings.EMBEDDINGS_NAME`` lazily, so
assigning it here is enough; no caller needs to know the pin exists.
Does nothing when the environment pins the name, and degrades to the
code-level default when the database cannot be reached.
Args:
log: Logger for the first-run notice; module logger when omitted.
"""
out = log or logger
if "EMBEDDINGS_NAME" in settings.model_fields_set:
return
try:
with db_session() as conn:
repo = AppMetadataRepository(conn)
stored = repo.get(PIN_KEY)
if stored:
settings.EMBEDDINGS_NAME = stored
return
upgrading = _has_sources(conn)
pinned = repo.setdefault(
PIN_KEY, DEFAULT_LEGACY if upgrading else DEFAULT_NEW_INSTALL
)
settings.EMBEDDINGS_NAME = pinned
out.info("Embeddings: pinned this installation to %s.", pinned)
if upgrading and repo.get(NOTICE_KEY) is None:
print(_legacy_notice(pinned), flush=True)
repo.set(NOTICE_KEY, "1")
except Exception as exc: # noqa: BLE001 — never block boot on the database
out.debug(
"Embeddings: could not resolve the pinned model (%s); using %s.",
exc,
settings.EMBEDDINGS_NAME,
exc_info=True,
)
def warn_on_source_model_mismatch(log: Optional[logging.Logger] = None) -> None:
"""Report sources whose vectors were built by a different model.
The width check cannot see this: mpnet and granite are both 768, so an
index queried by the wrong model returns worse answers and no error. This
is the only signal, so it names the sources and the command that fixes
them.
A source with no recorded model pre-dates the column and is therefore the
legacy model, not unknown.
Args:
log: Logger to warn on; module logger when omitted.
"""
out = log or logger
active = resolve(settings.EMBEDDINGS_NAME)
try:
with db_session() as conn:
if conn.execute(text("SELECT to_regclass('public.sources')")).scalar() is None:
return
rows = conn.execute(
text("SELECT COALESCE(model, :legacy) AS model, count(*) FROM sources GROUP BY 1"),
{"legacy": DEFAULT_LEGACY},
).fetchall()
except Exception as exc: # noqa: BLE001 — never block boot on the database
out.debug("Embeddings: could not check source models (%s)", exc, exc_info=True)
return
stale = [
(name, count)
for name, count in rows
# Compare through the registry so an alias is not read as a different
# model. An unregistered name resolves to None, which only matches
# another unregistered name if the strings agree.
for a, b in [(resolve(name), active)]
if (a or name) != (b or settings.EMBEDDINGS_NAME)
]
if not stale:
return
detail = ", ".join(f"{count} built with {name}" for name, count in stale)
out.warning(
"Embeddings: %s, but queries are embedded with %s. Retrieval against those "
"sources is degraded and will not raise. Re-embed them with "
"`python -m application.scripts.reembed`, or set EMBEDDINGS_NAME back.",
detail,
settings.EMBEDDINGS_NAME,
)
+6
View File
@@ -235,6 +235,12 @@ token_usage_table = Table(
# Added in ``0015_token_usage_model_id``. Canonical model id (catalog
# name for built-ins, UUID for BYOM); NULL on un-backfilled rows.
Column("model_id", Text),
# Added in ``0031_token_usage_cache_tokens``. Prompt-cache breakdowns of
# ``prompt_tokens`` as reported by the provider (reads; and writes on
# newer OpenAI-family models). NULL = not reported, distinct from 0 = no
# cache activity, so hit-rate queries stay honest across providers.
Column("cached_tokens", Integer),
Column("cache_write_tokens", Integer),
)
user_logs_table = Table(
@@ -37,6 +37,21 @@ class AppMetadataRepository:
{"key": key, "value": value},
)
def setdefault(self, key: str, value: str) -> str:
"""Store ``value`` under ``key`` only if absent, returning what stands.
Two workers racing on first startup converge on one value instead of
each overwriting the other, the way ``get_or_create_instance_id`` does.
"""
self._conn.execute(
text(
"INSERT INTO app_metadata (key, value) VALUES (:key, :value) "
"ON CONFLICT (key) DO NOTHING"
),
{"key": key, "value": value},
)
return self.get(key) or value
def get_or_create_instance_id(self) -> str:
"""Return the anonymous instance UUID, generating one if absent.
@@ -35,6 +35,8 @@ class TokenUsageRepository:
request_id: Optional[str] = None,
model_id: Optional[str] = None,
timestamp: Optional[datetime] = None,
cached_tokens: Optional[int] = None,
cache_write_tokens: Optional[int] = None,
) -> None:
# Attribution guard: the ``token_usage_attribution_chk`` CHECK
# constraint requires at least one of ``user_id`` / ``api_key``
@@ -60,12 +62,14 @@ class TokenUsageRepository:
INSERT INTO token_usage (
user_id, api_key, agent_id,
prompt_tokens, generated_tokens,
cached_tokens, cache_write_tokens,
source, request_id, model_id, timestamp
)
VALUES (
:user_id, :api_key,
CAST(:agent_id AS uuid),
:prompt_tokens, :generated_tokens,
:cached_tokens, :cache_write_tokens,
:source, :request_id, :model_id, COALESCE(:timestamp, now())
)
"""
@@ -76,6 +80,8 @@ class TokenUsageRepository:
"agent_id": agent_id_uuid,
"prompt_tokens": prompt_tokens,
"generated_tokens": generated_tokens,
"cached_tokens": cached_tokens,
"cache_write_tokens": cache_write_tokens,
"source": source,
"request_id": request_id,
"model_id": model_id,
+45 -8
View File
@@ -1,6 +1,7 @@
"""Local file system implementation."""
import os
import shutil
import tempfile
from typing import BinaryIO, List, Callable
from application.storage.base import BaseStorage
@@ -36,16 +37,52 @@ class LocalStorage(BaseStorage):
return resolved
def save_file(self, file_data: BinaryIO, path: str, **kwargs) -> dict:
"""Save a file to local storage."""
"""Save a file, replacing any existing one atomically.
The bytes land on a temporary file beside the destination and are moved
into place with ``os.replace``, so an interrupted write leaves the
previous file intact instead of a truncated one. Streaming straight onto
the destination is unrecoverable for a file that is rewritten in place:
a half-written ``index.faiss`` loads at neither the old width nor the
new one, and ``application.scripts.reembed`` rewrites every index it
touches.
"""
full_path = self._get_full_path(path)
directory = os.path.dirname(full_path)
os.makedirs(directory, exist_ok=True)
os.makedirs(os.path.dirname(full_path), exist_ok=True)
if hasattr(file_data, 'save'):
file_data.save(full_path)
else:
with open(full_path, 'wb') as f:
shutil.copyfileobj(file_data, f)
# Same directory as the destination, so the replace below is a rename
# within one filesystem rather than a copy.
fd, temp_path = tempfile.mkstemp(
dir=directory, prefix=f".{os.path.basename(full_path)}.", suffix=".tmp"
)
try:
with os.fdopen(fd, "wb") as f:
if hasattr(file_data, "save"):
file_data.save(f)
else:
shutil.copyfileobj(file_data, f)
f.flush()
os.fsync(f.fileno())
# mkstemp is 0600; keep whatever the file already had, or fall back
# to what a plain open() would have produced.
try:
mode = os.stat(full_path).st_mode & 0o777
except FileNotFoundError:
mode = 0o644
os.chmod(temp_path, mode)
os.replace(temp_path, full_path)
except BaseException:
# A successful replace consumes the temp file; every other exit
# leaves it behind.
try:
os.unlink(temp_path)
except OSError:
# Best-effort: the write already failed, and that exception is
# the one worth raising. A temp file we cannot remove must not
# mask it.
pass
raise
return {
'storage_type': 'local'
+169 -1
View File
@@ -2,12 +2,17 @@
from __future__ import annotations
import codecs
import io
import os
from contextlib import suppress
from typing import BinaryIO
from typing import BinaryIO, Container
from application.core.settings import settings
from application.parser.file.constants import (
attachment_extension,
has_attachment_parser,
)
_COPY_CHUNK_BYTES = 64 * 1024
@@ -26,6 +31,169 @@ class UploadTooLargeError(ValueError):
"""Raised when one uploaded file exceeds the configured byte cap."""
class UnsupportedUploadTypeError(ValueError):
"""Raised when an uploaded attachment has a file type the worker cannot parse."""
_UNSUPPORTED_UPLOAD_PREFIX = "Unsupported file type"
def unsupported_upload_message(filename: str | None) -> str:
"""Return the stable client-facing rejection message for an unparseable upload.
Args:
filename: The rejected upload's filename.
Returns:
``"Unsupported file type: .mp4"`` style text, naming the extension.
"""
extension = attachment_extension(filename)
return f"{_UNSUPPORTED_UPLOAD_PREFIX}: {extension or '(no extension)'}"
def is_unsupported_upload_message(message: str | None) -> bool:
"""Return whether ``message`` is one produced by :func:`unsupported_upload_message`."""
return bool(message) and str(message).startswith(_UNSUPPORTED_UPLOAD_PREFIX)
# A suffix with no parser is read by ``SimpleDirectoryReader``'s plain-text
# fallthrough, so it is admitted on content: enough of the head to recognise a
# container header, and a tolerance that keeps real text (UTF-8 accents, an
# ANSI-coloured log) in while a random binary — ~12.5% of bytes below 0x20 —
# stays out.
_TEXT_SNIFF_BYTES = 8192
_MAX_NONTEXT_RATIO = 0.10
# Control bytes that occur in ordinary text: tab, LF, VT, FF, CR, ESC.
_TEXT_CONTROL_BYTES = frozenset({0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x1B})
_TEXT_CONTROL_CHARS = frozenset("\t\n\v\f\r\x1b")
# A UTF-16/32 file is half NUL bytes, so the byte rules below would reject it.
# Its BOM says which encoding to read it as; the decoded characters are then
# judged instead. Longest BOM first — UTF-32-LE starts with the UTF-16-LE one.
# A BOM is never a verdict on its own: three bytes must not buy a video a pass.
_TEXT_BOMS = (
(codecs.BOM_UTF32_LE, "utf-32-le"),
(codecs.BOM_UTF32_BE, "utf-32-be"),
(codecs.BOM_UTF8, None),
(codecs.BOM_UTF16_LE, "utf-16-le"),
(codecs.BOM_UTF16_BE, "utf-16-be"),
)
def _decoded_looks_like_text(text: str) -> bool:
"""Return whether decoded characters read as text.
Args:
text: Characters decoded from a BOM-marked sample.
Returns:
False when a NUL character appears — the byte-level rule, one level
up, since no text holds one — or when too many characters are
unprintable: control, unassigned, private-use or surrogate, which is
what binary decodes into. Replacement characters count as unprintable,
so bytes the decoder could not read are evidence rather than silently
dropped (a truncated character at the sample boundary is one of
thousands and cannot swing the ratio).
"""
if not text:
return True
if "\x00" in text:
return False
nontext = sum(
1
for char in text
if char == "�"
or (not char.isprintable() and char not in _TEXT_CONTROL_CHARS)
)
return nontext / len(text) <= _MAX_NONTEXT_RATIO
def looks_like_text(sample: bytes) -> bool:
"""Return whether a leading byte sample reads as text rather than binary.
Args:
sample: The first bytes of a file; an empty sample counts as text.
Returns:
False when the sample holds a NUL byte or too many other non-text
control bytes, True otherwise. A UTF-16/32 BOM switches the test to
the decoded characters; the content is still what decides.
"""
if not sample:
return True
body = sample
for bom, encoding in _TEXT_BOMS:
if not sample.startswith(bom):
continue
body = sample[len(bom) :]
if encoding is not None:
return _decoded_looks_like_text(body.decode(encoding, errors="replace"))
# UTF-8 BOM: the byte rules still apply to everything after it.
break
if not body:
return True
if b"\x00" in body:
return False
nontext = sum(
1
for byte in body
if (byte < 0x20 and byte not in _TEXT_CONTROL_BYTES) or byte == 0x7F
)
return nontext / len(body) <= _MAX_NONTEXT_RATIO
def file_looks_like_text(path: str | os.PathLike[str]) -> bool:
"""Return whether a file on disk reads as text, by its leading bytes.
Args:
path: Filesystem path to sample.
Returns:
The :func:`looks_like_text` verdict for the file's head; True when the
file cannot be read, leaving that failure to the parser to report.
"""
try:
with open(path, "rb") as handle:
return looks_like_text(handle.read(_TEXT_SNIFF_BYTES))
except OSError:
return True
def enforce_parseable_attachment(
path: str | os.PathLike[str],
filename: str | None,
parser_extensions: Container[str] | None = None,
) -> None:
"""Reject an attachment that no parser handles and that is not plain text.
Suffixes with a parser are admitted unconditionally — a PDF is binary and
parses fine. Everything else, .txt included, has to read as text: that is
what keeps a video or an archive out of the plain-text fallthrough while
leaving source, config and log files in, whatever the file is named.
Args:
path: Local path to the staged upload, readable before it is stored.
filename: The upload's filename, used for the suffix and the message.
parser_extensions: Suffixes to treat as parser-backed. Defaults to
``ATTACHMENT_PARSER_EXTENSIONS``, which assumes the full parser
table; callers holding the live extractor (the worker) pass its
keys, so a docling-less install does not admit a .webp on trust
and then read it as text.
Raises:
UnsupportedUploadTypeError: When the file has no parser and its
contents are binary.
"""
if parser_extensions is None:
has_parser = has_attachment_parser(filename)
else:
has_parser = attachment_extension(filename) in parser_extensions
if has_parser:
return
if file_looks_like_text(path):
return
raise UnsupportedUploadTypeError(unsupported_upload_message(filename))
class _LimitedRawReader(io.RawIOBase):
"""Expose a binary stream while enforcing an exact cumulative read cap."""
+28 -1
View File
@@ -131,6 +131,11 @@ def _persist_call_usage(llm, call_usage):
agent_id=str(agent_id) if agent_id else None,
prompt_tokens=call_usage["prompt_tokens"],
generated_tokens=call_usage["generated_tokens"],
# Present only when the provider reported the breakdown;
# persisted as NULL otherwise so "unknown" never reads as
# "0% cache hits".
cached_tokens=call_usage.get("cached_tokens"),
cache_write_tokens=call_usage.get("cache_write_tokens"),
source=(
getattr(llm, "_token_usage_source", None) or "agent_stream"
),
@@ -151,6 +156,12 @@ def _prefer_provider_usage(llm: Any, call_usage: Dict[str, int]) -> Dict[str, in
``*_tokens_details`` breakdowns (``cached_tokens``,
``reasoning_tokens``) back out of these bins — that would break
parity with what providers bill.
The prompt-cache sub-bins ARE carried alongside (``cached_tokens``,
``cache_write_tokens``; Anthropic's ``cache_creation_tokens`` maps to
the latter) so persistence and the finish events can chart them. They
are added only when the provider reported them. The rest of
``call_usage`` (e.g. ``model``) is preserved rather than replaced.
"""
reported = getattr(llm, "_last_usage", None)
if not isinstance(reported, dict):
@@ -174,10 +185,22 @@ def _prefer_provider_usage(llm: Any, call_usage: Dict[str, int]) -> Dict[str, in
# Slotted/immutable LLM stand-ins can't record the claim; this
# call still gets the provider counts, which is correct for them.
pass
return {
merged = {
**call_usage,
"prompt_tokens": int(prompt or 0),
"generated_tokens": int(completion or 0),
}
details = reported.get("prompt_tokens_details")
if isinstance(details, dict):
cached = details.get("cached_tokens")
written = details.get("cache_write_tokens")
if written is None:
written = details.get("cache_creation_tokens")
if cached is not None:
merged["cached_tokens"] = int(cached or 0)
if written is not None:
merged["cache_write_tokens"] = int(written or 0)
return merged
def gen_token_usage(func):
@@ -224,6 +247,8 @@ def gen_token_usage(func):
prompt_tokens=call_usage["prompt_tokens"],
completion_tokens=call_usage["generated_tokens"],
latency_ms=int((time.monotonic() - started_at) * 1000),
cached_tokens=call_usage.get("cached_tokens"),
cache_write_tokens=call_usage.get("cache_write_tokens"),
error=error,
)
except Exception:
@@ -272,6 +297,8 @@ def stream_token_usage(func):
prompt_tokens=call_usage["prompt_tokens"],
completion_tokens=call_usage["generated_tokens"],
latency_ms=int((time.monotonic() - started_at) * 1000),
cached_tokens=call_usage.get("cached_tokens"),
cache_write_tokens=call_usage.get("cache_write_tokens"),
error=error,
)
except Exception:
+151 -77
View File
@@ -1,5 +1,4 @@
import logging
import os
from abc import ABC, abstractmethod
from typing import Optional
@@ -7,7 +6,29 @@ import requests
from application.core.settings import settings
from application.vectorstore.embeddings_openai import OpenAIEmbeddings
from application.utils import get_encoding
from application.vectorstore.model_registry import (
dimension_for,
max_input_tokens_for,
resolve,
)
def _embeddings_name_is_explicit() -> bool:
"""True when ``EMBEDDINGS_NAME`` names a model somebody actually chose.
Not ``model_fields_set``: pydantic marks a field as set for any value that
reached it, including one read from ``.env``, and every setup script has
always written ``EMBEDDINGS_NAME`` unconditionally. An install carrying the
legacy name a script wrote years ago would read as a deliberate choice and
lend a remote server that model's context window.
"""
fields = getattr(type(settings), "model_fields", None)
if not isinstance(fields, dict):
return False
field = fields.get("EMBEDDINGS_NAME")
if field is None:
return False
return settings.EMBEDDINGS_NAME != field.default
class RemoteEmbeddings:
@@ -23,45 +44,83 @@ class RemoteEmbeddings:
self.headers = {"Content-Type": "application/json"}
if api_key:
self.headers["Authorization"] = f"Bearer {api_key}"
self.dimension = 768
# Width comes from the registry. This used to be a hardcoded 768 that
# ``embed_query`` claimed to correct on first use -- but the correction
# was guarded by ``if self.dimension is None``, which the hardcode made
# unreachable, so a remote model of any other width silently produced a
# ``vector(768)`` column. ``None`` here means "unknown", and the probe
# below now genuinely runs.
self.dimension = dimension_for(model_name)
def _token_counter(self):
"""Counter matching the remote model's tokenizer, cached per process."""
from application.parser.tokenization import get_token_counter
return get_token_counter(self.model_name)
def _resolve_input_limit(self):
"""Token ceiling for a single embed input, or ``None`` for no limit.
``EMBEDDINGS_MAX_INPUT_TOKENS`` wins when set. Otherwise a registered
model contributes its own context window, so a request that the server
would reject -- or silently truncate -- is clipped here instead of
being sent and paid for.
That fallback needs the name to mean something. For a remote server it
is only a label forwarded as the ``model`` field, so the settings
default must not lend the server mpnet's 384-token window: a name
nobody chose describes nothing, and clipping on it would silently
discard most of every chunk.
"""
configured = settings.EMBEDDINGS_MAX_INPUT_TOKENS
if configured and configured > 0:
return configured
if not _embeddings_name_is_explicit():
return None
model_limit = max_input_tokens_for(self.model_name)
return model_limit if model_limit and model_limit > 0 else None
def _truncate_inputs(self, inputs):
"""Clip each input to ``EMBEDDINGS_MAX_INPUT_TOKENS`` tokens.
"""Clip each input to the resolved token limit.
The remote server (e.g. llama.cpp) hard-rejects any single input
larger than its physical batch size with a 500. When the setting is
configured, each input is truncated to that many tokens before the
request and the overflow is dropped (lossy by design). Token counts
use the shared tiktoken encoding, which differs from the server's
tokenizer, so set the limit with headroom under the server's true
limit to absorb tokenizer skew.
larger than its physical batch size with a 500, so oversized inputs are
truncated before the request and the overflow is dropped (lossy by
design).
Counting uses the embedding model's own tokenizer where it is known, so
the limit and the count are in the same unit. When it is not -- an
unregistered model, or no tokenizer available -- this falls back to
tiktoken, and the limit should then carry headroom to absorb the skew
between the two tokenizers.
Args:
inputs: A single string or a list of strings to embed.
Returns:
The inputs with each string clipped to the token limit, or the
inputs unchanged when the limit is unset or non-positive.
inputs unchanged when no limit applies.
"""
limit = settings.EMBEDDINGS_MAX_INPUT_TOKENS
if not limit or limit <= 0:
limit = self._resolve_input_limit()
if not limit:
return inputs
encoding = get_encoding()
counter = self._token_counter()
def clip(text):
if not isinstance(text, str):
return text
tokens = encoding.encode_ordinary(text)
if len(tokens) <= limit:
count = counter.count(text)
if count <= limit:
return text
logging.warning(
"Truncating remote embeddings input from %d to %d tokens (%d dropped)",
len(tokens),
count,
limit,
len(tokens) - limit,
count - limit,
)
return encoding.decode(tokens[:limit])
pieces = counter.split(text, limit)
return pieces[0] if pieces else text
if isinstance(inputs, list):
return [clip(text) for text in inputs]
@@ -129,7 +188,7 @@ class RemoteEmbeddings:
def _get_embeddings_wrapper():
"""Lazy import of EmbeddingsWrapper to avoid loading SentenceTransformer when using remote embeddings."""
"""Lazy import of EmbeddingsWrapper, so a remote setup never loads ONNX."""
from application.vectorstore.embeddings_local import EmbeddingsWrapper
return EmbeddingsWrapper
@@ -178,36 +237,28 @@ class EmbeddingsSingleton:
@staticmethod
def _create_instance(embeddings_name, *args, **kwargs):
if embeddings_name == "openai_text-embedding-ada-002":
"""Build the runner for ``embeddings_name``, per the model registry.
The registry replaced a hand-maintained factory dict whose entries
existed only to rewrite a configured name into a repository id. That
rewrite is now a registry field, so an unknown name needs no entry
here: it is passed through as a Hugging Face repository.
"""
spec = resolve(embeddings_name)
if spec is not None and spec.provider == "openai":
return OpenAIEmbeddings(*args, **kwargs)
# Lazy import EmbeddingsWrapper only when needed (avoids loading SentenceTransformer)
EmbeddingsWrapper = _get_embeddings_wrapper()
embeddings_factory = {
"huggingface_sentence-transformers/all-mpnet-base-v2": lambda: EmbeddingsWrapper(
"sentence-transformers/all-mpnet-base-v2"
),
"huggingface_sentence-transformers-all-mpnet-base-v2": lambda: EmbeddingsWrapper(
"sentence-transformers/all-mpnet-base-v2"
),
"huggingface_hkunlp/instructor-large": lambda: EmbeddingsWrapper(
"hkunlp/instructor-large"
),
}
if embeddings_name in embeddings_factory:
if args or kwargs:
logging.debug(
"Dropping %d positional and %d keyword argument(s) for pinned "
"embeddings model %s: its factory takes none.",
len(args),
len(kwargs),
embeddings_name,
)
return embeddings_factory[embeddings_name]()
else:
return EmbeddingsWrapper(embeddings_name, *args, **kwargs)
if spec is not None and (args or kwargs):
logging.debug(
"Dropping %d positional and %d keyword argument(s) for registered "
"embeddings model %s: the registry supplies its configuration.",
len(args),
len(kwargs),
embeddings_name,
)
return EmbeddingsWrapper(embeddings_name)
return EmbeddingsWrapper(embeddings_name, *args, **kwargs)
def _azure_configured() -> bool:
@@ -219,17 +270,29 @@ def _azure_configured() -> bool:
)
def _delegation_enabled() -> bool:
"""True when this process should embed on the worker rather than locally.
Compared against ``True`` rather than coerced: tests patch ``settings``
with a ``MagicMock``, whose every attribute is a truthy object, and
``bool()`` on that would silently route them through the broker.
"""
return getattr(settings, "EMBEDDINGS_DELEGATE_TO_WORKER", False) is True
def get_embeddings(
embeddings_name: Optional[str] = None, embeddings_key: Optional[str] = None
):
"""Resolve the configured embeddings instance. The single entry point.
Callers that reach for :meth:`EmbeddingsSingleton.get_instance` directly
reproduce neither the bundled local-model path (the Docker image ships
``/app/models/all-mpnet-base-v2``) nor the OpenAI/Azure key handling: they
download a second copy of the model from the hub, and passing the key
positionally to the pinned HuggingFace names raises ``TypeError`` because
those factories take no arguments. Route every caller through here.
skip the remote dispatch and the OpenAI/Azure key handling. Route every
caller through here.
With ``EMBEDDINGS_DELEGATE_TO_WORKER`` this returns a client that runs the
model on the Celery worker, so an API process never loads one. The client
embeds locally when it finds itself inside a worker task, so the worker is
unaffected.
Args:
embeddings_name: Model name; defaults to ``settings.EMBEDDINGS_NAME``.
@@ -239,6 +302,34 @@ def get_embeddings(
The shared embeddings instance for the resolved model.
"""
embeddings_name = embeddings_name or settings.EMBEDDINGS_NAME
if not settings.EMBEDDINGS_BASE_URL and _delegation_enabled():
cache_key = f"delegated_{embeddings_name}"
if cache_key not in EmbeddingsSingleton._instances:
from application.vectorstore.embeddings_delegated import DelegatedEmbeddings
EmbeddingsSingleton._instances[cache_key] = DelegatedEmbeddings(
embeddings_name, embeddings_key
)
return EmbeddingsSingleton._instances[cache_key]
return build_local_embeddings(embeddings_name, embeddings_key)
def build_local_embeddings(
embeddings_name: Optional[str] = None, embeddings_key: Optional[str] = None
):
"""Resolve the embeddings instance that runs in *this* process.
Bypasses worker delegation, so it is what the worker's embed task and the
boot hook use. Everything else should call :func:`get_embeddings`.
Args:
embeddings_name: Model name; defaults to ``settings.EMBEDDINGS_NAME``.
embeddings_key: API key; defaults to ``settings.EMBEDDINGS_KEY``.
Returns:
The shared in-process embeddings instance for the resolved model.
"""
embeddings_name = embeddings_name or settings.EMBEDDINGS_NAME
embeddings_key = (
embeddings_key if embeddings_key is not None else settings.EMBEDDINGS_KEY
)
@@ -250,7 +341,11 @@ def get_embeddings(
)
return EmbeddingsSingleton._remote_instance(embeddings_name, embeddings_key)
if embeddings_name == "openai_text-embedding-ada-002":
# Match through the registry, not on the canonical string: the bare
# ``text-embedding-ada-002`` alias resolves here too, and skipping this
# branch would drop the Azure deployment name and the API key.
spec = resolve(embeddings_name)
if spec is not None and spec.provider == "openai":
if _azure_configured():
embedding_instance = EmbeddingsSingleton.get_instance(
embeddings_name, model=settings.AZURE_EMBEDDINGS_DEPLOYMENT_NAME
@@ -259,31 +354,10 @@ def get_embeddings(
embedding_instance = EmbeddingsSingleton.get_instance(
embeddings_name, openai_api_key=embeddings_key
)
elif embeddings_name == "huggingface_sentence-transformers/all-mpnet-base-v2":
possible_paths = [
"/app/models/all-mpnet-base-v2", # Docker absolute path
"./models/all-mpnet-base-v2", # Relative path
]
local_model_path = None
for path in possible_paths:
if os.path.exists(path):
local_model_path = path
logging.info(f"Found local model at path: {path}")
break
else:
logging.info(f"Path does not exist: {path}")
if local_model_path:
embedding_instance = EmbeddingsSingleton.get_instance(
local_model_path,
)
else:
logging.warning(
f"Local model not found in any of the paths: {possible_paths}. Falling back to HuggingFace download."
)
embedding_instance = EmbeddingsSingleton.get_instance(
embeddings_name,
)
else:
# No per-model branching: the registry resolves names and FastEmbed
# caches artifacts under EMBEDDINGS_CACHE_DIR, which is where the
# image warms them at build time.
embedding_instance = EmbeddingsSingleton.get_instance(embeddings_name)
return embedding_instance
@@ -0,0 +1,254 @@
"""Query embedding executed in the Celery worker instead of in the API.
The API embeds every query it serves, so it needs an embedder -- and a local
one costs roughly 800 MB of ONNX Runtime per process. That is the whole
footprint of an API container that otherwise holds no model.
This client keeps the interface (``embed_query``/``embed_documents``/
``dimension``) and moves only the computation: the text goes to the worker over
Celery and the vector comes back. The API pays a broker round trip per query
and no resident model.
Inside a worker there is nothing to delegate to -- dispatching would queue work
behind the task already running and wait on itself -- so a call made while a
task is executing runs locally, on a model this process loads once and caches.
``DOCUMENT_PARSE_QUEUE`` exists for the same reason on the parsing side.
Production deployments should point ``EMBEDDINGS_BASE_URL`` at a real embedding
service instead: that removes the model from *both* processes and costs a
network hop rather than a broker round trip.
"""
from __future__ import annotations
import logging
import threading
import time
from typing import Any, List, Optional
from application.core.settings import settings
from application.vectorstore.model_registry import dimension_for
logger = logging.getLogger(__name__)
#: Dispatched by name so the API never imports the task module -- and through
#: it ``application.worker``, which pulls in the whole parsing stack.
EMBED_TASK = "application.vectorstore.embeddings_tasks.embed_texts"
#: How long after a failed dispatch to fail fast instead of waiting out another
#: full ``EMBEDDINGS_DELEGATE_TIMEOUT``. Short enough that a worker restart is
#: picked up within one query, long enough to collapse the retries inside a
#: single retrieval into one timeout rather than one per source.
_FAILURE_COOLDOWN = 30.0
#: How long a caller waits for the outcome of the dispatch already in flight
#: before giving up on its own. Only applies while the worker is unproven --
#: once one dispatch has succeeded, every caller goes straight to the broker.
#: Comfortably above a healthy round trip (~60 ms on a prefork worker) and far
#: below ``EMBEDDINGS_DELEGATE_TIMEOUT``, which is the point.
_PROBE_WAIT = 2.0
_NO_WORKER_HINT = (
"Start a worker consuming it, point EMBEDDINGS_BASE_URL at an embedding "
"service, or set EMBEDDINGS_DELEGATE_TO_WORKER=false to load the model in "
"this process instead."
)
def _forget(result) -> None:
"""Drop the task's stored vector from the result backend.
Nothing ever reads it back. The key is ``celery-task-meta-<uuid>``, minted
per dispatch rather than derived from the text, so a repeated query is a new
task and a new key -- the value is written once, read once by the ``get()``
already waiting on it, then dead. Left alone it occupies ~17 KB for
``result_expires`` (7 days), in the Redis the broker also runs on.
Also releases the backend's pub/sub subscription for the task, which
``get()`` alone does not.
Never raises: the vector is already in hand, and a backend that cannot
delete must not fail the search. On the timeout path the worker may still
store its result afterwards, leaving one orphaned key -- no worse than not
forgetting at all, and bounded by the same expiry.
"""
try:
result.forget()
except Exception as exc: # noqa: BLE001 — cleanup must never fail a query
logger.debug("Could not forget the embed task result: %s", exc)
def _in_worker() -> bool:
"""True when a Celery task is executing in this process."""
try:
from application.celery_init import celery
return celery.current_worker_task is not None
except Exception:
return False
class DelegatedEmbeddings:
"""Embeds by dispatching to the Celery worker, or locally inside one."""
def __init__(self, embeddings_name: str, embeddings_key: Optional[str] = None) -> None:
self.embeddings_name = embeddings_name
self.embeddings_key = embeddings_key
self._local: Any = None
self._dimension: Optional[int] = dimension_for(embeddings_name)
self._failed_at: Optional[float] = None
# A dispatch has completed successfully, so the worker is known to be
# consuming the queue and callers need not take turns proving it.
self._verified = False
self._probing = False
self._state_lock = threading.Lock()
self._probe_done = threading.Event()
def _cooldown_remaining(self) -> float:
"""Seconds left of the fail-fast window after a failed dispatch."""
# One load: a concurrent success clearing the latch between two reads
# would otherwise subtract from None.
failed_at = self._failed_at
if failed_at is None:
return 0.0
return max(0.0, _FAILURE_COOLDOWN - (time.monotonic() - failed_at))
def _local_embeddings(self):
"""The in-process model, built once, for use inside a worker task."""
if self._local is None:
from application.vectorstore.base import build_local_embeddings
self._local = build_local_embeddings(self.embeddings_name, self.embeddings_key)
return self._local
def _send(self, texts: List[str], queue: str, timeout: int) -> List[List[float]]:
"""Publish the embed task and wait for its vectors."""
from application.celery_init import celery
result = celery.send_task(EMBED_TASK, args=[texts, self.embeddings_name], queue=queue)
try:
vectors = result.get(timeout=timeout)
except Exception as exc:
self._failed_at = time.monotonic()
# Drop the proof with the worker that supplied it. ``_verified``
# short-circuits ahead of the probe gate, so leaving it set means
# the gate only ever covers a worker that was never healthy --
# while the case that actually happens is a healthy one being
# redeployed or OOM-killed. Every caller would then pay the full
# timeout, together, on every wave once the cooldown lapses.
self._verified = False
raise RuntimeError(
f"Embedding request to the Celery worker timed out or failed ({exc}). "
f"A worker must be consuming the {queue!r} queue for retrieval to "
f"work. {_NO_WORKER_HINT}"
) from exc
finally:
_forget(result)
self._failed_at = None
self._verified = True
return vectors
def _cooldown_error(self, queue: str, remaining: float) -> RuntimeError:
return RuntimeError(
f"Skipping the embed dispatch: a previous request to the {queue!r} "
f"queue failed and the {_FAILURE_COOLDOWN}s cooldown has "
f"{remaining:.0f}s left. {_NO_WORKER_HINT}"
)
def _dispatch(self, texts: List[str]) -> List[List[float]]:
"""Run the embed task on the worker and wait for its vectors.
A missing worker is a property of the deployment, not of this call, so
at most one caller waits out ``EMBEDDINGS_DELEGATE_TIMEOUT`` to discover
it. Two guards do that:
The cooldown latch covers requests arriving *after* a failure -- without
it a single retrieval pays the timeout twice, once in
``fanout.embed_questions`` and again per source when it falls back to
letting each store embed its own query.
The probe covers requests already in flight *alongside* the first one,
which the latch cannot: nothing is latched until that first ``get()``
returns, so every thread in the opening wave would otherwise block for
the full timeout at once -- at the shipped 60s and 96 WSGI threads, an
API that serves nothing at all, health checks included.
"""
queue = getattr(settings, "EMBEDDINGS_QUEUE", "embeddings")
timeout = getattr(settings, "EMBEDDINGS_DELEGATE_TIMEOUT", 60)
remaining = self._cooldown_remaining()
if remaining > 0:
raise self._cooldown_error(queue, remaining)
if self._verified:
return self._send(texts, queue, timeout)
with self._state_lock:
# "send" -- proven while we waited for the lock, just go.
# "wait" -- another caller is already finding out; don't pay a
# second full timeout to learn the same thing.
# "probe" -- nobody is; this call is the one that finds out.
role = "send" if self._verified else "wait" if self._probing else "probe"
if role == "probe":
self._probing = True
self._probe_done.clear()
if role == "probe":
try:
return self._send(texts, queue, timeout)
finally:
with self._state_lock:
self._probing = False
self._probe_done.set()
if role == "wait":
self._probe_done.wait(_PROBE_WAIT)
remaining = self._cooldown_remaining()
if remaining > 0:
raise self._cooldown_error(queue, remaining)
if not self._verified:
raise RuntimeError(
f"Skipping the embed dispatch: an earlier request to the "
f"{queue!r} queue is still unanswered after {_PROBE_WAIT}s, so "
f"no worker appears to be consuming it. {_NO_WORKER_HINT}"
)
return self._send(texts, queue, timeout)
def embed_documents(self, documents: List[str]) -> List[List[float]]:
"""Embed a list of texts, preserving order."""
if not documents:
return []
if _in_worker():
return self._local_embeddings().embed_documents(documents)
vectors = self._dispatch(list(documents))
if self._dimension is None and vectors:
self._dimension = len(vectors[0])
return vectors
def embed_query(self, query: str) -> List[float]:
"""Embed a single query string."""
return self.embed_documents([query])[0]
@property
def dimension(self) -> Optional[int]:
"""Vector width, from the registry where possible.
Falls back to one round trip for a model the registry does not
describe, and to ``None`` when even that fails -- callers already treat
an unknown width as "nothing to compare yet" rather than an error.
"""
if self._dimension is None:
try:
self._dimension = len(self.embed_query("dimension probe"))
except Exception as exc:
logger.warning("Could not determine embedding width: %s", exc)
return None
return self._dimension
def __call__(self, text):
if isinstance(text, str):
return self.embed_query(text)
elif isinstance(text, list):
return self.embed_documents(text)
raise ValueError("Input must be a string or a list of strings")
+336 -38
View File
@@ -1,54 +1,352 @@
"""
Local embeddings using SentenceTransformer.
This module is only imported when EMBEDDINGS_BASE_URL is not set,
to avoid loading SentenceTransformer into memory when using remote embeddings.
"""Local embeddings via FastEmbed (ONNX Runtime).
Replaces the previous SentenceTransformer implementation. Both run the same
weights; FastEmbed reaches them through ONNX Runtime instead of torch, which
removes torch, transformers and sentence-transformers from the dependency set
and measurably reduces both resident memory and import cost.
The swap is numerically transparent for existing indexes: embedding the same
text with SentenceTransformer and with FastEmbed's fp32 ONNX graph of mpnet
returns vectors at cosine 1.0, so an mpnet index built before this change
keeps working unchanged. That result is specific to the fp32 graph -- the
granite entries run an int8-quantised one and are not bit-comparable to a
fp32 index of the same model.
Models are described in :mod:`application.vectorstore.model_registry`. A name
the registry does not know is treated as a Hugging Face repository, which is
what someone configuring an arbitrary model expects; how to run it is read
from the repository itself rather than assumed.
"""
import json
import logging
import threading
from dataclasses import replace
from typing import Any, List, Optional
from sentence_transformers import SentenceTransformer
from application.core.settings import settings
from application.vectorstore.model_registry import EmbeddingModel, resolve, known_names
logger = logging.getLogger(__name__)
# ``add_custom_model`` mutates a process-global registry inside FastEmbed, so
# repeated registration of the same name is both wasteful and racy under the
# thread pool the API serves requests from.
_registered: set = set()
_register_lock = threading.Lock()
# Last-resort layout for a repository that declares nothing about itself.
_FALLBACK_ONNX_FILE = "onnx/model.onnx"
_FALLBACK_POOLING = "mean"
# Sentence-transformers records how a model turns token vectors into one
# vector, and whether it normalises the result, as files in the repository.
# Reading them is the difference between running a model and running something
# that merely shares its weights: mean-pooling a CLS model returns vectors at
# cosine ~0.95 to the correct ones -- close enough to look like it works, far
# enough to degrade retrieval, and silent either way.
_POOLING_CONFIG = "1_Pooling/config.json"
_MODULES_CONFIG = "modules.json"
def _pooling_type(pooling: str):
"""Map our ``"cls"``/``"mean"`` spelling onto FastEmbed's enum."""
from fastembed.common.model_description import PoolingType
return {"cls": PoolingType.CLS, "mean": PoolingType.MEAN}[pooling]
def _is_builtin(repo: str) -> bool:
"""True when FastEmbed already ships a description for ``repo``."""
from fastembed import TextEmbedding
lowered = repo.lower()
return any(
str(entry.get("model", "")).lower() == lowered
for entry in TextEmbedding.list_supported_models()
)
def _register(model: EmbeddingModel) -> None:
"""Teach FastEmbed about a model, exactly once per process."""
from fastembed import TextEmbedding
from fastembed.common.model_description import ModelSource
with _register_lock:
if model.repo in _registered:
return
if _is_builtin(model.repo):
# ``add_custom_model`` refuses a name FastEmbed already ships, and
# its own description carries the pooling, width and graph file we
# would be supplying, so there is nothing to add. Without this,
# configuring any of FastEmbed's ~30 built-in models (bge, e5,
# MiniLM, gte, ...) fails every embed call.
_registered.add(model.repo)
return
TextEmbedding.add_custom_model(
model=model.repo,
pooling=_pooling_type(model.pooling),
normalization=model.normalize,
sources=ModelSource(hf=model.repo),
dim=model.dimension,
model_file=model.onnx_file,
)
_registered.add(model.repo)
def _read_repo_json(repo: str, filename: str) -> Optional[dict]:
"""Fetch one small JSON from ``repo``, or ``None`` when it is not there.
Reads through the Hugging Face hub cache, so a warmed image finds it
offline. Every failure -- absent file, no network, malformed JSON -- is the
same answer to the caller: this repository does not tell us.
"""
try:
from huggingface_hub import hf_hub_download
with open(hf_hub_download(repo_id=repo, filename=filename), encoding="utf-8") as handle:
return json.load(handle)
except Exception as exc:
logger.debug("No %s for %s (%s)", filename, repo, exc)
return None
def _describe_from_repo(repo: str) -> Optional[EmbeddingModel]:
"""Build a spec from a repository's sentence-transformers metadata.
Args:
repo: Hugging Face repository id.
Returns:
The described model, or ``None`` when the repository carries no
metadata to read -- leaving the caller to fall back to assumptions.
Raises:
RuntimeError: If the model has a Dense projection head. FastEmbed runs
the transformer and pools it, and nothing else, so the projection
would be skipped and the vectors come out both the wrong width and
in a different space. There is no correct way to run it here.
"""
pooling_config = _read_repo_json(repo, _POOLING_CONFIG)
if pooling_config is None:
return None
kinds = {
str(module.get("type", "")).rsplit(".", 1)[-1]
for module in (_read_repo_json(repo, _MODULES_CONFIG) or [])
if isinstance(module, dict)
}
if "Dense" in kinds:
raise RuntimeError(
f"Embedding model {repo!r} has a Dense projection layer, which FastEmbed "
"cannot run: its vectors would be the wrong width and in a different "
"space than the model was trained to produce. Choose a model without "
"one, or serve this one over EMBEDDINGS_BASE_URL."
)
if pooling_config.get("pooling_mode_cls_token"):
pooling = "cls"
elif pooling_config.get("pooling_mode_mean_tokens"):
pooling = "mean"
else:
# max, mean_sqrt_len, weighted-mean: FastEmbed offers none of them, so
# there is nothing to describe and guessing is what we are avoiding.
logger.warning(
"Embedding model %s uses a pooling mode FastEmbed cannot reproduce (%s).",
repo,
", ".join(sorted(k for k, v in pooling_config.items() if v is True)) or "unknown",
)
return None
# ``Normalize`` appears in modules.json only when the model L2-normalises.
# Its absence is a fact, not missing data: a dot-product model is trained
# on unnormalised vectors and normalising re-ranks its results.
normalize = "Normalize" in kinds
dimension = int(pooling_config.get("word_embedding_dimension") or 0)
logger.info(
"Embedding model %s declares %s pooling, normalize=%s, dimension=%s.",
repo,
pooling,
normalize,
dimension or "unknown",
)
return EmbeddingModel(
name=repo,
dimension=dimension,
max_input_tokens=512,
pooling=pooling,
normalize=normalize,
repo=repo,
onnx_file=_FALLBACK_ONNX_FILE,
)
def _apply_overrides(spec: EmbeddingModel) -> EmbeddingModel:
"""Let ``EMBEDDINGS_POOLING``/``EMBEDDINGS_NORMALIZE`` win over any source."""
pooling = getattr(settings, "EMBEDDINGS_POOLING", None)
normalize = getattr(settings, "EMBEDDINGS_NORMALIZE", None)
changes = {}
if isinstance(pooling, str) and pooling.strip().lower() in ("cls", "mean"):
changes["pooling"] = pooling.strip().lower()
if isinstance(normalize, bool):
changes["normalize"] = normalize
if not changes:
return spec
logger.info("Overriding %s from settings: %s", spec.repo, changes)
return replace(spec, **changes)
def _spec_for(model_name: str) -> EmbeddingModel:
"""Registry entry for ``model_name``, or a best-effort one for a raw repo.
Args:
model_name: Configured ``EMBEDDINGS_NAME``.
Returns:
The registry entry, else one read from the repository's own
sentence-transformers metadata, else a last-resort entry assuming the
standard ONNX layout with mean pooling and ``dimension = 0`` so the
caller knows to probe for the real width. Settings overrides win over
all three.
"""
spec = resolve(model_name)
if spec is not None:
return _apply_overrides(spec)
described = _describe_from_repo(model_name)
if described is not None:
return _apply_overrides(described)
logger.warning(
"Embedding model %r is not in the registry (known: %s) and its repository "
"declares no pooling, so %s with L2 normalisation is assumed. If that is "
"wrong the vectors will be quietly poor rather than fail; set "
"EMBEDDINGS_POOLING=cls|mean and EMBEDDINGS_NORMALIZE to pin it.",
model_name,
", ".join(known_names()),
_FALLBACK_POOLING,
)
return _apply_overrides(
EmbeddingModel(
name=model_name,
dimension=0,
max_input_tokens=512,
pooling=_FALLBACK_POOLING,
normalize=True,
repo=model_name,
onnx_file=_FALLBACK_ONNX_FILE,
)
)
def _pad_to_longest_in_batch(model: Any) -> None:
"""Undo a fixed padding width baked into a model's ``tokenizer.json``."""
# FastEmbed enables padding only when the tokenizer declares none, so a
# fixed ``length`` survives loading. Shorter inputs are then padded to that
# width while longer ones keep their own, the batch is ragged, and the ONNX
# tensor build fails. mpnet ships ``length: 128``; granite does not.
# Mean pooling masks pad tokens, so the vectors are unaffected.
tokenizer = getattr(getattr(model, "model", None), "tokenizer", None)
padding = getattr(tokenizer, "padding", None)
if not isinstance(padding, dict) or padding.get("length") is None:
return
tokenizer.enable_padding(
direction=padding.get("direction", "right"),
pad_id=padding.get("pad_id", 0),
pad_type_id=padding.get("pad_type_id", 0),
pad_token=padding.get("pad_token", "<pad>"),
length=None,
pad_to_multiple_of=padding.get("pad_to_multiple_of"),
)
class EmbeddingsWrapper:
def __init__(self, model_name, *args, **kwargs):
logging.info(f"Initializing EmbeddingsWrapper with model: {model_name}")
"""Runs an embedding model locally through FastEmbed.
Exposes the ``embed_query``/``embed_documents``/``dimension`` interface the
vector stores rely on, matching ``RemoteEmbeddings`` and ``OpenAIEmbeddings``.
"""
def __init__(self, model_name: str, *args: Any, **kwargs: Any) -> None:
"""Load ``model_name`` locally.
Args:
model_name: Registry name, alias, or a Hugging Face repository id.
Raises:
RuntimeError: If the model cannot be loaded, with the configured
name and the registry's known names in the message.
"""
from fastembed import TextEmbedding
self.spec = _spec_for(model_name)
logger.info("Loading embeddings model %s via FastEmbed", self.spec.repo)
try:
kwargs.setdefault("trust_remote_code", True)
self.model = SentenceTransformer(
model_name,
config_kwargs={"allow_dangerous_deserialization": True},
*args,
**kwargs,
)
if self.model is None or self.model._first_module() is None:
raise ValueError(
f"SentenceTransformer model failed to load properly for: {model_name}"
)
# Renamed in sentence-transformers 5.4; keep the old name as a fallback.
get_dimension = getattr(
self.model,
"get_embedding_dimension",
getattr(self.model, "get_sentence_embedding_dimension", None),
)
self.dimension = get_dimension()
logging.info(f"Successfully loaded model with dimension: {self.dimension}")
except Exception as e:
logging.error(
f"Failed to initialize SentenceTransformer with model {model_name}: {str(e)}",
exc_info=True,
)
raise
_register(self.spec)
init_kwargs = {"model_name": self.spec.repo}
threads = getattr(settings, "EMBEDDINGS_THREADS", None)
if isinstance(threads, int) and threads > 0:
init_kwargs["threads"] = threads
cache_dir = getattr(settings, "EMBEDDINGS_CACHE_DIR", None)
if cache_dir:
init_kwargs["cache_dir"] = cache_dir
self.model = TextEmbedding(**init_kwargs)
except Exception as exc:
raise RuntimeError(
f"Could not load embeddings model {model_name!r} via FastEmbed: "
f"{exc}. Known models: {', '.join(known_names())}."
) from exc
def embed_query(self, query: str):
return self.model.encode(query).tolist()
_pad_to_longest_in_batch(self.model)
self.dimension = self.spec.dimension or self._probe_dimension()
logger.info("Embeddings model ready (dimension=%d)", self.dimension)
def embed_documents(self, documents: list):
return self.model.encode(documents).tolist()
def _probe_dimension(self) -> int:
"""Determine the vector width of a model the registry does not describe."""
return len(self.embed_query("dimension probe"))
def embed_query(self, query: str) -> List[float]:
"""Embed a single query string."""
return self.embed_documents([query])[0]
def embed_documents(self, documents: List[str]) -> List[List[float]]:
"""Embed a list of documents, preserving input order.
Batched by ``EMBEDDINGS_MODEL_BATCH_SIZE``, not by the pipeline's
``EMBEDDINGS_BATCH_SIZE``: one is documents per forward pass, the other
is chunks per store transaction, and sizing the forward pass from the
transaction is what made ingest peak at 6.6 GB.
Inputs are grouped by length first. ONNX needs a rectangular tensor, so
every input in a pass is padded up to the longest one in it; with mixed
lengths that padding is most of the work. The original order is
restored before returning, so callers zipping these against their texts
are unaffected.
"""
if not documents:
return []
batch_size: Optional[int] = None
raw = getattr(settings, "EMBEDDINGS_MODEL_BATCH_SIZE", None)
if isinstance(raw, int) and not isinstance(raw, bool) and raw > 0:
batch_size = raw
documents = list(documents)
if batch_size is None or len(documents) <= batch_size:
# One batch either way: sorting would only add work.
return [v.tolist() for v in self.model.embed(documents, batch_size=batch_size)]
order = sorted(range(len(documents)), key=lambda i: len(documents[i]))
grouped = [documents[i] for i in order]
vectors = [v.tolist() for v in self.model.embed(grouped, batch_size=batch_size)]
restored: List[Optional[List[float]]] = [None] * len(documents)
for position, original_index in enumerate(order):
restored[original_index] = vectors[position]
return restored
def __call__(self, text):
if isinstance(text, str):
return self.embed_query(text)
elif isinstance(text, list):
return self.embed_documents(text)
else:
raise ValueError("Input must be a string or a list of strings")
raise ValueError("Input must be a string or a list of strings")
@@ -0,0 +1,29 @@
"""The Celery task behind :mod:`application.vectorstore.embeddings_delegated`.
Kept out of ``application.api.user.tasks`` deliberately: that module imports
``application.worker`` and the whole parsing stack with it, which is the
opposite of what delegation is for.
"""
from __future__ import annotations
from typing import List, Optional
from application.celery_init import celery
from application.vectorstore.embeddings_delegated import EMBED_TASK
@celery.task(name=EMBED_TASK, acks_late=False, ignore_result=False)
def embed_texts(texts: List[str], embeddings_name: Optional[str] = None) -> List[List[float]]:
"""Embed ``texts`` with the worker's local model.
Args:
texts: Strings to embed.
embeddings_name: Model to use; the configured one when omitted.
Returns:
One vector per input, in input order.
"""
from application.vectorstore.base import get_embeddings
return get_embeddings(embeddings_name).embed_documents(list(texts))
+80 -27
View File
@@ -81,7 +81,29 @@ class FaissStore(BaseVectorStore):
# and must not be shown as one.
score_kind = "l2_distance"
def __init__(self, source_id: str, embeddings_key: str, docs_init=None):
def __init__(
self,
source_id: str,
embeddings_key: str,
docs_init=None,
ids=None,
batch_size=None,
skip_dimension_check: bool = False,
):
"""Open or build one source's FAISS index.
Args:
source_id: Source whose index to open.
embeddings_key: API key handed to the embeddings provider.
docs_init: Documents to build a fresh index from. Loads the stored
index instead when omitted.
ids: Chunk ids to keep when building. Generated when omitted.
batch_size: Documents per embed call when building.
skip_dimension_check: Open an index whose width does not match the
configured model. Only for a caller that is about to replace
that index, such as the re-embed script, which otherwise cannot
read the chunks it needs to rebuild from.
"""
super().__init__()
self.source_id = source_id
self.path = get_vectorstore(source_id)
@@ -94,27 +116,47 @@ class FaissStore(BaseVectorStore):
try:
if docs_init:
self._build_from_documents(docs_init)
self._build_from_documents(docs_init, ids=ids, batch_size=batch_size)
else:
self._load_from_storage()
except Exception as e:
raise Exception(f"Error loading FAISS index: {str(e)}")
self.assert_embedding_dimensions(self.embeddings)
if not skip_dimension_check:
self.assert_embedding_dimensions(self.embeddings)
# -- Construction ----------------------------------------------------
def _build_from_documents(self, docs_init) -> None:
"""Create a fresh index seeded with ``docs_init``."""
def _build_from_documents(self, docs_init, ids=None, batch_size=None) -> None:
"""Create a fresh index seeded with ``docs_init``.
Args:
docs_init: Documents to embed.
ids: Chunk ids to keep. Generated when omitted, which renumbers
every chunk and orphans anything referencing the old ids.
batch_size: Documents per embed call. Without it the whole index
goes out in one call, which a remote embeddings server rejects
or times out on.
"""
texts, metadatas = [], []
for doc in docs_init:
texts.append(getattr(doc, "page_content", None) or getattr(doc, "text", "") or "")
metadatas.append(getattr(doc, "metadata", None) or getattr(doc, "extra_info", None) or {})
faiss = _dependable_faiss_import()
vectors = self.embeddings.embed_documents(texts)
self.index = faiss.IndexFlatL2(len(vectors[0]))
self._append(texts, metadatas, vectors)
ids = list(ids) if ids else None
step = batch_size if batch_size and batch_size > 0 else len(texts)
for start in range(0, len(texts), max(1, step)):
stop = start + max(1, step)
vectors = self.embeddings.embed_documents(texts[start:stop])
if self.index is None:
self.index = faiss.IndexFlatL2(len(vectors[0]))
self._append(
texts[start:stop],
metadatas[start:stop],
vectors,
ids[start:stop] if ids else None,
)
def _load_from_storage(self) -> None:
"""Load the index and its sidecar, preferring JSON over the pickle."""
@@ -283,7 +325,13 @@ class FaissStore(BaseVectorStore):
f.write(dump_pickle_sidecar(self.documents, self.index_to_docstore_id))
def _save_to_storage(self) -> bool:
"""Persist the index through the configured storage backend."""
"""Persist the index through the configured storage backend.
Each file is replaced atomically by the backend, so an interrupted save
leaves the previous one readable. The three are still written in
sequence: a crash between them pairs a new index with an older sidecar,
which stays consistent because the ids and row order are preserved.
"""
with tempfile.TemporaryDirectory() as temp_dir:
self._write_index_files(temp_dir)
storage_path = get_vectorstore(self.source_id)
@@ -301,24 +349,29 @@ class FaissStore(BaseVectorStore):
# -- Introspection ---------------------------------------------------
def assert_embedding_dimensions(self, embeddings) -> None:
"""Check the index width matches the embedding model's width."""
if (
settings.EMBEDDINGS_NAME
== "huggingface_sentence-transformers/all-mpnet-base-v2"
):
word_embedding_dimension = getattr(embeddings, "dimension", None)
if word_embedding_dimension is None:
raise AttributeError(
"'dimension' attribute not found in embeddings instance."
)
if self.index is None:
return
if word_embedding_dimension != self.index.d:
raise ValueError(
f"Embedding dimension mismatch: embeddings.dimension "
f"({word_embedding_dimension}) != docsearch index dimension "
f"({self.index.d})"
)
"""Check the index width matches the embedding model's width.
This used to run only when ``EMBEDDINGS_NAME`` was mpnet, so every
other model skipped the check entirely -- exactly the models most
likely to differ from an index built earlier. It now runs for any
model that reports a width.
"""
word_embedding_dimension = getattr(embeddings, "dimension", None)
if word_embedding_dimension is None:
# A remote model of unknown width reports None until its first
# call; there is nothing to compare yet.
return
if self.index is None:
return
if word_embedding_dimension != self.index.d:
raise ValueError(
f"Embedding dimension mismatch: {settings.EMBEDDINGS_NAME} produces "
f"{word_embedding_dimension}-dim vectors but this FAISS index is "
f"{self.index.d}-dim. The index was built with a different "
f"embedding model; re-embed it with "
f"`python -m application.scripts.reembed` or point "
f"EMBEDDINGS_NAME back at the original model."
)
def get_chunks(self) -> List[Dict[str, Any]]:
"""Return every chunk held in the index."""
+5 -1
View File
@@ -2,6 +2,7 @@ from typing import List, Optional
import importlib
from application.vectorstore.base import BaseVectorStore
from application.core.settings import settings
from application.vectorstore.model_registry import DEFAULT_EMBEDDING_DIMENSION
class LanceDBVectorStore(BaseVectorStore):
"""Class for LanceDB Vector Store integration."""
@@ -54,8 +55,11 @@ class LanceDBVectorStore(BaseVectorStore):
"""Ensure the table exists before performing operations."""
if self.table is None:
embeddings = self._get_embeddings(settings.EMBEDDINGS_NAME, self.embeddings_key)
# A model outside the registry reports no width until it has run;
# ``list_size=None`` is a TypeError, not a permissive schema.
dimension = getattr(embeddings, "dimension", None) or DEFAULT_EMBEDDING_DIMENSION
schema = self.pa.schema([
self.pa.field("vector", self.pa.list_(self.pa.float32(), list_size=embeddings.dimension)),
self.pa.field("vector", self.pa.list_(self.pa.float32(), list_size=dimension)),
self.pa.field("text", self.pa.string()),
self.pa.field("metadata", self.pa.struct([
self.pa.field("key", self.pa.string()),
+177
View File
@@ -0,0 +1,177 @@
"""Canonical description of every embedding model DocsGPT knows how to run.
``EMBEDDINGS_NAME`` used to be a free-form string interpreted in half a dozen
places: a factory dict here, a bundled-model path probe there, a dimension
assertion gated on one model's name, and a hardcoded ``dimension = 768`` on the
remote client. Each of those encoded a different subset of the same facts, and
they drifted.
This module is the single place those facts live. A model is described once and
every consumer -- the local runner, the remote client, the schema bootstrap, the
chunker -- reads the same entry.
Unknown names are not an error: :func:`resolve` returns ``None`` and callers
fall back to treating the name as a Hugging Face repository, which is what a
user configuring an arbitrary model expects.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict, Optional, Tuple
@dataclass(frozen=True)
class EmbeddingModel:
"""Everything the application needs to know about one embedding model.
Attributes:
name: Canonical ``EMBEDDINGS_NAME`` value.
dimension: Width of the vectors it produces.
max_input_tokens: The model's own context window, counted in *its*
tokenizer. Used to bound what we send it, never to silently
reshape chunks.
pooling: ``"cls"`` or ``"mean"`` -- how token vectors become one vector.
normalize: Whether outputs are L2-normalised to unit length.
provider: Which runner handles it (``"fastembed"`` or ``"openai"``).
repo: Hugging Face repository holding weights and tokenizer.
onnx_file: Path within ``repo`` to the ONNX graph to run.
aliases: Other spellings of ``name`` accepted from configuration.
"""
name: str
dimension: int
max_input_tokens: int
pooling: str = "mean"
normalize: bool = True
provider: str = "fastembed"
repo: Optional[str] = None
onnx_file: Optional[str] = None
aliases: Tuple[str, ...] = field(default_factory=tuple)
#: The model DocsGPT installed before the granite migration. Kept as the
#: default so an existing deployment that upgrades keeps its index working;
#: new installs are pointed at granite by the setup script and env template.
MPNET = EmbeddingModel(
name="huggingface_sentence-transformers/all-mpnet-base-v2",
dimension=768,
max_input_tokens=384,
pooling="mean",
normalize=True,
repo="sentence-transformers/all-mpnet-base-v2",
onnx_file="onnx/model.onnx",
aliases=(
"huggingface_sentence-transformers-all-mpnet-base-v2",
"sentence-transformers/all-mpnet-base-v2",
"all-mpnet-base-v2",
),
)
#: Default for new installs: same 768 dimensions as mpnet, an 8x wider
#: effective input, and multilingual retrieval.
#:
#: Runs the int8-quantised graph, which is ~4x smaller than the fp32 one
#: (313 MB against 1247 MB) and keeps the image shippable now that both
#: defaults are baked in. It costs some numeric fidelity against fp32 --
#: measured at cosine 0.96 on short texts, with retrieval rank order
#: unchanged -- so it is not bit-comparable to a fp32 granite index.
GRANITE_311M = EmbeddingModel(
name="ibm-granite/granite-embedding-311m-multilingual-r2",
dimension=768,
max_input_tokens=32768,
pooling="cls",
normalize=True,
repo="ibm-granite/granite-embedding-311m-multilingual-r2",
onnx_file="onnx/model_quint8_avx2.onnx",
aliases=("granite-embedding-311m-multilingual-r2", "granite-311m"),
)
#: Smaller granite. Half the vector width, roughly three times the speed.
#: Int8-quantised on the same terms as GRANITE_311M above.
GRANITE_97M = EmbeddingModel(
name="ibm-granite/granite-embedding-97m-multilingual-r2",
dimension=384,
max_input_tokens=32768,
pooling="cls",
normalize=True,
repo="ibm-granite/granite-embedding-97m-multilingual-r2",
onnx_file="onnx/model_quint8_avx2.onnx",
aliases=("granite-embedding-97m-multilingual-r2", "granite-97m"),
)
OPENAI_ADA_002 = EmbeddingModel(
name="openai_text-embedding-ada-002",
dimension=1536,
max_input_tokens=8191,
pooling="mean",
normalize=True,
provider="openai",
repo=None,
onnx_file=None,
aliases=("text-embedding-ada-002",),
)
MODELS: Tuple[EmbeddingModel, ...] = (
MPNET,
GRANITE_311M,
GRANITE_97M,
OPENAI_ADA_002,
)
#: Fallback vector width when the configured model is unknown -- the width
#: every DocsGPT install has used to date, so an unrecognised model does not
#: silently reshape an existing table.
DEFAULT_EMBEDDING_DIMENSION = 768
#: Name that a fresh install should be configured with.
DEFAULT_NEW_INSTALL = GRANITE_311M.name
#: Name that ``settings.EMBEDDINGS_NAME`` defaults to, i.e. what an existing
#: deployment falls back to when it never pinned one.
DEFAULT_LEGACY = MPNET.name
def _index() -> Dict[str, EmbeddingModel]:
"""Build the lookup table of every accepted spelling."""
table: Dict[str, EmbeddingModel] = {}
for model in MODELS:
for key in (model.name, *model.aliases):
table[key.lower()] = model
return table
_LOOKUP = _index()
def resolve(name: Optional[str]) -> Optional[EmbeddingModel]:
"""Return the registry entry for ``name``, or ``None`` when unknown.
Args:
name: A configured ``EMBEDDINGS_NAME`` value, in any accepted spelling.
Returns:
The matching :class:`EmbeddingModel`, or ``None`` for a name the
registry does not describe -- which callers treat as a Hugging Face
repository rather than an error.
"""
if not name:
return None
return _LOOKUP.get(name.strip().lower())
def dimension_for(name: Optional[str]) -> Optional[int]:
"""Vector width for ``name``, or ``None`` when unknown."""
model = resolve(name)
return model.dimension if model else None
def max_input_tokens_for(name: Optional[str]) -> Optional[int]:
"""Context window for ``name``, or ``None`` when unknown."""
model = resolve(name)
return model.max_input_tokens if model else None
def known_names() -> Tuple[str, ...]:
"""Canonical names of every registered model, for error messages."""
return tuple(model.name for model in MODELS)
+37
View File
@@ -55,6 +55,10 @@ from application.storage.db.repositories.wiki_pages import (
from application.storage.db.session import db_readonly, db_session
from application.storage.db.source_config import SourceConfig
from application.storage.storage_creator import StorageCreator
from application.upload_limits import (
enforce_parseable_attachment,
UnsupportedUploadTypeError,
)
from application.utils import (
count_tokens_docs,
get_encoding,
@@ -1554,6 +1558,30 @@ class AttachmentRejectedError(Exception):
"""
def _reject_unparseable_attachment(
local_path: str, filename: str, parser_extensions
) -> None:
"""Reject an attachment with no parser whose contents are binary.
Defence in depth behind the route's gate, and stricter than it: this runs
against the parser table actually loaded, so a suffix the route trusted
(.webp with docling absent, say) is still refused rather than opened as
plain text by ``SimpleDirectoryReader`` and stored as garbage.
Args:
local_path: Filesystem path of the attachment about to be parsed.
filename: The upload's original filename, which carries the suffix.
parser_extensions: Suffixes the live ``file_extractor`` can parse.
Raises:
AttachmentRejectedError: If the file has no parser and is not text.
"""
try:
enforce_parseable_attachment(local_path, filename, parser_extensions)
except UnsupportedUploadTypeError as exc:
raise AttachmentRejectedError(str(exc)) from exc
def _reject_attachment_zip_bomb(local_path: str) -> None:
"""Reject a zip-container attachment that decompresses to too much.
@@ -1752,6 +1780,7 @@ def attachment_worker(self, file_info, user):
parser_name = type(_parser).__name__ if _parser is not None else "SimpleDirectoryReader"
def _parse_local_file(local_path: str, **kwargs) -> Document:
_reject_unparseable_attachment(local_path, filename, set(file_extractor))
_reject_attachment_zip_bomb(local_path)
parse_path, is_temp_copy = _bounded_attachment_copy(local_path)
try:
@@ -2524,6 +2553,14 @@ def reembed_wiki_page_worker(self, source_id, path, content_hash, user):
with db_session() as conn:
WikiPagesRepository(conn).set_embed_status(source_id, path, "embedded")
# These chunks were just embedded with the configured model, so the
# source now names it. Wiki sources created before this was recorded
# carry NULL, which the boot mismatch check reads as the legacy
# model and reports as stale; stamping here heals them on the next
# page edit.
SourcesRepository(conn).update(
source_id, user, {"model": settings.EMBEDDINGS_NAME}
)
except Exception:
with db_session() as conn:
WikiPagesRepository(conn).set_embed_status(source_id, path, "failed")