Cursor 写 Python 异步脚本总卡死?用 .cursorrules 根治 AI 的“异步幻觉”

用 Cursor 生成 Python 异步脚本时，最让人血压飙升的瞬间莫过于：代码看着挺美，一跑直接卡死，或者抛出 RuntimeError: This event loop is already running，仔细一看，AI 竟然在 async def 里塞了同步阻塞的 requests.get。

要解决 AI 缺乏上下文导致的“异步幻觉”，反复在对话框里纠正太低效了。我摸索下来的最终选型是：在项目根目录引入 .cursorrules 作为全局 Prompt，强制约束 AI 弃用 requests，统一用 httpx，并规范 asyncio 入口。配好后，Cursor 生成的代码直接具备生产级并发能力，无需二次重构。

锁死技术栈：编写 .cursorrules 终结 AI 幻觉

AI 偏爱 requests 是因为训练数据权重高。我之前就踩过坑，AI 偷偷在 async def 里塞了个 requests.get，导致整个事件循环卡死半小时。 要让 Cursor 长记性，必须在 .cursorrules 里画红线。

在项目根目录创建 .cursorrules，写入以下约束。关键是不光得告诉 AI 别干啥，还得把替代方案和代码范式喂给它。

# Python 异步工程规范 (Cursor Rules)

## 技术栈约束
- 严禁在任何 `async def` 函数或异步上下文中使用 `requests`、`urllib` 等同步网络库。
- 所有 HTTP 请求必须使用 `httpx` 库。同步环境使用 `httpx.Client`，异步环境必须使用 `httpx.AsyncClient`。
- 异步任务并发控制使用 `asyncio.gather` 或 `asyncio.TaskGroup` (Python 3.11+)，严禁使用 `time.sleep`，必须使用 `asyncio.sleep`。

## 异步上下文规范
- 实例化 `httpx.AsyncClient` 时，必须使用 `async with` 上下文管理器，确保连接池正确释放。
- 如果需要在 Jupyter Notebook 或已有事件循环的环境（如 Streamlit/FastAPI）中运行异步代码，必须引入 `nest_asyncio` 并调用 `nest_asyncio.apply()`，或者封装为独立的线程运行。
- 脚本入口必须统一使用 `if __name__ == "__main__": asyncio.run(main())`。

## 异常处理
- 网络请求必须捕获 `httpx.HTTPStatusError` 和 `httpx.RequestError`，并实现带有指数退避（Exponential Backoff）的重试机制，推荐使用 `tenacity` 库。

保存后，Cursor 生成代码前会自动读取。你会发现，AI 再也写不出 await requests.get() 这种让人啼笑皆非的代码了。

改造网络请求层：httpx 异步客户端的正确封装

规则定好了，接下来看实战落地。官方文档推荐每次请求都 async with httpx.AsyncClient()，但实际高并发项目里我更倾向于在类级别维护一个全局 Client 实例。因为前者频繁创建销毁 TCP 连接太慢了，后者能真正吃满连接池复用的红利。

下面是符合规范的请求层封装：

import httpx
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

class AsyncFetcher:
    def __init__(self, max_connections: int = 100):
        # 限制最大并发连接数，防止触发目标站点的反爬或导致本地端口耗尽
        limits = httpx.Limits(max_connections=max_connections, max_keepalive_connections=20)
        self.client = httpx.AsyncClient(limits=limits, timeout=10.0)

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    async def fetch(self, url: str) -> dict:
        try:
            response = await self.client.get(url)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            print(f"HTTP error {e.response.status_code} for {url}")
            raise
        except httpx.RequestError as e:
            print(f"Request error for {url}: {e}")
            raise

    async def close(self):
        await self.client.aclose()

async def main():
    fetcher = AsyncFetcher()
    urls = [f"https://api.example.com/data/{i}" for i in range(50)]

    try:
        # 使用 asyncio.gather 并发执行，共享同一个 client 连接池
        tasks = [fetcher.fetch(url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        print(f"成功获取 {sum(1 for r in results if not isinstance(r, Exception))} 条数据")
    finally:
        # 确保在退出前关闭客户端，释放资源
        await fetcher.close()

if __name__ == "__main__":
    asyncio.run(main())

这段代码主要解决了三个痛点：用 httpx.Limits 控连接池大小；靠 tenacity 搞定优雅重试；最后用 try...finally 兜底 aclose()，免得终端狂刷 Unclosed client session 警告。

统一事件循环入口：抹平环境差异

纯脚本用 asyncio.run() 没毛病。但平时做快速验证，我习惯直接在 Jupyter 里跑。我第一次在 Notebook 里跑异步代码时，直接抛出 RuntimeError: This event loop is already running 让我懵了半天，后来才反应过来这些环境底层已经有事件循环了。

为了让代码具备跨环境的鲁棒性，我一般在项目里封装一个统一的运行入口：

import asyncio
import sys

def run_async(coro):
    """
    抹平 Jupyter/IPython 与标准 Python 脚本的事件循环差异
    """
    try:
        # 尝试获取当前事件循环
        loop = asyncio.get_event_loop()
        if loop.is_running():
            # 如果在 Jupyter 等已有循环的环境中，注入 nest_asyncio
            import nest_asyncio
            nest_asyncio.apply()
            return loop.run_until_complete(coro)
    except RuntimeError:
        pass

    # 标准脚本环境，直接使用 asyncio.run
    if sys.version_info >= (3, 7):
        return asyncio.run(coro)
    else:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            return loop.run_until_complete(coro)
        finally:
            loop.close()

# 使用方式：
# run_async(main())

把这段代码扔进 utils.py，并在 .cursorrules 里指明优先用它。这能帮你省去 90% 切环境导致的调试时间。

避坑清单：实测报错与解法

用 Cursor 写异步代码，下面这几个坑我基本都替你们踩过了：

用 AI 写代码，上限全看你的工程约束能力。把这套 .cursorrules 喂给 Cursor，基本就能告别异步卡死这种低级错误，直接产出能上生产的代码了。

完整可运行版

# 安装依赖: pip install httpx tenacity nest_asyncio
# 运行前置条件: 需要网络连接以请求公共 API (jsonplaceholder.typicode.com)
# 运行方式: python full.py

import asyncio
import sys
import os
import httpx
import nest_asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

CURSOR_RULES_CONTENT = """# Python 异步工程规范 (Cursor Rules)

## 技术栈约束
- 严禁在任何 `async def` 函数或异步上下文中使用 `requests`、`urllib` 等同步网络库。
- 所有 HTTP 请求必须使用 `httpx` 库。同步环境使用 `httpx.Client`，异步环境必须使用 `httpx.AsyncClient`。
- 异步任务并发控制使用 `asyncio.gather` 或 `asyncio.TaskGroup` (Python 3.11+)，严禁使用 `time.sleep`，必须使用 `asyncio.sleep`。

## 异步上下文规范
- 实例化 `httpx.AsyncClient` 时，必须使用 `async with` 上下文管理器，确保连接池正确释放。
- 如果需要在 Jupyter Notebook 或已有事件循环的环境（如 Streamlit/FastAPI）中运行异步代码，必须引入 `nest_asyncio` 并调用 `nest_asyncio.apply()`，或者封装为独立的线程运行。
- 脚本入口必须统一使用 `if __name__ == "__main__": asyncio.run(main())` 或自定义的 `run_async`。

## 异常处理与并发控制
- 网络请求必须捕获 `httpx.HTTPStatusError` 和 `httpx.RequestError`，并实现带有指数退避的重试机制，推荐使用 `tenacity` 库。
- 批量请求时必须引入 `asyncio.Semaphore` 限制并发上限，防止触发反爬或端口耗尽。
"""

def generate_cursor_rules():
    """在项目根目录生成 .cursorrules 文件，约束 AI 编码规范"""
    filepath = ".cursorrules"
    with open(filepath, "w", encoding="utf-8") as f:
        f.write(CURSOR_RULES_CONTENT)
    print(f"[+] 已成功生成 {os.path.abspath(filepath)} 配置文件，Cursor 将自动读取该上下文。")

def run_async(coro):
    """
    抹平 Jupyter/IPython 与标准 Python 脚本的事件循环差异
    """
    try:
        loop = asyncio.get_running_loop()
        if loop.is_running():
            nest_asyncio.apply()
            return loop.run_until_complete(coro)
    except RuntimeError:
        pass

    if sys.version_info >= (3, 7):
        return asyncio.run(coro)
    else:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            return loop.run_until_complete(coro)
        finally:
            loop.close()

class AsyncFetcher:
    def __init__(self, max_connections: int = 100, max_concurrent_tasks: int = 10):
        limits = httpx.Limits(
            max_connections=max_connections, 
            max_keepalive_connections=20
        )
        self.client = httpx.AsyncClient(limits=limits, timeout=10.0)
        self.semaphore = asyncio.Semaphore(max_concurrent_tasks)

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    async def fetch(self, url: str) -> dict:
        async with self.semaphore:
            try:
                response = await self.client.get(url)
                response.raise_for_status()
                return response.json()
            except httpx.HTTPStatusError as e:
                print(f"[-] HTTP error {e.response.status_code} for {url}")
                raise
            except httpx.RequestError as e:
                print(f"[-] Request error for {url}: {e}")
                raise

    async def close(self):
        await self.client.aclose()

async def main():
    generate_cursor_rules()

    fetcher = AsyncFetcher(max_connections=50, max_concurrent_tasks=5)
    urls = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 11)]

    print(f"[*] 开始并发请求 {len(urls)} 个目标...")
    try:
        tasks = [fetcher.fetch(url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        success_count = sum(1 for r in results if not isinstance(r, Exception))
        print(f"[+] 成功获取 {success_count} 条数据")

        if success_count > 0:
            for r in results:
                if not isinstance(r, Exception):
                    print(f"[*] 示例数据标题: {r.get('title', 'No title')}")
                    break
    finally:
        await fetcher.close()
        print("[*] 客户端连接池已安全释放。")

if __name__ == "__main__":
    run_async(main())