Python异步编程实践：从asyncio到LangChain调用链

Python异步编程实践从asyncio到LangChain调用链摘要本文记录Python异步编程的学习过程与实践结论以「心屿树洞」AI心理陪伴项目为应用载体覆盖以下内容异步IO的动机与适用场景asyncio核心用法async/await、create_task、gather典型错误分析同步阻塞调用、串行等待、库兼容性项目实战对比同步requestsvs 异步httpx批量接口压测与LangChain生态的衔接invoke()与ainvoke()的选型依据一、背景与动机在基于LangChain开发「心屿树洞」项目时官方文档示例大量采用异步接口fromlangchain_openaiimportChatOpenAI modelChatOpenAI()asyncdefmain():resultawaitmodel.ainvoke(你好)# 异步调用若改用同步invoke()执行批量压力测试10用户并发性能存在明显瓶颈模式单次LLM耗时10用户总耗时结论同步invoke()约2s约20s线性累积无法支持多用户并发异步ainvoke()约2s约2-3s并发取最大值满足生产级并发要求因此需系统掌握Python异步IO机制作为后续Agent服务的技术基础。二、核心机制同步 vs 异步IO2.1 执行模型差异同步调用阻塞式线程发起IO请求后阻塞等待完成后再继续后续逻辑总耗时 Σ 各任务耗时适用场景脚本、单任务、对并发无要求的工具类程序异步调用非阻塞式线程发起IO请求后挂起当前协程转去执行其他就绪协程事件循环Event Loop统一调度所有协程总耗时 max(各任务耗时)适用场景网络IO密集型服务LLM调用、HTTP请求、数据库查询2.2 典型错误1阻塞调用未异步化错误示例在async函数中使用同步阻塞APIimportasyncioimporttimeasyncdeffetch_llm(prompt):time.sleep(2)# ❌ 同步阻塞async/await无法调度returnf【回复】{prompt}asyncdefmain():starttime.time()r1awaitfetch_llm(你好)r2awaitfetch_llm(今天天气怎么样)print(f耗时{time.time()-start:.2f}s)# 输出 4.00s串行asyncio.run(main())问题根因time.sleep()为线程级同步阻塞事件循环无法将其调度出去。三、asyncio 标准用法3.1 基础修正异步等待 任务调度正确示例使用 asyncio.sleep create_task 实现并发importasyncioimporttimeasyncdeffetch_llm(prompt):awaitasyncio.sleep(2)# ✅ 异步挂起事件循环可调度其他协程returnf【回复】{prompt}asyncdefmain():starttime.time()# 先创建任务立即进入事件循环调度队列t1asyncio.create_task(fetch_llm(你好))t2asyncio.create_task(fetch_llm(今天天气怎么样))# 再等待完成r1,r2awaitt1,awaitt2print(r1,r2,sep\n)print(f耗时{time.time()-start:.2f}s)# 输出 2.01s并发asyncio.run(main())3.2 批量并发asyncio.gather当任务数较多时使用gather一次性编排并发任务批量并发示例10个LLM请求同时发出importasyncioimporttimeasyncdeffetch_llm(prompt):awaitasyncio.sleep(2)returnf【回复】{prompt}asyncdefmain():starttime.time()prompts[f第{i}条消息foriinrange(10)]tasks[fetch_llm(p)forpinprompts]resultsawaitasyncio.gather(*tasks)# gather自动收集所有结果forrinresults:print(r)print(f10条消息耗时{time.time()-start:.2f}s)# 输出 2.00sasyncio.run(main())3.3 常见易错点汇总易错场景错误写法正确写法协程未执行hello()直接调用await hello()或asyncio.run(hello())嵌套事件循环在async函数内调用asyncio.run()直接await目标协程HTTP库选型requests.get()同步httpx.AsyncClient().get()异步文件IOopen().read()大文件阻塞aiofiles.open().read()数据库驱动psycopg2/sqlite3同步asyncpg/aiosqlite异步判断原则在async函数内所有IO密集操作前必须有await关键字否则需替换为异步实现。四、项目实战心屿树洞批量接口压测对比4.1 测试目标模拟10个用户同时调用「心屿树洞」聊天接口对比同步与异步实现的端到端耗时。接口信息创建会话POST /api/session/create发送消息POST /api/chat/sendMessage4.2 同步实现requests同步版压测脚本importrequestsimporttime BASEhttp://localhost:3000/apidefsend(session_id,content):rrequests.post(f{BASE}/chat/sendMessage,json{session_id:session_id,content:content},timeout30)returnr.json()defbench():starttime.time()foriinrange(10):requests.post(f{BASE}/session/create,json{emotion_tag:anxiety})respsend(fuser_{i},我最近学习压力好大睡不着觉)replyresp.get(data,{}).get(reply,)[:30]print(f用户{i}:{reply}...)print(f同步总耗时{time.time()-start:.2f}s)if__name____main__:bench()实测结果真实LLM调用10用户总耗时56.82sQPS ≈ 0.18。4.3 异步实现httpx asyncio异步版压测脚本 依赖pip install httpx importasyncioimporthttpximporttime BASEhttp://localhost:3000/apiasyncdefsend(client,session_id,content):rawaitclient.post(f{BASE}/chat/sendMessage,json{session_id:session_id,content:content},timeout30.0)returnr.json()asyncdefbench_async():starttime.time()asyncwithhttpx.AsyncClient(timeout60.0)asclient:# 并发创建会话create_tasks[client.post(f{BASE}/session/create,json{emotion_tag:anxiety})for_inrange(10)]awaitasyncio.gather(*create_tasks)# 并发发送消息chat_tasks[send(client,fuser_{i},我最近学习压力好大睡不着觉)foriinrange(10)]resultsawaitasyncio.gather(*chat_tasks)fori,respinenumerate(results):replyresp.get(data,{}).get(reply,)[:30]print(f用户{i}:{reply}...)print(f异步总耗时{time.time()-start:.2f}s)if__name____main__:asyncio.run(bench_async())实测结果真实LLM调用10用户总耗时8.73sQPS ≈ 1.15。4.4 性能对比汇总指标同步requests异步httpx asyncio提升幅度10用户总耗时56.82s8.73s↓ 84.6%并发度1串行10并发—适用性脚本、单用户调试生产服务、批量任务—五、与LangChain生态的衔接LangChain 0.3.x 版本所有核心组件均提供同步与异步双接口方法类型适用场景model.invoke()同步阻塞脚本、CLI工具、单用户原型model.ainvoke()异步非阻塞FastAPI服务、批量任务、多用户并发chain.invoke()/chain.ainvoke()同上链式调用Prompt → Model → Parseragent.invoke()/agent.ainvoke()同上Agent Tool 调用场景选型建议「心屿树洞」后续Python Agent后端将统一采用FastAPI LangChain ainvoke系列架构以满足多用户并发要求。对于Tool装饰器tool函数实现内部需使用异步版本SDK如异步HTTP、异步向量库。六、后续研究方向当前阶段已掌握基础并发能力后续结合实际需求深入以下内容并发控制asyncio.Semaphore限制LLM并发数避免触发API限流状态持久化LangGraph Checkpointer 与异步存储后端事件循环原理asyncio内部调度机制、协程状态机模型并发模型对比协程asynciovs 多线程ThreadPoolExecutorvs 多进程ProcessPoolExecutor的选型边界七、参考资料Python官方文档asyncio— Asynchronous I/Ohttpx 官方文档Async Client QuickStartLangChain 官方文档Async API / LCEL Expression Language##以下是github仓库https://github.com/kkk418/python-async-learn标签#Python异步编程#asyncio#httpx#LangChain#心屿树洞项目实战