BiSheng 数据模型与存储层深度解析24 个 SQLModel 模型、五种存储引擎与上下文生命周期管理【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng本篇技术指南聚焦 BiSheng开源 LLM DevOps 平台持久化层的完整设计MySQL 中的 24 个 SQLModel ORM 模型如何统一承载用户、应用、会话、权限等关系型数据Milvus、Elasticsearch、MinIO、Redis 四种异构存储如何与 MySQL 协同以及core/context/下上下文管理系统如何统一编排所有存储引擎的连接生命周期。读完本文你将掌握 BiSheng 的模型分类与关系、DAO 三层结构的编码规范、五种存储引擎的职责边界以及 BaseContextManager 的状态机与双锁机制实现原理可直接用于二次开发中的数据建模与基础设施接入。持久化层全景BiSheng 的持久化层由 MySQL 中的 24 个 SQLModel ORM 模型和 5 种异构存储引擎协同组成。关系型数据用户、应用、会话、权限等存储在 MySQL 中通过统一的 DAO 模式提供同步/异步访问向量数据存入 Milvus关键词索引交给 Elasticsearch文件对象托管于 MinIO会话状态和缓存则由 Redis 承载。所有存储引擎的连接生命周期由src/backend/bisheng/core/context/下的上下文管理系统统一编排。从源码结构看模型目录 src/backend/bisheng/database/models/ 除本文所述的 24 个核心模型外还包含随 v2.5 多租户与组织架构演进新增的tenant.py租户模型、department.py、department_admin_grant.py部门树与管理员授权、failed_tuple.pyOpenFGA 写入失败的元组留痕等文件可视为核心模型的扩展族。核心模型清单以下 24 个模型文件位于 src/backend/bisheng/database/models/ 目录下模型类文件用途Flowflow.py应用/工作流/助手的统一定义包含名称、JSON 画布数据、状态、类型FlowVersionflow_version.py应用版本控制每个版本独立保存画布数据快照Assistantassistant.pyAI 助手配置包含系统提示词、模型参数、温度等AssistantLinkassistant.py助手关联表连接助手与工具、技能、知识库Templatetemplate.py应用模板预置的工作流/助手模板ChatMessagemessage.py聊天消息记录支持 LONGTEXT 消息体、点赞、敏感词状态MessageSessionsession.py会话记录关联应用与用户汇总互动统计Rolerole.py角色定义内置管理员角色(ID1)和默认角色(ID2)RoleAccessrole_access.py角色权限映射定义角色对各类资源的读写权限Groupgroup.py用户组默认组 ID2GroupResourcegroup_resource.py用户组资源共享映射UserGroupuser_group.py用户-组关联表含组管理员标识UserLinkuser_link.py用户关联信息如常用应用等Tagtag.py标签定义区分知识库标签和应用标签TagLinktag.py标签-资源关联表支持多资源类型绑定Datasetdataset.py微调数据集元数据Evaluationevaluation.py评测任务记录执行状态、评分结果、结果文件路径Reportreport.py报告模板与生成记录VariableValuevariable_value.py工作流节点变量值持久化RecallChunkrecall_chunk.pyRAG 召回追踪记录每次检索的关键词与命中分块InviteCodeinvite_code.py邀请码管理支持批次、用量限制AuditLogaudit_log.py审计日志记录用户操作行为与 IP 地址MarkTaskmark_task.py数据标注任务定义MarkRecordmark_record.py标注记录关联任务与会话MarkAppUsermark_app_user.py标注任务的应用-用户分配模型关系图模型分类详解应用层模型Flow是系统中最核心的模型通过flow_type枚举统一承载六种应用类型枚举定义见 flow.py枚举值FlowType说明5ASSISTANTAI 助手10WORKFLOW工作流15WORKSTATION工作台20LINSIGHT灵思模式25CHANNEL_ARTICLE频道文章助手30KNOLEDGE_SPACE知识空间Flow.data字段以 JSON 格式存储完整的画布定义包含nodes和edges创建时会自动校验 JSON 结构——flow.py 中的validate_json字段校验器会在写入前强制检查nodes与edges两个键是否存在否则抛出ValueError。在数据库层data通过sa_columnColumn(JsonType)映射为 JSON 列方言无关的 JsonType见core/database/dialect_helpers.py。FlowVersion为每个应用维护独立的版本链is_current标记当前生效版本。值得注意的实现细节FlowDao.create_flowflow.py在创建 Flow 时会在同一事务内自动创建一个名为v0、is_current1的默认版本快照并根据flow_type将应用类型映射为ApplicationTypeEnumWORKFLOW/ASSISTANT/LINSIGHT/DAILY_CHAT上报遥测事件NEW_APPLICATIONdelete_flow则通过is_delete1软删除版本记录保留历史审计。Flow.status控制应用上下线状态OFFLINE(1)表示离线编辑中ONLINE(2)表示已上线可用。FlowBase中还包含tenant_id默认 1多租户归属、is_sharedF017 根资源向子租户共享的标记镜像 FGA 的 shared_with 元组、guide_word引导语等字段。查询侧get_flows默认只 select 元数据列而刻意排除庞大的data画布字段源码注释明确data 数据量太大对 MySQL 有影响列表排序统一按update_time倒序分页支持传统 OFFSET 与 F027 引入的 keyset 游标分页两种模式。Assistant独立存储助手的 LLM 配置model_name、temperature、max_token、system_prompt通过AssistantLink关联表将助手与工具(tool_id)、技能(flow_id)、知识库(knowledge_id)三类资源建立多对多关系。查看 assistant.py 可见其默认值temperature1、max_token32000、statusOFFLINE、is_delete0knowledge_auth字段F041控制运行时知识库检索是校验当前提问用户的 view_file 权限还是沿用配置作者的权限。AssistantLinkDao提供了insert_batch、update_assistant_tool、update_assistant_flow、update_assistant_knowledge等按资源维度全量替换关联的方法注意update_assistant_knowledge保存知识库关联时必须携带技能 ID。会话与消息模型MessageSession以chat_id为主键记录用户与应用的一次完整对话会话。包含会话统计字段like、dislike、copied和敏感词审核状态。group_ids以 JSON 数组存储所属用户组支持按组过滤会话。实现细节session.py 中的insert_one/async_insert_one在写入时若未显式传group_ids会自动查询该用户所属的用户组并回填touch_session使用asyncio.shield包裹 UPDATE→COMMIT防止请求取消导致行锁泄漏idle-in-transaction拖垮连接池。filter_session支持按 chat_ids、flow_ids、user_ids、反馈类型like/dislike/copied、时间范围、敏感词状态、flow_type 组合过滤并可选按 update_time 或 create_time 排序。ChatMessage存储具体的消息内容message字段使用 MySQL LONGTEXT 类型以支持超长回复message.py。关键字段包括is_bot-- 区分用户消息与 AI 回复type/category-- 消息类型分类如 question、answerliked/solved-- 用户反馈0 未评/1 赞/2 踩intermediate_steps-- 推理过程日志Text 类型files-- 关联的上传文件LargeTextsensitive_status-- 敏感词检测结果1 通过/2 违规mark_status/mark_user-- 标注状态与被标注人配合标注系统ChatMessage通过__table_args__ {mysql_charset: utf8mb4, mysql_collate: utf8mb4_unicode_ci}在表级别设置utf8mb4字符集以支持 emoji 等特殊字符。ChatMessageDao.aget_messages_by_chat_id有一个值得借鉴的细节先按create_time DESC, id DESC取最近 N 条再在内存中 reverse保证返回最新时间窗口、时间升序的消息且用自增id作为二级排序键避免同一秒内 question 与 answer 顺序错乱。RBAC 权限模型权限体系采用用户 - 用户组 - 角色 - 权限四层结构User ──┬── UserGroup ──── Group │ │ └── UserRole ───── Role ──── RoleAccess │ GroupResource ┘RoleAccess通过AccessType枚举role_access.py定义 13 种权限类型编码AccessType说明1KNOWLEDGE知识库读权限3KNOWLEDGE_WRITE知识库写权限5ASSISTANT_READ助手读权限6ASSISTANT_WRITE助手写权限7GPTS_TOOL_READ工具读权限8GPTS_TOOL_WRITE工具写权限9WORKFLOW工作流读权限10WORKFLOW_WRITE工作流写权限11DASHBOARD看板读权限12DASHBOARD_WRITE看板写权限99WEB_MENU前端菜单栏权限RoleAccess的third_id字段统一指向被授权资源的 ID如 Flow.id / Assistant.idrole_id type third_id构成语义上的三元组权限判定RoleAccessDao提供了judge_role_access/ajudge_role_access判定单条、get_role_access_batch批量、update_role_access_all先删后插全量刷新等 API。同一文件中的WebMenuResource枚举WORKSTATION、ADMIN、BUILD、KNOWLEDGE_SPACE、LINSIGHT_TASK_MODE 等则把WEB_MENU权限细化为前端菜单栏的可见性控制项。GroupResource通过ResourceTypeEnum定义资源类型KNOWLEDGE1, ASSISTANT3, GPTS_TOOL4, WORK_FLOW5, DASHBOARD6, WORKSTATION7, SPACE_FILE8将资源共享到指定用户组。UserGroup是用户与组的多对多关联表is_group_admin标识该用户是否为组管理员。从 user_group.py 可见其user_id与group_id构成复合主键且都声明了外键user.user_id、group.id与ondeleteCASCADE保证成员关系随主体删除级联清理。业务支撑模型Evaluation-- 评测任务模型通过exec_typeflow/assistant/workflow和unique_id关联被评测的应用status追踪执行状态1 运行中/2 失败/3 成功result_score以 JSON 存储评分结果Dataset-- 微调数据集object_name指向 MinIO 中的数据文件Report-- 报告模板与生成记录object_name指向 MinIO 中的模板文件Tag / TagLink-- 标签系统business_type区分知识库标签knowledge_space与应用标签applicationTagLink 通过唯一约束resource_id resource_type tag_id防止重复绑定VariableValue-- 工作流节点变量持久化记录 flow_id、version_id、node_id 和变量值value_type区分文本(1)、列表(2)、文件(3)RecallChunk-- RAG 召回追踪关联 message_id 和 chat_id记录检索关键词keywords和命中的文档分块chunk及元数据InviteCode-- 邀请码管理支持批次batch_id/batch_name、用量限制limit/used和用户绑定标注模型标注系统由三个模型协作MarkTask-- 标注任务定义包含创建者、关联应用 ID、标注人员列表process_users状态枚举为 DEFAULT(1)/DONE(2)/ING(3)MarkRecord-- 标注记录关联 task_id 和 session_id追踪每条会话的标注状态MarkAppUser-- 标注任务中的应用-用户分配关系审计模型AuditLog记录系统中的关键操作行为。system_id标识操作所属模块chat/build/knowledge/system/dashboard 等event_type记录具体行为如 create_chat、delete_knowledge、user_loginobject_type标识操作对象类型work_flow/assistant/knowledge 等并记录操作者 IP 地址。主键使用 UUID 格式通过generate_uuid工厂函数生成见 utils.py。DAO 模式所有模型文件遵循统一的三层结构Base Schema - Model(tableTrue) - Dao 类。基类所有模型继承自SQLModelSerializable定义在 common/models/base.py它扩展了 SQLModel 并默认以 JSON 模式序列化class SQLModelSerializable(SQLModel): model_config ConfigDict(from_attributesTrue) def model_dump(self, **kwargs) - Dict[str, Any]: if mode not in kwargs: kwargs[mode] json return super().model_dump(**kwargs)基类还提供了create_new(**data)工厂方法用于便捷构造实例。model_dump默认强制modejson保证所有 ORM 对象序列化输出为 JSON 兼容类型datetime → ISO 字符串、UUID → 字符串避免 FastAPI 响应序列化时因类型不匹配报错。三层结构以 Flow 为例说明典型的模型文件组织方式flow.py# 1. Base Schema -- 定义字段、校验逻辑不映射数据库表 class FlowBase(SQLModelSerializable): name: str Field(indexTrue) user_id: Optional[int] Field(defaultNone, indexTrue) data: Optional[Dict] Field(defaultNone) status: Optional[int] Field(default1) flow_type: Optional[int] Field(defaultFlowType.WORKFLOW.value) create_time: Optional[datetime] Field(...) update_time: Optional[datetime] Field(...) # 2. Model -- 映射数据库表声明主键和特殊列类型 class Flow(FlowBase, tableTrue): id: str Field(default_factorygenerate_uuid, primary_keyTrue, uniqueTrue) data: Optional[Dict] Field(defaultNone, sa_columnColumn(JSON)) # 3. Read/Create/Update Schema -- API 层的请求/响应模型 class FlowRead(FlowBase): id: str user_name: Optional[str] None class FlowCreate(FlowBase): flow_id: Optional[str] None class FlowUpdate(SQLModelSerializable): name: Optional[str] None description: Optional[str] None # 4. Dao 类 -- 数据访问对象封装 CRUD 操作 class FlowDao(FlowBase): classmethod def create_flow(cls, flow_info: Flow, flow_type: Optional[int]) - Flow: with get_sync_db_session() as session: session.add(flow_info) session.commit() session.refresh(flow_info) return flow_info classmethod async def aget_flow_by_id(cls, flow_id: str) - Flow: async with get_async_db_session() as session: statement select(Flow).where(Flow.id flow_id) result await session.exec(statement) return result.first()这套分层的工程意义在于Base Schema 只负责字段声明与校验逻辑不落表Model 负责主键、索引、特殊列类型等物理映射Read/Create/Update Schema 作为 FastAPI 的请求/响应模型可独立演化Dao 类则把所有 SQL 收拢为可复用的类方法。主键统一使用generate_uuid生成的字符串 UUIDFlow、Assistant或自增整数Role、RoleAccess、ChatMessage、Group等。Dao 方法命名约定前缀含义示例get_同步查询get_one_assistant()aget_异步查询aget_one_assistant()create_同步创建create_flow()update_同步更新update_version()delete_同步删除delete_assistant()filter_条件过滤查询filter_dataset_by_ids()所有 Dao 方法均为classmethod或staticmethod通过get_sync_db_session()/get_async_db_session()获取数据库会话无需实例化 Dao 对象。异步方法通常成对提供如get_flow_by_id/aget_flow_by_id同步版本供 Celery worker、命令行脚本等非 async 上下文调用异步版本供 FastAPI 请求路径使用部分高频写入方法还叠加了retry_on_transient_db_conflict()装饰器见 core/database/retry.py处理数据库瞬态锁冲突。存储引擎职责BiSheng 采用 5 种存储引擎各司其职存储引擎基础设施位置存储内容访问方式MySQL 8.0core/database/全部 ORM 模型数据应用定义、用户、权限、会话、消息、评测、审计等SQLModel/SQLAlchemy同步引擎(pymysql) 异步引擎(aiomysql)Redis 7.0core/cache/redis_manager.py配置缓存(100s TTL)、Celery 消息代理、Linsight 会话状态(1h 过期)、分布式锁RedisManager 上下文管理器Milvuscore/vectorstore/稠密向量索引知识库文档的 Embedding 向量用于语义相似度检索Collection 抽象支持 Milvus/Qdrant/Chroma 多后端Elasticsearchcore/search/elasticsearch/稀疏/关键词索引BM25 检索遥测统计数据EsConnManager 上下文管理器双实例业务 统计MinIOcore/storage/minio/文件对象上传文档、数据集文件、报告模板、应用 Logo、知识库原始文件MinioManager 上下文管理器S3 兼容 APIMySQL 连接管理DatabaseConnectionManagercore/database/connection.py负责管理数据库引擎的创建和连接池配置自动将同步 URLpymysql转换为异步 URLaiomysql连接池默认配置pool_size100、max_overflow20、pool_timeout30、pool_recycle36001 小时回收、pool_pre_pingTrue连接健康检查——以上默认值可直接在connection.py的get_default_engine_config中核对SQLite 等特殊引擎会改用StaticPoolpool_size1、max_overflow0并剥离无关的池参数保证测试环境可用通过get_sync_db_session()和get_async_db_session()两个上下文管理器向 Dao 层提供会话向量存储双通道知识库 RAG 采用稠密向量Milvus 稀疏检索Elasticsearch双通道架构Milvus-- 存储文档分块的 Embedding 向量支持 ANN近似最近邻语义检索Elasticsearch-- 存储文档分块的原文提供 BM25 关键词检索能力两个通道的检索结果经过融合排序后返回兼顾语义理解与精确匹配。Milvus 侧通过 Collection 抽象层屏蔽后端差异除 Milvus 外还支持 Qdrant、Chroma具体向量化与检索管道可参考 知识库与 RAG 管道 与 core/vectorstore/。Elasticsearch 双实例系统注册两个EsConnManager实例业务实例-- 处理知识库文档的关键词索引与检索统计实例statistics_es_name-- 存储遥测统计数据支持用户行为分析和系统运营指标查询上下文管理系统所有存储引擎的连接生命周期由core/context/下的上下文管理系统统一编排。BaseContextManager 生命周期BaseContextManager[T]core/context/base.py是所有基础设施管理器的抽象基类提供线程安全的延迟加载、缓存和生命周期管理。其状态由ContextState枚举定义流转如下UNINITIALIZED ── INITIALIZING ── READY │ │ v v ERROR CLOSING ── CLOSED核心特性均可从 base.py 源码核对双锁机制-- 同步锁threading.Lock和异步锁asyncio.Lock分别保护对应的初始化路径双检查模式--async_get_instance()/sync_get_instance()在获取锁前后各检查一次状态避免重复初始化快速路径直接返回 READY 实例拿到锁后二次确认重试机制-- 默认 3 次重试_default_retry_count3指数退避2^attempt秒超时时间默认 30 秒_default_timeout30.0等待事件--threading.Event和asyncio.Event让后续请求等待首次初始化完成而非重复触发初始化失败时也会 set 事件并抛出ContextInitializationError避免等待者无限阻塞异常体系--ContextError→ContextInitializationError/ContextTimeoutError/ContextStateError三级异常分别对应初始化失败、超时、状态非法ERROR/CLOSED 状态下访问实例辅助能力--is_ready()/get_state()/get_info()提供运行状态观测async_reset()/sync_reset()支持关闭后重置为 UNINITIALIZED 以便重新初始化sync_context()/async_context()上下文管理器配合with语句便捷取用实例ContextRegistry提供注册表式的批量管理支持health_check()健康检查与async_close_all()并行关闭FunctionContextManager-- 基于函数的快速实现允许只传init_func/cleanup_func即可构造管理器同步函数在异步路径下会通过run_in_executor丢入线程池执行ApplicationContextManager 编排ApplicationContextManagercore/context/manager.py作为顶层编排器持有全局app_context单例按依赖顺序注册并初始化所有基础设施上下文管理器实际注册逻辑见_register_default_contextsDatabaseManager -- MySQL 连接 | RedisManager -- Redis 缓存 | MinioManager -- MinIO 对象存储 | EsConnManager (业务) -- Elasticsearch 业务实例 EsConnManager (统计) -- Elasticsearch 统计实例 | HttpClientManager -- HTTP 客户端 | PromptManager -- 提示词管理初始化在 FastAPI lifespan 中触发initialize_app_context(config)关闭时按注册的逆序清理资源_close_contexts_in_reverse_order。编排器支持三类高级能力依赖声明register_context(context, dependencies[...])声明某管理器依赖的其他上下文初始化时递归先初始化依赖项可选降级optionalTrue注册的上下文如 OpenFGA 的FGAManager受config.openfga.enabled开关控制初始化失败仅告警不阻断启动实现降级模式运维接口health_check(include_details)返回每个上下文的健康状态与详细信息restart_context(name)支持单个上下文的重启list_contexts(state_filter)可按状态过滤列出模块级还暴露了get_context、sync_get_instance、async_get_instance、register_context、health_check、close_app_context等便捷函数业务代码通过manager.async_get_instance()或manager.sync_get_instance()获取已初始化的连接实例首次调用时自动触发延迟初始化。子类实现每个具体的 Manager 继承BaseContextManager[T]并实现四个抽象方法方法用途_async_initialize() - T异步创建连接/客户端实例_sync_initialize() - T同步创建连接/客户端实例_async_cleanup()异步释放资源关闭连接池等_sync_cleanup()同步释放资源业务代码通过manager.async_get_instance()或manager.sync_get_instance()获取已初始化的连接实例首次调用时自动触发延迟初始化。相关文档文档说明系统架构全景图运行时组件与请求数据流后端领域模块总览15 DDD 模块清单与分层约定知识库与 RAG 管道三阶段文档处理管道向量存储细节部署架构与配置存储服务部署与配置系统【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考