import aiohttp import aiofiles import requests from rag.db import MilvusOperate, MysqlOperate from rag.task_registry import task_registry, TaskCancelledException from rag.document_load.pdf_load import MinerUParsePdf from rag.document_load.office_load import MinerUParseOffice from rag.document_load.txt_load import TextLoad from rag.document_load.image_load import MinerUParseImage from utils.upload_file_to_oss import UploadMinio from utils.get_logger import setup_logger from config import minio_config import os import time from uuid import uuid1 from langchain_text_splitters import RecursiveCharacterTextSplitter from rag.document_load.document_format_conversion import * import re from typing import List pdf_parse = MinerUParsePdf() office_parse = MinerUParseOffice() text_parse = TextLoad() image_parse = MinerUParseImage() logger = setup_logger(__name__) class ProcessDocuments(): def __init__(self, file_json): self.file_json = file_json self.knowledge_id = self.file_json.get("knowledge_id") self.mysql_client = MysqlOperate() self.minio_client = UploadMinio() self.milvus_client = MilvusOperate(collection_name=self.knowledge_id) # def _get_file_type(self, name): # if name.endswith(".txt"): # return text_parse # elif name.endswith('.pdf'): # return pdf_parse # elif name.endswith((".doc", ".docx", "ppt", "pptx")): # return office_parse # elif name.endswith((".jpg", "png", "jpeg")): # return image_parse # else: # raise "不支持的文件格式" # 只接收txt和pdf格式 def _get_file_type(self, name): if name.endswith(".txt"): return text_parse elif name.endswith('.pdf'): return pdf_parse # async def save_file_temp(self, session, url, name): # down_file_path = "./tmp_file" + f"/{self.knowledge_id}" # # down_file_path = "./tmp_file" # os.makedirs(down_file_path, exist_ok=True) # down_file_name = down_file_path + f"/{name}" # # if os.path.exists(down_file_name): # # pass # # else: # async with session.get(url, ssl=False) as resp: # resp.raise_for_status() # content_length = resp.headers.get('Content-Length') # if content_length: # file_size = int(content_length) # else: # file_size = 0 # async with aiofiles.open(down_file_name, 'wb') as f: # async for chunk in resp.content.iter_chunked(1024): # await f.write(chunk) async def save_file_temp(self, session, url, name, max_retries=3): down_file_path = "./tmp_file" + f"/{self.knowledge_id}" os.makedirs(down_file_path, exist_ok=True) down_file_name = down_file_path + f"/{name}" attempt = 0 while attempt < max_retries: try: async with session.get(url, ssl=False) as resp: resp.raise_for_status() content_length = resp.headers.get('Content-Length') file_size = int(content_length) if content_length else 0 async with aiofiles.open(down_file_name, 'wb') as f: async for chunk in resp.content.iter_chunked(1024): await f.write(chunk) # 成功就 return return down_file_name, file_size except Exception as e: attempt += 1 # 最后一次失败直接抛出 if attempt >= max_retries: raise # 指数退避：0.5s ~ 3s 内，随机涨 wait = min(3, 0.5 * (2 ** attempt) + random.random()) await asyncio.sleep(wait) return down_file_name, file_size def send_post_request_sync(self, url, json_data=None, headers=None, timeout=30): """ 同步发送POST请求 参数: url: 请求的URL地址 json_data: JSON格式的请求体数据（字典类型） headers: 可选的请求头字典 timeout: 请求超时时间（秒），默认30秒 返回: dict: 包含状态码、响应数据等信息 """ try: # 设置默认的Content-Type为application/json if headers is None: headers = {} if 'Content-Type' not in headers: headers['Content-Type'] = 'application/json' resp = requests.post(url, json=json_data, headers=headers, timeout=timeout, verify=False) status_code = resp.status_code response_data = resp.text logger.info(f"同步POST请求成功 [url={url}]: {response_data} json_data={json_data}") return { "code": 200, "status_code": status_code, "data": response_data, "message": "POST请求成功" } except requests.exceptions.Timeout as e: logger.error(f"同步POST请求超时 [url={url}]: {e}") return { "code": 500, "message": f"POST请求超时: {str(e)}" } except requests.exceptions.RequestException as e: logger.error(f"同步POST请求失败 [url={url}]: {e}") return { "code": 500, "message": f"POST请求失败: {str(e)}" } def file_split_by_len(self, file_text): split_map = { "0": ["#"], # 按标题段落切片 "1": [""], # 按页切片 "2": ["\n"] # 按问答对 } separator_num = self.file_json.get("set_slice") slice_value = self.file_json.get("slice_value", "").replace("\\n", "\n") separator = split_map.get(separator_num) if split_map.get(separator_num) else [slice_value] logger.info(f"文本切分字符：{separator}") text_split = RecursiveCharacterTextSplitter( separators=separator, chunk_size=500, chunk_overlap=40, length_function=len ) texts = text_split.split_text(file_text) return texts def split_text(self, file_text): text_split = RecursiveCharacterTextSplitter( separators=["\n\n", "\n"], chunk_size=500, chunk_overlap=40, length_function=len ) texts = text_split.split_text(file_text) return texts def chunk_text_for_rag(self, text: str, max_chars: int = 20000) -> List[str]: """ 分割长文本 参数: text: 输入的长文本字符串 max_chars: 每块的最大字符数，默认20000 返回: List[str]: 分割后的文本块列表 分割规则: 1. 在标签x.x（如1.1, 2.3等）处分割 2. 下一块包含上一块最后一个x.x标签的完整内容（重叠） 3. 如果没有x.x标签或加上重叠内容超过max_chars，则按句号"。"分割 """ if len(text) <= max_chars: return [text] chunks = [] start_idx = 0 while start_idx < len(text): # 确定当前块的结束位置 end_idx = min(start_idx + max_chars, len(text)) if end_idx >= len(text): # 最后一块，直接添加 chunks.append(text[start_idx:]) break # 在当前块范围内查找x.x标签 current_chunk = text[start_idx:end_idx] split_pos, overlap_content = self.find_split_point(current_chunk, text, start_idx) if split_pos is not None: # 找到了合适的x.x标签分割点 actual_end = start_idx + split_pos # 检查加上重叠内容是否超过限制 if overlap_content and len(current_chunk[:split_pos]) + len(overlap_content) <= max_chars: chunks.append(text[start_idx:actual_end]) # 下一块从重叠内容的起始位置开始 overlap_start = text.rfind(overlap_content, start_idx, actual_end) if overlap_start != -1: start_idx = overlap_start else: start_idx = actual_end else: # 重叠内容过长，使用句号分割 split_pos, overlap_content = self.find_split_by_period(current_chunk, text, start_idx) if split_pos is not None: actual_end = start_idx + split_pos chunks.append(text[start_idx:actual_end]) if overlap_content: overlap_start = text.rfind(overlap_content, start_idx, actual_end) start_idx = overlap_start if overlap_start != -1 else actual_end else: start_idx = actual_end else: # 连句号都找不到，强制分割 chunks.append(text[start_idx:end_idx]) start_idx = end_idx else: # 没有找到x.x标签，使用句号分割 split_pos, overlap_content = self.find_split_by_period(current_chunk, text, start_idx) if split_pos is not None: actual_end = start_idx + split_pos chunks.append(text[start_idx:actual_end]) if overlap_content: overlap_start = text.rfind(overlap_content, start_idx, actual_end) start_idx = overlap_start if overlap_start != -1 else actual_end else: start_idx = actual_end else: # 连句号都找不到，强制分割 chunks.append(text[start_idx:end_idx]) start_idx = end_idx return chunks def find_split_point(self, chunk: str, full_text: str, chunk_start: int): """ 在块中查找最近的x.x标签作为分割点 返回: (split_position, overlap_content): 分割位置和重叠内容 """ # 匹配x.x格式的标签（如1.1, 2.3, 10.5等） pattern = r'\d+\.\d+' # pattern = r'\b\d+\.\d+\b' matches = list(re.finditer(pattern, chunk)) if not matches: return None, None # 找最后一个匹配的标签 last_match = matches[-1] tag_start = last_match.start() # 提取该标签的完整内容（从标签到下一个标签或块尾） tag_content_start = chunk_start + tag_start # 在全文中查找下一个x.x标签 remaining_text = full_text[tag_content_start:] next_tag_match = re.search(pattern, remaining_text[len(last_match.group()):]) if next_tag_match: tag_content_end = tag_content_start + len(last_match.group()) + next_tag_match.start() else: tag_content_end = len(full_text) overlap_content = full_text[tag_content_start:tag_content_end] # 检查重叠内容长度 if len(overlap_content) > 20000: # 重叠内容本身超过限制，返回None使用句号分割 return None, None return tag_start, overlap_content def find_split_by_period(self, chunk: str, full_text: str, chunk_start: int): """ 按句号"。"查找分割点 返回: (split_position, overlap_content): 分割位置和重叠内容 """ # 从后向前查找句号 last_period = chunk.rfind('。') if last_period == -1: return None, None # 分割点是句号之后 split_pos = last_period + 1 # 查找这个句子的开始位置作为重叠内容 # 向前查找上一个句号 prev_period = chunk.rfind('。', 0, last_period) if prev_period != -1: overlap_start = prev_period + 1 overlap_content = chunk[overlap_start:split_pos] else: # 如果前面没有句号，则从块开始到当前句号 overlap_content = chunk[:split_pos] # 检查重叠内容长度 if len(overlap_content) > 20000: overlap_content = "" return split_pos, overlap_content def split_by_title(self, file_content_list, set_table, doc_id): # TODO 处理根据标题切分逻辑 图片替换标识符，表格按照set table 0图片，1html数据 text_lists = [] text = "" image_num = 1 flag_img_info = {} level_1_text = "" level_2_text = "" for i, content_dict in enumerate(file_content_list): text_type = content_dict.get("type") content_text = content_dict.get("text") if text_type == "text": text_level = content_dict.get("text_level", "") if text_level == 1: if not level_1_text: level_1_text = f"# {content_text}\n" text += f"# {content_text}\n" else: if len(text) >20000: text_chunks = self.chunk_text_for_rag(text) text_lists.extend(text_chunks) else: text_lists.append(text) text = f"# {content_text}\n" level_1_text = f"# {content_text}\n" level_2_text = "" elif text_level == 2: if not level_2_text: text += f"## {content_text}\n" level_2_text = f"## {content_text}\n" else: if len(text) >20000: text_chunks = self.chunk_text_for_rag(text) text_lists.extend(text_chunks) else: text_lists.append(text) text = level_1_text + f"## {content_text}\n" else: if text_level: text += text_level*"#" + " " + content_text + "\n" else: text += content_text elif text_type == "table" and set_table == "1": text += " ".join(content_dict.get("table_caption")) + '\n' text += content_dict.get("table_body") + '\n' elif text_type in ("image", "table"): image_path = content_dict.get("img_path") if not image_path: continue image_name = image_path.split("/")[1] # save_image_path = "./tmp_file/images/" + f"/{image_name}" save_image_path = "./tmp_file/document/vlm/images" + f"/{image_name}" replace_text = f"【示意图序号_{doc_id}_{image_num}】" minio_file_path = f"/pdf/{self.knowledge_id}/{doc_id}/{replace_text}.jpg" self.minio_client.upload_file(save_image_path, minio_file_path) minio_url = minio_config.get("minio_url") minio_bucket = minio_config.get("minio_bucket") flag_img_info[replace_text] = f"{minio_url}/{minio_bucket}/{minio_file_path}" if text_type == "table": text += " ".join(content_dict.get("table_caption")) + '\n' text += replace_text image_num += 1 if i+1 == len(file_content_list): text_lists.append(text) elif text_type == "list": list_items = content_dict.get("list_items") if list_items: text += "\n".join(list_items) + "\n" return text_lists, flag_img_info def split_by_page(self, file_content_list, set_table, doc_id): # TODO 处理按照页面切分，图片处理成标识符，表格按照set table 0图片，1html数据 text_lists = [] current_page = "" text = "" image_num = 1 flag_img_info = {} for i,content_dict in enumerate(file_content_list): page_index = content_dict.get("page_idx") if i == 0: current_page = page_index elif page_index != current_page: text_lists.append(text) text = "" current_page = page_index text_type = content_dict.get("type") if text_type == "text": content_text = content_dict.get("text") text_level = content_dict.get("text_level") if text_level: text += "#" * text_level + " " + content_text else: text += content_text elif text_type == "table" and set_table == "1": text += content_dict.get("table_body") elif text_type in ("image", "table"): image_path = content_dict.get("img_path") image_name = image_path.split("/")[1] # save_image_path = "./tmp_file/images/" + f"/{image_name}" save_image_path = "./tmp_file/document/vlm/images" + f"/{image_name}" replace_text = f"【示意图序号_{doc_id}_{image_num}】" minio_file_path = f"/pdf/{self.knowledge_id}/{doc_id}/{replace_text}.jpg" self.minio_client.upload_file(save_image_path, minio_file_path) minio_url = minio_config.get("minio_url") minio_bucket = minio_config.get("minio_bucket") flag_img_info[replace_text] = f"{minio_url}/{minio_bucket}/{minio_file_path}" text += replace_text image_num += 1 if i+1 == len(file_content_list): text_lists.append(text) return text_lists, flag_img_info def split_by_self(self, file_content_list, set_table, slice_value, doc_id): # TODO 按照自定义的符号切分，图片处理成标识符，表格按照set table 0图片，1html数据，长度控制500以内，超过500切断 logger.info(f"自定义的分隔符：{slice_value}") text = "" image_num = 1 flag_img_info = {} for i, content_dict in enumerate(file_content_list): text_type = content_dict.get("type") if text_type == "text": content_text = content_dict.get("text") text_level = content_dict.get("text_level") if text_level: text += "#" * text_level + " " + content_text else: text += content_text elif text_type == "table" and set_table == "1": text += content_dict.get("table_body") elif text_type in ("image", "table"): image_path = content_dict.get("img_path") image_name = image_path.split("/")[1] # save_image_path = "./tmp_file/images/" + f"/{image_name}" save_image_path = "./tmp_file/document/vlm/images" + f"/{image_name}" replace_text = f"【示意图序号_{doc_id}_{image_num}】" minio_file_path = f"/pdf/{self.knowledge_id}/{doc_id}/{replace_text}.jpg" self.minio_client.upload_file(save_image_path, minio_file_path) minio_url = minio_config.get("minio_url") minio_bucket = minio_config.get("minio_bucket") flag_img_info[replace_text] = f"{minio_url}/{minio_bucket}/{minio_file_path}" text += replace_text image_num += 1 split_lists = text.split(slice_value) text_lists = [] for split_text in split_lists: r = len(split_text)//500 if r >= 1: for i in range(r+1): t = split_text[i*500:(i+1)*500] if t: text_lists.append(t) else: text_lists.append(split_text) return text_lists, flag_img_info def file_split(self, file_content_list, doc_id): # TODO 根据文本列表进行切分 返回切分列表和存储图片的链接 separator_num = self.file_json.get("set_slice") set_table = self.file_json.get("set_table") # separator = split_map.get(separator_num) if split_map.get(separator_num) else [slice_value] # logger.info(f"文本切分字符：{separator}") if isinstance(file_content_list, str): file_text = file_content_list text_lists = self.split_text(file_text) return text_lists, {} elif separator_num == "0": # 使用标题段落切分，使用text_level=1，2 切分即一个# 还是两个# text_lists, flag_img_info = self.split_by_title(file_content_list, set_table, doc_id) return text_lists, flag_img_info elif separator_num == "1": # 按照页面方式切分 text_lists, flag_img_info = self.split_by_page(file_content_list, set_table, doc_id) return text_lists, flag_img_info elif separator_num == "2": # 按照问答对切分 针对exce文档，暂不实现 return [], {} else: # 自定义切分的方式，按照自定义字符以及文本长度切分，超过500 slice_value = self.file_json.get("slice_value", "").replace("\\n", "\n") text_lists, flag_img_info = self.split_by_self(file_content_list, set_table, slice_value, doc_id) return text_lists, flag_img_info def process_data_to_milvus_schema(self, text_lists, doc_id, name): """组织数据格式： { "content": text, "doc_id": doc_id, "chunk_id": chunk_id, "metadata": {"source": file_name}, } """ docs = [] total_len = 0 for i, text in enumerate(text_lists): chunk_id = str(uuid1()) chunk_len = len(text) total_len += chunk_len d = { "content": text, "doc_id": doc_id, "chunk_id": chunk_id, "metadata": {"source": name, "chunk_index": i+1, "chunk_len": chunk_len} } docs.append(d) return docs, total_len async def process_documents(self, file_json): user_id = file_json.get("userId", "") task_id = file_json.get("docs", "")[0].get("document_id") # ===== 注册任务到全局任务表 ===== task_ctx = None if task_id: task_ctx = task_registry.register(task_id, user_id, self.knowledge_id) # 启动进度（独立线程） reporter = None if task_id: from rag.progress_reporter import ProgressReporter from config import progress_callback_config estimate_seconds = progress_callback_config.get("estimate_seconds", 120) callback_url = progress_callback_config.get("default_url") reporter = ProgressReporter(task_id, callback_url, estimate_seconds, user_id) reporter.start() # 关联 reporter 到任务上下文（用于取消时停止进度上报） if task_ctx: task_ctx.reporter = reporter # 初始化成功文档列表（在 try 外面定义，确保异常处理时可访问） success_doc = [] try: # 文档下载 separator_num = file_json.get("set_slice") if separator_num == "2": if reporter: reporter.complete(success=False) return {"code": 500, "message": "暂不支持解析"} docs = file_json.get("docs") flag = file_json.get("flag") for doc in docs: # ===== 检查点1：每个文档处理前检查取消标志 ===== if task_ctx and task_ctx.is_cancelled: logger.info(f"任务 {task_id} 在文档处理前被取消") raise TaskCancelledException("任务已被用户取消") url = doc.get("url") name = doc.get("name") doc_id = doc.get("document_id") async with aiohttp.ClientSession() as session: down_file_name, file_size = await self.save_file_temp(session, url, name) # 只接收txt和pdf格式，其它格式统一转换成pdf，不支持格式在convert_to_pdf中判断 if not os.path.exists(down_file_name): down_file_name, file_size = await self.save_file_temp(session, url, name) logger.info(f'{doc_id}文件下载失败，重试中~') logger.info(f'下载文件路径：{down_file_name}') if not name.endswith(".txt") and not name.endswith('.pdf'): down_file_name = convert_to_pdf(down_file_name) name = os.path.basename(down_file_name) file_parse = self._get_file_type(name) file_content_list = await file_parse.extract_text(down_file_name) logger.info(f"mineru解析的pdf数据：{file_content_list}") text_lists, flag_img_info = self.file_split(file_content_list, doc_id) docs, total_char_len = self.process_data_to_milvus_schema(text_lists, doc_id, name) logger.info(f"存储到milvus的文本数据：{docs}") # ===== 检查点2：解析完成、入库前检查取消标志 ===== if task_ctx and task_ctx.is_cancelled: logger.info(f"任务 {task_id} 在入库前被取消，无需清理数据库") raise TaskCancelledException("任务已被用户取消") if flag == "upload": # 插入到milvus库中 insert_slice_flag, insert_mysql_info = self.mysql_client.insert_to_slice(docs, self.knowledge_id, doc_id) if insert_slice_flag: # 插入到mysql的slice info数据库中 insert_img_flag, insert_mysql_info = self.mysql_client.insert_to_image_url(flag_img_info, self.knowledge_id, doc_id) else: insert_img_flag = False parse_file_status = False if insert_img_flag: insert_milvus_flag, insert_milvus_str = self.milvus_client._insert_data(docs) # 插入mysql中的bm_media_replacement表中 else: # self.milvus_client._delete_by_doc_id(doc_id=doc_id) insert_milvus_flag = False # return resp parse_file_status = False if insert_milvus_flag: parse_file_status = True # ===== 检查点3：入库后检查取消标志，若取消则清理已插入数据 ===== if task_ctx and task_ctx.is_cancelled: logger.info(f"任务 {task_id} 在入库后被取消，清理已插入数据: doc_id={doc_id}") self.milvus_client._delete_by_doc_id(doc_id=doc_id) self.mysql_client.delete_to_slice(doc_id=doc_id) self.mysql_client.delete_image_url(doc_id=doc_id) raise TaskCancelledException("任务已被用户取消") else: self.mysql_client.delete_to_slice(doc_id=doc_id) # self.milvus_client._delete_by_doc_id(doc_id=doc_id) self.mysql_client.delete_image_url(doc_id=doc_id) # resp = {"code": 500, "message": insert_mysql_info} parse_file_status = False # return resp elif flag == "update": # 更新切片方式 # 先把库中的数据删除 self.milvus_client._delete_by_doc_id(doc_id=doc_id) self.mysql_client.delete_to_slice(doc_id=doc_id) insert_milvus_start_time = time.time() insert_slice_flag, insert_mysql_info = self.mysql_client.insert_to_slice(docs, self.knowledge_id, doc_id) # insert_milvus_flag, insert_milvus_str = self.milvus_client._batch_insert_data(docs,text_lists) insert_milvus_end_time = time.time() logger.info(f"插入milvus数据库耗时：{insert_milvus_end_time - insert_milvus_start_time}") if insert_slice_flag: # 插入到mysql的slice info数据库中 insert_mysql_start_time = time.time() insert_milvus_flag, insert_milvus_str = self.milvus_client._insert_data(docs) insert_mysql_end_time = time.time() logger.info(f"插入mysql数据库耗时:{insert_mysql_end_time - insert_mysql_start_time}") else: # resp = {"code": 500, "message": insert_milvus_str} # return resp insert_milvus_flag = False parse_file_status = False if insert_milvus_flag: # resp = {"code": 200, "message": "切片修改成功"} parse_file_status = True # ===== 检查点3：入库后检查取消标志，若取消则清理已插入数据 ===== if task_ctx and task_ctx.is_cancelled: logger.info(f"任务 {task_id} 在入库后被取消，清理已插入数据: doc_id={doc_id}") self.milvus_client._delete_by_doc_id(doc_id=doc_id) self.mysql_client.delete_to_slice(doc_id=doc_id) raise TaskCancelledException("任务已被用户取消") else: self.mysql_client.delete_to_slice(doc_id=doc_id) # self.milvus_client._delete_by_doc_id(doc_id=doc_id) # resp = {"code":500, "message": insert_mysql_info} parse_file_status = False # return resp if parse_file_status: success_doc.append(doc_id) else: if flag == "upload": for del_id in success_doc: self.milvus_client._delete_by_doc_id(doc_id=del_id) self.mysql_client.delete_image_url(doc_id=del_id) self.mysql_client.delete_to_slice(doc_id=del_id) if reporter: reporter.complete(success=False) self.send_post_request_sync("http://10.10.10.2:8091/deepseek/api/updateDocumentByPython",{"status":"2","documentId": file_json.get("docs")[0].get("document_id")}) return {"code": 500, "message": "解析失败", "knowledge_id" : self.knowledge_id, "doc_info": {}} self.send_post_request_sync("http://10.10.10.2:8091/deepseek/api/updateDocumentByPython", {"knowledgeId": self.knowledge_id,"documentId": file_json.get("docs")[0].get("document_id"), "length": file_size, "wordNum": total_char_len, "sliceTotal": len(text_lists), "status":"1"}) # 任务完成：发送100%进度 if reporter: reporter.complete(success=True) return {"code": 200, "message": "解析成功", "knowledge_id" : self.knowledge_id, "doc_info": {"file_size": file_size, "total_char_len": total_char_len, "slice_num": len(text_lists)}} except TaskCancelledException as e: # ===== 处理任务取消异常 ===== logger.info(f"任务被取消 [task_id={task_id}]: {e}") self.send_post_request_sync("http://10.10.10.2:8091/deepseek/api/updateDocumentByPython",{"status":"2","documentId": file_json.get("docs")[0].get("document_id")}) # if task_id: # del_success, del_info = self.mysql_client.delete_document(task_id) # if del_success: # logger.info(f"已删除 bm_document 记录: task_id={task_id}") # else: # logger.warning(f"删除 bm_document 记录失败: {del_info}") # 清理之前已成功处理的文档数据（slice、image、milvus） for del_id in success_doc: logger.info(f"清理已入库的文档数据: doc_id={del_id}") self.milvus_client._delete_by_doc_id(doc_id=del_id) self.mysql_client.delete_image_url(doc_id=del_id) self.mysql_client.delete_to_slice(doc_id=del_id) if reporter: reporter.complete(success=False) return {"code": 499, "message": "任务已取消", "knowledge_id": self.knowledge_id, "doc_info": {}} except Exception as e: # 捕获所有异常，统一处理 logger.error(f"文件处理异常 [task_id={task_id}]: {e}", exc_info=True) self.send_post_request_sync("http://10.10.10.2:8091/deepseek/api/updateDocumentByPython",{"status":"2","documentId": file_json.get("docs")[0].get("document_id")}) if reporter: reporter.complete(success=False) return {"code": 500, "message": "error", "knowledge_id": self.knowledge_id, "doc_info": {}} finally: # ===== 注销任务 ===== if task_id: task_registry.unregister(task_id)