diff --git a/ai_knowledge_base_v2/apps/backend/app/api/admin_agent_records.py b/ai_knowledge_base_v2/apps/backend/app/api/admin_agent_records.py new file mode 100644 index 0000000..948b984 --- /dev/null +++ b/ai_knowledge_base_v2/apps/backend/app/api/admin_agent_records.py @@ -0,0 +1,203 @@ +from __future__ import annotations + +import json + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.core.database import get_db +from app.core.dependencies import get_current_admin +from app.core.responses import api_success +from app.models.admin import Admin +from app.models.knowledge import ( + HumanAttentionHistory, + HumanAttentionRecord, + KnowledgeRetrievalCandidate, + KnowledgeRetrievalLog, +) +from app.schemas.knowledge import AttentionUpdateRequest +from app.services.admin_service import OperationLogService + +router = APIRouter() + + +@router.get("/retrieval-log/list") +def retrieval_logs( + sessionId: int | None = Query(default=None), + knowledgeId: int | None = Query(default=None), + knowledgeCalled: bool | None = Query(default=None), + statusValue: str = Query(default="", alias="status"), + attentionCreated: bool | None = Query(default=None), + db: Session = Depends(get_db), + current_admin: Admin = Depends(get_current_admin), +) -> dict: + query = select(KnowledgeRetrievalLog).order_by(KnowledgeRetrievalLog.id.desc()).limit(300) + if sessionId is not None: + query = query.where(KnowledgeRetrievalLog.session_id == sessionId) + if knowledgeId is not None: + query = query.where(KnowledgeRetrievalLog.selected_knowledge_ids.like(f"%{knowledgeId}%")) + if knowledgeCalled is not None: + query = query.where(KnowledgeRetrievalLog.knowledge_called == (1 if knowledgeCalled else 0)) + if statusValue: + query = query.where(KnowledgeRetrievalLog.status == statusValue) + if attentionCreated is not None: + query = query.where(KnowledgeRetrievalLog.attention_created == (1 if attentionCreated else 0)) + return api_success([_retrieval_dict(item, detail=False) for item in db.scalars(query).all()]) + + +@router.get("/retrieval-log/{log_id}") +def retrieval_detail( + log_id: int, + db: Session = Depends(get_db), + current_admin: Admin = Depends(get_current_admin), +) -> dict: + log = db.get(KnowledgeRetrievalLog, log_id) + if log is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="检索日志不存在") + candidates = db.scalars( + select(KnowledgeRetrievalCandidate) + .where(KnowledgeRetrievalCandidate.retrieval_log_id == log.id) + .order_by(KnowledgeRetrievalCandidate.selected.desc(), KnowledgeRetrievalCandidate.id) + ).all() + data = _retrieval_dict(log, detail=True) + data["candidates"] = [ + { + "id": item.id, + "knowledgeId": item.knowledge_id, + "versionId": item.version_id, + "chunkId": item.chunk_id, + "sectionId": item.section_id, + "lexicalScore": item.lexical_score, + "rerankScore": item.rerank_score, + "selected": item.selected == 1, + "discardReason": item.discard_reason, + } + for item in candidates + ] + return api_success(data) + + +@router.get("/attention/list") +def attention_list( + priority: str = Query(default=""), + statusValue: str = Query(default="", alias="status"), + db: Session = Depends(get_db), + current_admin: Admin = Depends(get_current_admin), +) -> dict: + query = select(HumanAttentionRecord).order_by(HumanAttentionRecord.id.desc()).limit(300) + if priority: + query = query.where(HumanAttentionRecord.priority == priority) + if statusValue: + query = query.where(HumanAttentionRecord.status == statusValue) + return api_success([_attention_dict(item) for item in db.scalars(query).all()]) + + +@router.put("/attention/{attention_id}") +def update_attention( + attention_id: int, + payload: AttentionUpdateRequest, + db: Session = Depends(get_db), + current_admin: Admin = Depends(get_current_admin), +) -> dict: + item = db.get(HumanAttentionRecord, attention_id) + if item is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="人工关注记录不存在") + previous = item.status + item.status = payload.status + item.handler_id = current_admin.id + item.handler_note = payload.note + db.add_all( + [ + item, + HumanAttentionHistory( + attention_id=item.id, + from_status=previous, + to_status=payload.status, + note=payload.note, + operated_by=current_admin.id, + ), + ] + ) + OperationLogService.write(db, admin_id=current_admin.id, module="attention", action="update", target_id=item.id) + db.commit() + return api_success(_attention_dict(item)) + + +@router.delete("/attention/{attention_id}") +def delete_attention( + attention_id: int, + db: Session = Depends(get_db), + current_admin: Admin = Depends(get_current_admin), +) -> dict: + item = db.get(HumanAttentionRecord, attention_id) + if item is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="人工关注记录不存在") + histories = db.scalars( + select(HumanAttentionHistory).where(HumanAttentionHistory.attention_id == item.id) + ).all() + for history in histories: + db.delete(history) + db.delete(item) + OperationLogService.write(db, admin_id=current_admin.id, module="attention", action="delete", target_id=attention_id) + db.commit() + return api_success() + + +def _retrieval_dict(item: KnowledgeRetrievalLog, *, detail: bool) -> dict: + data = { + "id": item.id, + "sessionId": item.session_id, + "messageId": item.message_id, + "userId": item.user_id, + "question": item.question, + "knowledgeCalled": item.knowledge_called == 1, + "selectedKnowledgeIds": item.selected_knowledge_ids, + "queryTerms": _json(item.query_terms), + "finalSectionIds": item.final_section_ids, + "status": item.status, + "toolCostMs": item.tool_cost_ms, + "totalCostMs": item.total_cost_ms, + "errorMessage": item.error_message, + "attentionCreated": item.attention_created == 1, + "createdAt": item.created_at, + } + if detail: + data.update( + { + "catalogSnapshot": _json(item.catalog_snapshot), + "catalogHash": item.catalog_hash, + "safetyRuleVersion": item.safety_rule_version, + "toolTrace": _json(item.tool_trace), + "discardedConflicts": _json(item.discarded_conflicts), + "finalAnswer": item.final_answer, + } + ) + return data + + +def _attention_dict(item: HumanAttentionRecord) -> dict: + return { + "id": item.id, + "sessionId": item.session_id, + "messageId": item.message_id, + "userId": item.user_id, + "triggerMessage": item.trigger_message, + "problemSummary": item.problem_summary, + "triggerReason": item.trigger_reason, + "priority": item.priority, + "status": item.status, + "handlerId": item.handler_id, + "handlerNote": item.handler_note, + "createdAt": item.created_at, + "updatedAt": item.updated_at, + } + + +def _json(raw: str | None): + if not raw: + return None + try: + return json.loads(raw) + except json.JSONDecodeError: + return raw diff --git a/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge_lifecycle.py b/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge_lifecycle.py index bd2c6d3..d809e24 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge_lifecycle.py +++ b/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge_lifecycle.py @@ -291,6 +291,31 @@ def migration_summary( return api_success({"total": len(rows), "counts": counts, "items": [_knowledge_detail(item) for item in rows]}) +@router.post("/knowledge-migration/run") +async def run_migration( + knowledgeId: int | None = Query(default=None), + db: Session = Depends(get_db), + current_admin: Admin = Depends(get_current_admin), +) -> dict: + _require_super_admin(current_admin) + query = select(Knowledge).where(Knowledge.lifecycle_status == "active") + if knowledgeId is not None: + query = query.where(Knowledge.id == knowledgeId) + rows = db.scalars(query.order_by(Knowledge.id)).all() + results = [] + for knowledge in rows: + try: + job = await KnowledgePipelineService.synchronize( + db, knowledge, admin_id=current_admin.id, mode="review" + ) + results.append({"knowledgeId": knowledge.id, "ok": True, "job": _job_dict(job)}) + except Exception as exc: + results.append({"knowledgeId": knowledge.id, "ok": False, "error": str(exc)}) + OperationLogService.write(db, admin_id=current_admin.id, module="knowledge_migration", action="run", target_id=knowledgeId) + db.commit() + return api_success({"total": len(results), "success": sum(1 for item in results if item["ok"]), "results": results}) + + def _require_super_admin(admin: Admin) -> None: if admin.role is not None and admin.role.code != "super_admin": raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="仅超级管理员可以执行此操作") diff --git a/ai_knowledge_base_v2/apps/backend/app/api/admin_settings.py b/ai_knowledge_base_v2/apps/backend/app/api/admin_settings.py index 7015bda..6c07b89 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/admin_settings.py +++ b/ai_knowledge_base_v2/apps/backend/app/api/admin_settings.py @@ -19,6 +19,7 @@ from app.schemas.admin import ( ) from app.services.admin_service import OperationLogService from app.services.feishu_service import FeishuKnowledgeService +from app.services.knowledge_agent_service import KnowledgeAgentService from app.services.knowledge_service import KnowledgeScope from app.services.model_service import ModelClientService from app.services.rag_service import RagResult @@ -60,7 +61,7 @@ def reset_prompt(db: Session = Depends(get_db), current_admin: Admin = Depends(g @router.post("/agent/debug") -def debug_agent( +async def debug_agent( payload: AgentDebugRequest, db: Session = Depends(get_db), current_admin: Admin = Depends(get_current_admin), @@ -69,10 +70,21 @@ def debug_agent( if model is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="模型不存在") - scopes = _agent_knowledge_scopes(db, payload.knowledgeIds) - chunks = FeishuKnowledgeService.retrieve(payload.question, scopes, db) - prompt = _build_agent_debug_prompt(payload.promptContent, payload.question, chunks) - rag_result = RagResult(question=payload.question, knowledge_scopes=scopes, chunks=chunks, prompt=prompt) + rag_result = await KnowledgeAgentService.build_result( + db, + question=payload.question, + version_overrides=payload.knowledgeVersions, + preview_knowledge_ids=payload.knowledgeIds, + ) + rag_result = RagResult( + question=rag_result.question, + knowledge_scopes=rag_result.knowledge_scopes, + chunks=rag_result.chunks, + prompt=f"{payload.promptContent.strip()}\n\n{rag_result.prompt}", + allow_general_knowledge=rag_result.allow_general_knowledge, + retrieval_log_id=rag_result.retrieval_log_id, + tool_trace=rag_result.tool_trace, + ) result = ModelClientService.debug_model( model, rag_result, @@ -85,6 +97,8 @@ def debug_agent( "max_token": payload.maxToken, }, ) + result["retrievalTrace"] = rag_result.tool_trace or [] + result["retrievalLogId"] = rag_result.retrieval_log_id OperationLogService.write(db, admin_id=current_admin.id, module="agent", action="debug", target_id=model.id) db.commit() return api_success(result) diff --git a/ai_knowledge_base_v2/apps/backend/app/api/router.py b/ai_knowledge_base_v2/apps/backend/app/api/router.py index bcb8a4d..3e4c7d0 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/router.py +++ b/ai_knowledge_base_v2/apps/backend/app/api/router.py @@ -4,6 +4,7 @@ from fastapi import APIRouter from app.api import ( admin_auth, + admin_agent_records, admin_dashboard, admin_knowledge, admin_knowledge_lifecycle, @@ -22,6 +23,7 @@ api_router.include_router(auth.router, prefix="/auth", tags=["auth"]) api_router.include_router(user.router, prefix="/user", tags=["user"]) api_router.include_router(chat.router, prefix="/chat", tags=["chat"]) api_router.include_router(admin_auth.router, prefix="/admin", tags=["admin"]) +api_router.include_router(admin_agent_records.router, prefix="/admin", tags=["admin-agent-records"]) api_router.include_router(admin_dashboard.router, prefix="/admin", tags=["admin"]) api_router.include_router(admin_users.router, prefix="/admin", tags=["admin"]) api_router.include_router(admin_knowledge.router, prefix="/admin", tags=["admin"]) diff --git a/ai_knowledge_base_v2/apps/backend/app/models/knowledge.py b/ai_knowledge_base_v2/apps/backend/app/models/knowledge.py index 7f93f9c..6b504e3 100644 --- a/ai_knowledge_base_v2/apps/backend/app/models/knowledge.py +++ b/ai_knowledge_base_v2/apps/backend/app/models/knowledge.py @@ -7,11 +7,13 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship from app.models.base import Base, TimestampMixin +PRIMARY_KEY_TYPE = BigInteger().with_variant(Integer, "sqlite") + class Knowledge(Base, TimestampMixin): __tablename__ = "sys_knowledge" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column(String(100), nullable=False) feishu_space_id: Mapped[str] = mapped_column(String(100), nullable=False) feishu_node_id: Mapped[str] = mapped_column(String(100), nullable=False) @@ -38,7 +40,7 @@ class Knowledge(Base, TimestampMixin): class UserKnowledgePermission(Base, TimestampMixin): __tablename__ = "sys_user_kb" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) user_id: Mapped[int] = mapped_column(ForeignKey("sys_user.id"), index=True, nullable=False) knowledge_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge.id"), index=True, nullable=False) effective_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) @@ -52,7 +54,7 @@ class UserKnowledgePermission(Base, TimestampMixin): class KnowledgeSourceSnapshot(Base): __tablename__ = "sys_knowledge_source_snapshot" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) knowledge_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge.id"), index=True, nullable=False) sync_job_id: Mapped[int | None] = mapped_column(BigInteger, index=True, nullable=True) source_title: Mapped[str] = mapped_column(String(255), nullable=False) @@ -67,7 +69,7 @@ class KnowledgeVersion(Base): __tablename__ = "sys_knowledge_version" __table_args__ = (UniqueConstraint("knowledge_id", "version_no", name="uq_kb_version_no"),) - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) knowledge_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge.id"), index=True, nullable=False) snapshot_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge_source_snapshot.id"), nullable=False) version_no: Mapped[int] = mapped_column(Integer, nullable=False) @@ -86,7 +88,7 @@ class KnowledgeVersion(Base): class KnowledgeManifest(Base): __tablename__ = "sys_knowledge_manifest" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) knowledge_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge.id"), index=True, nullable=False) version_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge_version.id"), index=True, nullable=False) purpose: Mapped[str] = mapped_column(Text, nullable=False) @@ -105,7 +107,7 @@ class KnowledgeManifest(Base): class KnowledgeSection(Base): __tablename__ = "sys_knowledge_section" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) knowledge_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge.id"), index=True, nullable=False) version_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge_version.id"), index=True, nullable=False) section_key: Mapped[str] = mapped_column(String(64), nullable=False) @@ -120,7 +122,7 @@ class KnowledgeSection(Base): class KnowledgeChunk(Base): __tablename__ = "sys_knowledge_chunk" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) knowledge_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge.id"), index=True, nullable=False) version_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge_version.id"), index=True, nullable=False) section_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge_section.id"), index=True, nullable=False) @@ -138,7 +140,7 @@ class KnowledgeChunk(Base): class KnowledgeCard(Base): __tablename__ = "sys_knowledge_card" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) knowledge_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge.id"), index=True, nullable=False) version_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge_version.id"), index=True, nullable=False) section_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge_section.id"), index=True, nullable=False) @@ -161,7 +163,7 @@ class KnowledgeCard(Base): class KnowledgeSyncJob(Base): __tablename__ = "sys_knowledge_sync_job" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) knowledge_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge.id"), index=True, nullable=False) mode: Mapped[str] = mapped_column(String(30), nullable=False) status: Mapped[str] = mapped_column(String(30), index=True, nullable=False) @@ -178,7 +180,7 @@ class KnowledgeSyncJob(Base): class KnowledgeReviewRecord(Base): __tablename__ = "sys_knowledge_review_record" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) knowledge_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge.id"), index=True, nullable=False) version_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge_version.id"), index=True, nullable=False) card_id: Mapped[int | None] = mapped_column(ForeignKey("sys_knowledge_card.id"), nullable=True) @@ -191,7 +193,7 @@ class KnowledgeReviewRecord(Base): class KnowledgePublishLog(Base): __tablename__ = "sys_knowledge_publish_log" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) knowledge_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge.id"), index=True, nullable=False) from_version_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True) to_version_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True) @@ -205,7 +207,7 @@ class KnowledgePublishLog(Base): class KnowledgeRetrievalLog(Base): __tablename__ = "sys_knowledge_retrieval_log" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) session_id: Mapped[int | None] = mapped_column(BigInteger, index=True, nullable=True) message_id: Mapped[int | None] = mapped_column(BigInteger, index=True, nullable=True) user_id: Mapped[int | None] = mapped_column(BigInteger, index=True, nullable=True) @@ -231,7 +233,7 @@ class KnowledgeRetrievalLog(Base): class KnowledgeRetrievalCandidate(Base): __tablename__ = "sys_knowledge_retrieval_candidate" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) retrieval_log_id: Mapped[int] = mapped_column(ForeignKey("sys_knowledge_retrieval_log.id"), index=True, nullable=False) knowledge_id: Mapped[int] = mapped_column(BigInteger, index=True, nullable=False) version_id: Mapped[int] = mapped_column(BigInteger, index=True, nullable=False) @@ -246,7 +248,7 @@ class KnowledgeRetrievalCandidate(Base): class HumanAttentionRecord(Base): __tablename__ = "sys_human_attention_record" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) session_id: Mapped[int] = mapped_column(BigInteger, index=True, nullable=False) message_id: Mapped[int] = mapped_column(BigInteger, index=True, nullable=False) user_id: Mapped[int] = mapped_column(BigInteger, index=True, nullable=False) @@ -264,7 +266,7 @@ class HumanAttentionRecord(Base): class HumanAttentionHistory(Base): __tablename__ = "sys_human_attention_history" - id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True) attention_id: Mapped[int] = mapped_column(ForeignKey("sys_human_attention_record.id"), index=True, nullable=False) from_status: Mapped[str | None] = mapped_column(String(20), nullable=True) to_status: Mapped[str] = mapped_column(String(20), nullable=False) diff --git a/ai_knowledge_base_v2/apps/backend/app/schemas/admin.py b/ai_knowledge_base_v2/apps/backend/app/schemas/admin.py index 506af45..331078a 100644 --- a/ai_knowledge_base_v2/apps/backend/app/schemas/admin.py +++ b/ai_knowledge_base_v2/apps/backend/app/schemas/admin.py @@ -82,6 +82,7 @@ class AgentDebugRequest(BaseModel): promptContent: str = Field(min_length=1) modelId: int = Field(gt=0) knowledgeIds: list[int] = Field(default_factory=list) + knowledgeVersions: dict[int, int] = Field(default_factory=dict) question: str = Field(min_length=1, max_length=2000) temperature: float | None = Field(default=None, ge=0, le=2) topP: float | None = Field(default=None, ge=0, le=1) diff --git a/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py b/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py index 9e59fd9..4993b03 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py +++ b/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py @@ -11,10 +11,12 @@ from sqlalchemy import select from sqlalchemy.orm import Session from app.models.chat import ChatMessage +from app.models.knowledge import KnowledgeRetrievalLog from app.models.user import User from app.services.ai_request_log_service import AiRequestLogService from app.services.chat_service import ChatService, _title_from_question from app.services.external_errors import ExternalServiceError +from app.services.human_attention_service import HumanAttentionService from app.services.model_stream_service import ModelStreamService from app.services.rag_async_service import AsyncRagService from app.services.rag_service import RagService @@ -83,6 +85,7 @@ class ChatStreamService: error_message=str(exc), retrieved_chunks=rag_result.chunks if rag_result is not None else None, ) + _mark_retrieval_failed(db, rag_result, str(exc), cost_ms) db.commit() raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc @@ -102,6 +105,7 @@ class ChatStreamService: error_message="模型未返回有效内容", retrieved_chunks=rag_result.chunks if rag_result is not None else None, ) + _mark_retrieval_failed(db, rag_result, "模型未返回有效内容", cost_ms) db.commit() raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="模型未返回有效内容") @@ -188,6 +192,7 @@ class ChatStreamService: normalized_question, history=history, session_summary=getattr(session, "summary", None), + session_id=session.id, ) model_response = ModelStreamService.stream_async(db, rag_result) async for chunk in model_response.chunks: @@ -304,9 +309,52 @@ def _write_success( cost_ms=cost_ms, retrieved_chunks=rag_result.chunks if rag_result is not None else None, ) + if rag_result is not None and rag_result.retrieval_log_id: + retrieval_log = db.get(KnowledgeRetrievalLog, rag_result.retrieval_log_id) + if retrieval_log: + attention = HumanAttentionService.create_if_needed( + db, + session_id=session.id, + message_id=user_message_id_for_attention(db, session.id, assistant_message.id), + user_id=user.id, + question=question, + answer=answer, + knowledge_missing=not rag_result.allow_general_knowledge and not rag_result.is_hit, + ) + retrieval_log.message_id = assistant_message.id + retrieval_log.final_answer = answer + retrieval_log.status = "success" + retrieval_log.total_cost_ms = cost_ms + retrieval_log.attention_created = 1 if attention else 0 + db.add(retrieval_log) db.commit() +def user_message_id_for_attention(db: Session, session_id: int, assistant_message_id: int) -> int: + message_id = db.scalar( + select(ChatMessage.id) + .where( + ChatMessage.session_id == session_id, + ChatMessage.role == "user", + ChatMessage.id < assistant_message_id, + ) + .order_by(ChatMessage.id.desc()) + .limit(1) + ) + return int(message_id or assistant_message_id) + + +def _mark_retrieval_failed(db: Session, rag_result, error_message: str, cost_ms: int) -> None: + if rag_result is None or not rag_result.retrieval_log_id: + return + retrieval_log = db.get(KnowledgeRetrievalLog, rag_result.retrieval_log_id) + if retrieval_log: + retrieval_log.status = "failed" + retrieval_log.error_message = error_message + retrieval_log.total_cost_ms = cost_ms + db.add(retrieval_log) + + def _build_rag_result(db: Session, user: User, question: str, history: list[ChatMessage], summary: str | None): parameters = signature(RagService.build_result).parameters if "history" in parameters: diff --git a/ai_knowledge_base_v2/apps/backend/app/services/human_attention_service.py b/ai_knowledge_base_v2/apps/backend/app/services/human_attention_service.py new file mode 100644 index 0000000..ec41ada --- /dev/null +++ b/ai_knowledge_base_v2/apps/backend/app/services/human_attention_service.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from sqlalchemy.orm import Session + +from app.models.knowledge import HumanAttentionHistory, HumanAttentionRecord + +URGENT_TERMS = ("自杀", "不想活", "自伤", "伤害别人", "杀人", "现实危险") +IMPORTANT_TERMS = ("绝望", "撑不住", "崩溃", "非常痛苦", "反复失败", "没有办法") +CONTACT_TERMS = ("联系老师", "找老师", "人工帮助", "人工客服") + + +class HumanAttentionService: + @staticmethod + def create_if_needed( + db: Session, + *, + session_id: int, + message_id: int, + user_id: int, + question: str, + answer: str, + knowledge_missing: bool, + ) -> HumanAttentionRecord | None: + priority = None + reason = None + if any(term in question for term in URGENT_TERMS): + priority, reason = "urgent", "检测到现实危险或自伤伤人风险" + elif any(term in question for term in IMPORTANT_TERMS): + priority, reason = "important", "用户表达持续或强烈痛苦" + elif any(term in question for term in CONTACT_TERMS): + priority, reason = "normal", "用户主动要求联系老师或人工" + elif knowledge_missing: + priority, reason = "normal", "课程或业务问题缺少可靠正式知识" + if priority is None: + return None + record = HumanAttentionRecord( + session_id=session_id, + message_id=message_id, + user_id=user_id, + trigger_message=question, + problem_summary=_summary(question), + trigger_reason=reason, + priority=priority, + status="pending", + ) + db.add(record) + db.flush() + db.add( + HumanAttentionHistory( + attention_id=record.id, + from_status=None, + to_status="pending", + note=f"系统自动创建；回答摘要：{_summary(answer, 200)}", + operated_by=0, + ) + ) + return record + + +def _summary(text: str, limit: int = 120) -> str: + value = " ".join(text.split()) + return value if len(value) <= limit else value[:limit].rstrip() + "…" diff --git a/ai_knowledge_base_v2/apps/backend/app/services/knowledge_agent_service.py b/ai_knowledge_base_v2/apps/backend/app/services/knowledge_agent_service.py new file mode 100644 index 0000000..180bd28 --- /dev/null +++ b/ai_knowledge_base_v2/apps/backend/app/services/knowledge_agent_service.py @@ -0,0 +1,293 @@ +from __future__ import annotations + +import asyncio +import hashlib +import json +import math +import re +from dataclasses import dataclass +from time import perf_counter + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.core.config import get_settings +from app.models.ai_config import ModelConfig +from app.models.knowledge import ( + Knowledge, + KnowledgeChunk, + KnowledgeManifest, + KnowledgeRetrievalCandidate, + KnowledgeRetrievalLog, + KnowledgeSection, + KnowledgeVersion, +) +from app.services.knowledge_pipeline_service import expand_synonyms, extract_terms +from app.services.knowledge_service import KnowledgeScope +from app.services.model_service import _call_configured_model, _system_config_bool +from app.services.rag_service import PromptService, RagResult, RetrievedChunk + +SAFETY_RULE_VERSION = "minimum-safety-v1" +MAX_LEXICAL_CANDIDATES = 12 +MAX_SELECTED_SECTIONS = 4 +BUSINESS_MARKERS = {"课程", "大本营", "训练营", "老师", "卢慧", "功课", "学员", "课堂", "练习", "觉察", "内在"} + + +@dataclass +class Candidate: + chunk: KnowledgeChunk + section: KnowledgeSection + knowledge: Knowledge + version: KnowledgeVersion + lexical_score: float + rerank_score: float | None = None + selected: bool = False + discard_reason: str | None = None + + +class KnowledgeAgentService: + @classmethod + async def build_result( + cls, + db: Session, + *, + question: str, + history=None, + session_summary: str | None = None, + session_id: int | None = None, + user_id: int | None = None, + version_overrides: dict[int, int] | None = None, + preview_knowledge_ids: list[int] | None = None, + ) -> RagResult: + started = perf_counter() + catalog = cls.get_knowledge_catalog( + db, version_overrides=version_overrides, preview_knowledge_ids=preview_knowledge_ids + ) + catalog_json = json.dumps(catalog, ensure_ascii=False, sort_keys=True, default=str) + log = KnowledgeRetrievalLog( + session_id=session_id, + user_id=user_id, + question=question.strip(), + catalog_snapshot=catalog_json, + catalog_hash=hashlib.sha256(catalog_json.encode()).hexdigest(), + safety_rule_version=SAFETY_RULE_VERSION, + status="running", + ) + db.add(log) + db.flush() + scopes = [KnowledgeScope(id=item["knowledgeId"], name=item["name"], feishu_space_id="", feishu_node_id="") for item in catalog] + trace: list[dict] = [{"tool": "get_knowledge_catalog", "count": len(catalog)}] + need_knowledge, selected_ids, reason = cls._decide(question, catalog) + trace.append({"tool": "agent_decision", "needKnowledge": need_knowledge, "selectedKnowledgeIds": selected_ids, "reason": reason}) + chunks: list[RetrievedChunk] = [] + try: + if need_knowledge and selected_ids: + terms = cls._query_terms(question) + candidates = cls.search_knowledge(db, terms, selected_ids, catalog) + trace.append({"tool": "search_knowledge", "queryTerms": terms, "candidateCount": len(candidates)}) + await cls._rerank(db, question, candidates, trace) + selected = cls._select_sections(candidates) + chunks = [ + RetrievedChunk( + knowledge_id=item.knowledge.id, + knowledge_name=item.knowledge.name, + title=item.section.title, + content=cls._protect_content(item.section.content), + version_id=item.version.id, + section_id=item.section.id, + chunk_id=item.chunk.id, + score=item.rerank_score if item.rerank_score is not None else item.lexical_score, + ) + for item in selected + ] + trace.append({"tool": "read_knowledge_sections", "sectionIds": [item.section.id for item in selected]}) + cls._persist_candidates(db, log.id, candidates) + log.knowledge_called = 1 + log.selected_knowledge_ids = ",".join(str(item) for item in selected_ids) + log.query_terms = json.dumps(terms, ensure_ascii=False) + log.final_section_ids = ",".join(str(item.section.id) for item in selected) + log.tool_trace = json.dumps(trace, ensure_ascii=False) + log.tool_cost_ms = int((perf_counter() - started) * 1000) + log.status = "prepared" + db.add(log) + return RagResult( + question=question, + knowledge_scopes=scopes, + chunks=chunks, + prompt=PromptService.build_prompt(db, question, chunks, history, session_summary), + allow_general_knowledge=not need_knowledge, + retrieval_log_id=log.id, + tool_trace=trace, + ) + except Exception as exc: + log.status = "failed" + log.error_message = str(exc) + log.tool_trace = json.dumps(trace, ensure_ascii=False) + log.tool_cost_ms = int((perf_counter() - started) * 1000) + db.add(log) + raise + + @staticmethod + def get_knowledge_catalog( + db: Session, + *, + version_overrides: dict[int, int] | None = None, + preview_knowledge_ids: list[int] | None = None, + ) -> list[dict]: + query = select(Knowledge).where(Knowledge.lifecycle_status == "active") + if preview_knowledge_ids is None: + query = query.where( + Knowledge.status == 1, + Knowledge.current_version_id.is_not(None), + Knowledge.manifest_confirmed == 1, + Knowledge.source_status == "normal", + ) + elif preview_knowledge_ids: + query = query.where(Knowledge.id.in_(preview_knowledge_ids)) + rows = db.scalars(query.order_by(Knowledge.id)).all() + catalog: list[dict] = [] + for knowledge in rows: + version_id = (version_overrides or {}).get(knowledge.id) or knowledge.current_version_id or knowledge.pending_version_id + if not version_id: + continue + version = db.get(KnowledgeVersion, version_id) + manifest = db.scalar(select(KnowledgeManifest).where(KnowledgeManifest.version_id == version_id)) + if version is None or manifest is None: + continue + catalog.append( + { + "knowledgeId": knowledge.id, + "name": knowledge.name, + "type": knowledge.knowledge_type, + "purpose": manifest.purpose, + "applicableQuestions": manifest.applicable_questions, + "inapplicableQuestions": manifest.inapplicable_questions, + "coreTopics": manifest.core_topics, + "boundaries": manifest.boundaries, + "versionId": version.id, + "versionNo": version.version_no, + "publishedAt": version.published_at, + } + ) + return catalog + + @classmethod + def search_knowledge(cls, db: Session, terms: list[str], selected_ids: list[int], catalog: list[dict]) -> list[Candidate]: + versions_by_kb = {item["knowledgeId"]: item["versionId"] for item in catalog} + version_ids = [versions_by_kb[item] for item in selected_ids if item in versions_by_kb] + if not version_ids: + return [] + chunks = db.scalars(select(KnowledgeChunk).where(KnowledgeChunk.version_id.in_(version_ids))).all() + sections = {item.id: item for item in db.scalars(select(KnowledgeSection).where(KnowledgeSection.version_id.in_(version_ids))).all()} + knowledge_rows = {item.id: item for item in db.scalars(select(Knowledge).where(Knowledge.id.in_(selected_ids))).all()} + versions = {item.id: item for item in db.scalars(select(KnowledgeVersion).where(KnowledgeVersion.id.in_(version_ids))).all()} + frequencies = {term: sum(1 for chunk in chunks if term in f"{chunk.title}{chunk.normalized_text}") for term in terms} + candidates: list[Candidate] = [] + for chunk in chunks: + section, knowledge, version = sections.get(chunk.section_id), knowledge_rows.get(chunk.knowledge_id), versions.get(chunk.version_id) + if not section or not knowledge or not version: + continue + haystack = f"{chunk.title} {chunk.normalized_text} {chunk.content}".lower() + score = 0.0 + for term in terms: + count = haystack.count(term) + if count: + idf = math.log((len(chunks) + 1) / (frequencies.get(term, 0) + 1)) + 1 + score += (count / (count + 1.2)) * idf * (2.4 if term in chunk.title.lower() else 1.0) + if score > 0: + candidates.append(Candidate(chunk, section, knowledge, version, score)) + candidates.sort(key=lambda item: (item.lexical_score, item.version.published_at or item.version.created_at), reverse=True) + return candidates[:MAX_LEXICAL_CANDIDATES] + + @classmethod + async def _rerank(cls, db: Session, question: str, candidates: list[Candidate], trace: list[dict]) -> None: + if not candidates: + return + model = db.scalar(select(ModelConfig).where(ModelConfig.enabled == 1).order_by(ModelConfig.id.desc()).limit(1)) + if model is None or _system_config_bool(db, "mock_model_enabled", get_settings().mock_model_enabled): + trace.append({"tool": "model_rerank", "status": "lexical_fallback", "reason": "模型未启用或处于 mock"}) + return + prompt = ( + "你是知识检索重排器。根据用户问题给候选打0到100相关分，只返回JSON：" + '{"scores":[{"index":1,"score":90}]}。不要回答问题。\n' + f"用户问题：{question}\n候选：\n" + + "\n".join(f"[{index}] {item.knowledge.name}/{item.chunk.title}: {item.chunk.content[:350]}" for index, item in enumerate(candidates, 1)) + ) + rag = RagResult(question=question, knowledge_scopes=[], chunks=[], prompt=prompt, allow_general_knowledge=True) + try: + raw = await asyncio.to_thread(_call_configured_model, model, rag, allow_no_hit=True) + payload = json.loads(_extract_json(raw)) + score_map = {int(item["index"]): max(0.0, min(100.0, float(item["score"]))) for item in payload.get("scores", [])} + for index, item in enumerate(candidates, 1): + item.rerank_score = score_map.get(index) + candidates.sort(key=lambda item: item.rerank_score if item.rerank_score is not None else item.lexical_score, reverse=True) + trace.append({"tool": "model_rerank", "status": "success", "candidateCount": len(candidates)}) + except Exception as exc: + trace.append({"tool": "model_rerank", "status": "lexical_fallback", "reason": str(exc)[:300]}) + + @staticmethod + def _select_sections(candidates: list[Candidate]) -> list[Candidate]: + selected: list[Candidate] = [] + seen: set[int] = set() + for item in candidates: + score = item.rerank_score if item.rerank_score is not None else item.lexical_score * 10 + if score < 8: + item.discard_reason = "相关性不足" + elif item.section.id in seen: + item.discard_reason = "同一父章节已有更高分候选" + elif len(selected) >= MAX_SELECTED_SECTIONS: + item.discard_reason = "超过本轮章节数量限制" + else: + item.selected = True + selected.append(item) + seen.add(item.section.id) + return selected + + @staticmethod + def _persist_candidates(db: Session, log_id: int, candidates: list[Candidate]) -> None: + for item in candidates: + db.add(KnowledgeRetrievalCandidate( + retrieval_log_id=log_id, + knowledge_id=item.knowledge.id, + version_id=item.version.id, + chunk_id=item.chunk.id, + section_id=item.section.id, + lexical_score=f"{item.lexical_score:.6f}", + rerank_score=f"{item.rerank_score:.4f}" if item.rerank_score is not None else None, + selected=1 if item.selected else 0, + discard_reason=item.discard_reason, + )) + + @staticmethod + def _decide(question: str, catalog: list[dict]) -> tuple[bool, list[int], str]: + business = any(marker in question for marker in BUSINESS_MARKERS) + if not catalog: + return business, [], "当前没有可用正式知识库" + terms = set(extract_terms(question)) + ranked: list[tuple[float, int]] = [] + for item in catalog: + manifest_terms = set(extract_terms(" ".join(str(item.get(key, "")) for key in ("name", "purpose", "applicableQuestions", "coreTopics")))) + overlap = len(terms & manifest_terms) / max(1, min(len(terms), 18)) + if overlap > 0 or business: + ranked.append((overlap, int(item["knowledgeId"]))) + ranked.sort(reverse=True) + selected = [item_id for score, item_id in ranked if score >= 0.04][:4] + if business and not selected: + selected = [item_id for _, item_id in ranked[:3]] + return business or bool(selected), selected, "涉及课程/老师/业务知识" if business else "问题与知识目录主题相关" + + @staticmethod + def _query_terms(question: str) -> list[str]: + base = extract_terms(question) + return list(dict.fromkeys(base + expand_synonyms(base)))[:40] + + @staticmethod + def _protect_content(content: str) -> str: + return content if len(content) <= 3200 else content[:3200].rstrip() + "\n[章节内容已按保护规则截断]" + + +def _extract_json(value: str) -> str: + match = re.search(r"\{[\s\S]*\}", value) + if not match: + raise ValueError("重排模型未返回 JSON") + return match.group(0) diff --git a/ai_knowledge_base_v2/apps/backend/app/services/model_service.py b/ai_knowledge_base_v2/apps/backend/app/services/model_service.py index d7bd41f..aa2d463 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/model_service.py +++ b/ai_knowledge_base_v2/apps/backend/app/services/model_service.py @@ -37,7 +37,7 @@ class ModelClientService: if model is None: raise ExternalServiceError("未启用可用模型，请先在模型管理中启用一个模型。", provider="model") model_name = model.model_name - answer = _call_configured_model(model, rag_result) + answer = _call_configured_model(model, rag_result, allow_no_hit=rag_result.allow_general_knowledge) return ModelCompletion( answer=answer, model_id=model.id if model is not None else None, @@ -185,7 +185,7 @@ def _call_openai_compatible_model(model: ModelConfig, rag_result: RagResult) -> payload = { "model": model.model_name, "messages": [ - {"role": "system", "content": "你是大本营答疑助手，只能基于已提供的知识片段回答。"}, + {"role": "system", "content": _agent_system_prompt()}, {"role": "user", "content": rag_result.prompt}, ], "stream": False, @@ -218,7 +218,7 @@ def _call_anthropic_messages(model: ModelConfig, rag_result: RagResult) -> str: payload = { "model": model.model_name, "max_tokens": model.max_token or 1024, - "system": "你是大本营答疑助手，只能基于已提供的知识片段回答。", + "system": _agent_system_prompt(), "messages": [{"role": "user", "content": rag_result.prompt}], "stream": False, } @@ -252,7 +252,7 @@ def _call_gemini_generate_content(model: ModelConfig, rag_result: RagResult) -> payload = { "systemInstruction": { - "parts": [{"text": "你是大本营答疑助手，只能基于已提供的知识片段回答。"}] + "parts": [{"text": _agent_system_prompt()}] }, "contents": [{"role": "user", "parts": [{"text": rag_result.prompt}]}], } @@ -337,7 +337,7 @@ def _call_minimax(model: ModelConfig, rag_result: RagResult) -> str: "bot_setting": [ { "bot_name": "大本营答疑助手", - "content": "你是大本营答疑助手，只能基于已提供的知识片段回答。", + "content": _agent_system_prompt(), } ], } @@ -435,6 +435,13 @@ def _model_api_key(model: ModelConfig) -> str: return SecretService.decrypt(model.api_key) +def _agent_system_prompt() -> str: + return ( + "你是大本营答疑助手。必须遵守最低安全规则。涉及课程、老师观点和公司业务时只能依据提供的正式知识，" + "没有可靠知识必须明确依据不足；普通常识可以谨慎回答。不得编造老师观点，不得输出整篇课程资料或大段连续原文。" + ) + + def _put_if_not_none(target: dict[str, Any], key: str, value: Any) -> None: if value is not None: target[key] = value diff --git a/ai_knowledge_base_v2/apps/backend/app/services/model_stream_service.py b/ai_knowledge_base_v2/apps/backend/app/services/model_stream_service.py index 528053d..500d93b 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/model_stream_service.py +++ b/ai_knowledge_base_v2/apps/backend/app/services/model_stream_service.py @@ -15,6 +15,7 @@ from app.models.ai_config import ModelConfig from app.services.external_errors import ExternalServiceError from app.services.model_service import ( _anthropic_headers, + _agent_system_prompt, _auth_headers, _call_configured_model, _decimal_to_float, @@ -60,7 +61,7 @@ class ModelStreamService: chunks=_chunk_text(_mock_answer(rag_result)), ) - if not rag_result.is_hit: + if not rag_result.is_hit and not rag_result.allow_general_knowledge: return StreamingModelResponse( model_id=model.id if model is not None else None, model_name=model.model_name if model is not None else "no-hit", @@ -94,7 +95,7 @@ class ModelStreamService: chunks=_async_chunk_text(_mock_answer(rag_result)), ) - if not rag_result.is_hit: + if not rag_result.is_hit and not rag_result.allow_general_knowledge: return AsyncStreamingModelResponse( model_id=model.id if model is not None else None, model_name=model.model_name if model is not None else "no-hit", @@ -127,18 +128,20 @@ def _get_enabled_model(db: Session) -> ModelConfig | None: def _stream_configured_model(model: ModelConfig, rag_result: RagResult) -> Iterator[str]: api_type = model.api_type or "openai_compatible" if model.stream_enabled != 1: - return _chunk_text(_call_configured_model(model, rag_result)) + return _chunk_text(_call_configured_model(model, rag_result, allow_no_hit=rag_result.allow_general_knowledge)) if api_type == "anthropic_messages": return _stream_anthropic_messages(model, rag_result) if api_type == "openai_compatible": return _stream_openai_compatible_model(model, rag_result) - return _chunk_text(_call_configured_model(model, rag_result)) + return _chunk_text(_call_configured_model(model, rag_result, allow_no_hit=rag_result.allow_general_knowledge)) async def _stream_configured_model_async(model: ModelConfig, rag_result: RagResult) -> AsyncIterator[str]: api_type = model.api_type or "openai_compatible" if model.stream_enabled != 1: - answer = await asyncio.to_thread(_call_configured_model, model, rag_result) + answer = await asyncio.to_thread( + _call_configured_model, model, rag_result, allow_no_hit=rag_result.allow_general_knowledge + ) async for chunk in _async_chunk_text(answer): yield chunk return @@ -153,7 +156,9 @@ async def _stream_configured_model_async(model: ModelConfig, rag_result: RagResu yield chunk return - answer = await asyncio.to_thread(_call_configured_model, model, rag_result) + answer = await asyncio.to_thread( + _call_configured_model, model, rag_result, allow_no_hit=rag_result.allow_general_knowledge + ) async for chunk in _async_chunk_text(answer): yield chunk @@ -162,7 +167,7 @@ def _stream_openai_compatible_model(model: ModelConfig, rag_result: RagResult) - payload: dict[str, Any] = { "model": model.model_name, "messages": [ - {"role": "system", "content": "你是大本营答疑助手，只能基于已提供的知识片段回答。"}, + {"role": "system", "content": _agent_system_prompt()}, {"role": "user", "content": rag_result.prompt}, ], "max_tokens": model.max_token or 1024, @@ -211,7 +216,7 @@ def _openai_stream_payload(model: ModelConfig, rag_result: RagResult) -> dict[st payload: dict[str, Any] = { "model": model.model_name, "messages": [ - {"role": "system", "content": "你是大本营答疑助手，只能基于已提供的知识片段回答。"}, + {"role": "system", "content": _agent_system_prompt()}, {"role": "user", "content": rag_result.prompt}, ], "max_tokens": model.max_token or 1024, @@ -317,7 +322,7 @@ def _stream_anthropic_messages(model: ModelConfig, rag_result: RagResult) -> Ite payload: dict[str, Any] = { "model": model.model_name, "max_tokens": model.max_token or 1024, - "system": "你是大本营答疑助手，只能基于已提供的知识片段回答。", + "system": _agent_system_prompt(), "messages": [{"role": "user", "content": rag_result.prompt}], } _put_if_not_none(payload, "temperature", _decimal_to_float(model.temperature)) @@ -361,7 +366,7 @@ def _anthropic_stream_payload(model: ModelConfig, rag_result: RagResult) -> dict payload: dict[str, Any] = { "model": model.model_name, "max_tokens": model.max_token or 1024, - "system": "你是大本营答疑助手，只能基于已提供的知识片段回答。", + "system": _agent_system_prompt(), "messages": [{"role": "user", "content": rag_result.prompt}], } _put_if_not_none(payload, "temperature", _decimal_to_float(model.temperature)) diff --git a/ai_knowledge_base_v2/apps/backend/app/services/rag_async_service.py b/ai_knowledge_base_v2/apps/backend/app/services/rag_async_service.py index 2f50720..328ec50 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/rag_async_service.py +++ b/ai_knowledge_base_v2/apps/backend/app/services/rag_async_service.py @@ -1,14 +1,11 @@ from __future__ import annotations -from inspect import signature - from sqlalchemy.orm import Session from app.models.chat import ChatMessage from app.models.user import User -from app.services.feishu_async_service import AsyncFeishuKnowledgeService -from app.services.knowledge_service import KnowledgeAccessService -from app.services.rag_service import PromptService, RagResult +from app.services.knowledge_agent_service import KnowledgeAgentService +from app.services.rag_service import RagResult class AsyncRagService: @@ -19,21 +16,13 @@ class AsyncRagService: question: str, history: list[ChatMessage] | None = None, session_summary: str | None = None, + session_id: int | None = None, ) -> RagResult: - scopes = KnowledgeAccessService.get_allowed_knowledge(db, user) - chunks = await AsyncFeishuKnowledgeService.retrieve(question, scopes, db) - prompt = _build_prompt(db, question, chunks, history, session_summary) - return RagResult(question=question, knowledge_scopes=scopes, chunks=chunks, prompt=prompt) - - -def _build_prompt( - db: Session, - question: str, - chunks, - history: list[ChatMessage] | None, - session_summary: str | None, -) -> str: - parameters = signature(PromptService.build_prompt).parameters - if "history" in parameters: - return PromptService.build_prompt(db, question, chunks, history, session_summary) - return PromptService.build_prompt(db, question, chunks) + return await KnowledgeAgentService.build_result( + db, + question=question, + history=history, + session_summary=session_summary, + session_id=session_id, + user_id=user.id, + ) diff --git a/ai_knowledge_base_v2/apps/backend/app/services/rag_service.py b/ai_knowledge_base_v2/apps/backend/app/services/rag_service.py index f11116b..22b2efb 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/rag_service.py +++ b/ai_knowledge_base_v2/apps/backend/app/services/rag_service.py @@ -26,6 +26,10 @@ class RetrievedChunk: title: str content: str source_url: str | None = None + version_id: int | None = None + section_id: int | None = None + chunk_id: int | None = None + score: float | None = None @dataclass(frozen=True) @@ -34,6 +38,9 @@ class RagResult: knowledge_scopes: list[KnowledgeScope] chunks: list[RetrievedChunk] prompt: str + allow_general_knowledge: bool = False + retrieval_log_id: int | None = None + tool_trace: list[dict] | None = None @property def is_hit(self) -> bool: @@ -41,7 +48,8 @@ class RagResult: @property def knowledge_ids(self) -> str: - return ",".join(str(scope.id) for scope in self.knowledge_scopes) + ids = list(dict.fromkeys(chunk.knowledge_id for chunk in self.chunks)) + return ",".join(str(item) for item in ids) class RagService: @@ -61,8 +69,7 @@ class RagService: class PromptService: _default_prompt = ( - "你是大本营答疑助手。你只能基于提供的知识片段回答，不能编造。" - "如果知识片段之间有冲突，需要说明不同来源的观点。" + "你是大本营答疑助手。必须遵守系统安全边界，涉及课程、老师观点和公司业务时只能依据可靠知识，不能编造。" ) @classmethod @@ -78,13 +85,18 @@ class PromptService: # 知识库上下文 context = "\n\n".join( - f"[知识片段 {index}] 来源：{chunk.knowledge_name} / {chunk.title}\n{chunk.content}" + f"[已回读完整章节 {index}] {chunk.title}\n{chunk.content}" for index, chunk in enumerate(chunks, start=1) ) if not context: - context = "未检索到相关知识片段。" + context = "本轮没有可靠的正式知识章节。对于一般常识可以谨慎回答；涉及课程、老师观点或公司业务时必须说明依据不足，不得编造。" parts = [prompt] + parts.append( + "[不可关闭的最低安全规则 v1]\n" + "现实危险、自伤伤人风险应优先建议立即寻求线下专业帮助；医疗、法律、财务问题不得给出替代专业意见的结论；" + "不得伪造老师观点或课程内容；不得输出整篇课程文章、大段连续原文，也不得通过多轮拼接还原完整资料。" + ) # 对话历史：摘要 + 最近 N 轮原文 history_part = cls._build_history_section(history, session_summary) diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py b/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py new file mode 100644 index 0000000..3368992 --- /dev/null +++ b/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +import asyncio + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session +from sqlalchemy.pool import StaticPool + +from app.models import Base +from app.models.knowledge import ( + Knowledge, + KnowledgeChunk, + KnowledgeManifest, + KnowledgeSection, + KnowledgeSourceSnapshot, + KnowledgeVersion, +) +from app.services.knowledge_agent_service import KnowledgeAgentService + + +def _database() -> Session: + engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool) + Base.metadata.create_all(engine) + return Session(engine) + + +def _add_published_knowledge(db: Session, *, knowledge_id: int, name: str, open_status: int = 1) -> Knowledge: + knowledge = Knowledge( + id=knowledge_id, + name=name, + feishu_space_id="space", + feishu_node_id=f"node-{knowledge_id}", + status=open_status, + source_status="normal", + manifest_confirmed=1, + knowledge_type="course", + review_mode="manual", + ) + db.add(knowledge) + db.flush() + snapshot = KnowledgeSourceSnapshot( + knowledge_id=knowledge.id, + source_title=name, + source_content="家长沟通需要先稳定自己的情绪，再倾听孩子的真实困难。", + content_hash=f"hash-{knowledge_id}", + source_identifier=f"space/node-{knowledge_id}", + ) + db.add(snapshot) + db.flush() + version = KnowledgeVersion( + knowledge_id=knowledge.id, + snapshot_id=snapshot.id, + version_no=1, + status="published", + quality_status="passed", + processing_rule_version="test-v1", + ) + db.add(version) + db.flush() + knowledge.current_version_id = version.id + db.add( + KnowledgeManifest( + knowledge_id=knowledge.id, + version_id=version.id, + purpose="解答家长与孩子沟通、学习动力相关课程问题", + applicable_questions="孩子学习动力、亲子沟通", + inapplicable_questions="天气和交通", + core_topics="家长情绪、倾听孩子、学习动力", + boundaries="不输出完整课程资料", + content_hash=f"manifest-{knowledge_id}", + confirmed=1, + ) + ) + section = KnowledgeSection( + knowledge_id=knowledge.id, + version_id=version.id, + section_key="S0001", + title="家长沟通的第一步", + content="家长和学习动力不足的孩子沟通时，第一步是先稳定自己的焦虑，再倾听孩子遇到的具体困难。", + source_start=0, + source_end=44, + sort_order=1, + content_hash=f"section-{knowledge_id}", + ) + db.add(section) + db.flush() + db.add( + KnowledgeChunk( + knowledge_id=knowledge.id, + version_id=version.id, + section_id=section.id, + title=section.title, + content=section.content, + normalized_text="家长 沟通 学习 动力 孩子 焦虑 倾听", + keywords='["家长","沟通","学习动力"]', + synonyms='["父母","交流"]', + source_start=0, + source_end=44, + sort_order=1, + content_hash=f"chunk-{knowledge_id}", + ) + ) + db.commit() + return knowledge + + +def test_catalog_only_contains_open_published_knowledge(): + with _database() as db: + _add_published_knowledge(db, knowledge_id=1, name="开放课程", open_status=1) + _add_published_knowledge(db, knowledge_id=2, name="关闭课程", open_status=0) + catalog = KnowledgeAgentService.get_knowledge_catalog(db) + assert [item["knowledgeId"] for item in catalog] == [1] + + +def test_common_question_skips_knowledge_tools(): + with _database() as db: + _add_published_knowledge(db, knowledge_id=1, name="亲子课程") + result = asyncio.run(KnowledgeAgentService.build_result(db, question="今天北京天气怎么样？")) + assert result.allow_general_knowledge is True + assert result.chunks == [] + assert result.tool_trace[1]["needKnowledge"] is False + + +def test_course_question_without_catalog_does_not_fall_back_to_general_knowledge(): + with _database() as db: + result = asyncio.run(KnowledgeAgentService.build_result(db, question="卢慧老师课程里怎么解释亲子沟通？")) + assert result.allow_general_knowledge is False + assert result.chunks == [] + + +def test_course_question_searches_chunk_and_reads_parent_section(): + with _database() as db: + _add_published_knowledge(db, knowledge_id=1, name="亲子课程") + result = asyncio.run(KnowledgeAgentService.build_result(db, question="课程里家长怎么和学习动力不足的孩子沟通？")) + assert result.allow_general_knowledge is False + assert len(result.chunks) == 1 + assert "先稳定自己的焦虑" in result.chunks[0].content + assert any(item["tool"] == "read_knowledge_sections" for item in result.tool_trace)