我要投稿

基于 LLM 抽取与 Neo4j，从会议纪要构建自更新知识图谱

发布日期：2026-02-03 08:24:17 浏览次数： 1533

作者：AI大模型观察站

微信搜一搜，关注“AI大模型观察站”

把非结构化会议记录转化为可查询的知识图谱，并支持增量更新——无需每次全量重处理。

会议记录是组织智能的金矿。它们记录了决策、行动项、参与者信息，以及人与任务之间的关系。但多数组织仍将其当作静态文档，仅能进行基础的全文检索。

不如想象一下，像查询数据库那样查询你的会议：

• “谁参加过主题为‘预算规划’的会议？” • “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 —— 从公开文档与新闻中抽取数据；构建竞争对手关系、产品与市场定位的知识图谱，并处理持续更新

53AI，企业落地大模型首选服务商

产品：场景落地咨询+大模型应用平台+行业解决方案

承诺：免费POC验证，效果达标后再合作。零风险落地应用大模型，已交付160+中大型企业