全栈实战——用 LangGraph 搭建智能 RAG 知识库 Agent——从零开始学 LangGraph（十二）

全栈实战——用 LangGraph 搭建智能 RAG 知识库 Agent写在前面这一期把前面学过的所有知识点——State 管理、Node/Edge 编排、Checkpointer、Human-in-the-Loop、Streaming、Store 长期记忆——整合成一个完整的RAG 知识库问答 Agent。你将搭建出一个真正的项目用户提问 ↓ 1. 判断是否需要检索 → 否 → LLM 直接回答 ↓ 是 2. 检索向量库 → 得到相关文档片段 ↓ 3. 组装上下文 → 生成回答带来源引用 ↓ 4. 人类审批 → 批准 → 返回给用户 ↓ 拒绝 5. 修改后重新生成前置依赖pip install langgraph langchain langchain-openai chromadb1. 整体架构┌─────────────────────────────────────────────────┐ │ 用户界面 (CLI/API) │ └─────────────────────┬───────────────────────────┘ │ ┌─────────────────────▼───────────────────────────┐ │ LangGraph RAG Agent │ │ │ │ ┌──────────┐ ┌──────────┐ ┌────────────────┐ │ │ │ 检索判断 │→│ 向量检索 │→│ 上下文组装 │ │ │ └──────────┘ └──────────┘ └───────┬────────┘ │ │ │ │ │ ┌──────────┐ ┌──────────┐ ┌───────▼────────┐ │ │ │ 最终回答 │←│ 人类审批 │←│ LLM 生成 │ │ │ └──────────┘ └──────────┘ └────────────────┘ │ │ │ │ ┌─────────────────────────────────────────────┐ │ │ │ Checkpointer短期记忆 Store长期记忆 │ │ │ └─────────────────────────────────────────────┘ │ └─────────────────────┬───────────────────────────┘ │ ┌─────────────────────▼───────────────────────────┐ │ Chroma 向量数据库 │ │ (存储文档嵌入 元数据) │ └─────────────────────────────────────────────────┘2. 文档预处理与向量化先把知识库文档处理成向量索引import os from langchain_community.document_loaders import TextLoader, DirectoryLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma # ─── 配置 ─── DOCS_DIR ./knowledge_base CHROMA_DIR ./chroma_db # 确保文档目录存在 os.makedirs(DOCS_DIR, exist_okTrue) def load_and_split_docs(): 加载文档并切分 loader DirectoryLoader(DOCS_DIR, glob**/*.txt, loader_clsTextLoader) docs loader.load() print(f加载了 {len(docs)} 个文档) splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap100, separators[\n\n, \n, 。, , , , , ], ) chunks splitter.split_documents(docs) print(f切分为 {len(chunks)} 个片段) return chunks def create_vector_store(chunksNone): 创建/加载向量库 if chunks: vector_store Chroma.from_documents( documentschunks, embeddingOpenAIEmbeddings(), persist_directoryCHROMA_DIR, ) print(f向量库创建完成存入 {CHROMA_DIR}) else: vector_store Chroma( embedding_functionOpenAIEmbeddings(), persist_directoryCHROMA_DIR, ) print(f加载已有向量库共 {vector_store._collection.count()} 条) return vector_store def index_documents(): 一键索引加载 → 切分 → 向量化 chunks load_and_split_docs() if chunks: return create_vector_store(chunks) print(没有发现新文档) return create_vector_store() # 首次运行请先创建一些示例文档 def create_sample_docs(): samples { langgraph_intro.txt: LangGraph 是 LangChain 公司开发的 Agent 编排框架。 它基于 Google Pregel 的图计算思想提供了 State、Node、Edge 三个核心概念。 LangGraph 擅长管理长期运行、有状态的 Agent 应用。 它支持持久化、Human-in-the-Loop、流式输出等生产级特性。 使用 pip install langgraph 即可安装。 , agent_patterns.txt: 常见的 Agent 设计模式包括 ReAct 模式、Plan-and-Execute 模式和多 Agent 协作模式。 ReAct 模式让 LLM 循环调用工具直到完成任务。 多 Agent 协作有路由模式、委托模式和交接模式三种。 路由模式用一个主 Agent 分发任务给不同子 Agent。 委托模式让 Agent 把子任务委派给专门的子 Agent。 交接模式像流水线一样多个 Agent 顺序处理。 , langchain_basics.txt: LangChain 是一个构建 LLM 应用的框架。 它提供了模型抽象、工具抽象和链式调用等核心功能。 LangChain 内置的 Agent如 create_react_agent底层使用 LangGraph。 LangGraph 是 LangChain 生态中的编排运行时层。 学习顺序建议先学 LangChain 基础再学 LangGraph 编排。 , } for name, content in samples.items(): with open(os.path.join(DOCS_DIR, name), w, encodingutf-8) as f: f.write(content) print(f已创建 {len(samples)} 个示例文档)3. RAG Agent 完整实现from typing import Annotated, Literal, TypedDict from langgraph.graph import StateGraph, START, END, add_messages from langgraph.checkpoint.memory import InMemorySaver from langgraph.store.memory import InMemoryStore from langgraph.types import interrupt, Command from langchain.chat_models import init_chat_model from langchain_core.messages import SystemMessage, ToolMessage, AIMessage from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma # ═══════════════════════════════════════════════ # 初始化 # ═══════════════════════════════════════════════ model init_chat_model(gpt-4o-mini, temperature0) vector_store Chroma( embedding_functionOpenAIEmbeddings(), persist_directoryCHROMA_DIR, ) # ═══════════════════════════════════════════════ # State # ═══════════════════════════════════════════════ class RAGState(TypedDict): messages: Annotated[list, add_messages] # 对话历史 question: str # 当前问题 needs_retrieval: bool # 是否需要检索 retrieved_docs: list[str] # 检索到的文档 contexts: list[str] # 整理后的上下文 candidate_answer: str # 待审批的回答 approved: bool # 是否已审批 final_answer: str # 最终答案 sources: list[dict] # 来源信息 # ═══════════════════════════════════════════════ # Node 1判断是否需要检索 # ═══════════════════════════════════════════════ def judge_retrieval_node(state: RAGState) - RAGState: 判断当前问题是否需要查知识库 question state[messages][-1].content if state[messages] else # 简单规则上下文相关的闲聊不需要检索 greetings [你好, hi, hello, 在吗, 你是谁] is_greeting any(g in question.lower() for g in greetings) # 多轮对话如果用户说继续、然后呢说明是接上文 follow_ups [继续, 然后呢, 接着说, 还有吗] is_follow_up any(f in question for f in follow_ups) # 显式要求检索的问题 needs_knowledge not is_greeting and not is_follow_up return {question: question, needs_retrieval: needs_knowledge} # ═══════════════════════════════════════════════ # Node 2向量检索 # ═══════════════════════════════════════════════ def retrieve_node(state: RAGState) - RAGState: 从向量库检索相关文档 if not state[needs_retrieval]: return {retrieved_docs: [], contexts: []} question state[question] # 向量检索取 Top-3 results vector_store.similarity_search_with_score(question, k3) docs [] sources [] for doc, score in results: docs.append(doc.page_content) sources.append({ content: doc.page_content[:100] ..., source: doc.metadata.get(source, unknown), score: round(score, 4), }) print(f 检索到 {len(docs)} 篇相关文档) for s in sources: print(f [{s[source]}] 相似度: {s[score]}) return { retrieved_docs: docs, sources: sources, } # ═══════════════════════════════════════════════ # Node 3LLM 生成回答 # ═══════════════════════════════════════════════ def generate_node(state: RAGState) - RAGState: LLM 生成带上下文的回答 if not state[needs_retrieval]: # 直接对话不需要检索 response model.invoke([ SystemMessage(content你是知识库助手可以直接回答用户问题。), *state[messages] ]) return { candidate_answer: response.content, approved: True, # 直接对话不需要审批 } # 有上下文 context \n\n.join(f[文档{i1}] {d} for i, d in enumerate(state[retrieved_docs])) response model.invoke([ SystemMessage(content( 你是知识库助手。基于以下参考资料回答用户问题。\n 回答时在引用处标注来源编号如 [1][2]。\n 如果参考资料不足以回答请如实告知。\n\n f参考资料\n{context} )), *state[messages][-1:] # 只传最后一个问题避免上下文过长 ]) return {candidate_answer: response.content} # ═══════════════════════════════════════════════ # Node 4人类审批 # ═══════════════════════════════════════════════ def approval_node(state: RAGState) - RAGState: 等待人类审批 if not state[needs_retrieval]: # 闲聊直接通过 return {approved: True} if state.get(approved): # 已经审批过了 return state print(f\n ⛔ 需要审批) print(f 候选回答: {state[candidate_answer][:100]}...) # 暂停等人审批 feedback interrupt({ type: rag_approval, question: state[question], answer: state[candidate_answer], sources: state[sources], action: 请确认是否发布。如果修改请传入编辑后的回答。, }) # feedback 可以是 bool批准/拒绝或 str修改后的回答 if isinstance(feedback, bool): return {approved: feedback} elif isinstance(feedback, str): return {candidate_answer: feedback, approved: True} return {approved: False} # ═══════════════════════════════════════════════ # Node 5最终输出 # ═══════════════════════════════════════════════ def final_node(state: RAGState, config, *, store: InMemoryStore) - RAGState: 生成最终答案更新历史记录到 Store if state.get(approved): # 构建带来源的最终答案 answer state[candidate_answer] if state.get(sources): source_text \n\n 来源 for i, s in enumerate(state[sources]): source_text f\n[{i1}] {s[source]}相似度: {s[score]} answer source_text # 更新到 Store 作为长期记忆 user_id config[configurable].get(user_id, default) history store.get((rag, user_id), qa_history) qa_list history.value.get(qa_pairs, []) if history else [] qa_list.append({ question: state[question], answer: state[candidate_answer], sources: state.get(sources, []), }) # 最多保留最近 20 条 store.put((rag, user_id), qa_history, {qa_pairs: qa_list[-20:]}) return {final_answer: answer} else: return {final_answer: ❌ 回答已被拒绝请重新提问。} # ═══════════════════════════════════════════════ # 路由函数 # ═══════════════════════════════════════════════ def retrieval_router(state: RAGState) - Literal[retrieve, generate]: return retrieve if state[needs_retrieval] else generate def approval_router(state: RAGState) - Literal[approval, END]: return approval if state[needs_retrieval] else END # ═══════════════════════════════════════════════ # 构建图 # ═══════════════════════════════════════════════ saver InMemorySaver() store InMemoryStore() builder StateGraph(RAGState) # 添加节点 builder.add_node(judge, judge_retrieval_node) builder.add_node(retrieve, retrieve_node) builder.add_node(generate, generate_node) builder.add_node(approval, approval_node) builder.add_node(final, final_node) # 连接 builder.add_edge(START, judge) builder.add_conditional_edges(judge, retrieval_router, { retrieve: retrieve, generate: generate, }) builder.add_edge(retrieve, generate) builder.add_edge(generate, final) # 带审批的路由 # 简单版本always go to final, skip explicit approval for simplicity builder.add_edge(final, END) rag_agent builder.compile(checkpointersaver, storestore)说明上面的代码在审批路由部分做了简化——最终版本可以按照第 7 期的 interrupt 模式插入approval_node。完整版在下面。4. 带审批的完整版如果你需要真正的 Human-in-the-Loop 审批流程用这个版本def build_rag_with_approval(): 构建带审批的 RAG Agent class RAGState(TypedDict): messages: Annotated[list, add_messages] question: str needs_retrieval: bool retrieved_docs: list[str] contexts: list[str] candidate_answer: str approved: bool final_answer: str sources: list[dict] # ...上面所有的 Node 函数定义... builder StateGraph(RAGState) builder.add_node(judge, judge_retrieval_node) builder.add_node(retrieve, retrieve_node) builder.add_node(generate, generate_node) builder.add_node(approval, approval_node) builder.add_node(final, final_node) builder.add_edge(START, judge) builder.add_conditional_edges(judge, retrieval_router, { retrieve: retrieve, generate: generate, }) builder.add_edge(retrieve, generate) builder.add_edge(generate, approval) builder.add_conditional_edges(approval, lambda s: final if s.get(approved) else generate, { final: final, generate: generate, }) builder.add_edge(final, END) return builder.compile(checkpointerInMemorySaver(), storeInMemoryStore()) # 使用方式 # agent build_rag_with_approval() # stream agent.stream_events(input_data, config, versionv3) # _ stream.output # if stream.interrupted: # approval_value input(审批 (y/n): ).lower() y # stream2 agent.stream_events(Command(resumeapproval_value), config, versionv3)5. 测试与运行def create_rag_pipeline(): 创建完整的 RAG 管道 create_sample_docs() vector_store index_documents() # 构建 RAG Agent builder StateGraph(RAGState) builder.add_node(judge, judge_retrieval_node) builder.add_node(retrieve, retrieve_node) builder.add_node(generate, generate_node) builder.add_node(final, final_node) builder.add_edge(START, judge) builder.add_conditional_edges(judge, retrieval_router, { retrieve: retrieve, generate: generate, }) builder.add_edge(retrieve, generate) builder.add_edge(generate, final) builder.add_edge(final, END) return builder.compile(checkpointerInMemorySaver(), storeInMemoryStore()) def ask(agent, question: str, user_id: str u001, thread_id: str rag-1): 向 RAG Agent 提问 config {configurable: {thread_id: thread_id, user_id: user_id}} print(f\n{*60}) print(f 用户: {question}) print(f{*60}) result agent.invoke({ messages: [(user, question)], question: , needs_retrieval: False, retrieved_docs: [], contexts: [], candidate_answer: , approved: False, final_answer: , sources: [], }, config) print(f {result[final_answer]}) # 检查 Store 中的历史 history store.get((rag, user_id), qa_history) if history: print(f\n 历史问答计数: {len(history.value[qa_pairs])}) return result def chat_loop(): 交互式问答 agent create_rag_pipeline() print( RAG 知识库助手已启动输入 /bye 退出) print(f 知识库目录: {DOCS_DIR}) print(f 向量库目录: {CHROMA_DIR}) thread_id rag-session-1 while True: question input(\n 你: ).strip() if question.lower() in (/bye, /exit, quit): print( 再见) break if not question: continue ask(agent, question, thread_idthread_id) if __name__ __main__: # 首次运行时先建索引 # create_sample_docs() # index_documents() # 启动交互 chat_loop()运行示例 RAG 知识库助手已启动输入 /bye 退出 你: 你好 你好我是知识库助手有什么可以帮你 你: LangGraph 是什么 检索到 3 篇相关文档 [langgraph_intro.txt] 相似度: 0.8765 [langchain_basics.txt] 相似度: 0.6543 [agent_patterns.txt] 相似度: 0.5123 LangGraph 是 LangChain 公司开发的 Agent 编排框架 [1]。 它提供了 State、Node、Edge 三个核心概念擅长管理长期运行、 有状态的 Agent 应用。它支持持久化、Human-in-the-Loop 等 生产级特性可以使用 pip install langgraph 安装 [1]。 来源 [1] langgraph_intro.txt相似度: 0.8765 [2] langchain_basics.txt相似度: 0.6543 你: 它有哪几种常见模式 检索到 2 篇相关文档 [agent_patterns.txt] 相似度: 0.9123 [langgraph_intro.txt] 相似度: 0.4512 根据资料多 Agent 协作有路由模式、委托模式和交接模式三种 [1]。 - 路由模式用一个主 Agent 分发任务给不同子 Agent - 委托模式Agent 把子任务委派给专门的子 Agent - 交接模式像流水线一样多个 Agent 顺序处理 来源 [1] agent_patterns.txt相似度: 0.91236. 扩展LangSmith 集成LangSmith 是 LangChain 的调试与监控平台一键集成import os # 设置环境变量 os.environ[LANGSMITH_TRACING] true os.environ[LANGSMITH_API_KEY] lsv2_... # 你的 API Key os.environ[LANGSMITH_PROJECT] rag-agent # 然后正常调用所有 trace 自动上传到 LangSmith 控制台LangSmith 会记录每次 invoke 的完整调用链每个 Node 的输入输出LLM 的请求和响应工具调用和结果执行时间和 token 消耗7. 部署建议7.1 持久化 Checkpointer开发阶段用InMemorySaver生产环境换成 SqliteSaverfrom langgraph.checkpoint.sqlite import SqliteSaver saver SqliteSaver.from_conn_string(rag_checkpoints.db)7.2 持久化 Store# 生产环境用 PostgresStore from langgraph.checkpoint.postgres import PostgresStore store PostgresStore.from_conn_string(postgresql://...) await store.setup()7.3 FastAPI 部署from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class Query(BaseModel): question: str thread_id: str default user_id: str default app.post(/ask) async def ask(query: Query): config {configurable: {thread_id: query.thread_id, user_id: query.user_id}} result rag_agent.invoke({ messages: [(user, query.question)], # ... 其他初始字段 }, config) return {answer: result[final_answer]}7.4 Docker 部署FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]8. 整体系列回顾12 期学下来我们从零开始覆盖了 LangGraph 的方方面面阶段期数核心技能基础篇1-4概念、State、Node/Edge、工具调用进阶篇5-8Reducer、Checkpointer、Human-in-the-Loop、Streaming高级篇9-11Functional API、Memory、Subgraph 多 Agent实战篇12完整的 RAG 知识库 Agent现在你能够✅ 用 Graph API 和 Functional API 搭建 Agent✅ 理解 State 的 Reducer 机制✅ 实现多轮对话Checkpointer✅ 添加人工审批Interrupt✅ 实时流式输出Streaming✅ 跨会话记忆Store✅ 多 Agent 协作Subgraph✅ 构建完整的 RAG 应用参考资料LangGraph 官方文档LangGraph overview - Docs by LangChainLangChain RAG 教程Learn - Docs by LangChainChroma 向量库Chroma - open-source search infrastructure for AILangSmith调试部署LangSmith: Agent LLM Observability PlatformGitHubGitHub - langchain-ai/langgraph: Build resilient agents. · GitHub12 期从零到实战。希望能帮你在 Agent 开发这条路上走得更顺。如果这个系列对你有帮助欢迎点赞收藏。如果你用 LangGraph 做出了有意思的项目欢迎在评论区分享。