在現代Python開發中，異步編程已經成為提高程序性能的重要手段，特別是在處理網絡請求、數據庫操作或AI模型調用等耗時操作時。本文將通過實際的LangGraph 示例，深入解析async的真正作用，並揭示一個常見誤區：為什麼異步順序執行與同步執行時間相近？
async的核心作用
async的主要價值在於創建異步編程環境，讓程序在等待耗時操作時不被阻塞，從而提高執行效率。但是，很多開發者對異步編程存在一個根本性的誤解。
常見誤區：async ≠ 自動加速
許多人認為只要在函數前加上async，程序就會自動變快。這是錯誤的！
讓我們通過一個LangGraph的實際例子來説明：
from langchain.chat_models import init_chat_model
from langgraph.graph import MessagesState, StateGraph
from 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, StateGraph
from 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_time
print(f"同步執行完成，總耗時: {duration:.2f} 秒")
return results, duration
async 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_time
print(f"異步順序執行完成，總耗時: {duration:.2f} 秒")
return results, duration
async 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_time
print(f"異步併發執行完成，總耗時: {duration:.2f} 秒")
return results, duration
async 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_time
speedup_vs_async_seq = async_seq_time / async_con_time
print("性能提升:")
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) # 同時處理所有異步順序執行時間相近的原因：
- 都是順序執行：兩種方式都是"處理完第一個請求，再處理第二個"
- 等待時間相同：每個LLM調用的網絡延遲和處理時間是一樣的
- 沒有併發優勢：異步順序執行沒有利用異步的核心優勢——併發
實際運行和測試
將上述代碼保存為 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() # 依賴result1
result3 = await api_call_3() # 依賴result2
return [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總結
- async的真正價值：不在於讓單個任務變快，而在於讓多個任務可以同時進行
- 異步順序執行時間相近：因為還是逐個等待，沒有發揮併發優勢
- 性能提升的關鍵：使用
asyncio.gather()或類似機制實現真正的併發 - 實際應用：在設計異步程序時，要識別哪些任務可以併發執行
異步編程是一個強大的工具，但只有正確使用才能發揮其真正的威力。記住：異步的魅力不在於等待得更快，而在於可以同時等待多件事情。
延伸思考:在你的項目中，有哪些場景可以從順序執行改為併發執行？試着識別那些相互獨立的異步操作，這通常是性能優化的黃金機會。
