官方文档：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 f"gewe:sess:{appid}:{peer_id}" def append_turn(appid, peer_id, role, content, max_turns=12): 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 msgs

3.2 RAG 检索 + 带工具的 Agent

TOOLS = [ { "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_k=4) -> 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( model="gpt-4.1-mini", messages=messages, tools=TOOLS, temperature=0.2, timeout=25, ) 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(model="gpt-4.1-mini", messages=messages, temperature=0.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_ascii=False) 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）。

四、生产环境避坑与安全风控

1. 提示词注入：用户可能要求「忽略上述指令」；系统提示固定不可被用户覆盖，工具白名单强制。

2. PII：订单号、手机号进日志脱敏；不把完整聊天记录明文长期扔第三方。

3. 幻觉：涉及金额、发货、退款必须 Tool 实查，禁止模型直接承诺。

4. 环路：isSelf、机器人互加好友场景要熔断。

5. 成本：短问答走小模型；复杂走大模型；RAG 命中率监控。

6. 合规：AI 自动加好友/群发仍受微信规范约束，能力边界以官方文档为准。

五、本篇交付清单

• Agent 与 GeWe 异步解耦架构

• Session + RAG + Tool Calling 示例

• 置信度转人工与限速出站

• 注入攻击与幻觉防控要点