wecom-cs-automation

仓库: aaaaqwq/claude-code-skills
安装量: 46
排名: #16158

安装

npx skills add https://github.com/aaaaqwq/claude-code-skills --skill wecom-cs-automation
企业微信客服自动化系统
这是一个完整的企业微信客服 AI 助手解决方案，能够自动处理好友添加、智能问答、人工转接等场景。
核心功能
1. 自动同意好友添加
实时监听好友添加事件
自动通过好友请求
发送欢迎消息
标注用户来源和标签
2. 基于知识库的智能问答
向量知识库存储企业知识
语义搜索匹配问题
LLM 生成专业回复
支持多轮对话上下文
3. 未知问题人工介入
置信度阈值判断
自动提醒人工客服
转接对话给人工
记录未解决问题用于优化
技术架构
┌─────────────┐
│ 企业微信 │
│ 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: 自动生成 订阅所需事件 联系人变更事件 消息事件 外部联系人免验证添加事件 第二步：设置知识库

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 第三步：导入知识库数据

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;" 第四步：启动回调服务

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 第五步：验证服务

1. 检查服务状态

curl http://localhost:8000/health

2. 测试知识库搜索

curl -X POST http://localhost:8000/api/test_kb \ -H "Content-Type: application/json" \ -d '{"query": "如何退款？"}' 使用方法 场景 1：自动欢迎新好友 当用户添加客服为好友时：

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 ) 场景 2：知识库问答

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 ) 场景 3：人工介入提醒

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 ) 目录结构 ~/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 配置 所需密钥

企业微信

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

已有

配置文件 创建 ~/clawd/skills/wecom-cs-automation/.env ：

企业微信配置

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

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 }

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 ) 知识库管理 添加知识

方式 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": ["退货", "政策"]}' ) ; 更新知识

重新导入（自动去重）

python3 scripts/import_kb.py --input faq.md --refresh 测试搜索 python3 scripts/search_kb.py "如何退款？" 监控与维护 日志查看

服务日志

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

数据库日志

tail -f /var/log/postgresql/postgresql-14-main.log 性能监控

添加到 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 ( )

...

人工介入统计 -- 查看未解决问题分布 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 白名单限制 审计日志 记录所有人工介入 定期审查访问日志 故障排查 问题 1：回调接收不到消息

检查端口监听

ss -ltnp | grep 8000

检查 Nginx 配置（如有）

nginx -t

查看防火墙

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

检查数据

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

测试搜索

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

重新向量化

python3 scripts/import_kb.py --rebuild 问题 3：人工提醒未发送

测试 Telegram 连接

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

检查通知配置

cat .env | grep NOTIFICATION 扩展功能 1. 多轮对话记忆

使用 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 } " ) 2. 情感分析

检测用户情绪

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 3. 主动营销

定期推送

async
def
daily_promotion
(
)
:
users
=
get_active_users
(
days
=
7
)
for
user_id
in
users
:
await
wecom
.
send_text
(
user_id
,
"今日特惠：..."
)
相关技能
feishu-automation
飞书平台自动化
notion-automation
Notion 知识库集成
telegram-automation
Telegram 通知集成 参考资源 企业微信 API 文档 PostgreSQL pgvector FastAPI WebSocket Kimi API 文档
← 返回排行榜