회의록에서 자동 업데이트되는 Neo4j 지식 그래프를 구축한 방법 (LLM 비용을 99% 절감)
Source: Dev.to
문제: 회의 노트가 낭비되고 있다
미국만 해도 매일 6,200만 ~ 8,000만 건의 회의가 열립니다. 회의에서는 결정, 액션 아이템, 작업 할당이 이루어지지만, 대부분의 인사이트는 Google Docs 안에서 사라집니다.
“예산 회의에 누가 참석했지?” 혹은 “이번 달에 Alex에게 어떤 작업이 할당됐지?”를 알고 싶나요? 수천 개의 Markdown 파일을 뒤져야 하는 상황을 상상해 보세요.
진짜 문제는? 회의 노트는 살아있는 문서라는 점입니다. 사람들은 이름을 수정하고, 작업을 재할당하고, 결정을 업데이트합니다. 증분 처리 없이 전체를 다시 처리해야 하니 두 가지 선택지에 갇히게 됩니다:
- 💸 모든 문서를 다시 처리하면서 발생하는 거대한 LLM 비용
- 📉 오래되고 정체된 지식 그래프
저는 변경된 문서만 처리하는 자동 업데이트 Neo4j 지식 그래프를 구축해 LLM 비용을 99 % 절감했습니다.
우리가 만들고 있는 것
지저분한 회의 노트를 질의 가능한 그래프 데이터베이스로 변환하는 파이프라인:
Google Drive → Detect Changes → Split Meetings → LLM Extract → Neo4j결과: 세 가지 노드 타입(Meeting, Person, Task)과 세 가지 관계(ATTENDED, DECIDED, ASSIGNED_TO)를 통해 다음과 같은 질의를 할 수 있습니다:
- “Sarah는 어떤 회의에 참석했나요?”
- “이 작업은 어느 회의에서 결정됐나요?”
- “Q4 작업을 모두 누가 담당하고 있나요?”
핵심 비법: 증분 처리
1. 변경된 부분만 처리
Google Drive 소스는 마지막 수정 타임스탬프를 추적합니다. 100 000개의 회의 노트 중 매일 1 %만 바뀐다면, 1 000개의 파일만 처리하면 됩니다—100 000개를 모두 처리할 필요가 없습니다.
@cocoindex.flow_def(name="MeetingNotesGraph")
def meeting_notes_graph_flow(
flow_builder: cocoindex.FlowBuilder,
data_scope: cocoindex.DataScope
) -> None:
credential_path = os.environ["GOOGLE_SERVICE_ACCOUNT_CREDENTIAL"]
root_folder_ids = os.environ["GOOGLE_DRIVE_ROOT_FOLDER_IDS"].split(",")
data_scope["documents"] = flow_builder.add_source(
cocoindex.sources.GoogleDrive(
service_account_credential_path=credential_path,
root_folder_ids=root_folder_ids,
recent_changes_poll_interval=datetime.timedelta(seconds=10),
),
refresh_interval=datetime.timedelta(minutes=1),
)효과: 일반적인 1 % 일일 변동률에서 LLM API 비용을 99 % 절감합니다.
2. 스마트 문서 분할
회의 파일에는 여러 세션이 포함되는 경우가 많습니다. 헤더(## Meeting Title 등)를 각 섹션에 유지하면서 지능적으로 분할해 LLM이 컨텍스트를 잃지 않게 합니다.
with data_scope["documents"].row() as document:
document["meetings"] = document["content"].transform(
cocoindex.functions.SplitBySeparators(
separators_regex=[r"\n\n##?\ "],
keep_separator="RIGHT",
)
)3. 구조화된 LLM 추출
“어떤 JSON이라도”라고 모델에 요구하는 대신 구체적인 스키마를 정의합니다.
@dataclass
class Person:
name: str
@dataclass
class Task:
description: str
assigned_to: list[Person]
@dataclass
class Meeting:
time: datetime.date
note: str
organizer: Person
participants: list[Person]
tasks: list[Task]캐시와 함께 추출하면 동일 입력에 대해 캐시된 결과를 재사용해 불필요한 LLM 호출을 없앨 수 있습니다.
with document["meetings"].row() as meeting:
parsed = meeting["parsed"] = meeting["text"].transform(
cocoindex.functions.ExtractByLlm(
llm_spec=cocoindex.LlmSpec(
api_type=cocoindex.LlmApiType.OPENAI,
model="gpt-4",
),
output_type=Meeting,
)
)그래프 구축
노드와 관계 수집
meeting_nodes = data_scope.add_collector()
attended_rels = data_scope.add_collector()
decided_tasks_rels = data_scope.add_collector()
assigned_rels = data_scope.add_collector()
meeting_key = {"note_file": document["filename"], "time": parsed["time"]}
meeting_nodes.collect(**meeting_key, note=parsed["note"])
attended_rels.collect(
id=cocoindex.GeneratedField.UUID,
**meeting_key,
person=parsed["organizer"]["name"],
is_organizer=True,
)
with parsed["participants"].row() as participant:
attended_rels.collect(
id=cocoindex.GeneratedField.UUID,
**meeting_key,
person=participant["name"],
)Upsert 로직을 이용한 Neo4j 내보내기
meeting_nodes.export(
"meeting_nodes",
cocoindex.targets.Neo4j(
connection=conn_spec,
mapping=cocoindex.targets.Nodes(label="Meeting"),
),
primary_key_fields=["note_file", "time"],
)Person과 Task 노드 선언:
flow_builder.declare(
cocoindex.targets.Neo4jDeclaration(
connection=conn_spec,
nodes_label="Person",
primary_key_fields=["name"],
)
)
flow_builder.declare(
cocoindex.targets.Neo4jDeclaration(
connection=conn_spec,
nodes_label="Task",
primary_key_fields=["description"],
)
)관계 내보내기:
attended_rels.export(
"attended_rels",
cocoindex.targets.Neo4j(
connection=conn_spec,
mapping=cocoindex.targets.Relationships(
rel_type="ATTENDED",
source=cocoindex.targets.NodeFromFields(
label="Person",
fields=[cocoindex.targets.TargetFieldMapping(
source="person", target="name"
)],
),
target=cocoindex.targets.NodeFromFields(
label="Meeting",
fields=[
cocoindex.targets.TargetFieldMapping("note_file"),
cocoindex.targets.TargetFieldMapping("time"),
],
),
),
),
primary_key_fields=["id"],
)파이프라인 실행
설정
export OPENAI_API_KEY=sk-...
export GOOGLE_SERVICE_ACCOUNT_CREDENTIAL=/path/to/service_account.json
export GOOGLE_DRIVE_ROOT_FOLDER_IDS=folderId1,folderId2
pip install cocoindex그래프 구축
cocoindex update mainNeo4j Browser(http://localhost:7474)에서 질의
// 누가 어떤 회의에 참석했나요?
MATCH (p:Person)-[:ATTENDED]->(m:Meeting)
RETURN p, m
// 회의에서 결정된 작업
MATCH (m:Meeting)-[:DECIDED]->(t:Task)
RETURN m, t
// 사람별 작업 할당
MATCH (p:Person)-[:ASSIGNED_TO]->(t:Task)
RETURN p, t왜 중요한가
1. 규모에 따른 비용 절감
- 전통적 방식: 100 000개 문서 재처리 → 100 000번 LLM 호출
- 증분 방식: 1 000개 변경 문서만 처리 → 1 000번 LLM 호출
결과: 99 % 비용 절감.
2. 실시간 업데이트
라이브 모드로 전환하면 회의 노트가 바뀔 때마다 그래프가 자동으로 업데이트됩니다:
refresh_interval=datetime.timedelta(minutes=1)3. 데이터 라인리지
CocoIndex는 모든 변환 과정을 추적하므로, Neo4j 노드를 LLM 추출 단계와 원본 문서까지 역추적할 수 있습니다.
회의 노트를 넘어
이 패턴은 문서가 시간이 지나면서 진화하는 텍스트 중심 도메인 전반에 적용할 수 있으며, 비용 효율적이고 최신 상태의 지식 그래프를 제공합니다.