QzAgent智能体平台的飞书通道技术实现详解

一、架构概览

智能体平台 采用 WebSocket 长连接接收 + HTTP API 发送 的双通道架构，与飞书（Feishu/Lark）平台进行双向通信。

核心设计特点

二、用户识别与会话隔离

2.1 用户身份标识（open_id）

飞书为每个用户分配全局唯一的 open_id，系统通过它识别不同用户：

sender_id_obj = getattr(sender, "sender_id", None)
sender_id = str(getattr(sender_id_obj, "open_id", "")).strip()

open_id 格式：ou_9ba5cd58e37b46f9a59204ac0f578ccb

2.2 会话ID生成（session_id）

def resolve_session_id(self, sender_id: str, channel_meta: Dict) -> str:
    chat_id = (meta.get("feishu_chat_id") or "").strip()
    chat_type = (meta.get("feishu_chat_type") or "p2p").strip()
    
    if chat_type == "group" and chat_id:
        # 群聊：app_id后4位 + chat_id后8位
        app_suffix = self.app_id[-4:]
        return f"{app_suffix}_{short_session_id_from_full_id(chat_id)}"
    
    if sender_id:
        # 单聊：sender_id(open_id)后8位
        return short_session_id_from_full_id(sender_id)

三、消息处理完整流程

3.1 消息接收与去重

async def _on_message(self, data: "P2ImMessageReceiveV1") -> None:
    # 1. 消息去重（防止飞书重试导致重复处理）
    message_id = getattr(message, "message_id", None) or ""
    if message_id in self._processed_message_ids:
        return  # 已处理，丢弃
    self._processed_message_ids[message_id] = None
    
    # 2. 提取用户身份
    sender_id = str(getattr(sender_id_obj, "open_id", "")).strip()
    
    # 3. 构建请求
    request = self.build_agent_request_from_native(native_payload)

3.2 请求构建（build_agent_request_from_native）

def build_agent_request_from_native(self, native_payload: Any) -> "AgentRequest":
    # session_id 优先级：传入值 > 自动生成
    session_id = payload.get("session_id") or self.resolve_session_id(
        sender_id, meta
    )
    
    user_id = (
        meta.get("feishu_sender_id") 
        or payload.get("user_id") 
        or sender_id
    )
    
    return AgentRequest(
        channel_id=channel_id,
        sender_id=user_id,
        session_id=session_id,      # 会话隔离ID
        content_parts=content_parts,
        channel_meta=meta,
    )

关键设计：session_id 支持外部传入，便于 Cron 等场景复用会话上下文。

3.3 接收ID路由映射

系统通过 _receive_id_store 维护 session_id → receive_id 的映射，确保响应能准确回传：

async def _save_receive_id(self, session_id: str, receive_id: str, 
                            receive_id_type: str):
    # 核心映射：session_id -> (receive_id_type, receive_id)
    self._receive_id_store[session_id] = (receive_id_type, receive_id)
    
    # 同时按 open_id 索引，方便 Cron 主动发送
    if receive_id_type == "open_id":
        self._receive_id_store[receive_id] = (receive_id_type, receive_id)
    
    # 持久化到磁盘（重启不丢失）
    self._save_receive_id_store_to_disk()

存储结构：

{
    "0f578ccb": ("open_id", "ou_9ba5..."),      # 单聊
    "app1_abc123": ("chat_id", "oc_group..."),  # 群聊
    "ou_9ba5...": ("open_id", "ou_9ba5..."),   # open_id索引
}

四、响应回传机制

4.1 目标地址生成

def to_handle_from_target(self, *, user_id: str, session_id: str) -> str:
    if session_id:
        return f"feishu:sw:{session_id}"      # 优先用 session_id
    return f"feishu:open_id:{user_id}"        # fallback

4.2 响应路由解析

async def _get_receive_for_send(self, to_handle: str, meta: Dict):
    route = self._route_from_handle(to_handle)
    session_key = route.get("session_key")
    
    # 1. 从内存/磁盘查表
    if session_key:
        recv = await self._load_receive_id(session_key)
        if recv:
            return recv  # (receive_id_type, receive_id)
    
    # 2. fallback 到 open_id 直发
    # ...

4.3 消息发送

async def send(self, to_handle: str, text: str, meta=None):
    recv = await self._get_receive_for_send(to_handle, meta)
    receive_id_type, receive_id = recv
    
    # 调用飞书 API 发送
    await self._send_text(receive_id_type, receive_id, body)

五、高级特性

5.1 话题（Thread）模式

当消息在话题中时，系统将 user_id 覆盖为话题ID，实现话题内共享上下文：

if thread_id:
    thread_uid = f"thread:{short_session_id_from_full_id(thread_id)}"
    native["user_id"] = thread_uid
    meta["feishu_sender_id"] = thread_uid

效果：

• 话题内所有消息共享同一个 user_id（thread:xxx）
• 不同话题之间上下文完全隔离
• 话题不支持流式卡片（飞书 API 限制）

5.2 交互式卡片（Card）

架构：

• Dispatcher：路由分发（message_type / action_type）
• Context：上下文提取（session_ctx）
• Handler：业务处理（tool_guard 等）

用户追踪关键：卡片按钮的 value 中嵌入 session_ctx，确保点击后能正确路由：

{
    "type": "tool_guard_approval",
    "action": "approve",
    "request_id": "req_123",
    "session_ctx": {
        "sender_id": "ou_xxx",
        "session_id": "abc123",
        "receive_id": "oc_group...",
        "receive_id_type": "chat_id",
    }
}

5.3 定时任务（Cron）主动发送

Cron 通过已存储的 _receive_id_store 实现主动推送：

# Cron 执行器
async def execute(self, job: CronJobSpec):
    await self._channel_manager.send_text(
        channel="feishu",
        user_id=job.dispatch.target.user_id,
        session_id=job.dispatch.target.session_id,
        text=job.text,
        meta=dispatch_meta,
    )

前提：用户至少与 Bot 交互过一次（_receive_id_store 中已有映射）。

5.4 多机器人隔离

当同一系统接入多个飞书机器人时，通过 app_id 实现隔离：

def _on_message_sync(self, data):
    # 防跨实例分发
    event_app_id = getattr(header, "app_id", None)
    if event_app_id and event_app_id != self.app_id:
        return  # 丢弃其他机器人的事件

同时，群聊 session_id 包含 app_id 后缀，避免冲突：

if chat_type == "group" and chat_id:
    app_suffix = self.app_id[-4:]
    return f"{app_suffix}_{short_session_id_from_full_id(chat_id)}"

六、数据持久化

6.1 会话历史

文件路径：{workspace}/sessions/feishu/{open_id}_{session_id}.json

{
  "agent": {
    "memory": {
      "content": [
        {"role": "user", "content": "..."},
        {"role": "assistant", "content": "..."}
      ]
    }
  },
  "name": "Friday",
  "_sys_prompt": "..."
}

6.2 接收ID映射

文件路径：{workspace}/feishu_receive_id_store.json

{
  "0f578ccb": ["open_id", "ou_9ba5cd58e37b46f9a59204ac0f578ccb"],
  "app1_abc123": ["chat_id", "oc_xxxxxxxxxxxxxxxx"],
  "thread:abc123": ["open_id", "ou_xxxxxxxxxxxxxxxx"]
}

七、关键设计总结