LangGraph + LangChain + FAISS：手把手教你打造一个带记忆的PDF问答机器人

LangGraph实战构建带记忆的PDF问答机器人全流程解析在AI应用开发领域如何让对话系统具备长期记忆和精准检索能力一直是开发者面临的挑战。本文将带你从零开始使用LangGraph、LangChain和FAISS三大工具构建一个能记住对话历史并从PDF文档中精准查找答案的智能助手。不同于简单的问答系统我们将重点解决多轮对话中的上下文保持问题并通过可视化工作流展示整个系统的运行机制。1. 技术栈选型与核心组件在开始编码前我们需要理解每个技术组件在系统中的角色LangGraph负责工作流编排和状态管理让系统能够记住对话历史LangChain提供RAG检索增强生成链式处理能力FAISSFacebook开源的向量数据库用于高效相似性搜索这三个组件的协同工作流程可以形象地理解为用户提问进入LangGraph定义的工作流系统通过FAISS检索相关文档片段LangChain将检索结果与对话历史结合生成最终回复关键配置参数对比组件核心功能典型配置性能考量LangGraph工作流编排节点数、边定义状态更新频率FAISS向量检索索引类型(nlist, nprobe)查询延迟与内存占用LangChain生成优化chunk_size, chunk_overlap上下文窗口利用率2. 环境准备与基础搭建首先确保安装必要的Python包pip install langgraph langchain faiss-cpu pypdf对于生产环境建议使用GPU加速的faiss-gpu版本pip install faiss-gpu初始化基础组件from langgraph.graph import StateGraph from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import FAISS from langchain.document_loaders import PyPDFLoader # 初始化嵌入模型 embedding_model HuggingFaceEmbeddings( model_namesentence-transformers/all-mpnet-base-v2 )3. PDF文档处理与向量化文档处理是RAG系统的核心环节需要特别注意文本分块的策略from langchain.text_splitter import RecursiveCharacterTextSplitter def process_pdf(file_path): # 加载PDF文档 loader PyPDFLoader(file_path) pages loader.load() # 智能分块 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200, separators[\n\n, \n, 。, , ] ) chunks text_splitter.split_documents(pages) # 创建向量存储 vector_store FAISS.from_documents(chunks, embedding_model) return vector_store # 使用示例 pdf_path technical_document.pdf vector_db process_pdf(pdf_path) vector_db.save_local(faiss_index)分块策略优化建议技术文档建议使用较大的chunk_size(800-1200)对话记录较小的chunk_size(300-500)效果更好法律文本需要保持段落完整性可设置separators[\n\n]4. 构建带记忆的工作流LangGraph的核心价值在于状态管理下面我们定义一个能记住对话历史的图结构from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages # 定义状态结构 class ConversationState(TypedDict): messages: Annotated[List, add_messages] last_query: str search_results: List[str] # 初始化图 workflow StateGraph(ConversationState) # 添加检索节点 def retrieve_info(state: ConversationState): query state[messages][-1][content] results vector_db.similarity_search(query, k3) return { last_query: query, search_results: [doc.page_content for doc in results] } workflow.add_node(retriever, retrieve_info)5. 多轮对话与上下文管理实现带记忆的问答需要精心设计状态更新逻辑from langchain.chat_models import ChatOpenAI from langchain.prompts import ChatPromptTemplate llm ChatOpenAI(modelgpt-4) # 定义生成节点 def generate_response(state: ConversationState): prompt_template 根据以下上下文和对话历史回答问题 相关文档内容 {context} 对话历史 {history} 问题{question} prompt ChatPromptTemplate.from_template(prompt_template) chain prompt | llm context \n.join(state[search_results]) history \n.join([msg[content] for msg in state[messages][:-1]]) response chain.invoke({ context: context, history: history, question: state[last_query] }) return {messages: [{role: assistant, content: response.content}]} workflow.add_node(generator, generate_response)6. 高级功能扩展基础工作流搭建完成后可以添加以下增强功能条件分支处理def should_use_rag(state: ConversationState) - str: # 简单启发式问题包含特定关键词或超过一定长度 query state[last_query] if len(query.split()) 5 or ? in query: return use_rag return direct_response workflow.add_conditional_edges( retriever, should_use_rag, { use_rag: generator, direct_response: fallback } )对话历史压缩防止上下文窗口溢出def compress_history(messages: List) - List: # 实现基于重要性的历史消息筛选 return compressed_messages7. 系统集成与性能优化将各个组件组装成完整工作流# 定义边关系 workflow.add_edge(generator, END) workflow.add_edge(fallback, END) # 编译工作流 app workflow.compile() # 运行示例 def chat_loop(): state {messages: []} while True: user_input input(用户: ) if user_input.lower() exit: break state[messages].append({role: user, content: user_input}) for output in app.stream(state): for key, value in output.items(): if messages in value: print(f助手: {value[messages][-1][content]}) state[messages].extend(value[messages]) chat_loop()性能优化技巧FAISS索引优化index FAISS.IndexFlatIP(768) # 使用内积相似度 index FAISS.IndexIVFFlat(index, 768, 100) # 使用倒排索引加速LangChain缓存from langchain.cache import InMemoryCache llm.cache InMemoryCache()异步处理async def async_retrieve(state): # 实现异步检索 pass在实际项目中我们还需要考虑错误处理和回退机制对话质量监控用户反馈收集系统知识库自动更新策略这个项目的完整实现大约需要300-500行Python代码具体复杂度取决于所需功能的完整度。建议从最小可行产品开始逐步添加高级功能。