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

fix: invalidate agent cache on knowledge changes

Browse Source
This commit is contained in:
Nelson
2026-07-13 13:18:52 +08:00
parent 00d53c2569
commit 750a57a390
6 changed files with 178 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge.py
View File

@@ -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()

8
ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge_lifecycle.py
View File

@@ -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))

2
ai_knowledge_base_v2/apps/backend/app/core/database.py
View File

@@ -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]:

71
ai_knowledge_base_v2/apps/backend/app/services/knowledge_catalog_cache_service.py
View File

@@ -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)

6
ai_knowledge_base_v2/apps/backend/tests/test_knowledge_admin.py
View File

@@ -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(

110
ai_knowledge_base_v2/apps/backend/tests/test_knowledge_cache_invalidation.py Normal file
View File

@@ -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 == []