Function InitRAGJobEngine

api/service/rag_job_engine.go:38–80

InitRAGJobEngine 初始化RAG任务引擎

()

// InitRAGJobEngine 初始化RAG任务引擎
func InitRAGJobEngine() {
	initOnce.Do(func() {
		if !common.IsRedisEnabled() {
			logger.SysLog("Redis not enabled, skipping RAG job engine initialization")
			return
		}

		// 初始化 V2 引擎
		ragJobFactoryV2 = v2factory.NewJobFactory(model.DB, common.RDB)
		ragJobEngineV2 = v2engines.NewRagJobEngineV2(common.RDB, model.DB, ragJobFactoryV2)

		// 注册 V2 Handler
		ragJobEngineV2.RegisterHandler("document_parsing", v2steps.NewDocumentParsingHandler(model.DB))
		ragJobEngineV2.RegisterHandler("content_cleaning", func(ctx context.Context, job *model.RagJob, config json.RawMessage) error {
			logger.Info(ctx, "V2 Handler: content_cleaning executed")
			return nil
		})
		ragJobEngineV2.RegisterHandler("summary_generation", v2steps.NewSummaryGenerationHandler(model.DB))
		ragJobEngineV2.RegisterHandler("document_chunking", v2steps.NewDocumentChunkingHandler(model.DB))
		ragJobEngineV2.RegisterHandler("vector_indexing", v2steps.NewVectorIndexingHandler(model.DB))
		ragJobEngineV2.RegisterHandler("graph_generation", v2steps.NewGraphGenerationHandler(model.DB))

		// 注册 V2 Recovery Handler
		ragJobEngineV2.RegisterRecoveryHandler("document_parsing", v2steps.RecoverDocumentParsing(model.DB))
		ragJobEngineV2.RegisterRecoveryHandler("content_cleaning", v2steps.RecoverContentCleaning())
		ragJobEngineV2.RegisterRecoveryHandler("summary_generation", v2steps.RecoverSummaryGeneration(model.DB))
		ragJobEngineV2.RegisterRecoveryHandler("document_chunking", v2steps.RecoverDocumentChunking(model.DB))
		ragJobEngineV2.RegisterRecoveryHandler("vector_indexing", v2steps.RecoverVectorIndexing(model.DB))
		ragJobEngineV2.RegisterRecoveryHandler("graph_generation", v2steps.RecoverGraphGeneration(model.DB))

		// 在后台执行恢复+启动 Worker，不阻塞主流程启动
		go func() {
			ragJobEngineV2.RecoverStuckJobs(context.Background())
			rag.StartSiteEmbeddingReindexCoordinator(context.Background(), model.DB, 30*time.Second)
			ragJobEngineV2.StartWorkers()
			// 每小时清理一次超过 24h 无心跳的死 job
			ragJobEngineV2.StartStaleJobCleaner(context.Background(), 1*time.Hour)
			logger.SysLog("RAG job engine recovery, workers, and stale-job cleaner started (background)")
		}()

		logger.SysLog("RAG job engine initialized (recovery running in background)")
	})
}

type RetryJobStepOptionsV2 struct {
	Continue    bool