root
7f08d92958
Adds a new browser-fetch microservice that wraps Lightpanda for rendering JS-heavy pages, and wires it into the knowledge URL parser as a fallback when the upstream link reader returns 403/429/5xx for an allow-listed domain (e.g. zhuanlan.zhihu.com). Includes config/env plumbing, hot-reload diff support, and full deploy assets (Dockerfile target, docker-compose, k3s manifests, image-build/package scripts).
2949 lines
85 KiB
Go
2949 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{}
|
||
browserFetchMu sync.Mutex
|
||
browserFetchQ *browserFetchQueue
|
||
}
|
||
|
||
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 knowledgeRuntimeFromFullConfig(cfg)
|
||
}
|
||
}
|
||
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
|
||
}
|