LlamaIndex RAG知识库对话与OCR识别系统搭建详解

该文档围绕LlamaIndex框架，详细介绍了RAG（检索增强生成）知识库对话系统搭建与自定义OCR图片识别功能实现，同时包含Chainlit界面优化与FastAPI服务部署，以下从核心模块、代码解析、搭建流程三方面展开讲解。

一、核心模块概览

整个系统分为三大核心模块，各模块职责明确且相互协同，共同实现“文档识别-知识存储-智能问答”的完整流程：

模块名称	核心功能	关键技术/工具
Chainlit界面优化	美化UI、提供交互入口、支持会话管理	Chainlit框架、CSS自定义、会话状态管理
自定义OCR识别	图片/文档文字提取、批量处理、结果存储	Umi-OCR工具、HTTP接口调用、Base64编码
RAG知识库问答	向量存储、知识检索、智能对话生成	Milvus向量数据库、LlamaIndex、FastAPI

二、分模块详细解析与代码实操

（一）Chainlit界面优化：打造交互式对话入口

Chainlit是轻量级AI应用开发框架，本模块通过5个关键步骤优化界面，提升用户体验。

1.1 修改Logo图片：替换默认标识

• 核心目标：替换Chainlit默认Logo，增强品牌辨识度。
• 操作步骤：
  1. 在项目根目录创建public文件夹，放入3类图片文件：
     • favicon.png：浏览器标签页图标；
     • logo_dark.png：深色模式下的Logo；
     • logo_light.png：浅色模式下的Logo。
  2. 参考官方文档配置路径：Chainlit自定义Logo。

1.2 添加提示块：提供快速对话入口

• 核心目标：为用户提供预设对话主题，降低使用门槛。
• 关键代码（chainlit_ui.py）：

import chainlit as cl

@cl.set_starters
async def set_starters():
    """定义快速开始的对话主题，包含标题、详情、图标"""
    return [
        cl.Starter(
            label="大模型提高软件测试效率",  # 主题标题
            message="详细介绍如何借助大语言模型提高软件测试效率。",  # 对话详情
            icon="/public/apidog.svg"  # 图标路径（需放在public文件夹）
        ),
        cl.Starter(
            label="自动化测试思路",
            message="详细描述一下接口及UI自动化测试的基本思路。",
            icon="/public/pulumi.svg"
        ),
        # 可根据需求添加更多主题
    ]

# 注意：注释掉初始欢迎消息，避免与提示块重复
@cl.on_chat_start
async def start():
    # await cl.Message(author="Assistant", content="您好，我是AI智能助手...").send()  # 注释此行
    pass

• 效果：用户打开对话界面时，可直接点击预设主题，快速触发对应对话。

1.3 添加设置面板：支持模型与功能切换

• 核心目标：允许用户选择大模型、开启/关闭多模态RAG功能。
• 关键代码（chainlit_ui.py）：

from chainlit.input_widget import Select, Switch

@cl.on_chat_start
async def start():
    # 发送设置面板（模型选择+多模态开关）
    await cl.ChatSettings(
        [
            Select(
                id="Model",  # 配置ID，用于后续获取值
                label="模型选择",
                values=["DeepSeek", "Moonshot"],  # 支持的大模型列表
                initial_index=0  # 默认选中第一个模型
            ),
            Switch(
                id="multimodal",
                label="多模态RAG",
                initial=True  # 默认开启多模态（支持图片识别）
            )
        ]
    ).send()

# 监听设置更新，保存到会话
@cl.on_settings_update
async def setup_settings(settings):
    cl.user_session.set("settings", settings)  # 将用户选择的设置存入会话

• 效果：界面将显示“设置”按钮，点击可切换模型（如DeepSeek/Moonshot）或开启多模态RAG。

1.4 会话管理：支持“返回对话”与“恢复对话”

• 核心目标：用户刷新页面或重新进入时，可恢复历史对话记录。
• 关键代码（chainlit_ui.py）：

from chainlit.types import ThreadDict
from llama_index.core.base.llms.types import ChatMessage
from llama_index.core.chat_engine import SimpleChatEngine

@cl.on_chat_resume
async def on_chat_resume(thread: ThreadDict):
    """恢复历史对话：从线程信息中提取历史消息，重建聊天引擎"""
    # 初始化聊天引擎
    chat_engine = SimpleChatEngine.from_defaults()
    
    # 遍历历史消息，添加到聊天引擎的对话历史
    for message in thread.get("steps", []):
        if message["type"] == "user_message":
            # 用户消息：角色标记为"user"
            chat_engine.chat_history.append(
                ChatMessage(content=message["output"], role="user")
            )
        elif message["type"] == "assistant_message":
            # 助手消息：角色标记为"assistant"
            chat_engine.chat_history.append(
                ChatMessage(content=message["output"], role="assistant")
            )
    
    # 将重建的聊天引擎存入会话，供后续使用
    cl.user_session.set("chat_engine", chat_engine)

1.5 CSS自定义：隐藏冗余元素、美化界面

• 核心目标：隐藏默认的“README按钮”“Chat按钮”等冗余元素，优化界面简洁度。
• 操作步骤：
  1. 在public文件夹创建ui.css文件；
  2. 写入CSS样式（通过浏览器开发者工具获取元素类名）：

/* 隐藏README按钮 */
.css-8kxmdr {
    visibility: hidden !important;
}

/* 隐藏Chat按钮 */
.css-plyx71 {
    visibility: hidden !important;
}

/* 隐藏默认Logo（替换为自定义Logo时使用） */
.css-12hxhao {
    visibility: hidden !important;
}

/* 隐藏页脚 */
.css-1705j0v {
    visibility: hidden !important;
}

3. 在config.toml中配置CSS路径：

[UI]
custom_css = "/public/ui.css"  # 指向public文件夹中的ui.css

（二）自定义OCR识别：提取图片/文档中的文字

本模块基于Umi-OCR工具（免费、开源、离线），实现图片与文档的文字提取，核心是通过HTTP接口调用OCR服务，完成“上传-处理-下载-清理”全流程。

2.1 前置准备：部署Umi-OCR服务

1. 下载Umi-OCR：
   • 官网：Umi-OCR GitHub；
   • 安装包：Umi-OCR下载链接（支持Windows/Linux）。
2. 开启HTTP服务：
   • 打开Umi-OCR，进入「全局设置」→「高级」→「服务」；
   • 勾选“允许HTTP服务”，设置“主机为任何可用地址”，端口默认1314（后续配置需与该端口一致）。

2.2 核心配置：环境变量与参数定义

• 2.2.1 .env文件（环境变量配置）：
  在项目根目录创建.env，存储OCR服务地址与文件下载路径：

# OCR服务地址（与Umi-OCR设置的端口一致）
OCR_BASE_URL="http://127.0.0.1:1314"
# OCR识别结果下载路径（自动创建）
OCR_DOWNLOAD_PATH="./data/ocr_download"

• 2.2.2 rag/config.py（参数封装）：
  用Pydantic封装配置，便于全局调用：

import os
from pydantic import BaseModel, Field

class RAGConfig(BaseModel):
    # OCR基础URL（从.env读取）
    ocr_base_url: str = Field(default=os.getenv("OCR_BASE_URL"), description="OCR服务地址")
    # OCR下载目录（从.env读取）
    ocr_download_dir: str = Field(default=os.getenv("OCR_DOWNLOAD_PATH"), description="OCR结果存储路径")

# 单例模式：全局唯一配置实例
RagConfig = RAGConfig()

2.3 核心模块：创建ocr.py实现OCR逻辑

rag/ocr.py是OCR功能的核心，包含5个核心函数，实现“上传文件→处理OCR→生成下载链接→下载结果→清理任务”全流程：

import base64
import os
import json
import time
import zipfile
import requests
from .config import RagConfig  # 导入配置

# 初始化：从配置获取OCR服务地址、下载目录、请求头
base_url = RagConfig.ocr_base_url
download_dir = RagConfig.ocr_download_dir
headers = {"Content-Type": "application/json"}

def _upload_file(file_path):
    """1. 上传文件到OCR服务，处理非ASCII文件名问题"""
    url = f"{base_url}/api/doc/upload"
    # 上传配置（混合模式提取文本）
    options_json = json.dumps({"doc.extractionMode": "mixed"})
    
    try:
        # 尝试用原文件名上传
        with open(file_path, "rb") as f:
            response = requests.post(
                url,
                files={"file": f},
                data={"json": options_json}
            )
        response.raise_for_status()
        res_data = json.loads(response.text)
        if res_data["code"] == 101:  # 101错误：文件名含非ASCII字符
            raise Exception("非ASCII文件名")
    except:
        # 用临时ASCII文件名重新上传（如"temp.pdf"）
        file_name = os.path.basename(file_path)
        _, suffix = os.path.splitext(file_name)
        temp_name = f"temp{suffix}"
        with open(file_path, "rb") as f:
            response = requests.post(
                url,
                files={"file": (temp_name, f)},  # 临时文件名
                data={"json": options_json}
            )
        response.raise_for_status()
        res_data = json.loads(response.text)
    
    return res_data["data"]  # 返回文件ID（用于后续处理）

def _process_file(file_id):
    """2. 轮询OCR任务状态，直到处理完成"""
    url = f"{base_url}/api/doc/result"
    data_str = json.dumps({
        "id": file_id,
        "is_data": True,
        "format": "text"  # 结果格式为文本
    })
    
    print("===== 轮询OCR任务状态 =====")
    while True:
        time.sleep(1)  # 每秒轮询一次
        response = requests.post(url, data=data_str, headers=headers)
        response.raise_for_status()
        res_data = json.loads(response.text)
        
        # 断言任务状态正常
        assert res_data["code"] == 100, f"获取状态失败：{res_data}"
        # 打印进度（已处理页数/总页数）
        print(f"进度：{res_data['processed_count']}/{res_data['pages_count']}")
        
        # 任务完成则退出循环
        if res_data["is_done"]:
            assert res_data["state"] == "success", f"OCR失败：{res_data['message']}"
            print("OCR任务完成")
            break

def _generate_target_file(file_id):
    """3. 生成OCR结果下载链接"""
    url = f"{base_url}/api/doc/download"
    # 下载配置（支持txt、pdf等格式）
    download_options = {
        "id": file_id,
        "file_types": ["txt", "pdfLayered"],  # 下载txt和分层PDF
        "ingore_blank": False  # Umi-OCR 2.1.4及以下版本拼写错误，新版本用"ignore_blank"
    }
    
    response = requests.post(url, data=json.dumps(download_options), headers=headers)
    response.raise_for_status()
    res_data = json.loads(response.text)
    assert res_data["code"] == 100, f"获取下载链接失败：{res_data}"
    
    return res_data["name"], res_data["data"]  # 返回文件名和下载链接

def _download_file(url, name):
    """4. 下载OCR结果到本地目录"""
    # 创建下载目录（不存在则创建）
    if not os.path.exists(download_dir):
        os.makedirs(download_dir)
    download_path = os.path.join(download_dir, name)
    
    # 流式下载（支持大文件）
    response = requests.get(url, stream=True)
    response.raise_for_status()
    total_size = int(response.headers.get("content-length", 0))
    downloaded_size = 0
    log_size = 10 * 1024 * 1024  # 每下载10MB打印一次进度
    
    with open(download_path, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            if chunk:
                f.write(chunk)
                downloaded_size += len(chunk)
                # 打印进度
                if downloaded_size >= log_size:
                    log_size += 10 * 1024 * 1024
                    progress = (downloaded_size / total_size) * 100
                    print(f"下载进度：{downloaded_size//1024//1024}MB | {progress:.2f}%")
    
    print(f"OCR结果已保存：{download_path}")
    return download_path

def _clean_up(file_id):
    """5. 清理OCR服务中的临时任务（避免占用资源）"""
    url = f"{base_url}/api/doc/clear/{file_id}"
    response = requests.get(url)
    response.raise_for_status()
    res_data = json.loads(response.text)
    assert res_data["code"] == 100, f"清理任务失败：{res_data}"
    print("OCR任务清理完成")

def ocr_file_to_text(file_path):
    """对外暴露的核心函数：输入文件路径，返回OCR提取的文本文件路径"""
    # 1. 上传文件
    file_id = _upload_file(file_path)
    # 2. 处理OCR
    _process_file(file_id)
    # 3. 生成下载链接
    name, url = _generate_target_file(file_id)
    # 4. 下载结果
    zip_path = _download_file(url, name)
    # 5. 清理任务
    _clean_up(file_id)
    
    # 解压zip包，提取txt文件（OCR结果）
    with zipfile.ZipFile(zip_path, "r") as zip_ref:
        # 获取原始文件名（不含后缀）
        file_name = os.path.basename(file_path)
        file_name_no_ext = os.path.splitext(file_name)[0]
        # 提取OCR结果文件（命名格式：[OCR]_原始文件名.txt）
        target_file = f"[OCR]_{file_name_no_ext}.txt"
        zip_ref.extract(target_file, download_dir)
    
    # 返回OCR文本文件路径
    return os.path.join(download_dir, target_file)

def ocr_image_to_text(file_path):
    """图片OCR：输入图片路径，返回提取的文本（Base64编码传输）"""
    # 图片转Base64（便于HTTP传输）
    with open(file_path, "rb") as f:
        base64_str = base64.b64encode(f.read()).decode("utf-8")
    
    # 调用OCR图片接口
    url = f"{base_url}/api/ocr"
    data = {
        "base64": base64_str,
        "options": {"data.format": "text"}
    }
    response = requests.post(url, data=json.dumps(data), headers=headers)
    response.raise_for_status()
    res_data = json.loads(response.text)
    
    return res_data.get("data", "")  # 返回提取的文本

2.4 工具函数：rag/utils.py辅助功能

rag/utils.py提供3个辅助函数，支撑OCR与RAG的协同：

import base64
from pathlib import Path
from PIL import Image
from .llms import moonshot_llm  # 自定义大模型模块

def ocr_file_to_text_llm(file_path) -> str:
    """用大模型提取文件文本（补充OCR无法处理的场景）"""
    client = moonshot_llm()  # 初始化Moonshot大模型客户端
    # 上传文件并提取文本
    file_obj = client.files.create(file=Path(file_path), purpose="file-extract")
    file_content = client.files.content(file_obj.id).json()
    return file_content.get("content", "")  # 返回提取的文本

def get_b64_image_from_path(image_path):
    """图片转Base64（与ocr_image_to_text功能一致，供外部调用）"""
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")

def is_image(file_path):
    """判断文件是否为图片（避免非图片文件调用图片OCR）"""
    try:
        with Image.open(file_path) as img:
            img.verify()  # 验证图片有效性
        return True
    except Exception as e:
        print(f"非图片文件：{e}")
        return False

（三）RAG知识库问答：实现“检索-生成”智能对话

本模块基于Milvus向量数据库（高效存储向量）与LlamaIndex（RAG框架），实现“文档入库→向量检索→对话生成”流程，同时通过FastAPI提供文件上传与Web服务。

3.1 向量数据库管理：utils/milvus.py

Milvus是开源向量数据库，用于存储文档的向量表示（便于快速检索），utils/milvus.py实现集合（类似数据库表）的查询与删除：

import os
from pymilvus import MilvusClient

# 初始化Milvus客户端（从.env读取地址）
client = MilvusClient(uri=os.getenv("MILVUS_URI", "http://127.0.0.1:19530"))

def list_collections():
    """查询所有集合（知识库）名称"""
    return client.list_collections()

def drop_collection(collection_name):
    """删除指定集合（谨慎使用，会删除所有数据）"""
    client.drop_collection(collection_name=collection_name)

3.2 知识库加载与对话：rag/traditional_rag.py

traditional_rag.py是RAG核心，继承自RAG基类，实现“文档加载→OCR处理→向量入库→索引创建”流程：

import asyncio
import os
from datetime import datetime
from llama_index.core import SimpleDirectoryReader, Document
from .base_rag import RAG  # 自定义RAG基类
from .utils import is_image, ocr_file_to_text, ocr_image_to_text  # 导入OCR工具

class TraditionalRAG(RAG):
    async def load_data(self):
        """加载文件并处理：图片用ocr_image_to_text，文档用ocr_file_to_text"""
        docs = []  # 存储处理后的Document对象
        for file in self.files:  # self.files是传入的文件路径列表
            if is_image(file):
                # 1. 图片：OCR提取文本，保存为临时文件
                text = ocr_image_to_text(file)
                temp_file = f"{datetime.now().strftime('%Y%m%d%H%M%S')}.txt"
                with open(temp_file, "w", encoding="utf-8") as f:
                    f.write(text)
            else:
                # 2. 文档（如PDF）：OCR提取文本，返回文本文件路径
                temp_file = ocr_file_to_text(file)
            
            # 3. 读取临时文件，创建LlamaIndex的Document对象（含元数据）
            data = SimpleDirectoryReader(input_files=[temp_file]).load_data()
            doc = Document(
                text="\n\n".join([d.text for d in data]),  # 合并文本
                metadata={"path": file}  # 记录原始文件路径
            )
            docs.append(doc)
            
            # 4. 删除临时文件（避免占用空间）
            os.remove(temp_file)
        
        return docs  # 返回Document列表，供后续创建向量索引

# 测试：处理单张图片
if __name__ == "__main__":
    rag = TraditionalRAG(files=["../test_data/222.jpg"])  # 传入图片路径
    asyncio.run(rag.create_index_local())  # 创建本地索引（测试用）

3.3 Web服务部署：基于FastAPI

通过FastAPI提供文件上传接口与Web页面，支持用户上传文档并创建知识库索引。

• 3.3.1 主服务：main.py

import os.path
from fastapi import FastAPI, UploadFile, Form, Request, File
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from rag.traditional_rag import TraditionalRAG  # 导入RAG类
from utils.r import R  # 导入响应工具类

# 初始化FastAPI应用
app = FastAPI()
# 初始化模板引擎（用于渲染HTML页面）
templates = Jinja2Templates(directory="templates")
# 文档存储目录（自动创建）
documents_dir = os.path.join(os.getcwd(), "documents")
os.makedirs(documents_dir, exist_ok=True)

@app.post("/uploadfiles/")
async def create_upload_files(
    files: list[UploadFile] = File(...),  # 上传的文件列表
    collection_name: str = Form(None),    # 知识库名称（Milvus集合名）
    multimodal: bool = Form(False)       # 是否开启多模态
):
    """文件上传接口：接收文件，创建RAG索引"""
    file_list = []
    for file in files:
        # 保存上传的文件到documents目录
        file_path = os.path.join(documents_dir, file.filename)
        with open(file_path, "wb+") as f:
            f.write(await file.read())  # 异步读取文件内容
        file_list.append(file_path)
    
    # 初始化RAG对象，创建索引（存入Milvus）
    rag = TraditionalRAG(files=file_list)
    await rag.create_index(collection_name=collection_name)
    
    # 返回成功响应（用自定义R工具类）
    return R.ok("索引创建成功", data={"file_list": file_list})

@app.get("/", response_class=HTMLResponse)
async def main(request: Request):
    """首页：渲染文件上传表单"""
    return templates.TemplateResponse(
        "index.html",  # 模板文件路径
        {"request": request}  # 传递请求对象（FastAPI模板要求）
    )

# 启动命令：uvicorn main:app --host 0.0.0.0 --port 8080

• 3.3.2 响应工具类：utils/r.py
  标准化API响应格式，便于前端处理：

from starlette.responses import JSONResponse

class R:
    @staticmethod
    def ok(message="success", data=None):
        """成功响应：状态码200"""
        return JSONResponse(
            status_code=200,
            content={"code": 200, "message": message, "data": data}
        )
    
    @staticmethod
    def error(message="error", data=None, status_code=500):
        """错误响应：默认状态码500"""
        return JSONResponse(
            status_code=status_code,
            content={"code": status_code, "message": message, "data": data}
        )

• 3.3.3 前端页面：templates/index.html
  简单的文件上传表单，支持多文件上传与知识库命名：

<!DOCTYPE html>
<html>
<head>
    <title>RAG知识库文件上传</title>
</head>
<body>
    <h1>上传文件到RAG知识库</h1>
    <form action="/uploadfiles/" enctype="multipart/form-data" method="post">
        <!-- 知识库名称 -->
        <input name="collection_name" type="text" placeholder="输入知识库名称" required><br>
        <!-- 多文件上传 -->
        <input name="files" type="file" multiple required><br>
        <!-- 多模态开关 -->
        <input name="multimodal" type="checkbox" checked> 开启多模态RAG<br>
        <!-- 提交按钮 -->
        <input type="submit" value="上传并创建索引">
    </form>
</body>
</html>

3.4 对话界面整合：chainlit_ui.py完整逻辑

chainlit_ui.py整合RAG与Chainlit界面，实现“选择知识库→发送消息→智能回复”流程，核心代码如下：

import random
import chainlit as cl
from chainlit.input_widget import Select, Switch
from chainlit.types import ThreadDict
from dotenv import load_dotenv
from llama_index.core.base.llms.types import ChatMessage
from llama_index.core.chat_engine import SimpleChatEngine
from llama_index.core.chat_engine.types import ChatMode
from llama_index.core.memory import ChatMemoryBuffer
from rag.traditional_rag import TraditionalRAG
from rag.base_rag import RAG
from utils.milvus import list_collections

# 加载环境变量
load_dotenv()

# 1. 聊天配置：仅管理员可见知识库（示例认证）
@cl.set_chat_profiles
async def chat_profile(current_user: cl.User):
    # 仅管理员可查看知识库（实际项目需对接真实权限系统）
    if current_user.metadata.get("role") != "admin":
        return None
    
    # 获取Milvus中的所有知识库（集合）
    kb_list = list_collections()
    # 初始化配置：默认对话（无知识库）+ 所有知识库
    profiles = [
        cl.ChatProfile(
            name="default",
            markdown_description="直接与大模型对话（不使用知识库）",
            icon="/public/kbs/4.png"
        )
    ]
    # 为每个知识库添加配置
    for kb_name in kb_list:
        profiles.append(
            cl.ChatProfile(
                name=kb_name,
                markdown_description=f"基于「{kb_name}」知识库对话",
                icon=f"/public/kbs/{random.randint(1, 3)}.jpg"  # 随机图标
            )
        )
    return profiles

# 2. 初始化聊天：加载知识库与聊天引擎
@cl.on_chat_start
async def start():
    # 发送设置面板（模型选择+多模态开关）
    await cl.ChatSettings(
        [
            Select(id="Model", label="模型选择", values=["DeepSeek", "Moonshot"], initial_index=0),
            Switch(id="multimodal", label="多模态RAG", initial=True)
        ]
    ).send()
    
    # 获取用户选择的知识库（chat_profile）
    kb_name = cl.user_session.get("chat_profile")
    if kb_name in [None, "default"]:
        # 默认模式：无知识库，仅大模型对话
        memory = ChatMemoryBuffer.from_defaults(token_limit=1024)  # 对话记忆
        chat_engine = SimpleChatEngine.from_defaults(memory=memory)
    else:
        # 知识库模式：加载Milvus中的索引，创建RAG聊天引擎
        index = await RAG.load_index(collection_name=kb_name)
        chat_engine = index.as_chat_engine(chat_mode=ChatMode.CONTEXT)  # 基于上下文检索
    
    # 保存聊天引擎到会话
    cl.user_session.set("chat_engine", chat_engine)

# 3. 处理用户消息：支持文本与文件（图片/文档）
@cl.on_message
async def main(message: cl.Message):
    # 从会话获取聊天引擎
    chat_engine = cl.user_session.get("chat_engine")
    # 初始化助手回复
    msg = cl.Message(content="", author="Assistant")
    await msg.send()  # 先发送空消息，后续流式输出内容
    
    # 处理用户上传的文件（图片/文档）
    files = []
    for elem in message.elements:
        if isinstance(elem, (cl.File, cl.Image)):
            files.append(elem.path)
    
    # 如果有文件：创建临时索引（补充到现有知识库）
    if len(files) > 0:
        rag = TraditionalRAG(files=files)
        # 创建临时索引（不存入Milvus，仅当前会话使用）
        index = await rag.create_index_local()
        # 更新聊天引擎为RAG模式
        chat_engine = index.as_chat_engine(chat_mode=ChatMode.CONTEXT)
        cl.user_session.set("chat_engine", chat_engine)
    
    # 流式生成回复（提升用户体验）
    res = await cl.make_async(chat_engine.stream_chat)(message.content)
    for token in res.response_gen:
        await msg.stream_token(token)
    
    # 完成回复
    await msg.update()

# 4. 认证回调：示例管理员认证（实际项目需替换为真实认证）
@cl.password_auth_callback
def auth_callback(username: str, password: str) -> cl.User:
    if (username, password) == ("admin", "admin"):
        return cl.User(
            identifier="admin",
            metadata={"role": "admin", "provider": "credentials"}
        )
    return None  # 认证失败

三、完整搭建流程与测试

（一）环境准备

1. 安装依赖：
   创建requirements.txt，包含所有依赖包：

chainlit>=1.0.0
llama-index>=0.10.0
pymilvus>=2.4.0
fastapi>=0.100.0
uvicorn>=0.23.0
requests>=2.31.0
pydantic>=2.0.0
pillow>=10.0.0
python-dotenv>=1.0.0

执行安装命令：

pip install -r requirements.txt

2. 启动依赖服务：
   • Milvus向量数据库：
     参考Milvus快速启动，本地部署后默认地址为http://127.0.0.1:19530（与.env中MILVUS_URI一致）。
   • Umi-OCR服务：
     打开Umi-OCR，开启HTTP服务（端口1314，与.env中OCR_BASE_URL一致）。

（二）项目目录结构

确保项目目录如下（避免路径错误）：

your_project/
├── .env                  # 环境变量配置
├── config.toml           # Chainlit配置
├── main.py               # FastAPI主服务
├── chainlit_ui.py        # Chainlit对话界面
├── requirements.txt      # 依赖包列表
├── public/               # 静态资源（Logo、CSS、图标）
│   ├── favicon.png
│   ├── logo_dark.png
│   ├── logo_light.png
│   ├── ui.css
│   └── kbs/              # 知识库图标
│       ├── 1.jpg
│       ├── 2.jpg
│       ├── 3.jpg
│       └── 4.png
├── rag/                  # RAG与OCR核心模块
│   ├── __init__.py
│   ├── config.py         # 配置封装
│   ├── ocr.py            # OCR核心逻辑
│   ├── traditional_rag.py# RAG核心逻辑
│   ├── base_rag.py       # RAG基类（文档中未给出，需自定义）
│   └── llms.py           # 大模型客户端（文档中未给出，需自定义）
├── utils/                # 工具函数
│   ├── __init__.py
│   ├── milvus.py         # Milvus管理
│   ├── r.py              # 响应工具
│   └── utils.py          # 辅助函数
├── templates/            # HTML模板
│   └── index.html        # 文件上传页面
├── documents/            # 上传文件存储目录（自动创建）
└── data/                 # OCR结果存储目录（自动创建）
    └── ocr_download/

（三）启动与测试

1. 启动FastAPI服务（文件上传）：

uvicorn main:app --host 0.0.0.0 --port 8080

• 访问http://127.0.0.1:8080，上传文件并输入知识库名称（如test_kb），点击“上传并创建索引”；
• 索引创建成功后，文件内容会通过OCR提取并转为向量存入Milvus。

2. 启动Chainlit对话界面：

chainlit run chainlit_ui.py --port 8000

• 访问http://127.0.0.1:8000，输入管理员账号密码（admin/admin）；
• 在“聊天配置”中选择test_kb知识库，发送与文件相关的问题（如“文档中提到的软件测试方法有哪些？”），即可获得基于知识库的智能回复。

3. 测试OCR功能：
   • 在Chainlit界面上传一张含文字的图片，发送消息（如“提取图片中的文字”）；
   • 系统会自动调用OCR提取文字，并基于提取结果生成回复。

四、关键注意事项

1. Umi-OCR版本兼容：
   • 文档中使用Umi-OCR 2.1.4，_generate_target_file函数中ingore_blank为拼写错误；
   • 若使用新版本Umi-OCR，需将ingore_blank改为ignore_blank。
2. Milvus数据管理：
   • 每次创建知识库索引前，确保Milvus中无同名集合（可通过utils.milvus.drop_collection删除）；
   • Milvus默认不持久化数据，需在配置中开启持久化（参考Milvus官方文档）。
3. 大模型API密钥：
   • .env中需配置DEEPSEEK_API_KEY和MOONSHOT_API_KEY（从对应大模型平台获取），否则无法生成对话回复。
4. 权限控制：
   • 文档中auth_callback为示例认证，实际项目需对接企业权限系统（如OAuth2、LDAP），避免硬编码账号密码。

通过以上步骤，即可完整搭建一套支持OCR识别的RAG知识库对话系统，实现“图片/文档识别→知识存储→智能问答”的全流程。