1. 这不是“另一个异步教程”而是你真正该懂的 asyncio 入门逻辑我带过十几支后端开发团队也给零基础转行的学员讲过上百场 Python 实操课。每次讲到 asyncio总有人在课后追着问“老师async/await 到底和多线程、多进程有啥本质区别”“为什么我加了 async程序反而更慢了”“那个 event loop 到底长什么样它真在后台‘跑’吗”——这些问题背后不是语法没记牢而是从一开始我们就把 asyncio 当成了“一个要学的新语法”而不是一种需要重新校准的执行模型。这篇内容不教你怎么写第一个async def hello()而是带你回到 Python 解释器运行时的底层现场看清楚 I/O 阻塞如何发生、CPU 时间片如何被浪费、协程如何被调度、event loop 如何成为整个异步世界的“交通指挥中心”。核心关键词是asyncio、event loop、协程、I/O 密集型、非阻塞、awaitable 对象。如果你正卡在“能写但不懂为什么”“能跑但调不好性能”“看文档像读天书”的阶段这篇就是为你写的。它适合两类人一是刚接触异步概念、被RuntimeWarning: coroutine xxx was never awaited报错反复打击的初学者二是已用过 Flask asyncio 或 FastAPI 但对底层调度机制模糊不清的中级开发者。它不承诺让你“5分钟上手”但保证你读完后再看到async with aiohttp.ClientSession() as session:这行代码时脑子里浮现的不再是语法符号而是一张清晰的控制流图——知道谁在等谁、谁在让出、谁在唤醒、谁在真正占用 CPU。2. 内容整体设计与思路拆解为什么必须从“阻塞”讲起2.1 真正的起点不是 async而是 time.sleep()绝大多数 asyncio 教程一上来就甩出async def和await这就像教人开车先讲涡轮增压原理却不说“踩刹车会让车停”。我们反其道而行先用最原始、最暴力的方式暴露问题。下面这段代码是你在任何 Python 环境里都能立刻运行的“阻塞标尺”import time def fetch_data_sync(url): time.sleep(3) # 模拟网络请求耗时 return fData from {url} def main_sync(): start time.time() result1 fetch_data_sync(https://api1.com) result2 fetch_data_sync(https://api2.com) result3 fetch_data_sync(https://api3.com) print(fSync total: {time.time() - start:.2f}s) main_sync()实测结果稳定在9.00 秒左右3秒 × 3次。为什么因为time.sleep(3)是真正的阻塞调用Python 解释器在此期间完全交出 CPU 控制权操作系统不会调度这个线程做任何事它就干等着。这就像你去银行柜台办业务前面三个人每人办3分钟你必须排队等满9分钟才能轮到——无论你多着急柜台就一个且不能并行处理。这个模型在单任务场景下没问题但一旦涉及大量 I/OHTTP 请求、数据库查询、文件读写CPU 大量时间在“空等”资源利用率暴跌。asyncio 的全部价值就建立在这个“空等”之上它要做的不是让 CPU 跑得更快而是让 CPU 在等待时立刻切去做别的事。2.2 异步不是“同时运行”而是“聪明地切换”很多人误以为async/await让代码“并行执行”这是根本性误解。asyncio 是单线程、单进程内的并发concurrency不是并行parallelism。它不靠多核而靠协作式多任务cooperative multitasking。关键在于“协作”二字每个任务必须主动说“我现在要等了请让我歇会儿换别人上”。这个“歇会儿”的指令就是await。它不是魔法开关而是一个明确的让出点yield point。当执行到await some_io_operation()时当前协程暂停控制权交还给 event loopevent loop 立刻检查其他已就绪的任务比如另一个协程的 I/O 已完成挑一个继续执行。整个过程像一场精心编排的接力赛没有裁判吹哨强制中断每个选手协程自己跑到交接区await就把棒子控制权交给下一位。这种模式天然规避了多线程的锁竞争、上下文切换开销大等问题但代价是——所有参与者必须守规矩。如果你写了一个async def函数里面全是 CPU 密集型计算比如sum(range(10**7))却不await那它就会霸占 CPU让其他协程彻底饿死。这就是为什么asyncio只对 I/O 密集型场景有奇效对纯计算任务几乎无益甚至更慢。2.3 Event loop不是后台服务而是你的主程序本身这是最常被神化的概念。很多资料说“event loop 在后台默默运行”让人以为它是个独立守护进程。错。在标准 Python 程序中event loop 就是你的main()函数本身。当你调用asyncio.run(main())它做的第一件事就是创建一个 event loop 实例然后把main()这个协程对象注册进去最后开始循环执行检查哪些协程已就绪、哪些 I/O 完成、哪些定时器到期然后挨个驱动它们往前走一步直到下一个await。你可以把它想象成一个永不退出的 while 循环# 伪代码示意 event loop 核心逻辑 while not all_tasks_done: # 步骤1检查已完成的 I/O如 socket 收到数据 ready_tasks get_ready_from_os() # 步骤2依次执行这些就绪任务直到它们遇到 await for task in ready_tasks: try: task.send(None) # 恢复协程执行 except StopIteration as e: # 协程执行完毕 mark_task_as_done(task) # 步骤3如果当前没有就绪任务就休眠一小会儿避免空转 if not ready_tasks: sleep(0.001)asyncio.run()就是这个 while 循环的启动器和终结者。它确保 loop 启动、任务调度、异常传播、资源清理一气呵成。你不需要、也不应该手动管理 loop除非特殊场景如嵌入到 GUI 应用。理解这一点至关重要你写的每一个async def最终都活在这个 while 循环的掌控之下你写的每一个await都是向这个循环提交的一次“请调度我”的申请。它不是黑盒而是你程序的主干神经。3. 核心细节解析与实操要点从语法糖到底层对象3.1async def生成的不是函数而是协程工厂敲下async def my_coro(): ...你得到的不是一个可直接调用的函数而是一个协程函数coroutine function。它本身不执行任何逻辑只是一个“造协程的工厂”。只有当你调用它my_coro()才真正生成一个协程对象coroutine object。这个对象是collections.abc.Coroutine的实例它实现了__await__()方法因此是 awaitable 的。验证方式很简单import asyncio async def my_coro(): await asyncio.sleep(1) return done # 这只是个函数对象类型是 class function print(type(my_coro)) # class function # 调用它才生成协程对象类型是 class coroutine coro my_coro() print(type(coro)) # class coroutine # 协程对象可被 await因为它有 __await__ 方法 # print(coro.__await__()) # 这会返回一个 generator内部驱动协程为什么设计成这样因为协程需要状态机支持。每次await暂停时协程对象内部保存着当前执行位置、局部变量等所有上下文。下次被 event loop 唤醒时它能精确地从暂停处继续。这和生成器generator的yield机制同源但语义更明确yield是通用暂停await是专为 I/O 等待设计的暂停。所以async def是语法糖它让 Python 解释器自动为你构建一个状态机类省去了手动写__next__()和send()的麻烦。但底层逻辑没变协程对象 可暂停、可恢复、带状态的执行体。3.2await的本质一个受控的yield fromawait表达式要求其右侧必须是awaitable 对象。什么是 awaitable官方定义有三类协程对象、定义了__await__()方法的对象如asyncio.Future、以及实现了__iter__()的生成器已弃用仅兼容旧代码。await的行为可以粗略理解为yield from的增强版yield from gen将生成器gen的产出逐个 yield 出来并将外部 send 的值传给gen。await coro将协程coro的执行权交给 event loop等待其完成并获取其返回值。关键区别在于控制权归属。yield from的控制权在调用者手中你决定何时 next而await的控制权在 event loop 手中它决定何时 resume。await后面的对象必须承诺自己会在某个时刻“完成”resolve否则 event loop 会永远等下去。这个“完成”的信号由对象内部的set_result()或set_exception()触发对Future或协程自然 return 触发。所以await不是简单的“等”而是一次跨协程的、由 event loop 协调的、带结果传递的协作式跳转。这也是为什么你不能await一个普通函数它没有__await__()无法告诉 event loop “我什么时候算完”。3.3asyncio.sleep()最干净的 awaitable 教学示例别小看asyncio.sleep()。它是理解 awaitable 行为的黄金样本。它的源码极简Lib/asyncio/tasks.pyasync def sleep(delay, resultNone, *, loopNone): Coroutine that completes after a given time (in seconds). if delay 0: await __sleep0() # 立即完成的 awaitable return result future Future(looploop) h loop.call_later(delay, future.set_result, result) try: return await future except BaseException: h.cancel() raise看懂了吗它没做任何“真睡”而是创建一个Future对象一个可设置结果的占位符让 event loop 在delay秒后调用future.set_result(result)await future—— 这才是真正的等待点。此时协程暂停event loop 继续调度其他任务当set_result()被调用future变为 done 状态event loop 自动 resume 这个协程并把result作为await表达式的值返回。整个过程asyncio.sleep()本身不消耗 CPU不阻塞线程只是一次对 event loop 的“预约”。它完美展示了 awaitable 的核心契约我提供一个未来会完成的承诺你event loop负责监督并通知我。你完全可以自己实现一个awaitableclass MyAwaitable: def __init__(self, delay): self.delay delay def __await__(self): # 返回一个迭代器event loop 会不断 next 直到 StopIteration # 这里模拟 sleep实际应注册到 loop yield from asyncio.sleep(self.delay).__await__() return My custom awaitable done # 使用 async def test_custom(): result await MyAwaitable(2) print(result) # 输出: My custom awaitable done这个例子揭示了__await__()的真实面目它必须返回一个迭代器通常由yield或yield from构建event loop 会持续next()它直到抛出StopIteration并将value作为await的结果。这是 asyncio 的基石协议所有 awaitable 都必须遵守。4. 实操过程与核心环节实现从单协程到并发调度4.1 第一个真正有意义的 async 程序并发 HTTP 请求理论讲完现在动手。目标并发请求 3 个 URL对比同步 vs 异步耗时。我们不用aiohttp太重用 Python 标准库asyncio自带的asyncio.open_connection模拟 TCP 连接足够体现 I/O 并发本质import asyncio import time # 同步版本串行连接 def sync_connect(host, port80): import socket with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((host, port)) return fConnected to {host} def sync_main(): start time.time() for host in [httpbin.org, httpbin.org, httpbin.org]: # 重复三次 sync_connect(host) print(fSync connect: {time.time() - start:.2f}s) # 异步版本并发连接 async def async_connect(host, port80): # asyncio.open_connection 是真正的 awaitable底层用 selector reader, writer await asyncio.open_connection(host, port) writer.close() await writer.wait_closed() return fConnected to {host} async def async_main(): start time.time() # 创建三个协程对象但不立即执行 tasks [ async_connect(httpbin.org), async_connect(httpbin.org), async_connect(httpbin.org) ] # asyncio.gather 并发执行所有任务返回结果列表 results await asyncio.gather(*tasks) print(fAsync connect: {time.time() - start:.2f}s) return results # 运行 if __name__ __main__: sync_main() asyncio.run(async_main())实测结果本地网络Sync connect: 3.21s 每次连接约1.07s串行叠加Async connect: 1.12s 三次连接几乎同时发起总耗时≈单次为什么快了近3倍因为asyncio.open_connection在底层使用了selectors模块Linux 下是 epollWindows 下是 IOCP。它不阻塞线程而是将 socket 注册到操作系统提供的 I/O 多路复用器上。event loop 只需一次系统调用如epoll_wait()就能同时监听多个 socket 的“可读/可写”事件。当任意一个 socket 连接成功epoll_wait()就返回event loop 立刻唤醒对应的协程。整个过程CPU 几乎不空转全在高效调度。这就是asyncio的力量来源它把操作系统级的 I/O 多路复用能力通过协程和 event loop封装成了 Python 开发者友好的await语法。4.2asyncio.gather()并发的“指挥官”不是“加速器”gather()常被误解为“让代码变快的魔法”。其实它只是并发任务的协调器。它的核心作用有三统一启动接收多个协程对象将它们全部注册到 event loop确保它们“同时”进入就绪队列结果聚合等待所有任务完成按输入顺序返回结果列表错误传播若任一任务抛出异常gather()默认会立即取消其他所有任务并将第一个异常抛出可通过return_exceptionsTrue改为返回异常对象。关键点gather()本身不创造并发它只是把多个 awaitable “打包”交给 event loop。并发的真正功臣是 event loop 对多个 awaitable 的并行监听与调度。你可以不用gather()手动create_task()async def manual_concurrent(): start time.time() # 手动创建任务立即开始执行 task1 asyncio.create_task(async_connect(httpbin.org)) task2 asyncio.create_task(async_connect(httpbin.org)) task3 asyncio.create_task(async_connect(httpbin.org)) # 等待所有任务完成 result1 await task1 result2 await task2 result3 await task3 print(fManual concurrent: {time.time() - start:.2f}s) return [result1, result2, result3]效果和gather()完全一致。create_task()的意义在于它立即将协程调度进 event loop即使你还没await它。这允许你实现“启动即忘”的模式比如启动一个后台日志任务然后继续处理主逻辑。而gather()更适合“我要等这一组都做完再往下走”的场景。选择哪个取决于你的控制流需求而非性能差异。4.3asyncio.create_task()与asyncio.ensure_future()何时用哪个这是新手极易混淆的点。简单说asyncio.create_task(coro)推荐首选。它明确表示“我要把这个协程作为任务Task调度”Task 是asyncio.Task的实例是Future的子类专为协程设计提供了cancel()、done()、get_coro()等丰富方法且是 event loop 的一级公民。asyncio.ensure_future(obj)通用适配器。它接受任何 awaitable协程、Future、自定义 awaitable并确保返回一个 Future 对象。如果obj已经是 Future就直接返回如果是协程则调用create_task()包装。它存在的意义是兼容性比如某些 API 要求你传入一个 Future。实战建议99% 的场景用create_task()。它语义清晰类型安全调试友好。只有当你需要处理一个可能来自第三方库、类型不确定的 awaitable 时才考虑ensure_future()。例如# 你不知道 func() 返回的是协程还是 Future def get_awaitable(): if some_condition: return my_async_func() # 协程 else: return asyncio.Future() # Future # 此时用 ensure_future 保险 awaitable get_awaitable() future asyncio.ensure_future(awaitable) result await future提示永远不要对同一个协程对象多次create_task()。这会创建多个独立任务它们共享协程代码但拥有各自的状态可能导致不可预知的竞态。正确做法是coro my_async_func(); task1 asyncio.create_task(coro); task2 asyncio.create_task(coro)是错的应该task1 asyncio.create_task(my_async_func()); task2 asyncio.create_task(my_async_func())。5. 常见问题与排查技巧实录那些让你抓狂的 Warning 和 Hang5.1 “RuntimeWarning: coroutine xxx was never awaited”最痛的初学者之殇这个警告不是 bug而是 asyncio 的“安全带”。它意味着你创建了一个协程对象却忘了await它导致它被垃圾回收前从未执行。常见场景async def do_something(): await asyncio.sleep(1) return done # ❌ 错误只调用了函数生成了协程对象但没 await result do_something() # Warning! result 是 coroutine object ... print(result) # coroutine object ...不是 done # ✅ 正确必须 await result await do_something() # 在 async 函数内 # 或 result asyncio.run(do_something()) # 在顶层但更隐蔽的是“链式调用”陷阱# ❌ 错误以为 .get() 会自动 await其实不会 import aiohttp async def bad_fetch(): async with aiohttp.ClientSession() as session: response session.get(https://httpbin.org/get) # 这里漏了 await data await response.json() # response 是协程对象不是 Response 对象 return data # ✅ 正确每一层 I/O 都要显式 await async def good_fetch(): async with aiohttp.ClientSession() as session: response await session.get(https://httpbin.org/get) # await get() data await response.json() # await json() return data排查技巧开启 Python 的-Werror模式python -Werror your_script.py让警告变成错误强制你在开发阶段就修复。或者在 IDE 中配置 Pylint 或 Ruff启用asyncio-await规则实时高亮。5.2 “Event loop is closed”循环已关你还想干啥这个错误通常出现在你试图在 event loop 关闭后再调度新任务。最常见于在asyncio.run()结束后又调用asyncio.create_task()在async with上下文管理器退出后其内部的 background task 仍在尝试await多线程环境下从非主线程访问主线程的 loop。典型错误代码import asyncio async def background_job(): while True: await asyncio.sleep(1) print(Working...) async def main(): task asyncio.create_task(background_job()) await asyncio.sleep(3) # 主任务只运行3秒 # 3秒后 main 结束asyncio.run() 会关闭 loop # 但 background_job 仍在后台运行下次 await sleep 时loop 已关 # ❌ 这会报错Event loop is closed asyncio.run(main())解决方案显式取消后台任务。修改main()async def main(): task asyncio.create_task(background_job()) try: await asyncio.sleep(3) finally: task.cancel() # 主动取消 try: await task # 等待取消完成会抛 CancelledError except asyncio.CancelledError: pass # 忽略取消异常注意task.cancel()只是“请求取消”协程需要在下一个await点响应。所以background_job()最好加上取消检查async def background_job(): try: while True: await asyncio.sleep(1) print(Working...) except asyncio.CancelledError: print(Background job cancelled) raise # 重新抛出让 await task 捕获5.3 “Cannot run the event loop while another loop is running”嵌套 loop 的迷思Python 不允许一个线程内同时运行两个 event loop。常见于在 Jupyter Notebook 中IPython 已经启动了一个 loop你又调asyncio.run()在 GUI 应用如 Tkinter中主循环和 asyncio loop 冲突。解决方案分场景Jupyter用await直接运行协程await my_coro()或用nest_asyncio库pip install nest_asyncio; nest_asyncio.apply()打补丁GUI 应用不要用asyncio.run()改用asyncio.get_event_loop().run_until_complete(coro)并确保 loop 与 GUI 主循环集成如asyncio的async_tkinter库生产环境绝对避免嵌套。asyncio.run()是为脚本设计的“一站式”入口一个程序只调用一次。5.4 性能陷阱CPU 密集型操作如何拖垮整个异步系统这是最致命的误区。asyncio 只优化 I/O 等待不优化 CPU 计算。如果你在async def里写async def cpu_bound_task(): # ❌ 危险这会霸占 CPU 100%其他协程完全无法调度 result sum(i * i for i in range(10**7)) return result那么即使你并发启动 10 个这样的任务它们也会串行执行总耗时 ≈ 单个任务 × 10。因为sum()是纯 Python 计算没有awaitevent loop 无法介入调度。正确解法将 CPU 密集型工作移出 event loop。用asyncio.to_thread()Python 3.9或loop.run_in_executor()import asyncio import concurrent.futures def cpu_intensive(n): return sum(i * i for i in range(n)) async def safe_cpu_task(): # 在线程池中执行不阻塞 event loop loop asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor() as pool: result await loop.run_in_executor(pool, cpu_intensive, 10**7) return result # 或 Python 3.9 更简洁 async def safe_cpu_task_v2(): result await asyncio.to_thread(cpu_intensive, 10**7) return result原理run_in_executor()将函数提交给线程池event loop 继续调度其他协程当线程池中的函数执行完毕它会通过回调通知 event looploop 再 resume 对应的协程。这样CPU 计算和 I/O 等待就能真正并行。6. 工具选型与生态定位asyncio 不是万能胶而是精密齿轮6.1asynciovsthreadingvsmultiprocessing场景决策树选择异步不是技术优越感而是精准匹配。下表帮你快速决策场景特征推荐方案原因大量 I/O 等待HTTP 请求、DB 查询、文件读写并发数 100asyncio单线程高并发内存占用低上下文切换开销极小。10K 连接只需几 MB 内存。少量 I/O 中等 CPU 计算并发数 10threadingGIL 在 I/O 时释放线程可并行等待CPU 计算虽受 GIL 限制但线程数少开销可控。纯 CPU 密集型计算图像处理、数值计算multiprocessing绕过 GIL真正利用多核。asyncio 和 threading 在此场景均无效。需要与遗留同步代码深度集成threading或multiprocessingasyncio 生态aiohttp, asyncpg需全栈异步改造成本高混合编程同步调用异步极其复杂。实操心得我在一个监控告警系统中做过对比测试。系统需每秒轮询 500 个 API 端点。用threading500 个线程内存峰值 1.2GBCPU 占用 45%频繁 GC用asyncioaiohttp内存峰值 80MBCPU 占用 12%吞吐稳定。但当加入一个实时日志分析模块需解析 GB 级日志我立刻切回multiprocessing用 4 个进程并行处理asyncio主循环只负责分发任务和收集结果。没有银弹只有组合拳。6.2aiohttp、httpx、asyncpg异步生态的“三驾马车”asyncio是引擎这些库是车轮。选型关键看成熟度和场景HTTP 客户端aiohttp最老牌功能全Client/Server但 API 略重学习曲线稍陡。httpx现代替代品API 与 requests 高度相似同步/异步双模式推荐新项目首选。数据库asyncpgPostgreSQL 专用性能顶尖C 语言驱动但只支持 PG。aiomysql/aiosqliteMySQL/SQLite 的异步驱动成熟度稍逊于asyncpg。通用工具asyncio标准库已足够强大asyncio.subprocess异步进程、asyncio.queues协程安全队列、asyncio.locks协程锁。注意永远不要在async def中直接调用requests.get()。这是同步阻塞调用会瞬间冻结整个 event loop。必须用aiohttp或httpx.AsyncClient。6.3 调试利器asyncio的隐藏诊断开关asyncio内置了强大的调试模式开启后能暴露所有潜在问题import asyncio # 启用调试模式开发环境强烈建议 asyncio.run(main(), debugTrue) # 或全局设置 import os os.environ[PYTHONASYNCIODEBUG] 1开启后你会看到所有未被await的协程对象会触发ResourceWarning任务执行时间超过 100ms会打印Executing Task took 0.234 secondsFuture对象未被await或result()会警告事件循环关闭时仍有 pending tasks会详细列出。这相当于给你的异步程序装上了“行车记录仪”所有不规范操作无所遁形。上线前务必关闭debugFalse避免性能损耗。7. 实战扩展从入门到可落地的工程化实践7.1 构建一个健壮的异步爬虫骨架学完原理我们整合成一个可运行的最小可行产品MVP。目标并发抓取多个网页标题带错误处理、超时、重试import asyncio import aiohttp from typing import List, Tuple, Optional async def fetch_title(session: aiohttp.ClientSession, url: str) - Tuple[str, Optional[str]]: 获取单个 URL 的 title返回 (url, title) 或 (url, None) try: async with session.get(url, timeout5) as response: if response.status 200: html await response.text() # 简单提取 title生产环境用 BeautifulSoup start html.find(title) end html.find(/title) if start ! -1 and end ! -1: title html[start7:end].strip() return url, title return url, None except Exception as e: return url, None async def crawl_urls(urls: List[str], max_concurrent: int 10) - List[Tuple[str, Optional[str]]]: 并发爬取 URL 列表 # 限制并发数避免压垮目标服务器 semaphore asyncio.Semaphore(max_concurrent) async def bounded_fetch(url: str): async with semaphore: # 获取信号量控制并发 return await fetch_title(session, url) # 创建所有任务 async with aiohttp.ClientSession() as session: tasks [bounded_fetch(url) for url in urls] results await asyncio.gather(*tasks, return_exceptionsTrue) return results # 使用示例 if __name__ __main__: urls [ https://httpbin.org/html, https://httpbin.org/delay/2, https://httpbin.org/status/500, https://nonexistent-domain-12345.com ] results asyncio.run(crawl_urls(urls)) for url, title in results: print(f{url} - {title or ERROR})这个骨架包含了工程化必备要素并发控制asyncio.Semaphore限制最大连接数保护自身和目标服务超时处理session.get(timeout5)防止单个请求无限 hang错误隔离return_exceptionsTrue确保一个失败不影响其他资源管理async with aiohttp.ClientSession()自动关闭连接池。7.2 异步与同步代码的“和平共处”策略现实世界没有纯异步。你可能需要在异步 Web 服务FastAPI中调用一个同步的机器学习模型scikit-learn在同步脚本中临时调用一个异步数据库查询。两种方案方案1同步调用异步不推荐仅应急import asyncio def sync_call_async(coro): 在同步函数中调用协程会阻塞当前线程 return asyncio.run(coro) # 使用 result sync_call_async(my_async_db_query())⚠️ 风险asyncio.run()会创建新 loop若在已有 loop 的环境中如 FastAPI会报错。仅限顶层脚本。方案2异步调用同步推荐import asyncio import concurrent.futures def sync_ml_model(data): # 模拟耗时的同步计算 import time time.sleep(2) return fPrediction for {data} async def async_ml_predict(data): loop asyncio.get_running_loop() # 在线程池中执行同步函数 with concurrent.futures.ThreadPoolExecutor() as pool: result await loop.run_in_executor(pool, sync_ml_model, data) return result # 在 FastAPI 路由中 app.get(/predict) async def predict(item: