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

fix docker issue

Browse Source
This commit is contained in:
EduBrain Dev
2026-04-14 17:25:06 +08:00
parent 4c6a20e5fc
commit c5f96056fa
9 changed files with 170 additions and 537 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
app/api/v1/images.py
View File

@@ -1,6 +1,6 @@
"""
图片 OCR API 路由
OCR 识别 + BM25 倒排索引搜索
OCR 识别 + AI 搜索
"""
from __future__ import annotations
@@ -24,7 +24,6 @@ from app.config import settings
from app.database import get_db
from app.models.base import OCRImage
from app.services.ocr_service import OCRService
from app.services.search_engine import BM25Index, tokenize
from app.services.llm_service import LLMService
logger = logging.getLogger(__name__)
@@ -45,9 +44,6 @@ def _get_ocr_semaphore() -> asyncio.Semaphore:
ALLOWED_EXTENSIONS = {".png", ".jpg", ".jpeg", ".bmp", ".webp"}
# 全局 BM25 索引实例（在 main.py 中初始化）
bm25_index: Optional[BM25Index] = None
def _check_image_ext(filename: str) -> str:
suffix = os.path.splitext(filename)[1].lower()
@@ -62,41 +58,8 @@ def _save_ocr_result(record: OCRImage, ocr_result, db: AsyncSession):
record.confidence = ocr_result.confidence
record.provider = ocr_result.provider
record.status = "completed"
# 保留 keywords（如果有）
if hasattr(ocr_result, 'keywords') and ocr_result.keywords:
record.tags = json.dumps(ocr_result.keywords, ensure_ascii=False)
def _add_to_index(record: OCRImage):
"""将 OCR 结果添加到 BM25 索引"""
if bm25_index is None or not record.ocr_text:
return
# 索引内容 = OCR 文本（权重最高）
index_text = record.ocr_text
# 如果有 tags/story_summary 也加入索引
if record.tags:
try:
tags = json.loads(record.tags)
if tags:
index_text += "\n" + " ".join(tags)
except Exception:
pass
if record.story_summary:
index_text += "\n" + record.story_summary
bm25_index.add_document(
doc_id=record.id,
text=index_text,
metadata={
"file_path": record.file_path,
"tags": json.loads(record.tags) if record.tags else [],
"story_summary": record.story_summary or "",
"confidence": record.confidence,
"provider": record.provider,
"created_at": str(record.created_at),
},
)
async def _check_ocr_duplicate(ocr_text: str, db: AsyncSession) -> Optional[int]:
"""检查 OCR 文本是否已存在，返回已存在记录的 ID 或 None"""
@@ -144,13 +107,11 @@ async def recognize_image(image_id: int, db: AsyncSession = Depends(get_db)):
_save_ocr_result(image_record, ocr_result, db)
await db.flush()
await db.refresh(image_record)
_add_to_index(image_record)
return {
"id": image_record.id,
"file_path": image_record.file_path,
"ocr_text": image_record.ocr_text,
"tags": json.loads(image_record.tags) if image_record.tags else [],
"confidence": image_record.confidence,
"provider": image_record.provider,
"status": image_record.status,
@@ -208,12 +169,10 @@ async def batch_recognize(files: List[UploadFile], db: AsyncSession = Depends(ge
_save_ocr_result(image_record, ocr_result, db)
await db.flush()
await db.refresh(image_record)
_add_to_index(image_record)
results.append({
"id": image_record.id, "filename": file.filename,
"file_path": str(dest_path), "status": "completed",
"ocr_text": ocr_result.text,
"tags": json.loads(image_record.tags) if image_record.tags else [],
"confidence": ocr_result.confidence, "provider": ocr_result.provider,
"block_count": len(ocr_result.blocks),
})
@@ -266,8 +225,7 @@ async def import_from_paths(data: PathImportRequest, db: AsyncSession = Depends(
_save_ocr_result(image_record, ocr_result, db)
await db.flush()
await db.refresh(image_record)
_add_to_index(image_record)
results.append({"id": image_record.id, "filename": os.path.basename(file_path), "file_path": file_path, "status": "completed", "ocr_text": ocr_result.text, "tags": json.loads(image_record.tags) if image_record.tags else [], "confidence": ocr_result.confidence, "provider": ocr_result.provider})
results.append({"id": image_record.id, "filename": os.path.basename(file_path), "file_path": file_path, "status": "completed", "ocr_text": ocr_result.text, "confidence": ocr_result.confidence, "provider": ocr_result.provider})
except Exception as e:
image_record.status = "failed"
image_record.error_message = str(e)
@@ -315,12 +273,10 @@ async def recognize_direct(file: UploadFile, db: AsyncSession = Depends(get_db))
_save_ocr_result(image_record, ocr_result, db)
await db.flush()
await db.refresh(image_record)
_add_to_index(image_record)
return {
"id": image_record.id, "file_path": image_record.file_path,
"original_filename": file.filename,
"ocr_text": image_record.ocr_text,
"tags": json.loads(image_record.tags) if image_record.tags else [],
"confidence": image_record.confidence, "provider": image_record.provider,
"status": image_record.status,
"blocks": [{"text": b.text, "bbox": b.bbox, "confidence": b.confidence} for b in ocr_result.blocks],
@@ -420,7 +376,6 @@ async def search_images(
"image_base64": image_base64,
"ocr_text": item.ocr_text,
"ocr_text_preview": (item.ocr_text[:150] + "...") if item.ocr_text and len(item.ocr_text) > 150 else item.ocr_text,
"tags": json.loads(item.tags) if item.tags else [],
"confidence": item.confidence,
"provider": item.provider,
"created_at": str(item.created_at),
@@ -489,9 +444,6 @@ async def delete_image(image_id: int, db: AsyncSession = Depends(get_db)):
image_record = result.scalar_one_or_none()
if image_record is None:
raise HTTPException(status_code=404, detail="图片记录不存在")
# 从 BM25 索引移除
if bm25_index is not None:
bm25_index.remove_document(image_record.id)
# 删除磁盘上的图片文件
try:
os.unlink(image_record.file_path)
@@ -561,9 +513,6 @@ async def dedup_images(db: AsyncSession = Depends(get_db)):
duplicates_found += len(group) - 1
kept += 1
for record in group[1:]:
# 从 BM25 索引移除
if bm25_index is not None:
bm25_index.remove_document(record.id)
# 删除磁盘上的图片文件
try:
os.unlink(record.file_path)

49
app/main.py
View File

@@ -107,7 +107,7 @@ bot_manager = BotManager()
async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
"""
应用生命周期管理：
- 启动时：初始化数据目录、验证数据库连接、预热嵌入服务、初始化 BM25 索引
- 启动时：初始化数据目录、验证数据库连接、预热嵌入服务
- 关闭时：释放数据库连接
"""
_logger = logging.getLogger(__name__)
@@ -126,53 +126,6 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
except Exception as exc:
_logger.warning("嵌入服务预热失败（将在首次使用时重试）: %s", exc)
# ── 初始化 BM25 索引 ──
try:
from app.services.search_engine import BM25Index
import app.api.v1.images as images_module
_images_bm25 = BM25Index()
# 加载数据库中已有的 OCR 记录到索引
from app.database import async_session_factory
from app.models.base import OCRImage
from sqlalchemy import select
async with async_session_factory() as session:
result = await session.execute(
select(OCRImage).where(OCRImage.status == "completed")
)
records = result.scalars().all()
for record in records:
if record.ocr_text:
index_text = record.ocr_text
if record.tags:
try:
tags = json.loads(record.tags)
if tags:
index_text += "\n" + " ".join(tags)
except Exception:
pass
if record.story_summary:
index_text += "\n" + record.story_summary
_images_bm25.add_document(
doc_id=record.id,
text=index_text,
metadata={
"file_path": record.file_path,
"tags": json.loads(record.tags) if record.tags else [],
"story_summary": record.story_summary or "",
"confidence": record.confidence,
"provider": record.provider,
"created_at": str(record.created_at),
},
)
images_module.bm25_index = _images_bm25
_logger.info("BM25 索引加载完成: %d 条记录", _images_bm25.size)
except Exception as exc:
_logger.warning("BM25 索引初始化失败: %s", exc)
yield
# ── 关闭 ──

12
app/models/base.py
View File

@@ -66,20 +66,8 @@ class OCRImage(Base, TimestampMixin):
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）")
@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:

152
app/services/llm_service.py
View File

@@ -236,158 +236,6 @@ class LLMService:
return []
@classmethod
async def extract_tags(cls, text: str) -> List[str]:
"""
从文本中提取语义标签。
用于图片 OCR 内容的标签化和搜索查询的标签化。
Args:
text: 待提取标签的文本
Returns:
标签列表，如 ["教育", "高考数学", "二次方程", "解题技巧"]
"""
messages = [
{
"role": "system",
"content": (
"你是一个内容标签提取助手。用户会给你一段文字内容，"
"你需要从中提取 3-10 个能概括内容主题的标签。\n"
"标签要求：\n"
"1. 用简短的词语（2-6个字）\n"
"2. 涵盖主题、场景、情感、关键实体等维度\n"
"3. 考虑同义词和近义词（如'不想工作'也打上'躺平'标签）\n"
"4. 只输出 JSON 数组，不要其他文字\n"
'例如：["高考数学", "二次方程", "韦达定理", "解题技巧"]'
),
},
{
"role": "user",
"content": f"请从以下内容中提取标签：\n\n{text[:2000]}",
},
]
try:
result = await cls.chat(messages, temperature=0.3, max_tokens=256)
result = result.strip()
if result.startswith("```"):
result = result.split("\n", 1)[-1]
if result.endswith("```"):
result = result[:-3]
result = result.strip()
tags = json.loads(result)
if isinstance(tags, list):
# 去重、清洗
seen = set()
clean = []
for t in tags:
t = str(t).strip()
if 1 <= len(t) <= 20 and t not in seen:
seen.add(t)
clean.append(t)
return clean[:15]
except Exception as exc:
logger.warning("标签提取失败: %s", exc)
return []
@classmethod
async def extract_search_tags(cls, query: str) -> List[str]:
"""
从用户搜索查询中提取标签，用于匹配数据库中存储的标签。
和 extract_tags 类似，但更侧重于提取搜索意图相关的标签。
"""
messages = [
{
"role": "system",
"content": (
"你是一个搜索意图分析助手。用户会给你一个搜索查询，"
"你需要提取 3-8 个可能匹配的标签词。\n"
"标签要求：\n"
"1. 用简短的词语（2-6个字）\n"
"2. 包含同义词和近义词\n"
"3. 包含上义词和下义词\n"
"4. 只输出 JSON 数组\n"
'例如：用户搜"孩子躺平"，输出 ["躺平", "不上班", "年轻人", "就业", "摆烂", "啃老"]'
),
},
{
"role": "user",
"content": f"请从以下搜索查询中提取匹配标签：{query}",
},
]
try:
result = await cls.chat(messages, temperature=0.3, max_tokens=256)
result = result.strip()
if result.startswith("```"):
result = result.split("\n", 1)[-1]
if result.endswith("```"):
result = result[:-3]
result = result.strip()
tags = json.loads(result)
if isinstance(tags, list):
seen = set()
clean = []
for t in tags:
t = str(t).strip()
if 1 <= len(t) <= 20 and t not in seen:
seen.add(t)
clean.append(t)
return clean[:10]
except Exception as exc:
logger.warning("搜索标签提取失败: %s", exc)
# 降级：直接用原始查询词作为标签
return [query.strip()] if query.strip() else []
@classmethod
async def summarize_story(cls, text: str) -> str:
"""
用 Qwen3-8B 提炼图片 OCR 文本的故事内容。
非思考模式，直接输出。
"""
messages = [
{
"role": "system",
"content": (
"你是一个内容提炼助手。用户会给你一段从图片中识别出的文字内容，"
"你需要提炼出其中讲述的故事或事件。\n"
"要求：\n"
"1. 用简洁的自然语言描述图片内容讲述的故事或事件\n"
"2. 保留关键人物、事件、情感、场景等核心信息\n"
"3. 50-200字\n"
"4. 直接输出提炼内容，不要加前缀"
),
},
{
"role": "user",
"content": f"/no_think\n请提炼以下内容的故事：\n\n{text[:3000]}",
},
]
try:
from openai import AsyncOpenAI
client = AsyncOpenAI(
api_key=settings.OPENAI_API_KEY,
base_url=settings.OPENAI_BASE_URL,
)
response = await client.chat.completions.create(
model="Qwen/Qwen3-8B",
messages=messages,
temperature=0.3,
max_tokens=512,
extra_body={"chat_template_kwargs": {"enable_thinking": False}},
)
content = response.choices[0].message.content or ""
# 清理可能的 /no_think 回显
content = content.replace("/no_think", "").strip()
return content
except Exception as exc:
logger.warning("故事提炼失败: %s", exc)
return ""
@classmethod
async def judge_batch(cls, query: str, articles: List[dict]) -> List[int]:
"""

185
app/services/search_engine.py
View File

@@ -1,185 +0,0 @@
"""
BM25 倒排索引搜索引擎
用于图片 OCR 内容的全文搜索
参考 Meilisearch / Whoosh 设计
"""
from __future__ import annotations
import re
import math
import logging
from collections import Counter, defaultdict
from typing import List, Dict, Optional, Tuple
logger = logging.getLogger(__name__)
# 尝试导入 jieba，失败则用简单的字符分割
try:
import jieba
jieba.setLogLevel(logging.WARNING)
def tokenize(text: str) -> List[str]:
"""中文分词"""
return [w for w in jieba.cut_for_search(text) if len(w.strip()) > 1]
except ImportError:
logger.warning("jieba 未安装，使用简单分词（pip install jieba 获得更好效果）")
def tokenize(text: str) -> List[str]:
"""简单分词：按空格和标点分割"""
return [w for w in re.split(r'[\s,.\-;:!?"\'/\\|()\[\]{}<>，。；：！？""''、（）【】]+', text) if len(w.strip()) > 1]
class BM25Index:
"""
BM25 倒排索引
- 支持增量添加文档
- 支持删除文档
- 支持中文分词（jieba）
- BM25 评分排序
"""
def __init__(self, k1: float = 1.5, b: float = 0.75):
"""
Args:
k1: 词频饱和参数（默认 1.5）
b: 文档长度归一化参数（默认 0.75）
"""
self.k1 = k1
self.b = b
# 倒排索引：term -> {doc_id: tf}
self.inverted_index: Dict[str, Dict[int, int]] = defaultdict(lambda: defaultdict(int))
# 文档信息
self.doc_lengths: Dict[int, int] = {} # doc_id -> 文档长度（词数）
self.doc_count: int = 0
self.avg_doc_length: float = 0.0
# 文档存储（存完整数据）
self.documents: Dict[int, dict] = {}
# IDF 缓存
self._idf_cache: Dict[str, float] = {}
def add_document(self, doc_id: int, text: str, metadata: Optional[dict] = None):
"""
添加文档到索引
Args:
doc_id: 文档 ID
text: 待索引的文本内容
metadata: 附加元数据（如 file_path, tags 等）
"""
# 如果已存在，先删除
if doc_id in self.documents:
self.remove_document(doc_id)
tokens = tokenize(text)
self.doc_lengths[doc_id] = len(tokens)
self.documents[doc_id] = {
"text": text,
"tokens": tokens,
"metadata": metadata or {},
}
# 构建倒排索引
tf = Counter(tokens)
for term, count in tf.items():
self.inverted_index[term][doc_id] = count
self.doc_count += 1
self._update_avg_doc_length()
self._idf_cache.clear() # 文档数变了，IDF 需要重算
def remove_document(self, doc_id: int):
"""从索引中删除文档"""
if doc_id not in self.documents:
return
# 从倒排索引中移除
tokens = self.documents[doc_id].get("tokens", [])
for term in set(tokens):
if term in self.inverted_index and doc_id in self.inverted_index[term]:
del self.inverted_index[term][doc_id]
if not self.inverted_index[term]:
del self.inverted_index[term]
del self.documents[doc_id]
del self.doc_lengths[doc_id]
self.doc_count -= 1
self._update_avg_doc_length()
self._idf_cache.clear()
def _update_avg_doc_length(self):
"""更新平均文档长度"""
if self.doc_count > 0:
self.avg_doc_length = sum(self.doc_lengths.values()) / self.doc_count
else:
self.avg_doc_length = 0.0
def _idf(self, term: str) -> float:
"""
计算逆文档频率 IDF
IDF(t) = ln((N - df(t) + 0.5) / (df(t) + 0.5) + 1)
其中 N 是文档总数，df(t) 是包含 term 的文档数
"""
if term in self._idf_cache:
return self._idf_cache[term]
df = len(self.inverted_index.get(term, {}))
idf = math.log((self.doc_count - df + 0.5) / (df + 0.5) + 1)
self._idf_cache[term] = idf
return idf
def search(self, query: str, top_k: int = 10) -> List[Tuple[int, float]]:
"""
BM25 搜索
Args:
query: 搜索查询（会自动分词）
top_k: 返回前 K 个结果
Returns:
[(doc_id, score), ...] 按 BM25 分数降序排列
"""
if self.doc_count == 0 or self.avg_doc_length == 0:
return []
query_tokens = tokenize(query)
if not query_tokens:
return []
scores: Dict[int, float] = defaultdict(float)
for term in query_tokens:
idf = self._idf(term)
postings = self.inverted_index.get(term, {})
for doc_id, tf in postings.items():
dl = self.doc_lengths[doc_id]
# BM25 公式
numerator = tf * (self.k1 + 1)
denominator = tf + self.k1 * (1 - self.b + self.b * dl / self.avg_doc_length)
scores[doc_id] += idf * numerator / denominator
# 按分数降序排序
ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return ranked[:top_k]
def get_document(self, doc_id: int) -> Optional[dict]:
"""获取文档完整数据"""
return self.documents.get(doc_id)
def clear(self):
"""清空索引"""
self.inverted_index.clear()
self.doc_lengths.clear()
self.documents.clear()
self._idf_cache.clear()
self.doc_count = 0
self.avg_doc_length = 0.0
@property
def size(self) -> int:
"""索引中的文档数量"""
return self.doc_count