把非结构化会议记录转化为可查询的知识图谱，并支持增量更新——无需每次全量重处理。
会议记录是组织智能的金矿。它们记录了决策、行动项、参与者信息，以及人与任务之间的关系。但多数组织仍将其当作静态文档，仅能进行基础的全文检索。
不如想象一下，像查询数据库那样查询你的会议：
• “谁参加过主题为‘预算规划’的会议？” • “Sarah 在所有会议中被分配了哪些任务？” • “展示四季度所有涉及工程团队的决策。”
这正是知识图谱的用武之地。将非结构化会议记录抽取为结构化信息并构建图表示，你就能解锁强大的基于关系的查询，这是传统文档存储无法实现的。
本文将构建一个实用的 CocoIndex pipeline，它可以：
从 Google Drive 读取 Markdown 会议记录 使用 LLM 抽取结构化实体（meetings、participants、tasks） 将数据以知识图谱的形式写入 Neo4j 仅在源文档发生变化时自动更新
完整源码在 GitHub 提供。https://github.com/cocoindex-io/meeting-notes-knowledge-graph
架构总览
该 pipeline 遵循清晰的数据流，并在每个阶段内置增量处理：
Google Drive（Documents - 带变更追踪）
→ 识别变更的文档
→ 按会议拆分
→ 使用 LLM 抽取结构化数据（仅处理变更的文档）
→ 收集节点与关系
→ 导出到 Neo4j（带 upsert 逻辑）
前置条件
安装 Neo4j 并在本地启动 默认本地浏览器地址：http://localhost:7474本文示例默认凭证：用户名
neo4j，密码cocoindex配置你的 OpenAI API key
准备 Google Drive： 创建一个 Google Cloud service account 并下载其 JSON 凭证 将源文件夹共享给该 service account 的邮箱 收集你希望导入的根文件夹 ID
详情参见 Setup for Google Drive
环境变量
设置如下环境变量：
export OPENAI_API_KEY=sk-...
export GOOGLE_SERVICE_ACCOUNT_CREDENTIAL=/absolute/path/to/service_account.json
export GOOGLE_DRIVE_ROOT_FOLDER_IDS=folderId1,folderId2
说明：
GOOGLE_DRIVE_ROOT_FOLDER_IDS接受以逗号分隔的多个文件夹 ID该流程会轮询最近变更，并定期刷新
下面逐一拆解各组件：
Flow Definition
概览
添加 source 和 collector
@cocoindex.flow_def(name="MeetingNotesGraph")
defmeeting_notes_graph_flow(
flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope
) -> None:
"""
Define an example flow that extracts triples from files and builds knowledge graph.
"""
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),
)
该 pipeline 首先通过 service account 连接 Google Drive。CocoIndex 的内置 source 连接器负责鉴权，并提供增量变更检测。recent_changes_poll_interval 表示每 10 秒检查一次新增或修改文件，而 refresh_interval 决定整个 flow 的重跑频率（每分钟一次）。
这正是 CocoIndex 的超级能力之一：带自动变更跟踪的增量处理。框架不是每次都重处理全部文档，而是：
从 Google Drive 列出带最后修改时间的文件
仅识别自上次成功运行以来新增或修改的文件
完全跳过未变更的文件
只将变更的文档传递到下游
效果如何？在一个日均 1% 变更率的企业环境中，只有 1% 的文档会触发下游处理。未变更的文件不会调用你的 LLM API，不会生成 Neo4j 查询，更不会消耗计算资源。
添加 collector
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()
随后，pipeline 使用不同的 collector 来收集不同实体类型和关系的数据：
Meeting Nodes —— 存储会议本身，包含日期与记录
Attendance Relationships —— 捕捉谁参加了会议，以及其是否为组织者
Task Decision Relationships —— 将会议与决策（被决定的任务）关联
Task Assignment Relationships —— 将具体任务分配给人员
处理每个文档
抽取会议
with data_scope["documents"].row() as document:
document["meetings"] = document["content"].transform(
cocoindex.functions.SplitBySeparators(
separators_regex=[r"\n\n##?\ "], keep_separator="RIGHT"
)
)
会议文档常在同一文件中包含多场会议。本步骤基于 Markdown 标题（## 或 #，且前有空行）进行拆分，将每个段落视作一场独立会议。keep_separator="RIGHT" 表示分隔符（标题）保留在右侧段落，以保持上下文。
抽取单场会议
定义 Meeting schema
@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 提供了明确的抽取指引和 schema。相比让 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-5"
),
output_type=Meeting,
)
)
至关重要的是，这一步同样受益于增量处理。由于 ExtractByLlm 是重步骤，我们将其输出缓存；只要输入（输入文本、模型、输出类型定义）未发生变化，就会复用缓存结果，无需重新调用 LLM。
收集关系
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"],
)
with parsed["tasks"].row() as task:
decided_tasks_rels.collect(
id=cocoindex.GeneratedField.UUID,
**meeting_key,
description=task["description"],
)
with task["assigned_to"].row() as assigned_to:
assigned_rels.collect(
id=cocoindex.GeneratedField.UUID,
**meeting_key,
task=task["description"],
person=assigned_to["name"],
)
在 CocoIndex 中，Collectors 类似内存缓冲区：你为不同类别（meeting nodes、attendance、tasks、assignments）声明各自的 collector，然后在处理每个文档时“收集”相关条目。
这段代码从解析后的会议记录中收集节点与关系，以便使用 CocoIndex 在 Neo4j 中构建知识图谱：
Person → Meeting（ATTENDED） 将参与者（包括组织者）与他们参加的会议相连。
Meeting → Task（DECIDED） 将会议与所做出的任务/决策相连。
Person → Task（ASSIGNED_TO） 将任务与负责该任务的人员相连。
映射到图数据库
概览
我们将创建一个 property graph，包含如下节点与关系。关于 property graph 的更多信息，请参阅 CocoIndex 的文档：Property Graph Targets。
映射 Meeting Nodes
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 关系
ATTENDED relationships
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"],
)
该调用确保将 ATTENDED 关系——即“Person → Meeting”（组织者或参与者 → 会议）——明确编码为 Neo4j 图中的边。
它通过
ATTENDED关系将Person节点与Meeting节点相连，从而支持诸如“Alice 参加了哪些会议？”或“会议 X 都有谁参加？”之类的查询。通过一致且正确地映射
Person和Meeting节点（使用唯一键），可避免重复的人员或会议。由于关系具有唯一 ID，并以一致键导出，图可在增量更新中保持稳定：重跑不会产生重复的边或节点。
映射 DECIDED 关系
DECIDED relationships
decided_tasks_rels.export(
"decided_tasks_rels",
cocoindex.targets.Neo4j(
connection=conn_spec,
mapping=cocoindex.targets.Relationships(
rel_type="DECIDED",
source=cocoindex.targets.NodeFromFields(
label="Meeting",
fields=[
cocoindex.targets.TargetFieldMapping("note_file"),
cocoindex.targets.TargetFieldMapping("time"),
],
),
target=cocoindex.targets.NodeFromFields(
label="Task",
fields=[
cocoindex.targets.TargetFieldMapping("description"),
],
),
),
),
primary_key_fields=["id"],
)
该调用将 DECIDED 关系——即“Meeting → Task”——明确编码为 Neo4j 图中的边。
它通过
DECIDED关系将Meeting节点与Task节点相连，从而支持如下查询：“会议 X 决定了哪些任务？”
“任务 Y 源于哪场会议？”
通过一致的键来映射
Meeting与Task（会议使用note_file + time，任务使用description），可避免图中出现重复的任务或会议节点。由于关系具有唯一 ID，并以一致键导出，图可在增量更新中保持稳定：重跑 pipeline 不会产生重复的边或节点。
映射 ASSIGNED_TO 关系
ASSIGNED_TO relationships
assigned_rels.export(
"assigned_rels",
cocoindex.targets.Neo4j(
connection=conn_spec,
mapping=cocoindex.targets.Relationships(
rel_type="ASSIGNED_TO",
source=cocoindex.targets.NodeFromFields(
label="Person",
fields=[
cocoindex.targets.TargetFieldMapping(
source="person", target="name"
),
],
),
target=cocoindex.targets.NodeFromFields(
label="Task",
fields=[
cocoindex.targets.TargetFieldMapping(
source="task", target="description"
),
],
),
),
),
primary_key_fields=["id"],
)
最终图谱
运行该 pipeline 后，你的 Neo4j 数据库将包含一个丰富、可查询的图：
Nodes：
Meeting—— 表示单场会议，包含日期与记录等属性Person—— 表示参与会议的个人Task—— 表示在会议中决定的可执行事项
Relationships：
ATTENDED—— 连接人员与他们参加的会议DECIDED—— 连接会议与其决定的任务ASSIGNED_TO—— 连接人员与他们负责的任务
重要的是，在最终导出到知识图谱的步骤中，CocoIndex 也采用增量方式。CocoIndex 仅对发生变化的节点或关系进行变更；对于未变更的部分则不做任何操作。这避免了目标数据库的无谓震荡，并最小化写入成本。
运行
构建/更新图谱
安装依赖：
pip install -e .
更新索引（运行一次 flow 来构建/更新图谱）：
cocoindex update main
浏览知识图谱
打开 Neo4j Browser：http://localhost:7474。
示例 Cypher 查询：
// 所有关系
MATCH p=()-->() RETURN p
// 谁参加了哪些会议（包含组织者）
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
企业级真实场景
该模式远不止适用于会议记录：
Research Paper Analysis —— 从组织知识库中抽取论文，在成千上万份文档中构建概念与引用的知识图谱，并追踪引用与概念的更新
Customer Support Tickets —— 抽取问题、解决方案，以及工单与客户之间的关系；在处理频繁编辑与状态更新的同时，识别成千上万工单中的模式
Email Thread Summarization —— 在数百万封邮件中构建沟通模式与决策结果的图谱；应对团队转发、编辑并引用历史讨论的现实
Compliance Documentation —— 从政策文档中抽取监管要求；追踪政策变更并通过图结构级联其影响；维护文档版本的审计轨迹
Competitive Intelligence —— 从公开文档与新闻中抽取数据；构建竞争对手关系、产品与市场定位的知识图谱，并处理持续更新
