首先需要清楚什么是同步和异步同步调用通俗例子你去奶茶店点单1. 你下单站在柜台原地不动等待2. 店员做完你的奶茶交到你手上你才能走、做别的事整个过程你全程阻塞不能中途刷手机、买小吃必须等任务完成。程序逻辑发起调用后代码停住卡住必须等被调用的方法/接口完全执行完毕、返回结果才会继续往下执行后面代码。特点1. 顺序执行逻辑简单好写2. 阻塞等待浪费时间效率低3. 实时拿到返回结果简易代码示意同步同步调用喝水必须等喝水结束才能吃饭排队等做完一件才能下一件def drink(): print(喝水中...) time.sleep(3) # 模拟耗时 3 秒 print(喝完水了) def eat(): drink() # 同步调用卡住 3 秒 print(开始吃饭) 执行顺序喝水 → 等 3 秒 → 吃饭异步调用通俗例子还是奶茶店1. 你下单拿号不用站在原地等去旁边逛超市、刷手机2. 奶茶做好后喇叭喊你的号/短信通知你你再过去取等待的时间你可以并行做其他事情不会卡住。程序逻辑发起调用后代码不会阻塞立刻执行后面的逻辑耗时任务后台偷偷跑等任务完成后通过回调/通知拿到结果。特点1. 不阻塞主线程多个任务同时跑效率高2. 逻辑相对复杂需要处理回调、结果接收3. 无法立刻拿到结果要等通知简易代码示意异步异步调用喝水不用等待直接去吃饭不用等先干别的完成后通知你async def drink(): print(后台开始喝水) await asyncio.sleep(3) print(喝完水了通知你) async def eat(): task asyncio.create_task(drink()) # 异步启动任务不阻塞 print(我先吃饭不等喝水) await task # 后续需要时再等结果 执行顺序先打印“我先吃饭”3 秒后打印喝完水核心对比总结1. 同步调用者阻塞等待任务串行执行即时返回结果2. 异步调用者不阻塞任务并行执行完成后回调获取结果开发常见场景同步简单查询、本地小型计算短耗时操作异步网络请求、文件上传下载、接口调用、定时任务耗时操作防止页面卡死基于此模型调用包含同步调用和异步调用Messages - Docs by LangChain普通调用invoke处理单条输入等待 LLM 完全推理完成后再返回调用结果# 构建消息列表 messages [ SystemMessage(content你是一个法律助手只回答法律问题超出范围的统一回答非法律问题无可奉告), # HumanMessage(content简单介绍下广告法一句话告知 50 字以内) HumanMessage(content23 等于几?) ] 3. 调用模型 response model.invoke(messages) # ainvoke print(f响应类型{type(response)}) 打印结果 print(response.content) print(response.content_blocks)ainvokeLangChain 提供 ainvoke() 异步调用接口用于在异步环境async/await中高效并行地执行模型推理。它的核心作用是让你同时调用多个模型请求而不阻塞主线程——特别适合大批量请求或 Web 服务场景如 FastAPIasync def main(): # 异步调用一条请求 response await model.ainvoke(解释一下 LangChain 是什么简洁回答 100 字以内) print(f响应类型{type(response)}) print(response.content_blocks) 运行异步函数 if name main: asyncio.run(main())流式调用stream流式响应是一种逐步返回大模型生成结果的技术生成一点返回一点允许服务器将响应内容分批次实时传输给客户端而不是等待全部内容生成完毕后再一次性返回# 构建消息列表 messages [ SystemMessage(content你叫小问是一个乐于助人的 AI 人工助手), HumanMessage(content你是谁) ] 流式调用大模型 response model.stream(messages) print(f响应类型{type(response)}) 流式打印结果 for chunk in response: # 刷新缓冲区 (无换行符缓冲区未刷新内容可能不会立即显示) print(chunk.content, end, flushTrue) print(\n)涉及 Python 中的 generator 生成器需要掌握astream异步流式响应# 异步流式调用大模型定义异步函数 async def async_stream_call(): # astream 返回异步生成器无需 await 修饰直接赋值 response model.astream(messages) print(f响应类型{type(response)}) # 响应类型class async_generator # 异步遍历异步生成器必须使用 async for不可用普通 for async for chunk in response: # 刷新缓冲区实现流式打印无换行、即时输出 print(chunk.content, end, flushTrue) print(\n)stream 和 astream 只是异步同步方式不同运行结果是相同的批处理batch处理批量输入一次性向模型提交多个输入并并行处理从而显著提升吞吐量# 问题列表 questions [ 什么是 Redis? 简洁回答字数控制在 100 以内, Python 的生成器是做什么的简洁回答字数控制在 100 以内, 解释一下 Docker 和 Kubernetes 的关系? 简洁回答字数控制在 100 以内 ] 批量调用大模型 model.batch() response model.batch(questions) print(f响应类型{type(response)}) print() for q, r in zip(questions, response): print(f问题{q}\n回答{r.content}\n)abatch异步处理批量输入# 异步批量调用大模型定义异步函数封装异步操作 # abatch() 是异步方法需要基于 async/await 语法构建异步程序并用 asyncio 驱动运行 async def async_batch_call(): # 调用 model.abatch() 异步批量处理请求需用 await 修饰关键 response await model.abatch(questions) print(f响应类型{type(response)}) # 遍历结果并格式化输出与原来的同步版本格式一致 for q, r in zip(questions, response): print(f问题{q}\n回答{r.content}\n) 运行异步函数 if name main: asyncio.run(async_batch_call())总结invoke/ainvoke将单个输入转换为输出batch/abatch批量将多个输入转换为输出stream/astream从单个输入生成流式输出带有“a”前缀的方法是异步的需要与 asyncio 和 await 语法一起使用以实现并发