企业官网建设流程全解析

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/目录下包含两个配套示例与一份说明文档：

两个示例的核心共同点是：每个对话轮次都完整落盘聊天历史。官方注释也明确指出，这种"每轮都读写"的做法并非性能最优解，而是为了清晰地展示序列化机制本身的运作原理；更优的工程做法是仅在会话结束时写入一次，且存储介质通常也不应局限于文件。

运行示例前需要满足：

1. 选择一个支持函数调用（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。
2. 环境变量：各服务的模型 ID 与密钥均从环境变量读取（例如OPENAI_API_KEY、OPENAI_CHAT_MODEL_ID；Azure OpenAI 则对应AZURE_OPENAI_CHAT_DEPLOYMENT_NAME、AZURE_OPENAI_API_KEY、AZURE_OPENAI_ENDPOINT等），完整的变量对照表见 ALL_SETTINGS.md。
3. 运行方式：示例采用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（基于 pydantic），messages字段保存ChatMessageContent列表，并提供了四组序列化相关方法：

• serialize()：调用model_dump_json(exclude_none=True, indent=2)将整个历史序列化为格式化的 JSON 字符串（chat_history.py）。其中exclude_none=True会剔除值为None的字段，indent=2便于人工阅读与 diff。
• restore_chat_history(chat_history_json)：类方法，通过model_validate_json()将 JSON 字符串反序列化回ChatHistory实例；若 JSON 非法会抛出ContentInitializationError（chat_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_serialize（L276）、test_serialize_and_deserialize_to_chat_history（L300）、test_deserialize_invalid_json_raises_exception（L322）以及test_chat_history_serialize（L657），验证了"序列化-反序列化"往返一致性与非法输入的处理行为，可作为你自行扩展持久化逻辑时的参考基线。

三、示例一：基于临时文件的聊天历史序列化

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_tokens=2000, temperature=0.7, top_p=0.8），可以直接修改以满足业务需要。

3.2 每轮对话的读写循环

chat()函数实现了完整的"加载 → 对话 → 保存"循环：

async def chat(file) -> bool: try: # 尝试从文件加载历史；文件不存在则开启新会话 history = ChatHistory.load_chat_history_from_file(file_path=file) print(f"Chat 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(f"Mosscap:> {result}") history.add_message(result) # 追加助手回复 print(f"Saving {len(history.messages)} messages to the file.") history.store_chat_history_to_file(file_path=file) # 整段历史写回文件 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(mode="w+", dir=".", suffix=".json", delete=True) 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 3+3?')." f" Your chat history will be saved in: {file.name}") while chatting: chatting = await chat(file.name)

由于delete=True，程序退出后文件会被自动删除，因此不需要任何额外的环境配置即可运行——这正是官方文档强调"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 3+3?'). 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! I'm 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 NoSQL

store_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_indexed=True)] messages: Annotated[list[dict[str, str]], VectorStoreField("data", is_indexed=True)]

字段语义：

• session_id标记为"key"类型，作为记录的主键（对应 Cosmos DB 中的id）。
• user_id与messages标记为"data"类型，并设置is_indexed=True以便后续按用户或内容过滤查询。
• 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_name=collection_name, record_type=ChatHistoryModel, ) await self.collection.ensure_collection_exists() async def store_messages(self) -> None: if self.collection: await self.collection.upsert( ChatHistoryModel( session_id=self.session_id, user_id=self.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 步骤三：搭建带函数调用的 Kernel

kernel = Kernel() kernel.add_plugin(MathPlugin(), plugin_name="math") kernel.add_plugin(TimePlugin(), plugin_name="time") 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, kernel=kernel)在调用时显式传入kernel，使模型在需要时能够执行已注册插件。

4.4 步骤四：主对话循环

async def chat(history: ChatHistoryInCosmosDB) -> bool: await history.read_messages() # 先加载既有历史 print(f"Chat 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. I'm 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, kernel=kernel) if result: print(f"Mosscap:> {result}") history.add_message(result) print(f"Saving {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_database=True) as store: history = ChatHistoryInCosmosDB(store=store, session_id="session1", user_id="user") await history.create_collection(collection_name="chat_history") # ... 对话循环 ... if delete_when_done and history.collection: await history.collection.ensure_collection_deleted()

这里有两个关键点：

• CosmosNoSqlStore(create_database=True)：CosmosNoSqlStore是VectorStore的 Azure Cosmos DB NoSQL 实现（见 azure_cosmos_db.py）。create_database=True表示当目标数据库不存在时自动创建；若为False，而数据库不存在，则操作会抛出VectorStoreOperationException。
• 异步上下文管理器：CosmosNoSqlStore实现了__aexit__，退出时若客户端由 SDK 内部创建（managed_client=True）则自动close()底层连接（azure_cosmos_db.py），避免连接泄漏。

4.6 环境变量与认证方式

CosmosNoSqlStore的配置由CosmosNoSqlSettings类（azure_cosmos_db.py）从环境变量读取，前缀为AZURE_COSMOS_DB_NO_SQL_，支持从环境变量或.env文件加载：

认证方式有两种（可从源码 azure_cosmos_db.py 推断）：

1. 主密钥认证：设置AZURE_COSMOS_DB_NO_SQL_KEY，SDK 直接用密钥构造CosmosClient；
2. Entra ID 认证：不设置密钥，改为向CosmosNoSqlStore传入credential参数（AsyncTokenCredential类型），例如AzureCliCredential。这也是官方文档所说"你也可以依靠 Entra ID 认证而非密钥"的底层实现。

五、两种方案对比与生产化建议

结合两个示例的官方注释，可以提炼出以下工程建议：

1. 不要每轮都全量落盘。两个示例之所以"每轮读写"，是为了把机制讲清楚；生产环境更合理的做法是会话结束时批量写入一次，或在写入前判断内容是否有变化。
2. 为数据库增长做规划。接入历史压缩器（history reducers）控制单会话体积；按需清理过期会话。
3. 利用向量存储的增量价值。在ChatHistoryModel中追加会话摘要的 embedding 向量字段，即可借助VectorStore的向量搜索能力实现"相似历史会话检索"，这是文件方案无法比拟的。
4. 认证选型。本地开发可用密钥或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