Losita/mai-bot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

将所有必要内容完全迁移后删除原文件

Browse Source
This commit is contained in:
UnCLAS-Prommer
2026-03-12 09:16:51 +08:00
committed by DrSmoothl
parent 71a288983f
commit e303fbeb6b
6 changed files with 271 additions and 1619 deletions
Download Patch File Download Diff File Expand all files Collapse all files

596
src/bw_learner/expression_learner_old.py
View File

@@ -1,596 +0,0 @@
import time
import json
import os
import re
import asyncio
from typing import List, Optional, Tuple, Any, Dict
from src.common.logger import get_logger
from src.common.database.database_model import Expression
from src.llm_models.utils_model import LLMRequest
from src.config.config import model_config, global_config
from src.chat.utils.chat_message_builder import (
build_anonymous_messages,
)
from src.prompt.prompt_manager import prompt_manager
from src.chat.message_receive.chat_stream import get_chat_manager
from src.bw_learner.learner_utils_old import (
filter_message_content,
is_bot_message,
build_context_paragraph,
contains_bot_self_name,
calculate_similarity,
parse_expression_response,
)
from src.bw_learner.jargon_miner_old import miner_manager
from src.bw_learner.expression_auto_check_task import (
single_expression_check,
)
# MAX_EXPRESSION_COUNT = 300
logger = get_logger("expressor")
class ExpressionLearner:
def __init__(self, chat_id: str) -> None:
self.express_learn_model: LLMRequest = LLMRequest(
model_set=model_config.model_task_config.utils, request_type="expression.learner"
)
self.summary_model: LLMRequest = LLMRequest(
model_set=model_config.model_task_config.tool_use, request_type="expression.summary"
)
self.check_model: Optional[LLMRequest] = None # 检查用的 LLM 实例，延迟初始化
self.chat_id = chat_id
self.chat_stream = get_chat_manager().get_stream(chat_id)
self.chat_name = get_chat_manager().get_stream_name(chat_id) or chat_id
# 学习锁，防止并发执行学习任务
self._learning_lock = asyncio.Lock()
async def learn_and_store(
self,
messages: List[Any],
) -> Optional[List[Tuple[str, str, str]]]:
"""
学习并存储表达方式
Args:
messages: 外部传入的消息列表（必需）
num: 学习数量
timestamp_start: 学习开始的时间戳，如果为None则使用self.last_learning_time
"""
if not messages:
return None
random_msg = messages
# 学习用（开启行编号，便于溯源）
random_msg_str: str = await build_anonymous_messages(random_msg, show_ids=True)
prompt_template = prompt_manager.get_prompt("learn_style")
prompt_template.add_context("bot_name", global_config.bot.nickname)
prompt_template.add_context("chat_str", random_msg_str)
prompt = await prompt_manager.render_prompt(prompt_template)
# print(f"random_msg_str:{random_msg_str}")
# logger.info(f"学习{type_str}的prompt: {prompt}")
try:
response, _ = await self.express_learn_model.generate_response_async(prompt, temperature=0.3)
except Exception as e:
logger.error(f"学习表达方式失败,模型生成出错: {e}")
return None
# 解析 LLM 返回的表达方式列表和黑话列表（包含来源行编号）
expressions: List[Tuple[str, str, str]]
jargon_entries: List[Tuple[str, str]] # (content, source_id)
expressions, jargon_entries = parse_expression_response(response)
# 从缓存中检查 jargon 是否出现在 messages 中
cached_jargon_entries = self._check_cached_jargons_in_messages(random_msg)
if cached_jargon_entries:
# 合并缓存中的 jargon 条目（去重：如果 content 已存在则跳过）
existing_contents = {content for content, _ in jargon_entries}
for content, source_id in cached_jargon_entries:
if content not in existing_contents:
jargon_entries.append((content, source_id))
existing_contents.add(content)
logger.info(f"从缓存中检查到黑话: {content}")
# 检查表达方式数量，如果超过10个则放弃本次表达学习
if len(expressions) > 20:
logger.info(f"表达方式提取数量超过10个（实际{len(expressions)}个），放弃本次表达学习")
expressions = []
# 检查黑话数量，如果超过30个则放弃本次黑话学习
if len(jargon_entries) > 30:
logger.info(f"黑话提取数量超过30个（实际{len(jargon_entries)}个），放弃本次黑话学习")
jargon_entries = []
# 处理黑话条目，路由到 jargon_miner（即使没有表达方式也要处理黑话）
if jargon_entries:
await self._process_jargon_entries(jargon_entries, random_msg)
# 如果没有表达方式，直接返回
if not expressions:
logger.info("解析后没有可用的表达方式")
return []
logger.info(f"学习的prompt: {prompt}")
logger.info(f"学习的expressions: {expressions}")
logger.info(f"学习的jargon_entries: {jargon_entries}")
logger.info(f"学习的response: {response}")
# 过滤表达方式，根据 source_id 溯源并应用各种过滤规则
learnt_expressions = self._filter_expressions(expressions, random_msg)
if learnt_expressions is None:
logger.info("没有学习到表达风格")
return []
# 展示学到的表达方式
learnt_expressions_str = ""
for situation, style in learnt_expressions:
learnt_expressions_str += f"{situation}->{style}\n"
logger.info(f"在 {self.chat_name} 学习到表达风格:\n{learnt_expressions_str}")
current_time = time.time()
# 存储到数据库 Expression 表
for situation, style in learnt_expressions:
await self._upsert_expression_record(
situation=situation,
style=style,
current_time=current_time,
)
return learnt_expressions
def _filter_expressions(
self,
expressions: List[Tuple[str, str, str]],
messages: List[Any],
) -> List[Tuple[str, str, str]]:
"""
过滤表达方式，移除不符合条件的条目
Args:
expressions: 表达方式列表，每个元素是 (situation, style, source_id)
messages: 原始消息列表，用于溯源和验证
Returns:
过滤后的表达方式列表，每个元素是 (situation, style, context)
"""
filtered_expressions: List[Tuple[str, str, str]] = [] # (situation, style, context)
# 准备机器人名称集合（用于过滤 style 与机器人名称重复的表达）
banned_names = set()
bot_nickname = (global_config.bot.nickname or "").strip()
if bot_nickname:
banned_names.add(bot_nickname)
alias_names = global_config.bot.alias_names or []
for alias in alias_names:
alias = alias.strip()
if alias:
banned_names.add(alias)
banned_casefold = {name.casefold() for name in banned_names if name}
for situation, style, source_id in expressions:
source_id_str = (source_id or "").strip()
if not source_id_str.isdigit():
# 无效的来源行编号，跳过
continue
line_index = int(source_id_str) - 1 # build_anonymous_messages 的编号从 1 开始
if line_index < 0 or line_index >= len(messages):
# 超出范围，跳过
continue
# 当前行的原始内容
current_msg = messages[line_index]
# 过滤掉从bot自己发言中提取到的表达方式
if is_bot_message(current_msg):
continue
context = filter_message_content(current_msg.processed_plain_text or "")
if not context:
continue
# 过滤掉包含 SELF 的内容（不学习）
if "SELF" in (situation or "") or "SELF" in (style or "") or "SELF" in context:
logger.info(f"跳过包含 SELF 的表达方式: situation={situation}, style={style}, source_id={source_id}")
continue
# 过滤掉 style 与机器人名称/昵称重复的表达
normalized_style = (style or "").strip()
if normalized_style and normalized_style.casefold() in banned_casefold:
logger.debug(
f"跳过 style 与机器人名称重复的表达方式: situation={situation}, style={style}, source_id={source_id}"
)
continue
# 过滤掉包含 "表情：" 或 "表情:" 的内容
if (
"表情：" in (situation or "")
or "表情:" in (situation or "")
or "表情：" in (style or "")
or "表情:" in (style or "")
or "表情：" in context
or "表情:" in context
):
logger.info(f"跳过包含表情标记的表达方式: situation={situation}, style={style}, source_id={source_id}")
continue
# 过滤掉包含 "[图片" 的内容
if "[图片" in (situation or "") or "[图片" in (style or "") or "[图片" in context:
logger.info(f"跳过包含图片标记的表达方式: situation={situation}, style={style}, source_id={source_id}")
continue
filtered_expressions.append((situation, style))
return filtered_expressions
async def _upsert_expression_record(
self,
situation: str,
style: str,
current_time: float,
) -> None:
# 检查是否有相似的 situation（相似度 >= 0.75，检查 content_list）
# 完全匹配（相似度 == 1.0）和相似匹配（相似度 >= 0.75）统一处理
expr_obj, similarity = await self._find_similar_situation_expression(situation, similarity_threshold=0.75)
if expr_obj:
# 根据相似度决定是否使用 LLM 总结
# 完全匹配（相似度 == 1.0）时不总结，相似匹配时总结
use_llm_summary = similarity < 1.0
await self._update_existing_expression(
expr_obj=expr_obj,
situation=situation,
current_time=current_time,
use_llm_summary=use_llm_summary,
)
return
# 没有找到匹配的记录，创建新记录
await self._create_expression_record(
situation=situation,
style=style,
current_time=current_time,
)
async def _create_expression_record(
self,
situation: str,
style: str,
current_time: float,
) -> None:
content_list = [situation]
# 创建新记录时，直接使用原始的 situation，不进行总结
formatted_situation = situation
Expression.create(
situation=formatted_situation,
style=style,
content_list=json.dumps(content_list, ensure_ascii=False),
count=1,
last_active_time=current_time,
chat_id=self.chat_id,
create_date=current_time,
)
async def _update_existing_expression(
self,
expr_obj: Expression,
situation: str,
current_time: float,
use_llm_summary: bool = True,
) -> None:
"""
更新现有 Expression 记录（situation 完全匹配或相似的情况）
将新的 situation 添加到 content_list，不合并 style
Args:
use_llm_summary: 是否使用 LLM 进行总结，完全匹配时为 False，相似匹配时为 True
"""
# 更新 content_list（添加新的 situation）
content_list = self._parse_content_list(expr_obj.content_list)
content_list.append(situation)
expr_obj.content_list = json.dumps(content_list, ensure_ascii=False)
# 更新其他字段
expr_obj.count = (expr_obj.count or 0) + 1
expr_obj.checked = False # count 增加时重置 checked 为 False
expr_obj.last_active_time = current_time
if use_llm_summary:
# 相似匹配时，使用 LLM 重新组合 situation
new_situation = await self._compose_situation_text(
content_list=content_list,
fallback=expr_obj.situation,
)
expr_obj.situation = new_situation
expr_obj.save()
# count 增加后，立即进行一次检查
await self._check_expression_immediately(expr_obj)
def _parse_content_list(self, stored_list: Optional[str]) -> List[str]:
if not stored_list:
return []
try:
data = json.loads(stored_list)
except json.JSONDecodeError:
return []
return [str(item) for item in data if isinstance(item, str)] if isinstance(data, list) else []
async def _find_similar_situation_expression(
self, situation: str, similarity_threshold: float = 0.75
) -> Tuple[Optional[Expression], float]:
"""
查找具有相似 situation 的 Expression 记录
检查 content_list 中的每一项
Args:
situation: 要查找的 situation
similarity_threshold: 相似度阈值，默认 0.75
Returns:
Tuple[Optional[Expression], float]:
- 找到的最相似的 Expression 对象，如果没有找到则返回 None
- 相似度值（如果找到匹配，范围在 similarity_threshold 到 1.0 之间）
"""
# 查询同一 chat_id 的所有记录
all_expressions = Expression.select().where(Expression.chat_id == self.chat_id)
best_match = None
best_similarity = 0.0
for expr in all_expressions:
# 检查 content_list 中的每一项
content_list = self._parse_content_list(expr.content_list)
for existing_situation in content_list:
similarity = calculate_similarity(situation, existing_situation)
if similarity >= similarity_threshold and similarity > best_similarity:
best_similarity = similarity
best_match = expr
if best_match:
logger.debug(
f"找到相似的 situation: 相似度={best_similarity:.3f}, 现有='{best_match.situation}', 新='{situation}'"
)
return best_match, best_similarity
async def _compose_situation_text(self, content_list: List[str], fallback: str = "") -> str:
sanitized = [c.strip() for c in content_list if c.strip()]
if not sanitized:
return fallback
prompt = (
"请阅读以下多个聊天情境描述，并将它们概括成一句简短的话，"
"长度不超过20个字，保留共同特点：\n"
f"{chr(10).join(f'- {s}' for s in sanitized[-10:])}\n只输出概括内容。"
)
try:
summary, _ = await self.summary_model.generate_response_async(prompt, temperature=0.2)
summary = summary.strip()
if summary:
return summary
except Exception as e:
logger.error(f"概括表达情境失败: {e}")
return "/".join(sanitized) if sanitized else fallback
async def _init_check_model(self) -> None:
"""初始化检查用的 LLM 实例"""
if self.check_model is None:
try:
self.check_model = LLMRequest(
model_set=model_config.model_task_config.tool_use, request_type="expression.check"
)
logger.debug("检查用 LLM 实例初始化成功")
except Exception as e:
logger.error(f"创建检查用 LLM 实例失败: {e}")
async def _check_expression_immediately(self, expr_obj: Expression) -> None:
"""
立即检查表达方式（在 count 增加后调用）
Args:
expr_obj: 要检查的表达方式对象
"""
try:
# 检查是否启用自动检查
if not global_config.expression.expression_self_reflect:
logger.debug("表达方式自动检查未启用，跳过立即检查")
return
# 初始化检查用的 LLM
await self._init_check_model()
if self.check_model is None:
logger.warning("检查用 LLM 实例初始化失败，跳过立即检查")
return
# 执行 LLM 评估
suitable, reason, error = await single_expression_check(expr_obj.situation, expr_obj.style)
# 更新数据库
expr_obj.checked = True
expr_obj.rejected = not suitable # 通过则 rejected=False，不通过则 rejected=True
expr_obj.save()
status = "通过" if suitable else "不通过"
logger.info(
f"表达方式立即检查完成 [ID: {expr_obj.id}] - {status} | "
f"Situation: {expr_obj.situation[:30]}... | "
f"Style: {expr_obj.style[:30]}... | "
f"Reason: {reason[:50] if reason else '无'}..."
)
if error:
logger.warning(f"表达方式立即检查时出现错误 [ID: {expr_obj.id}]: {error}")
except Exception as e:
logger.error(f"立即检查表达方式失败 [ID: {expr_obj.id}]: {e}", exc_info=True)
# 检查失败时，保持 checked=False，等待后续自动检查任务处理
def _check_cached_jargons_in_messages(self, messages: List[Any]) -> List[Tuple[str, str]]:
"""
检查缓存中的 jargon 是否出现在 messages 中
Args:
messages: 消息列表
Returns:
List[Tuple[str, str]]: 匹配到的黑话条目列表，每个元素是 (content, source_id)
"""
if not messages:
return []
# 获取 jargon_miner 实例
jargon_miner = miner_manager.get_miner(self.chat_id)
# 获取缓存中的所有 jargon
cached_jargons = jargon_miner.get_cached_jargons()
if not cached_jargons:
return []
matched_entries: List[Tuple[str, str]] = []
# 遍历 messages，检查缓存中的 jargon 是否出现
for i, msg in enumerate(messages):
# 跳过机器人自己的消息
if is_bot_message(msg):
continue
# 获取消息文本
msg_text = (getattr(msg, "processed_plain_text", None) or "").strip()
if not msg_text:
continue
# 检查每个缓存中的 jargon 是否出现在消息文本中
for jargon in cached_jargons:
if not jargon or not jargon.strip():
continue
jargon_content = jargon.strip()
# 使用正则匹配，考虑单词边界（类似 jargon_explainer 中的逻辑）
pattern = re.escape(jargon_content)
# 对于中文，使用更宽松的匹配；对于英文/数字，使用单词边界
if re.search(r"[\u4e00-\u9fff]", jargon_content):
# 包含中文，使用更宽松的匹配
search_pattern = pattern
else:
# 纯英文/数字，使用单词边界
search_pattern = r"\b" + pattern + r"\b"
if re.search(search_pattern, msg_text, re.IGNORECASE):
# 找到匹配，构建条目（source_id 从 1 开始，因为 build_anonymous_messages 的编号从 1 开始）
source_id = str(i + 1)
matched_entries.append((jargon_content, source_id))
return matched_entries
async def _process_jargon_entries(self, jargon_entries: List[Tuple[str, str]], messages: List[Any]) -> None:
"""
处理从 expression learner 提取的黑话条目，路由到 jargon_miner
Args:
jargon_entries: 黑话条目列表，每个元素是 (content, source_id)
messages: 消息列表，用于构建上下文
"""
if not jargon_entries or not messages:
return
# 获取 jargon_miner 实例
jargon_miner = miner_manager.get_miner(self.chat_id)
# 构建黑话条目格式，与 jargon_miner.run_once 中的格式一致
entries: List[Dict[str, List[str]]] = []
for content, source_id in jargon_entries:
content = content.strip()
if not content:
continue
# 过滤掉包含 SELF 的黑话，不学习
if "SELF" in content:
logger.info(f"跳过包含 SELF 的黑话: {content}")
continue
# 检查是否包含机器人名称
if contains_bot_self_name(content):
logger.info(f"跳过包含机器人昵称/别名的黑话: {content}")
continue
# 解析 source_id
source_id_str = (source_id or "").strip()
if not source_id_str.isdigit():
logger.warning(f"黑话条目 source_id 无效: content={content}, source_id={source_id_str}")
continue
# build_anonymous_messages 的编号从 1 开始
line_index = int(source_id_str) - 1
if line_index < 0 or line_index >= len(messages):
logger.warning(f"黑话条目 source_id 超出范围: content={content}, source_id={source_id_str}")
continue
# 检查是否是机器人自己的消息
target_msg = messages[line_index]
if is_bot_message(target_msg):
logger.info(f"跳过引用机器人自身消息的黑话: content={content}, source_id={source_id_str}")
continue
# 构建上下文段落
context_paragraph = build_context_paragraph(messages, line_index)
if not context_paragraph:
logger.warning(f"黑话条目上下文为空: content={content}, source_id={source_id_str}")
continue
entries.append({"content": content, "raw_content": [context_paragraph]})
if not entries:
return
# 调用 jargon_miner 处理这些条目
await jargon_miner.process_extracted_entries(entries)
class ExpressionLearnerManager:
def __init__(self):
self.expression_learners = {}
self._ensure_expression_directories()
def get_expression_learner(self, chat_id: str) -> ExpressionLearner:
if chat_id not in self.expression_learners:
self.expression_learners[chat_id] = ExpressionLearner(chat_id)
return self.expression_learners[chat_id]
def _ensure_expression_directories(self):
"""
确保表达方式相关的目录结构存在
"""
base_dir = os.path.join("data", "expression")
directories_to_create = [
base_dir,
os.path.join(base_dir, "learnt_style"),
os.path.join(base_dir, "learnt_grammar"),
]
for directory in directories_to_create:
try:
os.makedirs(directory, exist_ok=True)
logger.debug(f"确保目录存在: {directory}")
except Exception as e:
logger.error(f"创建目录失败 {directory}: {e}")
expression_learner_manager = ExpressionLearnerManager()

86
src/bw_learner/jargon_explainer.py Normal file
View File

@@ -0,0 +1,86 @@
from typing import Optional, Dict, List
from sqlmodel import select, func as fn
import json
from src.common.database.database import get_db_session
from src.common.database.database_model import Jargon
from src.common.logger import get_logger
from src.config.config import global_config
logger = get_logger("jargon_explainer")
def search_jargon(
keyword: str,
chat_id: Optional[str] = None,
limit: int = 10,
case_sensitive: bool = False,
fuzzy: bool = True,
) -> List[Dict[str, str]]:
"""
搜索 jargon，支持大小写不敏感和模糊搜索
Args:
keyword: 搜索关键词
chat_id: 可选的聊天 ID（session_id）
- 如果开启了 all_global：此参数被忽略，查询所有 is_global=True 的记录
- 如果关闭了 all_global：如果提供则优先搜索该聊天或 global 的 jargon
limit: 返回结果数量限制，默认 10
case_sensitive: 是否大小写敏感，默认 False（不敏感）
fuzzy: 是否模糊搜索，默认 True（使用 LIKE 匹配）
Returns:
List[Dict[str, str]]: 包含 content, meaning 的字典列表
"""
if not keyword or not keyword.strip():
return []
keyword = keyword.strip()
# 构建搜索条件
if case_sensitive: # 大小写敏感
search_condition = Jargon.content.contains(keyword) if fuzzy else Jargon.content == keyword # type: ignore
else:
keyword_lower = keyword.lower()
search_condition = (
fn.LOWER(Jargon.content).contains(keyword_lower) if fuzzy else fn.LOWER(Jargon.content) == keyword_lower
)
# 根据 all_global 配置决定查询逻辑同时，限制结果数量（先多取一些，因为后面可能过滤）
if global_config.expression.all_global_jargon:
# 开启 all_global：所有记录都是全局的，查询所有 is_global=True 的记录（无视 chat_id）
query = select(Jargon).where(search_condition, Jargon.is_global).order_by(Jargon.count.desc()).limit(limit * 2) # type: ignore
else:
# 关闭 all_global：查询所有记录，chat_id 过滤在 Python 层面进行
query = select(Jargon).where(search_condition).order_by(Jargon.count.desc()).limit(limit * 2) # type: ignore
# 执行查询并返回结果
results: List[Dict[str, str]] = []
with get_db_session() as session:
jargons = session.exec(query).all()
for jargon in jargons:
# 如果提供了 chat_id 且 all_global=False，需要检查 session_id_dict 是否包含目标 chat_id
if chat_id and not global_config.expression.all_global_jargon and not jargon.is_global:
try: # 解析 session_id_dict
session_id_dict = json.loads(jargon.session_id_dict) if jargon.session_id_dict else {}
except (json.JSONDecodeError, TypeError):
session_id_dict = {}
logger.warning(
f"解析 session_id_dict 失败，jargon_id={jargon.id}，原始数据：{jargon.session_id_dict}"
)
# 检查是否包含目标 chat_id
if chat_id not in session_id_dict:
continue
# 只返回有 meaning 的记录
if not jargon.meaning.strip():
continue
results.append({"content": jargon.content or "", "meaning": jargon.meaning or ""})
# 达到限制数量后停止
if len(results) >= limit:
break
return results

589
src/bw_learner/jargon_miner_old.py
View File

@@ -1,589 +0,0 @@
import json
import asyncio
import random
from collections import OrderedDict
from typing import List, Dict, Optional, Callable
from json_repair import repair_json
from sqlalchemy import func as fn
from src.common.logger import get_logger
from src.common.database.database_model import Jargon
from src.llm_models.utils_model import LLMRequest
from src.config.config import model_config, global_config
from src.chat.message_receive.chat_stream import get_chat_manager
from src.prompt.prompt_manager import prompt_manager
from src.bw_learner.learner_utils_old import (
parse_chat_id_list,
chat_id_list_contains,
update_chat_id_list,
)
logger = get_logger("jargon")
def _is_single_char_jargon(content: str) -> bool:
"""
判断是否是单字黑话（单个汉字、英文或数字）
Args:
content: 词条内容
Returns:
bool: 如果是单字黑话返回True，否则返回False
"""
if not content or len(content) != 1:
return False
char = content[0]
# 判断是否是单个汉字、单个英文字母或单个数字
return (
"\u4e00" <= char <= "\u9fff" # 汉字
or "a" <= char <= "z" # 小写字母
or "A" <= char <= "Z" # 大写字母
or "0" <= char <= "9" # 数字
)
def _should_infer_meaning(jargon_obj: Jargon) -> bool:
"""
判断是否需要进行含义推断
在 count 达到 3,6, 10, 20, 40, 60, 100 时进行推断
并且count必须大于last_inference_count，避免重启后重复判定
如果is_complete为True，不再进行推断
"""
# 如果已完成所有推断，不再推断
if jargon_obj.is_complete:
return False
count = jargon_obj.count or 0
last_inference = jargon_obj.last_inference_count or 0
# 阈值列表：3,6, 10, 20, 40, 60, 100
thresholds = [2, 4, 8, 12, 24, 60, 100]
if count < thresholds[0]:
return False
# 如果count没有超过上次判定值，不需要判定
if count <= last_inference:
return False
# 找到第一个大于last_inference的阈值
next_threshold = None
for threshold in thresholds:
if threshold > last_inference:
next_threshold = threshold
break
# 如果没有找到下一个阈值，说明已经超过100，不应该再推断
if next_threshold is None:
return False
# 检查count是否达到或超过这个阈值
return count >= next_threshold
class JargonMiner:
def __init__(self, chat_id: str) -> None:
self.chat_id = chat_id
self.llm = LLMRequest(
model_set=model_config.model_task_config.utils,
request_type="jargon.extract",
)
self.llm_inference = LLMRequest(
model_set=model_config.model_task_config.utils,
request_type="jargon.inference",
)
# 初始化stream_name作为类属性，避免重复提取
chat_manager = get_chat_manager()
stream_name = chat_manager.get_stream_name(self.chat_id)
self.stream_name = stream_name if stream_name else self.chat_id
self.cache_limit = 50
self.cache: OrderedDict[str, None] = OrderedDict()
# 黑话提取锁，防止并发执行
self._extraction_lock = asyncio.Lock()
def _add_to_cache(self, content: str) -> None:
"""将提取到的黑话加入缓存，保持LRU语义"""
if not content:
return
key = content.strip()
if not key:
return
# 单字黑话（单个汉字、英文或数字）不记录到缓存
if _is_single_char_jargon(key):
return
if key in self.cache:
self.cache.move_to_end(key)
else:
self.cache[key] = None
if len(self.cache) > self.cache_limit:
self.cache.popitem(last=False)
def get_cached_jargons(self) -> List[str]:
"""获取缓存中的所有黑话列表"""
return list(self.cache.keys())
async def _infer_meaning_by_id(self, jargon_id: int) -> None:
"""通过ID加载对象并推断"""
try:
jargon_obj = Jargon.get_by_id(jargon_id)
# 再次检查is_complete，因为可能在异步任务执行时已被标记为完成
if jargon_obj.is_complete:
logger.debug(f"jargon {jargon_obj.content} 已完成所有推断，跳过")
return
await self.infer_meaning(jargon_obj)
except Exception as e:
logger.error(f"通过ID推断jargon失败: {e}")
async def infer_meaning(self, jargon_obj: Jargon) -> None:
"""
对jargon进行含义推断
"""
try:
content = jargon_obj.content
raw_content_str = jargon_obj.raw_content or ""
# 解析raw_content列表
raw_content_list = []
if raw_content_str:
try:
raw_content_list = (
json.loads(raw_content_str) if isinstance(raw_content_str, str) else raw_content_str
)
if not isinstance(raw_content_list, list):
raw_content_list = [raw_content_list] if raw_content_list else []
except (json.JSONDecodeError, TypeError):
raw_content_list = [raw_content_str] if raw_content_str else []
if not raw_content_list:
logger.warning(f"jargon {content} 没有raw_content，跳过推断")
return
# 获取当前count和上一次的meaning
current_count = jargon_obj.count or 0
previous_meaning = jargon_obj.meaning or ""
# 当count为24, 60时，随机移除一半的raw_content项目
if current_count in [24, 60] and len(raw_content_list) > 1:
# 计算要保留的数量（至少保留1个）
keep_count = max(1, len(raw_content_list) // 2)
raw_content_list = random.sample(raw_content_list, keep_count)
logger.info(
f"jargon {content} count={current_count}，随机移除后剩余 {len(raw_content_list)} 个raw_content项目"
)
# 步骤1: 基于raw_content和content推断
raw_content_text = "\n".join(raw_content_list)
# 当count为24, 60, 100时，在prompt中放入上一次推断出的meaning作为参考
previous_meaning_section = ""
previous_meaning_instruction = ""
if current_count in [24, 60, 100] and previous_meaning:
previous_meaning_section = f"\n**上一次推断的含义（仅供参考）**\n{previous_meaning}"
previous_meaning_instruction = (
"- 请参考上一次推断的含义，结合新的上下文信息，给出更准确或更新的推断结果"
)
prompt1_template = prompt_manager.get_prompt("jargon_inference_with_context")
prompt1_template.add_context("bot_name", global_config.bot.nickname)
prompt1_template.add_context("content", str(content))
prompt1_template.add_context("raw_content_list", raw_content_text)
prompt1_template.add_context("previous_meaning_section", previous_meaning_section)
prompt1_template.add_context("previous_meaning_instruction", previous_meaning_instruction)
prompt1 = await prompt_manager.render_prompt(prompt1_template)
response1, _ = await self.llm_inference.generate_response_async(prompt1, temperature=0.3)
if not response1:
logger.warning(f"jargon {content} 推断1失败：无响应")
return
# 解析推断1结果
inference1 = None
try:
resp1 = response1.strip()
if resp1.startswith("{") and resp1.endswith("}"):
inference1 = json.loads(resp1)
else:
repaired = repair_json(resp1)
inference1 = json.loads(repaired) if isinstance(repaired, str) else repaired
if not isinstance(inference1, dict):
logger.warning(f"jargon {content} 推断1结果格式错误")
return
except Exception as e:
logger.error(f"jargon {content} 推断1解析失败: {e}")
return
# 检查推断1是否表示信息不足无法推断
no_info = inference1.get("no_info", False)
meaning1 = inference1.get("meaning", "").strip()
if no_info or not meaning1:
logger.info(f"jargon {content} 推断1表示信息不足无法推断，放弃本次推断，待下次更新")
# 更新最后一次判定的count值，避免在同一阈值重复尝试
jargon_obj.last_inference_count = jargon_obj.count or 0
jargon_obj.save()
return
# 步骤2: 仅基于content推断
prompt2_template = prompt_manager.get_prompt("jargon_inference_content_only")
prompt2_template.add_context("content", str(content))
prompt2 = await prompt_manager.render_prompt(prompt2_template)
response2, _ = await self.llm_inference.generate_response_async(prompt2, temperature=0.3)
if not response2:
logger.warning(f"jargon {content} 推断2失败：无响应")
return
# 解析推断2结果
inference2 = None
try:
resp2 = response2.strip()
if resp2.startswith("{") and resp2.endswith("}"):
inference2 = json.loads(resp2)
else:
repaired = repair_json(resp2)
inference2 = json.loads(repaired) if isinstance(repaired, str) else repaired
if not isinstance(inference2, dict):
logger.warning(f"jargon {content} 推断2结果格式错误")
return
except Exception as e:
logger.error(f"jargon {content} 推断2解析失败: {e}")
return
# logger.info(f"jargon {content} 推断2提示词: {prompt2}")
# logger.info(f"jargon {content} 推断2结果: {response2}")
# logger.info(f"jargon {content} 推断1提示词: {prompt1}")
# logger.info(f"jargon {content} 推断1结果: {response1}")
if global_config.debug.show_jargon_prompt:
logger.info(f"jargon {content} 推断2提示词: {prompt2}")
logger.info(f"jargon {content} 推断2结果: {response2}")
logger.info(f"jargon {content} 推断1提示词: {prompt1}")
logger.info(f"jargon {content} 推断1结果: {response1}")
else:
logger.debug(f"jargon {content} 推断2提示词: {prompt2}")
logger.debug(f"jargon {content} 推断2结果: {response2}")
logger.debug(f"jargon {content} 推断1提示词: {prompt1}")
logger.debug(f"jargon {content} 推断1结果: {response1}")
# 步骤3: 比较两个推断结果
prompt3_template = prompt_manager.get_prompt("jargon_compare_inference")
prompt3_template.add_context("inference1", json.dumps(inference1, ensure_ascii=False))
prompt3_template.add_context("inference2", json.dumps(inference2, ensure_ascii=False))
prompt3 = await prompt_manager.render_prompt(prompt3_template)
if global_config.debug.show_jargon_prompt:
logger.info(f"jargon {content} 比较提示词: {prompt3}")
response3, _ = await self.llm_inference.generate_response_async(prompt3, temperature=0.3)
if not response3:
logger.warning(f"jargon {content} 比较失败：无响应")
return
# 解析比较结果
comparison = None
try:
resp3 = response3.strip()
if resp3.startswith("{") and resp3.endswith("}"):
comparison = json.loads(resp3)
else:
repaired = repair_json(resp3)
comparison = json.loads(repaired) if isinstance(repaired, str) else repaired
if not isinstance(comparison, dict):
logger.warning(f"jargon {content} 比较结果格式错误")
return
except Exception as e:
logger.error(f"jargon {content} 比较解析失败: {e}")
return
# 判断是否为黑话
is_similar = comparison.get("is_similar", False)
is_jargon = not is_similar # 如果相似，说明不是黑话；如果有差异，说明是黑话
# 更新数据库记录
jargon_obj.is_jargon = is_jargon
jargon_obj.meaning = inference1.get("meaning", "") if is_jargon else ""
# 更新最后一次判定的count值，避免重启后重复判定
jargon_obj.last_inference_count = jargon_obj.count or 0
# 如果count>=100，标记为完成，不再进行推断
if (jargon_obj.count or 0) >= 100:
jargon_obj.is_complete = True
jargon_obj.save()
logger.debug(
f"jargon {content} 推断完成: is_jargon={is_jargon}, meaning={jargon_obj.meaning}, last_inference_count={jargon_obj.last_inference_count}, is_complete={jargon_obj.is_complete}"
)
# 固定输出推断结果，格式化为可读形式
if is_jargon:
# 是黑话，输出格式：[聊天名]xxx的含义是 xxxxxxxxxxx
meaning = jargon_obj.meaning or "无详细说明"
is_global = jargon_obj.is_global
if is_global:
logger.info(f"[黑话]{content}的含义是 {meaning}")
else:
logger.info(f"[{self.stream_name}]{content}的含义是 {meaning}")
else:
# 不是黑话，输出格式：[聊天名]xxx 不是黑话
logger.info(f"[{self.stream_name}]{content} 不是黑话")
except Exception as e:
logger.error(f"jargon推断失败: {e}")
import traceback
traceback.print_exc()
async def process_extracted_entries(
self, entries: List[Dict[str, List[str]]], person_name_filter: Optional[Callable[[str], bool]] = None
) -> None:
"""
处理已提取的黑话条目（从 expression_learner 路由过来的）
Args:
entries: 黑话条目列表，每个元素格式为 {"content": "...", "raw_content": [...]}
person_name_filter: 可选的过滤函数，用于检查内容是否包含人物名称
"""
if not entries:
return
try:
# 去重并合并raw_content（按 content 聚合）
merged_entries: OrderedDict[str, Dict[str, List[str]]] = OrderedDict()
for entry in entries:
content_key = entry["content"]
# 检查是否包含人物名称
# logger.info(f"process_extracted_entries 检查是否包含人物名称: {content_key}")
# logger.info(f"person_name_filter: {person_name_filter}")
if person_name_filter and person_name_filter(content_key):
logger.info(f"process_extracted_entries 跳过包含人物名称的黑话: {content_key}")
continue
raw_list = entry.get("raw_content", []) or []
if content_key in merged_entries:
merged_entries[content_key]["raw_content"].extend(raw_list)
else:
merged_entries[content_key] = {
"content": content_key,
"raw_content": list(raw_list),
}
uniq_entries = []
for merged_entry in merged_entries.values():
raw_content_list = merged_entry["raw_content"]
if raw_content_list:
merged_entry["raw_content"] = list(dict.fromkeys(raw_content_list))
uniq_entries.append(merged_entry)
saved = 0
updated = 0
for entry in uniq_entries:
content = entry["content"]
raw_content_list = entry["raw_content"] # 已经是列表
try:
# 查询所有content匹配的记录
query = Jargon.select().where(Jargon.content == content)
# 查找匹配的记录
matched_obj = None
for obj in query:
if global_config.expression.all_global_jargon:
# 开启all_global：所有content匹配的记录都可以
matched_obj = obj
break
else:
# 关闭all_global：需要检查chat_id列表是否包含目标chat_id
chat_id_list = parse_chat_id_list(obj.chat_id)
if chat_id_list_contains(chat_id_list, self.chat_id):
matched_obj = obj
break
if matched_obj:
obj = matched_obj
try:
obj.count = (obj.count or 0) + 1
except Exception:
obj.count = 1
# 合并raw_content列表：读取现有列表，追加新值，去重
existing_raw_content = []
if obj.raw_content:
try:
existing_raw_content = (
json.loads(obj.raw_content) if isinstance(obj.raw_content, str) else obj.raw_content
)
if not isinstance(existing_raw_content, list):
existing_raw_content = [existing_raw_content] if existing_raw_content else []
except (json.JSONDecodeError, TypeError):
existing_raw_content = [obj.raw_content] if obj.raw_content else []
# 合并并去重
merged_list = list(dict.fromkeys(existing_raw_content + raw_content_list))
obj.raw_content = json.dumps(merged_list, ensure_ascii=False)
# 更新chat_id列表：增加当前chat_id的计数
chat_id_list = parse_chat_id_list(obj.chat_id)
updated_chat_id_list = update_chat_id_list(chat_id_list, self.chat_id, increment=1)
obj.chat_id = json.dumps(updated_chat_id_list, ensure_ascii=False)
# 开启all_global时，确保记录标记为is_global=True
if global_config.expression.all_global_jargon:
obj.is_global = True
# 关闭all_global时，保持原有is_global不变（不修改）
obj.save()
# 检查是否需要推断（达到阈值且超过上次判定值）
if _should_infer_meaning(obj):
# 异步触发推断，不阻塞主流程
# 重新加载对象以确保数据最新
jargon_id = obj.id
asyncio.create_task(self._infer_meaning_by_id(jargon_id))
updated += 1
else:
# 没找到匹配记录，创建新记录
if global_config.expression.all_global_jargon:
# 开启all_global：新记录默认为is_global=True
is_global_new = True
else:
# 关闭all_global：新记录is_global=False
is_global_new = False
# 使用新格式创建chat_id列表：[[chat_id, count]]
chat_id_list = [[self.chat_id, 1]]
chat_id_json = json.dumps(chat_id_list, ensure_ascii=False)
Jargon.create(
content=content,
raw_content=json.dumps(raw_content_list, ensure_ascii=False),
chat_id=chat_id_json,
is_global=is_global_new,
count=1,
)
saved += 1
except Exception as e:
logger.error(f"保存jargon失败: chat_id={self.chat_id}, content={content}, err={e}")
continue
finally:
self._add_to_cache(content)
# 固定输出提取的jargon结果，格式化为可读形式（只要有提取结果就输出）
if uniq_entries:
# 收集所有提取的jargon内容
jargon_list = [entry["content"] for entry in uniq_entries]
jargon_str = ",".join(jargon_list)
# 输出格式化的结果（使用logger.info会自动应用jargon模块的颜色）
logger.info(f"[{self.stream_name}]疑似黑话: {jargon_str}")
if saved or updated:
logger.debug(f"jargon写入: 新增 {saved} 条，更新 {updated} 条，chat_id={self.chat_id}")
except Exception as e:
logger.error(f"处理已提取的黑话条目失败: {e}")
class JargonMinerManager:
def __init__(self) -> None:
self._miners: dict[str, JargonMiner] = {}
def get_miner(self, chat_id: str) -> JargonMiner:
if chat_id not in self._miners:
self._miners[chat_id] = JargonMiner(chat_id)
return self._miners[chat_id]
miner_manager = JargonMinerManager()
def search_jargon(
keyword: str, chat_id: Optional[str] = None, limit: int = 10, case_sensitive: bool = False, fuzzy: bool = True
) -> List[Dict[str, str]]:
"""
搜索jargon，支持大小写不敏感和模糊搜索
Args:
keyword: 搜索关键词
chat_id: 可选的聊天ID
- 如果开启了all_global：此参数被忽略，查询所有is_global=True的记录
- 如果关闭了all_global：如果提供则优先搜索该聊天或global的jargon
limit: 返回结果数量限制，默认10
case_sensitive: 是否大小写敏感，默认False（不敏感）
fuzzy: 是否模糊搜索，默认True（使用LIKE匹配）
Returns:
List[Dict[str, str]]: 包含content, meaning的字典列表
"""
if not keyword or not keyword.strip():
return []
keyword = keyword.strip()
# 构建查询（选择所有需要的字段，以便后续过滤）
query = Jargon.select()
# 构建搜索条件
if case_sensitive:
# 大小写敏感
if fuzzy:
# 模糊搜索
search_condition = Jargon.content.contains(keyword)
else:
# 精确匹配
search_condition = Jargon.content == keyword
else:
# 大小写不敏感
if fuzzy:
# 模糊搜索（使用LOWER函数）
search_condition = fn.LOWER(Jargon.content).contains(keyword.lower())
else:
# 精确匹配（使用LOWER函数）
search_condition = fn.LOWER(Jargon.content) == keyword.lower()
query = query.where(search_condition)
# 根据all_global配置决定查询逻辑
if global_config.expression.all_global_jargon:
# 开启all_global：所有记录都是全局的，查询所有is_global=True的记录（无视chat_id）
query = query.where(Jargon.is_global)
# 注意：对于all_global=False的情况，chat_id过滤在Python层面进行，以便兼容新旧格式
# 注意：meaning的过滤移到Python层面，因为我们需要先过滤chat_id
# 按count降序排序，优先返回出现频率高的
query = query.order_by(Jargon.count.desc())
# 限制结果数量（先多取一些，因为后面可能过滤）
query = query.limit(limit * 2)
# 执行查询并返回结果，过滤chat_id
results = []
for jargon in query:
# 如果提供了chat_id且all_global=False，需要检查chat_id列表是否包含目标chat_id
if chat_id and not global_config.expression.all_global_jargon:
chat_id_list = parse_chat_id_list(jargon.chat_id)
# 如果记录是is_global=True，或者chat_id列表包含目标chat_id，则包含
if not jargon.is_global and not chat_id_list_contains(chat_id_list, chat_id):
continue
# 只返回有meaning的记录
if not jargon.meaning or jargon.meaning.strip() == "":
continue
results.append({"content": jargon.content or "", "meaning": jargon.meaning or ""})
# 达到限制数量后停止
if len(results) >= limit:
break
return results

439
src/bw_learner/learner_utils_old.py
View File

@@ -16,77 +16,6 @@ from json_repair import repair_json
logger = get_logger("learner_utils")
def filter_message_content(content: Optional[str]) -> str:
"""
过滤消息内容，移除回复、@、图片等格式
Args:
content: 原始消息内容
Returns:
str: 过滤后的内容
"""
if not content:
return ""
# 移除以[回复开头、]结尾的部分，包括后面的"，说："部分
content = re.sub(r"\[回复.*?\]，说：\s*", "", content)
# 移除@<...>格式的内容
content = re.sub(r"@<[^>]*>", "", content)
# 移除[picid:...]格式的图片ID
content = re.sub(r"\[picid:[^\]]*\]", "", content)
# 移除[表情包：...]格式的内容
content = re.sub(r"\[表情包：[^\]]*\]", "", content)
return content.strip()
def calculate_similarity(text1: str, text2: str) -> float:
"""
计算两个文本的相似度，返回0-1之间的值
使用SequenceMatcher计算相似度
Args:
text1: 第一个文本
text2: 第二个文本
Returns:
float: 相似度值，范围0-1
"""
return difflib.SequenceMatcher(None, text1, text2).ratio()
def calculate_style_similarity(style1: str, style2: str) -> float:
"""
计算两个 style 的相似度，返回0-1之间的值
在计算前会移除"使用"和"句式"这两个词（参考 expression_similarity_analysis.py）
Args:
style1: 第一个 style
style2: 第二个 style
Returns:
float: 相似度值，范围0-1
"""
if not style1 or not style2:
return 0.0
# 移除"使用"和"句式"这两个词
def remove_ignored_words(text: str) -> str:
"""移除需要忽略的词"""
text = text.replace("使用", "")
text = text.replace("句式", "")
return text.strip()
cleaned_style1 = remove_ignored_words(style1)
cleaned_style2 = remove_ignored_words(style2)
# 如果清理后文本为空，返回0
if not cleaned_style1 or not cleaned_style2:
return 0.0
return difflib.SequenceMatcher(None, cleaned_style1, cleaned_style2).ratio()
def _compute_weights(population: List[Dict]) -> List[float]:
"""
@@ -275,224 +204,224 @@ def contains_bot_self_name(content: str) -> bool:
return any(name in target for name in candidates)
def build_context_paragraph(messages: List[Any], center_index: int) -> Optional[str]:
"""
构建包含中心消息上下文的段落（前3条+后3条），使用标准的 readable builder 输出
"""
if not messages or center_index < 0 or center_index >= len(messages):
return None
# def build_context_paragraph(messages: List[Any], center_index: int) -> Optional[str]:
# """
# 构建包含中心消息上下文的段落（前3条+后3条），使用标准的 readable builder 输出
# """
# if not messages or center_index < 0 or center_index >= len(messages):
# return None
context_start = max(0, center_index - 3)
context_end = min(len(messages), center_index + 1 + 3)
context_messages = messages[context_start:context_end]
# context_start = max(0, center_index - 3)
# context_end = min(len(messages), center_index + 1 + 3)
# context_messages = messages[context_start:context_end]
if not context_messages:
return None
# if not context_messages:
# return None
try:
paragraph = build_readable_messages(
messages=context_messages,
replace_bot_name=True,
timestamp_mode="relative",
read_mark=0.0,
truncate=False,
show_actions=False,
show_pic=True,
message_id_list=None,
remove_emoji_stickers=False,
pic_single=True,
)
except Exception as e:
logger.warning(f"构建上下文段落失败: {e}")
return None
# try:
# paragraph = build_readable_messages(
# messages=context_messages,
# replace_bot_name=True,
# timestamp_mode="relative",
# read_mark=0.0,
# truncate=False,
# show_actions=False,
# show_pic=True,
# message_id_list=None,
# remove_emoji_stickers=False,
# pic_single=True,
# )
# except Exception as e:
# logger.warning(f"构建上下文段落失败: {e}")
# return None
paragraph = paragraph.strip()
return paragraph or None
# paragraph = paragraph.strip()
# return paragraph or None
def is_bot_message(msg: Any) -> bool:
"""判断消息是否来自机器人自身"""
if msg is None:
return False
# def is_bot_message(msg: Any) -> bool:
# """判断消息是否来自机器人自身"""
# if msg is None:
# return False
bot_config = getattr(global_config, "bot", None)
if not bot_config:
return False
# bot_config = getattr(global_config, "bot", None)
# if not bot_config:
# return False
platform = (
str(getattr(msg, "user_platform", "") or getattr(getattr(msg, "user_info", None), "platform", "") or "")
.strip()
.lower()
)
user_id = str(getattr(msg, "user_id", "") or getattr(getattr(msg, "user_info", None), "user_id", "") or "").strip()
# platform = (
# str(getattr(msg, "user_platform", "") or getattr(getattr(msg, "user_info", None), "platform", "") or "")
# .strip()
# .lower()
# )
# user_id = str(getattr(msg, "user_id", "") or getattr(getattr(msg, "user_info", None), "user_id", "") or "").strip()
if not platform or not user_id:
return False
# if not platform or not user_id:
# return False
platform_accounts = {}
try:
platform_accounts = parse_platform_accounts(getattr(bot_config, "platforms", []) or [])
except Exception:
platform_accounts = {}
# platform_accounts = {}
# try:
# platform_accounts = parse_platform_accounts(getattr(bot_config, "platforms", []) or [])
# except Exception:
# platform_accounts = {}
bot_accounts: Dict[str, str] = {}
qq_account = str(getattr(bot_config, "qq_account", "") or "").strip()
if qq_account:
bot_accounts["qq"] = qq_account
# bot_accounts: Dict[str, str] = {}
# qq_account = str(getattr(bot_config, "qq_account", "") or "").strip()
# if qq_account:
# bot_accounts["qq"] = qq_account
telegram_account = str(getattr(bot_config, "telegram_account", "") or "").strip()
if telegram_account:
bot_accounts["telegram"] = telegram_account
# telegram_account = str(getattr(bot_config, "telegram_account", "") or "").strip()
# if telegram_account:
# bot_accounts["telegram"] = telegram_account
for plat, account in platform_accounts.items():
if account and plat not in bot_accounts:
bot_accounts[plat] = account
# for plat, account in platform_accounts.items():
# if account and plat not in bot_accounts:
# bot_accounts[plat] = account
bot_account = bot_accounts.get(platform)
return bool(bot_account and user_id == bot_account)
# bot_account = bot_accounts.get(platform)
# return bool(bot_account and user_id == bot_account)
def parse_expression_response(response: str) -> Tuple[List[Tuple[str, str, str]], List[Tuple[str, str]]]:
"""
解析 LLM 返回的表达风格总结和黑话 JSON，提取两个列表。
# def parse_expression_response(response: str) -> Tuple[List[Tuple[str, str, str]], List[Tuple[str, str]]]:
# """
# 解析 LLM 返回的表达风格总结和黑话 JSON，提取两个列表。
期望的 JSON 结构：
[
{"situation": "AAAAA", "style": "BBBBB", "source_id": "3"}, // 表达方式
{"content": "词条", "source_id": "12"}, // 黑话
...
]
# 期望的 JSON 结构：
# [
# {"situation": "AAAAA", "style": "BBBBB", "source_id": "3"}, // 表达方式
# {"content": "词条", "source_id": "12"}, // 黑话
# ...
# ]
Returns:
Tuple[List[Tuple[str, str, str]], List[Tuple[str, str]]]:
第一个列表是表达方式 (situation, style, source_id)
第二个列表是黑话 (content, source_id)
"""
if not response:
return [], []
# Returns:
# Tuple[List[Tuple[str, str, str]], List[Tuple[str, str]]]:
# 第一个列表是表达方式 (situation, style, source_id)
# 第二个列表是黑话 (content, source_id)
# """
# if not response:
# return [], []
raw = response.strip()
# raw = response.strip()
# 尝试提取 ```json 代码块
json_block_pattern = r"```json\s*(.*?)\s*```"
match = re.search(json_block_pattern, raw, re.DOTALL)
if match:
raw = match.group(1).strip()
else:
# 去掉可能存在的通用 ``` 包裹
raw = re.sub(r"^```\s*", "", raw, flags=re.MULTILINE)
raw = re.sub(r"```\s*$", "", raw, flags=re.MULTILINE)
raw = raw.strip()
# # 尝试提取 ```json 代码块
# json_block_pattern = r"```json\s*(.*?)\s*```"
# match = re.search(json_block_pattern, raw, re.DOTALL)
# if match:
# raw = match.group(1).strip()
# else:
# # 去掉可能存在的通用 ``` 包裹
# raw = re.sub(r"^```\s*", "", raw, flags=re.MULTILINE)
# raw = re.sub(r"```\s*$", "", raw, flags=re.MULTILINE)
# raw = raw.strip()
parsed = None
expressions: List[Tuple[str, str, str]] = [] # (situation, style, source_id)
jargon_entries: List[Tuple[str, str]] = [] # (content, source_id)
# parsed = None
# expressions: List[Tuple[str, str, str]] = [] # (situation, style, source_id)
# jargon_entries: List[Tuple[str, str]] = [] # (content, source_id)
try:
# 优先尝试直接解析
if raw.startswith("[") and raw.endswith("]"):
parsed = json.loads(raw)
else:
repaired = repair_json(raw)
if isinstance(repaired, str):
parsed = json.loads(repaired)
else:
parsed = repaired
except Exception as parse_error:
# 如果解析失败，尝试修复中文引号问题
# 使用状态机方法，在 JSON 字符串值内部将中文引号替换为转义的英文引号
try:
# try:
# # 优先尝试直接解析
# if raw.startswith("[") and raw.endswith("]"):
# parsed = json.loads(raw)
# else:
# repaired = repair_json(raw)
# if isinstance(repaired, str):
# parsed = json.loads(repaired)
# else:
# parsed = repaired
# except Exception as parse_error:
# # 如果解析失败，尝试修复中文引号问题
# # 使用状态机方法，在 JSON 字符串值内部将中文引号替换为转义的英文引号
# try:
def fix_chinese_quotes_in_json(text):
"""使用状态机修复 JSON 字符串值中的中文引号"""
result = []
i = 0
in_string = False
escape_next = False
# def fix_chinese_quotes_in_json(text):
# """使用状态机修复 JSON 字符串值中的中文引号"""
# result = []
# i = 0
# in_string = False
# escape_next = False
while i < len(text):
char = text[i]
# while i < len(text):
# char = text[i]
if escape_next:
# 当前字符是转义字符后的字符，直接添加
result.append(char)
escape_next = False
i += 1
continue
# if escape_next:
# # 当前字符是转义字符后的字符，直接添加
# result.append(char)
# escape_next = False
# i += 1
# continue
if char == "\\":
# 转义字符
result.append(char)
escape_next = True
i += 1
continue
# if char == "\\":
# # 转义字符
# result.append(char)
# escape_next = True
# i += 1
# continue
if char == '"' and not escape_next:
# 遇到英文引号，切换字符串状态
in_string = not in_string
result.append(char)
i += 1
continue
# if char == '"' and not escape_next:
# # 遇到英文引号，切换字符串状态
# in_string = not in_string
# result.append(char)
# i += 1
# continue
if in_string:
# 在字符串值内部，将中文引号替换为转义的英文引号
if char == '"': # 中文左引号 U+201C
result.append('\\"')
elif char == '"': # 中文右引号 U+201D
result.append('\\"')
else:
result.append(char)
else:
# 不在字符串内，直接添加
result.append(char)
# if in_string:
# # 在字符串值内部，将中文引号替换为转义的英文引号
# if char == '"': # 中文左引号 U+201C
# result.append('\\"')
# elif char == '"': # 中文右引号 U+201D
# result.append('\\"')
# else:
# result.append(char)
# else:
# # 不在字符串内，直接添加
# result.append(char)
i += 1
# i += 1
return "".join(result)
# return "".join(result)
fixed_raw = fix_chinese_quotes_in_json(raw)
# fixed_raw = fix_chinese_quotes_in_json(raw)
# 再次尝试解析
if fixed_raw.startswith("[") and fixed_raw.endswith("]"):
parsed = json.loads(fixed_raw)
else:
repaired = repair_json(fixed_raw)
if isinstance(repaired, str):
parsed = json.loads(repaired)
else:
parsed = repaired
except Exception as fix_error:
logger.error(f"解析表达风格 JSON 失败，初始错误: {type(parse_error).__name__}: {str(parse_error)}")
logger.error(f"修复中文引号后仍失败，错误: {type(fix_error).__name__}: {str(fix_error)}")
logger.error(f"解析表达风格 JSON 失败，原始响应：{response}")
logger.error(f"处理后的 JSON 字符串（前500字符）：{raw[:500]}")
return [], []
# # 再次尝试解析
# if fixed_raw.startswith("[") and fixed_raw.endswith("]"):
# parsed = json.loads(fixed_raw)
# else:
# repaired = repair_json(fixed_raw)
# if isinstance(repaired, str):
# parsed = json.loads(repaired)
# else:
# parsed = repaired
# except Exception as fix_error:
# logger.error(f"解析表达风格 JSON 失败，初始错误: {type(parse_error).__name__}: {str(parse_error)}")
# logger.error(f"修复中文引号后仍失败，错误: {type(fix_error).__name__}: {str(fix_error)}")
# logger.error(f"解析表达风格 JSON 失败，原始响应：{response}")
# logger.error(f"处理后的 JSON 字符串（前500字符）：{raw[:500]}")
# return [], []
if isinstance(parsed, dict):
parsed_list = [parsed]
elif isinstance(parsed, list):
parsed_list = parsed
else:
logger.error(f"表达风格解析结果类型异常: {type(parsed)}, 内容: {parsed}")
return [], []
# if isinstance(parsed, dict):
# parsed_list = [parsed]
# elif isinstance(parsed, list):
# parsed_list = parsed
# else:
# logger.error(f"表达风格解析结果类型异常: {type(parsed)}, 内容: {parsed}")
# return [], []
for item in parsed_list:
if not isinstance(item, dict):
continue
# for item in parsed_list:
# if not isinstance(item, dict):
# continue
# 检查是否是表达方式条目（有 situation 和 style）
situation = str(item.get("situation", "")).strip()
style = str(item.get("style", "")).strip()
source_id = str(item.get("source_id", "")).strip()
# # 检查是否是表达方式条目（有 situation 和 style）
# situation = str(item.get("situation", "")).strip()
# style = str(item.get("style", "")).strip()
# source_id = str(item.get("source_id", "")).strip()
if situation and style and source_id:
# 表达方式条目
expressions.append((situation, style, source_id))
elif item.get("content"):
# 黑话条目（有 content 字段）
content = str(item.get("content", "")).strip()
source_id = str(item.get("source_id", "")).strip()
if content and source_id:
jargon_entries.append((content, source_id))
# if situation and style and source_id:
# # 表达方式条目
# expressions.append((situation, style, source_id))
# elif item.get("content"):
# # 黑话条目（有 content 字段）
# content = str(item.get("content", "")).strip()
# source_id = str(item.get("source_id", "")).strip()
# if content and source_id:
# jargon_entries.append((content, source_id))
return expressions, jargon_entries
# return expressions, jargon_entries

179
src/bw_learner/message_recorder_old.py
View File

@@ -1,179 +0,0 @@
import time
import asyncio
from typing import List, Any
from src.common.logger import get_logger
from src.chat.message_receive.chat_manager import chat_manager as _chat_manager
from src.chat.utils.chat_message_builder import get_raw_msg_by_timestamp_with_chat_inclusive
from src.chat.utils.common_utils import TempMethodsExpression
from src.bw_learner.expression_learner_old import expression_learner_manager
from src.bw_learner.jargon_miner_old import miner_manager
logger = get_logger("bw_learner")
class MessageRecorder:
"""
统一的消息记录器，负责管理时间窗口和消息提取，并将消息分发给 expression_learner 和 jargon_miner
"""
def __init__(self, chat_id: str) -> None:
self.chat_id = chat_id
self.chat_stream = _chat_manager.get_session_by_session_id(chat_id)
self.chat_name = _chat_manager.get_session_name(chat_id) or chat_id
# 维护每个chat的上次提取时间
self.last_extraction_time: float = time.time()
# 提取锁，防止并发执行
self._extraction_lock = asyncio.Lock()
# 获取 expression 和 jargon 的配置参数
self._init_parameters()
# 获取 expression_learner 和 jargon_miner 实例
self.expression_learner = expression_learner_manager.get_expression_learner(chat_id)
self.jargon_miner = miner_manager.get_miner(chat_id)
def _init_parameters(self) -> None:
"""初始化提取参数"""
# 获取 expression 配置
_, self.enable_expression_learning, self.enable_jargon_learning = (
TempMethodsExpression.get_expression_config_for_chat(self.chat_id)
)
self.min_messages_for_extraction = 30
self.min_extraction_interval = 60
logger.debug(
f"MessageRecorder 初始化: chat_id={self.chat_id}, "
f"min_messages={self.min_messages_for_extraction}, "
f"min_interval={self.min_extraction_interval}"
)
def should_trigger_extraction(self) -> bool:
"""
检查是否应该触发消息提取
Returns:
bool: 是否应该触发提取
"""
# 检查时间间隔
time_diff = time.time() - self.last_extraction_time
if time_diff < self.min_extraction_interval:
return False
# 检查消息数量
recent_messages = get_raw_msg_by_timestamp_with_chat_inclusive(
chat_id=self.chat_id,
timestamp_start=self.last_extraction_time,
timestamp_end=time.time(),
)
if not recent_messages or len(recent_messages) < self.min_messages_for_extraction:
return False
return True
async def extract_and_distribute(self) -> None:
"""
提取消息并分发给 expression_learner 和 jargon_miner
"""
# 使用异步锁防止并发执行
async with self._extraction_lock:
# 在锁内检查，避免并发触发
if not self.should_trigger_extraction():
return
# 检查 chat_stream 是否存在
if not self.chat_stream:
return
# 记录本次提取的时间窗口，避免重复提取
extraction_start_time = self.last_extraction_time
extraction_end_time = time.time()
# 立即更新提取时间，防止并发触发
self.last_extraction_time = extraction_end_time
try:
# logger.info(f"在聊天流 {self.chat_name} 开始统一消息提取和分发")
# 拉取提取窗口内的消息
messages = get_raw_msg_by_timestamp_with_chat_inclusive(
chat_id=self.chat_id,
timestamp_start=extraction_start_time,
timestamp_end=extraction_end_time,
)
if not messages:
logger.debug(f"聊天流 {self.chat_name} 没有新消息，跳过提取")
return
# 按时间排序，确保顺序一致
messages = sorted(messages, key=lambda msg: msg.time or 0)
logger.info(
f"聊天流 {self.chat_name} 提取到 {len(messages)} 条消息，"
f"时间窗口: {extraction_start_time:.2f} - {extraction_end_time:.2f}"
)
# 触发 expression_learner 和 jargon_miner 的处理
if self.enable_expression_learning:
asyncio.create_task(self._trigger_expression_learning(messages))
except Exception as e:
logger.error(f"为聊天流 {self.chat_name} 提取和分发消息失败: {e}")
import traceback
traceback.print_exc()
# 即使失败也保持时间戳更新，避免频繁重试
async def _trigger_expression_learning(self, messages: List[Any]) -> None:
"""
触发 expression 学习，使用指定的消息列表
Args:
timestamp_start: 开始时间戳
timestamp_end: 结束时间戳
messages: 消息列表
"""
try:
# 传递消息给 ExpressionLearner（必需参数）
learnt_style = await self.expression_learner.learn_and_store(messages=messages)
if learnt_style:
logger.info(f"聊天流 {self.chat_name} 表达学习完成")
else:
logger.debug(f"聊天流 {self.chat_name} 表达学习未获得有效结果")
except Exception as e:
logger.error(f"为聊天流 {self.chat_name} 触发表达学习失败: {e}")
import traceback
traceback.print_exc()
class MessageRecorderManager:
"""MessageRecorder 管理器"""
def __init__(self) -> None:
self._recorders: dict[str, MessageRecorder] = {}
def get_recorder(self, chat_id: str) -> MessageRecorder:
"""获取或创建指定 chat_id 的 MessageRecorder"""
if chat_id not in self._recorders:
self._recorders[chat_id] = MessageRecorder(chat_id)
return self._recorders[chat_id]
# 全局管理器实例
recorder_manager = MessageRecorderManager()
async def extract_and_distribute_messages(chat_id: str) -> None:
"""
统一的消息提取和分发入口函数
Args:
chat_id: 聊天流ID
"""
recorder = recorder_manager.get_recorder(chat_id)
await recorder.extract_and_distribute()