OpenHuman Memory Sources 深度解析：连接器注册表、Reader 抽象与按 Agent 作用域隔离

OpenHuman Memory Sources 深度解析连接器注册表、Reader 抽象与按 Agent 作用域隔离【免费下载链接】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导读本文围绕 OpenHuman 的memory_sources领域展开它负责回答什么在喂我的记忆以类型化注册表的形式把本地文件夹、GitHub 仓库、RSS 订阅、网页以及 Composio OAuth 集成统一管理起来持久化于config.toml的[[memory_sources]]表并提供运行时 CRUD、统一 Reader 抽象、按源同步状态以及openhuman.memory_sources_*RPC 面。读完本文你将掌握七种 Source 类型及其字段要求、RPC 增删改查与手动同步的调用方式、SourceReader的底层实现与分发机制、同步新鲜度Freshness与进度事件的判定规则以及按 Agent 档案作用域隔离这一隐私与聚焦控制的核心原理。一、定位Memory Sources 在 OpenHuman 记忆体系中的角色在 OpenHuman 中Memory Tree 负责怎么存储与摘要而memory_sources领域src/openhuman/memory/sources/负责上游问题——什么在喂我的记忆。二者分工如下Memory Sources只定义连接器并从中读取数据摄取引擎与同步调度位于memory/memory_syncSources 把工作分派给正确的后端本身不直接实现摄取管线。从源码结构看该领域在 mod.rs 中被拆分为四个职责清晰的模块readers统一读取抽象、registry注册表读写与写锁、schemasRPC 请求/响应结构与控制器注册、rpcJSON-RPC 表面外加status同步状态、sync手动同步与reconcileComposio 对账三个支撑模块。需要特别注意的是自 #5560 重构起该模块不再以 glob 方式依赖内存引擎 crate而是直接依赖引擎无关的tinymemory-sources类型 crateSourceKind、MemorySourceEntry等即来自types这消除了宿主程序对内存引擎的编译期耦合。二、七种 Source 类型SourceKind判别器与字段要求每一个 Source 都是一个扁平的MemorySourceEntry其类型定义位于tinymemory-sourcescrate 并由 mod.rs 重新导出。kind字段SourceKind枚举决定哪些字段是必填的——校验在 add/update 时由validate()强制执行而非依赖 Rust 类型系统。类型SourceKind摄取内容ComposioComposio已 OAuth 连接的 SaaS 集成Gmail、Slack、Notion 等同步由提供商驱动ConversationConversationAgent 自身的对话转录文本FolderFolder本地目录按 glob 匹配默认**/*.md单文件上限 10 MB带路径穿越防护GitHub repoGithubRepo项目活动commits、issues、PRs经ghCLI 或公开 REST 兜底RSS feedRssFeedRSS/Atom 订阅条目Web pageWebPage抓取的网页可用 CSSselector收窄Twitter queryTwitterQuery已保存的 Twitter 查询Reader 已搭建同步有意未实现等待凭据每个条目还可携带可选的全同步预算字段——max_tokens_per_sync、max_cost_per_sync_usd、sync_depth_days——防止话痨型源在一次运行中烧穿 token 预算。这与仓库中的预算测试如rpc_budget_tests相互印证。Folder 的路径安全与默认值FolderReader 是唯一直接与本地文件系统打交道的 Reader其安全语义值得单独说明默认 glob 为**/*.md单文件大小上限 10 MB内置路径穿越防护防止 glob 结果逃逸出配置的根目录该 Reader 是tinymemory_sources::readers::folder::FolderReader的产品态适配器把 OpenHuman 的Config转换为底层 crate 需要的workspace_dir路径。三、增删改查openhuman.memory_sources_*RPC 表面Sources 通过memory_sources控制器schemas.rs→rpc.rs完成 CRUD命名空间为openhuman.memory_sources_*RPC用途list列出已配置的源会先惰性对账 Composioget按id获取单个源add新增源类型专属字段平铺在请求上update通过MemorySourcePatch做部分更新remove按id删除源list_items经其 Reader 列出某源的可读条目read_item读取单个条目的内容sync排队手动同步立即返回进度经事件推送status_list每个源的同步状态所有变更操作都会重载活动Config、应用修改再通过config.save()原子写盘见registry.rs。从 registry.rs 可以看到一个关键工程细节注册表整体重写[[memory_sources]]表因此并发写者会各自读-改-写并互相丢更新——为此引入了一个进程级tokio::sync::Mutexmemory_sources_write_guard来串行化对注册表文件的写入。registry.rs还提供list_sources/get_source/list_enabled_by_kind等读取入口并区分两类变体默认变体如list_sources经config::rpc::load_config_with_timeout从进程环境解析配置适合 RPC 处理器服务当前活跃用户_in变体如list_sources_in、get_source_in接收显式Config供绑定到特定工作区的调用方使用避免工作区 B 的调用方读到工作区 A 的源这一跨工作区泄漏这正是 workspace-keyed 内存绑定的存在意义。在桌面应用中这些操作呈现在 Intelligence / Memory 标签页与 Auto-fetch 的 20 分钟抓取节奏并列。四、统一读取抽象SourceReadertrait 与reader_for分发每种 Source 类型实现同一个异步 traitSourceReadersrc/openhuman/memory/sources/readers/mod.rs#[async_trait] pub trait SourceReader: Send Sync { fn kind(self) - SourceKind; async fn list_items(self, source, config) - ResultVecSourceItem, String; async fn read_item(self, source, item_id, config) - ResultSourceContent, String; }reader_for(kind)分发器根据SourceKind返回对应实现readers/mod.rsSourceKind::Composio→ComposioReaderSourceKind::Conversation→ConversationReaderSourceKind::Folder→FolderReaderSourceKind::GithubRepo→GithubReaderSourceKind::TwitterQuery→TwitterReaderSourceKind::RssFeed→RssReader::new()SourceKind::WebPage→WebPageReader手动sync时Reader 型源会遍历list_items把每个条目经memory::ingest_pipeline::ingest_document摄取sync.rsComposio 源则整体委托给memory_sync::composio::run_connection_sync不做逐条读取因此ComposioReader::read_item只是一个说明性占位实现。网络 Reader 的分发决策重要安全语义模块文档readers/mod.rs明确指出底层引擎无关 crate 的reader_for故意不派发网络型 Reader——网络读取必须在调用方已明确决定允许这次抓取时显式构造这正是宿主程序掌控出站流量、OAuth 与成本预算的方式。本领域的reader_for之所以发放全部七种是因为调用方是响应显式用户请求的 RPC 处理器而非定时器。切勿从轮询循环中复用它若轮询循环需要 Reader应通过 crate 分发获取本地型 Reader并刻意构造网络型 Reader让该决策保持可见。五、同步状态与新鲜度status.rs为每个源计算SourceStatus查询mem_tree_chunks已同步/待处理块数、最近块时间戳使用source_id LIKE前缀——Reader 型源用mem_src:{id}:%Composio 用{toolkit}:%。每个源得到一枚FreshnessLabelstatus.rsActive最近块 ≤ 30 秒前Recent最近块 ≤ 5 分钟前Idle更旧或尚无块。FreshnessLabel::from_age_ms使用saturating_sub而非普通减法未来时间戳远端时钟偏移的导入条目会得到非正年龄而非溢出并读取为Active。同步进度以MemorySyncStageChanged事件流式推送Requested → Fetching → Stored → Ingesting → Completed/Failed标记为connection_id Some(source.id)因此 UI 无需轮询即可展示实时进度。status_list对单个源的查询失败会降级为一条Idle零行记录而不是让整个调用失败。SourceStatus还额外携带sync_stage与sync_detail两个在途字段来自宿主对阶段流的记忆见 openhuman#6019仅凭块计数无法表达连接器仍在分页、尚未写入任何内容的状态缺了它Sources 面板在运行中途挂载时会误把同步中的源显示为空闲。Composio 自动 upsert当 OAuth 连接创建时memory_sync::composio::bus调用upsert_composio_source使新连接的集成无需重启即成为 SourcelistRPC 还会在每次列出时执行惰性对账reconcile::ensure_composio_sources捕获该钩子存在之前就已建立的连接。六、按 Agent 档案的作用域隔离Source Scoping默认情况下Agent 从所有源回忆。Source Scoping 允许 Agent 档案把回忆限制在一个源 id 白名单内——客服风格的 Agent 永远看不到你的私人 Gmail研究风格的 Agent 则只聚焦于相关的仓库与订阅。这是一项隐私与聚焦控制而不仅仅是相关性微调。机制位于src/openhuman/memory/source_scope.rs。由于把 allowlist 贯穿每个记忆工具与深层select_trees检索层会触及数十个调用点实现借鉴了thread_context信道在 Agent 回合周围设置一个tokio::task_local!检索层环境式读取它无需显式管线source_scope.rs定义了with_source_scope另有current_source_scope读取当前值None在任何作用域之外或with_source_scope(None, …)无限制。这是 cron、子 Agent、CLI 以及任何未设置memory_sources的档案的默认行为。Some(set)把回忆限制在集合中的源作用域内。空集合意味着不暴露任何内容该档案未选择任何源。门控是标签判别 失败开放的门控函数chunk_source_allowedsource_scope.rs只触碰带memory_sources标签的块不带memory_sources标签的块工作记忆、对话转录、内部块始终通过——即使在空 allowlist 下带标签的 memory-source 块只有在其源 id 被允许时才通过。id 会匹配原始source_idComposio / 信道作用域如slack:#eng或从mem_src:id:item复合键提取出的注册表 idReader 型源。因此收紧档案的作用域只会隐藏其已连接的源绝不会让其丢失自身的对话上下文。七、配置示例与使用建议以下是一个贴合仓库实际结构的config.toml片段演示[[memory_sources]]表的形态字段以文档与源码中出现的名称为准具体必填字段由validate()按 kind 强制# 本地文件夹源默认 glob **/*.md单文件上限 10MB [[memory_sources]] id mem-src-docs kind Folder path ~/notes glob **/*.md max_tokens_per_sync 20000 # RSS 订阅源 [[memory_sources]] id mem-src-blog kind RssFeed url https://example.com/feed.xml # 网页源可用 CSS selector 收窄 [[memory_sources]] id mem-src-docs-page kind WebPage url https://docs.example.com selector main article注意以上字段为便于理解给出的示意形态实际必填字段以validate()与MemorySourceEntry定义为准正式使用时请通过openhuman.memory_sources_addRPC 添加由校验逻辑保证字段完整性。使用建议预算先行为高音量源RSS、GitHub、大型文件夹设置max_tokens_per_sync/max_cost_per_sync_usd防止单次同步烧穿成本善用 Scoping为每个 Agent 档案显式配置memory_sources白名单把隐私控制前置到档案配置而非事后补救依赖事件而非轮询手动sync立即返回进度通过MemorySyncStageChanged事件获取status_list只在需要快照时调用理解默认值未设置memory_sources的档案cron、子 Agent、CLI默认无限制回忆配置了空集合的档案则什么都不暴露。八、延伸阅读Auto-fetch让活跃源保持新鲜的 20 分钟节奏Memory Trees所有源汇入的管线Obsidian Wiki源落地所在的 Markdown 保险库IntegrationsComposio 源背后的 OAuth 提供商连接。【免费下载链接】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），仅供参考