当前位置: 首页 > wzjs >正文

做网站都需要买什么软件百度seo软件

做网站都需要买什么软件,百度seo软件,重庆的汽车网站建设,盘锦做网站这段 Python 代码是一个完整的 知识库数据库操作模块,用于对本地知识库系统中的知识库进行增删改查(CRUD)操作。它基于 SQLAlchemy ORM 框架 和一个自定义的装饰器 with_session 实现数据库会话管理。 📘 一、整体功能概述 该模块…

这段 Python 代码是一个完整的 知识库数据库操作模块,用于对本地知识库系统中的知识库进行增删改查(CRUD)操作。它基于 SQLAlchemy ORM 框架 和一个自定义的装饰器 @with_session 实现数据库会话管理。


📘 一、整体功能概述

该模块主要实现以下功能:

功能描述
✅ 添加知识库如果不存在则添加,否则更新信息
✅ 列出所有知识库可设置文件数量过滤条件
✅ 检查知识库是否存在使用不区分大小写的匹配方式
✅ 加载知识库基本信息获取名称、向量库类型、嵌入模型
✅ 删除知识库支持按名称删除
✅ 获取知识库详细信息返回字典格式数据,便于接口返回

此外还提供了一个测试入口函数,用于验证这些数据库操作是否正常工作。


🧠 二、代码结构详解

🔹 第一部分:导入依赖与模块说明

from server.db.models.knowledge_base_model import KnowledgeBaseModel
from server.db.session import with_session
  • KnowledgeBaseModel:SQLAlchemy ORM 定义的知识库表模型
  • with_session:一个装饰器,自动注入 SQLAlchemy 的 session 对象,避免手动打开/关闭数据库连接

🔹 第二部分:核心数据库操作函数

1. add_kb_to_db(session, kb_name, kb_info, vs_type, embed_model)

🎯 功能:
  • 向数据库中添加一个新的知识库
  • 若已存在同名知识库,则更新其信息
📌 参数说明:
  • kb_name: 知识库名称(唯一)
  • kb_info: 知识库简介(用于 AI agent 理解用途)
  • vs_type: 向量库类型(如 FAISS、Chroma)
  • embed_model: 嵌入模型名称(如 text-embedding-ada-002)
🔄 工作流程:
  1. 查询是否有同名知识库(忽略大小写)
  2. 若无,则创建并插入新记录
  3. 若有,则更新 info、向量库类型和嵌入模型字段
✅ 返回值:

始终返回 True,表示执行成功


2. list_kbs_from_db(session, min_file_count: int = -1)

🎯 功能:

列出所有文件数量大于指定值的知识库名称列表

📌 参数说明:
  • min_file_count: 最小文件数,用于筛选活跃的知识库
🔄 工作流程:
  • 查询所有满足条件的 kb_name
  • 将结果从 (kb_name,) 转换为 kb_name 字符串列表
✅ 示例输出:
["test_kb", "math_kb", "law_kb"]

3. kb_exists(session, kb_name)

🎯 功能:

检查某个知识库是否存在于数据库中

📌 参数说明:
  • kb_name: 知识库名称(支持模糊匹配,不区分大小写)
✅ 返回值:

布尔值,True 表示存在,False 表示不存在


4. load_kb_from_db(session, kb_name)

🎯 功能:

加载知识库的基本配置信息(名称、向量库类型、嵌入模型)

📌 参数说明:
  • kb_name: 知识库名称(支持模糊匹配)
✅ 返回值:

元组形式 (kb_name, vs_type, embed_model),如果不存在则返回 (None, None, None)


5. delete_kb_from_db(session, kb_name)

🎯 功能:

根据知识库名称删除对应条目

📌 参数说明:
  • kb_name: 知识库名称(支持模糊匹配)
✅ 返回值:

始终返回 True,表示操作成功(即使未找到要删除的对象)


6. get_kb_detail(session, kb_name)

🎯 功能:

获取知识库的完整信息,包括:

  • 名称
  • 简介
  • 向量库类型
  • 嵌入模型
  • 文件数量
  • 创建时间
✅ 返回值:
  • 存在时返回字典:
    {"kb_name": "test_kb","kb_info": "这是一个用于测试的知识库","vs_type": "FAISS","embed_model": "text-embedding-ada-002","file_count": 0,"create_time": datetime.datetime(...)
    }
    
  • 不存在时返回空字典 {}

🔁 三、装饰器机制:@with_session

@with_session
def add_kb_to_db(session, kb_name, kb_info, vs_type, embed_model):

🧩 作用:

  • 自动为你打开数据库会话(session)
  • 函数执行完毕后自动提交事务或回滚异常
  • 避免手动管理 session,提高代码可维护性

💡 类似于上下文管理器:

with get_db() as session:# 执行数据库操作

📐 四、ORM 模型结构(来自 knowledge_base_model.py)

你使用的 KnowledgeBaseModel 应该是如下结构(简化版):

class KnowledgeBaseModel(Base):__tablename__ = 'knowledge_base'id = Column(Integer, primary_key=True, autoincrement=True)kb_name = Column(String(50), unique=True, comment='知识库名称')kb_info = Column(String(200), comment='知识库简介')vs_type = Column(String(50), comment='向量库类型')embed_model = Column(String(50), comment='嵌入模型名称')file_count = Column(Integer, default=0, comment='文件数量')create_time = Column(DateTime, default=func.now(), comment='创建时间')

🧪 五、测试入口函数(if name == “main”)

这是程序的主入口,用于运行测试流程。

✅ 测试步骤如下:

  1. 添加知识库

    • 使用 add_kb_to_db() 添加一个名为 test_kb 的知识库
  2. 检查是否存在

    • 使用 kb_exists() 判断知识库是否入库成功
  3. 获取详细信息

    • 使用 get_kb_detail() 输出当前知识库的所有字段信息
  4. 加载基本信息

    • 使用 load_kb_from_db() 获取 kb_name, vs_type, embed_model
  5. 列出所有知识库

    • 使用 list_kbs_from_db() 获取当前数据库中所有知识库名称
  6. 删除知识库

    • 使用 delete_kb_from_db() 删除刚添加的知识库
  7. 再次检查是否存在

    • 确认删除是否成功

📋 六、测试流程示意

🧪 开始测试知识库数据库操作函数...
📌 正在添加知识库:test_kb
✅ 添加完成
🔍 知识库 test_kb 是否存在?是
📝 获取知识库 test_kb 的详细信息
🧠 加载知识库信息:test_kb, FAISS, text-embedding-ada-002
📋 当前数据库中的知识库列表:
['test_kb']
🗑️ 正在删除知识库:test_kb
✅ 删除完成
🔍 删除后,知识库 test_kb 是否还存在?否
🎉 所有测试通过!

📦 七、适用场景与扩展建议

✅ 适用场景:

场景描述
🧠 AI 助手后台管理多个知识库,支持文档问答
📄 RAG 架构记录每个知识库的向量库类型与嵌入模型
📈 数据统计统计知识库文件数量、创建时间等
🧩 多任务适配不同知识库使用不同 embedding 模型

🛠️ 推荐扩展方向:

扩展点描述
✅ 增加 hash 校验字段避免重复上传相同文件
✅ 添加文件路径字段file_path TEXT,方便定位实际文件位置
✅ 支持软删除is_deleted BOOLEAN DEFAULT False
✅ 支持异步状态标记status ENUM('pending', 'processing', 'done')
✅ 多线程安全优化提升大规模并发访问效率

📊 八、实战代码部分展示:

#!/usr/bin/env python
# coding=utf-8"""
@author: zgw
@date: 2025/6/7 16:07
@source from: 
"""
from server.db.models.knowledge_base_model import KnowledgeBaseModel
from server.db.session import with_session@with_session
def add_kb_to_db(session, kb_name, kb_info, vs_type, embed_model):# 创建知识库实例kb = session.query(KnowledgeBaseModel).filter(KnowledgeBaseModel.kb_name.ilike(kb_name)).first()if not kb:kb = KnowledgeBaseModel(kb_name=kb_name, kb_info=kb_info, vs_type=vs_type, embed_model=embed_model)session.add(kb)else:  # update kb with new vs_type and embed_modelkb.kb_info = kb_infokb.vs_type = vs_typekb.embed_model = embed_modelreturn True@with_session
def list_kbs_from_db(session, min_file_count: int = -1):kbs = session.query(KnowledgeBaseModel.kb_name).filter(KnowledgeBaseModel.file_count > min_file_count).all()kbs = [kb[0] for kb in kbs]return kbs@with_session
def kb_exists(session, kb_name):kb = session.query(KnowledgeBaseModel).filter(KnowledgeBaseModel.kb_name.ilike(kb_name)).first()status = True if kb else Falsereturn status@with_session
def load_kb_from_db(session, kb_name):kb = session.query(KnowledgeBaseModel).filter(KnowledgeBaseModel.kb_name.ilike(kb_name)).first()if kb:kb_name, vs_type, embed_model = kb.kb_name, kb.vs_type, kb.embed_modelelse:kb_name, vs_type, embed_model = None, None, Nonereturn kb_name, vs_type, embed_model@with_session
def delete_kb_from_db(session, kb_name):kb = session.query(KnowledgeBaseModel).filter(KnowledgeBaseModel.kb_name.ilike(kb_name)).first()if kb:session.delete(kb)return True@with_session
def get_kb_detail(session, kb_name: str) -> dict:kb: KnowledgeBaseModel = session.query(KnowledgeBaseModel).filter(KnowledgeBaseModel.kb_name.ilike(kb_name)).first()if kb:return {"kb_name": kb.kb_name,"kb_info": kb.kb_info,"vs_type": kb.vs_type,"embed_model": kb.embed_model,"file_count": kb.file_count,"create_time": kb.create_time,}else:return {}# ========================
# 测试入口
# ========================if __name__ == "__main__":test_kb_name = "test_kb"test_kb_info = "这是一个用于测试的知识库"test_vs_type = "FAISS"test_embed_model = "text-embedding-ada-002"print("🧪 开始测试知识库数据库操作函数...")# 1. 添加知识库print(f"📌 正在添加知识库:{test_kb_name}")add_kb_to_db(kb_name=test_kb_name,kb_info=test_kb_info,vs_type=test_vs_type,embed_model=test_embed_model)print("✅ 添加完成")# 2. 检查是否存在exists = kb_exists(kb_name=test_kb_name)print(f"🔍 知识库 {test_kb_name} 是否存在?{'是' if exists else '否'}")assert exists is True, "❌ 添加知识库失败"# 3. 获取详细信息print(f"📝 获取知识库 {test_kb_name} 的详细信息")detail = get_kb_detail(test_kb_name)print("Detail:", detail)# 4. 加载基本信息kb_name, vs_type, embed_model = load_kb_from_db(test_kb_name)print(f"🧠 加载知识库信息:{kb_name}, {vs_type}, {embed_model}")assert kb_name == test_kb_name, "❌ 加载知识库名称错误"assert vs_type == test_vs_type, "❌ 向量库类型不一致"assert embed_model == test_embed_model, "❌ 嵌入模型不一致"# 5. 列出所有知识库print("📋 当前数据库中的知识库列表:")all_kbs = list_kbs_from_db(min_file_count=-1)print(all_kbs)assert test_kb_name in all_kbs, "❌ 列表中未找到刚添加的知识库"# 6. 删除知识库print(f"🗑️ 正在删除知识库:{test_kb_name}")delete_kb_from_db(test_kb_name)print("✅ 删除完成")# 7. 再次检查是否存在exists_after_delete = kb_exists(kb_name=test_kb_name)print(f"🔍 删除后,知识库 {test_kb_name} 是否还存在?{'是' if exists_after_delete else '否'}")assert exists_after_delete is False, "❌ 删除失败"print("🎉 所有测试通过!")
http://www.dtcms.com/wzjs/185931.html

相关文章:

  • 怎么做移动网站抖音竞价推广怎么做
  • 网站开发名词解释广告网
  • 广东圆心科技网站开发需要多少钱打开百度搜索引擎
  • 如何做竞价网站数据监控公司网站优化
  • 网络科技公司名字起名大全seo在线教学
  • 国外最大的设计网站淘宝怎么推广自己的产品
  • 聊城集团网站建设价格网站宣传方法
  • 下做图软件在哪个网站下载器什么是信息流广告
  • 番禺人才网最新招聘信息四川最好的网络优化公司
  • 西安建设网站公司国产免费crm系统有哪些
  • 网站的广告语应该怎么做软文营销常用的方式
  • 网站制作租用空间百度广告登录入口
  • 平面设计创意网站建设高端网站公司
  • 郑州app网站开发成都十大营销策划公司
  • 天河网站建设公司商城网站建设
  • 中国网通做网站抖音搜索seo
  • 一级a做爰免费网站成都搜狗seo
  • 做网站有必要微博推广怎么做
  • 网页设计实训报告总结与收获南昌seo排名优化
  • 赤裸做爰游戏漫画网站四川百度推广排名查询
  • 济南网站设计公司推荐网络推广是做什么工作的
  • 长沙做网站nn微联讯点很好石家庄限号
  • 广州网站建设怎么样下载优化大师app
  • 湖北住房城乡建设厅网站首页百度站长提交网址
  • 比较出名的wordpress网站武汉网络推广优化
  • 昆山建设局图审中心网站沧州网站建设推广
  • 咸宁网站建设哪家好成功的网络营销案例有哪些
  • 做网站视频存储在哪里鞍山seo优化
  • 冠县网站建设全媒体运营师报名费多少钱
  • 网站生成app要多少钱跨境电商seo