Semantic Kernel Python 如何用向量存储与 Filter 实现提示语义缓存【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel如果你的应用会收到大量“换一种问法、意思相同”的提示词每次都调用 LLM 会带来不必要的延迟和 token 开销。Semantic Kernel Python 的做法是用向量存储保存“提示词向量 模型结果”记录再挂两个 Kernel Filter渲染阶段先拿渲染后的提示词去向量检索命中相似记录就直接用缓存结果未命中时才真正调用模型调用完成后再把新结果写回向量存储。完整可运行示例见 semantic_caching.py。适用前提Python 3.10按 python/README.md 安装semantic-kernel并配置 OpenAI 服务凭据。准备环境与 OpenAI 服务凭据安装命令来自 README 的 Quick Installpip install --upgrade semantic-kernel示例需要聊天补全和文本嵌入两个 OpenAI 服务且都通过环境变量读取配置。按 README 的说明把凭据设为环境变量或在项目根目录创建.env文件OPENAI_API_KEYsk-... OPENAI_CHAT_MODEL_ID...README 同时支持另一种方式把api_key等配置参数直接传给 AI 服务构造函数来覆盖环境变量。本文示例代码沿用“环境变量”方式与示例代码保持一致。嵌入模型有一个硬约束数据模型把向量字段声明为 1536 维见下文dimensions1536所以配置的嵌入模型必须能生成 1536 维向量否则与数据模型不匹配。定义缓存数据模型与向量存储示例用vectorstoremodel装饰器声明要写入向量存储的记录结构from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from typing import Annotated from uuid import uuid4 from semantic_kernel import Kernel from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAITextEmbedding from semantic_kernel.connectors.in_memory import InMemoryStore from semantic_kernel.data.vector import VectorStore, VectorStoreCollection, VectorStoreField, vectorstoremodel from semantic_kernel.filters import FilterTypes, FunctionInvocationContext, PromptRenderContext from semantic_kernel.functions import FunctionResult COLLECTION_NAME llm_responses RECORD_ID_KEY cache_record_id # Define a simple data model to store, the prompt and the result # we annotate the prompt field as the vector field, the prompt itself will not be stored. # and if you use include_vectors in the search, it will return the vector, but not the prompt. vectorstoremodel(collection_nameCOLLECTION_NAME) dataclass class CacheRecord: result: Annotated[str, VectorStoreField(data, is_full_text_indexedTrue)] prompt: Annotated[str | None, VectorStoreField(vector, dimensions1536)] None id: Annotated[str, VectorStoreField(key)] field(default_factorylambda: str(uuid4()))三个字段各有用途result存模型返回的结果文本建全文索引is_full_text_indexedTrueprompt向量字段。注意示例注释的说明——提示词原文不会存储存的是它的向量检索时即使使用include_vectors返回的也是向量而不是提示词id记录键默认用uuid4生成。向量存储使用内置的InMemoryStore构造时必须传入嵌入生成器embedding OpenAITextEmbedding(service_idembedder) vector_store InMemoryStore(embedding_generatorembedding)[InMemoryStore](https://link.gitcode.com/i/035988c19e62ee5887069bdb0c8dd10d)的源码 docstring 明确说明内存集合是临时的只存在于内存中不会持久化到磁盘或任何外部存储。也就是说进程重启后缓存清空这是本方案的已知边界见文末。实现语义缓存 Filter缓存逻辑放在一个类里分别挂到提示词渲染和函数调用两个 Filter 点上class PromptCacheFilter: A filter to cache the results of the prompt rendering and function invocation. def __init__( self, vector_store: VectorStore, score_threshold: float 0.2, ): if vector_store.embedding_generator is None: raise ValueError(The vector store must have an embedding generator.) self.vector_store vector_store self.collection: VectorStoreCollection[str, CacheRecord] vector_store.get_collection(record_typeCacheRecord) self.score_threshold score_threshold async def on_prompt_render( self, context: PromptRenderContext, next: Callable[[PromptRenderContext], Awaitable[None]] ): await next(context) await self.collection.ensure_collection_exists() results await self.collection.search(context.rendered_prompt, vector_property_nameprompt, top1) async for result in results.results: if result.score and result.score self.score_threshold: context.function_result FunctionResult( functioncontext.function.metadata, valueresult.record.result, rendered_promptcontext.rendered_prompt, metadata{RECORD_ID_KEY: result.record.id}, ) async def on_function_invocation( self, context: FunctionInvocationContext, next: Callable[[FunctionInvocationContext], Awaitable[None]] ): await next(context) result context.result if result and result.rendered_prompt and RECORD_ID_KEY not in result.metadata: cache_record CacheRecord(promptresult.rendered_prompt, resultstr(result)) await self.collection.ensure_collection_exists() await self.collection.upsert(cache_record)两个钩子的分工on_prompt_render查缓存。先执行await next(context)走完后续渲染环节拿到context.rendered_prompt然后以它为查询文本对prompt向量字段做top1检索。命中判据是result.score self.score_threshold。方向依据示例 docstring内存向量存储的默认距离度量是余弦距离cosine distance得分越接近 0 表示匹配越接近所以阈值比较用小于号。命中时直接把context.function_result设为缓存记录里的result并在 metadata 里写入cache_record_id这次调用就不再走模型。on_function_invocation写缓存。先执行await next(context)让真正的模型调用完成然后检查两点result.rendered_prompt存在、且 metadata 里没有cache_record_id。后者表示这次结果不是缓存命中产生的命中路径由on_prompt_render写入该键即是一条新结果于是构造CacheRecord并upsert进集合。两个 Filter 的上下文类型PromptRenderContext、FunctionInvocationContext以及FilterTypes枚举都从 filters 包 导入。组装 Kernel 并注册 Filterkernel Kernel() chat OpenAIChatCompletion(service_iddefault) kernel.add_service(chat) vector_store InMemoryStore(embedding_generatorembedding) cache PromptCacheFilter(vector_storevector_store) kernel.add_filter(FilterTypes.PROMPT_RENDERING, cache.on_prompt_render) kernel.add_filter(FilterTypes.FUNCTION_INVOCATION, cache.on_function_invocation)注意两点add_service只注册了聊天服务嵌入服务只作为embedding_generator传给向量存储不注册进 Kernelscore_threshold在构造PromptCacheFilter时可选默认0.2。运行并判断缓存是否生效示例的完整代码在 semantic_caching.py。把该文件放入你自己的项目已按上文安装semantic-kernel并配置好环境变量然后运行python semantic_caching.py示例main()连续发起三个调用并用一个计时辅助函数打印每次的Elapsed Timeasync def execute_async(kernel: Kernel, title: str, prompt: str): Helper method to execute and log time. print(f{title}: {prompt}) start time.time() result await kernel.invoke_prompt(prompt) elapsed time.time() - start print(f\tElapsed Time: {elapsed:.3f}) return result r1 await execute_async(kernel, First run, Whats the tallest building in New York?) r2 await execute_async(kernel, Second run, How are you today?) r3 await execute_async(kernel, Third run, What is the highest building in New York City?)三个提示词的意图是刻意的第一次提问建立缓存第二次问的是完全不同的问题应正常走模型并写入缓存第三次与第一次语义相近但措辞不同若命中阈值就由缓存直接给出结果。判断方式是对比三次打印出的Elapsed Time命中的那次不再发起模型请求耗时取决于你的网络与模型环境不存在固定的预期数值不要以某个固定秒数作为成功条件。仓库的集成测试也覆盖了这个示例在 test_concepts.py 中semantic_caching用例只有在设置了COMPLETIONS_CONCEPT_SAMPLE环境变量时才会运行否则被跳过。想按仓库测试方式回归时可在python/目录下执行COMPLETIONS_CONCEPT_SAMPLE只需设置为任意值COMPLETIONS_CONCEPT_SAMPLE1 pytest tests/samples/test_concepts.py -k semantic_caching阈值调整与限制命中阈值score_threshold默认0.2比较方向是score score_threshold余弦距离越接近 0 越接近。传入更大值会让更多“大致相似”的提示词命中缓存更小值则更严格。示例未讨论阈值如何按业务选取按你自己对“相似”的要求调整即可。缓存不持久InMemoryStore的数据只在内存中进程退出即丢失in_memory.py docstring。需要跨重启保留缓存时可换用实现同一VectorStore接口的其他连接器README 列出的 Vector DB 支持包括 Azure AI Search、Elasticsearch、Chroma 等。示例本身只演示了InMemoryStore替换后注意新存储的向量维度要与数据模型声明的 1536 维保持一致。嵌入维度必须匹配dimensions1536写在数据模型里嵌入服务生成的向量维度必须与它一致否则与模型声明不匹配。仓库 README 顶部注明Semantic Kernel 的后续方向是 Microsoft Agent FrameworkMAF。本文描述的缓存机制按当前仓库中的代码与示例为准。【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Semantic Kernel Python 如何用向量存储与 Filter 实现提示语义缓存
Semantic Kernel Python 如何用向量存储与 Filter 实现提示语义缓存【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel如果你的应用会收到大量“换一种问法、意思相同”的提示词每次都调用 LLM 会带来不必要的延迟和 token 开销。Semantic Kernel Python 的做法是用向量存储保存“提示词向量 模型结果”记录再挂两个 Kernel Filter渲染阶段先拿渲染后的提示词去向量检索命中相似记录就直接用缓存结果未命中时才真正调用模型调用完成后再把新结果写回向量存储。完整可运行示例见 semantic_caching.py。适用前提Python 3.10按 python/README.md 安装semantic-kernel并配置 OpenAI 服务凭据。准备环境与 OpenAI 服务凭据安装命令来自 README 的 Quick Installpip install --upgrade semantic-kernel示例需要聊天补全和文本嵌入两个 OpenAI 服务且都通过环境变量读取配置。按 README 的说明把凭据设为环境变量或在项目根目录创建.env文件OPENAI_API_KEYsk-... OPENAI_CHAT_MODEL_ID...README 同时支持另一种方式把api_key等配置参数直接传给 AI 服务构造函数来覆盖环境变量。本文示例代码沿用“环境变量”方式与示例代码保持一致。嵌入模型有一个硬约束数据模型把向量字段声明为 1536 维见下文dimensions1536所以配置的嵌入模型必须能生成 1536 维向量否则与数据模型不匹配。定义缓存数据模型与向量存储示例用vectorstoremodel装饰器声明要写入向量存储的记录结构from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from typing import Annotated from uuid import uuid4 from semantic_kernel import Kernel from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAITextEmbedding from semantic_kernel.connectors.in_memory import InMemoryStore from semantic_kernel.data.vector import VectorStore, VectorStoreCollection, VectorStoreField, vectorstoremodel from semantic_kernel.filters import FilterTypes, FunctionInvocationContext, PromptRenderContext from semantic_kernel.functions import FunctionResult COLLECTION_NAME llm_responses RECORD_ID_KEY cache_record_id # Define a simple data model to store, the prompt and the result # we annotate the prompt field as the vector field, the prompt itself will not be stored. # and if you use include_vectors in the search, it will return the vector, but not the prompt. vectorstoremodel(collection_nameCOLLECTION_NAME) dataclass class CacheRecord: result: Annotated[str, VectorStoreField(data, is_full_text_indexedTrue)] prompt: Annotated[str | None, VectorStoreField(vector, dimensions1536)] None id: Annotated[str, VectorStoreField(key)] field(default_factorylambda: str(uuid4()))三个字段各有用途result存模型返回的结果文本建全文索引is_full_text_indexedTrueprompt向量字段。注意示例注释的说明——提示词原文不会存储存的是它的向量检索时即使使用include_vectors返回的也是向量而不是提示词id记录键默认用uuid4生成。向量存储使用内置的InMemoryStore构造时必须传入嵌入生成器embedding OpenAITextEmbedding(service_idembedder) vector_store InMemoryStore(embedding_generatorembedding)[InMemoryStore](https://link.gitcode.com/i/035988c19e62ee5887069bdb0c8dd10d)的源码 docstring 明确说明内存集合是临时的只存在于内存中不会持久化到磁盘或任何外部存储。也就是说进程重启后缓存清空这是本方案的已知边界见文末。实现语义缓存 Filter缓存逻辑放在一个类里分别挂到提示词渲染和函数调用两个 Filter 点上class PromptCacheFilter: A filter to cache the results of the prompt rendering and function invocation. def __init__( self, vector_store: VectorStore, score_threshold: float 0.2, ): if vector_store.embedding_generator is None: raise ValueError(The vector store must have an embedding generator.) self.vector_store vector_store self.collection: VectorStoreCollection[str, CacheRecord] vector_store.get_collection(record_typeCacheRecord) self.score_threshold score_threshold async def on_prompt_render( self, context: PromptRenderContext, next: Callable[[PromptRenderContext], Awaitable[None]] ): await next(context) await self.collection.ensure_collection_exists() results await self.collection.search(context.rendered_prompt, vector_property_nameprompt, top1) async for result in results.results: if result.score and result.score self.score_threshold: context.function_result FunctionResult( functioncontext.function.metadata, valueresult.record.result, rendered_promptcontext.rendered_prompt, metadata{RECORD_ID_KEY: result.record.id}, ) async def on_function_invocation( self, context: FunctionInvocationContext, next: Callable[[FunctionInvocationContext], Awaitable[None]] ): await next(context) result context.result if result and result.rendered_prompt and RECORD_ID_KEY not in result.metadata: cache_record CacheRecord(promptresult.rendered_prompt, resultstr(result)) await self.collection.ensure_collection_exists() await self.collection.upsert(cache_record)两个钩子的分工on_prompt_render查缓存。先执行await next(context)走完后续渲染环节拿到context.rendered_prompt然后以它为查询文本对prompt向量字段做top1检索。命中判据是result.score self.score_threshold。方向依据示例 docstring内存向量存储的默认距离度量是余弦距离cosine distance得分越接近 0 表示匹配越接近所以阈值比较用小于号。命中时直接把context.function_result设为缓存记录里的result并在 metadata 里写入cache_record_id这次调用就不再走模型。on_function_invocation写缓存。先执行await next(context)让真正的模型调用完成然后检查两点result.rendered_prompt存在、且 metadata 里没有cache_record_id。后者表示这次结果不是缓存命中产生的命中路径由on_prompt_render写入该键即是一条新结果于是构造CacheRecord并upsert进集合。两个 Filter 的上下文类型PromptRenderContext、FunctionInvocationContext以及FilterTypes枚举都从 filters 包 导入。组装 Kernel 并注册 Filterkernel Kernel() chat OpenAIChatCompletion(service_iddefault) kernel.add_service(chat) vector_store InMemoryStore(embedding_generatorembedding) cache PromptCacheFilter(vector_storevector_store) kernel.add_filter(FilterTypes.PROMPT_RENDERING, cache.on_prompt_render) kernel.add_filter(FilterTypes.FUNCTION_INVOCATION, cache.on_function_invocation)注意两点add_service只注册了聊天服务嵌入服务只作为embedding_generator传给向量存储不注册进 Kernelscore_threshold在构造PromptCacheFilter时可选默认0.2。运行并判断缓存是否生效示例的完整代码在 semantic_caching.py。把该文件放入你自己的项目已按上文安装semantic-kernel并配置好环境变量然后运行python semantic_caching.py示例main()连续发起三个调用并用一个计时辅助函数打印每次的Elapsed Timeasync def execute_async(kernel: Kernel, title: str, prompt: str): Helper method to execute and log time. print(f{title}: {prompt}) start time.time() result await kernel.invoke_prompt(prompt) elapsed time.time() - start print(f\tElapsed Time: {elapsed:.3f}) return result r1 await execute_async(kernel, First run, Whats the tallest building in New York?) r2 await execute_async(kernel, Second run, How are you today?) r3 await execute_async(kernel, Third run, What is the highest building in New York City?)三个提示词的意图是刻意的第一次提问建立缓存第二次问的是完全不同的问题应正常走模型并写入缓存第三次与第一次语义相近但措辞不同若命中阈值就由缓存直接给出结果。判断方式是对比三次打印出的Elapsed Time命中的那次不再发起模型请求耗时取决于你的网络与模型环境不存在固定的预期数值不要以某个固定秒数作为成功条件。仓库的集成测试也覆盖了这个示例在 test_concepts.py 中semantic_caching用例只有在设置了COMPLETIONS_CONCEPT_SAMPLE环境变量时才会运行否则被跳过。想按仓库测试方式回归时可在python/目录下执行COMPLETIONS_CONCEPT_SAMPLE只需设置为任意值COMPLETIONS_CONCEPT_SAMPLE1 pytest tests/samples/test_concepts.py -k semantic_caching阈值调整与限制命中阈值score_threshold默认0.2比较方向是score score_threshold余弦距离越接近 0 越接近。传入更大值会让更多“大致相似”的提示词命中缓存更小值则更严格。示例未讨论阈值如何按业务选取按你自己对“相似”的要求调整即可。缓存不持久InMemoryStore的数据只在内存中进程退出即丢失in_memory.py docstring。需要跨重启保留缓存时可换用实现同一VectorStore接口的其他连接器README 列出的 Vector DB 支持包括 Azure AI Search、Elasticsearch、Chroma 等。示例本身只演示了InMemoryStore替换后注意新存储的向量维度要与数据模型声明的 1536 维保持一致。嵌入维度必须匹配dimensions1536写在数据模型里嵌入服务生成的向量维度必须与它一致否则与模型声明不匹配。仓库 README 顶部注明Semantic Kernel 的后续方向是 Microsoft Agent FrameworkMAF。本文描述的缓存机制按当前仓库中的代码与示例为准。【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考