examples/meeting_notes_graph_surrealdb/main.py
Rama main · réplica de lectura del repositorio original.
"""
Meeting Notes Graph (v1) — CocoIndex pipeline example.
Ingest Markdown meeting notes from a local folder, split each note into
per-meeting sections at heading boundaries, extract structured information
with LiteLLM + instructor, deduplicate person names with embedding-based
entity resolution, and build a knowledge graph in SurrealDB:
meeting records — one per meeting section
person records — canonical organizers, participants, and task assignees
task records — tasks decided in meetings
attended person -> meeting (with is_organizer flag)
decided meeting -> task
assigned_to person -> task
The pipeline runs in three phases:
1. Per-file extraction declares meeting and task records plus decided edges,
and publishes raw (un-resolved) person names into an in-memory LiveMap.
2. Person entity resolution maps raw names to canonical names.
3. Phases 2+3 run together in a periodically-refreshed component that reads
the LiveMap snapshot and declares canonical person records and the
person-touching edges (attended, assigned_to) using resolved names.
Run once (catch-up) or keep it watching the notes folder (live):
cocoindex update main # one-shot
cocoindex update -L main # live: file edits stream into the graph
"""
from __future__ import annotations
import datetime
import os
import pathlib
import re
from collections.abc import AsyncIterator
from dataclasses import dataclass
from typing import Annotated, Any, cast
import instructor
import litellm
import pydantic
import cocoindex as coco
from cocoindex.connectors import localfs, surrealdb
from cocoindex.ops.entity_resolution import ResolvedEntities, resolve_entities
from cocoindex.ops.entity_resolution.llm_resolver import LlmPairResolver
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder
from cocoindex.resources.file import FileLike, PatternFilePathMatcher
from cocoindex.resources.id import IdGenerator
from cocoindex.resources.live_map import LiveMap
litellm.drop_params = True
# ---------------------------------------------------------------------------
# Context keys
# ---------------------------------------------------------------------------
KG_DB = coco.ContextKey[surrealdb.ConnectionFactory]("kg_db")
LLM_MODEL = coco.ContextKey[str]("llm_model", detect_change=True)
RESOLUTION_LLM_MODEL = coco.ContextKey[str]("resolution_llm_model", detect_change=True)
EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder", detect_change=True)
# ---------------------------------------------------------------------------
# Lifespan
# ---------------------------------------------------------------------------
@coco.lifespan
async def coco_lifespan(
builder: coco.EnvironmentBuilder,
) -> AsyncIterator[None]:
kg_db = surrealdb.ConnectionFactory(
url=os.environ.get("SURREALDB_URL", "ws://localhost:8000/rpc"),
namespace=os.environ.get("SURREALDB_NS", "cocoindex"),
database=os.environ.get("SURREALDB_DB", "meeting_notes"),
credentials={
"username": os.environ.get("SURREALDB_USER", "root"),
"password": os.environ.get("SURREALDB_PASS", "root"),
},
)
builder.provide(KG_DB, kg_db)
builder.provide(LLM_MODEL, os.environ.get("LLM_MODEL", "openai/gpt-5-mini"))
builder.provide(
RESOLUTION_LLM_MODEL,
os.environ.get("RESOLUTION_LLM_MODEL", "openai/gpt-5-mini"),
)
builder.provide(
EMBEDDER,
SentenceTransformerEmbedder("Snowflake/snowflake-arctic-embed-xs"),
)
yield
await kg_db.close()
# ---------------------------------------------------------------------------
# SurrealDB row schemas (dataclasses for declare_record / declare_relation).
# The `id` field becomes the SurrealDB record id.
# ---------------------------------------------------------------------------
@dataclass
class Meeting:
id: int # Generated via generate_id((note_file, time_iso))
note_file: str
# Stored as an ISO string: the connector's datetime mapping requires
# SurrealDB datetime literals, which JSON transport doesn't produce.
time: Annotated[datetime.date, surrealdb.SurrealType("string")]
note: str
@dataclass
class Person:
id: str # canonical name — traversals read as person:`Alice Chen`
@dataclass
class Task:
id: str # task description
@dataclass
class AttendedRel:
"""attended edge payload. The relation record id is auto-derived from
(from_id=person, to_id=meeting_id) by the SurrealDB connector — we mount
this relation without a TableSchema so the connector's endpoint-based
fallback supplies the id, giving exactly one edge per (person, meeting).
"""
is_organizer: bool
# decided and assigned_to carry no payload — declared without schema or
# record, with the connector deriving record ids from (from_id, to_id).
# ---------------------------------------------------------------------------
# LLM extraction schemas (Pydantic, for instructor)
# ---------------------------------------------------------------------------
class ExtractedPerson(pydantic.BaseModel):
name: str = pydantic.Field(
description="Full name of the person, as written in the note."
)
class ExtractedTask(pydantic.BaseModel):
description: str = pydantic.Field(
description="Concise, standalone description of the task or action item."
)
assigned_to: list[ExtractedPerson] = pydantic.Field(
default_factory=list,
description="People the task is assigned to.",
)
class ExtractedMeeting(pydantic.BaseModel):
time: datetime.date = pydantic.Field(
description="Date of the meeting in ISO format (YYYY-MM-DD)."
)
note: str = pydantic.Field(
description="A brief summary or notes from the meeting section.",
)
organizer: ExtractedPerson = pydantic.Field(
description="The person who organized or led the meeting."
)
participants: list[ExtractedPerson] = pydantic.Field(
default_factory=list,
description=(
"People who attended the meeting other than the organizer. "
"Do not include the organizer here."
),
)
tasks: list[ExtractedTask] = pydantic.Field(
default_factory=list,
description="Action items or tasks decided in the meeting.",
)
EXTRACT_PROMPT = """\
You are an expert at reading meeting notes and extracting structured information.
Given a single meeting section (Markdown), extract:
- The meeting date (look for a date in the heading or body; required).
- A brief note summarizing what the meeting was about.
- The organizer (the person who ran the meeting). If unclear, pick the person
who appears most central to the meeting.
- Participants other than the organizer.
- Tasks or action items decided, including who they are assigned to.
Return only what is supported by the text. Use full names where available.
"""
# ---------------------------------------------------------------------------
# LLM extraction
# ---------------------------------------------------------------------------
@coco.fn(memo=True)
async def extract_meeting(section_text: str) -> ExtractedMeeting:
"""Extract a structured Meeting from a Markdown section via LiteLLM + instructor."""
client = cast(
instructor.AsyncInstructor,
instructor.from_litellm(litellm.acompletion, mode=instructor.Mode.JSON),
)
result = await client.chat.completions.create(
model=coco.use_context(LLM_MODEL),
response_model=ExtractedMeeting,
messages=[
{"role": "system", "content": EXTRACT_PROMPT},
{"role": "user", "content": section_text},
],
)
# Re-validate to restore class identity for pickling.
return ExtractedMeeting.model_validate(result.model_dump())
# ---------------------------------------------------------------------------
# Splitting — match v0's `\n\n##? ` heading regex
# ---------------------------------------------------------------------------
_HEADING_RE = re.compile(r"\n\n##?\s+")
def _split_meetings(text: str) -> list[str]:
parts = _HEADING_RE.split("\n\n" + text)
return [p.strip() for p in parts if p.strip()]
# ---------------------------------------------------------------------------
# Internal transfer types (Phase 1 → Phase 3)
# ---------------------------------------------------------------------------
@dataclass
class MeetingExtraction:
"""Raw per-meeting data carried forward to entity resolution + relation declaration."""
meeting_id: int
organizer: str # raw name
participants: list[str] # raw names
task_assignees: list[
tuple[str, list[str]]
] # (task_description, [raw assignee names])
# ---------------------------------------------------------------------------
# Phase 1: per-meeting and per-file processing
# ---------------------------------------------------------------------------
# Not memoized: LiveMap producers must re-run each session to refill the
# in-memory map. The expensive part — extract_meeting — is memoized per section.
@coco.fn
async def process_file(
file: FileLike,
meeting_table: surrealdb.TableTarget[Meeting],
task_table: surrealdb.TableTarget[Task],
decided_rel: surrealdb.RelationTarget[Any],
extractions: LiveMap[str, list[MeetingExtraction]],
) -> None:
text = await file.read_text()
note_file = file.file_path.path.as_posix()
id_generator = IdGenerator()
file_extractions = []
for section in _split_meetings(text):
extracted = await extract_meeting(section)
meeting_id = await id_generator.next_id(extracted.time)
meeting_table.declare_record(
row=Meeting(
id=meeting_id,
note_file=note_file,
time=extracted.time,
note=extracted.note,
)
)
for task in extracted.tasks:
task_table.declare_record(row=Task(id=task.description))
decided_rel.declare_relation(from_id=meeting_id, to_id=task.description)
file_extractions.append(
MeetingExtraction(
meeting_id=meeting_id,
organizer=extracted.organizer.name,
participants=[p.name for p in extracted.participants],
task_assignees=[
(t.description, [a.name for a in t.assigned_to])
for t in extracted.tasks
],
)
)
extractions.declare_entry(note_file, file_extractions)
# ---------------------------------------------------------------------------
# Phase 2: Person entity resolution
# ---------------------------------------------------------------------------
@coco.fn(memo=True)
async def _resolve_persons(raw_persons: set[str]) -> ResolvedEntities:
return await resolve_entities(
entities=raw_persons,
embedder=coco.use_context(EMBEDDER),
resolve_pair=LlmPairResolver(model=coco.use_context(RESOLUTION_LLM_MODEL)),
)
# ---------------------------------------------------------------------------
# Phase 3: declare canonical person records + person-touching relations.
# Runs as an auto-refreshed component: each cycle reads the LiveMap snapshot,
# so edits streamed in by phase 1 reach the graph within one interval.
# ---------------------------------------------------------------------------
async def sync_person_graph(
extractions: LiveMap[str, list[MeetingExtraction]],
person_table: surrealdb.TableTarget[Person],
attended_rel: surrealdb.RelationTarget[Any],
assigned_rel: surrealdb.RelationTarget[Any],
) -> None:
meetings: list[MeetingExtraction] = []
async for _note_file, file_extractions in extractions:
meetings.extend(file_extractions)
raw_persons: set[str] = set()
for m in meetings:
raw_persons.add(m.organizer)
raw_persons.update(m.participants)
for _task_desc, assignees in m.task_assignees:
raw_persons.update(assignees)
persons = await _resolve_persons(raw_persons)
# Declare canonical person records.
for canonical_name in persons.canonicals():
person_table.declare_record(row=Person(id=canonical_name))
for m in meetings:
# attended — aggregate organizer + participants. Organizer flag wins
# on collision so a person listed as both gets a single edge with
# is_organizer=true. Resolution happens before aggregation so two
# raw names that resolve to the same person also collapse.
attendees: dict[str, bool] = {persons.canonical_of(m.organizer): True}
for p in m.participants:
attendees.setdefault(persons.canonical_of(p), False)
for canonical, is_organizer in attendees.items():
attended_rel.declare_relation(
from_id=canonical,
to_id=m.meeting_id,
record=AttendedRel(is_organizer=is_organizer),
)
# assigned_to — dedup per (canonical person, task description).
for task_desc, assignees in m.task_assignees:
seen: set[str] = set()
for raw in assignees:
canonical = persons.canonical_of(raw)
if canonical in seen:
continue
seen.add(canonical)
assigned_rel.declare_relation(from_id=canonical, to_id=task_desc)
# ---------------------------------------------------------------------------
# App main
# ---------------------------------------------------------------------------
@coco.fn
async def app_main(notesdir: pathlib.Path) -> None:
# --- Mount node tables ---
meeting_table = await surrealdb.mount_table_target(
KG_DB,
"meeting",
await surrealdb.TableSchema.from_class(Meeting),
)
person_table = await surrealdb.mount_table_target(
KG_DB,
"person",
await surrealdb.TableSchema.from_class(Person),
)
task_table = await surrealdb.mount_table_target(
KG_DB,
"task",
await surrealdb.TableSchema.from_class(Task),
)
# --- Mount relation targets ---
# attended carries is_organizer; mounted without a schema so the connector
# auto-derives the relation record id from (from_id, to_id).
attended_rel = await surrealdb.mount_relation_target(
KG_DB, "attended", person_table, meeting_table
)
decided_rel = await surrealdb.mount_relation_target(
KG_DB, "decided", meeting_table, task_table
)
assigned_rel = await surrealdb.mount_relation_target(
KG_DB, "assigned_to", person_table, task_table
)
# --- Phase 1: per-file extraction, live-watching the notes folder ---
extractions: LiveMap[str, list[MeetingExtraction]] = await LiveMap.create()
source = localfs.walk_dir(
notesdir,
recursive=True,
path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]),
live=True,
)
handle = await coco.mount_each(
process_file,
source.items(),
meeting_table,
task_table,
decided_rel,
extractions,
)
# Wait for the initial scan so the first aggregation sees every note.
await handle.ready()
# --- Phases 2+3: entity resolution + person graph, refreshed periodically ---
await coco.mount(
coco.auto_refresh(sync_person_graph, interval=datetime.timedelta(seconds=5)),
extractions,
person_table,
attended_rel,
assigned_rel,
)
app = coco.App(
coco.AppConfig(name="MeetingNotesGraphSurrealDB"),
app_main,
notesdir=pathlib.Path("./notes"),
)