企业微信并发优化方案对比

# api_server.py 第 2373 行
_MAX_CONCURRENT_RUNS = 10  # 已内置并发限制

# 每个请求在独立线程池里跑 agent
return await loop.run_in_executor(None, _run)  # 第 2367 行

# wecom_bridge.py（新增服务）

from aiohttp import web
import aiohttp

API_SERVER_URL = "http://localhost:8000/v1/chat/completions"
API_SERVER_TOKEN = "your-token-here"

async def wecom_webhook(request):
    """企业微信消息回调入口"""
    msg = await request.json()
    user_id = msg.get("FromUserName")       # 发送者
    content = msg.get("Content")             # 文本内容
    chat_id = msg.get("ToUserName")         # 接收者（机器人）

    # 调用 api_server
    async with aiohttp.ClientSession() as sess:
        resp = await sess.post(
            API_SERVER_URL,
            json={
                "messages": [{"role": "user", "content": content}],
                "stream": False
            },
            headers={
                "Authorization": f"Bearer {API_SERVER_TOKEN}",
                "X-Tenant-ID": "wecom",
                "X-User-ID": user_id,
                "X-Chat-ID": chat_id,
            }
        )
        result = await resp.json()

    reply = result["choices"][0]["message"]["content"]

    # 通过企业微信主动推送回复
    await send_wecom_message(user_id, reply)
    return {"errcode": 0}

async def send_wecom_message(to_user, content):
    """调用企业微信应用消息接口"""
    # 实现参见企业微信官方API

app = web.Application()
app.router.add_post("/wecom/webhook", wecom_webhook)
web.run_app(app, host="0.0.0.0", port=8080)

# 在 api_server.py 的 _handle_chat_completions() 里新增

# 从 header 提取 tenant/user
tenant_id = request.headers.get("X-Tenant-ID", "default")
user_id = request.headers.get("X-User-ID", "default")
chat_id = request.headers.get("X-Chat-ID", "default")

# 构建隔离的 session_key
session_key = f"{tenant_id}:{user_id}:{chat_id}:dm"

# 每个 session_key 领取一个 sentinel，表示"这个 session 正在处理"
self._running_agents[_quick_key] = _AGENT_PENDING_SENTINEL
self._running_agents_ts[_quick_key] = time.time()

# 中间大量 await 点（vision、STT、hooks）...

# 真正处理（整个过程是串行的）
_agent_result = await self._handle_message_with_agent(...)

# run.py Gateway.__init__() 里新增

# 全局并发控制信号量（可配置）
self._agent_semaphore = asyncio.Semaphore(5)

# per-session 锁字典（不同 session 不互相阻塞）
self._session_locks: Dict[str, asyncio.Lock] = {}

# run.py _handle_message() 改造

async def _handle_message(self, event: MessageEvent) -> Optional[str]:
    session_key = self._session_key_for_source(source)

    # 每个 session 有自己的锁（同一 session 串行，不同 session 并发）
    session_lock = self._session_locks.setdefault(session_key, asyncio.Lock())

    # 全局限流：最多 5 个并发 Agent
    async with self._agent_semaphore:
        # session 级串行
        async with session_lock:
            return await self._handle_message_with_agent(
                event, source, _quick_key, run_generation
            )

# 方案A：写操作加全局锁（简单但影响性能）
self._session_write_lock = asyncio.Lock()

# 方案B：每个并发 slot 用独立 SQLite 文件
# 按 session_key 哈希到不同 DB 文件

# 方案C：改用 PostgreSQL（改动最大但最彻底）

# run.py __init__() 读取配置
max_concurrent = self.config.get("gateway", {}).get("max_concurrent_agents", 5)
self._agent_semaphore = asyncio.Semaphore(max_concurrent)