GuangYu/QuestionProject
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement chat queue fallback

Browse Source
This commit is contained in:
Nelson
2026-07-08 16:57:40 +08:00
parent 10b6de0622
commit 6f3bbd3803
10 changed files with 367 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
ai_knowledge_base_v2/apps/admin-web/src/App.vue
View File

@@ -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: "请求在排队中超过该时间后自动结束并提示用户稍后再试。",
},
],
},
{

3
ai_knowledge_base_v2/apps/backend/.env.example
View File

@@ -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

70
ai_knowledge_base_v2/apps/backend/app/api/chat.py
View File

@@ -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"

3
ai_knowledge_base_v2/apps/backend/app/core/config.py
View File

@@ -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 = "系统管理员"

146
ai_knowledge_base_v2/apps/backend/app/services/chat_queue_service.py Normal file
View File

@@ -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))

22
ai_knowledge_base_v2/apps/user-client/src/App.vue
View File

@@ -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) => {
await streamChat(
activeSessionId.value,
message,
async (chunk) => {
if (!hasContent) {
assistantMessage.content = "";
hasContent = true;
}
assistantMessage.content += chunk;
await scrollToBottom();
}, activeAbortController.value.signal);
},
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();

53
ai_knowledge_base_v2/apps/user-client/src/components/ChatMessage.vue
View File

@@ -18,24 +18,36 @@ const assistantParts = computed(() => {
if (props.role === "user") {
return { reasoning: "", answer: props.content };
}
const matched = props.content.match(/<think>([\s\S]*?)<\/think>/i);
if (matched) {
return {
reasoning: matched[1].trim(),
answer: props.content.replace(matched[0], "").trim(),
};
}
const opening = props.content.match(/<think>([\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;
// 匹配所有完整的 <think>...</think>（全局匹配，支持多个）
const thinkRegex = /<think>([\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();
}
// 处理未闭合的 <think>（流式输出时可能还没收到 </think>）
const openThink = answer.match(/<think>([\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(() => {
<div class="avatar">{{ role === "assistant" ? "AI" : "我" }}</div>
<div class="bubble">
<template v-if="role === 'assistant'">
<details v-if="assistantParts.reasoning" class="reasoning-panel" :open="streaming">
<summary>{{ streaming ? "思考中" : "思考过程" }}</summary>
<div class="markdown-content reasoning-content" v-html="renderedReasoning"></div>
</details>
<div v-if="streaming && assistantParts.reasoning" class="thinking-indicator">
<span class="thinking-dots">思考中</span>
</div>
<div class="content markdown-content" v-html="renderedContent"></div>
</template>
<div v-else class="content">{{ renderedContent }}</div>

10
ai_knowledge_base_v2/apps/user-client/src/services/api.ts
View File

@@ -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);
}
}

25
ai_knowledge_base_v2/apps/user-client/src/styles.css
View File

@@ -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;

35
ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md
View File

@@ -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` 事件。