企业官网建设流程全解析

一张月考题，7 个任务，100 分。没有历史包袱，从零开始，如何在一天内交出一个可运行的「基于 LangGraph 的智能文档问答系统（RAG）」？本文按真实开发顺序复盘全流程：拆题 → 验证风险 → 父子块入库 → 状态机问答 → 接口与验收，并附上所有踩过的坑。

一、拆题

表格

拆题后立刻能得到两个关键结论：

1. 分值就是优先级。任务 3~6 加起来 60 分，核心是 LangGraph 的状态图本身（State、四节点、条件边、循环）；任务 7 只有 10 分，接口最后做。
2. 必须有一个明确的验收标准。每个任务 "算做完" 的定义要在动手前写清楚：例如任务 2 的完成标准是 "文档入库后检索能命中"，任务 6 的完成标准是 "命中 / 不命中问题各跑一次并说明图走向"。

技术栈随之确定：LangGraph（状态图编排）+langchain_milvus.Milvus（向量库）+Milvus Lite（本地文件库，免部署）+FastAPI（接口）+ 硅基流动的 Embedding API + Moonshot 的 Kimi 模型（OpenAI 兼容协议）。

经验：拆题不是列清单，而是把 "模糊的考试要求" 翻译成 "可执行的验收门"。每个任务配一个 "怎么算完成"，后面每步都在往这些门上撞。

二、风险前置：写主代码前的 30 分钟验证

从零开始的优势是没有历史包袱，但风险一点不比改造少 —— 尤其是外部依赖。动手前先用最小脚本验证三类风险，能避免写到一半返工：

1. 向量库链路：langchain_milvus.Milvus在当前依赖版本（langchain-milvus 0.4.0 + pymilvus 3.0.1）下能否建库、写入、检索。注意：这个库的历史版本存在兼容性 bug，必须先实测。
2. Embedding API：硅基流动的Qwen/Qwen3-Embedding-0.6B是否可用、返回维度是否正常。
3. LLM API：Moonshot 的kimi-k2.7-code是否可调用，参数约束是什么。

这一轮验证会提前暴露大量坑，我在本项目里踩到的三个都在这里：

• Kimi 只允许temperature=1：传 0 会直接返回 400 invalid temperature。所有模型初始化必须用默认温度。
• Moonshot 组织级 RPM 上限约 3 次 / 分钟：图里 retrieve→grade→generate 连续调用必然撞 429。解决方案是写一个统一的safe_invoke封装：同进程内两次调用至少间隔 2 秒 + 遇到 429 做 1/2/4/8/16 秒指数退避。
• Milvus Lite 是单文件库、单进程锁：服务运行期间另开进程访问同一个.db文件会报DataDirLockedError。这是设计限制，不是 bug，验证时要记住先确认没有残留进程占用。

def safe_invoke(model, *args, max_retries: int = 5, **kwargs): """统一限速 + 指数退避重试，图内所有 LLM 调用必须走它""" for attempt in range(max_retries + 1): try: with _call_lock: wait = _MIN_CALL_INTERVAL - (time.time() - _last_call_time) if wait > 0: time.sleep(wait) result = model.invoke(*args, **kwargs) _last_call_time = time.time() return result except RateLimitError: if attempt == max_retries: raise time.sleep(2 ** attempt) # 1s, 2s, 4s, 8s, 16s

经验：外部依赖的风险前置验证，30 分钟能省下半天返工。别急着写主代码，先让最小链路 "写入→检索→调用" 跑通。

三、离线入库：父子块切分与向量化（任务 1+2）

这是整个系统的数据地基。题目要求 "文档加载与切分"，但切分策略直接决定检索质量。本项目采用父子块结构：先按页切、再按正则标题切出 "父块"（完整条款），再对父块二次切出 "子块"（喂给向量检索），检索时 "子检父回"—— 用子块做相似度匹配，命中后返回完整父块给大模型。

切分流水线：

1. 按页加载：PyPDFLoader逐页读取。
2. 正则匹配标题切父块：用^\s*第[一二三四五六七八九十百0-9]+条匹配每页的 "第 X 条"，把相邻标题之间的内容合并成一个父块（一条完整的制度条款）。
3. 跨页续接：某条内容跨页时，上一页的残留内容自动并入下一个父块；首页的文档名、章名并入第一条块。
4. 二次切子块：对每个父块用RecursiveCharacterTextSplitter按 500/50 切分，子块携带parent_id、parent_text、page、document元数据。
5. 入库：Milvus.add_documents写入集合company_milvus。

TITLE_PATTERN = re.compile(r"^\s*第[一二三四五六七八九十百0-9]+条", re.MULTILINE) CHILD_SPLITTER = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] ) # 每个父块（一条完整条款）→ 拆成若干子块 for idx, (parent_text, parent_page) in enumerate(parents, 1): parent_id = f"parent_{idx}" for child_text in CHILD_SPLITTER.split_text(parent_text): child_docs.append(Document( page_content=child_text, metadata={ "parent_id": parent_id, "parent_text": parent_text, # 检索命中后返回完整父块 "page": parent_page, "document": document_name, }, ))

检索端实现 "子检父回"：用子块向量做 top_k 召回，按parent_id去重，把命中的子块替换成完整父块文本返回。

class ParentChildRetriever: def invoke(self, query: str) -> list: child_docs = base_retriever.invoke(query) seen, parents = set(), [] for d in child_docs: pid = d.metadata.get("parent_id") if not pid or pid in seen: continue seen.add(pid) parents.append(Document( page_content=d.metadata.get("parent_text") or d.page_content, metadata={"parent_id": pid, "page": d.metadata.get("page"), "document": d.metadata.get("document")}, )) return parents

实测效果：一份 3 页的《员工守则》被切成16 个父块（对应 16 条制度）→ 16 个子块。因为每条制度本身较短（500 字内），父块没有发生二次切分；若遇到长条款，子块数量会大于父块数。

经验："怎么切" 比 "切多少" 重要。父子块的关键收益是 —— 检索用小块保证命中率，生成用大块保证上下文完整，这正是 "子检父回" 的意义。

四、在线问答：LangGraph 状态机（任务 3~6）

这是 60 分的核心。用StateGraph把 RAG 流程编排成一张可循环的状态图。

4.1 定义 State（任务 3）

class AgentState(TypedDict): question: str # 当前问题（rewrite 节点会改写它） documents: List[Document] # 检索到的文档片段 messages: Annotated[list, add_messages] # 对话消息（add_messages 累加） generation: str # 最终答案 iterations: int # 检索次数（循环终止条件） relevant: bool # 相关性判断结果

两个采分点：messages必须用Annotated[list, add_messages]做 reducer（LangGraph 会自动合并新旧消息），iterations是循环终止的关键计数。

4.2 四个节点（任务 4）

• retrieve：调用 retriever 检索，iterations + 1，记录命中条数。
• grade：用 LLM 判断检索片段是否与问题相关。注意 ——题目里的 "分数" 是步骤分，不是让 agent 打分，所以 grade 只输出 "相关 / 不相关 + 理由"，不做任何评分。

class GradeOutput(BaseModel): relevant: bool = Field(description="检索片段是否与用户问题相关") reason: str = Field(description="判断理由") # with_structured_output 强制输出 JSON 结构 model = get_chat_model().with_structured_output(GradeOutput) result = safe_invoke(model, f"{GRADE_PROMPT}\n\n用户问题：{question}\n\n检索片段：\n{docs_text}")

• rewrite：检索不相关时，让 LLM 结合已有片段把问题改写成更贴近文档表述的新查询，更新state["question"]后重新检索。
• generate：基于最终片段生成答案，系统提示词里硬性要求：不编造、标注来源（文档名 第 X 页）、检索为空时如实告知。

4.3 条件边与循环（任务 5）

def decide(state: AgentState) -> Literal["generate", "rewrite"]: if state.get("relevant", False): return "generate" if state.get("iterations", 0) < MAX_ITERATIONS: # MAX_ITERATIONS = 3 return "rewrite" return "generate" # 兜底：超上限直接生成，避免死循环 builder.add_edge(START, "retrieve") builder.add_edge("retrieve", "grade") builder.add_edge("rewrite", "retrieve") # 重写 → 重检索循环 builder.add_edge("generate", END) builder.add_conditional_edges("grade", decide, { "generate": "generate", # 相关 → 生成 "rewrite": "rewrite", # 不相关且未超上限 → 重写重检 })

这张图的核心价值在于：它不只是 "检索→回答" 的流水线，而是一个带自我纠错机制的循环—— 检索不相关就重写查询再试，最多 3 次，超限兜底进入生成。这大幅降低了大模型基于无关片段编造答案（幻觉）的概率。

4.4 编译与验证（任务 6）

builder.compile()后，用两个极端问题各跑一次并记录图走向：

• 命中问题："员工费用报销需要什么流程" → 第 1 次检索判相关 → 引用第 12 条父块生成答案，1 次迭代结束。
• 不命中问题："公司食堂中午吃什么" → 3 次检索全部判不相关 → 重写 2 次后达到上限 → 兜底进入 generate，如实回答 "知识库中未检索到相关内容"，3 次迭代结束。

这两条运行记录就是任务 6 的 10 分交付物 —— 它证明了循环、终止机制和兜底逻辑都真实工作。

五、接口与验收（任务 7）

最后用 FastAPI 把图包成 HTTP 接口：

• POST /ask：接收{"question": "..."}，返回{"answer": "...", "iterations": n}。
• POST /api/upload：multipart 上传 PDF，校验格式与大小后自动完成 "加载→切分→入库"。
• GET /api/chat：兼容旧接口的 GET 问答。
• 自带 Swagger/docs，可直接在浏览器调试。

验证方式：uvicorn启动后 curl 实测 ——/ask返回 200 且答案正确、上传真实 PDF 成功返回入库统计（3 页 → 16 父块 → 16 子块）、上传非 PDF 返回 400。

六、踩坑清单（全文最值钱的部分）

表格

七、结语

回头总结这套 "从零开发" 的方法论，其实只有六步：

拆题算分 → 验证风险 → 入库先行 → 图为核心 → 接口收尾 → 双向验证。

• 拆题定栈：把 100 分翻译成验收门，分值即优先级；
• 风险前置：外部依赖先跑最小脚本，坑提前踩；
• 入库先行：父子块切分 + 子检父回，先把数据地基打牢并验证检索命中；
• 图为核心：State、四节点、条件边、循环、兜底，一个都不能少；
• 接口收尾：FastAPI 三件套，10 分任务最后做；
• 双向验证：命中 / 不命中各跑一次，记录图走向，文档闭环。