langgraph의 공식문서를 번역해 놓은 자료입니다. 예제 일부는 변경하였고, 필요한 경우 부연 설명을 추가하였습니다. 문제가 되면 삭제하겠습니다.
https://langchain-ai.github.io/langgraph/tutorials/rag/langgraph_adaptive_rag/
Adaptive RAG는 (1) 쿼리 분석과 (2) active/self-corrective RAG를 결합한 RAG를 위한 전략이다.
논문에서는 쿼리 분석을 통해 다음과 같이 경로를 설정한다고 보고한다.
- No Retrieval
- Single-shot RAG
- Iterative RAG
이것을 LangGraph를 사용하여 발전시켜 보자.
아래 구현에서는 다음과 같이 경로를 설정할 것이다.
- 웹 검색(Web search): 최근 사건(event)과 관련된 질문
- 자기 수정(Self-corrective) RAG: 인덱스 관련 질문
Index 생성
### Build Index
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
### from langchain_cohere import CohereEmbeddings
# Set embeddings
embd = OpenAIEmbeddings()
# Docs to index
urls = [
"https://namu.wiki/w/%EC%95%BC%EA%B5%AC/%EA%B2%BD%EA%B8%B0%20%EB%B0%A9%EC%8B%9D" # 야구/경기 방식
]
# Load
docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]
# Split
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
chunk_size=500, chunk_overlap=0
)
doc_splits = text_splitter.split_documents(docs_list)
# Add to vectorstore
vectorstore = Chroma.from_documents(
documents=doc_splits,
collection_name="rag-chroma",
embedding=embd,
)
retriever = vectorstore.as_retriever()
LLMs
### Router
from typing import Literal
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
question = "야구에서 홈런?"
# Data model
class RouteQuery(BaseModel):
"""Route a user query to the most relevant datasource."""
datasource: Literal["vectorstore", "web_search"] = Field(
...,
description="Given a user question choose to route it to web search or a vectorstore.",
)
# 함수 호출
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm_router = llm.with_structured_output(RouteQuery)
# Prompt
system = """You are an expert at routing a user question to a vectorstore or web search.
The vectorstore contains documents related to baseball.
Use the vectorstore for questions on these topics. Otherwise, use web-search."""
route_prompt = ChatPromptTemplate.from_messages(
[
("system", system),
("human", "{question}"),
]
)
question_router = route_prompt | structured_llm_router
print(question_router.invoke({"question": question}))
RouteQueyr 노드에서는 야구와 관련된 질문은 vectorstore를 검색하고 그 외는 web_search를 검색한다.
"야구에서 홈런"를 실행해보면 아래와 같이 vectorstore로 나온다.
datasource='vectorstore'
### Retrieval Grader
# Data model
class GradeDocuments(BaseModel):
"""Binary score for relevance check on retrieved documents."""
binary_score: str = Field(
description="Documents are relevant to the question, 'yes' or 'no'"
)
# LLM with function call
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeDocuments)
# Prompt
system = """You are a grader assessing relevance of a retrieved document to a user question. \n
If the document contains keyword(s) or semantic meaning related to the user question, grade it as relevant. \n
It does not need to be a stringent test. The goal is to filter out erroneous retrievals. \n
Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question."""
grade_prompt = ChatPromptTemplate.from_messages(
[
("system", system),
("human", "Retrieved document: \n\n {document} \n\n User question: {question}"),
]
)
retrieval_grader = grade_prompt | structured_llm_grader
question = "야구에서 홈런"
docs = retriever.invoke(question)
doc_txt = docs[1].page_content
print(retrieval_grader.invoke({"question": question, "document": doc_txt}))
문서와의 관련성 평가를 한다.
binary_score='yes'
### Generate
from langchain import hub
from langchain_core.output_parsers import StrOutputParser
# Prompt
prompt = hub.pull("rlm/rag-prompt")
# LLM
llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0)
# Post-processing
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
# Chain
rag_chain = prompt | llm | StrOutputParser()
# Run
generation = rag_chain.invoke({"context": docs, "question": question})
print(generation)
야구에서 홈런은 타자가 공을 멀리 쳐서 외야를 넘어가는 경우를 말합니다. 홈런이 발생하면 주자가 홈 플레이트를 밟아 득점하게 되며, 만루홈런의 경우 최대 4점을 얻을 수 있습니다. 홈런은 현대 야구에서 중요한 득점 방법 중 하나입니다.
### Hallucination Grader
# Data model
class GradeHallucinations(BaseModel):
"""Binary score for hallucination present in generation answer."""
binary_score: str = Field(
description="Answer is grounded in the facts, 'yes' or 'no'"
)
# LLM with function call
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeHallucinations)
# Prompt
system = """You are a grader assessing whether an LLM generation is grounded in / supported by a set of retrieved facts. \n
Give a binary score 'yes' or 'no'. 'Yes' means that the answer is grounded in / supported by the set of facts."""
hallucination_prompt = ChatPromptTemplate.from_messages(
[
("system", system),
("human", "Set of facts: \n\n {documents} \n\n LLM generation: {generation}"),
]
)
hallucination_grader = hallucination_prompt | structured_llm_grader
hallucination_grader.invoke({"documents": docs, "generation": generation})
답변이 generation된 문서를 기반으로 만들어졌는지 hallucination 평가를 한다.
binary_score='yes'
### Answer Grader
# Data model
class GradeAnswer(BaseModel):
"""Binary score to assess answer addresses question."""
binary_score: str = Field(
description="Answer addresses the question, 'yes' or 'no'"
)
# LLM with function call
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeAnswer)
# Prompt
system = """You are a grader assessing whether an answer addresses / resolves a question \n
Give a binary score 'yes' or 'no'. Yes' means that the answer resolves the question."""
answer_prompt = ChatPromptTemplate.from_messages(
[
("system", system),
("human", "User question: \n\n {question} \n\n LLM generation: {generation}"),
]
)
answer_grader = answer_prompt | structured_llm_grader
answer_grader.invoke({"question": question, "generation": generation})
질문의 문제를 해결하는데 적절한 답변인지 체크를 한다.
binary_score='yes'
### Question Re-writer
# LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Prompt
system = """You a question re-writer that converts an input question to a better version that is optimized \n
for vectorstore retrieval. Look at the input and try to reason about the underlying semantic intent / meaning."""
re_write_prompt = ChatPromptTemplate.from_messages(
[
("system", system),
(
"human",
"Here is the initial question: \n\n {question} \n Formulate an improved question.",
),
]
)
question_rewriter = re_write_prompt | llm | StrOutputParser()
question_rewriter.invoke({"question": question})
질문의 의도와 의미를 강조하도록 질문 재작성을 한다.
"야구에서 홈런" --> "야구에서 홈런의 정의와 중요성은 무엇인가요?"으로 재작성이 되었다.
야구에서 홈런의 정의와 중요성은 무엇인가요?
Web 검색 도구
### Search
from langchain_community.tools.tavily_search import TavilySearchResults
web_search_tool = TavilySearchResults(k=3)
Graph 구성
Graph 상태 정의
from typing import List
from typing_extensions import TypedDict
class GraphState(TypedDict):
"""
Represents the state of our graph.
Attributes:
question: question
generation: LLM generation
documents: list of documents
"""
question: str
generation: str
documents: List[str]
Graph Flow 정의
from langchain.schema import Document
def retrieve(state):
"""
Retrieve documents
Args:
state (dict): The current graph state
Returns:
state (dict): New key added to state, documents, that contains retrieved documents
"""
print("---RETRIEVE---")
question = state["question"]
# Retrieval
documents = retriever.invoke(question)
return {"documents": documents, "question": question}
def generate(state):
"""
Generate answer
Args:
state (dict): The current graph state
Returns:
state (dict): New key added to state, generation, that contains LLM generation
"""
print("---GENERATE---")
question = state["question"]
documents = state["documents"]
# RAG generation
generation = rag_chain.invoke({"context": documents, "question": question})
return {"documents": documents, "question": question, "generation": generation}
def grade_documents(state):
"""
Determines whether the retrieved documents are relevant to the question.
Args:
state (dict): The current graph state
Returns:
state (dict): Updates documents key with only filtered relevant documents
"""
print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
question = state["question"]
documents = state["documents"]
# Score each doc
filtered_docs = []
for d in documents:
score = retrieval_grader.invoke(
{"question": question, "document": d.page_content}
)
grade = score.binary_score
if grade == "yes":
print("---GRADE: DOCUMENT RELEVANT---")
filtered_docs.append(d)
else:
print("---GRADE: DOCUMENT NOT RELEVANT---")
continue
return {"documents": filtered_docs, "question": question}
def transform_query(state):
"""
Transform the query to produce a better question.
Args:
state (dict): The current graph state
Returns:
state (dict): Updates question key with a re-phrased question
"""
print("---TRANSFORM QUERY---")
question = state["question"]
documents = state["documents"]
# Re-write question
better_question = question_rewriter.invoke({"question": question})
return {"documents": documents, "question": better_question}
def web_search(state):
"""
Web search based on the re-phrased question.
Args:
state (dict): The current graph state
Returns:
state (dict): Updates documents key with appended web results
"""
print("---WEB SEARCH---")
question = state["question"]
# Web search
docs = web_search_tool.invoke({"query": question})
web_results = "\n".join([d["content"] for d in docs])
web_results = Document(page_content=web_results)
return {"documents": web_results, "question": question}
### Edges ###
def route_question(state):
"""
Route question to web search or RAG.
Args:
state (dict): The current graph state
Returns:
str: Next node to call
"""
print("---ROUTE QUESTION---")
question = state["question"]
source = question_router.invoke({"question": question})
if source.datasource == "web_search":
print("---ROUTE QUESTION TO WEB SEARCH---")
return "web_search"
elif source.datasource == "vectorstore":
print("---ROUTE QUESTION TO RAG---")
return "vectorstore"
def decide_to_generate(state):
"""
Determines whether to generate an answer, or re-generate a question.
Args:
state (dict): The current graph state
Returns:
str: Binary decision for next node to call
"""
print("---ASSESS GRADED DOCUMENTS---")
state["question"]
filtered_documents = state["documents"]
if not filtered_documents:
# All documents have been filtered check_relevance
# We will re-generate a new query
print(
"---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---"
)
return "transform_query"
else:
# We have relevant documents, so generate answer
print("---DECISION: GENERATE---")
return "generate"
def grade_generation_v_documents_and_question(state):
"""
Determines whether the generation is grounded in the document and answers question.
Args:
state (dict): The current graph state
Returns:
str: Decision for next node to call
"""
print("---CHECK HALLUCINATIONS---")
question = state["question"]
documents = state["documents"]
generation = state["generation"]
score = hallucination_grader.invoke(
{"documents": documents, "generation": generation}
)
grade = score.binary_score
# Check hallucination
if grade == "yes":
print("---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---")
# Check question-answering
print("---GRADE GENERATION vs QUESTION---")
score = answer_grader.invoke({"question": question, "generation": generation})
grade = score.binary_score
if grade == "yes":
print("---DECISION: GENERATION ADDRESSES QUESTION---")
return "useful"
else:
print("---DECISION: GENERATION DOES NOT ADDRESS QUESTION---")
return "not useful"
else:
pprint("---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS, RE-TRY---")
return "not supported"
Graph 컴파일
from langgraph.graph import END, StateGraph, START
workflow = StateGraph(GraphState)
# Define the nodes
workflow.add_node("web_search", web_search) # web search
workflow.add_node("retrieve", retrieve) # retrieve
workflow.add_node("grade_documents", grade_documents) # grade documents
workflow.add_node("generate", generate) # generatae
workflow.add_node("transform_query", transform_query) # transform_query
# Build graph
workflow.add_conditional_edges(
START,
route_question,
{
"web_search": "web_search",
"vectorstore": "retrieve",
},
)
workflow.add_edge("web_search", "generate")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
"grade_documents",
decide_to_generate,
{
"transform_query": "transform_query",
"generate": "generate",
},
)
workflow.add_edge("transform_query", "retrieve")
workflow.add_conditional_edges(
"generate",
grade_generation_v_documents_and_question,
{
"not supported": "generate",
"useful": END,
"not useful": "transform_query",
},
)
# Compile
app = workflow.compile()
Graph 사용하기
from pprint import pprint
# Run
inputs = {
"question": "야구에서 홈런은 뭐야?"
}
for output in app.stream(inputs):
for key, value in output.items():
pprint(f"Node '{key}':")
pprint("\n---\n")
# Final generation
pprint(value["generation"])
아래는 실행된 로그를 출력한 내용이다.
# 질문을 라우팅 요청을 한다
---ROUTE QUESTION---
# 질문과 vectorstore와의 관련성이 있으므로 RAG로 라우팅한다.
---ROUTE QUESTION TO RAG---
# 검색을 수행한다.
---RETRIEVE---
"Node 'retrieve':"
'\n---\n'
# 검색된 문서와 질문의 관련성 평가를 한다.
---CHECK DOCUMENT RELEVANCE TO QUESTION---
# 관련 없음
---GRADE: DOCUMENT NOT RELEVANT---
# 관련 없음
---GRADE: DOCUMENT NOT RELEVANT---
# 관련 없음
---GRADE: DOCUMENT NOT RELEVANT---
# 관련 있는 문서 검색
---GRADE: DOCUMENT RELEVANT---
# 관련된 문서가 1개 이상 발견되었으므로 generate로 이동한다
---ASSESS GRADED DOCUMENTS---
# 답변을 생성한다.
---DECISION: GENERATE---
"Node 'grade_documents':"
'\n---\n'
---GENERATE---
# hallucination 체크를 한다.
---CHECK HALLUCINATIONS---
# 문서 내용이 질문의 내용을 설명하고 있다.
---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---
---GRADE GENERATION vs QUESTION---
# generation이 질문을 나타낸다. (useful)
---DECISION: GENERATION ADDRESSES QUESTION---
"Node 'generate':"
'\n---\n'
('야구에서 홈런은 타자가 공을 멀리 쳐서 외야를 넘어가는 경우를 말합니다. 주자가 있을 경우, 홈런을 치면 주자들이 순서대로 진루하여 '
'득점할 수 있습니다. 만루홈런을 치면 최대 4점을 득점할 수 있습니다.')
vectorstore와 관련없는 질문
from pprint import pprint
# Run
inputs = {
"question": "축구에서 파울은 뭐야?"
}
for output in app.stream(inputs):
for key, value in output.items():
pprint(f"Node '{key}':")
pprint("\n---\n")
# Final generation
pprint(value["generation"])
실행결과
---ROUTE QUESTION---
---ROUTE QUESTION TO WEB SEARCH---
---WEB SEARCH---
"Node 'web_search':"
'\n---\n'
---GENERATE---
---CHECK HALLUCINATIONS---
---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---
---GRADE GENERATION vs QUESTION---
---DECISION: GENERATION ADDRESSES QUESTION---
"Node 'generate':"
'\n---\n'
('축구에서 파울은 상대 선수의 플레이에 지장을 주는 행위로 정의됩니다. 대표적인 예로는 상대를 밀거나 차는 행위, 발을 걸거나 잡아당기는 '
'행위, 핸드볼 등이 있습니다. 이러한 규정은 공정하고 안전한 경기를 위해 국제 축구 연맹(FIFA)에 의해 정해집니다.')
여기서는 최초 라우팅 부터 Web search로 향했다. vectorstore가 야구와 관련된 내용인데, 축구를 질문했으니 당연히 vectorstore에는 없다고 판단한다.
vectorstore에 관련성만 있고 답은 없는 질문
from pprint import pprint
# Run
inputs = {
"question": "야구 선수 중에 가장 잘했던 선수는 누구야?"
}
for output in app.stream(inputs):
for key, value in output.items():
pprint(f"Node '{key}':")
pprint("\n---\n")
# Final generation
pprint(value["generation"])
vectorstore에는 야구의 규칙에 관련된 내용만 있다. 그래서 실제 답을 찾을 수 없다. 이를 실행하면 어떻게 되는지 알아보자.
실행결과
---ROUTE QUESTION---
---ROUTE QUESTION TO RAG---
---RETRIEVE---
"Node 'retrieve':"
'\n---\n'
---CHECK DOCUMENT RELEVANCE TO QUESTION---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---ASSESS GRADED DOCUMENTS---
---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---
"Node 'grade_documents':"
'\n---\n'
---TRANSFORM QUERY---
Better question: 역대 최고의 야구 선수는 누구인가요?
"Node 'transform_query':"
'\n---\n'
---RETRIEVE---
"Node 'retrieve':"
'\n---\n'
---CHECK DOCUMENT RELEVANCE TO QUESTION---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---ASSESS GRADED DOCUMENTS---
---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---
"Node 'grade_documents':"
'\n---\n'
---TRANSFORM QUERY---
Better question: 역대 최고의 야구 선수로 평가받는 인물은 누구인가요?
"Node 'transform_query':"
'\n---\n'
---RETRIEVE---
"Node 'retrieve':"
'\n---\n'
---CHECK DOCUMENT RELEVANCE TO QUESTION---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---ASSESS GRADED DOCUMENTS---
---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---
"Node 'grade_documents':"
'\n---\n'
---TRANSFORM QUERY---
Better question: 역대 최고의 야구 선수로 평가되는 인물은 누구이며, 그 이유는 무엇인가요?
"Node 'transform_query':"
'\n---\n'
---RETRIEVE---
"Node 'retrieve':"
'\n---\n'
---CHECK DOCUMENT RELEVANCE TO QUESTION---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---ASSESS GRADED DOCUMENTS---
---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---
"Node 'grade_documents':"
'\n---\n'
---TRANSFORM QUERY---
Better question: 역대 최고의 야구 선수로 평가받는 인물은 누구이며, 그 선수가 최고의 선수로 여겨지는 이유는 무엇인가요?
"Node 'transform_query':"
'\n---\n'
---RETRIEVE---
"Node 'retrieve':"
'\n---\n'
---CHECK DOCUMENT RELEVANCE TO QUESTION---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---ASSESS GRADED DOCUMENTS---
---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---
"Node 'grade_documents':"
'\n---\n'
---TRANSFORM QUERY---
Better question: 역대 최고의 야구 선수로 평가받는 인물은 누구이며, 그 선수가 최고의 선수로 여겨지는 이유는 무엇인지 구체적인 예시와 함께 설명해 주실 수 있나요?
"Node 'transform_query':"
'\n---\n'
---RETRIEVE---
"Node 'retrieve':"
'\n---\n'
---CHECK DOCUMENT RELEVANCE TO QUESTION---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---ASSESS GRADED DOCUMENTS---
---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---
"Node 'grade_documents':"
'\n---\n'
---TRANSFORM QUERY---
Better question: 역대 최고의 야구 선수로 평가받는 인물은 누구이며, 그 선수가 최고의 선수로 여겨지는 이유를 구체적인 예시와 함께 설명해 주실 수 있나요?
"Node 'transform_query':"
'\n---\n'
vectorstore를 검색하지만 실제로 관련있는 문서를 찾지 못해서 질문을 계속 rewrite 하는 것을 확인할 수 있다. 하지만 rewrite가 되더라도 적절한 문서를 찾지 못해서 계속 실행되는 현상이 발생한다. (무한루프)
[Rewrite 질문 목록]
- 야구 선수 중에 가장 잘했던 선수는 누구야?
- 역대 최고의 야구 선수는 누구인가요?
- 역대 최고의 야구 선수로 평가받는 인물은 누구인가요?
- 역대 최고의 야구 선수로 평가되는 인물은 누구이며, 그 이유는 무엇인가요?
- 역대 최고의 야구 선수로 평가받는 인물은 누구이며, 그 선수가 최고의 선수로 여겨지는 이유는 무엇인가요?
- 역대 최고의 야구 선수로 평가받는 인물은 누구이며, 그 선수가 최고의 선수로 여겨지는 이유는 무엇인지 구체적인 예시와 함께 설명해 주실 수 있나요?
- 역대 최고의 야구 선수로 평가받는 인물은 누구이며, 그 선수가 최고의 선수로 여겨지는 이유를 구체적인 예시와 함께 설명해 주실 수 있나요?
이런 경우는 무한루프 발생으로 문제가 될 수 있다. 그래서 순환 횟수를 recursion_limit을 사용해서 제한해야 한다.
inputs = {"question": "야구 선수 중에 가장 잘했던 선수는 누구야?"}
for output in app.stream(inputs, {"recursion_limit": 5}):
for key, value in output.items():
pprint(f"Node '{key}':")
pprint("\n---\n")
recursion_limit
는 실행되는 node 기준
으로 횟수를 제한한다.