构建基于ERNIE与Milvus的多文档高精度分析与问答系统

# backend.py (OnlinePDFParser 类核心逻辑摘要)
def predict(self, file_path):
    # 1. 文件转 Base64
    with open(file_path, "rb") as file:
        file_data = base64.b64encode(file.read()).decode("ascii")

    # 2. 构造请求 Payload
    payload = {
        "file": file_data,
        "fileType": 1, # PDF 类型
        "useChartRecognition": False, # 根据需求配置
        "useDocOrientationClassify": False
    }

    # 3. 发送请求获取 Layout Parsing 结果
    response = requests.post(self.api_url, json=payload, headers=headers)
    res_json = response.json()

    # 4. 提取 Markdown 文本与图片
    parsing_results = res_json.get("result", {}).get("layoutParsingResults", [])
    mock_outputs = []
    for item in parsing_results:
        md_text = item.get("markdown", {}).get("text", "")
        images = item.get("markdown", {}).get("images", {})
        # ... (后续图片下载与文本清洗逻辑)
        mock_outputs.append(MockResult(md_text, images))

    return mock_outputs, "Success"

# backend.py
def split_text_into_chunks(text: str, chunk_size: int = 300, overlap: int = 120) -> list:
    """基于滑动窗口的文本分块，保留 overlap 长度的重叠上下文"""
    if not text: return []
    lines = [line.strip() for line in text.split("\n") if line.strip()]
    chunks = []
    current_chunk = []
    current_length = 0

    for line in lines:
        while len(line) > chunk_size:
            # 处理超长单行
            part = line[:chunk_size]
            line = line[chunk_size:]
            current_chunk.append(part)
            # ... (切分逻辑) ...

        current_chunk.append(line)
        current_length += len(line)

        # 当累积长度超过阈值，生成一个 chunk
        if current_length > chunk_size:
            chunks.append("\n".join(current_chunk))
            # 回退：保留最后 overlap 长度的文本作为下一个 chunk 的开头
            overlap_text = current_chunk[-1][-overlap:] if current_chunk else ""
            current_chunk = [overlap_text] if overlap_text else []
            current_length = len(overlap_text)

    if current_chunk:
        chunks.append("\n".join(current_chunk).strip())
    return chunks

import binascii
import re

def encode_name(ui_name):
    """把中文名称转为 Milvus 合法的 Hex 字符串"""
    if not ui_name: return ""
    # 如果是纯英文/数字/下划线，直接返回
    if re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', ui_name):
        return ui_name

    # Hex 编码并加前缀 kb_
    hex_str = binascii.hexlify(ui_name.encode('utf-8')).decode('utf-8')
    return f"kb_{hex_str}"

def decode_name(real_name):
    """把 Hex 字符串转回中文"""
    if real_name.startswith("kb_"):
        try:
            hex_str = real_name[3:]
            return binascii.unhexlify(hex_str).decode('utf-8')
        except:
            return real_name
    return real_name

# vector_store.py
def insert_documents(self, documents):
    """批量向量化并写入 Milvus"""
    if not documents: return

    # 1. 提取纯文本列表，批量请求 Embedding 模型
    texts = [doc['content'] for doc in documents]
    embeddings = self.get_embeddings(texts)

    # 2. 数据清洗：过滤掉 Embedding 失败的坏数据
    valid_docs, valid_vectors = [], []
    for i, emb in enumerate(embeddings):
        if emb and len(emb) == 384: # 确保向量维度正确
            valid_docs.append(documents[i])
            valid_vectors.append(emb)

    # 3. 组装列式数据 (Columnar Format)
    # Milvus insert 接口要求各字段数据以列表形式传入
    data = [
        [doc['filename'] for doc in valid_docs],  # Scalar: 文件名
        [doc['page'] for doc in valid_docs],      # Scalar: 页码 (用于溯源)
        [doc['chunk_id'] for doc in valid_docs],  # Scalar: 切片ID
        [doc['content'] for doc in valid_docs],   # Scalar: 原始内容 (用于关键词检索)
        valid_vectors                             # Vector: 语义向量
    ]

    # 4. 执行插入与持久化
    self.collection.insert(data)
    self.collection.flush()

# vector_store.py 中的检索逻辑摘要

def search(self, query: str, top_k: int = 10, **kwargs):
   '''向量检索(Dense+Keyword)+RRF 融合'''
   # 1. 向量检索 (Dense)
   dense_results = []
   query_vector = self.embedding_client.get_embedding(query) # ... (Milvus search code) ...

    # 2. 关键词检索 (Keyword)
    # 通过 jieba 分词后构建 like "%keyword%" 查询
    keyword_results = self._keyword_search(query, top_k=top_k * 5, expr=expr)

    # 3. RRF 融合
    rank_dict = {}

    def apply_rrf(results_list, k=60, weight=1.0):
        for rank, item in enumerate(results_list):
            doc_id = item.get('id') or item.get('chunk_id')
            if doc_id not in rank_dict:
                rank_dict[doc_id] = {"data": item, "score": 0.0}
            # RRF 核心公式
            rank_dict[doc_id]["score"] += weight * (1.0 / (k + rank))

    apply_rrf(dense_results, weight=4.0)
    apply_rrf(keyword_results, weight=1.0)

    # 4. 排序输出
    sorted_docs = sorted(rank_dict.values(), key=lambda x: x['score'], reverse=True)
    return [item['data'] for item in sorted_docs[:top_k * 2]]

# reranker_v2.py

def _calculate_composite_score(self, query: str, chunk: Dict[str, Any]) -> float:
    content = chunk.get('content', '')

    # 1. 字面重合度 (FuzzyWuzzy)
    fuzzy_score = fuzz.partial_ratio(query, content)

    # 2. 关键词覆盖率
    query_keywords = self._extract_keywords(query)
    content_keywords = self._extract_keywords(content)
    keyword_coverage = (len(query_keywords & content_keywords) / len(query_keywords)) * 100 if query_keywords else 0

    # 3. 向量语义分 (归一化)
    milvus_distance = chunk.get('semantic_score', 0)
    milvus_similarity = 100 / (1 + milvus_distance * 0.1)

    # 4. 长度惩罚 (偏好 200-600 字的段落)
    content_len = len(content)
    if 200 <= content_len <= 600:
        length_score = 100
    else:
        # ... (惩罚逻辑)
        length_score = 100 - min(50, abs(content_len - 400) / 20)

    # 加权求和
    base_score = (
        fuzzy_score * 0.25 +
        keyword_coverage * 0.25 +
        milvus_similarity * 0.35 +
        length_score * 0.15
    )

    # 位置权重
    position_bonus = 0
    if 'milvus_rank' in chunk:
        rank = chunk['milvus_rank']
        position_bonus = max(0, 20 - rank)

    # 专有名词额外加分 (Bonus)
    proper_noun_bonus = 30 if self._check_proper_nouns(query, content) else 0

    return base_score + proper_noun_bonus

# main.py 配置 LaTeX 规则
latex_config = [
    {"left": "$$", "right": "$$", "display": True},   # 识别行间公式
    {"left": "$", "right": "$", "display": False},    # 识别行内公式
    {"left": "\\(", "right": "\\)", "display": False}, # 标准 LaTeX 行内
    {"left": "\\[", "right": "\\]", "display": True}   # 标准 LaTeX 行间
]

# 在 Chatbot 中启用 LaTeX
chatbot = gr.Chatbot(
    label="Conversation",
    # ... 其他参数 ...
    latex_delimiters=latex_config  # 关键配置：启用公式渲染
)

# 在摘要显示区启用 LaTeX
doc_summary = gr.Markdown(
    value="*暂无摘要*",
    latex_delimiters=latex_config
)

# backend.py - 构建参考来源列表
sources = "\n\n📚 **参考来源:**\n"
for c in final:
    # ... (去重逻辑) ...
    # 直接 Reranker 计算出的单片得分
    sources += f"- {key} [相关性:{c.get('composite_score',0):.0f}%]\n"

# backend.py - 计算整体置信度
# 1. 获取重排序后的 Top-1 片段
final = processed[:22]
top_score = final[0].get('composite_score', 0) if final else 0

# 2. 归一化处理 (封顶 100%)，作为本次问答的"置信度"
metric = f"{min(100, top_score):.1f}%"