code: import os
  
  from transformers import AutoTokenizer
  
  from configs import (
      EMBEDDING_MODEL,
      KB_ROOT_PATH,
      CHUNK_SIZE,
      OVERLAP_SIZE,
      ZH_TITLE_ENHANCE,
      logger,
      log_verbose,
      text_splitter_dict,
      LLM_MODEL,
      TEXT_SPLITTER_NAME,
  )
  import importlib
  from text_splitter import zh_title_enhance as func_zh_title_enhance
  import langchain.document_loaders
  from langchain.docstore.document import Document
  from langchain.text_splitter import TextSplitter
  from pathlib import Path
  import json
  from concurrent.futures import ThreadPoolExecutor
  from server.utils import run_in_thread_pool, embedding_device, get_model_worker_config
  import io
  from typing import List, Union, Callable, Dict, Optional, Tuple, Generator
  import chardet
  
  
  def validate_kb_name(knowledge_base_id: str) -> bool:
      # 检查是否包含预期外的字符或路径攻击关键字
      if "../" in knowledge_base_id:
          return False
      return True
  
  
  def get_kb_path(knowledge_base_name: str):
      return os.path.join(KB_ROOT_PATH, knowledge_base_name)
  
  
  def get_doc_path(knowledge_base_name: str):
      return os.path.join(get_kb_path(knowledge_base_name), "content")
  
  
  def get_vs_path(knowledge_base_name: str, vector_name: str):
      return os.path.join(get_kb_path(knowledge_base_name), vector_name)
  
  
  def get_file_path(knowledge_base_name: str, doc_name: str):
      return os.path.join(get_doc_path(knowledge_base_name), doc_name)
  
  
  def list_kbs_from_folder():
      return [f for f in os.listdir(KB_ROOT_PATH)
              if os.path.isdir(os.path.join(KB_ROOT_PATH, f))]
  
  
  def list_files_from_folder(kb_name: str):
      doc_path = get_doc_path(kb_name)
      return [file for file in os.listdir(doc_path)
              if os.path.isfile(os.path.join(doc_path, file))]
  
  
  def load_embeddings(model: str = EMBEDDING_MODEL, device: str = embedding_device()):
      '''
      从缓存中加载embeddings，可以避免多线程时竞争加载。
      '''
      from server.knowledge_base.kb_cache.base import embeddings_pool
      return embeddings_pool.load_embeddings(model=model, device=device)
  
  
  LOADER_DICT = {"UnstructuredHTMLLoader": ['.html'],
                 "UnstructuredMarkdownLoader": ['.md'],
                 "CustomJSONLoader": [".json"],
                 "CSVLoader": [".csv"],
                 "RapidOCRPDFLoader": [".pdf"],
                 "RapidOCRLoader": ['.png', '.jpg', '.jpeg', '.bmp'],
                 "UnstructuredFileLoader": ['.eml', '.msg', '.rst',
                                            '.rtf', '.txt', '.xml',
                                            '.docx', '.epub', '.odt',
                                            '.ppt', '.pptx', '.tsv'],
                 }
  SUPPORTED_EXTS = [ext for sublist in LOADER_DICT.values() for ext in sublist]
  
  
  class CustomJSONLoader(langchain.document_loaders.JSONLoader):
      '''
      langchain的JSONLoader需要jq，在win上使用不便，进行替代。针对langchain==0.0.286
      '''
  
      def __init__(
          self,
          file_path: Union[str, Path],
          content_key: Optional[str] = None,
          metadata_func: Optional[Callable[[Dict, Dict], Dict]] = None,
          text_content: bool = True,
          json_lines: bool = False,
      ):
          """Initialize the JSONLoader.
  
          Args:
              file_path (Union[str, Path]): The path to the JSON or JSON Lines file.
              content_key (str): The key to use to extract the content from the JSON if
                  results to a list of objects (dict).
              metadata_func (Callable[Dict, Dict]): A function that takes in the JSON
                  object extracted by the jq_schema and the default metadata and returns
                  a dict of the updated metadata.
              text_content (bool): Boolean flag to indicate whether the content is in
                  string format, default to True.
              json_lines (bool): Boolean flag to indicate whether the input is in
                  JSON Lines format.
          """
          self.file_path = Path(file_path).resolve()
          self._content_key = content_key
          self._metadata_func = metadata_func
          self._text_content = text_content
          self._json_lines = json_lines
  
      def _parse(self, content: str, docs: List[Document]) -> None:
          """Convert given content to documents."""
          data = json.loads(content)
  
          # Perform some validation
          # This is not a perfect validation, but it should catch most cases
          # and prevent the user from getting a cryptic error later on.
          if self._content_key is not None:
              self._validate_content_key(data)
          if self._metadata_func is not None:
              self._validate_metadata_func(data)
  
          for i, sample in enumerate(data, len(docs) + 1):
              text = self._get_text(sample=sample)
              metadata = self._get_metadata(
                  sample=sample, source=str(self.file_path), seq_num=i
              )
              docs.append(Document(page_content=text, metadata=metadata))
  
  
  langchain.document_loaders.CustomJSONLoader = CustomJSONLoader
  
  
  def get_LoaderClass(file_extension):
      for LoaderClass, extensions in LOADER_DICT.items():
          if file_extension in extensions:
              return LoaderClass
  
  
  # 把一些向量化共用逻辑从KnowledgeFile抽取出来，等langchain支持内存文件的时候，可以将非磁盘文件向量化
  def get_loader(loader_name: str, file_path_or_content: Union[str, bytes, io.StringIO, io.BytesIO]):
      '''
      根据loader_name和文件路径或内容返回文档加载器。
      '''
      try:
          if loader_name in ["RapidOCRPDFLoader", "RapidOCRLoader"]:
              document_loaders_module = importlib.import_module('document_loaders')
          else:
              document_loaders_module = importlib.import_module('langchain.document_loaders')
          DocumentLoader = getattr(document_loaders_module, loader_name)
      except Exception as e:
          msg = f"为文件{file_path_or_content}查找加载器{loader_name}时出错：{e}"
          logger.error(f'{e.__class__.__name__}: {msg}',
                       exc_info=e if log_verbose else None)
          document_loaders_module = importlib.import_module('langchain.document_loaders')
          DocumentLoader = getattr(document_loaders_module, "UnstructuredFileLoader")
  
      if loader_name == "UnstructuredFileLoader":
          loader = DocumentLoader(file_path_or_content, autodetect_encoding=True)
      elif loader_name == "CSVLoader":
          # 自动识别文件编码类型，避免langchain loader 加载文件报编码错误
          with open(file_path_or_content, 'rb') as struct_file:
              encode_detect = chardet.detect(struct_file.read())
          if encode_detect:
              loader = DocumentLoader(file_path_or_content, encoding=encode_detect["encoding"])
          else:
              loader = DocumentLoader(file_path_or_content, encoding="utf-8")
  
      elif loader_name == "JSONLoader":
          loader = DocumentLoader(file_path_or_content, jq_schema=".", text_content=False)
      elif loader_name == "CustomJSONLoader":
          loader = DocumentLoader(file_path_or_content, text_content=False)
      elif loader_name == "UnstructuredMarkdownLoader":
          loader = DocumentLoader(file_path_or_content, mode="elements")
      elif loader_name == "UnstructuredHTMLLoader":
          loader = DocumentLoader(file_path_or_content, mode="elements")
      else:
          loader = DocumentLoader(file_path_or_content)
      return loader
  
  
  def make_text_splitter(
      splitter_name: str = TEXT_SPLITTER_NAME,
      chunk_size: int = CHUNK_SIZE,
      chunk_overlap: int = OVERLAP_SIZE,
      llm_model: str = LLM_MODEL,
  ):
      """
      根据参数获取特定的分词器
      """
      splitter_name = splitter_name or "SpacyTextSplitter"
      try:
          if splitter_name == "MarkdownHeaderTextSplitter":  # MarkdownHeaderTextSplitter特殊判定
              headers_to_split_on = text_splitter_dict[splitter_name]['headers_to_split_on']
              text_splitter = langchain.text_splitter.MarkdownHeaderTextSplitter(
                  headers_to_split_on=headers_to_split_on)
          else:
  
              try:  ## 优先使用用户自定义的text_splitter
                  text_splitter_module = importlib.import_module('text_splitter')
                  TextSplitter = getattr(text_splitter_module, splitter_name)
              except:  ## 否则使用langchain的text_splitter
                  text_splitter_module = importlib.import_module('langchain.text_splitter')
                  TextSplitter = getattr(text_splitter_module, splitter_name)
  
              if text_splitter_dict[splitter_name]["source"] == "tiktoken":  ## 从tiktoken加载
                  try:
                      text_splitter = TextSplitter.from_tiktoken_encoder(
                          encoding_name=text_splitter_dict[splitter_name]["tokenizer_name_or_path"],
                          pipeline="zh_core_web_sm",
                          chunk_size=chunk_size,
                          chunk_overlap=chunk_overlap
                      )
                  except:
                      text_splitter = TextSplitter.from_tiktoken_encoder(
                          encoding_name=text_splitter_dict[splitter_name]["tokenizer_name_or_path"],
                          chunk_size=chunk_size,
                          chunk_overlap=chunk_overlap
                      )
              elif text_splitter_dict[splitter_name]["source"] == "huggingface":  ## 从huggingface加载
                  if text_splitter_dict[splitter_name]["tokenizer_name_or_path"] == "":
                      config = get_model_worker_config(llm_model)
                      text_splitter_dict[splitter_name]["tokenizer_name_or_path"] = \
                          config.get("model_path")
  
                  if text_splitter_dict[splitter_name]["tokenizer_name_or_path"] == "gpt2":
                      from transformers import GPT2TokenizerFast
                      from langchain.text_splitter import CharacterTextSplitter
                      tokenizer = GPT2TokenizerFast.from_pretrained("gpt2")
                  else:  ## 字符长度加载
                      tokenizer = AutoTokenizer.from_pretrained(
                          text_splitter_dict[splitter_name]["tokenizer_name_or_path"],
                          trust_remote_code=True)
                  text_splitter = TextSplitter.from_huggingface_tokenizer(
                      tokenizer=tokenizer,
                      chunk_size=chunk_size,
                      chunk_overlap=chunk_overlap
                  )
              else:
                  try:
                      text_splitter = TextSplitter(
                          pipeline="zh_core_web_sm",
                          chunk_size=chunk_size,
                          chunk_overlap=chunk_overlap
                      )
                  except:
                      text_splitter = TextSplitter(
                          chunk_size=chunk_size,
                          chunk_overlap=chunk_overlap
                      )
      except Exception as e:
          print(e)
          text_splitter_module = importlib.import_module('langchain.text_splitter')
          TextSplitter = getattr(text_splitter_module, "RecursiveCharacterTextSplitter")
          text_splitter = TextSplitter(chunk_size=250, chunk_overlap=50)
      return text_splitter
  
  class KnowledgeFile:
      def __init__(
              self,
              filename: str,
              knowledge_base_name: str
      ):
          '''
          对应知识库目录中的文件，必须是磁盘上存在的才能进行向量化等操作。
          '''
          self.kb_name = knowledge_base_name
          self.filename = filename
          self.ext = os.path.splitext(filename)[-1].lower()
          if self.ext not in SUPPORTED_EXTS:
              raise ValueError(f"暂未支持的文件格式 {self.ext}")
          self.filepath = get_file_path(knowledge_base_name, filename)
          self.docs = None
          self.splited_docs = None
          self.document_loader_name = get_LoaderClass(self.ext)
          self.text_splitter_name = TEXT_SPLITTER_NAME
  
      def file2docs(self, refresh: bool=False):
          if self.docs is None or refresh:
              logger.info(f"{self.document_loader_name} used for {self.filepath}")
              loader = get_loader(self.document_loader_name, self.filepath)
              self.docs = loader.load()
          return self.docs
  
      def docs2texts(
          self,
          docs: List[Document] = None,
          zh_title_enhance: bool = ZH_TITLE_ENHANCE,
          refresh: bool = False,
          chunk_size: int = CHUNK_SIZE,
          chunk_overlap: int = OVERLAP_SIZE,
          text_splitter: TextSplitter = None,
      ):
          docs = docs or self.file2docs(refresh=refresh)
          if not docs:
              return []
          if self.ext not in [".csv"]:
              if text_splitter is None:
                  text_splitter = make_text_splitter(splitter_name=self.text_splitter_name, chunk_size=chunk_size, chunk_overlap=chunk_overlap)
              if self.text_splitter_name == "MarkdownHeaderTextSplitter":
                  docs = text_splitter.split_text(docs[0].page_content)
                  for doc in docs:
                      # 如果文档有元数据
                      if doc.metadata:
                          doc.metadata["source"] = os.path.basename(self.filepath)
              else:
                  docs = text_splitter.split_documents(docs)
  
          print(f"文档切分示例：{docs[0]}")
          if zh_title_enhance:
              docs = func_zh_title_enhance(docs)
          self.splited_docs = docs
          return self.splited_docs
  
      def file2text(
          self,
          zh_title_enhance: bool = ZH_TITLE_ENHANCE,
          refresh: bool = False,
          chunk_size: int = CHUNK_SIZE,
          chunk_overlap: int = OVERLAP_SIZE,
          text_splitter: TextSplitter = None,
      ):
          if self.splited_docs is None or refresh:
              docs = self.file2docs()
              self.splited_docs = self.docs2texts(docs=docs,
                                                  zh_title_enhance=zh_title_enhance,
                                                  refresh=refresh,
                                                  chunk_size=chunk_size,
                                                  chunk_overlap=chunk_overlap,
                                                  text_splitter=text_splitter)
          return self.splited_docs
  
      def file_exist(self):
          return os.path.isfile(self.filepath)
  
      def get_mtime(self):
          return os.path.getmtime(self.filepath)
  
      def get_size(self):
          return os.path.getsize(self.filepath)
  
  
  def files2docs_in_thread(
          files: List[Union[KnowledgeFile, Tuple[str, str], Dict]],
          chunk_size: int = CHUNK_SIZE,
          chunk_overlap: int = OVERLAP_SIZE,
          zh_title_enhance: bool = ZH_TITLE_ENHANCE,
          pool: ThreadPoolExecutor = None,
  ) -> Generator:
      '''
      利用多线程批量将磁盘文件转化成langchain Document.
      如果传入参数是Tuple，形式为(filename, kb_name)
      生成器返回值为 status, (kb_name, file_name, docs | error)
      '''
      def file2docs(*, file: KnowledgeFile, **kwargs) -> Tuple[bool, Tuple[str, str, List[Document]]]:
          try:
              return True, (file.kb_name, file.filename, file.file2text(**kwargs))
          except Exception as e:
              msg = f"从文件 {file.kb_name}/{file.filename} 加载文档时出错：{e}"
              logger.error(f'{e.__class__.__name__}: {msg}',
                           exc_info=e if log_verbose else None)
              return False, (file.kb_name, file.filename, msg)
  
      kwargs_list = []
      for i, file in enumerate(files):
          kwargs = {}
          try:
              if isinstance(file, tuple) and len(file) >= 2:
                  filename=file[0]
                  kb_name=file[1]
                  file = KnowledgeFile(filename=filename, knowledge_base_name=kb_name)
              elif isinstance(file, dict):
                  filename = file.pop("filename")
                  kb_name = file.pop("kb_name")
                  kwargs.update(file)
                  file = KnowledgeFile(filename=filename, knowledge_base_name=kb_name)
              kwargs["file"] = file
              kwargs["chunk_size"] = chunk_size
              kwargs["chunk_overlap"] = chunk_overlap
              kwargs["zh_title_enhance"] = zh_title_enhance
              kwargs_list.append(kwargs)
          except Exception as e:
              yield False, (kb_name, filename, str(e))
  
      for result in run_in_thread_pool(func=file2docs, params=kwargs_list, pool=pool):
          yield result
  
  
  if __name__ == "__main__":
      from pprint import pprint
  
      kb_file = KnowledgeFile(filename="test.txt", knowledge_base_name="samples")
      # kb_file.text_splitter_name = "RecursiveCharacterTextSplitter"
      docs = kb_file.file2docs()
      pprint(docs[-1])
  
      docs = kb_file.file2text()
      pprint(docs[-1])
  
apis: langchain.docstore.document.Document langchain.text_splitter.TextSplitter.from_tiktoken_encoder langchain.text_splitter.TextSplitter langchain.text_splitter.TextSplitter.from_huggingface_tokenizer
extract_api: [((964, 1011), 'os.path.join', 'os.path.join', (['KB_ROOT_PATH', 'knowledge_base_name'], {}), '(KB_ROOT_PATH, knowledge_base_name)\n', (976, 1011), False, 'import os\n'), ((1789, 1807), 'server.utils.embedding_device', 'embedding_device', ([], {}), '()\n', (1805, 1807), False, 'from server.utils import run_in_thread_pool, embedding_device, get_model_worker_config\n'), ((1940, 1999), 'server.knowledge_base.kb_cache.base.embeddings_pool.load_embeddings', 'embeddings_pool.load_embeddings', ([], {'model': 'model', 'device': 'device'}), '(model=model, device=device)\n', (1971, 1999), False, 'from server.knowledge_base.kb_cache.base import embeddings_pool\n'), ((15816, 15881), 'server.utils.run_in_thread_pool', 'run_in_thread_pool', ([], {'func': 'file2docs', 'params': 'kwargs_list', 'pool': 'pool'}), '(func=file2docs, params=kwargs_list, pool=pool)\n', (15834, 15881), False, 'from server.utils import run_in_thread_pool, embedding_device, get_model_worker_config\n'), ((16147, 16163), 'pprint.pprint', 'pprint', (['docs[-1]'], {}), '(docs[-1])\n', (16153, 16163), False, 'from pprint import pprint\n'), ((16200, 16216), 'pprint.pprint', 'pprint', (['docs[-1]'], {}), '(docs[-1])\n', (16206, 16216), False, 'from pprint import pprint\n'), ((4174, 4193), 'json.loads', 'json.loads', (['content'], {}), '(content)\n', (4184, 4193), False, 'import json\n'), ((13844, 13873), 'os.path.isfile', 'os.path.isfile', (['self.filepath'], {}), '(self.filepath)\n', (13858, 13873), False, 'import os\n'), ((13915, 13946), 'os.path.getmtime', 'os.path.getmtime', (['self.filepath'], {}), '(self.filepath)\n', (13931, 13946), False, 'import os\n'), ((13987, 14017), 'os.path.getsize', 'os.path.getsize', (['self.filepath'], {}), '(self.filepath)\n', (14002, 14017), False, 'import os\n'), ((1445, 1469), 'os.listdir', 'os.listdir', (['KB_ROOT_PATH'], {}), '(KB_ROOT_PATH)\n', (1455, 1469), False, 'import os\n'), ((1641, 1661), 'os.listdir', 'os.listdir', (['doc_path'], {}), '(doc_path)\n', (1651, 1661), False, 'import os\n'), ((5418, 5461), 'importlib.import_module', 'importlib.import_module', (['"""document_loaders"""'], {}), "('document_loaders')\n", (5441, 5461), False, 'import importlib\n'), ((5514, 5567), 'importlib.import_module', 'importlib.import_module', (['"""langchain.document_loaders"""'], {}), "('langchain.document_loaders')\n", (5537, 5567), False, 'import importlib\n'), ((5742, 5829), 'configs.logger.error', 'logger.error', (['f"""{e.__class__.__name__}: {msg}"""'], {'exc_info': '(e if log_verbose else None)'}), "(f'{e.__class__.__name__}: {msg}', exc_info=e if log_verbose else\n    None)\n", (5754, 5829), False, 'from configs import EMBEDDING_MODEL, KB_ROOT_PATH, CHUNK_SIZE, OVERLAP_SIZE, ZH_TITLE_ENHANCE, logger, log_verbose, text_splitter_dict, LLM_MODEL, TEXT_SPLITTER_NAME\n'), ((5881, 5934), 'importlib.import_module', 'importlib.import_module', (['"""langchain.document_loaders"""'], {}), "('langchain.document_loaders')\n", (5904, 5934), False, 'import importlib\n'), ((10617, 10667), 'importlib.import_module', 'importlib.import_module', (['"""langchain.text_splitter"""'], {}), "('langchain.text_splitter')\n", (10640, 10667), False, 'import importlib\n'), ((10779, 10825), 'langchain.text_splitter.TextSplitter', 'TextSplitter', ([], {'chunk_size': '(250)', 'chunk_overlap': '(50)'}), '(chunk_size=250, chunk_overlap=50)\n', (10791, 10825), False, 'from langchain.text_splitter import TextSplitter\n'), ((11618, 11686), 'configs.logger.info', 'logger.info', (['f"""{self.document_loader_name} used for {self.filepath}"""'], {}), "(f'{self.document_loader_name} used for {self.filepath}')\n", (11629, 11686), False, 'from configs import EMBEDDING_MODEL, KB_ROOT_PATH, CHUNK_SIZE, OVERLAP_SIZE, ZH_TITLE_ENHANCE, logger, log_verbose, text_splitter_dict, LLM_MODEL, TEXT_SPLITTER_NAME\n'), ((12910, 12937), 'text_splitter.zh_title_enhance', 'func_zh_title_enhance', (['docs'], {}), '(docs)\n', (12931, 12937), True, 'from text_splitter import zh_title_enhance as func_zh_title_enhance\n'), ((1499, 1528), 'os.path.join', 'os.path.join', (['KB_ROOT_PATH', 'f'], {}), '(KB_ROOT_PATH, f)\n', (1511, 1528), False, 'import os\n'), ((1692, 1720), 'os.path.join', 'os.path.join', (['doc_path', 'file'], {}), '(doc_path, file)\n', (1704, 1720), False, 'import os\n'), ((3852, 3867), 'pathlib.Path', 'Path', (['file_path'], {}), '(file_path)\n', (3856, 3867), False, 'from pathlib import Path\n'), ((4809, 4855), 'langchain.docstore.document.Document', 'Document', ([], {'page_content': 'text', 'metadata': 'metadata'}), '(page_content=text, metadata=metadata)\n', (4817, 4855), False, 'from langchain.docstore.document import Document\n'), ((7841, 7881), 'importlib.import_module', 'importlib.import_module', (['"""text_splitter"""'], {}), "('text_splitter')\n", (7864, 7881), False, 'import importlib\n'), ((14757, 14844), 'configs.logger.error', 'logger.error', (['f"""{e.__class__.__name__}: {msg}"""'], {'exc_info': '(e if log_verbose else None)'}), "(f'{e.__class__.__name__}: {msg}', exc_info=e if log_verbose else\n    None)\n", (14769, 14844), False, 'from configs import EMBEDDING_MODEL, KB_ROOT_PATH, CHUNK_SIZE, OVERLAP_SIZE, ZH_TITLE_ENHANCE, logger, log_verbose, text_splitter_dict, LLM_MODEL, TEXT_SPLITTER_NAME\n'), ((8049, 8099), 'importlib.import_module', 'importlib.import_module', (['"""langchain.text_splitter"""'], {}), "('langchain.text_splitter')\n", (8072, 8099), False, 'import importlib\n'), ((8324, 8521), 'langchain.text_splitter.TextSplitter.from_tiktoken_encoder', 'TextSplitter.from_tiktoken_encoder', ([], {'encoding_name': "text_splitter_dict[splitter_name]['tokenizer_name_or_path']", 'pipeline': '"""zh_core_web_sm"""', 'chunk_size': 'chunk_size', 'chunk_overlap': 'chunk_overlap'}), "(encoding_name=text_splitter_dict[\n    splitter_name]['tokenizer_name_or_path'], pipeline='zh_core_web_sm',\n    chunk_size=chunk_size, chunk_overlap=chunk_overlap)\n", (8358, 8521), False, 'from langchain.text_splitter import TextSplitter\n'), ((9895, 10012), 'langchain.text_splitter.TextSplitter.from_huggingface_tokenizer', 'TextSplitter.from_huggingface_tokenizer', ([], {'tokenizer': 'tokenizer', 'chunk_size': 'chunk_size', 'chunk_overlap': 'chunk_overlap'}), '(tokenizer=tokenizer, chunk_size=\n    chunk_size, chunk_overlap=chunk_overlap)\n', (9934, 10012), False, 'from langchain.text_splitter import TextSplitter\n'), ((11140, 11166), 'os.path.splitext', 'os.path.splitext', (['filename'], {}), '(filename)\n', (11156, 11166), False, 'import os\n'), ((8691, 8861), 'langchain.text_splitter.TextSplitter.from_tiktoken_encoder', 'TextSplitter.from_tiktoken_encoder', ([], {'encoding_name': "text_splitter_dict[splitter_name]['tokenizer_name_or_path']", 'chunk_size': 'chunk_size', 'chunk_overlap': 'chunk_overlap'}), "(encoding_name=text_splitter_dict[\n    splitter_name]['tokenizer_name_or_path'], chunk_size=chunk_size,\n    chunk_overlap=chunk_overlap)\n", (8725, 8861), False, 'from langchain.text_splitter import TextSplitter\n'), ((9160, 9194), 'server.utils.get_model_worker_config', 'get_model_worker_config', (['llm_model'], {}), '(llm_model)\n', (9183, 9194), False, 'from server.utils import run_in_thread_pool, embedding_device, get_model_worker_config\n'), ((9592, 9633), 'transformers.GPT2TokenizerFast.from_pretrained', 'GPT2TokenizerFast.from_pretrained', (['"""gpt2"""'], {}), "('gpt2')\n", (9625, 9633), False, 'from transformers import GPT2TokenizerFast\n'), ((9699, 9818), 'transformers.AutoTokenizer.from_pretrained', 'AutoTokenizer.from_pretrained', (["text_splitter_dict[splitter_name]['tokenizer_name_or_path']"], {'trust_remote_code': '(True)'}), "(text_splitter_dict[splitter_name][\n    'tokenizer_name_or_path'], trust_remote_code=True)\n", (9728, 9818), False, 'from transformers import AutoTokenizer\n'), ((10161, 10256), 'langchain.text_splitter.TextSplitter', 'TextSplitter', ([], {'pipeline': '"""zh_core_web_sm"""', 'chunk_size': 'chunk_size', 'chunk_overlap': 'chunk_overlap'}), "(pipeline='zh_core_web_sm', chunk_size=chunk_size,\n    chunk_overlap=chunk_overlap)\n", (10173, 10256), False, 'from langchain.text_splitter import TextSplitter\n'), ((12717, 12748), 'os.path.basename', 'os.path.basename', (['self.filepath'], {}), '(self.filepath)\n', (12733, 12748), False, 'import os\n'), ((10407, 10471), 'langchain.text_splitter.TextSplitter', 'TextSplitter', ([], {'chunk_size': 'chunk_size', 'chunk_overlap': 'chunk_overlap'}), '(chunk_size=chunk_size, chunk_overlap=chunk_overlap)\n', (10419, 10471), False, 'from langchain.text_splitter import TextSplitter\n')]