2025-09-14 langchain学习-09 自定义文件读取器

from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document
from typing import Iterator,AsyncIterator
import asyncio

class MyDocLoader(BaseLoader):
  def __init__(self,file_path:str)->None:
    self.file_path=file_path
  
  def lazy_load(self)->Iterator[Document]:
    line_num=0
    with open(self.file_path,encoding='utf-8') as f:
      for line in f:
        #判断去除首尾空字符后返回值是否为空
        if not line.strip():
          continue
        yield(
          Document(
            page_content=line,
            matedata={"line_num":line_num,"source":self.file_path}
          )
        )
        line_num+=1
  
  async def alazy_load(self)->AsyncIterator[Document]:
    line_num=0
    import aiofiles
    async with aiofiles.open(self.file_path,encoding='utf-8') as f:
      async for line in f:
        yield Document(
          page_content=line,
          matedata={"line_num":line_num,"source":self.file_path}
        )
        line_num+=1
      
def func1():
  loader=MyDocLoader("C:/Users/eva/Desktop/test.txt")
  for doc in loader.lazy_load():
    print(doc)

#func1()

async def func2():
  loader=MyDocLoader("C:/Users/eva/Desktop/test.txt")
  async for doc in loader.alazy_load():
    print(doc)

asyncio.run(func2())