From a589a25bdc5569021a4422896f8cd9bc2f79e811 Mon Sep 17 00:00:00 2001
From: Nelson <1475262689@qq.com>
Date: Fri, 31 Jul 2026 16:50:11 +0800
Subject: [PATCH] feat: persist cleaned question insights
---
.../apps/admin-web/src/App.vue | 29 +-
.../apps/admin-web/src/services/api.ts | 5 +
.../apps/admin-web/src/types/api.ts | 10 +
.../0020_question_insight_persistence.py | 104 +++++++
.../apps/backend/app/api/admin_records.py | 25 ++
.../apps/backend/app/models/__init__.py | 2 +
.../apps/backend/app/models/insight.py | 55 ++++
.../app/services/question_insight_service.py | 271 +++++++++++++++++-
.../backend/tests/test_admin_pagination.py | 23 +-
.../docs/qianwen_product_todo.md | 2 +
10 files changed, 505 insertions(+), 21 deletions(-)
create mode 100644 ai_knowledge_base_v2/apps/backend/alembic/versions/0020_question_insight_persistence.py
create mode 100644 ai_knowledge_base_v2/apps/backend/app/models/insight.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 8d61ea2..8be6539 100644
--- a/ai_knowledge_base_v2/apps/admin-web/src/App.vue
+++ b/ai_knowledge_base_v2/apps/admin-web/src/App.vue
@@ -927,7 +927,26 @@ async function resetChatFilters() {
async function searchQuestionInsights() {
pagers.questionInsights.page = 1;
- await loadRecordTab("questionInsights");
+ loading.value = true;
+ try {
+ const refresh = await api.refreshQuestionInsights({
+ dateFrom: formatRecordDateTime(questionInsightFilters.dateFrom, "start"),
+ dateTo: formatRecordDateTime(questionInsightFilters.dateTo, "end"),
+ maxMessages: questionInsightFilters.maxMessages,
+ });
+ await loadQuestionInsights(1, pagers.questionInsights.pageSize);
+ if (refresh.processedMessages > 0) {
+ ElMessage.success(
+ `新增清洗 ${refresh.processedMessages} 条消息，得到 ${refresh.acceptedQuestions} 个有效问题${refresh.hasMore ? "；仍有历史消息待清洗" : ""}`,
+ );
+ } else {
+ ElMessage.success("没有新增消息，已直接使用持久化清洗结果统计");
+ }
+ } catch (error) {
+ ElMessage.error(error instanceof Error ? error.message : "问题洞察刷新失败");
+ } finally {
+ loading.value = false;
+ }
}
async function resetQuestionInsightFilters() {
@@ -1555,17 +1574,17 @@ function formatRecordDateTime(value: string, boundary: "start" | "end") {
- 统计问题
+ 刷新并统计
重置
- 一期先对用户消息做去噪、拆问、同义词归一和相似问法合并；后续可把清洗后的结果交给大模型做更细的主题命名。
+ 清洗结果会持久化保存；点击“刷新并统计”只处理尚未清洗的用户消息，再按当前时间范围聚合，翻页不会重新扫描聊天原文。
- 扫描用户消息{{ questionInsights.summary.scannedMessages }}
+ 纳入清洗消息{{ questionInsights.summary.scannedMessages }}
有效问题{{ questionInsights.summary.cleanedQuestions }}
过滤低价值{{ questionInsights.summary.filteredMessages }}
高频问题组{{ questionInsights.summary.visibleClusterCount }}
diff --git a/ai_knowledge_base_v2/apps/admin-web/src/services/api.ts b/ai_knowledge_base_v2/apps/admin-web/src/services/api.ts
index b2dc62e..a8c6f10 100644
--- a/ai_knowledge_base_v2/apps/admin-web/src/services/api.ts
+++ b/ai_knowledge_base_v2/apps/admin-web/src/services/api.ts
@@ -32,6 +32,7 @@ import type {
PeriodicReportRecord,
PromptDetail,
PromptHistoryItem,
+ QuestionInsightRefreshResult,
QuestionInsightSummary,
} from "../types/api";
@@ -230,6 +231,10 @@ export const api = {
operationLogs: (query: { module?: string; page?: number; pageSize?: number } = {}) => request>>(`/admin/log/list${queryString(query)}`),
questionInsights: (query: { dateFrom?: string; dateTo?: string; minCount?: number; maxMessages?: number; page?: number; pageSize?: number } = {}) =>
request(`/admin/question-insights/summary${queryString(query)}`),
+ refreshQuestionInsights: (query: { dateFrom?: string; dateTo?: string; maxMessages?: number } = {}) =>
+ request(`/admin/question-insights/refresh${queryString(query)}`, {
+ method: "POST",
+ }),
retrievalLogs: (query: { page?: number; pageSize?: number } = {}) => request>(`/admin/retrieval-log/list${queryString(query)}`),
estimateRetrievalCleanup: (before: string) => request<{ before: string; estimatedCount: number }>("/admin/retrieval-log/cleanup/estimate", { method: "POST", body: JSON.stringify({ before }) }),
cleanupRetrievalLogs: (before: string) => request<{ before: string; deleted: number }>("/admin/retrieval-log/cleanup", { method: "POST", body: JSON.stringify({ before }) }),
diff --git a/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts b/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts
index e91f81c..79f01de 100644
--- a/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts
+++ b/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts
@@ -556,6 +556,7 @@ export interface QuestionInsightSummary {
clusterCount: number;
visibleClusterCount: number;
minCount: number;
+ cleanerVersion: string;
};
items: QuestionInsightCluster[];
total: number;
@@ -563,6 +564,15 @@ export interface QuestionInsightSummary {
pageSize: number;
}
+export interface QuestionInsightRefreshResult {
+ processedMessages: number;
+ acceptedQuestions: number;
+ filteredMessages: number;
+ concurrentSkips: number;
+ hasMore: boolean;
+ cleanerVersion: string;
+}
+
export interface QuestionInsightCluster {
rank: number;
title: string;
diff --git a/ai_knowledge_base_v2/apps/backend/alembic/versions/0020_question_insight_persistence.py b/ai_knowledge_base_v2/apps/backend/alembic/versions/0020_question_insight_persistence.py
new file mode 100644
index 0000000..623c420
--- /dev/null
+++ b/ai_knowledge_base_v2/apps/backend/alembic/versions/0020_question_insight_persistence.py
@@ -0,0 +1,104 @@
+"""persist cleaned question insights
+
+Revision ID: 0020_question_insight_persistence
+Revises: 0019_periodic_reports
+Create Date: 2026-07-31 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0020_question_insight_persistence"
+down_revision = "0019_periodic_reports"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ op.create_table(
+ "sys_question_insight_cleaned_question",
+ sa.Column(
+ "id",
+ sa.BigInteger().with_variant(sa.Integer(), "sqlite"),
+ primary_key=True,
+ autoincrement=True,
+ ),
+ sa.Column("message_id", sa.BigInteger(), nullable=False),
+ sa.Column("session_id", sa.BigInteger(), nullable=False),
+ sa.Column("user_id", sa.BigInteger(), nullable=False),
+ sa.Column("part_index", sa.Integer(), nullable=False),
+ sa.Column("cleaner_version", sa.String(length=20), nullable=False),
+ sa.Column("source_hash", sa.String(length=64), nullable=False),
+ sa.Column("cleaned_text", sa.Text(), nullable=False),
+ sa.Column("normalized_text", sa.Text(), nullable=False),
+ sa.Column("category", sa.String(length=50), nullable=False, server_default="other"),
+ sa.Column("tokens_json", sa.Text(), nullable=False),
+ sa.Column("accepted", sa.Integer(), nullable=False, server_default="1"),
+ sa.Column("filtered_reason", sa.String(length=100), nullable=True),
+ sa.Column("source_created_at", sa.DateTime(), nullable=False),
+ sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+ sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+ sa.ForeignKeyConstraint(
+ ["message_id"],
+ ["sys_chat_message.id"],
+ ondelete="CASCADE",
+ ),
+ sa.UniqueConstraint(
+ "message_id",
+ "cleaner_version",
+ "part_index",
+ name="uq_question_insight_message_version_part",
+ ),
+ )
+ op.create_index(
+ "ix_sys_question_insight_cleaned_question_session_id",
+ "sys_question_insight_cleaned_question",
+ ["session_id"],
+ )
+ op.create_index(
+ "ix_sys_question_insight_cleaned_question_user_id",
+ "sys_question_insight_cleaned_question",
+ ["user_id"],
+ )
+ op.create_index(
+ "ix_question_insight_version_accepted_created",
+ "sys_question_insight_cleaned_question",
+ ["cleaner_version", "accepted", "source_created_at"],
+ )
+ op.create_index(
+ "ix_question_insight_category_created",
+ "sys_question_insight_cleaned_question",
+ ["category", "source_created_at"],
+ )
+ op.create_index(
+ "ix_question_insight_session_created",
+ "sys_question_insight_cleaned_question",
+ ["session_id", "source_created_at"],
+ )
+
+
+def downgrade() -> None:
+ op.drop_index(
+ "ix_question_insight_session_created",
+ table_name="sys_question_insight_cleaned_question",
+ )
+ op.drop_index(
+ "ix_question_insight_category_created",
+ table_name="sys_question_insight_cleaned_question",
+ )
+ op.drop_index(
+ "ix_question_insight_version_accepted_created",
+ table_name="sys_question_insight_cleaned_question",
+ )
+ op.drop_index(
+ "ix_sys_question_insight_cleaned_question_user_id",
+ table_name="sys_question_insight_cleaned_question",
+ )
+ op.drop_index(
+ "ix_sys_question_insight_cleaned_question_session_id",
+ table_name="sys_question_insight_cleaned_question",
+ )
+ op.drop_table("sys_question_insight_cleaned_question")
diff --git a/ai_knowledge_base_v2/apps/backend/app/api/admin_records.py b/ai_knowledge_base_v2/apps/backend/app/api/admin_records.py
index fe2605d..fc7bec5 100644
--- a/ai_knowledge_base_v2/apps/backend/app/api/admin_records.py
+++ b/ai_knowledge_base_v2/apps/backend/app/api/admin_records.py
@@ -20,6 +20,7 @@ from app.models.growth import ShareDraft, TeacherHelpCard, TopicSummary
from app.models.logs import AiRequestLog, OperationLog
from app.models.user import User
from app.api.pagination import page_result
+from app.services.admin_service import OperationLogService
from app.services.question_insight_service import QuestionInsightService
from app.services.growth_profile_service import topic_dict, topic_summary_dict
from app.services.help_card_service import help_card_dict
@@ -275,6 +276,30 @@ def question_insights(
)
+@router.post("/question-insights/refresh")
+def refresh_question_insights(
+ dateFrom: datetime | None = Query(default=None),
+ dateTo: datetime | None = Query(default=None),
+ maxMessages: int = Query(default=5000, ge=100, le=20000),
+ db: Session = Depends(get_db),
+ current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+ result = QuestionInsightService.refresh(
+ db,
+ date_from=dateFrom,
+ date_to=dateTo,
+ max_messages=maxMessages,
+ )
+ OperationLogService.write(
+ db,
+ admin_id=current_admin.id,
+ module="question_insight",
+ action=f"refresh:{result['processedMessages']}",
+ )
+ db.commit()
+ return api_success(result)
+
+
def _chat_query(
*,
keyword: str,
diff --git a/ai_knowledge_base_v2/apps/backend/app/models/__init__.py b/ai_knowledge_base_v2/apps/backend/app/models/__init__.py
index bb2e634..f9841e3 100644
--- a/ai_knowledge_base_v2/apps/backend/app/models/__init__.py
+++ b/ai_knowledge_base_v2/apps/backend/app/models/__init__.py
@@ -4,6 +4,7 @@ from app.models.base import Base
from app.models.chat import ChatMessage, ChatSession, TopicSession
from app.models.entitlement import EntitlementPlan, UserEntitlement, UserEntitlementLog
from app.models.growth import GrowthProfileRevision, PeriodicReport, ShareDraft, TeacherHelpCard, TopicSummary, UserGrowthProfile
+from app.models.insight import QuestionInsightCleanedQuestion
from app.models.knowledge import (
HumanAttentionHistory,
HumanAttentionRecord,
@@ -54,6 +55,7 @@ __all__ = [
"TopicSession",
"TopicSummary",
"Prompt",
+ "QuestionInsightCleanedQuestion",
"Role",
"SystemConfig",
"ShareDraft",
diff --git a/ai_knowledge_base_v2/apps/backend/app/models/insight.py b/ai_knowledge_base_v2/apps/backend/app/models/insight.py
new file mode 100644
index 0000000..d36610d
--- /dev/null
+++ b/ai_knowledge_base_v2/apps/backend/app/models/insight.py
@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.models.base import Base
+
+
+class QuestionInsightCleanedQuestion(Base):
+ """用户问题清洗后的持久化结果。
+
+ 每个原始消息至少写入一条记录。没有有效问题的消息会写入 accepted=0
+ 的占位记录，使增量清洗无需重复读取已经处理过的聊天原文。
+ """
+
+ __tablename__ = "sys_question_insight_cleaned_question"
+ __table_args__ = (
+ UniqueConstraint(
+ "message_id",
+ "cleaner_version",
+ "part_index",
+ name="uq_question_insight_message_version_part",
+ ),
+ )
+
+ id: Mapped[int] = mapped_column(
+ BigInteger().with_variant(Integer, "sqlite"),
+ primary_key=True,
+ autoincrement=True,
+ )
+ message_id: Mapped[int] = mapped_column(
+ ForeignKey("sys_chat_message.id", ondelete="CASCADE"),
+ nullable=False,
+ )
+ session_id: Mapped[int] = mapped_column(BigInteger, index=True, nullable=False)
+ user_id: Mapped[int] = mapped_column(BigInteger, index=True, nullable=False)
+ part_index: Mapped[int] = mapped_column(Integer, nullable=False)
+ cleaner_version: Mapped[str] = mapped_column(String(20), nullable=False)
+ source_hash: Mapped[str] = mapped_column(String(64), nullable=False)
+ cleaned_text: Mapped[str] = mapped_column(Text, nullable=False, default="")
+ normalized_text: Mapped[str] = mapped_column(Text, nullable=False, default="")
+ category: Mapped[str] = mapped_column(String(50), nullable=False, default="other")
+ tokens_json: Mapped[str] = mapped_column(Text, nullable=False, default="[]")
+ accepted: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+ filtered_reason: Mapped[str | None] = mapped_column(String(100), nullable=True)
+ source_created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
+ created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), nullable=False)
+ updated_at: Mapped[datetime] = mapped_column(
+ DateTime,
+ server_default=func.now(),
+ onupdate=func.now(),
+ nullable=False,
+ )
diff --git a/ai_knowledge_base_v2/apps/backend/app/services/question_insight_service.py b/ai_knowledge_base_v2/apps/backend/app/services/question_insight_service.py
index 73bcb8a..9f584a5 100644
--- a/ai_knowledge_base_v2/apps/backend/app/services/question_insight_service.py
+++ b/ai_knowledge_base_v2/apps/backend/app/services/question_insight_service.py
@@ -1,5 +1,7 @@
from __future__ import annotations
+import hashlib
+import json
import re
from collections import Counter
from dataclasses import dataclass, field
@@ -7,14 +9,18 @@ from datetime import datetime
from difflib import SequenceMatcher
from typing import Iterable
-from sqlalchemy import select
+from sqlalchemy import and_, func, select
+from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.models.chat import ChatMessage, ChatSession
+from app.models.insight import QuestionInsightCleanedQuestion
from app.models.logs import AiRequestLog
from app.models.user import User
+CLEANER_VERSION = "v1"
+
LOW_VALUE_EXACT = {
"你好",
"您好",
@@ -103,6 +109,60 @@ class QuestionCluster:
class QuestionInsightService:
+ @staticmethod
+ def refresh(
+ db: Session,
+ *,
+ date_from: datetime | None = None,
+ date_to: datetime | None = None,
+ max_messages: int = 5000,
+ ) -> dict:
+ """增量清洗尚未处理的用户消息，并把结果写入持久化清洗表。"""
+
+ messages = _load_unprocessed_user_messages(
+ db,
+ date_from=date_from,
+ date_to=date_to,
+ limit=max_messages,
+ )
+ processed_messages = 0
+ accepted_questions = 0
+ filtered_messages = 0
+ concurrent_skips = 0
+
+ for row in messages:
+ message = row[0]
+ cleaned, filtered_count = _clean_messages([row])
+ records = _cleaned_records(message, cleaned, filtered_count)
+ try:
+ # 多个管理员同时刷新时，唯一约束负责去重；单条消息冲突不会回滚整批。
+ with db.begin_nested():
+ db.add_all(records)
+ db.flush()
+ except IntegrityError:
+ concurrent_skips += 1
+ continue
+ processed_messages += 1
+ accepted_questions += len(cleaned)
+ filtered_messages += filtered_count
+
+ has_more = bool(
+ _load_unprocessed_user_messages(
+ db,
+ date_from=date_from,
+ date_to=date_to,
+ limit=1,
+ )
+ )
+ return {
+ "processedMessages": processed_messages,
+ "acceptedQuestions": accepted_questions,
+ "filteredMessages": filtered_messages,
+ "concurrentSkips": concurrent_skips,
+ "hasMore": has_more,
+ "cleanerVersion": CLEANER_VERSION,
+ }
+
@staticmethod
def summarize(
db: Session,
@@ -114,8 +174,16 @@ class QuestionInsightService:
page_size: int = 20,
max_messages: int = 5000,
) -> dict:
- messages = _load_user_messages(db, date_from=date_from, date_to=date_to, limit=max_messages)
- cleaned, filtered_count = _clean_messages(messages)
+ persisted_items = _load_persisted_cleaned_questions(
+ db,
+ date_from=date_from,
+ date_to=date_to,
+ limit=max_messages,
+ )
+ cleaned = [_question_from_persisted(item) for item in persisted_items if item.accepted]
+ source_message_ids = {item.message_id for item in persisted_items}
+ accepted_message_ids = {item.message_id for item in persisted_items if item.accepted}
+ filtered_count = len(source_message_ids - accepted_message_ids)
clusters = _cluster_questions(cleaned)
ai_logs = _load_ai_logs(db, date_from=date_from, date_to=date_to, limit=max_messages)
visible_clusters = [cluster for cluster in clusters if len(cluster.questions) >= min_count]
@@ -124,6 +192,7 @@ class QuestionInsightService:
total = len(visible_clusters)
offset = (page - 1) * page_size
page_clusters = visible_clusters[offset : offset + page_size]
+ message_contents, users = _load_sample_context(db, page_clusters)
return {
"range": {
@@ -132,21 +201,31 @@ class QuestionInsightService:
"maxMessages": max_messages,
},
"summary": {
- "scannedMessages": len(messages),
+ "scannedMessages": len(source_message_ids),
"cleanedQuestions": len(cleaned),
"filteredMessages": filtered_count,
"clusterCount": len(clusters),
"visibleClusterCount": total,
"minCount": min_count,
+ "cleanerVersion": CLEANER_VERSION,
},
- "items": [_cluster_dict(index + offset + 1, cluster, ai_logs) for index, cluster in enumerate(page_clusters)],
+ "items": [
+ _cluster_dict(
+ index + offset + 1,
+ cluster,
+ ai_logs,
+ message_contents=message_contents,
+ users=users,
+ )
+ for index, cluster in enumerate(page_clusters)
+ ],
"total": total,
"page": page,
"pageSize": page_size,
}
-def _load_user_messages(
+def _load_unprocessed_user_messages(
db: Session,
*,
date_from: datetime | None,
@@ -155,9 +234,19 @@ def _load_user_messages(
) -> list[tuple[ChatMessage, ChatSession | None, User | None]]:
query = (
select(ChatMessage, ChatSession, User)
+ .outerjoin(
+ QuestionInsightCleanedQuestion,
+ and_(
+ QuestionInsightCleanedQuestion.message_id == ChatMessage.id,
+ QuestionInsightCleanedQuestion.cleaner_version == CLEANER_VERSION,
+ ),
+ )
.join(ChatSession, ChatSession.id == ChatMessage.session_id, isouter=True)
.join(User, User.id == ChatMessage.user_id, isouter=True)
- .where(ChatMessage.role == "user")
+ .where(
+ ChatMessage.role == "user",
+ QuestionInsightCleanedQuestion.id.is_(None),
+ )
)
if date_from is not None:
query = query.where(ChatMessage.created_at >= date_from.replace(tzinfo=None))
@@ -170,6 +259,50 @@ def _load_user_messages(
)
+def _load_persisted_cleaned_questions(
+ db: Session,
+ *,
+ date_from: datetime | None,
+ date_to: datetime | None,
+ limit: int,
+) -> list[QuestionInsightCleanedQuestion]:
+ range_filters = [QuestionInsightCleanedQuestion.cleaner_version == CLEANER_VERSION]
+ if date_from is not None:
+ range_filters.append(QuestionInsightCleanedQuestion.source_created_at >= date_from.replace(tzinfo=None))
+ if date_to is not None:
+ range_filters.append(QuestionInsightCleanedQuestion.source_created_at <= date_to.replace(tzinfo=None))
+
+ latest_messages = (
+ select(
+ QuestionInsightCleanedQuestion.message_id.label("message_id"),
+ func.max(QuestionInsightCleanedQuestion.source_created_at).label("latest_at"),
+ )
+ .where(*range_filters)
+ .group_by(QuestionInsightCleanedQuestion.message_id)
+ .order_by(
+ func.max(QuestionInsightCleanedQuestion.source_created_at).desc(),
+ QuestionInsightCleanedQuestion.message_id.desc(),
+ )
+ .limit(limit)
+ .subquery()
+ )
+ return list(
+ db.scalars(
+ select(QuestionInsightCleanedQuestion)
+ .join(
+ latest_messages,
+ QuestionInsightCleanedQuestion.message_id == latest_messages.c.message_id,
+ )
+ .where(QuestionInsightCleanedQuestion.cleaner_version == CLEANER_VERSION)
+ .order_by(
+ QuestionInsightCleanedQuestion.source_created_at.desc(),
+ QuestionInsightCleanedQuestion.message_id.desc(),
+ QuestionInsightCleanedQuestion.part_index.asc(),
+ )
+ ).all()
+ )
+
+
def _load_ai_logs(
db: Session,
*,
@@ -223,6 +356,80 @@ def _clean_messages(messages: Iterable[tuple[ChatMessage, ChatSession | None, Us
return cleaned, filtered_count
+def _cleaned_records(
+ message: ChatMessage,
+ cleaned: list[CleanedQuestion],
+ filtered_count: int,
+) -> list[QuestionInsightCleanedQuestion]:
+ source_hash = hashlib.sha256((message.content or "").encode("utf-8")).hexdigest()
+ if filtered_count:
+ return [
+ QuestionInsightCleanedQuestion(
+ message_id=message.id,
+ session_id=message.session_id,
+ user_id=message.user_id,
+ part_index=-1,
+ cleaner_version=CLEANER_VERSION,
+ source_hash=source_hash,
+ cleaned_text="",
+ normalized_text="",
+ category="filtered",
+ tokens_json="[]",
+ accepted=0,
+ filtered_reason="低价值或无有效问题",
+ source_created_at=message.created_at,
+ )
+ ]
+
+ records: list[QuestionInsightCleanedQuestion] = []
+ for part_index, question in enumerate(cleaned):
+ category, _category_label = _classify_text(f"{question.text}{question.normalized}")
+ records.append(
+ QuestionInsightCleanedQuestion(
+ message_id=message.id,
+ session_id=message.session_id,
+ user_id=message.user_id,
+ part_index=part_index,
+ cleaner_version=CLEANER_VERSION,
+ source_hash=source_hash,
+ cleaned_text=question.text,
+ normalized_text=question.normalized,
+ category=category,
+ tokens_json=json.dumps(sorted(question.tokens), ensure_ascii=False),
+ accepted=1,
+ filtered_reason=None,
+ source_created_at=message.created_at,
+ )
+ )
+ return records
+
+
+def _question_from_persisted(item: QuestionInsightCleanedQuestion) -> CleanedQuestion:
+ try:
+ decoded_tokens = json.loads(item.tokens_json or "[]")
+ except (TypeError, ValueError, json.JSONDecodeError):
+ decoded_tokens = []
+ tokens = {
+ str(token)
+ for token in decoded_tokens
+ if isinstance(token, str) and token
+ }
+ if not tokens:
+ tokens = _tokens(item.normalized_text)
+ return CleanedQuestion(
+ raw="",
+ text=item.cleaned_text,
+ normalized=item.normalized_text,
+ user_id=item.user_id,
+ user_name="",
+ user_phone="",
+ session_id=item.session_id,
+ message_id=item.message_id,
+ created_at=item.source_created_at,
+ tokens=tokens,
+ )
+
+
def _split_questions(content: str) -> list[str]:
text = (content or "").strip()
if not text:
@@ -334,16 +541,47 @@ def _looks_more_question_like(candidate: str, current: str) -> bool:
return any(marker in candidate for marker in markers) and not any(marker in current for marker in markers)
-def _cluster_dict(rank: int, cluster: QuestionCluster, ai_logs: list[AiRequestLog]) -> dict:
+def _load_sample_context(
+ db: Session,
+ clusters: list[QuestionCluster],
+) -> tuple[dict[int, str], dict[int, User]]:
+ sample_questions: list[CleanedQuestion] = []
+ for cluster in clusters:
+ sample_questions.extend(
+ sorted(cluster.questions, key=lambda item: item.created_at, reverse=True)[:5]
+ )
+ message_ids = {item.message_id for item in sample_questions}
+ user_ids = {item.user_id for item in sample_questions}
+ message_contents = {
+ message_id: content
+ for message_id, content in db.execute(
+ select(ChatMessage.id, ChatMessage.content).where(ChatMessage.id.in_(message_ids))
+ ).all()
+ } if message_ids else {}
+ users = {
+ user.id: user
+ for user in db.scalars(select(User).where(User.id.in_(user_ids))).all()
+ } if user_ids else {}
+ return message_contents, users
+
+
+def _cluster_dict(
+ rank: int,
+ cluster: QuestionCluster,
+ ai_logs: list[AiRequestLog],
+ *,
+ message_contents: dict[int, str],
+ users: dict[int, User],
+) -> dict:
questions = sorted(cluster.questions, key=lambda item: item.created_at, reverse=True)
- users = {item.user_id for item in questions}
+ user_ids = {item.user_id for item in questions}
sessions = {item.session_id for item in questions}
variants = Counter(item.text for item in questions).most_common(6)
terms = _top_terms(questions)
category, category_label = _classify_cluster(cluster)
related_logs = _related_ai_logs(cluster, ai_logs)
no_hit_count = sum(1 for item in related_logs if not item.knowledge_hit)
- failed_count = sum(1 for item in related_logs if item.status != "SUCCESS")
+ failed_count = sum(1 for item in related_logs if (item.status or "").upper() != "SUCCESS")
return {
"rank": rank,
"title": cluster.title,
@@ -351,7 +589,7 @@ def _cluster_dict(rank: int, cluster: QuestionCluster, ai_logs: list[AiRequestLo
"category": category,
"categoryLabel": category_label,
"count": len(questions),
- "userCount": len(users),
+ "userCount": len(user_ids),
"sessionCount": len(sessions),
"aiRequestCount": len(related_logs),
"noHitCount": no_hit_count,
@@ -367,9 +605,9 @@ def _cluster_dict(rank: int, cluster: QuestionCluster, ai_logs: list[AiRequestLo
"messageId": item.message_id,
"sessionId": item.session_id,
"userId": item.user_id,
- "userName": item.user_name,
- "userPhone": item.user_phone,
- "raw": item.raw,
+ "userName": users[item.user_id].name if item.user_id in users else "",
+ "userPhone": users[item.user_id].phone if item.user_id in users else "",
+ "raw": message_contents.get(item.message_id, ""),
"cleaned": item.text,
"createdAt": item.created_at,
}
@@ -388,7 +626,10 @@ def _top_terms(questions: list[CleanedQuestion]) -> list[str]:
def _classify_cluster(cluster: QuestionCluster) -> tuple[str, str]:
- text = f"{cluster.title}{cluster.normalized}"
+ return _classify_text(f"{cluster.title}{cluster.normalized}")
+
+
+def _classify_text(text: str) -> tuple[str, str]:
rules = (
("fixed_info", "固定信息", ("上课安排", "回放", "会议链接", "课程助理", "时间", "链接", "权益", "联系方式", "安排")),
("homework", "功课操作", ("功课", "练习", "作业", "怎么做", "步骤", "操作")),
diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_admin_pagination.py b/ai_knowledge_base_v2/apps/backend/tests/test_admin_pagination.py
index 49a63a8..5ee06e8 100644
--- a/ai_knowledge_base_v2/apps/backend/tests/test_admin_pagination.py
+++ b/ai_knowledge_base_v2/apps/backend/tests/test_admin_pagination.py
@@ -2,12 +2,14 @@ from sqlalchemy import create_engine
from datetime import datetime, timedelta
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
+from types import SimpleNamespace
from app.api.admin_agent_records import attention_list, retrieval_logs
-from app.api.admin_records import ai_logs, chat_detail, chat_messages, question_insights
+from app.api.admin_records import ai_logs, chat_detail, chat_messages, question_insights, refresh_question_insights
from app.api.admin_users import list_users
from app.models import Base
from app.models.chat import ChatMessage, ChatSession
+from app.models.insight import QuestionInsightCleanedQuestion
from app.models.knowledge import HumanAttentionRecord, KnowledgeRetrievalLog
from app.models.logs import AiRequestLog
from app.models.user import User
@@ -129,6 +131,20 @@ def test_question_insights_clean_and_cluster_similar_user_questions():
db.add(AiRequestLog(session_id=1, user_id=1, status="SUCCESS", prompt="用户问题：心光有哪些作业？", knowledge_hit=0))
db.commit()
+ first_refresh = refresh_question_insights(
+ dateFrom=None,
+ dateTo=None,
+ maxMessages=100,
+ db=db,
+ current_admin=SimpleNamespace(id=1),
+ )["data"]
+ second_refresh = refresh_question_insights(
+ dateFrom=None,
+ dateTo=None,
+ maxMessages=100,
+ db=db,
+ current_admin=SimpleNamespace(id=1),
+ )["data"]
response = question_insights(
dateFrom=None,
dateTo=None,
@@ -141,6 +157,11 @@ def test_question_insights_clean_and_cluster_similar_user_questions():
)
data = response["data"]
+ assert first_refresh["processedMessages"] == 4
+ assert first_refresh["acceptedQuestions"] == 4
+ assert first_refresh["filteredMessages"] == 1
+ assert second_refresh["processedMessages"] == 0
+ assert db.query(QuestionInsightCleanedQuestion).count() == 5
assert data["summary"]["scannedMessages"] == 4
assert data["summary"]["filteredMessages"] == 1
assert data["summary"]["cleanedQuestions"] == 4
diff --git a/ai_knowledge_base_v2/docs/qianwen_product_todo.md b/ai_knowledge_base_v2/docs/qianwen_product_todo.md
index 61521c1..2edaa2e 100644
--- a/ai_knowledge_base_v2/docs/qianwen_product_todo.md
+++ b/ai_knowledge_base_v2/docs/qianwen_product_todo.md
@@ -830,6 +830,8 @@ AI 日志增加：
#### 开发进度
- 2026-07-31：二期第一步已在现有统计结果中增加问题分类、关联 AI 请求数、无知识命中次数、请求失败次数、是否需要知识跟进和运营处理建议；后台问题洞察卡片已展示分类标签、无命中/失败标记和建议动作。暂未新增持久化清洗表和人工合并/拆分能力，避免一次性扩大数据模型。
+- 2026-07-31：已新增 `sys_question_insight_cleaned_question` 持久化清洗表和清洗版本字段。后台“刷新并统计”只增量处理尚未清洗的用户消息；低价值消息也会写入过滤占位记录，避免后续反复读取聊天原文；统计、筛选和分页均直接读取清洗结果。并发刷新由数据库唯一约束和单消息事务隔离去重，不会重复沉淀同一消息。
+- 待继续：高频问题一键转知识库补充建议、人工合并/拆分问题组，以及清洗规则升级后的版本重建入口。
#### 验收标准