ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-style: normal;font-variant-ligatures: normal;font-variant-caps: normal;letter-spacing: normal;orphans: 2;text-align: justify;text-indent: 0px;text-transform: none;widows: 2;word-spacing: 0px;-webkit-text-stroke-width: 0px;white-space: normal;background-color: rgb(255, 255, 255);text-decoration-thickness: initial;text-decoration-style: initial;text-decoration-color: initial;">2.创建检索工具fromlangchain_community.document_loadersimportWebBaseLoader
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()forurlinurls]
对文档数据进行切分：
fromlangchain_text_splittersimportRecursiveCharacterTextSplitter
docs_list=[itemforsublistindocsforiteminsublist]
text_splitter=RecursiveCharacterTextSplitter.from_tiktoken_encoder(
chunk_size=100,chunk_overlap=50
)
doc_splits=text_splitter.split_documents(docs_list)
使用阿里QianWen的embedding模型将文档数据转化为向量，存储到内存向量数据库
fromlangchain_core.vectorstoresimportInMemoryVectorStore
fromlangchain_community.embeddingsimportDashScopeEmbeddings
vectorstore=InMemoryVectorStore.from_documents(
documents=doc_splits,embedding=DashScopeEmbeddings(model="text-embedding-v3")
)
retriever=vectorstore.as_retriever()
创建检索工具：
fromlangchain.tools.retrieverimportcreate_retriever_tool
retriever_tool=create_retriever_tool(
retriever,
"retrieve_blog_posts",
"Search and return information about Lilian Weng blog posts.",
)
使用阿里QianWen模型作为LLM，构建generate_query_or_respond节点
fromlanggraph.graphimportMessagesState
response_model=ChatTongyi(model="qwen-plus")
defgenerate_query_or_respond(state:MessagesState):
"""Callthe 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]}
定义grade_documents节点： 定义GradeDocumentsclass,使用QianWen模型结构化输出（返回yes和no），对检索工具的结果进行评分，如果返回yes则返回generate_answer节点，否则返回rewrite_question节点。
frompydanticimportBaseModel,Field
fromtypingimportLiteral
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")
defgrade_documents(
state:MessagesState,
)->Literal["generate_answer","rewrite_question"]:
"""Determine whether the retrieved documents are relevant to the question."""
formessageinstate["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
ifscore=="yes":
return"generate_answer"
else:
return"rewrite_question"
定义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:"
)
defrewrite_question(state:MessagesState):
"""Rewrite the original user question."""
formessageinstate["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}]}
定义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}"
)
defgenerate_answer(state:MessagesState):
"""Generate an answer."""
formessageinstate["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]}
将所有节点组装为Graph图：
fromlanggraph.graphimportStateGraph,START,END
fromlanggraph.prebuiltimportToolNode
fromlanggraph.prebuiltimporttools_condition
workflow=StateGraph(MessagesState)
#Definethe 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")
#Decidewhether to retrieve
workflow.add_conditional_edges(
"generate_query_or_respond",
#AssessLLMdecision(call`retriever_tool`tool or respond to the user)
tools_condition,
{
#Translatethe condition outputs to nodesinour graph
"tools":"retrieve",
END:END,
},
)
#Edgestaken after the`action`node is called.
workflow.add_conditional_edges(
"retrieve",
#Assessagent decision
grade_documents,
)
workflow.add_edge("generate_answer",END)
workflow.add_edge("rewrite_question","generate_query_or_respond")
#Compile
graph=workflow.compile()
| 欢迎光临 链载Ai (https://www.lianzai.com/) | Powered by Discuz! X3.5 |