HippoRAG는 인간 뇌의 해마(hippocampus) 기억 인덱싱 이론에서 영감을 받은 RAG 아키텍처입니다. LLM으로 문서에서 지식 트리플(주어-관계-목적어)을 추출하여 지식 그래프를 구축하고, Personalized PageRank(PPR)를 활용하여 검색하는 방식으로 기존 RAG보다 다단계 추론과 지식 통합에 뛰어납니다.
해마 기억 인덱싱 이론에 따르면, 해마는 기억의 내용 자체를 저장하는 것이 아니라 대뇌 신피질에 분산된 기억 조각들의 인덱스(색인)를 관리합니다.새로운 정보가 들어오면 해마는 기존 기억과의 연결(패턴 분리/패턴 완성)을 통해 관련 기억을 활성화합니다. HippoRAG는 이 과정을 지식 그래프의 엔티티 연결과 PPR 전파로 모사합니다.
from langchain.chat_models import init_chat_modelfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import StrOutputParserllm = init_chat_model("gpt-4o-mini", temperature=0)def extract_query_entities(state: HippoState) -> HippoState: """질문에서 핵심 Named Entity를 추출합니다.""" question = state["question"] prompt = ChatPromptTemplate.from_messages([ ("system", ( "질문에서 핵심 Named Entity(인물, 기술, 개념, 조직 등)를 추출하세요.\n" "쉼표로 구분하여 출력하세요.\n" "예: 트랜스포머, 셀프 어텐션, BERT" )), ("human", "{question}"), ]) chain = prompt | llm | StrOutputParser() result = chain.invoke({"question": question}) entities = [e.strip() for e in result.split(",") if e.strip()] return {"query_entities": entities}
Copy
def retrieve_from_kg(state: HippoState) -> HippoState: """Knowledge Graph에서 PPR 방식으로 관련 트리플과 노드를 탐색합니다.""" query_entities = state["query_entities"] kg = KNOWLEDGE_GRAPH # 1. 시드 노드 매칭 seed_nodes = [e for e in query_entities if e in kg["entities"]] # 2. PPR 시뮬레이션: 시드 노드와 연결된 트리플 탐색 related_triples = [] visited_entities = set(seed_nodes) for triple in kg["triples"]: subj, rel, obj = triple if subj in seed_nodes or obj in seed_nodes: related_triples.append({"subject": subj, "relation": rel, "object": obj}) visited_entities.add(subj) visited_entities.add(obj) # 3. 2-hop 확장: 1차 탐색에서 발견된 엔티티의 이웃도 포함 for triple in kg["triples"]: subj, rel, obj = triple if (subj in visited_entities or obj in visited_entities) and triple not in [ (t["subject"], t["relation"], t["object"]) for t in related_triples ]: related_triples.append({"subject": subj, "relation": rel, "object": obj}) visited_entities.add(subj) visited_entities.add(obj) return {"kg_results": related_triples, "query_entities": list(visited_entities)}
Copy
def retrieve_passages(state: HippoState) -> HippoState: """KG 탐색에서 발견된 엔티티와 연결된 원본 패시지를 수집합니다.""" entities = state["query_entities"] kg = KNOWLEDGE_GRAPH passage_ids = set() for entity in entities: if entity in kg["entities"]: passage_ids.update(kg["entities"][entity]["passages"]) documents = [ Document(page_content=PASSAGES[pid], metadata={"passage_id": pid}) for pid in passage_ids if pid in PASSAGES ] return {"documents": documents}
Copy
def generate(state: HippoState) -> HippoState: """수집된 패시지와 KG 트리플을 기반으로 답변을 생성합니다.""" question = state["question"] documents = state["documents"] kg_results = state["kg_results"] # KG 트리플 정보를 텍스트로 변환 triples_text = "\n".join( f"- {t['subject']} → {t['relation']} → {t['object']}" for t in kg_results ) context = "\n\n".join(doc.page_content for doc in documents) prompt = ChatPromptTemplate.from_messages([ ("system", ( "다음 지식 그래프 관계와 문서를 참고하여 질문에 답변하세요.\n\n" "지식 그래프 관계:\n{triples}\n\n" "관련 문서:\n{context}" )), ("human", "{question}"), ]) chain = prompt | llm | StrOutputParser() answer = chain.invoke({ "triples": triples_text, "context": context, "question": question, }) return {"answer": answer}