Files
ragflow/internal/dao/database.go
Zhichang Yu 01d667296d refactor(knowledge_compile): global compile pool, token-budget batching, and DocEngine-only deletion (#17679)
## Summary

This PR refactors the Go knowledge-compilation ingestion pipeline
(`internal/ingestion/knowledge_compile` +
`internal/ingestion/component/knowledge_compiler`) with three related
changes:

- **Token-budget batching for LLM merge decisions.**
`LLMMergeDecider.DecideBatch` previously stuffed every `(existing,
candidate)` pair into a single LLM call, risking `max_token` overflow.
It now splits pairs into token-bounded sub-batches (budget =
`llmMaxTokens * 0.85`) via `tokenizer.NumTokensFromString`, runs them
concurrently while preserving the global pair index, and never
reindexes.
- **Process-level global compile pool.** Introduces a single vCPU-sized
goroutine pool (`pool.go`, env `KC_COMPILE_CONCURRENCY`) dedicated to
*all* knowledge-compilation stages. KNN search loop, `DecideBatch`
sub-batches, `WriteMerged`/`DeleteMerged` internals, and the
component-level (structure/mindmap) per-call pools are all unified into
it via an injected submitter. No more per-job short-lived goroutines in
`runCompilerJobs` (futures are collected then awaited on the caller).
Fan-out stays bounded by the pool worker count; these stages are
docengine-bounded / LLM-bounded, not CPU-bounded.
- **DocEngine-only deletion.** `Consumer.processBatch` deletion no
longer loads the deleted docs' products into memory. Two sequential
DocEngine calls replace the old in-memory surgery:
- `DeleteDocLevelForDocs` — one `DeleteChunks` over `doc_id IN
deletedDocIDs` (merged rows carry `doc_id == kb`, so only per-doc
products match).
- `StripMergedSources` — one `Search` of `kc_merged=1` rows filtered by
`source_doc_ids IN deletedDocIDs` (intersection pushed down to the
engine), `UpdateChunks` the source array of survivors, and
`DeleteChunks` the rows whose array became empty.

## Changes

- `internal/ingestion/knowledge_compile/pool.go` (new): global
`compilerPool` +
`runCompilerJobs`/`SubmitCompilerJob`/`SubmitCompilerJobs`.
- `internal/ingestion/knowledge_compile/consumer.go`: deletion rewritten
to the two DocEngine calls;
`mergedBase`/`toDelete`/`stripDeletedSources` removed.
- `internal/ingestion/knowledge_compile/writer.go`:
`DeleteDocLevelForDocs` + `StripMergedSources` replace
`DeleteMergedForDoc`/`DeleteMerged`.
- `internal/ingestion/knowledge_compile/reader.go`: drop
`LoadMergedBySourceDoc` + `containsString` (keep `LoadDocProducts` for
the completion branch).
- `internal/ingestion/knowledge_compile/dedup.go`: `NewLLMDeduper` takes
`llmMaxTokens`; wires `SetMaxBatchTokens`/`SetSubmitter`.
- `internal/ingestion/knowledge_compiler/{structure,merge}.go`,
`mindmap/mindmap.go`, `pool_wiring.go`: token-budget split + submitter
injection.
- Tests: `structure_test.go` (token-budget split), `dedup_test.go`,
`consumer_test.go` (tombstone + DocEngine deletion assertions) updated.

## Validation

`bash build.sh --test -race ./internal/ingestion/knowledge_compile/...
./internal/ingestion/component/knowledge_compiler/...` passes (unit
tier, no external services).

🤖 Generated with [CodeBuddy](https://www.codebuddy.ai)

---------

Co-authored-by: yuzhichang <yuzhichang@infiniflow.ai>
2026-08-02 17:06:29 +08:00

290 lines
8.3 KiB
Go

//
// Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
package dao
import (
"context"
"fmt"
"os"
"path/filepath"
"ragflow/internal/common"
"ragflow/internal/entity"
"ragflow/internal/entity/models"
"strings"
"sync"
"time"
"ragflow/internal/server"
"go.uber.org/zap"
gormLogger "gorm.io/gorm/logger"
"gorm.io/driver/mysql"
"gorm.io/gorm"
)
var DB *gorm.DB
var modelProviderManager *models.ProviderManager
var modelProviderManagerMu sync.Mutex
// LLMFactoryConfig represents a single LLM factory configuration
type LLMFactoryConfig struct {
Name string `json:"name"`
Logo string `json:"logo"`
Tags string `json:"tags"`
Status string `json:"status"`
Rank string `json:"rank"`
LLM []LLMConfig `json:"llm"`
}
// LLMConfig represents a single LLM model configuration
type LLMConfig struct {
LLMName string `json:"llm_name"`
Tags string `json:"tags"`
MaxTokens int64 `json:"max_tokens"`
ModelType string `json:"model_type"`
IsTools bool `json:"is_tools"`
}
// LLMFactoriesFile represents the structure of llm_factories.json
type LLMFactoriesFile struct {
FactoryLLMInfos []LLMFactoryConfig `json:"factory_llm_infos"`
}
// InitDB initialize database connection
func InitDB(ctx context.Context, migrateDB bool) error {
globalConfig := server.GetConfig()
databaseConfig := globalConfig.GetMySQLConfig()
dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=%s&parseTime=True&loc=Local",
databaseConfig.User,
databaseConfig.Password,
databaseConfig.Host,
databaseConfig.Port,
databaseConfig.DatabaseName,
databaseConfig.Charset,
)
// Set log level
var gormLogLevel gormLogger.LogLevel
if globalConfig.GetMode() == "debug" {
gormLogLevel = gormLogger.Info
} else {
gormLogLevel = gormLogger.Silent
}
// Connect to database
var err error
DB, err = gorm.Open(mysql.Open(dsn), &gorm.Config{
Logger: gormLogger.Default.LogMode(gormLogLevel),
NowFunc: func() time.Time {
return time.Now().Local()
},
TranslateError: true,
})
if err != nil {
return fmt.Errorf("failed to connect database: %w", err)
}
// Get general database object sql.DB
sqlDB, err := DB.DB()
if err != nil {
return fmt.Errorf("failed to get database instance: %w", err)
}
// Set connection pool
sqlDB.SetMaxIdleConns(10)
sqlDB.SetMaxOpenConns(100)
sqlDB.SetConnMaxLifetime(time.Hour)
// Auto migrate all dataModels
dataModels := []interface{}{
&entity.User{},
&entity.Tenant{},
&entity.UserTenant{},
&entity.File{},
&entity.File2Document{},
&entity.TenantLLM{},
&entity.Chat{},
&entity.ChatSession{},
&entity.Task{},
&entity.APIToken{},
&entity.API4Conversation{},
&entity.Knowledgebase{},
&entity.InvitationCode{},
&entity.Document{},
&entity.UserCanvas{},
&entity.CanvasTemplate{},
&entity.UserCanvasVersion{},
&entity.LLMFactories{},
&entity.LLM{},
&entity.TenantLangfuse{},
&entity.SystemSettings{},
&entity.Connector{},
&entity.Connector2Kb{},
&entity.SyncLogs{},
&entity.MCPServer{},
&entity.Memory{},
&entity.Search{},
&entity.PipelineOperationLog{},
&entity.EvaluationDataset{},
&entity.EvaluationCase{},
&entity.EvaluationRun{},
&entity.EvaluationResult{},
&entity.TimeRecord{},
&entity.License{},
&entity.SkillSearchConfig{},
&entity.TenantModelInstance{},
&entity.TenantModel{},
&entity.TenantModelGroupMapping{},
&entity.TenantModelProvider{},
&entity.TenantModelGroup{},
&entity.IngestionTask{},
&entity.IngestionTaskLog{},
&entity.FileCommit{},
&entity.FileCommitItem{},
&entity.KnowledgeCompileDataset{},
// Knowledge-compile compilation templates and their groups. The Go
// KnowledgeCompilerComponent resolves a compilation_template (or group)
// from these tables at runtime, so the Go side must guarantee they exist.
&entity.CompilationTemplate{},
&entity.CompilationTemplateGroup{},
}
if migrateDB {
common.Info("Migrating database schema...")
for _, m := range dataModels {
if err = autoMigrateSafely(ctx, DB, m); err != nil {
return fmt.Errorf("failed to migrate model %T: %w", m, err)
}
}
// Run manual migrations for complex schema changes
if err = RunMigrations(ctx, DB); err != nil {
return fmt.Errorf("failed to run manual migrations: %w", err)
}
common.Info("Database schema migrated successfully")
}
// Seed built-in agent templates so the Go backend can serve the
// "create agent from template" catalogue without relying on Python-side
// initialization.
if err = SeedCanvasTemplates(ctx, DB); err != nil {
common.Warn("Failed to seed canvas templates", zap.Error(err))
}
// Seed the built-in compilation template group (c3aa748c...) for every
// tenant so compiler.json's default group resolves out of the box.
if err = SeedBuiltinCompilationTemplates(ctx, DB); err != nil {
common.Warn("Failed to seed built-in compilation templates", zap.Error(err))
}
common.Info("Database connected and migrated successfully")
err = models.InitProviderManager("conf/models")
if err != nil {
common.Fatal("Failed to load model providers", zap.Error(err))
}
modelProviderManager = models.GetProviderManager()
common.Info("Model providers loaded successfully")
return nil
}
// GetDB get database instance
func GetDB() *gorm.DB {
return DB
}
// GetModelProviderManager get database instance
func GetModelProviderManager() *models.ProviderManager {
if modelProviderManager != nil {
return modelProviderManager
}
modelProviderManagerMu.Lock()
defer modelProviderManagerMu.Unlock()
if modelProviderManager != nil {
return modelProviderManager
}
if existing := models.GetProviderManager(); existing != nil {
modelProviderManager = existing
return modelProviderManager
}
modelConfigDir, err := findModelConfigDir()
if err != nil {
common.Fatal("Failed to locate model providers", zap.Error(err))
}
if err = models.InitProviderManager(modelConfigDir); err != nil {
common.Fatal("Failed to load model providers", zap.Error(err))
}
modelProviderManager = models.GetProviderManager()
return modelProviderManager
}
func findModelConfigDir() (string, error) {
candidates := []string{
"conf/models",
filepath.Join("..", "..", "conf", "models"),
filepath.Join("..", "..", "..", "conf", "models"),
}
for _, candidate := range candidates {
if info, err := os.Stat(candidate); err == nil && info.IsDir() {
return candidate, nil
}
}
return "", fmt.Errorf("conf/models not found")
}
// autoMigrateSafely runs AutoMigrate and ignores duplicate index errors
// This handles cases where indexes already exist (e.g., created by Python backend)
func autoMigrateSafely(ctx context.Context, db *gorm.DB, model interface{}) error {
//err := db.Debug().AutoMigrate(model) // to print debug info
err := db.WithContext(ctx).AutoMigrate(model)
if err == nil {
return nil
}
// Check if error is MySQL duplicate index error (Error 1061)
errStr := err.Error()
if strings.Contains(errStr, "Error 1061") && strings.Contains(errStr, "Duplicate key name") {
common.Warn("Index already exists, skipping", zap.String("error", errStr))
return nil
}
if strings.Contains(errStr, "Error 1060") && strings.Contains(errStr, "Duplicate column name") {
common.Warn("Column already exists, skipping", zap.String("error", errStr))
return nil
}
if strings.Contains(errStr, "Error 1050") && strings.Contains(errStr, "Table") {
common.Warn("Table already exists, skipping", zap.String("error", errStr))
return nil
}
if strings.Contains(errStr, "Error 1091") && strings.Contains(errStr, "Can't DROP") {
common.Warn("Index/column already dropped, skipping", zap.String("error", errStr))
return nil
}
if strings.Contains(errStr, "Error 1138") && strings.Contains(errStr, "Invalid use of NULL") {
common.Warn("NULL value in existing rows, skipping migration change", zap.String("error", errStr))
return nil
}
return err
}