在现代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")
# 异步节点定义
asyncdef 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]}
# 异步版本的节点
asyncdef 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
asyncdef 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
asyncdef 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
asyncdef 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密集型计算
- 必须顺序执行的依赖任务
- 简单的单次操作
最佳实践
# 错误用法：异步但无性能提升
asyncdef bad\_example():
result1 = await api\_call\_1()
result2 = await api\_call\_2() # 依赖result1
result3 = await api\_call\_3() # 依赖result2
return [result1, result2, result3]
# 改进：部分并发
asyncdef 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]
# 最佳：完全并发（当任务独立时）
asyncdef 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()或类似机制实现真正的并发 - 实际应用 ：在设计异步程序时，要识别哪些任务可以并发执行
异步编程是一个强大的工具，但只有正确使用才能发挥其真正的威力。记住：异步的魅力不在于等待得更快，而在于可以同时等待多件事情 。
延伸思考:在你的项目中，有哪些场景可以从顺序执行改为并发执行？试着识别那些相互独立的异步操作，这通常是性能优化的黄金机会。
添加微信，备注” LLM “进入大模型技术交流群
如果你觉得这篇文章对你有帮助，别忘了点个赞、送个喜欢
/ 作者：致Great
/ 作者：欢迎转载，标注来源即可
Agent实战教程：Langgraph的StateGraph以及State怎么用
Agent实战教程：LangGraph核心概念节点、边以及状态详解
Agent实战教程：LangGraph关于智能体的架构模式与核心概念
Agent实战教程：LangGraph中工作流与智能体区别与实现
Agent实战教程：LangGraph相关概念介绍以及快速入门
Deep Agents：用于复杂任务自动化的 AI 代理框架
Kimi K2智能体能力的技术突破：大规模数据合成 + 通用强化学习
构建AI Agent的完整实战指南：从邮件助手案例看6步落地方法
Context Engineering：从Prompt Engineering到上下文工程的演进
如何用LangGraph打造Web Research多智能体系统
Anthropic关于智能体的经验分享：如何构建高效的Agent？
Gemini开源项目DeepResearch：基于LangGraph的智能研究Agent技术原理与实现
xAI 把 Grok 的系统提示词全部公开了，我们看看DeepResearch的系统提示词怎么设计的?