44d6175559
Lets users stop an in-flight document parse to free up LLM / worker
resources without losing the chunks and index already written. The
core insight is that the previous parse_status=completed flipped as
soon as primary chunks landed, while the most expensive subtasks
(graph extract = N LLM calls per chunk, plus summary, question
generation) were still running in the background — so "completed"
wasn't actually terminal from a resource standpoint.
State machine
pending -> processing -> finalizing -> completed
|
+-> cancelled (any of the three
in-flight states)
+-> failed
+-> deleting
`finalizing` is the new post-process fan-out window. parse_status
only promotes to `completed` once pending_subtasks_count (a new
column tracking summary + question + per-chunk graph extract)
drains to zero via atomic FinalizeSubtask. Wiki ingest is
intentionally excluded from the counter — it's a KB-scoped
debounced batch and would otherwise pin parse_status in
`finalizing` for the wiki batch window.
Backend
- New ParseStatusFinalizing + pending_subtasks_count column with
migration 000056.
- knowledgeRepository.SetFinalizing transitions processing -> finalizing
conditionally so a racing cancel cannot be clobbered.
- knowledgeRepository.FinalizeSubtask atomically decrements the
counter and self-promotes the row to completed when it hits zero.
- KnowledgePostProcess restructured to compute expected subtask
count up front, flip to finalizing (or completed when no
enrichment is enabled), and only then fan out subtasks. Subtask
handlers (summary, question, graph extract) defer-decrement on
terminal exit using the existing isFinalAsynqAttempt convention.
- New POST /api/v1/knowledge/{id}/cancel-parse handler accepting
pending / processing / finalizing. Marks the row cancelled,
zeroes the counter, best-effort dequeues asynq tasks via a new
TaskInspector abstraction (asynq-mode walks pending/scheduled/
retry queues; Lite-mode noop), and scrubs wiki ingest pending op.
- SpanTracker.AbortAttempt flat-sweeps every still-running span
for the attempt via a new repo.CancelAllOpenSpans helper so the
trace viewer's striped bars all flip to cancelled, even leaf
generations whose parent stage already EndSpan'd (multimodal
fan-out pattern). knowledge_post_process closes its postSpan
via SkipSpan on the cancel/deleting entry guard so a worker
that opens a span AFTER the cancel sweep doesn't leak it.
- Housekeeping and resetPendingTasks sweep finalizing rows
identically to processing so a crash/restart can't strand them.
- DeleteKnowledge/DeleteKnowledgeList proactively dequeue
downstream tasks via the same TaskInspector path.
- ChunkExtractService gets a cancel entry guard so the most
expensive enrichment (graph extract) bails immediately when the
parent knowledge is aborted.
Frontend
- New cancelKnowledgeParse API client + "Stop parsing" entry in
both list view and card view more menus, gated on
pending/processing/finalizing.
- Polling predicate refactored to a shared isParseInFlight helper
that recognises `finalizing` (previously the doc list silently
stopped polling once parse_status flipped from processing).
- Knowledge processing timeline: isPolling includes finalizing,
new isHardTerminal short-circuits LIVE for cancelled/failed/
completed so stranded child spans cannot pin LIVE on.
- DocumentListView.computeStatus distinguishes finalizing
("增强中") from completed and shows the previous "生成摘要中"
copy when summary_status is still pending under finalizing.
Added cancelled badge as well.
- i18n: statusFinalizing / statusCancelled / cancelParse* keys
across zh-CN, en-US, ko-KR, ru-RU.
Docs / SDK
- docs/api/knowledge.md: documents the new finalizing state,
cancel-parse semantics, and which statuses accept cancel.
- client (Go SDK): CancelKnowledgeParse with docstring listing
the cancellable statuses.
11 KiB
11 KiB
WeKnora HTTP 客户端
这个包提供了与WeKnora服务进行交互的客户端库,支持所有基于HTTP的接口调用,使其他模块更方便地集成WeKnora服务,无需直接编写HTTP请求代码。
主要功能
该客户端包含以下主要功能模块:
- 会话管理:创建、获取、更新和删除会话
- 知识库管理:创建、获取、更新和删除知识库
- 知识管理:添加、获取和删除知识内容
- 租户管理:租户的CRUD操作
- 知识问答:支持普通问答和流式问答
- Agent问答:支持基于Agent的智能问答,包含思考过程、工具调用和反思
- 分块管理:查询、更新和删除知识分块
- 消息管理:获取和删除会话消息
- 模型管理:创建、获取、更新和删除模型
使用方法
创建客户端实例
import (
"context"
"github.com/Tencent/WeKnora/client"
"time"
)
// 创建客户端实例
apiClient := client.NewClient(
"http://api.example.com",
client.WithToken("your-auth-token"),
client.WithTimeout(30*time.Second),
)
租户配置
客户端支持通过 WithTenantID 设置默认租户,请求时会自动携带 X-Tenant-ID 请求头:
tenantID := uint64(10000)
apiClient := client.NewClient(
"http://api.example.com",
client.WithToken("your-auth-token"),
client.WithTenantID(tenantID),
)
如果某个请求需要临时切换租户,可以在 context 中设置 TenantID,值可以是 uint64、*uint64 或字符串形式的数字,客户端会优先使用该值:
ctx := context.WithValue(context.Background(), "TenantID", uint64(10000))
// 调用任意客户端方法时传入 ctx,即可切换到租户 10000
示例:创建知识库并上传文件
// 创建知识库
kb := &client.KnowledgeBase{
Name: "测试知识库",
Description: "这是一个测试知识库",
ChunkingConfig: client.ChunkingConfig{
ChunkSize: 500,
ChunkOverlap: 50,
Separators: []string{"\n\n", "\n", ". ", "? ", "! "},
},
ImageProcessingConfig: client.ImageProcessingConfig{
ModelID: "image_model_id",
},
EmbeddingModelID: "embedding_model_id",
SummaryModelID: "summary_model_id",
}
kb, err := apiClient.CreateKnowledgeBase(context.Background(), kb)
if err != nil {
// 处理错误
}
// 上传知识文件并添加元数据
metadata := map[string]string{
"source": "local",
"type": "document",
}
knowledge, err := apiClient.CreateKnowledgeFromFile(context.Background(), kb.ID, "path/to/file.pdf", metadata)
if err != nil {
// 处理错误
}
示例:创建会话并进行问答
// 创建会话
sessionRequest := &client.CreateSessionRequest{
KnowledgeBaseID: knowledgeBaseID,
SessionStrategy: &client.SessionStrategy{
MaxRounds: 10,
EnableRewrite: true,
FallbackStrategy: "fixed_answer",
FallbackResponse: "抱歉,我无法回答这个问题",
EmbeddingTopK: 5,
KeywordThreshold: 0.5,
VectorThreshold: 0.7,
RerankModelID: "rerank_model_id",
RerankTopK: 3,
RerankThreshold: 0.8,
SummaryModelID: "summary_model_id",
},
}
session, err := apiClient.CreateSession(context.Background(), sessionRequest)
if err != nil {
// 处理错误
}
// 普通问答
answer, err := apiClient.KnowledgeQA(context.Background(), session.ID, &client.KnowledgeQARequest{
Query: "什么是人工智能?",
})
if err != nil {
// 处理错误
}
// 流式问答
err = apiClient.KnowledgeQAStream(context.Background(), session.ID, &client.KnowledgeQARequest{
Query: "什么是机器学习?",
KnowledgeBaseIDs: []string{knowledgeBaseID}, // 可选:指定知识库
WebSearchEnabled: false, // 可选:是否启用网络搜索
}, func(response *client.StreamResponse) error {
// 处理每个响应片段
fmt.Print(response.Content)
return nil
})
if err != nil {
// 处理错误
}
示例:Agent智能问答
Agent问答提供更强大的智能对话能力,支持工具调用、思考过程展示和自我反思。
// 创建Agent会话
agentSession := apiClient.NewAgentSession(session.ID)
// 进行Agent问答,带完整事件处理
err := agentSession.Ask(context.Background(), "搜索机器学习相关知识并总结要点",
func(resp *client.AgentStreamResponse) error {
switch resp.ResponseType {
case client.AgentResponseTypeThinking:
// Agent正在思考
if resp.Done {
fmt.Printf("💭 思考: %s\n", resp.Content)
}
case client.AgentResponseTypeToolCall:
// Agent调用工具
if resp.Data != nil {
toolName := resp.Data["tool_name"]
fmt.Printf("🔧 调用工具: %v\n", toolName)
}
case client.AgentResponseTypeToolResult:
// 工具执行结果
fmt.Printf("✓ 工具结果: %s\n", resp.Content)
case client.AgentResponseTypeReferences:
// 知识引用
if resp.KnowledgeReferences != nil {
fmt.Printf("📚 找到 %d 条相关知识\n", len(resp.KnowledgeReferences))
for _, ref := range resp.KnowledgeReferences {
fmt.Printf(" - [%.3f] %s\n", ref.Score, ref.KnowledgeTitle)
}
}
case client.AgentResponseTypeAnswer:
// 最终答案(流式输出)
fmt.Print(resp.Content)
if resp.Done {
fmt.Println() // 结束后换行
}
case client.AgentResponseTypeReflection:
// Agent的自我反思
if resp.Done {
fmt.Printf("🤔 反思: %s\n", resp.Content)
}
case client.AgentResponseTypeError:
// 错误信息
fmt.Printf("❌ 错误: %s\n", resp.Content)
}
return nil
})
if err != nil {
// 处理错误
}
// 简化版:只关心最终答案
var finalAnswer string
err = agentSession.Ask(context.Background(), "什么是深度学习?",
func(resp *client.AgentStreamResponse) error {
if resp.ResponseType == client.AgentResponseTypeAnswer {
finalAnswer += resp.Content
}
return nil
})
Agent事件类型说明
| 事件类型 | 说明 | 何时触发 |
|---|---|---|
AgentResponseTypeThinking |
Agent思考过程 | Agent分析问题和制定计划时 |
AgentResponseTypeToolCall |
工具调用 | Agent决定使用某个工具时 |
AgentResponseTypeToolResult |
工具执行结果 | 工具执行完成后 |
AgentResponseTypeReferences |
知识引用 | 检索到相关知识时 |
AgentResponseTypeAnswer |
最终答案 | Agent生成回答时(流式) |
AgentResponseTypeReflection |
自我反思 | Agent评估自己的回答时 |
AgentResponseTypeError |
错误 | 发生错误时 |
Agent问答测试工具
我们提供了一个交互式命令行工具用于测试Agent功能:
cd client/cmd/agent_test
go build -o agent_test
./agent_test -url http://localhost:8080 -kb <knowledge_base_id>
该工具支持:
- 创建和管理会话
- 交互式Agent问答
- 实时显示所有Agent事件
- 性能统计和调试信息
详细使用说明请参考 client/cmd/agent_test/README.md。
Agent问答的高级用法
更多高级用法示例,请参考 agent_example.go 文件,包括:
- 基础Agent问答
- 工具调用跟踪
- 知识引用捕获
- 完整事件跟踪
- 自定义错误处理
- 流取消控制
- 多会话管理
### 示例:管理模型
```go
// 创建模型
modelRequest := &client.CreateModelRequest{
Name: "测试模型",
Type: client.ModelTypeChat,
Source: client.ModelSourceInternal,
Description: "这是一个测试模型",
Parameters: client.ModelParameters{
"temperature": 0.7,
"top_p": 0.9,
},
IsDefault: true,
}
model, err := apiClient.CreateModel(context.Background(), modelRequest)
if err != nil {
// 处理错误
}
// 列出所有模型
models, err := apiClient.ListModels(context.Background())
if err != nil {
// 处理错误
}
示例:管理知识分块
// 列出知识分块
chunks, total, err := apiClient.ListKnowledgeChunks(context.Background(), knowledgeID, 1, 10)
if err != nil {
// 处理错误
}
// 更新分块
updateRequest := &client.UpdateChunkRequest{
Content: "更新后的分块内容",
IsEnabled: true,
}
updatedChunk, err := apiClient.UpdateChunk(context.Background(), knowledgeID, chunkID, updateRequest)
if err != nil {
// 处理错误
}
示例:重新解析知识
// 重新解析知识(删除现有内容并重新解析)
// 适用场景:
// 1. 原始解析失败,需要重试
// 2. 更新了解析配置(如分块策略、多模态设置等),需要重新解析
// 3. 知识内容已更新,需要刷新解析结果
knowledge, err := apiClient.ReparseKnowledge(context.Background(), knowledgeID)
if err != nil {
// 处理错误
}
// 知识将进入 "pending" 状态,异步重新解析
fmt.Printf("Knowledge ID: %s\n", knowledge.ID)
fmt.Printf("Parse Status: %s\n", knowledge.ParseStatus) // "pending"
fmt.Printf("Enable Status: %s\n", knowledge.EnableStatus) // "disabled"
// 可以轮询检查解析状态
for {
time.Sleep(5 * time.Second)
knowledge, err := apiClient.GetKnowledge(context.Background(), knowledgeID)
if err != nil {
// 处理错误
}
if knowledge.ParseStatus == "completed" {
fmt.Println("Knowledge re-parsing completed!")
break
} else if knowledge.ParseStatus == "failed" {
fmt.Printf("Knowledge re-parsing failed: %s\n", knowledge.ErrorMessage)
break
}
}
示例:取消解析
// 取消正在进行的解析任务(资源紧张 / 上传错误文件时使用)
// - 已经 completed / failed 的知识不能取消
// - 已写入的分块/索引会保留,可后续调用 ReparseKnowledge 重新解析
knowledge, err := apiClient.CancelKnowledgeParse(context.Background(), knowledgeID)
if err != nil {
// 处理错误
}
fmt.Printf("Parse Status: %s\n", knowledge.ParseStatus) // "cancelled"
示例:获取会话消息
// 获取最近消息
messages, err := apiClient.GetRecentMessages(context.Background(), sessionID, 10)
if err != nil {
// 处理错误
}
// 获取指定时间之前的消息
beforeTime := time.Now().Add(-24 * time.Hour)
olderMessages, err := apiClient.GetMessagesBefore(context.Background(), sessionID, beforeTime, 10)
if err != nil {
// 处理错误
}
完整示例
请参考 example.go 文件中的 ExampleUsage 函数,其中展示了客户端的完整使用流程。