尧图网站建设 
简介本资源是一套面向AI开发者与大模型应用实践者的完整本地化RAGAgent系统实现方案聚焦于轻量级、可快速部署的交互式问答应用开发。项目基于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人学习下载读者可直接运行获得可交互的本地RAGAgent原型系统掌握LangGraph状态管理、RAG流程编排、Streamlit多页面路由设计等关键实践能力。1. 为什么本地 RAG Agent 不再是“玩具级”方案LangGraph 编排 Streamlit 快速交付真能跑通完整闭环你手头有一堆 PDF、Word、内部文档想让大模型“真正懂业务”而不是泛泛而谈你试过 LangChain 的 Chain但一加记忆、一加工具调用就崩debug 日志像黑匣子你搭过 RAG但用户问“上季度华东区销售额同比变化”模型却只翻出销售制度文件——不是没检索到而是检索结果和生成逻辑断层了。这正是当前 RAG 瓶颈的典型表现检索与生成强耦合、状态不可控、多步决策无法追踪。而 LangGraph 的出现把 RAG 和 Agent 从“函数调用链”升级为“可观察、可中断、可回溯的状态机”。它不替代 LLM而是给 LLM 加上流程脑、记忆体和工具手Streamlit 则把这套逻辑变成一个按钮就能跑、改两行代码就能换模型、非工程师也能调试的界面。这不是 demo是我在三个客户现场落地的真实路径用 Ollama 拉起 Qwen2-7BMac M2 Pro 16GB 内存下全程本地运行接入 ChromaDB 向量库通过 LangGraph 定义“检索→重写→验证→生成→反思”五步循环最后用 Streamlit 封装成带 trace 可视化、支持上传新文档、实时查看每步中间结果的 Web 应用。适合正在被“RAG 结果不准”、“Agent 一跑就死”、“本地部署卡在环境配置”折磨的算法工程师、AI 产品负责人和懂 Python 的业务分析师。2. 从零构建LangGraph 编排核心——定义节点、边与状态管理LangGraph 的本质是将 LLM 调用、工具执行、条件判断封装为图中的节点Node再用边Edge定义它们之间的流转逻辑。它不依赖外部调度器所有状态都保存在一个可序列化的 State 对象中。这种设计让“RAGAgent”不再是线性 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(modelnomic-embed-text) # 本地轻量嵌入模型 vectorstore Chroma( persist_directory./chroma_db, embedding_functionembeddings ) 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(modelqwen2:7b, temperature0) 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(modelqwen2:7b, temperature0.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(modelnomic-embed-text)这是目前本地部署最轻量、效果尚可的开源嵌入模型100MB比all-MiniLM-L6-v2在中文长文本上更稳定Chroma(persist_directory./chroma_db)Chroma 默认使用内存数据库加persist_directory才能真正持久化否则重启应用数据全丢ChatOllama(modelqwen2: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节点传入初始GraphStateadd_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 初始化与状态同步让前端和后端共享同一份 GraphStateStreamlit 的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_count0, is_validFalse, step_log[] ) # 创建带 checkpoint 的 app支持断点续跑 app_with_mem workflow.compile(checkpointercheckpointer)注意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( questionquestion, documents[], generation, search_queryquestion, # 初始 query 就是用户问题 retry_count0, is_validFalse, step_log[] ) # 调用图启用流式 for event in app_with_mem.stream( initial_state, config{configurable: {thread_id: 1}}, # thread_id 是 checkpoint key stream_modevalues # 返回每次更新后的完整 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), expandedTrue): 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(请输入您的问题, valueA产品和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_modevalues是关键参数它让每次 yield 都返回完整的GraphState而不是只返回变更字段。这样你才能在前端拿到step_log并渲染config{configurable: {thread_id: 1}}中的thread_id必须唯一且稳定比如用用户 ID 哈希否则 checkpoint 会混乱st.expander(..., expandedTrue)让最新步骤默认展开用户无需手动点开就能看到实时进展这是调试体验的核心提升。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_filesTrue ) 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_size300, overlap50 text_splitter RecursiveCharacterTextSplitter( chunk_size300, chunk_overlap50, separators[\n\n, \n, 。, , , , , 、] ) documents.extend(text_splitter.split_documents(docs)) # 写入 Chroma embeddings OllamaEmbeddings(modelnomic-embed-text) vectorstore Chroma( persist_directory./chroma_db, embedding_functionembeddings ) vectorstore.add_documents(documents) st.success(f✅ 已添加 {len(documents)} 个文本块到知识库) st.toast(知识库已更新下次提问将自动使用新内容)玄学参数chunk_size300是针对中文文档的实测最优值。太大如 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.port8501 --server.headlesstrue --server.enableCORSfalse并在代码中用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_functionembeddings, collection_metadata{hnsw:space: cosine} # 显式指定距离空间 ) # 在初始化后立即调用一次 dummy 查询触发索引构建 _ vectorstore.similarity_search(dummy, k1)4.3 现象Qwen2-7B 本地推理时显存爆满即使 M2 Ultra 64GB原因Ollama 默认启用num_gpu1但 Apple Silicon 的 GPU 内存管理与 CUDA 不同num_gpu1实际占用全部 Unified Memory且未启用量化。解决拉取已量化模型 显式限制内存ollama pull qwen2:7b-q4_k_m # 4-bit 量化版本体积 4GBM2 Pro 可流畅运行 ollama run qwen2:7b-q4_k_m -p num_gpu0 # 强制 CPU 推理更稳并在ChatOllama初始化时加model_kwargs{num_gpu: 0}。4.4 现象LangGraph 图执行到rewrite节点后search_query字段被清空或覆盖为 None原因GraphState是TypedDictPython 的dict.update()在浅拷贝时会丢失类型注解导致字段被设为None。LangGraph 内部状态合并逻辑对此敏感。解决永远用copy.deepcopy()初始化 state或改用pydantic.BaseModel定义 state推荐from pydantic import BaseModel class GraphState(BaseModel): question: str documents: List[Document] Field(default_factorylist) # ... 其他字段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_contentd[content], metadatad[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(修正后的检索语句, valuest.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_modevalues ): 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成本高、精度低而是OCR 提取文字用pymupdffitz高效提取 PDF 图片区域文字CLIP 提取视觉语义用clip-interrogator生成图片描述caption双通道向量化将 OCR 文本和 CLIP caption 拼接一起送入nomic-embed-text检索时融合用户问图相关问题同时用文本 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(dpi150, clipfitz.Rect(*bbox)) img Image.frombytes(RGB, [pix.width, pix.height], pix.samples) # OCR import pytesseract ocr_text pytesseract.image_to_string(img, langchi_sim) # CLIP caption ci_config Config(clip_model_nameViT-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: fOCR:{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_contentcontent[combined_text], metadata{source: file_path, page: page_num, type: image_caption} ) documents.append(doc_obj) return documents为什么比直接喂图强成本CLIP caption 生成 1sOCR 3s而多模态 LLM如 Qwen-VL单图推理需 20s可控caption 和 OCR 可人工审核、清洗、增强RAG 友好文本向量检索成熟稳定图像向量检索如 CLIP embedding在长尾 query 上召回率波动大合规不上传用户图片到云端全部本地处理。5.3 我的习惯用langgraph.checkpoint.sqlite替代MemorySaver并每日备份MemorySaver适合开发但上线第一天就被客户问“昨天张经理问的问题今天还能查吗”——这逼我立刻切到 SQLite checkpointfrom langgraph.checkpoint.sqlite import SqliteSaver # 初始化 saver SqliteSaver.from_uri(checkpoint.db) # 编译时传入 app_with_saver workflow.compile(checkpointersaver) # 每日备份脚本crontab # 0 2 * * * cp checkpoint.db checkpoint_$(date \%Y\%m\%d).dbSQLite checkpoint 自动建表、自动迁移checkpoint.db文件就是你的对话历史数据库SELECT * FROM checkpoints ORDER BY thread_id就能看到所有会话。这比任何日志都真实。希望帮到你。本文还有配套的精品资源点击获取 LangGraph+Streamlit构建可调试本地RAG Agent闭环
简介本资源是一套面向AI开发者与大模型应用实践者的完整本地化RAGAgent系统实现方案聚焦于轻量级、可快速部署的交互式问答应用开发。项目基于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人学习下载读者可直接运行获得可交互的本地RAGAgent原型系统掌握LangGraph状态管理、RAG流程编排、Streamlit多页面路由设计等关键实践能力。1. 为什么本地 RAG Agent 不再是“玩具级”方案LangGraph 编排 Streamlit 快速交付真能跑通完整闭环你手头有一堆 PDF、Word、内部文档想让大模型“真正懂业务”而不是泛泛而谈你试过 LangChain 的 Chain但一加记忆、一加工具调用就崩debug 日志像黑匣子你搭过 RAG但用户问“上季度华东区销售额同比变化”模型却只翻出销售制度文件——不是没检索到而是检索结果和生成逻辑断层了。这正是当前 RAG 瓶颈的典型表现检索与生成强耦合、状态不可控、多步决策无法追踪。而 LangGraph 的出现把 RAG 和 Agent 从“函数调用链”升级为“可观察、可中断、可回溯的状态机”。它不替代 LLM而是给 LLM 加上流程脑、记忆体和工具手Streamlit 则把这套逻辑变成一个按钮就能跑、改两行代码就能换模型、非工程师也能调试的界面。这不是 demo是我在三个客户现场落地的真实路径用 Ollama 拉起 Qwen2-7BMac M2 Pro 16GB 内存下全程本地运行接入 ChromaDB 向量库通过 LangGraph 定义“检索→重写→验证→生成→反思”五步循环最后用 Streamlit 封装成带 trace 可视化、支持上传新文档、实时查看每步中间结果的 Web 应用。适合正在被“RAG 结果不准”、“Agent 一跑就死”、“本地部署卡在环境配置”折磨的算法工程师、AI 产品负责人和懂 Python 的业务分析师。2. 从零构建LangGraph 编排核心——定义节点、边与状态管理LangGraph 的本质是将 LLM 调用、工具执行、条件判断封装为图中的节点Node再用边Edge定义它们之间的流转逻辑。它不依赖外部调度器所有状态都保存在一个可序列化的 State 对象中。这种设计让“RAGAgent”不再是线性 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(modelnomic-embed-text) # 本地轻量嵌入模型 vectorstore Chroma( persist_directory./chroma_db, embedding_functionembeddings ) 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(modelqwen2:7b, temperature0) 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(modelqwen2:7b, temperature0.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(modelnomic-embed-text)这是目前本地部署最轻量、效果尚可的开源嵌入模型100MB比all-MiniLM-L6-v2在中文长文本上更稳定Chroma(persist_directory./chroma_db)Chroma 默认使用内存数据库加persist_directory才能真正持久化否则重启应用数据全丢ChatOllama(modelqwen2: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节点传入初始GraphStateadd_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 初始化与状态同步让前端和后端共享同一份 GraphStateStreamlit 的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_count0, is_validFalse, step_log[] ) # 创建带 checkpoint 的 app支持断点续跑 app_with_mem workflow.compile(checkpointercheckpointer)注意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( questionquestion, documents[], generation, search_queryquestion, # 初始 query 就是用户问题 retry_count0, is_validFalse, step_log[] ) # 调用图启用流式 for event in app_with_mem.stream( initial_state, config{configurable: {thread_id: 1}}, # thread_id 是 checkpoint key stream_modevalues # 返回每次更新后的完整 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), expandedTrue): 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(请输入您的问题, valueA产品和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_modevalues是关键参数它让每次 yield 都返回完整的GraphState而不是只返回变更字段。这样你才能在前端拿到step_log并渲染config{configurable: {thread_id: 1}}中的thread_id必须唯一且稳定比如用用户 ID 哈希否则 checkpoint 会混乱st.expander(..., expandedTrue)让最新步骤默认展开用户无需手动点开就能看到实时进展这是调试体验的核心提升。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_filesTrue ) 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_size300, overlap50 text_splitter RecursiveCharacterTextSplitter( chunk_size300, chunk_overlap50, separators[\n\n, \n, 。, , , , , 、] ) documents.extend(text_splitter.split_documents(docs)) # 写入 Chroma embeddings OllamaEmbeddings(modelnomic-embed-text) vectorstore Chroma( persist_directory./chroma_db, embedding_functionembeddings ) vectorstore.add_documents(documents) st.success(f✅ 已添加 {len(documents)} 个文本块到知识库) st.toast(知识库已更新下次提问将自动使用新内容)玄学参数chunk_size300是针对中文文档的实测最优值。太大如 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.port8501 --server.headlesstrue --server.enableCORSfalse并在代码中用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_functionembeddings, collection_metadata{hnsw:space: cosine} # 显式指定距离空间 ) # 在初始化后立即调用一次 dummy 查询触发索引构建 _ vectorstore.similarity_search(dummy, k1)4.3 现象Qwen2-7B 本地推理时显存爆满即使 M2 Ultra 64GB原因Ollama 默认启用num_gpu1但 Apple Silicon 的 GPU 内存管理与 CUDA 不同num_gpu1实际占用全部 Unified Memory且未启用量化。解决拉取已量化模型 显式限制内存ollama pull qwen2:7b-q4_k_m # 4-bit 量化版本体积 4GBM2 Pro 可流畅运行 ollama run qwen2:7b-q4_k_m -p num_gpu0 # 强制 CPU 推理更稳并在ChatOllama初始化时加model_kwargs{num_gpu: 0}。4.4 现象LangGraph 图执行到rewrite节点后search_query字段被清空或覆盖为 None原因GraphState是TypedDictPython 的dict.update()在浅拷贝时会丢失类型注解导致字段被设为None。LangGraph 内部状态合并逻辑对此敏感。解决永远用copy.deepcopy()初始化 state或改用pydantic.BaseModel定义 state推荐from pydantic import BaseModel class GraphState(BaseModel): question: str documents: List[Document] Field(default_factorylist) # ... 其他字段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_contentd[content], metadatad[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(修正后的检索语句, valuest.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_modevalues ): 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成本高、精度低而是OCR 提取文字用pymupdffitz高效提取 PDF 图片区域文字CLIP 提取视觉语义用clip-interrogator生成图片描述caption双通道向量化将 OCR 文本和 CLIP caption 拼接一起送入nomic-embed-text检索时融合用户问图相关问题同时用文本 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(dpi150, clipfitz.Rect(*bbox)) img Image.frombytes(RGB, [pix.width, pix.height], pix.samples) # OCR import pytesseract ocr_text pytesseract.image_to_string(img, langchi_sim) # CLIP caption ci_config Config(clip_model_nameViT-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: fOCR:{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_contentcontent[combined_text], metadata{source: file_path, page: page_num, type: image_caption} ) documents.append(doc_obj) return documents为什么比直接喂图强成本CLIP caption 生成 1sOCR 3s而多模态 LLM如 Qwen-VL单图推理需 20s可控caption 和 OCR 可人工审核、清洗、增强RAG 友好文本向量检索成熟稳定图像向量检索如 CLIP embedding在长尾 query 上召回率波动大合规不上传用户图片到云端全部本地处理。5.3 我的习惯用langgraph.checkpoint.sqlite替代MemorySaver并每日备份MemorySaver适合开发但上线第一天就被客户问“昨天张经理问的问题今天还能查吗”——这逼我立刻切到 SQLite checkpointfrom langgraph.checkpoint.sqlite import SqliteSaver # 初始化 saver SqliteSaver.from_uri(checkpoint.db) # 编译时传入 app_with_saver workflow.compile(checkpointersaver) # 每日备份脚本crontab # 0 2 * * * cp checkpoint.db checkpoint_$(date \%Y\%m\%d).dbSQLite checkpoint 自动建表、自动迁移checkpoint.db文件就是你的对话历史数据库SELECT * FROM checkpoints ORDER BY thread_id就能看到所有会话。这比任何日志都真实。希望帮到你。本文还有配套的精品资源点击获取