ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Studio: skip re-indexing an already-indexed document (content-hash dedup)

Browse source 
Re-uploading the same file into the same scope (KB or thread) used to
parse, chunk, caption and embed it all over again, creating a duplicate
set of chunks. Dedup by content hash instead:

  - schema: add rag_documents.content_hash (sha256 of the bytes) via the
    standard PRAGMA/ALTER migration, plus (scope, content_hash) indexes.
  - upload: _save_upload now streams the bytes through sha256 and returns
    the digest alongside path/name/size.
  - _start_ingestion: before inserting, look for a COMPLETED row in the
    same scope with the same hash. If found, delete the redundant upload
    from disk and return the existing document_id with already_indexed=
    true and an empty job_id — no ingestion job is started. Only
    'completed' rows dedup, so a failed/in-flight prior attempt can still
    retry. Scope-local: the same file in two KBs is indexed in each.
  - frontend: UploadResponse.already_indexed flows through the rag-store
    (skips job subscription) into both upload paths, which mark the chip
    ready immediately and toast '<file> is already indexed'.

Pre-existing rows have NULL content_hash and won't dedup until
re-uploaded once under the new path. Not build/UI-verified here (no bun
in this env); needs typecheck + browser check.
This commit is contained in:
Roland Tannous 2026-05-28 15:19:06 +04:00
commit 266342a64e
6 changed files with 131 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

54
studio/backend/routes/rag.py
View file

@ -6,6 +6,7 @@
from __future__ import annotations
import asyncio
import hashlib
import json
import os
import queue as queue_module
@ -121,6 +122,9 @@ class UploadResponse(BaseModel):
document_id: str
job_id: str
filename: str
# True when an identical file (same content hash) was already indexed
# in this scope, so no new ingestion job was started. job_id is "".
already_indexed: bool = False
class SearchRequest(BaseModel):
@ -248,7 +252,7 @@ def _document_or_404(document_id: str) -> Any:
return row
async def _save_upload(file: UploadFile) -> tuple[Path, str, int]:
async def _save_upload(file: UploadFile) -> tuple[Path, str, int, str]:
import anyio
filename = _sanitize_filename(file.filename or "document")
@ -264,6 +268,9 @@ async def _save_upload(file: UploadFile) -> tuple[Path, str, int]:
stored_path = upload_dir / stored_name
max_bytes = RAG_MAX_UPLOAD_MB * 1024 * 1024
written = 0
# Hash the bytes as they stream so we can dedup identical re-uploads
# within a scope without re-reading the file.
hasher = hashlib.sha256()
# Route writes through anyio worker thread so the event loop stays free.
# Outer try/except cleans up partial files after async-with closes the fd
# (Windows refuses unlink on an open fd).
@ -279,6 +286,7 @@ async def _save_upload(file: UploadFile) -> tuple[Path, str, int]:
status_code = 413,
detail = f"File exceeds {RAG_MAX_UPLOAD_MB} MB limit",
)
hasher.update(chunk)
await f.write(chunk)
except HTTPException:
stored_path.unlink(missing_ok = True)
@ -286,7 +294,7 @@ async def _save_upload(file: UploadFile) -> tuple[Path, str, int]:
if written == 0:
stored_path.unlink(missing_ok = True)
raise HTTPException(status_code = 400, detail = "Empty upload payload")
return stored_path, filename, written
return stored_path, filename, written, hasher.hexdigest()
def _start_ingestion(
@ -300,15 +308,46 @@ def _start_ingestion(
embedding_model: str,
chunking_strategy: str = "standard",
mode: str = "text",
content_hash: str | None = None,
) -> UploadResponse:
document_id = str(uuid4())
with get_connection() as conn:
# Dedup: if an identical file (same content hash) is already
# indexed in this scope, skip re-ingestion. Only a 'completed'
# row counts — a failed/in-flight prior attempt should be allowed
# to retry. Scope is the same kb_id or thread_id the upload
# targets (a file shared across two KBs is indexed in each).
if content_hash:
if kb_id is not None:
existing = conn.execute(
"SELECT id, filename FROM rag_documents "
"WHERE kb_id = ? AND content_hash = ? AND status = 'completed' "
"LIMIT 1",
(kb_id, content_hash),
).fetchone()
else:
existing = conn.execute(
"SELECT id, filename FROM rag_documents "
"WHERE thread_id = ? AND content_hash = ? AND status = 'completed' "
"LIMIT 1",
(thread_id, content_hash),
).fetchone()
if existing is not None:
# Drop the redundant upload we just wrote to disk; the
# already-indexed copy stays the source of truth.
_unlink_if_under_uploads(stored_path)
return UploadResponse(
document_id = existing["id"],
job_id = "",
filename = existing["filename"],
already_indexed = True,
)
conn.execute(
"""
INSERT INTO rag_documents
(id, kb_id, thread_id, filename, content_type, stored_path,
status, num_chunks, byte_size, created_at)
VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?)
status, num_chunks, byte_size, content_hash, created_at)
VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?, ?)
""",
(
document_id,
@ -318,6 +357,7 @@ def _start_ingestion(
content_type,
str(stored_path),
byte_size,
content_hash,
_now_ms(),
),
)
@ -806,7 +846,7 @@ async def upload_kb_document(
current_subject: str = Depends(get_current_subject),
) -> UploadResponse:
kb_row = _kb_or_404(kb_id)
stored_path, filename, byte_size = await _save_upload(file)
stored_path, filename, byte_size, content_hash = await _save_upload(file)
# Tolerate pre-Phase-3 rows missing chunking_strategy/mode.
kb_keys = kb_row.keys() if hasattr(kb_row, "keys") else ()
chunking_strategy = (
@ -823,6 +863,7 @@ async def upload_kb_document(
embedding_model = kb_row["embedding_model"],
chunking_strategy = chunking_strategy,
mode = mode,
content_hash = content_hash,
)
@ -835,7 +876,7 @@ async def upload_thread_document(
from utils.rag.config import resolve_embedder
# No chat_threads check — fresh threads aren't persisted until first run.
stored_path, filename, byte_size = await _save_upload(file)
stored_path, filename, byte_size, content_hash = await _save_upload(file)
settings = _load_thread_settings(thread_id)
embedder = settings.embedding_model or resolve_embedder(
settings.mode,
@ -851,6 +892,7 @@ async def upload_thread_document(
embedding_model = embedder,
chunking_strategy = settings.chunking_strategy,
mode = settings.mode,
content_hash = content_hash,
)

15
studio/backend/storage/studio_db.py
View file

@ -262,6 +262,21 @@ def _ensure_schema(conn: sqlite3.Connection) -> None:
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_rag_documents_thread_id ON rag_documents(thread_id)"
)
# content_hash: sha256 of the uploaded bytes, used to skip re-indexing a
# file that already exists in the same scope (kb_id / thread_id).
rag_documents_columns = {
row[1] for row in conn.execute("PRAGMA table_info(rag_documents)").fetchall()
}
if "content_hash" not in rag_documents_columns:
conn.execute("ALTER TABLE rag_documents ADD COLUMN content_hash TEXT")
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_rag_documents_kb_hash "
"ON rag_documents(kb_id, content_hash)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_rag_documents_thread_hash "
"ON rag_documents(thread_id, content_hash)"
)
# kind: text|image|caption. linked_chunk_id pairs image↔caption (both null for text).
conn.execute(
"""

23
studio/frontend/src/features/chat/hooks/use-thread-doc-uploads.ts
View file

@ -123,7 +123,28 @@ export function useThreadDocUploads(): UseThreadDocUploadsResult {
setChipScopeKeys((m) => ({ ...m, [localChipId]: scopeKey }));
const uploadDocument = useRagStore.getState().uploadDocument;
try {
const { documentId, jobId } = await uploadDocument(scope, file);
const { documentId, jobId, alreadyIndexed } = await uploadDocument(
scope,
file,
);
if (alreadyIndexed) {
// Identical file already in this scope — no re-index.
setPendingDocs((prev) =>
prev.map((d) =>
d.id === localChipId
? { ...d, status: "ready", documentId }
: d,
),
);
toast.info(`${file.name} is already indexed`);
if (
scope?.kind === "thread" &&
useChatRuntimeStore.getState().ragSource.kind === "off"
) {
useChatRuntimeStore.getState().setRagSource({ kind: "thread" });
}
return;
}
setPendingDocs((prev) =>
prev.map((d) =>
d.id === localChipId

22
studio/frontend/src/features/chat/shared-composer.tsx
View file

@ -612,7 +612,27 @@ export function SharedComposer({
}
const uploadDocument = useRagStore.getState().uploadDocument;
try {
const { documentId, jobId } = await uploadDocument(scope, file);
const { documentId, jobId, alreadyIndexed } = await uploadDocument(
scope,
file,
);
if (alreadyIndexed) {
setPendingDocs((prev) =>
prev.map((d) =>
d.id === localChipId
? { ...d, status: "ready", documentId }
: d,
),
);
toast.info(`${file.name} is already indexed`);
if (
scope?.kind === "thread" &&
useChatRuntimeStore.getState().ragSource.kind === "off"
) {
useChatRuntimeStore.getState().setRagSource({ kind: "thread" });
}
return;
}
setPendingDocs((prev) =>
prev.map((d) =>
d.id === localChipId

3
studio/frontend/src/features/rag/api/rag-api.ts
View file

@ -36,6 +36,9 @@ export interface UploadResponse {
document_id: string;
job_id: string;
filename: string;
/** True when an identical file was already indexed in this scope; no
* new ingestion job was started and job_id is "". */
already_indexed?: boolean;
}
export interface SearchHit {

31
studio/frontend/src/features/rag/stores/rag-store.ts
View file

@ -56,7 +56,11 @@ interface RagStoreState {
uploadDocument: (
scope: { kind: "kb"; kbId: string } | { kind: "thread"; threadId: string },
file: File,
) => Promise<{ documentId: string; jobId: string }>;
) => Promise<{
documentId: string;
jobId: string;
alreadyIndexed: boolean;
}>;
deleteDocument: (documentId: string, scopeKey: string) => Promise<void>;
loadThreadIndexes: () => Promise<void>;
@ -199,16 +203,25 @@ export const useRagStore = create<RagStoreState>((set, get) => ({
} else {
void get().loadThreadDocuments(scope.threadId);
}
get().subscribeJob(result.job_id, () => {
if (scope.kind === "kb") {
void get().loadKBDocuments(scope.kbId);
} else {
void get().loadThreadDocuments(scope.threadId);
}
});
return { documentId: result.document_id, jobId: result.job_id, scopeKey } as {
// Identical file already indexed in this scope: no job to track.
if (!result.already_indexed && result.job_id) {
get().subscribeJob(result.job_id, () => {
if (scope.kind === "kb") {
void get().loadKBDocuments(scope.kbId);
} else {
void get().loadThreadDocuments(scope.threadId);
}
});
}
return {
documentId: result.document_id,
jobId: result.job_id,
alreadyIndexed: result.already_indexed ?? false,
scopeKey,
} as {
documentId: string;
jobId: string;
alreadyIndexed: boolean;
};
},