[SYNAPSE 개발일지 #3] GraphRAG 파이프라인 구축 - 그래프 생성과 탐색, SSE 기반 실시간 통신까지
[SYNAPSE 개발일지 #2] Redis Streams로 서버 간 의존도 낮추기 - 이벤트 기반 아키텍처 설계
[SYNAPSE 개발일지 #2] Redis Streams로 서버 간 의존도 낮추기 - 이벤트 기반 아키텍처 설계
[SYNAPSE 개발일지 #1] GraphRAG를 위한 AI 서버 아키텍처 설계 - Orchestrator 도입부터 DB 접근 권한과 도메인 소유권까지 [SYNAPSE 개발일지 #1] GraphRAG를 위한 AI 서버 아키텍처 설계 - Orchestrator 도입부터 DB
blog.juyear.dev
이전 글 읽으러 가기!
👋 소개 및 회고
안녕하세요. 대학생 개발자 주이어입니다.
저번 글에서 Redis Streams를 이용하여 메인 서버와 AI 서버의 아키텍처를 어느정도 설계하였습니다.
오늘은 이어서 SYNAPSE의 핵심 기능인 GraphRAG 파이프라인 설계에 대해서 정리해 보려합니다.
- Graph 생성 구현
- Graph 탐색 구현
위 두 가지를 중심으로 작성할 예정이며, 중간마다 고민했던 설계 문제도 넣을 예정입니다.
⚙️ Graph 생성 구현
파이프라인 설계 및 구현
Graph 생성 파이프라인의 경우 이전 글에서도 몇 번 보여드렸습니다.
해당 파이프라인을 구축하기 위해 메인서버와 AI서버에 대해서 많은 아키텍처 설계가 있었고 이를 이전 글들에서 다뤄왔습니다.
이미 아키텍처 설계가 끝났기 때문에 구현은 많이 어렵지 않았습니다.
async def note_analyze(self, data: NoteAnalyzeRequest):
embedding = await self.note_service.create_note_embedding(data.content)
top_k_notes = await self.note_service.get_top_k_notes_by_embedding(data.user_id, embedding, 5)
await self.redis_service.add_save_embedding_task(data,embedding)
for note in top_k_notes:
result = await self.link_service.generate_link_reason(data, note)
link = json.loads(result)
if link.get("isLink"):
await self.redis_service.add_save_link_task(data.user_id, link, data.note_id, note["id"])
else:
print("연결 안됨")Orchestrator에서 note service, link service, redis service를 조합하여 구현하였습니다.
함수를 기준으로 진행 과정을 살펴보면,
create_note_embedding -> get_top_k_notes_by_embedding -> add_save_embedding_task
-> generate_link_reason -> add_save_link_task위와 같습니다.
이런식으로 Orchestrator가 여러 service를 조합하여 Graph 생성 파이프라인을 구축할 수 있도록 구현하였습니다.
DB 설계 문제
Graph 생성의 경우 크게 고민할 부분이 없었습니다.
오히려 제가 고민했던 부분은 DB 설계 부분이었습니다.
Graph 생성을 구현했으니 이제 생성된 link 정보를 활용해 실제로 화면에 띄워야 했고, 여기서 문제가 생겼습니다.
제가 처음에 생각한 구조는 위 그림과 같았습니다.
크게 고민할 부분도 아니고 어떻게 보면 당연한 설계였습니다.
User가 Note를 가지고 있고, Note가 Link를 가지고 있는 깔끔한 구조였습니다.
const graphData = useMemo(() => {
return {
nodes: filteredNotes.map((note) => ({
id: String(note.id),
name: note.title,
category: note.category,
color: categories.find((c) => c.title === note.category.title)?.color,
content: note.content,
})),
links: filteredLinks.map((link) => ({
...link,
source: String(link.sourceNote.id),
target: String(link.targetNote.id),
})),
};
}, [filteredNotes, filteredLinks, categories]);문제는 Graph를 화면에 띄우기 위해서는 Note 데이터와 Link 데이터를 따로 들고 있어야 했고, Link 데이터는 연결된 노트의 정보를 들고 있어야 했습니다.
그래야 ForceGraph2D(Graph 생성) 라이브러리가 인식하고 Graph를 생성할 수 있었습니다.
하지만 현재 DB 구조에서는 이를 구현하기에 두 가지 문제점이 있었습니다.
[1. Note와 Link 분리 문제]
현재 DB 구조에서는 한 번에 Link 데이터만 가져올 수 없었습니다.
Link 데이터를 가져오기 위해서는 Note를 거쳐야 했고, Note에 연결된 Link 데이터만 가져올 수 있었습니다.
이렇게 가져올 경우 프론트에서 Note와 Link를 분리하는 별도의 작업을 진행해야 했습니다.
Note 데이터가 한 두개라면 문제가 없겠지만 많아지면 많아질수록 분리하는 작업의 비용은 커질 수 밖에 없습니다.
[2. 데이터 중복 조회 문제]
또 다른 문제점 중 하나는 중복 조회가 일어난다는 점입니다.
ForceGraph2D에서 그래프를 생성하기 위해서는 Link가 Source와 Target에 대한 정보를 가지고 있어야 하고, 이를 위해서는 위 그림처럼 중복 조회가 일어날 수 있습니다.
수많은 Note와 수많은 Link에서는 훨씬 많은 중복 조회가 발생할 수 있습니다.
물론 잘 설계하여 중복 조회를 해결할 수 있겠지만, 결국 라이브러리가 요구하는 형태의 데이터를 만들기 위해서는 백엔드 또는 프론트엔드에서 별도의 데이터 가공 작업이 필요했습니다.
[문제 해결]
이러한 두 가지 문제점을 해결하기 위해 저는 위와 같은 구조를 떠올렸습니다.
이런식으로 DB를 설계한다면, 가져올 때 부터 Note 데이터와 Link 데이터를 분리해서 가져올 수 있었습니다.
하지만 처음부터 이렇게 설계하지 않았던 이유는 이러한 구조도 사실 완벽한 구조는 아닙니다.
사실 DB 구조를 제대로 확인해보면 위와 더 유사합니다.
이러한 구조는 Note와 Link 모두 user_id를 저장하게 되고, 이러한 설계는 정규화 관점에서 데이터 중복을 발생시키는 설계입니다.
user_id가 여러개라고 해서 무조건 데이터 중복은 아니지만, Link는 Note를 통해 사용자 정보를 유추할 수 있음에도 user_id를 별도로 저장하게 되므로 동일한 정보를 중복 저장하는 비정규화 구조가 됩니다.
또한 Note와 Link의 user_id가 항상 동일해야 하므로 데이터 일관성을 관리해야 하는 부담도 있습니다.
하지만 SYNAPSE의 경우 Note와 Link의 소유자가 변경되지 않는다는 특성이 있고, 이는 데이터 일관성 관리와 수정 비용과 같은 부담을 없앨 수 있었습니다.
어떠한 설계를 하냐에는 100% 정답이 없다고 하지만, SYNAPSE에서는 그래프를 구성하기 위해 사용자 기준으로 모든 Link를 조회하는 작업이 자주 발생했기 때문에 성능과 단순 조회 구조를 우선시하여 현재의 구조를 선택했습니다.
🔍 Graph 탐색 구현
파이프라인 설계 및 구현
Graph 탐색 파이프라인은 위와 같습니다.
단계별로 간단히 정리하자면 아래와 같습니다.
- User Query : 사용자가 질문을 작성합니다.
- Query Embedding : 사용자 질문을 Embedding 합니다.
- Vector Search : 사용자 질문을 기반으로 관련 있는 Note를 선별합니다.
- Find Connected Link : 선별된 Note에 연결된 Link를 찾습니다.
- Retrieve Connected Note : Link에 연결된 반대편 Note를 가져옵니다.
- Remove Duplicates : 중복되는 Note를 제거합니다.
- Graph Result : Note와 Link 데이터를 활용해 사용자 질문에 해당하는 Graph를 생성합니다.
추가로 AI 답변 생성 기능까지 들어가야 하지만, 단순 API 호출이기 때문에 이번 글에서는 제외하였습니다.
[RAG와의 차이점]
일반 RAG와의 차이에 대해서 간단히 정리해보면, top k 문서를 찾는 과정까지는 똑같습니다.
하지만 기존 RAG는 문서들이 서로 독립적으로 존재하기 때문에 검색된 문서만을 기반으로 답변을 생성할 수 있고, 이는 전체적인 문맥을 파악하기 어렵다는 문제점이 있습니다.
반면 GraphRAG는 top k 문서를 찾은 이후 각 문서에 연결되어 있는 다른 문서까지 함께 탐색하여 더 넓은 문맥을 구성할 수 있습니다.
이러한 GraphRAG의 특징은 단편적인 데이터 뿐만 아니라 관계까지 함께 활용할 수 있다는 점입니다.
예를 들어 "Redis"를 검색했을 때 기존 RAG는 Redis에 대한 노트만 참고하여 답변을 생성하지만, GraphRAG는 Redis와 연결된 "Pub/Sub", "Stream", "Redis를 사용한 프로젝트"까지 함께 가져와 하나의 맥락으로 이해할 수 있습니다.
이를 통해 검색 결과가 단순히 비슷한 문서의 집합이 아니라, 연결된 지식 그래프 형태의 컨텍스트로 확장되며, 이후 LLM은 이러한 관계를 참고하여 더욱 정확하고 풍부한 답변을 생성할 수 있습니다.
이제 다시 돌아와 파이프라인 구현에 대해서 정리해보겠습니다.
async def graph_search(self, data:GraphSearchRequest):
embedding = await create_embedding(data.content)
top_k_notes = await self.note_service.get_top_k_notes_by_embedding(data.user_id, embedding, 3)
notes = []
links = []
note_ids = set()
link_ids = set()
for note in top_k_notes:
if note["id"] not in note_ids:
notes.append(note)
note_ids.add(note["id"])
related_links = await self.link_service.find_by_note_id(note)
for link in related_links:
sourceNoteId = json.loads(link["sourceNote"])["id"]
targetNoteId = json.loads(link["targetNote"])["id"]
if link["id"] not in link_ids:
link_ids.add(link["id"])
links.append(link)
if sourceNoteId not in note_ids:
note_ids.add(sourceNoteId)
related_note = await self.note_service.find_by_id(sourceNoteId)
notes.append(related_note[0])
if targetNoteId not in note_ids:
note_ids.add(targetNoteId)
related_note = await self.note_service.find_by_id(targetNoteId)
notes.append(related_note[0])
print(len(notes),len(links))
await self.redis_service.addGraphSearchResultTask(data.user_id,notes,links)이러한 파이프라인을 구현하기 위해 graph search라는 Orchestrator을 만들어 주었습니다.
해당 Orchestrator는 위에서 말한 파이프라인을 전반을 수행합니다.
SSE 설계
해당 파이프라인을 거쳐 생성된 Graph를 화면에 보여주기 위해서는 한 가지 문제점이 있습니다.
일반적인 Graph의 경우 클라이언트에서 요청을 보내고 Graph Data를 받아 렌더링만 해주면 되었습니다.
하지만 사용자 질문의 경우 단순한 API 호출 방식이 아닙니다.
Graph 생성을 마냥 기다리고 있을 수 없기 때문에 Redis를 활용한 비동기로 처리가 진행되며, Graph가 생성되었을 때 클라이언트로 직접 보내줘야 합니다.
저는 이를 해결하기 위해 이전에 몇 번 사용해본 WebSocket을 가장 먼저 떠올렸습니다.
하지만 이에 대해서 좀 더 찾아보니 서버에서 클라이언트로 보내는 통신만 필요하다면 SSE(Server-Sent-Events)를 설계하는 것이 더 좋다고 하였습니다.
WebSocket은 HTTP 연결 이후 별도의 WebSocket 프로토콜로 연결을 유지하여 클라이언트와 서버가 하나의 연결에서 자유롭게 양방향 통신을 수행할 수 있게 도와준다면,
SSE는 HTTP 프로토콜을 그대로 유지한 채 응답(Response)을 끝내지 않고 연결을 지속하여 서버에서 지속적으로 event를 보낼 수 있도록 설계하는 것 입니다.
이러한 SSE는 브라우저에서 기본적으로 제공하는 EventSource API를 사용하여 구현할 수 있기 때문에 구현이 비교적 단순하며, 서버에서 event를 전송한다는 관점만 생각하면 되기 때문에 로직도 단순합니다.
최종적으로 현재 필요한 것은 서버에서 클라이언트로 Graph 데이터(이후에는 AI 답변까지)를 넘겨주는 것이기 때문에 SSE를 선택하였습니다.
useEffect(() => {
const eventSource = new EventSource(
`${process.env.NEXT_PUBLIC_BACKEND_URL}/chats/messages/stream`,
{ withCredentials: true },
);
eventSource.onmessage = (e) => {
const result = JSON.parse(e.data);
if (result.type === "graph") {
console.log(result.data);
setGraph(result.data);
} else if (result.type === "message") {
console.log(result);
const newAIMessage: MessageType = {
id: messages.length + 2,
role: "ai",
content: message,
createdAt: "2026-07-10",
};
setMessages((prev) => [...prev, newAIMessage]);
}
};
eventSourceRef.current = eventSource;
return () => {
eventSource.close();
};
}, []);
클라이언트에서는 위와 같이 EventSource를 이용하여 서버에 요청을 보내도록 설계하면 됩니다.
이후 onmessage를 통해 서버에서 오는 event를 감지할 수 있습니다.
저는 여러 종류의 event를 설계하고 있었기 때문에 type으로 구분할 수 있도록 해주었습니다.
@Sse('messages/stream')
stream(@User() user: UserEntity): Observable<MessageEvent> {
return this.chatsStreamService.connect(user.id);
}서버에서는 일반적인 @Get, @Post가 아닌 @Sse를 사용하여 SSE 요청이라는 것을 알려줍니다.
connect(userId: number) {
if (!this.streams.has(userId)) {
this.streams.set(userId, new Subject<MessageEvent>());
}
return this.streams.get(userId)!.asObservable();
}
sendGraph(userId: number, data: ChatsStreamData) {
this.streams.get(userId)?.next({
type: 'graph',
data: {
notes: data.notes,
links: data.links,
},
});
}
disconnect(userId: number) {
this.streams.get(userId)?.complete();
this.streams.delete(userId);
}그 후 저는 stream service를 별도로 분리하여 위와 같이 구현해주었습니다.
connect의 경우 연결 정보를 저장하기 위한 함수이고, user별로 저장하기 위해 user id를 받아와 같이 저장해주었습니다.
sendGraph의 경우 전달받은 data를 event를 통해 내보내는 함수입니다.
이때 내보내는 경로는 저장해둔 user id를 기반으로 찾아옵니다.
disconnect는 연결을 해제하기 위한 함수입니다.
최종적으로 클라이언트에서 Graph 데이터를 전달받을 수 있었고, 해당 데이터를 활용하여 위와 같이 사용자 질문에 맞는 Graph를 생성하여 보여줄 수 있었습니다.
*추가 - Orchestrator 최적화
async def graph_search(self, data:GraphSearchRequest):
embedding = await create_embedding(data.content)
top_k_notes = await self.note_service.get_top_k_notes_by_embedding(data.user_id, embedding, 3)
notes = []
links = []
note_ids = set()
link_ids = set()
for note in top_k_notes:
if note["id"] not in note_ids:
notes.append(note)
note_ids.add(note["id"])
related_links = await self.link_service.find_by_note_id(note)
for link in related_links:
sourceNoteId = json.loads(link["sourceNote"])["id"]
targetNoteId = json.loads(link["targetNote"])["id"]
if link["id"] not in link_ids:
link_ids.add(link["id"])
links.append(link)
if sourceNoteId not in note_ids:
note_ids.add(sourceNoteId)
related_note = await self.note_service.find_by_id(sourceNoteId)
notes.append(related_note[0])
if targetNoteId not in note_ids:
note_ids.add(targetNoteId)
related_note = await self.note_service.find_by_id(targetNoteId)
notes.append(related_note[0])
print(len(notes),len(links))
await self.redis_service.addGraphSearchResultTask(data.user_id,notes,links)Graph 탐색 파이프라인에서 보여드린 Orchestrator 코드를 보면 한 가지 아쉬운 점이 있습니다.
Orchestrator는 여러 Service를 조합하여 흐름을 만들어줄 뿐 비즈니스 로직에는 관여하면 안된다는 것이 역할 책임의 규칙입니다.
하지만 위 코드에서는 단순히 Service를 조합하는 것을 넘어 note와 link를 관리하고 중복을 제거하는 등의 비즈니스 로직에 관여하는 것을 알 수 있습니다.
이러한 부분을 알고 있지만서도... 구현을 위주로 개발을 하다보면 이렇게 코드가 복잡해지는 경우가 종종 있습니다.
그래서 이러한 부분을 graph service로 분리하여 최적화 해보도록 하겠습니다.
현재 Orchestrator는 note와 link 중복 처리, 그래프 생성과 같은 그래프의 전반적인 비즈니스 로직을 처리하고 있습니다.
class GraphService:
def __init__(self, note_service: NoteService, link_service: LinkService):
self.note_service = note_service
self.link_service = link_service
async def graph_expand(self, start_notes):
notes = []
links = []
note_ids = set()
link_ids = set()
for note in start_notes:
if note["id"] not in note_ids:
notes.append(note)
note_ids.add(note["id"])
related_links = await self.link_service.find_by_note_id(note)
for link in related_links:
sourceNoteId = json.loads(link["sourceNote"])["id"]
targetNoteId = json.loads(link["targetNote"])["id"]
if link["id"] not in link_ids:
link_ids.add(link["id"])
links.append(link)
if sourceNoteId not in note_ids:
note_ids.add(sourceNoteId)
related_note = await self.note_service.find_by_id(sourceNoteId)
notes.append(related_note[0])
if targetNoteId not in note_ids:
note_ids.add(targetNoteId)
related_note = await self.note_service.find_by_id(targetNoteId)
notes.append(related_note[0])
return notes, links우선 GraphService에 graph_expand 함수를 만들어 비즈니스 로직을 빼주었습니다.
class GraphSearchOrchestrator:
def __init__(self, note_service: NoteService, link_service: LinkService, graph_service: GraphService, redis_service: RedisService):
self.note_service = note_service
self.link_service = link_service
self.graph_service = graph_service
self.redis_service = redis_service
async def graph_search(self, data:GraphSearchRequest):
embedding = await create_embedding(data.content)
top_k_notes = await self.note_service.get_top_k_notes_by_embedding(data.user_id, embedding, 3)
notes, links = await self.graph_service.graph_expand(top_k_notes)
print(len(notes),len(links))
await self.redis_service.addGraphSearchResultTask(data.user_id,notes,links)그 후 Orchestrator에서는 GraphService를 의존받아 호출하는 형식으로 정리해주었습니다.
이제 Orchestrator는 Service의 흐름만 조절할 뿐 이전처럼 비즈니스 로직에 관여하지 않습니다.
대신 비즈니스 로직은 GraphService에서 처리하게 됩니다.
이렇게 코드를 정리해주면 이후에 로직이 수정되더라도 Orchestrator는 수정할 필요가 없으며, GraphService만 수정한 후 테스트 및 적용을 할 수 있습니다.
😊 마무리
✅ Graph 생성 파이프라인 구현
✅ Graph 탐색 파이프라인 구현
✅ SSE 설계, Orchestrator 최적화 등 리팩토링 진행
➡️ Graph 탐색 실시간 렌더링 구현
[진행 상황 정리]
이번 글에서는 SYNAPSE의 핵심 기능인 GraphRAG의 Graph 생성과 탐색에 대해서 설계하고 구현하는 것 까지 정리했습니다.
또한 설계 및 구현하면서 만난 다양한 고민과 문제를 어떻게 해결했는지 정리했습니다.
코드 양은 보여드린 것 보다 훨씬 많고 복잡하게 구조화 되어있지만, 요즘 시대에 중요한 것은 코드 작성이 아닌 어떻게 설계할 것인지가 더 중요하다고 생각해서 최대한 코드는 덜어내고 설계적인 관점 위주에서 정리해 봤습니다.
다음 글에서는 Graph 탐색 실시간 렌더링 구현에 대해서 적어보려 합니다.
현재는 Graph를 생성한 후 한번에 데이터를 보내주면 렌더링하는 방식입니다.
하지만 제가 원하는 것은 Graph를 탐색하는 과정을 실시간으로 렌더링하여 사용자에게 와우 포인트를 주는 것입니다.
그럼 지금까지
읽어주셔서 감사드리며, 다음에 더 좋은 글로 돌아오도록 하겠습니다.
by. 대학생 개발자 주이어
KYT CODING COMMUNITY Discord 서버에 가입하세요!
Discord에서 KYT CODING COMMUNITY 커뮤니티를 확인하세요. 27명과 어울리며 무료 음성 및 텍스트 채팅을 즐기세요.
discord.com
KYT CODING COMMUNITY 가입하기!