飞书文档 → Memory DPE 开发模式指南

本文档梳理「连接飞书 → 获取文档列表 → 用户选择文档 → 同步至本地 → 学习至 Memory DPE」的完整开发模式，供后续新增类似远程文档源接入时参考。

全局数据流概览

┌──────────────────────────────────────────────────────────────────────────┐
│                          Phase 1: 连接与授权                              │
├──────────────────────────────────────────────────────────────────────────┤
│  前端 → GET /utils/feishu/authorization                                  │
│       → 飞书 OAuth 授权页                                                │
│       → GET /utils/feishu/callback                                       │
│       → Redis 存储 user_access_token / tenant_access_token              │
│       → Celery 定时刷新 Token                                            │
├──────────────────────────────────────────────────────────────────────────┤
│                          Phase 2: 浏览文档列表                            │
├──────────────────────────────────────────────────────────────────────────┤
│  前端 → GET /dashboard/doc-center/source/feishu/{drive|wiki}/list       │
│       → FeishuDrive/WikiManageService.list_remote_docs_and_folders()    │
│       → lark-oapi SDK 调用飞书 API                                       │
│       → [DocModel | FolderModel] → RemoteDriveListDto                   │
├──────────────────────────────────────────────────────────────────────────┤
│                          Phase 3: 同步文档内容                            │
├──────────────────────────────────────────────────────────────────────────┤
│  前端 → POST /dashboard/doc-center/source/feishu/{category}/sync        │
│       → acreate_sync_task() → Celery: memory_manage.add_document        │
│       → 派发至 TFRSUtilsCelery: feishu.sync_drive_obj / sync_wiki_obj  │
│       → 外部解析 → DPE dict 回调 → Celery: memory_manage.add_dpe        │
│       → Document.model_validate(dpe) → robot.brain.memory.add_doc(doc)  │
├──────────────────────────────────────────────────────────────────────────┤
│                          Phase 4: 学习至 Memory                          │
├──────────────────────────────────────────────────────────────────────────┤
│  前端 → PUT /factory/online/{setting_id}/action/learn_document           │
│       → OnlineFactoryService → Celery: memory_manage.learn_document     │
│       → DPELearnMixin.learn_document(session, doc_id)                   │
│       → DSLV2GraphIndex.learn() → OWL/Vector/Keyword 索引写入           │
│       → LearnResult 统计返回                                             │
└──────────────────────────────────────────────────────────────────────────┘

Phase 1: 飞书连接与授权

涉及文件

层级	文件	职责
API	`api/utils/feishu.py`	OAuth 授权/回调/解绑端点
Service	`services/doc_source_service/feishu_manage_service.py`	认证模型、Token 管理、Client 构建
Config	`services/doc_source_service/doc_source_config_service.py`	`FeiShuConfigModel` 持久化
DB	`db/models/doc_source_setting.py`	`FeiShuConfig` TypedDict
DTO	`dtos/v1/utils/feishu_dto.py`	`FeishuAuthDto`
DI	`dependencies/tfscontainer.py`	`feishu_manage_service` Provider

认证模型

# feishu_manage_service.py
class FeishuAuthModel:
    app_id: str
    app_secret: SecretStr
    redirect_uri: str = "/utils/feishu/callback"
    scope: list[str]  # 31+ 个权限，包括 docs/drive/wiki/contact/im

双 Token 体系

Token 类型	Redis Key 模式	获取方式	用途
`tenant_access_token`	`feishu:tenant_access_token`	App 凭证直接获取	企业级 API 调用
`user_access_token`	`feishu:{robot_id}:user_access_token`	OAuth 用户授权	用户维度文档访问

Token 刷新通过 Celery 定时任务 schedule_refresh_feishu_token 自动维护。

开发要点

Client 构建：get_client() 返回 lark.Client，启用 enable_set_token(True) 以支持动态 Token 切换
Token 传递模式：所有 API 调用通过 RequestOption 条件注入 user_access_token，无则回退 tenant_token
跨域安全：OAuth 回调返回 JSON {"location": "..."} 而非 302，由前端控制跳转

Phase 2: 文档列表浏览

涉及文件

层级	文件	职责
API	`api/v1/dashboard/doc_source.py`	`DocSourceView` CBV 端点
Protocol	`protocols/doc_manage_protocol.py`	`RemoteDocManageProtocol` 统一接口
Service (Wiki)	`services/doc_source_service/feishu_wiki_manage_service.py`	知识库目录树
Service (Drive)	`services/doc_source_service/feishu_drive_manage_service.py`	云文档目录树
URI	`protocols/uri_protocol.py`	`FeishuURI` 统一资源标识
DTO	`dtos/v1/dashboard/doc_center.py`	`RemoteDriveListDto`, `DocItem`, `FolderItem`

核心协议：RemoteDocManageProtocol

# protocols/doc_manage_protocol.py
class RemoteDocManageProtocol(Protocol):
    async def list_remote_docs_and_folders(
        self,
        parent_original_uri: Optional[URL] = None,
        page_token: Optional[str] = None,
        page_size: int = 10,
    ) -> tuple[Sequence[DocModel | FolderModel], Optional[str], bool]:
        """返回 (items, next_page_token, has_more)"""
        ...

    async def list_all_remote_docs_and_folders(
        self, parent_original_uri: Optional[URL] = None
    ) -> AsyncGenerator[DocModel | FolderModel, None]: ...

    async def get_remote_doc_meta(self, original_uri: str) -> dict: ...
    async def get_remote_folder_meta(self, original_uri: str) -> dict: ...

任何新的远程文档源只需实现此 Protocol 即可接入。

FeishuURI 统一资源标识

Wiki 模式:
  feishu://wiki/space/{space_id}                              → 知识库空间
  feishu://wiki/{doc_type}/{space_id}/{doc_id}                → 知识库文档

Drive 模式:
  feishu://drive/folder/{folder_token}                        → 云文档文件夹
  feishu://drive/folder/{folder_token}/{doc_type}/{doc_token} → 云文档文件

FeishuURI 解析出 mode, group_type, group_id, doc_type, original_id, category 等字段，贯穿整个流程。

API 端点调用链

GET /source/feishu/drive/list?parentOriginalUri=...&pageToken=...
  ↓
DocSourceView.get_feishu_file_list()
  ↓ match category
FeishuDriveManageService.list_remote_docs_and_folders(FeishuURI(...), page_token)
  ↓ lark-oapi SDK
client.drive.v1.file.list(ListFileRequest(folder_token=...))
  ↓
[DocModel | FolderModel] → .to_dto() → RemoteDriveListDto
  ↓
TFSResponse[RemoteDriveListDto] → 前端

数据转换层

飞书 API Response
  ↓ Service 层转换
DocModel / FolderModel  (ServiceModel — 内部流转)
  ↓ .to_dto()
DocItem / FolderItem    (DTO — API 响应)
  ↓ .from_orm()
DocMetadata / Folder    (ORM — 数据库存储)

Phase 3: 文档同步（飞书 → DPE 本地存储）

涉及文件

层级	文件	职责
API	`api/v1/dashboard/doc_source.py`	`sync_feishu_file` 端点
Service	`services/doc_source_service/doc_source_manage_service.py`	`RedisBaseDocSourceManageService` 基类
Protocol	`protocols/sync_task_manage_protocol.py`	`SyncTaskManageProtocol`
Celery Task	`protocols/sync_task_manage_celery.py`	`add_document_to_robot`, `add_dpe_to_robot`
Redis Model	`db/redis_models/sync_task.py`	`SyncTaskMetaARedisModel`
DB Model	`db/models/task.py`	`TFSTask`

同步任务创建流程

POST /source/feishu/{category}/sync  body: { uri: "feishu://wiki/docx/..." }
  ↓
FeishuDriveManageService.acreate_sync_task(target_uri, DocSource.FEISHU, cascade)
  ↓ 基类 RedisBaseDocSourceManageService.acreate_sync_task()
  ├─ 检查是否已有运行中任务（幂等保护）
  ├─ _create_sync_task() → Celery send_task("memory_manage.add_document")
  ├─ SyncTaskMetaARedisModel 保存 Redis 元数据
  └─ cascade=True 时递归创建子任务
  ↓
返回 taskId: UUID4

Celery 任务链

memory_manage.add_document (TFRobotServer memory_tasks 队列)
  ↓ 解析 FeishuURI，按 mode 分发
  ├─ drive/doc → send_task("feishu.sync_drive_obj", link=add_dpe_to_robot)
  └─ wiki/doc  → send_task("feishu.sync_wiki_obj", link=add_dpe_to_robot)
         ↓ TFRSUtilsCelery 执行文档解析
         ↓ 返回 DPE dict
memory_manage.add_dpe (回调，memory_tasks 队列)
  ↓
  ├─ Document.model_validate(dpe)        # 校验 DPE 结构
  ├─ _complete_image_paths_in_document() # 补全图片绝对路径 + 安全校验
  └─ robot.brain.memory.add_doc(doc)     # 写入 Memory 存储

关键设计决策

Celery 链式回调：send_task(..., link=add_dpe_to_robot.s(robot_id=...).set(queue="memory_tasks")) — 上游任务返回值作为下游第一个位置参数
图片路径安全校验：相对路径必须以 {robot_id}/ 开头，防止跨 Robot 读取
COS 统一化：COS 文档通过 presigned URL 转换为 url.sync_file_obj 任务，复用 Web 同步通道
级联同步：文件夹同步时 cascade=True，通过 list_target_uri() 递归创建子任务

Phase 4: 文档学习（DPE → Memory 索引）

涉及文件

层级	文件	职责
API	`api/v1/robot_factory/online_setting.py`	`trigger_action` 端点
Service	`services/robot_factory_service/online_factory_service.py`	Action 路由与 Celery 派发
Constants	`services/robot_factory_service/utils.py`	`CELERY_ACTIONS` 定义
Celery Task	`protocols/learn_task_celery.py`	`learn_document` Task
Mixin	`robot/factory/brain/memory/dpe_learn_mixin.py`	`DPELearnMixin` 学习逻辑
Factory	`robot/factory/base.py`	`trigger_tfs_action()` Action 分派

Action 触发链

PUT /factory/online/{setting_id}/action/learn_document  body: { doc_id: 123 }
  ↓
OnlineFactoryService.atrigger_online_action(setting_id, "learn_document", {doc_id: 123})
  ↓ action in CELERY_ACTIONS?
  ↓ YES
acreate_learn_task(setting_id, doc_id=123)
  ↓
Celery send_task("memory_manage.learn_document",
                  kwargs={setting_id, action, params},
                  queue="memory_tasks")
  ↓
TFSTask 入库（category=LEARN, meta={setting_id, doc_id}）
  ↓
返回 taskId: UUID4 → 前端轮询状态

Celery Worker 执行

# learn_task_celery.py
@memory_manage_celery.task(bind=True, name="memory_manage.learn_document")
def learn_document(self: Task, setting_id, action, params):
    # Phase 1: PROGRESS → building
    self.update_state(state="PROGRESS", meta={"phase": "building"})

    # Phase 2: 从 ORM 还原 OnlineSetting
    orm = repo.get_one(setting_id)
    factory = get_factory(orm.scene, orm.factory)
    online_setting = factory.from_orm(orm)

    # Phase 3: PROGRESS → learning
    self.update_state(state="PROGRESS", meta={"phase": "learning"})

    # Phase 4: 触发 TFSAction
    result = online_setting.trigger_tfs_action(action, session, **params)

    # Phase 5: 返回 LearnResult 摘要（duck-type 按字段存在性提取，OntoLearnResult 同形兼容）
    return {setting_id, action, duration_seconds, total_elements, succeeded, failed, ...}

    # Phase 6（TFRS-357）: 中断契约信号（TFROB-694）—— 用户中断 / drain 取消在途 learn
    # 时，内核检查点抛 TFUserInterruptError（已处理元素落库 + 断点可续学）。任务捕获后：
    #   update_state(state="INTERRUPTED", meta=partial_summary)  → 状态机 PROGRESS→INTERRUPTED
    #   raise Ignore()  → Celery 不覆写已存 state、仅 ACK（不重投、不重试、不计失败）
    # partial_summary 含已完成索引的聚合统计（mixin 挂到 TFUserInterruptError.partial_result）；
    # INTERRUPTED 为自定义终态，任务可重新提交（TERMINAL_STATES 含 INTERRUPTED）断点续学。

DPELearnMixin 学习过程

# dpe_learn_mixin.py
async def alearn_document(self, async_session, doc_id):
    memory = await self.abuild(async_session, db_container=default_db_container)
    try:
        for index in memory.unified_store.indexes:
            if isinstance(index, DSLV2GraphIndex):
                doc = memory.get_doc(doc_id)
                elements = [await async_session.merge(ele.to_orm()) for ele in doc.elements]
                result = await index.alearn(async_session, elements=elements)
                # 聚合结果到 LearnResult
    finally:
        await safe_adispose_memory(memory)

Action 三级路由

OnlineFactoryService.atrigger_online_action() 按 action 类型分三条路径：

路由	匹配条件	执行方式	示例
ROBOT_WORKER	`action in ROBOT_WORKER_ACTIONS`	同步等待 robot_tasks 队列	`current_tools`, `retrieve_tools`
CELERY	`action in CELERY_ACTIONS`	异步 fire-and-forget + 轮询	`learn_document`, `learn_all`
General	其它	直接调用 `trigger_tfs_action`	自定义 Action

开发模式总结：新增远程文档源的 Checklist

当需要接入一个新的远程文档源（如 Notion、Google Drive、Confluence 等），按以下步骤进行：

Step 1: URI 协议扩展

[ ] 在 protocols/uri_protocol.py 新增 URI 类（如 NotionURI）
[ ] 定义 URI scheme 和解析规则

Step 2: Service 层实现

[ ] 新建 services/doc_source_service/{source}_manage_service.py
[ ] 继承 RedisBaseDocSourceManageService
[ ] 实现 RemoteDocManageProtocol 的四个方法：
list_remote_docs_and_folders()
list_all_remote_docs_and_folders()
get_remote_doc_meta()
get_remote_folder_meta()
[ ] 实现 _create_sync_task() 抽象方法
[ ] 实现 list_target_uri() 抽象方法（支持级联同步）

Step 3: 认证与 Token 管理

[ ] 实现 OAuth 或 API Key 认证流程
[ ] Redis Token 存储与自动刷新（参考 FeishuAuthModel 模式）
[ ] 如需 OAuth：新增 /utils/{source}/authorization 和 /callback 端点

Step 4: 配置持久化

[ ] 在 db/models/doc_source_setting.py 新增配置 TypedDict
[ ] 在 doc_source_config_service.py 新增 ConfigModel
[ ] 新增对应 DTO（dtos/v1/dashboard/doc_center.py）

Step 5: DI 注册

[ ] 在 dependencies/tfscontainer.py 注册新的 Service Provider
[ ] 构造参数统一为 (redis, db, celery_app)

Step 6: API 端点

[ ] 在 api/v1/dashboard/doc_source.py 的 DocSourceView 中新增列表/同步端点
[ ] 或在 DocSourceView.__async_init__ 中注入新 Service

Step 7: Celery 同步任务

[ ] 在 protocols/sync_task_manage_celery.py 的 add_document_to_robot 中新增 case DocSource.XXX
[ ] 对应的文档解析任务在 TFRSUtilsCelery 中实现
[ ] 通过 link=add_dpe_to_robot.s(...) 回调写入 Memory

Step 8: 无需修改的部分

以下模块是通用的，新增文档源时不需要修改：

add_dpe_to_robot — DPE → Memory 写入是通用的
DPELearnMixin — 学习逻辑与文档源无关
learn_task_celery.py — 学习任务调度是通用的
Factory Action 路由 — CELERY_ACTIONS 已覆盖
RemoteDriveListDto — 列表 DTO 是通用的

关键架构约束

单进程文件系统：TFRobot 配置依赖文件系统，服务以单线程模式运行
Celery 线程池：TFRobotServer Celery worker 使用 --pool=threads（非 prefork）
跨进程解析：文档解析由 TFRSUtilsCelery 独立进程完成，通过 Celery 链式回调返回 DPE
DPE 第一参数约束：add_dpe_to_robot(self, dpe, robot_id) 中 dpe 必须是第一个参数，因为 Celery 链式回调将上游返回值作为下游第一个位置参数
幂等同步：acreate_sync_task 检查已有运行中任务，避免重复同步
Memory 生命周期：DPELearnMixin 在 finally 中必须调用 safe_adispose_memory() 释放资源