企业微信客服自动化系统

这是一个完整的企业微信客服 AI 助手解决方案，能够自动处理好友添加、智能问答、人工转接等场景。

核心功能

1. 自动同意好友添加

实时监听好友添加事件
自动通过好友请求
发送欢迎消息
标注用户来源和标签

实时监听好友添加事件
自动通过好友请求
发送欢迎消息
标注用户来源和标签

2. 基于知识库的智能问答

向量知识库存储企业知识
语义搜索匹配问题
LLM 生成专业回复
支持多轮对话上下文

向量知识库存储企业知识
语义搜索匹配问题
LLM 生成专业回复
支持多轮对话上下文

3. 未知问题人工介入

置信度阈值判断
自动提醒人工客服
转接对话给人工
记录未解决问题用于优化

置信度阈值判断
自动提醒人工客服
转接对话给人工
记录未解决问题用于优化

技术架构

┌─────────────┐
│ 企业微信     │
│  Webhook    │
└──────┬──────┘
       │
       ▼
┌─────────────────┐
│  回调服务器      │
│  (Go/Python)    │
└──────┬──────────┘
       │
       ├──────────────────┐
       │                  │
       ▼                  ▼
┌──────────────┐   ┌──────────────┐
│ 向量知识库    │   │  LLM API     │
│ (PG+pgvector)│   │ (Kimi/GPT-4) │
└──────────────┘   └──────────────┘
       │
       ▼
┌──────────────┐
│  人工提醒     │
│  (Telegram)  │
└──────────────┘

┌─────────────┐
│ 企业微信     │
│  Webhook    │
└──────┬──────┘
       │
       ▼
┌─────────────────┐
│  回调服务器      │
│  (Go/Python)    │
└──────┬──────────┘
       │
       ├──────────────────┐
       │                  │
       ▼                  ▼
┌──────────────┐   ┌──────────────┐
│ 向量知识库    │   │  LLM API     │
│ (PG+pgvector)│   │ (Kimi/GPT-4) │
└──────────────┘   └──────────────┘
       │
       ▼
┌──────────────┐
│  人工提醒     │
│  (Telegram)  │
└──────────────┘

快速开始

第一步：配置企业微信应用

创建企业微信应用
- 登录企业微信管理后台
- 应用管理 → 创建应用 → 选择"微信客服"
- 记录以下信息：
  - ```
  corp_id
```
  : 企业 ID
- ```
agent_id
```
    : 应用 AgentId
  - ```
  secret
```
  : 应用 Secret

配置回调地址

URL: https://your-domain.com/wecom/callback
Token: 自定义验证令牌
EncodingAESKey: 自动生成

订阅所需事件
- 联系人变更事件
- 消息事件
- 外部联系人免验证添加事件

创建企业微信应用
- 登录企业微信管理后台
- 应用管理 → 创建应用 → 选择"微信客服"
- 记录以下信息：
  - ```
  corp_id
```
  : 企业 ID
- ```
agent_id
```
    : 应用 AgentId
  - ```
  secret
```
  : 应用 Secret

配置回调地址

URL: https://your-domain.com/wecom/callback
Token: 自定义验证令牌
EncodingAESKey: 自动生成

订阅所需事件
- 联系人变更事件
- 消息事件
- 外部联系人免验证添加事件

第二步：设置知识库

bash

undefined

bash

undefined

1. 安装 PostgreSQL + pgvector

sudo apt install postgresql-14 sudo -u postgres psql -c "CREATE EXTENSION vector;"

2. 创建数据库

sudo -u postgres createdb wecom_kb

3. 初始化表结构

psql wecom_kb < ~/clawd/skills/wecom-cs-automation/schema.sql

undefined

psql wecom_kb < ~/clawd/skills/wecom-cs-automation/schema.sql

undefined

第三步：导入知识库数据

bash

undefined

bash

undefined

1. 准备知识文档（Markdown 格式）

2. 切片并向量化

python3 ~/clawd/skills/wecom-cs-automation/scripts/import_kb.py
--input knowledge.md
--token $(pass show api/kimi)

3. 验证导入

psql wecom_kb -c "SELECT COUNT(*) FROM knowledge_chunks;"

undefined

psql wecom_kb -c "SELECT COUNT(*) FROM knowledge_chunks;"

undefined

第四步：启动回调服务

bash

undefined

bash

undefined

1. 配置环境变量

cat > .env << EOF WECOM_CORP_ID=$(pass show api/wecom-corp-id) WECOM_AGENT_SECRET=$(pass show api/wecom-agent-secret) WECOM_TOKEN=your_webhook_token WECOM_AES_KEY=your_aes_key KB_DB_URL=postgresql://localhost/wecom_kb LLM_API_KEY=$(pass show api/kimi) LLM_API_BASE=https://api.moonshot.cn/v1 NOTIFICATION_CHANNEL=telegram:8518085684 EOF

2. 启动服务（Python FastAPI）

uvicorn wecom_callback_server:app --host 0.0.0.0 --port 8000

或使用 Go

go run cmd/server/main.go

undefined

go run cmd/server/main.go

undefined

第五步：验证服务

bash

undefined

bash

undefined

1. 检查服务状态

curl http://localhost:8000/health

2. 测试知识库搜索

curl -X POST http://localhost:8000/api/test_kb
-H "Content-Type: application/json"
-d '{"query": "如何退款？"}'

undefined

curl -X POST http://localhost:8000/api/test_kb
-H "Content-Type: application/json"
-d '{"query": "如何退款？"}'

undefined

使用方法

场景 1：自动欢迎新好友

当用户添加客服为好友时：

python

undefined

当用户添加客服为好友时：

python

undefined

skills/wecom-cs-automation/workflows/on_friend_add.py

async def handle_friend_add(user_id, user_name): # 1. 通过好友请求 await wecom.accept_friend(user_id)

# 2. 添加用户标签
await wecom.add_external_tag(user_id, tags=["新客户"])

# 3. 发送欢迎消息
welcome_msg = f"""👋 欢迎来到{name}！

我是智能客服小助手，可以帮您： • 查询订单状态 • 解答常见问题 • 处理售后问题

如有复杂问题，我会转接人工客服为您服务。"""

await wecom.send_text(user_id, welcome_msg)

undefined

async def handle_friend_add(user_id, user_name): # 1. 通过好友请求 await wecom.accept_friend(user_id)

# 2. 添加用户标签
await wecom.add_external_tag(user_id, tags=["新客户"])

# 3. 发送欢迎消息
welcome_msg = f"""👋 欢迎来到{name}！

我是智能客服小助手，可以帮您： • 查询订单状态 • 解答常见问题 • 处理售后问题

如有复杂问题，我会转接人工客服为您服务。"""

await wecom.send_text(user_id, welcome_msg)

undefined

场景 2：知识库问答

python

undefined

python

undefined

skills/wecom-cs-automation/workflows/answer_question.py

async def handle_question(user_id, question): # 1. 搜索知识库 chunks = await search_knowledge(question, top_k=3)

# 2. 构建提示词
context = "\n\n".join([c.content for c in chunks])
prompt = f"""基于以下知识库内容回答用户问题：

知识库： {context}

用户问题：{question}

如果知识库中没有相关信息，请回复"抱歉，这个问题我暂时无法回答，已为您转接人工客服。""""

# 3. 调用 LLM
answer = await call_llm(prompt)

# 4. 判断是否需要人工介入
if "暂时无法回答" in answer or chunks[0].similarity < 0.7:
    await escalate_to_human(user_id, question)
else:
    await wecom.send_text(user_id, answer)

undefined

async def handle_question(user_id, question): # 1. 搜索知识库 chunks = await search_knowledge(question, top_k=3)

# 2. 构建提示词
context = "\n\n".join([c.content for c in chunks])
prompt = f"""基于以下知识库内容回答用户问题：

知识库： {context}

用户问题：{question}

如果知识库中没有相关信息，请回复"抱歉，这个问题我暂时无法回答，已为您转接人工客服。""""

# 3. 调用 LLM
answer = await call_llm(prompt)

# 4. 判断是否需要人工介入
if "暂时无法回答" in answer or chunks[0].similarity < 0.7:
    await escalate_to_human(user_id, question)
else:
    await wecom.send_text(user_id, answer)

undefined

场景 3：人工介入提醒

python

undefined

python

undefined

skills/wecom-cs-automation/workflows/escalate.py

async def escalate_to_human(user_id, question): # 1. 发送用户消息 await wecom.send_text(user_id, "⏳ 已为您转接人工客服，请稍候...")

# 2. 通过 Telegram 通知人工客服
user_info = await wecom.get_user_info(user_id)
notification = f"""🚨 需要人工介入

用户：{user_info.name} ({user_info.id}) 问题：{question} 时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}

请及时处理。"""

await send_telegram_message(notification)

# 3. 记录未解决问题
await save_unknown_question(user_id, question)

undefined

async def escalate_to_human(user_id, question): # 1. 发送用户消息 await wecom.send_text(user_id, "⏳ 已为您转接人工客服，请稍候...")

# 2. 通过 Telegram 通知人工客服
user_info = await wecom.get_user_info(user_id)
notification = f"""🚨 需要人工介入

用户：{user_info.name} ({user_info.id}) 问题：{question} 时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}

请及时处理。"""

await send_telegram_message(notification)

# 3. 记录未解决问题
await save_unknown_question(user_id, question)

undefined

目录结构

~/clawd/skills/wecom-cs-automation/
├── SKILL.md                    # 本文件
├── schema.sql                  # 数据库表结构
├── config/
│   ├── kb_config.yaml          # 知识库配置
│   └── escalation_rules.yaml   # 转人工规则
├── scripts/
│   ├── import_kb.py            # 导入知识库
│   ├── search_kb.py            # 测试知识库搜索
│   └── init_db.py              # 初始化数据库
├── workflows/
│   ├── on_friend_add.py        # 好友添加处理
│   ├── answer_question.py      # 问答处理
│   └── escalate.py             # 人工转接
├── server/
│   ├── main.py                 # FastAPI 主服务
│   ├── wecom_client.py         # 企业微信 API 客户端
│   ├── kb_searcher.py          # 知识库搜索
│   └── notification.py         # 通知服务
└── knowledge/
    └── sample.md               # 示例知识文档

~/clawd/skills/wecom-cs-automation/
├── SKILL.md                    # 本文件
├── schema.sql                  # 数据库表结构
├── config/
│   ├── kb_config.yaml          # 知识库配置
│   └── escalation_rules.yaml   # 转人工规则
├── scripts/
│   ├── import_kb.py            # 导入知识库
│   ├── search_kb.py            # 测试知识库搜索
│   └── init_db.py              # 初始化数据库
├── workflows/
│   ├── on_friend_add.py        # 好友添加处理
│   ├── answer_question.py      # 问答处理
│   └── escalate.py             # 人工转接
├── server/
│   ├── main.py                 # FastAPI 主服务
│   ├── wecom_client.py         # 企业微信 API 客户端
│   ├── kb_searcher.py          # 知识库搜索
│   └── notification.py         # 通知服务
└── knowledge/
    └── sample.md               # 示例知识文档

API 配置

所需密钥

bash

undefined

bash

undefined

企业微信

pass insert api/wecom-corp-id # 企业 ID pass insert api/wecom-agent-secret # 应用 Secret

LLM（推荐 Kimi，中文优化）

pass insert api/kimi # 已有

Telegram 通知（可选）

pass insert api/telegram-bot # 已有

undefined

pass insert api/telegram-bot # 已有

undefined

配置文件

创建

~/clawd/skills/wecom-cs-automation/.env

：

env

undefined

创建

~/clawd/skills/wecom-cs-automation/.env

：

env

undefined

企业微信配置

WECOM_CORP_ID=${WECOM_CORP_ID} WECOM_AGENT_ID=1000002 WECOM_AGENT_SECRET=${WECOM_AGENT_SECRET} WECOM_TOKEN=random_token_here WECOM_ENCODING_AES_KEY=base64_key_here

数据库

KB_DB_URL=postgresql://postgres@localhost/wecom_kb

LLM

LLM_PROVIDER=kimi LLM_API_KEY=${LLM_API_KEY} LLM_API_BASE=https://api.moonshot.cn/v1 LLM_MODEL=moonshot-v1-8k

知识库搜索

KB_SIMILARITY_THRESHOLD=0.7 KB_TOP_K=3

人工介入

NOTIFICATION_ENABLED=true NOTIFICATION_CHANNEL=telegram:8518085684

undefined

NOTIFICATION_ENABLED=true NOTIFICATION_CHANNEL=telegram:8518085684

undefined

工作流程详解

完整消息处理流程

mermaid

graph TD
    A[接收消息] --> B{是否为文本?}
    B -->|是| C[搜索知识库]
    B -->|否| D[其他类型处理]
    C --> E{相似度 > 阈值?}
    E -->|是| F[生成回答]
    E -->|否| G[转人工]
    F --> H[发送回复]
    G --> I[通知人工客服]
    D --> J[按类型处理]

mermaid

graph TD
    A[接收消息] --> B{是否为文本?}
    B -->|是| C[搜索知识库]
    B -->|否| D[其他类型处理]
    C --> E{相似度 > 阈值?}
    E -->|是| F[生成回答]
    E -->|否| G[转人工]
    F --> H[发送回复]
    G --> I[通知人工客服]
    D --> J[按类型处理]

数据流

python

undefined

python

undefined

1. 接收 Webhook

@app.post("/wecom/callback") async def wecom_callback(payload: WebhookPayload): event = payload.Event[0]

# 2. 路由事件
if event.Event == "add_external_contact":
    await handle_friend_add(event.UserId)
elif event.Event == "msg":
    await handle_message(event)

return {"errcode": 0}

@app.post("/wecom/callback") async def wecom_callback(payload: WebhookPayload): event = payload.Event[0]

# 2. 路由事件
if event.Event == "add_external_contact":
    await handle_friend_add(event.UserId)
elif event.Event == "msg":
    await handle_message(event)

return {"errcode": 0}

3. 处理消息

async def handle_message(event): user_id = event.FromUserName content = event.Content

# 搜索知识库
results = search_kb(content)

# 判断置信度
if results[0].score > CONFIDENCE_THRESHOLD:
    # 自动回复
    answer = generate_answer(results, content)
    send_message(user_id, answer)
else:
    # 转人工
    escalate_to_human(user_id, content)

undefined

async def handle_message(event): user_id = event.FromUserName content = event.Content

# 搜索知识库
results = search_kb(content)

# 判断置信度
if results[0].score > CONFIDENCE_THRESHOLD:
    # 自动回复
    answer = generate_answer(results, content)
    send_message(user_id, answer)
else:
    # 转人工
    escalate_to_human(user_id, content)

undefined

知识库管理

添加知识

bash

undefined

bash

undefined

方式 1：从 Markdown 导入

python3 scripts/import_kb.py
--input ~/clawd/knowledge/faq.md
--category "常见问题"

方式 2：直接插入数据库

psql wecom_kb INSERT INTO knowledge_chunks (content, metadata) VALUES ( '退货政策：7天无理由退货', '{"category": "售后", "tags": ["退货", "政策"]}' );

undefined

psql wecom_kb INSERT INTO knowledge_chunks (content, metadata) VALUES ( '退货政策：7天无理由退货', '{"category": "售后", "tags": ["退货", "政策"]}' );

undefined

更新知识

bash

undefined

bash

undefined

重新导入（自动去重）

python3 scripts/import_kb.py --input faq.md --refresh

undefined

python3 scripts/import_kb.py --input faq.md --refresh

undefined

测试搜索

bash

python3 scripts/search_kb.py "如何退款？"

bash

python3 scripts/search_kb.py "如何退款？"

监控与维护

日志查看

bash

undefined

bash

undefined

服务日志

tail -f /var/log/wecom-cs/server.log

数据库日志

tail -f /var/log/postgresql/postgresql-14-main.log

undefined

tail -f /var/log/postgresql/postgresql-14-main.log

undefined

性能监控

python

undefined

python

undefined

添加到 server/main.py

from prometheus_client import Counter, Histogram

message_counter = Counter('messages_total', 'Total messages') answer_latency = Histogram('answer_latency_seconds', 'Answer latency')

@answer_latency.time() def handle_message(): message_counter.inc() # ...

undefined

from prometheus_client import Counter, Histogram

message_counter = Counter('messages_total', 'Total messages') answer_latency = Histogram('answer_latency_seconds', 'Answer latency')

@answer_latency.time() def handle_message(): message_counter.inc() # ...

undefined

人工介入统计

sql

-- 查看未解决问题分布
SELECT
    COUNT(*) as count,
    SUBSTRING(content, 1, 30) as question_preview
FROM unknown_questions
GROUP BY question_preview
ORDER BY count DESC
LIMIT 10;

sql

-- 查看未解决问题分布
SELECT
    COUNT(*) as count,
    SUBSTRING(content, 1, 30) as question_preview
FROM unknown_questions
GROUP BY question_preview
ORDER BY count DESC
LIMIT 10;

安全最佳实践

密钥管理
- 所有密钥使用
```
pass
```
  存储
- 环境变量引用，不硬编码
数据隐私
- 客户信息加密存储
- 定期清理敏感日志
访问控制
- Webhook 验证签名
- IP 白名单限制
审计日志
- 记录所有人工介入
- 定期审查访问日志

密钥管理
- 所有密钥使用
```
pass
```
  存储
- 环境变量引用，不硬编码
数据隐私
- 客户信息加密存储
- 定期清理敏感日志
访问控制
- Webhook 验证签名
- IP 白名单限制
审计日志
- 记录所有人工介入
- 定期审查访问日志

故障排查

问题 1：回调接收不到消息

bash

undefined

bash

undefined

检查端口监听

ss -ltnp | grep 8000

检查 Nginx 配置（如有）

nginx -t

查看防火墙

sudo ufw status

undefined

sudo ufw status

undefined

问题 2：知识库搜索无结果

bash

undefined

bash

undefined

检查数据

psql wecom_kb -c "SELECT COUNT(*) FROM knowledge_chunks;"

测试搜索

python3 scripts/search_kb.py "测试查询"

重新向量化

python3 scripts/import_kb.py --rebuild

undefined

python3 scripts/import_kb.py --rebuild

undefined

问题 3：人工提醒未发送

bash

undefined

bash

undefined

测试 Telegram 连接

curl -X POST "https://api.telegram.org/bot$TELEGRAM_TOKEN/sendMessage"
-d "chat_id=8518085684&text=测试"

检查通知配置

cat .env | grep NOTIFICATION

undefined

cat .env | grep NOTIFICATION

undefined

扩展功能

1. 多轮对话记忆

python

undefined

python

undefined

使用 Redis 存储会话上下文

async def get_conversation_history(user_id): return redis.get(f"conv:{user_id}")

async def append_message(user_id, role, content): redis.rpush(f"conv:{user_id}", f"{role}:{content}")

undefined

async def get_conversation_history(user_id): return redis.get(f"conv:{user_id}")

async def append_message(user_id, role, content): redis.rpush(f"conv:{user_id}", f"{role}:{content}")

undefined

2. 情感分析

python

undefined

python

undefined

检测用户情绪

async def analyze_sentiment(text): result = openai.ChatCompletion.create( model="gpt-4", messages=[{ "role": "system", "content": "判断用户情绪（正面/负面/中性），只返回一个词。" }, { "role": "user", "content": text }] ) return result.choices[0].message.content

undefined

async def analyze_sentiment(text): result = openai.ChatCompletion.create( model="gpt-4", messages=[{ "role": "system", "content": "判断用户情绪（正面/负面/中性），只返回一个词。" }, { "role": "user", "content": text }] ) return result.choices[0].message.content

undefined

3. 主动营销

python

undefined

python

undefined

定期推送

async def daily_promotion(): users = get_active_users(days=7) for user_id in users: await wecom.send_text(user_id, "今日特惠：...")

undefined

async def daily_promotion(): users = get_active_users(days=7) for user_id in users: await wecom.send_text(user_id, "今日特惠：...")

undefined

wecom-cs-automation

Original

Translation

企业微信客服自动化系统

企业微信客服自动化系统

核心功能

核心功能

1. 自动同意好友添加

1. 自动同意好友添加

2. 基于知识库的智能问答

2. 基于知识库的智能问答

3. 未知问题人工介入

3. 未知问题人工介入

技术架构

技术架构

快速开始

快速开始

第一步：配置企业微信应用

第一步：配置企业微信应用

第二步：设置知识库

第二步：设置知识库

1. 安装 PostgreSQL + pgvector

1. 安装 PostgreSQL + pgvector

2. 创建数据库

2. 创建数据库

3. 初始化表结构

3. 初始化表结构

第三步：导入知识库数据

第三步：导入知识库数据

1. 准备知识文档（Markdown 格式）

1. 准备知识文档（Markdown 格式）

2. 切片并向量化

2. 切片并向量化

3. 验证导入

3. 验证导入

第四步：启动回调服务

第四步：启动回调服务

1. 配置环境变量

1. 配置环境变量

2. 启动服务（Python FastAPI）

2. 启动服务（Python FastAPI）

或使用 Go

或使用 Go

第五步：验证服务

第五步：验证服务

1. 检查服务状态

1. 检查服务状态

2. 测试知识库搜索

2. 测试知识库搜索

使用方法

使用方法

场景 1：自动欢迎新好友

场景 1：自动欢迎新好友

skills/wecom-cs-automation/workflows/on_friend_add.py

skills/wecom-cs-automation/workflows/on_friend_add.py

场景 2：知识库问答

场景 2：知识库问答

skills/wecom-cs-automation/workflows/answer_question.py

skills/wecom-cs-automation/workflows/answer_question.py

场景 3：人工介入提醒

场景 3：人工介入提醒

skills/wecom-cs-automation/workflows/escalate.py

skills/wecom-cs-automation/workflows/escalate.py

目录结构

目录结构

API 配置

API 配置

所需密钥

所需密钥

企业微信

企业微信

LLM（推荐 Kimi，中文优化）

LLM（推荐 Kimi，中文优化）

Telegram 通知（可选）

Telegram 通知（可选）

配置文件

配置文件

企业微信配置

企业微信配置

数据库