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

feat: persist cleaned question insights

Browse Source
This commit is contained in:
Nelson
2026-07-31 16:50:11 +08:00
parent 6c03faf10c
commit a589a25bdc
10 changed files with 505 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

271
ai_knowledge_base_v2/apps/backend/app/services/question_insight_service.py
View File

@@ -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", "功课操作", ("功课", "练习", "作业", "怎么做", "步骤", "操作")),