langchain - 💡(How to fix) Fix RunnableSequence.astream() leaks TCP socket FDs on client disconnect

Submission checklist

• This is a bug, not a usage question.
• I added a clear and descriptive title that summarizes this issue.
• 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).
• This is not related to the langchain-community package.
• I posted a self-contained, minimal, reproducible example. A maintainer can copy it and run it AS IS.

Package (Required)

• langchain
• langchain-openai
• langchain-anthropic
• langchain-classic
• langchain-core
• langchain-model-profiles
• langchain-tests
• langchain-text-splitters
• langchain-chroma
• langchain-deepseek
• langchain-exa
• langchain-fireworks
• langchain-groq
• langchain-huggingface
• langchain-mistralai
• langchain-nomic
• langchain-ollama
• langchain-openrouter
• langchain-perplexity
• langchain-qdrant
• langchain-xai
• Other / not sure / general

Related Issues / PRs

No response

Reproduction Steps / Example Code (Python)

import asyncio, multiprocessing, os, time
import httpx

def run_server():
    import uvicorn
    from fastapi import FastAPI
    from fastapi.responses import StreamingResponse
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI

    app = FastAPI()

    @app.get("/health")
    def health():
        return {"fd": len(os.listdir(f"/proc/{os.getpid()}/fd"))}

    @app.post("/stream")
    async def stream():
        model = ChatOpenAI(model="gpt-4o-mini", streaming=True)
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are helpful."),
            ("user", "{message}"),
        ])
        chain = prompt | model

        async def generate():
            async for chunk in chain.astream({"message": "Tell me a very long story. At least 1000 words."}):
                if chunk.content:
                    yield chunk.content

        return StreamingResponse(generate(), media_type="text/event-stream")

    uvicorn.run(app, host="127.0.0.1", port=5099, log_level="warning")

async def run_client():
    url = "http://127.0.0.1:5099"

    async with httpx.AsyncClient() as c:
        r = await c.get(f"{url}/health")
        print(f"BEFORE: {r.json()}")

    for i in range(20):
        async with httpx.AsyncClient() as c:
            async with c.stream("POST", f"{url}/stream", timeout=30) as resp:
                count = 0
                async for _ in resp.aiter_bytes():
                    count += 1
                    if count >= 5:
                        break  # disconnect mid-stream

        if i % 5 == 4:
            await asyncio.sleep(1)
            async with httpx.AsyncClient() as c:
                r = await c.get(f"{url}/health")
                print(f"  After {i+1} interrupts: {r.json()}")
        else:
            await asyncio.sleep(0.3)

    await asyncio.sleep(3)
    async with httpx.AsyncClient() as c:
        r = await c.get(f"{url}/health")
        print(f"FINAL: {r.json()}")

if __name__ == "__main__":
    server = multiprocessing.Process(target=run_server, daemon=True)
    server.start()
    time.sleep(3)
    try:
        asyncio.run(run_client())
    finally:
        server.terminate()

Error Message and Stack Trace (if applicable)

No error is raised. The leak is silent — FD count monotonically increases:

BEFORE: {'fd': 12}
  After 5 interrupts: {'fd': 17}
  After 10 interrupts: {'fd': 22}
  After 15 interrupts: {'fd': 27}
  After 20 interrupts: {'fd': 32}
FINAL: {'fd': 32}

FD increases by ~1 per interrupted stream. GC does not reclaim them.

Description

chain.astream() (where chain = prompt | model, a RunnableSequence) leaks 1 TCP socket FD per client disconnect when served via FastAPI + uvicorn. Normal completions do not leak. GC does not reclaim.

The leak is specific to RunnableSequence:

Root cause

In langchain_core/runnables/base.py, RunnableSequence.astream() wraps model.astream() in multiple async generator layers via _atransform_stream_with_config() (line 2537) → _atransform() (line 3666). On client disconnect, starlette cancels the response task via anyio cancel scope, throwing CancelledError through the generator chain. The httpx response's __aexit__() is invoked, but it internally awaits stream.aclose() to return the connection to the pool — and this inner await is also cancelled by the same cancel scope. The socket is never returned.

Without RunnableSequence, model.astream() has a shorter cleanup path and __aexit__ completes before the cancel scope takes effect.

In production (voice AI app with frequent mid-stream interruptions), this led to FD exhaustion and service outage.

Workaround

Call the openai SDK directly instead of chain.astream().

System Info

System Information

OS: Linux OS Version: #1 SMP Fri Nov 29 17:22:03 UTC 2024 Python Version: 3.12.13 (main, May 19 2026, 23:55:49) [GCC 14.2.0]

Package Information

langchain_core: 1.4.2 langsmith: 0.8.11 langchain_openai: 1.2.2 langchain_protocol: 0.0.16

Optional packages not installed

deepagents deepagents-cli

Other Dependencies

httpx: 0.28.1 jsonpatch: 1.33 openai: 2.41.0 orjson: 3.11.9 packaging: 26.2 pydantic: 2.13.4 pyyaml: 6.0.3 requests: 2.34.2 requests-toolbelt: 1.0.0 tenacity: 9.1.4 tiktoken: 0.13.0 typing-extensions: 4.15.0 uuid-utils: 0.16.0 websockets: 16.0 xxhash: 3.7.0 zstandard: 0.25.0