尧图网站建设 
一张月考题7 个任务100 分。没有历史包袱从零开始如何在一天内交出一个可运行的「基于 LangGraph 的智能文档问答系统RAG」本文按真实开发顺序复盘全流程拆题 → 验证风险 → 父子块入库 → 状态机问答 → 接口与验收并附上所有踩过的坑。一、拆题表格任务内容分值任务 1文档加载与切分15任务 2向量化并写入 Milvus15任务 3定义 State15任务 4实现四个节点15任务 5条件边与循环20任务 6编译与验证10任务 7FastAPI 进阶10拆题后立刻能得到两个关键结论分值就是优先级。任务 3~6 加起来 60 分核心是 LangGraph 的状态图本身State、四节点、条件边、循环任务 7 只有 10 分接口最后做。必须有一个明确的验收标准。每个任务 算做完 的定义要在动手前写清楚例如任务 2 的完成标准是 文档入库后检索能命中任务 6 的完成标准是 命中 / 不命中问题各跑一次并说明图走向。技术栈随之确定LangGraph状态图编排langchain_milvus.Milvus向量库Milvus Lite本地文件库免部署FastAPI接口 硅基流动的 Embedding API Moonshot 的 Kimi 模型OpenAI 兼容协议。经验拆题不是列清单而是把 模糊的考试要求 翻译成 可执行的验收门。每个任务配一个 怎么算完成后面每步都在往这些门上撞。二、风险前置写主代码前的 30 分钟验证从零开始的优势是没有历史包袱但风险一点不比改造少 —— 尤其是外部依赖。动手前先用最小脚本验证三类风险能避免写到一半返工向量库链路langchain_milvus.Milvus在当前依赖版本langchain-milvus 0.4.0 pymilvus 3.0.1下能否建库、写入、检索。注意这个库的历史版本存在兼容性 bug必须先实测。Embedding API硅基流动的Qwen/Qwen3-Embedding-0.6B是否可用、返回维度是否正常。LLM APIMoonshot 的kimi-k2.7-code是否可调用参数约束是什么。这一轮验证会提前暴露大量坑我在本项目里踩到的三个都在这里Kimi 只允许temperature1传 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 分钟能省下半天返工。别急着写主代码先让最小链路 写入→检索→调用 跑通。三、离线入库父子块切分与向量化任务 12这是整个系统的数据地基。题目要求 文档加载与切分但切分策略直接决定检索质量。本项目采用父子块结构先按页切、再按正则标题切出 父块完整条款再对父块二次切出 子块喂给向量检索检索时 子检父回—— 用子块做相似度匹配命中后返回完整父块给大模型。切分流水线按页加载PyPDFLoader逐页读取。正则匹配标题切父块用^\s*第[一二三四五六七八九十百0-9]条匹配每页的 第 X 条把相邻标题之间的内容合并成一个父块一条完整的制度条款。跨页续接某条内容跨页时上一页的残留内容自动并入下一个父块首页的文档名、章名并入第一条块。二次切子块对每个父块用RecursiveCharacterTextSplitter按 500/50 切分子块携带parent_id、parent_text、page、document元数据。入库Milvus.add_documents写入集合company_milvus。TITLE_PATTERN re.compile(r^\s*第[一二三四五六七八九十百0-9]条, re.MULTILINE) CHILD_SPLITTER RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , , ] ) # 每个父块一条完整条款→ 拆成若干子块 for idx, (parent_text, parent_page) in enumerate(parents, 1): parent_id fparent_{idx} for child_text in CHILD_SPLITTER.split_text(parent_text): child_docs.append(Document( page_contentchild_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_contentd.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任务 3class 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]做 reducerLangGraph 会自动合并新旧消息iterations是循环终止的关键计数。4.2 四个节点任务 4retrieve调用 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 条件边与循环任务 5def 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 编译与验证任务 6builder.compile()后用两个极端问题各跑一次并记录图走向命中问题员工费用报销需要什么流程 → 第 1 次检索判相关 → 引用第 12 条父块生成答案1 次迭代结束。不命中问题公司食堂中午吃什么 → 3 次检索全部判不相关 → 重写 2 次后达到上限 → 兜底进入 generate如实回答 知识库中未检索到相关内容3 次迭代结束。这两条运行记录就是任务 6 的 10 分交付物 —— 它证明了循环、终止机制和兜底逻辑都真实工作。五、接口与验收任务 7最后用 FastAPI 把图包成 HTTP 接口POST /ask接收{question: ...}返回{answer: ..., iterations: n}。POST /api/uploadmultipart 上传 PDF校验格式与大小后自动完成 加载→切分→入库。GET /api/chat兼容旧接口的 GET 问答。自带 Swagger/docs可直接在浏览器调试。验证方式uvicorn启动后 curl 实测 ——/ask返回 200 且答案正确、上传真实 PDF 成功返回入库统计3 页 → 16 父块 → 16 子块、上传非 PDF 返回 400。六、踩坑清单全文最值钱的部分表格坑现象解法Kimi 温度限制temperature0报 400一律用默认temperature1Moonshot RPM 限流连续调用 429safe_invoke2s 间隔 指数退避重试Milvus Lite 单进程锁二次进程访问报DataDirLockedError设计限制验证前确认无残留进程langchain_milvus 版本兼容旧版本写入 / 检索报错实测当前版本0.4.0后再切换PyPDFLoader 页码 0 基元数据page从 0 开始用page_label转成印刷页码循环死锁风险不相关时无限重试MAX_ITERATIONS3decide兜底七、结语回头总结这套 从零开发 的方法论其实只有六步拆题算分 → 验证风险 → 入库先行 → 图为核心 → 接口收尾 → 双向验证。拆题定栈把 100 分翻译成验收门分值即优先级风险前置外部依赖先跑最小脚本坑提前踩入库先行父子块切分 子检父回先把数据地基打牢并验证检索命中图为核心State、四节点、条件边、循环、兜底一个都不能少接口收尾FastAPI 三件套10 分任务最后做双向验证命中 / 不命中各跑一次记录图走向文档闭环。 LangGraph 文档问答 RAG 的完整工程复盘
一张月考题7 个任务100 分。没有历史包袱从零开始如何在一天内交出一个可运行的「基于 LangGraph 的智能文档问答系统RAG」本文按真实开发顺序复盘全流程拆题 → 验证风险 → 父子块入库 → 状态机问答 → 接口与验收并附上所有踩过的坑。一、拆题表格任务内容分值任务 1文档加载与切分15任务 2向量化并写入 Milvus15任务 3定义 State15任务 4实现四个节点15任务 5条件边与循环20任务 6编译与验证10任务 7FastAPI 进阶10拆题后立刻能得到两个关键结论分值就是优先级。任务 3~6 加起来 60 分核心是 LangGraph 的状态图本身State、四节点、条件边、循环任务 7 只有 10 分接口最后做。必须有一个明确的验收标准。每个任务 算做完 的定义要在动手前写清楚例如任务 2 的完成标准是 文档入库后检索能命中任务 6 的完成标准是 命中 / 不命中问题各跑一次并说明图走向。技术栈随之确定LangGraph状态图编排langchain_milvus.Milvus向量库Milvus Lite本地文件库免部署FastAPI接口 硅基流动的 Embedding API Moonshot 的 Kimi 模型OpenAI 兼容协议。经验拆题不是列清单而是把 模糊的考试要求 翻译成 可执行的验收门。每个任务配一个 怎么算完成后面每步都在往这些门上撞。二、风险前置写主代码前的 30 分钟验证从零开始的优势是没有历史包袱但风险一点不比改造少 —— 尤其是外部依赖。动手前先用最小脚本验证三类风险能避免写到一半返工向量库链路langchain_milvus.Milvus在当前依赖版本langchain-milvus 0.4.0 pymilvus 3.0.1下能否建库、写入、检索。注意这个库的历史版本存在兼容性 bug必须先实测。Embedding API硅基流动的Qwen/Qwen3-Embedding-0.6B是否可用、返回维度是否正常。LLM APIMoonshot 的kimi-k2.7-code是否可调用参数约束是什么。这一轮验证会提前暴露大量坑我在本项目里踩到的三个都在这里Kimi 只允许temperature1传 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 分钟能省下半天返工。别急着写主代码先让最小链路 写入→检索→调用 跑通。三、离线入库父子块切分与向量化任务 12这是整个系统的数据地基。题目要求 文档加载与切分但切分策略直接决定检索质量。本项目采用父子块结构先按页切、再按正则标题切出 父块完整条款再对父块二次切出 子块喂给向量检索检索时 子检父回—— 用子块做相似度匹配命中后返回完整父块给大模型。切分流水线按页加载PyPDFLoader逐页读取。正则匹配标题切父块用^\s*第[一二三四五六七八九十百0-9]条匹配每页的 第 X 条把相邻标题之间的内容合并成一个父块一条完整的制度条款。跨页续接某条内容跨页时上一页的残留内容自动并入下一个父块首页的文档名、章名并入第一条块。二次切子块对每个父块用RecursiveCharacterTextSplitter按 500/50 切分子块携带parent_id、parent_text、page、document元数据。入库Milvus.add_documents写入集合company_milvus。TITLE_PATTERN re.compile(r^\s*第[一二三四五六七八九十百0-9]条, re.MULTILINE) CHILD_SPLITTER RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , , ] ) # 每个父块一条完整条款→ 拆成若干子块 for idx, (parent_text, parent_page) in enumerate(parents, 1): parent_id fparent_{idx} for child_text in CHILD_SPLITTER.split_text(parent_text): child_docs.append(Document( page_contentchild_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_contentd.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任务 3class 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]做 reducerLangGraph 会自动合并新旧消息iterations是循环终止的关键计数。4.2 四个节点任务 4retrieve调用 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 条件边与循环任务 5def 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 编译与验证任务 6builder.compile()后用两个极端问题各跑一次并记录图走向命中问题员工费用报销需要什么流程 → 第 1 次检索判相关 → 引用第 12 条父块生成答案1 次迭代结束。不命中问题公司食堂中午吃什么 → 3 次检索全部判不相关 → 重写 2 次后达到上限 → 兜底进入 generate如实回答 知识库中未检索到相关内容3 次迭代结束。这两条运行记录就是任务 6 的 10 分交付物 —— 它证明了循环、终止机制和兜底逻辑都真实工作。五、接口与验收任务 7最后用 FastAPI 把图包成 HTTP 接口POST /ask接收{question: ...}返回{answer: ..., iterations: n}。POST /api/uploadmultipart 上传 PDF校验格式与大小后自动完成 加载→切分→入库。GET /api/chat兼容旧接口的 GET 问答。自带 Swagger/docs可直接在浏览器调试。验证方式uvicorn启动后 curl 实测 ——/ask返回 200 且答案正确、上传真实 PDF 成功返回入库统计3 页 → 16 父块 → 16 子块、上传非 PDF 返回 400。六、踩坑清单全文最值钱的部分表格坑现象解法Kimi 温度限制temperature0报 400一律用默认temperature1Moonshot RPM 限流连续调用 429safe_invoke2s 间隔 指数退避重试Milvus Lite 单进程锁二次进程访问报DataDirLockedError设计限制验证前确认无残留进程langchain_milvus 版本兼容旧版本写入 / 检索报错实测当前版本0.4.0后再切换PyPDFLoader 页码 0 基元数据page从 0 开始用page_label转成印刷页码循环死锁风险不相关时无限重试MAX_ITERATIONS3decide兜底七、结语回头总结这套 从零开发 的方法论其实只有六步拆题算分 → 验证风险 → 入库先行 → 图为核心 → 接口收尾 → 双向验证。拆题定栈把 100 分翻译成验收门分值即优先级风险前置外部依赖先跑最小脚本坑提前踩入库先行父子块切分 子检父回先把数据地基打牢并验证检索命中图为核心State、四节点、条件边、循环、兜底一个都不能少接口收尾FastAPI 三件套10 分任务最后做双向验证命中 / 不命中各跑一次记录图走向文档闭环。