简介这是一套面向企业客服场景的微信在线AI客服系统开源源码基于PHP开发可与企业微信客服集成帮助开发者与中小企业搭建7×24小时智能应答服务适合具备一定PHP基础、希望二次开发或私有化部署客服系统的技术人员。压缩包共43个文件以31个PHP源码文件为核心涵盖AI服务、会话管理、鉴权与接口逻辑另含3个HTML页面、3个TXT说明、1个Markdown功能文档及图标、配置文件等整体约20.58MB。系统支持上下文理解、产品知识库与FAQ优先应答、图片和视频内容分析并提供关键词触发转人工、后台一键介入等对话管理能力目录中api、admin、includes、media、logs等模块划分清晰便于按功能定位代码。目前已有144人学习下载可作为智能客服项目落地与二次开发的参考实现。1. 从零搭一套微信在线 AI 客服这套源码到底能省掉哪些重复活如果你正在做私域运营、SaaS 工具或者企业内部系统大概率绕不开一个需求把微信生态里的用户咨询接进来先用 AI 挡一轮挡不住的再转人工。市面上现成的 SaaS 客服按坐席按月收费量一大成本就压不住自己从零写光是微信消息加解密、多客服分配、会话上下文管理这几块就够折腾两周。这套 2026 最新微信在线 AI 客服系统开源源码解决的就是这个从 0 到 1 的重复建设问题——它把微信侧的接入层、AI 对话层、人工坐席调度层和后台管理界面都串好了你拿到手改的是业务逻辑和模型配置不是通信底层。它适合三类人一是想快速验证 AI 客服产品形态的独立开发者二是需要私有化部署、数据不能出内网的企业技术团队三是手上已有大模型 API、只缺一套微信接入壳子的后端工程师。不适合完全没碰过服务端部署的新手直接上生产因为里面涉及回调验签、消息队列和数据库得有点后端底子。下面我按「这套东西怎么跑起来 → 关键模块怎么改 → 哪里最容易翻车」的顺序拆一遍都是能直接抄的步骤。2. 环境准备与首次启动把服务在本地跑通2.1 技术栈确认与依赖安装拿到源码包先别急着改代码第一步是确认技术栈和本机环境对不对得上。这类微信 AI 客服系统常见做法是后端用 PythonFastAPI 或 Flask Redis MySQL/PostgreSQL前端管理后台用 Vue 或 ReactAI 层通过 HTTP 调大模型接口。先看项目根目录的requirements.txt或package.json确认版本约束。# 以 Python 后端为例创建独立虚拟环境避免污染全局包 python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 安装依赖建议加国内镜像加速 pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple # 前端管理后台依赖 cd admin-web npm install --registryhttps://registry.npmmirror.com逻辑说明虚拟环境是必须的因为客服系统往往依赖特定版本的cryptography微信消息加解密用和redis客户端跟本机其他项目冲突是血泪经验。参数上-i指定镜像源只影响下载速度不改包内容。如果requirements.txt里锁了python_version用pyenv切到对应版本再装否则cryptography编译会报错。2.2 配置文件与数据库初始化依赖装完接下来是配置。这类项目一般有个.env.example或config.yaml需要你填数据库连接、Redis 地址、微信 AppID/Secret、大模型 API Key。复制一份改成自己的cp .env.example .env# .env 关键项说明 DB_HOST127.0.0.1 DB_PORT3306 DB_NAMEai_customer_service DB_USERroot DB_PASSWORDyour_password REDIS_HOST127.0.0.1 REDIS_PORT6379 REDIS_DB0 WECHAT_APPIDwx开头的一串 WECHAT_SECRET公众号或小程序的密钥 WECHAT_TOKEN自定义令牌填什么后台就配什么 WECHAT_AES_KEY43位随机字符串 LLM_API_BASEhttps://你的模型服务地址/v1 LLM_API_KEYsk-xxxx LLM_MODELqwen-plus逻辑说明WECHAT_TOKEN和WECHAT_AES_KEY是微信服务器验证回调时用的必须和微信公众平台后台「服务器配置」里填的完全一致差一个字符就验签失败。LLM_API_BASE指向你的模型服务兼容 OpenAI 格式的接口都能接。参数上REDIS_DB建议单独开一个库别和别的业务混用不然后期清缓存容易误伤。数据库初始化一般项目会带init.sql或迁移脚本# 建库 mysql -u root -p -e CREATE DATABASE ai_customer_service DEFAULT CHARSET utf8mb4; # 导入表结构 mysql -u root -p ai_customer_service init.sqlutf8mb4是必须的微信昵称里有 emoji用utf8存进去会变问号这个坑我踩过不止一次。2.3 启动服务与回调验证配置就绪后启动后端# 开发模式启动带热重载 uvicorn main:app --host 0.0.0.0 --port 8000 --reload启动成功后微信公众平台的服务器配置需要填一个公网可访问的 URL。本地开发常见做法是用内网穿透工具把 8000 端口映射出去拿到一个临时域名填进微信后台。填完后微信会发一个 GET 请求做验签你的服务要能正确返回echostr。如果后台提示「token 验证失败」九成是WECHAT_TOKEN对不上或者服务没起来。提示验签逻辑在源码的wechat/callback.py里核心是hashlib.sha1对 token、timestamp、nonce 排序后拼接再摘要跟微信传的 signature 比对。调试时把这三个值和算出来的 signature 打日志一眼就能看出问题。3. 消息流转与 AI 对话链路核心模块怎么改3.1 微信消息接收与解密流程微信推送到你服务器的消息是加密的源码里一般封装了解密函数。理解这条链路对排查「AI 不回复」至关重要。流程是微信 POST 加密 XML → 你的服务用 AESKey 解密 → 解析出用户 OpenID、消息类型、内容 → 交给业务层。# 消息解密核心逻辑简化示意实际以源码为准 import base64 from Crypto.Cipher import AES def decrypt_message(encrypted, aes_key, appid): # aes_key 是 43 位补一个 做 base64 解码得到 32 字节密钥 key base64.b64decode(aes_key ) iv key[:16] # 初始向量取密钥前 16 字节 cipher AES.new(key, AES.MODE_CBC, iv) decrypted cipher.decrypt(base64.b64decode(encrypted)) # 去 PKCS7 填充 pad decrypted[-1] content decrypted[:-pad].decode(utf-8) # 末尾 16 字节是 appid校验防止串号 if not content.endswith(appid): raise ValueError(appid 不匹配可能是配置串了) return content[:-len(appid)]逻辑说明aes_key后面补是因为微信给的 EncodingAESKey 是 43 位base64 解码需要长度是 4 的倍数。iv取密钥前 16 字节是微信规定的不是随便设。最后校验 appid 很关键——如果你同时接了多个公众号配置串了会导致 A 号的消息被 B 号处理这个 bug 隐蔽性极强。参数上MODE_CBC和 PKCS7 填充都是微信协议固定的不能改。3.2 接入大模型实现自动回复消息解析出来后业务层判断如果当前会话没有人工坐席接入就丢给 AI。源码里通常有个ai_service.py或类似模块核心是把用户消息 历史上下文拼成 prompt 发给模型。# AI 回复生成以兼容 OpenAI 格式的接口为例 import httpx async def generate_reply(user_message, history, config): messages [ {role: system, content: 你是微信客服助手回答简洁不确定时引导转人工。} ] # 只带最近 5 轮上下文防止 token 超限 messages.extend(history[-10:]) messages.append({role: user, content: user_message}) async with httpx.AsyncClient(timeout30) as client: resp await client.post( f{config.LLM_API_BASE}/chat/completions, headers{Authorization: fBearer {config.LLM_API_KEY}}, json{ model: config.LLM_MODEL, messages: messages, temperature: 0.3, # 客服场景要稳别太发散 max_tokens: 500 } ) return resp.json()[choices][0][message][content]逻辑说明history[-10:]是只取最近 10 条消息约 5 轮对话因为客服场景上下文太长既费 token 又容易让模型跑偏。temperature设 0.3 而不是默认的 0.7是因为客服回答要稳定可控太高会出现答非所问的玄学输出。timeout30是防止模型服务卡住导致微信那边超时重试重试多了用户会收到重复回复。参数上max_tokens限制回复长度微信消息太长体验差500 够用了。3.3 人工坐席接管与消息分流AI 不是万能的用户一句「转人工」或者 AI 连续两轮没解决就该切人工。源码里一般用 Redis 存会话状态标记session_mode是ai还是human。# 会话状态管理 import redis r redis.Redis(host127.0.0.1, port6379, db0) def should_transfer_to_human(openid, user_message, ai_reply): # 用户主动要求转人工 if 转人工 in user_message or 人工客服 in user_message: return True # AI 回复里包含兜底话术说明它没把握 if 抱歉 in ai_reply and 无法 in ai_reply: return True # 检查该用户是否已被人工接管 mode r.get(fsession_mode:{openid}) return mode bhuman def set_session_mode(openid, mode): # 人工接管状态保留 30 分钟超时自动回 AI r.setex(fsession_mode:{openid}, 1800, mode)逻辑说明setex的 1800 秒是人工会话的保鲜期坐席下班或忘记切回时30 分钟后自动回落到 AI避免用户发消息没人理。参数上这个时间可以根据你的坐席在线时段调整白天短一点、夜间长一点都行。分流逻辑要放在 AI 回复之前判断否则用户说了「转人工」还先收到一条 AI 回复体验很割裂。4. 后台管理与数据落库让客服记录可查可追4.1 会话记录表结构与写入客服系统跑起来只是第一步能不能查到历史会话、能不能统计 AI 解决率取决于数据落库设计。源码里一般有conversations和messages两张核心表。-- 会话表一个用户一个会话周期 CREATE TABLE conversations ( id BIGINT PRIMARY KEY AUTO_INCREMENT, openid VARCHAR(64) NOT NULL, mode VARCHAR(10) DEFAULT ai, -- ai 或 human agent_id INT DEFAULT NULL, -- 接管的人工坐席 ID created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_openid (openid) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 消息表每条收发消息一行 CREATE TABLE messages ( id BIGINT PRIMARY KEY AUTO_INCREMENT, conversation_id BIGINT NOT NULL, role VARCHAR(10) NOT NULL, -- user / ai / agent content TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_conv (conversation_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;逻辑说明conversations和messages拆开是为了避免单表膨胀——一个活跃用户一天可能几十条消息全塞一张表查询会越来越慢。mode字段记录当前是 AI 还是人工统计 AI 解决率时直接按这个字段分组就行。content用TEXT不用VARCHAR因为用户可能发长文本VARCHAR(255)会截断。索引建在openid和conversation_id上后台按用户查会话时走索引不然数据量上万后查询会明显卡顿。4.2 管理后台的坐席分配逻辑后台管理界面通常包含坐席列表、会话列表、手动接管按钮。坐席分配常见做法是轮询或按当前会话数最少分配。# 简单轮询分配坐席 def assign_agent(): agents get_online_agents() # 从 Redis 或数据库取在线坐席 if not agents: return None # 用 Redis 计数器做轮询保证分配均匀 idx r.incr(agent_round_robin) % len(agents) return agents[idx]逻辑说明incr是原子操作多个请求同时进来不会分配重复。参数上get_online_agents的在线判断一般靠坐席端心跳心跳超时 60 秒就认为离线。如果坐席少、咨询量大轮询会导致每个人同时接好几个会话这时候要加一个「当前会话数上限」判断超过阈值就不再分配新会话让用户排队或继续由 AI 接待。4.3 微信消息推送与模板消息有些场景需要主动给用户发消息比如工单处理完成通知。微信生态里这属于模板消息或订阅消息源码里一般封装了发送接口。# 发送模板消息以公众号为例 async def send_template_message(openid, template_id, data): # 先拿 access_token注意缓存微信限制每天获取次数 token await get_access_token() url fhttps://api.weixin.qq.com/cgi-bin/message/template/send?access_token{token} payload { touser: openid, template_id: template_id, data: data } async with httpx.AsyncClient() as client: resp await client.post(url, jsonpayload) return resp.json()逻辑说明access_token必须缓存微信规定有效期 7200 秒但频繁获取会触发限流。常见做法是存 Redis设 7000 秒过期快到期时刷新。参数上template_id要在公众号后台申请不同行业模板不一样测试时用沙箱环境的模板 ID。如果返回errcode: 40001就是 token 失效或没缓存好重新获取即可。5. 避坑与排查这几处翻车点我替你踩过了5.1 回调验签一直失败现象微信后台保存服务器配置时提示「token 验证失败」服务日志里看不到请求进来。原因三种可能——服务没监听公网、WECHAT_TOKEN和后台填的不一致、验签逻辑里参数排序写错。最常见的是内网穿透工具挂了微信请求根本没到你的服务。解决先在浏览器直接访问你的回调 URL 加?echostrtest看服务有没有响应。有响应再对 token把微信传来的signature、timestamp、nonce和本地算的对比打日志。排序必须是字典序不是按参数名长度。5.2 AI 回复重复发送现象用户收到两条一模一样的 AI 回复。原因模型响应超过微信要求的 5 秒微信判定超时后重试推送你的服务又处理了一遍。解决在消息处理入口加去重用MsgId做 Redis 锁处理过的消息 5 分钟内不再处理。同时把模型调用超时设短一点或者先回一个「正在思考」的占位异步再推结果。5.3 中文乱码或 emoji 丢失现象数据库里用户昵称显示成问号或者 AI 回复里的 emoji 变成方块。原因数据库、表、连接字符集没统一成utf8mb4。解决建库建表都用utf8mb4连接串加charsetutf8mb4Python 里确保decode(utf-8)。三层缺一层都会出问题这个坑排查起来最费时间建议一开始就定死。5.4 人工接管后 AI 还在抢答现象坐席已经接管会话用户发消息还是先收到 AI 回复。原因分流判断放在了 AI 回复之后或者 Redis 里session_mode没设成功。解决把should_transfer_to_human的判断提到 AI 调用之前先查 Redis 状态再决定走哪条路。检查set_session_mode有没有被调用坐席点击「接管」时就要立刻写入状态不能等用户下一条消息才写。5.5 模型接口超时导致消息丢失现象用户发了消息既没收到 AI 回复也没转人工像石沉大海。原因模型接口超时抛异常代码没做兜底消息处理中断。解决AI 调用外面包一层 try/except超时或报错时返回一句兜底话术并触发转人工。常见做法是设 8 秒超时超过就放弃 AI 直接转人工宁可人工慢一点也不能让用户干等。6. 进阶把 AI 客服接进企业微信与多模型切换跑通基础版之后很多人会想再往前一步能不能同时接企业微信能不能在多个模型之间切换做效果对比这两个需求这套源码的架构都留了口子但需要你自己补一点胶水代码。先说企业微信接入。企业微信的消息加解密协议和公众号基本一致区别在于回调 URL 的验证参数多了AgentID以及发送消息的接口地址不同。常见做法是抽一个WeChatAdapter基类把decrypt、encrypt、send_message定义成抽象方法公众号和企业微信各实现一个子类。这样业务层不用改换适配器就行。配置上企业微信的CorpID对应公众号的AppIDSecret用应用的 Secret 而不是通讯录的填错了会报60020错误。# 适配器模式示意 class WeChatAdapter: def decrypt(self, raw): raise NotImplementedError def send_message(self, openid, content): raise NotImplementedError class MPAdapter(WeChatAdapter): def decrypt(self, raw): return decrypt_message(raw, self.aes_key, self.appid) def send_message(self, openid, content): return send_customer_service_msg(openid, content) class WorkWeChatAdapter(WeChatAdapter): def decrypt(self, raw): return decrypt_work_message(raw, self.aes_key, self.corp_id) def send_message(self, openid, content): return send_work_msg(openid, content, self.agent_id)多模型切换更简单把LLM_MODEL和LLM_API_BASE做成后台可配的每个模型存一条记录加个权重字段做灰度。我一般会同时挂两个模型一个主力一个备用主力超时就自动切备用这样单点故障不至于让客服瘫痪。验证方法上别只看「能回复」就完事。我会跑三个指标一是 AI 首响时间从收到消息到发出回复的毫秒数超过 3 秒就要优化二是转人工率如果超过 40% 说明 prompt 或模型选型有问题三是重复消息率正常应该为 0。这三个数在后台加个统计页就能看比拍脑袋判断靠谱得多。从那以后我每次部署这类客服系统都强制先跑一遍「发消息 → 收回复 → 查库 → 看日志」的闭环确认四个环节都通再接真实流量。希望帮到你。本文还有配套的精品资源点击获取