查看知识库及数据源

查询已有知识库，并使用返回的知识库 ID 检查数据源和处理任务。创建语义模型及条目介绍同一资源在结构化语义配置中的创建与校验，本页侧重已有资源的数据源检查。

准备

1. 配置 SDK 和个人访问令牌。

2. 选择工作区并获取工作区 ID。

列出知识库并检查数据源

示例读取一页知识库，显示 ID、名称和类型，再检查首个知识库。知识库 ID 在清单中为整数，详情方法接收字符串，示例显式转换。

Python

import os

import moi_product_sdk as sdk

client = sdk.new_with_personal_access_token(
    os.environ["PRODUCT_API_BASE_URL"], os.environ["PRODUCT_API_KEY"]
)
workspace_id = os.environ["WORKSPACE_ID"]
workspace = client.workspace(workspace_id)
knowledge = workspace.knowledge()
page = knowledge.list()
print("总数:", page.total, "下一页标记:", page.next_page_token)
for item in page.items:
    print(item.id, item.name, item.semantic_model_kind)
if not page.items:
    raise SystemExit("当前页没有知识库")
model_id = str(page.items[0].id)
info = knowledge.get(model_id)
sources = knowledge.list_sources(model_id)
jobs = knowledge.list_source_jobs(model_id)
print("知识库:", info.id, info.name)
for source in sources.items:
    print("数据源:", source.source_id, source.display_name, source.ingest_status)
for job in jobs.items:
    print("处理任务:", job.job_id, job.source_id, job.job_status, job.error)

Go

package main

import (
	"context"
	"fmt"
	sdk "github.com/matrixorigin/matrixflow/sdk/go-sdk"
	"os"
	"strconv"
)

func main() {
	ctx := context.Background()
	client, err := sdk.NewWithPersonalAccessToken(os.Getenv("PRODUCT_API_BASE_URL"), os.Getenv("PRODUCT_API_KEY"))
	if err != nil {
		panic(err)
	}
	workspaceID := os.Getenv("WORKSPACE_ID")
	workspace, err := client.Workspace(workspaceID)
	if err != nil {
		panic(err)
	}
	_ = workspace
	knowledge := workspace.Knowledge()
	page, err := knowledge.List(ctx)
	if err != nil {
		panic(err)
	}
	fmt.Println("总数:", page.GetTotal(), "下一页标记:", page.GetNextPageToken())
	for _, item := range page.GetItems() {
		fmt.Println(item.GetId(), item.GetName(), item.GetSemanticModelKind())
	}
	if len(page.GetItems()) == 0 {
		fmt.Println("当前页没有知识库")
		return
	}
	modelID := strconv.FormatInt(page.GetItems()[0].GetId(), 10)
	info, err := knowledge.Get(ctx, modelID)
	if err != nil {
		panic(err)
	}
	sources, err := knowledge.ListSources(ctx, modelID)
	if err != nil {
		panic(err)
	}
	jobs, err := knowledge.ListSourceJobs(ctx, modelID)
	if err != nil {
		panic(err)
	}
	fmt.Println("知识库:", info.GetId(), info.GetName())
	for _, source := range sources.GetItems() {
		fmt.Println("数据源:", source.GetSourceId(), source.GetDisplayName(), source.GetIngestStatus())
	}
	for _, job := range jobs.GetItems() {
		fmt.Println("处理任务:", job.GetJobId(), job.GetSourceId(), job.GetJobStatus(), job.GetError())
	}
}

查看所选知识库的数据源处理状态、任务状态和错误信息。字段含义见知识库数据源。

最后更新于 2026 年 11 月 11 日