使用langchain_postgres.v2.vectorstores时BlockingError问题求助

在自定义chat-langchain仓库中使用PGVectorStore V2时，尝试PGVectorStore.create_sync、PGVectorStore.create()、AsyncPGVectorStore.create()三种方法均触发blockbuster.blockbuster.BlockingError: Blocking call to socket.socket.accept错误，以下是问题复现代码、报错栈及针对性修复方案：

同步代码示例 #

from langchain_community.vectorstores import PGVectorStore

# 数据库连接配置
CONNECTION_STRING = "postgresql+psycopg2://user:pass@localhost:5432/dbname"
COLLECTION_NAME = "test_collection"

# 同步创建向量存储
vector_store = PGVectorStore.create_sync(
    connection_string=CONNECTION_STRING,
    collection_name=COLLECTION_NAME,
    embedding_function=embeddings
)

异步代码示例 #

from langchain_community.vectorstores import AsyncPGVectorStore
import asyncio

async def create_async_store():
    CONNECTION_STRING = "postgresql+psycopg2://user:pass@localhost:5432/dbname"
    COLLECTION_NAME = "test_collection"
    
    # 异步创建向量存储
    vector_store = await AsyncPGVectorStore.create(
        connection_string=CONNECTION_STRING,
        collection_name=COLLECTION_NAME,
        embedding_function=embeddings
    )
    return vector_store

asyncio.run(create_async_store())

报错栈 #

Traceback (most recent call last):
  File "main.py", line XX, in <module>
    asyncio.run(create_async_store())
  File "/usr/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
  File "main.py", line XX, in create_async_store
    vector_store = await AsyncPGVectorStore.create(...)
  File "/path/to/langchain_community/vectorstores/pgvector.py", line XX, in create
    await cls._create_table_if_not_exists(...)
  File "/path/to/langchain_community/vectorstores/pgvector.py", line XX, in _create_table_if_not_exists
    await conn.run_sync(_create_table_sync)
  File "/path/to/sqlalchemy/ext/asyncio/engine.py", line 922, in run_sync
    return await loop.run_in_executor(
  File "/usr/lib/python3.10/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/path/to/langchain_community/vectorstores/pgvector.py", line XX, in _create_table_sync
    conn.execute(text(create_table_query))
  File "/path/to/sqlalchemy/engine/base.py", line 1414, in execute
    return meth(self, multiparams, params)
  ...
  blockbuster.blockbuster.BlockingError: Blocking call to socket.socket.accept

修复方法 #

1. 替换为异步数据库驱动 #

同步驱动psycopg2在异步上下文执行会触发阻塞错误，将连接字符串改为异步驱动asyncpg：

# 修改后的异步连接字符串
CONNECTION_STRING = "postgresql+asyncpg://user:pass@localhost:5432/dbname"

2. 隔离同步操作到独立线程 #

若必须使用同步驱动，用asyncio.to_thread将同步创建逻辑隔离到线程池，避免阻塞异步事件循环：

import asyncio
from langchain_community.vectorstores import PGVectorStore

async def create_store():
    CONNECTION_STRING = "postgresql+psycopg2://user:pass@localhost:5432/dbname"
    COLLECTION_NAME = "test_collection"
    
    # 用线程隔离同步调用
    vector_store = await asyncio.to_thread(
        PGVectorStore.create_sync,
        connection_string=CONNECTION_STRING,
        collection_name=COLLECTION_NAME,
        embedding_function=embeddings
    )
    return vector_store

3. 临时禁用blockbuster阻塞检测（未尝试的选项） #

作为调试临时方案，禁用blockbuster的阻塞检测机制：

import blockbuster
blockbuster.disable()

注意：此方法仅用于调试，生产环境需优先修复异步/同步代码兼容性问题。

4. 升级依赖包 #

更新langchain-community、pgvector及asyncpg到最新版本，修复已知的异步兼容bug：

pip install --upgrade langchain-community pgvector asyncpg

内容的提问来源于stack exchange，提问作者Tom Tom