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 流水线从"一串怕一个崩"的紧耦合变成了"各司其职、独立容错"的松耦合。

核心收益有三点：

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

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

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