官方文档GeWe API - GeWe API微信 API 开发文档一、业务痛点与技术背景私域场景要的不是「能聊天的 Bot」而是可控、可审计、可降级的 AI 助手会话粘性映射故障转移与健康摘除容量规划与演练剧本多账号舰队调度GeWe 提供稳定的消息入出站AI 层负责理解与决策。二者通过「异步 Agent Runtime」解耦。二、核心架构设计与数据流转Webhook → MQ → Dialog Orchestrator │ ┌───────────┼────────────┐ ▼ ▼ ▼ Intent Router RAG Retriever Tool Registry │ │ │(查单/建群/打标) └───────────┼────────────┘ ▼ LLM Planner (Function Call) │ ┌──────────┼──────────┐ ▼ ▼ ▼ 直接回复 调 GeWe API 转人工工单 │ │ └──── Outbound Sender限速/风控关键约束Orchestrator 消费 MQ永不在 Webhook 内调模型。RAG 只检索已审批知识库工具调用走白名单。对用户最终回复必须过「策略引擎」敏感词、承诺词、夜间模式。三、关键代码与配置示例3.1 会话记忆Redis 摘要import json, redis, time from openai import OpenAI r redis.Redis.from_url(os.environ[REDIS_URL]) llm OpenAI() def session_key(appid: str, peer_id: str) - str: return fgewe:sess:{appid}:{peer_id} def append_turn(appid, peer_id, role, content, max_turns12): k session_key(appid, peer_id) r.rpush(k, json.dumps({role: role, content: content, ts: time.time()})) r.ltrim(k, -max_turns, -1) r.expire(k, 7 * 86400) def build_messages(appid, peer_id, system: str, user_text: str): history [json.loads(x) for x in r.lrange(session_key(appid, peer_id), 0, -1)] msgs [{role: system, content: system}] history [ {role: user, content: user_text} ] return msgs3.2 RAG 检索 带工具的 AgentTOOLS [ { type: function, function: { name: query_order, description: 按订单号查询物流状态, parameters: { type: object, properties: {order_id: {type: string}}, required: [order_id], }, }, }, { type: function, function: { name: tag_friend, description: 给微信好友打业务标签, parameters: { type: object, properties: { wxid: {type: string}, tag: {type: string}, }, required: [wxid, tag], }, }, }, ] def retrieve(kb: str, query: str, top_k4) - str: # 伪代码向量库检索 docs vector_search(kb, query, top_k) return \n---\n.join(d[text] for d in docs) def run_agent(appid, peer_id, user_text: str) - str: knowledge retrieve(after_sales_kb, user_text) system f你是企业售后助手。仅依据知识库与工具结果回答。 禁止编造物流与价格。不确定时回复将转人工。 知识库片段\n{knowledge} messages build_messages(appid, peer_id, system, user_text) resp llm.chat.completions.create( modelgpt-4.1-mini, messagesmessages, toolsTOOLS, temperature0.2, timeout25, ) msg resp.choices[0].message if msg.tool_calls: for call in msg.tool_calls: result dispatch_tool(call.function.name, call.function.arguments) messages.append(msg) messages.append({ role: tool, tool_call_id: call.id, content: result, }) resp llm.chat.completions.create(modelgpt-4.1-mini, messagesmessages, temperature0.2) msg resp.choices[0].message answer policy_filter(msg.content or ) append_turn(appid, peer_id, user, user_text) append_turn(appid, peer_id, assistant, answer) return answer def dispatch_tool(name: str, args_json: str) - str: args json.loads(args_json) if name query_order: return json.dumps(order_service.get(args[order_id]), ensure_asciiFalse) if name tag_friend: # 调用 GeWe 标签能力字段以文档为准 return json.dumps(gewe.tag_friend(args[wxid], args[tag])) return json.dumps({error: tool_not_allowed})3.3 回复出站与人工接管async function replyOrEscalate(ctx: Ctx, answer: string, confidence: number) { if (confidence 0.55 || /转人工|人工客服/.test(answer)) { await ticketService.create({ peerId: ctx.peerId, reason: low_confidence }); await gewe.sendText(ctx.appid, ctx.peerId, 已为您转接人工请稍候。); return; } await rateLimiter.take(ctx.appid); // 与风控策略联动 await gewe.sendText(ctx.appid, ctx.peerId, answer); }3.4 延迟与降级agent: llm_timeout_ms: 25000 fallback_reply: 正在为您查询请稍等片刻 max_tool_rounds: 3 night_mode: enabled: true only_faq: true超时先发fallback_reply后台继续跑完再补一条注意防双发用 client_msg_id。四、生产环境避坑与安全风控提示词注入用户可能要求「忽略上述指令」系统提示固定不可被用户覆盖工具白名单强制。PII订单号、手机号进日志脱敏不把完整聊天记录明文长期扔第三方。幻觉涉及金额、发货、退款必须 Tool 实查禁止模型直接承诺。环路isSelf、机器人互加好友场景要熔断。成本短问答走小模型复杂走大模型RAG 命中率监控。合规AI 自动加好友/群发仍受微信规范约束能力边界以官方文档为准。五、本篇交付清单Agent 与 GeWe 异步解耦架构Session RAG Tool Calling 示例置信度转人工与限速出站注入攻击与幻觉防控要点