Semantic Kernel Python 聊天历史持久化实战从文件序列化到 Azure Cosmos DB 存储【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel本文以 Semantic Kernel Python SDK 仓库中的python/samples/concepts/chat_history示例为核心系统讲解ChatHistory对象的持久化机制先展示基于临时文件的 JSON 序列化方案再深入 Azure Cosmos DB NoSQL 的向量存储集成方案。读完本文你将掌握ChatHistory内建的文件读写 API、如何自定义数据模型并通过VectorStore扩展ChatHistory子类实现云端存储以及两种方案在生产场景中的取舍与演进方向。一、示例概览与运行前置条件python/samples/concepts/chat_history/目录下包含两个配套示例与一份说明文档文件说明README.md本主题的官方说明文档serialize_chat_history.py基于文件序列化聊天历史的对话机器人示例store_chat_history_in_cosmosdb.py使用 Azure Cosmos DB NoSQL 存储聊天历史的进阶示例两个示例的核心共同点是每个对话轮次都完整落盘聊天历史。官方注释也明确指出这种每轮都读写的做法并非性能最优解而是为了清晰地展示序列化机制本身的运作原理更优的工程做法是仅在会话结束时写入一次且存储介质通常也不应局限于文件。运行示例前需要满足选择一个支持函数调用function calling的聊天补全服务并配置好对应密钥。示例代码通过 chat_completion_services.py 中的Services枚举与get_chat_completion_service_and_request_settings()工厂函数统一创建服务实例可选服务包括OPENAI、AZURE_OPENAI、AZURE_AI_INFERENCE、ANTHROPIC、BEDROCK、GOOGLE_AI、MISTRAL_AI、OLLAMA、ONNX、VERTEX_AI、DEEPSEEK与NVIDIA。环境变量各服务的模型 ID 与密钥均从环境变量读取例如OPENAI_API_KEY、OPENAI_CHAT_MODEL_IDAzure OpenAI 则对应AZURE_OPENAI_CHAT_DEPLOYMENT_NAME、AZURE_OPENAI_API_KEY、AZURE_OPENAI_ENDPOINT等完整的变量对照表见 ALL_SETTINGS.md。运行方式示例采用from samples.concepts.setup.chat_completion_services import ...的相对导入因此需要在python/目录下以python -m方式执行例如python -m samples.concepts.chat_history.serialize_chat_history。二、ChatHistory 的内建序列化能力源码基础在深入两个示例之前先理解ChatHistory类本身提供的持久化基础设施相关实现集中在 chat_history.py。2.1 核心方法ChatHistory继承自KernelBaseModel基于 pydanticmessages字段保存ChatMessageContent列表并提供了四组序列化相关方法serialize()调用model_dump_json(exclude_noneTrue, indent2)将整个历史序列化为格式化的 JSON 字符串chat_history.py。其中exclude_noneTrue会剔除值为None的字段indent2便于人工阅读与 diff。restore_chat_history(chat_history_json)类方法通过model_validate_json()将 JSON 字符串反序列化回ChatHistory实例若 JSON 非法会抛出ContentInitializationErrorchat_history.py。store_chat_history_to_file(file_path)以w模式写入文件——文件不存在则创建存在则整体截断覆盖chat_history.py。load_chat_history_from_file(file_path)类方法以r模式读取文件并调用restore_chat_history完成反序列化chat_history.py。这组 API 正是第一个示例的基础序列化 → 落盘 → 读取 → 反序列化环环相扣。2.2 消息构造辅助方法ChatHistory还提供了一组便捷的追加消息方法两个示例中都会用到add_system_message(content)追加系统消息角色SYSTEM。add_user_message(content)追加用户消息角色USER。add_assistant_message(content)追加助手消息角色ASSISTANT。add_message(message)追加一条ChatMessageContent实例或由 dict 构造的消息。这些方法通过singledispatchmethod实现重载既支持纯文本字符串也支持KernelContent列表用于多模态内容与工具调用结果。当函数调用function calling发生时FunctionResultContent会以TOOL角色消息进入历史而这些结构化内容同样可被serialize()完整保存——这正是带函数调用的对话历史也能持久化的关键。2.3 单元测试印证仓库中的单元测试 test_chat_history.py 覆盖了序列化路径test_serializeL276、test_serialize_and_deserialize_to_chat_historyL300、test_deserialize_invalid_json_raises_exceptionL322以及test_chat_history_serializeL657验证了序列化-反序列化往返一致性与非法输入的处理行为可作为你自行扩展持久化逻辑时的参考基线。三、示例一基于临时文件的聊天历史序列化serialize_chat_history.py 构建了一个带自动函数调用的对话机器人其核心设计是每一轮对话后把历史写入临时 JSON 文件下一轮开始时再读取回来。3.1 服务选择chat_completion_service, request_settings get_chat_completion_service_and_request_settings(Services.OPENAI)该行位于文件第 33 行serialize_chat_history.py。切换服务只需把Services.OPENAI换成前文枚举中的任意一个并保证对应环境变量已配置。request_settings返回的是服务对应的PromptExecutionSettings例如 OpenAI 默认max_tokens2000, temperature0.7, top_p0.8可以直接修改以满足业务需要。3.2 每轮对话的读写循环chat()函数实现了完整的加载 → 对话 → 保存循环async def chat(file) - bool: try: # 尝试从文件加载历史文件不存在则开启新会话 history ChatHistory.load_chat_history_from_file(file_pathfile) print(fChat history successfully loaded {len(history.messages)} messages.) except Exception: print(Chat history file not found. Starting a new conversation.) history ChatHistory() history.add_system_message( You are a chat bot. Your name is Mosscap and you have one goal: figure out what people need. ) user_input input(User: ) # 读取用户输入 if user_input.lower().strip() exit: return False # 输入 exit 退出 history.add_user_message(user_input) # 追加用户消息 result await chat_completion_service.get_chat_message_content(history, request_settings) if result: print(fMosscap: {result}) history.add_message(result) # 追加助手回复 print(fSaving {len(history.messages)} messages to the file.) history.store_chat_history_to_file(file_pathfile) # 整段历史写回文件 return True值得注意的细节首次运行时文件不存在load_chat_history_from_file会抛出异常代码捕获后新建ChatHistory并注入系统提示词机器人名为 Mosscap。add_message(result)直接接收get_chat_message_content返回的ChatMessageContent对象与add_user_message相比更完整地保留了消息内容结构包括函数调用相关的 items。读取与写入对称load_chat_history_from_file/store_chat_history_to_file分别封装了反序列化与序列化调用方无需关心 JSON 细节。3.3 临时文件的创建与清理main()使用tempfile.NamedTemporaryFile在当前目录创建带.json后缀的临时文件with tempfile.NamedTemporaryFile(modew, dir., suffix.json, deleteTrue) as file: print(Welcome to the chat bot!\n Type exit to exit.\n Try a math question to see function calling in action (e.g. what is 33?). f Your chat history will be saved in: {file.name}) while chatting: chatting await chat(file.name)由于deleteTrue程序退出后文件会被自动删除因此不需要任何额外的环境配置即可运行——这正是官方文档强调no additional setup is required的原因。3.4 示例运行输出源码 docstring 中给出了完整交互样例Welcome to the chat bot! Type exit to exit. Try a math question to see function calling in action (e.g. what is 33?). Your chat history will be saved in: local working directory/tmpq1n1f6qk.json Chat history file not found. Starting a new conversation. User: Hello, how are you? Mosscap: Hello! Im here and ready to help. What do you need today? Saving 3 messages to the file. Chat history successfully loaded 3 messages. User: exit可以看到第二轮启动时历史被成功加载3 条消息 系统消息 用户消息 助手消息验证了每轮持久化确实生效。四、示例二将聊天历史存入 Azure Cosmos DB NoSQLstore_chat_history_in_cosmosdb.py 是前一个示例的进阶版用 Azure Cosmos DB NoSQL 替代临时文件作为存储后端并引入了VectorStore抽象。示例代码将整个过程组织为五个步骤下面逐一拆解。4.1 步骤一定义数据模型使用vectorstoremodel装饰器与dataclass定义一个不含向量的简单记录模型vectorstoremodel dataclass class ChatHistoryModel: session_id: Annotated[str, VectorStoreField(key)] user_id: Annotated[str, VectorStoreField(data, is_indexedTrue)] messages: Annotated[list[dict[str, str]], VectorStoreField(data, is_indexedTrue)]字段语义session_id标记为key类型作为记录的主键对应 Cosmos DB 中的id。user_id与messages标记为data类型并设置is_indexedTrue以便后续按用户或内容过滤查询。messages存储为list[dict]因为ChatMessageContent需要先经过model_dump()转成纯 JSON 字典才能序列化入库。从源码结构看VectorStoreField与vectorstoremodel来自 semantic_kernel/data/vector 模块是 Semantic Kernel Python 中向量存储数据模型的通用声明方式即使当前模型不含向量字段后续也可以随时追加VectorStoreField声明为vector类型的字段例如会话摘要的 embedding用于按语义相似度检索相似对话——示例注释中明确指出了这一演进路径。4.2 步骤二扩展 ChatHistory 实现 store/read示例创建ChatHistoryInCosmosDB子类在ChatHistory基础上增加了session_id、user_id、store、collection四个字段与三个方法class ChatHistoryInCosmosDB(ChatHistory): session_id: str user_id: str store: VectorStore collection: VectorStoreCollection[str, ChatHistoryModel] | None None async def create_collection(self, collection_name: str) - None: self.collection self.store.get_collection( collection_namecollection_name, record_typeChatHistoryModel, ) await self.collection.ensure_collection_exists() async def store_messages(self) - None: if self.collection: await self.collection.upsert( ChatHistoryModel( session_idself.session_id, user_idself.user_id, messages[msg.model_dump() for msg in self.messages], ) ) async def read_messages(self) - None: if self.collection: record await self.collection.get(self.session_id) if record: for message in record.messages: self.messages.append(ChatMessageContent.model_validate(message))方法职责create_collection通过store.get_collection()拿到类型化集合再调用ensure_collection_exists()确保底层容器存在。store_messages写入方向。msg.model_dump()把每条ChatMessageContent转为可序列化字典整体upsert进 Cosmos DB按session_id主键覆盖写。read_messages读取方向。collection.get(self.session_id)按主键取回记录再用ChatMessageContent.model_validate(message)将字典还原为消息对象保证反序列化是序列化的严格逆操作。此外示例注释还提醒了两个生产化方向可以使用历史压缩器history reducers控制数据库体积增长也可以接入会话摘要与向量字段实现相似对话的语义检索。4.3 步骤三搭建带函数调用的 Kernelkernel Kernel() kernel.add_plugin(MathPlugin(), plugin_namemath) kernel.add_plugin(TimePlugin(), plugin_nametime) chat_completion_service, request_settings get_chat_completion_service_and_request_settings(Services.AZURE_OPENAI) request_settings.function_choice_behavior FunctionChoiceBehavior.Auto(filters{excluded_plugins: [ChatBot]}) kernel.add_service(chat_completion_service)注册了MathPlugin与TimePlugin来自 semantic_kernel/core_plugins用于演示函数调用。通过FunctionChoiceBehavior.Auto(filters{excluded_plugins: [ChatBot]})开启自动函数调用并排除名为ChatBot的插件避免递归调用机器人自身插件。get_chat_message_content(history, request_settings, kernelkernel)在调用时显式传入kernel使模型在需要时能够执行已注册插件。4.4 步骤四主对话循环async def chat(history: ChatHistoryInCosmosDB) - bool: await history.read_messages() # 先加载既有历史 print(fChat history successfully loaded {len(history.messages)} messages.) if len(history.messages) 0: # 新会话注入系统消息与开场白 history.add_system_message( You are a chat bot. Your name is Mosscap and you have one goal: figure out what people need. ) history.add_user_message(Hi there, who are you?) history.add_assistant_message(I am Mosscap, a chat bot. Im trying to figure out what people need.) user_input input(User: ) if user_input.lower().strip() exit: return False history.add_user_message(user_input) result await chat_completion_service.get_chat_message_content(history, request_settings, kernelkernel) if result: print(fMosscap: {result}) history.add_message(result) print(fSaving {len(history.messages)} messages to AzureCosmosDB.) await history.store_messages() # 每轮结束写回 Cosmos DB return True与文件方案相比差异点在于历史加载改成了异步的read_messages()按session_id从云端拉取且新会话会额外注入一对示例开场白让模型立刻进入角色。4.5 步骤五Store 生命周期管理async with CosmosNoSqlStore(create_databaseTrue) as store: history ChatHistoryInCosmosDB(storestore, session_idsession1, user_iduser) await history.create_collection(collection_namechat_history) # ... 对话循环 ... if delete_when_done and history.collection: await history.collection.ensure_collection_deleted()这里有两个关键点CosmosNoSqlStore(create_databaseTrue)CosmosNoSqlStore是VectorStore的 Azure Cosmos DB NoSQL 实现见 azure_cosmos_db.py。create_databaseTrue表示当目标数据库不存在时自动创建若为False而数据库不存在则操作会抛出VectorStoreOperationException。异步上下文管理器CosmosNoSqlStore实现了__aexit__退出时若客户端由 SDK 内部创建managed_clientTrue则自动close()底层连接azure_cosmos_db.py避免连接泄漏。4.6 环境变量与认证方式CosmosNoSqlStore的配置由CosmosNoSqlSettings类azure_cosmos_db.py从环境变量读取前缀为AZURE_COSMOS_DB_NO_SQL_支持从环境变量或.env文件加载环境变量必填说明AZURE_COSMOS_DB_NO_SQL_URL是Cosmos DB NoSQL 账户的 URI可在 Azure 门户的 Keys Endpoint 中查看AZURE_COSMOS_DB_NO_SQL_KEY否账户主密钥不提供时可改用 Entra ID 认证AZURE_COSMOS_DB_NO_SQL_DATABASE_NAME否数据库名不设置时会使用默认名认证方式有两种可从源码 azure_cosmos_db.py 推断主密钥认证设置AZURE_COSMOS_DB_NO_SQL_KEYSDK 直接用密钥构造CosmosClientEntra ID 认证不设置密钥改为向CosmosNoSqlStore传入credential参数AsyncTokenCredential类型例如AzureCliCredential。这也是官方文档所说你也可以依靠 Entra ID 认证而非密钥的底层实现。五、两种方案对比与生产化建议维度文件序列化方案Cosmos DB NoSQL 方案存储介质本地临时 JSON 文件Azure 云端容器额外配置无自动清理临时文件需要AZURE_COSMOS_DB_NO_SQL_URL及密钥或 Entra ID核心 APIstore_chat_history_to_file/load_chat_history_from_file自定义store_messages/read_messages基于VectorStore数据模型无直接序列化ChatHistoryvectorstoremodel数据类 VectorStoreField声明会话标识文件路径session_id主键扩展性弱可加向量字段、索引过滤、语义检索结合两个示例的官方注释可以提炼出以下工程建议不要每轮都全量落盘。两个示例之所以每轮读写是为了把机制讲清楚生产环境更合理的做法是会话结束时批量写入一次或在写入前判断内容是否有变化。为数据库增长做规划。接入历史压缩器history reducers控制单会话体积按需清理过期会话。利用向量存储的增量价值。在ChatHistoryModel中追加会话摘要的 embedding 向量字段即可借助VectorStore的向量搜索能力实现相似历史会话检索这是文件方案无法比拟的。认证选型。本地开发可用密钥或AzureCliCredential生产环境优先 Entra ID 托管身份。六、进一步探索想要深入ChatHistory的完整 API消息追加、移除、迭代、from_rendered_prompt等阅读 chat_history.py。想了解各聊天服务对应的全部环境变量与配置项参考 ALL_SETTINGS.md。想掌握CosmosNoSqlStore/CosmosNoSqlCollection的向量索引、过滤查询与混合搜索实现研读 azure_cosmos_db.py。想学习VectorStore抽象与数据模型声明方式的通用用法可查看 python/semantic_kernel/data/vector 目录及其单元测试。【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
