从零搭建 RAG 核心链路：文本切片 + Ollama 向量化 + Milvus 存储（附源码）

本片文章仅作为我自己学习记录顺便也分享给大家。一、为什么做这个构建一个 Agent 开发的基础设施的其中之一巩固基础完成文档到向量数据库的全链路熟悉开发与设计流程二、技术栈层技术Web 框架Flask向量模型Ollama qwen3-embedding:4b2560 维向量数据库Milvus部署在腾讯云文件解析python-docx只处理word文档管理工具Attumilvus可视化工具三、核心实现1.模块拆分docToMilvus.py → Flask 路由接收入口 处理文本.py → 滑窗切片 向量化 存 Milvus核心引擎 处理文件.py → 文件解析复用处理文本逻辑2.滑窗切片策略- window_size500, slide400 - 相邻切片 100 字重叠保证语义不断层3.向量化- 调 Ollama /api/embed 接口 - qwen3-embedding:4b 输出 2560 维向量4.Milvus存储- 使用新版 MilvusClient API - 指定 db_name 连接正确的 Database5.统一返回值{success: true/false, msg: ..., chunk_count: n}四、遇到的问题及解决方法问题原因解决cant find collectionDatabase 名不是默认 default加 db_namemy_milvusuri is illegalURI 缺少协议前缀加上 http://500 返回 None文件分支忘了 return补上返回逻辑SchemaNotReadyException表没在 Milvus 上建好Attu 里建表或代码自动创建五、架构总结数据源文件/文本 → 解析器 → 处理文本.py ├─ 滑窗切片 ├─ Ollama 向量化 └─ Milvus 写入六、下一步计划- 搜索查询向量数据接口 - 接入大模型做 RAG 对话 - 搭配前端 Vue 2 页面作为小demo七、源码docToMilvus.pyfrom flask import Flask, request import 处理文本 import 处理文件 app Flask(__name__) app.route(/) def hello(): return {msg: ok} app.route(/document, methods[POST]) def document(): # 两种模式第一种是上传文件第二个是上传大长文本只能用其中一个 file request.files.get(file) text request.form.get(text) if file and file.filename: result 处理文件.process_file(file) if not result[success]: return {msg: result[msg], code: 400} return {msg: ok, code: 200, chunk_count: result[chunk_count]} elif text: result 处理文本.process_text(text) if not result[success]: return {msg: result[msg], code: 400} return {msg: ok, code: 200, chunk_count: result[chunk_count]} else: return {msg: file and text all None, code: 400} if __name__ __main__: app.run(host0.0.0.0, port8000)处理文本.pyimport requests from pymilvus import MilvusClient MILVUS_URI http://124.223.163.254:19530 MILVUS_DB my_milvus COLLECTION_NAME doc_chunks OLLAMA_URL http://localhost:11434/api/embed OLLAMA_MODEL qwen3-embedding:4b WINDOW_SIZE 500 SLIDE 400 def process_text(text): chunks [] start 0 while start len(text): chunk text[start:start WINDOW_SIZE] if chunk.strip(): chunks.append(chunk.strip()) start SLIDE if not chunks: return {success: False, msg: empty text, chunk_count: 0} response requests.post(OLLAMA_URL, json{ model: OLLAMA_MODEL, input: chunks }) if response.status_code ! 200: return {success: False, msg: follama error {response.status_code}, chunk_count: 0} vector_list response.json()[embeddings] client MilvusClient(uriMILVUS_URI, db_nameMILVUS_DB) data [{text: chunks[i], vector: vector_list[i]} for i in range(len(chunks))] client.insert(collection_nameCOLLECTION_NAME, datadata) return {success: True, msg: ok, chunk_count: len(chunks)}处理文件.pyimport os from docx import Document import 处理文本 # 暂时不考虑有表格的情况 def process_file(file): ext os.path.splitext(file.filename)[1].lower() if ext not in (.docx, .doc): return {success: False, msg: only support .docx file, chunk_count: 0} doc Document(file) text \n.join(p.text for p in doc.paragraphs) if not text.strip(): return {success: False, msg: empty file, chunk_count: 0} return 处理文本.process_text(text)