GitHub 소스코드 기반 Q&A 챗봇
랭그래프 리포지토리를 적당한 폴더에 클론합니다.
git clone https://github.com/langchain-ai/langgraph%pip install -qU unstructured langchain_text_splitters langchain-community langchain_cohere faiss-cpu /Users/jeongsk/Workspace/Wantedlab/langchain-academy/.venv/bin/python: No module named pip
Note: you may need to restart the kernel to use updated packages.
환경변수
from dotenv import load_dotenv
load_dotenv("../../.env", override=True)True
import os
import getpass
def _set_env(var: str):
env_value = os.environ.get(var)
if not env_value:
env_value = getpass.getpass(f"{var}: ")
os.environ[var] = env_value
_set_env("LANGSMITH_API_KEY")
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "langchain-academy"
_set_env("OPENAI_API_KEY")데이터 전처리
먼저, 다운로드 받은 langgraph의 GitHub 저장소에서 파일을 로드합니다.
여기서 로드하는 파일은 파이썬 파일(.py), 마크다운 파일(.md), 노트북 파일(.ipynb)입니다.
ROOT_PATH = "../../langgraph"
libs_path = os.path.join(ROOT_PATH, "libs")
docs_path = os.path.join(ROOT_PATH, "docs")
examples_path = os.path.join(ROOT_PATH, "examples")
all_repos = [libs_path, docs_path, examples_path]문서 로더
from langchain_core.documents import Document
from langchain_community.document_loaders import (
DirectoryLoader,
UnstructuredMarkdownLoader,
NotebookLoader,
)
from langchain_community.document_loaders.generic import GenericLoader
from langchain_community.document_loaders.parsers import LanguageParser
# 파이썬 파일 로더
def load_python_files(paths: list[str]) -> list[Document]:
documents: list[Document] = []
for path in paths:
loader = GenericLoader.from_filesystem(
path,
glob="**/[!.]*",
suffixes=[".py"],
show_progress=True,
parser=LanguageParser(language="python", parser_threshold=30),
)
documents.extend(loader.load())
return documents
# 마크다운 파일 로더
def load_markdown_files(paths: list[str]) -> list[Document]:
documents: list[Document] = []
for path in paths:
try:
loader = DirectoryLoader(
path,
glob="**/*.md",
loader_cls=UnstructuredMarkdownLoader,
loader_kwargs={"mode": "single"},
recursive=True,
)
documents.extend(loader.load())
except Exception as e:
print(f"Error loading Markdown files from {path}: {str(e)}")
continue
return documents
# 주피터 노트북 파일 로더
def load_notebook_files(paths: list[str]) -> list[Document]:
documents: list[Document] = []
exclude_dirs = [
".git",
"__pycache__",
".ipynb_checkpoints",
"node_modules",
".venv",
"venv",
]
notebook_files = []
for path in paths:
for root, dirs, files in os.walk(path):
# 제외할 디렉토리 필터링
dirs[:] = [d for d in dirs if d not in exclude_dirs]
# .ipynb 파일 찾기
for file in files:
if file.endswith(".ipynb"):
full_path = os.path.join(root, file)
notebook_files.append(full_path)
for file in notebook_files:
loader = NotebookLoader(
file,
include_outputs=False,
max_output_length=10,
remove_newline=True,
traceback=False,
)
documents.extend(loader.load())
return documentspython_documents = load_python_files(all_repos)
print(len(python_documents))/Users/jeongsk/Workspace/Wantedlab/langchain-academy/.venv/lib/python3.12/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html
from .autonotebook import tqdm as notebook_tqdm
64%|██████▎ | 149/234 [00:00<00:00, 228.05it/s]<unknown>:39: SyntaxWarning: invalid escape sequence '\('
<unknown>:39: SyntaxWarning: invalid escape sequence '\('
<unknown>:39: SyntaxWarning: invalid escape sequence '\('
100%|██████████| 234/234 [00:00<00:00, 264.95it/s]
100%|██████████| 25/25 [00:00<00:00, 753.17it/s]
100%|██████████| 1/1 [00:00<00:00, 344.84it/s]
1894
markdown_documents = load_markdown_files(all_repos)
print(len(markdown_documents))174
notebook_documents = load_notebook_files(all_repos)
print(len(notebook_documents))111
all_docs = python_documents + markdown_documents + notebook_documents
print(len(all_docs))2179
문서 분할
from langchain_text_splitters import (
PythonCodeTextSplitter,
MarkdownTextSplitter,
RecursiveCharacterTextSplitter,
Language,
)
# 파이썬 코드 스플릿터
python_splitter = PythonCodeTextSplitter(
chunk_size=1000,
chunk_overlap=200,
)
# 마크다운 스플릿터
markdown_splitter = MarkdownTextSplitter(
chunk_size=1000,
chunk_overlap=200,
)
# 주피터 노트북 스필릿터
notebook_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=RecursiveCharacterTextSplitter.get_separators_for_language(
Language.PYTHON
)
+ RecursiveCharacterTextSplitter.get_separators_for_language(Language.MARKDOWN),
length_function=len,
)# 파이썬 문서 분할
splited_python_documents = python_splitter.split_documents(python_documents)
len(splited_python_documents)5681
# 마크다운 문서 분할
splited_markdown_documents = markdown_splitter.split_documents(markdown_documents)
len(splited_markdown_documents)2377
# 주피터 노트북 문서 분할
splited_notebook_documents = notebook_splitter.split_documents(notebook_documents)
len(splited_notebook_documents)1261
all_splited_documents = (
splited_python_documents + splited_markdown_documents + splited_python_documents
)
len(all_splited_documents)13739
임베딩 및 벡터 DB에 저장
CacheBackedEmbeddings을 사용하면 임베딩을 다시 계산할 필요없이 일시적으로 캐시할 수 있습니다.
FAISS는 효율적인 유사성 검색 및 밀집 벡터 클러스터링을 위한 라이브러리입니다.
임베딩 모델 생성 및 임베딩 캐시
from langchain.storage import LocalFileStore
from langchain_openai import OpenAIEmbeddings
from langchain.embeddings import CacheBackedEmbeddings
# 임베딩 모델 생성
underlying_embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
store = LocalFileStore("./cache")
embeddings = CacheBackedEmbeddings.from_bytes_store(
underlying_embeddings,
store,
namespace=underlying_embeddings.model,
query_embedding_cache=True,
)/Users/jeongsk/Workspace/Wantedlab/langchain-academy/.venv/lib/python3.12/site-packages/langchain/embeddings/cache.py:58: UserWarning: Using default key encoder: SHA-1 is *not* collision-resistant. While acceptable for most cache scenarios, a motivated attacker can craft two different payloads that map to the same cache key. If that risk matters in your environment, supply a stronger encoder (e.g. SHA-256 or BLAKE2) via the `key_encoder` argument. If you change the key encoder, consider also creating a new cache, to avoid (the potential for) collisions with existing keys.
_warn_about_sha1_encoder()
# 임베딩하기 전에 캐시가 비어 있습니다.
# list(store.yield_keys())
# []벡터 저장소 초기화
# 벡터 스토어 생성
import faiss
from langchain_community.docstore.in_memory import InMemoryDocstore
from langchain_community.vectorstores import FAISS
index = faiss.IndexFlatL2(len(embeddings.embed_query("hello")))
vector_store = FAISS(
embedding_function=embeddings,
index=index,
docstore=InMemoryDocstore(),
index_to_docstore_id={},
)문서를 벡터 저장소에 추가
results = vector_store.add_documents(documents=all_splited_documents)
results[:5]['0e93a970-2a9f-42df-840f-e993d08ab152',
'5ef48fc5-d060-4eaf-a96c-26489dbbda5f',
'2e54b739-73d2-4710-af81-10e6eb0924ec',
'790c5a7e-59ae-400f-949b-218ccfb0f9b2',
'3f1e2c01-e15b-4ced-9a86-87f63dd0d96c']
벡터 스토어 저장 및 불러오기
FAISS 인덱스를 저장하고 불러올 수도 있습니다. 이렇게 하면 매번 사용할 때마다 새로 생성할 필요가 없어서 편리합니다.
# 벡터 스토어를 저장하고 불러오기
vector_store.save_local("faiss_index")
new_vector_store = FAISS.load_local(
"faiss_index",
embeddings,
allow_dangerous_deserialization=True,
)docs = new_vector_store.similarity_search("self-rag")
print(docs[0].page_content)def _strip_self(sig: inspect.Signature) -> inspect.Signature:
params = list(sig.parameters.values())
if params and params[0].name == "self":
params = params[1:]
return sig.replace(parameters=params)
RAG 체인 구축
여기서는 Naive RAG를 구성합니다. 그리고 검색 정확도를 향상시키기 위해 Cohere Rerank를 사용합니다.
retriever = vector_store.as_retriever(search_kwargs={"k": 200})retriever.invoke("self-rag")[:5][Document(id='3d70aeb9-c402-4f2b-9b10-2930058419ed', metadata={'source': '../../langgraph/libs/sdk-py/tests/test_api_parity.py', 'content_type': 'functions_classes', 'language': 'python'}, page_content='def _strip_self(sig: inspect.Signature) -> inspect.Signature:\n params = list(sig.parameters.values())\n if params and params[0].name == "self":\n params = params[1:]\n return sig.replace(parameters=params)'),
Document(id='d5bba9b5-6b90-4ecf-8796-73c63f7d9d53', metadata={'source': '../../langgraph/libs/sdk-py/tests/test_api_parity.py', 'content_type': 'functions_classes', 'language': 'python'}, page_content='def _strip_self(sig: inspect.Signature) -> inspect.Signature:\n params = list(sig.parameters.values())\n if params and params[0].name == "self":\n params = params[1:]\n return sig.replace(parameters=params)'),
Document(id='1837a9df-f44b-4b54-9e00-ffb17272c99d', metadata={'source': '../../langgraph/libs/checkpoint/langgraph/store/base/batch.py', 'content_type': 'functions_classes', 'language': 'python'}, page_content='return wrapper'),
Document(id='7f6d5219-0c43-43b4-b6e5-d77d6d3bdf15', metadata={'source': '../../langgraph/libs/checkpoint/langgraph/store/base/batch.py', 'content_type': 'functions_classes', 'language': 'python'}, page_content='return wrapper'),
Document(id='d155eaa0-9ab4-41a5-8e11-6094b8ccf8a4', metadata={'source': '../../langgraph/docs/docs/tutorials/workflows.md'}, page_content='```\n\n**LangSmith Trace**\n\nhttps://smith.langchain.com/public/86ab3e60-2000-4bff-b988-9b89a3269789/r\n\n**Resources:**\n\n**Examples**\n\n[Here](https://github.com/langchain-ai/local-deep-researcher) is an assistant that uses evaluator-optimizer to improve a report. See our video [here](https://www.youtube.com/watch?v=XGuTzHoqlj8).\n\n[Here](https://langchain-ai.github.io/langgraph/tutorials/rag/langgraph_adaptive_rag_local/) is a RAG workflow that grades answers for hallucinations or errors. See our video [here](https://www.youtube.com/watch?v=bq1Plo2RhYI).\n:::\n\n:::js\n```typescript\n// Graph state\nconst State = z.object({\n joke: z.string().optional(),\n topic: z.string(),\n feedback: z.string().optional(),\n funny_or_not: z.string().optional(),\n});')]
# Reranker 설정
from langchain_cohere import CohereRerank
from langchain.retrievers import ContextualCompressionRetriever
compressor = CohereRerank(model="rerank-v3.5", top_n=20)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=retriever,
)# prompt 설정
system_prompt = """당신은 CODE Copilot 어시스턴트입니다.
질문에 답변하기 위해 다음에 제시된 소스 코드 또는 문서의 일부를 반드시 사용해야 합니다.
RAG(Retrieval Augmented Generation) 소스 코드 및 문서와 관련된 질문이 주어집니다.
답변을 모르는 경우에는 모른다고 말하십시오.
답변은 한국어로 작성하십시오.
질문에 답변할 때 다음 지침을 따르십시오:
1. 제공된 CONTEXT 내의 정보만 사용하십시오.
2. 가능한 한 많은 예제 코드 조각을 포함하십시오.
3. 전체 코드 조각을 작성하는 것을 매우 권장합니다.
4. CONTEXT에 명시적으로 언급된 내용을 넘어 외부 정보를 도입하거나 추측하지 마십시오.
5. CONTEXT에는 각 개별 문서의 주제별 출처가 포함되어 있습니다.
6. 답변 내 관련 진술 옆에 해당 출처를 포함하십시오. 예를 들어 출처 #1의 경우 [1]을 사용하십시오.
7. 답변 하단에 출처를 순서대로 나열하십시오. [1] 출처 1, [2] 출처 2, 등
8. 출처가 <source>assistant/docs/llama3_1.md" page="7"</source>인 경우 다음과 같이 나열하십시오: [1] llama3_1.md
인용 시 괄호 추가 및 문서 출처 서두 문구를 생략하십시오.
----
### 출처
출처 섹션에서는:
- 답변에 사용된 모든 출처를 포함하십시오
- 관련 웹사이트 또는 문서 이름에 대한 전체 링크를 제공하십시오
- 각 출처는 새 줄로 구분하십시오. 마크다운에서 새 줄을 생성하려면 각 줄 끝에 공백 두 개를 사용하십시오.
- 다음과 같이 표시됩니다:
**출처**
- [1] 링크 또는 문서명
- [2] 링크 또는 문서명
출처를 반드시 통합하세요. 예를 들어 다음은 올바르지 않습니다:
- [3] https://ai.meta.com/blog/meta-llama-3-1/
- [4] https://ai.meta.com/blog/meta-llama-3-1/
중복 출처는 없어야 합니다. 다음과 같이 간결하게 작성하세요:
- [3] https://ai.meta.com/blog/meta-llama-3-1/
-----
### CONTEXT
질문에 답변할 때 사용할 수 있는 CONTEXT 입니다:
{context}
----
### 질문
사용자의 질문은 다음과 같습니다:
{question}
----
최종 검토 사항:
- 보고서가 요구되는 구조를 따르고 있는지 확인하세요
- 모든 가이드라인이 준수되었는지 확인하세요
- 해당되는 경우 답변에 전체 코드 스니펫이 포함되었는지 확인하세요
- 답변은 한국어로 작성되어야 합니다
- 많은 예제 코드 스니펫을 사용하면 사용자로부터 높은 평가를 받을 수 있습니다
- 단계별로 생각하세요.
----
질문에 대한 답변과 출처:"""from operator import itemgetter
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
template = ChatPromptTemplate(
[
("system", system_prompt),
("user", "{user_input}"),
]
)
llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
chain = (
{
"user_input": itemgetter("input"),
"context": itemgetter("context"),
}
| template
| llm
| StrOutputParser()
)LangGraph 구성하기
상태 정의
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
question: Annotated[str, "question"]
documents: Annotated[list[str], "documents"]Routing 노드
사용자 질문의 의도를 파악하고 routing를 수행합니다. 벡터 검색이 필요한지 아닌지 여부만 판단합니다.
from typing import Literal
from pydantic import BaseModel, Field
class RouteQuery(BaseModel):
binary_score: Annotated[
Literal[1, 0],
Field(
...,
description="사용자 질문이 벡터 스토어 검색이 필요한지 여부를 판단합니다."
"벡터 스토어에는 LangGraph 와 RAG(Retrieval Augmented Generation) 관련 소스 코드와 문서가 포함되어 있습니다."
"관련 있는 경우는 1를 반환합니다."
"질문이 소스 코드 또는 문서와 관련된지 판단할 수 없는 경우 1을 반환하십시오."
"답변을 모르는 경우 1를 반환하십시오."
"그 외에는 모두 0을 반환합니다.",
),
]
# 라우팅 노드
def routing_node(state):
routing_llm = llm.with_structured_output(RouteQuery)
response = routing_llm.invoke([state["messages"][-1]])
if response.binary_score == 1:
return "query_expansion"
else:
return "general_answer"routing_node({"messages": [("user", "self-rag에 대해서 설명해주세요")]})'query_expansion'
routing_node({"messages": [("user", "오늘 날씨는?")]})'general_answer'
질문 재작성 노드
from pydoc import describe
from langchain_core.messages import HumanMessage
class RewriteQuery(BaseModel):
question: Annotated[
str,
Field(
...,
description="질문 재작성 도구로, 입력된 질문을 CODE SEARCH(github repository)에 최적화된 더 나은 버전으로 변환합니다."
"기존 질문의 근본적인 의미적 의도/의미를 추론하세요."
"영어로 작성하세요.",
),
]
def rewrite_query_node(state: State):
question = state.get("question") if state.get("question") else state["messages"][-1]
rewriter_llm = llm.with_structured_output(RewriteQuery)
if isinstance(question, str):
response = rewriter_llm.invoke([HumanMessage(content=question)])
else:
response = rewriter_llm.invoke([question])
return {"question": response.question}rewrite_query_node({"messages": [("user", "self-rag?")]}){'question': "Could you please provide a detailed explanation of 'self-rag'? Additionally, if possible, include related concepts or methods to better understand its context and applications."}
평가 노드
문서의 관련성 여부와 환각 여부를 평가합니다.
- 검색된 문서의 관련성 평가
- 답변의 환각 여부 평가
- 답변 질문에 대한 관련성 평가
# 문서 평가를 위한 데이터 모델 정의
class GradeDocuments(BaseModel):
"""검색된 문서의 관련성 검증"""
binary_score: Annotated[
Literal[1, 0],
Field(
..., description="Context 문서가 질문과 관련이 있는가? 1 또는 0 으로 답변"
),
]
# 문서 관련성 평가 노드
def filtering_documents_node(state: State):
question = state.get("question")
documents = state.get("documents")
grader_llm = llm.with_structured_output(GradeDocuments)
filtered_docs = []
for doc in documents:
response = grader_llm.invoke(f"{question}\n\n{doc}")
if response.binary_score == 1:
filtered_docs.append(doc)
continue
return {"documents": filtered_docs}벡터 스토어 검색 노드
def retrieve_node(state: State):
question = state.get("question")
documents = compression_retriever.invoke(question)
return {"documents": documents}retrieve_node({"question": "self-rag에 대해서 설명해주세요."}){'documents': [Document(metadata={'source': '../../langgraph/libs/cli/js-examples/README.md', 'relevance_score': 0.12906231}, page_content='Add retrieval-augmented generation (RAG) capabilities by integrating external APIs or databases to provide more customized responses.\n\nDevelopment\n\nWhile iterating on your graph, you can edit past state and rerun your app from previous states to debug specific nodes. Local changes will be automatically applied via hot reload. Try experimenting with:\n\nModifying the system prompt to give your chatbot a unique personality.\n\nAdding new nodes to the graph for more complex conversation flows.\n\nImplementing conditional logic to handle different types of user inputs.\n\nFollow-up requests will be appended to the same thread. You can create an entirely new thread, clearing previous history, using the + button in the top right.\n\nFor more advanced features and examples, refer to the LangGraph.js documentation. These resources can help you adapt this template for your specific use case and build more sophisticated conversational agents.'),
Document(metadata={'source': '../../langgraph/docs/docs/tutorials/workflows.md', 'relevance_score': 0.10923949}, page_content='```\n\n**LangSmith Trace**\n\nhttps://smith.langchain.com/public/c4580b74-fe91-47e4-96fe-7fac598d509c/r\n\n**Resources:**\n\n**LangChain Academy**\n\nSee our lesson on routing [here](https://github.com/langchain-ai/langchain-academy/blob/main/module-1/router.ipynb).\n\n**Examples**\n\n[Here](https://langchain-ai.github.io/langgraph/tutorials/rag/langgraph_adaptive_rag_local/) is RAG workflow that routes questions. See our video [here](https://www.youtube.com/watch?v=bq1Plo2RhYI).\n:::\n\n:::js\n```typescript\nimport { SystemMessage, HumanMessage } from "@langchain/core/messages";\n\n// Schema for structured output to use as routing logic\nconst Route = z.object({\n step: z.enum(["poem", "story", "joke"]).describe("The next step in the routing process"),\n});\n\n// Augment the LLM with schema for structured output\nconst router = llm.withStructuredOutput(Route);\n\n// State\nconst State = z.object({\n input: z.string(),\n decision: z.string().optional(),\n output: z.string().optional(),\n});'),
Document(metadata={'source': '../../langgraph/docs/docs/tutorials/workflows.md', 'relevance_score': 0.103885114}, page_content='```\n\n**LangSmith Trace**\n\nhttps://smith.langchain.com/public/86ab3e60-2000-4bff-b988-9b89a3269789/r\n\n**Resources:**\n\n**Examples**\n\n[Here](https://github.com/langchain-ai/local-deep-researcher) is an assistant that uses evaluator-optimizer to improve a report. See our video [here](https://www.youtube.com/watch?v=XGuTzHoqlj8).\n\n[Here](https://langchain-ai.github.io/langgraph/tutorials/rag/langgraph_adaptive_rag_local/) is a RAG workflow that grades answers for hallucinations or errors. See our video [here](https://www.youtube.com/watch?v=bq1Plo2RhYI).\n:::\n\n:::js\n```typescript\n// Graph state\nconst State = z.object({\n joke: z.string().optional(),\n topic: z.string(),\n feedback: z.string().optional(),\n funny_or_not: z.string().optional(),\n});'),
Document(metadata={'source': '../../langgraph/docs/docs/tutorials/rag/langgraph_agentic_rag.md', 'relevance_score': 0.09664629}, page_content='Agentic RAG\n\nIn this tutorial we will build a retrieval agent. Retrieval agents are useful when you want an LLM to make a decision about whether to retrieve context from a vectorstore or respond to the user directly.\n\nBy the end of the tutorial we will have done the following:\n\nFetch and preprocess documents that will be used for retrieval.\n\nIndex those documents for semantic search and create a retriever tool for the agent.\n\nBuild an agentic RAG system that can decide when to use the retriever tool.\n\nScreenshot 2024-02-14 at 3.43.58 PM.png\n\nSetup\n\nLet\'s download the required packages and set our API keys:\n\n%%capture --no-stderr\n%pip install -U --quiet langgraph "langchain[openai]" langchain-community langchain-text-splitters\n\nimport getpass\nimport os\n\n\ndef _set_env(key: str):\n if key not in os.environ:\n os.environ[key] = getpass.getpass(f"{key}:")\n\n\n_set_env("OPENAI_API_KEY")'),
Document(metadata={'source': '../../langgraph/docs/docs/tutorials/rag/langgraph_agentic_rag.md', 'relevance_score': 0.08306123}, page_content='# Edges taken after the `action` node is called.\nworkflow.add_conditional_edges(\n "retrieve",\n # Assess agent decision\n grade_documents,\n)\nworkflow.add_edge("generate_answer", END)\nworkflow.add_edge("rewrite_question", "generate_query_or_respond")\n\n# Compile\ngraph = workflow.compile()\n\nVisualize the graph:\n\nfrom IPython.display import Image, display\n\ndisplay(Image(graph.get_graph().draw_mermaid_png()))\n\nGraph\n\n8. Run the agentic RAG\n\nfor chunk in graph.stream(\n {\n "messages": [\n {\n "role": "user",\n "content": "What does Lilian Weng say about types of reward hacking?",\n }\n ]\n }\n):\n for node, update in chunk.items():\n print("Update from node", node)\n update["messages"][-1].pretty_print()\n print("\\n\\n")\n\nOutput:'),
Document(metadata={'source': '../../langgraph/docs/docs/llms-txt-overview.md', 'relevance_score': 0.08234999}, page_content='This MCP server allows integrating llms.txt into tools like Cursor, Windsurf, Claude, and Claude Code.\n\n📘 Setup instructions and usage examples are available in the repository.\n\nUsing llms-full.txt\n\nThe LangGraph llms-full.txt file typically contains several hundred thousand tokens, exceeding the context window limitations of most LLMs. To effectively use this file:\n\nWith IDEs (e.g., Cursor, Windsurf):\n\nAdd the llms-full.txt as custom documentation. The IDE will automatically chunk and index the content, implementing Retrieval-Augmented Generation (RAG).\n\nWithout IDE support:\n\nUse a chat model with a large context window.\n\nImplement a RAG strategy to manage and query the documentation efficiently.'),
Document(metadata={'source': '../../langgraph/docs/_scripts/notebook_hooks.py', 'content_type': 'simplified_code', 'language': 'python', 'relevance_score': 0.072328284}, page_content='"how-tos/create-react-agent-system-prompt.ipynb": "agents/context.md#prompts",\n "how-tos/create-react-agent-structured-output.ipynb": "agents/agents.md#structured-output",\n # misc\n "prebuilt.md": "agents/prebuilt.md",\n "reference/prebuilt.md": "reference/agents.md",\n "concepts/high_level.md": "index.md",\n "concepts/index.md": "index.md",\n "concepts/v0-human-in-the-loop.md": "concepts/human-in-the-loop.md",\n "how-tos/index.md": "index.md",\n "tutorials/introduction.ipynb": "concepts/why-langgraph.md",\n "agents/deployment.md": "tutorials/langgraph-platform/local-server.md",\n # deployment redirects\n "how-tos/deploy-self-hosted.md": "cloud/deployment/self_hosted_data_plane.md",\n "concepts/self_hosted.md": "concepts/langgraph_self_hosted_data_plane.md",\n "tutorials/deployment.md": "concepts/deployment_options.md",\n # assistant redirects\n "cloud/how-tos/assistant_versioning.md": "cloud/how-tos/configuration_cloud.md",'),
Document(metadata={'source': '../../langgraph/docs/_scripts/notebook_hooks.py', 'content_type': 'simplified_code', 'language': 'python', 'relevance_score': 0.072328284}, page_content='"how-tos/create-react-agent-system-prompt.ipynb": "agents/context.md#prompts",\n "how-tos/create-react-agent-structured-output.ipynb": "agents/agents.md#structured-output",\n # misc\n "prebuilt.md": "agents/prebuilt.md",\n "reference/prebuilt.md": "reference/agents.md",\n "concepts/high_level.md": "index.md",\n "concepts/index.md": "index.md",\n "concepts/v0-human-in-the-loop.md": "concepts/human-in-the-loop.md",\n "how-tos/index.md": "index.md",\n "tutorials/introduction.ipynb": "concepts/why-langgraph.md",\n "agents/deployment.md": "tutorials/langgraph-platform/local-server.md",\n # deployment redirects\n "how-tos/deploy-self-hosted.md": "cloud/deployment/self_hosted_data_plane.md",\n "concepts/self_hosted.md": "concepts/langgraph_self_hosted_data_plane.md",\n "tutorials/deployment.md": "concepts/deployment_options.md",\n # assistant redirects\n "cloud/how-tos/assistant_versioning.md": "cloud/how-tos/configuration_cloud.md",'),
Document(metadata={'source': '../../langgraph/docs/docs/how-tos/tool-calling.md', 'relevance_score': 0.04794482}, page_content='```'),
Document(metadata={'source': '../../langgraph/docs/docs/how-tos/tool-calling.md', 'relevance_score': 0.04794482}, page_content='```'),
Document(metadata={'source': '../../langgraph/docs/docs/how-tos/tool-calling.md', 'relevance_score': 0.04794482}, page_content='```'),
Document(metadata={'source': '../../langgraph/docs/docs/how-tos/tool-calling.md', 'relevance_score': 0.04794482}, page_content='```'),
Document(metadata={'source': '../../langgraph/docs/docs/how-tos/multi_agent.md', 'relevance_score': 0.04794482}, page_content='```'),
Document(metadata={'source': '../../langgraph/docs/docs/how-tos/multi_agent.md', 'relevance_score': 0.04794482}, page_content='```'),
Document(metadata={'source': '../../langgraph/docs/docs/how-tos/memory/add-memory.md', 'relevance_score': 0.04794482}, page_content='```'),
Document(metadata={'source': '../../langgraph/docs/docs/how-tos/memory/add-memory.md', 'relevance_score': 0.04794482}, page_content='```'),
Document(metadata={'source': '../../langgraph/libs/langgraph/langgraph/pregel/main.py', 'content_type': 'functions_classes', 'language': 'python', 'relevance_score': 0.043208163}, page_content='self.debug = debug if debug is not None else get_debug()\n self.checkpointer = checkpointer\n self.store = store\n self.cache = cache\n self.retry_policy = (\n (retry_policy,) if isinstance(retry_policy, RetryPolicy) else retry_policy\n )\n self.cache_policy = cache_policy\n self.context_schema = context_schema\n self.config = config\n self.trigger_to_nodes = trigger_to_nodes or {}\n self.name = name\n if auto_validate:\n self.validate()'),
Document(metadata={'source': '../../langgraph/libs/langgraph/langgraph/pregel/main.py', 'content_type': 'functions_classes', 'language': 'python', 'relevance_score': 0.043208163}, page_content='self.debug = debug if debug is not None else get_debug()\n self.checkpointer = checkpointer\n self.store = store\n self.cache = cache\n self.retry_policy = (\n (retry_policy,) if isinstance(retry_policy, RetryPolicy) else retry_policy\n )\n self.cache_policy = cache_policy\n self.context_schema = context_schema\n self.config = config\n self.trigger_to_nodes = trigger_to_nodes or {}\n self.name = name\n if auto_validate:\n self.validate()'),
Document(metadata={'source': '../../langgraph/libs/langgraph/langgraph/_internal/_runnable.py', 'content_type': 'functions_classes', 'language': 'python', 'relevance_score': 0.04201308}, page_content='self.afunc = afunc\n self.tags = tags\n self.kwargs = kwargs\n self.trace = trace\n self.recurse = recurse\n self.explode_args = explode_args\n # check signature\n if func is None and afunc is None:\n raise ValueError("At least one of func or afunc must be provided.")'),
Document(metadata={'source': '../../langgraph/libs/langgraph/langgraph/_internal/_runnable.py', 'content_type': 'functions_classes', 'language': 'python', 'relevance_score': 0.04201308}, page_content='self.afunc = afunc\n self.tags = tags\n self.kwargs = kwargs\n self.trace = trace\n self.recurse = recurse\n self.explode_args = explode_args\n # check signature\n if func is None and afunc is None:\n raise ValueError("At least one of func or afunc must be provided.")')]}
웹 검색 노드
from langchain_tavily import TavilySearch
search_tool = TavilySearch(max_results=3)
def web_search(state: State):
question = state.get("question")
web_results = search_tool.invoke({"query": question})
documents: list[Document] = []
for r in web_results["results"]:
documents.append(
Document(
page_content=f"# {r['title']}\n\n{r['content']}",
metadata={
"source": r["url"],
"score": r["score"],
},
)
)
return {"documents": documents}web_search({"question": "대한민국의 수도는?"}){'documents': [Document(metadata={'source': 'https://quizlet.com/kr/591770273/lesson-2-대한민국의-수도는-서울입니다-flash-cards/', 'score': 0.8997663}, page_content='# Lesson 2: 대한민국의 수도는 서울입니다 낱말 카드\n\n대한민국의 수도는서울입니다. 서울은 한반도의 중심에 있고 인구는 약천만 명입니다. 서울에는 구경할 곳이 많습니다. 먼저 외국인들이 많이 찾아가는 남산이'),
Document(metadata={'source': 'https://namu.wiki/w/수도(도시)', 'score': 0.86981434}, page_content='# 수도(도시)\n\n6 days ago—대한민국은 조선과 대한제국의 수도 한성이 이어진서울특별시를 계속 수도로 설정하고 있다. 일제강점기에도 행정중심지인 조선총독부가 서울(경성)에'),
Document(metadata={'source': 'https://m.kin.naver.com/qna/dirs/111001/docs/478501757?d1id=11', 'score': 0.8622012}, page_content='# 대한민국 수도 서울\n\n대한민국 수도 서울 대한민국의 수도인 서울이 어떻게 수도가 됐는지가 궁금합니다. 서울이 조선시대 때 한양이라 불리우며 조선의 수도였던 것은 아는데 그 조선의 수도가 자연스레 한국의 수도가 되어 대한민국의 수도는 서울이다라는 공식 선언같은 건 없이 지금의 대한민국의 수도가 되었다는 말도 있던데 그게 사실인가요?? #문제풀이 우리나라 수도 명칭인 서울은 원래 한 나라의 수도를 뜻하는 순수한 우리말의 일반명사였어요. 그래서 신라의 수도였던 서라벌, 고려의 수도였던 개성도 모두 서울이라고 불렀지요. 조선의 수도였던 한양 역시 사람들은 대부분 지금처럼 서울이라고 불렀어요. 그러다가 서울이 우리나라의 수도이자 지명을 겸하는 제도상 공식명칭으로 쓰이기 시작한 것은 그렇다면 지금의 서울지역은 시대에 따라 어떻게 불렸을까요? 조선시대에는 한양(漢陽) 혹은 한성부(漢城府)로 불렸다는 이야기가 나와요. 우리나라의 수도인 서울 지역의 명칭도 시대에 학점은행제 Expert "학점은행제 잘 몰랐는데 이해하기 쉽게 설명해 주셔서 감사합니다" 학점은행제 QnA "오늘도 문제풀이 감사합니다! 학점은행제 IN Expert')]}
백터 스토어 기반 답변 생성 노드
def answer_rag_node(state: State):
question = state.get("question")
documents = state.get("documents")
response = llm.invoke(
system_prompt.format(
context=documents,
question=question,
)
)
return {
"messages": [response],
"generation": response.content,
"question": "",
"documents": [],
}answer_rag_node(
{
"question": "대한민국 수도는?",
"documents": [Document(page_content="서울")],
},
){'messages': [AIMessage(content='대한민국의 수도는 "서울"입니다. \n\n예를 들어, 다음과 같이 간단한 파이썬 코드로 수도를 출력할 수 있습니다:\n\n```python\ncapital = "서울"\nprint("대한민국의 수도는", capital, "입니다.")\n```\n\n출력 결과:\n```\n대한민국의 수도는 서울 입니다.\n```\n\n또한, 만약 데이터베이스나 문서에서 수도 정보를 조회하는 경우, 다음과 같이 사용할 수 있습니다:\n\n```python\n# 수도 정보가 저장된 변수\ncountry_capital = {\n "대한민국": "서울",\n "일본": "도쿄",\n "중국": "베이징"\n}\n\nprint("대한민국의 수도는", country_capital["대한민국"], "입니다.")\n```\n\n출력 결과:\n```\n대한민국의 수도는 서울 입니다.\n```\n\n이처럼 "서울"이라는 정보는 CONTEXT 내에 명확히 포함되어 있습니다.\n\n**출처** \n- [1] 서울', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 208, 'prompt_tokens': 689, 'total_tokens': 897, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4.1-mini-2025-04-14', 'system_fingerprint': 'fp_95d112f245', 'id': 'chatcmpl-CN00vr6o2L7JC7VNNEUv2ZToG9kWf', 'service_tier': 'default', 'finish_reason': 'stop', 'logprobs': None}, id='run--770f58c2-4687-4e5e-a7e6-2291d781bbb9-0', usage_metadata={'input_tokens': 689, 'output_tokens': 208, 'total_tokens': 897, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})],
'generation': '대한민국의 수도는 "서울"입니다. \n\n예를 들어, 다음과 같이 간단한 파이썬 코드로 수도를 출력할 수 있습니다:\n\n```python\ncapital = "서울"\nprint("대한민국의 수도는", capital, "입니다.")\n```\n\n출력 결과:\n```\n대한민국의 수도는 서울 입니다.\n```\n\n또한, 만약 데이터베이스나 문서에서 수도 정보를 조회하는 경우, 다음과 같이 사용할 수 있습니다:\n\n```python\n# 수도 정보가 저장된 변수\ncountry_capital = {\n "대한민국": "서울",\n "일본": "도쿄",\n "중국": "베이징"\n}\n\nprint("대한민국의 수도는", country_capital["대한민국"], "입니다.")\n```\n\n출력 결과:\n```\n대한민국의 수도는 서울 입니다.\n```\n\n이처럼 "서울"이라는 정보는 CONTEXT 내에 명확히 포함되어 있습니다.\n\n**출처** \n- [1] 서울'}
일반적인 답변 생성 노드
def answer_general_node(state: State):
response = llm.invoke(state["messages"])
return {"messages": [response]}# 웹검색이 필요한지 여부 검사
def decide_to_web_search_node(state):
filtered_docs = state["documents"]
if len(filtered_docs) < 2:
return "web_search"
else:
return "rag_answer"class GroundednessChecker(BaseModel):
binary_score: Annotated[
Literal[1, 0],
Field(
...,
description="LLM 답변이 검색된 사실 집합에 근거하거나 이를 뒷받침하는지 평가하는 채점 도구입니다."
"답변이 사실 집합에 근거하거나 이를 뒷받침하면 1을 반환하고, 그렇지 않으면 0을 반환하세요.",
),
]
class RelevantAnswerChecker(BaseModel):
binary_score: Annotated[
Literal[1, 0],
Field(
...,
description="답변이 질문을 해결했는지 평가하는 채점 도구입니다."
"질문에 대한 답변이 해결되었다면 1을 반환하고, 그렇지 않다면 0을 반환하세요.",
),
]
# 답변의 환각 여부/관련성 여부 평가 노드
def answer_groundedness_check(state):
# 질문과 문서 검색 결과 가져오기
question = state["question"]
documents = state["documents"]
generation = state["generation"]
groundedness_checker = llm.with_structured_output(GroundednessChecker)
relevant_answer_checker = llm.with_structured_output(RelevantAnswerChecker)
# Groundedness 평가
score = groundedness_checker.invoke(
f"Set of facts: \n\n {documents} \n\n LLM generation: {generation}"
)
grade = score.binary_score
# Groundedness 평가 결과에 따른 처리
if grade == 1:
# 답변의 관련성(Relevance) 평가
score = relevant_answer_checker.invoke(
f"User question: \n\n {question} \n\n LLM generation: {generation}"
)
grade = score.binary_score
# 관련성 평가 결과에 따른 처리
if grade == 1:
return "relevant"
else:
return "not relevant"
else:
return "not grounded"그래프 구성
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import InMemorySaver
builder = StateGraph(State)
# 노드
builder.add_node("query_expansion", rewrite_query_node)
builder.add_node("query_rewrite", rewrite_query_node)
builder.add_node("web_search", web_search)
builder.add_node("retrieve", retrieve_node)
builder.add_node("grade_documents", filtering_documents_node)
builder.add_node("general_answer", answer_general_node)
builder.add_node("rag_answer", answer_rag_node)
# 엣지
builder.set_conditional_entry_point(
routing_node,
{
"query_expansion": "query_expansion",
"general_answer": "general_answer",
},
)
builder.add_edge("query_expansion", "retrieve")
builder.add_edge("retrieve", "grade_documents")
builder.add_conditional_edges(
"grade_documents",
decide_to_web_search_node,
{
"web_search": "web_search",
"rag_answer": "rag_answer",
},
)
builder.add_edge("query_rewrite", "rag_answer")
builder.add_conditional_edges(
"rag_answer",
answer_groundedness_check,
{
"relevant": END,
"not relevant": "web_search",
"not grounded": "query_rewrite",
},
)
graph = builder.compile(checkpointer=InMemorySaver())from IPython.display import display, Image
display(Image(graph.get_graph().draw_mermaid_png()))
그래프 실행
from langchain_core.runnables import RunnableConfig
config = RunnableConfig(
recursion_limit=20,
configurable={"thread_id": "2"},
)
query = "Self-RAG 에서 사용되는 관련성 평가 노드 예제를 찾아주세요."
for event in graph.stream(
{"messages": [HumanMessage(content=query)]},
stream_mode="updates",
config=config,
):
for key, value in event.items():
print(f"==== {key} ===")
if "messages" in value:
value["messages"][-1].pretty_print()==== query_expansion ===
==== retrieve ===
==== grade_documents ===
==== rag_answer ===
==================================[1m Ai Message [0m==================================
Self-RAG에서 사용되는 관련성 평가 노드는 검색된 문서들이 실제로 질문에 적합한지를 판단하는 역할을 합니다. 즉, 단순히 키워드 매칭이 아니라 문서의 의미적 관련도(semantic relevance)를 평가하여, 더 정확하고 유의미한 정보를 선택하는 과정입니다. 이 과정은 Retrieval Augmented Generation에서 매우 중요하며, 관련성 평가는 문서의 내용이 질문 의도와 얼마나 일치하는지를 수치화하는 작업입니다.
기존 Self-RAG 관련성 평가 노드 예제는 다음과 같이 구성되어 있습니다:
```python
input = {
"messages": convert_to_messages([
{
"role": "user",
"content": "What does Lilian Weng say about types of reward hacking?",
},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "1",
"name": "retrieve_blog_posts",
"args": {"query": "types of reward hacking"},
}
],
},
{
"role": "tool",
"content": "reward hacking can be categorized into two types: environment or goal misspecification, and reward tampering",
"tool_call_id": "1",
},
])
}
grade_documents(input)
```
이 코드는 사용자 질문에 대해 검색된 문서가 실제로 관련성이 있는지 평가하는 `grade_documents` 함수를 호출하는 예입니다. 여기서 `convert_to_messages`는 대화 형식으로 메시지를 변환하는 유틸리티입니다. 이 노드는 검색된 결과가 질문에 적합한지 판단하여 이후 단계에서 더 나은 답변 생성을 돕습니다.
---
이 예제를 GitHub의 CODE SEARCH 스타일로 최적화된 버전으로 변환하면 다음과 같은 점들을 개선할 수 있습니다:
1. **명확한 입력/출력 타입 지정**
2. **관련성 점수 계산 및 반환**
3. **비동기 처리 지원 (필요시)**
4. **재사용 가능한 함수 구조화**
5. **의미적 관련도 평가를 위한 임베딩 또는 벡터 유사도 활용 (Self-RAG의 핵심)**
아래는 최적화된 예제 코드입니다:
```python
from typing import List, Dict, Any
def convert_to_messages(conversation: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
# 기존 메시지 포맷 변환 함수 (예시)
return conversation
def grade_documents(input_data: Dict[str, Any]) -> List[Dict[str, Any]]:
"""
Evaluate the relevance of retrieved documents to the user's query.
Returns a list of documents with relevance scores.
"""
messages = input_data.get("messages", [])
user_query = next((m["content"] for m in messages if m["role"] == "user"), "")
retrieved_docs = [m for m in messages if m["role"] == "tool"]
# 예: 임베딩 기반 유사도 계산 (의사 코드)
def semantic_similarity(query: str, doc: str) -> float:
# 실제 구현은 임베딩 벡터 간 코사인 유사도 등 사용
return 0.85 # 예시 점수
graded_docs = []
for doc in retrieved_docs:
content = doc.get("content", "")
score = semantic_similarity(user_query, content)
graded_docs.append({
"content": content,
"score": score,
"tool_call_id": doc.get("tool_call_id")
})
# 관련성 높은 순 정렬
graded_docs.sort(key=lambda x: x["score"], reverse=True)
return graded_docs
# 사용 예시
input_example = {
"messages": convert_to_messages([
{"role": "user", "content": "What does Lilian Weng say about types of reward hacking?"},
{"role": "assistant", "content": "", "tool_calls": [{"id": "1", "name": "retrieve_blog_posts", "args": {"query": "types of reward hacking"}}]},
{"role": "tool", "content": "reward hacking can be categorized into two types: environment or goal misspecification, and reward tampering", "tool_call_id": "1"},
])
}
graded_results = grade_documents(input_example)
for doc in graded_results:
print(f"Content: {doc['content']}\nRelevance Score: {doc['score']}\n")
```
---
### 관련도/의미 설명 (Relevance / Semantic Meaning)
- **Relevance (관련도)**: 사용자의 질문과 문서 내용이 얼마나 밀접하게 연관되어 있는지를 나타내는 척도입니다. 단순 키워드 일치가 아니라 문서가 질문의 의도와 주제를 얼마나 잘 반영하는지를 평가합니다.
- **Semantic Meaning (의미적 의미)**: 문서와 질문 간의 의미적 유사성을 측정하는 것으로, 자연어 처리에서 임베딩 벡터를 활용해 문장이나 문서의 의미를 수치화합니다. 이를 통해 단어 단위가 아닌 문장 전체의 의미를 비교할 수 있습니다.
Self-RAG에서는 이 두 가지를 결합하여, 검색된 문서가 단순히 키워드가 포함된 문서가 아니라 실제로 질문에 답변할 수 있는 의미적 관련성을 가진 문서인지 평가합니다. 이 과정이 잘 되어야 최종 생성되는 답변의 품질이 높아집니다.
---
요약하면, Self-RAG의 관련성 평가 노드는 검색된 문서들의 의미적 관련도를 평가하여, 질문에 가장 적합한 정보를 선별하는 역할을 하며, 이를 위해 임베딩 기반 유사도 계산과 같은 의미적 평가 기법을 활용하는 것이 최적화된 접근법입니다.
[1] langgraph/docs/docs/tutorials/rag/langgraph_agentic_rag.md
[2] langgraph/libs/langgraph/tests/test_checkpoint_migration.py
[3] langgraph/docs/docs/tutorials/rag/langgraph_agentic_rag.md
---
**출처**
- [1] ../../langgraph/docs/docs/tutorials/rag/langgraph_agentic_rag.md
- [2] ../../langgraph/libs/langgraph/tests/test_checkpoint_migration.py