尧图网站建设 
简介这是一套基于PHP开发的微信在线AI客服系统源码面向需要为企业微信搭建7×24小时智能客服的中小团队与个人开发者尤其适合具备一定PHP基础、希望快速落地客服机器人的技术人员。系统集成企业微信客服支持文本对话、图片分析与视频分析等多种交互方式并内置对话管理、人工转接、咨询提醒等高级功能AI无法处理时可平滑转接人工保障服务质量。压缩包共38个文件约20.57MB以31个PHP文件为主体涵盖微信交互、AI算法逻辑与对话流程管理等模块另含md说明文档、html页面、jpg图片及gitignore、htaccess等配置辅助文件目录结构清晰便于按模块二次开发与参数调整。目前已有65人学习下载。源码模块化程度较高开发者可参照说明文档快速完成部署并根据业务场景定制回复模板与对话策略适合作为企业客服智能化改造的起步方案。1. 微信在线AI客服系统到底在解决什么从人工排队到秒回微信生态里的客服长期处在一个尴尬位置用户消息进来人工客服要么忙不过来要么下班了没人接第二天再回复客户早跑了。2026 年做微信在线 AI 客服系统源码核心要解决的就是这件事——把微信消息通道和 AI 对话能力接起来让用户发来的每一条消息都能被即时理解、即时响应同时把复杂问题转给人工。这套系统适合谁做私域运营的团队、有微信小程序或公众号的中小企业、想给现有客服系统加 AI 能力的开发者。它不是一个开箱即用的成品软件而是一套源码工程你需要自己部署服务端、配置微信侧的接入参数、接上大模型 API。源码的价值在于可控消息怎么存、AI 怎么调、转人工的阈值怎么设全在你手里。接下来我会按「先跑通最小链路再补工程细节最后说坑」的顺序把这条路走一遍。2. 微信消息通道怎么接公众号、小程序与客服消息三条路2.1 三种接入方式的选型对比微信侧能接收用户消息的入口主要有三个选错了后面全是返工。接入方式适用场景消息类型接入门槛公众号被动回复用户主动发消息文本、图片、语音需认证服务号小程序客服消息小程序内联系客服文本、图片、卡片小程序需开通客服企业微信会话存档员工与客户对话全类型需企业认证公众号被动回复是最常见的起点用户在对话框发消息微信服务器把 XML 推到你的回调地址你在 5 秒内返回响应。小程序客服消息走的是另一套接口消息通过微信服务器的 JSON 推送过来需要先调用customservice相关接口获取会话。企业微信会话存档适合已经有员工在用企微跟客户聊的场景能拿到完整对话记录做 AI 分析。我一般建议先从公众号被动回复跑通因为它的调试链路最短微信官方有测试号可以直接用不用等认证。2.2 配置回调地址与验证签名公众号后台配置服务器地址时微信会先发一个 GET 请求做签名验证。这一步的代码必须写对否则后面所有消息都收不到。import hashlib def check_signature(token, signature, timestamp, nonce): # 将 token、timestamp、nonce 三个参数按字典序排序 params sorted([token, timestamp, nonce]) # 拼接成一个字符串后进行 sha1 加密 raw .join(params) sha1 hashlib.sha1(raw.encode(utf-8)).hexdigest() # 与微信传来的 signature 比对 return sha1 signature逻辑说明微信服务器会把signature、timestamp、nonce、echostr四个参数以 GET 方式发到你的回调地址。你用自己的token在公众号后台设置加上后两个参数做同样的 sha1 计算结果一致就原样返回echostr验证通过。参数说明token是你自己在公众号后台填的任意字符串不是微信分配的但要和代码里保持一致。timestamp和nonce是微信每次请求随机生成的不需要存储。注意signature比对要用常量时间比较避免时序攻击生产环境可以用hmac.compare_digest。2.3 接收消息并解析 XML 体验证通过后用户发的每条消息都会以 POST XML 形式推到你的地址。解析这个 XML 是接 AI 的第一步。import xml.etree.ElementTree as ET def parse_wechat_message(xml_body): # 解析微信推送的 XML 消息体 root ET.fromstring(xml_body) msg { to_user: root.find(ToUserName).text, # 公众号原始ID from_user: root.find(FromUserName).text, # 用户OpenID create_time: root.find(CreateTime).text, msg_type: root.find(MsgType).text, # text/image/voice等 content: root.find(Content).text if root.find(Content) is not None else , msg_id: root.find(MsgId).text } return msg逻辑说明FromUserName是用户的 OpenID这是你在微信侧唯一能标识用户的东西后续做会话管理、查历史记录都靠它。MsgType决定你后面怎么处理——文本直接送 AI图片和语音需要先转成文字。参数说明Content字段只在文本消息里存在图片消息里没有所以要做空判断。MsgId是微信侧的消息 ID可以用来做去重防止微信重试导致重复回复。提示微信服务器要求 5 秒内返回响应如果 AI 接口响应慢先返回空串或「正在思考」的提示再用客服消息接口异步推送结果。3. AI 对话引擎怎么搭从 Prompt 到多轮上下文管理3.1 大模型 API 的接入与流式输出拿到用户消息后下一步是调大模型。2026 年主流做法是走 OpenAI 兼容接口国内多家厂商都支持这个格式切换成本低。import requests def call_llm(user_message, history, api_key, base_url): # 构造对话历史system prompt 定义客服角色 messages [ {role: system, content: 你是一名微信在线客服回答要简洁、准确不确定的问题引导用户转人工。} ] # 追加历史对话保持多轮上下文 messages.extend(history) messages.append({role: user, content: user_message}) resp requests.post( f{base_url}/v1/chat/completions, headers{Authorization: fBearer {api_key}}, json{ model: gpt-4o-mini, # 按实际可用模型替换 messages: messages, temperature: 0.3, # 客服场景降低随机性 max_tokens: 500 }, timeout10 ) return resp.json()[choices][0][message][content]逻辑说明system消息定义 AI 的角色边界这一步直接决定回复风格。history是之前几轮的对话记录按[{role:user,content:...},{role:assistant,content:...}]格式传入。temperature设低一些客服场景不需要创意需要稳定。参数说明max_tokens控制回复长度微信消息太长用户不会看500 足够。timeout设 10 秒超过就降级到人工。base_url和api_key从环境变量读不要硬编码在源码里。3.2 多轮对话的上下文窗口与截断策略多轮对话不能无限往 messages 里塞token 有上限成本也在涨。常见做法是保留最近 N 轮或者按 token 数截断。def trim_history(history, max_turns6): # 只保留最近 max_turns 轮对话每轮包含 user 和 assistant 两条 if len(history) max_turns * 2: return history[-(max_turns * 2):] return history逻辑说明max_turns6意味着保留最近 6 轮问答大约 12 条消息。这个数字不是固定的取决于你的模型上下文窗口和业务复杂度。售前咨询通常 3 到 4 轮就够售后排查可能需要 8 到 10 轮。参数说明如果对话涉及订单号、手机号等关键信息不要只靠截断应该把关键实体抽出来存到会话状态里每次请求时重新注入 system prompt。3.3 意图识别与转人工的触发条件AI 不是万能的什么时候转人工需要明确规则。我一般设三个触发条件用户明确说「转人工」、AI 连续两轮回答置信度低、用户情绪检测为负面。def should_transfer_to_human(user_message, ai_reply, history): # 条件一用户明确要求转人工 if any(kw in user_message for kw in [转人工, 人工客服, 找真人]): return True # 条件二AI 回复中包含不确定表述 if any(kw in ai_reply for kw in [不确定, 不清楚, 建议咨询]): return True # 条件三连续两轮用户重复同一问题 if len(history) 4: last_user history[-2][content] if history[-2][role] user else if last_user and last_user.strip() user_message.strip(): return True return False逻辑说明这三个条件覆盖了大部分需要人工介入的场景。条件二的关键词列表可以根据你的业务补充比如「退款」「投诉」这类词也应该触发转人工。参数说明转人工后AI 应该停止自动回复把会话标记为「人工接管中」同时把之前的对话摘要推给人工客服避免用户重复描述问题。4. 会话存储与用户管理OpenID 映射与消息落库4.1 用 OpenID 做用户唯一标识微信侧拿不到用户的手机号除非用户主动授权OpenID 是唯一稳定的用户标识。同一个用户在不同公众号下的 OpenID 不同如果有多公众号需要用 UnionID 做跨号统一。CREATE TABLE wechat_user ( id BIGINT PRIMARY KEY AUTO_INCREMENT, openid VARCHAR(64) NOT NULL UNIQUE, unionid VARCHAR(64) DEFAULT NULL, nickname VARCHAR(128) DEFAULT NULL, first_seen DATETIME DEFAULT CURRENT_TIMESTAMP, last_active DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_unionid (unionid) ); CREATE TABLE chat_message ( id BIGINT PRIMARY KEY AUTO_INCREMENT, openid VARCHAR(64) NOT NULL, role ENUM(user, assistant, human) NOT NULL, content TEXT NOT NULL, msg_id VARCHAR(64) DEFAULT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_openid_time (openid, created_at) );逻辑说明wechat_user表存用户基本信息openid加唯一索引防止重复插入。chat_message表存所有对话记录role字段区分是用户发的、AI 回的、还是人工客服回的。idx_openid_time联合索引让「查某个用户最近的消息」这个高频查询走索引。参数说明content用 TEXT 类型微信消息最长 2048 字节TEXT 足够。如果要做全文检索可以加 FULLTEXT 索引或外接搜索引擎。4.2 会话上下文的读写与缓存每次调 AI 都要读历史消息如果每次都查数据库QPS 一高就扛不住。常见做法是用 Redis 缓存最近几轮对话。import redis import json r redis.Redis(hostlocalhost, port6379, db0) def get_history(openid, max_turns6): # 从 Redis 列表读取最近对话key 按用户维度隔离 key fchat:history:{openid} raw r.lrange(key, -max_turns * 2, -1) return [json.loads(item) for item in raw] def append_history(openid, role, content): # 追加一条对话到 Redis并裁剪到固定长度 key fchat:history:{openid} r.rpush(key, json.dumps({role: role, content: content})) r.ltrim(key, -20, -1) # 只保留最近 20 条 r.expire(key, 3600 * 24) # 24 小时过期逻辑说明用 Redis List 存对话rpush追加lrange读取。ltrim保证列表不会无限增长expire保证不活跃用户的缓存自动清理。参数说明max_turns和ltrim的数值要配合读取时取最近 12 条存储时保留 20 条留一些余量。过期时间 24 小时是经验值超过这个时间用户大概率是新问题不需要旧上下文。4.3 消息落库与异步写入Redis 是缓存数据库才是持久化。但每次对话都同步写库会拖慢响应我一般用异步队列。from celery import Celery app Celery(tasks, brokerredis://localhost:6379/1) app.task def save_message_async(openid, role, content, msg_idNone): # 异步写入 MySQL不阻塞主流程 conn get_db_connection() cursor conn.cursor() cursor.execute( INSERT INTO chat_message (openid, role, content, msg_id) VALUES (%s, %s, %s, %s), (openid, role, content, msg_id) ) conn.commit() cursor.close() conn.close()逻辑说明主流程只写 Redis 并返回响应落库交给 Celery 异步做。这样即使数据库短暂不可用用户侧也不会感知到延迟。参数说明broker用 Redis 的另一个 db和缓存分开避免互相影响。如果量不大也可以直接用线程池做异步不一定上 Celery。5. 避坑与排查微信 AI 客服上线后最容易翻车的五件事5.1 回调地址验证通过但收不到消息现象公众号后台显示「服务器配置成功」但用户发消息没有任何反应。原因微信验证时发的是 GET 请求实际消息推送是 POST。很多框架默认只处理 GET 或只处理 POST导致验证过了但消息进不来。解决确认你的路由同时接受 GET 和 POST。GET 走签名验证逻辑POST 走消息解析逻辑。另外检查服务器防火墙是否放行了微信服务器 IP 段微信官方文档有公布 IP 列表。5.2 AI 回复超时导致微信重试现象用户收到重复回复或者回复延迟很高。原因微信要求 5 秒内响应大模型 API 偶尔超过这个时间。微信没收到响应会重试三次每次间隔 5 秒导致同一条消息被处理多次。解决收到消息后立即返回空串微信不会展示空回复然后用客服消息接口异步推送 AI 结果。同时在处理消息前用MsgId做去重已经处理过的直接跳过。5.3 多轮对话串号现象A 用户的对话历史出现在 B 用户的会话里。原因会话 key 设计有问题比如用了公众号 ID 而不是用户 OpenID 做 key或者 Redis 连接池串了 db。解决会话 key 必须包含 OpenID格式如chat:history:{openid}。如果用了多进程或多线程确认 Redis 连接是每个请求独立获取的不要用全局单例。5.4 转人工后 AI 还在自动回复现象用户已经和人工客服聊上了AI 还在插话。原因转人工状态没有持久化或者状态存在内存里多实例部署时不同实例状态不一致。解决转人工状态存 Rediskey 如chat:human:{openid}设置合理的过期时间比如 30 分钟无交互自动释放。每次 AI 回复前先检查这个 key 是否存在。5.5 敏感词过滤缺失导致合规风险现象AI 回复了不该回复的内容或者用户输入了违规内容被原样回显。原因没有做输入输出双向过滤。解决在用户消息进 AI 之前做一次敏感词检测AI 回复发出之前再做一次。敏感词库可以用开源的也可以自己维护。检测到敏感词时直接返回预设的安全话术并记录日志。6. 让 AI 客服更懂业务知识库注入与效果验证6.1 用 RAG 把产品文档喂给 AI通用大模型不知道你的产品细节直接问「你们支持退货吗」它只能瞎编。常见做法是 RAG把产品文档、FAQ、历史工单切片存到向量库用户提问时先检索相关片段拼到 prompt 里。def build_rag_prompt(user_question, vector_store, top_k3): # 检索最相关的文档片段 docs vector_store.similarity_search(user_question, ktop_k) context \n.join([doc.page_content for doc in docs]) # 把检索结果拼到 system prompt 里 system_prompt f你是一名微信在线客服。请根据以下资料回答用户问题 {context} 如果资料中没有相关信息请引导用户转人工不要编造。 return system_prompt逻辑说明similarity_search返回和问题最相关的 top_k 个文档片段拼到 system prompt 里。这样 AI 的回答有据可依不会胡编。参数说明top_k3是经验值太多会挤占上下文窗口太少可能漏掉关键信息。文档切片大小建议 300 到 500 字太大检索精度下降太小上下文不完整。6.2 用日志和人工抽检验证回复质量上线后不能只看「有没有回复」要看「回复得对不对」。我一般做两件事一是把 AI 回复和人工回复做对比抽检二是监控转人工率。-- 统计每天的转人工率超过 30% 说明 AI 回答质量有问题 SELECT DATE(created_at) AS day, COUNT(DISTINCT openid) AS total_users, SUM(CASE WHEN role human THEN 1 ELSE 0 END) AS human_count, ROUND(SUM(CASE WHEN role human THEN 1 ELSE 0 END) / COUNT(*), 3) AS transfer_rate FROM chat_message GROUP BY DATE(created_at) ORDER BY day DESC;逻辑说明转人工率是 AI 客服最核心的指标。如果超过 30%说明要么知识库覆盖不够要么 prompt 写得不好要么转人工规则太敏感。参数说明这个查询按天聚合transfer_rate是人工消息占总消息的比例。实际看的时候要结合业务售前咨询转人工率天然比售后低。6.3 一个我踩过的坑别让 AI 替你做承诺最后说一个血泪教训。早期版本里AI 为了「显得有用」会主动说「可以退货」「三天内到账」这类话。结果用户截图来找客服兑现人工根本不知道这回事。后来我在 system prompt 里加了一条硬规则任何涉及金额、时效、政策的承诺必须回复「具体以人工客服确认为准」。这条规则救了我好几次。AI 客服的边界是「解答已知问题」不是「替公司做决定」。这个习惯我一直保持到现在希望帮到你。本文还有配套的精品资源点击获取 微信在线AI客服系统源码实战：从消息接入到多轮对话与转人工
简介这是一套基于PHP开发的微信在线AI客服系统源码面向需要为企业微信搭建7×24小时智能客服的中小团队与个人开发者尤其适合具备一定PHP基础、希望快速落地客服机器人的技术人员。系统集成企业微信客服支持文本对话、图片分析与视频分析等多种交互方式并内置对话管理、人工转接、咨询提醒等高级功能AI无法处理时可平滑转接人工保障服务质量。压缩包共38个文件约20.57MB以31个PHP文件为主体涵盖微信交互、AI算法逻辑与对话流程管理等模块另含md说明文档、html页面、jpg图片及gitignore、htaccess等配置辅助文件目录结构清晰便于按模块二次开发与参数调整。目前已有65人学习下载。源码模块化程度较高开发者可参照说明文档快速完成部署并根据业务场景定制回复模板与对话策略适合作为企业客服智能化改造的起步方案。1. 微信在线AI客服系统到底在解决什么从人工排队到秒回微信生态里的客服长期处在一个尴尬位置用户消息进来人工客服要么忙不过来要么下班了没人接第二天再回复客户早跑了。2026 年做微信在线 AI 客服系统源码核心要解决的就是这件事——把微信消息通道和 AI 对话能力接起来让用户发来的每一条消息都能被即时理解、即时响应同时把复杂问题转给人工。这套系统适合谁做私域运营的团队、有微信小程序或公众号的中小企业、想给现有客服系统加 AI 能力的开发者。它不是一个开箱即用的成品软件而是一套源码工程你需要自己部署服务端、配置微信侧的接入参数、接上大模型 API。源码的价值在于可控消息怎么存、AI 怎么调、转人工的阈值怎么设全在你手里。接下来我会按「先跑通最小链路再补工程细节最后说坑」的顺序把这条路走一遍。2. 微信消息通道怎么接公众号、小程序与客服消息三条路2.1 三种接入方式的选型对比微信侧能接收用户消息的入口主要有三个选错了后面全是返工。接入方式适用场景消息类型接入门槛公众号被动回复用户主动发消息文本、图片、语音需认证服务号小程序客服消息小程序内联系客服文本、图片、卡片小程序需开通客服企业微信会话存档员工与客户对话全类型需企业认证公众号被动回复是最常见的起点用户在对话框发消息微信服务器把 XML 推到你的回调地址你在 5 秒内返回响应。小程序客服消息走的是另一套接口消息通过微信服务器的 JSON 推送过来需要先调用customservice相关接口获取会话。企业微信会话存档适合已经有员工在用企微跟客户聊的场景能拿到完整对话记录做 AI 分析。我一般建议先从公众号被动回复跑通因为它的调试链路最短微信官方有测试号可以直接用不用等认证。2.2 配置回调地址与验证签名公众号后台配置服务器地址时微信会先发一个 GET 请求做签名验证。这一步的代码必须写对否则后面所有消息都收不到。import hashlib def check_signature(token, signature, timestamp, nonce): # 将 token、timestamp、nonce 三个参数按字典序排序 params sorted([token, timestamp, nonce]) # 拼接成一个字符串后进行 sha1 加密 raw .join(params) sha1 hashlib.sha1(raw.encode(utf-8)).hexdigest() # 与微信传来的 signature 比对 return sha1 signature逻辑说明微信服务器会把signature、timestamp、nonce、echostr四个参数以 GET 方式发到你的回调地址。你用自己的token在公众号后台设置加上后两个参数做同样的 sha1 计算结果一致就原样返回echostr验证通过。参数说明token是你自己在公众号后台填的任意字符串不是微信分配的但要和代码里保持一致。timestamp和nonce是微信每次请求随机生成的不需要存储。注意signature比对要用常量时间比较避免时序攻击生产环境可以用hmac.compare_digest。2.3 接收消息并解析 XML 体验证通过后用户发的每条消息都会以 POST XML 形式推到你的地址。解析这个 XML 是接 AI 的第一步。import xml.etree.ElementTree as ET def parse_wechat_message(xml_body): # 解析微信推送的 XML 消息体 root ET.fromstring(xml_body) msg { to_user: root.find(ToUserName).text, # 公众号原始ID from_user: root.find(FromUserName).text, # 用户OpenID create_time: root.find(CreateTime).text, msg_type: root.find(MsgType).text, # text/image/voice等 content: root.find(Content).text if root.find(Content) is not None else , msg_id: root.find(MsgId).text } return msg逻辑说明FromUserName是用户的 OpenID这是你在微信侧唯一能标识用户的东西后续做会话管理、查历史记录都靠它。MsgType决定你后面怎么处理——文本直接送 AI图片和语音需要先转成文字。参数说明Content字段只在文本消息里存在图片消息里没有所以要做空判断。MsgId是微信侧的消息 ID可以用来做去重防止微信重试导致重复回复。提示微信服务器要求 5 秒内返回响应如果 AI 接口响应慢先返回空串或「正在思考」的提示再用客服消息接口异步推送结果。3. AI 对话引擎怎么搭从 Prompt 到多轮上下文管理3.1 大模型 API 的接入与流式输出拿到用户消息后下一步是调大模型。2026 年主流做法是走 OpenAI 兼容接口国内多家厂商都支持这个格式切换成本低。import requests def call_llm(user_message, history, api_key, base_url): # 构造对话历史system prompt 定义客服角色 messages [ {role: system, content: 你是一名微信在线客服回答要简洁、准确不确定的问题引导用户转人工。} ] # 追加历史对话保持多轮上下文 messages.extend(history) messages.append({role: user, content: user_message}) resp requests.post( f{base_url}/v1/chat/completions, headers{Authorization: fBearer {api_key}}, json{ model: gpt-4o-mini, # 按实际可用模型替换 messages: messages, temperature: 0.3, # 客服场景降低随机性 max_tokens: 500 }, timeout10 ) return resp.json()[choices][0][message][content]逻辑说明system消息定义 AI 的角色边界这一步直接决定回复风格。history是之前几轮的对话记录按[{role:user,content:...},{role:assistant,content:...}]格式传入。temperature设低一些客服场景不需要创意需要稳定。参数说明max_tokens控制回复长度微信消息太长用户不会看500 足够。timeout设 10 秒超过就降级到人工。base_url和api_key从环境变量读不要硬编码在源码里。3.2 多轮对话的上下文窗口与截断策略多轮对话不能无限往 messages 里塞token 有上限成本也在涨。常见做法是保留最近 N 轮或者按 token 数截断。def trim_history(history, max_turns6): # 只保留最近 max_turns 轮对话每轮包含 user 和 assistant 两条 if len(history) max_turns * 2: return history[-(max_turns * 2):] return history逻辑说明max_turns6意味着保留最近 6 轮问答大约 12 条消息。这个数字不是固定的取决于你的模型上下文窗口和业务复杂度。售前咨询通常 3 到 4 轮就够售后排查可能需要 8 到 10 轮。参数说明如果对话涉及订单号、手机号等关键信息不要只靠截断应该把关键实体抽出来存到会话状态里每次请求时重新注入 system prompt。3.3 意图识别与转人工的触发条件AI 不是万能的什么时候转人工需要明确规则。我一般设三个触发条件用户明确说「转人工」、AI 连续两轮回答置信度低、用户情绪检测为负面。def should_transfer_to_human(user_message, ai_reply, history): # 条件一用户明确要求转人工 if any(kw in user_message for kw in [转人工, 人工客服, 找真人]): return True # 条件二AI 回复中包含不确定表述 if any(kw in ai_reply for kw in [不确定, 不清楚, 建议咨询]): return True # 条件三连续两轮用户重复同一问题 if len(history) 4: last_user history[-2][content] if history[-2][role] user else if last_user and last_user.strip() user_message.strip(): return True return False逻辑说明这三个条件覆盖了大部分需要人工介入的场景。条件二的关键词列表可以根据你的业务补充比如「退款」「投诉」这类词也应该触发转人工。参数说明转人工后AI 应该停止自动回复把会话标记为「人工接管中」同时把之前的对话摘要推给人工客服避免用户重复描述问题。4. 会话存储与用户管理OpenID 映射与消息落库4.1 用 OpenID 做用户唯一标识微信侧拿不到用户的手机号除非用户主动授权OpenID 是唯一稳定的用户标识。同一个用户在不同公众号下的 OpenID 不同如果有多公众号需要用 UnionID 做跨号统一。CREATE TABLE wechat_user ( id BIGINT PRIMARY KEY AUTO_INCREMENT, openid VARCHAR(64) NOT NULL UNIQUE, unionid VARCHAR(64) DEFAULT NULL, nickname VARCHAR(128) DEFAULT NULL, first_seen DATETIME DEFAULT CURRENT_TIMESTAMP, last_active DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_unionid (unionid) ); CREATE TABLE chat_message ( id BIGINT PRIMARY KEY AUTO_INCREMENT, openid VARCHAR(64) NOT NULL, role ENUM(user, assistant, human) NOT NULL, content TEXT NOT NULL, msg_id VARCHAR(64) DEFAULT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_openid_time (openid, created_at) );逻辑说明wechat_user表存用户基本信息openid加唯一索引防止重复插入。chat_message表存所有对话记录role字段区分是用户发的、AI 回的、还是人工客服回的。idx_openid_time联合索引让「查某个用户最近的消息」这个高频查询走索引。参数说明content用 TEXT 类型微信消息最长 2048 字节TEXT 足够。如果要做全文检索可以加 FULLTEXT 索引或外接搜索引擎。4.2 会话上下文的读写与缓存每次调 AI 都要读历史消息如果每次都查数据库QPS 一高就扛不住。常见做法是用 Redis 缓存最近几轮对话。import redis import json r redis.Redis(hostlocalhost, port6379, db0) def get_history(openid, max_turns6): # 从 Redis 列表读取最近对话key 按用户维度隔离 key fchat:history:{openid} raw r.lrange(key, -max_turns * 2, -1) return [json.loads(item) for item in raw] def append_history(openid, role, content): # 追加一条对话到 Redis并裁剪到固定长度 key fchat:history:{openid} r.rpush(key, json.dumps({role: role, content: content})) r.ltrim(key, -20, -1) # 只保留最近 20 条 r.expire(key, 3600 * 24) # 24 小时过期逻辑说明用 Redis List 存对话rpush追加lrange读取。ltrim保证列表不会无限增长expire保证不活跃用户的缓存自动清理。参数说明max_turns和ltrim的数值要配合读取时取最近 12 条存储时保留 20 条留一些余量。过期时间 24 小时是经验值超过这个时间用户大概率是新问题不需要旧上下文。4.3 消息落库与异步写入Redis 是缓存数据库才是持久化。但每次对话都同步写库会拖慢响应我一般用异步队列。from celery import Celery app Celery(tasks, brokerredis://localhost:6379/1) app.task def save_message_async(openid, role, content, msg_idNone): # 异步写入 MySQL不阻塞主流程 conn get_db_connection() cursor conn.cursor() cursor.execute( INSERT INTO chat_message (openid, role, content, msg_id) VALUES (%s, %s, %s, %s), (openid, role, content, msg_id) ) conn.commit() cursor.close() conn.close()逻辑说明主流程只写 Redis 并返回响应落库交给 Celery 异步做。这样即使数据库短暂不可用用户侧也不会感知到延迟。参数说明broker用 Redis 的另一个 db和缓存分开避免互相影响。如果量不大也可以直接用线程池做异步不一定上 Celery。5. 避坑与排查微信 AI 客服上线后最容易翻车的五件事5.1 回调地址验证通过但收不到消息现象公众号后台显示「服务器配置成功」但用户发消息没有任何反应。原因微信验证时发的是 GET 请求实际消息推送是 POST。很多框架默认只处理 GET 或只处理 POST导致验证过了但消息进不来。解决确认你的路由同时接受 GET 和 POST。GET 走签名验证逻辑POST 走消息解析逻辑。另外检查服务器防火墙是否放行了微信服务器 IP 段微信官方文档有公布 IP 列表。5.2 AI 回复超时导致微信重试现象用户收到重复回复或者回复延迟很高。原因微信要求 5 秒内响应大模型 API 偶尔超过这个时间。微信没收到响应会重试三次每次间隔 5 秒导致同一条消息被处理多次。解决收到消息后立即返回空串微信不会展示空回复然后用客服消息接口异步推送 AI 结果。同时在处理消息前用MsgId做去重已经处理过的直接跳过。5.3 多轮对话串号现象A 用户的对话历史出现在 B 用户的会话里。原因会话 key 设计有问题比如用了公众号 ID 而不是用户 OpenID 做 key或者 Redis 连接池串了 db。解决会话 key 必须包含 OpenID格式如chat:history:{openid}。如果用了多进程或多线程确认 Redis 连接是每个请求独立获取的不要用全局单例。5.4 转人工后 AI 还在自动回复现象用户已经和人工客服聊上了AI 还在插话。原因转人工状态没有持久化或者状态存在内存里多实例部署时不同实例状态不一致。解决转人工状态存 Rediskey 如chat:human:{openid}设置合理的过期时间比如 30 分钟无交互自动释放。每次 AI 回复前先检查这个 key 是否存在。5.5 敏感词过滤缺失导致合规风险现象AI 回复了不该回复的内容或者用户输入了违规内容被原样回显。原因没有做输入输出双向过滤。解决在用户消息进 AI 之前做一次敏感词检测AI 回复发出之前再做一次。敏感词库可以用开源的也可以自己维护。检测到敏感词时直接返回预设的安全话术并记录日志。6. 让 AI 客服更懂业务知识库注入与效果验证6.1 用 RAG 把产品文档喂给 AI通用大模型不知道你的产品细节直接问「你们支持退货吗」它只能瞎编。常见做法是 RAG把产品文档、FAQ、历史工单切片存到向量库用户提问时先检索相关片段拼到 prompt 里。def build_rag_prompt(user_question, vector_store, top_k3): # 检索最相关的文档片段 docs vector_store.similarity_search(user_question, ktop_k) context \n.join([doc.page_content for doc in docs]) # 把检索结果拼到 system prompt 里 system_prompt f你是一名微信在线客服。请根据以下资料回答用户问题 {context} 如果资料中没有相关信息请引导用户转人工不要编造。 return system_prompt逻辑说明similarity_search返回和问题最相关的 top_k 个文档片段拼到 system prompt 里。这样 AI 的回答有据可依不会胡编。参数说明top_k3是经验值太多会挤占上下文窗口太少可能漏掉关键信息。文档切片大小建议 300 到 500 字太大检索精度下降太小上下文不完整。6.2 用日志和人工抽检验证回复质量上线后不能只看「有没有回复」要看「回复得对不对」。我一般做两件事一是把 AI 回复和人工回复做对比抽检二是监控转人工率。-- 统计每天的转人工率超过 30% 说明 AI 回答质量有问题 SELECT DATE(created_at) AS day, COUNT(DISTINCT openid) AS total_users, SUM(CASE WHEN role human THEN 1 ELSE 0 END) AS human_count, ROUND(SUM(CASE WHEN role human THEN 1 ELSE 0 END) / COUNT(*), 3) AS transfer_rate FROM chat_message GROUP BY DATE(created_at) ORDER BY day DESC;逻辑说明转人工率是 AI 客服最核心的指标。如果超过 30%说明要么知识库覆盖不够要么 prompt 写得不好要么转人工规则太敏感。参数说明这个查询按天聚合transfer_rate是人工消息占总消息的比例。实际看的时候要结合业务售前咨询转人工率天然比售后低。6.3 一个我踩过的坑别让 AI 替你做承诺最后说一个血泪教训。早期版本里AI 为了「显得有用」会主动说「可以退货」「三天内到账」这类话。结果用户截图来找客服兑现人工根本不知道这回事。后来我在 system prompt 里加了一条硬规则任何涉及金额、时效、政策的承诺必须回复「具体以人工客服确认为准」。这条规则救了我好几次。AI 客服的边界是「解答已知问题」不是「替公司做决定」。这个习惯我一直保持到现在希望帮到你。本文还有配套的精品资源点击获取