本片文章仅作为我自己学习记录，顺便也分享给大家。

一、为什么做这个

        构建一个 Agent 开发的基础设施的其中之一，巩固基础

        完成文档到向量数据库的全链路，熟悉开发与设计流程

二、技术栈

层	技术
Web 框架	Flask
向量模型	Ollama + qwen3-embedding:4b（2560 维）
向量数据库	Milvus（部署在腾讯云）
文件解析	python-docx（只处理word文档）
管理工具	Attu（milvus可视化工具）

三、核心实现

        1.模块拆分

docToMilvus.py    → Flask 路由，接收入口
处理文本.py        → 滑窗切片 + 向量化 + 存 Milvus（核心引擎）
处理文件.py        → 文件解析，复用处理文本逻辑

        2.滑窗切片策略

- window_size=500, slide=400
- 相邻切片 100 字重叠，保证语义不断层

        3.向量化

- 调 Ollama /api/embed 接口
- qwen3-embedding:4b 输出 2560 维向量

        4.Milvus存储

- 使用新版 MilvusClient API
- 指定 db_name 连接正确的 Database

        5.统一返回值

{"success": true/false, "msg": "...", "chunk_count": n}

四、遇到的问题及解决方法

问题	原因	解决
can't find collection	Database 名不是默认 default	加 db_name="my_milvus"
uri is illegal	URI 缺少协议前缀	加上 http://
500 返回 None	文件分支忘了 return	补上返回逻辑
SchemaNotReadyException	表没在 Milvus 上建好	Attu 里建表或代码自动创建

五、架构总结

数据源（文件/文本） → 解析器 → 处理文本.py
                                    ├─ 滑窗切片
                                    ├─ Ollama 向量化
                                    └─ Milvus 写入

六、下一步计划

- 搜索查询向量数据接口
- 接入大模型做 RAG 对话
- 搭配前端 Vue 2 页面作为小demo

七、源码

docToMilvus.py

from 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(host="0.0.0.0", port=8000)

处理文本.py

import 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": f"ollama error {response.status_code}", "chunk_count": 0}

    vector_list = response.json()["embeddings"]

    client = MilvusClient(uri=MILVUS_URI, db_name=MILVUS_DB)
    data = [{"text": chunks[i], "vector": vector_list[i]} for i in range(len(chunks))]
    client.insert(collection_name=COLLECTION_NAME, data=data)

    return {"success": True, "msg": "ok", "chunk_count": len(chunks)}

处理文件.py

import 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)