GuangYu/HuiBrain
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 移除 PostgreSQL 支持，简化为纯 SQLite 部署

Browse Source 
- 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 等资源受限环境的轻量级部署
This commit is contained in:
EduBrain Dev
2026-04-14 14:50:53 +08:00
parent f60b45358d
commit 496e11e26e
14 changed files with 112 additions and 697 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
README.md
View File

@@ -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 |

63
alembic.ini
View File

@@ -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

67
alembic/env.py
View File

@@ -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()

26
alembic/script.py.mako
View File

@@ -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"}

0
alembic/versions/.gitkeep
View File

4
app/config.py
View File

@@ -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 数据库连接字符串",
)
# ──────────────────────────── 嵌入模型 ────────────────────────────

61
app/database.py
View File

@@ -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")
# ──────────────────────────── 异步引擎 ────────────────────────────
if IS_SQLITE:
# 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,
)
# ──────────────────────────── 会话工厂 ────────────────────────────
@@ -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
# ──────────────────────────── 生命周期辅助 ────────────────────────────
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"))
# 创建所有表
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()

145
app/models/base.py
View File

@@ -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="更新时间"
)

17
app/services/import_service.py
View File

@@ -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,7 +342,6 @@ 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
@@ -351,13 +349,6 @@ class ImportService:
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)
""")
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)

196
app/services/search_service.py
View File

@@ -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),
)

36
docker-compose.yml
View File

@@ -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

4
pyproject.toml
View File

@@ -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",

3
requirements.txt
View File

@@ -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

129
sql/init.sql
View File

@@ -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 $$;