diff --git a/wechat-bot/bot/agent_client.py b/wechat-bot/bot/agent_client.py index 4262078..0c6cd0c 100644 --- a/wechat-bot/bot/agent_client.py +++ b/wechat-bot/bot/agent_client.py @@ -13,7 +13,7 @@ from typing import Optional import httpx from config import Config -from bot.reply_parser import extract_final_reply +from bot.reply_parser import parse_reply logger = logging.getLogger(__name__) @@ -72,6 +72,46 @@ def clear_session(user_id: str): _user_sessions.pop(user_id, None) +# ============================================================ +# 显示格式化 +# ============================================================ + +def _format_turn_for_display(text: str, use_emoji: bool = True) -> str: + """ + 将一轮的原始文本（ thinking... response...）格式化为微信展示文本。 + use_emoji=True 用 💭 前缀（流式阶段，企微会渲染为可折叠的"已深度思考"） + use_emoji=False 用纯文本标题（最终回复，避免触发"思考中"指示器） + """ + if not text: + return "" + + prefix = "💭 思考过程:" if use_emoji else "【思考过程】" + + # 查找 response 分隔标记 + parts = text.split('\n response\n', 1) + if len(parts) == 2: + think_raw, body_raw = parts + think_clean = think_raw.strip() + if think_clean.startswith(' thinking'): + think_clean = think_clean[len(' thinking'):].lstrip('\n') + think_clean = think_clean.strip() + body_clean = body_raw.strip() + + result = "" + if think_clean: + result = f"{prefix}\n{think_clean}\n\n" + if body_clean: + result += body_clean + return result + else: + # 还没有 response，全是思考内容 + think_clean = text.strip() + if think_clean.startswith(' thinking'): + think_clean = think_clean[len(' thinking'):].lstrip('\n') + think_clean = think_clean.strip() + return f"{prefix}\n{think_clean}" if think_clean else "" + + # ============================================================ # Agent 调用 # ============================================================ @@ -118,20 +158,44 @@ async def call_agent_stream( return await call_agent_sync(user_id, message) stream_id = generate_req_id("stream") - full_text = "" + full_text = "" # 累积非 STOP 的增量文本 + turn_start = 0 # 当前轮在 full_text 中的起始位置 + stop_texts: list[str] = [] # 每轮 STOP 时的完整文本快照 + event_count = 0 + last_sent_display = "" # 上次已发送到微信的展示文本 has_sent_final = False - has_sent_thinking = False + first_event_sent = False # 是否已发送过流式更新（用于5秒超时保护） async for line in response.aiter_lines(): if not line.startswith("data: "): continue data_str = line[6:] + event_count += 1 try: data = json.loads(data_str) except json.JSONDecodeError: + logger.warning("SSE #%d JSON 解析失败: %s", event_count, data_str[:200]) continue + finish_reason = data.get("finishReason", "") + content = data.get("content", {}) + parts = content.get("parts", []) + + # 日志：每个 SSE 事件的摘要 + part_summary = [] + for p in parts: + t = p.get("text", "") + part_summary.append( + f"text_len={len(t)}, thought={p.get('thought', False)}, " + f"preview={t[:50]!r}" + ) + logger.info( + "SSE #%d | finish=%s | parts=%d | %s", + event_count, finish_reason, len(parts), + " | ".join(part_summary) if part_summary else "(无parts)", + ) + # 错误处理 if "error" in data: logger.error("Agent SSE 错误: %s", data["error"]) @@ -139,51 +203,82 @@ async def call_agent_stream( 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 + is_final = finish_reason == "STOP" - # 只发送一次"正在思考"状态 - if full_text and not has_sent_thinking and not has_sent_final: - try: - await ws_client.reply_stream( - frame, stream_id, "💭 正在思考...", False + if is_final: + # STOP 事件：ADK 会重复发送完整文本，不追加到 full_text + # 保存当前轮的文本快照 + turn_text = full_text[turn_start:] + if turn_text: + stop_texts.append(turn_text) + logger.info( + ">>> STOP #%d，当前轮文本长度: %d，前200字: %s", + len(stop_texts), len(turn_text), turn_text[:200], ) - has_sent_thinking = True - except Exception: - pass + turn_start = len(full_text) + last_sent_display = "" # 新轮次重置展示缓存 + else: + # 非 STOP：增量文本 + for part in parts: + text = part.get("text", "") + if text: + full_text += text - # 最终完成时提取正式回复并发送 - 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) + # 流式更新微信消息 + turn_text = full_text[turn_start:] + display = _format_turn_for_display(turn_text) + if display and display != last_sent_display: + try: + await ws_client.reply_stream( + frame, stream_id, display, False + ) + last_sent_display = display + first_event_sent = True + except Exception: + pass - # 兜底：如果流结束但未发送最终回复 - if full_text and not has_sent_final: - clean_reply = extract_final_reply(full_text) - logger.info("Agent SSE 兜底发送，长度: %d", len(clean_reply)) + # === SSE 流结束，发送最终回复 === + logger.info( + "=== SSE 流结束 === %d 个事件, %d 个 STOP, full_text 总长: %d", + event_count, len(stop_texts), len(full_text), + ) + for idx, st in enumerate(stop_texts): + tc, bc = parse_reply(st) + logger.info( + "STOP[%d] 解析: 思考=%d, 正文=%d, 正文前100字: %s", + idx, len(tc), len(bc), bc[:100] if bc else "(空)", + ) + + # 从最后一个有正文的 STOP 构建最终内容 + final_body = "" + for stop_text in reversed(stop_texts): + _, body_content = parse_reply(stop_text) + if body_content: + final_body = body_content + break + + # 兜底 + if not final_body and stop_texts: + _, final_body = parse_reply(stop_texts[-1]) + if not final_body and full_text: + _, final_body = parse_reply(full_text[turn_start:]) + if not final_body and full_text: + final_body = full_text[turn_start:] + + if final_body: try: + # 用纯正文 + finish=True 正常关闭流 + # 思考过程在流式阶段已通过 💭 展示，finish=True 替换为正文 await ws_client.reply_stream( - frame, stream_id, clean_reply, True + frame, stream_id, final_body, True ) + has_sent_final = True + logger.info(">>> 最终回复发送成功，正文长度: %d", len(final_body)) except Exception as e: - logger.warning("推送兜底回复失败: %s", e) + logger.warning("推送最终回复失败: %s", e) - return extract_final_reply(full_text) if full_text else "" + _, body = parse_reply(full_text[turn_start:] if turn_start else full_text) + return body if body else "" except httpx.ConnectError: logger.error("无法连接到 Agent 服务: %s", Config.AGENT_BASE_URL) @@ -236,4 +331,4 @@ async def call_agent_sync(user_id: str, message: str) -> str: return "抱歉，服务暂时不可用，请稍后重试。" except Exception as e: logger.error("同步调用 Agent 失败: %s", e) - return "抱歉，处理消息时出现错误，请稍后重试。" + return "抱歉，处理消息时出现错误，请稍后重试。" \ No newline at end of file diff --git a/wechat-bot/bot/handlers.py b/wechat-bot/bot/handlers.py index ce8ddd7..fed5c8b 100644 --- a/wechat-bot/bot/handlers.py +++ b/wechat-bot/bot/handlers.py @@ -9,8 +9,6 @@ """ 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 @@ -33,11 +31,8 @@ async def handle_message(ws_client, frame: dict, msg_type: str): # ---- 调用 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) + logger.warning("Agent 未返回有效回复，用户: %s", user_id) async def _resolve_content( diff --git a/wechat-bot/bot/reply_parser.py b/wechat-bot/bot/reply_parser.py index 87da722..8a6075f 100644 --- a/wechat-bot/bot/reply_parser.py +++ b/wechat-bot/bot/reply_parser.py @@ -1,55 +1,108 @@ """ 回复文本解析模块 -负责处理 Agent 返回的原始文本，提取最终正式回复。 +负责处理 Agent 返回的原始文本，分离思考内容和正文。 -模型思考标记格式（按优先级检测）： -1. str: +def parse_reply(text: str) -> tuple[str, str]: """ - 从 SSE 收集的完整文本中提取最终正式回复。 + 从文本中分离思考内容和正文。 - 策略： - 1. 去除 ?", "", text) + think_content = "" + body_content = text - # 步骤2：兼容 💭 emoji 分割（旧版格式） - if "💭" in text: - text = text.split("💭")[-1] + # 格式1：匹配 " thinking\\n ... \\n response\\n" 纯文本标记 + # 从 SSE 日志确认: 思考内容以 "\\n response\\n" 结束 + parts = _RESPONSE_MARKER.split(text, maxsplit=1) + if len(parts) == 2: + think_part, body_part = parts + # 去掉开头的 " thinking\\n" + think_part = think_part.strip() + if think_part.startswith(' thinking'): + think_part = think_part[len(' thinking'):].lstrip('\n') + think_content = think_part.strip() + body_content = body_part.strip() - # 清理前导空白 - text = text.strip() + # 格式2：匹配 " thinking... response" HTML 标签风格（兜底） + if not think_content and "", text) + if match: + think_content = match.group(1).strip() + body_content = re.sub(r"", "", text).strip() + else: + think_content = text.replace(" str: + """ + 从文本中只提取正文内容（不含思考）。 + 用于流式更新时过滤思考内容。 + + 如果文本中还没有出现 "\\n response\\n" 标记， + 说明还在思考阶段，返回空字符串避免把思考内容当正文发送。 + """ + if not text: + return "" + + # 检查正文标记是否已出现 + split_pos = _find_response_marker(text) + if split_pos < 0: + # 还没出现 response 标记，仍在思考阶段，不返回任何内容 + return "" + + # 返回 response 之后的内容 + body = text[split_pos + 1:] + # 去掉开头的 response\n + body = body.lstrip('\n') + if body.startswith(' response'): + body = body[len(' response'):].lstrip('\n') + + # 再次检查是否有新的 response 标记（多轮场景） + pos2 = _find_response_marker(body) + if pos2 >= 0: + body = body[pos2 + 1:] + body = body.lstrip('\n') + if body.startswith(' response'): + body = body[len(' response'):].lstrip('\n') + + return body.strip() + + +def _find_response_marker(text: str) -> int: + """查找 \\n response\\n 标记的位置，返回 response 中 'r' 的索引""" + m = _RESPONSE_MARKER.search(text) + return m.start() if m else -1 def _deduplicate(text: str) -> str: - """ - 对回复文本去重。 - ADK 最后一个 SSE event 会重复发送完整内容（含思考+回复）， - 去除 后可能出现连续两份相同的正式回复。 - 支持行级和字符级去重。 - """ + """对回复文本去重。""" if not text: return text @@ -66,13 +119,13 @@ def _deduplicate(text: str) -> str: 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:] @@ -81,4 +134,4 @@ def _deduplicate(text: str) -> str: if trimmed_len % sub_len == 0 and trimmed[:sub_len] * (trimmed_len // sub_len) == trimmed: return trimmed[:sub_len] - return text + return text \ No newline at end of file