推荐语

探索LangGraph在RAG Agent中的应用，实现知识与生成的完美融合。

核心内容：
1. RAG Agent核心逻辑：检索、增强、生成
2. 文档预处理与检索工具创建
3. 使用LangGraph构建RAG Agent的详细步骤

杨芳贤

53A创始人/腾讯云(TVP)最具价值专家

介绍

这篇文章来介绍下使用 LangGraph 构建 RAG Agent。

RAG通过引入外部知识库，将动态检索与生成能力结合，让LLM既能“博学”又能“可信”。它的核心逻辑是：
1️⃣ 检索 → 从知识库中精准拉取相关文档；
2️⃣ 增强 → 将检索结果融入提示（Prompt），辅助模型生成；
3️⃣ 生成 → 输出兼具准确性与透明度的答案。

1.预处理文档

使用WebBaseLoader工具加载web资源，读取文档

from langchain_community.document_loaders import WebBaseLoader

urls =[
"https://lilianweng.github.io/posts/2024-11-28-reward-hacking/",
"https://lilianweng.github.io/posts/2024-07-07-hallucination/",
"https://lilianweng.github.io/posts/2024-04-12-diffusion-video/",
]

docs =[WebBaseLoader(url).load()for url in urls]

2.创建检索工具

对文档数据进行切分：

from langchain_text_splitters import RecursiveCharacterTextSplitter

docs_list =[item for sublist in docs for item in sublist]

text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=100, chunk_overlap=50
)
doc_splits = text_splitter.split_documents(docs_list)

使用阿里QianWen的embedding模型将文档数据转化为向量，存储到内存向量数据库

from langchain_core.vectorstoresimportInMemoryVectorStore
from langchain_community.embeddingsimportDashScopeEmbeddings

vectorstore =InMemoryVectorStore.from_documents(
    documents=doc_splits, embedding=DashScopeEmbeddings(model="text-embedding-v3")
)
retriever = vectorstore.as_retriever()

创建检索工具：

from langchain.tools.retrieverimport create_retriever_tool

retriever_tool =create_retriever_tool(
    retriever,
"retrieve_blog_posts",
"Search and return information about Lilian Weng blog posts.",
)

3.生成查询

使用阿里QianWen模型作为LLM，构建 generate_query_or_respond 节点

from langgraph.graphimportMessagesState

response_model =ChatTongyi(model="qwen-plus")


def generate_query_or_respond(state:MessagesState):
"""Call the model to generate a response based on the current state.Given
    the question, it will decide to retrieve using the retriever tool, or simply respond to the user.
"""
    response =(
        response_model
.bind_tools([retriever_tool]).invoke(state["messages"])
)
return{"messages":[response]}

3.对文档评分

定义grade_documents节点： 定义GradeDocuments class,使用QianWen模型结构化输出（返回yes和no），对检索工具的结果进行评分，如果返回yes则返回generate_answer节点，否则返回rewrite_question节点。

from pydantic importBaseModel,Field
from typing importLiteral

GRADE_PROMPT=(
"You are a grader assessing relevance of a retrieved document to a user question. \n "
"Here is the retrieved document: \n\n {context} \n\n"
"Here is the user question: {question} \n"
"If the document contains keyword(s) or semantic meaning related to the user question, grade it as relevant. \n"
"Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question."
)


classGradeDocuments(BaseModel):
"""Grade documents using a binary score for relevance check."""

binary_score: str =Field(
        description="Relevance score: 'yes' if relevant, or 'no' if not relevant"
)


grader_model =ChatTongyi(model="qwen-plus")


def grade_documents(
state:MessagesState,
)->Literal["generate_answer","rewrite_question"]:
"""Determine whether the retrieved documents are relevant to the question."""
for message in state["messages"]:
ifisinstance(message,HumanMessage):
            question = message.content
    context = state["messages"][-1].content

    prompt =GRADE_PROMPT.format(question=question, context=context)
    response =(
        grader_model
.with_structured_output(GradeDocuments).invoke(
[{"role":"user","content": prompt}]
)
)
    score = response.binary_score

if score =="yes":
return"generate_answer"
else:
return"rewrite_question"

4.重写问题

定义rewrite_question节点，如果文档评分不相关，则重新根据用户问题生成查询检索：

REWRITE_PROMPT=(
"Look at the input and try to reason about the underlying semantic intent / meaning.\n"
"Here is the initial question:"
"\n ------- \n"
"{question}"
"\n ------- \n"
"Formulate an improved question:"
)


def rewrite_question(state:MessagesState):
"""Rewrite the original user question."""
for message in state["messages"]:
ifisinstance(message,HumanMessage):
            question = message.content
    prompt =REWRITE_PROMPT.format(question=question)
    response = response_model.invoke([{"role":"user","content": prompt}])
return{"messages":[{"role":"user","content": response.content}]}

5.生成答案

定义generate_answer节点, 根据检索结果和用户问题生成答案：

GENERATE_PROMPT=(
"You are an assistant for question-answering tasks. "
"Use the following pieces of retrieved context to answer the question. "
"If you don't know the answer, just say that you don't know. "
"Use three sentences maximum and keep the answer concise.\n"
"Question: {question} \n"
"Context: {context}"
)


def generate_answer(state:MessagesState):
"""Generate an answer."""
for message in state["messages"]:
ifisinstance(message,HumanMessage):
            question = message.content
    context = state["messages"][-1].content
    prompt =GENERATE_PROMPT.format(question=question, context=context)
    response = response_model.invoke([{"role":"user","content": prompt}])
return{"messages":[response]}

6.组装Graph

将所有节点组装为Graph图：

from langgraph.graphimportStateGraph,START,END
from langgraph.prebuiltimportToolNode
from langgraph.prebuiltimport tools_condition

workflow =StateGraph(MessagesState)

# Define the nodes we will cycle between
workflow.add_node(generate_query_or_respond)
workflow.add_node("retrieve",ToolNode([retriever_tool]))
workflow.add_node(rewrite_question)
workflow.add_node(generate_answer)

workflow.add_edge(START,"generate_query_or_respond")

# Decide whether to retrieve
workflow.add_conditional_edges(
"generate_query_or_respond",
    # AssessLLMdecision(call `retriever_tool` tool or respond to the user)
    tools_condition,
{
        # Translate the condition outputs to nodes in our graph
"tools":"retrieve",
END:END,
},
)

# Edges taken after the `action` node is called.
workflow.add_conditional_edges(
"retrieve",
    # Assess agent decision
    grade_documents,
)
workflow.add_edge("generate_answer",END)
workflow.add_edge("rewrite_question","generate_query_or_respond")

# Compile
graph = workflow.compile()