从零开始做一个高校课程资料 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_modelDocumentRead,status_codestatus.HTTP_201_CREATED)asyncdefupload_document(file:UploadFile,db:SessionDepends(get_db))-Document:这个接口挂在routerAPIRouter(prefix/api/documents,tags[documents])所以完整路径是POST /api/documents前端或测试代码会用multipart/form-data上传文件例如POST /api/documents Content-Type: multipart/form-data filelogin-lab.md后端通过 FastAPI 的UploadFile接收文件。2. 初始化数据库表上传函数第一步执行init_db()init_db()在 database.py 里Base.metadata.create_all(bindengine)它会确保数据库表存在包括documentschunksuserschat_messages如果表已经存在不会重复创建。数据库地址来自配置settings.database_url默认是sqlite:///./rag_assistant.db如果.env里配置了DATABASE_URL会优先使用.env中的值。3. 保存原始上传文件到本地 uploads 目录上传接口接着执行stored_pathawaitsave_upload_file(file)对应 storage.py。3.1 获取上传目录defget_upload_root()-Path:rootget_settings().upload_dir root.mkdir(parentsTrue,exist_okTrue)returnroot上传目录来自配置upload_dir:PathPath(uploads)默认情况下文件会保存到backend/uploads/如果.env里写了UPLOAD_DIRsomewhere_else则会保存到你配置的位置。3.2 生成安全文件名safe_namePath(file.filenameorupload.bin).name targetroot/f{uuid4().hex}-{safe_name}这里做了两件事Path(...).name只取文件名去掉路径避免用户传入类似../../xxx的路径。前面加一个 UUID避免同名文件互相覆盖。例如用户上传login-lab.md实际保存后可能变成uploads/8f2c9c1e2e3d4a7b9a-login-lab.md3.3 读取并写入文件contentawaitfile.read()target.write_bytes(content)当前实现是一次性把上传文件完整读入内存然后写入磁盘。这对小文件没问题如果以后支持很大的 PDF/PPT可能需要改成流式写入避免占用过多内存。4. 在documents表创建资料记录文件保存后接口会创建一个Document数据库记录documentDocument(filenamefile.filenameorstored_path.name,file_typedetect_file_type(file.filenameorstored_path.name),storage_pathstr(stored_path),statusuploaded,uploaded_byNone,)对应数据库表模型classDocument(Base):__tablename__documentsid:intfilename:strfile_type:strstorage_path:strstatus:strerror_message:str|Noneuploaded_by:int|Nonecreated_at:datetime updated_at:datetime字段含义字段含义filename用户上传时的原始文件名file_type根据后缀推断出的文件类型storage_path文件实际保存路径status当前处理状态error_message解析失败时记录错误uploaded_by上传用户目前是Nonecreated_at创建时间updated_at更新时间文件类型如何判断defdetect_file_type(filename:str)-str:suffixPath(filename).suffix.lower().lstrip(.)returnsuffixorunknown例如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改成processingingest_document()第一段逻辑withSessionLocal.begin()assession:documentsession.get(Document,document_id)storage_pathdocument.storage_path document.statusprocessingdocument.error_messageNone它会重新打开一个数据库 session根据document_id找到刚刚上传的资料记录然后把状态改成processing表示正在处理。此时状态流转为uploaded - processing7. 根据文件类型解析资料内容接下来执行parsedparse_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_typecode,languagejava)它会用 UTF-8 读取整个文件textpath.read_text(encodingutf-8)然后返回一个ParsedDocumentParsedDocument(titlepath.name,sections[ParsedSection(texttext,content_typecode,source_pathstr(path),languagejava,)],)也就是说代码文件会被标记为chunk_type code language java / xml / sql / jsp / ...后续问答时可以知道这段资料是代码。9. Markdown 和 TXT 解析文本文件后缀是TEXT_EXTENSIONS{.md,.txt}如果上传login-lab.md也会走_parse_plain_file()但类型是content_typetextlanguageNone也就是说 Markdown 当前没有做标题结构解析只是当作普通文本整体读取。10. PDF 解析如果上传.pdfifsuffix.pdf:return_parse_pdf(path)PDF 解析逻辑readerPdfReader(str(path))forindex,pageinenumerate(reader.pages,start1):textpage.extract_text()oriftext.strip():sections.append(...)它会按页读取 PDF对每一页执行extract_text()跳过空白页每一页生成一个ParsedSection记录页码source_page生成的 section 大致是ParsedSection(text这一页提取出来的文字,content_typetext,source_pathuploads/xxx.pdf,source_page1,)所以 PDF 后续引用时可以知道内容来自第几页。限制是它只提取 PDF 中可抽取的文字。如果 PDF 是扫描图片当前代码不会 OCR。11. Word.docx解析如果上传.docxifsuffix.docx:return_parse_docx(path)解析逻辑documentDocxDocument(str(path))text\n.join(paragraph.textforparagraphindocument.paragraphsifparagraph.text.strip())它会读取 Word 文档中的段落文本并用换行拼起来。当前不会解析表格结构图片页码批注页眉页脚最后生成一个整体 sectionParsedSection(texttext,content_typetext,source_pathstr(path),)12. PowerPoint.pptx解析如果上传.pptxifsuffix.pptx:return_parse_pptx(path)解析逻辑presentationPresentation(str(path))forindex,slideinenumerate(presentation.slides,start1):texts[]forshapeinslide.shapes:ifhasattr(shape,text)andshape.text.strip():texts.append(shape.text)它会遍历每一页幻灯片遍历幻灯片里的 shape如果 shape 有文本就提取出来每一页幻灯片生成一个 sectionsource_page记录为幻灯片页码也就是说PPTX 的一页幻灯片大致会变成一个文本 section。13. 不支持的文件类型会失败如果文件后缀不在支持范围内例如bad.exe会走到raiseValueError(fUnsupported file type:{suffix})这会让摄取流程失败。当前上传接口没有捕获这个异常所以接口层会返回500。但是失败前会把数据库里的资料状态更新为failed并记录错误信息。测试里也验证了这个行为assertresponse.status_code500assertdocuments[0][status]failedassertdocuments[0][error_message]14. 解析结果被切成 chunks文件解析完成后执行chunkschunk_parsed_document(parsed)切块逻辑在 chunking.py。默认参数max_chars900overlap_chars120意思是每个 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 包含哪些信息每个切片会被封装成ChunkDataChunkData(contentpiece,chunk_typesection.content_type,source_titleparsed.title,source_pathsection.source_path,source_pagesection.source_page,languagesection.language,metadata{title:parsed.title},)字段含义字段含义content这一小段资料内容chunk_typetext或codesource_title原始文件名source_path原始文件保存路径source_pagePDF 页码或 PPT 页码language代码语言例如javametadata额外元数据目前只有标题16. 删除旧 chunks写入新 chunks解析和切块成功后系统重新打开数据库事务withSessionLocal.begin()assession:documentsession.get(Document,document_id)session.query(Chunk).filter(Chunk.document_iddocument_id).delete()它会先删除这个 document 旧的 chunks。然后逐个写入新的 chunksession.add(Chunk(document_iddocument_id,contentchunk.content,chunk_typechunk.chunk_type,source_titlechunk.source_title,source_pagechunk.source_page,source_pathchunk.source_path,languagechunk.language,metadata_jsonchunk.metadata,))对应数据库表是classChunk(Base):__tablename__chunksid: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.statusindexeddocument.error_messageNone状态变成indexed完整成功状态流转是uploaded - processing - indexed这表示资料已经上传、解析、切块并写入数据库可以被后续问答检索使用。18. 如果中间失败状态改成failedingest_document()外层有异常处理exceptExceptionasexc:withSessionLocal.begin()assession:documentsession.get(Document,document_id)ifdocumentisnotNone:document.statusfaileddocument.error_messagestr(exc)raise如果解析、切块或入库任何一步失败document 状态改成failed错误信息写入error_message异常继续抛出所以数据库会保留失败记录但 HTTP 接口当前会返回服务器错误。失败状态流转一般是uploaded - processing - failed19. 上传接口最终返回什么如果成功接口返回DocumentReadclassDocumentRead(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 接口返回 DocumentRead22. 当前实现的几个关键特点1. 上传和解析是同步的接口会等解析完成后才返回。优点前端拿到结果时资料已经可用。缺点大文件会让接口响应变慢。2. 原始文件会保存在本地默认目录是backend/uploads/数据库里只保存文件路径不保存文件二进制内容。3. 文件名会加 UUID 防冲突用户上传的原始文件名保存在documents.filename。实际磁盘文件名会变成{uuid}-{原始文件名}4. 当前没有向量化虽然项目叫 RAG assistant但当前上传后没有 embedding 流程。当前资料入库单位是普通文本 chunks后续检索主要依赖chunks表中的文本内容。5. 失败文件也会留下记录例如上传.exe文件会先保存到 uploadsdocument 记录会创建ingest 阶段失败document 状态变成failederror_message记录失败原因HTTP 返回 50023. 简化版总结上传资料后系统内部实际发生的是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