Skill 清单

企微组织架构识别场景的全部能力定义，按清海 AI 实际工作方式分层组织。会话场景概览见企微组织架构识别，角色维度的会话旅程见角色场景。

架构总览

企微组织架构识别的能力分为四层：

A — Skill（AI 对话侧）

⬜ org-query（待创建）

当前状态：尚未独立创建此 Skill。AI 依赖 search_contacts 工具 + system prompt 中的身份层完成组织查询，覆盖 80% 场景。

Skill 路径（规划）：skills/org-query/SKILL.md

触发意图：

「公司有哪些部门」「技术部有谁」「张三的上级是谁」
「最近有人入职吗」「谁离职了」「前端组几个人」

需要的工具：search_contacts、query_org_structure、query_department_members、query_person_relations、query_staff_changes

B — Tool（暴露给 AI 调用）

✅ search_contacts（复用）

通用人员搜索工具，按姓名/部门/职位模糊查询，底层走 PersonIndex。

python

search_contacts(query: str, department: str = None, limit: int = 10)
→ [{ name, department, position, userid, telephone, ... }]

✅ save_contact_alias（复用）

保存人员别名（昵称、英文名等），补全模糊匹配召回率。

✅ link_platform_identity（复用）

绑定跨平台身份，将企微 userid 与其他系统（GitLab、飞书等）账号关联。

⬜ query_org_structure（待实现）

查询组织架构树，支持指定层级展开。

python

query_org_structure(department_id: int = 1, depth: int = 2)
→ { id, name, children: [...], member_count }

底层：PersonIndex 部门树（同步时已构建）

⬜ query_department_members（待实现）

查询指定部门的成员列表。

python

query_department_members(department: str, include_sub: bool = False)
→ [{ name, userid, position, is_leader }]

底层：PersonIndex + channels.wecom.department 字段

⬜ query_person_relations（待实现）

查询上下级关系：获取某人的直属上级或直属下属。

python

query_person_relations(name: str, relation_type: "leader" | "members")
→ { person, leader } 或 { person, members: [...] }

底层：PersonIndex 的 direct_leader 字段（企微直接提供）

⬜ query_staff_changes（待实现）

查询人员变动记录（入职/离职/调岗）。

python

query_staff_changes(days: int = 7, change_type: "join" | "leave" | "transfer" | None = None)
→ [{ name, change_type, date, department }]

底层：unified_users.change_log 字段（同步时追加写入）

C — 系统能力

⬜ 企微通讯录同步（WeComSync）

需新建，对应飞书的 FeishuSync、钉钉的 DingTalkSync。

核心逻辑：

python

class WeComSync:
    async def sync_contacts(self):
        token = await self._get_access_token()
        depts = await self._fetch_all_departments(token)      # GET /cgi-bin/department/list?id=1
        for dept in depts:
            members = await self._fetch_dept_members(token, dept.id)  # GET /cgi-bin/user/list
            await self._upsert_unified_users(members, channel="wecom")
        await self._rebuild_person_index()

    async def _get_access_token(self):
        # 检查缓存是否过期（7200s），过期则重新获取
        # GET https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=X&corpsecret=Y
        pass

同步字段映射（unified_users.channels.wecom）：

企微字段	存储字段	说明
`userid`	`channels.wecom.userid`	主键
`name`	`display_name`	显示名
`department`	`channels.wecom.department`	部门 ID 列表
`position`	`channels.wecom.position`	职位
`is_leader_in_dept`	`channels.wecom.is_leader`	部门主管标识（数组）
`direct_leader`	`channels.wecom.direct_leader`	直属上级 userid 列表
`status`	`is_active`	1=True，其余=False
`telephone`	`channels.wecom.telephone`	手机号
`hire_date`	`channels.wecom.hire_date`	入职日期

⬜ AES 消息解密（企微特有）

企微所有消息回调均经过 AES-256-CBC 加密，需先解密才能处理。

解密流程：

python

# 1. 验证签名（防伪造）
# GET 请求中 msg_signature = sha1(token + timestamp + nonce + echostr)
# POST 请求中 msg_signature = sha1(token + timestamp + nonce + encrypt_msg)

# 2. Base64 解码 Encrypt 字段
# 3. AES-256-CBC 解密（key = base64.b64decode(EncodingAESKey + "=")）
# 4. 解析内部 XML，取出真实 MsgType / Event / ChangeType 等字段

配置参数（存入 .config.json）：

wecom.token — 验签 Token（自定义）
wecom.encoding_aes_key — 43 位加密密钥
wecom.corpid — 企业 ID
wecom.corpsecret — 应用密钥

⬜ access_token 自动刷新（企微特有）

企微 access_token 有效期 7200 秒，每应用独立，需自动刷新缓存。

python

class WeComTokenManager:
    _cache: dict = {}  # { corpsecret_hash: (token, expire_at) }

    async def get_token(self, corpid, corpsecret) -> str:
        key = hash(corpsecret)
        if key not in self._cache or time.time() > self._cache[key][1] - 300:
            # 提前 5 分钟刷新
            token = await self._fetch_token(corpid, corpsecret)
            self._cache[key] = (token, time.time() + 7200)
        return self._cache[key][0]

✅ PersonIndex 索引（复用）

已有实现，无需改动。扩展 channels.wecom 字段后自动支持企微查询。

✅ PersonContext 档案（复用）

PersonContextResolver.resolve_by_channel("wecom", userid) 已通过通道前缀路由，复用现有实现。

⬜ userid 身份解析适配层

企微有两种消息格式，需在 avatar_core 入口统一提取 userid。

python

def extract_wecom_userid(payload: dict | str) -> str:
    if isinstance(payload, dict):
        # 智能机器人：JSON 格式
        return payload["from"]["userid"]
    else:
        # 自建应用：解密后的 XML 格式
        return parse_xml(payload)["FromUserName"]

⬜ 权限过滤

与飞书/钉钉版本一致，暂未实现，规划中。

D — 主动推送

⬜ 通讯录变更事件接收端点

接收企微推送的通讯录变更事件，触发增量更新。

事件路由逻辑：

python

# 解密后的 XML 事件路由
async def handle_wecom_event(xml: dict):
    if xml["MsgType"] == "event" and xml["Event"] == "change_contact":
        change_type = xml["ChangeType"]
        if change_type == "create_user":
            await sync_single_user(xml["UserID"])
        elif change_type == "update_user":
            await sync_single_user(xml["UserID"])
        elif change_type == "delete_user":
            await mark_user_inactive(xml["UserID"])
        elif change_type in ("create_party", "update_party"):
            await sync_department(xml["Id"])
        elif change_type == "delete_party":
            await mark_department_inactive(xml["Id"])

关键约束：

事件接收端点必须在 5 秒内返回 200 OK，否则企微会重试（最多 3 次）
重复事件用 MsgId + CreateTime 去重

⬜ 每日全量同步（兜底）

凌晨 2:30 触发全量同步，补偿漏掉的事件推送。

python

# daemon.py 中注册定时任务
schedule.every().day.at("02:30").do(WeComSync().sync_contacts)

⬜ 数据质量自检

同步完成后检测数据质量：userid 为空、direct_leader 指向不存在用户等。

⬜ 人员变动通知

通过 notification.py 向老板推送人员变动摘要（可选，每日汇总或实时）。

复用的现有能力

能力	模块路径	复用方式
PersonIndex	`scripts/memory/entity_manager.py`	直接复用，扩展 `channels.wecom` 字段
PersonContextResolver	`scripts/memory/entity_manager.py`	`resolve_by_channel("wecom", userid)`
search_contacts Tool	`scripts/direct/tools/`	无需修改
save_contact_alias Tool	`scripts/direct/tools/`	无需修改
link_platform_identity Tool	`scripts/direct/tools/`	无需修改
notification.py	`scripts/notification.py`	人员变动推送复用防骚扰约束

全局约束

约束	说明
数据权威性	企微通讯录为权威源（`authoritative_sync`），不被 AI 推断覆盖
AES 解密必须	企微所有回调消息必须先验签再解密，跳过验签属于安全漏洞
token 缓存	access_token 必须全局缓存，禁止每次调用时重新获取（会触发频率限制）
离职处理	`status!=1` 或 `ChangeType=delete_user` → `is_active=False`，消息静默
权限最小化	申请企微 API 权限时遵循最小必要原则，通讯录读取权限独立于消息发送权限
5 秒响应	事件回调必须在 5 秒内返回 200，耗时操作异步处理

文档

电子表格

多维表格

知识库

Skill 清单

架构总览

A — Skill（AI 对话侧）

⬜ org-query（待创建）

B — Tool（暴露给 AI 调用）

✅ search_contacts（复用）

✅ save_contact_alias（复用）

✅ link_platform_identity（复用）

⬜ query_org_structure（待实现）

⬜ query_department_members（待实现）

⬜ query_person_relations（待实现）

⬜ query_staff_changes（待实现）

C — 系统能力

⬜ 企微通讯录同步（WeComSync）

⬜ AES 消息解密（企微特有）

⬜ access_token 自动刷新（企微特有）

✅ PersonIndex 索引（复用）

✅ PersonContext 档案（复用）

⬜ userid 身份解析适配层

⬜ 权限过滤

D — 主动推送

⬜ 通讯录变更事件接收端点

⬜ 每日全量同步（兜底）

⬜ 数据质量自检

⬜ 人员变动通知

复用的现有能力

全局约束

Skill 清单 ​

架构总览 ​

A — Skill（AI 对话侧） ​

⬜ org-query（待创建） ​

B — Tool（暴露给 AI 调用） ​

✅ search_contacts（复用） ​

✅ save_contact_alias（复用） ​

✅ link_platform_identity（复用） ​

⬜ query_org_structure（待实现） ​

⬜ query_department_members（待实现） ​

⬜ query_person_relations（待实现） ​

⬜ query_staff_changes（待实现） ​

C — 系统能力 ​

⬜ 企微通讯录同步（WeComSync） ​

⬜ AES 消息解密（企微特有） ​

⬜ access_token 自动刷新（企微特有） ​

✅ PersonIndex 索引（复用） ​

✅ PersonContext 档案（复用） ​

⬜ userid 身份解析适配层 ​

⬜ 权限过滤 ​

D — 主动推送 ​

⬜ 通讯录变更事件接收端点 ​

⬜ 每日全量同步（兜底） ​

⬜ 数据质量自检 ​

⬜ 人员变动通知 ​

复用的现有能力 ​

全局约束 ​

Skill 清单

架构总览

A — Skill（AI 对话侧）

⬜ org-query（待创建）

B — Tool（暴露给 AI 调用）

✅ search_contacts（复用）

✅ save_contact_alias（复用）

✅ link_platform_identity（复用）

⬜ query_org_structure（待实现）

⬜ query_department_members（待实现）

⬜ query_person_relations（待实现）

⬜ query_staff_changes（待实现）

C — 系统能力

⬜ 企微通讯录同步（WeComSync）

⬜ AES 消息解密（企微特有）

⬜ access_token 自动刷新（企微特有）

✅ PersonIndex 索引（复用）

✅ PersonContext 档案（复用）

⬜ userid 身份解析适配层

⬜ 权限过滤

D — 主动推送

⬜ 通讯录变更事件接收端点

⬜ 每日全量同步（兜底）

⬜ 数据质量自检

⬜ 人员变动通知

复用的现有能力

全局约束