init

wechat-bot/bot/agent_client.py

@@ -0,0 +1,237 @@
+"""
+Agent HTTP 客户端模块
+负责与远程 ADK Agent 服务通信，包括：
+- HTTP 客户端生命周期管理
+- 用户会话管理（创建、缓存、清除）
+- SSE 流式调用（主方案）
+- 同步调用（降级方案）
+"""
+import json
+import logging
+from typing import Optional
+
+import httpx
+
+from config import Config
+from bot.reply_parser import extract_final_reply
+
+logger = logging.getLogger(__name__)
+
+# 用户会话映射: user_id -> session_id
+_user_sessions: dict[str, str] = {}
+
+# Agent HTTP 客户端
+_http_client: Optional[httpx.AsyncClient] = None
+
+
+def get_http_client() -> httpx.AsyncClient:
+    """获取或创建 HTTP 客户端（全局单例）"""
+    global _http_client
+    if _http_client is None:
+        _http_client = httpx.AsyncClient(
+            base_url=Config.AGENT_BASE_URL,
+            timeout=Config.AGENT_TIMEOUT,
+        )
+    return _http_client
+
+
+async def close_http_client():
+    """关闭 HTTP 客户端"""
+    global _http_client
+    if _http_client:
+        await _http_client.aclose()
+        _http_client = None
+
+
+# ============================================================
+# 会话管理
+# ============================================================
+
+async def get_or_create_session(user_id: str) -> str:
+    """获取或创建用户的 Agent 会话"""
+    if user_id in _user_sessions:
+        return _user_sessions[user_id]
+
+    client = get_http_client()
+    try:
+        url = f"/apps/{Config.AGENT_APP_NAME}/users/{user_id}/sessions"
+        resp = await client.post(url, json={})
+        resp.raise_for_status()
+        data = resp.json()
+        session_id = data["id"]
+        _user_sessions[user_id] = session_id
+        logger.info("为用户 %s 创建 Agent 会话: %s", user_id, session_id)
+        return session_id
+    except Exception as e:
+        logger.error("创建 Agent 会话失败: %s", e)
+        raise
+
+
+def clear_session(user_id: str):
+    """清除用户的会话缓存（会话失效时调用）"""
+    _user_sessions.pop(user_id, None)
+
+
+# ============================================================
+# Agent 调用
+# ============================================================
+
+async def call_agent_stream(
+    ws_client,
+    frame: dict,
+    user_id: str,
+    message: str,
+) -> str:
+    """
+    通过 SSE 流式调用 Agent，实现打字机效果。
+    失败时降级为同步调用。
+    返回完整的回复文本。
+    """
+    from aibot import generate_req_id
+
+    session_id = await get_or_create_session(user_id)
+    client = get_http_client()
+
+    try:
+        req_body = {
+            "app_name": Config.AGENT_APP_NAME,
+            "user_id": user_id,
+            "session_id": session_id,
+            "new_message": {
+                "role": "user",
+                "parts": [{"text": message}],
+            },
+            "streaming": True,
+        }
+
+        async with client.stream(
+            "POST",
+            "/run_sse",
+            json=req_body,
+            timeout=Config.AGENT_TIMEOUT,
+        ) as response:
+            if response.status_code != 200:
+                logger.warning(
+                    "SSE 流式调用返回 %d，降级为同步调用",
+                    response.status_code,
+                )
+                return await call_agent_sync(user_id, message)
+
+            stream_id = generate_req_id("stream")
+            full_text = ""
+            has_sent_final = False
+
+            async for line in response.aiter_lines():
+                if not line.startswith("data: "):
+                    continue
+
+                data_str = line[6:]
+                try:
+                    data = json.loads(data_str)
+                except json.JSONDecodeError:
+                    continue
+
+                # 错误处理
+                if "error" in data:
+                    logger.error("Agent SSE 错误: %s", data["error"])
+                    if "not found" in data["error"].lower():
+                        clear_session(user_id)
+                    return "抱歉，处理消息时出现错误，请稍后重试。"
+
+                # 收集所有文本（含思考过程）
+                parts = data.get("content", {}).get("parts", [])
+                for part in parts:
+                    text = part.get("text", "")
+                    if not text:
+                        continue
+                    if part.get("thought"):
+                        continue
+                    full_text += text
+
+                # 实时推送"正在思考"状态
+                if full_text and not has_sent_final:
+                    try:
+                        await ws_client.reply_stream(
+                            frame, stream_id, "💭 正在思考...", False
+                        )
+                    except Exception:
+                        pass
+
+                # 最终完成时提取正式回复并发送
+                is_final = data.get("finishReason") == "STOP"
+                if is_final and full_text and not has_sent_final:
+                    clean_reply = extract_final_reply(full_text)
+                    logger.info("Agent 回复完成，长度: %d", len(clean_reply))
+                    try:
+                        await ws_client.reply_stream(
+                            frame, stream_id, clean_reply, True
+                        )
+                        has_sent_final = True
+                    except Exception as e:
+                        logger.warning("推送最终回复失败: %s", e)
+
+            # 兜底：如果流结束但未发送最终回复
+            if full_text and not has_sent_final:
+                clean_reply = extract_final_reply(full_text)
+                logger.info("Agent SSE 兜底发送，长度: %d", len(clean_reply))
+                try:
+                    await ws_client.reply_stream(
+                        frame, stream_id, clean_reply, True
+                    )
+                except Exception as e:
+                    logger.warning("推送兜底回复失败: %s", e)
+
+            return extract_final_reply(full_text) if full_text else ""
+
+    except httpx.ConnectError:
+        logger.error("无法连接到 Agent 服务: %s", Config.AGENT_BASE_URL)
+        return "抱歉，Agent 服务连接失败，请稍后重试。"
+    except Exception as e:
+        logger.error("SSE 流式调用异常，降级为同步: %s", e)
+        return await call_agent_sync(user_id, message)
+
+
+async def call_agent_sync(user_id: str, message: str) -> str:
+    """
+    同步调用 Agent（/run 接口），作为流式调用的降级方案。
+    """
+    session_id = await get_or_create_session(user_id)
+    client = get_http_client()
+
+    def _build_body(sid: str) -> dict:
+        return {
+            "app_name": Config.AGENT_APP_NAME,
+            "user_id": user_id,
+            "session_id": sid,
+            "new_message": {
+                "role": "user",
+                "parts": [{"text": message}],
+            },
+        }
+
+    def _extract_reply(events: list) -> str:
+        """从 Event[] 中提取最终回复（跳过 thought）"""
+        for event in events:
+            parts = event.get("content", {}).get("parts", [])
+            for part in parts:
+                if part.get("text") and not part.get("thought"):
+                    return part["text"]
+        return ""
+
+    try:
+        resp = await client.post("/run", json=_build_body(session_id))
+        resp.raise_for_status()
+        return _extract_reply(resp.json())
+
+    except httpx.HTTPStatusError as e:
+        if e.response.status_code == 404:
+            clear_session(user_id)
+            new_session = await get_or_create_session(user_id)
+            resp = await client.post("/run", json=_build_body(new_session))
+            resp.raise_for_status()
+            return _extract_reply(resp.json())
+        logger.error("Agent HTTP 错误 [%d]: %s", e.response.status_code, e)
+        return "抱歉，服务暂时不可用，请稍后重试。"
+    except Exception as e:
+        logger.error("同步调用 Agent 失败: %s", e)
+        return "抱歉，处理消息时出现错误，请稍后重试。"

wechat-bot/bot/handlers.py

@@ -0,0 +1,91 @@
+"""
+消息处理器模块
+提供统一的消息处理流程，消除各消息类型处理器之间的重复代码。
+
+处理流程：
+1. 解析消息元数据和内容
+2. （可选）下载附件（图片/文件/视频）
+3. 调用 Agent 获取回复
+"""
+import logging
+
+from aibot import generate_req_id
+
+from bot.agent_client import call_agent_stream
+from bot.message_parser import extract_metadata, extract_text_content
+
+logger = logging.getLogger(__name__)
+
+
+async def handle_message(ws_client, frame: dict, msg_type: str):
+    """
+    通用消息处理器，适用于所有消息类型。
+
+    对于需要下载附件的消息类型（image/file/video），
+    会先尝试下载附件并附加描述信息。
+    """
+    meta = extract_metadata(frame)
+    user_id = meta["user_id"]
+
+    # ---- 提取消息内容 ----
+    content = await _resolve_content(ws_client, frame, msg_type, meta)
+    logger.info("收到%s消息 - 用户: %s, 内容: %s", msg_type, user_id, content[:100])
+
+    # ---- 调用 Agent ----
+    reply = await call_agent_stream(ws_client, frame, user_id, content)
+
+    # ---- 兜底：未收到回复 ----
+    if not reply:
+        stream_id = generate_req_id("stream")
+        await ws_client.reply_stream(frame, stream_id, "（未收到回复）", True)
+
+
+async def _resolve_content(
+    ws_client, frame: dict, msg_type: str, meta: dict
+) -> str:
+    """
+    解析消息内容。
+    对于文本类消息直接提取文本；对于附件类消息尝试下载并附加描述。
+    """
+    if msg_type in ("text", "voice", "mixed"):
+        return extract_text_content(frame)
+
+    if msg_type == "image":
+        return await _download_and_describe(ws_client, frame, "image", "图片")
+
+    if msg_type == "file":
+        return await _download_and_describe(ws_client, frame, "file", "文件")
+
+    if msg_type == "video":
+        return await _download_and_describe(ws_client, frame, "video", "视频")
+
+    return extract_text_content(frame)
+
+
+async def _download_and_describe(
+    ws_client, frame: dict, media_type: str, label: str
+) -> str:
+    """
+    尝试下载附件并返回描述文本。
+    下载失败时返回简单描述，不阻断流程。
+    """
+    body = frame.get("body", {})
+    media_info = body.get(media_type, {})
+
+    if not media_info:
+        return f"[用户发送了一个{label}]"
+
+    url = media_info.get("url")
+    aeskey = media_info.get("aeskey")
+
+    if not url:
+        return f"[用户发送了一个{label}]"
+
+    try:
+        buffer, filename = await ws_client.download_file(url, aeskey)
+        desc = f"[用户发送了{label}: {filename}, 大小: {len(buffer)} bytes]"
+        logger.info("%s下载成功: %s, 大小: %d bytes", label, filename, len(buffer))
+        return desc
+    except Exception as e:
+        logger.warning("%s下载失败: %s", label, e)
+        return f"[用户发送了一个{label}]"

wechat-bot/bot/message_parser.py

@@ -0,0 +1,57 @@
+"""
+消息解析模块
+负责从企业微信 SDK 的消息帧中提取结构化数据。
+
+frame 结构: {cmd, headers, body: {msgtype, text, from, ...}}
+注意：msgtype 在 body 内部，不在 frame 顶层。
+"""
+
+
+def extract_text_content(frame: dict) -> str:
+    """
+    从消息帧中提取文本内容。
+    根据不同的消息类型，提取可用于 Agent 处理的文本描述。
+    """
+    body = frame.get("body", {})
+    msg_type = body.get("msgtype", "")
+
+    if msg_type == "text":
+        return body.get("text", {}).get("content", "")
+
+    elif msg_type == "image":
+        return "[用户发送了一张图片]"
+
+    elif msg_type == "voice":
+        return body.get("voice", {}).get("content", "[用户发送了一条语音消息]")
+
+    elif msg_type == "file":
+        filename = body.get("file", {}).get("filename", "未知文件")
+        return f"[用户发送了一个文件: {filename}]"
+
+    elif msg_type == "video":
+        return "[用户发送了一段视频]"
+
+    elif msg_type == "mixed":
+        mixed_items = body.get("mixed", {}).get("content", [])
+        text_parts = []
+        for item in mixed_items:
+            if isinstance(item, dict) and item.get("type") == "text":
+                text_parts.append(item.get("content", ""))
+            elif isinstance(item, str):
+                text_parts.append(item)
+        return "\n".join(text_parts) if text_parts else "[用户发送了图文混排消息]"
+
+    return f"[未知消息类型: {msg_type}]"
+
+
+def extract_metadata(frame: dict) -> dict:
+    """提取消息的元数据（msgtype 在 body 内部）"""
+    body = frame.get("body", {})
+    from_info = body.get("from", {})
+    return {
+        "user_id": from_info.get("userid", ""),
+        "chat_id": body.get("chatid", ""),
+        "chat_type": body.get("chattype", ""),
+        "msg_id": body.get("msgid", ""),
+        "msg_type": body.get("msgtype", ""),
+    }

wechat-bot/bot/reply_parser.py

@@ -0,0 +1,84 @@
+"""
+回复文本解析模块
+负责处理 Agent 返回的原始文本，提取最终正式回复。
+
+模型思考标记格式（按优先级检测）：
+1. <think...</think 标签对（MiniMax M2.7 等模型）
+2. 💭 emoji 前缀（旧版格式，保留兼容）
+
+ADK 最后一个 event 会重复发送完整内容，需要去重。
+"""
+import logging
+import re
+
+logger = logging.getLogger(__name__)
+
+
+def extract_final_reply(text: str) -> str:
+    """
+    从 SSE 收集的完整文本中提取最终正式回复。
+
+    策略：
+    1. 去除 <think...</think 标签及其内容
+    2. 兼容 💭 emoji 分割（旧版格式）
+    3. 去重
+    """
+    if not text:
+        return text
+
+    # 步骤1：去除 <think...</think 标签块
+    if "<think" in text:
+        text = re.sub(r"<think[\s\S]*?</think\s*>?", "", text)
+
+    # 步骤2：兼容 💭 emoji 分割（旧版格式）
+    if "💭" in text:
+        text = text.split("💭")[-1]
+
+    # 清理前导空白
+    text = text.strip()
+
+    # 步骤3：去重
+    text = _deduplicate(text)
+
+    return text
+
+
+def _deduplicate(text: str) -> str:
+    """
+    对回复文本去重。
+    ADK 最后一个 SSE event 会重复发送完整内容（含思考+回复），
+    去除 <think...> 后可能出现连续两份相同的正式回复。
+    支持行级和字符级去重。
+    """
+    if not text:
+        return text
+
+    # 行级去重
+    lines = text.split("\n")
+    non_empty = [l for l in lines if l.strip()]
+
+    if len(non_empty) > 1:
+        mid = len(non_empty) // 2
+        if non_empty[:mid] == non_empty[mid:]:
+            return "\n".join(non_empty[:mid])
+
+        for overlap_len in range(min(len(non_empty) // 2, 20), 0, -1):
+            if non_empty[-overlap_len:] == non_empty[:overlap_len]:
+                return "\n".join(non_empty[overlap_len:])
+
+    # 字符级去重：检查文本是否由两个相同的子串拼接而成
+    length = len(text)
+    for sub_len in range(length // 2, 0, -1):
+        if length % sub_len == 0 and text[:sub_len] * (length // sub_len) == text:
+            return text[:sub_len]
+
+    # 部分重叠：前缀和后缀相同，中间有重复
+    for overlap in range(min(length // 2, 500), 0, -1):
+        if text[:overlap] == text[-overlap:]:
+            trimmed = text[overlap:]
+            trimmed_len = len(trimmed)
+            for sub_len in range(trimmed_len // 2, 0, -1):
+                if trimmed_len % sub_len == 0 and trimmed[:sub_len] * (trimmed_len // sub_len) == trimmed:
+                    return trimmed[:sub_len]
+
+    return text

wechat-bot/bot/wecom_bot.py

@@ -0,0 +1,104 @@
+"""
+企业微信桥接模块
+使用 wecom-aibot-python-sdk 与企业微信建立 WebSocket 长连接，
+接收用户消息后通过 ADK API 调用远程 Agent，再将 Agent 回复发送给用户。
+
+本模块仅负责：
+- 创建 WSClient 实例
+- 注册事件和消息处理器
+"""
+import json
+import logging
+import traceback
+
+from aibot import WSClient, WSClientOptions
+
+from config import Config
+from bot.agent_client import close_http_client
+from bot.handlers import handle_message
+from bot.message_parser import extract_metadata
+
+logger = logging.getLogger(__name__)
+
+
+def create_bot_client() -> WSClient:
+    """
+    创建企业微信机器人 WebSocket 客户端，并注册所有事件和消息处理器。
+    """
+    ws_client = WSClient(
+        WSClientOptions(
+            bot_id=Config.WECHAT_BOT_ID,
+            secret=Config.WECHAT_BOT_SECRET,
+        )
+    )
+
+    # ---- 事件处理器 ----
+
+    @ws_client.on("authenticated")
+    def on_authenticated():
+        logger.info("企业微信机器人认证成功")
+
+    @ws_client.on("event.enter_chat")
+    async def on_enter_chat(frame):
+        """用户进入会话，发送欢迎语"""
+        meta = extract_metadata(frame)
+        user_id = meta.get("user_id", "")
+        logger.info("用户进入会话: %s", user_id)
+
+        try:
+            await ws_client.reply_welcome(
+                frame,
+                {
+                    "msgtype": "text",
+                    "text": {
+                        "content": "您好！我是智能助手，有什么可以帮您的吗？"
+                    },
+                },
+            )
+        except Exception as e:
+            logger.error("发送欢迎语失败: %s", e)
+
+    @ws_client.on("event.feedback_event")
+    async def on_feedback(frame):
+        """用户反馈事件"""
+        body = frame.get("body", {})
+        logger.info("收到用户反馈: %s", json.dumps(body, ensure_ascii=False))
+
+    @ws_client.on("disconnected")
+    def on_disconnected(reason):
+        logger.warning("连接被断开（可能被新连接踢掉）: %s", reason)
+
+    # ---- 消息处理器 ----
+
+    @ws_client.on("message.text")
+    async def on_text(frame):
+        await _safe_handle(ws_client, frame, "text")
+
+    @ws_client.on("message.image")
+    async def on_image(frame):
+        await _safe_handle(ws_client, frame, "image")
+
+    @ws_client.on("message.voice")
+    async def on_voice(frame):
+        await _safe_handle(ws_client, frame, "voice")
+
+    @ws_client.on("message.file")
+    async def on_file(frame):
+        # file 类型可能包含视频
+        body = frame.get("body", {})
+        msg_type = "video" if body.get("video") else "file"
+        await _safe_handle(ws_client, frame, msg_type)
+
+    @ws_client.on("message.mixed")
+    async def on_mixed(frame):
+        await _safe_handle(ws_client, frame, "mixed")
+
+    return ws_client
+
+
+async def _safe_handle(ws_client, frame: dict, msg_type: str):
+    """安全地处理消息，捕获异常防止崩溃"""
+    try:
+        await handle_message(ws_client, frame, msg_type)
+    except Exception as e:
+        logger.error("处理%s消息异常: %s\n%s", msg_type, e, traceback.format_exc())