隨著大模型（LLM）能力越來越強，RAG（Retrieval Augmented Generation，檢索增強生成）技術(shù)成為增強大模型知識準(zhǔn)確性的關(guān)鍵手段。

通過檢索實時數(shù)據(jù)、外部文檔，模型能回答更多基于事實的問題，降低“幻覺”概率。

而 LangChain 的 LangGraph 能將 LLM、RAG、工具調(diào)用（Tools）整合成一個智能 Agent 流程圖，極大提升了問答系統(tǒng)的動態(tài)能力。

本文通過一個完整示例，展示如何用 LangChain 構(gòu)建一個「RAG + Agent」的問答系統(tǒng)，代碼可直接復(fù)用，幫助大家快速落地智能應(yīng)用。

工程結(jié)構(gòu)

llm_env.py          # 初始化 LLM

rag_agent.py        # 結(jié)合 RAG 與 Agent 的主邏輯

初始化 LLM

首先通過 llm_env.py 初始化一個 LLM 模型對象，供整個流程使用：

from langchain.chat_models import init_chat_model

llm = init_chat_model("gpt-4o-mini", model_provider="openai")

RAG + Agent 系統(tǒng)搭建

導(dǎo)入依賴

import os

import sys

import time

sys.path.append(os.getcwd())

from llm_set import llm_env

from langchain.embeddings import OpenAIEmbeddings

from langchain_postgres import PGVector

from langchain_community.document_loaders import WebBaseLoader

from langchain_core.documents import Document

from langchain_text_splitters import RecursiveCharacterTextSplitter

from langgraph.graph import MessagesState, StateGraph

from langchain_core.tools import tool

from langchain_core.messages import HumanMessage, SystemMessage

from langgraph.prebuilt import ToolNode, tools_condition

from langgraph.graph import END

from langgraph.checkpoint.postgres import PostgresSaver

初始化 LLM 與 Embedding

llm = llm_env.llm

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

初始化向量數(shù)據(jù)庫

vector_store = PGVector(

    embeddings=embeddings,

    collection_name="my_rag_agent_docs",

    connection="postgresql+psycopg2://postgres:123456@localhost:5433/langchainvector",

)

加載網(wǎng)頁文檔

url = "https://www.cnblogs.com/chenyishi/p/18926783"

loader = WebBaseLoader(

    web_paths=(url,),

)

docs = loader.load()

for doc in docs:

    doc.metadata["source"] = url

文本分割 & 入庫

text_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=50)

all_splits = text_splitter.split_documents(docs)

existing = vector_store.similarity_search(url, k=1, filter={"source": url})

if not existing:

    _ = vector_store.add_documents(documents=all_splits)

    print("文檔向量化完成")

定義 RAG 檢索工具

通過 @tool 裝飾器，定義一個文檔檢索工具，供 Agent 動態(tài)調(diào)用：

@tool(response_format="content_and_artifact")

def retrieve(query: str) -> tuple[str, dict]:

    """Retrieve relevant documents from the vector store."""

    retrieved_docs = vector_store.similarity_search(query, k=2)

    if not retrieved_docs:

        return "No relevant documents found.", {}

    return "\n\n".join(

        (f"Source: {doc.metadata}\n" f"Content: {doc.page_content}")

        for doc in retrieved_docs

    ), retrieved_docs

定義 Agent Graph 節(jié)點

LLM 調(diào)用工具節(jié)點

def query_or_respond(state: MessagesState):

    llm_with_tools = llm.bind_tools([retrieve])

    response = llm_with_tools.invoke(state["messages"])

    return {"messages": [response]}

工具節(jié)點

tools = ToolNode([retrieve])

生成響應(yīng)節(jié)點

def generate(state: MessagesState):

    recent_tool_messages = []

    for message in reversed(state["messages"]):

        if message.type == "tool":

            recent_tool_messages.append(message)

        else:

            break

    tool_messages = recent_tool_messages[::-1]

    system_message_content = "\n\n".join(doc.content for doc in tool_messages)

    conversation_messages = [

        message

        for message in state["messages"]

        if message.type in ("human", "system")

        or (message.type == "ai" and not message.tool_calls)

    ]

    prompt = [SystemMessage(system_message_content)] + conversation_messages

    response = llm.invoke(prompt)

    return {"messages": [response]}

組裝 Agent 流程圖

graph_builder = StateGraph(MessagesState)

graph_builder.add_node(query_or_respond)

graph_builder.add_node(tools)

graph_builder.add_node(generate)

graph_builder.set_entry_point("query_or_respond")

graph_builder.add_conditional_edges(

    "query_or_respond",

    tools_condition,

    path_map={END: END, "tools": "tools"},

)

graph_builder.add_edge("tools", "generate")

graph_builder.add_edge("generate", END)

啟用 Checkpoint & 運行流程

數(shù)據(jù)庫存儲器

DB_URI = "postgresql://postgres:123456@localhost:5433/langchaindemo?sslmode=disable"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:

    checkpointer.setup()

    graph = graph_builder.compile(checkpointer=checkpointer)

啟動交互循環(huán)

input_thread_id = input("輸入thread_id:")

time_str = time.strftime("%Y%m%d", time.localtime())

config = {"configurable": {"thread_id": f"rag-{time_str}-demo-{input_thread_id}"}}

print("輸入問題，輸入 exit 退出。")

while True:

    query = input("你: ")

    if query.strip().lower() == "exit":

        break

    response = graph.invoke({"messages": [HumanMessage(content=query)]}, config=config)

    print(response)

總結(jié)

本文完整展示了如何用 LangChain + LangGraph，結(jié)合：

LLM（大模型）
Embedding 檢索（RAG）
Agent 動態(tài)調(diào)用工具
流程圖編排
Checkpoint 存儲

構(gòu)建一個智能問答系統(tǒng)。通過將工具（RAG 檢索）和 Agent 機制結(jié)合，可以讓 LLM 在需要的時候 自主調(diào)用檢索能力，有效增強對知識的引用能力，解決“幻覺”問題，具備很好的落地應(yīng)用價值。

總結(jié)

以上是生活随笔為你收集整理的AI大模型应用开发-用LangChain构建带Agen流程的RAG系统的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。