第 5 章 并发体系、异步编程与 Agent 后端实战5.1 并发基石GIL 机制与并发选型矩阵Java 工程师非常熟悉 JVM 内存模型JMM、内核线程与 JUC 线程池在 Java 里多线程可以轻松打满多核 CPU。而在 Python 中首先需要理解GILGlobal Interpreter Lock全局解释器锁的运行机制1. GIL 的本质与运行机制CPython 解释器的互斥锁为了简化内存管理引用计数安全CPython 在进程级别设置了一把全局锁。同一时刻无论机器有多少个 CPU 核心同一个 Python 进程内只能有一个线程执行 Python 字节码。为什么 Java 没有 GILJVM 采用精密的分代垃圾回收与细粒度对象头锁机制支持真正的多线程多核并行。2. I/O 密集 vs CPU 密集I/O 密集型Agent 后端绝大多数场景当线程执行网络调用LLM API、HTTP 请求、文件读写或数据库查询等待时Python 会主动释放 GIL。因此在 I/O 等待期间其他线程或协程可以继续执行。在此场景下asyncio协程或多线程完全能支撑极高并发。CPU 密集型分词、向量计算、复杂文本解析线程持续执行 Python 计算GIL 无法释放多线程不但无法利用多核还会因为频繁的线程上下文切换与争锁导致性能比单线程更慢。唯一解法多进程Multiprocessing。每个进程拥有独立的 Python 解释器、独立的内存空间和独立的 GIL由操作系统调度到不同 CPU 核心并行执行。3. Python 后端并发选型并发模式核心技术载体并发上限内存开销核心应用场景 (Agent 后端)对标 Java 机制异步协程 (Asyncio)async/await 单线程事件循环万级 ~ 十万级极低 (~KB 级/协程)Agent 核心标配海量 LLM API 交互、FastAPI 请求、WebSocket/SSE 流式推送WebFlux / Netty / CompletableFuture (非阻塞)多线程 (Threading)concurrent.futures.ThreadPoolExecutor千级较高 (~MB 级/线程)包装遗留的同步阻塞 SDK如老旧数据库驱动、同步文件读写Executors.newFixedThreadPool多进程 (Multiprocessing)concurrent.futures.ProcessPoolExecutorCPU 核心数 (十级)高 (独立进程内存)CPU 密集计算本地 Embedding 计算、大文本正则清洗、分词加密ForkJoinPool / 多进程隔离服务在asyncio中运行阻塞代码的最佳桥梁# 遇到无法异步化的第三方同步库丢给线程池避免卡死事件循环resultawaitasyncio.to_thread(legacy_blocking_io_call,arg1)5.2 为什么 agent 后端必须 async一次 Agent 请求通常包含多次 LLM API 调用、知识库检索与工具执行大部分时间处于 I/O 等待。Java 通常采用线程池 CompletableFuturePython 则采用事件循环Event Loop在单线程内调度并发协程发生 I/O 等待时自动让出执行权。JavaPython并发单元线程OS 级MB 级栈协程用户态KB 级编写方式CompletableFuture.supplyAsyncasync def/await并发上限千级线程万级协程注意点线程安全、锁单处同步阻塞会导致整个事件循环挂起等待最大禁忌在async函数里调用同步阻塞调用如time.sleep、同步requests.get——整个事件循环都会被卡住所有并发请求一起卡死。要么用异步库httpx/asyncpg要么用asyncio.to_thread丢线程池resultawaitasyncio.to_thread(blocking_func,arg)# ↔ executor.submit5.3 async/await ↔ CompletableFutureimportasyncio# async def 定义协程调用它只是创建协程对象不执行和生成器同理asyncdefcall_llm(prompt:str)-str:...# 遇到 I/O 就 await 让出resultawaitcall_llm(hi)# await ≈ .join()等结果但不阻塞 loop// Java 等价心智varfutureCompletableFuture.supplyAsync(()-callLlm(hi));varresultfuture.join();关键差异Java 的 future 是启动即执行、后面再 joinPython 的协程对象创建时不执行必须被 await或 gather/create_task 调度才运行。并发执行 ↔ allOf# Java: CompletableFuture.allOf(f1, f2, f3).thenApply(...)resultsawaitasyncio.gather(retrieve_from_db(query),# 协程 1call_llm(prompt),# 协程 2fetch_tools(),# 协程 3)# 按传入顺序返回结果列表db_result,llm_result,toolsresults# deep research 场景并行处理列表asyncdefsearch_all(queries:list[str])-list[str]:returnawaitasyncio.gather(*[search(q)forqinqueries])启动后台任务 ↔ executor.submittaskasyncio.create_task(call_llm(hi))# 立即调度不等待↔ submit# ... 干别的事 ...resultawaittask# 之后需要结果时再 await超时与取消# 超时 ↔ future.orTimeouttry:resultawaitasyncio.wait_for(call_llm(prompt),timeout30)exceptTimeoutError:...# Python 3.11 更简洁的写法asyncwithasyncio.timeout(30):resultawaitcall_llm(prompt)5.4 异步生成器与流式传输Agent SSE 实战在 Agent 后端开发中大模型打字机流式输出的本质是异步生成器Async Generator。在 4.4 节我们学习了同步生成器defyield它是阻塞拉取的在异步事件循环中大模型分块数据是随网络数据包异步到达的必须结合async def与yieldimportasynciofromcollections.abcimportAsyncIteratorfromopenaiimportAsyncOpenAI clientAsyncOpenAI()# 异步生成器函数返回类型标注为 AsyncIterator[str]asyncdefstream_agent_reasoning(prompt:str)-AsyncIterator[str]:# 模拟调用大模型流式接口streamawaitclient.chat.completions.create(modelgpt-4o,messages[{role:user,content:prompt}],streamTrue,)asyncforchunkinstream:# 异步迭代逐个等待网络 chunkcontentchunk.choices[0].delta.contentifcontent:yieldcontent# 每到达一段内容立即异步产出# 消费端必须使用 async for 循环消费asyncdefmain_consumer():asyncfortokeninstream_agent_reasoning(写一首诗):print(token,end,flushTrue)# 控制台流式打字效果FastAPI 中端到端 Server-Sent Events (SSE) 接口落地在微服务接口层直接将异步生成器包装为StreamingResponse即可原生输出符合标准 SSE 格式的 HTTP 长连接流fromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponse appFastAPI()asyncdefformat_sse_stream(prompt:str)-AsyncIterator[str]:asyncfortokeninstream_agent_reasoning(prompt):# 遵循 SSE 规范每条数据以 data: ...\n\n 结尾yieldfdata:{token}\n\nyielddata: [DONE]\n\napp.get(/api/chat/stream)asyncdefchat_endpoint(prompt:str):returnStreamingResponse(format_sse_stream(prompt),media_typetext/event-stream,headers{Cache-Control:no-cache,Connection:keep-alive,})Java Web 对应物Spring WebFlux 的FluxServerSentEventString。5.5 异步生态速查用途同步库异步库agent 后端用这套HTTP 客户端requestshttpxWeb 框架FlaskFastAPI uvicorn数据库psycopgasyncpg/ SQLAlchemy asyncRedisredisredis有 async 模式LLM SDKopenaiopenai.AsyncOpenAIimporthttpxasyncwithhttpx.AsyncClient(timeout30)asclient:# ↔ WebClient/OkHttprespawaitclient.post(https://api.openai.com/v1/chat/completions,json{...},headers{Authorization:fBearer{key}},)resp.raise_for_status()# ↔ 判断 2xxdataresp.json()注意LLM SDK 的异步客户端类名带Async前缀且实例应全局复用内含连接池fromopenaiimportAsyncOpenAI clientAsyncOpenAI()# 模块级创建一次别每请求新建asyncdefchat(messages:list[dict])-str:respawaitclient.chat.completions.create(modelgpt-4o,messagesmessages)returnresp.choices[0].message.content5.6 并发控制 ↔ 线程池限流Java 用固定大小线程池天然限流Python 协程想开多少开多少限流要显式用Semaphoresemasyncio.Semaphore(10)# 最多 10 个并发 LLM 调用asyncdefcall_llm_limited(prompt:str)-str:asyncwithsem:# ↔ acquire/releasereturnawaitcall_llm(prompt)# 深度研究批量并发带限流 全部执行asyncdefresearch_all(queries:list[str])-list[str]:returnawaitasyncio.gather(*[call_llm_limited(q)forqinqueries])其他常用工具asyncio.Queue↔BlockingQueue生产者-消费者常用于事件流asyncio.Lock↔ReentrantLockasyncio.Event↔CountDownLatch的简化版。第 6 章 后端工程生态与持久层6.1 测试pytest ↔ JUnit 5# tests/test_tools.pyimportpytestfrommy_agent.tools.calculatorimportcalculate# 一条测试就是一个 test_ 开头的普通函数无需类、无断言 API直接 assertdeftest_addition():assertcalculate(1 2)3# assert 表达式失败自动打印两侧值deftest_division_by_zero():withpytest.raises(ZeroDivisionError):# ↔ assertThrowscalculate(1 / 0)# parametrize ↔ ParameterizedTestpytest.mark.parametrize(expr, expected,[(11,2),(2*3,6),])deftest_expressions(expr,expected):assertcalculate(expr)expected# fixture ↔ BeforeEach 依赖注入自动发现、按需注入pytest.fixturedefagent():aToolAgent(nametest)yielda# yield 前是 setup后是 teardowna.cleanup()deftest_agent_runs(agent):# 参数名匹配 fixture 即自动注入assertdoneinagent.run(task)uv run pytest# 跑全部uv run pytest-kaddition# 按名过滤 ↔ Taguv run pytest--cov# 覆盖率需 pytest-cov↔ JaCoCo与 JUnit 的心智差异没有注解驱动的生命周期就是普通函数 fixture 注入mock 用unittest.mockmocker.patch(...)替换任意对象属性比 Mockito 更加直接动态。6.2 日志importlogging loggerlogging.getLogger(__name__)# ↔ LoggerFactory.getLogger按模块命名logger.info(tool %s finished in %.2fs,name,elapsed)# 惰性格式化与 slf4j 同理logger.exception(failed)# 在 except 内用自动附 traceback# 应用入口配置一次FastAPI 项目通常在启动处或用 loguru 简化logging.basicConfig(levellogging.INFO,format%(asctime)s %(name)s %(levelname)s %(message)s)标准库logging层级 logger 传播机制 ↔ logback 的层级 appender。社区更爱loguru零配置、彩色、自动 tracebackfromloguruimportlogger logger.info(tool {name} done,namename)logger.add(app.log,rotation10 MB)# ↔ logback RollingFileAppender注意Python 的 print 不该出现在后端代码里脚本除外一律 logger。6.3 常用标准库速查后端开发最常碰的标准库都是内置无需安装库用途Java 对应jsonJSON 编解码Jacksonos/os.environ环境变量、系统交互System.getenvpathlib.Path路径操作面向对象替代 os.pathjava.nio.file.Pathdatetime时间日期另有zoneinfo处理时区java.timere正则java.util.regexdataclasses数据类Lombokenum枚举Enumcollectionsdefaultdict、Counter、deque补充容器functoolscache、wraps、partial函数工具itertools迭代工具islice、chain、productStream/Collectors 部分contextlib上下文管理器工具try-with-resources 泛化asyncio异步运行时CompletableFuture/Executortyping类型注解工具泛型hashlib/secrets摘要与安全随机MessageDigest/SecureRandomuuidUUIDUUIDlogging日志SLF4Junittest.mock测试替身Mockitosqlite3内置 SQLite无JDBC 需驱动# 几个高频惯用法fromcollectionsimportdefaultdict tools_by_tag:dict[str,list[Tool]]defaultdict(list)# 等价 computeIfAbsenttools_by_tag[search].append(tool)fromcollectionsimportCounter Counter(roles)# 频次统计 ↔ Collectors.groupingBycountingfrompathlibimportPath configjson.loads(Path(config.json).read_text(encodingutf-8))Path(out)/report.md# 路径拼接用 / 运算符fromfunctoolsimportcachecachedefload_schema(name:str)-dict:...6.4 数据库访问SQLAlchemy 2.0Python 生态没有 MyBatis 式的 XML mapper 主流玩法主流是SQLAlchemyJPA 地位 轻量的原生 SQL 支持agent 项目常配asyncpg SQLAlchemy async或直接SQLModelSQLAlchemy pydantic 合体fromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnclassBase(DeclarativeBase):...classConversation(Base):# ↔ Entity__tablename__conversationsid:Mapped[str]mapped_column(primary_keyTrue)# 类型由注解推导user_id:Mapped[str]mapped_column(indexTrue)created_at:Mapped[datetime]# 查询2.0 风格接近 JPA Criteriafromsqlalchemyimportselect stmtselect(Conversation).where(Conversation.user_iduid).order_by(Conversation.created_at.desc())rows(awaitsession.execute(stmt)).scalars().all()# 支持直接执行原生 SQLrows(awaitsession.execute(text(SELECT * FROM conversations WHERE user_id :uid),{uid:uid})).mappings().all()