【导读】会议纪要沉淀着决策、行动项、责任人和跨团队协作关系，却常被当作静态文档做全文检索。随着 LLM 抽取与图数据库能力成熟，一种更“可计算”的组织记忆正在形成：把非结构化纪要转成结构化节点与关系，并在源文档变更时自动增量更新。本文拆解一套基于 CocoIndex pipeline、Google Drive 变更追踪与 Neo4j upsert 的实现思路，展示从抽取 schema 到 Cypher 查询的完整链路。
一、从“文档检索”到“关系查询”：会议纪要为何适合做知识图谱
会议记录在组织中天然具备“图结构”属性：人参与会议（参与关系）、会议产出任务与决策（决策关系）、任务再分配给具体责任人（分配关系）。但在多数企业里，它通常以 Markdown/Doc/PDF 的形态散落在协作盘里，即便做了企业搜索，也往往停留在关键词命中和片段回显。
知识图谱（尤其是 Neo4j 的 property graph）提供的是另一种能力：围绕实体（nodes）与关系（relationships）组织信息，使查询更接近业务问题本身。例如：
- “谁参加过主题为‘预算规划’的会议？”
- “Sarah 在所有会议中被分配了哪些任务？”
- “展示四季度所有涉及工程团队的决策。”
这类问题的难点并不在于“有没有文本”，而在于需要跨文档、跨段落抽取稳定的实体，并把它们以可追溯的关系网络连接起来。对会议纪要而言，至少存在三类核心对象足以支撑第一版图谱：
- Meeting：单场会议（时间、纪要正文、来源文件等）
- Person：组织者与参与者
- Task：会议中决定的可执行事项（含 assigned_to）
因此，一个可落地的目标是：把会议纪要从“全文检索的静态文本”，升级为“可通过 Cypher 进行关系查询的动态图谱”，并且要能适应现实世界的频繁修改——这正是“自更新/增量更新”的关键。
二、LLM 抽取 + CocoIndex 增量处理：自更新 pipeline 的核心机制
要让图谱持续可用，工程上通常会遇到两个成本黑洞：
1）会议纪要体量增长后，全量重处理会造成 LLM 调用与写库成本失控；
2）文档内容经常被补写、修订、追记，图谱需要跟随更新，同时避免重复节点/重复关系。
一套较清晰的数据流可以按以下链路组织，并在每个阶段植入增量策略：
Google Drive（Documents，带变更追踪）
→ 识别变更的文档
→ 按会议拆分（一个文件多场会议）
→ 使用 LLM 抽取结构化数据（仅处理变更文档）
→ 汇总节点与关系（collectors）
→ 导出到 Neo4j（upsert 逻辑）
其中，“增量更新”的实现通常来自两层机制叠加：
- Source 端变更检测：只把自上次成功运行以来“新增或修改”的文件传给下游；未变更文件直接跳过，从根源上减少 LLM 调用与计算浪费。
- LLM 抽取结果缓存：对 ExtractByLlm 这类重步骤做缓存；只要输入（文本内容、模型、output_type schema）不变，就复用历史结果。
- Target 端 upsert：写入 Neo4j 时用稳定的 primary key 做去重与更新，避免重跑产生重复 nodes/relationships，并把变化“收敛”为最小写入集。
在实现上，pipeline 会通过 service account 接入 Google Drive，并设置两个典型参数：
- recent_changes_poll_interval：例如每 10 秒轮询一次最近变更，用于快速捕捉新增/编辑
- refresh_interval：例如每分钟触发一次 flow 刷新，用于周期性跑批与状态对齐
这类设计的价值在于：当企业环境的日变更率只有 1% 左右时，真正触发下游 LLM 抽取与 Neo4j 写入的也只有约 1% 的文档，计算与 API 成本在规模化时更可控。
三、从 schema 到图谱映射：Meeting/Person/Task 与三类关系如何落到 Neo4j
1）按 Markdown 标题拆分“单场会议”
现实里一个会议纪要文件可能记录多次会议，因此需要先把文档切成“以会议为单位”的段落。常见策略是利用 Markdown 标题作为分隔符，例如根据 ## 或 #（并要求前置空行）拆分，并保留标题到右侧段落以保持上下文。
with data_scope["documents"].row() as document: document["meetings"] = document["content"].transform( cocoindex.functions.SplitBySeparators( separators_regex=[r"\n\n##?\ "], keep_separator="RIGHT" ) )
2）用 dataclass 明确 LLM 抽取的 output_type
相比让 LLM 输出自由格式文本，再做二次解析，直接给出强约束 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]
3）ExtractByLlm 抽取 + collectors 汇总节点与关系
LLM 抽取执行时指定 llm_spec、模型与 output_type，并把结果写入 parsed，随后使用多个 collector 分别收集 Meeting nodes 与三类关系数据。
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, ) )
collectors 收集的结构通常对应“图谱越写入的最小单元”，例如：
- meeting_nodes：会议节点（唯一键、note 等属性）
- attended_rels：Person → Meeting 的 ATTENDED 关系（含组织者标记）
- decided_tasks_rels：Meeting → Task 的 DECIDED 关系
- assigned_rels：Person → Task 的 ASSIGNED_TO 关系
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"], )
这里有两个决定增量稳定性的细节：
- Meeting 的主键：示例使用 note_file + time，确保同一文件同一日期会议可被稳定定位。
- Relationship 的唯一性：关系端用 GeneratedField.UUID 生成 id 并作为 primary_key_fields，配合一致的节点映射，重跑时不会重复插入边。
4）Neo4j 的 nodes/relationships 映射与 upsert
把 collectors 导出到 Neo4j 时，需要分别声明 nodes label、primary key 与关系映射方式。
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 节点（用于关系写入时自动 upsert 目标节点）：
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 关系（Person → Meeting）：
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.TargetField图("note_file"), cocoindex.targets.TargetFieldMapping("time"), ], ), ), ), primary_key_fields=["id"], )
DECIDED 关系（Meeting → Task）：
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"], )
ASSIGNED_TO 关系（Person → Task）：
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"], )
当这些映射成立后，图谱结构就相对清晰：
- Nodes：Meeting、Person、Task
- Relationships：ATTENDED、DECIDED、ASSIGNED_TO
并且在增量重跑时，只对发生变化的 nodes/relationships 做变更，未变更部分不写入，从而降低 Neo4j 的写入震荡与成本。
5）运行与查询：用 Cypher 把“会议”变成可组合的数据
构建/更新图谱常见命令包括安装依赖与触发一次更新：
pip install -e . cocoindex update main
进入 Neo4j Browser 后，可以用基础 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
这些查询之所以“更接近业务语言”，本质在于：图谱让“人-会-任务”的网络关系成为一等公民，而不是从一堆文档里临时拼接结果。
结语：技术背后的管理思考
将会议纪要做成可增量更新的知识图谱，表面是 LLM、ExtractByLlm、Neo4j、upsert、recent_changes_poll_interval 等技术组合，背后则是在重构组织的“可追溯协作链路”：决策从哪场会议产生、任务由谁认领、跨团队议题如何演进，都能用关系查询快速还原。这会直接影响组织效能——减少信息在群聊与文档夹中的沉没成本，让复盘、交接、审计与跨部门协同更接近“查数”而非“问人”。同时，它也会抬高人才能力要求：不仅需要会写纪要，更需要理解结构化表达、数据治理与权限边界，避免把敏感信息无差别写入图谱。正如红海云在探索新一代人力资源管理解决方案时所强调的，技术的终极价值在于赋能组织把知识资产沉淀为流程、指标与责任体系，从而让协作更透明、执行更可控、人才发展更有据可依。
