From 496e11e26e28ba3798ec115dd3f7d55827f418e7 Mon Sep 17 00:00:00 2001 From: EduBrain Dev Date: Tue, 14 Apr 2026 14:50:53 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E7=A7=BB=E9=99=A4=20PostgreSQL=20?= =?UTF-8?q?=E6=94=AF=E6=8C=81=EF=BC=8C=E7=AE=80=E5=8C=96=E4=B8=BA=E7=BA=AF?= =?UTF-8?q?=20SQLite=20=E9=83=A8=E7=BD=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - config.py: DATABASE_URL 默认值改为 SQLite - database.py: 移除 PostgreSQL 分支，简化为纯 SQLite - models/base.py: 移除 pgvector 导入和条件分支 - search_service.py: 移除 _search_postgres 方法 - import_service.py: 移除 pgvector 相关代码 - requirements.txt: 移除 asyncpg/alembic/pgvector 依赖 - pyproject.toml: 同步移除相关依赖 - docker-compose.yml: 移除 db 服务，- 删除 alembic.ini/Dockerfile.db/sql 目录 - README.md: 更新文档，移除 PostgreSQL 相关内容 适合 NAS 等资源受限环境的轻量级部署 --- README.md | 24 ++-- alembic.ini | 63 ----------- alembic/env.py | 67 ----------- alembic/script.py.mako | 26 ----- alembic/versions/.gitkeep | 0 app/config.py | 4 +- app/database.py | 81 ++++---------- app/models/base.py | 145 +++++++++++------------- app/services/import_service.py | 31 ++---- app/services/search_service.py | 196 +-------------------------------- docker-compose.yml | 36 +----- pyproject.toml | 4 +- requirements.txt | 3 - sql/init.sql | 129 ---------------------- 14 files changed, 112 insertions(+), 697 deletions(-) delete mode 100644 alembic.ini delete mode 100644 alembic/env.py delete mode 100644 alembic/script.py.mako delete mode 100644 alembic/versions/.gitkeep delete mode 100644 sql/init.sql diff --git a/README.md b/README.md index ebed8ab..d777dd6 100644 --- a/README.md +++ b/README.md @@ -19,10 +19,10 @@ - **BM25 索引**: 内置倒排索引搜索引擎，jieba 中文分词 ### AI 搜索 -- **知识库搜索**: 向量语义搜索 + 全文检索混合排序 +- **知识库搜索**: 关键词搜索 - **图片搜索**: MiniMax M2.7 语义匹配，SSE 流式实时返回结果 - **并发池**: 最多 10 个 LLM 请求并发，每批数量可配置 -- **可配置参数**: 搜索返回数量、LLM 批量判断数量均可在网页端设置 +- **可配置参数**: 搜索返回数量、LLM 批量判断数量、OCR 并发数量均可在网页端设置 ### 企业微信机器人 - **WebSocket 长连接**: 基于官方 SDK，无需公网 IP @@ -33,7 +33,7 @@ ### 其他 - **MCP 协议**: 支持 Model Context Protocol，可被 AI Agent 调用 -- **Docker 部署**: 完整的 Docker Compose 编排（应用 + PostgreSQL + pgvector） +- **Docker 部署**: 单容器部署，轻量级 ## 技术栈 @@ -41,13 +41,13 @@ |------|------| | 后端框架 | FastAPI + Uvicorn | | ORM | SQLAlchemy 2.0 (async) | -| 数据库 | PostgreSQL (pgvector) / SQLite | +| 数据库 | SQLite | | 前端 | 纯 HTML + CSS + JS（苹果风格 SPA） | | OCR | DeepSeek Vision / PaddleOCR / 阿里云 / 腾讯云 | | LLM | MiniMax M2.7 / DeepSeek / Qwen3-8B | -| 搜索 | BM25 倒排索引 + 向量语义搜索 | +| 搜索 | BM25 倒排索引 + 关键词搜索 | | 企业微信 | wecom-aibot-python-sdk (WebSocket 长连接) | -| 部署 | Docker + Docker Compose | +| 部署 | Docker | ## 项目结构 @@ -56,12 +56,12 @@ edu-brain/ ├── app/ │ ├── main.py # FastAPI 入口，路由注册，生命周期管理，BotManager │ ├── config.py # pydantic-settings 配置管理 -│ ├── database.py # 数据库连接（PG/SQLite 双后端） +│ ├── database.py # SQLite 数据库连接 │ ├── models/base.py # ORM 模型（KnowledgePage, OCRImage 等） │ ├── schemas/ # Pydantic 请求/响应 Schema │ ├── api/v1/ # API 路由 │ │ ├── pages.py # 知识页面 CRUD -│ │ ├── search.py # 语义搜索 +│ │ ├── search.py # 关键词搜索 │ │ ├── images.py # 图片 OCR + AI 搜索（SSE）+ 去重 + 删除 │ │ ├── import_export.py # 文件导入导出 │ │ └── settings.py # 系统设置 + 机器人管理 @@ -69,7 +69,7 @@ edu-brain/ │ │ ├── ocr_service.py # OCR 识别服务（多 Provider） │ │ ├── llm_service.py # LLM 服务（标签提取/搜索匹配） │ │ ├── search_engine.py # BM25 倒排索引引擎 -│ │ ├── search_service.py # 混合搜索服务 +│ │ ├── search_service.py # 搜索服务 │ │ ├── embedding_service.py # 嵌入模型服务 │ │ ├── import_service.py # 文件导入服务 │ │ └── page_service.py # 页面 CRUD 服务 @@ -80,7 +80,6 @@ edu-brain/ ├── .env.example # 环境变量模板 ├── requirements.txt # Python 依赖 ├── Dockerfile # 应用镜像 -├── Dockerfile.db # 数据库镜像（PG + pgvector） └── docker-compose.yml # Docker 编排 ``` @@ -192,7 +191,7 @@ docker-compose up -d | POST | `/api/v1/images/dedup` | 一键去重 | | GET | `/api/v1/pages` | 知识页面列表 | | POST | `/api/v1/pages` | 创建知识页面 | -| POST | `/api/v1/search` | 语义搜索知识库 | +| POST | `/api/v1/search` | 关键词搜索知识库 | | POST | `/api/v1/import/file` | 导入文件 | | GET | `/api/v1/settings` | 获取系统设置 | | PUT | `/api/v1/settings` | 更新系统设置 | @@ -216,7 +215,7 @@ docker-compose up -d | Provider | 说明 | 额外配置 | |----------|------|----------| | `minimax` | MiniMax 嵌入 | `MINIMAX_API_KEY` | -| `openai` | OpenAI 嵌入 | `OPENAI_API_KEY` | +| `openai` | OpenAI 嵌入（兼容硅基流动等第三方接口） | `OPENAI_API_KEY` | | `zhipu` | 智谱 AI 嵌入 | `ZHIPU_API_KEY` | | `dashscope` | 阿里云 DashScope | `DASHSCOPE_API_KEY` | | `local_bge` | 本地 BGE 模型 | `LOCAL_BGE_MODEL_PATH` | @@ -227,3 +226,4 @@ docker-compose up -d |------|------|--------|------| | 搜索返回数量 | 每次搜索最多返回多少条匹配结果 | 3 | 1-10 | | LLM 批量判断数量 | 每批发给 LLM 判断的图片数量 | 10 | 1-50 | +| OCR 并发数量 | 多用户同时上传时 OCR 并发识别数量 | 1 | 1-10 | diff --git a/alembic.ini b/alembic.ini deleted file mode 100644 index 5163e7c..0000000 --- a/alembic.ini +++ /dev/null @@ -1,63 +0,0 @@ -# Alembic 数据库迁移配置 - -[alembic] -# 迁移脚本目录 -script_location = alembic - -# 数据库连接 URL（通过环境变量覆盖） -# 注意：Alembic 使用同步驱动，需要将 asyncpg 替换为 psycopg2 -sqlalchemy.url = postgresql://postgres:postgres@localhost:5432/edu_brain - -# 模板文件 -file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s - -# 是否自动应用迁移 -# prepend_sys_path = . - -# 时区设置 -timezone = Asia/Shanghai - -# 截断长迁移消息 -truncate_slug_length = 40 - -# 修订 ID 格式 -# revision_environment = false - -# 输出编码 -# output_encoding = utf-8 - -[post_write_hooks] - -[loggers] -keys = root,sqlalchemy,alembic - -[handlers] -keys = console - -[formatters] -keys = generic - -[logger_root] -level = WARN -handlers = console -qualname = - -[logger_sqlalchemy] -level = WARN -handlers = -qualname = sqlalchemy.engine - -[logger_alembic] -level = INFO -handlers = -qualname = alembic - -[handler_console] -class = StreamHandler -args = (sys.stderr,) -level = NOTSET -formatter = generic - -[formatter_generic] -format = %(levelname)-5.5s [%(name)s] %(message)s -datefmt = %H:%M:%S diff --git a/alembic/env.py b/alembic/env.py deleted file mode 100644 index 1bccfc2..0000000 --- a/alembic/env.py +++ /dev/null @@ -1,67 +0,0 @@ -""" -Alembic 迁移环境配置 -支持异步数据库连接 -""" - -from logging.config import fileConfig - -from alembic import context -from sqlalchemy import engine_from_config, pool - -# 导入所有 ORM 模型以确保 Base.metadata 包含完整的表定义 -from app.models.base import Base - -# Alembic Config 对象 -config = context.config - -# 设置日志 -if config.config_file_name is not None: - fileConfig(config.config_file_name) - -# 元数据目标，用于自动生成迁移 -target_metadata = Base.metadata - - -def run_migrations_offline() -> None: - """ - 以 'offline' 模式运行迁移。 - 只需要 URL，不需要 Engine。调用 context.execute() 将迁移 - 直接发送到数据库。 - """ - url = config.get_main_option("sqlalchemy.url") - context.configure( - url=url, - target_metadata=target_metadata, - literal_binds=True, - dialect_opts={"paramstyle": "named"}, - ) - - with context.begin_transaction(): - context.run_migrations() - - -def run_migrations_online() -> None: - """ - 以 'online' 模式运行迁移。 - 创建 Engine 并关联 connection 到 context。 - """ - connectable = engine_from_config( - config.get_section(config.config_ini_section, {}), - prefix="sqlalchemy.", - poolclass=pool.NullPool, - ) - - with connectable.connect() as connection: - context.configure( - connection=connection, - target_metadata=target_metadata, - ) - - with context.begin_transaction(): - context.run_migrations() - - -if context.is_offline_mode(): - run_migrations_offline() -else: - run_migrations_online() diff --git a/alembic/script.py.mako b/alembic/script.py.mako deleted file mode 100644 index fbc4b07..0000000 --- a/alembic/script.py.mako +++ /dev/null @@ -1,26 +0,0 @@ -"""${message} - -Revision ID: ${up_revision} -Revises: ${down_revision | comma,n} -Create Date: ${create_date} - -""" -from typing import Sequence, Union - -from alembic import op -import sqlalchemy as sa -${imports if imports else ""} - -# revision identifiers, used by Alembic. -revision: str = ${repr(up_revision)} -down_revision: Union[str, None] = ${repr(down_revision)} -branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)} -depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)} - - -def upgrade() -> None: - ${upgrades if upgrades else "pass"} - - -def downgrade() -> None: - ${downgrades if downgrades else "pass"} diff --git a/alembic/versions/.gitkeep b/alembic/versions/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/app/config.py b/app/config.py index e4375f5..c372add 100644 --- a/app/config.py +++ b/app/config.py @@ -46,8 +46,8 @@ class Settings(BaseSettings): # ──────────────────────────── 数据库 ──────────────────────────── DATABASE_URL: str = Field( - default="postgresql+asyncpg://postgres:postgres@db:5432/edu_brain", - description="PostgreSQL 异步连接字符串", + default="sqlite+aiosqlite:///./edu_brain.db", + description="SQLite 数据库连接字符串", ) # ──────────────────────────── 嵌入模型 ──────────────────────────── diff --git a/app/database.py b/app/database.py index 0586d02..d9d3cd3 100644 --- a/app/database.py +++ b/app/database.py @@ -1,12 +1,13 @@ """ -数据库连接模块 -支持 PostgreSQL（生产）和 SQLite（本地开发）两种后端 +数据库连接管理 + +SQLite 数据库，支持异步操作。 """ from __future__ import annotations -from collections.abc import AsyncGenerator from pathlib import Path +from typing import AsyncGenerator from sqlalchemy import text from sqlalchemy.ext.asyncio import ( @@ -14,35 +15,25 @@ from sqlalchemy.ext.asyncio import ( async_sessionmaker, create_async_engine, ) -from sqlalchemy.pool import NullPool from app.config import settings -# ──────────────────────────── 判断数据库类型 ──────────────────────────── +# ──────────────────────────── 数据库引擎 ──────────────────────────── DB_URL = settings.DATABASE_URL -IS_SQLITE = DB_URL.startswith("sqlite") -# ──────────────────────────── 异步引擎 ──────────────────────────── +# 解析 SQLite 文件路径 +sqlite_path = DB_URL.replace("sqlite+aiosqlite:///", "") +if not sqlite_path.startswith("/"): + sqlite_path = str(Path(__file__).parent.parent / sqlite_path) -if IS_SQLITE: - # SQLite 配置 - sqlite_path = DB_URL.replace("sqlite+aiosqlite:///", "") - if not sqlite_path.startswith("/"): - sqlite_path = str(Path(__file__).parent.parent / sqlite_path) +# 确保数据目录存在 +Path(sqlite_path).parent.mkdir(parents=True, exist_ok=True) - engine = create_async_engine( - f"sqlite+aiosqlite:///{sqlite_path}", - echo=False, - ) -else: - # PostgreSQL 配置 - engine = create_async_engine( - DB_URL, - echo=False, - poolclass=NullPool, - pool_pre_ping=True, - ) +engine = create_async_engine( + f"sqlite+aiosqlite:///{sqlite_path}", + echo=False, +) # ──────────────────────────── 会话工厂 ──────────────────────────── @@ -61,47 +52,21 @@ async def get_db() -> AsyncGenerator[AsyncSession, None]: FastAPI 依赖项：获取异步数据库会话。 """ async with async_session_factory() as session: - try: - yield session - await session.commit() - except Exception: - await session.rollback() - raise + yield session -# ──────────────────────────── 生命周期辅助 ──────────────────────────── +async def close_db() -> None: + """关闭数据库连接池""" + await engine.dispose() + async def init_db() -> None: """初始化数据库：建表""" from app.models.base import Base async with engine.begin() as conn: - if IS_SQLITE: - # SQLite 启用 WAL 模式和外键约束 - await conn.execute(text("PRAGMA journal_mode=WAL")) - await conn.execute(text("PRAGMA foreign_keys=ON")) + # SQLite 启用 WAL 模式和外键约束 + await conn.execute(text("PRAGMA journal_mode=WAL")) + await conn.execute(text("PRAGMA foreign_keys=ON")) # 创建所有表 await conn.run_sync(Base.metadata.create_all) - - if not IS_SQLITE: - # PostgreSQL 需要创建扩展 - async with engine.begin() as conn: - try: - await conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector")) - except Exception: - pass - try: - await conn.execute(text("CREATE EXTENSION IF NOT EXISTS zhparser")) - except Exception: - pass - try: - await conn.execute(text( - "CREATE TEXT SEARCH CONFIGURATION IF NOT EXISTS chinese_zh (PARSER = zhparser)" - )) - except Exception: - pass - - -async def close_db() -> None: - """关闭数据库引擎""" - await engine.dispose() diff --git a/app/models/base.py b/app/models/base.py index 7c0cfad..e7b750a 100644 --- a/app/models/base.py +++ b/app/models/base.py @@ -1,26 +1,18 @@ """ -SQLAlchemy 基础模型 + 通用 Mixin -提供 DeclarativeBase、时间戳 Mixin 和公共序列化方法 -兼容 PostgreSQL（pgvector）和 SQLite（本地开发） +ORM 模型定义 + +SQLite 数据库模型，使用 Text 存储 JSON 格式的向量。 """ from __future__ import annotations +import json from datetime import datetime from typing import Any, Dict from sqlalchemy import DateTime, Float, Integer, String, Text, func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column -from app.database import IS_SQLITE - -# SQLite 下不使用 pgvector，用 Text 存储向量（JSON 格式） -if IS_SQLITE: - VectorType = Text -else: - from pgvector.sqlalchemy import Vector - VectorType = Vector - class Base(DeclarativeBase): """所有 ORM 模型的基类""" @@ -28,96 +20,83 @@ class Base(DeclarativeBase): class TimestampMixin: - """ - 时间戳 Mixin，为模型自动添加 created_at / updated_at 字段。 - """ + """时间戳 Mixin""" + created_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), - server_default=func.now(), - comment="创建时间", + DateTime, default=func.now(), comment="创建时间" ) updated_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), - server_default=func.now(), - onupdate=func.now(), - comment="更新时间", + DateTime, default=func.now(), onupdate=func.now(), comment="更新时间" ) - def to_dict(self) -> Dict[str, Any]: - """将模型实例转换为字典""" - result: Dict[str, Any] = {} - for column in self.__table__.columns: # type: ignore[attr-defined] - value = getattr(self, column.name, None) - if isinstance(value, datetime): - value = value.isoformat() - elif hasattr(value, "tolist"): - value = value.tolist() - result[column.name] = value - return result - - -# ──────────────────────────── 业务模型 ──────────────────────────── class KnowledgePage(Base, TimestampMixin): - """知识页面模型""" + """知识页面""" __tablename__ = "knowledge_pages" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) title: Mapped[str] = mapped_column(String(500), nullable=False, comment="页面标题") - content: Mapped[str] = mapped_column(Text, nullable=False, comment="页面正文内容") - source_file: Mapped[str | None] = mapped_column(String(500), nullable=True, comment="来源文件名") - course_name: Mapped[str | None] = mapped_column(String(200), nullable=True, comment="课程名称") - teacher_name: Mapped[str | None] = mapped_column(String(100), nullable=True, comment="讲师名称") - live_date: Mapped[str | None] = mapped_column(String(20), nullable=True, comment="直播日期") - page_number: Mapped[int | None] = mapped_column(Integer, nullable=True, comment="原始页码") - metadata_json: Mapped[str | None] = mapped_column(Text, nullable=True, comment="额外元数据") + content: Mapped[str] = mapped_column(Text, nullable=False, comment="页面内容") + source: Mapped[str | None] = mapped_column(String(200), nullable=True, comment="来源文件名") + page_number: Mapped[int | None] = mapped_column(Integer, nullable=True, comment="原文件页码") + # 向量嵌入（JSON 格式存储） + embedding: Mapped[str | None] = mapped_column(Text, nullable=True, comment="向量嵌入（JSON）") + @property + def embedding_vector(self) -> list[float] | None: + if self.embedding is None: + return None + return json.loads(self.embedding) -class KnowledgeChunk(Base, TimestampMixin): - """知识分块模型""" - __tablename__ = "knowledge_chunks" - - id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) - page_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True, comment="关联页面 ID") - chunk_index: Mapped[int] = mapped_column(Integer, nullable=False, comment="分块序号") - content: Mapped[str] = mapped_column(Text, nullable=False, comment="分块文本内容") - embedding: Mapped[list | None] = mapped_column( - VectorType(768) if not IS_SQLITE else Text, - nullable=True, - comment="嵌入向量", - ) - search_vector: Mapped[Any | None] = mapped_column( - Text, nullable=True, comment="全文搜索向量（仅 PG）", - ) - - def to_dict(self) -> Dict[str, Any]: - result = super().to_dict() - result.pop("embedding", None) - result.pop("search_vector", None) - return result + @embedding_vector.setter + def embedding_vector(self, value: list[float] | None): + self.embedding = json.dumps(value) if value else None class OCRImage(Base, TimestampMixin): - """OCR 图片记录模型""" + """OCR 图片记录""" __tablename__ = "ocr_images" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) file_path: Mapped[str] = mapped_column(String(500), nullable=False, comment="图片文件路径") - ocr_text: Mapped[str | None] = mapped_column(Text, nullable=True, comment="OCR 识别全文") - confidence: Mapped[float | None] = mapped_column(Float, nullable=True, comment="平均置信度") + original_filename: Mapped[str | None] = mapped_column(String(200), nullable=True, comment="原始文件名") + ocr_text: Mapped[str | None] = mapped_column(Text, nullable=True, comment="OCR 识别文本") + confidence: Mapped[float | None] = mapped_column(Float, nullable=True, comment="OCR 置信度") provider: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="OCR 提供商") - status: Mapped[str] = mapped_column(String(20), default="pending", comment="处理状态") - error_message: Mapped[str | None] = mapped_column(Text, nullable=True, comment="错误信息") - tags: Mapped[str | None] = mapped_column(Text, nullable=True, comment="LLM 提取的标签，JSON 数组格式") - story_summary: Mapped[str | None] = mapped_column(Text, nullable=True, comment="Qwen3-8B 提炼的故事摘要") + status: Mapped[str] = mapped_column( + String(20), default="pending", comment="状态: pending/processing/completed/failed" + ) + tags: Mapped[str | None] = mapped_column(Text, nullable=True, comment="关键词标签（JSON 数组）") + story_summary: Mapped[str | None] = mapped_column(Text, nullable=True, comment="故事摘要") + blocks: Mapped[str | None] = mapped_column(Text, nullable=True, comment="OCR 文本块（JSON）") - def to_dict(self) -> Dict[str, Any]: - result = super().to_dict() - # 将 tags 从 JSON 字符串解析为列表返回 - if result.get("tags") and isinstance(result["tags"], str): - try: - import json - result["tags"] = json.loads(result["tags"]) - except (json.JSONDecodeError, TypeError): - pass - return result + @property + def tags_list(self) -> list[str]: + if self.tags is None: + return [] + return json.loads(self.tags) + + @tags_list.setter + def tags_list(self, value: list[str]): + self.tags = json.dumps(value) if value else None + + @property + def blocks_list(self) -> list[dict]: + if self.blocks is None: + return [] + return json.loads(self.blocks) + + @blocks_list.setter + def blocks_list(self, value: list[dict]): + self.blocks = json.dumps(value) if value else None + + +class WebsiteSettings(Base): + """网站设置（单例）""" + __tablename__ = "website_settings" + + key: Mapped[str] = mapped_column(String(100), primary_key=True, comment="设置键") + value: Mapped[str | None] = mapped_column(Text, nullable=True, comment="设置值（JSON）") + updated_at: Mapped[datetime] = mapped_column( + DateTime, default=func.now(), onupdate=func.now(), comment="更新时间" + ) diff --git a/app/services/import_service.py b/app/services/import_service.py index 0938cb7..c23b227 100644 --- a/app/services/import_service.py +++ b/app/services/import_service.py @@ -14,7 +14,6 @@ from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings -from app.database import IS_SQLITE logger = logging.getLogger(__name__) @@ -343,21 +342,13 @@ class ImportService: # 批量插入 for idx, (chunk_text, embedding) in enumerate(zip(chunks, embeddings)): - if IS_SQLITE: - # SQLite: 向量存为 JSON 文本 - import json as _json - embedding_str = _json.dumps(embedding) if embedding else None - sql = text(""" - INSERT INTO knowledge_chunks (page_id, chunk_index, content, embedding) - VALUES (:page_id, :chunk_index, :content, :embedding) - """) - else: - # PostgreSQL: 使用 pgvector 类型 - embedding_str = "[" + ",".join(str(x) for x in embedding) + "]" if embedding else None - sql = text(""" - INSERT INTO knowledge_chunks (page_id, chunk_index, content, embedding) - VALUES (:page_id, :chunk_index, :content, :embedding::vector) - """) + # SQLite: 向量存为 JSON 文本 + import json as _json + embedding_str = _json.dumps(embedding) if embedding else None + sql = text(""" + INSERT INTO knowledge_chunks (page_id, chunk_index, content, embedding) + VALUES (:page_id, :chunk_index, :content, :embedding) + """) await self.db.execute(sql, { "page_id": page_id, "chunk_index": idx, @@ -365,14 +356,6 @@ class ImportService: "embedding": embedding_str, }) - # 更新全文搜索向量（仅 PostgreSQL） - if not IS_SQLITE: - await self.db.execute(text(""" - UPDATE knowledge_chunks - SET search_vector = to_tsvector('chinese_zh', content) - WHERE page_id = :page_id - """), {"page_id": page_id}) - await self.db.flush() return len(chunks) diff --git a/app/services/search_service.py b/app/services/search_service.py index 7b59b10..9337488 100644 --- a/app/services/search_service.py +++ b/app/services/search_service.py @@ -1,20 +1,18 @@ """ 语义搜索服务 -支持 PostgreSQL（向量+全文混合）和 SQLite（LIKE 关键词搜索） + +SQLite 数据库，使用 LIKE 关键词搜索。 """ from __future__ import annotations -import json import logging import time -from typing import List, Optional +from typing import List from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession -from app.config import settings -from app.database import IS_SQLITE from app.schemas.search import SearchRequest, SearchResult, SearchResponse logger = logging.getLogger(__name__) @@ -28,21 +26,10 @@ class SearchService: async def search(self, request: SearchRequest) -> SearchResponse: """ - 执行搜索。 - - PostgreSQL: 向量搜索 + 中文全文搜索混合排序 - - SQLite: LIKE 关键词搜索（无向量能力） + 执行搜索：使用 LIKE 关键词搜索 """ start_time = time.time() - if IS_SQLITE: - return await self._search_sqlite(request, start_time) - else: - return await self._search_postgres(request, start_time) - - # ──────────────────────────── SQLite 搜索 ──────────────────────────── - - async def _search_sqlite(self, request: SearchRequest, start_time: float) -> SearchResponse: - """SQLite 模式：使用 LIKE 关键词搜索""" # LLM 查询扩展（可选） expanded_queries = [request.query] try: @@ -109,178 +96,3 @@ class SearchService: results=results, elapsed_ms=round(elapsed_ms, 2), ) - - # ──────────────────────────── PostgreSQL 搜索 ──────────────────────────── - - async def _search_postgres(self, request: SearchRequest, start_time: float) -> SearchResponse: - """PostgreSQL 模式：向量搜索 + 中文全文搜索混合""" - # LLM 查询扩展 - expanded_queries = [request.query] - if request.use_fulltext: - try: - from app.services.llm_service import LLMService - expanded_queries = await LLMService.expand_query(request.query) - except Exception as exc: - logger.debug("查询扩展跳过: %s", exc) - - # 生成查询向量 - from app.services.embedding_service import EmbeddingService - all_query_embeddings = await EmbeddingService.embed_batch(expanded_queries) - - # 构建过滤条件 - where_clauses = [] - params: dict = {} - - if request.course_name: - where_clauses.append("kp.course_name = :course_name") - params["course_name"] = request.course_name - if request.teacher_name: - where_clauses.append("kp.teacher_name = :teacher_name") - params["teacher_name"] = request.teacher_name - if request.live_date_from: - where_clauses.append("kp.live_date >= :live_date_from") - params["live_date_from"] = request.live_date_from - if request.live_date_to: - where_clauses.append("kp.live_date <= :live_date_to") - params["live_date_to"] = request.live_date_to - - where_sql = "" - if where_clauses: - where_sql = "AND " + " AND ".join(where_clauses) - - # 向量搜索 - vector_results = [] - params["limit"] = request.top_k * 2 - for q_embedding in all_query_embeddings: - embedding_str = "[" + ",".join(str(x) for x in q_embedding) + "]" - vector_sql = f""" - SELECT - kc.id AS chunk_id, - kp.id AS page_id, - kp.title AS page_title, - kc.content, - 1 - (kc.embedding <=> :embedding::vector) AS vector_score, - 0.0 AS text_score, - kp.course_name, - kp.teacher_name, - kp.live_date - FROM knowledge_chunks kc - JOIN knowledge_pages kp ON kc.page_id = kp.id - WHERE kc.embedding IS NOT NULL - {where_sql} - ORDER BY kc.embedding <=> :embedding::vector - LIMIT :limit - """ - v_params = {**params, "embedding": embedding_str} - result = await self.db.execute(text(vector_sql), v_params) - vector_results.extend(result.fetchall()) - - # 全文搜索 - text_results = [] - if request.use_fulltext: - for q in expanded_queries: - clean_query = q.replace("'", "''") - fulltext_sql = f""" - SELECT - kc.id AS chunk_id, - kp.id AS page_id, - kp.title AS page_title, - kc.content, - 0.0 AS vector_score, - ts_rank_cd(kc.search_vector, to_tsquery('chinese_zh', :query)) AS text_score, - kp.course_name, - kp.teacher_name, - kp.live_date, - ts_headline('chinese_zh', kc.content, to_tsquery('chinese_zh', :query), - 'MaxWords=50, MinWords=20, ShortWord=2') AS highlight - FROM knowledge_chunks kc - JOIN knowledge_pages kp ON kc.page_id = kp.id - WHERE kc.search_vector @@ to_tsquery('chinese_zh', :query) - {where_sql} - ORDER BY text_score DESC - LIMIT :limit - """ - ft_params = {**params, "query": clean_query} - ft_result = await self.db.execute(text(fulltext_sql), ft_params) - text_results.extend(ft_result.fetchall()) - - # 合并结果 - merged: dict[int, dict] = {} - - for row in vector_results: - chunk_id = row.chunk_id - if chunk_id not in merged: - merged[chunk_id] = { - "chunk_id": chunk_id, - "page_id": row.page_id, - "page_title": row.page_title, - "content": row.content, - "vector_score": row.vector_score, - "text_score": 0.0, - "course_name": row.course_name, - "teacher_name": row.teacher_name, - "live_date": row.live_date, - "highlight": None, - } - else: - merged[chunk_id]["vector_score"] = max( - merged[chunk_id]["vector_score"], row.vector_score - ) - - for row in text_results: - chunk_id = row.chunk_id - if chunk_id not in merged: - merged[chunk_id] = { - "chunk_id": chunk_id, - "page_id": row.page_id, - "page_title": row.page_title, - "content": row.content, - "vector_score": 0.0, - "text_score": row.text_score or 0.0, - "course_name": row.course_name, - "teacher_name": row.teacher_name, - "live_date": row.live_date, - "highlight": row.highlight, - } - else: - merged[chunk_id]["text_score"] = max( - merged[chunk_id]["text_score"], row.text_score or 0.0 - ) - if row.highlight: - merged[chunk_id]["highlight"] = row.highlight - - # 计算综合得分 - max_text_score = max( - (r["text_score"] for r in merged.values()), default=1.0 - ) or 1.0 - - scored_results = [] - for item in merged.values(): - normalized_text = item["text_score"] / max_text_score if max_text_score > 0 else 0 - combined_score = 0.7 * item["vector_score"] + 0.3 * normalized_text - - if combined_score >= request.threshold: - scored_results.append( - SearchResult( - chunk_id=item["chunk_id"], - page_id=item["page_id"], - page_title=item["page_title"], - content=item["content"], - score=round(combined_score, 4), - course_name=item["course_name"], - teacher_name=item["teacher_name"], - live_date=item["live_date"], - highlight=item["highlight"], - ) - ) - - scored_results.sort(key=lambda x: x.score, reverse=True) - scored_results = scored_results[:request.top_k] - - elapsed_ms = (time.time() - start_time) * 1000 - return SearchResponse( - query=request.query, - total=len(scored_results), - results=scored_results, - elapsed_ms=round(elapsed_ms, 2), - ) diff --git a/docker-compose.yml b/docker-compose.yml index 5d1a29c..1a14397 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,7 +1,6 @@ version: "3.8" services: - # ── 应用服务 ── app: build: context: . @@ -11,45 +10,12 @@ services: ports: - "${APP_PORT:-8000}:8000" volumes: - - ./data:/app/data # 数据目录挂载 + - ./data:/app/data env_file: - .env - depends_on: - db: - condition: service_healthy - environment: - - DATABASE_URL=postgresql+asyncpg://postgres:${POSTGRES_PASSWORD:-postgres}@db:5432/edu_brain healthcheck: test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"] interval: 30s timeout: 10s retries: 3 start_period: 60s - - # ── PostgreSQL 数据库（pgvector + zhparser 自定义镜像） ── - db: - build: - context: . - dockerfile: Dockerfile.db - container_name: edu-brain-db - restart: unless-stopped - ports: - - "${DB_PORT:-5432}:5432" - volumes: - - postgres_data:/var/lib/postgresql/data - - ./sql/init.sql:/docker-entrypoint-initdb.d/01-init.sql:ro - environment: - POSTGRES_USER: postgres - POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-postgres} - POSTGRES_DB: edu_brain - POSTGRES_HOST_AUTH_METHOD: scram-sha-256 - healthcheck: - test: ["CMD-SHELL", "pg_isready -U postgres -d edu_brain"] - interval: 10s - timeout: 5s - retries: 5 - start_period: 30s - -volumes: - postgres_data: - driver: local diff --git a/pyproject.toml b/pyproject.toml index 6f805a2..d7ac88b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,9 +24,7 @@ dependencies = [ "uvicorn[standard]>=0.32.0", "pydantic-settings>=2.6.0", "sqlalchemy[asyncio]>=2.0.36", - "asyncpg>=0.30.0", - "alembic>=1.14.0", - "pgvector>=0.3.6", + "aiosqlite>=0.20.0", "openai>=1.58.0", "zhipuai>=2.4.0", "dashscope>=1.20.0", diff --git a/requirements.txt b/requirements.txt index afa07c7..ee4d393 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,9 +7,6 @@ python-multipart>=0.0.18 # ── 数据库 ── sqlalchemy[asyncio]>=2.0.36,<3.0.0 aiosqlite>=0.20.0,<1.0.0 -asyncpg>=0.30.0,<1.0.0 -alembic>=1.14.0,<2.0.0 -pgvector>=0.3.6,<1.0.0 # ── 嵌入模型 SDK ── openai>=1.58.0,<2.0.0 diff --git a/sql/init.sql b/sql/init.sql deleted file mode 100644 index 53a5d89..0000000 --- a/sql/init.sql +++ /dev/null @@ -1,129 +0,0 @@ --- ============================================================ --- EduBrain 数据库初始化脚本 --- 包含 pgvector 扩展、zhparser 中文分词扩展 --- ============================================================ - --- 创建 pgvector 扩展（向量相似度搜索） -CREATE EXTENSION IF NOT EXISTS vector; - --- 创建 zhparser 中文分词扩展（全文搜索） -CREATE EXTENSION IF NOT EXISTS zhparser; - --- 创建基于 zhparser 的中文全文搜索配置 -CREATE TEXT SEARCH CONFIGURATION chinese_zh (PARSER = zhparser); - --- 配置中文全文搜索映射：名词、动词、形容词、成语、叹词、习语使用 simple 字典 -ALTER TEXT SEARCH CONFIGURATION chinese_zh ADD MAPPING FOR n,v,a,i,e,l WITH simple; - --- ============================================================ --- 知识页面表 --- ============================================================ -CREATE TABLE IF NOT EXISTS knowledge_pages ( - id SERIAL PRIMARY KEY, - title VARCHAR(500) NOT NULL, - content TEXT NOT NULL, - source_file VARCHAR(500), - course_name VARCHAR(200), - teacher_name VARCHAR(100), - live_date VARCHAR(20), - page_number INTEGER, - metadata_json TEXT, - created_at TIMESTAMPTZ DEFAULT NOW(), - updated_at TIMESTAMPTZ DEFAULT NOW() -); - --- 索引 -CREATE INDEX IF NOT EXISTS idx_pages_course ON knowledge_pages(course_name); -CREATE INDEX IF NOT EXISTS idx_pages_teacher ON knowledge_pages(teacher_name); -CREATE INDEX IF NOT EXISTS idx_pages_date ON knowledge_pages(live_date); -CREATE INDEX IF NOT EXISTS idx_pages_created ON knowledge_pages(created_at DESC); - --- ============================================================ --- 知识分块表（向量检索单元） --- ============================================================ -CREATE TABLE IF NOT EXISTS knowledge_chunks ( - id SERIAL PRIMARY KEY, - page_id INTEGER NOT NULL REFERENCES knowledge_pages(id) ON DELETE CASCADE, - chunk_index INTEGER NOT NULL, - content TEXT NOT NULL, - embedding vector(1024), -- 嵌入向量（默认 1024 维） - search_vector TSVECTOR, -- 中文全文搜索向量 - created_at TIMESTAMPTZ DEFAULT NOW(), - updated_at TIMESTAMPTZ DEFAULT NOW() -); - --- 索引 -CREATE INDEX IF NOT EXISTS idx_chunks_page ON knowledge_chunks(page_id); -CREATE INDEX IF NOT EXISTS idx_chunks_page_index ON knowledge_chunks(page_id, chunk_index); - --- HNSW 向量索引（适合高维向量，查询速度快） -CREATE INDEX IF NOT EXISTS idx_chunks_embedding - ON knowledge_chunks - USING hnsw (embedding vector_cosine_ops) - WITH (m = 16, ef_construction = 200); - --- GIN 全文搜索索引 -CREATE INDEX IF NOT EXISTS idx_chunks_search - ON knowledge_chunks - USING gin(search_vector); - --- ============================================================ --- OCR 图片记录表 --- ============================================================ -CREATE TABLE IF NOT EXISTS ocr_images ( - id SERIAL PRIMARY KEY, - file_path VARCHAR(500) NOT NULL, - ocr_text TEXT, - confidence FLOAT, - provider VARCHAR(50), - status VARCHAR(20) DEFAULT 'pending', - error_message TEXT, - created_at TIMESTAMPTZ DEFAULT NOW(), - updated_at TIMESTAMPTZ DEFAULT NOW() -); - --- 索引 -CREATE INDEX IF NOT EXISTS idx_ocr_status ON ocr_images(status); -CREATE INDEX IF NOT EXISTS idx_ocr_created ON ocr_images(created_at DESC); - --- ============================================================ --- updated_at 自动更新触发器 --- ============================================================ -CREATE OR REPLACE FUNCTION update_updated_at_column() -RETURNS TRIGGER AS $$ -BEGIN - NEW.updated_at = NOW(); - RETURN NEW; -END; -$$ LANGUAGE plpgsql; - --- 为所有表添加 updated_at 触发器 -DO $$ -BEGIN - IF NOT EXISTS ( - SELECT 1 FROM pg_trigger - WHERE tgname = 'tr_pages_updated_at' - ) THEN - CREATE TRIGGER tr_pages_updated_at - BEFORE UPDATE ON knowledge_pages - FOR EACH ROW EXECUTE FUNCTION update_updated_at_column(); - END IF; - - IF NOT EXISTS ( - SELECT 1 FROM pg_trigger - WHERE tgname = 'tr_chunks_updated_at' - ) THEN - CREATE TRIGGER tr_chunks_updated_at - BEFORE UPDATE ON knowledge_chunks - FOR EACH ROW EXECUTE FUNCTION update_updated_at_column(); - END IF; - - IF NOT EXISTS ( - SELECT 1 FROM pg_trigger - WHERE tgname = 'tr_ocr_updated_at' - ) THEN - CREATE TRIGGER tr_ocr_updated_at - BEFORE UPDATE ON ocr_images - FOR EACH ROW EXECUTE FUNCTION update_updated_at_column(); - END IF; -END $$;