在現代Python開發中，異步編程已經成為提高程序性能的重要手段，特別是在處理網絡請求、數據庫操作或AI模型調用等耗時操作時。本文將通過實際的LangGraph 示例，深入解析async的真正作用，并揭示一個常見誤區：為什么異步順序執行與同步執行時間相近？

async的核心作用

async的主要價值在于創建異步編程環境，讓程序在等待耗時操作時不被阻塞，從而提高執行效率。但是，很多開發者對異步編程存在一個根本性的誤解。

常見誤區：async ≠ 自動加速

許多人認為只要在函數前加上async，程序就會自動變快。這是錯誤的！

讓我們通過一個LangGraph的實際例子來說明：

from langchain.chat_models import init_chat_model
from langgraph.graph import MessagesState, StateGraphfrom dotenv import load_dotenv
load_dotenv()
# 初始化 LLM 模型
llm = ChatDeepSeek(model="deepseek-chat")# 異步節點定義
async def async_node(state: MessagesState): new_message = await llm.ainvoke(state["messages"]) return {"messages": [new_message]}builder = StateGraph(MessagesState).add_node(async_node).set_entry_point("node")
graph = builder.compile()

完整的性能對比示例

以下是一個可以完整運行的性能測試示例：

import asyncio
import time
from langchain.chat_models import init_chat_model
from langgraph.graph import MessagesState, StateGraphfrom dotenv import load_dotenv
load_dotenv()
# 初始化 LLM 模型
llm = ChatDeepSeek(model="deepseek-chat")# 同步版本的節點
def sync_node(state: MessagesState):"""同步版本：會阻塞等待"""new_message = llm.invoke(state["messages"])return {"messages": [new_message]}# 異步版本的節點
async def async_node(state: MessagesState):"""異步版本：可以并發執行"""new_message = await llm.ainvoke(state["messages"])return {"messages": [new_message]}# 創建同步圖
sync_builder = StateGraph(MessagesState).add_node("node", sync_node).set_entry_point("node")
sync_graph = sync_builder.compile()# 創建異步圖
async_builder = StateGraph(MessagesState).add_node("node", async_node).set_entry_point("node")
async_graph = async_builder.compile()# 測試消息
messages = [{"role": "user", "content": "你好，請介紹一下自己"},{"role": "user", "content": "請解釋一下什么是人工智能"},{"role": "user", "content": "給我講個笑話吧"},{"role": "user", "content": "請推薦幾本好書"},{"role": "user", "content": "今天天氣怎么樣？"}
]def test_sync_sequential():"""測試同步順序執行"""print("同步順序執行測試...")start_time = time.time()results = []for i, msg in enumerate(messages):print(f"  處理消息 {i+1}/{len(messages)}...")result = sync_graph.invoke({"messages": [msg]})results.append(result)end_time = time.time()duration = end_time - start_timeprint(f"同步執行完成，總耗時: {duration:.2f} 秒")return results, durationasync def test_async_sequential():"""測試異步順序執行"""print("異步順序執行測試...")start_time = time.time()results = []for i, msg in enumerate(messages):print(f"  處理消息 {i+1}/{len(messages)}...")result = await async_graph.ainvoke({"messages": [msg]})results.append(result)end_time = time.time()duration = end_time - start_timeprint(f"異步順序執行完成，總耗時: {duration:.2f} 秒")return results, durationasync def test_async_concurrent():"""測試異步并發執行"""print("異步并發執行測試...")start_time = time.time()# 創建所有任務tasks = []for i, msg in enumerate(messages):print(f"  啟動任務 {i+1}/{len(messages)}...")task = async_graph.ainvoke({"messages": [msg]})tasks.append(task)# 并發執行所有任務print("  所有任務并發運行中...")results = await asyncio.gather(*tasks)end_time = time.time()duration = end_time - start_timeprint(f"異步并發執行完成，總耗時: {duration:.2f} 秒")return results, durationasync def main():"""主函數：運行所有測試"""print("=" * 60)print("LangGraph 異步 vs 同步性能測試")print("=" * 60)print(f"測試場景：處理 {len(messages)} 個 LLM 請求")print()# 1. 同步順序執行sync_results, sync_time = test_sync_sequential()print()# 2. 異步順序執行async_seq_results, async_seq_time = await test_async_sequential()print()# 3. 異步并發執行async_con_results, async_con_time = await test_async_concurrent()print()# 性能對比分析print("=" * 60)print("性能對比分析")print("=" * 60)print(f"同步順序執行:     {sync_time:.2f} 秒")print(f"異步順序執行:     {async_seq_time:.2f} 秒")print(f"異步并發執行:     {async_con_time:.2f} 秒")print()# 計算性能提升if async_con_time > 0:speedup_vs_sync = sync_time / async_con_timespeedup_vs_async_seq = async_seq_time / async_con_timeprint("性能提升:")print(f"異步并發 vs 同步順序: {speedup_vs_sync:.1f}x 倍速提升")print(f"異步并發 vs 異步順序: {speedup_vs_async_seq:.1f}x 倍速提升")# 運行測試
if __name__ == "__main__":asyncio.run(main())

三種執行方式的性能對比

1. 同步順序執行

def test_sync_sequential():results = []for msg in messages:result = sync_graph.invoke({"messages": [msg]})results.append(result)return results# 執行時間線：
# [請求1---等待---響應1] [請求2---等待---響應2] [請求3---等待---響應3] ...
# 總耗時：約 10-15 秒（5個請求 × 每個2-3秒）

2. 異步順序執行

async def test_async_sequential():results = []for msg in messages:result = await async_graph.ainvoke({"messages": [msg]})  # 還是逐個等待results.append(result)return results# 執行時間線：
# [請求1---等待---響應1] [請求2---等待---響應2] [請求3---等待---響應3] ...
# 總耗時：約 10-15 秒（與同步執行相近）

3. 異步并發執行

async def test_async_concurrent():# 關鍵：同時啟動所有任務tasks = [async_graph.ainvoke({"messages": [msg]}) for msg in messages]# 并發執行所有任務results = await asyncio.gather(*tasks)return results# 執行時間線：
# [請求1---等待---響應1]
# [請求2---等待---響應2]  ← 同時進行
# [請求3---等待---響應3]  ← 同時進行
# [請求4---等待---響應4]  ← 同時進行
# [請求5---等待---響應5]  ← 同時進行
# 總耗時：約 2-3 秒（接近單個請求時間）

為什么異步順序執行時間相近？

這個現象困惑了很多開發者。讓我們深入分析原因：

控制權的概念

在異步編程中，控制權指的是CPU當前正在執行哪段代碼的決定權。

同步執行中的控制權

def sync_function():print("開始")result = llm.invoke(messages)  # CPU 在這里"卡住"等待print("結束")return result# 執行流程：
# 1. CPU 執行 print("開始")
# 2. CPU 調用 llm.invoke()
# 3. CPU 完全停止，等待網絡響應（2-3秒）
# 4. 收到響應后，CPU 繼續執行 print("結束")

在步驟3中，CPU被完全占用但什么都不做，這就是"阻塞"。

異步執行中的控制權轉移

async def async_function():print("開始")result = await llm.ainvoke(messages)  # 讓出控制權print("結束")return result# 執行流程：
# 1. CPU 執行 print("開始")
# 2. CPU 調用 llm.ainvoke()
# 3. 遇到 await，CPU 說："我先去做別的事，響應來了再叫我"
# 4. CPU 可以執行其他任務
# 5. 網絡響應到達，CPU 重新獲得控制權
# 6. CPU 繼續執行 print("結束")

關鍵洞察：讓出控制權 ≠ 時間節省

# 異步但沒有性能提升（順序執行）
for msg in messages:result = await process_message(msg)  # 還是一個接一個等待# 異步真正的優勢（并發執行）
tasks = [process_message(msg) for msg in messages]
results = await asyncio.gather(*tasks)  # 同時處理所有

異步順序執行時間相近的原因：

1. 都是順序執行：兩種方式都是"處理完第一個請求，再處理第二個"
2. 等待時間相同：每個LLM調用的網絡延遲和處理時間是一樣的
3. 沒有并發優勢：異步順序執行沒有利用異步的核心優勢——并發

實際運行和測試

將上述代碼保存為 async_test.py，運行后會看到類似輸出：

============================================================
🧪 LangGraph 異步 vs 同步性能測試
============================================================
📝 測試場景：處理 5 個 LLM 請求🔄 同步順序執行測試...處理消息 1/5...處理消息 2/5...處理消息 3/5...處理消息 4/5...處理消息 5/5...
? 同步執行完成，總耗時: 148.37 秒? 異步順序執行測試...處理消息 1/5...處理消息 2/5...處理消息 3/5...處理消息 4/5...處理消息 5/5...
? 異步順序執行完成，總耗時: 147.72 秒🚀 異步并發執行測試...啟動任務 1/5...啟動任務 2/5...啟動任務 3/5...啟動任務 4/5...啟動任務 5/5...🔥 所有任務并發運行中...
? 異步執行完成，總耗時: 67.24 秒============================================================
📊 性能對比分析
============================================================
同步順序執行:     148.37 秒
異步順序執行:     147.72 秒
異步并發執行:     67.24 秒🎯 性能提升:
異步并發 vs 同步順序: 2.2x 倍速提升
異步并發 vs 異步順序: 2.2x 倍速提升💡 關鍵發現:
? 異步并發執行可以顯著減少總耗時
? 當有多個獨立的 LLM 調用時，并發執行效果最明顯
? 異步順序執行與同步執行時間相近（都是逐個等待）
? 實際加速比取決于網絡延遲和 LLM 響應時間

實際應用指導

何時使用異步？

適合使用異步的場景：

• 多個獨立的網絡請求（如批量API調用）
• 并發的數據庫查詢
• 同時處理多個用戶請求
• I/O密集型任務

不適合使用異步的場景：

• CPU密集型計算
• 必須順序執行的依賴任務
• 簡單的單次操作

最佳實踐

# 錯誤用法：異步但無性能提升
async def bad_example():result1 = await api_call_1()result2 = await api_call_2()  # 依賴result1result3 = await api_call_3()  # 依賴result2return [result1, result2, result3]# 改進：部分并發
async def better_example():# 可以并發的部分task1 = api_call_1()task2 = independent_api_call()result1, result2 = await asyncio.gather(task1, task2)# 依賴前面結果的部分result3 = await api_call_3(result1)return [result1, result2, result3]# 最佳：完全并發（當任務獨立時）
async def best_example():tasks = [api_call_1(),api_call_2(),api_call_3(),api_call_4(),api_call_5()]results = await asyncio.gather(*tasks)return results

總結

1. async的真正價值：不在于讓單個任務變快，而在于讓多個任務可以同時進行
2. 異步順序執行時間相近：因為還是逐個等待，沒有發揮并發優勢
3. 性能提升的關鍵：使用asyncio.gather()或類似機制實現真正的并發
4. 實際應用：在設計異步程序時，要識別哪些任務可以并發執行

異步編程是一個強大的工具，但只有正確使用才能發揮其真正的威力。記住：異步的魅力不在于等待得更快，而在于可以同時等待多件事情。

延伸思考:在你的項目中，有哪些場景可以從順序執行改為并發執行？試著識別那些相互獨立的異步操作，這通常是性能優化的黃金機會。