From d2e08ab36c5def194664f85fa58da77c5a4c5fc9 Mon Sep 17 00:00:00 2001 From: Nelson <1475262689@qq.com> Date: Mon, 3 Aug 2026 11:35:58 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=BC=82=E6=AD=A5=E6=B2=89=E6=B7=80?= =?UTF-8?q?=E4=B8=BB=E9=A2=98=E6=91=98=E8=A6=81=E5=92=8C=E6=88=90=E9=95=BF?= =?UTF-8?q?=E6=A1=A3=E6=A1=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ai_knowledge_base_v2/.env.example | 4 + .../apps/admin-web/src/App.vue | 35 ++- .../apps/admin-web/src/services/api.ts | 3 + .../apps/admin-web/src/types/api.ts | 5 + .../apps/backend/.env.example | 4 + .../0024_topic_settlement_async_jobs.py | 59 +++++ .../apps/backend/app/api/admin_users.py | 28 +++ .../apps/backend/app/api/chat.py | 9 + .../apps/backend/app/api/user.py | 7 + .../apps/backend/app/core/config.py | 4 + ai_knowledge_base_v2/apps/backend/app/main.py | 8 +- .../apps/backend/app/models/growth.py | 13 +- .../app/services/growth_profile_service.py | 231 +++++++++++++++--- .../app/services/topic_settlement_worker.py | 217 ++++++++++++++++ .../backend/tests/test_growth_profiles.py | 140 ++++++++++- .../apps/user-client/src/App.vue | 47 +++- .../apps/user-client/src/services/api.ts | 1 + .../apps/user-client/src/types/api.ts | 8 + .../deployment/DEPLOYMENT_GUIDE.md | 20 ++ .../deployment/RELEASE_CHECKLIST.md | 3 + ai_knowledge_base_v2/docker-compose.dev.yml | 4 + ai_knowledge_base_v2/docker-compose.prod.yml | 4 + .../docs/qianwen_product_todo.md | 5 +- .../docs/一期全链路验收记录.md | 24 +- 24 files changed, 825 insertions(+), 58 deletions(-) create mode 100644 ai_knowledge_base_v2/apps/backend/alembic/versions/0024_topic_settlement_async_jobs.py create mode 100644 ai_knowledge_base_v2/apps/backend/app/services/topic_settlement_worker.py diff --git a/ai_knowledge_base_v2/.env.example b/ai_knowledge_base_v2/.env.example index a450ca0..d49c19e 100644 --- a/ai_knowledge_base_v2/.env.example +++ b/ai_knowledge_base_v2/.env.example @@ -28,6 +28,10 @@ PERIODIC_REPORT_MAX_ATTEMPTS=3 PERIODIC_REPORT_WEEKLY_ENABLED=true PERIODIC_REPORT_MONTHLY_ENABLED=true PERIODIC_REPORT_TIMEZONE=Asia/Shanghai +TOPIC_SETTLEMENT_WORKER_ENABLED=true +TOPIC_SETTLEMENT_POLL_SECONDS=2 +TOPIC_SETTLEMENT_STALE_MINUTES=30 +TOPIC_SETTLEMENT_MAX_ATTEMPTS=3 # ============================================ # 内网穿透（frpc sidecar） 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 e301432..4f723db 100644 --- a/ai_knowledge_base_v2/apps/admin-web/src/App.vue +++ b/ai_knowledge_base_v2/apps/admin-web/src/App.vue @@ -47,6 +47,7 @@ const selectedUserDetail = ref(null); const userDetailOpen = ref(false); const userDetailLoading = ref(false); const userReportGenerating = ref(""); +const userSettlementRetrying = ref(null); const userKeyword = ref(""); const entitlementPlans = ref([]); const models = ref([]); @@ -375,11 +376,26 @@ async function refreshSelectedUserDetail() { } } +async function retryTopicSettlement(topicId: number) { + if (!selectedUserDetail.value || userSettlementRetrying.value !== null) return; + userSettlementRetrying.value = topicId; + try { + await api.retryTopicSettlement(selectedUserDetail.value.user.id, topicId); + selectedUserDetail.value = await api.userDetail(selectedUserDetail.value.user.id); + ElMessage.success("主题沉淀任务已重新进入后台队列"); + } catch (error) { + ElMessage.error(error instanceof Error ? error.message : "主题沉淀重试失败"); + } finally { + userSettlementRetrying.value = null; + } +} + function reportStatusLabel(status: string) { const labels: Record = { pending: "等待生成", running: "生成中", success: "已完成", + fallback: "已降级沉淀", empty: "暂无沉淀", failed: "生成失败", }; @@ -1714,13 +1730,28 @@ function formatRecordDateTime(value: string, boundary: "start" | "end") { 更新：{{ topic.updatedAt }} 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 7399ae1..92d8128 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 @@ -35,6 +35,7 @@ import type { QuestionInsightRefreshResult, QuestionInsightSummary, TopicSessionRecord, + TopicSummaryRecord, } from "../types/api"; const API_BASE = import.meta.env.VITE_API_BASE_URL ?? "/api"; @@ -135,6 +136,8 @@ export const api = { userReports: (id: number, limit = 20) => request(`/admin/user/${id}/reports${queryString({ limit })}`), generateUserReport: (id: number, payload: { reportType: string; periodStart?: string | null; periodEnd?: string | null }) => request(`/admin/user/${id}/reports/generate`, { method: "POST", body: JSON.stringify(payload) }), + retryTopicSettlement: (userId: number, topicId: number) => + request(`/admin/user/${userId}/topic/${topicId}/settlement/retry`, { method: "POST", body: "{}" }), createUser: (payload: Record) => request("/admin/user", { method: "POST", body: JSON.stringify(payload) }), importUsers: (students: Record[]) => 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 87310f7..6f6a0da 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 @@ -500,6 +500,11 @@ export interface TopicSummaryRecord { modelName?: string | null; status: string; errorMessage?: string | null; + attemptCount: number; + maxAttempts: number; + nextRunAt?: string | null; + lastStartedAt?: string | null; + finishedAt?: string | null; generatedAt: string; } diff --git a/ai_knowledge_base_v2/apps/backend/.env.example b/ai_knowledge_base_v2/apps/backend/.env.example index 2a5aabc..a6c25b2 100644 --- a/ai_knowledge_base_v2/apps/backend/.env.example +++ b/ai_knowledge_base_v2/apps/backend/.env.example @@ -49,6 +49,10 @@ PERIODIC_REPORT_MAX_ATTEMPTS=3 PERIODIC_REPORT_WEEKLY_ENABLED=true PERIODIC_REPORT_MONTHLY_ENABLED=true PERIODIC_REPORT_TIMEZONE=Asia/Shanghai +TOPIC_SETTLEMENT_WORKER_ENABLED=true +TOPIC_SETTLEMENT_POLL_SECONDS=2 +TOPIC_SETTLEMENT_STALE_MINUTES=30 +TOPIC_SETTLEMENT_MAX_ATTEMPTS=3 # 本地开发可以使用开发密码；生产环境必须改成高强度密码，且 APP_ENV=production 时不能使用 admin123456 BOOTSTRAP_ADMIN_USERNAME=admin diff --git a/ai_knowledge_base_v2/apps/backend/alembic/versions/0024_topic_settlement_async_jobs.py b/ai_knowledge_base_v2/apps/backend/alembic/versions/0024_topic_settlement_async_jobs.py new file mode 100644 index 0000000..10c8531 --- /dev/null +++ b/ai_knowledge_base_v2/apps/backend/alembic/versions/0024_topic_settlement_async_jobs.py @@ -0,0 +1,59 @@ +"""add durable topic settlement job fields + +Revision ID: 0024_topic_settlement_jobs +Revises: 0023_simple_knowledge_route +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "0024_topic_settlement_jobs" +down_revision = "0023_simple_knowledge_route" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "sys_topic_summary", + sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"), + ) + op.add_column( + "sys_topic_summary", + sa.Column("max_attempts", sa.Integer(), nullable=False, server_default="3"), + ) + op.add_column("sys_topic_summary", sa.Column("next_run_at", sa.DateTime(), nullable=True)) + op.add_column("sys_topic_summary", sa.Column("locked_at", sa.DateTime(), nullable=True)) + op.add_column("sys_topic_summary", sa.Column("locked_by", sa.String(length=120), nullable=True)) + op.add_column("sys_topic_summary", sa.Column("last_started_at", sa.DateTime(), nullable=True)) + op.add_column("sys_topic_summary", sa.Column("finished_at", sa.DateTime(), nullable=True)) + op.execute( + "UPDATE sys_topic_summary " + "SET finished_at = generated_at " + "WHERE status IN ('success', 'fallback', 'failed')" + ) + op.create_index( + "ix_topic_summary_job_due", + "sys_topic_summary", + ["status", "next_run_at", "id"], + ) + op.create_index( + "ix_topic_summary_job_stale", + "sys_topic_summary", + ["status", "locked_at"], + ) + + +def downgrade() -> None: + op.drop_index("ix_topic_summary_job_stale", table_name="sys_topic_summary") + op.drop_index("ix_topic_summary_job_due", table_name="sys_topic_summary") + op.drop_column("sys_topic_summary", "finished_at") + op.drop_column("sys_topic_summary", "last_started_at") + op.drop_column("sys_topic_summary", "locked_by") + op.drop_column("sys_topic_summary", "locked_at") + op.drop_column("sys_topic_summary", "next_run_at") + op.drop_column("sys_topic_summary", "max_attempts") + op.drop_column("sys_topic_summary", "attempt_count") diff --git a/ai_knowledge_base_v2/apps/backend/app/api/admin_users.py b/ai_knowledge_base_v2/apps/backend/app/api/admin_users.py index 7154536..61e2345 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/admin_users.py +++ b/ai_knowledge_base_v2/apps/backend/app/api/admin_users.py @@ -349,6 +349,34 @@ def generate_user_report( return api_success(periodic_report_dict(report), message="报告任务已进入后台队列") +@router.post("/user/{user_id}/topic/{topic_id}/settlement/retry") +def retry_topic_settlement( + user_id: int, + topic_id: int, + db: Session = Depends(get_db), + current_admin: Admin = Depends(get_current_admin), +) -> dict: + user = _get_user(db, user_id) + topic = db.scalar( + select(TopicSession).where(TopicSession.id == topic_id, TopicSession.user_id == user.id) + ) + if topic is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="主题不存在") + if topic.status != "completed": + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="只能重试已结束主题的沉淀任务") + summary = GrowthProfileService.retry_topic_settlement(db, user=user, topic=topic) + OperationLogService.write( + db, + admin_id=current_admin.id, + module="topic_settlement", + action="retry", + target_id=summary.id, + ) + db.commit() + db.refresh(summary) + return api_success(topic_summary_dict(summary), message="主题沉淀任务已重新进入队列") + + @router.put("/user/{user_id}") def update_user( user_id: int, 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 5109087..a5477a8 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/chat.py +++ b/ai_knowledge_base_v2/apps/backend/app/api/chat.py @@ -98,6 +98,15 @@ def finish_topic( return api_success(GrowthProfileService.finish_active_topic(db, user=current_user, session=session)) +@router.get("/topic/settlement/{summary_id}") +def topic_settlement( + summary_id: int, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +) -> dict: + return api_success(GrowthProfileService.topic_settlement_result(db, user=current_user, summary_id=summary_id)) + + @router.post("/session/{session_id}/help-card") def generate_help_card( session_id: int, diff --git a/ai_knowledge_base_v2/apps/backend/app/api/user.py b/ai_knowledge_base_v2/apps/backend/app/api/user.py index 55af97b..906ab3e 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/user.py +++ b/ai_knowledge_base_v2/apps/backend/app/api/user.py @@ -48,6 +48,13 @@ def growth_profile( "summary": item.summary, "recommendedHomework": item.recommended_homework, "nextObservation": item.next_observation, + "status": item.status, + "errorMessage": item.error_message, + "attemptCount": item.attempt_count, + "maxAttempts": item.max_attempts, + "nextRunAt": item.next_run_at, + "lastStartedAt": item.last_started_at, + "finishedAt": item.finished_at, "generatedAt": item.generated_at, } for item in summaries 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 e7c0f6e..5fd8425 100644 --- a/ai_knowledge_base_v2/apps/backend/app/core/config.py +++ b/ai_knowledge_base_v2/apps/backend/app/core/config.py @@ -70,6 +70,10 @@ class Settings(BaseSettings): periodic_report_weekly_enabled: bool = True periodic_report_monthly_enabled: bool = True periodic_report_timezone: str = "Asia/Shanghai" + topic_settlement_worker_enabled: bool = True + topic_settlement_poll_seconds: int = 2 + topic_settlement_stale_minutes: int = 30 + topic_settlement_max_attempts: int = 3 bootstrap_admin_username: str = "" bootstrap_admin_password: str = "" bootstrap_admin_name: str = "系统管理员" diff --git a/ai_knowledge_base_v2/apps/backend/app/main.py b/ai_knowledge_base_v2/apps/backend/app/main.py index 874079c..247571d 100644 --- a/ai_knowledge_base_v2/apps/backend/app/main.py +++ b/ai_knowledge_base_v2/apps/backend/app/main.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio from contextlib import asynccontextmanager from fastapi import FastAPI @@ -13,7 +14,7 @@ from app.core.observability import RequestObservabilityMiddleware, configure_log from app.services.secret_service import SecretService from app.services.maintenance_service import MaintenanceService from app.services.periodic_report_worker import PeriodicReportWorker -import asyncio +from app.services.topic_settlement_worker import TopicSettlementWorker @asynccontextmanager @@ -24,12 +25,13 @@ async def lifespan(app: FastAPI): create_tables() maintenance_task = asyncio.create_task(MaintenanceService.run_forever()) periodic_report_task = asyncio.create_task(PeriodicReportWorker.run_forever()) + topic_settlement_task = asyncio.create_task(TopicSettlementWorker.run_forever()) try: yield finally: - for task in (maintenance_task, periodic_report_task): + for task in (maintenance_task, periodic_report_task, topic_settlement_task): task.cancel() - for task in (maintenance_task, periodic_report_task): + for task in (maintenance_task, periodic_report_task, topic_settlement_task): try: await task except asyncio.CancelledError: diff --git a/ai_knowledge_base_v2/apps/backend/app/models/growth.py b/ai_knowledge_base_v2/apps/backend/app/models/growth.py index 11a3b33..13b7395 100644 --- a/ai_knowledge_base_v2/apps/backend/app/models/growth.py +++ b/ai_knowledge_base_v2/apps/backend/app/models/growth.py @@ -12,7 +12,11 @@ PRIMARY_KEY_TYPE = BigInteger().with_variant(Integer, "sqlite") class TopicSummary(Base): __tablename__ = "sys_topic_summary" - __table_args__ = (UniqueConstraint("topic_session_id", name="uq_sys_topic_summary_topic"),) + __table_args__ = ( + UniqueConstraint("topic_session_id", name="uq_sys_topic_summary_topic"), + Index("ix_topic_summary_job_due", "status", "next_run_at", "id"), + Index("ix_topic_summary_job_stale", "status", "locked_at"), + ) id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) topic_session_id: Mapped[int] = mapped_column(ForeignKey("sys_topic_session.id"), index=True, nullable=False) @@ -28,6 +32,13 @@ class TopicSummary(Base): model_name: Mapped[str | None] = mapped_column(String(100), nullable=True) status: Mapped[str] = mapped_column(String(20), default="success", index=True, nullable=False) error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + attempt_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + max_attempts: Mapped[int] = mapped_column(Integer, default=3, nullable=False) + next_run_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + locked_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + locked_by: Mapped[str | None] = mapped_column(String(120), nullable=True) + last_started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + finished_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) generated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), 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/growth_profile_service.py b/ai_knowledge_base_v2/apps/backend/app/services/growth_profile_service.py index 818b401..5d306a3 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/growth_profile_service.py +++ b/ai_knowledge_base_v2/apps/backend/app/services/growth_profile_service.py @@ -9,48 +9,153 @@ from fastapi import HTTPException, status from sqlalchemy import select from sqlalchemy.orm import Session +from app.core.config import get_settings from app.models.chat import ChatMessage, ChatSession, TopicSession from app.models.growth import GrowthProfileRevision, TopicSummary, UserGrowthProfile from app.models.user import User from app.services.entitlement_service import EntitlementService from app.services.external_errors import ExternalServiceError from app.services.tracked_generation_service import TrackedGenerationService -from app.services.topic_session_service import TopicSessionService class GrowthProfileService: @staticmethod def finish_active_topic(db: Session, *, user: User, session: ChatSession, force: bool = False) -> dict: - topic = TopicSessionService.active_for_session(db, user=user, session=session) + topic = db.scalar( + select(TopicSession) + .where( + TopicSession.user_id == user.id, + TopicSession.chat_session_id == session.id, + TopicSession.status == "active", + ) + .order_by(TopicSession.created_at.desc(), TopicSession.id.desc()) + .with_for_update() + .limit(1) + ) if topic is None: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="当前会话没有进行中的主题") - summary = GrowthProfileService.generate_topic_summary(db, user=user, topic=topic, force=force) + # 重复点击或客户端超时重试时返回最近一次任务，不重复生成摘要。 + topic = db.scalar( + select(TopicSession) + .where( + TopicSession.user_id == user.id, + TopicSession.chat_session_id == session.id, + TopicSession.status == "completed", + ) + .order_by(TopicSession.ended_at.desc(), TopicSession.id.desc()) + .limit(1) + ) + summary = ( + db.scalar(select(TopicSummary).where(TopicSummary.topic_session_id == topic.id)) + if topic is not None + else None + ) + if topic is None or summary is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="当前会话没有进行中的主题") + entitlement = EntitlementService.active_entitlement(db, user) + return _settlement_result(db, topic=topic, summary=summary, user=user, growth_enabled=entitlement.enable_growth_profile) + + has_messages = db.scalar( + select(ChatMessage.id) + .where(ChatMessage.topic_session_id == topic.id, ChatMessage.user_id == user.id) + .limit(1) + ) + if has_messages is None: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="当前主题还没有可沉淀的对话内容") + + summary = db.scalar(select(TopicSummary).where(TopicSummary.topic_session_id == topic.id)) + is_new_summary = summary is None + if is_new_summary: + summary = TopicSummary(topic_session_id=topic.id, user_id=user.id, summary="") + if is_new_summary or force or summary.status not in {"pending", "running", "success", "fallback"}: + _reset_summary_job(summary) + summary.max_attempts = max(1, get_settings().topic_settlement_max_attempts) + db.add(summary) + topic.status = "completed" topic.ended_at = _now() - topic.recommended_homework = summary.recommended_homework db.add(topic) entitlement = EntitlementService.active_entitlement(db, user) - profile = None - if entitlement.enable_growth_profile: - profile = GrowthProfileService.update_growth_profile(db, user=user, topic_summary=summary) db.commit() db.refresh(topic) db.refresh(summary) - if profile is not None: - db.refresh(profile) - return { - "topic": topic_dict(topic), - "summary": topic_summary_dict(summary), - "profile": growth_profile_dict(profile) if profile is not None else None, - "growthProfileEnabled": entitlement.enable_growth_profile, - } + return _settlement_result(db, topic=topic, summary=summary, user=user, growth_enabled=entitlement.enable_growth_profile) @staticmethod - def generate_topic_summary(db: Session, *, user: User, topic: TopicSession, force: bool = False) -> TopicSummary: + def retry_topic_settlement(db: Session, *, user: User, topic: TopicSession) -> TopicSummary: + summary = db.scalar(select(TopicSummary).where(TopicSummary.topic_session_id == topic.id).with_for_update()) + if summary is None: + summary = TopicSummary(topic_session_id=topic.id, user_id=user.id, summary="") + if summary.status == "running": + return summary + _reset_summary_job(summary) + summary.max_attempts = max(1, get_settings().topic_settlement_max_attempts) + db.add(summary) + db.flush() + return summary + + @staticmethod + def topic_settlement_result(db: Session, *, user: User, summary_id: int) -> dict: + summary = db.scalar( + select(TopicSummary).where(TopicSummary.id == summary_id, TopicSummary.user_id == user.id) + ) + if summary is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="主题沉淀任务不存在") + topic = db.get(TopicSession, summary.topic_session_id) + if topic is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="主题不存在") + entitlement = EntitlementService.active_entitlement(db, user) + return _settlement_result( + db, + topic=topic, + summary=summary, + user=user, + growth_enabled=entitlement.enable_growth_profile, + ) + + @staticmethod + def process_topic_settlement(db: Session, *, summary: TopicSummary) -> tuple[TopicSummary, UserGrowthProfile | None]: + topic = db.get(TopicSession, summary.topic_session_id) + # Redis 失效时允许多个进程依靠数据库继续消费；用户行锁保证同一学员的 + # 多个主题不会并发覆盖长期成长档案。 + user = db.scalar(select(User).where(User.id == summary.user_id).with_for_update()) + if topic is None or user is None or user.is_deleted: + raise ValueError("主题或用户不存在") + generated = GrowthProfileService.generate_topic_summary( + db, + user=user, + topic=topic, + force=True, + process_job=True, + allow_fallback=False, + ) + topic.recommended_homework = generated.recommended_homework + db.add(topic) + entitlement = EntitlementService.active_entitlement(db, user) + profile = None + if entitlement.enable_growth_profile: + profile = GrowthProfileService.update_growth_profile(db, user=user, topic_summary=generated) + return generated, profile + + @staticmethod + def generate_topic_summary( + db: Session, + *, + user: User, + topic: TopicSession, + force: bool = False, + process_job: bool = False, + allow_fallback: bool = True, + ) -> TopicSummary: existing = db.scalar(select(TopicSummary).where(TopicSummary.topic_session_id == topic.id)) if existing is not None and not force: - return existing + if existing.status in {"success", "fallback"}: + return existing + if existing.status in {"pending", "running"}: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="主题正在后台沉淀，请稍后再试") + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="主题沉淀失败，请稍后重试") + if existing is not None and existing.status in {"pending", "running"} and not process_job: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="主题正在后台沉淀，请稍后再试") messages = list( db.scalars( select(ChatMessage) @@ -77,6 +182,8 @@ class GrowthProfileService: status_value = "success" error_message = None except ExternalServiceError as exc: + if not allow_fallback: + raise data = _fallback_summary(topic, conversation, "") status_value = "fallback" error_message = str(exc) @@ -88,6 +195,10 @@ class GrowthProfileService: existing.status = status_value existing.error_message = error_message existing.generated_at = _now() + existing.finished_at = _now() + existing.next_run_at = None + existing.locked_at = None + existing.locked_by = None db.add(existing) db.flush() return existing @@ -95,6 +206,8 @@ class GrowthProfileService: @staticmethod def update_growth_profile(db: Session, *, user: User, topic_summary: TopicSummary) -> UserGrowthProfile: profile = db.scalar(select(UserGrowthProfile).where(UserGrowthProfile.user_id == user.id)) + if profile is not None and profile.last_topic_summary_id == topic_summary.id: + return profile before = growth_profile_dict(profile) if profile is not None else None if profile is None: profile = UserGrowthProfile(user_id=user.id, profile_text="") @@ -197,6 +310,11 @@ def topic_summary_dict(summary: TopicSummary) -> dict: "modelName": summary.model_name, "status": summary.status, "errorMessage": summary.error_message, + "attemptCount": summary.attempt_count, + "maxAttempts": summary.max_attempts, + "nextRunAt": summary.next_run_at, + "lastStartedAt": summary.last_started_at, + "finishedAt": summary.finished_at, "generatedAt": summary.generated_at, } @@ -303,17 +421,32 @@ def _parse_summary_json(raw: str) -> dict[str, Any] | None: text = raw.strip() if not text: return None - if text.startswith("```"): - text = re.sub(r"^```(?:json)?", "", text, flags=re.IGNORECASE).strip() - text = re.sub(r"```$", "", text).strip() - match = re.search(r"\{[\s\S]*\}", text) - if match: - text = match.group(0) - try: - parsed = json.loads(text) - except json.JSONDecodeError: - return None - return parsed if isinstance(parsed, dict) else None + # 部分模型会把推理过程和最终 JSON 一起放进 answer，甚至在推理中先给出一份 + # 草稿 JSON。先移除推理块，再从后向前选取可完整解析的对象，避免把思考过程 + # 或多个 JSON 拼接后写入用户成长档案。 + text = re.sub(r"<(?:think|analysis)>[\s\S]*?", "", text, flags=re.IGNORECASE).strip() + fenced = re.findall(r"```(?:json)?\s*([\s\S]*?)\s*```", text, flags=re.IGNORECASE) + candidates = [*fenced, text] + for candidate in reversed(candidates): + parsed = _last_json_object(candidate) + if parsed is not None: + return parsed + return None + + +def _last_json_object(text: str) -> dict[str, Any] | None: + decoder = json.JSONDecoder() + parsed_objects: list[dict[str, Any]] = [] + for index, character in enumerate(text): + if character != "{": + continue + try: + value, _ = decoder.raw_decode(text[index:]) + except json.JSONDecodeError: + continue + if isinstance(value, dict): + parsed_objects.append(value) + return parsed_objects[-1] if parsed_objects else None def _messages_text(messages: list[ChatMessage]) -> str: @@ -341,3 +474,43 @@ def _limit(text: str, max_len: int) -> str: def _now() -> datetime: return datetime.now(UTC).replace(tzinfo=None) + + +def _reset_summary_job(summary: TopicSummary) -> None: + summary.summary = "" + summary.main_events = None + summary.emotions = None + summary.body_feelings = None + summary.beliefs = None + summary.recommended_homework = None + summary.insights = None + summary.next_observation = None + summary.model_name = None + summary.status = "pending" + summary.error_message = None + summary.attempt_count = 0 + summary.next_run_at = _now() + summary.locked_at = None + summary.locked_by = None + summary.last_started_at = None + summary.finished_at = None + + +def _settlement_result( + db: Session, + *, + topic: TopicSession, + summary: TopicSummary, + user: User, + growth_enabled: bool, +) -> dict: + profile = None + if growth_enabled and summary.status in {"success", "fallback"}: + profile = GrowthProfileService.get_growth_profile(db, user.id) + return { + "topic": topic_dict(topic), + "summary": topic_summary_dict(summary), + "profile": growth_profile_dict(profile), + "growthProfileEnabled": growth_enabled, + "settlementStatus": summary.status, + } diff --git a/ai_knowledge_base_v2/apps/backend/app/services/topic_settlement_worker.py b/ai_knowledge_base_v2/apps/backend/app/services/topic_settlement_worker.py new file mode 100644 index 0000000..727fb23 --- /dev/null +++ b/ai_knowledge_base_v2/apps/backend/app/services/topic_settlement_worker.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +import asyncio +import logging +import os +import socket +import uuid +from datetime import UTC, datetime, timedelta + +from sqlalchemy import or_, select +from sqlalchemy.orm import Session + +from app.core.config import get_settings +from app.core.database import SessionLocal +from app.models.chat import TopicSession +from app.models.growth import TopicSummary +from app.models.user import User +from app.services.growth_profile_service import GrowthProfileService +from app.services.redis_client import get_sync_redis_client + + +logger = logging.getLogger(__name__) +WORKER_LOCK_KEY = "topic-settlement:worker:lock" +WORKER_LOCK_TTL_SECONDS = 900 + + +class TopicSettlementWorker: + """持久化主题沉淀队列。 + + MySQL 保存任务、重试和执行状态，Redis 仅限制多进程同时执行数量。 + 即使 Redis 暂时不可用或服务重启，已结束主题也不会丢失沉淀任务。 + """ + + @classmethod + async def run_forever(cls) -> None: + settings = get_settings() + if not settings.topic_settlement_worker_enabled: + logger.info("topic settlement worker disabled") + return + worker_id = f"{socket.gethostname()}:{os.getpid()}:{uuid.uuid4().hex[:8]}" + poll_seconds = max(1, settings.topic_settlement_poll_seconds) + while True: + try: + processed = await asyncio.to_thread(cls.run_once, worker_id) + except Exception: + processed = False + logger.exception("topic settlement worker iteration failed") + await asyncio.sleep(0 if processed else poll_seconds) + + @classmethod + def run_once(cls, worker_id: str) -> bool: + redis, acquired = _acquire_worker_lock(worker_id) + if not acquired: + return False + try: + now = _now() + with SessionLocal() as db: + cls.recover_stale_jobs(db, now=now) + db.commit() + with SessionLocal() as db: + summary_id = cls.claim_next(db, worker_id=worker_id, now=now) + if summary_id is None: + return False + with SessionLocal() as db: + cls.execute_claimed(db, summary_id=summary_id, worker_id=worker_id) + return True + finally: + _release_worker_lock(redis, worker_id) + + @staticmethod + def claim_next(db: Session, *, worker_id: str, now: datetime | None = None) -> int | None: + current = now or _now() + summary = db.scalar( + select(TopicSummary) + .where( + TopicSummary.status == "pending", + TopicSummary.attempt_count < TopicSummary.max_attempts, + or_(TopicSummary.next_run_at.is_(None), TopicSummary.next_run_at <= current), + ) + .order_by(TopicSummary.next_run_at.asc(), TopicSummary.id.asc()) + .with_for_update(skip_locked=True) + .limit(1) + ) + if summary is None: + db.rollback() + return None + summary.status = "running" + summary.attempt_count += 1 + summary.locked_at = current + summary.locked_by = worker_id + summary.last_started_at = current + summary.finished_at = None + db.add(summary) + db.commit() + return summary.id + + @staticmethod + def execute_claimed(db: Session, *, summary_id: int, worker_id: str) -> TopicSummary | None: + summary = db.get(TopicSummary, summary_id) + if summary is None or summary.status != "running" or summary.locked_by != worker_id: + return summary + topic = db.get(TopicSession, summary.topic_session_id) + user = db.get(User, summary.user_id) + if topic is None or user is None or user.is_deleted: + _finish_permanent_failure(summary, "主题或用户不存在") + db.commit() + return summary + + try: + GrowthProfileService.process_topic_settlement(db, summary=summary) + now = _now() + summary.next_run_at = None + summary.finished_at = now + summary.locked_at = None + summary.locked_by = None + db.add(summary) + db.commit() + except Exception as exc: + # 撤销本轮可能已经写入的半成品摘要、档案和修订记录，再单独更新任务状态。 + db.rollback() + summary = db.get(TopicSummary, summary_id) + if summary is None: + return None + summary.error_message = str(exc)[:2000] + now = _now() + if summary.attempt_count < summary.max_attempts: + retry_seconds = min(300, 30 * (2 ** max(0, summary.attempt_count - 1))) + summary.status = "pending" + summary.next_run_at = now + timedelta(seconds=retry_seconds) + summary.finished_at = None + else: + summary.status = "failed" + summary.next_run_at = None + summary.finished_at = now + summary.locked_at = None + summary.locked_by = None + db.add(summary) + db.commit() + db.refresh(summary) + return summary + + @staticmethod + def recover_stale_jobs(db: Session, *, now: datetime | None = None) -> int: + current = now or _now() + stale_before = current - timedelta(minutes=max(5, get_settings().topic_settlement_stale_minutes)) + summaries = list( + db.scalars( + select(TopicSummary) + .where( + TopicSummary.status == "running", + TopicSummary.locked_at.is_not(None), + TopicSummary.locked_at < stale_before, + ) + .order_by(TopicSummary.id.asc()) + .limit(100) + .with_for_update(skip_locked=True) + ) + ) + for summary in summaries: + if summary.attempt_count >= summary.max_attempts: + summary.status = "failed" + summary.finished_at = current + summary.next_run_at = None + else: + summary.status = "pending" + summary.next_run_at = current + summary.finished_at = None + summary.error_message = _append_error(summary.error_message, "任务执行进程中断，系统已自动恢复") + summary.locked_at = None + summary.locked_by = None + db.add(summary) + return len(summaries) + + +def _finish_permanent_failure(summary: TopicSummary, message: str) -> None: + summary.status = "failed" + summary.error_message = message + summary.next_run_at = None + summary.locked_at = None + summary.locked_by = None + summary.finished_at = _now() + + +def _append_error(current: str | None, message: str) -> str: + if not current: + return message + return f"{current}\n{message}"[-2000:] + + +def _acquire_worker_lock(worker_id: str): + redis = get_sync_redis_client() + if redis is None: + return None, True + try: + return redis, bool(redis.set(WORKER_LOCK_KEY, worker_id, nx=True, ex=WORKER_LOCK_TTL_SECONDS)) + except Exception: + logger.warning("redis unavailable for topic settlement lock; falling back to database claim", exc_info=True) + return None, True + + +def _release_worker_lock(redis, worker_id: str) -> None: + if redis is None: + return + try: + redis.eval( + "if redis.call('get', KEYS[1]) == ARGV[1] then " + "return redis.call('del', KEYS[1]) else return 0 end", + 1, + WORKER_LOCK_KEY, + worker_id, + ) + except Exception: + logger.warning("failed to release topic settlement worker lock", exc_info=True) + + +def _now() -> datetime: + return datetime.now(UTC).replace(tzinfo=None) diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_growth_profiles.py b/ai_knowledge_base_v2/apps/backend/tests/test_growth_profiles.py index bdd3f06..9acabfe 100644 --- a/ai_knowledge_base_v2/apps/backend/tests/test_growth_profiles.py +++ b/ai_knowledge_base_v2/apps/backend/tests/test_growth_profiles.py @@ -1,6 +1,6 @@ from __future__ import annotations -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from sqlalchemy import create_engine from sqlalchemy.orm import Session @@ -10,11 +10,12 @@ from app.models import Base from app.models.ai_config import SystemConfig from app.models.chat import ChatMessage, ChatSession, TopicSession from app.models.entitlement import EntitlementPlan -from app.models.growth import GrowthProfileRevision, UserGrowthProfile +from app.models.growth import GrowthProfileRevision, TopicSummary, UserGrowthProfile from app.models.user import User from app.services.entitlement_service import EntitlementService -from app.services.growth_profile_service import GrowthProfileService +from app.services.growth_profile_service import GrowthProfileService, _parse_summary_json from app.services.rag_service import PromptService +from app.services.topic_settlement_worker import TopicSettlementWorker def _db() -> Session: @@ -27,7 +28,7 @@ def _now() -> datetime: return datetime.now(UTC).replace(tzinfo=None) -def test_finish_topic_generates_summary_and_updates_growth_profile_for_enabled_plan(): +def test_finish_topic_enqueues_summary_and_worker_updates_growth_profile_for_enabled_plan(): with _db() as db: db.add(SystemConfig(config_key="mock_model_enabled", config_value="true")) db.add(EntitlementPlan(id=10, name="深度陪伴版", plan_type="deep", monthly_topic_limit=90, enable_growth_profile=1, status=1)) @@ -58,13 +59,114 @@ def test_finish_topic_generates_summary_and_updates_growth_profile_for_enabled_p result = GrowthProfileService.finish_active_topic(db, user=user, session=session) assert result["topic"]["status"] == "completed" - assert result["summary"]["summary"] - assert result["profile"] is not None - assert "最近主题" in result["profile"]["profileText"] + assert result["settlementStatus"] == "pending" + assert result["summary"]["summary"] == "" + assert result["profile"] is None + assert db.query(UserGrowthProfile).filter_by(user_id=1).count() == 0 + + summary_id = TopicSettlementWorker.claim_next(db, worker_id="test-worker", now=_now()) + assert summary_id == result["summary"]["id"] + completed = TopicSettlementWorker.execute_claimed(db, summary_id=summary_id, worker_id="test-worker") + + assert completed is not None + assert completed.status == "success" + assert completed.summary + profile = db.query(UserGrowthProfile).filter_by(user_id=1).one() + assert "最近主题" in profile.profile_text assert db.query(UserGrowthProfile).filter_by(user_id=1).count() == 1 assert db.query(GrowthProfileRevision).filter_by(user_id=1).count() == 1 +def test_finish_topic_is_idempotent_and_does_not_call_model_in_request(monkeypatch): + with _db() as db: + now = _now() + user = User(id=1, phone="13800000001", name="学员", daily_chat_limit=100, daily_chat_used=0) + session = ChatSession(id=1, user_id=1, title="测试", message_count=1, last_message_at=now, is_deleted=0) + topic = TopicSession( + id=1, + user_id=1, + chat_session_id=1, + title="测试主题", + core_question="我想沉淀一下", + status="active", + message_count=1, + started_at=now, + ) + message = ChatMessage(id=1, session_id=1, topic_session_id=1, user_id=1, role="user", content="我想沉淀一下", created_at=now) + db.add_all([user, session, topic, message]) + db.commit() + monkeypatch.setattr( + "app.services.tracked_generation_service.TrackedGenerationService.generate", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("结束主题接口不应调用模型")), + ) + + first = GrowthProfileService.finish_active_topic(db, user=user, session=session) + second = GrowthProfileService.finish_active_topic(db, user=user, session=session) + + assert first["summary"]["id"] == second["summary"]["id"] + assert first["settlementStatus"] == "pending" + assert second["settlementStatus"] == "pending" + assert db.query(TopicSummary).count() == 1 + + +def test_failed_topic_settlement_is_retried(monkeypatch): + with _db() as db: + now = _now() + user = User(id=1, phone="13800000001", name="学员", daily_chat_limit=100, daily_chat_used=0) + session = ChatSession(id=1, user_id=1, title="测试", message_count=1, last_message_at=now, is_deleted=0) + topic = TopicSession(id=1, user_id=1, chat_session_id=1, title="测试", core_question="测试", status="active", started_at=now) + message = ChatMessage(id=1, session_id=1, topic_session_id=1, user_id=1, role="user", content="测试", created_at=now) + db.add_all([user, session, topic, message]) + db.commit() + result = GrowthProfileService.finish_active_topic(db, user=user, session=session) + monkeypatch.setattr( + GrowthProfileService, + "process_topic_settlement", + staticmethod(lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("模型暂时不可用"))), + ) + + summary_id = TopicSettlementWorker.claim_next(db, worker_id="retry-worker", now=_now()) + retried = TopicSettlementWorker.execute_claimed(db, summary_id=summary_id, worker_id="retry-worker") + + assert summary_id == result["summary"]["id"] + assert retried is not None + assert retried.status == "pending" + assert retried.attempt_count == 1 + assert retried.next_run_at is not None + assert "模型暂时不可用" in (retried.error_message or "") + + +def test_stale_topic_settlement_is_recovered_after_restart(): + with _db() as db: + now = _now() + user = User(id=1, phone="13800000001", name="学员", daily_chat_limit=100, daily_chat_used=0) + session = ChatSession(id=1, user_id=1, title="测试", message_count=1, last_message_at=now, is_deleted=0) + topic = TopicSession(id=1, user_id=1, chat_session_id=1, title="测试", core_question="测试", status="completed", started_at=now, ended_at=now) + summary = TopicSummary( + id=1, + topic_session_id=1, + user_id=1, + summary="", + status="running", + attempt_count=1, + max_attempts=3, + locked_by="stopped-worker", + locked_at=now - timedelta(minutes=31), + ) + db.add_all([user, session, topic, summary]) + db.commit() + + recovered = TopicSettlementWorker.recover_stale_jobs(db, now=now) + db.commit() + db.refresh(summary) + + assert recovered == 1 + assert summary.status == "pending" + assert summary.locked_by is None + assert summary.next_run_at == now + assert "自动恢复" in (summary.error_message or "") + + def test_prompt_can_include_growth_profile_context_without_replacing_knowledge_context(): with _db() as db: user = User(id=1, phone="13800000001", name="学员", daily_chat_limit=100, daily_chat_used=0) @@ -84,3 +186,27 @@ def test_prompt_can_include_growth_profile_context_without_replacing_knowledge_c assert "[长期成长档案]" in content assert "表达障碍" in content assert "[本轮可靠知识上下文]" in content + + +def test_summary_json_parser_ignores_model_reasoning_and_uses_final_json(): + raw = """ + +先分析一下，并给出一个未完成草稿： +```json +{"profileText": "不应采用的草稿"} +``` + +```json +{ + "profileText": "只保留最终成长档案", + "commonEmotions": ["紧张"], + "recentProgress": ["开始观察身体感受"] +} +``` +""" + + parsed = _parse_summary_json(raw) + + assert parsed is not None + assert parsed["profileText"] == "只保留最终成长档案" + assert "think" not in str(parsed).lower() 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 141f16c..6ccf094 100644 --- a/ai_knowledge_base_v2/apps/user-client/src/App.vue +++ b/ai_knowledge_base_v2/apps/user-client/src/App.vue @@ -43,6 +43,7 @@ const followingOutput = ref(true); const messageList = ref | null>(null); const activeAbortController = ref(null); let toastTimer: number | null = null; +let settlementPollVersion = 0; onMounted(async () => { if (!getToken()) { @@ -62,6 +63,7 @@ onMounted(async () => { }); onBeforeUnmount(() => { + settlementPollVersion += 1; activeAbortController.value?.abort(); if (toastTimer) window.clearTimeout(toastTimer); document.body.classList.remove("drawer-open"); @@ -215,7 +217,12 @@ async function finishCurrentTopic() { finishingTopic.value = true; try { const result = await api.finishTopic(activeSessionId.value); - showToast(result.growthProfileEnabled ? "本主题已沉淀，并更新了你的成长档案" : "本主题已沉淀"); + if (result.settlementStatus === "success" || result.settlementStatus === "fallback") { + showToast(result.growthProfileEnabled ? "本主题已沉淀，并更新了你的成长档案" : "本主题已沉淀"); + } else { + showToast("主题已结束，实修记录正在后台沉淀，可以继续使用其他功能"); + void watchTopicSettlement(result.summary.id); + } await refreshSessionList(); await refreshProfile(); } catch (error) { @@ -225,6 +232,38 @@ async function finishCurrentTopic() { } } +async function watchTopicSettlement(summaryId: number) { + const version = ++settlementPollVersion; + for (let attempt = 0; attempt < 40; attempt += 1) { + await new Promise((resolve) => window.setTimeout(resolve, 3000)); + if (version !== settlementPollVersion || !user.value) return; + try { + const result = await api.topicSettlement(summaryId); + if (result.settlementStatus === "success" || result.settlementStatus === "fallback") { + showToast(result.growthProfileEnabled ? "实修记录已沉淀，并更新了成长档案" : "实修记录已沉淀完成"); + await refreshProfile(); + return; + } + if (result.settlementStatus === "failed") { + showToast("实修记录暂时沉淀失败，系统已保留对话，可稍后重试"); + return; + } + } catch { + // 轮询失败不影响聊天；下一轮继续查询持久化任务状态。 + } + } +} + +function settlementStatusLabel(status: string) { + return { + pending: "等待沉淀", + running: "沉淀中", + success: "已完成", + fallback: "已降级完成", + failed: "沉淀失败", + }[status] || status; +} + async function generateHelpCard() { if (!activeSessionId.value || sending.value || generatingHelpCard.value) return; generatingHelpCard.value = true; @@ -490,8 +529,10 @@ async function copyText(text: string) {

最近主题沉淀

- -

{{ item.summary }}

+ +

{{ item.summary }}

+

沉淀暂时失败，对话内容仍已保留。

+

正在后台整理本主题的实修记录，可以先关闭页面或继续聊天。

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 1252e2d..3e6f4f8 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 @@ -89,6 +89,7 @@ export const api = { request("/chat/session/title", { method: "PUT", body: JSON.stringify({ sessionId, title }) }), deleteSession: (sessionId: number) => request(`/chat/session/${sessionId}`, { method: "DELETE" }), finishTopic: (sessionId: number) => request(`/chat/session/${sessionId}/topic/finish`, { method: "POST", body: JSON.stringify({}) }), + topicSettlement: (summaryId: number) => request(`/chat/topic/settlement/${summaryId}`), generateHelpCard: (sessionId: number) => request(`/chat/session/${sessionId}/help-card`, { method: "POST", body: JSON.stringify({}) }), markHelpCardCopied: (cardId: number) => request(`/chat/help-card/${cardId}/copied`, { method: "POST", body: JSON.stringify({}) }), helpCards: (limit = 20) => request(`/chat/help-card/list?limit=${limit}`), diff --git a/ai_knowledge_base_v2/apps/user-client/src/types/api.ts b/ai_knowledge_base_v2/apps/user-client/src/types/api.ts index 9064bab..86889b8 100644 --- a/ai_knowledge_base_v2/apps/user-client/src/types/api.ts +++ b/ai_knowledge_base_v2/apps/user-client/src/types/api.ts @@ -38,6 +38,13 @@ export interface TopicSummary { summary: string; recommendedHomework?: string | null; nextObservation?: string | null; + status: "pending" | "running" | "success" | "fallback" | "failed" | string; + errorMessage?: string | null; + attemptCount: number; + maxAttempts: number; + nextRunAt?: string | null; + lastStartedAt?: string | null; + finishedAt?: string | null; generatedAt: string; } @@ -77,6 +84,7 @@ export interface FinishTopicResult { summary: TopicSummary & Record; profile: GrowthProfile | null; growthProfileEnabled: boolean; + settlementStatus: string; } export interface TeacherHelpCard { diff --git a/ai_knowledge_base_v2/deployment/DEPLOYMENT_GUIDE.md b/ai_knowledge_base_v2/deployment/DEPLOYMENT_GUIDE.md index 9af4263..69d0715 100644 --- a/ai_knowledge_base_v2/deployment/DEPLOYMENT_GUIDE.md +++ b/ai_knowledge_base_v2/deployment/DEPLOYMENT_GUIDE.md @@ -163,6 +163,26 @@ PERIODIC_REPORT_STALE_MINUTES=30 PERIODIC_REPORT_MAX_ATTEMPTS=3 ``` +## 主题沉淀后台任务 + +用户点击“沉淀本主题”后，接口只结束当前主题并创建持久化任务，不再等待模型生成摘要和成长档案。用户可以立即继续聊天、切换会话或关闭页面。 + +任务状态保存在 `sys_topic_summary`： + +- `pending`：等待后台执行； +- `running`：正在生成摘要和成长档案； +- `success` / `fallback`：沉淀完成； +- `failed`：达到最大重试次数后仍失败，可在后台用户详情中手动重试。 + +多进程部署时 Redis 用于限制同一时间只有一个主题沉淀 Worker 执行，MySQL 负责持久化任务、幂等和失败重试。服务中断超过配置时间后，任务会自动恢复，不需要用户重复结束主题。 + +```text +TOPIC_SETTLEMENT_WORKER_ENABLED=true +TOPIC_SETTLEMENT_POLL_SECONDS=2 +TOPIC_SETTLEMENT_STALE_MINUTES=30 +TOPIC_SETTLEMENT_MAX_ATTEMPTS=3 +``` + ## 多模型分流 模型管理现在区分两个概念： diff --git a/ai_knowledge_base_v2/deployment/RELEASE_CHECKLIST.md b/ai_knowledge_base_v2/deployment/RELEASE_CHECKLIST.md index 307213c..9231e9b 100644 --- a/ai_knowledge_base_v2/deployment/RELEASE_CHECKLIST.md +++ b/ai_knowledge_base_v2/deployment/RELEASE_CHECKLIST.md @@ -23,6 +23,8 @@ - [ ] 已确认权益版本、能力开关和学员当前权益配置正确。 - [ ] 已验证后台 Agent 预览能够按模拟学员和具体主题加载上下文。 - [ ] 已确认“结束主题”在当前网关超时范围内完成；如已异步化，已验证任务状态、幂等和失败重试。 +- [ ] 已确认主题沉淀 Worker 已启用，等待、执行中、成功和失败状态均可查询。 +- [ ] 已验证主题沉淀任务在服务重启后自动恢复，后台可以手动重试最终失败任务。 ## 发布中 @@ -45,6 +47,7 @@ - [ ] 用户端可以提问并收到回答。 - [ ] 用户发送后输入框立即清空，思考状态与流式 Markdown 回答显示正常。 - [ ] 连续提问能够继续使用同一主题的上下文。 +- [ ] 结束主题接口快速返回，用户可以立即新建主题或继续使用系统。 - [ ] 无命中问题返回固定兜底。 - [ ] 管理后台可以登录。 - [ ] 管理后台可以查看 Dashboard。 diff --git a/ai_knowledge_base_v2/docker-compose.dev.yml b/ai_knowledge_base_v2/docker-compose.dev.yml index ed5c7dd..e332e8f 100644 --- a/ai_knowledge_base_v2/docker-compose.dev.yml +++ b/ai_knowledge_base_v2/docker-compose.dev.yml @@ -53,6 +53,10 @@ services: PERIODIC_REPORT_WEEKLY_ENABLED: "true" PERIODIC_REPORT_MONTHLY_ENABLED: "true" PERIODIC_REPORT_TIMEZONE: Asia/Shanghai + TOPIC_SETTLEMENT_WORKER_ENABLED: "true" + TOPIC_SETTLEMENT_POLL_SECONDS: "2" + TOPIC_SETTLEMENT_STALE_MINUTES: "30" + TOPIC_SETTLEMENT_MAX_ATTEMPTS: "3" JWT_SECRET_KEY: local-dev-secret-change-before-production MOCK_SMS_ENABLED: "true" MOCK_SMS_CODE: "123456" diff --git a/ai_knowledge_base_v2/docker-compose.prod.yml b/ai_knowledge_base_v2/docker-compose.prod.yml index 86be6f5..4a6abec 100644 --- a/ai_knowledge_base_v2/docker-compose.prod.yml +++ b/ai_knowledge_base_v2/docker-compose.prod.yml @@ -52,6 +52,10 @@ services: PERIODIC_REPORT_WEEKLY_ENABLED: ${PERIODIC_REPORT_WEEKLY_ENABLED:-true} PERIODIC_REPORT_MONTHLY_ENABLED: ${PERIODIC_REPORT_MONTHLY_ENABLED:-true} PERIODIC_REPORT_TIMEZONE: ${PERIODIC_REPORT_TIMEZONE:-Asia/Shanghai} + TOPIC_SETTLEMENT_WORKER_ENABLED: ${TOPIC_SETTLEMENT_WORKER_ENABLED:-true} + TOPIC_SETTLEMENT_POLL_SECONDS: ${TOPIC_SETTLEMENT_POLL_SECONDS:-2} + TOPIC_SETTLEMENT_STALE_MINUTES: ${TOPIC_SETTLEMENT_STALE_MINUTES:-30} + TOPIC_SETTLEMENT_MAX_ATTEMPTS: ${TOPIC_SETTLEMENT_MAX_ATTEMPTS:-3} JWT_SECRET_KEY: ${JWT_SECRET_KEY:?必须配置 JWT_SECRET_KEY} CONFIG_ENCRYPTION_KEY: ${CONFIG_ENCRYPTION_KEY:?必须配置 CONFIG_ENCRYPTION_KEY} MOCK_SMS_ENABLED: "false" diff --git a/ai_knowledge_base_v2/docs/qianwen_product_todo.md b/ai_knowledge_base_v2/docs/qianwen_product_todo.md index 7851ac4..432ba2f 100644 --- a/ai_knowledge_base_v2/docs/qianwen_product_todo.md +++ b/ai_knowledge_base_v2/docs/qianwen_product_todo.md @@ -357,7 +357,8 @@ Agent 已经可以： - 2026-07-31：已补齐后台预览的“模拟学员”入口；选择学员后，预览链路会加载该学员权益、主题额度使用量和长期成长档案，并在检索追踪中记录 `load_debug_user_context`。默认不选择学员时，保持原后台调试行为。 - 2026-07-31：后台预览已支持继续选择具体主题，加载主题摘要、滚动摘要和最近消息；正式聊天与后台预览均会按权益注入成长档案、求助卡和分享稿能力，并在追踪中记录模型路由与上下文来源。 -- 2026-07-31：一期全链路已完成隔离生产编排、真实模型问答、流式响应、主题上下文、成长档案、求助卡、分享稿、异步报告和记录审计验收。详细结果见《一期全链路验收记录》；主题沉淀接口的同步耗时仍需作为上线前性能项处理。 +- 2026-07-31：一期全链路已完成隔离生产编排、真实模型问答、流式响应、主题上下文、成长档案、求助卡、分享稿、异步报告和记录审计验收。详细结果见《一期全链路验收记录》。 +- 2026-08-03：主题结束已改为持久化异步沉淀。接口快速结束主题并创建唯一任务，后台独立生成摘要和成长档案；支持自动重试、服务重启恢复、用户端状态查询和后台手动重试，原同步模型调用不再阻塞用户操作。 --- @@ -896,6 +897,6 @@ AI 日志增加： - 千问千答不建设老师端或“老师工作版”，老师仍由学员自行联系； - 周报、月报的异步与定时生成已经完成，本计划中的连续陪伴主链路已完成一期闭环； -- 一期主要功能链路已验收通过；主题结束时同步生成摘要和成长档案的耗时仍是上线前必须关注的性能风险； +- 一期主要功能链路已验收通过；主题摘要和成长档案已改为持久化异步生成，原同步等待风险已经关闭； - 问题洞察的知识库补充建议、人工合并和拆分暂缓； - 下一阶段切换到其他产品板块时，再按对应板块重新确认范围、依赖和验收标准。 diff --git a/ai_knowledge_base_v2/docs/一期全链路验收记录.md b/ai_knowledge_base_v2/docs/一期全链路验收记录.md index 2b192eb..eda1ee7 100644 --- a/ai_knowledge_base_v2/docs/一期全链路验收记录.md +++ b/ai_knowledge_base_v2/docs/一期全链路验收记录.md @@ -11,7 +11,7 @@ 一期功能主链路已打通，权益管理页面已完成布局重构，用户端和后台 Agent 预览均能够按学员、主题和权益运行。自动化测试、前后端构建、数据库迁移和隔离生产环境冒烟均通过。 -当前仍有一项上线前性能风险：结束主题时会同步生成主题摘要和成长档案，本次真实模型验收耗时约 42 秒。结果能够正确生成，但等待时间容易让用户误认为操作无响应，也会放大重复点击和网络超时风险。本项不影响数据正确性，但不应按“完全无风险”关闭。 +2026-08-03 已完成主题沉淀异步化：结束主题接口只完成状态变更和任务入队，摘要、成长档案由持久化后台任务生成；重复点击返回同一任务，失败自动重试，服务重启能够恢复，最终失败可由管理员手动重试。原先约 42 秒的同步等待不再阻塞用户请求。 ## 3. 功能验收结果 @@ -24,7 +24,7 @@ | 用户端主题上下文 | 通过 | 连续消息均绑定同一主题，历史消息与主题上下文可继续使用 | | 后台 Agent 预览 | 通过 | 可选择模拟学员和具体主题，能够加载权益、主题摘要、最近消息和成长档案 | | Agent 运行追踪 | 通过 | 能看到 `load_debug_user_context`、模型路由等追踪节点 | -| 主题沉淀 | 功能通过，性能待优化 | 主题摘要与成长档案均生成成功；真实模型调用约 42 秒 | +| 主题沉淀 | 通过 | 结束主题快速入队；摘要与成长档案后台生成；支持幂等、自动重试、重启恢复和管理员手动重试 | | 求助卡 | 通过 | 可生成非空内容并记录复制状态 | | 分享稿 | 通过 | 可生成非空内容并记录复制状态 | | 周期报告 | 通过 | 入队接口约 7 毫秒返回，后台任务能够完成并保存报告 | @@ -36,12 +36,13 @@ | 验收项 | 结果 | 实际结果 | | --- | --- | --- | -| 后端自动化测试 | 通过 | 107 项测试全部通过 | -| OpenAPI 检查 | 通过 | 成功生成并检查 93 个接口路径 | +| 后端自动化测试 | 通过 | 111 项测试全部通过，包含异步入队、幂等、重试、重启恢复及推理内容过滤 | +| OpenAPI 检查 | 通过 | 成功生成并检查 95 个接口路径 | | 管理后台构建 | 通过 | 生产构建成功 | | 用户端构建 | 通过 | 生产构建成功 | -| Alembic 离线 SQL | 通过 | 成功生成 1106 行迁移 SQL，并覆盖到 `0023_simple_knowledge_route` | -| 全新数据库迁移 | 通过 | 在隔离 MySQL 数据库从零升级至最新版本，共生成 39 张表 | +| Alembic 离线 SQL | 通过 | 成功生成 1130 行迁移 SQL，并覆盖到 `0024_topic_settlement_jobs` | +| 现有数据库迁移 | 通过 | 开发 MySQL 已从 `0023_simple_knowledge_route` 升级至 `0024_topic_settlement_jobs` | +| 全新数据库迁移 | 通过 | 隔离 MySQL 从零升级到 `0024_topic_settlement_jobs`，主题任务字段和索引均已生成，验收库随后清理 | | 数据库与 Redis 就绪检查 | 通过 | `/api/ready` 返回数据库和 Redis 均已就绪，并返回 `X-Request-ID` | | 隔离生产编排 | 通过 | 使用生产镜像、独立数据库和 Redis 启动，所有容器健康 | | 生产冒烟 | 通过 | 用户端、管理后台、静态资源、健康接口和就绪接口均可访问 | @@ -52,14 +53,15 @@ 全链路验证脚本在生成 Alembic 离线 SQL 时，被部分历史迁移中的在线数据库检查阻断。已为相关迁移补充分离的离线模式逻辑，并为默认权益数据补充明确字段类型。在线迁移行为保持不变，且已通过全新 MySQL 数据库实际迁移验证。 +2026-08-03 浏览器验收时发现，个别模型可能把 `` 推理过程和多个 JSON 草稿一起返回，旧解析逻辑可能将其降级为成长档案正文。现已在结构化解析前剔除推理块，并从后向前选择最后一个可完整解析的 JSON；已补回归测试，思考过程不会再写入成长档案。 + ## 6. 上线前风险与建议 -### P1：结束主题接口等待时间过长 +### 已关闭：结束主题接口等待时间过长 -- 现象：结束主题后同步执行主题摘要和成长档案两次模型处理，本次耗时约 42 秒。 -- 影响：用户容易认为按钮没有生效；弱网或网关超时会使前端收到失败，但后端可能已经成功；重复点击可能产生重复任务。 -- 建议：把主题结束本身改为快速提交，摘要和成长档案转为后台任务；增加处理中状态、幂等键和失败重试；前端轮询或通过现有任务状态刷新结果。 -- 上线判断：内部小范围试用可继续；正式扩大用户前建议完成该项。 +- 处理：主题结束与模型生成已拆分，持久化任务由独立 Worker 执行。 +- 保护：数据库唯一约束和行锁保证重复请求只产生一条任务；最多自动重试 3 次；执行进程中断后自动恢复；后台支持手动重新入队。 +- 用户体验：主题结束后立即提示“实修记录正在后台沉淀”，用户可以继续聊天；页面停留期间会自动查询完成状态。 ### P2：前端构建体积提示