LangGraph+Streamlit构建可调试本地RAG Agent闭环

简介：本资源是一套面向AI开发者与大模型应用实践者的完整本地化RAG+Agent系统实现方案，聚焦于轻量级、可快速部署的交互式问答应用开发。项目基于LangGraph构建状态可控的Agent工作流，结合本地LLM与RAG检索增强机制，并通过Streamlit封装为直观Web界面，适用于智能客服、教育问答、个人知识助手等场景，适合具备Python基础及初步大模型概念的中阶学习者上手实践。压缩包共12个文件（10个Python核心模块、1个依赖说明txt、1个LICENSE），总大小仅12KB，结构精炼：包含naive_rag.py检索逻辑、agent_chat_page.py与rag_chat_page.py双模式页面、tools目录下的工具函数及st_main.py主入口，便于理解模块职责与调用关系。目前已有230人学习下载，读者可直接运行获得可交互的本地RAG+Agent原型系统，掌握LangGraph状态管理、RAG流程编排、Streamlit多页面路由设计等关键实践能力。

1. 为什么本地 RAG + Agent 不再是“玩具级”方案：LangGraph 编排 + Streamlit 快速交付，真能跑通完整闭环？

你手头有一堆 PDF、Word、内部文档，想让大模型“真正懂业务”，而不是泛泛而谈；你试过 LangChain 的 Chain，但一加记忆、一加工具调用就崩，debug 日志像黑匣子；你搭过 RAG，但用户问“上季度华东区销售额同比变化”，模型却只翻出销售制度文件——不是没检索到，而是检索结果和生成逻辑断层了。这正是当前 RAG 瓶颈的典型表现：检索与生成强耦合、状态不可控、多步决策无法追踪。而 LangGraph 的出现，把 RAG 和 Agent 从“函数调用链”升级为“可观察、可中断、可回溯的状态机”。它不替代 LLM，而是给 LLM 加上流程脑、记忆体和工具手；Streamlit 则把这套逻辑变成一个按钮就能跑、改两行代码就能换模型、非工程师也能调试的界面。这不是 demo，是我在三个客户现场落地的真实路径：用 Ollama 拉起 Qwen2-7B（Mac M2 Pro 16GB 内存下全程本地运行），接入 ChromaDB 向量库，通过 LangGraph 定义“检索→重写→验证→生成→反思”五步循环，最后用 Streamlit 封装成带 trace 可视化、支持上传新文档、实时查看每步中间结果的 Web 应用。适合正在被“RAG 结果不准”、“Agent 一跑就死”、“本地部署卡在环境配置”折磨的算法工程师、AI 产品负责人和懂 Python 的业务分析师。

2. 从零构建：LangGraph 编排核心——定义节点、边与状态管理

LangGraph 的本质是将 LLM 调用、工具执行、条件判断封装为图中的节点（Node），再用边（Edge）定义它们之间的流转逻辑。它不依赖外部调度器，所有状态都保存在一个可序列化的 State 对象中。这种设计让“RAG+Agent”不再是线性 pipeline，而是可分支、可循环、可人工干预的有向无环图（DAG）。下面以一个真实业务场景为例：用户提问“请对比 A 产品和 B 产品的售后服务政策差异”，系统需先识别实体（A/B）、检索对应文档、提取关键条款、结构化比对、最后生成表格。这个过程天然包含并行检索（A/B 分别查）、条件跳转（若某产品无政策文档则跳过比对）、失败重试（检索无结果时触发重写查询）——LangGraph 正是为此而生。

2.1 定义 State：让每一步操作都有迹可循

LangGraph 要求你显式声明整个流程中需要维护的所有字段。这不是冗余，而是可调试性的基石。我们定义一个State类，继承自TypedDict，明确标注每个字段的类型和用途：

from typing import TypedDict, List, Optional, Dict, Any from langgraph.graph import StateGraph, END from langchain_core.documents import Document class GraphState(TypedDict): question: str # 用户原始问题 documents: List[Document] # 检索到的文档列表 generation: str # 最终生成的回答 search_query: str # 当前用于检索的查询语句（可能被重写） retry_count: int # 检索失败重试次数 is_valid: bool # 检索结果是否足够支撑回答 step_log: List[Dict[str, Any]] # 每步执行日志，用于 Streamlit 可视化

提示：step_log是关键设计。它不是为了日志而日志，而是为 Streamlit 前端提供实时 trace 数据源。每执行一个节点，就往里 append 一条{ "step": "retrieve", "input": "...", "output_len": 3 }这样的记录。后续 Streamlit 页面直接读取该字段渲染流程图，无需额外 WebSocket 或后端 API。

2.2 构建节点：每个函数都是一个可测试、可替换的原子单元

LangGraph 节点必须是纯函数（pure function）：输入GraphState，输出更新后的GraphState。这意味着你可以单独测试retrieve函数，而不必启动整个图。我们实现四个核心节点：

# retrieval.py from langchain_chroma import Chroma from langchain_community.embeddings import OllamaEmbeddings def retrieve(state: GraphState) -> GraphState: """执行向量检索，返回 top_k 文档""" embeddings = OllamaEmbeddings(model="nomic-embed-text") # 本地轻量嵌入模型 vectorstore = Chroma( persist_directory="./chroma_db", embedding_function=embeddings ) retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) docs = retriever.invoke(state["search_query"]) # 记录日志 state["step_log"].append({ "step": "retrieve", "input": state["search_query"], "output_len": len(docs), "docs_preview": [d.page_content[:50] for d in docs[:2]] }) state["documents"] = docs return state # rewrite.py from langchain_core.prompts import ChatPromptTemplate from langchain_ollama import ChatOllama def rewrite_query(state: GraphState) -> GraphState: """当检索结果不足时，重写原始问题以提升召回率""" llm = ChatOllama(model="qwen2:7b", temperature=0) prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个专业的信息检索优化助手。请根据用户问题和当前检索结果，生成一个更精准、更具体的检索查询语句。只输出查询语句，不要解释。"), ("human", "用户问题：{question}\n当前检索结果摘要：{doc_summary}") ]) doc_summary = "\n".join([d.page_content[:100] for d in state["documents"][:2]]) chain = prompt | llm | (lambda x: x.content.strip()) new_query = chain.invoke({"question": state["question"], "doc_summary": doc_summary}) state["search_query"] = new_query state["retry_count"] += 1 state["step_log"].append({ "step": "rewrite_query", "input": state["question"], "output": new_query }) return state # generate.py def generate_answer(state: GraphState) -> GraphState: """基于检索文档生成最终回答""" llm = ChatOllama(model="qwen2:7b", temperature=0.3) prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个专业客服助手。请严格依据以下检索到的文档内容回答问题，禁止编造。如果文档中无相关信息，请明确告知‘未找到相关信息’。"), ("human", "问题：{question}\n文档：{context}") ]) context = "\n\n".join([d.page_content for d in state["documents"]]) chain = prompt | llm | (lambda x: x.content) answer = chain.invoke({"question": state["question"], "context": context}) state["generation"] = answer state["step_log"].append({ "step": "generate_answer", "input_context_len": len(context), "output": answer[:100] + "..." }) return state # validate.py def validate_retrieval(state: GraphState) -> str: """判断检索结果是否有效，决定下一步流向""" if len(state["documents"]) == 0 or state["retry_count"] >= 2: return "rewrite" elif len(state["documents"]) >= 3: return "generate" else: return "rewrite"

参数说明：

• OllamaEmbeddings(model="nomic-embed-text")：这是目前本地部署最轻量、效果尚可的开源嵌入模型（<100MB），比all-MiniLM-L6-v2在中文长文本上更稳定；
• Chroma(persist_directory="./chroma_db")：Chroma 默认使用内存数据库，加persist_directory才能真正持久化，否则重启应用数据全丢；
• ChatOllama(model="qwen2:7b")：Qwen2-7B 是当前本地推理性价比最高的中文模型之一（M2 Pro 上推理速度约 8 token/s），比 Llama3-8B 中文弱但更稳；
• validate_retrieval返回字符串"rewrite"或"generate"，这是 LangGraph 边路由的关键——它不返回GraphState，而是告诉图“下一步去哪个节点”。

2.3 组装图：用 add_node / add_edge 构建可执行流程

LangGraph 图的组装非常直白，没有魔法。add_node注册函数，add_edge定义无条件流转，add_conditional_edges定义分支逻辑：

from langgraph.graph import StateGraph, END workflow = StateGraph(GraphState) # 注册所有节点 workflow.add_node("retrieve", retrieve) workflow.add_node("rewrite", rewrite_query) workflow.add_node("generate", generate_answer) # 设置入口：从用户问题开始，先生成初始检索 query workflow.set_entry_point("retrieve") # 定义边：retrieve → 根据 validate_retrieval 判断去 rewrite 还是 generate workflow.add_conditional_edges( "retrieve", validate_retrieval, { "rewrite": "rewrite", "generate": "generate" } ) # rewrite 后必须回到 retrieve（重试检索） workflow.add_edge("rewrite", "retrieve") # generate 后结束 workflow.add_edge("generate", END) # 编译图，得到可调用的 app app = workflow.compile()

关键逻辑说明：

• set_entry_point("retrieve")表示图启动时自动调用retrieve节点，传入初始GraphState；
• add_conditional_edges的第三个参数是字典，key 是validate_retrieval函数的返回值，value 是目标节点名；
• add_edge("rewrite", "retrieve")实现了“重写→重检”的循环，但 LangGraph 会自动检测循环深度（默认 25 层），避免无限递归；
• app.compile()返回的是一个CompiledGraph对象，它支持.invoke()（单次执行）、.stream()（流式输出）、.get_graph().draw_mermaid_png()（导出流程图）等方法，这才是真正可交付的“Agent 引擎”。

3. Streamlit 封装：不只是 UI，更是调试器与交付界面

Streamlit 的优势在于“写 Python 就是写前端”。它不强制你学 React/Vue，也不要求你部署 Nginx，一个streamlit run app.py就能生成带交互、带状态、带实时日志的 Web 页面。更重要的是，Streamlit 天然适配 LangGraph 的stream()方法——你可以逐帧获取图中每个节点的输出，并实时渲染到页面上，这比任何 debug 工具都直观。

3.1 初始化与状态同步：让前端和后端共享同一份 GraphState

Streamlit 的st.session_state是跨请求保持状态的唯一可靠方式。我们将 LangGraph 的GraphState映射到st.session_state，确保每次用户提问、每次点击“重试”，后端图执行的输入都来自前端最新状态：

import streamlit as st from langgraph.checkpoint.memory import MemorySaver # 初始化 checkpoint，支持对话历史记忆（非必须，但强烈建议） checkpointer = MemorySaver() # 从 session_state 获取或初始化 state if "messages" not in st.session_state: st.session_state.messages = [] if "graph_state" not in st.session_state: st.session_state.graph_state = GraphState( question="", documents=[], generation="", search_query="", retry_count=0, is_valid=False, step_log=[] ) # 创建带 checkpoint 的 app（支持断点续跑） app_with_mem = workflow.compile(checkpointer=checkpointer)

注意：MemorySaver()是 LangGraph 内置的内存型 checkpoint，适合开发调试。生产环境应换为PostgresSaver或MongoDBSaver，否则重启服务后所有对话历史丢失。

3.2 流式渲染：用 st.empty() 实现节点级 trace 可视化

LangGraph 的.stream()方法返回一个生成器，每次 yield 一个(node_name, state_update)元组。我们用st.empty()占位符逐帧更新 UI，让用户亲眼看到“检索→重写→再检索→生成”的全过程：

def run_graph_stream(question: str): """执行图并流式更新 UI""" # 清空旧日志 st.session_state.graph_state["step_log"] = [] # 构建初始 state initial_state = GraphState( question=question, documents=[], generation="", search_query=question, # 初始 query 就是用户问题 retry_count=0, is_valid=False, step_log=[] ) # 调用图，启用流式 for event in app_with_mem.stream( initial_state, config={"configurable": {"thread_id": "1"}}, # thread_id 是 checkpoint key stream_mode="values" # 返回每次更新后的完整 state ): # 更新 session_state st.session_state.graph_state = event # 渲染当前步骤日志 if event["step_log"]: latest_log = event["step_log"][-1] with st.expander(f"🔍 {latest_log['step']} ({len(event['step_log'])} steps)", expanded=True): st.write(f"**Input**: {latest_log.get('input', 'N/A')}") if "output" in latest_log: st.write(f"**Output**: {latest_log['output'][:200]}...") if "docs_preview" in latest_log: st.write(f"**Retrieved Docs Preview**: {latest_log['docs_preview']}") # 主界面 st.title("🔍 本地 RAG + Agent 调试台") question = st.text_input("请输入您的问题：", value="A产品和B产品的售后服务政策有什么差异？") if st.button("🚀 开始执行"): if question.strip(): with st.spinner("Agent 正在思考中..."): run_graph_stream(question) else: st.warning("请输入有效问题") # 渲染最终答案 if st.session_state.graph_state["generation"]: st.subheader("✅ 最终回答") st.markdown(st.session_state.graph_state["generation"])

血泪经验：

• stream_mode="values"是关键参数，它让每次 yield 都返回完整的GraphState，而不是只返回变更字段。这样你才能在前端拿到step_log并渲染；
• config={"configurable": {"thread_id": "1"}}中的thread_id必须唯一且稳定（比如用用户 ID 哈希），否则 checkpoint 会混乱；
• st.expander(..., expanded=True)让最新步骤默认展开，用户无需手动点开就能看到实时进展，这是调试体验的核心提升。

3.3 文件上传与知识库热更新：让 RAG 真正“活”起来

RAG 的价值在于知识可更新。Streamlit 的st.file_uploader支持多文件、拖拽上传，配合 Chroma 的add_documents()，即可实现“上传即生效”：

st.subheader("📚 知识库管理") uploaded_files = st.file_uploader( "上传 PDF/DOCX/TXT 文件（支持批量）", type=["pdf", "docx", "txt"], accept_multiple_files=True ) if uploaded_files and st.button("🔄 添加到知识库"): from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter documents = [] for file in uploaded_files: if file.name.endswith(".pdf"): loader = PyPDFLoader(file) elif file.name.endswith(".docx"): loader = Docx2txtLoader(file) else: loader = TextLoader(file) docs = loader.load() # 分块：按中文习惯设 chunk_size=300, overlap=50 text_splitter = RecursiveCharacterTextSplitter( chunk_size=300, chunk_overlap=50, separators=["\n\n", "\n", "。", "！", "？", "；", "，", "、"] ) documents.extend(text_splitter.split_documents(docs)) # 写入 Chroma embeddings = OllamaEmbeddings(model="nomic-embed-text") vectorstore = Chroma( persist_directory="./chroma_db", embedding_function=embeddings ) vectorstore.add_documents(documents) st.success(f"✅ 已添加 {len(documents)} 个文本块到知识库") st.toast("知识库已更新，下次提问将自动使用新内容！")

玄学参数：

• chunk_size=300是针对中文文档的实测最优值。太大（如 1000）会导致单块信息过载，LLM 无法聚焦；太小（如 100）则割裂语义，检索召回率下降；
• separators显式指定中文标点，比默认英文分隔符更准——这是很多 RAG 项目翻车的隐形坑；
• st.toast()提供瞬时反馈，比st.success()更轻量，用户不会觉得页面被刷新打断。

4. 避坑指南：LangGraph + Streamlit 本地部署的 5 个真实翻车现场

LangGraph 和 Streamlit 都是“简单上手，深水踩坑”。以下是我在三个客户现场亲手填平的坑，每一条都附带复现方式、根本原因和一行解决命令。

4.1 现象：Streamlit 页面卡死，浏览器 console 报WebSocket connection failed

原因：LangGraph 的stream()默认使用异步 generator，而 Streamlit 的st.spinner和st.empty()在同步上下文中调用时，会阻塞事件循环，导致 WebSocket 断连。这不是网络问题，是执行模型错配。
解决：强制 Streamlit 使用线程模式运行，绕过 async 限制：

streamlit run app.py --server.port=8501 --server.headless=true --server.enableCORS=false

并在代码中用threading.Thread包裹app.stream()调用，或直接改用app.invoke()（牺牲流式，换稳定性）。

4.2 现象：Chroma 向量库首次加载极慢（>2分钟），后续查询又很快

原因：Chroma 默认使用hnswlib作为索引后端，首次as_retriever()会触发索引构建（build index），且hnswlib在 Apple Silicon 上编译优化不足。
解决：预构建索引 + 指定更优参数：

vectorstore = Chroma( persist_directory="./chroma_db", embedding_function=embeddings, collection_metadata={"hnsw:space": "cosine"} # 显式指定距离空间 ) # 在初始化后立即调用一次 dummy 查询，触发索引构建 _ = vectorstore.similarity_search("dummy", k=1)

4.3 现象：Qwen2-7B 本地推理时显存爆满（即使 M2 Ultra 64GB）

原因：Ollama 默认启用num_gpu=1，但 Apple Silicon 的 GPU 内存管理与 CUDA 不同，num_gpu=1实际占用全部 Unified Memory，且未启用量化。
解决：拉取已量化模型 + 显式限制内存：

ollama pull qwen2:7b-q4_k_m # 4-bit 量化版本，体积 <4GB，M2 Pro 可流畅运行 ollama run qwen2:7b-q4_k_m -p "num_gpu=0" # 强制 CPU 推理，更稳

并在ChatOllama初始化时加model_kwargs={"num_gpu": 0}。

4.4 现象：LangGraph 图执行到rewrite节点后，search_query字段被清空或覆盖为 None

原因：GraphState是TypedDict，Python 的dict.update()在浅拷贝时会丢失类型注解，导致字段被设为None。LangGraph 内部状态合并逻辑对此敏感。
解决：永远用copy.deepcopy()初始化 state，或改用pydantic.BaseModel定义 state（推荐）：

from pydantic import BaseModel class GraphState(BaseModel): question: str documents: List[Document] = Field(default_factory=list) # ... 其他字段

LangGraph 原生支持 Pydantic 模型，类型安全且无浅拷贝风险。

4.5 现象：Streamlit 页面刷新后，st.session_state中的graph_state丢失，报KeyError: 'step_log'

原因：Streamlit 在页面重载时会重置st.session_state，但GraphState实例不是 JSON serializable（含Document对象），导致st.session_state无法持久化。
解决：不在st.session_state中存Document对象，只存其page_content和元数据：

# 存储时 state["documents"] = [{"content": d.page_content, "metadata": d.metadata} for d in docs] # 使用时重建 docs = [Document(page_content=d["content"], metadata=d["metadata"]) for d in state["documents"]]

这是本地 RAG 必须接受的 trade-off：内存换序列化安全。

5. 进阶技巧：用 LangGraph Checkpoint 实现“可回溯 Agent”，以及如何让 RAG 真正理解图片

LangGraph 的checkpointer不只是存对话历史，它是让 Agent 具备“反思能力”的基础设施。而 RAG 知识库能否存图片？答案是：不能直接存图片，但能存图片的语义描述，且效果远超 raw pixel。这两点，是区分玩具和生产级 RAG 的分水岭。

5.1 Checkpoint + Manual Intervention：让 Agent 在卡点时喊你“救命”

LangGraph 的 checkpoint 机制允许你在任意节点中断执行，并手动修改GraphState后继续。这在调试复杂 RAG 场景时是后悔药。例如，当validate_retrieval判断“检索无效”进入rewrite，但重写后的 query 更差时，你可以在 Streamlit 页面加一个“人工修正 query”输入框：

# 在 Streamlit 页面中 if st.session_state.graph_state["step_log"] and st.session_state.graph_state["step_log"][-1]["step"] == "rewrite": st.info("⚠️ 检索效果不佳，系统已重写 query。您可手动修正：") manual_query = st.text_input("修正后的检索语句：", value=st.session_state.graph_state["search_query"]) if st.button("✅ 强制使用此 query"): st.session_state.graph_state["search_query"] = manual_query # 调用图，从 retrieve 节点重新开始 for event in app_with_mem.stream( st.session_state.graph_state, config={"configurable": {"thread_id": "1"}}, stream_mode="values" ): st.session_state.graph_state = event

底层原理：app_with_mem.stream()的config参数中的thread_id是 checkpoint 的 key。只要thread_id不变，LangGraph 就会从上次中断处恢复状态，并应用你手动修改的search_query。这相当于给 Agent 装了一个“暂停/快进/重拍”按钮。

5.2 RAG 知识库存图片？用 CLIP Embedding + OCR 文本双通道

RAG 知识库本身是文本向量库，无法直接索引图片二进制。但业务中大量 PDF 含图表、扫描件含发票，用户会问“请分析这张图里的数据趋势”。正确解法不是把图片喂给 LLM（成本高、精度低），而是：

1. OCR 提取文字：用pymupdf（fitz）高效提取 PDF 图片区域文字；
2. CLIP 提取视觉语义：用clip-interrogator生成图片描述（caption）；
3. 双通道向量化：将 OCR 文本和 CLIP caption 拼接，一起送入nomic-embed-text；
4. 检索时融合：用户问图相关问题，同时用文本 query 和 CLIP-generated query 检索。

# image_processor.py from PIL import Image import fitz # PyMuPDF from clip_interrogator import Config, Interrogator def extract_image_content(pdf_path: str, page_num: int, bbox: tuple) -> dict: """从 PDF 指定页、指定区域提取图片内容""" doc = fitz.open(pdf_path) page = doc[page_num] pix = page.get_pixmap(dpi=150, clip=fitz.Rect(*bbox)) img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples) # OCR import pytesseract ocr_text = pytesseract.image_to_string(img, lang='chi_sim') # CLIP caption ci_config = Config(clip_model_name="ViT-L-14/openai", cache_path="./clip_cache") ci_config.apply_low_vram_defaults() if not torch.cuda.is_available() else None interrogator = Interrogator(ci_config) caption = interrogator.interrogate(img) return { "ocr_text": ocr_text.strip(), "caption": caption, "combined_text": f"OCR:{ocr_text.strip()} CAPTION:{caption}" } # 在文档加载时调用 def load_pdf_with_images(file_path: str): documents = [] # ... 原有文本加载逻辑 # 额外提取图片区域 for page_num in range(len(doc)): image_list = doc.get_page_images(page_num) for img_info in image_list: xref = img_info[0] base_image = doc.extract_image(xref) # 这里需计算 bbox，略 content = extract_image_content(file_path, page_num, bbox) doc_obj = Document( page_content=content["combined_text"], metadata={"source": file_path, "page": page_num, "type": "image_caption"} ) documents.append(doc_obj) return documents

为什么比直接喂图强？

• 成本：CLIP caption 生成 <1s，OCR <3s；而多模态 LLM（如 Qwen-VL）单图推理需 20s+；
• 可控：caption 和 OCR 可人工审核、清洗、增强；
• RAG 友好：文本向量检索成熟稳定，图像向量检索（如 CLIP embedding）在长尾 query 上召回率波动大；
• 合规：不上传用户图片到云端，全部本地处理。

5.3 我的习惯：用langgraph.checkpoint.sqlite替代MemorySaver，并每日备份

MemorySaver适合开发，但上线第一天就被客户问：“昨天张经理问的问题，今天还能查吗？”——这逼我立刻切到 SQLite checkpoint：

from langgraph.checkpoint.sqlite import SqliteSaver # 初始化 saver = SqliteSaver.from_uri("checkpoint.db") # 编译时传入 app_with_saver = workflow.compile(checkpointer=saver) # 每日备份脚本（crontab） # 0 2 * * * cp checkpoint.db checkpoint_$(date +\%Y\%m\%d).db

SQLite checkpoint 自动建表、自动迁移，checkpoint.db文件就是你的对话历史数据库，SELECT * FROM checkpoints ORDER BY thread_id就能看到所有会话。这比任何日志都真实。

希望帮到你。

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