Async text splitter

[pprados]

The documentation of BaseLoader
say: Implementations should implement the lazy-loading method using generators to avoid loading all Documents into memory at once.

I fully agree with this objective. But the current API doesn't make it possible.

The TextSplitter API is incompatible with an async approach.

async def aload() -> VectorStore:
    db_url = "postgresql+psycopg://postgres:password_postgres@localhost:5432/"
    engine = create_async_engine(db_url, echo=True)

    loader = CSVLoader(
        "data/faq/faq.csv",
        source_column="source",
        metadata_columns=["question", "source"],
        encoding=None,
        csv_args={},
        autodetect_encoding=True,
    )
    text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
    raw_documents = loader.alazy_load()

    # TypeError: 'async_generator' object is not iterable
    documents = await text_splitter.atransform_documents(raw_documents)
    
    db = await PGVector.afrom_documents(
        documents,
        embedding=FakeEmbeddings(),
        connection=engine,
    )
    return db

With this patch in langchain_text_splitters/base.py#textSplitter

    async def atransform_documents(
        self, documents: AsyncIterator[Document], **kwargs: Any
    ) -> AsyncIterator[Document]:
        """Asynchronously transform a list of documents.

        Args:
            documents: A sequence of Documents to be transformed.

        Returns:
            A list of transformed Documents.
        """
        return await run_in_executor(
            None, self.transform_documents, documents, **kwargs
        )

It is then possible to have a Splitter from a lazy load. But it's no longer lazy.
To remain lazy, if must be implemented this way:

    async def atransform_documents(
        self, documents: AsyncIterator[Document], **kwargs: Any
    ) -> AsyncIterator[Document]:
        """Transform sequence of documents by splitting them."""
        async for doc in documents:
            yield self.split_documents([doc])

In this case, the caller must be modified accordingly:

    documents = [doc async for doc in text_splitter.atransform_documents(raw_documents)]
    db = await PGVector.afrom_documents(
        documents,
        embedding=FakeEmbeddings(),
        connection=engine,
    )

It's still not lazy. For this to be the case, you need to modify afrom_documents() from :

    def from_documents(
        cls: Type[PGVector],
        documents: List[Document],
        embedding: Embeddings,
        *,
        connection: Optional[Connection] = None,
        collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,
        distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,
        ids: Optional[List[str]] = None,
        pre_delete_collection: bool = False,
        use_jsonb: bool = True,
        **kwargs: Any,
    ) -> PGVector:

to

    def from_documents(
        cls: Type[PGVector],
        documents: AsyncIterable[Document],
        embedding: Embeddings,
        *,
        connection: Optional[Connection] = None,
        collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,
        distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,
        ids: Optional[List[str]] = None,
        pre_delete_collection: bool = False,
        use_jsonb: bool = True,
        **kwargs: Any,
    ) -> PGVector:

possibly with:

        documents: AsyncIterable[Document]|Iterable[Document],

and review all implementations.
What do you think? This conforms to the BaseLoader approach.

Note that I had already proposed a solution for making transformations lazy, while making them Runnable. Shall I try it?

[pprados]

@baskaryan, @hwchase17 ?