OpenHuman 自动抓取机制解析周期性集成同步如何把数据持续折叠进 Memory Tree【免费下载链接】openhumanOpenHuman is an open source personal AI for Mac, Windows and Linux — local-first memory, agent orchestration, and deep research.项目地址: https://gitcode.com/GitHub_Trending/op/openhumanOpenHuman 的集成层不是你问我才查的被动设计而是一个宿主侧周期性调度器它按固定节律轮询每个活跃连接Gmail、Notion、GitHub 等在满足该工具自身的最小同步间隔后调用原生同步逻辑把新数据规范化后写入 Memory Tree。读完本文你将理解这套 auto-fetch 的完整工作流程、每个连接独立维护的同步状态最后同步时间、日预算、去重集合、游标、周期与预算两类调优手段以及当前仓库中该调度器的 Rust 实现细节。核心理念从被动应答到主动拉取大多数AI 助手是反应式的你提问、它思考、它回答。OpenHuman 反其道而行——它持续从你的技术栈中拉取数据以至于当你问出我昨晚邮箱里来了什么时答案已经在 Memory Tree见 Memory Tree 文档里了。这个结论不是靠回答时现场查 API 得到的而是靠回答前数据就已经被周期性同步进来得到的。auto-fetch 正是 第三方集成层 之上的自动化能力连接建好后不需要你写任何提示词或轮询脚本数据就会自己流动。工作原理一个全局 tick逐连接判定整个 auto-fetch 由一个单一周期调度器驱动。每个 tick 到来时它会遍历每个活跃的集成连接查找对应的原生 provider如果距离该连接上次同步的时间已经足够长就调用provider.sync(ctx, SyncReason::Periodic)every 20 min | v for each active connection (Gmail, Notion, GitHub, ...) | -- check sync_state (toolkit, connection_id) | - last sync timestamp | - daily budget | - dedup set | - cursor | -- if interval elapsed - provider.sync() | -- on success - record_sync_success(ts)有四个设计要点值得注意一个全局 tick而不是一连接一个任务。每个用户的连接数量很少单个 20 分钟粒度的 tick 就足够了也让记账逻辑保持极简。状态以(toolkit, connection_id)为键。每个连接拥有自己独立的游标cursor、最后同步时间戳、去重集合和每日预算。重启后这些状态从本地 KV 重建漏掉的一次周期性同步是无害的因为重启后的下一个 tick 会把它补回来。原生同步与事件驱动路径共享状态。当 webhook 或on_connection_created事件触发一次非周期性的原生同步时它会写入同一份 sync_state因此调度器不会重复触发同一连接。错误只记录、被吞掉。调度器绝不允许自己 panic 出循环否则周期性同步会在整个进程生命周期里静默停摆。源码纵深调度器在当前仓库中的实现官方文档描述的是设计意图而 调度器实现 展示了这些设计约束在 Rust 中是如何落地的。全局循环与幂等启动start_periodic_sync()是一个刻意设计为廉价的入口用OnceLock守卫防止重复 spawn 第二个循环——这在 渠道启动路径 中体现得很清楚start_channels与bootstrap_core_runtime两条启动路径都可能调用它后到的调用直接 no-op 返回。循环本体很简单tokio::spawn(async move { loop { tokio::time::sleep(TICK_INTERVAL).await; if let Err(error) run_one_tick().await { log::warn!([composio:periodic] tick failed: {error}); } } });这里印证了文档中错误被记录并吞掉的约束单次 tick 失败只会打一条warn日志循环继续存活。从源码结构看有一个值得指出的文档与实现之间的演进差异periodic.rs中的模块注释明确记载早期引擎tinymemory 进程内管线v1.13.4 已删除曾在 tick 中列出所有活跃连接、跳过刚同步过的 toolkit/连接、对剩余部分执行 readingest并且跨重启持久化了每连接的游标与成本统计而当前宿主侧实现把该不该同步的判定放在一个进程内的last_synced()映射表中进程重启会忘记每个连接上次运行的时间从而在下一个 tick 直接补同步——这与文档中错过的一次同步无害重启后下个 tick 捡起来的表述是一致的行为结果只是记账介质从持久化游标变成了进程内状态。逐连接判定与最小同步间隔每个 tick 的核心逻辑是先取配置、列出所有活跃连接然后对每个连接做两步筛选该 toolkit 是否声明了周期性间隔——通过 catalog 查询native_provider_sync_interval_secs(toolkit)返回None表示该 toolkit 主动退出了自动同步对应文档中低流量工具可以不同步的思路。这个查询来自合同 crate 的静态 catalog 表见 provider 表面 中 re-export 的tinymemory_api::composio::catalogs即每个原生 provider 声明自己的sync_interval_secs高流量工具如 Gmail的间隔可以比低流量工具如 Stripe更短。距上次同步是否已到达间隔——查last_synced映射last.elapsed() interval才执行同步。共享同步状态为什么事件驱动与周期驱动不打架文档强调webhook 或on_connection_created触发的非周期同步会打上同一份 sync_state调度器不会重复触发。这在源码中有直接对应record_sync_success(toolkit, connection_id)是公开函数在 composio 模块入口 中 re-export周期性循环自己的成功会调用它而 事件总线驱动的同步路径 在 trigger 驱动同步成功后也调用同一个函数把当前时间写进同一张last_synced表。两个入口、一份状态去重自然成立。还有一个源码中额外可见的细节即使某次同步失败循环也会记录这次尝试——注释解释得很直白一个持续失败连接如果在每个 tick 都被重试会把一个坏账号变成对模块的热循环。这与文档错误被吞掉的原则是同一枚硬币的两面既保证循环不死也保证资源不被单个坏连接耗尽。日预算守门文档中 sync_state 里的 daily budget 字段在当前实现里由 run_sync_within_budget 承担tick 判定连接到期后实际同步走run_sync_within_budget(config, toolkit, conn.id, periodic)在预算范围内执行tinyconnectors→MemorySourceSink::accept_source_items的同一条读取摄取路径成功日志会带上records_read/written/more_pending三个计数debug 级别这对应文档Tuning and visibility一节中同步活动在核心日志的 debug 级别可见的说明。数据最终落在哪里每个 provider 自己塑形每个 provider 负责自己 ingestion 的形状。以 Gmail provider 为例拉取一页新消息 → 跑邮件 canonicalizer → 把结果送进与手动 UI 相同的ingest路径 → 分块落进 SQLite、摘要桶填充、被触及实体的主题树被标记为 dirty。其他 providerGitHub、Slack、Notion……遵循同样的形状自游标起抓取新条目 → 规范化 → 摄入 Memory Tree。这意味着 auto-fetch 产出的数据与手动触发同步、或用户在 UI 里操作产生的数据走完全一致的管道检索与压缩逻辑如 Smart Token Compression 让什么都抓保持低成本无需为数据来源做任何特判。为什么是 20 分钟 tick文档给出了明确的历史依据最初设计是 60 秒。多个 provider 同时连接时这意味着持续的 HTTP 抓取与数据库写入在笔记本上肉眼可见地忙碌。改为 20 分钟是用一点点新鲜度换取明显更低的前台负载。而每个 provider 声明的sync_interval_secs仍然封顶了两次实际同步之间的最小延迟全局 tick 只是放松了上限见 auto-fetch 原文档。从源码结构看这一分层在当前实现中对应为全局循环的唤醒间隔TICK_INTERVAL只是轮询粒度真正决定同步节奏的是 catalog 中每个 toolkit 自己的间隔值——注释原话是它与任何单个 toolkit 的自身间隔无关这只是轮询粒度。文档中提到的相关配置面还包括仓库同时暴露了sync_interval_secs配置项见 agent 配置 ops可写入Config::memory_sync_interval_secs并在每次 tick 时被调度器重新读取允许在运行期调整同步节奏。调优与可见性汇总把文档的调优清单与仓库实现对照得到一张可操作的面板手段文档描述仓库中的对应实现每 provider 间隔每个原生 provider 声明自己的sync_interval_secs高流量工具同步更频繁native_provider_sync_interval_secscatalog 静态表None即退出自动同步每日预算每个连接有每日请求预算控制 API 成本与限流pass_budget.rs 中的run_sync_within_budget日志同步活动记录在核心日志的 debug 级别循环内log::debug!输出records_read/written/more_pending状态可见性sync_state 按(toolkit, connection_id)维护进程内last_synced映射 record_sync_success公开写入设计取舍小结这套 auto-fetch 的取舍可以归结为三点全局单 tick 换取极简记账按连接粒度隔离游标与预算换取故障隔离错误吞掉并记录换取调度器永不死亡。代价是新鲜度的上限被 20 分钟 tick 放松由 provider 间隔兜住下限以及当前实现中同步记账不跨重启持久化——但正如文档所述这两点都在设计上被判定为无害或可接受。延伸阅读第三方集成auto-fetch 所运行的连接器层。Memory Tree所有被折叠进的记忆树的最终归宿。Smart Token Compression让什么都抓保持便宜的成本机制。auto-fetch 原文档 与 周期同步实现本文所依据的文档与源码。【免费下载链接】openhumanOpenHuman is an open source personal AI for Mac, Windows and Linux — local-first memory, agent orchestration, and deep research.项目地址: https://gitcode.com/GitHub_Trending/op/openhuman创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考