Конвейер индексации документов (Document Ingestion Pipeline)

Документы попадают в систему через загрузку файла, импорт по URL или ручное создание. Сервер сохраняет содержимое и ставит задачу обработки, которая последовательно выполняет разбор, разбиение на блоки и индексацию, а затем в соответствии с конфигурацией базы знаний генерирует сводку, вопросы, граф и Wiki. Состояние и прогресс обновляются по этапам обработки; неуспешные задачи восстанавливаются механизмами повторных попыток и инспекции.

Общая архитектура

Цепочка индексации aihelpme.ru — это распределённый асинхронный конвейер на базе Asynq (Redis). HTTP Handler только пишет в БД и ставит задачу в очередь; вся длительная работа (разбор, векторизация, обогащение через LLM) выполняется независимыми пулами Worker, потребляющими очереди.

Слой входа: три способа создания

Маршруты регистрируются в internal/router/routes_knowledge.go:

kb.POST("/file",   g.OwnedKBOrAdmin(), g.KBAccessWrite("id"), handler.CreateKnowledgeFromFile)
kb.POST("/url",    g.OwnedKBOrAdmin(), g.KBAccessWrite("id"), handler.CreateKnowledgeFromURL)
kb.POST("/manual", g.OwnedKBOrAdmin(), g.KBAccessWrite("id"), handler.CreateManualKnowledge)

Сопутствующие управляющие эндпоинты: POST /knowledge/:id/reparse (повторный разбор), POST /knowledge/:id/cancel-parse (отмена разбора), POST /knowledge/batch-reparse, POST /knowledge/batch-delete, POST /knowledge/move (перемещение между базами).

Загрузка файла (CreateKnowledgeFromFile)

Параметры формы: file, fileName, metadata, enable_multimodel, tag_ids, process_config (каждая загрузка может переопределить конфигурацию обработки уровня KB, см. Конфигурация обработки: значения KB по умолчанию + переопределение при отдельной загрузке).
Процесс: проверка расширения → дедупликация по MD5 → сохранение через FileService.SaveFile → создание записи Knowledge → постановка в очередь.

Единый шлюз по расширениям

supportedImportFileExtensions в internal/application/service/knowledge_util.go — единственный источник истины для всех путей импорта: прямая загрузка, скачивание файла по URL и повторная проверка после скачивания worker'ом обращаются к одной и той же таблице:

pdf txt docx doc epub html htm mhtml md markdown xmind
png jpg jpeg gif csv xlsx xls pptx ppt json
mp3 wav m4a flac ogg

Ранее импорт по URL поддерживал отдельный укороченный белый список, что приводило к расхождениям вроде «прямая загрузка xlsx проходит, а импорт xlsx по URL отклоняется» (#2447); теперь всё определяется единообразно через isSupportedImportExtension() / validateImportFileType(), а для видео выдаётся явное сообщение «загрузка видеофайлов пока не поддерживается».

Табличные расширения (csv / xlsx / xls, dataTableFileExtensions) после задачи обработки документа дополнительно вешают задачу построения сводки по таблице (enqueueDataTableSummaryIfNeeded).

Дополнительные предварительные проверки для изображений и аудио (полнота конфигурации объектного хранилища, настроены ли модели VLM / ASR) вместе с проверкой process_config сходятся в resolveFileImportProcessConfig(), которая общая для загрузки и импорта по URL.

Импорт по URL (CreateKnowledgeFromURL)

JSON Body: {url, file_name?, file_type?, enable_multimodel?, title?, tag_ids?, channel?, process_config?}.
isFileURL() по указанному выше единому набору расширений определяет, это «скачивание файла» или «загрузка веб-страницы».
Handler и Service на двух уровнях выполняют защиту от SSRF (internal/handler/knowledge.go и knowledge_create.go вызывают оба):

if err := secutils.ValidateURLForSSRF(req.URL); err != nil {
    c.Error(errors.NewBadRequestError(secutils.FormatSSRFError("URL", req.URL, err)))
    return
}

На стороне worker перед фактической загрузкой проверка выполняется ещё раз (convert() в knowledge_process.go) — тройной рубеж против TOCTOU.

Ручное создание (CreateManualKnowledge)

JSON Body — types.ManualKnowledgePayload{Title, Content, Status, TagIDs, Channel, ProcessConfig}, поддерживается черновиковое состояние (Draft); при публикации triggerManualProcessing() заводит в тот же конвейер разбиения/индексации, что и файл (этап DocReader пропускается).

Механизм дедупликации

knowledge_create.go вычисляет MD5 загруженного файла и ищет в БД по четвёрке:

hash, err := calculateFileHash(file) // MD5
exists, existingKnowledge, err := s.repo.CheckKnowledgeExists(ctx, tenantID, kbID,
    &types.KnowledgeCheckParams{
        Type:     "file",
        FileName: fileName,
        FileType: getFileType(fileName),
        FileSize: file.Size,
        FileHash: hash,
    })
if exists {
    return existingKnowledge, types.NewDuplicateFileError(existingKnowledge)
}

При совпадении документ повторно не индексируется, возвращается существующий Knowledge с DuplicateFileError (фронтенд по нему показывает «файл уже существует»). Строки с parse_status равным failed или deleting в проверке дубликатов не участвуют, поэтому файл, удаление которого ещё не завершилось или застряло, можно загрузить снова. FileType участвует в определении: дубликат считается только в пределах одного типа файла, поэтому полностью идентичные по содержимому notes.md и notes.txt сосуществуют как два независимых знания (CheckKnowledgeExists добавляет условие LOWER(file_type) в обе ветки — и по хэшу, и по «имя файла + размер»).

Начальное состояние

Ключевые начальные поля новой записи Knowledge (knowledge_create.go):

knowledge := &types.Knowledge{
    ID:           uuid.New().String(),
    Type:         "file",        // или "url" / "manual"
    ParseStatus:  "pending",     // начальное состояние разбора
    EnableStatus: "disabled",    // недоступно для поиска до завершения индексации
    FileHash:     hash,
    ...
}

Для знаний табличного типа CSV/Excel после создания дополнительно ставится задача TypeDataTableSummary (datatable:summary), порождающая Chunk типа table_summary / table_column для вопросов по таблицам.

Слой файлового хранилища (FileService и бэкенды хранилища)

Определение интерфейса

internal/types/interfaces/file.go:

type FileService interface {
    CheckConnectivity(ctx context.Context) error
    SaveFile(ctx context.Context, file *multipart.FileHeader, tenantID uint64, knowledgeID string) (string, error)
    SaveBytes(ctx context.Context, data []byte, tenantID uint64, fileName string, temp bool) (string, error)
    GetFile(ctx context.Context, filePath string) (io.ReadCloser, error)
    GetFileURL(ctx context.Context, filePath string) (string, error)
    DeleteFile(ctx context.Context, filePath string) error
    CopyFile(ctx context.Context, srcPath string, tenantID uint64, knowledgeID string) (string, error)
}

Поддерживаемые бэкенды хранилища

Фабричная функция NewFileServiceFromStorageConfig() (internal/application/service/file/factory.go) выбирает бэкенд по types.StorageEngineConfig.DefaultProvider. Фактический список поддерживаемых бэкендов:

Provider	Префикс пути	Файл реализации	Описание	Ключевая конфигурация
`local`	`local://`	`file/local.go`	Локальный диск одной машины	`LocalEngineConfig.PathPrefix`, базовый каталог берётся из `LOCAL_STORAGE_BASE_DIR`, подпись внешних ссылок — из `APP_EXTERNAL_URL`
`minio`	`minio://`	`file/minio.go`	MinIO / S3-совместимое	`MinIOEngineConfig` (при `mode: docker` читаются переменные `MINIO_ENDPOINT` / `MINIO_ACCESS_KEY_ID` / `MINIO_SECRET_ACCESS_KEY` / `MINIO_BUCKET_NAME`; при `mode: remote` — поля конфигурации)
`cos`	`cos://`	`file/cos.go`	COS	`SecretID/SecretKey/Region/BucketName/AppID`, поддерживается отдельная временная корзина `TempBucketName/TempRegion`
`oss`	`oss://`	`file/oss.go`	OSS	`Endpoint/Region/AccessKey/SecretKey/BucketName`, поддерживается временная корзина
`s3`	`s3://`	`file/s3.go`	AWS S3 / совместимый протокол	`Endpoint/Region/AccessKey/SecretKey/BucketName/UseSSL/ForcePathStyle`
`tos`	`tos://`	`file/tos.go`	TOS	То же, поддерживается временная корзина
`obs`	`obs://`	`file/obs.go`	OBS	`Endpoint/Region/AccessKey/SecretKey/BucketName/UseSSL`
`ks3`	`ks3://`	`file/ks3.go`	KS3	`Endpoint/Region/AccessKey/SecretKey/BucketName`
`dummy`	`dummy://`	`file/dummy.go`	Пустая реализация для тестов	нет

Правила организации ключей объектов

Штатные файлы: {tenantID}/{knowledgeID}/{uuid или метка времени в наносекундах}{ext}, например local://12345/kb-001/1722045600000000000.pdf.
Продукты экспорта/временные/клонированные: {tenantID}/exports/{fileName}_{timestamp}{ext}.
Безопасность путей: secutils.SafePathUnderBase (защита от обхода каталогов), secutils.SafeFileName, на стороне объектного хранилища — utils.SafeObjectKey.

Два слоя-обёртки

backend_scoped.go: при развёртывании с несколькими бэкендами хранилища добавляет к пути префикс экземпляра вида storage://{backendID}/{innerPath}, кодирует/декодирует через wrap/unwrap и запрещает операции между бэкендами. KB может привязываться к конкретному экземпляру бэкенда через StorageBackendID.
resource_catalog.go: регистрирует физический путь как стабильную ссылку resource://{uuid}, поддерживает Bind (связывание ресурса с владельцем вроде knowledge), MarkDeleted, CreateAccessGrant (генерация временного токена доступа с URL вида /r/{token}). Уровень приложения, держа ссылку resource://, может незаметно переносить нижележащее хранилище.

Механизм асинхронных задач (Asynq + Redis)

Постановка в очередь

knowledge_create.go собирает types.DocumentProcessPayload (содержит TenantID/KnowledgeID/KnowledgeBaseID/FilePath/FileName/FileType/EnableMultimodel/EnableQuestionGeneration/QuestionCount/Language/Attempt и др.), параметры задачи берутся из knowledge_task_options.go:

opts := []asynq.Option{
    asynq.Queue(types.QueueDefault),
    asynq.Timeout(config.DocumentProcessTimeout(cfg)), // по умолчанию 2 часа
    asynq.MaxRetry(3),                                  // при сбое максимум 3 повторных попытки
}
task := asynq.NewTask(types.TypeDocumentProcess, payloadBytes, opts...)
info, err := s.task.Enqueue(task)

При сбое постановки в очередь ParseStatus устанавливается в failed (файл уже сохранён, повторный запуск возможен через reparse).

Топология очередей и пулы Worker

Очереди, определённые в internal/types/task.go:

Константа очереди	Имя	Назначение
`QueueDefault`	`default`	Основная обработка документов (разбор/разбиение/эмбеддинг/индексация)
`QueueChatAttachment`	`chat_attachment`	Разбор временных документов сессии
`QueuePostProcess`	`postprocess`	Задачи оркестрации постобработки
`QueueSummary`	`summary`	Сводки, сводки по таблицам, автотеги, описание базы знаний
`QueueMultimodal`	`multimodal`	OCR изображений / VLM Caption
`QueueQuestion`	`question`	Генерация вопросов
`QueueGraph`	`graph`	Извлечение графа
`QueueWiki`	`wiki`	Генерация и завершение Wiki
`QueueMaintenance`	`low`	Обслуживающие задачи (массовый импорт FAQ, копирование/удаление баз знаний, массовое удаление/повторный разбор, перемещение и др.)

Есть также очереди sync (синхронизация источников данных) и memory (извлечение личной памяти). Конкурентность по умолчанию (internal/types/task.go): основной пул DefaultCoreWorkerConcurrency = 8, пул постобработки 2, пул обогащения 12, пул обслуживания 4, эластичный общий пул 6, пул Wiki 8; полная топология — в Система асинхронных задач.

Семантика повторных попыток при сбое

TypeDocumentProcess: MaxRetry(3) → исходная попытка + 3 повторные, всего 4 попытки; каждая попытка ограничена DocumentProcessTimeout (по умолчанию 2 часа, переменная окружения WEKNORA_DOCUMENT_PROCESS_TIMEOUT), причём отдельный вызов docreader дополнительно ограничен WEKNORA_DOCREADER_CALL_TIMEOUT (по умолчанию 30 минут).
panic внутри обработчика преобразуется в ошибку и передаётся в механизм повторных попыток и dead letter; при последнем сбое документ помечается failed; исполнитель режима Lite тоже перехватывает panic и не приводит к завершению процесса.
Payload несёт Attempt (при повторном разборе берётся исторический максимум attempt+1); Span Tracker изолирует по attempt дерево прогресса каждого раунда обработки, новый attempt «заменяет» (supersede) завершающие действия старой задачи.
Обработчик различает «последняя ли это попытка asynq» (isLastRetry): не последний сбой просто возвращает ошибку, чтобы asynq повторил, а на последней попытке ParseStatus записывается как failed и пишется ErrorMessage. При чтении тенанта, чтении строки знания или записи состояния processing транзиентная ошибка БД тоже возвращает ошибку для повтора, а не сразу подтверждает задачу; тихо завершается только когда строки знания действительно нет.
Повторный разбор: новый attempt назначается только после прохождения проверки переопределённой конфигурации и очистки старых ресурсов, отклонённый повторный разбор не влияет на предыдущий идущий раунд. Если предыдущий раунд ещё идёт (pending / processing / finalizing), сначала отменяются его задачи в очереди и работающим задачам сообщается об остановке. Задачи старого attempt всегда пропускаются: ProcessDocument, запись блоков в БД, постобработка, мультимодальные задачи по изображениям больше не пишут строки, не удаляют chunk и не уменьшают счётчики; операции Wiki несут attempt, взятый при постановке в очередь, и по его истечении больше не освобождают счётчики нового раунда; коллбэк dead letter тоже не помечает новый раунд как failed.

Конфигурация обработки: значения KB по умолчанию + переопределение при отдельной загрузке

ResolveProcessConfig(kb, overrides) из knowledge_process_config.go объединяет конфигурацию KB по умолчанию и process_config, переданный при загрузке (types.KnowledgeProcessOverrides), в types.EffectiveProcessConfig:

Переопределяемые пункты: ChunkingConfig (размер/перекрытие/стратегия блоков/родительско-дочерние блоки и т. д.), EnableMultimodel, VLMConfig, ASRConfig, QuestionGenerationConfig, GraphEnabled, ExtractConfig, ParserEngineRules.
Ограничение: eff.GraphEnabled = eff.GraphEnabled && eff.ExtractConfig.Enabled (граф зависит от включённой конфигурации извлечения).
ValidateProcessOverrides выполняет предварительную проверку по типу файла: для загрузки изображения обязательна модель VLM, для аудио — модель ASR, мультимодальность дополнительно требует полной конфигурации объектного хранилища (validateImageMultimodalConfig).
Переопределённая конфигурация сохраняется через knowledge.SetProcessOverrides в строке Knowledge и используется при reparse.

Какие конвейеры выполняются, определяется IndexingStrategy KB (internal/types/indexing_strategy.go):

type IndexingStrategy struct {
    VectorEnabled  bool // семантический векторный индекс
    KeywordEnabled bool // индекс ключевых слов BM25
    WikiEnabled    bool // автоматическая генерация страниц Wiki
    GraphEnabled   bool // извлечение графа знаний
}

NeedsEmbedding() = Vector || Keyword, NeedsChunks() = включён хотя бы один. По умолчанию включены vector+keyword.

Основной конвейер обработки (knowledge_process.go)

После потребления TypeDocumentProcess worker продвигается по пяти нормализованным этапам, каждый этап соответствует одному Span (см. Самовосстановление Housekeeping (knowledge_housekeeping.go)):

docreader → chunking → embedding → multimodal → postprocess

Разбор (convert, Stage: docreader)

beginStage(StageDocReader) записывает вход (file_name/file_type/is_url).
В режиме URL повторно ValidateURLForSSRF, при сбое сразу failStage + ParseStatus=failed.
Выбор движка: eff.ChunkingConfig.ResolveParserEngine(fileType) (для URL используется виртуальный тип "url"), маршрутизация по настроенным в KB ParserEngineRules (тип файла → движок); MergeParserEngineOverrides объединяет переопределения параметров движка уровня тенанта и уровня загрузки.
resolveDocReader возвращает interfaces.DocReader:

builtin: вызов Python-сервиса docreader через gRPC (docparser/grpc_parser.go) или HTTP (http_parser.go);
simple: нативный разбор на Go md/txt/csv/json/изображений/аудио (builtin_converter.go, CSV→таблица Markdown, JSON→рекурсивно разбиваемые блоки кода, изображения/аудио→ссылки-заглушки);
anydoc: разбор docx/doc/pptx/ppt/xlsx/xls/odf/rtf/epub/csv/pdf внутри процесса Go (anydoc_reader.go), в основе — Rust-библиотека anydoc, линкуемая через cgo. Встроенные изображения офисных документов вставляются обратно в Markdown на место по модели документа; сканированный PDF без текстового слоя при доступности DocReader откатывается на полностраничный рендеринг builtin. При линковке anydoc сложные форматы без настроенных правил по умолчанию идут через anydoc, но PDF по умолчанию всё равно через builtin. Доступно только в бинарниках со сборочным тегом anydoc, в остальных сборках этот движок отображается в списке движков как недоступный;
weknoracloud / mineru / mineru_cloud / paddleocr_vl / paddleocr_vl_cloud: HTTP-конвертеры (регистрируются в engines.go, доступность определяется по конфигурации mineru_endpoint, mineru_api_key, paddleocr_vl_endpoint и др.; самостоятельный mineru автоматически распознаёт V1 API MinerU 4.0 и старый /file_parse, см. Сервис разбора документов).

Каталог движков сосредоточен в internal/infrastructure/docparser/engines.go: каждый движок одновременно объявляет метаданные (имя, описание, типы файлов, пробу доступности) и фабрику NewReader, docparser.NewReader диспетчеризует по имени, а незарегистрированные имена (например markitdown, существующий только в docreader) попадают в клиент docreader. 5. Режим файла: байты читаются обратно через FileService.GetFile(payload.FilePath) и кладутся в ReadRequest.FileContent.

Сторона сервиса docreader (docreader/, Python gRPC): proto определён в docreader/proto/docreader.proto, методы сервиса Read / ReadStream (потоково: первый кадр — meta, затем по одному кадру на изображение, чтобы большие сканированные PDF не упирались в лимит сообщений gRPC) / ListEngines. Встроенные парсеры покрывают docx/doc/pdf/md/xlsx/xls/pptx/ppt/epub/html/htm/mhtml/xmind/изображения/веб-страницы (за URL отвечает WebParser), дополнительно зарегистрированы движки markitdown (Microsoft MarkItDown) и opendataloader (анализ вёрстки PDF, требует Java 11+); список движков ведётся на стороне Go в engine_registry.go, Go App больше не вызывает ListEngines. Возвращается единый ReadResult{MarkdownContent, ImageRefs, Metadata, IsAudio, AudioData} — продукт разбора всегда единообразен: текст Markdown + байты изображений, а сохранение изображений — обязанность стороны Go.

Транскрипция ASR (аудиофайлы)

Когда convertResult.IsAudio истинно (аудиофайл разбирается в заглушку + исходные байты):

asrModel, err := s.modelService.GetASRModel(ctx, eff.ASRConfig.ModelID)
transcriptionResult, err := asrModel.Transcribe(ctx, convertResult.AudioData, knowledge.FileName)

Текст транскрипции заменяет MarkdownContent, после чего обработка идёт по обычному текстовому конвейеру; при не настроенном ASR происходит немедленный сбой.

Извлечение и загрузка изображений

ImageResolver.ResolveAndStore из docparser/image_resolver.go:

Последовательно обрабатывает изображения с обёрткой <!link>, URI data:, встроенный base64 в HTML, «голый» base64, встроенные байты из возвращённых docreader ImageRefs;
отфильтровывает маленькие изображения уровня иконок (ширина/высота < 64px или < 512 байт, кроме исходно загруженных файлов с IsOriginal=true);
SaveBytes загружает в бэкенд хранилища текущей KB, savedRefs кэширует для дедупликации;
перезаписывает ссылки в Markdown на URL хранилища (markdown_image_scanner.go точно находит позиции ![alt](target)).

Затем ResolveRemoteImages дополнительно скачивает и переносит внешние изображения http(s) из Markdown (также под защитой SSRF). На выходе — storedImages []docparser.StoredImage для этапа мультимодальности.

Разбиение на блоки (Stage: chunking)

Разбиение выполняется на стороне Go (internal/infrastructure/chunker, подробнее в главе «Механизм разбиения на блоки»):

chunkCfg := buildSplitterConfigFromChunking(eff.ChunkingConfig)
if eff.ChunkingConfig.EnableParentChild {
    parentCfg, childCfg := buildParentChildConfigs(eff.ChunkingConfig, chunkCfg)
    pcResult := chunker.SplitParentChild(convertResult.MarkdownContent, parentCfg, childCfg)
    // children → types.ParsedChunk (с ParentIndex); parents → ParsedParentChunk
} else {
    splitChunks := chunker.Split(convertResult.MarkdownContent, chunkCfg)
}

Запись в БД и индексация (processChunks, Stage: chunking + embedding)

processChunks — ключевая функция сборки:

Родительские блоки (режим родительско-дочерних блоков): для каждого parent создаётся запись ChunkTypeParentText, связывается списком PreChunkID/NextChunkID; родительские блоки только попадают в БД, но не в векторный индекс (после попадания в дочерний блок содержимое родителя поднимается при поиске).
Текстовые блоки: для каждого ParsedChunk создаётся запись ChunkTypeText с StartAt/EndAt (смещение в rune исходного текста, пригодно для восстановления/подсветки) и ContextHeader (хлебные крошки заголовка, хранятся в chunks.context_header, в ответе API не возвращаются); в родительско-дочернем режиме пишется ParentChunkID. Перед разбиением встроенные HTML-таблицы от всех движков разбора единообразно преобразуются в таблицы Markdown (NormalizeHTMLTables).
chunkService.CreateChunks(ctx, insertChunks) пакетно пишет в БД; при сбое ParseStatus=failed + failStage(StageChunking).
Векторизация и индексация (при kb.NeedsEmbeddingModel()). Если модель Embedding или векторное хранилище, привязанное к базе знаний, не удаётся разрешить (модель удалена, учётные данные недействительны и т. п.), текущая попытка сразу завершается сбоем и причина пишется в error_message, не оставаясь в processing:

indexContent := titlePrefix + chunk.EmbeddingContent() // заголовок + хлебные крошки + содержимое
indexInfoList = append(indexInfoList, &types.IndexInfo{
    Content: indexContent, SourceID: chunk.ID, SourceType: types.ChunkSourceType,
    ChunkID: chunk.ID, KnowledgeID: knowledge.ID, KnowledgeBaseID: ..., IsEnabled: true,
})
err = retrieveEngine.BatchIndex(ctx, embeddingModel, indexInfoList)

При сбое индексации выполняется компенсирующий откат: удаление уже записанных chunks (DeleteChunksByKnowledgeID) и очистка векторного индекса (DeleteByKnowledgeIDList), установка failed, чтобы не оставалось незавершённых продуктов. 5. Веерная рассылка мультимодальных задач по изображениям: при enableMultimodel && len(storedImages) > 0 enqueueImageMultimodalTasks ставит по одной задаче TypeImageMultimodal (QueueMultimodal) на каждое изображение, payload содержит ImageURL/EnableOCR/EnableCaption/Attempt/ImageIndex. При сбое постановки отдельного изображения немедленно освобождается соответствующий счётчик; если не удалось поставить все, сразу переходят к постобработке, чтобы документ не застрял в processing. 6. finalizeIndexedKnowledgeState: если ещё должны выполняться мультимодальность/постобработка, состояние остаётся processing, иначе сразу completed; одновременно выставляется EnableStatus="enabled" (с этого момента документ доступен для поиска) и накапливается использованный объём хранилища тенанта.

Оркестрация постобработки (knowledge_post_process.go, Stage: postprocess)

После завершения всей мультимодальности (или при её отсутствии) в очередь ставится TypeKnowledgePostProcess. Эта задача — оркестратор подзадач обогащения, использующий атомарные счётчики для сходимости к конечному состоянию:

willSpawnSummary  := eff.SummaryEnabled && len(textChunks) > 0   // при отдельной загрузке можно отключить сводку
willSpawnQuestion := len(textChunks) > 0 && kb.NeedsEmbeddingModel() && eff.QuestionGenerationConfig.Enabled
willSpawnWiki     := kb.IndexingStrategy.WikiEnabled && len(textChunks) > 0
graphChunks       := selectGraphChunks(textChunks)                // см. ниже
// questionGenChunkBatchSize = 20: генерация вопросов пачками по 20 chunk (пропускаются chunk только со ссылками на изображения)
expectedSubtasks = summary(0/1) + questionBatchCount + wiki(0/1) + len(graphChunks)(при включённом графе)

// атомарно повышает parse_status из processing в finalizing и записывает pending_subtasks_count
promoted, err := s.knowledgeRepo.SetFinalizing(ctx, payload.KnowledgeID, expectedSubtasks)

expectedSubtasks == 0 идёт по быстрому пути сразу к completed.
Вход извлечения графа (selectGraphChunks): текстовые блоки с основным содержимым; текстовые блоки только со ссылками на изображения (например, страницы сканированных PDF) используют свои дочерние блоки image_ocr; дочерние блоки image_caption не участвуют, чтобы не дублировать OCR.
При сбое записи состояния processing → finalizing задача возвращает ошибку, чтобы asynq повторил, и не рапортует ложно об успехе, оставляя документ в processing.
Каждая подзадача при выходе в конечное состояние вызывает FinalizeSubtask, атомарно уменьшающую pending_subtasks_count; при достижении нуля автоматически происходит переход в completed.
Согласование недостачи: если фактически поставлено в очередь меньше, чем запланировано (например, не удалось поставить в какую-то очередь), сразу компенсируется разница уменьшением счётчика, чтобы не застрять навсегда в finalizing.
finalizeSubtaskDetached (knowledge.go): уменьшение выполняется вне контекста через context.WithoutCancel + 10-секундный таймаут, чтобы при корректном завершении worker отмена ctx не привела к потере счётчика и вечному зависанию знания в finalizing.

Четыре вида подзадач обогащения:

Задача	Очередь	Гранулярность	Описание
`TypeSummaryGeneration`	`summary`	1 на знание	Генерация сводки документа, независимый конечный автомат `summary_status`; при загрузке с `process_config.summary_enabled=false` не генерируется
`TypeQuestionGeneration`	очередь question	пачка на 20 chunk	Генерация поисковых вопросов для chunk
`TypeChunkExtract`	очередь graph	1 на chunk	Извлечение сущностей/связей и запись в движок графа
`TypeWikiIngest`	очередь wiki	пакетно с дебаунсом	Генерация/обновление страниц Wiki

Автоматические теги документа

Постобработка после разбора по auto_tag_config базы знаний решает, ставить ли в очередь knowledge:auto_tag (очередь summary). Обработчик выбирает из уже существующих тегов текущей KB, модель по умолчанию — summary_model_id; max_tags по умолчанию 3, верхний предел 10, документы с уже имеющимися тегами по умолчанию пропускаются. Он только инкрементально связывает теги, не создаёт новых категорий и не удаляет ручные теги; сбой не блокирует индексацию документа. Область действия конфигурации и нового разбора/повторного разбора см. в Управление базами знаний.

Обновление описания базы знаний (knowledgebase_profile.go)

При каждом выходе задачи сводки в конечное состояние, завершении планирования постобработки, удалении документа, завершении межбазового перемещения вызывается requestKnowledgeBaseProfileRefresh, ставящая в очередь одну задачу kb:profile (очередь summary). ID задачи бакетируется по базе знаний и 30-секундному окну, пакет загрузок или удалений вызывает агрегацию один раз; обработчик пересчитывает агрегат портрета документов и не вызывает модель, если хэш не изменился. В очередь попадают только документальные базы знаний с profile_config.enabled, ручной запуск (POST /knowledge-bases/:id/profile/generate) этим ограничением не связан.

Обновление сводки (knowledge_summary_refresh.go)

Помимо первой индексации, редактирование содержимого блока, включение/выключение блока и изменение пользовательских метаданных делают существующую сводку устаревшей; тогда в очередь ставится одно обновление сводки (его можно запустить вручную через POST /knowledge/:id/regenerate-summary):

В начале задачи записывается снимок входа: content_revision / is_enabled каждого исходного блока, а также версия custom_metadata;
После генерации снимок перепроверяется через summarySourceChanged(). Если за это время снова редактировали, возвращается ErrSummaryRefreshStale, этот результат отбрасывается и summary_status не меняется, а завершает более свежее обновление — иначе старая сводка перезаписала бы новую;
Сбой чтения из БД и «вход изменился» обрабатываются по-разному, чтобы транзиентная ошибка чтения не отбрасывалась тихо как устаревшая задача;
Задачу обновления выполняет Asynq worker без контекста тенанта, внедрённого HTTP middleware, поэтому restoreSummaryRefreshTenantInfo() восстанавливает полную конфигурацию тенанта — она нужна фабрике поисковых движков.

Мультимодальность изображений (image_multimodal.go)

ImageMultimodalService.Handle потребляет задачу по одному изображению:

readImageBytes берёт изображение из хранилища/по URL; resolveVLM берёт конфигурацию VLM базы знаний;
Генерируется Caption (VLM, prompt собирается buildVLMCaptionPrompt по DescriptionLanguage/CustomInstructions) и текст OCR;
Результат пишется обратно в ImageInfo (JSON) соответствующего текстового Chunk и создаются/обновляются два дочерних Chunk: ChunkTypeImageCaption и ChunkTypeImageOCR с ParentChunkID, указывающим на текстовый блок, после чего отдельно вызывается indexChunks для записи в векторный индекс — чтобы «поиск по описанию изображения тоже возвращал исходный блок»;

Вектор изображения: если в базе знаний включены векторы изображений (ImageProcessingConfig.ImageVectorEnabled, по умолчанию выключено, см. KnowledgeBase.IsImageVectorEnabled) и модель умеет обрабатывать изображения (возможность модели input содержит image, а протокол провайдера умеет передавать изображения, см. embedding.AsImageEmbedder), indexImageVector дополнительно кодирует само изображение этой моделью и сохраняет его как третий дочерний Chunk ChunkTypeImageVector: в строке индекса SourceType равно ImageSourceType, вектор — от самого изображения (записывается заранее вычисленным через CompositeRetrieveEngine.BatchIndexVectors), а Content переиспользует описание изображения (или при его отсутствии — текст OCR) для использования в переранжировании и контексте ответа. Сначала изображение проходит imageprep.Prepare, где формат и размер приводятся к ограничениям провайдера; для страниц сканированных PDF это не делается (OCR там точнее). Сбой на этом шаге только пишется в журнал и не влияет на задачу;
У ChunkTypeImageVector вектор нельзя восстановить из текста: syncChunkIndex синхронизирует для него только состояние включения, а updateChunkVector пропускает его; при изменении описания изображения обновляется только его Content;

shouldDropOrphanedMultimodal проверяет, не удалён ли/не заменён ли родительский блок, и осиротевшие задачи отбрасываются;
checkAndFinalizeAllImages: после обработки всех изображений enqueueKnowledgePostProcessTask запускает оркестрацию постобработки из Оркестрация постобработки (knowledge_post_process.go, Stage: postprocess). Постановка задачи постобработки повторяется на месте 3 раза, и если всё равно не удаётся, текущая задача возвращает ошибку для повтора, не подтверждая задачу и не оставляя документ в processing. На последней попытке даже при сбое чтения строки знания это изображение учитывается как завершённое. В режиме Lite Redis нет, счётчики хранятся в памяти процесса, и постобработка запускается только после завершения всех изображений.

Конечный автомат

Основное состояние Knowledge (ParseStatus)

Полный набор значений, определённый в internal/types/knowledge.go:

Значение	Значение
`pending`	Создан, ожидает, пока worker заберёт задачу
`processing`	Выполняются разбор/разбиение/эмбеддинг/мультимодальность
`finalizing`	Основной процесс завершён, ожидаются подзадачи обогащения (`pending_subtasks_count > 0`)
`completed`	Всё завершено
`failed`	Обработка неуспешна (причина в `ErrorMessage`)
`deleting`	Удаляется (маркер против параллелизма; не учитывается в числе документов базы знаний и не участвует в определении дубликатов файлов)
`cancelled`	Пользователь отменил разбор

Вспомогательные состояния: EnableStatus ∈ {enabled, disabled} (доступно ли для поиска; при успешной индексации сразу enabled, не дожидаясь обогащения); SummaryStatus ∈ {none, pending, processing, completed, failed}.

Прогресс по этапам (Span Tracker)

knowledge_span_tracker.go + internal/types/knowledge_span.go предоставляют дерево прогресса по этапам (именно по нему фронтенд рисует таймлайн):

Пять нормализованных этапов: StageDocReader / StageChunking / StageEmbedding / StageMultimodal / StagePostProcess (types.AllStages).
Состояния Span: pending / running / done / failed / skipped / cancelled. skipped используется для намеренного пропуска (например, мультимодальность не включена), cancelled — для каскадной отмены из-за сбоя выше по потоку.
У каждого раунда обработки независимый Attempt (repo.NextAttempt), корневой Span имеет name="knowledge_processing", Kind=SpanKindRoot; этапы отмечаются через beginStage / endStage / failStage / skipStage, вход и выход записываются в JSONMap (например chunks_planned / chunks_written / total_text_chars).
При каждой записи прогресса touchKnowledgeHeartbeat синхронно обновляет heartbeat, чтобы Housekeeping отличал всё ещё обрабатываемые задачи от переставших обновляться.

Самовосстановление Housekeeping (knowledge_housekeeping.go)

Каждые 5 минут (можно отключить через WEKNORA_HOUSEKEEPING_ENABLED) выполняется фоновый раунд, исправляющий зомби-состояния из-за падения worker / потери задач в Redis:

Sweep A — восстановление зависших знаний, трёхэтапная фильтрация:

Черновой отбор: parse_status IN (pending, processing, finalizing) AND updated_at < cutoff;
filterByLastSpanActivity: проверка heartbeat MAX(updated_at) в knowledge_processing_spans; записи с heartbeat в пределах порога сохраняются (всё ещё обрабатываются), а не имеющие ни одного span тоже признаются зависшими;
filterOutQueued: через asynq TaskInspector проверяется, есть ли ещё задачи в очереди; если есть — сохраняются (просто находятся в очереди).

Признанные зависшими знания построчно обновляются на parse_status = 'failed', pending_subtasks_count = 0, а в error_message указывается, на каком этапе застряло и время последнего прогресса, например:

text

task stuck in processing at docreader stage: no progress since 2026-09-22T09:37:54Z (> 2h10m0s), recovered by housekeeping

Одновременно закрываются span'ы, всё ещё находящиеся в pending/running в последнем attempt этого знания, с кодом ошибки TASK_STALLED, и дописывается duration_ms: место зависания помечается failed, остальные — cancelled. Место зависания — это работающий span этапа; если работающего этапа нет (в finalizing span постобработки уже закрыт), берётся самый внутренний всё ещё работающий дочерний span, и в тексте указываются их имена, например at postprocess stage (postprocess.summary). При сбое запроса heartbeat span'а время в текст не пишется. Так на таймлайне видно, где именно застряло, вместо бесконечного вращения.

Фронтенд до вердикта сервера сначала показывает подсказку: интерфейсы пакетного запроса и spans для обрабатываемых знаний возвращают last_activity_at; при отсутствии прогресса более 20 минут дополнительно возвращается серверный вердикт stall_state (тот же критерий накопления, что и в Sweep A). stalled в списке, карточке и на таймлайне отображается как «возможно, зависло», на таймлайне есть вход «остановить разбор»; queued отображается как «в очереди», без входа для остановки; при сбое пробы вердикта нет и показывается как обычный разбор. В период затишья интервал опроса увеличивается до 15 секунд.

Порог staleThreshold() = max(1h, DocumentProcessTimeout) + 10min.

Sweep B — восстановление зависшей сводки: summary_status = 'processing' AND updated_at < 1 час назад → установка failed. Если summary_status = 'pending', а разбор уже завершён (completed / failed / cancelled), прошло более 1 часа и в очереди нет её задачи, он также устанавливается в failed, чтобы после преждевременного отказа задачи сводки фронтенд не показывал вечно «генерация сводки». При отборе зависших документов в Sweep A и отмене разбора пользователем незавершённые состояния сводки тоже закрываются (в первом случае как failed, во втором как none).

Sweep C — восстановление зависшего удаления: если parse_status = 'deleting' и порог превышен, а в очереди нет охватывающей его задачи knowledge:list_delete, состояние устанавливается в failed с указанием причины, документ снова появляется в списке и его можно удалить повторно. При сбое пробы очереди обработка откладывается до следующего раунда; в режиме Lite задач в очереди нет, поэтому восстановление идёт сразу по фактическому положению.

Подстраховка очереди Wiki: для документов, застрявших в finalizing только из-за того, что постоянная очередь Wiki не потребляется, инспекция заново запускает генерацию Wiki для соответствующей базы знаний (не чаще одного раза на базу знаний за пороговый период). Если при запуске задачи Wiki обнаруживается, что база знаний отключила Wiki, в ней нет модели синтеза или модель удалена, незаявленные операции Wiki очищаются и соответствующие документы освобождаются, корректно завершая finalizing. Если модель синтеза есть, но её не удаётся собрать (например, адрес отклонён проверкой SSRF), задача Wiki на последней повторной попытке освобождает документы тем же способом. Операции Wiki защищают документ максимум 48 часов: документы без прогресса более 48 часов отбираются Sweep A как failed, а их операции Wiki вслед за этим удаляются.

Цепочка очистки при удалении (knowledge_delete.go)

Порядок DeleteKnowledge(ctx, id) тщательно продуман (сначала удаление строки, потом файлов, при сбое возможен повтор):

Пометить ParseStatus = deleting (блокирует запись параллельных задач);
Для знаний в состоянии pending/processing выполнить dequeueKnowledgeTasks() для отмены нижестоящих задач в очереди;
Параллельная очистка через errgroup четырёх видов ресурсов:

векторный/ключевой индекс: retrieveEngine.DeleteByKnowledgeIDList (маршрутизация по размерности embedding и типу KB);
Wiki: cleanupWikiOnKnowledgeDelete (запись Redis tombstone → очистка pending ingest → reconcile существующих страниц → постановка WikiRetract в очередь);
Chunks: chunkService.DeleteChunksByKnowledgeID;
граф: graphEngine.DelGraph;

Удаление связей Tag → удаление строки Knowledge в БД;
В самом конце best-effort очистка физических файлов: исходный файл + все извлечённые изображения, собранные из chunk_image_info (collectImageURLs + deleteExtractedImages), и возврат статистики хранилища тенанта.

Пакетная версия DeleteKnowledgeList предварительно загружает FileService каждой KB, группирует URL изображений по KB и группирует удаление индексов по модели embedding, избегая повторных запросов внутри горутин.

Импорт знаний типа FAQ (knowledge_faq.go / knowledge_faq_import.go)

База знаний FAQ не идёт по конвейеру разбора документов: у каждой FAQ KB есть только один экземпляр Knowledge (ensureFAQKnowledge), каждая пара вопрос-ответ — это Chunk типа ChunkTypeFAQ, а метаданные хранятся в Chunk.Metadata:

type FAQChunkMetadata struct {
    StandardQuestion  string   // стандартный вопрос
    SimilarQuestions  []string // похожие вопросы
    NegativeQuestions []string // негативные вопросы (фильтрация негативных примеров, не участвуют в индексации)
    Answers           []string
    AnswerStrategy    AnswerStrategy // "all" | "random"
    ...
}

Создание одной записи CreateFAQEntry: очистка и проверка → проверка дубликатов (checkFAQQuestionDuplicate) → построение Chunk (buildFAQChunkContent по FAQIndexMode решает, писать ли ответ в Content) → синхронная индексация indexFAQChunks → ChunkStatusIndexed.
Режим индексации (конфигурация уровня KB): FAQIndexModeQuestionOnly (question_only, индексируются только вопросы) / FAQIndexModeQuestionAnswer (question_answer, вопрос+ответ); индексация вопросов делится на FAQQuestionIndexModeCombined (стандартный + похожие вопросы объединяются в один вектор) и FAQQuestionIndexModeSeparate (каждый похожий вопрос — отдельный вектор, source_id вида {chunkID}-{index}, поддерживается инкрементальная индексация incrementalIndexFAQEntry).
Массовый импорт UpsertFAQEntries:
режим append (добавление/слияние) или replace (полная замена), поддерживается DryRun только с проверкой;
при более чем 200 записях или 50KB записи сначала загружаются SaveBytes в объектное хранилище, а payload несёт только EntriesURL;
постановка TypeFAQImport → QueueMaintenance, MaxRetry 5 (dry-run 3), Timeout 2 часа; для одной KB одновременно допускается только одна задача импорта (блокировка Redis);
дедупликация на основе CalculateFAQContentHash: к стандартному/похожим/негативным вопросам/ответам применяется нормализация (удаление URL, приведение к нижнему регистру, упрощение традиционных иероглифов, перевод полноширинных в полуширинные, умные пробелы), затем SHA256;
режим append выполняет четырёхэтапную проверку (конфликт стандартного вопроса → сбой всей записи; конфликт похожего вопроса/негативного вопроса → частичный сбой с отбрасыванием только конфликтующих пунктов; стандартный вопрос уже существует → слияние объединения);
прогресс пишется в Redis (FAQImportProgress: pending/processing/completed/failed, счётчики успеха/сбоя/частичного сбоя/пропуска), неуспешные записи экспортируются в CSV с UTF-8 BOM для скачивания.

Клонирование и перемещение знаний (knowledge_clone_move.go)

Клонирование (CloneKnowledgeBase / CloneChunk)

Клонирование уровня KB сначала копирует конфигурацию KB, затем по разности множеств (AminusB) добавляет/удаляет Knowledge, обрабатывая параллельно (удаление пачками по 10, клонирование по одному).
Клонирование уровня Chunk (пачками по 100) копирует пять типов chunk Text/ParentText/Summary/ImageCaption/ImageOCR:
Глубокое копирование изображений: cloneChunkImageInfo читает байты из исходного хранилища → пишет в пространство имён exports/ целевого тенанта, urlCache обеспечивает дедупликацию; rewriteContentImageURLs заменяет все старые URL в Content (сначала самые длинные URL, чтобы избежать частичного совпадения);
отображение тегов getOrCreateTagInTarget (при совпадении имени переиспользуется, иначе создаётся);
после перестроения отображений PreChunkID/NextChunkID/ParentChunkID — пакетная вставка;
векторный индекс копируется напрямую через retrieveEngine.CopyIndices(), без пересчёта embedding.
Клонирование FAQ KB идёт через разностную синхронизацию: chunkRepo.FAQChunkDiff по content_hash вычисляет три группы — добавление/удаление/совпадение, для совпадающих пар синхронизируется только состояние (IsEnabled/Flags/TagID/AnswerStrategy).
Прогресс пишется в Redis (KBCloneProgress).

Перемещение (ProcessKnowledgeMove)

Проверка порогов: тип исходной и целевой KB должен совпадать, и EmbeddingModelID должен совпадать. Два режима:

reuse_vectors: требует sourceKB.SharesStoreWith(targetKB) (один и тот же экземпляр векторного хранилища), CopyIndices копирует индекс → удаление исходного индекса → MoveChunksByKnowledgeID меняет принадлежность chunk → очистка связей Tag → обновление KB ID у Knowledge;
reparse: используется при разных векторных хранилищах. cleanupKnowledgeResources (удаление индекса/chunks/графа/возврат статистики хранилища) → Knowledge сбрасывается в pending и прикрепляется к целевой KB → повторная постановка TypeDocumentProcess (для типа manual идёт triggerManualProcessing).

Итоговый сквозной сценарий

Полный путь PDF с включёнными мультимодальностью, генерацией вопросов и графом таков:

POST /knowledge-bases/:id/knowledge/file → дедупликация по MD5 → cos://tenant/kb/uuid.pdf → Knowledge(pending) → asynq document:process;
Worker: Span attempt=1 открывает корень → этап docreader вызывает Python-сервис по gRPC и получает Markdown+байты изображений → изображения загружаются в хранилище и URL переписываются → этап chunking режет Go chunker → запись chunks → этап embedding BatchIndex → EnableStatus=enabled (с этого момента доступно для поиска) → постановка мультимодальной задачи на каждое изображение;
мультимодальный worker обрабатывает изображения по одному OCR+Caption, создаёт дочерние chunk'и image_caption/image_ocr и индексирует их; после завершения всех запускается post-process;
оркестратор вычисляет expectedSubtasks (1 сводка + N/20 пачек вопросов + M граф + 0/1 Wiki) → SetFinalizing → веерная рассылка; каждая подзадача в конечном состоянии вызывает FinalizeSubtask с уменьшением, и при достижении 0 → completed;
если на любом из этапов процесс зависает, Housekeeping каждые 5 минут по трём критериям «updated_at + heartbeat span'а + проверка очереди» отбирает его как failed, и пользователь может запустить reparse (attempt+1) заново.

Ссылки на реализацию

Расположение исходного кода, соответствующее каждому этапу:

Этап	Расположение исходного кода
HTTP-вход	`internal/handler/knowledge.go`, `internal/router/routes_knowledge.go`
Создание и постановка в очередь	`internal/application/service/knowledge_create.go`, `knowledge_task_options.go`
Файловое хранилище	`internal/application/service/file/` (`factory.go`, реализации каждого бэкенда)
Инфраструктура разбора	`internal/infrastructure/docparser/`, `docreader/` (Python-сервис)
Основной конвейер обработки	`internal/application/service/knowledge_process.go`
Слияние конфигурации обработки	`internal/application/service/knowledge_process_config.go`
Постобработка	`internal/application/service/knowledge_post_process.go`, `image_multimodal.go`
Отслеживание прогресса	`internal/application/service/knowledge_span_tracker.go`, `internal/types/knowledge_span.go`
Самовосстановление	`internal/application/service/knowledge_housekeeping.go`
Удаление	`internal/application/service/knowledge_delete.go`
FAQ	`internal/application/service/knowledge_faq.go`, `knowledge_faq_import.go`
Клонирование/перемещение	`internal/application/service/knowledge_clone_move.go`

Конвейер индексации документов (Document Ingestion Pipeline) ​

Общая архитектура ​

Слой входа: три способа создания ​

Загрузка файла (CreateKnowledgeFromFile) ​

Единый шлюз по расширениям ​

Импорт по URL (CreateKnowledgeFromURL) ​

Ручное создание (CreateManualKnowledge) ​

Механизм дедупликации ​

Начальное состояние ​

Слой файлового хранилища (FileService и бэкенды хранилища) ​

Определение интерфейса ​

Поддерживаемые бэкенды хранилища ​

Правила организации ключей объектов ​

Два слоя-обёртки ​

Механизм асинхронных задач (Asynq + Redis) ​

Постановка в очередь ​

Топология очередей и пулы Worker ​

Семантика повторных попыток при сбое ​

Конфигурация обработки: значения KB по умолчанию + переопределение при отдельной загрузке ​

Основной конвейер обработки (knowledge_process.go) ​

Разбор (convert, Stage: docreader) ​

Транскрипция ASR (аудиофайлы) ​

Извлечение и загрузка изображений ​

Разбиение на блоки (Stage: chunking) ​

Запись в БД и индексация (processChunks, Stage: chunking + embedding) ​

Оркестрация постобработки (knowledge_post_process.go, Stage: postprocess) ​

Автоматические теги документа ​

Обновление описания базы знаний (knowledgebase_profile.go) ​

Обновление сводки (knowledge_summary_refresh.go) ​

Мультимодальность изображений (image_multimodal.go) ​

Конечный автомат ​

Основное состояние Knowledge (ParseStatus) ​

Прогресс по этапам (Span Tracker) ​

Самовосстановление Housekeeping (knowledge_housekeeping.go) ​

Цепочка очистки при удалении (knowledge_delete.go) ​

Импорт знаний типа FAQ (knowledge_faq.go / knowledge_faq_import.go) ​

Клонирование и перемещение знаний (knowledge_clone_move.go) ​

Клонирование (CloneKnowledgeBase / CloneChunk) ​

Перемещение (ProcessKnowledgeMove) ​

Итоговый сквозной сценарий ​

Ссылки на реализацию ​