root
0a7b216513
Poll the knowledge item list every 5s while any item is still pending or processing so the UI reflects ingestion progress without manual refresh, and exclude items that have no active chunks from group counters so groups with no usable content do not appear populated. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2947 lines
85 KiB
Go
2947 lines
85 KiB
Go
package app
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"net/http"
|
||
"net/url"
|
||
"path/filepath"
|
||
"runtime/debug"
|
||
"slices"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
"unicode/utf8"
|
||
|
||
"github.com/google/uuid"
|
||
"github.com/jackc/pgx/v5"
|
||
"github.com/jackc/pgx/v5/pgxpool"
|
||
"go.uber.org/zap"
|
||
|
||
"github.com/geo-platform/tenant-api/internal/shared/auth"
|
||
"github.com/geo-platform/tenant-api/internal/shared/config"
|
||
"github.com/geo-platform/tenant-api/internal/shared/llm"
|
||
"github.com/geo-platform/tenant-api/internal/shared/objectstorage"
|
||
"github.com/geo-platform/tenant-api/internal/shared/response"
|
||
"github.com/geo-platform/tenant-api/internal/shared/retrieval"
|
||
"github.com/geo-platform/tenant-api/internal/tenant/prompts"
|
||
)
|
||
|
||
const (
|
||
defaultKnowledgeQueueSize = 128
|
||
defaultKnowledgeCleanupPoll = 15 * time.Second
|
||
defaultKnowledgeChildGroupName = "默认目录"
|
||
maxKnowledgeSnippetsPerItem = 2
|
||
maxKnowledgeTextNameRunes = 20
|
||
)
|
||
|
||
type KnowledgeService struct {
|
||
pool *pgxpool.Pool
|
||
provider retrieval.Provider
|
||
store retrieval.VectorStore
|
||
objectStorage objectstorage.Client
|
||
llmClient llm.Client
|
||
logger *zap.Logger
|
||
cfg knowledgeRuntimeConfig
|
||
configProvider runtimeConfigProvider
|
||
httpClient *http.Client
|
||
jobs chan knowledgeParseJob
|
||
cleanupMu sync.Mutex
|
||
cleanupInFlight map[string]struct{}
|
||
}
|
||
|
||
type knowledgeParseJob struct {
|
||
TenantID int64
|
||
ItemID int64
|
||
ParseTaskID int64
|
||
}
|
||
|
||
type knowledgeCleanupJob struct {
|
||
TenantID int64
|
||
ItemID int64
|
||
}
|
||
|
||
type knowledgeItemRecord struct {
|
||
ID int64
|
||
TenantID int64
|
||
GroupID int64
|
||
GroupName string
|
||
Name string
|
||
SourceType string
|
||
SourceURI *string
|
||
StorageKey string
|
||
ContentText *string
|
||
Markdown *string
|
||
ItemVersion int
|
||
}
|
||
|
||
type KnowledgeGroupRequest struct {
|
||
Name string `json:"name" binding:"required"`
|
||
ParentID *int64 `json:"parent_id"`
|
||
SortOrder *int `json:"sort_order"`
|
||
}
|
||
|
||
type KnowledgeGroupResponse struct {
|
||
ID int64 `json:"id"`
|
||
Name string `json:"name"`
|
||
ParentID *int64 `json:"parent_id"`
|
||
SortOrder int `json:"sort_order"`
|
||
ItemCount int64 `json:"item_count"`
|
||
CreatedAt string `json:"created_at"`
|
||
Children []KnowledgeGroupResponse `json:"children"`
|
||
}
|
||
|
||
type KnowledgeItemResponse struct {
|
||
ID int64 `json:"id"`
|
||
GroupID int64 `json:"group_id"`
|
||
GroupName string `json:"group_name"`
|
||
Name string `json:"name"`
|
||
SourceType string `json:"source_type"`
|
||
SourceURI *string `json:"source_uri"`
|
||
Status string `json:"status"`
|
||
SizeBytes int64 `json:"size_bytes"`
|
||
ErrorMessage *string `json:"error_message"`
|
||
CreatedAt string `json:"created_at"`
|
||
UpdatedAt string `json:"updated_at"`
|
||
}
|
||
|
||
type KnowledgeItemDetailResponse struct {
|
||
ID int64 `json:"id"`
|
||
GroupID int64 `json:"group_id"`
|
||
GroupName string `json:"group_name"`
|
||
Name string `json:"name"`
|
||
SourceType string `json:"source_type"`
|
||
SourceURI *string `json:"source_uri"`
|
||
Status string `json:"status"`
|
||
SizeBytes int64 `json:"size_bytes"`
|
||
ErrorMessage *string `json:"error_message"`
|
||
MarkdownContent *string `json:"markdown_content"`
|
||
CreatedAt string `json:"created_at"`
|
||
UpdatedAt string `json:"updated_at"`
|
||
}
|
||
|
||
type KnowledgeTextItemRequest struct {
|
||
GroupID int64 `json:"group_id" binding:"required"`
|
||
Name string `json:"name" binding:"required"`
|
||
Content string `json:"content" binding:"required"`
|
||
}
|
||
|
||
type KnowledgeURLItemRequest struct {
|
||
GroupID int64 `json:"group_id" binding:"required"`
|
||
Name string `json:"name" binding:"required"`
|
||
URL string `json:"url" binding:"required"`
|
||
}
|
||
|
||
type KnowledgeContext struct {
|
||
GroupIDs []int64
|
||
GroupNames []string
|
||
Snippets []KnowledgeSnippet
|
||
PreciseFacts []string
|
||
}
|
||
|
||
type KnowledgeSnippet struct {
|
||
PointID string
|
||
GroupID int64
|
||
GroupName string
|
||
KnowledgeItemID int64
|
||
ItemName string
|
||
SourceType string
|
||
SourceURI *string
|
||
Text string
|
||
Score float64
|
||
PreciseFacts []string
|
||
}
|
||
|
||
func NewKnowledgeService(
|
||
pool *pgxpool.Pool,
|
||
provider retrieval.Provider,
|
||
store retrieval.VectorStore,
|
||
objectStorage objectstorage.Client,
|
||
llmClient llm.Client,
|
||
logger *zap.Logger,
|
||
cfg config.RetrievalConfig,
|
||
llmCfg config.LLMConfig,
|
||
queueSize int,
|
||
workerCount int,
|
||
) *KnowledgeService {
|
||
svc := &KnowledgeService{
|
||
pool: pool,
|
||
provider: provider,
|
||
store: store,
|
||
objectStorage: objectStorage,
|
||
llmClient: llmClient,
|
||
logger: logger,
|
||
cfg: knowledgeRuntimeFromConfig(cfg, llmCfg),
|
||
httpClient: &http.Client{Timeout: defaultKnowledgeURLParseTimeout},
|
||
cleanupInFlight: make(map[string]struct{}),
|
||
}
|
||
|
||
if workerCount > 0 {
|
||
if queueSize <= 0 {
|
||
queueSize = defaultKnowledgeQueueSize
|
||
}
|
||
svc.jobs = make(chan knowledgeParseJob, queueSize)
|
||
for i := 0; i < workerCount; i++ {
|
||
go svc.runParseWorker()
|
||
}
|
||
go svc.recoverPendingParseTasks()
|
||
}
|
||
|
||
return svc
|
||
}
|
||
|
||
func (s *KnowledgeService) WithConfigProvider(provider runtimeConfigProvider) *KnowledgeService {
|
||
s.configProvider = provider
|
||
return s
|
||
}
|
||
|
||
func (s *KnowledgeService) runtimeConfig() knowledgeRuntimeConfig {
|
||
if s != nil && s.configProvider != nil {
|
||
if cfg := s.configProvider.Current(); cfg != nil {
|
||
return knowledgeRuntimeFromConfig(cfg.Retrieval, cfg.LLM)
|
||
}
|
||
}
|
||
if s == nil {
|
||
return knowledgeRuntimeFromConfig(config.RetrievalConfig{}, config.LLMConfig{})
|
||
}
|
||
return s.cfg
|
||
}
|
||
|
||
func (s *KnowledgeService) ListGroups(ctx context.Context) ([]KnowledgeGroupResponse, error) {
|
||
actor := auth.MustActor(ctx)
|
||
rows, err := s.pool.Query(ctx, `
|
||
SELECT id, name, parent_id, sort_order, created_at
|
||
FROM knowledge_groups
|
||
WHERE tenant_id = $1 AND deleted_at IS NULL
|
||
ORDER BY sort_order, created_at, id
|
||
`, actor.TenantID)
|
||
if err != nil {
|
||
return nil, response.ErrInternal(50050, "knowledge_groups_query_failed", "failed to list knowledge groups")
|
||
}
|
||
defer rows.Close()
|
||
|
||
nodeMap := make(map[int64]*KnowledgeGroupResponse)
|
||
parentMap := make(map[int64][]int64)
|
||
orderedIDs := make([]int64, 0)
|
||
for rows.Next() {
|
||
node := &KnowledgeGroupResponse{}
|
||
var createdAt time.Time
|
||
if err := rows.Scan(&node.ID, &node.Name, &node.ParentID, &node.SortOrder, &createdAt); err != nil {
|
||
return nil, response.ErrInternal(50050, "knowledge_groups_scan_failed", err.Error())
|
||
}
|
||
node.CreatedAt = createdAt.Format(time.RFC3339)
|
||
node.Children = []KnowledgeGroupResponse{}
|
||
nodeMap[node.ID] = node
|
||
orderedIDs = append(orderedIDs, node.ID)
|
||
if node.ParentID != nil {
|
||
parentMap[*node.ParentID] = append(parentMap[*node.ParentID], node.ID)
|
||
}
|
||
}
|
||
|
||
countRows, err := s.pool.Query(ctx, `
|
||
SELECT ki.group_id, COUNT(DISTINCT ki.id)::bigint
|
||
FROM knowledge_items ki
|
||
WHERE ki.tenant_id = $1
|
||
AND ki.deleted_at IS NULL
|
||
AND ki.status = 'completed'
|
||
AND EXISTS (
|
||
SELECT 1
|
||
FROM knowledge_chunks_meta kcm
|
||
WHERE kcm.tenant_id = ki.tenant_id
|
||
AND kcm.knowledge_item_id = ki.id
|
||
AND kcm.item_version = ki.item_version
|
||
AND kcm.status = 'active'
|
||
)
|
||
GROUP BY ki.group_id
|
||
`, actor.TenantID)
|
||
if err != nil {
|
||
return nil, response.ErrInternal(50050, "knowledge_group_counts_query_failed", "failed to list knowledge group counts")
|
||
}
|
||
defer countRows.Close()
|
||
|
||
directCounts := make(map[int64]int64)
|
||
for countRows.Next() {
|
||
var groupID, itemCount int64
|
||
if err := countRows.Scan(&groupID, &itemCount); err != nil {
|
||
return nil, response.ErrInternal(50050, "knowledge_group_counts_scan_failed", err.Error())
|
||
}
|
||
directCounts[groupID] = itemCount
|
||
}
|
||
|
||
var sumCounts func(int64) int64
|
||
sumCounts = func(groupID int64) int64 {
|
||
total := directCounts[groupID]
|
||
for _, childID := range parentMap[groupID] {
|
||
total += sumCounts(childID)
|
||
}
|
||
if node := nodeMap[groupID]; node != nil {
|
||
node.ItemCount = total
|
||
}
|
||
return total
|
||
}
|
||
|
||
var buildTree func(int64) KnowledgeGroupResponse
|
||
buildTree = func(groupID int64) KnowledgeGroupResponse {
|
||
node := nodeMap[groupID]
|
||
if node == nil {
|
||
return KnowledgeGroupResponse{}
|
||
}
|
||
result := *node
|
||
childIDs := parentMap[groupID]
|
||
result.Children = make([]KnowledgeGroupResponse, 0, len(childIDs))
|
||
for _, childID := range childIDs {
|
||
result.Children = append(result.Children, buildTree(childID))
|
||
}
|
||
return result
|
||
}
|
||
|
||
roots := make([]KnowledgeGroupResponse, 0)
|
||
for _, groupID := range orderedIDs {
|
||
node := nodeMap[groupID]
|
||
if node == nil {
|
||
continue
|
||
}
|
||
if node.ParentID == nil {
|
||
sumCounts(groupID)
|
||
roots = append(roots, buildTree(groupID))
|
||
}
|
||
}
|
||
|
||
return roots, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) CreateGroup(ctx context.Context, req KnowledgeGroupRequest) (*KnowledgeGroupResponse, error) {
|
||
actor := auth.MustActor(ctx)
|
||
|
||
name := strings.TrimSpace(req.Name)
|
||
if name == "" {
|
||
return nil, response.ErrBadRequest(40050, "knowledge_group_name_required", "knowledge group name is required")
|
||
}
|
||
|
||
var parentID *int64
|
||
if req.ParentID != nil && *req.ParentID > 0 {
|
||
if err := s.ensureRootGroup(ctx, actor.TenantID, *req.ParentID); err != nil {
|
||
return nil, err
|
||
}
|
||
parentID = req.ParentID
|
||
}
|
||
|
||
sortOrder := 0
|
||
if req.SortOrder != nil {
|
||
sortOrder = *req.SortOrder
|
||
}
|
||
|
||
var (
|
||
groupID int64
|
||
createdAt time.Time
|
||
)
|
||
err := s.pool.QueryRow(ctx, `
|
||
INSERT INTO knowledge_groups (tenant_id, name, parent_id, sort_order)
|
||
VALUES ($1, $2, $3, $4)
|
||
RETURNING id, created_at
|
||
`, actor.TenantID, name, parentID, sortOrder).Scan(&groupID, &createdAt)
|
||
if err != nil {
|
||
return nil, response.ErrConflict(40950, "knowledge_group_create_failed", "knowledge group already exists or cannot be created")
|
||
}
|
||
|
||
return &KnowledgeGroupResponse{
|
||
ID: groupID,
|
||
Name: name,
|
||
ParentID: parentID,
|
||
SortOrder: sortOrder,
|
||
ItemCount: 0,
|
||
CreatedAt: createdAt.Format(time.RFC3339),
|
||
Children: []KnowledgeGroupResponse{},
|
||
}, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) UpdateGroup(ctx context.Context, groupID int64, req KnowledgeGroupRequest) error {
|
||
actor := auth.MustActor(ctx)
|
||
|
||
name := strings.TrimSpace(req.Name)
|
||
if name == "" {
|
||
return response.ErrBadRequest(40050, "knowledge_group_name_required", "knowledge group name is required")
|
||
}
|
||
|
||
sortOrder := 0
|
||
if req.SortOrder != nil {
|
||
sortOrder = *req.SortOrder
|
||
}
|
||
|
||
tag, err := s.pool.Exec(ctx, `
|
||
UPDATE knowledge_groups
|
||
SET name = $1, sort_order = $2, updated_at = NOW()
|
||
WHERE id = $3 AND tenant_id = $4 AND deleted_at IS NULL
|
||
`, name, sortOrder, groupID, actor.TenantID)
|
||
if err != nil {
|
||
return response.ErrConflict(40950, "knowledge_group_update_failed", "knowledge group already exists or cannot be updated")
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return response.ErrNotFound(40451, "knowledge_group_not_found", "knowledge group not found")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *KnowledgeService) DeleteGroup(ctx context.Context, groupID int64) error {
|
||
actor := auth.MustActor(ctx)
|
||
|
||
tx, err := s.pool.Begin(ctx)
|
||
if err != nil {
|
||
return response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to begin delete transaction")
|
||
}
|
||
defer func() { _ = tx.Rollback(ctx) }()
|
||
|
||
groupIDs, err := s.listGroupSubtreeIDs(ctx, tx, actor.TenantID, groupID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(groupIDs) == 0 {
|
||
return response.ErrNotFound(40451, "knowledge_group_not_found", "knowledge group not found")
|
||
}
|
||
|
||
itemIDs, err := s.listActiveKnowledgeItemIDs(ctx, tx, actor.TenantID, groupIDs)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
if len(itemIDs) > 0 {
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE knowledge_items
|
||
SET deleted_at = NOW(), updated_at = NOW(), status = 'deleted'
|
||
WHERE tenant_id = $1
|
||
AND deleted_at IS NULL
|
||
AND id = ANY($2::bigint[])
|
||
`, actor.TenantID, itemIDs); err != nil {
|
||
return response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to delete knowledge items")
|
||
}
|
||
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE knowledge_parse_tasks
|
||
SET status = 'deleted',
|
||
error_message = 'knowledge group deleted',
|
||
completed_at = COALESCE(completed_at, NOW()),
|
||
updated_at = NOW()
|
||
WHERE tenant_id = $1
|
||
AND knowledge_item_id = ANY($2::bigint[])
|
||
AND status <> 'deleted'
|
||
`, actor.TenantID, itemIDs); err != nil {
|
||
return response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to cancel knowledge parse tasks")
|
||
}
|
||
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE knowledge_chunks_meta
|
||
SET status = 'deleting', updated_at = NOW()
|
||
WHERE tenant_id = $1
|
||
AND knowledge_item_id = ANY($2::bigint[])
|
||
AND status <> 'deleted'
|
||
`, actor.TenantID, itemIDs); err != nil {
|
||
return response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to mark knowledge chunks for deletion")
|
||
}
|
||
}
|
||
|
||
if _, err := tx.Exec(ctx, `
|
||
DELETE FROM prompt_rule_knowledge_groups
|
||
WHERE knowledge_group_id = ANY($1::bigint[])
|
||
`, groupIDs); err != nil {
|
||
return response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to delete knowledge group references")
|
||
}
|
||
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE knowledge_groups
|
||
SET deleted_at = NOW(), updated_at = NOW()
|
||
WHERE tenant_id = $1
|
||
AND deleted_at IS NULL
|
||
AND id = ANY($2::bigint[])
|
||
`, actor.TenantID, groupIDs); err != nil {
|
||
return response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to delete knowledge groups")
|
||
}
|
||
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to commit delete transaction")
|
||
}
|
||
|
||
for _, itemID := range itemIDs {
|
||
go s.cleanupDeletedKnowledgeItemAsync(actor.TenantID, itemID)
|
||
}
|
||
|
||
if s.logger != nil {
|
||
s.logger.Info("knowledge group soft deleted",
|
||
zap.Int64("tenant_id", actor.TenantID),
|
||
zap.Int64("knowledge_group_id", groupID),
|
||
zap.Int("group_count", len(groupIDs)),
|
||
zap.Int("item_count", len(itemIDs)),
|
||
)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *KnowledgeService) ListItems(ctx context.Context, groupID *int64) ([]KnowledgeItemResponse, error) {
|
||
actor := auth.MustActor(ctx)
|
||
if groupID != nil && *groupID > 0 {
|
||
if err := s.ensureGroupAccessible(ctx, actor.TenantID, *groupID); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
query := `
|
||
SELECT ki.id, ki.group_id, kg.name, ki.name, ki.source_type, ki.source_uri, ki.status,
|
||
ki.size_bytes, ki.error_message, ki.created_at, ki.updated_at
|
||
FROM knowledge_items ki
|
||
JOIN knowledge_groups kg ON kg.id = ki.group_id AND kg.deleted_at IS NULL
|
||
WHERE ki.tenant_id = $1
|
||
AND ki.deleted_at IS NULL
|
||
`
|
||
args := []any{actor.TenantID}
|
||
if groupID != nil && *groupID > 0 {
|
||
query += `
|
||
AND ki.group_id IN (
|
||
WITH RECURSIVE selected_groups AS (
|
||
SELECT id
|
||
FROM knowledge_groups
|
||
WHERE id = $2 AND tenant_id = $1 AND deleted_at IS NULL
|
||
UNION ALL
|
||
SELECT child.id
|
||
FROM knowledge_groups child
|
||
JOIN selected_groups parent ON child.parent_id = parent.id
|
||
WHERE child.tenant_id = $1 AND child.deleted_at IS NULL
|
||
)
|
||
SELECT id FROM selected_groups
|
||
)
|
||
`
|
||
args = append(args, *groupID)
|
||
}
|
||
query += ` ORDER BY ki.created_at DESC, ki.id DESC`
|
||
|
||
rows, err := s.pool.Query(ctx, query, args...)
|
||
if err != nil {
|
||
return nil, response.ErrInternal(50051, "knowledge_items_query_failed", "failed to list knowledge items")
|
||
}
|
||
defer rows.Close()
|
||
|
||
items := make([]KnowledgeItemResponse, 0)
|
||
for rows.Next() {
|
||
var (
|
||
item KnowledgeItemResponse
|
||
sourceURI *string
|
||
errorMessage *string
|
||
createdAt, updatedAt time.Time
|
||
)
|
||
if err := rows.Scan(
|
||
&item.ID,
|
||
&item.GroupID,
|
||
&item.GroupName,
|
||
&item.Name,
|
||
&item.SourceType,
|
||
&sourceURI,
|
||
&item.Status,
|
||
&item.SizeBytes,
|
||
&errorMessage,
|
||
&createdAt,
|
||
&updatedAt,
|
||
); err != nil {
|
||
return nil, response.ErrInternal(50051, "knowledge_items_scan_failed", err.Error())
|
||
}
|
||
item.SourceURI = sourceURI
|
||
item.ErrorMessage = errorMessage
|
||
item.CreatedAt = createdAt.Format(time.RFC3339)
|
||
item.UpdatedAt = updatedAt.Format(time.RFC3339)
|
||
items = append(items, item)
|
||
}
|
||
|
||
return items, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) GetItemDetail(ctx context.Context, itemID int64) (*KnowledgeItemDetailResponse, error) {
|
||
actor := auth.MustActor(ctx)
|
||
|
||
var (
|
||
item KnowledgeItemDetailResponse
|
||
sourceURI *string
|
||
errorMessage *string
|
||
markdownContent *string
|
||
createdAt, updatedAt time.Time
|
||
)
|
||
|
||
err := s.pool.QueryRow(ctx, `
|
||
SELECT ki.id, ki.group_id, kg.name, ki.name, ki.source_type, ki.source_uri, ki.status,
|
||
ki.size_bytes, ki.error_message, ki.markdown_content, ki.created_at, ki.updated_at
|
||
FROM knowledge_items ki
|
||
JOIN knowledge_groups kg ON kg.id = ki.group_id AND kg.deleted_at IS NULL
|
||
WHERE ki.id = $1 AND ki.tenant_id = $2 AND ki.deleted_at IS NULL
|
||
`, itemID, actor.TenantID).Scan(
|
||
&item.ID,
|
||
&item.GroupID,
|
||
&item.GroupName,
|
||
&item.Name,
|
||
&item.SourceType,
|
||
&sourceURI,
|
||
&item.Status,
|
||
&item.SizeBytes,
|
||
&errorMessage,
|
||
&markdownContent,
|
||
&createdAt,
|
||
&updatedAt,
|
||
)
|
||
if err != nil {
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, response.ErrNotFound(40450, "knowledge_item_not_found", "knowledge item not found")
|
||
}
|
||
return nil, response.ErrInternal(50051, "knowledge_items_query_failed", "failed to load knowledge item")
|
||
}
|
||
|
||
item.SourceURI = sourceURI
|
||
item.ErrorMessage = errorMessage
|
||
item.MarkdownContent = markdownContent
|
||
item.CreatedAt = createdAt.Format(time.RFC3339)
|
||
item.UpdatedAt = updatedAt.Format(time.RFC3339)
|
||
return &item, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) CreateTextItem(ctx context.Context, req KnowledgeTextItemRequest) (*KnowledgeItemResponse, error) {
|
||
actor := auth.MustActor(ctx)
|
||
|
||
if err := s.validateIngestionDependencies("text"); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
content := normalizeKnowledgeText(req.Content)
|
||
if content == "" {
|
||
return nil, response.ErrBadRequest(40051, "knowledge_content_required", "knowledge content is required")
|
||
}
|
||
|
||
name := strings.TrimSpace(req.Name)
|
||
if utf8.RuneCountInString(name) > maxKnowledgeTextNameRunes {
|
||
return nil, response.ErrBadRequest(40055, "knowledge_text_name_too_long", "knowledge text name must be at most 20 characters")
|
||
}
|
||
|
||
return s.queueKnowledgeItem(ctx, actor.TenantID, knowledgeIngestInput{
|
||
GroupID: req.GroupID,
|
||
Name: name,
|
||
SourceType: "text",
|
||
ContentText: &content,
|
||
SizeBytes: int64(len([]byte(content))),
|
||
})
|
||
}
|
||
|
||
func (s *KnowledgeService) CreateURLItem(ctx context.Context, req KnowledgeURLItemRequest) (*KnowledgeItemResponse, error) {
|
||
actor := auth.MustActor(ctx)
|
||
|
||
if err := s.validateIngestionDependencies("website"); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
rawURL := strings.TrimSpace(req.URL)
|
||
if _, err := url.ParseRequestURI(rawURL); err != nil {
|
||
return nil, response.ErrBadRequest(40052, "knowledge_url_invalid", "knowledge url is invalid")
|
||
}
|
||
|
||
return s.queueKnowledgeItem(ctx, actor.TenantID, knowledgeIngestInput{
|
||
GroupID: req.GroupID,
|
||
Name: strings.TrimSpace(req.Name),
|
||
SourceType: "website",
|
||
SourceURI: &rawURL,
|
||
})
|
||
}
|
||
|
||
func (s *KnowledgeService) CreateFileItem(
|
||
ctx context.Context,
|
||
groupID int64,
|
||
name string,
|
||
fileName string,
|
||
content []byte,
|
||
) (*KnowledgeItemResponse, error) {
|
||
actor := auth.MustActor(ctx)
|
||
|
||
if err := s.validateIngestionDependencies("document"); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
displayName := strings.TrimSpace(name)
|
||
if displayName == "" {
|
||
displayName = strings.TrimSuffix(filepath.Base(fileName), filepath.Ext(fileName))
|
||
}
|
||
|
||
objectKey := buildKnowledgeObjectKey(actor.TenantID, "document", fileName)
|
||
if err := s.objectStorage.PutBytes(ctx, objectKey, content, detectKnowledgeContentType(fileName)); err != nil {
|
||
return nil, response.ErrServiceUnavailable(50357, "knowledge_storage_unavailable", err.Error())
|
||
}
|
||
|
||
return s.queueKnowledgeItem(ctx, actor.TenantID, knowledgeIngestInput{
|
||
GroupID: groupID,
|
||
Name: displayName,
|
||
SourceType: "document",
|
||
StorageKey: objectKey,
|
||
SizeBytes: int64(len(content)),
|
||
})
|
||
}
|
||
|
||
func (s *KnowledgeService) DeleteItem(ctx context.Context, itemID int64) error {
|
||
actor := auth.MustActor(ctx)
|
||
|
||
tx, err := s.pool.Begin(ctx)
|
||
if err != nil {
|
||
return response.ErrInternal(50054, "knowledge_item_delete_failed", "failed to begin delete transaction")
|
||
}
|
||
defer func() { _ = tx.Rollback(ctx) }()
|
||
|
||
tag, err := tx.Exec(ctx, `
|
||
UPDATE knowledge_items
|
||
SET deleted_at = NOW(), updated_at = NOW(), status = 'deleted'
|
||
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
|
||
`, itemID, actor.TenantID)
|
||
if err != nil {
|
||
return response.ErrInternal(50054, "knowledge_item_delete_failed", "failed to delete knowledge item")
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return response.ErrNotFound(40450, "knowledge_item_not_found", "knowledge item not found")
|
||
}
|
||
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE knowledge_parse_tasks
|
||
SET status = 'deleted',
|
||
error_message = 'knowledge item deleted',
|
||
completed_at = COALESCE(completed_at, NOW()),
|
||
updated_at = NOW()
|
||
WHERE knowledge_item_id = $1
|
||
AND tenant_id = $2
|
||
AND status IN ('pending', 'running')
|
||
`, itemID, actor.TenantID); err != nil {
|
||
return response.ErrInternal(50054, "knowledge_item_delete_failed", "failed to cancel knowledge parse task")
|
||
}
|
||
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE knowledge_chunks_meta
|
||
SET status = 'deleting', updated_at = NOW()
|
||
WHERE knowledge_item_id = $1 AND tenant_id = $2 AND status <> 'deleted'
|
||
`, itemID, actor.TenantID); err != nil {
|
||
return response.ErrInternal(50054, "knowledge_item_delete_failed", "failed to mark knowledge chunks for deletion")
|
||
}
|
||
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return response.ErrInternal(50054, "knowledge_item_delete_failed", "failed to commit delete transaction")
|
||
}
|
||
|
||
if s.logger != nil {
|
||
s.logger.Info("knowledge item soft deleted",
|
||
zap.Int64("tenant_id", actor.TenantID),
|
||
zap.Int64("knowledge_item_id", itemID),
|
||
)
|
||
}
|
||
go s.cleanupDeletedKnowledgeItemAsync(actor.TenantID, itemID)
|
||
return nil
|
||
}
|
||
|
||
func (s *KnowledgeService) RetryItem(ctx context.Context, itemID int64) (*KnowledgeItemResponse, error) {
|
||
actor := auth.MustActor(ctx)
|
||
|
||
var sourceType string
|
||
var status string
|
||
err := s.pool.QueryRow(ctx, `
|
||
SELECT source_type, status
|
||
FROM knowledge_items
|
||
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
|
||
`, itemID, actor.TenantID).Scan(&sourceType, &status)
|
||
if err != nil {
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, response.ErrNotFound(40450, "knowledge_item_not_found", "knowledge item not found")
|
||
}
|
||
return nil, response.ErrInternal(50051, "knowledge_items_query_failed", "failed to load knowledge item")
|
||
}
|
||
if status != "failed" {
|
||
return nil, response.ErrConflict(40952, "knowledge_item_retry_unavailable", "only failed knowledge items can be retried")
|
||
}
|
||
if err := s.validateIngestionDependencies(sourceType); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
tx, err := s.pool.Begin(ctx)
|
||
if err != nil {
|
||
return nil, response.ErrInternal(50055, "knowledge_item_retry_failed", "failed to begin retry transaction")
|
||
}
|
||
defer func() { _ = tx.Rollback(ctx) }()
|
||
|
||
var parseTaskID int64
|
||
err = tx.QueryRow(ctx, `
|
||
UPDATE knowledge_parse_tasks
|
||
SET status = 'pending',
|
||
error_message = NULL,
|
||
started_at = NULL,
|
||
completed_at = NULL,
|
||
updated_at = NOW()
|
||
WHERE id = (
|
||
SELECT id
|
||
FROM knowledge_parse_tasks
|
||
WHERE tenant_id = $1
|
||
AND knowledge_item_id = $2
|
||
ORDER BY created_at DESC, id DESC
|
||
LIMIT 1
|
||
)
|
||
RETURNING id
|
||
`, actor.TenantID, itemID).Scan(&parseTaskID)
|
||
if err != nil {
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, response.ErrConflict(40952, "knowledge_item_retry_unavailable", "knowledge parse task is unavailable")
|
||
}
|
||
return nil, response.ErrInternal(50055, "knowledge_item_retry_failed", "failed to reset knowledge parse task")
|
||
}
|
||
|
||
tag, err := tx.Exec(ctx, `
|
||
UPDATE knowledge_items
|
||
SET status = 'pending',
|
||
error_message = NULL,
|
||
updated_at = NOW()
|
||
WHERE id = $1
|
||
AND tenant_id = $2
|
||
AND deleted_at IS NULL
|
||
AND status = 'failed'
|
||
`, itemID, actor.TenantID)
|
||
if err != nil {
|
||
return nil, response.ErrInternal(50055, "knowledge_item_retry_failed", "failed to reset knowledge item")
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return nil, response.ErrConflict(40952, "knowledge_item_retry_unavailable", "only failed knowledge items can be retried")
|
||
}
|
||
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return nil, response.ErrInternal(50055, "knowledge_item_retry_failed", "failed to commit retry transaction")
|
||
}
|
||
|
||
if err := s.enqueueParseJob(knowledgeParseJob{
|
||
TenantID: actor.TenantID,
|
||
ItemID: itemID,
|
||
ParseTaskID: parseTaskID,
|
||
}); err != nil {
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), actor.TenantID, itemID, parseTaskID, err)
|
||
return nil, response.ErrServiceUnavailable(50356, "knowledge_queue_unavailable", "knowledge parse queue is busy, please retry")
|
||
}
|
||
|
||
items, err := s.listKnowledgeItemsByIDs(ctx, actor.TenantID, []int64{itemID})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(items) == 0 {
|
||
return nil, response.ErrNotFound(40450, "knowledge_item_not_found", "knowledge item not found")
|
||
}
|
||
return &items[0], nil
|
||
}
|
||
|
||
func (s *KnowledgeService) ResolveContext(
|
||
ctx context.Context,
|
||
tenantID int64,
|
||
groupIDs []int64,
|
||
query string,
|
||
) (*KnowledgeContext, error) {
|
||
requestedGroupIDs := normalizeInt64IDs(groupIDs)
|
||
if len(requestedGroupIDs) == 0 {
|
||
return nil, nil
|
||
}
|
||
if strings.TrimSpace(query) == "" {
|
||
return nil, nil
|
||
}
|
||
if err := s.provider.Validate(); err != nil {
|
||
return nil, response.ErrServiceUnavailable(50351, "retrieval_unavailable", err.Error())
|
||
}
|
||
if err := s.store.Validate(); err != nil {
|
||
return nil, response.ErrServiceUnavailable(50352, "qdrant_unavailable", err.Error())
|
||
}
|
||
|
||
searchGroupIDs, groupNames, err := s.resolveKnowledgeSearchScope(ctx, tenantID, requestedGroupIDs)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(searchGroupIDs) == 0 || len(groupNames) == 0 {
|
||
return nil, nil
|
||
}
|
||
|
||
baseContext := &KnowledgeContext{
|
||
GroupIDs: searchGroupIDs,
|
||
GroupNames: groupNames,
|
||
Snippets: []KnowledgeSnippet{},
|
||
PreciseFacts: s.collectKnowledgeContextPreciseFacts(ctx, tenantID, searchGroupIDs, nil),
|
||
}
|
||
|
||
vectors, err := s.provider.Embed(ctx, []string{query})
|
||
if err != nil {
|
||
return nil, response.ErrServiceUnavailable(50353, "embedding_failed", err.Error())
|
||
}
|
||
if len(vectors) == 0 || len(vectors[0]) == 0 {
|
||
return nil, response.ErrServiceUnavailable(50353, "embedding_failed", "embedding vector is empty")
|
||
}
|
||
|
||
runtimeCfg := s.runtimeConfig()
|
||
searchPoints, err := s.store.Search(ctx, retrieval.SearchRequest{
|
||
Vector: vectors[0],
|
||
TenantID: tenantID,
|
||
GroupIDs: searchGroupIDs,
|
||
Limit: runtimeCfg.RecallLimit,
|
||
})
|
||
if err != nil {
|
||
return nil, response.ErrServiceUnavailable(50354, "qdrant_search_failed", err.Error())
|
||
}
|
||
if len(searchPoints) == 0 {
|
||
s.logKnowledgeResolve(baseContext, tenantID, query, "no_search_points", nil, nil)
|
||
return baseContext, nil
|
||
}
|
||
|
||
candidates := dedupeKnowledgeSearchCandidates(knowledgeSnippetsFromSearchPoints(searchPoints))
|
||
if len(candidates) == 0 {
|
||
s.logKnowledgeResolve(baseContext, tenantID, query, "no_candidate_text", nil, nil)
|
||
return baseContext, nil
|
||
}
|
||
|
||
candidates, err = s.filterActiveKnowledgeSnippets(ctx, tenantID, candidates)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(candidates) == 0 {
|
||
s.logKnowledgeResolve(baseContext, tenantID, query, "no_active_candidates", nil, nil)
|
||
return baseContext, nil
|
||
}
|
||
|
||
documents := make([]string, 0, len(candidates))
|
||
for _, candidate := range candidates {
|
||
documents = append(documents, candidate.Text)
|
||
}
|
||
|
||
reranked, err := s.provider.Rerank(ctx, query, documents, minInt(runtimeCfg.RerankTopN, len(documents)))
|
||
if err != nil {
|
||
selected := selectKnowledgeCandidateSnippets(candidates, runtimeCfg.RerankTopN)
|
||
selected = s.enrichKnowledgePromptSnippets(ctx, tenantID, selected)
|
||
baseContext.Snippets = selected
|
||
baseContext.PreciseFacts = s.collectKnowledgeContextPreciseFacts(ctx, tenantID, searchGroupIDs, selected)
|
||
s.logKnowledgeResolve(baseContext, tenantID, query, "rerank_failed_fallback", candidates, selected)
|
||
return baseContext, nil
|
||
}
|
||
|
||
selected := selectKnowledgeRerankedSnippets(candidates, reranked, runtimeCfg.RerankTopN)
|
||
selected = s.enrichKnowledgePromptSnippets(ctx, tenantID, selected)
|
||
baseContext.Snippets = selected
|
||
baseContext.PreciseFacts = s.collectKnowledgeContextPreciseFacts(ctx, tenantID, searchGroupIDs, selected)
|
||
s.logKnowledgeResolve(baseContext, tenantID, query, "resolved", candidates, selected)
|
||
return baseContext, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) logKnowledgeResolve(
|
||
knowledgeCtx *KnowledgeContext,
|
||
tenantID int64,
|
||
query string,
|
||
stage string,
|
||
candidates []KnowledgeSnippet,
|
||
selected []KnowledgeSnippet,
|
||
) {
|
||
if s == nil || s.logger == nil || knowledgeCtx == nil {
|
||
return
|
||
}
|
||
|
||
s.logger.Info("knowledge resolve result",
|
||
zap.String("stage", strings.TrimSpace(stage)),
|
||
zap.Int64("tenant_id", tenantID),
|
||
zap.Int64s("group_ids", knowledgeCtx.GroupIDs),
|
||
zap.Strings("group_names", knowledgeCtx.GroupNames),
|
||
zap.String("query", knowledgeLogPreview(query, 600)),
|
||
zap.Int("candidate_count", len(candidates)),
|
||
zap.Int("selected_count", len(selected)),
|
||
zap.Int("precise_fact_count", len(knowledgeCtx.PreciseFacts)),
|
||
// zap.Strings("precise_facts", summarizeKnowledgeFactsForLog(knowledgeCtx.PreciseFacts)),
|
||
)
|
||
}
|
||
|
||
// for debug
|
||
func summarizeKnowledgeSnippetsForLog(snippets []KnowledgeSnippet) []map[string]any {
|
||
if len(snippets) == 0 {
|
||
return []map[string]any{}
|
||
}
|
||
|
||
summary := make([]map[string]any, 0, len(snippets))
|
||
for _, snippet := range snippets {
|
||
entry := map[string]any{
|
||
"point_id": snippet.PointID,
|
||
"group_id": snippet.GroupID,
|
||
"group_name": strings.TrimSpace(snippet.GroupName),
|
||
"item_id": snippet.KnowledgeItemID,
|
||
"item_name": strings.TrimSpace(snippet.ItemName),
|
||
"score": snippet.Score,
|
||
"text": knowledgeLogPreview(snippet.Text, 240),
|
||
}
|
||
if sourceType := strings.TrimSpace(snippet.SourceType); sourceType != "" {
|
||
entry["source_type"] = sourceType
|
||
}
|
||
if sourceURI := strings.TrimSpace(derefKnowledgeString(snippet.SourceURI)); sourceURI != "" {
|
||
entry["source_uri"] = sourceURI
|
||
}
|
||
if len(snippet.PreciseFacts) > 0 {
|
||
entry["precise_facts"] = summarizeKnowledgeFactsForLog(snippet.PreciseFacts)
|
||
}
|
||
summary = append(summary, entry)
|
||
}
|
||
return summary
|
||
}
|
||
|
||
func summarizeKnowledgeFactsForLog(facts []string) []string {
|
||
if len(facts) == 0 {
|
||
return []string{}
|
||
}
|
||
|
||
limit := len(facts)
|
||
if limit > 12 {
|
||
limit = 12
|
||
}
|
||
|
||
summary := make([]string, 0, limit)
|
||
for _, fact := range facts[:limit] {
|
||
summary = append(summary, knowledgeLogPreview(fact, 240))
|
||
}
|
||
return summary
|
||
}
|
||
|
||
func knowledgeLogPreview(value string, maxRunes int) string {
|
||
text := strings.TrimSpace(value)
|
||
if text == "" {
|
||
return ""
|
||
}
|
||
|
||
runes := []rune(text)
|
||
if maxRunes <= 0 || len(runes) <= maxRunes {
|
||
return text
|
||
}
|
||
return strings.TrimSpace(string(runes[:maxRunes])) + "..."
|
||
}
|
||
|
||
func knowledgeSnippetsFromSearchPoints(points []retrieval.SearchPoint) []KnowledgeSnippet {
|
||
if len(points) == 0 {
|
||
return []KnowledgeSnippet{}
|
||
}
|
||
|
||
snippets := make([]KnowledgeSnippet, 0, len(points))
|
||
for _, point := range points {
|
||
text := strings.TrimSpace(anyString(point.Payload["text"]))
|
||
if text == "" {
|
||
continue
|
||
}
|
||
snippets = append(snippets, KnowledgeSnippet{
|
||
PointID: strings.TrimSpace(point.ID),
|
||
GroupID: anyInt64(point.Payload["group_id"]),
|
||
GroupName: strings.TrimSpace(anyString(point.Payload["group_name"])),
|
||
KnowledgeItemID: anyInt64(point.Payload["item_id"]),
|
||
ItemName: strings.TrimSpace(anyString(point.Payload["item_name"])),
|
||
Text: text,
|
||
Score: point.Score,
|
||
})
|
||
}
|
||
return snippets
|
||
}
|
||
|
||
func dedupeKnowledgeSearchCandidates(candidates []KnowledgeSnippet) []KnowledgeSnippet {
|
||
if len(candidates) == 0 {
|
||
return []KnowledgeSnippet{}
|
||
}
|
||
|
||
seenPoints := make(map[string]struct{}, len(candidates))
|
||
seenTexts := make(map[string]struct{}, len(candidates))
|
||
deduped := make([]KnowledgeSnippet, 0, len(candidates))
|
||
for _, candidate := range candidates {
|
||
if pointID := strings.TrimSpace(candidate.PointID); pointID != "" {
|
||
if _, exists := seenPoints[pointID]; exists {
|
||
continue
|
||
}
|
||
seenPoints[pointID] = struct{}{}
|
||
}
|
||
textKey := normalizeKnowledgeSnippetTextKey(candidate.Text)
|
||
if textKey == "" {
|
||
continue
|
||
}
|
||
if _, exists := seenTexts[textKey]; exists {
|
||
continue
|
||
}
|
||
seenTexts[textKey] = struct{}{}
|
||
candidate.Text = strings.TrimSpace(candidate.Text)
|
||
deduped = append(deduped, candidate)
|
||
}
|
||
return deduped
|
||
}
|
||
|
||
func selectKnowledgeCandidateSnippets(candidates []KnowledgeSnippet, limit int) []KnowledgeSnippet {
|
||
return selectKnowledgeRerankedSnippets(candidates, nil, limit)
|
||
}
|
||
|
||
func selectKnowledgeRerankedSnippets(
|
||
candidates []KnowledgeSnippet,
|
||
reranked []retrieval.RerankResult,
|
||
limit int,
|
||
) []KnowledgeSnippet {
|
||
if len(candidates) == 0 {
|
||
return []KnowledgeSnippet{}
|
||
}
|
||
if limit <= 0 || limit > len(candidates) {
|
||
limit = len(candidates)
|
||
}
|
||
|
||
selected := make([]KnowledgeSnippet, 0, limit)
|
||
seenPoints := make(map[string]struct{}, limit)
|
||
seenTexts := make(map[string]struct{}, limit)
|
||
itemCounts := make(map[int64]int, limit)
|
||
|
||
appendSnippet := func(snippet KnowledgeSnippet, enforceItemLimit bool) {
|
||
if len(selected) >= limit {
|
||
return
|
||
}
|
||
if pointID := strings.TrimSpace(snippet.PointID); pointID != "" {
|
||
if _, exists := seenPoints[pointID]; exists {
|
||
return
|
||
}
|
||
}
|
||
textKey := normalizeKnowledgeSnippetTextKey(snippet.Text)
|
||
if textKey == "" {
|
||
return
|
||
}
|
||
if _, exists := seenTexts[textKey]; exists {
|
||
return
|
||
}
|
||
if enforceItemLimit && snippet.KnowledgeItemID > 0 && itemCounts[snippet.KnowledgeItemID] >= maxKnowledgeSnippetsPerItem {
|
||
return
|
||
}
|
||
|
||
snippet.Text = strings.TrimSpace(snippet.Text)
|
||
selected = append(selected, snippet)
|
||
if pointID := strings.TrimSpace(snippet.PointID); pointID != "" {
|
||
seenPoints[pointID] = struct{}{}
|
||
}
|
||
seenTexts[textKey] = struct{}{}
|
||
if snippet.KnowledgeItemID > 0 {
|
||
itemCounts[snippet.KnowledgeItemID]++
|
||
}
|
||
}
|
||
|
||
for _, item := range reranked {
|
||
if item.Index < 0 || item.Index >= len(candidates) {
|
||
continue
|
||
}
|
||
snippet := candidates[item.Index]
|
||
snippet.Score = item.RelevanceScore
|
||
if text := strings.TrimSpace(item.Text); text != "" {
|
||
snippet.Text = text
|
||
}
|
||
appendSnippet(snippet, true)
|
||
}
|
||
|
||
for _, candidate := range candidates {
|
||
appendSnippet(candidate, true)
|
||
}
|
||
if len(selected) < limit {
|
||
for _, candidate := range candidates {
|
||
appendSnippet(candidate, false)
|
||
}
|
||
}
|
||
|
||
return selected
|
||
}
|
||
|
||
func normalizeKnowledgeSnippetTextKey(text string) string {
|
||
text = strings.TrimSpace(text)
|
||
if text == "" {
|
||
return ""
|
||
}
|
||
return strings.ToLower(strings.Join(strings.Fields(text), " "))
|
||
}
|
||
|
||
func (s *KnowledgeService) RenderPromptSection(ctx *KnowledgeContext) string {
|
||
if ctx == nil || (len(ctx.Snippets) == 0 && len(ctx.PreciseFacts) == 0) {
|
||
return ""
|
||
}
|
||
|
||
lines := prompts.KnowledgePromptIntroLines()
|
||
preciseFacts := ctx.PreciseFacts
|
||
if len(preciseFacts) == 0 {
|
||
preciseFacts = collectKnowledgePromptPreciseFacts(ctx.Snippets)
|
||
}
|
||
if len(preciseFacts) > 0 {
|
||
lines = append(lines,
|
||
"以下为必须严格按原文保留的高精度事实；如果正文需要引用，必须逐字一致，不得改写、翻译、推测或补全：",
|
||
)
|
||
for _, fact := range preciseFacts {
|
||
lines = append(lines, "- "+fact)
|
||
}
|
||
lines = append(lines, "")
|
||
}
|
||
|
||
for index, snippet := range ctx.Snippets {
|
||
label := knowledgeSnippetPromptLabel(snippet, true, index)
|
||
referenceID := knowledgeSnippetReferenceID(index)
|
||
block := fmt.Sprintf("%d. [%s] %s\n%s", index+1, referenceID, label, snippet.Text)
|
||
if sourceURI := strings.TrimSpace(derefKnowledgeString(snippet.SourceURI)); sourceURI != "" {
|
||
block = fmt.Sprintf("%s\n来源链接: %s", block, sourceURI)
|
||
}
|
||
lines = append(lines, block)
|
||
}
|
||
return strings.Join(lines, "\n")
|
||
}
|
||
|
||
type knowledgePromptItemRecord struct {
|
||
ID int64
|
||
GroupID int64
|
||
GroupName string
|
||
Name string
|
||
SourceType string
|
||
SourceURI *string
|
||
ContentText *string
|
||
MarkdownContent *string
|
||
}
|
||
|
||
func (s *KnowledgeService) enrichKnowledgePromptSnippets(
|
||
ctx context.Context,
|
||
tenantID int64,
|
||
snippets []KnowledgeSnippet,
|
||
) []KnowledgeSnippet {
|
||
if len(snippets) == 0 {
|
||
return []KnowledgeSnippet{}
|
||
}
|
||
|
||
itemIDs := make([]int64, 0, len(snippets))
|
||
for _, snippet := range snippets {
|
||
if snippet.KnowledgeItemID > 0 {
|
||
itemIDs = append(itemIDs, snippet.KnowledgeItemID)
|
||
}
|
||
}
|
||
itemIDs = normalizeInt64IDs(itemIDs)
|
||
if len(itemIDs) == 0 {
|
||
return snippets
|
||
}
|
||
|
||
items, err := s.loadKnowledgePromptItems(ctx, tenantID, itemIDs)
|
||
if err != nil {
|
||
if s.logger != nil {
|
||
s.logger.Warn("load knowledge prompt items failed", zap.Error(err))
|
||
}
|
||
return snippets
|
||
}
|
||
|
||
itemMap := make(map[int64]knowledgePromptItemRecord, len(items))
|
||
for _, item := range items {
|
||
itemMap[item.ID] = item
|
||
}
|
||
|
||
enriched := make([]KnowledgeSnippet, 0, len(snippets))
|
||
for _, snippet := range snippets {
|
||
item, ok := itemMap[snippet.KnowledgeItemID]
|
||
if ok {
|
||
snippet.GroupID = item.GroupID
|
||
snippet.GroupName = item.GroupName
|
||
snippet.ItemName = item.Name
|
||
snippet.SourceType = item.SourceType
|
||
snippet.SourceURI = item.SourceURI
|
||
sourceText := derefKnowledgeString(item.MarkdownContent)
|
||
if strings.TrimSpace(sourceText) == "" {
|
||
sourceText = derefKnowledgeString(item.ContentText)
|
||
}
|
||
snippet.PreciseFacts = extractKnowledgePreciseFacts(sourceText)
|
||
}
|
||
enriched = append(enriched, snippet)
|
||
}
|
||
|
||
return enriched
|
||
}
|
||
|
||
func (s *KnowledgeService) loadKnowledgePromptItems(
|
||
ctx context.Context,
|
||
tenantID int64,
|
||
itemIDs []int64,
|
||
) ([]knowledgePromptItemRecord, error) {
|
||
if len(itemIDs) == 0 {
|
||
return []knowledgePromptItemRecord{}, nil
|
||
}
|
||
|
||
rows, err := s.pool.Query(ctx, `
|
||
SELECT ki.id, ki.group_id, kg.name, ki.name, ki.source_type, ki.source_uri, ki.content_text, ki.markdown_content
|
||
FROM knowledge_items ki
|
||
JOIN knowledge_groups kg ON kg.id = ki.group_id AND kg.deleted_at IS NULL
|
||
WHERE ki.tenant_id = $1
|
||
AND ki.id = ANY($2::bigint[])
|
||
AND ki.status = 'completed'
|
||
AND ki.deleted_at IS NULL
|
||
`, tenantID, itemIDs)
|
||
if err != nil {
|
||
return nil, response.ErrInternal(50051, "knowledge_items_query_failed", "failed to load knowledge prompt items")
|
||
}
|
||
defer rows.Close()
|
||
|
||
items := make([]knowledgePromptItemRecord, 0, len(itemIDs))
|
||
for rows.Next() {
|
||
var item knowledgePromptItemRecord
|
||
if err := rows.Scan(
|
||
&item.ID,
|
||
&item.GroupID,
|
||
&item.GroupName,
|
||
&item.Name,
|
||
&item.SourceType,
|
||
&item.SourceURI,
|
||
&item.ContentText,
|
||
&item.MarkdownContent,
|
||
); err != nil {
|
||
return nil, response.ErrInternal(50051, "knowledge_items_scan_failed", err.Error())
|
||
}
|
||
items = append(items, item)
|
||
}
|
||
return items, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) collectKnowledgeContextPreciseFacts(
|
||
ctx context.Context,
|
||
tenantID int64,
|
||
groupIDs []int64,
|
||
snippets []KnowledgeSnippet,
|
||
) []string {
|
||
groupIDs = normalizeInt64IDs(groupIDs)
|
||
if len(groupIDs) == 0 {
|
||
return collectKnowledgePromptPreciseFacts(snippets)
|
||
}
|
||
|
||
matchedItemIDs := make([]int64, 0, len(snippets))
|
||
for _, snippet := range snippets {
|
||
if snippet.KnowledgeItemID > 0 {
|
||
matchedItemIDs = append(matchedItemIDs, snippet.KnowledgeItemID)
|
||
}
|
||
}
|
||
matchedItemIDs = normalizeInt64IDs(matchedItemIDs)
|
||
|
||
items, err := s.loadKnowledgePromptItemsByGroupIDs(ctx, tenantID, groupIDs, matchedItemIDs, 24)
|
||
if err != nil {
|
||
if s.logger != nil {
|
||
s.logger.Warn("collect knowledge context precise facts failed", zap.Error(err))
|
||
}
|
||
return collectKnowledgePromptPreciseFacts(snippets)
|
||
}
|
||
|
||
facts := make([]string, 0, 24)
|
||
seen := make(map[string]struct{}, 24)
|
||
for _, item := range items {
|
||
sourceText := derefKnowledgeString(item.MarkdownContent)
|
||
if strings.TrimSpace(sourceText) == "" {
|
||
sourceText = derefKnowledgeString(item.ContentText)
|
||
}
|
||
|
||
label := strings.TrimSpace(item.Name)
|
||
if strings.TrimSpace(item.GroupName) != "" && label != "" {
|
||
label = fmt.Sprintf("%s / %s", strings.TrimSpace(item.GroupName), label)
|
||
} else if strings.TrimSpace(item.GroupName) != "" {
|
||
label = strings.TrimSpace(item.GroupName)
|
||
}
|
||
|
||
for _, fact := range extractKnowledgePreciseFacts(sourceText) {
|
||
value := strings.TrimSpace(fact)
|
||
if value == "" {
|
||
continue
|
||
}
|
||
entry := value
|
||
if label != "" {
|
||
entry = fmt.Sprintf("%s：%s", label, value)
|
||
}
|
||
if _, exists := seen[entry]; exists {
|
||
continue
|
||
}
|
||
seen[entry] = struct{}{}
|
||
facts = append(facts, entry)
|
||
if len(facts) >= 24 {
|
||
return facts
|
||
}
|
||
}
|
||
}
|
||
|
||
if len(facts) == 0 {
|
||
return collectKnowledgePromptPreciseFacts(snippets)
|
||
}
|
||
return facts
|
||
}
|
||
|
||
func (s *KnowledgeService) loadKnowledgePromptItemsByGroupIDs(
|
||
ctx context.Context,
|
||
tenantID int64,
|
||
groupIDs []int64,
|
||
prioritizedItemIDs []int64,
|
||
limit int,
|
||
) ([]knowledgePromptItemRecord, error) {
|
||
groupIDs = normalizeInt64IDs(groupIDs)
|
||
prioritizedItemIDs = normalizeInt64IDs(prioritizedItemIDs)
|
||
if len(groupIDs) == 0 {
|
||
return []knowledgePromptItemRecord{}, nil
|
||
}
|
||
if limit <= 0 {
|
||
limit = 24
|
||
}
|
||
|
||
rows, err := s.pool.Query(ctx, `
|
||
WITH RECURSIVE selected_groups AS (
|
||
SELECT id
|
||
FROM knowledge_groups
|
||
WHERE tenant_id = $1
|
||
AND id = ANY($2::bigint[])
|
||
AND deleted_at IS NULL
|
||
UNION ALL
|
||
SELECT child.id
|
||
FROM knowledge_groups child
|
||
JOIN selected_groups parent ON child.parent_id = parent.id
|
||
WHERE child.tenant_id = $1
|
||
AND child.deleted_at IS NULL
|
||
)
|
||
SELECT ki.id, ki.group_id, kg.name, ki.name, ki.source_type, ki.source_uri, ki.content_text, ki.markdown_content
|
||
FROM knowledge_items ki
|
||
JOIN knowledge_groups kg ON kg.id = ki.group_id AND kg.deleted_at IS NULL
|
||
WHERE ki.tenant_id = $1
|
||
AND ki.status = 'completed'
|
||
AND ki.deleted_at IS NULL
|
||
AND ki.group_id IN (SELECT id FROM selected_groups)
|
||
ORDER BY
|
||
CASE
|
||
WHEN ki.id = ANY($3::bigint[]) THEN 0
|
||
ELSE 1
|
||
END,
|
||
ki.updated_at DESC,
|
||
ki.id DESC
|
||
LIMIT $4
|
||
`, tenantID, groupIDs, prioritizedItemIDs, limit)
|
||
if err != nil {
|
||
return nil, response.ErrInternal(50051, "knowledge_items_query_failed", "failed to load knowledge prompt items by groups")
|
||
}
|
||
defer rows.Close()
|
||
|
||
items := make([]knowledgePromptItemRecord, 0, limit)
|
||
for rows.Next() {
|
||
var item knowledgePromptItemRecord
|
||
if err := rows.Scan(
|
||
&item.ID,
|
||
&item.GroupID,
|
||
&item.GroupName,
|
||
&item.Name,
|
||
&item.SourceType,
|
||
&item.SourceURI,
|
||
&item.ContentText,
|
||
&item.MarkdownContent,
|
||
); err != nil {
|
||
return nil, response.ErrInternal(50051, "knowledge_items_scan_failed", err.Error())
|
||
}
|
||
items = append(items, item)
|
||
}
|
||
return items, nil
|
||
}
|
||
|
||
func collectKnowledgePromptPreciseFacts(snippets []KnowledgeSnippet) []string {
|
||
if len(snippets) == 0 {
|
||
return nil
|
||
}
|
||
|
||
facts := make([]string, 0, len(snippets)*2)
|
||
seen := make(map[string]struct{}, len(snippets)*2)
|
||
for index, snippet := range snippets {
|
||
label := knowledgeSnippetPromptLabel(snippet, false, index)
|
||
for _, fact := range snippet.PreciseFacts {
|
||
value := strings.TrimSpace(fact)
|
||
if value == "" {
|
||
continue
|
||
}
|
||
entry := fmt.Sprintf("%s：%s", label, value)
|
||
if _, exists := seen[entry]; exists {
|
||
continue
|
||
}
|
||
seen[entry] = struct{}{}
|
||
facts = append(facts, entry)
|
||
}
|
||
}
|
||
return facts
|
||
}
|
||
|
||
func knowledgeSnippetPromptLabel(snippet KnowledgeSnippet, fallbackToDefault bool, index int) string {
|
||
label := strings.TrimSpace(snippet.ItemName)
|
||
if label == "" && fallbackToDefault {
|
||
label = prompts.KnowledgeSnippetLabel(index)
|
||
}
|
||
groupName := strings.TrimSpace(snippet.GroupName)
|
||
if groupName != "" && label != "" {
|
||
return fmt.Sprintf("%s / %s", groupName, label)
|
||
}
|
||
if groupName != "" {
|
||
return groupName
|
||
}
|
||
return label
|
||
}
|
||
|
||
func knowledgeSnippetReferenceID(index int) string {
|
||
return fmt.Sprintf("K%d", index+1)
|
||
}
|
||
|
||
func extractKnowledgePreciseFacts(text string) []string {
|
||
normalized := normalizeKnowledgeMarkdown(text)
|
||
if normalized == "" {
|
||
normalized = normalizeKnowledgeText(text)
|
||
}
|
||
if normalized == "" {
|
||
return nil
|
||
}
|
||
|
||
lines := strings.Split(normalized, "\n")
|
||
facts := make([]string, 0, 12)
|
||
seen := make(map[string]struct{}, 12)
|
||
|
||
appendFact := func(value string) {
|
||
value = normalizeKnowledgeInlineText(trimKnowledgeFactLine(value))
|
||
if value == "" {
|
||
return
|
||
}
|
||
if _, exists := seen[value]; exists {
|
||
return
|
||
}
|
||
seen[value] = struct{}{}
|
||
facts = append(facts, value)
|
||
}
|
||
|
||
for index := 0; index < len(lines); index++ {
|
||
line := trimKnowledgeFactLine(lines[index])
|
||
if line == "" || strings.HasPrefix(line, "#") {
|
||
continue
|
||
}
|
||
|
||
if knowledgePreciseFactLabelOnly(line) {
|
||
sectionContent := collectKnowledgePreciseFactSection(lines, index, line)
|
||
if sectionContent != "" {
|
||
appendFact(strings.TrimRight(strings.TrimSpace(line), ":：") + "：" + sectionContent)
|
||
}
|
||
continue
|
||
}
|
||
|
||
if knowledgeLineContainsPreciseFact(line) {
|
||
appendFact(line)
|
||
}
|
||
}
|
||
|
||
if len(facts) > 12 {
|
||
return facts[:12]
|
||
}
|
||
return facts
|
||
}
|
||
|
||
func collectKnowledgePreciseFactSection(lines []string, startIndex int, label string) string {
|
||
maxLines := knowledgePreciseFactSectionLineLimit(label)
|
||
if maxLines <= 0 {
|
||
maxLines = 1
|
||
}
|
||
|
||
items := make([]string, 0, maxLines)
|
||
for index := startIndex + 1; index < len(lines) && len(items) < maxLines; index++ {
|
||
line := trimKnowledgeFactLine(lines[index])
|
||
if line == "" {
|
||
if len(items) > 0 {
|
||
break
|
||
}
|
||
continue
|
||
}
|
||
if strings.HasPrefix(line, "#") {
|
||
break
|
||
}
|
||
if knowledgePreciseFactLabelOnly(line) {
|
||
break
|
||
}
|
||
if len(items) > 0 && knowledgeLineStartsAnotherPreciseFact(line) && !knowledgeLineContainsBusinessFact(line) {
|
||
break
|
||
}
|
||
items = append(items, line)
|
||
}
|
||
|
||
return strings.Join(items, "；")
|
||
}
|
||
|
||
func knowledgePreciseFactSectionLineLimit(line string) int {
|
||
if knowledgeLineContainsBusinessFact(line) {
|
||
return 4
|
||
}
|
||
return 1
|
||
}
|
||
|
||
func trimKnowledgeFactLine(value string) string {
|
||
line := strings.TrimSpace(value)
|
||
line = strings.TrimLeft(line, "-*• \t")
|
||
line = strings.TrimLeft(line, "0123456789.、)） \t")
|
||
return strings.TrimSpace(line)
|
||
}
|
||
|
||
func knowledgePreciseFactLabelOnly(line string) bool {
|
||
normalized := strings.ToLower(strings.Trim(strings.TrimSpace(line), ":："))
|
||
switch normalized {
|
||
case "地址", "门店地址", "公司地址", "办公地址", "总部地址", "联系地址", "location", "address",
|
||
"电话", "联系电话", "客服热线", "phone", "tel", "telephone",
|
||
"官网", "网址", "网站", "website", "url",
|
||
"邮箱", "电子邮箱", "email", "e-mail",
|
||
"营业时间", "开放时间", "服务时间", "办公时间", "hours", "opening hours",
|
||
"主营业务", "主要业务", "核心业务", "业务范围", "经营范围", "服务范围",
|
||
"主营产品", "主要产品", "核心产品", "产品范围",
|
||
"main business", "core business", "business scope", "service scope",
|
||
"main products", "core products", "product scope":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func knowledgeLineContainsPreciseFact(line string) bool {
|
||
lower := strings.ToLower(strings.TrimSpace(line))
|
||
if lower == "" {
|
||
return false
|
||
}
|
||
|
||
keywords := []string{
|
||
"地址", "门店地址", "公司地址", "办公地址", "总部地址", "联系地址", "邮编", "位置", "坐标",
|
||
"电话", "联系电话", "客服热线", "热线", "传真",
|
||
"官网", "网址", "网站", "链接",
|
||
"邮箱", "电子邮箱", "邮件",
|
||
"营业时间", "开放时间", "服务时间", "办公时间",
|
||
"主营业务", "主要业务", "核心业务", "业务范围", "经营范围", "服务范围",
|
||
"主营产品", "主要产品", "核心产品", "产品范围",
|
||
"address", "location", "phone", "tel", "telephone", "website", "email", "e-mail",
|
||
"opening hours", "business hours", "postcode", "zip code",
|
||
"main business", "core business", "business scope", "service scope",
|
||
"main products", "core products", "product scope",
|
||
}
|
||
for _, keyword := range keywords {
|
||
if strings.Contains(lower, keyword) {
|
||
return true
|
||
}
|
||
}
|
||
|
||
if strings.Contains(lower, "http://") || strings.Contains(lower, "https://") {
|
||
return true
|
||
}
|
||
if strings.Contains(lower, "@") {
|
||
return true
|
||
}
|
||
|
||
digits := 0
|
||
for _, r := range lower {
|
||
if r >= '0' && r <= '9' {
|
||
digits++
|
||
}
|
||
}
|
||
return digits >= 7 && (strings.Contains(lower, "-") || strings.Contains(lower, " ") || strings.Contains(lower, "("))
|
||
}
|
||
|
||
func knowledgeLineStartsAnotherPreciseFact(line string) bool {
|
||
trimmed := strings.TrimSpace(line)
|
||
if trimmed == "" {
|
||
return false
|
||
}
|
||
if knowledgePreciseFactLabelOnly(trimmed) {
|
||
return true
|
||
}
|
||
|
||
lower := strings.ToLower(trimmed)
|
||
labels := []string{
|
||
"地址：", "门店地址：", "公司地址：", "办公地址：", "总部地址：", "联系地址：",
|
||
"电话：", "联系电话：", "客服热线：", "传真：",
|
||
"官网：", "网址：", "网站：", "邮箱：", "电子邮箱：",
|
||
"营业时间：", "开放时间：", "服务时间：", "办公时间：",
|
||
"主营业务：", "主要业务：", "核心业务：", "业务范围：", "经营范围：", "服务范围：",
|
||
"主营产品：", "主要产品：", "核心产品：", "产品范围：",
|
||
"address:", "location:", "phone:", "tel:", "telephone:",
|
||
"website:", "email:", "e-mail:",
|
||
"opening hours:", "business hours:",
|
||
"main business:", "core business:", "business scope:", "service scope:",
|
||
"main products:", "core products:", "product scope:",
|
||
}
|
||
for _, label := range labels {
|
||
if strings.HasPrefix(lower, label) {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func knowledgeLineContainsBusinessFact(line string) bool {
|
||
lower := strings.ToLower(strings.TrimSpace(line))
|
||
if lower == "" {
|
||
return false
|
||
}
|
||
|
||
keywords := []string{
|
||
"主营业务", "主要业务", "核心业务", "业务范围", "经营范围", "服务范围",
|
||
"主营产品", "主要产品", "核心产品", "产品范围",
|
||
"main business", "core business", "business scope", "service scope",
|
||
"main products", "core products", "product scope",
|
||
}
|
||
for _, keyword := range keywords {
|
||
if strings.Contains(lower, keyword) {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
type knowledgeIngestInput struct {
|
||
GroupID int64
|
||
Name string
|
||
SourceType string
|
||
SourceURI *string
|
||
StorageKey string
|
||
ContentText *string
|
||
SizeBytes int64
|
||
}
|
||
|
||
func (s *KnowledgeService) validateIngestionDependencies(sourceType string) error {
|
||
if err := s.provider.Validate(); err != nil {
|
||
return response.ErrServiceUnavailable(50351, "retrieval_unavailable", err.Error())
|
||
}
|
||
if err := s.store.Validate(); err != nil {
|
||
return response.ErrServiceUnavailable(50352, "qdrant_unavailable", err.Error())
|
||
}
|
||
if sourceType == "document" {
|
||
if err := s.objectStorage.Validate(); err != nil {
|
||
return response.ErrServiceUnavailable(50357, "knowledge_storage_unavailable", err.Error())
|
||
}
|
||
}
|
||
if sourceType == "website" {
|
||
runtimeCfg := s.runtimeConfig()
|
||
if strings.TrimSpace(runtimeCfg.ArkAPIKey) == "" {
|
||
return response.ErrServiceUnavailable(50358, "knowledge_webpage_parser_unavailable", "ark api key is not configured")
|
||
}
|
||
if err := s.llmClient.Validate(); err != nil {
|
||
return response.ErrServiceUnavailable(50359, "knowledge_markdown_formatter_unavailable", err.Error())
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *KnowledgeService) queueKnowledgeItem(
|
||
ctx context.Context,
|
||
tenantID int64,
|
||
input knowledgeIngestInput,
|
||
) (*KnowledgeItemResponse, error) {
|
||
resolvedGroupID, err := s.resolveKnowledgeStorageGroupID(ctx, tenantID, input.GroupID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
input.GroupID = resolvedGroupID
|
||
|
||
input.Name = strings.TrimSpace(input.Name)
|
||
if input.Name == "" {
|
||
return nil, response.ErrBadRequest(40054, "knowledge_name_required", "knowledge item name is required")
|
||
}
|
||
|
||
if input.ContentText != nil {
|
||
contentText := normalizeKnowledgeText(*input.ContentText)
|
||
if contentText == "" {
|
||
return nil, response.ErrBadRequest(40051, "knowledge_content_required", "knowledge content is required")
|
||
}
|
||
input.ContentText = &contentText
|
||
}
|
||
|
||
itemID, parseTaskID, err := s.createPendingKnowledgeItem(ctx, tenantID, input)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if err := s.enqueueParseJob(knowledgeParseJob{
|
||
TenantID: tenantID,
|
||
ItemID: itemID,
|
||
ParseTaskID: parseTaskID,
|
||
}); err != nil {
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), tenantID, itemID, parseTaskID, err)
|
||
return nil, response.ErrServiceUnavailable(50356, "knowledge_queue_unavailable", "knowledge parse queue is busy, please retry")
|
||
}
|
||
|
||
items, err := s.listKnowledgeItemsByIDs(ctx, tenantID, []int64{itemID})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(items) == 0 {
|
||
return nil, response.ErrInternal(50055, "knowledge_item_create_failed", "knowledge item was not found after creation")
|
||
}
|
||
return &items[0], nil
|
||
}
|
||
|
||
func (s *KnowledgeService) enqueueParseJob(job knowledgeParseJob) error {
|
||
if s.jobs == nil {
|
||
return fmt.Errorf("knowledge parse workers are not enabled")
|
||
}
|
||
|
||
select {
|
||
case s.jobs <- job:
|
||
return nil
|
||
default:
|
||
return fmt.Errorf("knowledge parse queue is full")
|
||
}
|
||
}
|
||
|
||
func (s *KnowledgeService) runParseWorker() {
|
||
for job := range s.jobs {
|
||
func(job knowledgeParseJob) {
|
||
defer func() {
|
||
if recovered := recover(); recovered != nil {
|
||
err := fmt.Errorf("knowledge parse panic: %v", recovered)
|
||
if s.logger != nil {
|
||
s.logger.Error("knowledge parse worker panicked",
|
||
zap.Int64("tenant_id", job.TenantID),
|
||
zap.Int64("knowledge_item_id", job.ItemID),
|
||
zap.Int64("parse_task_id", job.ParseTaskID),
|
||
zap.Any("panic", recovered),
|
||
zap.ByteString("stack", debug.Stack()),
|
||
)
|
||
}
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), job.TenantID, job.ItemID, job.ParseTaskID, err)
|
||
}
|
||
}()
|
||
|
||
s.executeParseJob(context.Background(), job)
|
||
}(job)
|
||
}
|
||
}
|
||
|
||
func (s *KnowledgeService) recoverPendingParseTasks() {
|
||
if s.jobs == nil {
|
||
return
|
||
}
|
||
|
||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||
defer cancel()
|
||
|
||
rows, err := s.pool.Query(ctx, `
|
||
SELECT kpt.tenant_id, kpt.knowledge_item_id, kpt.id
|
||
FROM knowledge_parse_tasks kpt
|
||
JOIN knowledge_items ki ON ki.id = kpt.knowledge_item_id
|
||
WHERE kpt.status IN ('pending', 'running')
|
||
AND ki.deleted_at IS NULL
|
||
AND ki.status IN ('pending', 'processing')
|
||
ORDER BY kpt.created_at ASC, kpt.id ASC
|
||
`)
|
||
if err != nil {
|
||
return
|
||
}
|
||
defer rows.Close()
|
||
|
||
for rows.Next() {
|
||
var job knowledgeParseJob
|
||
if scanErr := rows.Scan(&job.TenantID, &job.ItemID, &job.ParseTaskID); scanErr != nil {
|
||
continue
|
||
}
|
||
if err := s.enqueueParseJob(job); err != nil {
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *KnowledgeService) executeParseJob(ctx context.Context, job knowledgeParseJob) {
|
||
item, err := s.loadKnowledgeItemForParse(ctx, job.TenantID, job.ItemID)
|
||
if err != nil || item == nil {
|
||
return
|
||
}
|
||
|
||
if err := s.markKnowledgeIngestionRunning(ctx, job.TenantID, job.ItemID, job.ParseTaskID); err != nil {
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), job.TenantID, job.ItemID, job.ParseTaskID, err)
|
||
return
|
||
}
|
||
|
||
parsedContent, err := s.readKnowledgeItemParsedContent(ctx, item)
|
||
if err != nil {
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), job.TenantID, job.ItemID, job.ParseTaskID, err)
|
||
return
|
||
}
|
||
if item.SourceType == "website" {
|
||
if resolvedName := extractKnowledgeMarkdownTitle(parsedContent.Markdown); resolvedName != "" {
|
||
item.Name = resolvedName
|
||
}
|
||
}
|
||
|
||
chunkSource := strings.TrimSpace(parsedContent.Markdown)
|
||
if chunkSource == "" {
|
||
chunkSource = strings.TrimSpace(parsedContent.Text)
|
||
}
|
||
|
||
runtimeCfg := s.runtimeConfig()
|
||
chunks := splitKnowledgeTextIntoChunks(chunkSource, runtimeCfg.ChunkSize, runtimeCfg.ChunkOverlap)
|
||
if len(chunks) == 0 {
|
||
err = errors.New("knowledge content is empty after parsing")
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), job.TenantID, job.ItemID, job.ParseTaskID, err)
|
||
return
|
||
}
|
||
|
||
pointIDs, err := s.listKnowledgePointIDs(ctx, job.TenantID, job.ItemID)
|
||
if err != nil {
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), job.TenantID, job.ItemID, job.ParseTaskID, err)
|
||
return
|
||
}
|
||
if len(pointIDs) > 0 {
|
||
if err := s.store.DeletePoints(ctx, pointIDs); err != nil {
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), job.TenantID, job.ItemID, job.ParseTaskID, err)
|
||
return
|
||
}
|
||
}
|
||
|
||
embeddings, err := s.embedChunks(ctx, chunks)
|
||
if err != nil {
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), job.TenantID, job.ItemID, job.ParseTaskID, err)
|
||
return
|
||
}
|
||
if len(embeddings) == 0 || len(embeddings[0]) == 0 {
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), job.TenantID, job.ItemID, job.ParseTaskID, errors.New("embedding vector is empty"))
|
||
return
|
||
}
|
||
|
||
if err := s.store.EnsureCollection(ctx, len(embeddings[0])); err != nil {
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), job.TenantID, job.ItemID, job.ParseTaskID, err)
|
||
return
|
||
}
|
||
|
||
insertedPointIDs := make([]string, 0, len(chunks))
|
||
defer func() {
|
||
if err != nil && len(insertedPointIDs) > 0 {
|
||
_ = s.store.DeletePoints(context.Background(), insertedPointIDs)
|
||
}
|
||
}()
|
||
|
||
points := make([]retrieval.VectorPoint, 0, len(chunks))
|
||
metas := make([]knowledgeChunkMetaInput, 0, len(chunks))
|
||
for index, chunk := range chunks {
|
||
pointID := uuid.NewString()
|
||
insertedPointIDs = append(insertedPointIDs, pointID)
|
||
points = append(points, retrieval.VectorPoint{
|
||
ID: pointID,
|
||
Vector: embeddings[index],
|
||
Payload: map[string]any{
|
||
"tenant_id": job.TenantID,
|
||
"group_id": item.GroupID,
|
||
"group_name": item.GroupName,
|
||
"item_id": item.ID,
|
||
"item_name": item.Name,
|
||
"chunk_index": index,
|
||
"item_version": item.ItemVersion,
|
||
"status": "active",
|
||
"source_type": item.SourceType,
|
||
"text": chunk,
|
||
},
|
||
})
|
||
metas = append(metas, knowledgeChunkMetaInput{
|
||
ChunkIndex: index,
|
||
TokenCount: estimateKnowledgeTokenCount(chunk),
|
||
QdrantPointID: pointID,
|
||
ItemVersion: item.ItemVersion,
|
||
Status: "active",
|
||
})
|
||
}
|
||
|
||
if err = s.store.Upsert(ctx, points); err != nil {
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), job.TenantID, job.ItemID, job.ParseTaskID, err)
|
||
return
|
||
}
|
||
|
||
if err = s.markKnowledgeIngestionCompleted(ctx, job.TenantID, job.ItemID, job.ParseTaskID, item.Name, parsedContent, metas); err != nil {
|
||
_ = s.markKnowledgeIngestionFailed(context.Background(), job.TenantID, job.ItemID, job.ParseTaskID, err)
|
||
return
|
||
}
|
||
}
|
||
|
||
func (s *KnowledgeService) loadKnowledgeItemForParse(
|
||
ctx context.Context,
|
||
tenantID, itemID int64,
|
||
) (*knowledgeItemRecord, error) {
|
||
record := &knowledgeItemRecord{}
|
||
var contentText *string
|
||
var markdownContent *string
|
||
err := s.pool.QueryRow(ctx, `
|
||
SELECT ki.id, ki.tenant_id, ki.group_id, kg.name, ki.name, ki.source_type,
|
||
ki.source_uri, ki.storage_key, ki.content_text, ki.markdown_content, ki.item_version
|
||
FROM knowledge_items ki
|
||
JOIN knowledge_groups kg ON kg.id = ki.group_id
|
||
WHERE ki.id = $1 AND ki.tenant_id = $2 AND ki.deleted_at IS NULL
|
||
`, itemID, tenantID).Scan(
|
||
&record.ID,
|
||
&record.TenantID,
|
||
&record.GroupID,
|
||
&record.GroupName,
|
||
&record.Name,
|
||
&record.SourceType,
|
||
&record.SourceURI,
|
||
&record.StorageKey,
|
||
&contentText,
|
||
&markdownContent,
|
||
&record.ItemVersion,
|
||
)
|
||
if err != nil {
|
||
return nil, nil
|
||
}
|
||
record.ContentText = contentText
|
||
record.Markdown = markdownContent
|
||
return record, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) readKnowledgeItemParsedContent(ctx context.Context, item *knowledgeItemRecord) (*knowledgeParsedContent, error) {
|
||
if item == nil {
|
||
return nil, fmt.Errorf("knowledge item is empty")
|
||
}
|
||
|
||
switch item.SourceType {
|
||
case "text":
|
||
return parseKnowledgeTextInput(item.Name, derefKnowledgeString(item.ContentText)), nil
|
||
case "document":
|
||
data, err := s.objectStorage.GetBytes(ctx, item.StorageKey)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return extractKnowledgeContentFromFile(item.StorageKey, data)
|
||
case "website":
|
||
if item.SourceURI == nil || strings.TrimSpace(*item.SourceURI) == "" {
|
||
return nil, fmt.Errorf("knowledge website url is empty")
|
||
}
|
||
return s.parseWebsiteKnowledge(ctx, strings.TrimSpace(*item.SourceURI))
|
||
default:
|
||
return nil, fmt.Errorf("unsupported knowledge source type %q", item.SourceType)
|
||
}
|
||
}
|
||
|
||
type knowledgeChunkMetaInput struct {
|
||
ChunkIndex int
|
||
TokenCount int
|
||
QdrantPointID string
|
||
ItemVersion int
|
||
Status string
|
||
}
|
||
|
||
func (s *KnowledgeService) createPendingKnowledgeItem(
|
||
ctx context.Context,
|
||
tenantID int64,
|
||
input knowledgeIngestInput,
|
||
) (int64, int64, error) {
|
||
tx, err := s.pool.Begin(ctx)
|
||
if err != nil {
|
||
return 0, 0, response.ErrInternal(50055, "knowledge_item_create_failed", "failed to begin knowledge transaction")
|
||
}
|
||
defer func() { _ = tx.Rollback(ctx) }()
|
||
|
||
var itemID int64
|
||
err = tx.QueryRow(ctx, `
|
||
INSERT INTO knowledge_items (
|
||
tenant_id, group_id, source_type, name, source_uri, storage_key,
|
||
content_text, markdown_content, status, size_bytes, item_version
|
||
)
|
||
VALUES ($1, $2, $3, $4, $5, $6, $7, NULL, 'pending', $8, 1)
|
||
RETURNING id
|
||
`, tenantID, input.GroupID, input.SourceType, input.Name, input.SourceURI, input.StorageKey, input.ContentText, input.SizeBytes).Scan(&itemID)
|
||
if err != nil {
|
||
return 0, 0, response.ErrConflict(40951, "knowledge_item_create_failed", "failed to create knowledge item")
|
||
}
|
||
|
||
var taskID int64
|
||
err = tx.QueryRow(ctx, `
|
||
INSERT INTO knowledge_parse_tasks (
|
||
tenant_id, knowledge_item_id, source_type, status
|
||
)
|
||
VALUES ($1, $2, $3, 'pending')
|
||
RETURNING id
|
||
`, tenantID, itemID, input.SourceType).Scan(&taskID)
|
||
if err != nil {
|
||
return 0, 0, response.ErrInternal(50055, "knowledge_item_create_failed", "failed to create knowledge parse task")
|
||
}
|
||
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return 0, 0, response.ErrInternal(50055, "knowledge_item_create_failed", "failed to commit knowledge transaction")
|
||
}
|
||
return itemID, taskID, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) markKnowledgeIngestionRunning(
|
||
ctx context.Context,
|
||
tenantID, itemID, parseTaskID int64,
|
||
) error {
|
||
if _, err := s.pool.Exec(ctx, `
|
||
UPDATE knowledge_items
|
||
SET status = 'processing', error_message = NULL, updated_at = NOW()
|
||
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
|
||
`, itemID, tenantID); err != nil {
|
||
return err
|
||
}
|
||
|
||
if _, err := s.pool.Exec(ctx, `
|
||
UPDATE knowledge_parse_tasks
|
||
SET status = 'running',
|
||
error_message = NULL,
|
||
started_at = COALESCE(started_at, NOW()),
|
||
updated_at = NOW()
|
||
WHERE id = $1 AND tenant_id = $2
|
||
`, parseTaskID, tenantID); err != nil {
|
||
return err
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (s *KnowledgeService) markKnowledgeIngestionCompleted(
|
||
ctx context.Context,
|
||
tenantID, itemID, parseTaskID int64,
|
||
itemName string,
|
||
content *knowledgeParsedContent,
|
||
metas []knowledgeChunkMetaInput,
|
||
) error {
|
||
if content == nil {
|
||
return errors.New("knowledge content is empty")
|
||
}
|
||
if strings.TrimSpace(content.Text) == "" {
|
||
return errors.New("knowledge text content is empty")
|
||
}
|
||
if strings.TrimSpace(content.Markdown) == "" {
|
||
content.Markdown = buildKnowledgeMarkdownFromText("", content.Text)
|
||
}
|
||
itemName = strings.TrimSpace(itemName)
|
||
|
||
tx, err := s.pool.Begin(ctx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer func() { _ = tx.Rollback(ctx) }()
|
||
|
||
if _, err := tx.Exec(ctx, `
|
||
DELETE FROM knowledge_chunks_meta
|
||
WHERE tenant_id = $1 AND knowledge_item_id = $2
|
||
`, tenantID, itemID); err != nil {
|
||
return err
|
||
}
|
||
|
||
for _, meta := range metas {
|
||
if _, err := tx.Exec(ctx, `
|
||
INSERT INTO knowledge_chunks_meta (
|
||
tenant_id, knowledge_item_id, chunk_index, token_count,
|
||
qdrant_point_id, item_version, status
|
||
)
|
||
VALUES ($1, $2, $3, $4, $5, $6, $7)
|
||
`, tenantID, itemID, meta.ChunkIndex, meta.TokenCount, meta.QdrantPointID, meta.ItemVersion, meta.Status); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE knowledge_items
|
||
SET status = 'completed',
|
||
error_message = NULL,
|
||
name = CASE WHEN $3 <> '' THEN $3 ELSE name END,
|
||
content_text = $4,
|
||
markdown_content = $5,
|
||
updated_at = NOW()
|
||
WHERE id = $1 AND tenant_id = $2
|
||
`, itemID, tenantID, itemName, content.Text, content.Markdown); err != nil {
|
||
return err
|
||
}
|
||
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE knowledge_parse_tasks
|
||
SET status = 'completed', error_message = NULL, completed_at = NOW(), updated_at = NOW()
|
||
WHERE id = $1 AND tenant_id = $2
|
||
`, parseTaskID, tenantID); err != nil {
|
||
return err
|
||
}
|
||
|
||
return tx.Commit(ctx)
|
||
}
|
||
|
||
func (s *KnowledgeService) markKnowledgeIngestionFailed(
|
||
ctx context.Context,
|
||
tenantID, itemID, parseTaskID int64,
|
||
cause error,
|
||
) error {
|
||
errMessage := "knowledge ingestion failed"
|
||
if cause != nil {
|
||
errMessage = strings.TrimSpace(cause.Error())
|
||
}
|
||
|
||
if s.logger != nil {
|
||
s.logger.Error("knowledge parse failed",
|
||
zap.Int64("tenant_id", tenantID),
|
||
zap.Int64("knowledge_item_id", itemID),
|
||
zap.Int64("parse_task_id", parseTaskID),
|
||
zap.String("error_message", errMessage),
|
||
)
|
||
}
|
||
|
||
_, err := s.pool.Exec(ctx, `
|
||
UPDATE knowledge_items
|
||
SET status = 'failed', error_message = $1, updated_at = NOW()
|
||
WHERE id = $2 AND tenant_id = $3
|
||
`, errMessage, itemID, tenantID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
_, err = s.pool.Exec(ctx, `
|
||
UPDATE knowledge_parse_tasks
|
||
SET status = 'failed', error_message = $1, completed_at = NOW(), updated_at = NOW()
|
||
WHERE id = $2 AND tenant_id = $3
|
||
`, errMessage, parseTaskID, tenantID)
|
||
return err
|
||
}
|
||
|
||
func buildKnowledgeObjectKey(tenantID int64, sourceType, fileName string) string {
|
||
fileName = sanitizeKnowledgeFileName(fileName)
|
||
if fileName == "" {
|
||
fileName = sourceType + ".bin"
|
||
}
|
||
return fmt.Sprintf(
|
||
"knowledge/%d/%s/%s/%s-%s",
|
||
tenantID,
|
||
sourceType,
|
||
time.Now().UTC().Format("2006/01/02"),
|
||
uuid.NewString(),
|
||
fileName,
|
||
)
|
||
}
|
||
|
||
func sanitizeKnowledgeFileName(fileName string) string {
|
||
fileName = filepath.Base(strings.TrimSpace(fileName))
|
||
replacer := strings.NewReplacer(
|
||
"/", "-",
|
||
"\\", "-",
|
||
":", "-",
|
||
"?", "-",
|
||
"&", "-",
|
||
"=", "-",
|
||
"#", "-",
|
||
)
|
||
return replacer.Replace(fileName)
|
||
}
|
||
|
||
func detectKnowledgeContentType(fileName string) string {
|
||
switch strings.ToLower(filepath.Ext(fileName)) {
|
||
case ".txt", ".md", ".markdown":
|
||
return "text/plain; charset=utf-8"
|
||
case ".html", ".htm":
|
||
return "text/html; charset=utf-8"
|
||
case ".pdf":
|
||
return "application/pdf"
|
||
case ".docx":
|
||
return "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
|
||
case ".xlsx":
|
||
return "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||
case ".json":
|
||
return "application/json"
|
||
case ".csv":
|
||
return "text/csv; charset=utf-8"
|
||
default:
|
||
return "application/octet-stream"
|
||
}
|
||
}
|
||
|
||
func derefKnowledgeString(value *string) string {
|
||
if value == nil {
|
||
return ""
|
||
}
|
||
return strings.TrimSpace(*value)
|
||
}
|
||
|
||
func (s *KnowledgeService) ensureGroupAccessible(ctx context.Context, tenantID, groupID int64) error {
|
||
var exists bool
|
||
err := s.pool.QueryRow(ctx, `
|
||
SELECT EXISTS(
|
||
SELECT 1
|
||
FROM knowledge_groups
|
||
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
|
||
)
|
||
`, groupID, tenantID).Scan(&exists)
|
||
if err != nil {
|
||
return response.ErrInternal(50050, "knowledge_group_query_failed", "failed to query knowledge group")
|
||
}
|
||
if !exists {
|
||
return response.ErrNotFound(40451, "knowledge_group_not_found", "knowledge group not found")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *KnowledgeService) ensureRootGroup(ctx context.Context, tenantID, groupID int64) error {
|
||
if err := s.ensureGroupAccessible(ctx, tenantID, groupID); err != nil {
|
||
return err
|
||
}
|
||
|
||
var isRoot bool
|
||
err := s.pool.QueryRow(ctx, `
|
||
SELECT EXISTS(
|
||
SELECT 1
|
||
FROM knowledge_groups
|
||
WHERE id = $1
|
||
AND tenant_id = $2
|
||
AND deleted_at IS NULL
|
||
AND parent_id IS NULL
|
||
)
|
||
`, groupID, tenantID).Scan(&isRoot)
|
||
if err != nil {
|
||
return response.ErrInternal(50050, "knowledge_group_query_failed", "failed to query knowledge group")
|
||
}
|
||
if !isRoot {
|
||
return response.ErrBadRequest(40050, "knowledge_group_depth_exceeded", "third-level knowledge groups are not supported")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *KnowledgeService) ensureStorableGroup(ctx context.Context, tenantID, groupID int64) error {
|
||
if err := s.ensureGroupAccessible(ctx, tenantID, groupID); err != nil {
|
||
return err
|
||
}
|
||
|
||
var storable bool
|
||
err := s.pool.QueryRow(ctx, `
|
||
SELECT EXISTS(
|
||
SELECT 1
|
||
FROM knowledge_groups child
|
||
JOIN knowledge_groups parent
|
||
ON parent.id = child.parent_id
|
||
AND parent.tenant_id = child.tenant_id
|
||
AND parent.deleted_at IS NULL
|
||
WHERE child.id = $1
|
||
AND child.tenant_id = $2
|
||
AND child.deleted_at IS NULL
|
||
AND parent.parent_id IS NULL
|
||
)
|
||
`, groupID, tenantID).Scan(&storable)
|
||
if err != nil {
|
||
return response.ErrInternal(50050, "knowledge_group_query_failed", "failed to query knowledge group")
|
||
}
|
||
if !storable {
|
||
return response.ErrBadRequest(40053, "knowledge_group_storage_invalid", "knowledge items can only be stored in second-level groups")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *KnowledgeService) resolveKnowledgeStorageGroupID(ctx context.Context, tenantID, groupID int64) (int64, error) {
|
||
tx, err := s.pool.Begin(ctx)
|
||
if err != nil {
|
||
return 0, response.ErrInternal(50055, "knowledge_group_resolve_failed", "failed to begin knowledge group transaction")
|
||
}
|
||
defer func() { _ = tx.Rollback(ctx) }()
|
||
|
||
resolvedGroupID, err := s.resolveKnowledgeStorageGroupIDTx(ctx, tx, tenantID, groupID)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return 0, response.ErrInternal(50055, "knowledge_group_resolve_failed", "failed to commit knowledge group transaction")
|
||
}
|
||
return resolvedGroupID, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) resolveKnowledgeStorageGroupIDTx(
|
||
ctx context.Context,
|
||
tx pgx.Tx,
|
||
tenantID, groupID int64,
|
||
) (int64, error) {
|
||
var parentID *int64
|
||
if err := tx.QueryRow(ctx, `
|
||
SELECT parent_id
|
||
FROM knowledge_groups
|
||
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
|
||
FOR UPDATE
|
||
`, groupID, tenantID).Scan(&parentID); err != nil {
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return 0, response.ErrNotFound(40451, "knowledge_group_not_found", "knowledge group not found")
|
||
}
|
||
return 0, response.ErrInternal(50050, "knowledge_group_query_failed", "failed to query knowledge group")
|
||
}
|
||
|
||
if parentID != nil {
|
||
var grandParentID *int64
|
||
if err := tx.QueryRow(ctx, `
|
||
SELECT parent_id
|
||
FROM knowledge_groups
|
||
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
|
||
`, *parentID, tenantID).Scan(&grandParentID); err != nil {
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return 0, response.ErrNotFound(40451, "knowledge_group_not_found", "knowledge group not found")
|
||
}
|
||
return 0, response.ErrInternal(50050, "knowledge_group_query_failed", "failed to query knowledge group")
|
||
}
|
||
if grandParentID != nil {
|
||
return 0, response.ErrBadRequest(40053, "knowledge_group_storage_invalid", "knowledge items can only be stored in second-level groups")
|
||
}
|
||
return groupID, nil
|
||
}
|
||
|
||
var childGroupID int64
|
||
if err := tx.QueryRow(ctx, `
|
||
SELECT id
|
||
FROM knowledge_groups
|
||
WHERE tenant_id = $1
|
||
AND parent_id = $2
|
||
AND name = $3
|
||
AND deleted_at IS NULL
|
||
ORDER BY id
|
||
LIMIT 1
|
||
`, tenantID, groupID, defaultKnowledgeChildGroupName).Scan(&childGroupID); err == nil {
|
||
return childGroupID, nil
|
||
} else if !errors.Is(err, pgx.ErrNoRows) {
|
||
return 0, response.ErrInternal(50050, "knowledge_group_query_failed", "failed to query knowledge group")
|
||
}
|
||
|
||
if err := tx.QueryRow(ctx, `
|
||
INSERT INTO knowledge_groups (tenant_id, name, parent_id, sort_order)
|
||
VALUES ($1, $2, $3, 0)
|
||
RETURNING id
|
||
`, tenantID, defaultKnowledgeChildGroupName, groupID).Scan(&childGroupID); err == nil {
|
||
return childGroupID, nil
|
||
}
|
||
|
||
if err := tx.QueryRow(ctx, `
|
||
SELECT id
|
||
FROM knowledge_groups
|
||
WHERE tenant_id = $1
|
||
AND parent_id = $2
|
||
AND name = $3
|
||
AND deleted_at IS NULL
|
||
ORDER BY id
|
||
LIMIT 1
|
||
`, tenantID, groupID, defaultKnowledgeChildGroupName).Scan(&childGroupID); err != nil {
|
||
return 0, response.ErrConflict(40950, "knowledge_group_create_failed", "failed to create default knowledge group")
|
||
}
|
||
return childGroupID, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) listGroupSubtreeIDs(
|
||
ctx context.Context,
|
||
tx pgx.Tx,
|
||
tenantID, groupID int64,
|
||
) ([]int64, error) {
|
||
rows, err := tx.Query(ctx, `
|
||
WITH RECURSIVE selected_groups AS (
|
||
SELECT id
|
||
FROM knowledge_groups
|
||
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
|
||
UNION ALL
|
||
SELECT child.id
|
||
FROM knowledge_groups child
|
||
JOIN selected_groups parent ON child.parent_id = parent.id
|
||
WHERE child.tenant_id = $2 AND child.deleted_at IS NULL
|
||
)
|
||
SELECT id
|
||
FROM selected_groups
|
||
`, groupID, tenantID)
|
||
if err != nil {
|
||
return nil, response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to query knowledge groups")
|
||
}
|
||
defer rows.Close()
|
||
|
||
groupIDs := make([]int64, 0)
|
||
for rows.Next() {
|
||
var id int64
|
||
if err := rows.Scan(&id); err != nil {
|
||
return nil, response.ErrInternal(50054, "knowledge_group_delete_failed", err.Error())
|
||
}
|
||
groupIDs = append(groupIDs, id)
|
||
}
|
||
return groupIDs, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) listActiveKnowledgeItemIDs(
|
||
ctx context.Context,
|
||
tx pgx.Tx,
|
||
tenantID int64,
|
||
groupIDs []int64,
|
||
) ([]int64, error) {
|
||
if len(groupIDs) == 0 {
|
||
return []int64{}, nil
|
||
}
|
||
|
||
rows, err := tx.Query(ctx, `
|
||
SELECT id
|
||
FROM knowledge_items
|
||
WHERE tenant_id = $1
|
||
AND deleted_at IS NULL
|
||
AND group_id = ANY($2::bigint[])
|
||
`, tenantID, groupIDs)
|
||
if err != nil {
|
||
return nil, response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to query knowledge items")
|
||
}
|
||
defer rows.Close()
|
||
|
||
itemIDs := make([]int64, 0)
|
||
for rows.Next() {
|
||
var id int64
|
||
if err := rows.Scan(&id); err != nil {
|
||
return nil, response.ErrInternal(50054, "knowledge_group_delete_failed", err.Error())
|
||
}
|
||
itemIDs = append(itemIDs, id)
|
||
}
|
||
return itemIDs, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) embedChunks(ctx context.Context, chunks []string) ([][]float64, error) {
|
||
runtimeCfg := s.runtimeConfig()
|
||
results := make([][]float64, 0, len(chunks))
|
||
for start := 0; start < len(chunks); start += runtimeCfg.EmbeddingBatchSize {
|
||
end := start + runtimeCfg.EmbeddingBatchSize
|
||
if end > len(chunks) {
|
||
end = len(chunks)
|
||
}
|
||
vectors, err := s.provider.Embed(ctx, chunks[start:end])
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
results = append(results, vectors...)
|
||
}
|
||
if len(results) != len(chunks) {
|
||
return nil, errors.New("embedding result count mismatch")
|
||
}
|
||
return results, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) listKnowledgePointIDs(ctx context.Context, tenantID, itemID int64) ([]string, error) {
|
||
rows, err := s.pool.Query(ctx, `
|
||
SELECT qdrant_point_id
|
||
FROM knowledge_chunks_meta
|
||
WHERE tenant_id = $1 AND knowledge_item_id = $2 AND status <> 'deleted'
|
||
`, tenantID, itemID)
|
||
if err != nil {
|
||
return nil, response.ErrInternal(50054, "knowledge_item_delete_failed", "failed to query knowledge item points")
|
||
}
|
||
defer rows.Close()
|
||
|
||
pointIDs := make([]string, 0)
|
||
for rows.Next() {
|
||
var pointID string
|
||
if err := rows.Scan(&pointID); err != nil {
|
||
return nil, response.ErrInternal(50054, "knowledge_item_delete_failed", err.Error())
|
||
}
|
||
if strings.TrimSpace(pointID) != "" {
|
||
pointIDs = append(pointIDs, pointID)
|
||
}
|
||
}
|
||
return pointIDs, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) filterActiveKnowledgeSnippets(
|
||
ctx context.Context,
|
||
tenantID int64,
|
||
candidates []KnowledgeSnippet,
|
||
) ([]KnowledgeSnippet, error) {
|
||
if len(candidates) == 0 {
|
||
return []KnowledgeSnippet{}, nil
|
||
}
|
||
|
||
itemIDs := make([]int64, 0, len(candidates))
|
||
for _, candidate := range candidates {
|
||
if candidate.KnowledgeItemID > 0 {
|
||
itemIDs = append(itemIDs, candidate.KnowledgeItemID)
|
||
}
|
||
}
|
||
itemIDs = normalizeInt64IDs(itemIDs)
|
||
if len(itemIDs) == 0 {
|
||
return []KnowledgeSnippet{}, nil
|
||
}
|
||
|
||
rows, err := s.pool.Query(ctx, `
|
||
SELECT id
|
||
FROM knowledge_items
|
||
WHERE tenant_id = $1
|
||
AND id = ANY($2::bigint[])
|
||
AND status = 'completed'
|
||
AND deleted_at IS NULL
|
||
`, tenantID, itemIDs)
|
||
if err != nil {
|
||
return nil, response.ErrInternal(50051, "knowledge_items_query_failed", "failed to filter deleted knowledge items")
|
||
}
|
||
defer rows.Close()
|
||
|
||
active := make(map[int64]struct{}, len(itemIDs))
|
||
for rows.Next() {
|
||
var itemID int64
|
||
if err := rows.Scan(&itemID); err != nil {
|
||
return nil, response.ErrInternal(50051, "knowledge_items_scan_failed", err.Error())
|
||
}
|
||
active[itemID] = struct{}{}
|
||
}
|
||
|
||
filtered := make([]KnowledgeSnippet, 0, len(candidates))
|
||
for _, candidate := range candidates {
|
||
if _, ok := active[candidate.KnowledgeItemID]; ok {
|
||
filtered = append(filtered, candidate)
|
||
}
|
||
}
|
||
return filtered, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) cleanupDeletedKnowledgeItemAsync(tenantID, itemID int64) {
|
||
if !s.startCleanupJob(tenantID, itemID) {
|
||
return
|
||
}
|
||
|
||
go func() {
|
||
defer s.finishCleanupJob(tenantID, itemID)
|
||
|
||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||
defer cancel()
|
||
s.cleanupDeletedKnowledgeItem(ctx, tenantID, itemID)
|
||
}()
|
||
}
|
||
|
||
func (s *KnowledgeService) scheduleDeletedKnowledgeCleanup() {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||
defer cancel()
|
||
|
||
jobs, err := s.listKnowledgeCleanupCandidates(ctx, 20)
|
||
if err != nil {
|
||
return
|
||
}
|
||
|
||
for _, job := range jobs {
|
||
s.cleanupDeletedKnowledgeItemAsync(job.TenantID, job.ItemID)
|
||
}
|
||
}
|
||
|
||
func (s *KnowledgeService) RunDeletedCleanupSweep() {
|
||
if s == nil {
|
||
return
|
||
}
|
||
s.scheduleDeletedKnowledgeCleanup()
|
||
}
|
||
|
||
func (s *KnowledgeService) listKnowledgeCleanupCandidates(
|
||
ctx context.Context,
|
||
limit int,
|
||
) ([]knowledgeCleanupJob, error) {
|
||
if limit <= 0 {
|
||
limit = 20
|
||
}
|
||
|
||
rows, err := s.pool.Query(ctx, `
|
||
SELECT ki.tenant_id, ki.id
|
||
FROM knowledge_items ki
|
||
WHERE ki.deleted_at IS NOT NULL
|
||
ORDER BY ki.updated_at ASC, ki.id ASC
|
||
LIMIT $1
|
||
`, limit)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
jobs := make([]knowledgeCleanupJob, 0, limit)
|
||
for rows.Next() {
|
||
var job knowledgeCleanupJob
|
||
if err := rows.Scan(&job.TenantID, &job.ItemID); err != nil {
|
||
return nil, err
|
||
}
|
||
jobs = append(jobs, job)
|
||
}
|
||
return jobs, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) startCleanupJob(tenantID, itemID int64) bool {
|
||
key := fmt.Sprintf("%d:%d", tenantID, itemID)
|
||
|
||
s.cleanupMu.Lock()
|
||
defer s.cleanupMu.Unlock()
|
||
|
||
if _, exists := s.cleanupInFlight[key]; exists {
|
||
return false
|
||
}
|
||
s.cleanupInFlight[key] = struct{}{}
|
||
return true
|
||
}
|
||
|
||
func (s *KnowledgeService) finishCleanupJob(tenantID, itemID int64) {
|
||
key := fmt.Sprintf("%d:%d", tenantID, itemID)
|
||
|
||
s.cleanupMu.Lock()
|
||
delete(s.cleanupInFlight, key)
|
||
s.cleanupMu.Unlock()
|
||
}
|
||
|
||
func (s *KnowledgeService) cleanupDeletedKnowledgeItem(ctx context.Context, tenantID, itemID int64) {
|
||
item, err := s.loadKnowledgeItemForCleanup(ctx, tenantID, itemID)
|
||
if err != nil || item == nil {
|
||
return
|
||
}
|
||
|
||
errs := make([]string, 0, 2)
|
||
|
||
if strings.TrimSpace(item.StorageKey) != "" {
|
||
if err := s.objectStorage.Delete(ctx, item.StorageKey); err != nil {
|
||
errs = append(errs, err.Error())
|
||
} else {
|
||
_ = s.markKnowledgeStorageDeleted(ctx, tenantID, itemID)
|
||
}
|
||
}
|
||
|
||
pointIDs, err := s.listKnowledgePointIDs(ctx, tenantID, itemID)
|
||
if err != nil {
|
||
errs = append(errs, err.Error())
|
||
} else {
|
||
if len(pointIDs) > 0 {
|
||
if err := s.store.DeletePoints(ctx, pointIDs); err != nil {
|
||
errs = append(errs, err.Error())
|
||
}
|
||
}
|
||
}
|
||
|
||
if len(errs) > 0 {
|
||
if s.logger != nil {
|
||
s.logger.Error("knowledge cleanup failed",
|
||
zap.Int64("tenant_id", tenantID),
|
||
zap.Int64("knowledge_item_id", itemID),
|
||
zap.Strings("errors", errs),
|
||
)
|
||
}
|
||
_ = s.setKnowledgeCleanupError(ctx, tenantID, itemID, strings.Join(errs, "; "))
|
||
return
|
||
}
|
||
|
||
if err := s.deleteKnowledgeItemPermanently(ctx, tenantID, itemID); err != nil {
|
||
if s.logger != nil {
|
||
s.logger.Error("knowledge cleanup finalize failed",
|
||
zap.Int64("tenant_id", tenantID),
|
||
zap.Int64("knowledge_item_id", itemID),
|
||
zap.Error(err),
|
||
)
|
||
}
|
||
_ = s.setKnowledgeCleanupError(ctx, tenantID, itemID, err.Error())
|
||
return
|
||
}
|
||
|
||
if s.logger != nil {
|
||
s.logger.Info("knowledge cleanup completed",
|
||
zap.Int64("tenant_id", tenantID),
|
||
zap.Int64("knowledge_item_id", itemID),
|
||
)
|
||
}
|
||
_ = s.clearKnowledgeCleanupError(ctx, tenantID, itemID)
|
||
}
|
||
|
||
func (s *KnowledgeService) loadKnowledgeItemForCleanup(
|
||
ctx context.Context,
|
||
tenantID, itemID int64,
|
||
) (*knowledgeItemRecord, error) {
|
||
record := &knowledgeItemRecord{}
|
||
var contentText *string
|
||
var markdownContent *string
|
||
err := s.pool.QueryRow(ctx, `
|
||
SELECT ki.id, ki.tenant_id, ki.group_id, kg.name, ki.name, ki.source_type,
|
||
ki.source_uri, ki.storage_key, ki.content_text, ki.markdown_content, ki.item_version
|
||
FROM knowledge_items ki
|
||
JOIN knowledge_groups kg ON kg.id = ki.group_id
|
||
WHERE ki.id = $1 AND ki.tenant_id = $2 AND ki.deleted_at IS NOT NULL
|
||
`, itemID, tenantID).Scan(
|
||
&record.ID,
|
||
&record.TenantID,
|
||
&record.GroupID,
|
||
&record.GroupName,
|
||
&record.Name,
|
||
&record.SourceType,
|
||
&record.SourceURI,
|
||
&record.StorageKey,
|
||
&contentText,
|
||
&markdownContent,
|
||
&record.ItemVersion,
|
||
)
|
||
if err != nil {
|
||
return nil, nil
|
||
}
|
||
record.ContentText = contentText
|
||
record.Markdown = markdownContent
|
||
return record, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) markKnowledgeStorageDeleted(ctx context.Context, tenantID, itemID int64) error {
|
||
_, err := s.pool.Exec(ctx, `
|
||
UPDATE knowledge_items
|
||
SET storage_key = '', updated_at = NOW()
|
||
WHERE id = $1 AND tenant_id = $2
|
||
`, itemID, tenantID)
|
||
return err
|
||
}
|
||
|
||
func (s *KnowledgeService) deleteKnowledgeItemPermanently(ctx context.Context, tenantID, itemID int64) error {
|
||
_, err := s.pool.Exec(ctx, `
|
||
DELETE FROM knowledge_items
|
||
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NOT NULL
|
||
`, itemID, tenantID)
|
||
return err
|
||
}
|
||
|
||
func (s *KnowledgeService) markKnowledgeChunkMetaDeleted(ctx context.Context, tenantID, itemID int64) error {
|
||
_, err := s.pool.Exec(ctx, `
|
||
UPDATE knowledge_chunks_meta
|
||
SET status = 'deleted', updated_at = NOW()
|
||
WHERE knowledge_item_id = $1 AND tenant_id = $2 AND status <> 'deleted'
|
||
`, itemID, tenantID)
|
||
return err
|
||
}
|
||
|
||
func (s *KnowledgeService) setKnowledgeCleanupError(ctx context.Context, tenantID, itemID int64, message string) error {
|
||
_, err := s.pool.Exec(ctx, `
|
||
UPDATE knowledge_items
|
||
SET error_message = $1, updated_at = NOW()
|
||
WHERE id = $2 AND tenant_id = $3
|
||
`, strings.TrimSpace(message), itemID, tenantID)
|
||
return err
|
||
}
|
||
|
||
func (s *KnowledgeService) clearKnowledgeCleanupError(ctx context.Context, tenantID, itemID int64) error {
|
||
_, err := s.pool.Exec(ctx, `
|
||
UPDATE knowledge_items
|
||
SET error_message = NULL, updated_at = NOW()
|
||
WHERE id = $1 AND tenant_id = $2
|
||
`, itemID, tenantID)
|
||
return err
|
||
}
|
||
|
||
func (s *KnowledgeService) listKnowledgeItemsByIDs(ctx context.Context, tenantID int64, itemIDs []int64) ([]KnowledgeItemResponse, error) {
|
||
if len(itemIDs) == 0 {
|
||
return []KnowledgeItemResponse{}, nil
|
||
}
|
||
rows, err := s.pool.Query(ctx, `
|
||
SELECT ki.id, ki.group_id, kg.name, ki.name, ki.source_type, ki.source_uri, ki.status,
|
||
ki.size_bytes, ki.error_message, ki.created_at, ki.updated_at
|
||
FROM knowledge_items ki
|
||
JOIN knowledge_groups kg ON kg.id = ki.group_id
|
||
WHERE ki.tenant_id = $1 AND ki.id = ANY($2::bigint[]) AND ki.deleted_at IS NULL
|
||
ORDER BY ki.created_at DESC, ki.id DESC
|
||
`, tenantID, itemIDs)
|
||
if err != nil {
|
||
return nil, response.ErrInternal(50051, "knowledge_items_query_failed", "failed to list knowledge items")
|
||
}
|
||
defer rows.Close()
|
||
|
||
items := make([]KnowledgeItemResponse, 0)
|
||
for rows.Next() {
|
||
var (
|
||
item KnowledgeItemResponse
|
||
sourceURI *string
|
||
errorMessage *string
|
||
createdAt, updatedAt time.Time
|
||
)
|
||
if err := rows.Scan(
|
||
&item.ID,
|
||
&item.GroupID,
|
||
&item.GroupName,
|
||
&item.Name,
|
||
&item.SourceType,
|
||
&sourceURI,
|
||
&item.Status,
|
||
&item.SizeBytes,
|
||
&errorMessage,
|
||
&createdAt,
|
||
&updatedAt,
|
||
); err != nil {
|
||
return nil, response.ErrInternal(50051, "knowledge_items_scan_failed", err.Error())
|
||
}
|
||
item.SourceURI = sourceURI
|
||
item.ErrorMessage = errorMessage
|
||
item.CreatedAt = createdAt.Format(time.RFC3339)
|
||
item.UpdatedAt = updatedAt.Format(time.RFC3339)
|
||
items = append(items, item)
|
||
}
|
||
return items, nil
|
||
}
|
||
|
||
func (s *KnowledgeService) resolveKnowledgeSearchScope(ctx context.Context, tenantID int64, groupIDs []int64) ([]int64, []string, error) {
|
||
groupIDs = normalizeInt64IDs(groupIDs)
|
||
if len(groupIDs) == 0 {
|
||
return []int64{}, []string{}, nil
|
||
}
|
||
|
||
rows, err := s.pool.Query(ctx, `
|
||
WITH RECURSIVE selected_groups AS (
|
||
SELECT id, name, parent_id, sort_order, created_at
|
||
FROM knowledge_groups
|
||
WHERE tenant_id = $1
|
||
AND id = ANY($2::bigint[])
|
||
AND deleted_at IS NULL
|
||
UNION
|
||
SELECT child.id, child.name, child.parent_id, child.sort_order, child.created_at
|
||
FROM knowledge_groups child
|
||
JOIN selected_groups parent ON child.parent_id = parent.id
|
||
WHERE child.tenant_id = $1
|
||
AND child.deleted_at IS NULL
|
||
)
|
||
SELECT id, name
|
||
FROM selected_groups
|
||
ORDER BY sort_order, created_at, id
|
||
`, tenantID, groupIDs)
|
||
if err != nil {
|
||
return nil, nil, response.ErrInternal(50050, "knowledge_group_query_failed", "failed to query knowledge groups")
|
||
}
|
||
defer rows.Close()
|
||
|
||
resolvedIDs := make([]int64, 0, len(groupIDs))
|
||
names := make([]string, 0, len(groupIDs))
|
||
seenIDs := make(map[int64]struct{}, len(groupIDs))
|
||
seenNames := make(map[string]struct{}, len(groupIDs))
|
||
for rows.Next() {
|
||
var (
|
||
groupID int64
|
||
name string
|
||
)
|
||
if err := rows.Scan(&groupID, &name); err != nil {
|
||
return nil, nil, response.ErrInternal(50050, "knowledge_group_query_failed", err.Error())
|
||
}
|
||
if groupID > 0 {
|
||
if _, exists := seenIDs[groupID]; !exists {
|
||
seenIDs[groupID] = struct{}{}
|
||
resolvedIDs = append(resolvedIDs, groupID)
|
||
}
|
||
}
|
||
name = strings.TrimSpace(name)
|
||
if name != "" {
|
||
if _, exists := seenNames[name]; !exists {
|
||
seenNames[name] = struct{}{}
|
||
names = append(names, name)
|
||
}
|
||
}
|
||
}
|
||
return resolvedIDs, names, nil
|
||
}
|
||
|
||
func normalizeInt64IDs(ids []int64) []int64 {
|
||
if len(ids) == 0 {
|
||
return nil
|
||
}
|
||
set := make(map[int64]struct{}, len(ids))
|
||
result := make([]int64, 0, len(ids))
|
||
for _, id := range ids {
|
||
if id <= 0 {
|
||
continue
|
||
}
|
||
if _, exists := set[id]; exists {
|
||
continue
|
||
}
|
||
set[id] = struct{}{}
|
||
result = append(result, id)
|
||
}
|
||
slices.Sort(result)
|
||
return result
|
||
}
|
||
|
||
func anyString(value any) string {
|
||
switch typed := value.(type) {
|
||
case nil:
|
||
return ""
|
||
case string:
|
||
return typed
|
||
case []byte:
|
||
return string(typed)
|
||
default:
|
||
return fmt.Sprintf("%v", typed)
|
||
}
|
||
}
|
||
|
||
func anyInt64(value any) int64 {
|
||
switch typed := value.(type) {
|
||
case int64:
|
||
return typed
|
||
case int32:
|
||
return int64(typed)
|
||
case int:
|
||
return int64(typed)
|
||
case float64:
|
||
return int64(typed)
|
||
case float32:
|
||
return int64(typed)
|
||
default:
|
||
return 0
|
||
}
|
||
}
|
||
|
||
func minInt(a, b int) int {
|
||
if a < b {
|
||
return a
|
||
}
|
||
return b
|
||
}
|