从零搭建轻量图文联动 AI 工程助手|多模态 RAG 知识库实战落地

python==3.10numpy==1.26.4pandas==2.2.2PyPDF2==3.0.1python-docx==1.1.2markdown==3.6langchain==0.2.10langchain-community==0.2.9faiss-cpu==1.8.0rank_bm25==0.2.21paddlepaddle==2.6.1paddleocr==2.7.3transformers==4.41.2torch==2.3.0+cpupillow==10.3.0opencv-python==4.9.0.80openai==1.30.1tqdm==4.66.4

pip install -r requirements.txt

visual_rag_demo/├── data/              # 存放所有图文学习素材├── preprocess/        # 文本+图片预处理模块│   ├── __init__.py│   ├── text_processor.py   # 多格式文本解析清洗│   └── image_processor.py  # OpenCV降噪+PaddleOCR解析├── text_chunk/        # 双层分块核心逻辑（原名chunk，避坑重命名）│   ├── __init__.py│   └── chunk_manager.py    # 父子块分片实现├── embedding/         # 向量模型加载│   ├── __init__.py│   └── embedding_manager.py├── vector_store/      # FAISS本地向量库│   ├── __init__.py│   └── faiss_store.py├── retrieve/          # 混合检索实现│   ├── __init__.py│   └── retriever.py├── qa/                # LLM问答生成│   ├── __init__.py│   └── qa_engine.py├── main.py            # 项目主入口├── test_demo.py       # 效果对照测试└── requirements.txt   # 依赖清单

import osimport refrom PyPDF2 import PdfReaderfrom docx import Documentclass TextProcessor:    def __init__(self):        self.supported_extensions = [".txt", ".md", ".pdf", ".docx"]    def load_file(self, file_path):        """统一读取不同格式文档内容"""        ext = os.path.splitext(file_path)[1].lower()        if ext not in self.supported_extensions:            raise ValueError(f"不支持的文件格式: {ext}")        if ext == ".txt":            return self._read_txt(file_path)        elif ext == ".md":            return self._read_md(file_path)        elif ext == ".pdf":            return self._read_pdf(file_path)        elif ext == ".docx":            return self._read_docx(file_path)    def _read_txt(self, file_path):        with open(file_path, "r", encoding="utf-8", errors="ignore") as f:            return f.read()    def _read_md(self, file_path):        """读取Markdown文件，保留标题结构和代码块"""        with open(file_path, "r", encoding="utf-8", errors="ignore") as f:            return f.read()    def _read_pdf(self, file_path):        """读取PDF文件，提取纯文本"""        reader = PdfReader(file_path)        text = ""        for page in reader.pages:            page_text = page.extract_text()            if page_text:                text += page_text + "\n"        return text    def _read_docx(self, file_path):        """读取DOCX文件"""        doc = Document(file_path)        text = ""        for para in doc.paragraphs:            text += para.text + "\n"        return text    def clean_text(self, text):        """文本清洗：过滤空行、乱码、冗余符号，保留技术文档有效信息"""        if not text:            return ""        lines = text.split("\n")        cleaned_lines = []        for line in lines:            line = line.strip()            if not line:                continue            if len(line) < 2 and not line.strip():                continue            cleaned_lines.append(line)        cleaned_text = "\n".join(cleaned_lines)        cleaned_text = re.sub(r"\n{3,}", "\n\n", cleaned_text)        cleaned_text = re.sub(r"[ \t]+", " ", cleaned_text)        return cleaned_text.strip()    def process_file(self, file_path):        """完整处理流程：读取→清洗"""        raw_text = self.load_file(file_path)        cleaned_text = self.clean_text(raw_text)        return cleaned_text

from langchain_core.documents import Documentclass RecursiveTextSplitter:    def __init__(self, chunk_size=256, chunk_overlap=32, separators=None):        self.chunk_size = chunk_size        self.chunk_overlap = chunk_overlap        self.separators = separators or ["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""]    def _split_text(self, text, separators):        if not separators:            chunks = [text[i:i + self.chunk_size] for i in range(0, len(text), self.chunk_size - self.chunk_overlap)]            return chunks        separator = separators[0]        if separator == "":            chunks = [text[i:i + self.chunk_size] for i in range(0, len(text), self.chunk_size - self.chunk_overlap)]            return chunks        parts = text.split(separator)        chunks = []        current_chunk = parts[0] if parts else ""        for part in parts[1:]:            combined = current_chunk + separator + part            if len(combined) <= self.chunk_size:                current_chunk = combined            else:                if current_chunk:                    chunks.append(current_chunk)                if len(part) > self.chunk_size:                    sub_chunks = self._split_text(part, separators[1:])                    chunks.extend(sub_chunks)                    current_chunk = ""                else:                    current_chunk = part        if current_chunk:            chunks.append(current_chunk)        return chunks    def split_text(self, text):        return self._split_text(text, self.separators)class ChunkManager:    def __init__(self, child_chunk_size=256, child_chunk_overlap=32,                  parent_chunk_size=768, parent_chunk_overlap=64):        self.child_splitter = RecursiveTextSplitter(            chunk_size=child_chunk_size,            chunk_overlap=child_chunk_overlap        )        self.parent_splitter = RecursiveTextSplitter(            chunk_size=parent_chunk_size,            chunk_overlap=parent_chunk_overlap        )    def split_text(self, text, source="", source_type="text"):        """双层分块：先生成父块，再对每个父块生成子块"""        parent_chunks = self.parent_splitter.split_text(text)        all_chunks = []        parent_id = 0        for parent_text in parent_chunks:            child_chunks = self.child_splitter.split_text(parent_text)            parent_doc = Document(                page_content=parent_text,                metadata={                    "chunk_type": "parent",                    "parent_id": None,                    "child_ids": [],                    "source": source,                    "source_type": source_type,                    "chunk_id": f"parent_{parent_id}"                }            )            all_chunks.append(parent_doc)            child_ids = []            for child_idx, child_text in enumerate(child_chunks):                child_id = f"child_{parent_id}_{child_idx}"                child_ids.append(child_id)                child_doc = Document(                    page_content=child_text,                    metadata={                        "chunk_type": "child",                        "parent_id": f"parent_{parent_id}",                        "source": source,                        "source_type": source_type,                        "chunk_id": child_id                    }                )                all_chunks.append(child_doc)            parent_doc.metadata["child_ids"] = child_ids            parent_id += 1        return all_chunks

import cv2import numpy as npclass ImageProcessor:    def __init__(self):        self.ocr = PaddleOCR(use_angle_cls=True, lang="ch", use_gpu=False)    def preprocess_image(self, image_path):        """OpenCV图像预处理：灰度转换、高斯模糊降噪、二值化"""        img = cv2.imread(image_path)        if img is None:            raise ValueError(f"无法读取图片: {image_path}")        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)        blurred = cv2.GaussianBlur(gray, (3, 3), 0)        _, thresh = cv2.threshold(blurred, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)        return thresh

    def extract_text_from_image(self, image_path, use_preprocess=True):        """使用PaddleOCR提取图片中的文字内容"""        if use_preprocess:            processed_img = self.preprocess_image(image_path)            result = self.ocr.ocr(processed_img, cls=True)        else:            result = self.ocr.ocr(image_path, cls=True)        text_lines = []        if result and result[0]:            for line in result[0]:                if line[1] and len(line[1]) > 0:                    text_lines.append(line[1][0])        return "\n".join(text_lines)    def analyze_code_screenshot(self, image_path):        """代码截图专项解析：预处理+OCR提取，过滤低置信度结果"""        processed_img = self.preprocess_image(image_path)        result = self.ocr.ocr(processed_img, cls=True)        code_lines = []        if result and result[0]:            for line in result[0]:                if line[1] and len(line[1]) > 0:                    confidence = line[1][1]                    if confidence > 0.6:                        code_lines.append(line[1][0])        code_text = "\n".join(code_lines)        return {"type": "code", "content": code_text, "source": image_path}    def analyze_flowchart(self, image_path):        """架构流程图解析：提取文字并生成结构化描述"""        text = self.extract_text_from_image(image_path, use_preprocess=True)        structured_desc = f"【架构流程图描述】\n"        structured_desc += f"图片来源: {os.path.basename(image_path)}\n"        structured_desc += f"图中文字内容:\n{text}\n"        structured_desc += "\n【解析说明】\n"        structured_desc += "该图片为技术架构图或流程图，包含上述文字描述的模块和流程信息。"        return {"type": "flowchart", "content": structured_desc, "source": image_path}    def process_image(self, image_path):        """统一处理图片：自动识别类型并解析"""        text = self.extract_text_from_image(image_path, use_preprocess=True)        code_keywords = ["def ", "class ", "import ", "from ", "=", ";", "{", "}", "//", "#"]        has_code = any(keyword in text for keyword in code_keywords)        if has_code:            return self.analyze_code_screenshot(image_path)        else:            return self.analyze_flowchart(image_path)

import numpy as npfrom rank_bm25 import BM25Okapiclass HybridRetriever:    def __init__(self, vector_store, embedding_manager, bm25_weight=0.4, vector_weight=0.6):        self.vector_store = vector_store        self.embedding_manager = embedding_manager        self.bm25_weight = bm25_weight        self.vector_weight = vector_weight        self.bm25 = None        self.corpus = []        self._build_bm25_index()    def _tokenize_chinese(self, text):        """中文文本分词：字符级分词，保留英文单词和数字"""        tokens = []        i = 0        while i < len(text):            if text[i].isascii():                j = i                while j < len(text) and text[j].isascii():                    j += 1                tokens.append(text[i:j])                i = j            else:                tokens.append(text[i])                i += 1        return tokens    def _build_bm25_index(self):        """构建BM25索引"""        documents = self.vector_store.get_all_documents()        self.corpus = [doc.page_content for doc in documents]        if self.corpus:            tokenized_corpus = [self._tokenize_chinese(doc) for doc in self.corpus]            self.bm25 = BM25Okapi(tokenized_corpus)    def _normalize_scores(self, results, score_key="score"):        """归一化分数到[0,1]"""        if not results:            return results        scores = [r[score_key] for r in results]        max_score = max(scores)        min_score = min(scores)        if max_score == min_score:            for r in results:                r[score_key + "_normalized"] = 0.5        else:            for r in results:                r[score_key + "_normalized"] = (r[score_key] - min_score) / (max_score - min_score)        return results    def hybrid_search(self, query, k=10):        """混合检索：BM25 + 向量检索加权融合"""        bm25_results = self._bm25_search(query, k=k)        vector_results = self._vector_search(query, k=k)        bm25_results = self._normalize_scores(bm25_results, "score")        vector_results = self._normalize_scores(vector_results, "score")        merged_scores = {}        for r in bm25_results:            doc_id = id(r["document"])            merged_scores[doc_id] = {                "document": r["document"],                "bm25_score": r["score_normalized"],                "vector_score": 0.0            }        for r in vector_results:            doc_id = id(r["document"])            if doc_id in merged_scores:                merged_scores[doc_id]["vector_score"] = r["score_normalized"]            else:                merged_scores[doc_id] = {                    "document": r["document"],                    "bm25_score": 0.0,                    "vector_score": r["score_normalized"]                }        final_results = []        for doc_id, scores in merged_scores.items():            final_score = (scores["bm25_score"] * self.bm25_weight +                          scores["vector_score"] * self.vector_weight)            final_results.append({                "document": scores["document"],                "bm25_score": scores["bm25_score"],                "vector_score": scores["vector_score"],                "final_score": final_score            })        final_results.sort(key=lambda x: x["final_score"], reverse=True)        final_results = self._expand_to_parent(final_results)        return final_results[:k]

# 构建知识库（无需API Key）python main.py --build --data data# 交互式问答（需要API Key）python main.py --interactive --api-key "your-api-key" --base-url "https://qianfan.baidubce.com/v2/"