Python.五.(一)--1. 并发编程、异步IO与多进程

• Python里有多线程吗？

• 线程池与进程池的区别是什么？

• multiprocessing 模块怎么使用？

• ProcessPoolExecutor 怎么使用？

• Python多进程中的fork和spawn模式有什么区别？

• Python中使用async def定义函数有什么作用？

• Python协程、asyncIO和异步IO在AI Agent中如何使用？

并发编程、异步IO与多进程是一组从属关系

并发编程是上位概念，多进程和异步IO都是实现"并发"的不同手段：多进程靠多核并行提升 CPU 密集型吞吐，异步IO靠单线程事件循环提升 IO 密集型吞吐，多线程在 Python 中因 GIL 介于两者之间。三者的选择本质是"任务类型 + 是否受 GIL 限制"的权衡。

Python 里有多线程吗

有。threading 模块提供线程 API，concurrent.futures.ThreadPoolExecutor 提供线程池。但 CPython 受 GIL（全局解释器锁） 限制，同一时刻只有一个线程执行 Python 字节码，所以多线程做不到真正的 CPU 并行，主要用于 IO 密集型任务（网络、磁盘、阻塞等待）。

补充：Python 3.13 引入了实验性的 free-threaded（无 GIL）构建，可真正多线程并行，但目前仍是实验特性，生态尚未全面适配。

线程池与进程池的区别

核心区别在隔离性与是否受 GIL 限制：

multiprocessing 模块怎么使用

基本用法是创建 Process 对象，start() 启动、join() 等待：

from multiprocessing import Process, Queue

def worker(name, q):
q.put(f"{name} done")

if __name__ == "__main__":
q = Queue()
ps = [Process(target=worker, args=(f"w{i}", q)) for i in range(4)]
for p in ps:
p.start()
for p in ps:
p.join()
while not q.empty():
print(q.get())

注意 if __name__ == "__main__" 保护在 Windows（spawn 模式）下是必须的，否则会递归导入导致报错。进程间通信用 Queue、Pipe，共享状态用 Manager、Value、Array。

ProcessPoolExecutor 怎么使用

它来自 concurrent.futures，是更高级的进程池接口，推荐用它替代手写 Process：

from concurrent.futures import ProcessPoolExecutor

def task(n):
return n * n

if __name__ == "__main__":
with ProcessPoolExecutor(max_workers=4) as ex:
# 方式一：submit + as_completed（顺序灵活）
futures = [ex.submit(task, i) for i in range(10)]
for f in futures:
print(f.result())

# 方式二：map（按输入顺序返回，更简洁）
results = list(ex.map(task, range(10)))

with 语句会自动管理进程池的创建与关闭。submit 适合任务参数和返回顺序不一致的场景，map 适合批量同构任务。

fork 和 spawn 模式的区别

这是多进程的启动方式（multiprocessing.set_start_method）：

实践建议：跨平台代码默认用 spawn；fork 在多线程父进程中 fork 可能出问题（C10K 类 fork-safety 隐患），Python 3.14 起已在 POSIX 上弃用多线程 fork。

async def 定义函数的作用

async def 定义的是协程函数，调用它不会立即执行，而是返回一个协程对象，必须放进事件循环（await 或 asyncio.run）中才会运行。它让函数能在遇到 IO 等待时主动挂起、交出控制权，从而在单线程内并发处理多个 IO 任务。

import asyncio

async def fetch(url):
await asyncio.sleep(1) # 遇到等待就挂起，让出控制权
return url

async def main():
# 两个请求并发执行，总耗时约 1 秒而非 2 秒
results = await asyncio.gather(fetch("a"), fetch("b"))
print(results)

asyncio.run(main())

关键点：await 只能出现在 async def 内；协程必须被调度才执行；并发而非并行（单线程内切换）。

协程、asyncIO 与异步IO在 AI Agent 中如何使用

AI Agent 的典型瓶颈是大量 IO 等待：调用 LLM API、检索向量库、访问数据库、调用外部工具——这些每个都要几百毫秒到几秒，CPU 却基本空闲。异步IO正好对症：用单线程事件循环把多个 IO 等待重叠起来，极大提升吞吐、降低延迟。

典型应用场景：

• 并发调用多个工具/子任务：一个 Agent 同时查搜索、查数据库、调计算工具，用 asyncio.gather 并发执行而非串行等待。
• 流式响应处理：LLM 的 streaming 输出本质是异步流，async for 逐块接收并转写。
• 并发调用多个 LLM 或多轮采样：self-consistency、多路径推理同时发起。
• 同时服务多用户：后端框架（FastAPI、aiohttp）天然异步，配合 async Agent 可高并发处理请求。
• 多 Agent 协作：多个 Agent 节点并行规划/执行，通过异步通道通信。

一个最小示例：并发调用两个"工具"模拟 Agent 工作流：

import asyncio

async def call_llm(prompt):
await asyncio.sleep(0.8) # 模拟 API 网络等待
return f"LLM回复: {prompt}"

async def search(query):
await asyncio.sleep(0.5) # 模拟检索等待
return ["相关文档1", "相关文档2"]

async def agent_run(question):
# 检索与 LLM 调用并发，重叠等待时间
docs, answer = await asyncio.gather(
search(question),
call_llm(question),
)
return f"{answer}（参考: {docs}）"

asyncio.run(agent_run("什么是多进程"))

实战中要注意：CPU 密集型子任务（如本地模型推理、大向量计算）会阻塞事件循环，应丢进 ProcessPoolExecutor，用 loop.run_in_executor 调度——这正是多进程 + 异步IO混用的典型模式，各取所长。

一句话收尾：CPU 密集用多进程，IO 密集用异步IO，Agent 通常是异步IO为主、多进程兜底重计算