from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

store = InMemoryVectorStore(OpenAIEmbeddings())
await store.aadd_documents(documents)  # 5000+ documents


### Proposed Solution

### Proposed Solution

Replace sequential `await` loops with `asyncio.gather` in two methods of 
`OpenAIEmbeddings` (`libs/partners/openai/langchain_openai/embeddings/base.py`):

**1. `_aget_len_safe_embeddings`** (main async path)
**2. `aembed_documents` fast path** (when `check_embedding_ctx_length=False`)

Both currently do:
```python
# Sequential — each batch waits for the previous
while i < len(tokens):
    response = await self.async_client.create(input=batch, ...)
    i = batch_end
Change to:


# Pre-compute batch boundaries (no I/O dependency between batches)
batch_ranges = [...]

# Fire all API calls concurrently, results stay in order
batch_results = await asyncio.gather(
    *[_embed_batch(s, e) for s, e in batch_ranges]
)

Why this approach:

asyncio.gather preserves input order — no result reordering needed
Fix lives in the embedding provider (not the vector store), so every consumer (InMemoryVectorStore, Chroma, Pinecone, etc.) benefits automatically
No changes to the Embeddings ABC or any vector store code
No new parameters or breaking changes
Matches the existing pattern in MistralAIEmbeddings.aembed_documents
Sync methods (embed_documents) are left unchanged — no threading risk
OpenAI SDK's built-in retry/backoff handles rate limiting naturally
Scope: Only libs/partners/openai is touched (1 source file + tests).

### Alternatives Considered

### Alternatives Considered

**1. Multithreading with `ThreadPoolExecutor` in sync `embed_documents`**
Rejected — threads are heavyweight (~8MB stack each), require GIL 
contention management, and add thread-safety risks for the HTTP client. 
The async path with `asyncio.gather` achieves the same concurrency at 
near-zero overhead (~240 bytes per coroutine).

**2. Concurrency at the vector store layer (`InMemoryVectorStore.aadd_documents`)**
Rejected — the vector store doesn't know the provider's optimal batch size, 
token limits, or API constraints. Splitting texts at this layer would create 
redundant nested batching (vector store batches → provider batches internally). 
Wrong abstraction layer.

**3. Adding a concurrent method to the `Embeddings` ABC**
Rejected — would force every embedding provider to implement batching/
concurrency logic, even providers where it doesn't apply (local models, 
single-request APIs). The ABC should stay minimal.

**4. `asyncio.as_completed` instead of `asyncio.gather`**
Rejected — returns results in completion order, not submission order. Would 
require extra bookkeeping to maintain the positional mapping that 
`InMemoryVectorStore.aadd_documents` relies on (it zips vectors with 
documents by index). `asyncio.gather` preserves order by contract.

**5. `gather_with_concurrency` with semaphore (bounded concurrency)**
Considered but deferred — `langchain-core` has this utility at 
`runnables/utils.py`, but unbounded `asyncio.gather` matches the existing 
MistralAI precedent and the OpenAI SDK already handles rate limiting via 
built-in retry/backoff. A `max_concurrency` parameter can be added later 
if needed without breaking changes.


### Additional Context

### Additional Context

**Existing precedent in this repo:**
MistralAI embeddings already implement this exact pattern — concurrent 
batch API calls via `asyncio.gather`:
https://github.com/langchain-ai/langchain/blob/master/libs/partners/mistralai/langchain_mistralai/embeddings.py#L312-L314

```python
# MistralAI's existing implementation
batch_responses = await asyncio.gather(
    *[_aembed_batch(batch) for batch in self._get_batches(texts)]
)