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

3次阅读

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

在企业办公场景中，IM 系统与 AI 助手的集成已经成为提升效率的重要手段。飞书作为企业级 IM 平台，接入 ChatGPT 可以带来诸多便利，比如智能工单处理、会议纪要生成、自动问答等。然而，原生对接过程中存在几个关键挑战：

认证管理复杂 ：飞书使用 JWT 进行身份验证，需要正确处理签名和时效性
流量控制困难 ：ChatGPT API 有严格的速率限制，直接调用容易触发限流
消息格式差异 ：飞书的消息体采用加密和特定结构，需要额外解析处理
上下文管理 ：多轮对话场景下需要维护会话状态

直接调用方案
优点：实现简单，开发速度快
缺点：无法应对高并发，缺乏弹性容错机制
中间件代理方案
优点：可以加入队列缓冲、请求合并等优化策略
缺点：架构复杂度提高，需要额外运维成本

我们推荐采用中间件代理方案，具体架构如下：

graph TD
    A[飞书客户端] -->| 加密消息 | B[飞书事件订阅]
    B --> C[API 网关]
    C --> D[请求队列]
    D --> E[ChatGPT 处理模块]
    E --> F[响应返回]

JWT 验证 ：使用飞书提供的公钥验证请求签名
速率限制 ：采用令牌桶算法控制请求频率
消息队列 ：使用 Redis 或 RabbitMQ 缓冲高峰流量

from typing import Dict, Any
import json
import base64
import hashlib
import hmac
from cryptography.hazmat.primitives import serialization

class FeishuEventParser:
    def __init__(self, verification_token: str, encrypt_key: str):
        self.verification_token = verification_token
        self.encrypt_key = encrypt_key

    def verify_signature(self, timestamp: str, nonce: str, signature: str, body: str) -> bool:
        content = f"{timestamp}{nonce}{self.verification_token}{body}".encode('utf-8')
        sign = hmac.new(self.verification_token.encode('utf-8'), content, hashlib.sha256).hexdigest()
        return sign == signature

    def decrypt_event(self, encrypted_data: str) -> Dict[str, Any]:
        # 解密逻辑实现
        pass

import openai
import backoff
import logging
from typing import Optional

class ChatGPTClient:
    def __init__(self, api_key: str, max_retries: int = 3):
        openai.api_key = api_key
        self.max_retries = max_retries
        self.logger = logging.getLogger(__name__)

    @backoff.on_exception(backoff.expo, openai.error.RateLimitError, max_tries=3)
    async def chat_completion(
        self,
        messages: list,
        temperature: float = 0.7,
        max_tokens: Optional[int] = None
    ) -> str:
        try:
            response = await openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=messages,
                temperature=temperature,
                max_tokens=max_tokens
            )
            return response.choices[0].message.content
        except Exception as e:
            self.logger.error(f"ChatGPT API error: {str(e)}")
            raise

使用 Redis 存储已处理消息的 message_id，设置合理的过期时间。处理消息前先检查是否已存在，避免重复处理。