diff --git a/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge.py b/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge.py index e9b1e7d..6871a1a 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge.py +++ b/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge.py @@ -25,7 +25,6 @@ from app.models.knowledge import ( from app.schemas.admin import KnowledgeSaveRequest from app.services.admin_service import OperationLogService from app.services.feishu_service import FeishuKnowledgeService -from app.services.knowledge_catalog_cache_service import KnowledgeCatalogCacheService from app.api.pagination import page_result router = APIRouter() @@ -134,7 +133,6 @@ def import_knowledge_from_space( db.rollback() raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="该飞书文章已创建知识库") from exc db.refresh(knowledge) - KnowledgeCatalogCacheService.invalidate() return api_success({"imported": 1, "skipped": 0, "id": knowledge.id, "name": knowledge.name}) @@ -166,7 +164,6 @@ def create_knowledge( db.rollback() raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="该飞书文章已创建知识库") from exc db.refresh(knowledge) - KnowledgeCatalogCacheService.invalidate() return api_success(_knowledge_dict(knowledge)) @@ -187,7 +184,6 @@ def update_knowledge( OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="update", target_id=knowledge.id) db.commit() db.refresh(knowledge) - KnowledgeCatalogCacheService.invalidate() return api_success(_knowledge_dict(knowledge)) @@ -222,7 +218,6 @@ def delete_knowledge( db.delete(knowledge) OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="delete", target_id=knowledge.id) db.commit() - KnowledgeCatalogCacheService.invalidate() return api_success() 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 8226393..ce9d875 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 @@ -30,7 +30,6 @@ from app.schemas.knowledge import ( ) from app.services.admin_service import OperationLogService from app.services.knowledge_pipeline_service import KnowledgePipelineService -from app.services.knowledge_catalog_cache_service import KnowledgeCatalogCacheService router = APIRouter() @@ -69,7 +68,6 @@ def batch_update_metadata( ) results.append({"knowledgeId": knowledge_id, "ok": True}) db.commit() - KnowledgeCatalogCacheService.invalidate() success = sum(1 for item in results if item["ok"]) return api_success( {"total": len(results), "success": success, "failed": len(results) - success, "results": results} @@ -94,7 +92,6 @@ def update_metadata( ) db.commit() db.refresh(knowledge) - KnowledgeCatalogCacheService.invalidate() return api_success(_knowledge_detail(knowledge)) @@ -225,7 +222,6 @@ async def synchronize( raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"知识同步失败：{exc}") from exc OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="sync", target_id=knowledge.id) db.commit() - KnowledgeCatalogCacheService.invalidate() return api_success(_job_dict(job)) @@ -287,7 +283,6 @@ def publish( raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="publish", target_id=knowledge.id) db.commit() - KnowledgeCatalogCacheService.invalidate() return api_success(_knowledge_detail(knowledge)) @@ -311,7 +306,6 @@ def rollback( raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="rollback", target_id=knowledge.id) db.commit() - KnowledgeCatalogCacheService.invalidate() return api_success(_knowledge_detail(knowledge)) @@ -337,7 +331,6 @@ def open_status( db.add(knowledge) OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="open" if payload.open else "close", target_id=knowledge.id) db.commit() - KnowledgeCatalogCacheService.invalidate() return api_success(_knowledge_detail(knowledge)) @@ -358,7 +351,6 @@ def lifecycle( db.add(knowledge) OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action=payload.status, target_id=knowledge.id) db.commit() - KnowledgeCatalogCacheService.invalidate() return api_success(_knowledge_detail(knowledge)) diff --git a/ai_knowledge_base_v2/apps/backend/app/core/database.py b/ai_knowledge_base_v2/apps/backend/app/core/database.py index 01c942c..d3e1ab6 100644 --- a/ai_knowledge_base_v2/apps/backend/app/core/database.py +++ b/ai_knowledge_base_v2/apps/backend/app/core/database.py @@ -7,6 +7,7 @@ from sqlalchemy.orm import Session, sessionmaker from app.core.config import get_settings from app.models import Base +from app.services.knowledge_catalog_cache_service import register_knowledge_cache_invalidation_events settings = get_settings() @@ -20,6 +21,7 @@ if not settings.database_url.startswith("sqlite"): ) engine = create_engine(settings.database_url, **engine_options) SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True) +register_knowledge_cache_invalidation_events() def get_db() -> Generator[Session, None, None]: diff --git a/ai_knowledge_base_v2/apps/backend/app/services/knowledge_catalog_cache_service.py b/ai_knowledge_base_v2/apps/backend/app/services/knowledge_catalog_cache_service.py index 620e2f9..bd293da 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/knowledge_catalog_cache_service.py +++ b/ai_knowledge_base_v2/apps/backend/app/services/knowledge_catalog_cache_service.py @@ -4,15 +4,47 @@ import json import time from typing import Callable +from sqlalchemy import event +from sqlalchemy.orm import Session + +from app.models.knowledge import ( + Knowledge, + KnowledgeCard, + KnowledgeChunk, + KnowledgeManifest, + KnowledgePublishLog, + KnowledgeReviewRecord, + KnowledgeSection, + KnowledgeSourceSnapshot, + KnowledgeSyncJob, + KnowledgeVersion, + UserKnowledgePermission, +) from app.services.redis_client import get_sync_redis_client -class KnowledgeCatalogCacheService: - """Central invalidation point for current and future KnowledgeList caches.""" +_CACHE_INVALIDATION_PENDING = "agent_knowledge_cache_invalidation_pending" +_CACHE_RELEVANT_MODELS = ( + Knowledge, + KnowledgeCard, + KnowledgeChunk, + KnowledgeManifest, + KnowledgePublishLog, + KnowledgeReviewRecord, + KnowledgeSection, + KnowledgeSourceSnapshot, + KnowledgeSyncJob, + KnowledgeVersion, + UserKnowledgePermission, +) - VERSION_KEY = "kb:catalog:generation" - KEY_PREFIX = "kb:catalog:v1" - METRICS_KEY = "kb:catalog:metrics" + +class KnowledgeCatalogCacheService: + """Shared generation for KnowledgeList and all Agent knowledge caches.""" + + VERSION_KEY = "agent:knowledge:generation" + KEY_PREFIX = "agent:knowledge:catalog:v2" + METRICS_KEY = "agent:knowledge:cache:metrics" TTL_SECONDS = 300 @classmethod @@ -44,12 +76,37 @@ class KnowledgeCatalogCacheService: @classmethod def invalidate(cls) -> None: + """Logically expires every Agent knowledge cache in one atomic operation.""" client = get_sync_redis_client() if client is None: return try: client.incr(cls.VERSION_KEY) except Exception: - # The catalog currently reads MySQL on every request, so Redis failure - # must not make a committed metadata change look unsuccessful. + # Redis failure must not make a committed database change look unsuccessful; + # subsequent reads automatically fall back to MySQL when Redis is unavailable. return + + +def register_knowledge_cache_invalidation_events() -> None: + if event.contains(Session, "before_flush", _mark_knowledge_cache_dirty): + return + event.listen(Session, "before_flush", _mark_knowledge_cache_dirty) + event.listen(Session, "after_commit", _invalidate_knowledge_cache_after_commit) + event.listen(Session, "after_rollback", _clear_knowledge_cache_invalidation) + + +def _mark_knowledge_cache_dirty(session: Session, _flush_context: object, _instances: object) -> None: + changed = list(session.new) + list(session.deleted) + changed.extend(item for item in session.dirty if session.is_modified(item, include_collections=True)) + if any(isinstance(item, _CACHE_RELEVANT_MODELS) for item in changed): + session.info[_CACHE_INVALIDATION_PENDING] = True + + +def _invalidate_knowledge_cache_after_commit(session: Session) -> None: + if session.info.pop(_CACHE_INVALIDATION_PENDING, False): + KnowledgeCatalogCacheService.invalidate() + + +def _clear_knowledge_cache_invalidation(session: Session) -> None: + session.info.pop(_CACHE_INVALIDATION_PENDING, None) diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_admin.py b/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_admin.py index 928103c..5e10432 100644 --- a/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_admin.py +++ b/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_admin.py @@ -76,8 +76,7 @@ def _knowledge(db: Session, knowledge_id: int = 1) -> Knowledge: return item -def test_metadata_update_preserves_source_version_and_open_state(monkeypatch: pytest.MonkeyPatch): - monkeypatch.setattr("app.api.admin_knowledge_lifecycle.KnowledgeCatalogCacheService.invalidate", lambda: None) +def test_metadata_update_preserves_source_version_and_open_state(): with _database() as db: item = _knowledge(db) original = (item.feishu_space_id, item.feishu_node_id, item.current_version_id, item.status) @@ -106,8 +105,7 @@ def test_metadata_update_requires_super_admin(): assert exc.value.status_code == 403 -def test_batch_metadata_update_reports_missing_items(monkeypatch: pytest.MonkeyPatch): - monkeypatch.setattr("app.api.admin_knowledge_lifecycle.KnowledgeCatalogCacheService.invalidate", lambda: None) +def test_batch_metadata_update_reports_missing_items(): with _database() as db: item = _knowledge(db) response = batch_update_metadata( diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_cache_invalidation.py b/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_cache_invalidation.py new file mode 100644 index 0000000..a18baf8 --- /dev/null +++ b/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_cache_invalidation.py @@ -0,0 +1,110 @@ +from sqlalchemy import create_engine +from sqlalchemy.orm import Session +from sqlalchemy.pool import StaticPool + +from app.models import Base +from app.models.logs import OperationLog +from app.models.knowledge import Knowledge, KnowledgeManifest, KnowledgeSourceSnapshot, KnowledgeVersion +from app.services.knowledge_catalog_cache_service import KnowledgeCatalogCacheService + + +class _RedisRecorder: + def __init__(self) -> None: + self.increments: list[str] = [] + + def incr(self, key: str) -> None: + self.increments.append(key) + + +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 _knowledge() -> Knowledge: + return Knowledge( + id=1, + name="测试知识库", + feishu_space_id="space-1", + feishu_node_id="node-1", + status=0, + knowledge_type="general", + review_mode="manual", + ) + + +def test_all_agent_knowledge_cache_is_invalidated_after_each_committed_change(monkeypatch): + redis = _RedisRecorder() + monkeypatch.setattr( + "app.services.knowledge_catalog_cache_service.get_sync_redis_client", lambda: redis + ) + with _database() as db: + knowledge = _knowledge() + db.add(knowledge) + db.commit() + + knowledge.status = 1 + db.commit() + + snapshot = KnowledgeSourceSnapshot( + id=1, + knowledge_id=knowledge.id, + source_title="标题", + source_content="内容", + content_hash="snapshot-hash", + source_identifier="space-1/node-1", + ) + db.add(snapshot) + db.flush() + version = KnowledgeVersion( + id=1, + 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() + db.add( + KnowledgeManifest( + id=1, + knowledge_id=knowledge.id, + version_id=version.id, + purpose="用途", + applicable_questions="适用", + inapplicable_questions="不适用", + core_topics="主题", + content_hash="manifest-hash", + ) + ) + db.commit() + + assert redis.increments == [KnowledgeCatalogCacheService.VERSION_KEY] * 3 + + +def test_unrelated_log_changes_do_not_invalidate_agent_knowledge_cache(monkeypatch): + redis = _RedisRecorder() + monkeypatch.setattr( + "app.services.knowledge_catalog_cache_service.get_sync_redis_client", lambda: redis + ) + with _database() as db: + db.add(OperationLog(module="auth", action="login", result="success")) + db.commit() + + assert redis.increments == [] + + +def test_rolled_back_knowledge_change_does_not_invalidate_cache(monkeypatch): + redis = _RedisRecorder() + monkeypatch.setattr( + "app.services.knowledge_catalog_cache_service.get_sync_redis_client", lambda: redis + ) + with _database() as db: + db.add(_knowledge()) + db.flush() + db.rollback() + + assert redis.increments == []