企业官网建设流程全解析

简介：面向希望了解知识库系统如何落地的Python开发者与Web项目学习者，这份源码提供了一套可本地运行的AI知识库参考实现，重点覆盖文档上传、内容管理与问答检索等核心流程。包内共22个文件，以10个html页面模板、6个py业务脚本、4个pyc缓存文件为主，另含1个txt说明文档和1个db数据库文件，压缩包约53KB，规模轻量但结构完整。已有66人学习下载。读者可从中观察数据库初始化、配置参数管理、路由与模型组织，以及登录、文章录入、文档管理与问答日志等后台入口的协作方式，并借助说明文档快速理解系统用途。它适合作为课程设计、毕业项目原型或二次开发的起点，也便于对照梳理知识库系统从数据存储到页面交互的实现脉络。

1. 从一份能跑起来的 Python 知识库源码说起：它到底解决了什么

前阵子有人拿来一份 AI 知识库系统 Python 源码，说接口能起、页面上传 PDF 也不报错，可问什么都是「资料中未提及」。翻日志才发现，切片确实生成了，但一条都没写进向量库——.env里向量库目录写成了相对路径，服务在容器里跑，落到了另一个挂载点。这类项目常被当成「大模型套壳」，真正吃功夫的却是工程链路：解析、切分、向量化、召回、拼上下文、流式吐字，哪一环掉链子，用户看到的都是同一句「答不出来」。这份源码把链路拆成了可替换的模块，后端用 FastAPI，向量库和 LLM 都留了适配层，适合想给团队搭一套内部资料问答、又不想从零写 RAG 骨架的 Python 开发者，也适合拿它当样本，把 RAG 从概念读到能改代码的程度。

2. 技术选型与目录结构：一套 RAG 知识库的最小闭环

先把「这项目由哪些东西拼起来」讲清楚，后面读代码才不会迷路。整套系统按四层切开：接入层负责 HTTP 与流式响应，应用层管知识库、文档、会话三类业务对象，能力层是解析器、切分器、向量化、向量库、LLM 客户端这些可替换件，存储层则是关系库、向量库和原始文件目录三份数据。分层的好处很直接——换 embedding 模型只动能力层，换前端只动接入层的响应格式，业务代码基本不用碰。

技术选型上，常见做法是 Python 3.10 起步，FastAPI + Uvicorn 做服务，SQLAlchemy 管元数据，默认 SQLite 方便本地起步、上生产换 MySQL 或 PostgreSQL 只改连接串；向量库默认 Chroma，本地持久化、零运维，数据量上去或者要多人并发再换 Milvus、Qdrant，适配层已经把接口收敛成add / query / delete三个方法。文档解析用 PyMuPDF 处理 PDF、python-docx 处理 Word、纯文本和 Markdown 直接读，扫描件另外挂 OCR。LLM 侧统一走 OpenAI 兼容协议，本地推理服务和云端 API 用同一套客户端，切换只改 base_url 和 model 两个配置。

2.1 目录结构：每个目录负责哪一段链路

先看骨架，我一般拿到新项目第一件事就是把目录打印出来，对着 README 找入口，比直接翻代码快得多。

ai-knowledge-base/ ├── app/ │ ├── main.py # FastAPI 入口：注册路由、异常处理、启动钩子 │ ├── core/ │ │ ├── config.py # pydantic-settings 读 .env，全局唯一配置源 │ │ ├── logging.py # 统一日志格式，入库失败靠它定位 │ │ └── deps.py # 依赖注入：db session、向量库客户端、当前用户 │ ├── api/v1/ │ │ ├── knowledge.py # 知识库的增删改查 │ │ ├── document.py # 上传、解析状态查询、切片预览 │ │ └── chat.py # 问答、SSE 流式输出、会话历史 │ ├── models/ # SQLAlchemy ORM：知识库、文档、切片、会话 │ ├── schemas/ # Pydantic 请求/响应模型，入参校验都在这 │ ├── services/ │ │ ├── parser/ # 各格式解析器，统一返回纯文本 │ │ ├── splitter.py # 切分策略，chunk_size / overlap 在这 │ │ ├── embedding.py # 向量化封装，本地模型与远程 API 二选一 │ │ ├── vector_store.py # 向量库适配层，屏蔽 Chroma/FAISS 差异 │ │ ├── retriever.py # 召回、阈值过滤、可选重排 │ │ └── llm.py # LLM 客户端，流式与非流式两套方法 │ └── prompts/ # Prompt 模板，按场景拆文件 ├── data/ │ ├── uploads/ # 原始文件，按 kb_id 分目录 │ └── chroma/ # 向量库持久化目录 ├── web/ # 前端页面，独立工程或静态文件 ├── tests/ ├── .env.example ├── requirements.txt ├── Dockerfile └── docker-compose.yml

这份结构里最值得留意的是services/和api/的分界：路由层只做参数校验和响应组装，所有带状态的逻辑都压在 service 里。好处是问答链路可以脱离 HTTP 单独跑测试——传一个问题和 kb_id，直接断言召回结果和拼出来的 prompt，不用起服务、不用造请求。二次开发时想加「文档摘要」「问答评价」这类功能，照着 service 加文件、api 加路由就行，不会把入口文件写成一千行的面条。

models/里四张核心表撑起了整个业务：知识库表存名字和向量库 collection 名的映射，文档表存文件路径、解析状态、切片数、失败原因，切片表存 chunk 文本与在原文中的偏移量（用于回答时高亮出处），会话表存多轮历史。切片文本在向量库里也有一份，关系库这份主要是给「切片预览」和「删除文档时级联清理」用的，两边靠doc_id + chunk_index对齐。

2.2 配置与启动：一份 settings 管住所有可调项

RAG 系统的参数特别碎，切分长度、召回条数、相似度阈值、上下文预算，散落在代码里会非常难调。这个项目用一份 pydantic-settings 收口，.env覆盖默认值，改参数不用动代码。

from functools import lru_cache from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): # extra="ignore" 很重要：.env 里多写无关变量不会导致启动失败 model_config = SettingsConfigDict(env_file=".env", extra="ignore") app_host: str = "0.0.0.0" app_port: int = 8000 database_url: str = "sqlite:///./data/app.db" # 向量库：backend 决定走哪个适配器，目录必须用绝对路径或容器内一致路径 vector_backend: str = "chroma" chroma_dir: str = "/app/data/chroma" # 向量化：本地模型与远程 API 二选一 embedding_provider: str = "local" # local | remote embedding_model: str = "BAAI/bge-small-zh-v1.5" embedding_dim: int = 512 embedding_batch_size: int = 16 # 切分参数，中文按字符数算更直观 chunk_size: int = 500 chunk_overlap: int = 80 # 检索参数 top_k: int = 5 recall_multiplier: int = 2 # 先多召回再过滤 score_threshold: float = 0.35 # LLM：统一 OpenAI 兼容协议 llm_base_url: str = "http://127.0.0.1:11434/v1" llm_api_key: str = "sk-local" llm_model: str = "qwen2.5-7b-instruct" llm_temperature: float = 0.2 max_context_tokens: int = 3000 @lru_cache def get_settings() -> Settings: return Settings()

代码里几个点值得说。@lru_cache保证全进程只有一个 Settings 实例，配置读一次就够，避免每个请求都解析一遍.env；extra="ignore"是血泪经验，团队协作时.env里常混进别人调试用的变量，不加这个，pydantic 会直接抛 ValidationError 把服务拦在启动阶段。chroma_dir默认写成容器内的绝对路径，就是我前面说的那个坑——本地跑用相对路径没事，一进容器工作目录变了，向量库写到别处，读的时候自然是空的。

启动方式两条路，本地开发直接热重载，线上走容器：

# 本地开发：--reload 监听代码变更，日志级别调到 info 方便看入库过程 uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload --log-level info # 容器启动：先确认 data 目录挂载出来，否则重启后向量库和上传文件全丢 docker compose up -d --build

提示：启动后先访问/docs看 Swagger 是否完整加载，路由缺失通常意味着某个api/v1/*.py里的router没在main.py注册。

3. 文档入库链路：从上传 PDF 到切片落进向量库

入库是整条链路里最容易出问题的一段，因为它跨了文件系统、关系库、向量库三处状态，中间任何一步失败，用户看到的只是「文档状态一直是解析中」。我把这段拆成三步看：先把各种格式统一成纯文本，再按语义边界切成块，最后批量向量化连同元数据一起写进去。

3.1 解析层：多格式统一成纯文本

解析器的设计目标是「对上层只暴露一个函数」，上层不关心传进来的是 PDF 还是 Word。

from pathlib import Path from .pdf import parse_pdf # PyMuPDF 逐页提取 from .docx import parse_docx # python-docx 读段落 + 表格 from .text import parse_text # txt / md 直读，md 可选去标记 PARSER_MAP = { ".pdf": parse_pdf, ".docx": parse_docx, ".md": parse_text, ".txt": parse_text, } def parse_document(path: str | Path) -> str: path = Path(path) parser = PARSER_MAP.get(path.suffix.lower()) if parser is None: raise ValueError(f"不支持的文件类型: {path.suffix}") text = parser(path) # 统一清洗：去零宽字符和不断行空格，合并连续空行，去行首尾空白 text = text.replace("\u200b", "").replace("\xa0", " ") lines = [line.strip() for line in text.splitlines()] return "\n".join(line for line in lines if line)

PARSER_MAP用后缀查函数，加新格式只需要写一个函数、注册一行，parse_document不用改。清洗那几行别小看：PDF 复制出来的文本里零宽字符、不断行空格非常多，不处理会在切分时被当成正常字符算进长度，还会污染 embedding 的输入，让相似度莫名偏低。Word 里的表格要单独处理，python-docx的paragraphs拿不到表格内容，得遍历tables把单元格拼成「列名: 值」的形式塞回文本，否则制度类文档里的参数表整块丢失。

PDF 这块的现实是：文字版 PDF 用 PyMuPDF 提取效果很好，扫描件提取出来是空字符串。常见做法是在解析后判一个字符数下限，低于阈值就标记文档为「疑似扫描件」，走 OCR 分支或者直接返回失败原因让用户自己处理，别让它静默入库成零切片。

3.2 切分策略：chunk_size 与 overlap 的取舍

切分决定了召回的粒度上限。切太大，一个 chunk 里混着几个主题，向量被平均掉，召回不准；切太小，一句话被拦腰截断，模型拿到半句话答不出来。

from dataclasses import dataclass # 从粗到细的语义边界，中文标点一定要放进候选 SEPARATORS = ["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] @dataclass class Chunk: text: str index: int start: int # 在清洗后全文中的起始字符偏移 end: int def split_text(text: str, chunk_size: int = 500, overlap: int = 80) -> list[Chunk]: chunks: list[Chunk] = [] start, idx, n = 0, 0, len(text) while start < n: end = min(start + chunk_size, n) # 窗口尾部回退到最近的语义边界，避免把句子切成两半 if end < n: for sep in SEPARATORS: pos = text.rfind(sep, start + chunk_size // 2, end) if pos != -1: end = pos + len(sep) break piece = text[start:end].strip() if piece: chunks.append(Chunk(text=piece, index=idx, start=start, end=end)) idx += 1 if end >= n: break # 回退 overlap 个字符做重叠；max 防止 overlap 过大导致原地打转 start = max(end - overlap, start + 1) return chunks

逐个参数看：chunk_size=500是中文场景的常用起点，约等于三四百 token，正好覆盖一两个自然段，再大就超出小模型的注意力重点区。overlap=80保证跨块的信息不断裂，比如一张表的说明文字落在上一块、表头落在下一块，有重叠才有可能一起被召回。SEPARATORS的顺序是从段落级到字符级，rfind从窗口后半段往前找，能找到句号就不在逗号处断。

start = max(end - overlap, start + 1)这行是防死循环的保险。如果 overlap 配得比 chunk_size 还大，或者文本里有超长无标点串，没有这个保护就会原地循环、进程卡死。我在别的项目里真见过有人把 overlap 设成 1000 配 chunk_size 500，服务起来后 CPU 打满，查了半天才定位到这里。

切完每个 chunk 带上start/end偏移量，后面回答时能把出处定位到原文位置做高亮，这在内部知识库里是刚需——用户看到答案的第一反应是「哪份文件第几段说的」。

3.3 向量化与写入：批处理与元数据设计

向量化是整条链路里最慢的一步，也是内存最容易爆的一步。写法上一定要分批，别把一篇三百页的 PDF 切出来的几百个 chunk 一次性丢给模型。

def embed_texts(texts: list[str], batch_size: int = 16) -> list[list[float]]: vectors: list[list[float]] = [] for i in range(0, len(texts), batch_size): batch = texts[i:i + batch_size] if settings.embedding_provider == "local": # 本地模型：模型实例在模块级只加载一次，绝不能每个请求 new 一个 vecs = local_model.encode(batch, normalize_embeddings=True) vectors.extend(vecs.tolist()) else: resp = remote_client.embeddings(model=settings.embedding_model, input=batch) vectors.extend([d.embedding for d in resp.data]) return vectors def upsert_chunks(kb_id: str, doc_id: str, filename: str, chunks: list[Chunk], vectors: list[list[float]]) -> None: # collection 按知识库隔离，命名带前缀方便运维时按前缀清理 collection = chroma_client.get_or_create_collection( name=f"kb_{kb_id}", metadata={"hnsw:space": "cosine"} ) collection.upsert( # ids 用 doc_id:index 保证同一文档重新入库是覆盖而不是重复 ids=[f"{doc_id}:{c.index}" for c in chunks], documents=[c.text for c in chunks], embeddings=vectors, metadatas=[{ "doc_id": doc_id, "chunk_index": c.index, "char_start": c.start, "char_end": c.end, "source": filename, } for c in chunks], )

normalize_embeddings=True配合 cosine 距离，是检索打分能对齐的前提。ids的构造方式决定了幂等性：同一份文档被重复上传，用upsert会覆盖旧切片，不会在库里堆两份；如果换成add，Chroma 遇到重复 id 会报错，很多人第一次重传文件就卡在这。删除文档时按where={"doc_id": doc_id}批量删，比逐个 id 删快得多。

元数据里的char_start / char_end是我强烈建议保留的字段。省这点存储的代价是，回答生成后你没法做「点击引用跳转到原文对应位置」，而这恰恰是内部知识库区别于普通聊天机器人的关键体验。

入库这一步别放在上传接口里同步执行。一份大 PDF 解析加向量化可能要几十秒，HTTP 早超时了。常见做法是上传接口只落文件、建文档记录、状态置pending，然后丢给后台任务（FastAPI 的BackgroundTasks或者独立的 Celery / RQ 队列）跑完整条链路，前端轮询状态接口。任务里要用 try/except 包住整段，失败时把异常信息写进文档表的error_msg，不然用户只能看到无限转圈。

4. 检索问答链路：从一句提问到流式吐字

问答是用户真正接触的部分，链路比入库短，但参数敏感度高得多。一次提问要走：查询处理、向量化、召回、过滤、拼 Prompt、调模型、流式返回，中间还有会话历史要不要带、多知识库怎么隔离的问题。

4.1 召回：query 前缀、多召回再过滤

召回这块有一个特别容易翻车的细节，跟模型强相关。BGE 系列的中文模型在训练时对 query 侧加了指令前缀，doc 侧不加，如果两边都加或者都不加，相似度会明显掉一截。

def retrieve(kb_id: str, query: str) -> list[dict]: # BGE 中文系列：query 侧加指令前缀，doc 侧入库时不加 q = "为这个句子生成表示以用于检索相关文章：" + query q_vec = embed_texts([q])[0] collection = chroma_client.get_collection(f"kb_{kb_id}") res = collection.query( query_embeddings=[q_vec], n_results=settings.top_k * settings.recall_multiplier, include=["documents", "metadatas", "distances"], ) hits = [] for text, meta, dist in zip(res["documents"][0], res["metadatas"][0], res["distances"][0]): score = 1 - dist # cosine 距离转相似度，范围大致 0~1 if score >= settings.score_threshold: hits.append({"text": text, "meta": meta, "score": score}) return hits[:settings.top_k]

recall_multiplier=2的意思是先捞 10 条再过滤到 5 条。直接按 top_k 捞再卡阈值，经常出现「捞了 5 条、卡掉 3 条、只剩 2 条进上下文」，回答质量自然差。多捞一倍的代价很小，收益是过滤后仍有足够上下文。

score = 1 - dist只对 cosine 空间成立。如果你把 collection 的hnsw:space换成l2，这个换算就错了，阈值 0.35 会变得毫无意义，表现为「要么全过要么全不过」。换距离度量的时候，阈值必须重新标定——拿十几条已知相关的问答对跑一遍，看正样本的分数分布落在哪，再定阈值，别拍脑袋。

4.2 上下文拼装：预算内裁剪与引用编号

召回结果不能无脑全塞。模型上下文有上限，塞爆了要么报错，要么前面的指令被挤掉，回答开始胡说。

SYSTEM_PROMPT = """你是知识库问答助手，只能依据下面提供的资料回答。 资料中未提及的内容，直接回复「资料中未提及」，不要编造。 回答时在句末用 [1] [2] 标注引用的资料编号。 资料： {context} """ def build_prompt(question: str, hits: list[dict], max_tokens: int = 3000) -> str: blocks, used = [], 0 for i, h in enumerate(hits, start=1): # 中文粗略估算：1 个汉字约 1 token，留 20% 余量给问题和历史 cost = len(h["text"]) + 20 if used + cost > max_tokens * 0.8: break src = h["meta"].get("source", "未知来源") blocks.append(f"[{i}] 来源：{src}\n{h['text']}") used += cost return SYSTEM_PROMPT.format(context="\n\n".join(blocks))

几个设计点：每条资料带编号和来源名，模型才有东西可引用，前端拿这个编号去映射原文位置；按顺序累加、超预算就截断，而不是把每条都截一半，后者会让所有资料都变成残句；max_tokens * 0.8留出的余量是给用户问题和多轮历史的，不然长会话几轮之后上下文就满了。

Prompt 里那句「未提及就直说」不是可选项。不加这句，模型在资料无关时也会顺着常识编一段，用户第一次被骗到就会对整个系统的信任度归零。

4.3 流式输出：SSE 的三个必配项

问答体验里，首字延迟比总耗时更影响感受。非流式接口等十几秒才吐一整段，用户会以为卡死了；流式一个字一个字出，体感完全不同。

import json from fastapi.responses import StreamingResponse async def event_stream(question: str, kb_id: str, history: list[dict]): hits = retrieve(kb_id, question) prompt = build_prompt(question, hits) # 先把引用来源推给前端，用户能立刻看到「参考了哪几份资料」 refs = [{"index": i, "source": h["meta"].get("source"), "score": round(h["score"], 3)} for i, h in enumerate(hits, start=1)] yield f"data: {json.dumps({'type': 'references', 'data': refs}, ensure_ascii=False)}\n\n" async for delta in llm_stream(prompt, history): yield f"data: {json.dumps({'type': 'delta', 'data': delta}, ensure_ascii=False)}\n\n" yield "data: [DONE]\n\n" @router.post("/chat/stream") async def chat_stream(req: ChatRequest): return StreamingResponse( event_stream(req.question, req.kb_id, req.history), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # 关键：告诉 nginx 不要缓冲 }, )

三个必配项：media_type必须是text/event-stream，否则浏览器不会按 SSE 解析；X-Accel-Buffering: no是给反向代理看的，不写这条，nginx 默认会把响应缓冲到积累一定大小才下发，表现就是「明明后端在流式吐字，前端却一次性蹦出全文」；响应体每条消息以\n\n结尾，格式错一个换行前端就解析不出来。

ensure_ascii=False不能省。默认True会把中文转成\uXXXX，传输量翻几倍不说，前端如果处理不当会出现乱码。

4.4 会话与知识库隔离：别让答案串库

会话表里要存kb_id，每次提问都带着走时召回，别用全局变量或者缓存 key 里漏掉 kb_id，不然 A 部门的资料会出现在 B 部门的回答里，这在内部系统里是事故。向量库层面靠 collection 名隔离，kb_{kb_id}这个前缀保证查询范围天然收窄；关系库层面查会话历史时带上kb_id条件。

多轮对话还有个细节：用户说「那它的价格呢」，单看这句话向量化出来是没法召回的。常见做法是把最近两三轮的问答拼成一句改写后的 query 再检索，或者干脆把上一轮问题和当前问题拼接后一起向量化。前者效果好但多一次模型调用，后者零成本但容易召回偏。小团队起步用拼接法就够，等用户开始抱怨「上下文接不上」再上改写。

5. 常见问题排查：五个把新手卡住的现场

这套系统出问题的表现高度同质化，几乎都是「答不出来」或者「答得不对」，但根因分散在五个完全不同的位置。下面按我实际排查的频率排，每条都给出定位手法。

5.1 入库静默失败：上传成功但永远检索不到

现象：接口返回 200，文档列表里状态一直停在「解析中」或者显示「已完成」，但问相关问题一律回「资料中未提及」。

原因：入库流程被拆成了异步任务，上传接口只负责落文件和建记录，真正的解析、切分、向量化在后台跑。后台任务抛异常时如果没有写回状态和错误信息，前端看到的就是「成功但没数据」。最常见的异常是向量库目录不存在或无写权限，其次是解析器抛了未捕获的异常。

解决：先查文档表的error_msg字段，空的话去翻服务日志里该doc_id附近的打印。再确认向量库目录挂载和权限，容器里跑ls -l /app/data/chroma看有没有写权限。最后直接调切片数接口，如果切片数是 0，问题一定在解析或切分，不在向量化。我习惯在入库任务的开头、切分完成、写入完成三个点各打一条带doc_id和数量的日志，出问题一眼能看出卡在哪一步。

5.2 召回为空：切片有了但搜不出来

现象：文档状态正常、切片数也对，切片预览能看到内容，可提问就是召回不到。

原因：三种可能。一是 embedding 侧 query 和 doc 用了不同的处理路径，比如 doc 入库时加了指令前缀而 query 没加，两边向量空间对不齐，相似度普遍偏低，全被阈值卡掉。二是score_threshold定得太高，正样本分数在 0.3 附近，阈值设了 0.5 就全灭。三是换过 embedding 模型但没重建 collection，旧向量维度或语义空间和新模型不匹配。

解决：临时把阈值调到 0，打印召回结果和分数，看正样本到底落在什么区间，再回来定阈值。查一下入库和查询是不是走同一个embed_texts，BGE 系列的 query 前缀只能加在查询侧。换模型必须删掉旧 collection 全量重入，没有捷径——维度不同时 Chroma 会直接报错，维度碰巧相同更麻烦，它不报错，只是默默返回一堆无关结果。

5.3 答案串库：A 知识库问出 B 知识库的内容

现象：两个部门各建了一个知识库，提问 A 的问题，引用来源里出现了 B 的文件。

原因：collection 命名没有按知识库隔离，所有知识库共用一个 collection，靠 metadata 过滤；而查询时忘了带where条件，或者前端传的 kb_id 在后端被默认值覆盖了。

解决：优先用 collection 隔离，物理上分开，查询范围天然收窄，不依赖每个查询都记得写过滤条件。如果因为数据量原因必须共用 collection，那就在适配层的query方法里把where={"kb_id": kb_id}写死成必传参数，从接口签名上就不给漏掉的机会。顺手在会话表里加个kb_id字段，出问题时能快速定位是哪个环节把 kb_id 丢了。

5.4 PDF 解析出空白或乱码：扫描件与字体编码

现象：文字版 PDF 解析正常，某些 PDF 解析出来是空字符串或者一堆方块字符。

原因：扫描件没有文本层，PyMuPDF 提取自然是空；部分 PDF 用了非标准字体编码，提取出来是乱码；还有些 PDF 每页带页眉页脚，提取后被切成大量重复的短行，污染切片。

解决：解析后判字符数，低于页面数乘以某个下限就标记「疑似扫描件」，走 OCR 分支或者明确告诉用户不支持。乱码的先换解析库试，PyMuPDF 和 pdfplumber 对不同编码的容忍度不一样，两个都试通常能救回来一部分。页眉页脚用规则清洗：统计每页重复出现的短行，出现频率超过八成页数的直接删掉。这一步不做，切出来的 chunk 里有一半是「XX公司内部资料 第 N 页」，召回精度会被严重拉低。

5.5 流式输出变成一次性返回

现象：后端日志显示模型在逐字吐，前端却等十几秒后一次性显示全文。

原因：反向代理缓冲、响应头缺X-Accel-Buffering、前端没用流式读取方式这三个位置之一。另外还有一种隐蔽情况——框架的中间件把StreamingResponse包了一层，读取完整个 body 才转发。

解决：先直连后端端口测试，绕过代理，如果直连正常就说明是代理层，加X-Accel-Buffering: no并确认代理配置里对应 location 关了缓冲。前端用fetch配response.body.getReader()逐块 decode，别用await response.text()，那个方法会等整个响应结束。中间件问题相对少见，通过逐层注释中间件能快速二分出来。

注意：排查流式问题时别在浏览器开发者工具里看「响应」面板，它是等完整响应才渲染的，看「事件流」或网络请求的 Timing 更准。

6. 二次开发：换模型、加混合检索、接业务系统

这套骨架真正值钱的地方是可替换性，落地时最常改的三个方向，我按改动成本和风险排一下。

换 embedding 模型。改settings.embedding_model和embedding_dim，然后必须全量重建向量库——删掉data/chroma下对应 collection，把所有文档状态重置为 pending 重新入库。风险和成本都在这步重入上：文档多的话耗时几个小时，期间检索不可用。我的做法是新旧 collection 并存，新 collection 写完之后改配置里的 collection 前缀切过去，验证没问题再删旧的，回滚成本几乎为零。别信「维度一样就能混用」，不同模型即使维度相同，向量空间也是两个世界。

加混合检索。纯向量检索对专有名词、编号、型号这类精确匹配很弱，用户搜「HT-2048」经常召回一堆语义相近但编号不对的内容。补一路 BM25 关键词召回，再用 RRF 融合两路结果，是比较稳的做法。

def rrf_fuse(vector_hits: list[dict], bm25_hits: list[dict], k: int = 60) -> list[dict]: # RRF：按排名而非分数融合，避免两路分数量纲不同无法直接加权 scores: dict[str, float] = {} pool: dict[str, dict] = {} for rank, hit in enumerate(vector_hits, start=1): key = f"{hit['meta']['doc_id']}:{hit['meta']['chunk_index']}" scores[key] = scores.get(key, 0) + 1 / (k + rank) pool[key] = hit for rank, hit in enumerate(bm25_hits, start=1): key = f"{hit['meta']['doc_id']}:{hit['meta']['chunk_index']}" scores[key] = scores.get(key, 0) + 1 / (k + rank) pool[key] = hit ordered = sorted(scores.items(), key=lambda x: x[1], reverse=True) return [pool[key] for key, _ in ordered]

k=60是 RRF 的经验常数，作用是压平排名靠前几位的分差，让两路结果都能进最终列表。BM25 那一路用rank_bm25在内存里建索引就够，文档量到十万级再考虑换 Elasticsearch。融合后仍要走一遍top_k截断和阈值过滤。

接业务系统。对外暴露/v1/chat这种稳定接口，鉴权用 API Key 或者内网 token，别让业务方直接调前端用的会话接口。返回体里除了答案，把引用来源的doc_id、source、char_start/char_end一起返回，业务系统才能做「点击跳原文」。再加一层限流，按 kb_id 维度限制 QPS，防止某个业务方把模型配额吃干净。

有回我图省事，在同一份 collection 上直接换 embedding 模型，维度碰巧一样，Chroma 一声不吭收了，结果整整两天都在查「为什么召回全是无关内容」，最后才发现是新旧向量混在一个空间里。从那以后我每次动 embedding 配置，都强制先备份向量库目录、写清迁移步骤、用新 collection 跑一批已知问答对验证命中率，确认无误再切。希望帮到你。

本文还有配套的精品资源，点击获取