GraphRAG Cache 缓存模块完全指南内置缓存实现、工厂注册机制与自定义扩展实战【免费下载链接】graphragA modular graph-based Retrieval-Augmented Generation (RAG) system项目地址: https://gitcode.com/GitHub_Trending/gr/graphragGraphRAG Cachegraphrag-cache是模块化 GraphRAG 系统中的缓存基础设施包通过统一抽象、工厂模式与类型化配置为索引流水线与查询链路提供可插拔的 LLM 调用结果缓存能力。读完本文你将掌握Cache抽象接口与三个内置缓存实现JsonCache、MemoryCache、NoopCache的原理理解create_cache的懒加载注册机制并能通过继承Cache基类与CacheConfig扩展字段注册出自己的缓存实现。包定位与模块结构graphrag-cache位于仓库的 packages/graphrag-cache 目录是 GraphRAG 从单体应用拆分为多包架构workspace后的核心基础组件之一。根据其 pyproject.toml该包版本与主项目保持一致3.1.1运行时仅依赖另外两个同架构包graphrag-common3.1.1提供Factory基类与hash_data等通用工具graphrag-storage3.1.1提供Storage存储抽象供JsonCache等落盘型缓存使用。工程内的包目录布局如下graphrag_cache/ ├── __init__.py # 包统一出口Cache/CacheConfig/CacheType/create_cache 等 ├── cache.py # Cache 抽象基类abc.ABC ├── cache_config.py # CacheConfig类型化配置模型 ├── cache_type.py # CacheType 枚举json/memory/none ├── cache_factory.py # CacheFactory、cache_factory 单例、create_cache、register_cache ├── cache_key.py # CacheKeyCreator 协议与 create_cache_key 工具 ├── json_cache.py # JsonCache基于 Storage 的落盘缓存 ├── memory_cache.py # MemoryCache进程内字典缓存 └── noop_cache.py # NoopCache空操作缓存包的设计哲学与其 README 描述一致对外只暴露最少的语义接口Cache其余全部通过“配置CacheConfig→ 工厂cache_factory→ 实现”的链路按需装配。这样上层调用方如graphrag-llm的中间件层永远不必关心缓存究竟落在文件、内存还是被禁用。Cache 抽象接口所有缓存的统一契约所有内置及自定义缓存都必须实现 cache.py 中定义的抽象基类Cache。它是一个全异步接口初始化签名固定为__init__(self, *, storage: Storage | None, **kwargs)方法契约如下方法签名语义getasync get(key: str) - Any取回给定 key 的值不存在时返回Nonesetasync set(key: str, value: Any, debug_data: dict | None None) - None将 value 写入缓存debug_data用于附带调试信息JsonCache会将其一并落盘hasasync has(key: str) - bool判断 key 是否已缓存deleteasync delete(key: str) - None删除指定 keyclearasync clear() - None清空整个缓存childdef child(name: str) - Cache返回指定名字的子缓存用于缓存命名空间隔离值得注意该基类在TYPE_CHECKING块中才导入graphrag_storage.Storage类型说明Cache并不强制要求底层具备存储——是否使用Storage完全由具体实现决定例如MemoryCache与NoopCache并不依赖任何存储。这为“内存缓存”“禁用缓存”等轻量场景留出了空间。三种内置缓存实现的行为差异JsonCache落盘缓存默认JsonCache 是create_cache()的默认实现。它的构造要求传入Storage实例或一个等价于StorageConfig的字典若二者皆缺则直接抛出ValueError。其行为细节决定了它在生产中的可靠性序列化格式set时把value与debug_data合并成{result: value, ...}后执行json.dumps(..., ensure_asciiFalse)再交给Storage.set写盘读时自愈get时若底层存储抛UnicodeDecodeError或json.JSONDecodeError例如文件被截断或损坏会自动delete该坏键并返回None避免一次脏数据阻塞整条流水线跳过空值set(None)直接被忽略不会产生空缓存记录命名空间隔离child(name)通过self._storage.child(name)得到子存储天然实现按域隔离存储类型解耦究竟写本地文件、内存还是 Azure Blob完全由外部注入的Storage决定graphrag_storage 内置的StorageType枚举包括File、Memory、AzureBlob、AzureCosmos见 storage_type.py。MemoryCache进程内字典缓存MemoryCache 用一个普通 Python 字典self._cache充当存储所有get/set/has/delete/clear都是O(1)的字典操作。它不需要任何Storage与序列化速度快但生命周期与进程一致、不跨重启持久。它的child(name)简单地返回一个新的空MemoryCache不维护父子层级。适合单次进程内重复调用如同一索引任务内多次相同的 LLM 调用去重。NoopCache空操作缓存NoopCache 的注释直言其用途“usually useful for testing”。它的get永远返回None、has永远返回Falseset/delete/clear均为空操作child返回自身。将缓存类型切换为Noop等价于“关闭缓存”且对上层代码零改动——这对基准测试、故障排查和关闭缓存的调试场景极其方便。下表对照三种内置实现的关键差异实现CacheType值依赖 Storage持久化适用场景JsonCacheCacheType.Json(json)是必须注入是默认生产选项跨进程/重启复用结果MemoryCacheCacheType.Memory(memory)否否单进程内的重复调用去重NoopCacheCacheType.Noop(none)否否测试、关闭缓存、基准对照CacheConfig 与 CacheType类型化配置入口CacheConfig 是基于 Pydantic 的类型化配置模型是create_cache的唯一推荐入口字段如下type: str缓存实现类型默认CacheType.Json。内置类型取值见 cache_type.py 中的CacheType枚举json/memory/noneNoop。由于类型是字符串而非枚举它也天然支持传入自定义注册名见下文“自定义缓存”。storage: StorageConfig | None仅供JsonCache这类落盘缓存使用的存储配置默认StorageConfig(typeStorageType.File, base_dircache)即默认在相对路径cache/目录下读写文件。model_config ConfigDict(extraallow)允许额外字段。这是自定义缓存实现的关键支撑——自定义实现构造时所需的专有参数如some_setting可直接作为额外字段挂在CacheConfig上随工厂注入。工厂侧则用config.model_dump()把配置转成普通字典后透传给实现构造函数。若用户未传storage但配置里带有storage配置段工厂会调用graphrag_storage的create_storage(config.storage)自动创建存储实例见 cache_factory.py。create_cache懒加载注册 工厂装配在 cache_factory.py 中核心逻辑分三层单例工厂cache_factory CacheFactory()是全局唯一实例。CacheFactory继承自 graphrag-common 的Factory基类后者用__new__保证单例用_service_initializers字典按strategy名称登记构造器register()支持scopesingleton | transient默认transient其中singleton作用域会按 “strategy init_args” 哈希缓存实例保证相同入参复用同一实例。懒加载注册create_cache在拿到config.type后先判断cache_strategy not in cache_factory若未注册再通过match语句现场import对应模块并register_cache(...)。例如只有请求CacheType.Json时才导入并注册JsonCache——README 称之为“动态预注册”preregistration happens dynamically。因此用create_cache时完全无需手动注册内置实现对未识别类型则抛出带已注册类型清单的ValueError。参数装配把storage实例若存在回填进配置字典后调用cache_factory.create(strategycache_strategy, init_argsconfig_model)产出最终缓存实例。register_cache(cache_type, cache_initializer, scopetransient)是对cache_factory.register的薄封装供上层注册自定义实现。基础用法默认 JSON 文件缓存最简单的用法是不传任何参数直接create_cache()——它会落到默认配置JsonCache 本地cache/文件存储。这一流程可完整对照 basic_cache_example.ipynb 中给出的完整可运行示例from graphrag_cache import CacheConfig, CacheType, create_cache, create_cache_key from graphrag_storage import StorageConfig, StorageType async def run(): Demonstrate basic cache usage with graphrag_cache. cache create_cache() # The above is equivalent to the following: cache create_cache( CacheConfig( typeCacheType.Json, storageStorageConfig(typeStorageType.File, base_dircache), ), ) await cache.set(my_key, {k1: object to cache}) print(Value stored in cache for my_key:) print(await cache.get(my_key)) # create cache key from data dict. cache_key create_cache_key({some_arg: some_value, something_else: 5}) await cache.set(cache_key, {k2: object to cache}) print(\nValue stored in cache for cache_key using data dict:) print(await cache.get(cache_key)) if __name__ __main__: await run()注意if __name__ __main__: await run()的写法说明该包接口为async/await风格调用方需运行在事件循环如asyncio中。示例还展示了 cache_key.py 提供的create_cache_key(input_args: dict) - str它把参数字典交给graphrag_common的hash_data默认基于 sha256 的确定性哈希见 hasher.py生成稳定缓存键非常适合“以 LLM 请求入参作为缓存 key”的场景。该模块同时定义了CacheKeyCreator这一runtime_checkable协议允许上层用任意符合(input_args: dict[str, Any]) - str签名的函数作为自定义键生成器。直接使用 cache_factory跳过预注册与类型化配置create_cache的好处是类型化配置 内置实现懒注册但它也隐含了“全局单例工厂里总会有内置实现被注册”这一前提。若你希望得到一个完全干净、没有任何预注册项的工厂可绕过create_cache直接使用cache_factoryREADME 明确提示其代价是create用普通字典传参而非强类型CacheConfig。原文给出的最小用法如下from graphrag_cache.cache_factory import cache_factory from graphrag_cache.json_cache import JsonCache # cache_factory has no preregistered providers so you must register any # providers you plan on using. # May also register a custom implementation, see above for example. cache_factory.register(my_cache_impl, JsonCache) cache cache_factory.create(strategymy_cache_impl, init_args{some_setting: ...}) ...注意这里必须先手动注册再用由于cache_factory没有预注册任何 provider直接create未注册的 strategy 会在基类 Factory.create 处抛出“Strategy is not registered”的ValueError。另外手动注册时还可以传入scope参数控制生命周期默认transient每次创建新实例若改为singleton则相同 init 参数会复用同一实例——若你的缓存实现是有状态的且希望全局共享可显式利用这一点。自定义缓存实现继承 Cache register_cache当内置三种缓存无法满足需求例如需要接入 Redis、自定义 TTL 策略、或与自有存储联动时仓库推荐的扩展路线是继承Cache基类并实现全部抽象方法通过register_cache(name, MyCache)把实现注册进单例cache_factory之后即可用create_cache(CacheConfig(typeMyCache, ...额外参数...))或直接cache_factory.create(...)实例化。README 明确点出这一模式“registering it with the GraphRAG cache system…allowing for extensible caching solutions tailored to specific needs”。custom_cache_example.ipynb 给出了完整的自定义实现与注册示例from typing import Any from graphrag_cache import Cache, CacheConfig, create_cache, register_cache class MyCache(Cache): Custom cache implementation for storing and retrieving cached data. def __init__( self, some_setting: str, optional_setting: str default setting, **kwargs: Any, ): # Validate settings and initialize # View the JsonCache implementation to see how to create a cache that relies on a Storage provider. self.some_setting some_setting self.optional_setting optional_setting self._cache: dict[str, Any] {} self._child: Cache self async def get(self, key: str) - Any: Retrieve a value from the cache by key. return self._cache.get(key) async def set(self, key: str, value: Any, debug_data: Any None) - None: Store a value in the cache with the specified key. self._cache[key] value async def has(self, key: str) - bool: Check if a key exists in the cache. return key in self._cache async def delete(self, key: str) - None: Remove a key and its value from the cache. self._cache.pop(key, None) async def clear(self) - None: Clear all items from the cache. self._cache.clear() def child(self, name: str) - Cache: Create or access a child cache (not implemented in this example). return self._child register_cache(MyCache, MyCache) async def run(): Demonstrate usage of the custom cache implementation. cache create_cache( CacheConfig( typeMyCache, some_settingimportant setting value, # type: ignore ) ) await cache.set(my_key, {k1: object to cache}) print(Value stored in cache for my_key:) print(await cache.get(my_key)) if __name__ __main__: await run()该示例揭示了自定义缓存的三个关键机制构造参数来自CacheConfig额外字段MyCache.__init__的some_setting、optional_setting之所以能被自动注入正是因为 CacheConfig 设置了extraallow任意扩展字段都会原样进入model_dump()后的 init 字典。源码注释中用# type: ignore标注some_setting是“超出模型定义”的自定义字段。可选参数 默认值构造器里的optional_setting: str default setting能正常工作得益于Factory.create在派发前会“删除值为None的条目”从而让未提供的可选参数走实现自身的默认值见 factory.py。注册后即可通过CacheConfig.type字符串寻址register_cache(MyCache, MyCache)之后CacheConfig(typeMyCache)就能在create_cache中被解析工厂的match语句对未知名分支统一抛出“not registered”错误因此自定义类型必须在调用前完成注册。若自定义实现需要落盘示例注释给出了指引参考 JsonCache 的做法在构造器中接收Storage实例或StorageConfig字典通过graphrag_storage的create_storage建立存储连接——这样即可复用仓库整套存储后端本地文件 / Azure Blob / CosmosDB。包统一出口与工程集成视角从init.py 看包对外只导出七个符号Cache、CacheConfig、CacheType、CacheKeyCreator、create_cache、create_cache_key、register_cache。实践中推荐按此分层使用大多数场景配置驱动、快速上手从graphrag_cache顶层导入CacheConfig/CacheType/create_cache/create_cache_key深度定制绕过预注册从graphrag_cache.cache_factory导入单例cache_factory直接调用register/create。在整个 GraphRAG 工程中graphrag-cache位于依赖链的中下层它依赖graphrag-storage与graphrag-common本身不依赖 LLM 层而graphrag-llm等上层包通过本包完成结果缓存。若要独立安装使用可按其 pyproject.toml 的约束Python3.11,3.14在本地 workspace 中以pip install -e ./packages/graphrag-cache方式引入并保证同版本号的graphrag-common、graphrag-storage一并安装否则版本钉死3.1.1会阻碍解析。整体而言GraphRAG Cache 用“一个小抽象 一个单例工厂 一份可扩展配置”就解决了缓存的可插拔问题默认JsonCache开箱即用、MemoryCache极速去重、NoopCache一键关闭而自定义实现通过CacheConfig(extraallow)与register_cache可无缝接入工厂体系让缓存从“功能”真正演化为“可通过配置选型的基础设施”。【免费下载链接】graphragA modular graph-based Retrieval-Augmented Generation (RAG) system项目地址: https://gitcode.com/GitHub_Trending/gr/graphrag创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 GraphRAG Cache 缓存模块完全指南：内置缓存实现、工厂注册机制与自定义扩展实战
GraphRAG Cache 缓存模块完全指南内置缓存实现、工厂注册机制与自定义扩展实战【免费下载链接】graphragA modular graph-based Retrieval-Augmented Generation (RAG) system项目地址: https://gitcode.com/GitHub_Trending/gr/graphragGraphRAG Cachegraphrag-cache是模块化 GraphRAG 系统中的缓存基础设施包通过统一抽象、工厂模式与类型化配置为索引流水线与查询链路提供可插拔的 LLM 调用结果缓存能力。读完本文你将掌握Cache抽象接口与三个内置缓存实现JsonCache、MemoryCache、NoopCache的原理理解create_cache的懒加载注册机制并能通过继承Cache基类与CacheConfig扩展字段注册出自己的缓存实现。包定位与模块结构graphrag-cache位于仓库的 packages/graphrag-cache 目录是 GraphRAG 从单体应用拆分为多包架构workspace后的核心基础组件之一。根据其 pyproject.toml该包版本与主项目保持一致3.1.1运行时仅依赖另外两个同架构包graphrag-common3.1.1提供Factory基类与hash_data等通用工具graphrag-storage3.1.1提供Storage存储抽象供JsonCache等落盘型缓存使用。工程内的包目录布局如下graphrag_cache/ ├── __init__.py # 包统一出口Cache/CacheConfig/CacheType/create_cache 等 ├── cache.py # Cache 抽象基类abc.ABC ├── cache_config.py # CacheConfig类型化配置模型 ├── cache_type.py # CacheType 枚举json/memory/none ├── cache_factory.py # CacheFactory、cache_factory 单例、create_cache、register_cache ├── cache_key.py # CacheKeyCreator 协议与 create_cache_key 工具 ├── json_cache.py # JsonCache基于 Storage 的落盘缓存 ├── memory_cache.py # MemoryCache进程内字典缓存 └── noop_cache.py # NoopCache空操作缓存包的设计哲学与其 README 描述一致对外只暴露最少的语义接口Cache其余全部通过“配置CacheConfig→ 工厂cache_factory→ 实现”的链路按需装配。这样上层调用方如graphrag-llm的中间件层永远不必关心缓存究竟落在文件、内存还是被禁用。Cache 抽象接口所有缓存的统一契约所有内置及自定义缓存都必须实现 cache.py 中定义的抽象基类Cache。它是一个全异步接口初始化签名固定为__init__(self, *, storage: Storage | None, **kwargs)方法契约如下方法签名语义getasync get(key: str) - Any取回给定 key 的值不存在时返回Nonesetasync set(key: str, value: Any, debug_data: dict | None None) - None将 value 写入缓存debug_data用于附带调试信息JsonCache会将其一并落盘hasasync has(key: str) - bool判断 key 是否已缓存deleteasync delete(key: str) - None删除指定 keyclearasync clear() - None清空整个缓存childdef child(name: str) - Cache返回指定名字的子缓存用于缓存命名空间隔离值得注意该基类在TYPE_CHECKING块中才导入graphrag_storage.Storage类型说明Cache并不强制要求底层具备存储——是否使用Storage完全由具体实现决定例如MemoryCache与NoopCache并不依赖任何存储。这为“内存缓存”“禁用缓存”等轻量场景留出了空间。三种内置缓存实现的行为差异JsonCache落盘缓存默认JsonCache 是create_cache()的默认实现。它的构造要求传入Storage实例或一个等价于StorageConfig的字典若二者皆缺则直接抛出ValueError。其行为细节决定了它在生产中的可靠性序列化格式set时把value与debug_data合并成{result: value, ...}后执行json.dumps(..., ensure_asciiFalse)再交给Storage.set写盘读时自愈get时若底层存储抛UnicodeDecodeError或json.JSONDecodeError例如文件被截断或损坏会自动delete该坏键并返回None避免一次脏数据阻塞整条流水线跳过空值set(None)直接被忽略不会产生空缓存记录命名空间隔离child(name)通过self._storage.child(name)得到子存储天然实现按域隔离存储类型解耦究竟写本地文件、内存还是 Azure Blob完全由外部注入的Storage决定graphrag_storage 内置的StorageType枚举包括File、Memory、AzureBlob、AzureCosmos见 storage_type.py。MemoryCache进程内字典缓存MemoryCache 用一个普通 Python 字典self._cache充当存储所有get/set/has/delete/clear都是O(1)的字典操作。它不需要任何Storage与序列化速度快但生命周期与进程一致、不跨重启持久。它的child(name)简单地返回一个新的空MemoryCache不维护父子层级。适合单次进程内重复调用如同一索引任务内多次相同的 LLM 调用去重。NoopCache空操作缓存NoopCache 的注释直言其用途“usually useful for testing”。它的get永远返回None、has永远返回Falseset/delete/clear均为空操作child返回自身。将缓存类型切换为Noop等价于“关闭缓存”且对上层代码零改动——这对基准测试、故障排查和关闭缓存的调试场景极其方便。下表对照三种内置实现的关键差异实现CacheType值依赖 Storage持久化适用场景JsonCacheCacheType.Json(json)是必须注入是默认生产选项跨进程/重启复用结果MemoryCacheCacheType.Memory(memory)否否单进程内的重复调用去重NoopCacheCacheType.Noop(none)否否测试、关闭缓存、基准对照CacheConfig 与 CacheType类型化配置入口CacheConfig 是基于 Pydantic 的类型化配置模型是create_cache的唯一推荐入口字段如下type: str缓存实现类型默认CacheType.Json。内置类型取值见 cache_type.py 中的CacheType枚举json/memory/noneNoop。由于类型是字符串而非枚举它也天然支持传入自定义注册名见下文“自定义缓存”。storage: StorageConfig | None仅供JsonCache这类落盘缓存使用的存储配置默认StorageConfig(typeStorageType.File, base_dircache)即默认在相对路径cache/目录下读写文件。model_config ConfigDict(extraallow)允许额外字段。这是自定义缓存实现的关键支撑——自定义实现构造时所需的专有参数如some_setting可直接作为额外字段挂在CacheConfig上随工厂注入。工厂侧则用config.model_dump()把配置转成普通字典后透传给实现构造函数。若用户未传storage但配置里带有storage配置段工厂会调用graphrag_storage的create_storage(config.storage)自动创建存储实例见 cache_factory.py。create_cache懒加载注册 工厂装配在 cache_factory.py 中核心逻辑分三层单例工厂cache_factory CacheFactory()是全局唯一实例。CacheFactory继承自 graphrag-common 的Factory基类后者用__new__保证单例用_service_initializers字典按strategy名称登记构造器register()支持scopesingleton | transient默认transient其中singleton作用域会按 “strategy init_args” 哈希缓存实例保证相同入参复用同一实例。懒加载注册create_cache在拿到config.type后先判断cache_strategy not in cache_factory若未注册再通过match语句现场import对应模块并register_cache(...)。例如只有请求CacheType.Json时才导入并注册JsonCache——README 称之为“动态预注册”preregistration happens dynamically。因此用create_cache时完全无需手动注册内置实现对未识别类型则抛出带已注册类型清单的ValueError。参数装配把storage实例若存在回填进配置字典后调用cache_factory.create(strategycache_strategy, init_argsconfig_model)产出最终缓存实例。register_cache(cache_type, cache_initializer, scopetransient)是对cache_factory.register的薄封装供上层注册自定义实现。基础用法默认 JSON 文件缓存最简单的用法是不传任何参数直接create_cache()——它会落到默认配置JsonCache 本地cache/文件存储。这一流程可完整对照 basic_cache_example.ipynb 中给出的完整可运行示例from graphrag_cache import CacheConfig, CacheType, create_cache, create_cache_key from graphrag_storage import StorageConfig, StorageType async def run(): Demonstrate basic cache usage with graphrag_cache. cache create_cache() # The above is equivalent to the following: cache create_cache( CacheConfig( typeCacheType.Json, storageStorageConfig(typeStorageType.File, base_dircache), ), ) await cache.set(my_key, {k1: object to cache}) print(Value stored in cache for my_key:) print(await cache.get(my_key)) # create cache key from data dict. cache_key create_cache_key({some_arg: some_value, something_else: 5}) await cache.set(cache_key, {k2: object to cache}) print(\nValue stored in cache for cache_key using data dict:) print(await cache.get(cache_key)) if __name__ __main__: await run()注意if __name__ __main__: await run()的写法说明该包接口为async/await风格调用方需运行在事件循环如asyncio中。示例还展示了 cache_key.py 提供的create_cache_key(input_args: dict) - str它把参数字典交给graphrag_common的hash_data默认基于 sha256 的确定性哈希见 hasher.py生成稳定缓存键非常适合“以 LLM 请求入参作为缓存 key”的场景。该模块同时定义了CacheKeyCreator这一runtime_checkable协议允许上层用任意符合(input_args: dict[str, Any]) - str签名的函数作为自定义键生成器。直接使用 cache_factory跳过预注册与类型化配置create_cache的好处是类型化配置 内置实现懒注册但它也隐含了“全局单例工厂里总会有内置实现被注册”这一前提。若你希望得到一个完全干净、没有任何预注册项的工厂可绕过create_cache直接使用cache_factoryREADME 明确提示其代价是create用普通字典传参而非强类型CacheConfig。原文给出的最小用法如下from graphrag_cache.cache_factory import cache_factory from graphrag_cache.json_cache import JsonCache # cache_factory has no preregistered providers so you must register any # providers you plan on using. # May also register a custom implementation, see above for example. cache_factory.register(my_cache_impl, JsonCache) cache cache_factory.create(strategymy_cache_impl, init_args{some_setting: ...}) ...注意这里必须先手动注册再用由于cache_factory没有预注册任何 provider直接create未注册的 strategy 会在基类 Factory.create 处抛出“Strategy is not registered”的ValueError。另外手动注册时还可以传入scope参数控制生命周期默认transient每次创建新实例若改为singleton则相同 init 参数会复用同一实例——若你的缓存实现是有状态的且希望全局共享可显式利用这一点。自定义缓存实现继承 Cache register_cache当内置三种缓存无法满足需求例如需要接入 Redis、自定义 TTL 策略、或与自有存储联动时仓库推荐的扩展路线是继承Cache基类并实现全部抽象方法通过register_cache(name, MyCache)把实现注册进单例cache_factory之后即可用create_cache(CacheConfig(typeMyCache, ...额外参数...))或直接cache_factory.create(...)实例化。README 明确点出这一模式“registering it with the GraphRAG cache system…allowing for extensible caching solutions tailored to specific needs”。custom_cache_example.ipynb 给出了完整的自定义实现与注册示例from typing import Any from graphrag_cache import Cache, CacheConfig, create_cache, register_cache class MyCache(Cache): Custom cache implementation for storing and retrieving cached data. def __init__( self, some_setting: str, optional_setting: str default setting, **kwargs: Any, ): # Validate settings and initialize # View the JsonCache implementation to see how to create a cache that relies on a Storage provider. self.some_setting some_setting self.optional_setting optional_setting self._cache: dict[str, Any] {} self._child: Cache self async def get(self, key: str) - Any: Retrieve a value from the cache by key. return self._cache.get(key) async def set(self, key: str, value: Any, debug_data: Any None) - None: Store a value in the cache with the specified key. self._cache[key] value async def has(self, key: str) - bool: Check if a key exists in the cache. return key in self._cache async def delete(self, key: str) - None: Remove a key and its value from the cache. self._cache.pop(key, None) async def clear(self) - None: Clear all items from the cache. self._cache.clear() def child(self, name: str) - Cache: Create or access a child cache (not implemented in this example). return self._child register_cache(MyCache, MyCache) async def run(): Demonstrate usage of the custom cache implementation. cache create_cache( CacheConfig( typeMyCache, some_settingimportant setting value, # type: ignore ) ) await cache.set(my_key, {k1: object to cache}) print(Value stored in cache for my_key:) print(await cache.get(my_key)) if __name__ __main__: await run()该示例揭示了自定义缓存的三个关键机制构造参数来自CacheConfig额外字段MyCache.__init__的some_setting、optional_setting之所以能被自动注入正是因为 CacheConfig 设置了extraallow任意扩展字段都会原样进入model_dump()后的 init 字典。源码注释中用# type: ignore标注some_setting是“超出模型定义”的自定义字段。可选参数 默认值构造器里的optional_setting: str default setting能正常工作得益于Factory.create在派发前会“删除值为None的条目”从而让未提供的可选参数走实现自身的默认值见 factory.py。注册后即可通过CacheConfig.type字符串寻址register_cache(MyCache, MyCache)之后CacheConfig(typeMyCache)就能在create_cache中被解析工厂的match语句对未知名分支统一抛出“not registered”错误因此自定义类型必须在调用前完成注册。若自定义实现需要落盘示例注释给出了指引参考 JsonCache 的做法在构造器中接收Storage实例或StorageConfig字典通过graphrag_storage的create_storage建立存储连接——这样即可复用仓库整套存储后端本地文件 / Azure Blob / CosmosDB。包统一出口与工程集成视角从init.py 看包对外只导出七个符号Cache、CacheConfig、CacheType、CacheKeyCreator、create_cache、create_cache_key、register_cache。实践中推荐按此分层使用大多数场景配置驱动、快速上手从graphrag_cache顶层导入CacheConfig/CacheType/create_cache/create_cache_key深度定制绕过预注册从graphrag_cache.cache_factory导入单例cache_factory直接调用register/create。在整个 GraphRAG 工程中graphrag-cache位于依赖链的中下层它依赖graphrag-storage与graphrag-common本身不依赖 LLM 层而graphrag-llm等上层包通过本包完成结果缓存。若要独立安装使用可按其 pyproject.toml 的约束Python3.11,3.14在本地 workspace 中以pip install -e ./packages/graphrag-cache方式引入并保证同版本号的graphrag-common、graphrag-storage一并安装否则版本钉死3.1.1会阻碍解析。整体而言GraphRAG Cache 用“一个小抽象 一个单例工厂 一份可扩展配置”就解决了缓存的可插拔问题默认JsonCache开箱即用、MemoryCache极速去重、NoopCache一键关闭而自定义实现通过CacheConfig(extraallow)与register_cache可无缝接入工厂体系让缓存从“功能”真正演化为“可通过配置选型的基础设施”。【免费下载链接】graphragA modular graph-based Retrieval-Augmented Generation (RAG) system项目地址: https://gitcode.com/GitHub_Trending/gr/graphrag创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考