[!info] related notes

• 所属 MOC: Agent Runtime MOC
• 相关: Backpressure, Timeout Control

Concurrency Control

一句话定义

Concurrency Control 是 Agent Runtime 中控制并发度的机制。它限制同时执行的 Run 数量、并行工具调用数量和 LLM 请求并发度，防止资源耗尽和 API 限流。

核心原理

并发限制维度

维度	限制	原因
同时运行的 Run	10-50	服务器资源有限
单 Run 内并行工具调用	3-5	避免工具互相干扰
LLM API 并发请求	10-20	Provider 限流
同一用户的并发 Run	1-3	防止滥用

Python 实现

class ConcurrencyController:
    def __init__(self, max_concurrent_runs=20, max_parallel_tools=3):
        self.run_semaphore = asyncio.Semaphore(max_concurrent_runs)
        self.tool_semaphore = asyncio.Semaphore(max_parallel_tools)

    async def acquire_run_slot(self):
        return self.run_semaphore.acquire()

    def release_run_slot(self):
        self.run_semaphore.release()

    async def with_run_limit(self, coro):
        async with self.run_semaphore:
            return await coro

    async def with_tool_limit(self, coro):
        async with self.tool_semaphore:
            return await coro

# 使用
controller = ConcurrencyController()

async def handle_run(run):
    async with controller.run_semaphore:
        # 执行 Run
        ...

# 并行工具调用
async def execute_tools_parallel(tool_calls):
    tasks = [controller.with_tool_limit(execute(tc)) for tc in tool_calls]
    return await asyncio.gather(*tasks)

Go 实现

type ConcurrencyController struct {
    runSemaphore chan struct{}
}

func NewConcurrencyController(maxConcurrent int) *ConcurrencyController {
    return &ConcurrencyController{
        runSemaphore: make(chan struct{}, maxConcurrent),
    }
}

func (c *ConcurrencyController) Acquire() {
    c.runSemaphore <- struct{}{}
}

func (c *ConcurrencyController) Release() {
    <-c.runSemaphore
}

常见坑

1. 不做并发限制: 100 个用户同时发请求，服务器崩溃
2. 限流太严: 用户体验差
3. 不做用户级限制: 一个用户占满所有并发

参考资料

• Backpressure
• Python asyncio.Semaphore