[Python] Unclosed Client Session and Connector Issue When Using LangChain with AlloyDB for...

I'm encountering an issue when using LangChain with Google AlloyDB for vector searching. I implemented an asynchronous context manager to handle the initialization and cleanup of resources such as the AlloyDB engine and an aiohttp.ClientSession. Despite calling close() on these resources, I still see the following errors in my logs:

2025-01-22 10:22:19 - [BOT-ESPECIALIST] - INFO: Closed connections with AlloyDB.
2025-01-22 10:22:20 - [BOT-ESPECIALIST] - ERROR: Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7cccbee73490>
2025-01-22 10:22:20 - [BOT-ESPECIALIST] - ERROR: Unclosed connector
connections: ['deque([(<aiohttp.client_proto.ResponseHandler object at 0x7cccbee67ac0>, 112391.933409363), (<aiohttp.client_proto.ResponseHandler object at 0x7cccbee1e040>, 112392.240980781)])']
connector: <aiohttp.connector.TCPConnector object at 0x7cccbee734c0>

What I'm Using:


LangChain: To interact with AlloyDB for vector storage and searching. Google AlloyDB: A managed database where vector data is stored and queried. aiohttp.ClientSession: Used internally by AlloyDB's LangChain integration for asynchronous HTTP requests. Relevant Code: Here’s my AlloyDB class with the context manager implementation:

import aiohttp
import asyncio
import logging
from typing import List, Union
from langchain_google_alloydb_pg import AlloyDBEngine, AlloyDBVectorStore
from langchain_core.documents import Document
from langchain_core.embeddings.embeddings import Embeddings

class AlloyDB:
def __init__(self, connection, embedding_model):
self.engine: Union[AlloyDBEngine, None] = None
self.connection = connection
self.embedding_model = embedding_model
self.vector_store: Union[AlloyDBVectorStore, None] = None
self.session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=4))

async def __aenter__(self):
"""Initialize resources."""
self.engine = await AlloyDBEngine.afrom_instance(
project_id=self.connection.project_id,
region=self.connection.region,
cluster=self.connection.cluster,
instance=self.connection.instance,
database=self.connection.database,
user=self.connection.db_user,
password=self.connection.db_password,
)
return self

async def __aexit__(self, exc_type, exc_value, traceback):
"""Clean up resources."""
if self.engine:
self.engine.close()
if not self.session.closed:
await self.session.close()
logging.info("Closed connections with AlloyDB.")


And here's how I use it in my main function:

async def main():
async with AlloyDB(connection, embedding_model) as db:
await db.init_vector_storage_table(table="products", table_config=table_config)
query = "I'd like a fruit."
docs = await db.search_documents(query)
print(docs)

loop = asyncio.get_event_loop()
try:
loop.run_until_complete(main())
finally:
loop.close()

Observations:


Resource Cleanup: I explicitly call self.engine.close() and await self.session.close() in the aexit method, yet the error persists.

Error Details: The error indicates there’s an unclosed ClientSession and associated connectors, even though await self.session.close() is called.

Questions:

• What could be causing the aiohttp.ClientSession to remain unclosed despite explicitly calling close()?
• Is there a potential issue with how LangChain's AlloyDB integration uses aiohttp internally that might lead to this error?
• Are there any recommended patterns or best practices for managing aiohttp.ClientSession within an async context manager like this?

Continue reading...