AI大模型应用开发-用LangChain构建带Agen流程的RAG系统

随着大模型（LLM）能力越来越强，RAG（Retrieval Augmented Generation，检索增强生成）技术成为增强大模型知识准确性的关键手段。
通过检索实时数据、外部文档，模型能回答更多基于事实的问题，降低“幻觉”概率。
而 LangChain 的 LangGraph 能将 LLM、RAG、工具调用（Tools）整合成一个智能 Agent 流程图，极大提升了问答系统的动态能力。
本文通过一个完整示例，展示如何用 LangChain 构建一个「RAG + Agent」的问答系统，代码可直接复用，帮助大家快速落地智能应用。
工程结构

1. llm_env.py          # 初始化 LLM
2. rag_agent.py        # 结合 RAG 与 Agent 的主逻辑

复制代码

初始化 LLM

首先通过 llm_env.py 初始化一个 LLM 模型对象，供整个流程使用：

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

复制代码

RAG + Agent 系统搭建

导入依赖

1. import os
2. import sys
3. import time
5. sys.path.append(os.getcwd())
7. from llm_set import llm_env
8. from langchain.embeddings import OpenAIEmbeddings
9. from langchain_postgres import PGVector
10. from langchain_community.document_loaders import WebBaseLoader
11. from langchain_core.documents import Document
12. from langchain_text_splitters import RecursiveCharacterTextSplitter
13. from langgraph.graph import MessagesState, StateGraph
14. from langchain_core.tools import tool
15. from langchain_core.messages import HumanMessage, SystemMessage
16. from langgraph.prebuilt import ToolNode, tools_condition
17. from langgraph.graph import END
18. from langgraph.checkpoint.postgres import PostgresSaver

复制代码

初始化 LLM 与 Embedding

1. llm = llm_env.llm
3. embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

复制代码

初始化向量数据库

1. vector_store = PGVector(
2.     embeddings=embeddings,
3.     collection_name="my_rag_agent_docs",
4.     connection="postgresql+psycopg2://postgres:123456@localhost:5433/langchainvector",
5. )

复制代码

加载网页文档

1. url = "https://www.cnblogs.com/chenyishi/p/18926783"
2. loader = WebBaseLoader(
3.     web_paths=(url,),
4. )
5. docs = loader.load()
6. for doc in docs:
7.     doc.metadata["source"] = url

复制代码

文本分割 & 入库

1. text_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=50)
2. all_splits = text_splitter.split_documents(docs)
4. existing = vector_store.similarity_search(url, k=1, filter={"source": url})
5. if not existing:
6.     _ = vector_store.add_documents(documents=all_splits)
7.     print("文档向量化完成")

复制代码

定义 RAG 检索工具

通过 @tool 装饰器，定义一个文档检索工具，供 Agent 动态调用：

1. @tool(response_format="content_and_artifact")
2. def retrieve(query: str) -> tuple[str, dict]:
3.     """Retrieve relevant documents from the vector store."""
4.     retrieved_docs = vector_store.similarity_search(query, k=2)
5.     if not retrieved_docs:
6.         return "No relevant documents found.", {}
7.     return "\n\n".join(
8.         (f"Source: {doc.metadata}\n" f"Content: {doc.page_content}")
9.         for doc in retrieved_docs
10.     ), retrieved_docs

复制代码

定义 Agent Graph 节点

LLM 调用工具节点

1. def query_or_respond(state: MessagesState):
2.     llm_with_tools = llm.bind_tools([retrieve])
3.     response = llm_with_tools.invoke(state["messages"])
4.     return {"messages": [response]}

复制代码

工具节点

1. tools = ToolNode([retrieve])

复制代码

生成响应节点

1. def generate(state: MessagesState):
2.     recent_tool_messages = []
3.     for message in reversed(state["messages"]):
4.         if message.type == "tool":
5.             recent_tool_messages.append(message)
6.         else:
7.             break
9.     tool_messages = recent_tool_messages[::-1]
11.     system_message_content = "\n\n".join(doc.content for doc in tool_messages)
13.     conversation_messages = [
14.         message
15.         for message in state["messages"]
16.         if message.type in ("human", "system")
17.         or (message.type == "ai" and not message.tool_calls)
18.     ]
19.     prompt = [SystemMessage(system_message_content)] + conversation_messages
21.     response = llm.invoke(prompt)
22.     return {"messages": [response]}

复制代码

组装 Agent 流程图

1. graph_builder = StateGraph(MessagesState)
2. graph_builder.add_node(query_or_respond)
3. graph_builder.add_node(tools)
4. graph_builder.add_node(generate)
6. graph_builder.set_entry_point("query_or_respond")
7. graph_builder.add_conditional_edges(
8.     "query_or_respond",
9.     tools_condition,
10.     path_map={END: END, "tools": "tools"},
11. )
12. graph_builder.add_edge("tools", "generate")
13. graph_builder.add_edge("generate", END)

复制代码

启用 Checkpoint & 运行流程

数据库存储器

1. DB_URI = "postgresql://postgres:123456@localhost:5433/langchaindemo?sslmode=disable"
3. with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
4.     checkpointer.setup()
6.     graph = graph_builder.compile(checkpointer=checkpointer)

复制代码

启动交互循环

1. input_thread_id = input("输入thread_id:")
2. time_str = time.strftime("%Y%m%d", time.localtime())
3. config = {"configurable": {"thread_id": f"rag-{time_str}-demo-{input_thread_id}"}}
5. print("输入问题，输入 exit 退出。")
6. while True:
7.     query = input("你: ")
8.     if query.strip().lower() == "exit":
9.         break
10.     response = graph.invoke({"messages": [HumanMessage(content=query)]}, config=config)
11.     print(response)

复制代码

总结

本文完整展示了如何用 LangChain + LangGraph，结合：
LLM（大模型）
Embedding 检索（RAG）
Agent 动态调用工具
流程图编排
Checkpoint 存储
构建一个智能问答系统。通过将工具（RAG 检索）和 Agent 机制结合，可以让 LLM 在需要的时候 自主调用检索能力，有效增强对知识的引用能力，解决“幻觉”问题，具备很好的落地应用价值。

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