这篇文章来介绍下使用 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()
