asyncio/selector_events.py: ResourceWarning: unclosed transport _SelectorSocketTransport

[ziqizhang]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

import asyncio,sys
from typing import Any, Dict, List
from datetime import datetime
from util import langchain_util
from langchain.callbacks.base import AsyncCallbackHandler, BaseCallbackHandler
from langchain_community.callbacks import get_openai_callback
from langchain_core.messages import HumanMessage
from langchain_core.outputs import LLMResult
from langchain_openai import ChatOpenAI

import weaviate # conflict


class MyCustomAsyncHandler(AsyncCallbackHandler):
    """Async callback handler that can be used to handle callbacks from langchain."""

    async def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
    ) -> None:
        """Run when chain starts running."""
        await asyncio.sleep(0.3)
        class_name = serialized["name"]
        #print("(started))")

    async def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        """Run when chain ends running."""
        await asyncio.sleep(0.3)
        print(" >>> Your joke: {}".format(response.generations[0][0].text))

async def test(api):
    chat = ChatOpenAI(
        openai_api_key=api,
        max_tokens=25,
        callbacks=[langchain_util.CustomAsyncHandler()],
    )
    with get_openai_callback() as cb:
        res=await chat.agenerate([[HumanMessage(content="Tell me a joke")]])
        #print("ok")


async def test_batch(api):
    runs=5
    print(">{} runs at a time...".format(runs))
    tasks1 = []
    for i in range(0, runs):
        tasks1.append(test(api))

    await asyncio.gather(*tasks1)
    print("\t tasks1 done@ {}".format(datetime.now()))

    tasks2 = []
    for i in range(0, runs):
        tasks2.append(test(api))
    await asyncio.gather(*tasks2)
    print("\t tasks2 done@ {}".format(datetime.now()))

    tasks3 = []
    for i in range(0, runs):
        tasks3.append(test(api))
    await asyncio.gather(*tasks3)
    print("\t tasks3 done@ {}".format(datetime.now()))
    return "ok"

def run_async(api):
    count = 1
    loop = asyncio.new_event_loop()
    while True:
        print("[batch={}]".format(count))
        res = loop.run_until_complete(test_batch(api))
        count += 1

if __name__=='__main__':

    print("testing running async from a sync method")
    run_async("my open ai api key")
    print("ok")

Description

I followed Langchain documentation to write a testing code for making asynchronous calls. I have a simple AsyncCallBackHandler that prints the response on completion. This passed to the ChatOpenAI object in the method 'test'. Then in 'test_batch' I build 3 lists each makes 5 calls to 'test'.

The code seems to conflict with weaviate (https://weaviate.io/) the vector DB I need to use. This is because, when running the above code, I keep getting multiple warning message below and the program hangs after a random time:

/home/zz/Programs/miniconda3/lib/python3.10/asyncio/selector_events.py:710: ResourceWarning: unclosed transport <_SelectorSocketTransport fd=19 read=idle write=<idle, bufsize=0>>
  _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
ResourceWarning: Enable tracemalloc to get the object allocation traceback

However, the problem is gone if I comment out the last import line:

import weaviate # conflict
That's it, a single, import line that imports the weaviate package causes this issue. Removing the import the warning messages are gone.

This seems to indicate some conflict between weaviate and langchain. Any help?

System Info

langchain=0.1.9
python=3.10.12
ubuntu 22.04
weaviate-client = "3.24.1"

[dosubot[bot]]

🤖

Hey there @ziqizhang! Great to see you diving into another challenge with LangChain. Let’s unravel this mystery together.

The issue you're experiencing seems to be related to the management of asynchronous resources or connections in the Weaviate client when used within the LangChain framework. The Weaviate client might be opening network connections that are not being properly closed, which can lead to the ResourceWarning you're seeing.

In Python, especially with asynchronous code, it's common to manage such resources using context managers (async with) or by explicitly closing connections in a finally block or using a destructor method (__del__). However, from the code you've provided, it's not clear if such practices are being implemented.

To resolve this issue, you should ensure that all network connections or asynchronous tasks initiated by the Weaviate client are properly awaited and closed. If the Weaviate client provides a method to close connections or a context manager interface, it should be used whenever the client is no longer needed. If the client does not provide such mechanisms, you might need to implement a workaround that ensures all resources are properly managed and released.

Without specific details on how the Weaviate client manages its connections and whether it provides a way to close them, it's challenging to provide a precise solution. However, the general advice is to review and ensure proper management of asynchronous resources and connections when integrating external clients into frameworks or applications.

As for your question about known issues or conflicts between the LangChain framework and other Python packages, especially those that use asynchronous operations, I wasn't able to find any specific information in the LangChain repository. However, it's always a good practice to ensure proper resource management when integrating external libraries or clients into your application, as improper resource management can lead to issues like the one you're experiencing.

For more information, you might want to review the Weaviate client's documentation or source code to understand how it manages its connections and whether it provides a way to close them. If you're still experiencing issues, please provide more details about how you're using the Weaviate client in your application, and I'll do my best to assist you further.

Sources

• libs/langchain/langchain/vectorstores/weaviate.py

About Dosu

This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

If you want to continue the conversation, start your reply with @dosu-bot.

[knkski]

I'm getting the same issue with weaviate client 3.26.2 and langchain 0.1.16. Even if I don't trigger any of the code that uses the weaviate library, simply having import weaviate in a file that gets imported but not used is enough to trigger this behavior.

Looking at the weaviate codebase, I don't see anything that jumps out to me that would trigger this behavior. It might be in one of weaviate's dependencies instead of weaviate itself, it's hard to tell.

[axelstram]

Have you found a solution for this?