PDF论文解析

前面咱们做的系统使用的数据集是txt文件，这样做的目的是为了让系统暂时先跑通，今天咱们需要给系统再做一个升级，让系统真的可以阅读咱们平时可以看到的论文PDF文件。不过首先，咱们需要准备一个数据集文件，像下面这样，在根目录下创建一个data文件夹，里面放入两篇测试用的论文pdf：

data/
  paper1.pdf
  paper2.pdf

这里有个坑需要注意一下！ 由于咱们搭建的是一个系统，目前只是测试一下系统能否跑通，保证功能可实现就好了，放两篇几页的小论文就好，千万不要放几十页的综述论文，文本数量太多了，需要消耗大量token，钱包扛不住的。紧接着，咱们需要安装一个可以解析PDF的库。

pip install pypdf

随后，像咱们前面写的加载txt文档的函数一样，咱们需要写一个可以读取PDF的函数，并备注文献来源：

from pypdf import PdfReader

def load_pdfs(folder_path):
    documents = []

    for filename in os.listdir(folder_path):
        if filename.endswith(".pdf"):
            path = os.path.join(folder_path, filename)
            reader = PdfReader(path)

            text = ""
            for page in reader.pages:
                text += page.extract_text() or ""

            documents.append({
                "text": text,
                "source": filename
            })

    return documents

接着修改主函数，将原来读取txt的部分改成读取PDF，这样咱们就像模像样实现了将论文内容读入：

if __name__ == '__main__':
    docs = load_pdfs("data")   # ← 改这里

    chunks = process_documents(docs)

    rag = RAGSystem(chunks)
    rag.build_index()

    answer = rag.ask("What is the core contribution of paper1?")
    print(answer)

文本清洗

但是前面读取的PDF数据其实是有一些问题的，你可以在pycharm或者其他工具中打一个断点，可以看到读取到的PDF文件数据里面有很多回车符以及一些空格，这样数据组织形式比较乱，作者不喜欢这种乱做一团的数据，这些数据断句很奇怪，所以咱们需要对数据进行数据清洗：

import re

def clean_text(text):
    text = re.sub(r'\n+', '\n', text)
    text = re.sub(r'\s+', ' ', text)
    return text.strip()

最后在load_pdfs函数中，加上text = clean_text(text)就可以了，完整代码像这样：

def load_pdfs(folder_path):
    documents = []

    for filename in os.listdir(folder_path):
        if filename.endswith(".pdf"):
            path = os.path.join(folder_path, filename)
            reader = PdfReader(path)

            text = ""
            for page in reader.pages:
                text += page.extract_text() or ""

            text = clean_text(text)

            documents.append({
                "text": text,
                "source": filename
            })

    return documents

其实这一步我们的代码已经可以自己运行了，但是我们还可以对它进行一次升级，让他变成一个服务。

API服务化

将我们的程序进行API服务化需要用到一个叫做FastAPI的工具，这个工具简单，可以做到几行代码起服务，而且是专门为AI接口设计的。首先安装依赖：

pip install fastapi uvicorn

随后我们需要建一个app.py文件，然后再里面写：

from fastapi import FastAPI
from pydantic import BaseModel

# ===== 引入RAG系统 =====
from server import RAGSystem, load_pdfs, process_documents

# ===== 初始化 =====
app = FastAPI()

docs = load_pdfs("data")
chunks = process_documents(docs)

rag = RAGSystem(chunks)
rag.build_index()


# ===== 请求格式 =====
class QueryRequest(BaseModel):
    question: str


# ===== API接口 =====
@app.post("/ask")
def ask_question(req: QueryRequest):
    answer = rag.ask(req.question)
    return {
        "question": req.question,
        "answer": answer
    }

通过下面的命令就可以在终端启动服务了：

uvicorn app:app --reload

通过访问http://127.0.0.1:8000/docs就可以看到我们的服务，点页面上的ask服务，然后再点Try it out，就可以再页面上输入你的问题与系统交互。

这里有个坑！ 作者排查了好久，原来是我的网络不稳定，所以在进行chunk词嵌入的时候，有的chunk没返回embedding值，结果导致系统启动失败，所以需要对咱们的嵌入函数进行一定的修改，增加函数的鲁棒性：

def get_embedding(text, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = client2.embeddings.create(
                model="text-embedding-3-small",
                input=text
            )
            return np.array(response.data[0].embedding, dtype="float32")

        except Exception as e:
            print(f"Embedding失败，第{attempt+1}次重试...")
            time.sleep(2)

    print("Embedding最终失败，返回零向量")
    return np.zeros(1536, dtype="float32")  # embedding维度

同时对app.py进行修改，前面的代码是先初始化rag，然后再启动FastAPI，这样当初始化遇到任何问题，那么FastAPI都启动不了，所以咱们需要对app.py进行修改，通过装饰器@app.on_event("startup") 让服务正式上线前，完成一些必要的准备工作。

from fastapi import FastAPI
from pydantic import BaseModel

from server import RAGSystem, load_pdfs, process_documents

# 初始化
app = FastAPI()

rag = None  # 先不初始化

class QueryRequest(BaseModel):
    question: str

@app.on_event("startup")
def load_rag():
    global rag
    print("🚀 Loading RAG system...")

    docs = load_pdfs("data")
    print("docs数量:", len(docs))

    chunks = process_documents(docs)
    print("chunks数量:", len(chunks))

    rag = RAGSystem(chunks)
    rag.build_index()

    print("✅ RAG ready!")

# API接口
@app.post('/ask')
def ask_question(req: QueryRequest):
    answer = rag.ask(req.question)
    return {
        "question": req.question,
        "answer": answer
    }

这种服务启动前操作，其实还有配对的服务后操作，也有对应的装饰器# 关闭事件处理器 @app.on_event("shutdown") 不过咱们这里用不上，只给大家科普提一嘴。可以这样理解当你输入uvicorn app:app --reload时，启动事件处理器的装饰器装饰的函数会执行，这是系统会执行启动事件；当你按住Ctrl+ C结束服务时，关闭事件装饰器装饰的函数将会启动。

如果这篇文章对你有帮助，可以点个赞～
完整代码地址：https://github.com/1186141415/A-Paper-Rag-Agent