[!info] related notes

• 所属 MOC: Agent Runtime MOC
• 相关: Concurrency Control, Streaming Protocol

Backpressure

一句话定义

Backpressure 是当生产者速度超过消费者速度时的流控机制。LLM 产生 token 的速度可能很快，但前端渲染或 Go 后端转发可能跟不上，需要背压控制防止内存溢出和数据丢失。

核心原理

背压场景

LLM 产生 token (快) → Python 缓冲 → Go 缓冲 → 前端渲染 (慢)
                        ↑ 内存增长 ↑

Python 实现

import asyncio

class BackpressuredStream:
    def __init__(self, max_buffer_size=100):
        self.queue = asyncio.Queue(maxsize=max_buffer_size)

    async def produce(self, item):
        """生产者：如果队列满了，等待"""
        await self.queue.put(item)

    async def consume(self):
        """消费者：从队列取数据"""
        return await self.queue.get()

# 使用
stream = BackpressuredStream(max_buffer_size=50)

# 生产者（LLM token）
async def token_producer(llm_stream):
    async for token in llm_stream:
        await stream.produce(token)  # 队列满了会等待

# 消费者（SSE 转发）
async def sse_consumer(writer):
    while True:
        token = await stream.consume()
        writer.write(token)

Go 实现

func sseProxy(w http.ResponseWriter, aiStream <-chan Event) {
    // 使用带缓冲的 channel
    buffered := make(chan Event, 100)

    // 生产者
    go func() {
        for event := range aiStream {
            buffered <- event  // channel 满了会阻塞
        }
        close(buffered)
    }()

    // 消费者
    for event := range buffered {
        writeSSE(w, event)
        w.(http.Flusher).Flush()
    }
}

常见设计模式

1. 有界队列

队列有最大容量，满了就阻塞生产者。

2. 丢弃策略

队列满了丢弃最旧的数据（适合实时场景）。

3. 采样

数据太多时采样消费（如每 10 个 token 只消费 1 个）。

常见坑

1. 无界缓冲: token 无限累积，内存溢出
2. 不做流控: 生产者太快，消费者来不及处理
3. 阻塞主循环: 背压阻塞了 Agent 主循环

参考资料

• Backpressure - Reactive Streams
• Concurrency Control