AI大模型应用开发入门-LangChain开发RAG增强检索生成

检索增强生成（RAG）是一种结合“向量检索”与“大语言模型”的技术路线，能在问答、摘要、文档分析等场景中大幅提升准确性与上下文利用率。
本文将基于 LangChain 构建一个完整的 RAG 流程，结合 PGVector 作为向量数据库，并用 LangGraph 构建状态图控制流程。
大语言模型初始化（llm_env.py）

我们首先使用 LangChain 提供的模型初始化器加载 gpt-4o-mini 模型，供后续问答使用。

1. # llm_env.py
2. from langchain.chat_models import init_chat_model
4. llm = init_chat_model("gpt-4o-mini", model_provider="openai")

复制代码

RAG 主体流程（rag.py）

以下是整个 RAG 系统的主流程代码，主要包括：文档加载与切分、向量存储、状态图建模（analyze→retrieve→generate）、交互式问答。

1. # rag.py
2. import os
3. import sys
4. import time
6. sys.path.append(os.getcwd())
8. from llm_set import llm_env
9. from langchain_openai import OpenAIEmbeddings
10. from langchain_postgres import PGVector
11. from langchain_community.document_loaders import WebBaseLoader
12. from langchain_core.documents import Document
13. from langchain_text_splitters import RecursiveCharacterTextSplitter
14. from langgraph.graph import START, StateGraph
15. from typing_extensions import List, TypedDict, Annotated
16. from typing import Literal
17. from langgraph.checkpoint.postgres import PostgresSaver
18. from langgraph.graph.message import add_messages
19. from langchain_core.messages import HumanMessage, BaseMessage
20. from langchain_core.prompts import ChatPromptTemplate
22. # 初始化 LLM
23. llm = llm_env.llm
25. # 嵌入模型
26. embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
28. # 向量数据库初始化
29. vector_store = PGVector(
30.     embeddings=embeddings,
31.     collection_name="my_rag_docs",
32.     connection="postgresql+psycopg2://postgres:123456@localhost:5433/langchainvector",
33. )
35. # 加载网页内容
36. url = "https://python.langchain.com/docs/tutorials/qa_chat_history/"
37. loader = WebBaseLoader(web_paths=(url,))
38. docs = loader.load()
39. for doc in docs:
40.     doc.metadata["source"] = url
42. # 文本分割
43. text_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=50)
44. all_splits = text_splitter.split_documents(docs)
46. # 添加 section 元数据
47. total_documents = len(all_splits)
48. third = total_documents // 3
49. for i, document in enumerate(all_splits):
50.     if i < third:
51.         document.metadata["section"] = "beginning"
52.     elif i < 2 * third:
53.         document.metadata["section"] = "middle"
54.     else:
55.         document.metadata["section"] = "end"
57. # 检查是否已存在向量
58. existing = vector_store.similarity_search(url, k=1, filter={"source": url})
59. if not existing:
60.     _ = vector_store.add_documents(documents=all_splits)
61.     print("文档向量化完成")

复制代码

分析、检索与生成模块

接下来，我们定义三个函数构成 LangGraph 的流程：analyze → retrieve → generate。

1. class Search(TypedDict):
2.     query: Annotated[str, "The question to be answered"]
3.     section: Annotated[
4.         Literal["beginning", "middle", "end"],
5.         ...,
6.         "Section to query.",
7.     ]
9. class State(TypedDict):
10.     messages: Annotated[list[BaseMessage], add_messages]
11.     query: Search
12.     context: List[Document]
13.     answer: set
15. # 分析意图 → 获取 query 与 section
16. def analyze(state: State):
17.     structtured_llm = llm.with_structured_output(Search)
18.     query = structtured_llm.invoke(state["messages"])
19.     return {"query": query}
21. # 相似度检索
22. def retrieve(state: State):
23.     query = state["query"]
24.     if hasattr(query, 'section'):
25.         filter = {"section": query["section"]}
26.     else:
27.         filter = None
28.     retrieved_docs = vector_store.similarity_search(query["query"], filter=filter)
29.     return {"context": retrieved_docs}

复制代码

生成模块基于 ChatPromptTemplate 和当前上下文生成回答：

1. prompt_template = ChatPromptTemplate.from_messages(
2.     [
3.         ("system", "尽你所能按照上下文:{context}，回答问题：{question}。"),
4.     ]
5. )
7. def generate(state: State):
8.     docs_content = "\n\n".join(doc.page_content for doc in state["context"])
9.     messages = prompt_template.invoke({
10.         "question": state["query"]["query"],
11.         "context": docs_content,
12.     })
13.     response = llm.invoke(messages)
14.     return {"answer": response.content, "messages": [response]}

复制代码

构建 LangGraph 流程图

定义好状态结构后，我们构建 LangGraph：

1. graph_builder = StateGraph(State).add_sequence([analyze, retrieve, generate])
2. graph_builder.add_edge(START, "analyze")

复制代码

PG 数据库中保存中间状态（Checkpoint）

我们通过 PostgresSaver 记录每次对话的中间状态：

1. DB_URI = "postgresql://postgres:123456@localhost:5433/langchaindemo?sslmode=disable"
3. with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
4.     checkpointer.setup()
5.     graph = graph_builder.compile(checkpointer=checkpointer)
6.     input_thread_id = input("输入thread_id:")
7.     time_str = time.strftime("%Y%m%d", time.localtime())
8.     config = {"configurable": {"thread_id": f"rag-{time_str}-demo-{input_thread_id}"}}
10.     print("输入问题，输入 exit 退出。")
11.     while True:
12.         query = input("你: ")
13.         if query.strip().lower() == "exit":
14.             break
15.         input_messages = [HumanMessage(query)]
16.         response = graph.invoke({"messages": input_messages}, config=config)
17.         print(response["answer"])

复制代码

效果

总结

本文通过 LangChain 的模块式能力，结合 PGVector 向量库与 LangGraph 有状态控制系统，实现了一个可交互、可持久化、支持多文档结构的 RAG 系统。其优势包括：

• 支持结构化提问理解（分区查询）
  
• 自动化分段与元数据标记
  
• 状态流追踪与恢复
  
• 可拓展支持文档上传、缓存优化、多用户配置
  

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！