From 6f3bbd380387d076efe1062161e6824a691b0114 Mon Sep 17 00:00:00 2001 From: Nelson <1475262689@qq.com> Date: Wed, 8 Jul 2026 16:57:40 +0800 Subject: [PATCH] Implement chat queue fallback --- .../apps/admin-web/src/App.vue | 27 ++++ .../apps/backend/.env.example | 3 + .../apps/backend/app/api/chat.py | 70 ++++++++- .../apps/backend/app/core/config.py | 3 + .../app/services/chat_queue_service.py | 146 ++++++++++++++++++ .../apps/user-client/src/App.vue | 26 +++- .../src/components/ChatMessage.vue | 53 ++++--- .../apps/user-client/src/services/api.ts | 10 +- .../apps/user-client/src/styles.css | 25 +++ ...026-07-08-concurrency-optimization-plan.md | 35 +++++ 10 files changed, 367 insertions(+), 31 deletions(-) create mode 100644 ai_knowledge_base_v2/apps/backend/app/services/chat_queue_service.py diff --git a/ai_knowledge_base_v2/apps/admin-web/src/App.vue b/ai_knowledge_base_v2/apps/admin-web/src/App.vue index 10656fc..51b172f 100644 --- a/ai_knowledge_base_v2/apps/admin-web/src/App.vue +++ b/ai_knowledge_base_v2/apps/admin-web/src/App.vue @@ -275,6 +275,33 @@ const systemSettingSections: SystemSettingSection[] = [ defaultValue: true, description: "后台始终记录引用；此项控制用户端是否展示。", }, + { + key: "chat_max_active_requests", + label: "问答最大并发数", + type: "number", + defaultValue: 2, + min: 1, + max: 1000, + description: "同一后端进程内同时进入飞书和模型生成链路的最大请求数。", + }, + { + key: "chat_max_queue_size", + label: "问答最大排队数", + type: "number", + defaultValue: 20, + min: 0, + max: 10000, + description: "超过最大并发后允许等待的请求数量，超过后直接提示稍后再试。", + }, + { + key: "chat_queue_timeout_seconds", + label: "问答排队超时（秒）", + type: "number", + defaultValue: 60, + min: 1, + max: 3600, + description: "请求在排队中超过该时间后自动结束并提示用户稍后再试。", + }, ], }, { diff --git a/ai_knowledge_base_v2/apps/backend/.env.example b/ai_knowledge_base_v2/apps/backend/.env.example index 5978034..916c55a 100644 --- a/ai_knowledge_base_v2/apps/backend/.env.example +++ b/ai_knowledge_base_v2/apps/backend/.env.example @@ -24,6 +24,9 @@ FEISHU_RETRY_COUNT=2 DEFAULT_DAILY_CHAT_LIMIT=100 DEFAULT_USER_NAME_PREFIX=用户 +CHAT_MAX_ACTIVE_REQUESTS=2 +CHAT_MAX_QUEUE_SIZE=20 +CHAT_QUEUE_TIMEOUT_SECONDS=60 BOOTSTRAP_ADMIN_USERNAME=admin BOOTSTRAP_ADMIN_PASSWORD=admin123456 diff --git a/ai_knowledge_base_v2/apps/backend/app/api/chat.py b/ai_knowledge_base_v2/apps/backend/app/api/chat.py index aa6f37f..d497f33 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/chat.py +++ b/ai_knowledge_base_v2/apps/backend/app/api/chat.py @@ -3,7 +3,7 @@ from __future__ import annotations import json from collections.abc import Iterator -from fastapi import APIRouter, Depends, Query +from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import StreamingResponse from sqlalchemy.orm import Session @@ -20,6 +20,7 @@ from app.schemas.chat import ( UpdateSessionTitleRequest, ) from app.services.chat_service import ChatService +from app.services.chat_queue_service import chat_queue_manager, load_chat_queue_config router = APIRouter() @@ -72,8 +73,7 @@ def completions( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ) -> StreamingResponse: - answer = ChatService.create_answer(db, current_user, payload.sessionId, payload.message) - return StreamingResponse(_sse_chunks(answer), media_type="text/event-stream") + return StreamingResponse(_chat_stream(payload, db, current_user), media_type="text/event-stream") @router.post("/stop") @@ -86,9 +86,69 @@ def stop( return api_success() +def _chat_stream(payload: ChatCompletionRequest, db: Session, current_user: User) -> Iterator[str]: + config = load_chat_queue_config(db) + queue_request = chat_queue_manager.request_slot(config) + acquired = queue_request.status == "acquired" + queued_token = queue_request.token + + if queue_request.status == "rejected": + yield _sse_event( + "error", + message="当前请求过多，请稍后再试。", + activeCount=queue_request.active_count, + waitingCount=queue_request.waiting_count, + ) + yield _sse_done() + return + + if queue_request.status == "queued" and queued_token is not None: + yield _sse_event( + "queued", + message=f"当前请求较多，正在排队中，前方约 {max(queue_request.position - 1, 0)} 个请求。", + position=queue_request.position, + activeCount=queue_request.active_count, + waitingCount=queue_request.waiting_count, + ) + queue_request = chat_queue_manager.wait_for_slot(queued_token, config) + acquired = queue_request.status == "acquired" + if queue_request.status == "timeout": + yield _sse_event("error", message="排队等待超时，请稍后再试。") + yield _sse_done() + return + + try: + yield _sse_event( + "generating", + message="已进入生成队列，正在生成回答。", + activeCount=queue_request.active_count, + waitingCount=queue_request.waiting_count, + ) + answer = ChatService.create_answer(db, current_user, payload.sessionId, payload.message) + yield from _sse_chunks(answer) + except HTTPException as exc: + yield _sse_event("error", message=str(exc.detail)) + except Exception: + yield _sse_event("error", message="AI 回复生成失败，请稍后再试。") + finally: + if acquired: + chat_queue_manager.release_slot() + elif queued_token is not None: + chat_queue_manager.cancel_waiter(queued_token) + + yield _sse_done() + + def _sse_chunks(answer: str) -> Iterator[str]: chunk_size = 12 for index in range(0, len(answer), chunk_size): chunk = answer[index : index + chunk_size] - yield f"data: {json.dumps({'content': chunk}, ensure_ascii=False)}\n\n" - yield "data: [DONE]\n\n" + yield _sse_event("content", content=chunk) + + +def _sse_event(event_type: str, **payload: object) -> str: + return f"data: {json.dumps({'type': event_type, **payload}, ensure_ascii=False)}\n\n" + + +def _sse_done() -> str: + return "data: [DONE]\n\n" diff --git a/ai_knowledge_base_v2/apps/backend/app/core/config.py b/ai_knowledge_base_v2/apps/backend/app/core/config.py index b582a79..305c769 100644 --- a/ai_knowledge_base_v2/apps/backend/app/core/config.py +++ b/ai_knowledge_base_v2/apps/backend/app/core/config.py @@ -45,6 +45,9 @@ class Settings(BaseSettings): default_daily_chat_limit: int = 100 default_user_name_prefix: str = "用户" + chat_max_active_requests: int = 2 + chat_max_queue_size: int = 20 + chat_queue_timeout_seconds: int = 60 bootstrap_admin_username: str = "admin" bootstrap_admin_password: str = "admin123456" bootstrap_admin_name: str = "系统管理员" diff --git a/ai_knowledge_base_v2/apps/backend/app/services/chat_queue_service.py b/ai_knowledge_base_v2/apps/backend/app/services/chat_queue_service.py new file mode 100644 index 0000000..e1f94f1 --- /dev/null +++ b/ai_knowledge_base_v2/apps/backend/app/services/chat_queue_service.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +from collections import deque +from dataclasses import dataclass +from threading import Condition +import time + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.core.config import get_settings +from app.models.ai_config import SystemConfig + + +@dataclass(frozen=True) +class ChatQueueConfig: + max_active_requests: int + max_queue_size: int + queue_timeout_seconds: int + + +@dataclass(frozen=True) +class QueueRequest: + status: str + token: object | None = None + position: int = 0 + active_count: int = 0 + waiting_count: int = 0 + + +class ChatQueueManager: + def __init__(self) -> None: + self._condition = Condition() + self._active_count = 0 + self._waiters: deque[object] = deque() + + def request_slot(self, config: ChatQueueConfig) -> QueueRequest: + with self._condition: + if self._active_count < config.max_active_requests and not self._waiters: + self._active_count += 1 + return QueueRequest( + status="acquired", + active_count=self._active_count, + waiting_count=len(self._waiters), + ) + + if len(self._waiters) >= config.max_queue_size: + return QueueRequest( + status="rejected", + active_count=self._active_count, + waiting_count=len(self._waiters), + ) + + token = object() + self._waiters.append(token) + return QueueRequest( + status="queued", + token=token, + position=len(self._waiters), + active_count=self._active_count, + waiting_count=len(self._waiters), + ) + + def wait_for_slot(self, token: object, config: ChatQueueConfig) -> QueueRequest: + deadline = time.monotonic() + config.queue_timeout_seconds + with self._condition: + while True: + if self._waiters and self._waiters[0] is token and self._active_count < config.max_active_requests: + self._waiters.popleft() + self._active_count += 1 + self._condition.notify_all() + return QueueRequest( + status="acquired", + active_count=self._active_count, + waiting_count=len(self._waiters), + ) + + remaining = deadline - time.monotonic() + if remaining <= 0: + try: + self._waiters.remove(token) + except ValueError: + pass + self._condition.notify_all() + return QueueRequest( + status="timeout", + active_count=self._active_count, + waiting_count=len(self._waiters), + ) + + self._condition.wait(timeout=remaining) + + def cancel_waiter(self, token: object) -> None: + with self._condition: + try: + self._waiters.remove(token) + except ValueError: + return + self._condition.notify_all() + + def release_slot(self) -> None: + with self._condition: + if self._active_count > 0: + self._active_count -= 1 + self._condition.notify_all() + + +chat_queue_manager = ChatQueueManager() + + +def load_chat_queue_config(db: Session) -> ChatQueueConfig: + settings = get_settings() + return ChatQueueConfig( + max_active_requests=_config_int( + db, + "chat_max_active_requests", + settings.chat_max_active_requests, + minimum=1, + maximum=1000, + ), + max_queue_size=_config_int( + db, + "chat_max_queue_size", + settings.chat_max_queue_size, + minimum=0, + maximum=10000, + ), + queue_timeout_seconds=_config_int( + db, + "chat_queue_timeout_seconds", + settings.chat_queue_timeout_seconds, + minimum=1, + maximum=3600, + ), + ) + + +def _config_int(db: Session, key: str, default: int, *, minimum: int, maximum: int) -> int: + config = db.scalar(select(SystemConfig).where(SystemConfig.config_key == key)) + if config is None or not config.config_value.strip(): + return default + try: + value = int(float(config.config_value.strip())) + except ValueError: + return default + return max(minimum, min(maximum, value)) diff --git a/ai_knowledge_base_v2/apps/user-client/src/App.vue b/ai_knowledge_base_v2/apps/user-client/src/App.vue index 7a35d70..b6c48f3 100644 --- a/ai_knowledge_base_v2/apps/user-client/src/App.vue +++ b/ai_knowledge_base_v2/apps/user-client/src/App.vue @@ -84,12 +84,30 @@ async function send(message: string) { messages.value.push(userMessage, assistantMessage); loading.value = true; activeAbortController.value = new AbortController(); + let hasContent = false; await scrollToBottom(); try { - await streamChat(activeSessionId.value, message, async (chunk) => { - assistantMessage.content += chunk; - await scrollToBottom(); - }, activeAbortController.value.signal); + await streamChat( + activeSessionId.value, + message, + async (chunk) => { + if (!hasContent) { + assistantMessage.content = ""; + hasContent = true; + } + assistantMessage.content += chunk; + await scrollToBottom(); + }, + async (statusMessage, statusType) => { + if (statusType === "queued") { + assistantMessage.content = statusMessage || "当前请求较多，正在排队中。"; + } else if (statusType === "generating" && !hasContent) { + assistantMessage.content = ""; + } + await scrollToBottom(); + }, + activeAbortController.value.signal, + ); assistantMessage.streaming = false; await refreshSessions(activeSessionId.value); user.value = await api.profile(); diff --git a/ai_knowledge_base_v2/apps/user-client/src/components/ChatMessage.vue b/ai_knowledge_base_v2/apps/user-client/src/components/ChatMessage.vue index fa63177..95d57d5 100644 --- a/ai_knowledge_base_v2/apps/user-client/src/components/ChatMessage.vue +++ b/ai_knowledge_base_v2/apps/user-client/src/components/ChatMessage.vue @@ -18,24 +18,36 @@ const assistantParts = computed(() => { if (props.role === "user") { return { reasoning: "", answer: props.content }; } - const matched = props.content.match(/([\s\S]*?)<\/think>/i); - if (matched) { - return { - reasoning: matched[1].trim(), - answer: props.content.replace(matched[0], "").trim(), - }; - } - const opening = props.content.match(/([\s\S]*)$/i); - if (opening) { - return { - reasoning: opening[1].trim(), - answer: props.content.slice(0, opening.index).trim(), - }; - } - return { reasoning: "", answer: props.content }; -}); -const renderedReasoning = computed(() => markdown.render(assistantParts.value.reasoning || "")); + const content = props.content; + let reasoning = ""; + let answer = content; + + // 匹配所有完整的 ...（全局匹配，支持多个） + const thinkRegex = /([\s\S]*?)<\/think>/gi; + const matches = Array.from(answer.matchAll(thinkRegex)); + + if (matches.length > 0) { + reasoning = matches.map((m) => m[1].trim()).join("\n\n"); + answer = answer.replace(thinkRegex, "").trim(); + } + + // 处理未闭合的 （流式输出时可能还没收到 ） + const openThink = answer.match(/([\s\S]*)$/i); + if (openThink) { + reasoning += (reasoning ? "\n\n" : "") + openThink[1].trim(); + answer = answer.slice(0, openThink.index).trim(); + } + + // 兜底：如果 answer 为空但 reasoning 有内容，说明模型可能把全部内容放在 think 里了 + // 此时把 reasoning 当 answer 显示，不显示 thinking 面板 + if (!answer && reasoning) { + answer = reasoning; + reasoning = ""; + } + + return { reasoning, answer }; +}); const renderedContent = computed(() => { if (props.role === "user") return props.content; @@ -48,10 +60,9 @@ const renderedContent = computed(() => {
{{ role === "assistant" ? "AI" : "我" }}
{{ renderedContent }}
diff --git a/ai_knowledge_base_v2/apps/user-client/src/services/api.ts b/ai_knowledge_base_v2/apps/user-client/src/services/api.ts index 04c5041..a355e0a 100644 --- a/ai_knowledge_base_v2/apps/user-client/src/services/api.ts +++ b/ai_knowledge_base_v2/apps/user-client/src/services/api.ts @@ -49,6 +49,7 @@ export async function streamChat( sessionId: number, message: string, onChunk: (chunk: string) => void, + onStatus?: (message: string, type: string) => void, signal?: AbortSignal, ) { const token = getToken(); @@ -79,7 +80,14 @@ export async function streamChat( if (!line.startsWith("data:")) continue; const payload = line.slice(5).trim(); if (payload === "[DONE]") return; - const parsed = JSON.parse(payload) as { content?: string }; + const parsed = JSON.parse(payload) as { type?: string; content?: string; message?: string }; + if (parsed.type === "queued" || parsed.type === "generating") { + onStatus?.(parsed.message || "", parsed.type); + continue; + } + if (parsed.type === "error") { + throw new Error(parsed.message || "AI 回复失败"); + } if (parsed.content) onChunk(parsed.content); } } diff --git a/ai_knowledge_base_v2/apps/user-client/src/styles.css b/ai_knowledge_base_v2/apps/user-client/src/styles.css index 5d1e907..2d0da5f 100644 --- a/ai_knowledge_base_v2/apps/user-client/src/styles.css +++ b/ai_knowledge_base_v2/apps/user-client/src/styles.css @@ -393,6 +393,31 @@ button:disabled { font-size: 12px; } +.thinking-indicator { + margin-bottom: 10px; + padding: 6px 10px; + border-radius: 8px; + background: #f2f7f5; + color: #5d7169; + font-size: 13px; +} + +.thinking-dots::after { + content: ""; + display: inline-block; + width: 3px; + height: 3px; + margin-left: 4px; + border-radius: 50%; + background: #0f8b6f; + animation: thinking-dot 1.2s infinite; +} + +@keyframes thinking-dot { + 0%, 100% { opacity: 0.2; } + 50% { opacity: 1; } +} + .composer { position: absolute; left: 0; diff --git a/ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md b/ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md index 6d375c2..b822a90 100644 --- a/ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md +++ b/ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md @@ -124,3 +124,38 @@ 2. 当前模型和飞书外部 API 仍可能限流，排队只能保护本系统，不能突破外部限额。 3. 排队请求持有 SSE 连接，网关和浏览器超时时间要一起配置。 4. 真实流式前，用户端虽然能看到排队状态，但进入生成后仍要等完整模型结果才能输出内容。 + +## 阶段 1 实施记录 + +2026-07-08 开始实现第一阶段“并发准入与排队兜底”。 + +已完成： + +1. 后端新增进程内问答队列管理器，支持最大生成并发、最大等待队列和排队超时。 +2. `/chat/completions` 改为立即返回 SSE 流，并通过 SSE 事件通知用户端： + - `queued`：请求进入排队。 + - `generating`：请求获得执行名额。 + - `content`：回答内容片段。 + - `error`：队列满、排队超时或生成失败。 +3. 系统设置新增三个配置项： + - `chat_max_active_requests` + - `chat_max_queue_size` + - `chat_queue_timeout_seconds` +4. 用户端 H5 支持排队状态展示，排队时在 AI 气泡中显示等待提示，进入生成后切换到正常回答。 +5. 队列满时返回“当前请求过多，请稍后再试”，避免请求无限堆积。 + +阶段限制： + +1. 当前队列是单进程内存队列，适合当前开发环境和单 worker 兜底。 +2. 多 worker / 多实例部署前，必须升级为 Redis 全局队列，否则每个 worker 只会管理自己的排队状态。 +3. 当前仍不是真模型流式，进入生成后首字时间仍取决于飞书检索和模型完整返回时间。 + +验证记录： + +1. 后端 `compileall` 检查通过。 +2. 用户端 H5 `npm run build` 通过。 +3. 管理后台 `npm run build` 通过；仅存在第三方依赖 Rollup 注释告警，不影响构建。 +4. `git diff --check` 通过。 +5. Docker dev 环境已重新构建并启动：backend、user-client、admin-web、mysql 均正常运行。 +6. 队列管理器验证通过：`max_active_requests=1`、`max_queue_size=1` 时，第一个请求获得执行名额，第二个请求进入排队，第三个请求被拒绝，释放名额后第二个请求恢复执行。 +7. 真实 `/chat/completions` SSE 验证通过：接口先返回 `generating` 事件，随后返回连续 `content` 事件。