从零开始做一个高校课程资料 AI Agent 问答系统（三）上传资料全流程

从零开始做一个高校课程资料 AI Agent 问答系统

本系列博客将带你从零开始，使用Python + FastAPI + RAG + AI Agent搭建一个面向Java Web 课程资料问答与智能学习辅助的后端系统，涵盖课程资料上传、文档解析、文本切块、本地检索、问答生成、Agent 工具调用、执行链路追踪、运行记录回放等实战场景。无论你是 AI 应用初学者，还是希望掌握 RAG 与 Agent 工程化落地的开发者，都能通过本教程理解一个教学场景 AI Agent 系统的规划、开发、测试与升级过程。

内容主要包括：

✅基础架构搭建：FastAPI 后端项目初始化、接口路由设计、Pydantic Schema、SQLAlchemy 数据模型、SQLite 本地数据库配置。

✅课程资料知识库构建：支持 Markdown、TXT、代码文件、PDF、Word、PPTX 等资料上传，完成文档解析、文本切块、来源元数据保存与资料入库。

✅RAG 问答流程实现：基于课程资料 chunk 进行本地检索，构造 grounded prompt，生成带引用来源的回答，并在资料不足时安全拒答，降低大模型幻觉。

✅LLM 接入与兜底机制：支持 OpenAI-compatible 接口，可对接 OpenAI、Ollama、本地大模型等，同时保留 stub fallback，保证系统在无模型环境下也能稳定测试。

✅AI Agent 能力升级：从简单 Agent Harness 逐步升级为具备 Planner、Executor、Tool Registry、Memory、Verifier 的课程学习 Agent，实现任务规划、工具调用、短期记忆、结果校验和执行追踪。

✅Agent 执行记录与回放：新增 AgentRun、AgentStep、AgentToolCall 数据模型，保存每次 Agent 的执行计划、工具调用、引用来源、校验结果，并提供运行历史查询接口。

✅接口调试与测试验证：通过 Swagger UI 手动测试文档上传、问答、Agent 运行、工具列表和运行记录接口，并使用 pytest 编写自动化测试，覆盖 RAG、Agent、工具、记忆和校验流程。

✅系统演进与工程实践：从 MVP 到 v1.2.0 Agent 升级，逐步讲解如何控制范围、拆分模块、保持接口兼容、设计可测试代码，并为后续前端 Trace 面板、向量检索、LangGraph 工作流和 SaaS 化扩展打基础。

上传资料后的系统全流程

相关核心文件主要是：

• documents.py：上传接口
• storage.py：保存上传文件
• ingestion.py：解析、分块、入库
• parsers.py：不同文件类型解析
• chunking.py：文本切块
• document.py：资料表模型
• chunk.py：资料切片表模型

1. 用户上传文件，请求进入/api/documents

上传接口定义在：

@router.post("",response_model=DocumentRead,status_code=status.HTTP_201_CREATED)asyncdefupload_document(file:UploadFile,db:Session=Depends(get_db))->Document:

这个接口挂在：

router=APIRouter(prefix="/api/documents",tags=["documents"])

所以完整路径是：

POST /api/documents

前端或测试代码会用multipart/form-data上传文件，例如：

POST /api/documents Content-Type: multipart/form-data file=login-lab.md

后端通过 FastAPI 的UploadFile接收文件。

2. 初始化数据库表

上传函数第一步执行：

init_db()

init_db()在 database.py 里：

Base.metadata.create_all(bind=engine)

它会确保数据库表存在，包括：

• documents
• chunks
• users
• chat_messages

如果表已经存在，不会重复创建。

数据库地址来自配置：

settings.database_url

默认是：

sqlite:///./rag_assistant.db

如果.env里配置了DATABASE_URL，会优先使用.env中的值。

3. 保存原始上传文件到本地 uploads 目录

上传接口接着执行：

stored_path=awaitsave_upload_file(file)

对应 storage.py。

3.1 获取上传目录

defget_upload_root()->Path:root=get_settings().upload_dir root.mkdir(parents=True,exist_ok=True)returnroot

上传目录来自配置：

upload_dir:Path=Path("uploads")

默认情况下，文件会保存到：

backend/uploads/

如果.env里写了：

UPLOAD_DIR=somewhere_else

则会保存到你配置的位置。

3.2 生成安全文件名

safe_name=Path(file.filenameor"upload.bin").name target=root/f"{uuid4().hex}-{safe_name}"

这里做了两件事：

1. Path(...).name只取文件名，去掉路径，避免用户传入类似../../xxx的路径。
2. 前面加一个 UUID，避免同名文件互相覆盖。

例如用户上传：

login-lab.md

实际保存后可能变成：

uploads/8f2c9c1e2e3d4a7b9a-login-lab.md

3.3 读取并写入文件

content=awaitfile.read()target.write_bytes(content)

当前实现是一次性把上传文件完整读入内存，然后写入磁盘。

这对小文件没问题；如果以后支持很大的 PDF/PPT，可能需要改成流式写入，避免占用过多内存。

4. 在documents表创建资料记录

文件保存后，接口会创建一个Document数据库记录：

document=Document(filename=file.filenameorstored_path.name,file_type=detect_file_type(file.filenameorstored_path.name),storage_path=str(stored_path),status="uploaded",uploaded_by=None,)

对应数据库表模型：

classDocument(Base):__tablename__="documents"id:intfilename:strfile_type:strstorage_path:strstatus:strerror_message:str|Noneuploaded_by:int|Nonecreated_at:datetime updated_at:datetime

字段含义

文件类型如何判断

defdetect_file_type(filename:str)->str:suffix=Path(filename).suffix.lower().lstrip(".")returnsuffixor"unknown"

例如：

login-lab.md -> md slides.pptx -> pptx bad.exe -> exe

注意：这里仅根据文件后缀判断，不检查真实 MIME 类型。

然后保存数据库：

db.add(document)db.commit()db.refresh(document)

此时数据库里已经有一条状态为：

uploaded

的资料记录。

5. 进入资料摄取流程ingest_document

上传记录创建成功后，接口立刻调用：

ingest_document(document.id)

这个函数在 ingestion.py。

注意：当前项目是同步处理，不是后台任务。

也就是说：

上传请求不会在文件保存后立刻返回 而是会等解析、分块、入库全部完成后才返回

如果资料很大，接口响应会变慢。

6. 状态从uploaded改成processing

ingest_document()第一段逻辑：

withSessionLocal.begin()assession:document=session.get(Document,document_id)storage_path=document.storage_path document.status="processing"document.error_message=None

它会重新打开一个数据库 session，根据document_id找到刚刚上传的资料记录，然后把状态改成：

processing

表示正在处理。

此时状态流转为：

uploaded -> processing

7. 根据文件类型解析资料内容

接下来执行：

parsed=parse_file(Path(storage_path))

解析逻辑在 parsers.py。

支持的文件类型分几类。

8. 代码文件解析

代码文件后缀定义在：

CODE_EXTENSIONS={".java":"java",".xml":"xml",".properties":"properties",".yml":"yaml",".yaml":"yaml",".sql":"sql",".html":"html",".jsp":"jsp",".js":"javascript",".css":"css",}

如果上传的是：

LoginServlet.java

会走：

_parse_plain_file(path,content_type="code",language="java")

它会用 UTF-8 读取整个文件：

text=path.read_text(encoding="utf-8")

然后返回一个ParsedDocument：

ParsedDocument(title=path.name,sections=[ParsedSection(text=text,content_type="code",source_path=str(path),language="java",)],)

也就是说，代码文件会被标记为：

chunk_type = code language = java / xml / sql / jsp / ...

后续问答时可以知道这段资料是代码。

9. Markdown 和 TXT 解析

文本文件后缀是：

TEXT_EXTENSIONS={".md",".txt"}

如果上传：

login-lab.md

也会走_parse_plain_file()，但类型是：

content_type="text"language=None

也就是说 Markdown 当前没有做标题结构解析，只是当作普通文本整体读取。

10. PDF 解析

如果上传.pdf：

ifsuffix==".pdf":return_parse_pdf(path)

PDF 解析逻辑：

reader=PdfReader(str(path))forindex,pageinenumerate(reader.pages,start=1):text=page.extract_text()or""iftext.strip():sections.append(...)

它会：

1. 按页读取 PDF
2. 对每一页执行extract_text()
3. 跳过空白页
4. 每一页生成一个ParsedSection
5. 记录页码source_page

生成的 section 大致是：

ParsedSection(text="这一页提取出来的文字",content_type="text",source_path="uploads/xxx.pdf",source_page=1,)

所以 PDF 后续引用时可以知道内容来自第几页。

限制是：它只提取 PDF 中可抽取的文字。如果 PDF 是扫描图片，当前代码不会 OCR。

11. Word.docx解析

如果上传.docx：

ifsuffix==".docx":return_parse_docx(path)

解析逻辑：

document=DocxDocument(str(path))text="\n".join(paragraph.textforparagraphindocument.paragraphsifparagraph.text.strip())

它会读取 Word 文档中的段落文本，并用换行拼起来。

当前不会解析：

• 表格结构
• 图片
• 页码
• 批注
• 页眉页脚

最后生成一个整体 section：

ParsedSection(text=text,content_type="text",source_path=str(path),)

12. PowerPoint.pptx解析

如果上传.pptx：

ifsuffix==".pptx":return_parse_pptx(path)

解析逻辑：

presentation=Presentation(str(path))forindex,slideinenumerate(presentation.slides,start=1):texts=[]forshapeinslide.shapes:ifhasattr(shape,"text")andshape.text.strip():texts.append(shape.text)

它会：

1. 遍历每一页幻灯片
2. 遍历幻灯片里的 shape
3. 如果 shape 有文本，就提取出来
4. 每一页幻灯片生成一个 section
5. source_page记录为幻灯片页码

也就是说，PPTX 的一页幻灯片大致会变成一个文本 section。

13. 不支持的文件类型会失败

如果文件后缀不在支持范围内，例如：

bad.exe

会走到：

raiseValueError(f"Unsupported file type:{suffix}")

这会让摄取流程失败。

当前上传接口没有捕获这个异常，所以接口层会返回500。但是失败前会把数据库里的资料状态更新为：

failed

并记录错误信息。

测试里也验证了这个行为：

assertresponse.status_code==500assertdocuments[0]["status"]=="failed"assertdocuments[0]["error_message"]

14. 解析结果被切成 chunks

文件解析完成后，执行：

chunks=chunk_parsed_document(parsed)

切块逻辑在 chunking.py。

默认参数：

max_chars=900overlap_chars=120

意思是：

• 每个 chunk 最多 900 个字符
• 相邻 chunk 之间重叠 120 个字符

例如一段很长的资料会被切成：

chunk 1: 第 0 ~ 900 字 chunk 2: 第 780 ~ 1680 字 chunk 3: 第 1560 ~ 2460 字

中间有 120 字重叠，目的是避免重要上下文刚好被切断。

切块前会先清理文本

normalized="\n".join(line.rstrip()forlineintext.splitlines()).strip()

它会：

• 去掉每行右侧空白
• 保留换行结构
• 去掉整体首尾空白

如果清理后没有内容：

ifnotnormalized:return[]

则不会生成 chunk。

15. 每个 chunk 包含哪些信息

每个切片会被封装成ChunkData：

ChunkData(content=piece,chunk_type=section.content_type,source_title=parsed.title,source_path=section.source_path,source_page=section.source_page,language=section.language,metadata={"title":parsed.title},)

字段含义：

16. 删除旧 chunks，写入新 chunks

解析和切块成功后，系统重新打开数据库事务：

withSessionLocal.begin()assession:document=session.get(Document,document_id)session.query(Chunk).filter(Chunk.document_id==document_id).delete()

它会先删除这个 document 旧的 chunks。

然后逐个写入新的 chunk：

session.add(Chunk(document_id=document_id,content=chunk.content,chunk_type=chunk.chunk_type,source_title=chunk.source_title,source_page=chunk.source_page,source_path=chunk.source_path,language=chunk.language,metadata_json=chunk.metadata,))

对应数据库表是：

classChunk(Base):__tablename__="chunks"id:intdocument_id:intcontent:strchunk_type:strsource_title:str|Nonesource_page:int|Nonesource_path:str|Nonelanguage:str|Nonemetadata_json:dictcreated_at:datetime

这张表就是后续问答检索的基础。

17. 资料状态改成indexed

所有 chunks 写入成功后：

document.status="indexed"document.error_message=None

状态变成：

indexed

完整成功状态流转是：

uploaded -> processing -> indexed

这表示资料已经上传、解析、切块，并写入数据库，可以被后续问答检索使用。

18. 如果中间失败，状态改成failed

ingest_document()外层有异常处理：

exceptExceptionasexc:withSessionLocal.begin()assession:document=session.get(Document,document_id)ifdocumentisnotNone:document.status="failed"document.error_message=str(exc)raise

如果解析、切块或入库任何一步失败：

1. document 状态改成failed
2. 错误信息写入error_message
3. 异常继续抛出

所以数据库会保留失败记录，但 HTTP 接口当前会返回服务器错误。

失败状态流转一般是：

uploaded -> processing -> failed

19. 上传接口最终返回什么

如果成功，接口返回DocumentRead：

classDocumentRead(BaseModel):id:intfilename:strfile_type:strstatus:strerror_message:str|Nonecreated_at:datetime updated_at:datetime

成功返回示例：

{"id":1,"filename":"login-lab.md","file_type":"md","status":"indexed","error_message":null,"created_at":"2026-06-20T19:30:08","updated_at":"2026-06-20T19:30:09"}

注意：返回结果里不会包含 chunks 内容，只返回资料记录本身。

20. 上传后资料如何被后续问答使用

上传流程本身只做到：

保存文件 -> 建 document 记录 -> 解析文件 -> 切 chunks -> 存 chunks

后续用户提问时，系统会从chunks表里检索相关内容。

当前项目的 RAG 不是向量数据库版本，而是简单关键词/分词匹配检索。相关逻辑在：

app/services/retrieval.py app/services/chat.py

也就是说，当前上传资料后并没有生成 embedding，也没有写入向量库。

它目前更像是：

本地文件存储 + SQLite 元数据 + 文本切块 + 关键词检索

不是完整的：

上传 -> 向量化 -> 存入向量数据库 -> 语义检索

21. 整体流程图

用户上传文件 | v POST /api/documents | v init_db() 确保数据库表存在 | v save_upload_file() | |-- 读取 UPLOAD_DIR 配置 |-- 创建 uploads 目录 |-- 生成 UUID 文件名 |-- 写入原始文件 v documents 表插入记录 | |-- filename |-- file_type |-- storage_path |-- status = uploaded v ingest_document(document.id) | v status = processing | v parse_file(storage_path) | |-- .java/.xml/.sql/... -> code |-- .md/.txt -> text |-- .pdf -> 按页提取文本 |-- .docx -> 提取段落文本 |-- .pptx -> 按幻灯片提取文本 |-- 不支持 -> 抛异常 v chunk_parsed_document() | |-- 每块最多 900 字符 |-- 相邻块重叠 120 字符 v chunks 表写入切片 | v status = indexed | v 接口返回 DocumentRead

22. 当前实现的几个关键特点

1. 上传和解析是同步的

接口会等解析完成后才返回。

优点：前端拿到结果时，资料已经可用。

缺点：大文件会让接口响应变慢。

2. 原始文件会保存在本地

默认目录是：

backend/uploads/

数据库里只保存文件路径，不保存文件二进制内容。

3. 文件名会加 UUID 防冲突

用户上传的原始文件名保存在documents.filename。

实际磁盘文件名会变成：

{uuid}-{原始文件名}

4. 当前没有向量化

虽然项目叫 RAG assistant，但当前上传后没有 embedding 流程。

当前资料入库单位是普通文本 chunks，后续检索主要依赖chunks表中的文本内容。

5. 失败文件也会留下记录

例如上传.exe：

• 文件会先保存到 uploads
• document 记录会创建
• ingest 阶段失败
• document 状态变成failed
• error_message记录失败原因
• HTTP 返回 500

23. 简化版总结

上传资料后，系统内部实际发生的是：

1. FastAPI 接收上传文件 2. 把文件保存到 uploads 目录 3. 在 documents 表插入一条资料记录，状态为 uploaded 4. 进入 ingest_document() 5. 状态改为 processing 6. 根据文件后缀解析内容 7. 把解析出的文本按 900 字符切块，块之间重叠 120 字符 8. 把 chunks 写入数据库 9. document 状态改为 indexed 10. 接口返回资料记录

如果解析失败：

1. document 状态改为 failed 2. error_message 保存错误原因 3. 接口当前返回 500