用动态知识图谱重构 RAG：Cognee v2 让 AI 代理学会自主推理与实时演化

摘要
传统 RAG 系统依赖向量检索，面对多跳推理和持续更新的业务知识时往往力不从心。Cognee v2 将 Agentic RAG 与动态知识图谱深度融合，打造了一个能够自主构建、推理并实时演化的智能检索框架。本文通过一个最小可运行 Demo，还原其三大核心能力：从非结构化文本中自动抽取实体与关系、将复杂问题分解为图谱查询链、以及基于内存事件流即时更新知识。整个 Demo 使用规则模拟 LLM，无需任何外部 API，但架构完全兼容 GPT‑4、Claude 等真实大模型，让读者可以零成本体验下一代 RAG 的设计哲学。

为什么 RAG 需要“代理”和“知识图谱”？

过去两年，RAG（检索增强生成）已经成为大模型落地的事实标准。典型方案是：将文档切片存入向量数据库，查询时检索相似片段，拼接后交给 LLM 作答。这种模式在单跳事实查询上表现不错，但面对下面两类业务场景时却暴露出明显短板：

  • 多跳推理，例如“Alice 创立的公司的 CEO 是谁？”。答案需要先找公司，再找 CEO，仅靠向量相似度无法完成跨越两个检索步骤的逻辑链。
  • 知识持续演化，例如早上 CEO 是 Bob，下午变成 David。传统 RAG 往往需要重新索引文档，缓存命中与更新延迟让实时性很难保障。

于是社区提出了 Agentic RAG：让 LLM 扮演一个能制定计划、调用工具、迭代推理的“代理”。但在向量数据之上做多步检索，依然缺乏天然的结构化关联。真正适合关系推理的载体是知识图谱——它以节点和边直接表达实体与关系，一张图天然就是“多跳”的。

Cognee v2 正是在这一背景下诞生的 Agentic RAG 框架。它将动态知识图谱作为主要检索后端，让代理具备自主构建图谱、分解查询计划、并利用事件流实现知识实时演化的能力。下面通过一个不到 200 行代码的 Demo，演示其完整工作流。

技术方案：三大核心能力如何协同工作

Cognee v2 的设计可以抽象为三个互相协作的模块：

🧩 自主图谱构建

当一段文本传入系统时，代理会调用信息抽取函数（本 Demo 使用正则规则模拟 LLM）识别出主、谓、宾三元组，如 (Alice, founded, Acme Corp)。这些三元组被即时写入 KnowledgeGraph（基于 NetworkX 的有向图），同时生成一条带有时间戳的事件记录到内存记忆流中。

🧠 多级推理

用户输入问题后，代理不会直接去图里随机搜索。它会先将问题“翻译”成一系列图查询子任务，例如：

  1. 查找 Alice 的 founded 关系所指向的公司；
  2. 再查找该公司的 is_CEO_of 关系所指向的人物。
    这种 查询计划分解 让代理每一步都只做一件事，逐步逼近最终答案。对于更复杂的问题，可以自然扩展到三层、四层。

🌀 记忆流与实时演化

所有添加、删除、更新的操作都以事件流的形式被记录。当新文档说明“David 成为了 CEO”时，系统不但会更新图谱，还会保留旧版本的历史事件。后续问题若带时间限定（如“现在”），代理可以根据事件时间线，自动选取最新有效的关系。不需要重建索引，不需要暂停服务——知识在查询的那一刻就是最新的。

Demo 为了保证零外部依赖，使用了基于规则的 extract_triplets、decompose_question 和 generate_answer 三个函数来模拟 LLM。它们的接口和真实大模型调用完全一致：输入文本，输出结构化结果。替换成任何 LLM API 后，整套架构无需任何改动。

核心实现解析：从代码看 Cognee 的灵魂

整个 Demo 由两个文件构成：cognee.py 负责核心逻辑，main.py 编排演示流程。我们重点关注 cognee.py 中的关键设计。

知识图谱的轻量封装

类 KnowledgeGraph 基于 NetworkX 的有向图，并提供了语义化的增查接口：

1
2
3
4
5
6
7
8
9
10
11
12
13
class KnowledgeGraph:
    def __init__(self):
        self.graph = nx.DiGraph()
    
    def add_relation(self, head, relation, tail):
        self.add_entity(head)
        self.add_entity(tail)
        self.graph.add_edge(head, tail, relation=relation)
    
    def query_relation(self, head, relation):
        """返回所有与 head 具有给定 relation 的尾部节点"""
        return [v for u, v, data in self.graph.edges(data=True)
                if u == head and data.get('relation') == relation]

这里没有引入复杂的属性图模型，只保留最基本的节点与带标签的边。add_relation 自动确保节点存在，query_relation 则是多跳推理的基础原子操作。

代理大脑：Cognee 类

Cognee 把图谱、记忆流、以及三个模拟 LLM 的函数组织在一起。核心方法有两个：add_document 和 query。

文档摄入与三元组抽取

1
2
3
4
5
6
7
8
9
10
11
def add_document(self, doc_id, text):
    triplets = extract_triplets(text)   # 模拟 LLM 抽取
    timestamp = datetime.now().isoformat()
    for head, relation, tail in triplets:
        self.kg.add_relation(head, relation, tail)
        self.memory_stream.append({
            'action': 'add',
            'triplet': (head, relation, tail),
            'doc_id': doc_id,
            'time': timestamp
        })

extract_triplets 在 Demo 中实际使用正则实现：

1
2
3
4
5
6
7
8
9
def extract_triplets(text):
    # 简化：匹配 "A founded B", "B is CEO of C" 等模式
    patterns = [
        (r'(\w+) founded (\w+)', 'founded'),
        (r'(\w+) is CEO of (\w+)', 'is_CEO_of'),
        (r'(\w+) works for (\w+)', 'works_for'),
        (r'(\w+) is a (\w+) (\w+)', 'is_a'),
    ]
    ...

真实场景下，只需将这里的正则替换为一次 LLM 调用，提示词要求输出 JSON 三元组数组即可。

问题分解与图查询

query 方法是多级推理的入口：先调 decompose_question 得到子查询序列，然后逐步执行图查询，最后把查到的中间结果交给 generate_answer 生成最终回答。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def query(self, question):
    plan = decompose_question(question)      # 如 [('founded', 'Alice'), ('is_CEO_of', None)]
    current_results = None
    for step in plan:
        relation, entity_hint = step
        if entity_hint:
            current_results = self.kg.query_relation(entity_hint, relation)
        else:
            # 使用上一步的结果作为实体
            current_results = []
            for ent in current_results:
                current_results.extend(self.kg.query_relation(ent, relation))
    answer = generate_answer(question, current_results, self.memory_stream)
    return answer

例如问题“Who is the CEO of the company that Alice founded?”，模拟的 decompose_question 会输出 [('founded', 'Alice'), ('is_CEO_of', None)]。第一步找到“Acme Corp”，第二步对“Acme Corp”查询 is_CEO_of，最终获取“Bob”。整个推理链条完全由代理自主规划和执行，对上层透明。

记忆流辅助“现在”语义

当用户问“Who is the CEO of Acme Corp now?”时，generate_answer 除了查看图谱当前状态，还会检查记忆流中是否有更新的 is_CEO_of 事件。若存在更晚时间戳的 David 记录，则覆盖图谱静态答案。这展示了在真实系统中如何通过事件溯源机制实现实时知识演化，而不需要对图谱进行破坏性的“删除后插入”。

运行效果：一步步看推理如何发生

执行 python main.py，终端会依次输出以下关键结果（文字描述还原实际打印）：

  1. 添加文档

    • 文档1：Alice founded Acme Corp. Bob is CEO of Acme Corp.
    • 文档2：Acme Corp is a technology company. Charlie works for Acme Corp.
      代理在后台抽取的典型三元组：
      (Alice, founded, Acme Corp) → (Bob, is_CEO_of, Acme Corp) → (Acme Corp, is_a, technology company) → (Charlie, works_for, Acme Corp)

  2. 单跳查询
    问题：Who is the CEO of Acme Corp?
    代理直接在图谱中查找 Acme Corp 的 is_CEO_of 关系，返回 Bob。

  3. 多跳查询（演示核心亮点）
    问题：Who is the CEO of the company that Alice founded?
    代理的推理路径输出为：
    → “Alice founded Acme Corp”
    → “Acme Corp has CEO Bob”
    最终答案：Bob。整个过程没有一次性向量召回，而是两步精准图查询。

  4. 知识演化更新
    添加新文档：David is CEO of Acme Corp. Bob left the company.
    图谱中 Acme Corp 的 CEO 关系被更新为 David，同时记忆流记录下这一变更。

  5. 演化后查询
    问题：Who is the CEO of Acme Corp now?
    代理查看记忆流中最近的事件，返回 David。
    若接下来问“Who was the CEO before David?”，同样可以从记忆流历史中找到 Bob。这种按时间切片的能力在合同管理、招聘履历等场景中价值极高。

总结与展望

Cognee v2 用不到 200 行的 Demo 演示了一个完整的 Agentic RAG 工作流：从文本到图谱的自动构建，到多跳问题的计划式推理，再到基于事件流的实时知识演化。它彻底分离了“检索的结构”与“生成的智能”，让知识图谱承载关系，让代理负责逻辑。

实际生产环境中，只需将模拟函数替换为 LLM API，并接入真正的向量数据库（如 Qdrant）实现由语义触发的图谱混合检索，就能获得一个可扩展、可演化的企业级知识引擎。未来，Cognee 还可以进一步引入置信度评估、冲突检测、子图可视化等高级特性，让知识不再是静态的文档堆砌，而是一个不断生长、可推理、可信赖的“活系统”。

如果你正在设计下一代的 RAG 应用，不妨亲自跑一遍这个 Demo，感受用图形思考的美妙。