每天都有大量重复问题在客服群里刷屏？凌晨3点还有人问"怎么退换货"？传统客服成本高、响应慢、夜班难招？今天用 DeepSeek + 飞书机器人 + FastAPI，从零搭一个 24h 在线的 AI 智能客服。

---

📖 目录

• 一、为什么需要 AI 客服？
• 二、技术选型与架构设计
• 三、环境准备
• 四、创建飞书机器人
• 五、后端实现：FastAPI 接收消息并调用 DeepSeek
• 六、对话上下文管理
• 七、知识库接入（让AI知道你的业务）
• 八、部署到服务器
• 九、运行效果展示
• 十、踩坑记录与经验总结

---

一、为什么需要 AI 客服？

先看一组数据：

┌─────────────────────────────────────────────────┐
│           传统客服 vs AI客服                       │
├──────────────┬──────────────┬────────────────────┤
│     维度      │   传统客服    │     AI 客服        │
├──────────────┼──────────────┼────────────────────┤
│  响应速度     │  1-5分钟     │  秒级               │
│  服务时间     │  工作时间     │  7×24h             │
│  并发能力     │  一人服务N个  │  无限并发           │
│  人力成本     │  3000-8000/月│  API费用≈几十/月    │
│  知识一致性   │  因人而异     │  永远标准答案        │
│  情绪管理     │  会疲惫/暴躁  │  永远耐心            │
└──────────────┴──────────────┴────────────────────┘

我们的方案是：常见问题 AI 自动回复，复杂问题转人工。70-80%的客服问题都是重复的（怎么注册、怎么退款、运费多少……），这部分完全可以交给 AI。

---

二、技术选型与架构设计

2.1 技术栈

层级	技术选型	理由
AI 大模型	DeepSeek Chat	性价比高，中文能力强
后端框架	FastAPI (Python)	异步支持好，适合处理飞书事件回调
消息通道	飞书自定义机器人	国内企业常用，免费
部署	Docker + Nginx	标准化部署，方便维护

2.2 系统架构图

┌──────────────────────────────────────────────────────────┐
│                      用户（飞书）                          │
│                   发送消息到群聊/私聊                       │
└─────────────────────────┬────────────────────────────────┘
                          │ 飞书事件回调
                          ▼
┌──────────────────────────────────────────────────────────┐
│                    飞书开放平台                             │
│            接收消息 → 推送到你的服务器                       │
└─────────────────────────┬────────────────────────────────┘
                          │ POST /feishu/webhook
                          ▼
┌──────────────────────────────────────────────────────────┐
│                  FastAPI 后端服务                           │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐   │
│  │  飞书消息解析  │→│  意图判断     │→│  路由分发      │   │
│  │  (验证签名等)  │  │  (关键词/AI)  │  │  (AI/转人工)  │   │
│  └──────────────┘  └──────────────┘  └──────┬───────┘   │
│                                               │           │
│                    ┌──────────────────────────┤           │
│                    ▼                          ▼           │
│           ┌──────────────┐          ┌──────────────┐     │
│           │  DeepSeek API│          │  转人工通知    │     │
│           │  (AI回复)     │          │  (飞书@管理员) │     │
│           └──────┬───────┘          └──────────────┘     │
│                  │                                        │
│                  ▼                                        │
│           ┌──────────────┐                                │
│           │  知识库检索    │ (可选，RAG增强)                 │
│           │  (企业FAQ)    │                                │
│           └──────────────┘                                │
└──────────────────────────────────────────────────────────┘
                          │ 飞书 API 发送消息
                          ▼
┌──────────────────────────────────────────────────────────┐
│                      飞书（回复用户）                       │
└──────────────────────────────────────────────────────────┘

2.3 项目结构

feishu-ai-customer-service/
├── app/
│   ├── main.py              # FastAPI 入口
│   ├── config.py            # 配置管理（环境变量）
│   ├── routers/
│   │   └── feishu_router.py # 飞书事件回调路由
│   ├── services/
│   │   ├── feishu_service.py    # 飞书 API 封装
│   │   ├── deepseek_service.py  # DeepSeek API 封装
│   │   └── knowledge_service.py # 知识库检索（可选）
│   ├── models/
│   │   └── feishu_models.py  # Pydantic 数据模型
│   └── utils/
│       └── signature.py      # 飞书签名验证
├── requirements.txt
├── Dockerfile
├── .env.example
└── README.md

---

三、环境准备

3.1 安装依赖

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装依赖
pip install fastapi uvicorn httpx pydantic python-dotenv

3.2 环境变量配置

创建 .env 文件：

# DeepSeek API
DEEPSEEK_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx

# 飞书应用
FEISHU_APP_ID=cli_xxxxxxxxxxxx
FEISHU_APP_SECRET=xxxxxxxxxxxxxxxxxxxxxxxx

# 飞书机器人配置
FEISHU_BOT_NAME=智能客服小助手
FEISHU_VERIFICATION_TOKEN=xxxxxxxxxxxxxxxx
FEISHU_ENCRYPT_KEY=xxxxxxxxxxxxxxxx

# 服务配置
SERVER_HOST=0.0.0.0
SERVER_PORT=8000

---

四、创建飞书机器人

4.1 创建企业自建应用

1. 登录 飞书开放平台
2. 点击「创建企业自建应用」
3. 填写应用名称（如"智能客服"）和描述
4. 记录 App ID 和 App Secret

4.2 开通权限

在「权限管理」中，添加以下权限：

权限	权限标识	用途
接收群聊消息	im:message:receive_v1	接收用户在群里的消息
发送消息	im:message:send_as_bot	机器人发送回复
获取群信息	im:chat:readonly	获取群聊信息
读取消息中…的用户	im:message.resource	解析@人的消息

💡 小技巧：在「权限管理」→「批量导入/导出权限」中，粘贴以下JSON快速添加：

[
  "im:message:receive_v1",
  "im:message:send_as_bot",
  "im:chat:readonly",
  "im:message.resource"
]

4.3 配置事件订阅

1. 进入「事件与回调」→「事件配置」
2. 请求地址填写：https://你的域名/feishu/webhook
3. 添加事件：接收消息 im.message.receive_v1
4. 记录 Verification Token 和 Encrypt Key

4.4 发布应用

1. 进入「版本管理与发布」
2. 创建版本，填写版本号和更新说明
3. 提交审核（如果只需要内部使用，可以直接发布）

---

五、后端实现：FastAPI 接收消息并调用 DeepSeek

5.1 配置管理

# app/config.py
import os
from dotenv import load_dotenv

load_dotenv()

class Settings:
    """全局配置"""
    # DeepSeek
    DEEPSEEK_API_KEY: str = os.getenv("DEEPSEEK_API_KEY", "")
    DEEPSEEK_BASE_URL: str = "https://api.deepseek.com"
    DEEPSEEK_MODEL: str = "deepseek-chat"

    # 飞书
    FEISHU_APP_ID: str = os.getenv("FEISHU_APP_ID", "")
    FEISHU_APP_SECRET: str = os.getenv("FEISHU_APP_SECRET", "")
    FEISHU_VERIFICATION_TOKEN: str = os.getenv("FEISHU_VERIFICATION_TOKEN", "")
    FEISHU_ENCRYPT_KEY: str = os.getenv("FEISHU_ENCRYPT_KEY", "")

    # 客服配置
    BOT_NAME: str = os.getenv("FEISHU_BOT_NAME", "智能客服")
    SYSTEM_PROMPT: str = """你是一个专业的客服助手。请遵循以下规则：
1. 回答用户的问题要准确、简洁、有礼貌
2. 如果不确定答案，诚实告知用户
3. 当用户问题过于复杂时，建议转人工客服
4. 回答控制在200字以内，超出时建议用户具体说明需求
5. 不要编造不存在的政策或规则
"""

settings = Settings()

5.2 DeepSeek 服务

# app/services/deepseek_service.py
import httpx
import json
from app.config import settings

async def chat_completion(
    messages: list[dict],
    temperature: float = 0.7,
    max_tokens: int = 1024,
    stream: bool = False,
) -> str:
    """
    调用 DeepSeek API 进行对话
    """
    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(
            f"{settings.DEEPSEEK_BASE_URL}/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {settings.DEEPSEEK_API_KEY}",
                "Content-Type": "application/json",
            },
            json={
                "model": settings.DEEPSEEK_MODEL,
                "messages": messages,
                "temperature": temperature,
                "max_tokens": max_tokens,
                "stream": stream,
            },
        )

    if response.status_code != 200:
        error_msg = response.json().get("error", {}).get("message", "未知错误")
        raise Exception(f"DeepSeek API 调用失败: {error_msg}")

    return response.json()["choices"][0]["message"]["content"]


async def chat_with_context(user_message: str, chat_history: list[dict] = None) -> str:
    """
    带上下文的对话（支持多轮）
    """
    messages = [{"role": "system", "content": settings.SYSTEM_PROMPT}]

    # 加入历史对话（最多保留最近10轮）
    if chat_history:
        recent = chat_history[-20:]  # 10轮 = 20条消息
        messages.extend(recent)

    # 加入当前用户消息
    messages.append({"role": "user", "content": user_message})

    return await chat_completion(messages)

5.3 飞书服务

# app/services/feishu_service.py
import httpx
import hashlib
import base64
from app.config import settings

# 租户访问令牌缓存
_tenant_token = {"token": "", "expires_at": 0}


async def get_tenant_token() -> str:
    """获取飞书 tenant_access_token（带缓存）"""
    import time

    # 如果缓存未过期，直接返回
    if _tenant_token["token"] and time.time() < _tenant_token["expires_at"]:
        return _tenant_token["token"]

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(
            "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
            json={
                "app_id": settings.FEISHU_APP_ID,
                "app_secret": settings.FEISHU_APP_SECRET,
            },
        )

    data = response.json()
    if data.get("code") != 0:
        raise Exception(f"获取飞书token失败: {data}")

    _tenant_token["token"] = data["tenant_access_token"]
    _tenant_token["expires_at"] = time.time() + data["expire"] - 60  # 提前60秒刷新

    return _tenant_token["token"]


async def send_message(chat_id: str, text: str, msg_type: str = "text") -> dict:
    """
    发送消息到飞书群聊或私聊
    chat_id: 群聊ID (oc_xxx) 或 用户ID (ou_xxx)
    """
    token = await get_tenant_token()

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(
            "https://open.feishu.cn/open-apis/im/v1/messages",
            headers={"Authorization": f"Bearer {token}"},
            params={"receive_id_type": "chat_id"},
            json={
                "receive_id": chat_id,
                "msg_type": msg_type,
                "content": json.dumps({"text": text}),
            },
        )

    return response.json()


async def reply_message(message_id: str, text: str) -> dict:
    """回复指定消息（用于在话题中回复）"""
    token = await get_tenant_token()

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(
            f"https://open.feishu.cn/open-apis/im/v1/messages/{message_id}/reply",
            headers={"Authorization": f"Bearer {token}"},
            json={
                "msg_type": "text",
                "content": json.dumps({"text": text}),
            },
        )

    return response.json()

5.4 核心路由：处理飞书事件回调

# app/routers/feishu_router.py
import json
import hashlib
from fastapi import APIRouter, Request, HTTPException
from app.services.feishu_service import reply_message
from app.services.deepseek_service import chat_with_context
from app.config import settings

router = APIRouter(prefix="/feishu", tags=["feishu"])

# 对话上下文存储（简易版，生产环境建议用Redis）
chat_histories: dict[str, list[dict]] = {}


@router.post("/webhook")
async def feishu_webhook(request: Request):
    """
    飞书事件回调入口
    处理所有从飞书推送过来的事件
    """
    body = await request.json()

    # 1. URL 验证（首次配置时飞书会发一个 challenge 请求）
    if "challenge" in body:
        return {"challenge": body["challenge"]}

    # 2. 解析事件
    event = body.get("event", {})

    # 只处理文本消息
    if event.get("message_type") != "text":
        return {"code": 0, "msg": "ignore non-text message"}

    message_id = event.get("message_id", "")
    chat_id = event.get("chat_id", "")
    content_str = event.get("message", {}).get("content", "{}")

    # 3. 解析消息内容（飞书文本消息是 JSON 格式）
    try:
        content = json.loads(content_str)
        user_text = content.get("text", "").strip()
    except json.JSONDecodeError:
        return {"code": 0, "msg": "invalid content"}

    if not user_text:
        return {"code": 0, "msg": "empty message"}

    # 4. 去掉@机器人（群聊中会自动带上）
    user_text = user_text.replace(f"@_user_{settings.FEISHU_APP_ID}", "").strip()

    if not user_text:
        return {"code": 0, "msg": "empty after removing mention"}

    # 5. 获取或创建对话上下文
    if chat_id not in chat_histories:
        chat_histories[chat_id] = []

    # 6. 判断是否需要转人工
    if _should_transfer_to_human(user_text):
        ai_reply = "您的需求比较复杂，已为您转接人工客服，请稍等 ⏳"
        await reply_message(message_id, ai_reply)
        # TODO: 发送通知给人工客服
        return {"code": 0, "msg": "transferred to human"}

    # 7. 调用 DeepSeek 生成回复
    try:
        ai_reply = await chat_with_context(
            user_message=user_text,
            chat_history=chat_histories[chat_id],
        )
    except Exception as e:
        ai_reply = "抱歉，我遇到了一点问题，请稍后再试 🙏"
        print(f"DeepSeek 调用失败: {e}")

    # 8. 更新对话历史
    chat_histories[chat_id].append({"role": "user", "content": user_text})
    chat_histories[chat_id].append({"role": "assistant", "content": ai_reply})

    # 限制历史长度（最多保留最近10轮）
    if len(chat_histories[chat_id]) > 20:
        chat_histories[chat_id] = chat_histories[chat_id][-20:]

    # 9. 回复消息
    await reply_message(message_id, ai_reply)

    return {"code": 0, "msg": "ok"}


def _should_transfer_to_human(text: str) -> bool:
    """
    判断是否需要转人工客服
    匹配关键词列表
    """
    transfer_keywords = [
        "转人工", "人工客服", "找人工", "投诉",
        "退款", "赔偿", "法律", "律师",
    ]
    return any(kw in text for kw in transfer_keywords)

5.5 FastAPI 入口

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers.feishu_router import router as feishu_router

app = FastAPI(title="飞书AI客服", version="1.0.0")

# 跨域配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(feishu_router)


@app.get("/health")
async def health_check():
    return {"status": "ok"}

---

六、对话上下文管理

6.1 为什么需要上下文管理？

没有上下文管理的对话：

用户：你们支持什么支付方式？
AI：我们支持微信、支付宝、银行卡。

用户：第一种怎么用？       ← AI 不知道"第一种"是什么
AI：抱歉，我不太理解您的问题。

有上下文管理后：

用户：你们支持什么支付方式？
AI：我们支持微信、支付宝、银行卡。

用户：第一种怎么用？       ← AI 知道上文是支付方式
AI：微信支付很简单：打开微信→我→支付→扫码...

6.2 生产环境建议

当前我们用 dict 存储在内存里，重启就丢失了。生产环境建议：

# 使用 Redis 存储对话历史
import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def save_chat_history(chat_id: str, history: list[dict]):
    """保存对话历史到 Redis，设置1小时过期"""
    redis_client.setex(
        f"chat:{chat_id}",
        3600,  # 1小时过期
        json.dumps(history)
    )

def get_chat_history(chat_id: str) -> list[dict]:
    """从 Redis 读取对话历史"""
    data = redis_client.get(f"chat:{chat_id}")
    return json.loads(data) if data else []

---

七、知识库接入（让AI知道你的业务）

单纯依赖 DeepSeek 的通用知识是不够的，你的客服需要回答的是你的业务问题。

7.1 方案：Prompt 注入业务知识

最简单有效的方式：把你的业务FAQ直接写进 System Prompt。

SYSTEM_PROMPT = """你是一个专业的客服助手。

【公司信息】
- 公司名称：XX科技有限公司
- 客服电话：400-123-4567
- 工作时间：周一至周五 9:00-18:00
- 官网：www.example.com

【常见问题 FAQ】
Q: 如何注册账号？
A: 访问官网首页，点击右上角"免费注册"，使用手机号验证即可。

Q: 支持哪些支付方式？
A: 支持微信支付、支付宝、银联卡支付。

Q: 如何申请退款？
A: 登录账户→订单管理→选择订单→申请退款，1-3个工作日内处理。

Q: 运费怎么算？
A: 单笔订单满99元包邮，不满99元收取8元运费。

Q: 发票怎么开？
A: 下单时选择"需要发票"，填写开票信息，电子发票1-2天发送到邮箱。

【回答规则】
1. 优先根据以上FAQ回答
2. 如果FAQ中没有，根据常识回答，但要说明"以实际情况为准"
3. 回答控制在200字以内
4. 不确定的问题建议转人工客服
"""

7.2 进阶：RAG 检索增强

如果FAQ很多（几百条），直接塞进 Prompt 不现实。进阶方案是用向量数据库：

用户提问 → 向量化 → 在知识库中检索相关FAQ → 注入Prompt → DeepSeek生成回答

这部分可以参考我的另一篇：手把手搭建本地AI知识库（DeepSeek + 向量数据库）

---

八、部署到服务器

8.1 Docker 部署

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY app/ ./app/

# 暴露端口
EXPOSE 8000

# 启动
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# 构建镜像
docker build -t feishu-ai-cs .

# 运行容器
docker run -d \
  --name feishu-ai-cs \
  -p 8000:8000 \
  --env-file .env \
  feishu-ai-cs

8.2 Nginx 反向代理 + HTTPS

飞书的事件回调要求 HTTPS，所以需要 Nginx + SSL 证书：

server {
    listen 443 ssl;
    server_name your-domain.com;

    ssl_certificate /etc/nginx/ssl/cert.pem;
    ssl_certificate_key /etc/nginx/ssl/key.pem;

    location /feishu/ {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

💡 SSL 证书可以用 Let’s Encrypt 免费申请：certbot certonly --nginx -d your-domain.com

---

九、运行效果展示

┌─────────────────────────────────────────────┐
│ 飞书群聊                                    │
├─────────────────────────────────────────────┤
│                                             │
│ 👤 用户A: 你们支持什么支付方式？              │
│                                             │
│ 🤖 智能客服:                                │
│ 您好！我们目前支持以下支付方式：              │
│ 1️⃣ 微信支付                                │
│ 2️⃣ 支付宝                                  │
│ 3️⃣ 银联卡支付                              │
│ 单笔订单满99元还免运费哦~                    │
│                                             │
│ 👤 用户B: 第一种怎么用？                     │
│                                             │
│ 🤖 智能客服:                                │
│ 微信支付很简单：                             │
│ 1. 下单时选择"微信支付"                      │
│ 2. 系统会自动跳转到微信支付页面              │
│ 3. 确认金额并输入密码即可完成支付             │
│                                             │
│ 👤 用户C: 我要投诉！                         │
│                                             │
│ 🤖 智能客服:                                │
│ 您的需求比较复杂，已为您转接人工客服，        │
│ 请稍等 ⏳                                   │
│                                             │
│ 👨‍💼 客服小王: [收到转接通知]               │
│ 您好，我是人工客服小王，请问有什么可以帮您？   │
└─────────────────────────────────────────────┘

---

十、踩坑记录与经验总结

踩坑1：飞书消息内容是 JSON 字符串

飞书推送的事件中，message.content 是一个 JSON 字符串，不是对象。

# ❌ 错误写法
text = event["message"]["content"]["text"]  # TypeError!

# ✅ 正确写法
content = json.loads(event["message"]["content"])
text = content["text"]

踩坑2：群聊消息需要去掉 @机器人

在群里 @机器人 时，text 内容会包含 @_user_{app_id}，必须去掉，否则 AI 会把这段也当作用户输入。

踩坑3：飞书回调要求 3 秒内响应

飞书要求回调接口在 3 秒内 返回响应，否则会重试。DeepSeek 生成回复可能超过 3 秒，所以建议：

• 先立即返回 {"code": 0}，再异步处理
• 或者用消息队列（Celery）异步执行

踩坑4：tenant_access_token 有过期时间

Token 有效期 2 小时，需要做好缓存和自动刷新。提前 60 秒刷新避免使用过期 Token。

---

如果觉得有帮助，点赞 + 收藏支持一下！有问题欢迎评论区讨论 💬