update

appbuilder/core/console/knowledge_base/data_class.py

@@ -16,6 +16,7 @@
 from pydantic import Field
 from typing import Union
 from typing import Optional
+import datetime
 
 
 class KnowledgeBaseUploadFileResponse(BaseModel):
@@ -119,7 +120,7 @@ class KnowledgeBaseGetDetailRequest(BaseModel):
 class KnowledgeBaseDetailResponse(BaseModel):
     id: str = Field(..., description="知识库ID")
     name: str = Field(..., description="知识库名称")
-    description: str = Field(..., description="知识库描述")
+    description: Optional[str] = Field(None, description="知识库描述")
     config: Optional[KnowledgeBaseConfig] = Field(..., description="知识库配置")
 
 
@@ -205,3 +206,58 @@ class KnowledgeBaseCreateDocumentsRequest(BaseModel):
     processOption: Optional[DocumentProcessOption] = Field(
         None, description="文档处理选项"
     )
+
+
+class CreateChunkRequest(BaseModel):
+    documentId: str = Field(..., description="文档ID")
+    content: str = Field(..., description="文档内容")
+
+
+class CreateChunkResponse(BaseModel):
+    id: str = Field(..., description="切片ID")
+
+
+class ModifyChunkRequest(BaseModel):
+    chunkId: str = Field(..., description="切片ID")
+    content: str = Field(..., description="文档内容")
+    enable: bool = Field(..., description="是否启用")
+
+
+class DeleteChunkRequest(BaseModel):
+    chunkId: str = Field(..., description="切片ID")
+
+
+class DescribeChunkRequest(BaseModel):
+    chunkId: str = Field(..., description="切片ID")
+
+
+class DescribeChunkResponse(BaseModel):
+    id: str = Field(..., description="切片ID")
+    type: int = Field(..., description="切片类型")
+    knowledgeBaseId: str = Field(..., description="知识库ID")
+    documentId: str = Field(..., description="文档ID")
+    content: str = Field(..., description="文档内容")
+    enabled: bool = Field(..., description="是否启用")
+    wordCount: int = Field(..., description="切片内字符数量")
+    tokenCount: int = Field(..., description="切片内token数量")
+    status: str = Field(..., description="切片状态")
+    statusMessage: str = Field(..., description="切片状态信息")
+    createdAt: int = Field(..., description="创建时间")
+    updatedAt: int = Field(..., description="更新时间")
+
+
+class DescribeChunksRequest(BaseModel):
+    documentId: str = Field(..., description="文档ID")
+    marker: str = Field(..., description="起始位置")
+    maxKeys: int = Field(..., description="返回文档数量大小，默认10，最大值100")
+    type: Optional[int] = Field(None, description="切片类型")
+
+
+class DescribeChunksResponse(BaseModel):
+    data: list[DescribeChunkResponse] = Field(..., description="切片列表")
+    marker: str = Field(..., description="起始位置")
+    isTruncated: bool = Field(
+        ..., description="true表示后面还有数据，false表示已经是最后一页"
+    )
+    nextMarker: str = Field(..., description="下一页起始位置")
+    maxKeys: int = Field(..., description="本次查询包含的最大结果集数量")

appbuilder/core/console/knowledge_base/knowledge_base.py

@@ -211,6 +211,8 @@ def create_knowledge_base(
         data = response.json()
 
         resp = data_class.KnowledgeBaseDetailResponse(**data)
+        self.knowledge_id = resp.id
+        self.knowledge_name = resp.name
         return resp
 
     def get_knowledge_base_detail(
@@ -409,3 +411,133 @@ def upload_documents(
             data = response.json()
 
         return data
+
+    def create_chunk(
+        self,
+        documentId: str,
+        content: str,
+    ) -> data_class.CreateChunkResponse:
+        headers = self.http_client.auth_header_v2()
+        headers["content-type"] = "application/json"
+
+        url = self.http_client.service_url_v2("/knowledgeBase?Action=CreateChunk")
+
+        request = data_class.CreateChunkRequest(
+            documentId=documentId,
+            content=content,
+        )
+
+        response = self.http_client.session.post(
+            url=url, headers=headers, json=request.model_dump(exclude_none=True)
+        )
+
+        self.http_client.check_response_header(response)
+        self.http_client.check_console_response(response)
+        data = response.json()
+
+        resp = data_class.CreateChunkResponse(**data)
+        return resp
+
+    def modify_chunk(
+        self,
+        chunkId: str,
+        content: str,
+        enable: bool,
+    ):
+        headers = self.http_client.auth_header_v2()
+        headers["content-type"] = "application/json"
+
+        url = self.http_client.service_url_v2("/knowledgeBase?Action=ModifyChunk")
+
+        request = data_class.ModifyChunkRequest(
+            chunkId=chunkId,
+            content=content,
+            enable=enable,
+        )
+
+        response = self.http_client.session.post(
+            url=url, headers=headers, json=request.model_dump(exclude_none=True)
+        )
+
+        self.http_client.check_response_header(response)
+        self.http_client.check_console_response(response)
+        data = response.json()
+
+        return data
+
+    def delete_chunk(
+        self,
+        chunkId: str,
+    ):
+        headers = self.http_client.auth_header_v2()
+        headers["content-type"] = "application/json"
+
+        url = self.http_client.service_url_v2("/knowledgeBase?Action=DeleteChunk")
+
+        request = data_class.DeleteChunkRequest(
+            chunkId=chunkId,
+        )
+
+        response = self.http_client.session.post(
+            url=url, headers=headers, json=request.model_dump(exclude_none=True)
+        )
+
+        self.http_client.check_response_header(response)
+        self.http_client.check_console_response(response)
+        data = response.json()
+
+        return data
+
+    def describe_chunk(
+        self,
+        chunkId: str,
+    ) -> data_class.DescribeChunkResponse:
+        headers = self.http_client.auth_header_v2()
+        headers["content-type"] = "application/json"
+
+        url = self.http_client.service_url_v2("/knowledgeBase?Action=DescribeChunk")
+
+        request = data_class.DescribeChunkRequest(
+            chunkId=chunkId,
+        )
+
+        response = self.http_client.session.post(
+            url=url, headers=headers, json=request.model_dump(exclude_none=True)
+        )
+
+        self.http_client.check_response_header(response)
+        self.http_client.check_console_response(response)
+        data = response.json()
+
+        resp = data_class.DescribeChunkResponse(**data)
+        return resp
+
+    def describe_chunks(
+        self,
+        documentId: str,
+        marker: str,
+        maxKeys: int,
+        type: int = None,
+    ) -> data_class.DescribeChunkResponse:
+        headers = self.http_client.auth_header_v2()
+        headers["content-type"] = "application/json"
+
+        url = self.http_client.service_url_v2("/knowledgeBase?Action=DescribeChunks")
+
+        request = data_class.DescribeChunksRequest(
+            documentId=documentId,
+            marker=marker,
+            maxKeys=maxKeys,
+            type=type,
+        )
+
+        response = self.http_client.session.post(
+            url=url, headers=headers, json=request.model_dump(exclude_none=True)
+        )
+
+        self.http_client.check_response_header(response)
+        self.http_client.check_console_response(response)
+        data = response.json()
+
+        resp = data_class.DescribeChunksResponse(**data)
+        return resp