aiohttp RuntimeError: Session is closed with InfinityEmbeddings in async

[baptiste-pasquier]

Checked other resources

• I added a very descriptive title to this issue.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.
• I am sure that this is a bug in LangChain rather than my code.
• The bug is not resolved by updating to the latest stable version of LangChain (or the specific integration package).

Example Code

from langchain_community.embeddings import InfinityEmbeddings

query = "Where is Paris?"

infinity_api_url = "http://localhost:7997"

embeddings = InfinityEmbeddings(
    model="BAAI/bge-small-en-v1.5", infinity_api_url=infinity_api_url
)

The first async query works:

query_result = await embeddings.aembed_query(query)
print("embeddings created successful")

Output:

embeddings created successful

But the second one raises an error:

query_result = await embeddings.aembed_query(query)
print("embeddings created successful")

Error trace

Error Message and Stack Trace (if applicable)

---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)
Cell In[4], line 1
----> 1 query_result = await embeddings.aembed_query(query)
    2 print("embeddings created successful")

File ~/Library/Caches/pypoetry/virtualenvs/bug-infinity-cYOuLYza-py3.10/lib/python3.10/site-packages/langchain_community/embeddings/infinity.py:115, in InfinityEmbeddings.aembed_query(self, text)
    106 async def aembed_query(self, text: str) -> List[float]:
    107     """Async call out to Infinity's embedding endpoint.
    108 
    109     Args:
(...)
    113         Embeddings for the text.
    114     """
--> 115     embeddings = await self.aembed_documents([text])
    116     return embeddings[0]

File ~/Library/Caches/pypoetry/virtualenvs/bug-infinity-cYOuLYza-py3.10/lib/python3.10/site-packages/langchain_community/embeddings/infinity.py:89, in InfinityEmbeddings.aembed_documents(self, texts)
    80 async def aembed_documents(self, texts: List[str]) -> List[List[float]]:
    81     """Async call out to Infinity's embedding endpoint.
    82 
    83     Args:
(...)
    87         List of embeddings, one for each text.
    88     """
---> 89     embeddings = await self.client.aembed(
    90         model=self.model,
    91         texts=texts,
    92     )
    93     return embeddings

File ~/Library/Caches/pypoetry/virtualenvs/bug-infinity-cYOuLYza-py3.10/lib/python3.10/site-packages/langchain_community/embeddings/infinity.py:317, in TinyAsyncOpenAIInfinityEmbeddingClient.aembed(self, model, texts)
    310     self.aiosession = aiohttp.ClientSession(
    311         trust_env=True, connector=aiohttp.TCPConnector(limit=32)
    312     )
    313 async with self.aiosession as session:
    314 # async with aiohttp.ClientSession(
    315 #             connector=aiohttp.TCPConnector(limit=32)
    316 #         ) as session:
--> 317     embeddings_batch_perm = await asyncio.gather(
    318         *[
    319             self._async_request(
    320                 session=session,
    321                 kwargs=self._kwargs_post_request(model=model, texts=t),
    322             )
    323             for t in perm_texts_batched
    324         ]
    325     )
    327 embeddings_perm = self._unbatch(embeddings_batch_perm)
    328 embeddings = unpermute_func(embeddings_perm)

File ~/Library/Caches/pypoetry/virtualenvs/bug-infinity-cYOuLYza-py3.10/lib/python3.10/site-packages/langchain_community/embeddings/infinity.py:286, in TinyAsyncOpenAIInfinityEmbeddingClient._async_request(self, session, kwargs)
    283 async def _async_request(
    284     self, session: aiohttp.ClientSession, kwargs: Dict[str, Any]
    285 ) -> List[List[float]]:
--> 286     async with session.post(**kwargs) as response:
    287         if response.status != 200:
    288             raise Exception(
    289                 f"Infinity returned an unexpected response with status "
    290                 f"{response.status}: {response.text}"
    291             )

File ~/Library/Caches/pypoetry/virtualenvs/bug-infinity-cYOuLYza-py3.10/lib/python3.10/site-packages/aiohttp/client.py:1355, in _BaseRequestContextManager.__aenter__(self)
1354 async def __aenter__(self) -> _RetType:
-> 1355     self._resp: _RetType = await self._coro
1356     return await self._resp.__aenter__()

File ~/Library/Caches/pypoetry/virtualenvs/bug-infinity-cYOuLYza-py3.10/lib/python3.10/site-packages/aiohttp/client.py:492, in ClientSession._request(self, method, str_or_url, params, data, json, cookies, headers, skip_auto_headers, auth, allow_redirects, max_redirects, compress, chunked, expect100, raise_for_status, read_until_eof, proxy, proxy_auth, timeout, verify_ssl, fingerprint, ssl_context, ssl, server_hostname, proxy_headers, trace_request_ctx, read_bufsize, auto_decompress, max_line_size, max_field_size)
    450 async def _request(
    451     self,
    452     method: str,
(...)
    488     # set the default to None because we need to detect if the user wants
    489     # to use the existing timeouts by setting timeout to None.
    491     if self.closed:
--> 492         raise RuntimeError("Session is closed")
    494     ssl = _merge_ssl_params(ssl, verify_ssl, ssl_context, fingerprint)
    496     if data is not None and json is not None:

RuntimeError: Session is closed

Description

I'm trying to use InfinityEmbeddings asynchronously. The first async query works, but the second one consistently fails with a RuntimeError: Session is closed.

The problem comes from the aiohttp.ClientSession which is closed at the end of the with statement.

The session should be redefined directly within the with statement.

langchain/libs/community/langchain_community/embeddings/infinity.py

Lines 309 to 322 in c6350d6

	if self.aiosession is None:
	self.aiosession = aiohttp.ClientSession(
	trust_env=True, connector=aiohttp.TCPConnector(limit=32)
	)
	async with self.aiosession as session:
	embeddings_batch_perm = await asyncio.gather(
	*[
	self._async_request(
	session=session,
	kwargs=self._kwargs_post_request(model=model, texts=t),
	)
	for t in perm_texts_batched
	]
	)

System Info

System Information

OS: Darwin
OS Version: Darwin Kernel Version 23.5.0: Wed May 1 20:16:51 PDT 2024; root:xnu-10063.121.3~5/RELEASE_ARM64_T8103
Python Version: 3.10.13 (main, Feb 6 2024, 09:59:07) [Clang 15.0.0 (clang-1500.1.0.2.5)]

Package Information

langchain_core: 0.3.6
langchain: 0.3.1
langchain_community: 0.3.1
langsmith: 0.1.129
langchain_text_splitters: 0.3.0

Optional packages not installed

langgraph
langserve

Other Dependencies

aiohttp: 3.10.6
async-timeout: 4.0.3
dataclasses-json: 0.6.7
httpx: 0.27.2
jsonpatch: 1.33
numpy: 1.26.4
orjson: 3.10.7
packaging: 24.1
pydantic: 2.9.2
pydantic-settings: 2.5.2
PyYAML: 6.0.2
requests: 2.32.3
SQLAlchemy: 2.0.35
tenacity: 8.5.0
typing-extensions: 4.12.2