85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
"""
|
||
回复文本解析模块
|
||
负责处理 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
|