# -*- coding: utf-8 -*- """ 数据模型 """ from datetime import datetime from typing import Any from pydantic_ai._uuid import uuid7 from pydantic_ai.messages import ModelMessage, ModelMessagesTypeAdapter from sqlmodel import Field, SQLModel, JSON # 会话数据表模型 class Conversations(SQLModel, table=True): id: str = Field( default_factory=lambda: str(uuid7()), primary_key=True, description="会话唯一标识", ) user_id: str = Field(index=True, description="用户唯一标识") description: str = Field(description="会话描述") is_deleted: bool = Field(default=False, description="是否删除") dialogs: dict[str, dict[str, Any]] = Field(default_factory=dict, sa_type=JSON, description="对话字典") created_at: datetime = Field(default_factory=datetime.now, description="创建时间") # 运行结果数据表模型 class RunResults(SQLModel, table=True): conversation_id: str = Field(primary_key=True, description="会话唯一标识") dialog_id: str = Field(primary_key=True, description="对话唯一标识") new_messages: str = Field(description="新增消息") @staticmethod def adapt( conversation_id: str, dialog_id: str, new_messages: list[ModelMessage] ) -> "RunResults": """ 适配为运行结果记录 :param conversation_id: 会话唯一标识 :param dialog_id: 对话唯一标识 :param new_messages: 新增消息 :return: 运行结果记录 """ return RunResults( conversation_id=conversation_id, dialog_id=dialog_id, new_messages=ModelMessagesTypeAdapter.dump_json(new_messages).decode( "utf-8" ), # 序列化为 JSON 字符串 )