root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add knowledge markdown formatting and detail retrieval

Browse Source 
- Implemented KnowledgeWebsiteMarkdownPrompt for formatting webpage content into Markdown.
- Added GetItemDetail method in KnowledgeHandler to retrieve knowledge item details.
- Introduced KnowledgeDetailDrawer component for displaying knowledge item details in the admin web interface.
- Created MarkdownPreview component for rendering Markdown content.
- Added knowledge chunking and URL parsing logic to handle webpage content extraction and formatting.
- Updated database schema to include markdown_content in knowledge_items table.
This commit is contained in:
liangxu
2026-04-05 20:41:42 +08:00
parent 446f865cdf
commit dbd7747742
26 changed files with 1556 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/knowledge_service.go
+258 -47
View File
@@ -4,8 +4,10 @@ import (
"context"
"errors"
"fmt"
"net/http"
"net/url"
"path/filepath"
"runtime/debug"
"slices"
"strings"
"sync"
@@ -18,6 +20,7 @@ import (
"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"
@@ -25,8 +28,9 @@ import (
)
const (
defaultKnowledgeQueueSize = 128
defaultKnowledgeCleanupPoll = 15 * time.Second
defaultKnowledgeQueueSize = 128
defaultKnowledgeCleanupPoll = 15 * time.Second
defaultKnowledgeChildGroupName = "默认目录"
)
type KnowledgeService struct {
@@ -34,12 +38,17 @@ type KnowledgeService struct {
provider retrieval.Provider
store retrieval.VectorStore
objectStorage objectstorage.Client
llmClient llm.Client
logger *zap.Logger
chunkSize int
chunkOverlap int
embeddingBatchSize int
recallLimit int
rerankTopN int
arkBaseURL string
arkAPIKey string
urlMarkdownModel string
httpClient *http.Client
jobs chan knowledgeParseJob
cleanupMu sync.Mutex
cleanupInFlight map[string]struct{}
@@ -66,6 +75,7 @@ type knowledgeItemRecord struct {
SourceURI *string
StorageKey string
ContentText *string
Markdown *string
ItemVersion int
}
@@ -99,6 +109,21 @@ type KnowledgeItemResponse struct {
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"`
@@ -132,8 +157,10 @@ func NewKnowledgeService(
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 {
@@ -163,14 +190,25 @@ func NewKnowledgeService(
provider: provider,
store: store,
objectStorage: objectStorage,
llmClient: llmClient,
logger: logger,
chunkSize: chunkSize,
chunkOverlap: chunkOverlap,
embeddingBatchSize: batchSize,
recallLimit: recallLimit,
rerankTopN: rerankTopN,
arkBaseURL: strings.TrimSpace(llmCfg.BaseURL),
arkAPIKey: strings.TrimSpace(llmCfg.APIKey),
urlMarkdownModel: strings.TrimSpace(llmCfg.KnowledgeURLModel),
httpClient: &http.Client{Timeout: defaultKnowledgeURLParseTimeout},
cleanupInFlight: make(map[string]struct{}),
}
if svc.urlMarkdownModel == "" {
svc.urlMarkdownModel = strings.TrimSpace(llmCfg.Model)
}
if svc.arkBaseURL == "" {
svc.arkBaseURL = defaultKnowledgeArkBaseURL
}
if workerCount > 0 {
if queueSize <= 0 {
@@ -521,15 +559,58 @@ func (s *KnowledgeService) ListItems(ctx context.Context, groupID *int64) ([]Kno
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
}
if err := s.ensureStorableGroup(ctx, actor.TenantID, req.GroupID); err != nil {
return nil, err
}
content := normalizeKnowledgeText(req.Content)
if content == "" {
@@ -551,33 +632,17 @@ func (s *KnowledgeService) CreateURLItem(ctx context.Context, req KnowledgeURLIt
if err := s.validateIngestionDependencies("website"); err != nil {
return nil, err
}
if err := s.ensureStorableGroup(ctx, actor.TenantID, req.GroupID); err != nil {
return nil, err
}
rawURL := strings.TrimSpace(req.URL)
parsed, err := url.ParseRequestURI(rawURL)
if err != nil {
if _, err := url.ParseRequestURI(rawURL); err != nil {
return nil, response.ErrBadRequest(40052, "knowledge_url_invalid", "knowledge url is invalid")
}
snapshot, contentType, err := fetchKnowledgeURLSnapshot(ctx, parsed.String())
if err != nil {
return nil, response.ErrBadRequest(40052, "knowledge_url_fetch_failed", err.Error())
}
objectKey := buildKnowledgeObjectKey(actor.TenantID, "website", resolveSnapshotFileName(parsed.String(), contentType))
if err := s.objectStorage.PutBytes(ctx, objectKey, snapshot, contentType); err != nil {
return nil, response.ErrServiceUnavailable(50357, "knowledge_storage_unavailable", err.Error())
}
return s.queueKnowledgeItem(ctx, actor.TenantID, knowledgeIngestInput{
GroupID: req.GroupID,
Name: strings.TrimSpace(req.Name),
SourceType: "website",
SourceURI: &rawURL,
StorageKey: objectKey,
SizeBytes: int64(len(snapshot)),
})
}
@@ -593,9 +658,6 @@ func (s *KnowledgeService) CreateFileItem(
if err := s.validateIngestionDependencies("document"); err != nil {
return nil, err
}
if err := s.ensureStorableGroup(ctx, actor.TenantID, groupID); err != nil {
return nil, err
}
displayName := strings.TrimSpace(name)
if displayName == "" {
@@ -819,11 +881,19 @@ func (s *KnowledgeService) validateIngestionDependencies(sourceType string) erro
if err := s.store.Validate(); err != nil {
return response.ErrServiceUnavailable(50352, "qdrant_unavailable", err.Error())
}
if sourceType == "document" || sourceType == "website" {
if sourceType == "document" {
if err := s.objectStorage.Validate(); err != nil {
return response.ErrServiceUnavailable(50357, "knowledge_storage_unavailable", err.Error())
}
}
if sourceType == "website" {
if strings.TrimSpace(s.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
}
@@ -832,9 +902,11 @@ func (s *KnowledgeService) queueKnowledgeItem(
tenantID int64,
input knowledgeIngestInput,
) (*KnowledgeItemResponse, error) {
if err := s.ensureStorableGroup(ctx, tenantID, input.GroupID); err != nil {
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 == "" {
@@ -888,7 +960,25 @@ func (s *KnowledgeService) enqueueParseJob(job knowledgeParseJob) error {
func (s *KnowledgeService) runParseWorker() {
for job := range s.jobs {
s.executeParseJob(context.Background(), job)
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)
}
}
@@ -936,13 +1026,18 @@ func (s *KnowledgeService) executeParseJob(ctx context.Context, job knowledgePar
return
}
content, err := s.readKnowledgeItemContent(ctx, item)
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
}
}
chunks := splitKnowledgeTextIntoChunks(content, s.chunkSize, s.chunkOverlap)
chunks := splitKnowledgeTextIntoChunks(parsedContent.Text, s.chunkSize, s.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)
@@ -1018,7 +1113,7 @@ func (s *KnowledgeService) executeParseJob(ctx context.Context, job knowledgePar
return
}
if err = s.markKnowledgeIngestionCompleted(ctx, job.TenantID, job.ItemID, job.ParseTaskID, metas); err != nil {
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
}
@@ -1030,9 +1125,10 @@ func (s *KnowledgeService) loadKnowledgeItemForParse(
) (*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.item_version
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
@@ -1046,37 +1142,38 @@ func (s *KnowledgeService) loadKnowledgeItemForParse(
&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) readKnowledgeItemContent(ctx context.Context, item *knowledgeItemRecord) (string, error) {
func (s *KnowledgeService) readKnowledgeItemParsedContent(ctx context.Context, item *knowledgeItemRecord) (*knowledgeParsedContent, error) {
if item == nil {
return "", fmt.Errorf("knowledge item is empty")
return nil, fmt.Errorf("knowledge item is empty")
}
switch item.SourceType {
case "text":
return normalizeKnowledgeText(derefKnowledgeString(item.ContentText)), nil
return parseKnowledgeTextInput(item.Name, derefKnowledgeString(item.ContentText)), nil
case "document":
data, err := s.objectStorage.GetBytes(ctx, item.StorageKey)
if err != nil {
return "", err
return nil, err
}
return extractKnowledgeTextFromFile(item.StorageKey, data)
return extractKnowledgeContentFromFile(item.StorageKey, data)
case "website":
data, err := s.objectStorage.GetBytes(ctx, item.StorageKey)
if err != nil {
return "", err
if item.SourceURI == nil || strings.TrimSpace(*item.SourceURI) == "" {
return nil, fmt.Errorf("knowledge website url is empty")
}
return extractKnowledgeTextFromSnapshot(item.StorageKey, data)
return s.parseWebsiteKnowledge(ctx, strings.TrimSpace(*item.SourceURI))
default:
return "", fmt.Errorf("unsupported knowledge source type %q", item.SourceType)
return nil, fmt.Errorf("unsupported knowledge source type %q", item.SourceType)
}
}
@@ -1103,9 +1200,9 @@ func (s *KnowledgeService) createPendingKnowledgeItem(
err = tx.QueryRow(ctx, `
INSERT INTO knowledge_items (
tenant_id, group_id, source_type, name, source_uri, storage_key,
content_text, status, size_bytes, item_version
content_text, markdown_content, status, size_bytes, item_version
)
VALUES ($1, $2, $3, $4, $5, $6, $7, 'pending', $8, 1)
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 {
@@ -1159,8 +1256,21 @@ func (s *KnowledgeService) markKnowledgeIngestionRunning(
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
@@ -1188,9 +1298,14 @@ func (s *KnowledgeService) markKnowledgeIngestionCompleted(
if _, err := tx.Exec(ctx, `
UPDATE knowledge_items
SET status = 'completed', error_message = NULL, updated_at = NOW()
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); err != nil {
`, itemID, tenantID, itemName, content.Text, content.Markdown); err != nil {
return err
}
@@ -1370,6 +1485,99 @@ func (s *KnowledgeService) ensureStorableGroup(ctx context.Context, tenantID, gr
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,
@@ -1688,9 +1896,10 @@ func (s *KnowledgeService) loadKnowledgeItemForCleanup(
) (*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.item_version
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
@@ -1704,12 +1913,14 @@ func (s *KnowledgeService) loadKnowledgeItemForCleanup(
&record.SourceURI,
&record.StorageKey,
&contentText,
&markdownContent,
&record.ItemVersion,
)
if err != nil {
return nil, nil
}
record.ContentText = contentText
record.Markdown = markdownContent
return record, nil
}