飞书接入ChatGPT实战指南：从API集成到生产环境部署

2次阅读

没有评论

共计 2971 个字符，预计需要花费 8 分钟才能阅读完成。

最近在帮公司做飞书机器人接入 ChatGPT 的项目，踩了不少坑也积累了些经验，记录下完整实现过程。这种企业级 IM 工具对接 AI 服务的场景，和普通个人开发还是有很多不同的，特别在鉴权、消息处理和稳定性方面需要特别注意。

传统企业 IM 的痛点很明显：

员工每天要切换多个系统查数据，效率低下
标准化问答占用人力（如 IT Helpdesk）
历史对话无法形成知识沉淀

但改造过程中会遇到几个技术坎儿：

鉴权流程复杂：飞书企业自建应用需要多层审批和加密验证
消息异步处理：飞书要求 5 秒内必须响应，但 AI 生成可能需要 10+ 秒
上下文保持：多轮对话需要维护会话状态，并发时容易混乱

飞书开放平台主要提供三种接入方式：

自建应用（适合企业内部使用）
优点：权限灵活，可获取组织架构
缺点：需要企业管理员审批
应用商店（适合对外分发）
优点：免审批安装
缺点：功能受限
小程序（轻量级场景）
优点：快速上线
缺点：无法使用高级 API

我们选择自建应用方案，因为需要读取部门信息和 @人员功能。这里特别提醒：如果只是简单收发消息，应用商店方案会更省事。

飞书用的是标准的 OAuth2.0 流程，但有几个参数比较特殊：

# 获取预授权码示例
async def get_pre_auth_code():
    headers = {"Authorization": f"Bearer {APP_ACCESS_TOKEN}",
        "Content-Type": "application/json"
    }
    payload = {
        "redirect_uri": CALLBACK_URL,
        "app_id": APP_ID
    }
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://open.feishu.cn/open-apis/authen/v1/index",
            headers=headers,
            json=payload
        ) as resp:
            return await resp.json()

注意点：

APP_ACCESS_TOKEN 需要每 2 小时刷新一次
回调地址必须完全匹配注册时填写的 URL
生产环境一定要做 CSRF 防护

飞书所有事件回调都会带签名，必须验证才能处理：

def verify_signature(timestamp, nonce, signature, body):
    content = f"{timestamp}\n{nonce}\n{body}".encode('utf-8')
    key = APP_SECRET.encode('utf-8')
    hash = hmac.new(key, content, hashlib.sha256).digest()
    return base64.b64encode(hash).decode('utf-8') == signature

这里有个坑：body 必须是原始请求体字符串，不能是解析后的 JSON 对象。

用 Redis 存储上下文信息，结构设计如下：

# 存储结构示例
{
    "session_id": "user123_chat456",
    "context": [{"role": "user", "content": "如何申请年假"},
        {"role": "assistant", "content": "请访问 HR 系统..."}
    ],
    "created_at": 1689926400,
    "updated_at": 1689926420,
    "expire_in": 3600  # 1 小时过期
}

使用 Lua 脚本保证操作的原子性：

-- 更新上下文的 Lua 脚本
local key = KEYS[1]
local new_msg = ARGV[1]
local max_len = tonumber(ARGV[2])
local expire = tonumber(ARGV[3])

local data = redis.call('GET', key)
if not data then
    return 0
end

data = cjson.decode(data)
table.insert(data.context, new_msg)

-- 控制上下文长度
while #data.context > max_len do
    table.remove(data.context, 1)
end

data.updated_at = os.time()
redis.call('SETEX', key, expire, cjson.encode(data))
return 1

ChatGPT API 有严格限制（免费版 3 次 / 分钟），需要用令牌桶控制：

class TokenBucket:
    def __init__(self, capacity, refill_rate):
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = time.time()
        self.refill_rate = refill_rate  # 令牌 / 秒
        self.lock = threading.Lock()

    def consume(self):
        with self.lock:
            now = time.time()
            elapsed = now - self.last_refill
            self.tokens = min(
                self.capacity,
                self.tokens + elapsed * self.refill_rate
            )
            self.last_refill = now
            if self.tokens >= 1:
                self.tokens -= 1
                return True
            return False

用正则匹配身份证、手机号等：

SENSITIVE_PATTERNS = [(r'\b[1-9]\d{5}(18|19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[0-9Xx]\b', '[ID]'),
    (r'\b1[3-9]\d{9}\b', '[PHONE]')
]

def sanitize_text(text):
    for pattern, replacement in SENSITIVE_PATTERNS:
        text = re.sub(pattern, replacement, text)
    return text