diff --git a/appbuilder/core/console/knowledge_base/data_class.py b/appbuilder/core/console/knowledge_base/data_class.py index 438091d37..53d2cdfa0 100644 --- a/appbuilder/core/console/knowledge_base/data_class.py +++ b/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="本次查询包含的最大结果集数量") diff --git a/appbuilder/core/console/knowledge_base/knowledge_base.py b/appbuilder/core/console/knowledge_base/knowledge_base.py index dfe9862d3..e1d9b8522 100644 --- a/appbuilder/core/console/knowledge_base/knowledge_base.py +++ b/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