Python 事件驱动架构：用 asyncio.Queue 构建解耦的 RAG 处理流水线

Python 事件驱动架构：用 asyncio.Queue 构建解耦的 RAG 处理流水线

一、深度引言与场景痛点

大家好，我是赵咕咕。

你有没有见过这种 RAG 代码？

async def process_query(query: str): docs = await retrieve(query) context = await rerank(query, docs) answer = await generate(query, context) await log(query, answer) await cache(query, answer) return answer

看起来还行？那如果加上"检索失败要 fallback 到关键词搜索"、"重排序超时要降级为原始排序"、"日志写入失败不能阻塞主流程"、"缓存要异步批量写入"——五个if-else加上三层try-except，这段代码就变成了灾难。

这就是紧耦合流水线的通病：每个步骤直接依赖前一步的结果，异常处理和降级逻辑散落在各处，改一个步骤要看完整个流程。

解决这个问题的方法是事件驱动架构。用asyncio.Queue把各个处理步骤解耦成独立的生产者-消费者，让流水线从"串行硬编码"变成"消息驱动组合"。

二、底层机制与原理深度剖析

2.1 为什么要解耦？

紧耦合流水线的核心问题是步骤间的依赖是隐式的。你看到generate(query, context)，但它依赖了上面两步的产出。当任何一个步骤的异常需要被处理时，这个隐式依赖链就变成了噩梦。

事件驱动架构的核心思路是：步骤之间不直接通信，通过消息队列传递数据。每个步骤是一个独立的 Worker，它只关心"我从哪个队列读、我往哪个队列写"，不知道上游是谁、下游是谁。

2.2 asyncio.Queue 的运作模型

asyncio.Queue是 Python asyncio 内置的异步队列。跟queue.Queue不同的是，它的put和get不会阻塞线程，而是await等待。这让它天然适合在协程之间传递消息。

具体而言，整个流水线由输入层、处理层和中间队列组成。用户请求首先进入query_queue，随后被RetrieveWorker消费进行向量检索与关键词 fallback，结果写入retrieved_queue。接着RerankWorker从该队列读取数据进行重排序与超时降级，产出存入reranked_queue。GenerateWorker随后合成答案并写入generated_queue。最后，LogWorker和CacheWorker并行从生成队列读取数据，分别执行异步日志写入和批量缓存操作。

每个 Worker 是一个独立的asyncio.Task，持续从上游队列取消息、处理、写入下游队列。它们之间唯一的耦合点是队列的message数据格式。

2.3 消息协议的标准化

要让队列真正解耦，消息格式必须标准化。我定义了一个简单的PipelineMessage数据结构：

@dataclass class PipelineMessage: query_id: str # 唯一请求 ID，用于追踪整个链路

query: str # 原始查询 retrieved_docs: list | None = None reranked_docs: list | None = None answer: str | None = None metadata: dict | None = None # 各步骤可以往里塞元信息

每个 Worker 只读取和处理自己关心的字段，其他字段透传。这样新增一个 Worker 不需要修改已有的任何 Worker 代码。 ## 三、生产级代码实现 下面是完整的事件驱动 RAG 流水线实现： ```python import asyncio import uuid import logging from dataclasses import dataclass, field from typing import Any logger = logging.getLogger(__name__) @dataclass class PipelineMessage: """在流水线各 Worker 间传递的消息体。""" query_id: str query: str retrieved_docs: list[dict] = field(default_factory=list) reranked_docs: list[dict] = field(default_factory=list) answer: str | None = None error: str | None = None metadata: dict[str, Any] = field(default_factory=dict) class RAGPipeline: """事件驱动的 RAG 处理流水线。""" def __init__( self, retriever, # 替换为你的检索器 reranker, # 替换为你的重排序器 llm_generator, # 替换为你的 LLM cache_backend=None, # 替换为你的缓存 worker_count: int = 4, ): self._retriever = retriever self._reranker = reranker self._generator = llm_generator self._cache = cache_backend # 解耦队列 self._query_queue: asyncio.Queue[PipelineMessage] = asyncio.Queue( maxsize=100 ) self._retrieved_queue: asyncio.Queue[PipelineMessage] = asyncio.Queue( maxsize=200 ) self._reranked_queue: asyncio.Queue[PipelineMessage] = asyncio.Queue( maxsize=200 ) self._generated_queue: asyncio.Queue[PipelineMessage] = asyncio.Queue( maxsize=200 ) self._worker_count = worker_count self._workers: list[asyncio.Task] = [] self._results: dict[str, asyncio.Future] = {} self._running = False # ─── 对外接口 ─── async def start(self) -> None: """启动所有 Worker。""" self._running = True self._workers = [ asyncio.create_task(self._retrieve_worker()), asyncio.create_task(self._rerank_worker()), asyncio.create_task(self._generate_worker()), asyncio.create_task(self._log_worker()), asyncio.create_task(self._cache_worker()), ] logger.info("RAG Pipeline 已启动，%d 个 Worker 就绪", len(self._workers)) async def stop(self) -> None: """优雅关闭所有 Worker。""" self._running = False # 向每个队列发送哨兵信号 queues = [ self._query_queue, self._retrieved_queue, self._reranked_queue, self._generated_queue, ] for q in queues: await q.put(None) # None 作为停止信号 for worker in self._workers: worker.cancel() await asyncio.gather(*self._workers, return_exceptions=True) logger.info("RAG Pipeline 已停止") async def process(self, query: str, timeout: float = 30.0) -> PipelineMessage: """提交查询并等待结果。""" msg = PipelineMessage(query_id=str(uuid.uuid4()), query=query) future: asyncio.Future = asyncio.get_running_loop().create_future() self._results[msg.query_id] = future try: await self._query_queue.put(msg) result = await asyncio.wait_for(future, timeout=timeout) return result except asyncio.TimeoutError: self._results.pop(msg.query_id, None) return PipelineMessage( query_id=msg.query_id, query=query, error="流水线处理超时", ) except Exception: self._results.pop(msg.query_id, None) raise # ─── Worker 实现 ─── async def _retrieve_worker(self) -> None: """检索 Worker：向量检索 + 关键词 fallback。""" while self._running: msg = await self._query_queue.get() if msg is None: self._query_queue.task_done() break try: # 主路径：向量检索 docs = await asyncio.wait_for( self._retriever.asearch(msg.query), timeout=5.0 ) msg.retrieved_docs = docs msg.metadata["retrieval_method"] = "vector" except (asyncio.TimeoutError, Exception) as e: logger.warning("向量检索失败，降级到关键词检索: %s", e) try: docs = await self._retriever.keyword_search(msg.query) msg.retrieved_docs = docs msg.metadata["retrieval_method"] = "keyword_fallback" except Exception as e2: msg.error = f"检索失败: {e2}" msg.metadata["retrieval_error"] = str(e2) self._deliver_result(msg) self._query_queue.task_done() continue await self._retrieved_queue.put(msg) self._query_queue.task_done() async def _rerank_worker(self) -> None: """重排序 Worker：交叉编码器 + 超时降级。""" while self._running: msg = await self._retrieved_queue.get() if msg is None: self._retrieved_queue.task_done() break if msg.error: await self._reranked_queue.put(msg) self._retrieved_queue.task_done() continue try: reranked = await asyncio.wait_for( self._reranker.rerank(msg.query, msg.retrieved_docs), timeout=3.0, ) msg.reranked_docs = reranked msg.metadata["rerank_method"] = "cross_encoder" except asyncio.TimeoutError: logger.warning("重排序超时，使用原始排序") msg.reranked_docs = msg.retrieved_docs msg.metadata["rerank_method"] = "original_order" except Exception as e: logger.error("重排序异常: %s", e) msg.reranked_docs = msg.retrieved_docs msg.metadata["rerank_method"] = "original_order_fallback" msg.metadata["rerank_error"] = str(e) await self._reranked_queue.put(msg) self._retrieved_queue.task_done() async def _generate_worker(self) -> None: """生成 Worker：答案合成。""" while self._running: msg = await self._reranked_queue.get() if msg is None: self._reranked_queue.task_done() break if msg.error: await self._generated_queue.put(msg) self._reranked_queue.task_done() continue try: context = "\n\n".join( d.get("content", "")[:500] for d in msg.reranked_docs[:5] ) answer = await self._generator.generate(msg.query, context) msg.answer = answer except Exception as e: msg.error = f"生成失败: {e}" self._deliver_result(msg) await self._generated_queue.put(msg) self._reranked_queue.task_done() async def _log_worker(self) -> None: """日志 Worker：异步写入，不阻塞主流程。""" while self._running: msg = await self._generated_queue.get() if msg is None: self._generated_queue.task_done() break try: logger.info( "query_id=%s query=%.50s answer_len=%d error=%s", msg.query_id, msg.query, len(msg.answer or ""), msg.error, ) except Exception: pass # 日志失败不影响主流程 self._generated_queue.task_done() async def _cache_worker(self) -> None: """缓存 Worker：批量写入。""" if self._cache is None: return batch: list[PipelineMessage] = [] while self._running: msg = await self._generated_queue.get() if msg is None: self._generated_queue.task_done() break batch.append(msg) self._generated_queue.task_done() # 每 10 条或每 5 秒批量写入 if len(batch) >= 10: await self._flush_cache(batch) batch.clear() # 退出前刷掉剩余 if batch: await self._flush_cache(batch) async def _flush_cache(self, batch: list[PipelineMessage]) -> None: try: await asyncio.wait_for( self._cache.batch_set( {m.query: m.answer for m in batch if m.answer} ), timeout=2.0, ) except Exception as e: logger.warning("缓存批量写入失败: %s", e) def _deliver_result(self, msg: PipelineMessage) -> None: """将结果通知给等待的调用方。""" future = self._results.pop(msg.query_id, None) if future and not future.done(): future.set_result(msg)

设计要点解析：

• 每个 Worker 是独立协程：通过asyncio.create_task启动，在后台持续运行。新增处理步骤只需要新增一个 Worker 和对应的队列。
• 哨兵机制：stop()向每个队列发送None作为停止信号。Worker 收到None后退出循环。这比直接cancel()更优雅——数据不会丢。
• 超时 + 降级：检索和重排序都做了超时保护。检索超时降级到关键词搜索，重排序超时降级到原始排序。生产环境中"部分降级"比"整体崩溃"好一万倍。
• 错误穿透：PipelineMessage.error字段让错误消息沿着队列正常流转，下游 Worker 检测到error就直接透传。这样失败的消息也能被日志和监控捕获。
• 日志和缓存不阻塞：_log_worker和_cache_worker异常被吞掉了，因为日志和缓存失败不应该影响用户的问答结果。它们是最低优先级的旁路 Worker。
• maxsize 背压：每个队列都设了maxsize。当某个 Worker 处理不过来时，put会await等待，天然形成背压。

四、边界分析与架构权衡

4.1 事件驱动 vs 串行流水线

4.2 什么时候不该用事件驱动？

• 低并发、低复杂度：如果一天的请求量不超过几百条，Worker 之间的队列交换反而是多余的复杂度。
• 实时性要求极高：消息在多个队列间传递引入了额外延迟（通常 1-5ms），对实时场景可能不可接受。
• 强事务一致性：如果每个步骤必须全部成功才算一次有效处理，那么事件驱动的"部分降级"反而会引入数据不一致。
• 团队不熟悉异步编程：事件驱动架构对asyncio.Queue的理解要求较高，团队学习成本需要考虑。

4.3 消息协议的版本管理

当流水线越来越复杂时，消息格式一定会演化。建议在PipelineMessage.metadata里加一个version字段。每个 Worker 读取消息时检查版本，兼容旧版本消息：

if msg.metadata.get("version") == "v1": docs = msg.retrieved_docs # v1 格式 elif msg.metadata.get("version") == "v2": docs = msg.retrieved_chunks # v2 改名了

4.4 监控与可观测性

事件驱动架构的调试是个痛点。消息在队列间流转，一个请求走了多少步、卡在哪一步，不容易追踪。

建议在每个 Worker 的入口和出口打结构化日志，包含query_id。用 ELK 或 Grafana Loki 聚合，可以通过query_id串联整个处理链路。

另外，asyncio.Queue.qsize()可以作为队列积压的监控指标。如果某个队列长期大于某个阈值，说明下游 Worker 处理能力不足。

五、总结

事件驱动架构用asyncio.Queue实现，把 RAG 流水线从"一串怕一个崩"的紧耦合变成了"各司其职、独立容错"的松耦合。

核心收益有三点：

1. 异常隔离：检索挂了不影响还在队列里的其他请求，降级到关键词搜索继续跑。
2. 独立扩展：如果重排序成为瓶颈，把_rerank_worker从 1 个扩到 3 个，只改一行代码（create_task调三次）。
3. 旁路不阻塞：日志和缓存等旁路任务不再影响用户感知的响应时间。

代价是系统复杂度增加——你需要管理队列、Worker 生命周期、消息版本和链路追踪。但对于高并发、高可用的生产级 RAG 服务来说，这个代价是值得的。

事件驱动不是什么新鲜概念，后端工程师用消息队列解耦服务已经几十年了。但把同样的思想用在单进程内的 RAG 流水线上，效果出奇地好。asyncio.Queue就是你的"进程内消息中间件"。

下一篇预告：RAG 幻觉检测，在 Agent 输出答案前自动交叉验证引用和事实的一致性。