企业官网建设流程全解析

前言

写 FastAPI 后端、LangChain 异步向量库接口的时候，异步方法是绕不开的知识点。 很多新手照着网上代码抄，定义了async def函数，调用的时候忘记写await，打印出来看到<coroutine object xxx at ...>，一脸懵，不知道为什么拿不到返回结果。 还有的同学分不清：哪里可以写 await，同步函数里面能不能直接 await，asyncio.run()什么时候用。今天手把手，带大家吃透异步方法的编写与调用。

说明：基于标准 Python asyncio 语法，适配 FastAPI 开发环境。

一、怎么编写异步方法

定义异步函数很简单，函数声明前面加上关键字async。 用async def定义出来的函数，就叫异步方法（异步协程函数）。 在异步函数内部，才可以使用await去等待耗时操作。

import asyncio # ✅ 异步方法编写示例 async def async_get_msg(name: str): print(f"开始执行：{name}") # await 只能写在async def里面，用来等待IO操作 await asyncio.sleep(2) return f"Hello {name}, 异步执行完成"

重点记住：

1. async def：标记这是协程函数；
2. await：用来挂起当前协程，等待 IO 任务（网络请求、文件读写、向量库调用、数据库查询），不能放在普通 def 函数里；
3. asyncio.sleep()是异步休眠，不要用time.sleep()，time.sleep 会阻塞整个事件循环，异步项目大忌。

二、异步方法的调用规则

⚠️ 非常重要：调用async def定义的函数，不会直接执行函数里面的代码，而是返回一个协程对象。 只有对协程对象使用await，或者交给 asyncio 运行，函数体才真正执行！

方式 1：在其他异步函数内调用（FastAPI 接口里面就是这种场景）

FastAPI 的接口函数如果写成async def，这个接口函数本身就在异步事件循环中，可以直接 await。

from fastapi import FastAPI import asyncio app = FastAPI() async def async_get_msg(name: str): await asyncio.sleep(2) return f"Hello {name}" # FastAPI异步接口，可以直接await调用异步方法 @app.get("/test") async def test(): res = await async_get_msg("RAG") return {"result": res}

这是我们 RAG 后端最常用的写法。

方式 2：脚本入口，同步环境启动协程：asyncio.run ()

写独立测试脚本，入口是同步代码，用asyncio.run(协程对象)来启动事件循环。

import asyncio async def async_get_msg(name: str): await asyncio.sleep(2) return f"Hello {name}" async def main(): # 在异步main函数中await调用 ret = await async_get_msg("测试") print(ret) if __name__ == "__main__": # 启动事件循环，执行协程 asyncio.run(main())

❌ 新手经典错误写法（必踩坑）

# 错误！！ async def async_get_msg(name: str): await asyncio.sleep(2) return f"Hello {name}" if __name__ == "__main__": res = async_get_msg("测试") # 只拿到协程对象，代码不执行！ print(res)

运行输出：coroutine object async_get_msg at 0x...，没有执行函数内部打印和返回。

三、同步函数里面，怎么调用异步方法？

这个是很多 FastAPI 项目、SQLAlchemy 开发时高频问题。 普通def同步函数，不能直接写 await。 这时可以用asyncio.run()，但是注意：事件循环不能嵌套。

如果当前已经处在异步事件循环中（async 接口内），再调用asyncio.run()会直接报错。

import asyncio async def async_add(a, b): await asyncio.sleep(1) return a + b # 同步函数 def sync_call_async(): # 在同步函数中启动事件循环执行异步代码 result = asyncio.run(async_add(1, 2)) print(result) sync_call_async()

划重点：asyncio.run()适合还没有事件循环的地方，脚本入口、普通同步函数。在已经存在事件循环（async 接口里面）禁止使用。 如果在异步环境里，想同步等待协程，不能用 asyncio.run ()，可以用asyncio.create_task()并发执行。

四、并发调用多个异步任务

异步最大优势就是 IO 并发，多个异步方法可以一起跑，而不是串行等待。

import asyncio async def work(name, delay): await asyncio.sleep(delay) print(f"{name} 执行完毕") async def main(): # 创建多个任务，并发执行 task1 = asyncio.create_task(work("任务A", 2)) task2 = asyncio.create_task(work("任务B", 2)) await task1 await task2 asyncio.run(main())

两个任务几乎同时开始，总共耗时约 2 秒，而不是 4 秒。这就是异步 IO 的优势，适合数据库查询、大模型 invoke、向量库检索这类 IO 密集场景。

五、新手高频踩坑清单

❌坑 1：async 函数调用不加 await，拿到协程对象，误以为是返回结果。

现象：打印看到 coroutine object，没有执行函数体。

❌坑 2：在普通 def 函数里面直接写 await。

报错：SyntaxError: 'await' outside async function

❌坑 3：异步代码里面使用 time.sleep ()。

time.sleep 是阻塞式，会卡住整个事件循环，其他请求全部卡住。异步里面统一用await asyncio.sleep()。

❌坑 4：在 FastAPI 的 async 接口内部调用 asyncio.run ()。

报错：RuntimeError: asyncio.run () cannot be called from a running event loop，事件循环嵌套异常。

✅坑 5：混淆 CPU 密集和 IO 密集

异步适合 IO 密集（网络、数据库、文件、大模型调用）。CPU 密集计算不适合 async，async 不能加速计算。

📖答辩口述小抄（直接背）

问：简述 Python 异步方法的编写与调用？ 答：使用 async def 定义异步协程方法，在异步函数内部可以用 await 等待 IO 操作。调用异步函数会返回协程对象，必须通过 await 或者 asyncio.run () 执行。在 FastAPI 异步接口内直接 await 调用；在同步代码中，可以用 asyncio.run 启动事件循环。异步适合 IO 密集任务，实现并发，但是要注意不能在正在运行的事件循环里再次调用 asyncio.run，防止循环嵌套报错。

总结

记住核心口诀：async def 定义异步函数，await 只能写在 async 里面； 调用不加 await，得到协程对象； 脚本入口用 asyncio.run，异步环境直接 await； IO 密集用异步，CPU 计算别硬上。

在 RAG 项目里，异步大模型调用、异步数据库、异步向量库，都是基于这套 async/await。合理使用异步，可以提高接口并发能力，提升后端吞吐量。