feat: add video multimodal functionality

This commit is contained in:
DaWesen
2026-04-08 19:52:28 +08:00
committed by lyingbug
parent 2b10cc54de
commit b80bdb1079
20 changed files with 1213 additions and 28 deletions
+1
View File
@@ -67,6 +67,7 @@ RUN if [ -n "$APK_MIRROR_ARG" ]; then \
python3 python3-pip python3-dev libffi-dev libssl-dev \
nodejs npm \
gosu && \
libsqlite3-0 ffmpeg \
python3 -m pip install --break-system-packages --upgrade pip setuptools wheel && \
mkdir -p /home/appuser/.local/bin && \
curl -LsSf https://astral.sh/uv/install.sh | CARGO_HOME=/home/appuser/.cargo UV_INSTALL_DIR=/home/appuser/.local/bin sh && \
+4 -2
View File
@@ -103,15 +103,17 @@ export default {
noValidFilesSelected: 'All selected files are unsupported',
hiddenFilesFiltered: 'Filtered {count} hidden files',
imagesFilteredNoVLM: 'Filtered {count} image files (VLM not enabled)',
videosFilteredNoVLM: 'Filtered {count} video files (VLM not enabled)',
audiosFilteredNoASR: 'Filtered {count} audio files (ASR not enabled)',
invalidFilesFiltered: 'Filtered {count} unsupported files',
unsupportedFileType: 'Unsupported file type',
unsupportedTypesHint: 'Some document types ({types}) have no available parser engine and cannot be processed',
goToParserSettings: 'Configure',
failedFilesList: 'Failed files:',
failedFilesList: 'Failed files:',
andMoreFiles: '...and {count} more files',
duplicateFilesSkipped: '{count} duplicate files skipped',
uploadFile: 'Upload File',
uploadFileDesc: 'Supports PDF, Word, TXT, etc.',
uploadFileDesc: 'Supports PDF, Word, TXT, images, videos, audio, etc.',
importURL: 'Import from URL',
addDocument: 'Add Document',
importURLDesc: 'Import via URL link',
+3 -1
View File
@@ -102,6 +102,8 @@ export default {
noValidFilesSelected: "선택한 파일 모두 지원되지 않음",
hiddenFilesFiltered: "{count}개 숨김 파일 필터링됨",
imagesFilteredNoVLM: "{count}개 이미지 파일 필터링됨(VLM 미활성화)",
videosFilteredNoVLM: "{count}개 동영상 파일 필터링됨(VLM 미활성화)",
audiosFilteredNoASR: "{count}개 오디오 파일 필터링됨(ASR 미활성화)",
invalidFilesFiltered: "{count}개 지원되지 않는 파일 필터링됨",
unsupportedFileType: "지원되지 않는 파일 형식",
unsupportedTypesHint: "일부 문서 유형({types})에 사용 가능한 파서 엔진이 없어 처리할 수 없습니다",
@@ -110,7 +112,7 @@ export default {
andMoreFiles: "...외 {count}개 파일",
duplicateFilesSkipped: "{count}개 중복 파일 무시됨",
uploadFile: "파일 업로드",
uploadFileDesc: "PDF, Word, TXT 등 지원",
uploadFileDesc: "PDF, Word, TXT, 이미지, 동영상, 오디오 등 지원",
importURL: "웹페이지 가져오기",
addDocument: "문서 추가",
importURLDesc: "URL 링크로 가져오기",
+3 -1
View File
@@ -85,6 +85,8 @@ export default {
noValidFilesSelected: 'Все выбранные файлы не поддерживаются',
hiddenFilesFiltered: 'Отфильтровано {count} скрытых файлов',
imagesFilteredNoVLM: 'Отфильтровано {count} изображений (VLM не включен)',
videosFilteredNoVLM: 'Отфильтровано {count} видеофайлов (VLM не включен)',
audiosFilteredNoASR: 'Отфильтровано {count} аудиофайлов (ASR не включен)',
invalidFilesFiltered: 'Отфильтровано {count} неподдерживаемых файлов',
unsupportedFileType: 'Неподдерживаемый тип файла',
unsupportedTypesHint: 'Некоторые типы документов ({types}) не имеют доступного парсера и не могут быть обработаны',
@@ -93,7 +95,7 @@ export default {
andMoreFiles: '...и ещё {count} файлов',
duplicateFilesSkipped: 'Пропущено {count} повторяющихся файлов',
uploadFile: 'Загрузить файл',
uploadFileDesc: 'Поддерживает PDF, Word, TXT и т.д.',
uploadFileDesc: 'Поддерживает PDF, Word, TXT, изображения, видео, аудио и т.д.',
importURL: 'Импорт из URL',
importURLDesc: 'Импорт по ссылке URL',
importURLTitle: 'Импорт из URL',
+3 -1
View File
@@ -102,6 +102,8 @@ export default {
noValidFilesSelected: "选中的文件均不支持",
hiddenFilesFiltered: "已过滤 {count} 个隐藏文件",
imagesFilteredNoVLM: "已过滤 {count} 个图片文件(未启用VLM)",
videosFilteredNoVLM: "已过滤 {count} 个视频文件(未启用VLM)",
audiosFilteredNoASR: "已过滤 {count} 个音频文件(未启用ASR)",
invalidFilesFiltered: "已过滤 {count} 个不支持的文件",
unsupportedFileType: "不支持的文件格式",
unsupportedTypesHint: "部分文档类型({types})暂无可用解析引擎,上传后将无法解析",
@@ -110,7 +112,7 @@ export default {
andMoreFiles: "...及其他 {count} 个文件",
duplicateFilesSkipped: "已忽略 {count} 个重复文件",
uploadFile: "上传文件",
uploadFileDesc: "支持 PDF、Word、TXT 等",
uploadFileDesc: "支持 PDF、Word、TXT、图片、视频、音频等",
importURL: "导入网页",
addDocument: "添加文档",
importURLDesc: "通过URL链接导入",
+1 -1
View File
@@ -40,7 +40,7 @@ export function formatStringDate(date: any) {
year + "-" + month + "-" + day + " " + hour + ":" + minute + ":" + second
);
}
const DEFAULT_VALID_TYPES = new Set(["pdf", "txt", "md", "docx", "doc", "pptx", "ppt", "jpg", "jpeg", "png", "csv", "xlsx", "xls"]);
const DEFAULT_VALID_TYPES = new Set(["pdf", "txt", "md", "docx", "doc", "pptx", "ppt", "jpg", "jpeg", "png", "csv", "xlsx", "xls", "mp4", "mov", "avi", "mkv", "webm", "wmv", "flv", "mp3", "wav", "m4a", "flac", "ogg"]);
/**
* Returns true when the file should be **rejected**.
+30 -4
View File
@@ -210,6 +210,18 @@ const fileTypeOptions = computed(() => [
{ content: 'MD', value: 'md' },
{ content: 'URL', value: 'url' },
{ content: t('knowledgeBase.typeManual'), value: 'manual' },
{ content: 'MP4', value: 'mp4' },
{ content: 'MOV', value: 'mov' },
{ content: 'AVI', value: 'avi' },
{ content: 'MKV', value: 'mkv' },
{ content: 'WEBM', value: 'webm' },
{ content: 'WMV', value: 'wmv' },
{ content: 'FLV', value: 'flv' },
{ content: 'MP3', value: 'mp3' },
{ content: 'WAV', value: 'wav' },
{ content: 'M4A', value: 'm4a' },
{ content: 'FLAC', value: 'flac' },
{ content: 'OGG', value: 'ogg' },
]);
type TagInputInstance = ComponentPublicInstance<{ focus: () => void; select: () => void }>;
const tagDropdownOptions = computed(() =>
@@ -1115,11 +1127,14 @@ const handleFolderUpload = async (event: Event) => {
}
const vlmEnabled = kbInfo.value?.vlm_config?.enabled || false;
const asrEnabled = kbInfo.value?.asr_config?.enabled || false;
const dynamicTypes = supportedFileTypes.value.size > 0 ? supportedFileTypes.value : undefined
const validFiles: File[] = [];
let hiddenFileCount = 0;
let imageFilteredCount = 0;
let videoFilteredCount = 0;
let audioFilteredCount = 0;
for (let i = 0; i < files.length; i++) {
const file = files[i];
@@ -1132,13 +1147,25 @@ const handleFolderUpload = async (event: Event) => {
continue;
}
const fileExt = file.name.substring(file.name.lastIndexOf('.') + 1).toLowerCase();
const imageTypes = ['jpg', 'jpeg', 'png', 'gif', 'bmp', 'webp'];
const videoTypes = ['mp4', 'mov', 'avi', 'mkv', 'webm', 'wmv', 'flv'];
const audioTypes = ['mp3', 'wav', 'm4a', 'flac', 'ogg'];
if (!vlmEnabled) {
const fileExt = file.name.substring(file.name.lastIndexOf('.') + 1).toLowerCase();
const imageTypes = ['jpg', 'jpeg', 'png', 'gif', 'bmp', 'webp'];
if (imageTypes.includes(fileExt)) {
imageFilteredCount++;
continue;
}
if (videoTypes.includes(fileExt)) {
videoFilteredCount++;
continue;
}
}
if (!asrEnabled && audioTypes.includes(fileExt)) {
audioFilteredCount++;
continue;
}
if (!kbFileTypeVerification(file, true, dynamicTypes)) {
@@ -1151,7 +1178,6 @@ const handleFolderUpload = async (event: Event) => {
if (input) input.value = '';
return;
}
MessagePlugin.info(t('knowledgeBase.uploadingFolder', { total: validFiles.length }));
// 批量上传
@@ -1514,7 +1540,7 @@ async function createNewSession(value: string): Promise<void> {
ref="uploadInputRef"
type="file"
class="document-upload-input"
:accept="acceptFileTypes || '.pdf,.docx,.doc,.txt,.md,.json,.jpg,.jpeg,.png,.csv,.xlsx,.xls,.pptx,.ppt,.mp3,.wav,.m4a,.flac,.ogg'"
:accept="acceptFileTypes || '.pdf,.docx,.doc,.txt,.md,.json,.jpg,.jpeg,.png,.csv,.xlsx,.xls,.pptx,.ppt,.mp3,.wav,.m4a,.flac,.ogg,.mp4,.mov,.avi,.mkv,.webm,.wmv,.flv'"
multiple
@change="handleDocumentUpload"
/>
+75 -9
View File
@@ -1479,6 +1479,8 @@ type ProcessChunksOptions struct {
QuestionCount int
EnableMultimodel bool
StoredImages []docparser.StoredImage
IsVideo bool // true when knowledge is a standalone video file
VideoFilePath string // storage path to video file (provider:// URL)
// ParentChunks holds parent chunk data when parent-child chunking is enabled.
// When set, the chunks passed to processChunks are child chunks, and each
// child's ParentIndex references an entry in this slice.
@@ -1864,13 +1866,15 @@ func (s *knowledgeService) processChunks(ctx context.Context,
return
}
// Skip summary/question generation for image-type knowledge — the text chunk
// is just a markdown image reference, so LLM summary would be useless.
// The multimodal task will provide a caption as the description instead.
// Skip summary/question generation for image-type and video-type knowledge — the text chunk
// is just a markdown reference, so LLM summary would be useless.
// The multimodal task will provide a caption/summary as the description instead.
isImage := IsImageType(knowledge.FileType)
pendingMultimodal := isImage && options.EnableMultimodel && len(options.StoredImages) > 0
isVideo := IsVideoType(knowledge.FileType)
pendingMultimodal := (isImage && options.EnableMultimodel && len(options.StoredImages) > 0) ||
(isVideo && options.EnableMultimodel && options.IsVideo && options.VideoFilePath != "")
// For image files with pending multimodal processing, keep "processing" status
// For image/video files with pending multimodal processing, keep "processing" status
// so the frontend waits until the description is ready before showing "completed".
if pendingMultimodal {
knowledge.ParseStatus = types.ParseStatusProcessing
@@ -1884,7 +1888,7 @@ func (s *knowledgeService) processChunks(ctx context.Context,
knowledge.UpdatedAt = now
// Set summary status based on whether summary generation will be triggered
if len(textChunks) > 0 && !isImage {
if len(textChunks) > 0 && !isImage && !isVideo {
knowledge.SummaryStatus = types.SummaryStatusPending
} else {
knowledge.SummaryStatus = types.SummaryStatusNone
@@ -1895,7 +1899,7 @@ func (s *knowledgeService) processChunks(ctx context.Context,
}
// Enqueue question generation task if enabled (async, non-blocking)
if options.EnableQuestionGeneration && len(textChunks) > 0 && !isImage {
if options.EnableQuestionGeneration && len(textChunks) > 0 && !isImage && !isVideo {
questionCount := options.QuestionCount
if questionCount <= 0 {
questionCount = 3
@@ -1907,7 +1911,7 @@ func (s *knowledgeService) processChunks(ctx context.Context,
}
// Enqueue summary generation task (async, non-blocking)
if len(textChunks) > 0 && !isImage {
if len(textChunks) > 0 && !isImage && !isVideo {
s.enqueueSummaryGenerationTask(ctx, knowledge.KnowledgeBaseID, knowledge.ID)
}
@@ -1916,6 +1920,11 @@ func (s *knowledgeService) processChunks(ctx context.Context,
s.enqueueImageMultimodalTasks(ctx, knowledge, kb, options.StoredImages, chunks)
}
// Enqueue multimodal tasks for videos (async, non-blocking)
if options.EnableMultimodel && options.IsVideo && options.VideoFilePath != "" {
s.enqueueVideoMultimodalTasks(ctx, knowledge, kb, options.VideoFilePath, chunks)
}
// Update tenant's storage usage
tenantInfo.StorageUsed += totalStorageSize
if err := s.tenantRepo.AdjustStorageUsed(ctx, tenantInfo.ID, totalStorageSize); err != nil {
@@ -3002,7 +3011,8 @@ func (s *knowledgeService) ReparseKnowledge(ctx context.Context, knowledgeID str
func isValidFileType(filename string) bool {
switch strings.ToLower(getFileType(filename)) {
case "pdf", "txt", "docx", "doc", "md", "markdown", "png", "jpg", "jpeg", "gif", "csv", "xlsx", "xls", "pptx", "ppt", "json",
"mp3", "wav", "m4a", "flac", "ogg":
"mp3", "wav", "m4a", "flac", "ogg",
"mp4", "mov", "avi", "mkv", "webm", "wmv", "flv":
return true
default:
return false
@@ -7424,6 +7434,16 @@ func IsAudioType(fileType string) bool {
}
}
// IsVideoType checks if a file type is a video format
func IsVideoType(fileType string) bool {
switch strings.ToLower(fileType) {
case "mp4", "mov", "avi", "mkv", "webm", "wmv", "flv":
return true
default:
return false
}
}
// downloadFileFromURL downloads a remote file to a temp file and returns its binary content.
// payloadFileName and payloadFileType are in/out pointers: if they point to an empty string,
// the function resolves the value from Content-Disposition / URL path and writes it back.
@@ -7899,6 +7919,8 @@ func (s *knowledgeService) ProcessDocument(ctx context.Context, t *asynq.Task) e
QuestionCount: payload.QuestionCount,
EnableMultimodel: payload.EnableMultimodel,
StoredImages: storedImages,
IsVideo: convertResult != nil && convertResult.IsVideo,
VideoFilePath: payload.FilePath,
}
if kb.ChunkingConfig.EnableParentChild {
@@ -8111,6 +8133,50 @@ func (s *knowledgeService) enqueueImageMultimodalTasks(
}
}
// enqueueVideoMultimodalTasks enqueues asynq tasks for multimodal video processing.
func (s *knowledgeService) enqueueVideoMultimodalTasks(
ctx context.Context,
knowledge *types.Knowledge,
kb *types.KnowledgeBase,
videoFilePath string,
chunks []types.ParsedChunk,
) {
if s.task == nil || videoFilePath == "" {
return
}
// Get the first chunk's ID as the parent chunk
chunkID := ""
if len(chunks) > 0 {
chunkID = chunks[0].ChunkID
}
lang, _ := types.LanguageFromContext(ctx)
payload := types.VideoMultimodalPayload{
TenantID: knowledge.TenantID,
KnowledgeID: knowledge.ID,
KnowledgeBaseID: kb.ID,
ChunkID: chunkID,
VideoURL: videoFilePath,
EnableVLM: true,
EnableASR: true,
Language: lang,
}
payloadBytes, err := json.Marshal(payload)
if err != nil {
logger.Warnf(ctx, "Failed to marshal video multimodal payload: %v", err)
return
}
task := asynq.NewTask(types.TypeVideoMultimodal, payloadBytes)
if _, err := s.task.Enqueue(task); err != nil {
logger.Warnf(ctx, "Failed to enqueue video multimodal task for %s: %v", videoFilePath, err)
} else {
logger.Infof(ctx, "Enqueued video:multimodal task for %s", videoFilePath)
}
}
// ProcessFAQImport handles Asynq FAQ import tasks (including dry run mode)
func (s *knowledgeService) ProcessFAQImport(ctx context.Context, t *asynq.Task) error {
var payload types.FAQImportPayload
@@ -0,0 +1,596 @@
package service
import (
"context"
"encoding/json"
"fmt"
"io"
"os"
"strings"
"time"
filesvc "github.com/Tencent/WeKnora/internal/application/service/file"
"github.com/Tencent/WeKnora/internal/application/service/retriever"
"github.com/Tencent/WeKnora/internal/logger"
"github.com/Tencent/WeKnora/internal/models/asr"
"github.com/Tencent/WeKnora/internal/models/utils/ollama"
"github.com/Tencent/WeKnora/internal/models/video"
"github.com/Tencent/WeKnora/internal/models/vlm"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/types/interfaces"
secutils "github.com/Tencent/WeKnora/internal/utils"
"github.com/google/uuid"
"github.com/hibiken/asynq"
)
const (
vlmFramePrompt = "Analyze this video frame and provide a detailed description in Chinese. Requirements:\n" +
"1. Describe visible objects, people, and their actions\n" +
"2. Describe the scene and environment\n" +
"3. Extract and describe any text on the screen\n" +
"4. Note any notable events or changes\n" +
"5. Be concise but comprehensive"
vlmSummaryPrompt = "Summarize the following video frame descriptions into a coherent video summary in Chinese. Requirements:\n" +
"1. The descriptions are ordered chronologically by timestamp\n" +
"2. Focus on the main storyline and key events\n" +
"3. Maintain the temporal flow of the video\n" +
"4. Be concise but comprehensive"
)
// VideoMultimodalService handles video:multimodal asynq tasks.
// It reads videos from storage (via FileService for provider:// URLs),
// extracts frames using FFmpeg, performs VLM analysis and ASR transcription,
// and creates child chunks.
type VideoMultimodalService struct {
chunkService interfaces.ChunkService
modelService interfaces.ModelService
kbService interfaces.KnowledgeBaseService
knowledgeRepo interfaces.KnowledgeRepository
tenantRepo interfaces.TenantRepository
retrieveEngine interfaces.RetrieveEngineRegistry
ollamaService *ollama.OllamaService
taskEnqueuer interfaces.TaskEnqueuer
}
func NewVideoMultimodalService(
chunkService interfaces.ChunkService,
modelService interfaces.ModelService,
kbService interfaces.KnowledgeBaseService,
knowledgeRepo interfaces.KnowledgeRepository,
tenantRepo interfaces.TenantRepository,
retrieveEngine interfaces.RetrieveEngineRegistry,
ollamaService *ollama.OllamaService,
taskEnqueuer interfaces.TaskEnqueuer,
) interfaces.TaskHandler {
return &VideoMultimodalService{
chunkService: chunkService,
modelService: modelService,
kbService: kbService,
knowledgeRepo: knowledgeRepo,
tenantRepo: tenantRepo,
retrieveEngine: retrieveEngine,
ollamaService: ollamaService,
taskEnqueuer: taskEnqueuer,
}
}
// Handle implements asynq handler for TypeVideoMultimodal.
func (s *VideoMultimodalService) Handle(ctx context.Context, task *asynq.Task) error {
var payload types.VideoMultimodalPayload
if err := json.Unmarshal(task.Payload(), &payload); err != nil {
return fmt.Errorf("unmarshal video multimodal payload: %w", err)
}
logger.Infof(ctx, "[VideoMultimodal] Processing video: chunk=%s, url=%s, vlm=%v, asr=%v",
payload.ChunkID, payload.VideoURL, payload.EnableVLM, payload.EnableASR)
ctx = context.WithValue(ctx, types.TenantIDContextKey, payload.TenantID)
if payload.Language != "" {
ctx = context.WithValue(ctx, types.LanguageContextKey, payload.Language)
}
var videoBytes []byte
videoBytes, err := s.readVideoBytes(ctx, payload)
if err != nil {
return fmt.Errorf("read video bytes: %w", err)
}
if !payload.EnableVLM && !payload.EnableASR {
logger.Infof(ctx, "[VideoMultimodal] Both VLM and ASR are disabled, nothing to do")
return nil
}
var frames []*video.Frame
var extractor video.Extractor
if payload.EnableVLM {
var extErr error
extractor, extErr = video.NewExtractor(nil)
if extErr != nil {
logger.Warnf(ctx, "[VideoMultimodal] FFmpeg not found or not executable, frame extraction disabled: %v", extErr)
logger.Warnf(ctx, "[VideoMultimodal] Note: Please install FFmpeg to enable VLM frame analysis")
frames = []*video.Frame{}
} else {
options := video.DefaultExtractOptions()
if payload.MaxFrames > 0 {
options.MaxFrames = payload.MaxFrames
}
var extractErr error
frames, extractErr = extractor.ExtractFrames(ctx, videoBytes, options)
if extractErr != nil {
logger.Warnf(ctx, "[VideoMultimodal] Frame extraction failed: %v", extractErr)
frames = []*video.Frame{}
}
}
} else {
logger.Infof(ctx, "[VideoMultimodal] VLM disabled, skipping frame extraction")
frames = []*video.Frame{}
}
frameDescriptions := make([]string, 0, len(frames))
if payload.EnableVLM {
vlmMdl, vlmErr := s.resolveVLM(ctx, payload.KnowledgeBaseID)
if vlmErr != nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to resolve VLM: %v", vlmErr)
} else {
for i, frame := range frames {
desc, descErr := vlmMdl.Predict(ctx, frame.ImageData, vlmFramePrompt)
if descErr != nil {
logger.Warnf(ctx, "[VideoMultimodal] VLM analysis failed for frame %d at %.2fs: %v",
i, frame.Timestamp, descErr)
continue
}
frameDesc := fmt.Sprintf("[%.2fs] %s", frame.Timestamp, desc)
frameDescriptions = append(frameDescriptions, frameDesc)
logger.Infof(ctx, "[VideoMultimodal] Frame %d analyzed at %.2fs, desc len=%d",
i, frame.Timestamp, len(desc))
}
}
}
var asrText string
if payload.EnableASR {
asrText, err = s.transcribeAudio(ctx, videoBytes, payload.KnowledgeBaseID)
if err != nil {
logger.Warnf(ctx, "[VideoMultimodal] ASR transcription failed: %v", err)
} else if asrText != "" {
logger.Infof(ctx, "[VideoMultimodal] ASR transcription completed, len=%d", len(asrText))
}
}
var videoSummary string
if len(frameDescriptions) > 0 && payload.EnableVLM {
vlmMdl, _ := s.resolveVLM(ctx, payload.KnowledgeBaseID)
if vlmMdl != nil {
combinedInput := strings.Join(frameDescriptions, "\n")
summary, sumErr := vlmMdl.Predict(ctx, []byte(combinedInput), vlmSummaryPrompt)
if sumErr != nil {
logger.Warnf(ctx, "[VideoMultimodal] Video summarization failed: %v", sumErr)
videoSummary = combinedInput
} else {
videoSummary = summary
}
} else {
videoSummary = strings.Join(frameDescriptions, "\n")
}
}
videoInfo := types.VideoInfo{
URL: payload.VideoURL,
FrameCount: len(frameDescriptions),
HasVLMAnalysis: len(frameDescriptions) > 0,
HasASR: asrText != "",
VideoSummary: videoSummary,
ASRText: asrText,
FrameDescriptions: frameDescriptions,
}
videoInfoJSON, _ := json.Marshal([]types.VideoInfo{videoInfo})
newChunks := s.buildVideoChunks(payload, frameDescriptions, videoSummary, asrText, string(videoInfoJSON))
if len(newChunks) == 0 {
// Even if VLM/ASR both failed, mark knowledge as completed
s.finalizeVideoKnowledge(ctx, payload, "")
return nil
}
if err := s.chunkService.CreateChunks(ctx, newChunks); err != nil {
return fmt.Errorf("create video multimodal chunks: %w", err)
}
for _, c := range newChunks {
logger.Infof(ctx, "[VideoMultimodal] Created %s chunk %s for video %s, len=%d",
c.ChunkType, c.ID, payload.VideoURL, len(c.Content))
}
// Index chunks so they can be retrieved
s.indexChunks(ctx, payload, newChunks)
// Update the parent text chunk's VideoInfo (mirrors old docreader behaviour)
s.updateParentChunkVideoInfo(ctx, payload, videoInfo)
// For standalone video files, use summary as the knowledge description
// and mark the knowledge as completed (it was kept in "processing" until now).
s.finalizeVideoKnowledge(ctx, payload, videoSummary)
// Enqueue question generation for the frame descriptions/summary/ASR content if KB has it enabled.
// During initial processChunks, question generation is skipped for video-type
// knowledge because the text chunk is just a markdown reference. Now that we
// have real textual content (frame descriptions, summary, ASR), we can generate questions.
s.enqueueQuestionGenerationIfEnabled(ctx, payload)
return nil
}
func (s *VideoMultimodalService) readVideoBytes(ctx context.Context, payload types.VideoMultimodalPayload) ([]byte, error) {
var err error
var videoBytes []byte
if types.ParseProviderScheme(payload.VideoURL) != "" {
fileSvc := s.resolveFileServiceForPayload(ctx, payload)
if fileSvc == nil {
logger.Warnf(ctx, "[VideoMultimodal] Resolve tenant file service failed, fallback to URL/local: tenant=%d kb=%s",
payload.TenantID, payload.KnowledgeBaseID)
} else {
// provider:// scheme — read via FileService
reader, getErr := fileSvc.GetFile(ctx, payload.VideoURL)
if getErr != nil {
logger.Warnf(ctx, "[VideoMultimodal] FileService.GetFile(%s) failed: %v", payload.VideoURL, getErr)
} else {
videoBytes, err = io.ReadAll(reader)
reader.Close()
if err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Read provider file %s failed: %v", payload.VideoURL, err)
videoBytes = nil
}
}
}
}
if videoBytes == nil && payload.VideoLocalPath != "" {
videoBytes, err = os.ReadFile(payload.VideoLocalPath)
if err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Local file %s not available (%v), trying URL", payload.VideoLocalPath, err)
videoBytes = nil
}
}
if videoBytes == nil {
videoBytes, err = downloadVideoFromURL(payload.VideoURL)
if err != nil {
logger.Errorf(ctx, "[VideoMultimodal] Failed to download video from URL %s: %v", payload.VideoURL, err)
return nil, fmt.Errorf("read video from URL %s failed: %w", payload.VideoURL, err)
}
logger.Infof(ctx, "[VideoMultimodal] Video downloaded from URL, len=%d", len(videoBytes))
}
return videoBytes, nil
}
// downloadVideoFromURL downloads video bytes from an HTTP(S) URL.
func downloadVideoFromURL(videoURL string) ([]byte, error) {
return secutils.DownloadBytes(videoURL)
}
// transcribeAudio extracts and transcribes audio from a video using ASR.
func (s *VideoMultimodalService) transcribeAudio(ctx context.Context, videoBytes []byte, kbID string) (string, error) {
asrMdl, err := s.resolveASR(ctx, kbID)
if err != nil {
return "", fmt.Errorf("resolve ASR: %w", err)
}
return asrMdl.Transcribe(ctx, videoBytes, "video.mp4")
}
// buildVideoChunks creates child chunks for video multimodal analysis results:
// - Frame descriptions (one chunk per frame)
// - Video summary (single chunk)
// - ASR transcription (single chunk)
func (s *VideoMultimodalService) buildVideoChunks(
payload types.VideoMultimodalPayload,
frameDescriptions []string,
videoSummary string,
asrText string,
videoInfoJSON string,
) []*types.Chunk {
var newChunks []*types.Chunk
for _, desc := range frameDescriptions {
chunk := &types.Chunk{
ID: uuid.New().String(),
TenantID: payload.TenantID,
KnowledgeID: payload.KnowledgeID,
KnowledgeBaseID: payload.KnowledgeBaseID,
Content: desc,
ChunkType: types.ChunkTypeVideoFrame,
ParentChunkID: payload.ChunkID,
IsEnabled: true,
Flags: types.ChunkFlagRecommended,
VideoInfo: videoInfoJSON,
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
}
newChunks = append(newChunks, chunk)
}
if videoSummary != "" {
newChunks = append(newChunks, &types.Chunk{
ID: uuid.New().String(),
TenantID: payload.TenantID,
KnowledgeID: payload.KnowledgeID,
KnowledgeBaseID: payload.KnowledgeBaseID,
Content: videoSummary,
ChunkType: types.ChunkTypeVideoCaption,
ParentChunkID: payload.ChunkID,
IsEnabled: true,
Flags: types.ChunkFlagRecommended,
VideoInfo: videoInfoJSON,
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
})
}
if asrText != "" {
newChunks = append(newChunks, &types.Chunk{
ID: uuid.New().String(),
TenantID: payload.TenantID,
KnowledgeID: payload.KnowledgeID,
KnowledgeBaseID: payload.KnowledgeBaseID,
Content: asrText,
ChunkType: types.ChunkTypeVideoASR,
ParentChunkID: payload.ChunkID,
IsEnabled: true,
Flags: types.ChunkFlagRecommended,
VideoInfo: videoInfoJSON,
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
})
}
return newChunks
}
// finalizeVideoKnowledge updates the knowledge after multimodal processing:
// - For standalone video files: sets Description from summary and marks ParseStatus as completed.
// - For videos extracted from documents: no-op (description comes from summary generation).
func (s *VideoMultimodalService) finalizeVideoKnowledge(ctx context.Context, payload types.VideoMultimodalPayload, summary string) {
knowledge, err := s.knowledgeRepo.GetKnowledgeByIDOnly(ctx, payload.KnowledgeID)
if err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to get knowledge %s: %v", payload.KnowledgeID, err)
return
}
if knowledge == nil {
return
}
if !IsVideoType(knowledge.FileType) {
return
}
if summary != "" {
knowledge.Description = summary
}
knowledge.ParseStatus = types.ParseStatusCompleted
knowledge.UpdatedAt = time.Now()
if err := s.knowledgeRepo.UpdateKnowledge(ctx, knowledge); err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to finalize knowledge: %v", err)
} else {
logger.Infof(ctx, "[VideoMultimodal] Finalized video knowledge %s (status=completed, description=%d chars)",
payload.KnowledgeID, len(knowledge.Description))
}
}
// indexChunks indexes the newly created video chunks into the retrieval engine
// so they can participate in semantic search.
func (s *VideoMultimodalService) indexChunks(ctx context.Context, payload types.VideoMultimodalPayload, chunks []*types.Chunk) {
kb, err := s.kbService.GetKnowledgeBaseByIDOnly(ctx, payload.KnowledgeBaseID)
if err != nil || kb == nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to get KB for indexing: %v", err)
return
}
embeddingModel, err := s.modelService.GetEmbeddingModel(ctx, kb.EmbeddingModelID)
if err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to get embedding model for indexing: %v", err)
return
}
tenantInfo, err := s.tenantRepo.GetTenantByID(ctx, payload.TenantID)
if err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to get tenant for indexing: %v", err)
return
}
engine, err := retriever.NewCompositeRetrieveEngine(s.retrieveEngine, tenantInfo.GetEffectiveEngines())
if err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to init retrieve engine: %v", err)
return
}
indexInfoList := make([]*types.IndexInfo, 0, len(chunks))
for _, chunk := range chunks {
indexInfoList = append(indexInfoList, &types.IndexInfo{
Content: chunk.Content,
SourceID: chunk.ID,
SourceType: types.ChunkSourceType,
ChunkID: chunk.ID,
KnowledgeID: chunk.KnowledgeID,
KnowledgeBaseID: chunk.KnowledgeBaseID,
})
}
if err := engine.BatchIndex(ctx, embeddingModel, indexInfoList); err != nil {
logger.Errorf(ctx, "[VideoMultimodal] Failed to index video chunks: %v", err)
return
}
// Mark chunks as indexed.
// Must re-fetch from DB because the in-memory objects lack auto-generated fields
// (e.g. seq_id), and GORM Save would overwrite them with zero values.
for _, chunk := range chunks {
dbChunk, err := s.chunkService.GetChunkByIDOnly(ctx, chunk.ID)
if err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to fetch chunk %s for status update: %v", chunk.ID, err)
continue
}
dbChunk.Status = int(types.ChunkStatusIndexed)
if err := s.chunkService.UpdateChunk(ctx, dbChunk); err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to update chunk %s status to indexed: %v", chunk.ID, err)
}
}
logger.Infof(ctx, "[VideoMultimodal] Indexed %d video chunks for video %s", len(chunks), payload.VideoURL)
}
// updateParentChunkVideoInfo updates the parent text chunk's VideoInfo field,
// replicating the behaviour of the old docreader flow where the parent chunk
// carried the full video metadata (URL, frame descriptions, summary, ASR).
func (s *VideoMultimodalService) updateParentChunkVideoInfo(
ctx context.Context,
payload types.VideoMultimodalPayload,
videoInfo types.VideoInfo,
) {
if payload.ChunkID == "" {
return
}
chunk, err := s.chunkService.GetChunkByIDOnly(ctx, payload.ChunkID)
if err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to get parent chunk %s: %v", payload.ChunkID, err)
return
}
var existingInfos []types.VideoInfo
if chunk.VideoInfo != "" {
_ = json.Unmarshal([]byte(chunk.VideoInfo), &existingInfos)
}
found := false
for i, info := range existingInfos {
if info.URL == videoInfo.URL {
existingInfos[i] = videoInfo
found = true
break
}
}
if !found {
existingInfos = append(existingInfos, videoInfo)
}
videoInfoJSON, _ := json.Marshal(existingInfos)
chunk.VideoInfo = string(videoInfoJSON)
chunk.UpdatedAt = time.Now()
if err := s.chunkService.UpdateChunk(ctx, chunk); err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to update parent chunk %s VideoInfo: %v", chunk.ID, err)
} else {
logger.Infof(ctx, "[VideoMultimodal] Updated parent chunk %s VideoInfo for video", chunk.ID)
}
}
// resolveVLM creates a vlm.VLM instance for the given knowledge base,
// supporting both new-style (ModelID) and legacy (inline BaseURL) configs.
func (s *VideoMultimodalService) resolveVLM(ctx context.Context, kbID string) (vlm.VLM, error) {
kb, err := s.kbService.GetKnowledgeBaseByIDOnly(ctx, kbID)
if err != nil {
return nil, fmt.Errorf("get knowledge base %s: %w", kbID, err)
}
if kb == nil {
return nil, fmt.Errorf("knowledge base %s not found", kbID)
}
vlmCfg := kb.VLMConfig
if !vlmCfg.IsEnabled() {
return nil, fmt.Errorf("VLM is not enabled for knowledge base %s", kbID)
}
// New-style: resolve model through ModelService
if vlmCfg.ModelID != "" {
return s.modelService.GetVLMModel(ctx, vlmCfg.ModelID)
}
// Legacy: create VLM from inline config
return vlm.NewVLMFromLegacyConfig(vlmCfg, s.ollamaService)
}
// resolveASR creates an asr.ASR instance for the given knowledge base.
func (s *VideoMultimodalService) resolveASR(ctx context.Context, kbID string) (asr.ASR, error) {
kb, err := s.kbService.GetKnowledgeBaseByIDOnly(ctx, kbID)
if err != nil {
return nil, fmt.Errorf("get knowledge base %s: %w", kbID, err)
}
if kb == nil {
return nil, fmt.Errorf("knowledge base %s not found", kbID)
}
asrCfg := kb.ASRConfig
if !asrCfg.IsASREnabled() {
return nil, fmt.Errorf("ASR is not enabled for knowledge base %s", kbID)
}
return s.modelService.GetASRModel(ctx, asrCfg.ModelID)
}
// resolveFileServiceForPayload resolves tenant/KB scoped file service for reading provider:// URLs.
func (s *VideoMultimodalService) resolveFileServiceForPayload(ctx context.Context, payload types.VideoMultimodalPayload) interfaces.FileService {
tenant, err := s.tenantRepo.GetTenantByID(ctx, payload.TenantID)
if err != nil || tenant == nil {
logger.Warnf(ctx, "[VideoMultimodal] GetTenantByID failed: tenant=%d err=%v", payload.TenantID, err)
return nil
}
provider := types.ParseProviderScheme(payload.VideoURL)
if provider == "" {
kb, kbErr := s.kbService.GetKnowledgeBaseByIDOnly(ctx, payload.KnowledgeBaseID)
if kbErr != nil {
logger.Warnf(ctx, "[VideoMultimodal] GetKnowledgeBaseByIDOnly failed: kb=%s err=%v", payload.KnowledgeBaseID, kbErr)
} else if kb != nil {
provider = strings.ToLower(strings.TrimSpace(kb.GetStorageProvider()))
}
}
baseDir := strings.TrimSpace(os.Getenv("LOCAL_STORAGE_BASE_DIR"))
fileSvc, _, svcErr := filesvc.NewFileServiceFromStorageConfig(provider, tenant.StorageEngineConfig, baseDir)
if svcErr != nil {
logger.Warnf(ctx, "[VideoMultimodal] resolve file service failed: tenant=%d provider=%s err=%v", payload.TenantID, provider, svcErr)
return nil
}
return fileSvc
}
// enqueueQuestionGenerationIfEnabled checks if the knowledge base has question
// generation enabled and, if so, enqueues a task for the video knowledge.
func (s *VideoMultimodalService) enqueueQuestionGenerationIfEnabled(ctx context.Context, payload types.VideoMultimodalPayload) {
if s.taskEnqueuer == nil {
return
}
kb, err := s.kbService.GetKnowledgeBaseByIDOnly(ctx, payload.KnowledgeBaseID)
if err != nil || kb == nil {
return
}
if kb.QuestionGenerationConfig == nil || !kb.QuestionGenerationConfig.Enabled {
return
}
questionCount := kb.QuestionGenerationConfig.QuestionCount
if questionCount <= 0 {
questionCount = 3
}
if questionCount > 10 {
questionCount = 10
}
taskPayload := types.QuestionGenerationPayload{
TenantID: payload.TenantID,
KnowledgeBaseID: payload.KnowledgeBaseID,
KnowledgeID: payload.KnowledgeID,
QuestionCount: questionCount,
Language: payload.Language,
}
payloadBytes, err := json.Marshal(taskPayload)
if err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to marshal question generation payload: %v", err)
return
}
task := asynq.NewTask(types.TypeQuestionGeneration, payloadBytes, asynq.Queue("low"), asynq.MaxRetry(3))
if _, err := s.taskEnqueuer.Enqueue(task); err != nil {
logger.Warnf(ctx, "[VideoMultimodal] Failed to enqueue question generation for %s: %v", payload.KnowledgeID, err)
} else {
logger.Infof(ctx, "[VideoMultimodal] Enqueued question generation task for video knowledge %s (count=%d)",
payload.KnowledgeID, questionCount)
}
}
+2 -1
View File
@@ -47,7 +47,6 @@ import (
"github.com/Tencent/WeKnora/internal/application/service/llmcontext"
memoryService "github.com/Tencent/WeKnora/internal/application/service/memory"
"github.com/Tencent/WeKnora/internal/application/service/retriever"
infra_web_search "github.com/Tencent/WeKnora/internal/infrastructure/web_search"
"github.com/Tencent/WeKnora/internal/config"
"github.com/Tencent/WeKnora/internal/database"
"github.com/Tencent/WeKnora/internal/datasource"
@@ -64,6 +63,7 @@ import (
"github.com/Tencent/WeKnora/internal/im/telegram"
"github.com/Tencent/WeKnora/internal/im/wecom"
"github.com/Tencent/WeKnora/internal/infrastructure/docparser"
infra_web_search "github.com/Tencent/WeKnora/internal/infrastructure/web_search"
"github.com/Tencent/WeKnora/internal/logger"
"github.com/Tencent/WeKnora/internal/mcp"
"github.com/Tencent/WeKnora/internal/models/embedding"
@@ -173,6 +173,7 @@ func BuildContainer(container *dig.Container) *dig.Container {
must(container.Provide(service.NewChunkExtractService, dig.Name("chunkExtractor")))
must(container.Provide(service.NewDataTableSummaryService, dig.Name("dataTableSummary")))
must(container.Provide(service.NewImageMultimodalService, dig.Name("imageMultimodal")))
must(container.Provide(service.NewVideoMultimodalService, dig.Name("videoMultimodal")))
must(container.Provide(service.NewMessageService))
must(container.Provide(service.NewMCPServiceService))
@@ -28,6 +28,10 @@ var audioFormats = map[string]bool{
"mp3": true, "wav": true, "m4a": true, "flac": true, "ogg": true,
}
var videoFormats = map[string]bool{
"mp4": true, "mov": true, "avi": true, "mkv": true, "webm": true, "wmv": true, "flv": true,
}
func init() {
for k := range imageFormats {
simpleFormats[k] = true
@@ -35,6 +39,9 @@ func init() {
for k := range audioFormats {
simpleFormats[k] = true
}
for k := range videoFormats {
simpleFormats[k] = true
}
}
// IsSimpleFormat returns true if the file type can be handled by the Go SimpleFormatReader.
@@ -74,6 +81,8 @@ func (b *SimpleFormatReader) Read(_ context.Context, req *types.ReadRequest) (*t
return imageToResult(req.FileName, req.FileContent), nil
case audioFormats[ft]:
return audioToResult(req.FileName, req.FileContent), nil
case videoFormats[ft]:
return videoToResult(req.FileName, req.FileContent), nil
default:
return nil, fmt.Errorf("unsupported simple format: %s", ft)
}
@@ -113,6 +122,11 @@ func IsAudioFormat(fileType string) bool {
return audioFormats[strings.ToLower(strings.TrimPrefix(fileType, "."))]
}
// IsVideoFormat returns true if the file type is a recognized video format.
func IsVideoFormat(fileType string) bool {
return videoFormats[strings.ToLower(strings.TrimPrefix(fileType, "."))]
}
// audioToResult wraps a standalone audio file. The actual transcription is
// handled by the ASR model in the knowledge service pipeline. Here we just
// return a placeholder markdown with the raw bytes preserved for upstream
@@ -130,6 +144,22 @@ func audioToResult(fileName string, data []byte) *types.ReadResult {
}
}
// videoToResult wraps a standalone video file. The actual frame extraction,
// VLM analysis and ASR transcription is handled by the VideoMultimodalService
// in the knowledge service pipeline. Here we just return a placeholder markdown
// with the raw bytes preserved for upstream processing.
func videoToResult(fileName string, data []byte) *types.ReadResult {
if fileName == "" {
fileName = "video.mp4"
}
// Return a placeholder; the knowledge service will handle the actual processing.
return &types.ReadResult{
MarkdownContent: fmt.Sprintf("[Video file: %s]", fileName),
IsVideo: true,
VideoData: data,
}
}
// ensureOriginalImageRef checks whether the input file is an image and, if the
// returned markdown does not already contain a markdown image reference for it,
// prepends one and appends the raw bytes to imageRefs. This guarantees that
@@ -42,7 +42,7 @@ func (e *builtinEngine) Description() string {
return "DocReader built-in parser engine"
}
func (e *builtinEngine) FileTypes(_ bool) []string {
return []string{"docx", "doc", "pdf", "md", "markdown", "xlsx", "xls", "jpg", "jpeg", "png", "gif", "bmp", "tiff", "webp"}
return []string{"docx", "doc", "pdf", "md", "markdown", "xlsx", "xls", "jpg", "jpeg", "png", "gif", "bmp", "tiff", "webp", "mp3", "wav", "m4a", "flac", "ogg", "mp4", "mov", "avi", "mkv", "webm", "wmv", "flv"}
}
func (e *builtinEngine) CheckAvailable(docreaderConnected bool, _ map[string]string) (bool, string) {
if docreaderConnected {
@@ -67,7 +67,7 @@ func (e *simpleEngine) Description() string {
return "Simple format & image parsing (no external service required)"
}
func (e *simpleEngine) FileTypes(_ bool) []string {
return []string{"md", "markdown", "txt", "csv", "json", "jpg", "jpeg", "png", "gif", "bmp", "tiff", "webp", "mp3", "wav", "m4a", "flac", "ogg"}
return []string{"md", "markdown", "txt", "csv", "json", "jpg", "jpeg", "png", "gif", "bmp", "tiff", "webp", "mp3", "wav", "m4a", "flac", "ogg", "mp4", "mov", "avi", "mkv", "webm", "wmv", "flv"}
}
func (e *simpleEngine) CheckAvailable(_ bool, _ map[string]string) (bool, string) {
return true, ""
+95
View File
@@ -0,0 +1,95 @@
package video
import (
"context"
"io"
)
// Extractor defines the interface for video frame extraction operations.
type Extractor interface {
// ExtractFrames extracts frames from a video file.
// It returns a slice of Frame objects representing the extracted frames.
ExtractFrames(ctx context.Context, videoBytes []byte, options *ExtractOptions) ([]*Frame, error)
// ExtractFramesToReader extracts frames from a video and returns an io.Reader
// for each frame's image data. This is useful for streaming processing.
ExtractFramesToReaders(ctx context.Context, videoBytes []byte, options *ExtractOptions) ([]io.Reader, error)
// GetVideoInfo retrieves basic information about a video file.
// It returns a map containing duration, width, height, etc.
GetVideoInfo(ctx context.Context, videoBytes []byte) (map[string]interface{}, error)
}
// Frame represents a single extracted frame from a video.
type Frame struct {
// Index is the sequential index of the frame (0-based).
Index int
// Timestamp is the time in seconds from the start of the video.
Timestamp float64
// ImageData contains the raw image bytes of the frame.
ImageData []byte
// Format specifies the image format (e.g., "jpeg", "png").
Format string
// Width is the width of the frame in pixels.
Width int
// Height is the height of the frame in pixels.
Height int
}
// ExtractOptions configures how frames are extracted from a video.
type ExtractOptions struct {
// FrameInterval specifies how often to extract frames (in seconds).
// For example, 1.0 means extract one frame per second.
FrameInterval float64
// MaxFrames specifies the maximum number of frames to extract.
// If 0, extract all frames according to FrameInterval.
MaxFrames int
// Format specifies the output image format for extracted frames.
// Supported formats: "jpeg" (default), "png".
Format string
// Quality specifies the image quality for JPEG output (1-100, default 80).
Quality int
// Width specifies the target width for resizing frames (0 = no resize).
Width int
// Height specifies the target height for resizing frames (0 = no resize).
// If only one dimension is specified, the aspect ratio is preserved.
Height int
}
// DefaultExtractOptions returns the default frame extraction options.
func DefaultExtractOptions() *ExtractOptions {
return &ExtractOptions{
FrameInterval: 1.0,
MaxFrames: 0,
Format: "jpeg",
Quality: 80,
Width: 0,
Height: 0,
}
}
// Config holds the configuration needed to create a video extractor.
type Config struct {
// FFMpegPath is the path to the ffmpeg executable.
// If empty, it will try to find ffmpeg in the system PATH.
FFMpegPath string
// TempDir is the directory to use for temporary files.
// If empty, the system default temporary directory is used.
TempDir string
}
// NewExtractor creates a new video frame extractor based on the provided configuration.
func NewExtractor(config *Config) (Extractor, error) {
return NewFFMpegExtractor(config)
}
+298
View File
@@ -0,0 +1,298 @@
package video
import (
"bytes"
"context"
"fmt"
"io"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"github.com/Tencent/WeKnora/internal/logger"
)
// FFMpegExtractor implements the Extractor interface using FFmpeg.
type FFMpegExtractor struct {
ffmpegPath string
tempDir string
}
// NewFFMpegExtractor creates a new FFmpeg-based video extractor.
func NewFFMpegExtractor(config *Config) (Extractor, error) {
ffmpegPath := "ffmpeg"
tempDir := os.TempDir()
if config != nil {
if config.FFMpegPath != "" {
ffmpegPath = config.FFMpegPath
}
if config.TempDir != "" {
tempDir = config.TempDir
}
}
extractor := &FFMpegExtractor{ffmpegPath: ffmpegPath, tempDir: tempDir}
if err := extractor.checkFFmpeg(); err != nil {
return nil, err
}
return extractor, nil
}
// checkFFmpeg verifies that FFmpeg is available.
func (e *FFMpegExtractor) checkFFmpeg() error {
cmd := exec.Command(e.ffmpegPath, "-version")
_, err := cmd.CombinedOutput()
if err != nil {
return fmt.Errorf("ffmpeg not found or not executable: %w", err)
}
return nil
}
// ExtractFrames extracts frames from a video using FFmpeg.
func (e *FFMpegExtractor) ExtractFrames(ctx context.Context, videoBytes []byte, options *ExtractOptions) ([]*Frame, error) {
if options == nil {
options = DefaultExtractOptions()
}
tempVideoPath, err := e.writeTempVideo(videoBytes)
if err != nil {
return nil, err
}
defer os.Remove(tempVideoPath)
tempFrameDir, err := os.MkdirTemp(e.tempDir, "video-frames-")
if err != nil {
return nil, fmt.Errorf("failed to create temp frame directory: %w", err)
}
defer os.RemoveAll(tempFrameDir)
if _, err := e.getVideoDuration(ctx, tempVideoPath); err != nil {
logger.Warnf(ctx, "[VideoFFMpeg] Failed to get video duration: %v", err)
}
if err := e.extractFramesWithFFmpeg(ctx, tempVideoPath, tempFrameDir, options); err != nil {
return nil, err
}
return e.readExtractedFrames(ctx, tempFrameDir, options)
}
// ExtractFramesToReaders extracts frames and returns them as io.Readers.
func (e *FFMpegExtractor) ExtractFramesToReaders(ctx context.Context, videoBytes []byte, options *ExtractOptions) ([]io.Reader, error) {
frames, err := e.ExtractFrames(ctx, videoBytes, options)
if err != nil {
return nil, err
}
readers := make([]io.Reader, 0, len(frames))
for _, frame := range frames {
readers = append(readers, bytes.NewReader(frame.ImageData))
}
return readers, nil
}
// writeTempVideo writes the video bytes to a temporary file.
func (e *FFMpegExtractor) writeTempVideo(videoBytes []byte) (string, error) {
tempFile, err := os.CreateTemp(e.tempDir, "video-input-*.mp4")
if err != nil {
return "", fmt.Errorf("failed to create temp video file: %w", err)
}
defer tempFile.Close()
if _, err := tempFile.Write(videoBytes); err != nil {
os.Remove(tempFile.Name())
return "", fmt.Errorf("failed to write video bytes: %w", err)
}
return tempFile.Name(), nil
}
// getVideoDuration gets the duration of a video using FFmpeg.
func (e *FFMpegExtractor) getVideoDuration(ctx context.Context, videoPath string) (float64, error) {
cmd := exec.CommandContext(ctx, e.ffmpegPath,
"-i", videoPath,
"-f", "null",
"-")
output, err := cmd.CombinedOutput()
if err != nil {
return 0, err
}
outputStr := string(output)
durationStr := ""
lines := strings.Split(outputStr, "\n")
for _, line := range lines {
if strings.Contains(line, "Duration:") {
parts := strings.Split(line, "Duration:")
if len(parts) > 1 {
durationPart := strings.TrimSpace(parts[1])
durationStr = strings.Split(durationPart, ",")[0]
break
}
}
}
if durationStr == "" {
return 0, fmt.Errorf("could not parse duration")
}
timeParts := strings.Split(durationStr, ":")
if len(timeParts) != 3 {
return 0, fmt.Errorf("invalid duration format: %s", durationStr)
}
hours, _ := strconv.Atoi(timeParts[0])
minutes, _ := strconv.Atoi(timeParts[1])
seconds, _ := strconv.ParseFloat(timeParts[2], 64)
totalSeconds := float64(hours*3600) + float64(minutes*60) + seconds
return totalSeconds, nil
}
// extractFramesWithFFmpeg uses FFmpeg to extract frames from a video.
func (e *FFMpegExtractor) extractFramesWithFFmpeg(ctx context.Context, videoPath, outputDir string, options *ExtractOptions) error {
filters := []string{fmt.Sprintf("fps=1/%f", options.FrameInterval)}
if options.Width > 0 || options.Height > 0 {
scaleFilter := "scale="
if options.Width > 0 {
scaleFilter += strconv.Itoa(options.Width)
} else {
scaleFilter += "-1"
}
scaleFilter += ":"
if options.Height > 0 {
scaleFilter += strconv.Itoa(options.Height)
} else {
scaleFilter += "-1"
}
filters = append(filters, scaleFilter)
}
args := []string{
"-i", videoPath,
"-vf", strings.Join(filters, ","),
}
quality := options.Quality
if quality < 1 {
quality = 1
} else if quality > 100 {
quality = 100
}
if options.Format == "jpeg" {
args = append(args, "-q:v", strconv.Itoa(quality))
}
outputPattern := filepath.Join(outputDir, "frame-%05d."+options.Format)
args = append(args, outputPattern)
cmd := exec.CommandContext(ctx, e.ffmpegPath, args...)
var stderr bytes.Buffer
cmd.Stderr = &stderr
if err := cmd.Run(); err != nil {
return fmt.Errorf("ffmpeg extraction failed: %w, stderr: %s", err, stderr.String())
}
logger.Infof(ctx, "[VideoFFMpeg] Extracted frames to: %s", outputDir)
return nil
}
// readExtractedFrames reads the extracted frames from the temporary directory.
func (e *FFMpegExtractor) readExtractedFrames(ctx context.Context, frameDir string, options *ExtractOptions) ([]*Frame, error) {
files, err := filepath.Glob(filepath.Join(frameDir, "frame-*."+options.Format))
if err != nil {
return nil, fmt.Errorf("failed to list frame files: %w", err)
}
if len(files) == 0 {
return nil, fmt.Errorf("no frames were extracted")
}
frames := make([]*Frame, 0, len(files))
for i, filePath := range files {
imgData, err := os.ReadFile(filePath)
if err != nil {
logger.Warnf(ctx, "[VideoFFMpeg] Failed to read frame %d: %v", i, err)
continue
}
frame := &Frame{
Index: i,
Timestamp: float64(i) * options.FrameInterval,
ImageData: imgData,
Format: options.Format,
Width: options.Width,
Height: options.Height,
}
frames = append(frames, frame)
if options.MaxFrames > 0 && len(frames) >= options.MaxFrames {
break
}
}
if len(frames) == 0 {
return nil, fmt.Errorf("no frames could be read")
}
logger.Infof(ctx, "[VideoFFMpeg] Successfully read %d frames", len(frames))
return frames, nil
}
// GetVideoInfo returns basic information about a video.
func (e *FFMpegExtractor) GetVideoInfo(ctx context.Context, videoBytes []byte) (map[string]interface{}, error) {
tempVideoPath, err := e.writeTempVideo(videoBytes)
if err != nil {
return nil, err
}
defer os.Remove(tempVideoPath)
duration, err := e.getVideoDuration(ctx, tempVideoPath)
if err != nil {
return nil, err
}
cmd := exec.CommandContext(ctx, e.ffmpegPath, "-i", tempVideoPath, "-f", "null", "-")
output, _ := cmd.CombinedOutput()
outputStr := string(output)
width, height := 0, 0
lines := strings.Split(outputStr, "\n")
for _, line := range lines {
if strings.Contains(line, "Video:") {
parts := strings.Split(line, ",")
for _, part := range parts {
part = strings.TrimSpace(part)
if strings.Contains(part, "x") && !strings.Contains(part, "SAR") {
resParts := strings.Split(part, "x")
if len(resParts) == 2 {
width, _ = strconv.Atoi(strings.TrimSpace(resParts[0]))
heightStr := strings.TrimSpace(resParts[1])
heightParts := strings.Split(heightStr, " ")
if len(heightParts) > 0 {
height, _ = strconv.Atoi(heightParts[0])
}
}
break
}
}
break
}
}
videoInfo := map[string]interface{}{
"duration": duration,
"width": width,
"height": height,
}
return videoInfo, nil
}
+4
View File
@@ -23,6 +23,7 @@ type AsynqTaskParams struct {
ChunkExtractor interfaces.TaskHandler `name:"chunkExtractor"`
DataTableSummary interfaces.TaskHandler `name:"dataTableSummary"`
ImageMultimodal interfaces.TaskHandler `name:"imageMultimodal"`
VideoMultimodal interfaces.TaskHandler `name:"videoMultimodal"`
}
func getAsynqRedisClientOpt() *asynq.RedisClientOpt {
@@ -109,6 +110,9 @@ func RunAsynqServer(params AsynqTaskParams) *asynq.ServeMux {
// Register image multimodal handler
mux.HandleFunc(types.TypeImageMultimodal, params.ImageMultimodal.Handle)
// Register video multimodal handler
mux.HandleFunc(types.TypeVideoMultimodal, params.VideoMultimodal.Handle)
// Register data source sync handler
mux.HandleFunc(types.TypeDataSourceSync, params.DataSourceService.ProcessSync)
+26
View File
@@ -20,6 +20,12 @@ const (
ChunkTypeImageOCR ChunkType = "image_ocr"
// ChunkTypeImageCaption 表示图片描述的 Chunk
ChunkTypeImageCaption ChunkType = "image_caption"
// ChunkTypeVideoFrame 表示视频帧分析内容的 Chunk
ChunkTypeVideoFrame ChunkType = "video_frame"
// ChunkTypeVideoCaption 表示视频整体描述的 Chunk
ChunkTypeVideoCaption ChunkType = "video_caption"
// ChunkTypeVideoASR 表示视频音频转录的 Chunk
ChunkTypeVideoASR ChunkType = "video_asr"
// ChunkTypeSummary 表示摘要类型的 Chunk
ChunkTypeSummary = "summary"
// ChunkTypeEntity 表示实体类型的 Chunk
@@ -95,6 +101,24 @@ type ImageInfo struct {
OCRText string `json:"ocr_text"`
}
// VideoInfo 表示与 Chunk 关联的视频信息
type VideoInfo struct {
// 视频URL
URL string `json:"url" gorm:"type:text"`
// 帧数量
FrameCount int `json:"frame_count"`
// 是否有VLM分析
HasVLMAnalysis bool `json:"has_vlm_analysis"`
// 是否有ASR转录
HasASR bool `json:"has_asr"`
// 视频摘要
VideoSummary string `json:"video_summary"`
// ASR转录文本
ASRText string `json:"asr_text"`
// 帧描述列表
FrameDescriptions []string `json:"frame_descriptions"`
}
// Chunk represents a document chunk
// Chunks are meaningful text segments extracted from original documents
// and are the basic units of knowledge base retrieval
@@ -147,6 +171,8 @@ type Chunk struct {
ContentHash string `json:"content_hash" gorm:"type:varchar(64);index"`
// 图片信息,存储为 JSON
ImageInfo string `json:"image_info" gorm:"type:text"`
// 视频信息,存储为 JSON
VideoInfo string `json:"video_info" gorm:"type:text"`
// Chunk creation time
CreatedAt time.Time `json:"created_at"`
// Chunk last update time
+2
View File
@@ -22,6 +22,8 @@ type ReadResult struct {
Error string
IsAudio bool // true when the result contains raw audio data needing ASR transcription
AudioData []byte // raw audio bytes for ASR processing
IsVideo bool // true when the result contains raw video data needing multimodal processing
VideoData []byte // raw video bytes for video multimodal processing
}
// ImageRef represents an image reference extracted from the document.
+21 -6
View File
@@ -13,6 +13,7 @@ const (
TypeKnowledgeMove = "knowledge:move" // 知识移动任务
TypeDataTableSummary = "datatable:summary" // 表格摘要任务
TypeImageMultimodal = "image:multimodal" // 图片多模态处理任务(OCR + VLM Caption)
TypeVideoMultimodal = "video:multimodal" // 视频多模态处理任务(抽帧 + VLM + ASR)
TypeManualProcess = "manual:process" // 手工知识更新任务(cleanup + 重新索引)
TypeDataSourceSync = "datasource:sync" // 数据源同步任务
)
@@ -121,7 +122,7 @@ type KnowledgeMoveProgress struct {
SourceKBID string `json:"source_kb_id"`
TargetKBID string `json:"target_kb_id"`
Status KBCloneTaskStatus `json:"status"`
Progress int `json:"progress"` // 0-100
Progress int `json:"progress"` // 0-100
Total int `json:"total"` // 总知识数
Processed int `json:"processed"` // 已处理数
Failed int `json:"failed"` // 失败数
@@ -138,8 +139,8 @@ type ManualProcessPayload struct {
TenantID uint64 `json:"tenant_id"`
KnowledgeID string `json:"knowledge_id"`
KnowledgeBaseID string `json:"knowledge_base_id"`
Content string `json:"content"` // cleaned markdown content
NeedCleanup bool `json:"need_cleanup"` // true for update, false for create
Content string `json:"content"` // cleaned markdown content
NeedCleanup bool `json:"need_cleanup"` // true for update, false for create
}
// ImageMultimodalPayload represents the image multimodal processing task payload.
@@ -147,14 +148,28 @@ type ImageMultimodalPayload struct {
TenantID uint64 `json:"tenant_id"`
KnowledgeID string `json:"knowledge_id"`
KnowledgeBaseID string `json:"knowledge_base_id"`
ChunkID string `json:"chunk_id"` // parent text chunk
ImageURL string `json:"image_url"` // provider:// URL (e.g. local://..., minio://...)
ImageLocalPath string `json:"image_local_path"` // deprecated: kept for backward compat with in-flight tasks
ChunkID string `json:"chunk_id"` // parent text chunk
ImageURL string `json:"image_url"` // provider:// URL (e.g. local://..., minio://...)
ImageLocalPath string `json:"image_local_path"` // deprecated: kept for backward compat with in-flight tasks
EnableOCR bool `json:"enable_ocr"`
EnableCaption bool `json:"enable_caption"`
Language string `json:"language,omitempty"` // Request locale for {{language}} in prompt templates
}
// VideoMultimodalPayload represents the video multimodal processing task payload.
type VideoMultimodalPayload struct {
TenantID uint64 `json:"tenant_id"`
KnowledgeID string `json:"knowledge_id"`
KnowledgeBaseID string `json:"knowledge_base_id"`
ChunkID string `json:"chunk_id"` // parent text chunk
VideoURL string `json:"video_url"` // provider:// URL (e.g. local://..., minio://...)
VideoLocalPath string `json:"video_local_path"` // deprecated: kept for backward compat with in-flight tasks
EnableVLM bool `json:"enable_vlm"` // enable VLM frame analysis
EnableASR bool `json:"enable_asr"` // enable ASR audio transcription
Language string `json:"language,omitempty"` // Request locale for {{language}} in prompt templates
MaxFrames int `json:"max_frames,omitempty"` // optional: max frames to extract, 0 = no limit
}
// KBCloneTaskStatus represents the status of a knowledge base clone task
type KBCloneTaskStatus string
@@ -0,0 +1,7 @@
-- Migration: 000032_add_video_info_to_chunks (down)
-- Description: Remove video_info column from chunks table.
DO $$ BEGIN RAISE NOTICE '[Migration 000032 down] Removing video_info column from chunks'; END $$;
ALTER TABLE chunks DROP COLUMN IF EXISTS video_info;
DO $$ BEGIN RAISE NOTICE '[Migration 000032 down] video_info column removed successfully'; END $$;
@@ -0,0 +1,10 @@
-- Migration: 000032_add_video_info_to_chunks
-- Description: Add video_info column to chunks table for video multimodal processing.
-- This column stores JSON-serialized VideoInfo struct containing video analysis results.
DO $$ BEGIN RAISE NOTICE '[Migration 000032] Adding video_info column to chunks'; END $$;
ALTER TABLE chunks ADD COLUMN IF NOT EXISTS video_info TEXT;
COMMENT ON COLUMN chunks.video_info IS 'Video information in JSON format: {"url": string, "frame_count": int, "has_vlm_analysis": bool, "has_asr": bool, "video_summary": string, "asr_text": string, "frame_descriptions": string[]}';
DO $$ BEGIN RAISE NOTICE '[Migration 000032] video_info column added successfully'; END $$;