AutoGPT CoPilot 的 Graphiti 记忆系统：FalkorDB 图谱排查手册、配置全解与源码剖析

本文围绕 autogpt_platform/backend/backend/copilot/graphiti 目录下的开发者文档 AGENTS.md（由同目录 CLAUDE.md 以 @AGENTS.md 引用加载）展开，讲清 AutoGPT Platform 中 CoPilot 的 Graphiti 时序知识图谱记忆模块的职责边界、设计原则、全部配置项，以及一份可直接复制运行的 FalkorDB 查询手册，并结合 client.py、config.py、falkordb_driver.py 等源码剖析其隔离、缓存与容错机制。读完你既能独立排查某个用户的图谱数据质量，也能理解记忆检索为何要“优雅降级”而不是直接报错。

模块定位与职责边界

该目录是 CoPilot 的 Graphiti 记忆集成（Graphiti-backed memory integration），文档开篇即给出三条边界约定（AGENTS.md）：

• 文档性质：该文件仅面向开发者，不会被注入到 LLM 提示词中；真正运行时的提示词补充由 prompting.py 中的 get_graphiti_supplement() 生成——当 Graphiti 启用时，它会向系统提示词追加 “Memory System (Graphiti)” 一节，要求模型在回答前先检索持久化记忆工具。
• Scope 原则：所有 Graphiti 与 FalkorDB 相关的逻辑都应收敛在本包内，而不是散落到无关的 copilot 模块；改动优先发生在这里。
• 调试方法：在修改任何检索行为之前，先用原生 FalkorDB 查询检查已存储的节点、episode 和 RELATES_TO 事实；评估记忆质量时要区分三类来源——用户直接提供的事实、助手生成的结论、来源/元信息（provenance/meta）实体。

设计意图（Design Intent）有三条，且都能在源码中找到对应实现：

1. 按用户隔离：通过 group_id 限定的数据库与客户端实现每用户隔离——对应 client.py 的 derive_group_id() 与每用户独立 Graphiti 实例；
2. 警惕记忆污染：助手/工具措辞会污染记忆，抽取质量与写入成功同样重要——对应 ingest.py 中的 MemoryEnvelope/MemoryKind 记忆封包模型（memory_model.py）；
3. 检索必须优雅降级：warm-context 与工具驱动的召回失败时不能打断聊天执行——对应 context.py 中 fetch_warm_context() 对超时与任意异常统一返回 None 的行为。

配置全解：GRAPHITI_* 环境变量

所有配置集中在 config.py 的 GraphitiConfig（基于 pydantic_settings，前缀 GRAPHITI_）。开发环境默认值见 .env.default。

连接与模型参数

环境变量	默认值	说明
GRAPHITI_FALKORDB_HOST	localhost	FalkorDB 主机
GRAPHITI_FALKORDB_PORT	6380	FalkorDB 端口
GRAPHITI_FALKORDB_PASSWORD	空	密码；开发环境为 local-dev-password
GRAPHITI_LLM_MODEL	gpt-4.1-mini	实体抽取 LLM，必须支持结构化输出
GRAPHITI_EMBEDDER_MODEL	text-embedding-3-small	向量嵌入模型（与 LLM 相互独立，直连 OpenAI）
GRAPHITI_RERANKER_MODEL	gpt-4.1-nano	warm-context 检索的 cross-encoder 重排模型
GRAPHITI_LLM_BASE_URL / GRAPHITI_LLM_API_KEY	空	LLM 端点与密钥；空时走回退链（见下）
GRAPHITI_EMBEDDER_BASE_URL / GRAPHITI_EMBEDDER_API_KEY	空	嵌入端点与密钥；空时回退 CHAT_OPENAI_API_KEY → OPENAI_API_KEY
GRAPHITI_SEMAPHORE_LIMIT	5	摄取期间最大并发 LLM 调用数，防限流
GRAPHITI_CONTEXT_MAX_FACTS	20	warm context 一次预载的最大事实条数
GRAPHITI_CONTEXT_TIMEOUT	8.0 秒	warm context 抓取超时（为 FalkorDB 冷连接留了余量）
GRAPHITI_CLIENT_CACHE_MAXSIZE / _TTL	500 / 1800 秒	Graphiti 客户端 LRU 缓存上限与 TTL
GRAPHITI_FALKORDB_QUERY_MAX_ATTEMPTS	3	待处理队列背压错误（“Max pending queries exceeded”）的总尝试次数（含首次）
GRAPHITI_FALKORDB_QUERY_BACKOFF_BASE	0.1 秒	指数抖动退避的基础延迟
GRAPHITI_COMMUNITY_REBUILD_MIN_NEW_EPISODES	5	距上次重建新增 episode 少于该值时跳过社区重建（避免为几乎不变的图支付 LLM 摘要成本，且 Leiden 平局裁决非确定、摘要文本会漂移）
GRAPHITI_COMMUNITY_REBUILD_USE_FLEX_TIER	true	夜间社区重建走 OpenAI flex 服务档（约 5 折）；交互式摄取保持 sync 档，因为用户期望下一轮立即可用去重后的事实

API Key 回退链

从 config.py 的 resolve_llm_api_key() / resolve_embedder_api_key() 可以看到明确的回退顺序，设计目的是让记忆成本记在 AutoPilot 名下，而不是平台级公共 key：

• LLM key：GRAPHITI_LLM_API_KEY → CHAT_API_KEY → OPEN_ROUTER_API_KEY →（本地传输时）chat 配置的占位 key；
• Embedder key：GRAPHITI_EMBEDDER_API_KEY → CHAT_OPENAI_API_KEY → OPENAI_API_KEY →（本地传输时）chat 配置的占位 key。

本地 Ollama 传输下的模型自动改写

当 CHAT_USE_LOCAL=true 时，模型校验器 _apply_local_graphiti_models()（config.py）会把仍处于云端默认值、且未被操作者显式覆盖的三个模型改写为本地友好 slug：

• llm_model：gpt-4.1-mini → hf.co/unsloth/Qwen3.5-4B-GGUF:Q4_K_M（与聊天路径共用同一个 Ollama pull，已验证支持结构化输出）；
• reranker_model：gpt-4.1-nano → 同一 Qwen slug（重排提示更简单，复用聊天模型可避免为它单独拉一个模型）；
• embedder_model：text-embedding-3-small → nomic-embed-text（约 270MB、768 维；聊天路径不需要它，所以安装器的 --with-ollama 不覆盖，需要手动 ollama pull nomic-embed-text，详见 copilot-local-llm.md）。

操作者自行设置的非云端 slug（如 qwen3:8b）不受影响——只有字面量云端默认值才会触发改写。.env.default 中对此有完整注释。

特性开关

Graphiti 记忆并非默认开启。config.py 提供两个 LaunchDarkly 门控：

• is_enabled_for_user()：由 flag graphiti-memory 门控，决定该用户的记忆读写；未配置 LD 时默认关闭；
• is_communities_enabled_for_user()：由 flag GRAPHITI_COMMUNITIES_ENABLED 门控，控制每周 Leiden 社区重建 + LLM 摘要——与基础记忆开关解耦，即用户可以有记忆读写但不跑社区重建。

查询手册（Query Cookbook）

文档给出的排查命令要求统一从 autogpt_platform/backend 目录执行，并使用 poetry run ...。以下命令中的 USER_ID 应替换为实际用户 ID。

1. 推导用户的 group_id

poetry run python - <<'PY'
from backend.copilot.graphiti.client import derive_group_id
print(derive_group_id("883cc9da-fe37-4863-839b-acba022bf3ef"))
PY

derive_group_id() 会剥离非法字符、限制长度并加上 user_ 前缀；注意源码中它在净化改变了输入时会抛 ValueError（client.py），只有完全符合 [a-zA-Z0-9_-] 的 UUID 才能直接通过。

2. 查看图谱规模统计

poetry run python - <<'PY'
import asyncio
from backend.copilot.graphiti.client import derive_group_id
from backend.copilot.graphiti.config import graphiti_config
from backend.copilot.graphiti.falkordb_driver import AutoGPTFalkorDriver

USER_ID = "883cc9da-fe37-4863-839b-acba022bf3ef"
GROUP_ID = derive_group_id(USER_ID)

QUERIES = {
    "entities": "MATCH (n:Entity) RETURN count(n) AS count",
    "episodes": "MATCH (n:Episodic) RETURN count(n) AS count",
    "communities": "MATCH (n:Community) RETURN count(n) AS count",
    "relates_to_edges": "MATCH ()-[e:RELATES_TO]->() RETURN count(e) AS count",
}

async def run():
    driver = AutoGPTFalkorDriver(
        host=graphiti_config.falkordb_host,
        port=graphiti_config.falkordb_port,
        password=graphiti_config.falkordb_password or None,
        database=GROUP_ID,
    )
    try:
        for name, query in QUERIES.items():
            records, _, _ = await driver.execute_query(query)
            print(name, records[0]["count"])
    finally:
        await driver.close()

asyncio.run(run())
PY

四个查询分别统计实体节点、Episodic episode、社区节点与 RELATES_TO 事实边，对应图谱的四大组成部分。

3. 列出实体与关系名分布

poetry run python - <<'PY'
import asyncio
from backend.copilot.graphiti.client import derive_group_id
from backend.copilot.graphiti.config import graphiti_config
from backend.copilot.graphiti.falkordb_driver import AutoGPTFalkorDriver

USER_ID = "883cc9da-fe37-4863-839b-acba022bf3ef"
GROUP_ID = derive_group_id(USER_ID)

async def run():
    driver = AutoGPTFalkorDriver(
        host=graphiti_config.falkordb_host,
        port=graphiti_config.falkordb_port,
        password=graphiti_config.falkordb_password or None,
        database=GROUP_ID,
    )
    try:
        records, _, _ = await driver.execute_query(
            "MATCH (n:Entity) RETURN n.name AS name, n.summary AS summary ORDER BY n.name"
        )
        print("## entities")
        for row in records:
            print(row)

        records, _, _ = await driver.execute_query(
            """
            MATCH ()-[e:RELATES_TO]->()
            RETURN e.name AS relation, count(e) AS count
            ORDER BY count DESC, relation
            """
        )
        print("\n## relation_counts")
        for row in records:
            print(row)
    finally:
        await driver.close()

asyncio.run(run())
PY

第二段查询按关系类型聚合 RELATES_TO 边数，适合快速判断关系抽取是否失衡（比如 WORKS_AT 类关系被过度抽取）。

4. 检查某个节点周边的全部事实

poetry run python - <<'PY'
import asyncio
from backend.copilot.graphiti.client import derive_group_id
from backend.copilot.graphiti.config import graphiti_config
from backend.copilot.graphiti.falkordb_driver import AutoGPTFalkorDriver

USER_ID = "883cc9da-fe37-4863-839b-acba022bf3ef"
GROUP_ID = derive_group_id(USER_ID)
TARGET = "sarah"

async def run():
    driver = AutoGPTFalkorDriver(
        host=graphiti_config.falkordb_host,
        port=graphiti_config.falkordb_port,
        password=graphiti_config.falkordb_password or None,
        database=GROUP_ID,
    )
    try:
        records, _, _ = await driver.execute_query(
            """
            MATCH (a)-[e:RELATES_TO]->(b)
            WHERE (exists(a.name) AND toLower(a.name) = $target)
               OR (exists(b.name) AND toLower(b.name) = $target)
            RETURN a.name AS source, e.name AS relation, e.fact AS fact, b.name AS target
            ORDER BY e.created_at
            """,
            target=TARGET,
        )
        for row in records:
            print(row)
    finally:
        await driver.close()

asyncio.run(run())
PY

该查询同时匹配目标节点作为源端或目标端的有向事实，按创建时间排序，便于发现重复抽取与时间线上的事实演变。

5. 查看用户在 PostgreSQL 中的全部聊天消息

图谱之外，原始对话存于 Prisma 管理的 PostgreSQL。文档提供了一条联表查询（ChatMessage join ChatSession），按 createdAt, sequence 排序，截取前 260 字符的内容预览：

poetry run python - <<'PY'
import asyncio
from prisma import Prisma

USER_ID = "883cc9da-fe37-4863-839b-acba022bf3ef"

async def run():
    db = Prisma()
    await db.connect()
    try:
        rows = await db.query_raw(
            '''
            select cm."sessionId" as session_id,
                   cm.sequence,
                   cm.role,
                   left(cm.content, 260) as content,
                   cm."createdAt" as created_at
            from "ChatMessage" cm
            join "ChatSession" cs on cs.id = cm."sessionId"
            where cs."userId" = $1
            order by cm."createdAt", cm.sequence
            ''',
            USER_ID,
        )
        for row in rows:
            print(row)
    finally:
        await db.disconnect()

asyncio.run(run())
PY

对照图谱与原始消息，是验证“抽取是否忠实于用户原话”的标准动作。

图谱模型速记

文档 Notes 部分给出三条排查准则，理解它们能避免误判：

• RELATES_TO 边承载语义事实，检查时同时看 e.name（关系类型）与 e.fact（事实文本）；
• MENTIONS 边是 episode 到抽取节点的溯源（provenance），不是语义事实；
• 查重时优先使用有向查询 ->——无向匹配会把镜像边双重计数。

源码级机制剖析

每用户隔离与客户端缓存

get_graphiti_client(group_id)（client.py）为每个 group_id 构建独立 Graphiti 实例，理由有二：避免不同 group 并发访问时上游 self.driver 的变更竞态；以及 FalkorDB 连接内部 Future 绑定创建它的 event loop——CoPilot executor 每个 worker 线程跑一个独立 asyncio loop，进程级缓存会把 loop-1 绑定的连接交给 loop-2 上的任务，触发 “Future attached to a different loop” 错误。因此缓存以 weakref.WeakKeyDictionary 按 running loop 分区（loop 被回收后自动清理），并在缓存失效时通过 _EvictingTTLCache 重写 expire()/popitem() 顺带关闭 driver，防止连接泄漏。

定制 FalkorDB driver 的三处改动

AutoGPTFalkorDriver 继承 graphiti-core 的 FalkorDriver，做了三处 AutoGPT 专属调整：

1. 多租户全文检索过滤：build_fulltext_query() 追加 @group_id:... 过滤，防止多租户全文搜索跨用户图串数据（falkordb_driver.py）；
2. 可选跳过建索引后台任务：上游 driver 构造时会 fire-and-forget 地启动 build_indices_and_constraints；对短生命周期 driver（如管理端内存可视化每请求新开连接），这段顺序 CREATE INDEX 会与路由自身查询、连接关闭竞态，产生 “Connection closed by server” 日志噪音。只读路径可传 build_indices=False 跳过——索引已由长生命周期的聊天写入 client 建好；
3. 背压重试：FalkorDB 在共享服务端 pending-query 队列满时会以 “Max pending queries exceeded” 拒绝查询——这是执行前的拒绝，查询根本没跑过，因此重试无副作用。execute_query() 对该错误做有界抖动指数退避（base * 2^attempt + uniform(0, base)，单次等待封顶 2 秒），中间尝试保持静默，只有重试预算耗尽才通过上游 logger 上报，从而保住 Sentry 分组与良性清理过滤器；其余错误（Cypher 拼写错误、缺图、连接拆除）首次即失败。另外注意：终态错误日志故意不含查询参数，因为参数里可能携带用户记忆内容（姓名、事实），避免 PII 泄漏到 Sentry。

Warm Context：会话首轮的时序预载

fetch_warm_context()（context.py）在会话首轮调用，流程为：derive_group_id → 取缓存 client → 用 graphiti-core 的 EDGE_HYBRID_SEARCH_CROSS_ENCODER 配方（BM25 + 余弦 + BFS 边检索 + cross-encoder 重排，默认 limit=10 被覆盖为 context_max_facts）检索相关事实边，同时 retrieve_episodes 取最近 5 条 episode；二者通过 asyncio.gather 并发执行，整体受 context_timeout（默认 8 秒）约束。检索到的每个处于 status='tentative' 的边会被 fire-and-forget 地提升到 active 并累加命中计数（ratification 机制，见 dream/ratification.py）。最终 _format_context() 把结果拼成 <temporal_context> 块（内含 <FACTS> 与 <RECENT_EPISODES> 两节，事实带 valid_from — valid_to 时间窗），追加进系统提示词。任何超时或异常都只记 warning 并返回 None——这正是文档 Design Intent 中“失败优雅降级”的实现。

摄取的每用户串行化

ingest.py 头部注释说明了关键约束：graphiti-core 要求同一 group_id 内 add_episode() 顺序调用，因此本模块为每用户维护一个 asyncio.Queue，对调用方表现为 fire-and-forget。与客户端缓存同理，队列/worker 注册表也按 running loop 分区；空闲 worker 在 60 秒无活动后被清理；单条 episode body 硬上限 64KB（MAX_EPISODE_BODY_BYTES），超限直接拒绝而非截断，因为截断后的 JSON 封包下游必然解析失败。IngestionCompletion（ingest.py）则允许调用方只等待自己入队的那批 episode 落地，而不被同队列上其他活动（聊天与 dream-pass 写入共享队列）拖累。

小结

Graphiti 记忆模块是 AutoGPT CoPilot 的跨会话事实记忆层：以每用户 FalkorDB 数据库（user_<uuid>）隔离数据，写入经每用户串行队列摄取，读取在会话首轮做带 cross-encoder 重排的 warm-context 预载，社区级摘要由带活动门控的周重建维护。排查数据质量时，按 AGENTS.md 的手册从 group_id 推导、规模统计、实体/关系枚举、单节点事实、原始聊天消息五步走，即可覆盖绝大多数“记忆答非所问”问题的定位。相关源码入口：graphiti 包、prompting.py 的记忆提示词、.env.default 的 GRAPHITI 段 与 copilot-local-llm.md 的本地 LLM 文档。