使用LangChain和LangGraph大幅提升RAG效果
一、簡介
LangGraph是LangChain、LangServe和LangSmith系列的最新成員，旨在使用LLM構(gòu)建生成式人工智能應(yīng)用程序。請記住，所有這些都是獨立的包，必須單獨進行pip安裝。
在深入學(xué)習(xí)LangGraph之前，需要了解LangChain的兩個主要概念。
1. 鏈：圍繞LLM編寫的程序，用于執(zhí)行任務(wù)，例如自動SQL編寫或NER提取鏈等。請注意，鏈不能用于任何其他任務(wù)（甚至不能用于一般用例），如果嘗試這樣做，可能會損壞鏈。鏈中要遵循的步驟是預(yù)定義的，不可靈活調(diào)整。
2. 代理：鏈的更加靈活版本，代理通常是啟用第三方工具（例如谷歌搜索、YouTube）的LLM，由LLM本身決定下一步如何解決給定的查詢。
現(xiàn)在，當(dāng)處理現(xiàn)實世界的問題時，一個常見的問題是希望找到介于鏈和代理之間的解決方案。即不像鏈那樣硬編碼，但也不像代理那樣完全由LLM驅(qū)動。
二、LangGraph
LangGraph是以LangChain為核心，用于創(chuàng)建工作流程中的循環(huán)圖的工具。因此，我們假設(shè)以下示例：
你希望在知識庫上搭建一個基于RAG的檢索系統(tǒng)?，F(xiàn)在，你希望引入這樣一種情況：如果RAG的輸出未滿足特定質(zhì)量要求，代理/鏈應(yīng)該再次檢索數(shù)據(jù)，但這次是自行更改提示。并且重復(fù)此過程，直到達到質(zhì)量閾值為止。
使用LangGraph可以實現(xiàn)這種循環(huán)邏輯。這只是一個示例，使用LangGraph還可以做更多事情。
注：可以將其視為向鏈中引入循環(huán)邏輯，使其成為循環(huán)鏈。
- LangGraph對于構(gòu)建Autogen或MetaGPT等多代理應(yīng)用程序至關(guān)重要。
顧名思義，LangGraph具有一般圖形所具有的所有組件，例如節(jié)點、邊等，接下來通過一個示例來了解。
三、使用LangGraph改善RAG
在此示例中，希望將RAG系統(tǒng)在數(shù)據(jù)庫中的最終輸出減少到不超過30個字符。如果輸出長度大于30個字符，則希望引入循環(huán)，使用不同的提示再次嘗試，直到長度小于30個字符為止。這是一個演示目的的基本邏輯。你甚至可以實現(xiàn)復(fù)雜的邏輯來改善RAG結(jié)果。
我們將創(chuàng)建的圖形如下所示。
圖片
此處使用的版本為 langchain===0.0.349， openai===1.3.8， langgraph===0.0.26。
3.1 首先，讓我們導(dǎo)入重要的內(nèi)容并初始化LLM。這里使用的是OpenAI API，但你也可以使用其他LLM。
from typing import Dict, TypedDict, Optional
from langgraph.graph import StateGraph, END
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.embeddings.openai import OpenAIEmbeddings
llm = OpenAI(openai_api_key='your API')接下來，我們將定義一個StateGraph。
class GraphState(TypedDict):
question: Optional[str] = None
classification: Optional[str] = None
response: Optional[str] = None
length: Optional[int] = None
greeting: Optional[str] = None
workflow = StateGraph(GraphState)什么是StateGraph？
StateGraph是任何LangGraph流程的核心，它存儲了在執(zhí)行工作流程時我們將存儲的各種變量的狀態(tài)。在本例中，我們有5個變量，其值在執(zhí)行圖形時將進行更新，并將與所有邊和節(jié)點共享。
3.2 接下來，讓我們從現(xiàn)有向量數(shù)據(jù)庫中初始化一個RAG檢索鏈。代碼已在以下視頻中進行了解釋。
def retriever_qa_creation():
embeddings = OpenAIEmbeddings()
db = Chroma(embedding_functinotallow=embeddings,persist_directory='/database',collection_name='details')
qa = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=db.as_retriever())
return qa
rag_chain = retriever_qa_creation()3.3 接下來，我們將向該圖形添加節(jié)點。
def classify(question):
return llm("classify intent of given input as greeting or not_greeting. Output just the class.Input:{}".format(question)).strip()
def classify_input_node(state):
question = state.get('question', '').strip()
classification = classify(question)
return {"classification": classification}
def handle_greeting_node(state):
return {"greeting": "Hello! How can I help you today?"}
def handle_RAG(state):
question = state.get('question', '').strip()
prompt = question
if state.get("length")<30:
search_result = rag_chain.run(prompt)
else:
search_result = rag_chain.run(prompt+'. Return total count only.')
return {"response": search_result,"length":len(search_result)}
def bye(state):
return{"greeting":"The graph has finished"}
workflow.add_node("classify_input", classify_input_node)
workflow.add_node("handle_greeting", handle_greeting_node)
workflow.add_node("handle_RAG", handle_RAG)
workflow.add_node("bye", bye)這需要一些解釋。
- 每個節(jié)點都是一個Python函數(shù)，它可以：
①讀取任何狀態(tài)變量。
②更新任何狀態(tài)變量。在這種情況下，每個節(jié)點的返回函數(shù)都會更新某個或某些狀態(tài)變量的狀態(tài)/值。
- 使用state.get()來讀取任何狀態(tài)變量。
- handle_RAG節(jié)點可以幫助我們實現(xiàn)我們希望的循環(huán)自定義邏輯。如果輸出的長度<30，則使用提示符A；否則使用提示符B。對于第一種情況（當(dāng)RAG節(jié)點尚未執(zhí)行時），我們將傳遞length=0，并提供一個提示。
3.4 接下來，我們將添加入口點和邊緣。
workflow.set_entry_point("classify_input")
workflow.add_edge('handle_greeting', END)
workflow.add_edge('bye', END)在上述的代碼片段中，
- 我們向圖中添加了一個入口點，即無論輸入提示是什么，都會執(zhí)行的第一個節(jié)點函數(shù)。
- A節(jié)點和B節(jié)點之間的邊界定義了在節(jié)點A之后執(zhí)行節(jié)點B。在這種情況下，如果在我們的工作流中出現(xiàn)了handle_greeting或bye，則圖形應(yīng)該END（一個特殊的節(jié)點來終止工作流）。
3.5 接下來，讓我們添加條件邊界。
def decide_next_node(state):
return "handle_greeting" if state.get('classification') == "greeting" else "handle_RAG"
def check_RAG_length(state):
return "handle_RAG" if state.get("length")>30 else "bye"
workflow.add_conditional_edges(
"classify_input",
decide_next_node,
{
"handle_greeting": "handle_greeting",
"handle_RAG": "handle_RAG"
}
)
workflow.add_conditional_edges(
"handle_RAG",
check_RAG_length,
{
"bye": "bye",
"handle_RAG": "handle_RAG"
}
)條件邊界可根據(jù)條件（例如if-else）在兩個節(jié)點之間進行選擇。在創(chuàng)建的兩個條件邊界中：
第一個條件邊界
當(dāng)遇到classify_input時，根據(jù)decide_next_node函數(shù)的輸出選擇handle_greeting或handle_RAG。
第二個條件邊界
如果遇到handle_RAG，則根據(jù)check_RAG_length條件選擇handle_RAG或bye。
3.6 編譯并調(diào)用提示。初始時保持length變量設(shè)置為0。
app = workflow.compile()
app.invoke({'question':'Mehul developed which projects?','length':0})# 輸出
{'question': 'Mehul developed which projects?',
'classification': 'not_greeting',
'response': ' 4',
'length': 2,
'greeting': 'The graph has finished'}對于上述提示，圖形流程如下所示：
classify_input: 情感將為not_greeting。
由于第一個條件邊界，移至handle_RAG。
由于length=0，使用第一個提示并檢索答案（總長度將大于30）。
由于第二個條件邊界，再次移至handle_RAG。
由于length>30，使用第二個提示符。
由于第二個條件邊界，移至bye。
END。
如果沒有使用LangGraph：
rag_chain.run("Mehul developed which projects?")
# 輸出
"Mehul developed projects like ABC, XYZ, QWERTY. Not only these, he has major contribution in many other projects as well at OOO organization"3.7 下一個輸入。
app.invoke({'question':'Hello bot','length':0})
# 輸出
{'question': 'Hello bot',
'classification': 'greeting',
'response': None,
'length': 0,
'greeting': 'Hello! How can I help you today?'}這里的流程會更簡單。
classify_input: 情感將為greeting。
由于第一個條件邊界，移至handle_greeting。
END。
雖然我在這里應(yīng)用的條件相當(dāng)簡單，但通過添加更復(fù)雜的條件，這個框架可以很容易地用于改進你的結(jié)果。
