首先需要清楚什么是同步和異步同步調(diào)用通俗例子你去奶茶店點單1. 你下單站在柜臺原地不動等待2. 店員做完你的奶茶交到你手上你才能走、做別的事整個過程你全程阻塞不能中途刷手機、買小吃必須等任務(wù)完成。程序邏輯發(fā)起調(diào)用后代碼停住卡住必須等被調(diào)用的方法/接口完全執(zhí)行完畢、返回結(jié)果才會繼續(xù)往下執(zhí)行后面代碼。特點1. 順序執(zhí)行邏輯簡單好寫2. 阻塞等待浪費時間效率低3. 實時拿到返回結(jié)果簡易代碼示意同步同步調(diào)用喝水必須等喝水結(jié)束才能吃飯排隊等做完一件才能下一件def drink(): print(喝水中...) time.sleep(3) # 模擬耗時 3 秒 print(喝完水了) def eat(): drink() # 同步調(diào)用卡住 3 秒 print(開始吃飯) 執(zhí)行順序喝水 → 等 3 秒 → 吃飯異步調(diào)用通俗例子還是奶茶店1. 你下單拿號不用站在原地等去旁邊逛超市、刷手機2. 奶茶做好后喇叭喊你的號/短信通知你你再過去取等待的時間你可以并行做其他事情不會卡住。程序邏輯發(fā)起調(diào)用后代碼不會阻塞立刻執(zhí)行后面的邏輯耗時任務(wù)后臺偷偷跑等任務(wù)完成后通過回調(diào)/通知拿到結(jié)果。特點1. 不阻塞主線程多個任務(wù)同時跑效率高2. 邏輯相對復(fù)雜需要處理回調(diào)、結(jié)果接收3. 無法立刻拿到結(jié)果要等通知簡易代碼示意異步異步調(diào)用喝水不用等待直接去吃飯不用等先干別的完成后通知你async def drink(): print(后臺開始喝水) await asyncio.sleep(3) print(喝完水了通知你) async def eat(): task asyncio.create_task(drink()) # 異步啟動任務(wù)不阻塞 print(我先吃飯不等喝水) await task # 后續(xù)需要時再等結(jié)果 執(zhí)行順序先打印“我先吃飯”3 秒后打印喝完水核心對比總結(jié)1. 同步調(diào)用者阻塞等待任務(wù)串行執(zhí)行即時返回結(jié)果2. 異步調(diào)用者不阻塞任務(wù)并行執(zhí)行完成后回調(diào)獲取結(jié)果開發(fā)常見場景同步簡單查詢、本地小型計算短耗時操作異步網(wǎng)絡(luò)請求、文件上傳下載、接口調(diào)用、定時任務(wù)耗時操作防止頁面卡死基于此模型調(diào)用包含同步調(diào)用和異步調(diào)用Messages - Docs by LangChain普通調(diào)用invoke處理單條輸入等待 LLM 完全推理完成后再返回調(diào)用結(jié)果# 構(gòu)建消息列表 messages [ SystemMessage(content你是一個法律助手只回答法律問題超出范圍的統(tǒng)一回答非法律問題無可奉告), # HumanMessage(content簡單介紹下廣告法一句話告知 50 字以內(nèi)) HumanMessage(content23 等于幾?) ] 3. 調(diào)用模型 response model.invoke(messages) # ainvoke print(f響應(yīng)類型{type(response)}) 打印結(jié)果 print(response.content) print(response.content_blocks)ainvokeLangChain 提供 ainvoke() 異步調(diào)用接口用于在異步環(huán)境async/await中高效并行地執(zhí)行模型推理。它的核心作用是讓你同時調(diào)用多個模型請求而不阻塞主線程——特別適合大批量請求或 Web 服務(wù)場景如 FastAPIasync def main(): # 異步調(diào)用一條請求 response await model.ainvoke(解釋一下 LangChain 是什么簡潔回答 100 字以內(nèi)) print(f響應(yīng)類型{type(response)}) print(response.content_blocks) 運行異步函數(shù) if name main: asyncio.run(main())流式調(diào)用stream流式響應(yīng)是一種逐步返回大模型生成結(jié)果的技術(shù)生成一點返回一點允許服務(wù)器將響應(yīng)內(nèi)容分批次實時傳輸給客戶端而不是等待全部內(nèi)容生成完畢后再一次性返回# 構(gòu)建消息列表 messages [ SystemMessage(content你叫小問是一個樂于助人的 AI 人工助手), HumanMessage(content你是誰) ] 流式調(diào)用大模型 response model.stream(messages) print(f響應(yīng)類型{type(response)}) 流式打印結(jié)果 for chunk in response: # 刷新緩沖區(qū) (無換行符緩沖區(qū)未刷新內(nèi)容可能不會立即顯示) print(chunk.content, end, flushTrue) print(\n)涉及 Python 中的 generator 生成器需要掌握astream異步流式響應(yīng)# 異步流式調(diào)用大模型定義異步函數(shù) async def async_stream_call(): # astream 返回異步生成器無需 await 修飾直接賦值 response model.astream(messages) print(f響應(yīng)類型{type(response)}) # 響應(yīng)類型class async_generator # 異步遍歷異步生成器必須使用 async for不可用普通 for async for chunk in response: # 刷新緩沖區(qū)實現(xiàn)流式打印無換行、即時輸出 print(chunk.content, end, flushTrue) print(\n)stream 和 astream 只是異步同步方式不同運行結(jié)果是相同的批處理batch處理批量輸入一次性向模型提交多個輸入并并行處理從而顯著提升吞吐量# 問題列表 questions [ 什么是 Redis? 簡潔回答字數(shù)控制在 100 以內(nèi), Python 的生成器是做什么的簡潔回答字數(shù)控制在 100 以內(nèi), 解釋一下 Docker 和 Kubernetes 的關(guān)系? 簡潔回答字數(shù)控制在 100 以內(nèi) ] 批量調(diào)用大模型 model.batch() response model.batch(questions) print(f響應(yīng)類型{type(response)}) print() for q, r in zip(questions, response): print(f問題{q}\n回答{r.content}\n)abatch異步處理批量輸入# 異步批量調(diào)用大模型定義異步函數(shù)封裝異步操作 # abatch() 是異步方法需要基于 async/await 語法構(gòu)建異步程序并用 asyncio 驅(qū)動運行 async def async_batch_call(): # 調(diào)用 model.abatch() 異步批量處理請求需用 await 修飾關(guān)鍵 response await model.abatch(questions) print(f響應(yīng)類型{type(response)}) # 遍歷結(jié)果并格式化輸出與原來的同步版本格式一致 for q, r in zip(questions, response): print(f問題{q}\n回答{r.content}\n) 運行異步函數(shù) if name main: asyncio.run(async_batch_call())總結(jié)invoke/ainvoke將單個輸入轉(zhuǎn)換為輸出batch/abatch批量將多個輸入轉(zhuǎn)換為輸出stream/astream從單個輸入生成流式輸出帶有“a”前綴的方法是異步的需要與 asyncio 和 await 語法一起使用以實現(xiàn)并發(fā)