diff --git a/rag/services/kb_service.py b/rag/services/kb_service.py index 0212d14..023ec28 100644 --- a/rag/services/kb_service.py +++ b/rag/services/kb_service.py @@ -6,6 +6,7 @@ - Поиск релевантной информации по запросу (семантический поиск в Qdrant с фильтром по room_jid). - Управление доступом к документам (личные / глобальные / комнатные). - Поддержка обновления версий документов в комнатах (атомарное обновление: сначала создаётся новая версия, затем удаляется старая). +- Добавлен метод reindex_failed_documents для повторной индексации документов с indexed=False. Добавлена транзакционность индексации через флаг indexed в PostgreSQL. ВСЕ ВЫЗОВЫ QDRANT ОБЁРНУТЫ В asyncio.to_thread ДЛЯ ПРЕДОТВРАЩЕНИЯ БЛОКИРОВКИ EVENT LOOP. @@ -226,7 +227,6 @@ class KBService: dense_vectors = await self.embedding.embed_batch(chunks) if dense_vectors is None or len(dense_vectors) != len(chunks): logger.error(f"Не удалось получить dense-эмбеддинги для документа {new_doc_id}, оставляем indexed=False") - # Можно удалить новую запись, но оставим для диагностики return new_doc_id, 0 sparse_vectors = await self.embedding.embed_sparse_batch(chunks) @@ -289,6 +289,61 @@ class KBService: return new_doc_id, chunk_count + # ------------------------------------------------------------------ + # Повторная индексация (для reindex_worker) + # ------------------------------------------------------------------ + + async def reindex_failed_documents(self, batch_size: int = 10) -> None: + """ + Переиндексирует документы с indexed=False. + Использует метаданные для восстановления текста. + """ + import datetime + cutoff = datetime.datetime.now() - datetime.timedelta(days=7) + async with self.db.pool.acquire() as conn: + rows = await conn.fetch( + """ + SELECT id, source_name, owner_jid, room_jid, is_global, metadata, file_hash + FROM documents + WHERE indexed = FALSE + AND upload_time > $1 + LIMIT $2 + """, + cutoff, + batch_size + ) + if not rows: + return + + logger.info(f"Найдено {len(rows)} документов для переиндексации") + + for row in rows: + doc_id = row['id'] + metadata = row.get('metadata') or {} + file_text = metadata.get('file_text') + if not file_text: + logger.error(f"Документ {doc_id} не содержит текста для переиндексации, пропускаем") + continue + + try: + doc_id_new, chunk_count = await self.add_document( + file_name=row['source_name'], + file_text=file_text, + user_jid=row['owner_jid'], + is_global=row['is_global'], + title=row['source_name'], + metadata=metadata, + room_jid=row['room_jid'], + file_hash=row['file_hash'], + update_if_exists=True + ) + if chunk_count > 0: + logger.info(f"Документ {doc_id} успешно переиндексирован (новый doc_id={doc_id_new})") + else: + logger.error(f"Не удалось переиндексировать документ {doc_id}, chunk_count=0") + except Exception as e: + logger.exception(f"Ошибка переиндексации документа {doc_id}: {e}") + # ------------------------------------------------------------------ # Получение списков документов (используют методы db с фильтром indexed=TRUE) # ------------------------------------------------------------------