diff --git a/data/business.db b/data/business.db index d68251d..9365533 100644 Binary files a/data/business.db and b/data/business.db differ diff --git a/data/system.db b/data/system.db index 9fd4175..e704f2b 100644 Binary files a/data/system.db and b/data/system.db differ diff --git a/kb/consts/status.go b/kb/consts/status.go index e359873..1f6cb97 100644 --- a/kb/consts/status.go +++ b/kb/consts/status.go @@ -2,10 +2,12 @@ package consts // 文档状态 const ( - DocumentStatusPending = 0 // 待处理 - DocumentStatusParsing = 1 // 处理中 - DocumentStatusDone = 2 // 完成 - DocumentStatusFailed = 3 // 失败 + DocumentStatusPending = 0 // 待处理 + DocumentStatusParsing = 1 // 解析中（读取文件 + 分块） + DocumentStatusEmbedding = 2 // 向量生成中 + DocumentStatusKgBuilding = 3 // 知识图谱构建中 + DocumentStatusDone = 4 // 已完成 + DocumentStatusFailed = 5 // 失败 ) // 解析任务状态 diff --git a/kb/service/chunk_service.go b/kb/service/chunk_service.go index dc53a0b..2dd1d4b 100644 --- a/kb/service/chunk_service.go +++ b/kb/service/chunk_service.go @@ -409,7 +409,6 @@ func (s *chunkService) InsertAll(ctx context.Context, datasetId, documentId int6 } return dao.Document.UpdateFields(ctx, documentId, g.Map{ "chunk_count": len(chunks), - "status": consts.DocumentStatusDone, }) } diff --git a/kb/service/kg_entity_service.go b/kb/service/kg_entity_service.go index ebcbe2d..82f26f3 100644 --- a/kb/service/kg_entity_service.go +++ b/kb/service/kg_entity_service.go @@ -35,23 +35,24 @@ type kgExtractResult struct { } // ExtractDocument 对文档全部新分块做 LLM 抽取（挂在解析流水线分块落库之后）。 -// 每个分块一次调用，任何失败只记日志，不阻断解析流水线；未配置默认对话模型时直接跳过。 -func (s *kgEntityService) ExtractDocument(ctx context.Context, datasetId, documentId int64) error { +// 每个分块一次调用，任何失败只记日志，不阻断解析流水线；返回未成功抽取的分块数，供调用方标记图谱未构建。 +func (s *kgEntityService) ExtractDocument(ctx context.Context, datasetId, documentId int64) (int, error) { chunks, _, err := dao.Chunk.ListByDocument(ctx, documentId, 1, 100000) if err != nil { - return err + return 0, err } model, err := s.buildModel(ctx) if err != nil { - g.Log().Infof(ctx, "kg extract skipped: %v", err) - return nil + return 0, err } + failed := 0 for _, c := range chunks { if err := s.extractChunk(ctx, model, datasetId, c); err != nil { + failed++ g.Log().Warningf(ctx, "kg extract chunk %d failed: %v", c.Id, err) } } - return nil + return failed, nil } func (s *kgEntityService) buildModel(ctx context.Context) (*OpenAIChatModel, error) { diff --git a/kb/service/parse_task_service.go b/kb/service/parse_task_service.go index f223cdf..18bdc0e 100644 --- a/kb/service/parse_task_service.go +++ b/kb/service/parse_task_service.go @@ -2,6 +2,7 @@ package service import ( "context" + "fmt" "path/filepath" "time" @@ -117,15 +118,33 @@ func (s *parseTaskService) processOne(ctx context.Context) { g.Log().Warningf(ctx, "save detected structure patterns failed: %v", err) } } + if err := dao.Document.UpdateFields(ctx, doc.Id, g.Map{"status": consts.DocumentStatusEmbedding}); err != nil { + s.fail(ctx, task, "更新文档状态失败: "+err.Error()) + return + } if err := ChunkService.InsertAll(ctx, task.DatasetId, doc.Id, chunks, embedder); err != nil { s.fail(ctx, task, "写入分块失败: "+err.Error()) return } - // 第 3.5 步：知识图谱抽取（失败不阻断流水线） - if err := KgEntityService.ExtractDocument(ctx, task.DatasetId, doc.Id); err != nil { - g.Log().Warningf(ctx, "kg extract skipped for doc %d: %v", doc.Id, err) + // 第 3.5 步：知识图谱抽取（失败不阻断流水线，但文档标记图谱未构建，避免误报已完成） + if err := dao.Document.UpdateFields(ctx, doc.Id, g.Map{"status": consts.DocumentStatusKgBuilding}); err != nil { + g.Log().Warningf(ctx, "mark kg building failed: %v", err) } - if err := dao.ParseTask.UpdateStatus(ctx, task.Id, consts.TaskStatusDone, ""); err != nil { + kgFailed, kgErr := KgEntityService.ExtractDocument(ctx, task.DatasetId, doc.Id) + kgMsg := "" + switch { + case kgErr != nil: + kgMsg = "知识图谱未构建: " + kgErr.Error() + case kgFailed > 0: + kgMsg = fmt.Sprintf("知识图谱未构建：%d 个分块抽取失败", kgFailed) + } + if kgMsg != "" { + g.Log().Warningf(ctx, "kg extract incomplete for doc %d: %s", doc.Id, kgMsg) + } + if err := dao.Document.UpdateFields(ctx, doc.Id, g.Map{"status": consts.DocumentStatusDone, "error_msg": kgMsg}); err != nil { + g.Log().Warningf(ctx, "mark document done failed: %v", err) + } + if err := dao.ParseTask.UpdateStatus(ctx, task.Id, consts.TaskStatusDone, kgMsg); err != nil { g.Log().Errorf(ctx, "mark task done failed: %v", err) } } diff --git a/ui-src/src/views/DatasetDetail.vue b/ui-src/src/views/DatasetDetail.vue index 396de11..a450c00 100644 --- a/ui-src/src/views/DatasetDetail.vue +++ b/ui-src/src/views/DatasetDetail.vue @@ -23,6 +23,7 @@ {{ statusText(row.status) }} + 图谱未构建 @@ -30,7 +31,7 @@ @@ -126,8 +127,10 @@ let editingChunk = null const statusMap = { 0: { text: '待处理', type: 'info' }, 1: { text: '解析中', type: 'warning' }, - 2: { text: '已完成', type: 'success' }, - 3: { text: '失败', type: 'danger' } + 2: { text: '向量生成中', type: 'warning' }, + 3: { text: '图谱构建中', type: 'warning' }, + 4: { text: '已完成', type: 'success' }, + 5: { text: '失败', type: 'danger' } } function statusText(s) { return (statusMap[s] || {}).text || s } @@ -172,7 +175,7 @@ async function startPolling() { stopPolling() pollTimer = setInterval(async () => { await load() - if (!documents.value.some(d => d.status === 0 || d.status === 1)) stopPolling() + if (!documents.value.some(d => d.status < 4)) stopPolling() }, 3000) } function stopPolling() { @@ -189,7 +192,7 @@ onMounted(async () => { datasetName.value = cur ? cur.name : '数据集' } catch { /* 忽略 */ } await load() - if (documents.value.some(d => d.status === 0 || d.status === 1)) startPolling() + if (documents.value.some(d => d.status < 4)) startPolling() }) onUnmounted(stopPolling)