BruceSun
/
MaiBot
mirror of https://github.com/Mai-with-u/MaiBot.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'dev' of github.com:MaiM-with-u/MaiBot into dev

pull/1239/head
UnCLAS-Prommer 2025-09-15 15:23:17 +08:00
parent 48ed58c4a8 a55ce70c0d
commit 4f5dc810ef
No known key found for this signature in database
16 changed files with 186 additions and 1055 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

613
scripts/test_interest_embedding.py
View File

@ -1,613 +0,0 @@
#!/usr/bin/env python3
"""
基于Embedding的兴趣度计算测试脚本
使用MaiBot-Core的EmbeddingStore计算兴趣描述与目标文本的关联度
"""
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from typing import List, Dict, Tuple, Optional
import time
import json
import asyncio
from src.chat.knowledge.embedding_store import EmbeddingStore, cosine_similarity
from src.chat.knowledge.embedding_store import EMBEDDING_DATA_DIR_STR
from src.llm_models.utils_model import LLMRequest
from src.config.config import model_config
class InterestScorer:
"""基于Embedding的兴趣度计算器"""
def __init__(self, namespace: str = "interest_test"):
"""初始化兴趣度计算器"""
self.embedding_store = EmbeddingStore(namespace, EMBEDDING_DATA_DIR_STR)
async def get_embedding(self, text: str) -> Tuple[Optional[List[float]], float]:
"""获取文本的嵌入向量"""
start_time = time.time()
try:
# 直接使用异步方式获取嵌入
from src.llm_models.utils_model import LLMRequest
from src.config.config import model_config
llm = LLMRequest(model_set=model_config.model_task_config.embedding, request_type="embedding")
embedding, _ = await llm.get_embedding(text)
end_time = time.time()
elapsed = end_time - start_time
if embedding and len(embedding) > 0:
return embedding, elapsed
return None, elapsed
except Exception as e:
print(f"获取嵌入向量失败: {e}")
return None, 0.0
async def calculate_similarity(self, text1: str, text2: str) -> Tuple[float, float, float]:
"""计算两段文本的余弦相似度，返回(相似度, 文本1耗时, 文本2耗时)"""
emb1, time1 = await self.get_embedding(text1)
emb2, time2 = await self.get_embedding(text2)
if emb1 is None or emb2 is None:
return 0.0, time1, time2
return cosine_similarity(emb1, emb2), time1, time2
async def calculate_interest_score(self, interest_text: str, target_text: str) -> Dict:
"""
计算兴趣度分数
Args:
interest_text: 兴趣描述文本
target_text: 目标文本
Returns:
包含各种分数的字典
"""
# 只计算语义相似度（嵌入分数）
semantic_score, interest_time, target_time = await self.calculate_similarity(interest_text, target_text)
# 直接使用语义相似度作为最终分数
final_score = semantic_score
return {
"final_score": final_score,
"semantic_score": semantic_score,
"timing": {
"interest_embedding_time": interest_time,
"target_embedding_time": target_time,
"total_time": interest_time + target_time
}
}
async def batch_calculate(self, interest_text: str, target_texts: List[str]) -> List[Dict]:
"""批量计算兴趣度"""
results = []
total_start_time = time.time()
print(f"开始批量计算兴趣度...")
print(f"兴趣文本: {interest_text}")
print(f"目标文本数量: {len(target_texts)}")
# 获取兴趣文本的嵌入向量（只需要一次）
interest_embedding, interest_time = await self.get_embedding(interest_text)
if interest_embedding is None:
print("无法获取兴趣文本的嵌入向量")
return []
print(f"兴趣文本嵌入计算耗时: {interest_time:.3f}秒")
total_target_time = 0.0
for i, target_text in enumerate(target_texts):
print(f"处理第 {i+1}/{len(target_texts)} 个文本...")
# 获取目标文本的嵌入向量
target_embedding, target_time = await self.get_embedding(target_text)
total_target_time += target_time
if target_embedding is None:
semantic_score = 0.0
else:
semantic_score = cosine_similarity(interest_embedding, target_embedding)
# 直接使用语义相似度作为最终分数
final_score = semantic_score
results.append({
"target_text": target_text,
"final_score": final_score,
"semantic_score": semantic_score,
"timing": {
"target_embedding_time": target_time,
"item_total_time": target_time
}
})
# 按分数排序
results.sort(key=lambda x: x["final_score"], reverse=True)
total_time = time.time() - total_start_time
avg_target_time = total_target_time / len(target_texts) if target_texts else 0
print(f"\n=== 性能统计 ===")
print(f"兴趣文本嵌入计算耗时: {interest_time:.3f}秒")
print(f"目标文本嵌入计算总耗时: {total_target_time:.3f}秒")
print(f"目标文本嵌入计算平均耗时: {avg_target_time:.3f}秒")
print(f"总耗时: {total_time:.3f}秒")
print(f"平均每个目标文本处理耗时: {total_time / len(target_texts):.3f}秒")
return results
async def generate_paraphrases(self, original_text: str, num_sentences: int = 5) -> List[str]:
"""
使用LLM生成近义句子
Args:
original_text: 原始文本
num_sentences: 生成句子数量
Returns:
近义句子列表
"""
try:
# 创建LLM请求实例
llm_request = LLMRequest(
model_set=model_config.model_task_config.replyer,
request_type="paraphrase_generator"
)
# 构建生成近义句子的提示词
prompt = f"""请为以下兴趣描述生成{num_sentences}个意义相近但表达不同的句子：
原始兴趣描述：{original_text}
要求：
1. 保持原意不变，但尽量自由发挥，使用不同的表达方式，内容也可以有差异
2. 句子结构要有所变化
3. 可以适当调整语气和重点
4. 每个句子都要完整且自然
5. 只返回句子，不要编号，每行一个句子
生成的近义句子："""
print(f"正在生成近义句子...")
content, (reasoning, model_name, tool_calls) = await llm_request.generate_response_async(prompt)
# 解析生成的句子
sentences = []
for line in content.strip().split('\n'):
line = line.strip()
if line and not line.startswith('生成') and not line.startswith('近义'):
sentences.append(line)
# 确保返回指定数量的句子
sentences = sentences[:num_sentences]
print(f"成功生成 {len(sentences)} 个近义句子")
print(f"使用的模型: {model_name}")
return sentences
except Exception as e:
print(f"生成近义句子失败: {e}")
return []
async def evaluate_all_paraphrases(self, original_text: str, target_texts: List[str], num_sentences: int = 5) -> Dict:
"""
评估原始文本和所有近义句子的兴趣度
Args:
original_text: 原始兴趣描述文本
target_texts: 目标文本列表
num_sentences: 生成近义句子数量
Returns:
包含所有评估结果的字典
"""
print(f"\n=== 开始近义句子兴趣度评估 ===")
print(f"原始兴趣描述: {original_text}")
print(f"目标文本数量: {len(target_texts)}")
print(f"生成近义句子数量: {num_sentences}")
# 生成近义句子
paraphrases = await self.generate_paraphrases(original_text, num_sentences)
if not paraphrases:
print("生成近义句子失败，使用原始文本进行评估")
paraphrases = []
# 所有待评估的文本（原始文本 + 近义句子）
all_texts = [original_text] + paraphrases
# 对每个文本进行兴趣度评估
evaluation_results = {}
for i, text in enumerate(all_texts):
text_type = "原始文本" if i == 0 else f"近义句子{i}"
print(f"\n--- 评估 {text_type} ---")
print(f"文本内容: {text}")
# 计算兴趣度
results = await self.batch_calculate(text, target_texts)
evaluation_results[text_type] = {
"text": text,
"results": results,
"top_score": results[0]["final_score"] if results else 0.0,
"average_score": sum(r["final_score"] for r in results) / len(results) if results else 0.0
}
return {
"original_text": original_text,
"paraphrases": paraphrases,
"evaluations": evaluation_results,
"summary": self._generate_summary(evaluation_results, target_texts)
}
def _generate_summary(self, evaluation_results: Dict, target_texts: List[str]) -> Dict:
"""生成评估摘要 - 关注目标句子的表现"""
summary = {
"best_performer": None,
"worst_performer": None,
"average_scores": {},
"max_scores": {},
"rankings": [],
"target_stats": {},
"target_rankings": []
}
scores = []
for text_type, data in evaluation_results.items():
scores.append({
"text_type": text_type,
"text": data["text"],
"top_score": data["top_score"],
"average_score": data["average_score"]
})
# 按top_score排序
scores.sort(key=lambda x: x["top_score"], reverse=True)
summary["rankings"] = scores
summary["best_performer"] = scores[0] if scores else None
summary["worst_performer"] = scores[-1] if scores else None
# 计算原始文本统计
original_score = next((s for s in scores if s["text_type"] == "原始文本"), None)
if original_score:
summary["average_scores"]["original"] = original_score["average_score"]
summary["max_scores"]["original"] = original_score["top_score"]
# 计算目标句子的统计信息
target_stats = {}
for i, target_text in enumerate(target_texts):
target_key = f"目标{i+1}"
scores_for_target = []
# 收集所有兴趣描述对该目标文本的分数
for text_type, data in evaluation_results.items():
for result in data["results"]:
if result["target_text"] == target_text:
scores_for_target.append(result["final_score"])
if scores_for_target:
target_stats[target_key] = {
"target_text": target_text,
"scores": scores_for_target,
"average": sum(scores_for_target) / len(scores_for_target),
"max": max(scores_for_target),
"min": min(scores_for_target),
"std": (sum((x - sum(scores_for_target) / len(scores_for_target)) ** 2 for x in scores_for_target) / len(scores_for_target)) ** 0.5
}
summary["target_stats"] = target_stats
# 按平均分对目标文本排序
target_rankings = []
for target_key, stats in target_stats.items():
target_rankings.append({
"target_key": target_key,
"target_text": stats["target_text"],
"average_score": stats["average"],
"max_score": stats["max"],
"min_score": stats["min"],
"std_score": stats["std"]
})
target_rankings.sort(key=lambda x: x["average_score"], reverse=True)
summary["target_rankings"] = target_rankings
# 计算目标文本的整体统计
if target_rankings:
all_target_averages = [t["average_score"] for t in target_rankings]
all_target_scores = []
for stats in target_stats.values():
all_target_scores.extend(stats["scores"])
summary["target_overall"] = {
"avg_of_averages": sum(all_target_averages) / len(all_target_averages),
"overall_max": max(all_target_scores),
"overall_min": min(all_target_scores),
"best_target": target_rankings[0]["target_text"],
"worst_target": target_rankings[-1]["target_text"]
}
return summary
async def run_single_test():
"""运行单个测试"""
print("单个兴趣度测试")
print("=" * 40)
# 输入兴趣文本
# interest_text = input("请输入兴趣描述文本: ").strip()
# if not interest_text:
# print("兴趣描述不能为空")
# return
interest_text ="对技术相关话题，游戏和动漫相关话题感兴趣，也对日常话题感兴趣，不喜欢太过沉重严肃的话题"
# 输入目标文本
print("请输入目标文本 (输入空行结束):")
import random
target_texts = [
"AveMujica非常好看，你看了吗",
"明日方舟这个游戏挺好玩的",
"你能不能说点正经的",
"明日方舟挺好玩的",
"你的名字非常好看，你看了吗",
"《你的名字》非常好看，你看了吗",
"我们来聊聊苏联政治吧",
"轻音少女非常好看，你看了吗",
"我还挺喜欢打游戏的",
"我嘞个原神玩家啊",
"我心买了PlayStation5",
"直接Steam",
"有没有R"
]
random.shuffle(target_texts)
# while True:
# line = input().strip()
# if not line:
# break
# target_texts.append(line)
# if not target_texts:
# print("目标文本不能为空")
# return
# 计算兴趣度
scorer = InterestScorer()
results = await scorer.batch_calculate(interest_text, target_texts)
# 显示结果
print(f"\n兴趣度排序结果:")
print("-" * 80)
print(f"{'排名':<4} {'最终分数':<10} {'语义分数':<10} {'耗时(秒)':<10} {'目标文本'}")
print("-" * 80)
for j, result in enumerate(results):
target_text = result['target_text']
if len(target_text) > 40:
target_text = target_text[:37] + "..."
timing = result.get('timing', {})
item_time = timing.get('item_total_time', 0.0)
print(f"{j+1:<4} {result['final_score']:<10.3f} {result['semantic_score']:<10.3f} "
f"{item_time:<10.3f} {target_text}")
async def run_paraphrase_test():
"""运行近义句子测试"""
print("近义句子兴趣度对比测试")
print("=" * 40)
# 输入兴趣文本
interest_text = "对技术相关话题，游戏和动漫相关话题感兴趣，比如明日方舟和原神，也对日常话题感兴趣，不喜欢太过沉重严肃的话题"
# 输入目标文本
print("请输入目标文本 (输入空行结束):")
# target_texts = []
# while True:
# line = input().strip()
# if not line:
# break
# target_texts.append(line)
target_texts = [
"AveMujica非常好看，你看了吗",
"明日方舟这个游戏挺好玩的",
"你能不能说点正经的",
"明日方舟挺好玩的",
"你的名字非常好看，你看了吗",
"《你的名字》非常好看，你看了吗",
"我们来聊聊苏联政治吧",
"轻音少女非常好看，你看了吗",
"我还挺喜欢打游戏的",
"刚加好友就视奸空间14条",
"可乐老大加我好友，我先日一遍空间",
"鸟一茬茬的",
"可乐可以是m，群友可以是s"
]
if not target_texts:
print("目标文本不能为空")
return
# 创建评估器
scorer = InterestScorer()
# 运行评估
result = await scorer.evaluate_all_paraphrases(interest_text, target_texts, num_sentences=5)
# 显示结果
display_paraphrase_results(result, target_texts)
def display_paraphrase_results(result: Dict, target_texts: List[str]):
"""显示近义句子评估结果"""
print("\n" + "=" * 80)
print("近义句子兴趣度评估结果")
print("=" * 80)
# 显示目标文本
print(f"\n📋 目标文本列表:")
print("-" * 40)
for i, target in enumerate(target_texts):
print(f"{i+1}. {target}")
# 显示生成的近义句子
print(f"\n📝 生成的近义句子 (作为兴趣描述):")
print("-" * 40)
for i, paraphrase in enumerate(result["paraphrases"]):
print(f"{i+1}. {paraphrase}")
# 显示摘要
summary = result["summary"]
print(f"\n📊 评估摘要:")
print("-" * 40)
if summary["best_performer"]:
print(f"最佳表现: {summary['best_performer']['text_type']} (最高分: {summary['best_performer']['top_score']:.3f})")
if summary["worst_performer"]:
print(f"最差表现: {summary['worst_performer']['text_type']} (最高分: {summary['worst_performer']['top_score']:.3f})")
print(f"原始文本平均分: {summary['average_scores'].get('original', 0):.3f}")
# 显示目标文本的整体统计
if "target_overall" in summary:
overall = summary["target_overall"]
print(f"\n📈 目标文本整体统计:")
print("-" * 40)
print(f"目标文本数量: {len(summary['target_rankings'])}")
print(f"平均分的平均值: {overall['avg_of_averages']:.3f}")
print(f"所有匹配中的最高分: {overall['overall_max']:.3f}")
print(f"所有匹配中的最低分: {overall['overall_min']:.3f}")
print(f"最佳匹配目标: {overall['best_target'][:50]}...")
print(f"最差匹配目标: {overall['worst_target'][:50]}...")
# 显示目标文本排名
if "target_rankings" in summary and summary["target_rankings"]:
print(f"\n🏆 目标文本排名 (按平均分):")
print("-" * 80)
print(f"{'排名':<4} {'平均分':<8} {'最高分':<8} {'最低分':<8} {'标准差':<8} {'目标文本'}")
print("-" * 80)
for i, target in enumerate(summary["target_rankings"]):
target_text = target["target_text"][:40] + "..." if len(target["target_text"]) > 40 else target["target_text"]
print(f"{i+1:<4} {target['average_score']:<8.3f} {target['max_score']:<8.3f} {target['min_score']:<8.3f} {target['std_score']:<8.3f} {target_text}")
# 显示每个目标文本的详细分数分布
if "target_stats" in summary:
print(f"\n📊 目标文本详细分数分布:")
print("-" * 80)
for target_key, stats in summary["target_stats"].items():
print(f"\n{target_key}: {stats['target_text']}")
print(f" 平均分: {stats['average']:.3f}")
print(f" 最高分: {stats['max']:.3f}")
print(f" 最低分: {stats['min']:.3f}")
print(f" 标准差: {stats['std']:.3f}")
print(f" 所有分数: {[f'{s:.3f}' for s in stats['scores']]}")
# 显示最佳和最差兴趣描述的目标表现对比
if summary["best_performer"] and summary["worst_performer"]:
print(f"\n🔍 最佳 vs 最差兴趣描述对比:")
print("-" * 80)
best_data = result["evaluations"][summary["best_performer"]["text_type"]]
worst_data = result["evaluations"][summary["worst_performer"]["text_type"]]
print(f"最佳兴趣描述: {summary['best_performer']['text']}")
print(f"最差兴趣描述: {summary['worst_performer']['text']}")
print(f"")
print(f"{'目标文本':<30} {'最佳分数':<10} {'最差分数':<10} {'差值'}")
print("-" * 60)
for best_result, worst_result in zip(best_data["results"], worst_data["results"]):
if best_result["target_text"] == worst_result["target_text"]:
diff = best_result["final_score"] - worst_result["final_score"]
target_text = best_result["target_text"][:27] + "..." if len(best_result["target_text"]) > 30 else best_result["target_text"]
print(f"{target_text:<30} {best_result['final_score']:<10.3f} {worst_result['final_score']:<10.3f} {diff:+.3f}")
# 显示排名
print(f"\n🏆 兴趣描述性能排名:")
print("-" * 80)
print(f"{'排名':<4} {'文本类型':<10} {'最高分':<8} {'平均分':<8} {'兴趣描述内容'}")
print("-" * 80)
for i, item in enumerate(summary["rankings"]):
text_content = item["text"][:40] + "..." if len(item["text"]) > 40 else item["text"]
print(f"{i+1:<4} {item['text_type']:<10} {item['top_score']:<8.3f} {item['average_score']:<8.3f} {text_content}")
# 显示每个兴趣描述的详细结果
print(f"\n🔍 详细结果:")
print("-" * 80)
for text_type, data in result["evaluations"].items():
print(f"\n--- {text_type} ---")
print(f"兴趣描述: {data['text']}")
print(f"最高分: {data['top_score']:.3f}")
print(f"平均分: {data['average_score']:.3f}")
# 显示前3个匹配结果
top_results = data["results"][:3]
print(f"前3个匹配的目标文本:")
for j, result_item in enumerate(top_results):
print(f" {j+1}. 分数: {result_item['final_score']:.3f} - {result_item['target_text']}")
# 显示对比表格
print(f"\n📈 兴趣描述对比表格:")
print("-" * 100)
header = f"{'兴趣描述':<20}"
for i, target in enumerate(target_texts):
target_name = f"目标{i+1}"
header += f" {target_name:<12}"
print(header)
print("-" * 100)
# 原始文本行
original_line = f"{'原始文本':<20}"
original_data = result["evaluations"]["原始文本"]["results"]
for i in range(len(target_texts)):
if i < len(original_data):
original_line += f" {original_data[i]['final_score']:<12.3f}"
else:
original_line += f" {'-':<12}"
print(original_line)
# 近义句子行
for i, paraphrase in enumerate(result["paraphrases"]):
text_type = f"近义句子{i+1}"
line = f"{text_type:<20}"
paraphrase_data = result["evaluations"][text_type]["results"]
for j in range(len(target_texts)):
if j < len(paraphrase_data):
line += f" {paraphrase_data[j]['final_score']:<12.3f}"
else:
line += f" {'-':<12}"
print(line)
def main():
"""主函数"""
print("基于Embedding的兴趣度计算测试工具")
print("1. 单个兴趣度测试")
print("2. 近义句子兴趣度对比测试")
choice = input("\n请选择 (1/2): ").strip()
if choice == "1":
asyncio.run(run_single_test())
elif choice == "2":
asyncio.run(run_paraphrase_test())
else:
print("无效选择")
if __name__ == "__main__":
main()

10
src/chat/express/expression_learner.py
View File

@ -16,7 +16,7 @@ from src.chat.message_receive.chat_stream import get_chat_manager
MAX_EXPRESSION_COUNT = 300 MAX_EXPRESSION_COUNT = 300
DECAY_DAYS = 30 # 30天衰减到0.01 DECAY_DAYS = 15 # 30天衰减到0.01
DECAY_MIN = 0.01 # 最小衰减值 DECAY_MIN = 0.01 # 最小衰减值
logger = get_logger("expressor") logger = get_logger("expressor")
@ -45,10 +45,10 @@ def init_prompt() -> None:
例如：当"AAAAA"时，可以"BBBBB", AAAAA代表某个具体的场景，不超过20个字。BBBBB代表对应的语言风格，特定句式或表达方式，不超过20个字。 例如：当"AAAAA"时，可以"BBBBB", AAAAA代表某个具体的场景，不超过20个字。BBBBB代表对应的语言风格，特定句式或表达方式，不超过20个字。
例如： 例如：
当"对某件事表示十分惊叹，有些意外"时，使用"我嘞个xxxx" 当"对某件事表示十分惊叹"时，使用"我嘞个xxxx"
当"表示讽刺的赞同，不想讲道理"时，使用"对对对" 当"表示讽刺的赞同，不讲道理"时，使用"对对对"
当"想说明某个具体的事实观点，但懒得明说，或者不便明说，或表达一种默契"，使用"懂的都懂" 当"想说明某个具体的事实观点，但懒得明说，使用"懂的都懂"
当"当涉及游戏相关时，表示意外的夸赞，略带戏谑意味"时，使用"这么强！" 当"当涉及游戏相关时，夸赞，略带戏谑意味"时，使用"这么强！"
请注意：不要总结你自己（SELF）的发言，尽量保证总结内容的逻辑性 请注意：不要总结你自己（SELF）的发言，尽量保证总结内容的逻辑性
现在请你概括 现在请你概括

1
src/chat/frequency_control/frequency_control.py
View File

@ -3,7 +3,6 @@ from typing import Optional, Dict
from src.plugin_system.apis import message_api from src.plugin_system.apis import message_api
from src.chat.message_receive.chat_stream import ChatStream, get_chat_manager from src.chat.message_receive.chat_stream import ChatStream, get_chat_manager
from src.common.logger import get_logger from src.common.logger import get_logger
from src.chat.frequency_control.talk_frequency_control import get_config_base_talk_frequency
from src.chat.frequency_control.focus_value_control import get_config_base_focus_value from src.chat.frequency_control.focus_value_control import get_config_base_focus_value
logger = get_logger("frequency_control") logger = get_logger("frequency_control")

272
src/chat/frequency_control/talk_frequency_control.py
View File

@ -1,272 +0,0 @@
from typing import Optional
from datetime import datetime, timedelta
import statistics
from src.config.config import global_config
from src.chat.frequency_control.utils import parse_stream_config_to_chat_id
from src.common.database.database_model import Messages
def get_config_base_talk_frequency(chat_id: Optional[str] = None) -> float:
"""
根据当前时间和聊天流获取对应的 talk_frequency
Args:
chat_stream_id: 聊天流ID，格式为 "platform:chat_id:type"
Returns:
float: 对应的频率值
"""
if not global_config.chat.talk_frequency_adjust:
return global_config.chat.talk_frequency
# 优先检查聊天流特定的配置
if chat_id:
stream_frequency = get_stream_specific_frequency(chat_id)
if stream_frequency is not None:
return stream_frequency
# 检查全局时段配置（第一个元素为空字符串的配置）
global_frequency = get_global_frequency()
return global_config.chat.talk_frequency if global_frequency is None else global_frequency
def get_time_based_frequency(time_freq_list: list[str]) -> Optional[float]:
"""
根据时间配置列表获取当前时段的频率
Args:
time_freq_list: 时间频率配置列表，格式为 ["HH:MM,frequency", ...]
Returns:
float: 频率值，如果没有配置则返回 None
"""
from datetime import datetime
current_time = datetime.now().strftime("%H:%M")
current_hour, current_minute = map(int, current_time.split(":"))
current_minutes = current_hour * 60 + current_minute
# 解析时间频率配置
time_freq_pairs = []
for time_freq_str in time_freq_list:
try:
time_str, freq_str = time_freq_str.split(",")
hour, minute = map(int, time_str.split(":"))
frequency = float(freq_str)
minutes = hour * 60 + minute
time_freq_pairs.append((minutes, frequency))
except (ValueError, IndexError):
continue
if not time_freq_pairs:
return None
# 按时间排序
time_freq_pairs.sort(key=lambda x: x[0])
# 查找当前时间对应的频率
current_frequency = None
for minutes, frequency in time_freq_pairs:
if current_minutes >= minutes:
current_frequency = frequency
else:
break
# 如果当前时间在所有配置时间之前，使用最后一个时间段的频率（跨天逻辑）
if current_frequency is None and time_freq_pairs:
current_frequency = time_freq_pairs[-1][1]
return current_frequency
def get_stream_specific_frequency(chat_stream_id: str):
"""
获取特定聊天流在当前时间的频率
Args:
chat_stream_id: 聊天流ID（哈希值）
Returns:
float: 频率值，如果没有配置则返回 None
"""
# 查找匹配的聊天流配置
for config_item in global_config.chat.talk_frequency_adjust:
if not config_item or len(config_item) < 2:
continue
stream_config_str = config_item[0] # 例如 "qq:1026294844:group"
# 解析配置字符串并生成对应的 chat_id
config_chat_id = parse_stream_config_to_chat_id(stream_config_str)
if config_chat_id is None:
continue
# 比较生成的 chat_id
if config_chat_id != chat_stream_id:
continue
# 使用通用的时间频率解析方法
return get_time_based_frequency(config_item[1:])
return None
def get_global_frequency() -> Optional[float]:
"""
获取全局默认频率配置
Returns:
float: 频率值，如果没有配置则返回 None
"""
for config_item in global_config.chat.talk_frequency_adjust:
if not config_item or len(config_item) < 2:
continue
# 检查是否为全局默认配置（第一个元素为空字符串）
if config_item[0] == "":
return get_time_based_frequency(config_item[1:])
return None
def get_weekly_hourly_message_stats(chat_id: str):
"""
计算指定聊天最近一周每个小时的消息数量和用户数量
Args:
chat_id: 聊天ID（对应 Messages 表的 chat_id 字段）
Returns:
dict: 包含24个小时统计数据，格式为:
{
"0": {"message_count": [5, 8, 3, 12, 6, 9, 7], "message_std_dev": 2.1},
"1": {"message_count": [10, 15, 8, 20, 12, 18, 14], "message_std_dev": 3.2},
...
}
"""
# 计算一周前的时间戳
one_week_ago = datetime.now() - timedelta(days=7)
one_week_ago_timestamp = one_week_ago.timestamp()
# 初始化数据结构：按小时存储每天的消息计数
hourly_data = {}
for hour in range(24):
hourly_data[f"hour_{hour}"] = {"daily_counts": []}
try:
# 查询指定聊天最近一周的消息
messages = Messages.select().where(
(Messages.time >= one_week_ago_timestamp) &
(Messages.chat_id == chat_id)
)
# 统计每个小时的数据
for message in messages:
# 将时间戳转换为datetime
msg_time = datetime.fromtimestamp(message.time)
hour = msg_time.hour
# 记录每天的消息计数（按日期分组）
day_key = msg_time.strftime("%Y-%m-%d")
hour_key = f"{hour}"
# 为该小时添加当天的消息计数
found = False
for day_count in hourly_data[hour_key]["daily_counts"]:
if day_count["date"] == day_key:
day_count["count"] += 1
found = True
break
if not found:
hourly_data[hour_key]["daily_counts"].append({"date": day_key, "count": 1})
except Exception as e:
# 如果查询失败，返回空的统计结果
print(f"Error getting weekly hourly message stats for chat {chat_id}: {e}")
hourly_stats = {}
for hour in range(24):
hourly_stats[f"hour_{hour}"] = {
"message_count": [],
"message_std_dev": 0.0
}
return hourly_stats
# 计算每个小时的统计结果
hourly_stats = {}
for hour in range(24):
hour_key = f"hour_{hour}"
daily_counts = [day["count"] for day in hourly_data[hour_key]["daily_counts"]]
# 计算总消息数
total_messages = sum(daily_counts)
# 计算标准差
message_std_dev = 0.0
if len(daily_counts) > 1:
message_std_dev = statistics.stdev(daily_counts)
elif len(daily_counts) == 1:
message_std_dev = 0.0
# 按日期排序每日消息计数
daily_counts_sorted = sorted(hourly_data[hour_key]["daily_counts"], key=lambda x: x["date"])
hourly_stats[hour_key] = {
"message_count": [day["count"] for day in daily_counts_sorted],
"message_std_dev": message_std_dev
}
return hourly_stats
def get_recent_15min_stats(chat_id: str):
"""
获取最近15分钟指定聊天的消息数量和发言人数
Args:
chat_id: 聊天ID（对应 Messages 表的 chat_id 字段）
Returns:
dict: 包含消息数量和发言人数，格式为:
{
"message_count": 25,
"user_count": 8,
"time_range": "2025-01-01 14:30:00 - 2025-01-01 14:45:00"
}
"""
# 计算15分钟前的时间戳
fifteen_min_ago = datetime.now() - timedelta(minutes=15)
fifteen_min_ago_timestamp = fifteen_min_ago.timestamp()
current_time = datetime.now()
# 初始化统计结果
message_count = 0
user_set = set()
try:
# 查询最近15分钟的消息
messages = Messages.select().where(
(Messages.time >= fifteen_min_ago_timestamp) &
(Messages.chat_id == chat_id)
)
# 统计消息数量和用户
for message in messages:
message_count += 1
if message.user_id:
user_set.add(message.user_id)
except Exception as e:
# 如果查询失败，返回空结果
print(f"Error getting recent 15min stats for chat {chat_id}: {e}")
return {
"message_count": 0,
"user_count": 0,
"time_range": f"{fifteen_min_ago.strftime('%Y-%m-%d %H:%M:%S')} - {current_time.strftime('%Y-%m-%d %H:%M:%S')}"
}
return {
"message_count": message_count,
"user_count": len(user_set),
"time_range": f"{fifteen_min_ago.strftime('%Y-%m-%d %H:%M:%S')} - {current_time.strftime('%Y-%m-%d %H:%M:%S')}"
}

5
src/chat/heart_flow/heartFC_chat.py
View File

@ -1,7 +1,6 @@
import asyncio import asyncio
import time import time
import traceback import traceback
import math
import random import random
from typing import List, Optional, Dict, Any, Tuple, TYPE_CHECKING from typing import List, Optional, Dict, Any, Tuple, TYPE_CHECKING
from rich.traceback import install from rich.traceback import install
@ -20,7 +19,7 @@ from src.chat.heart_flow.hfc_utils import CycleDetail
from src.chat.heart_flow.hfc_utils import send_typing, stop_typing from src.chat.heart_flow.hfc_utils import send_typing, stop_typing
from src.chat.express.expression_learner import expression_learner_manager from src.chat.express.expression_learner import expression_learner_manager
from src.person_info.person_info import Person from src.person_info.person_info import Person
from src.plugin_system.base.component_types import ChatMode, EventType, ActionInfo from src.plugin_system.base.component_types import EventType, ActionInfo
from src.plugin_system.core import events_manager from src.plugin_system.core import events_manager
from src.plugin_system.apis import generator_api, send_api, message_api, database_api from src.plugin_system.apis import generator_api, send_api, message_api, database_api
from src.mais4u.mai_think import mai_thinking_manager from src.mais4u.mai_think import mai_thinking_manager
@ -376,7 +375,7 @@ class HeartFChatting:
action_success = False action_success = False
action_reply_text = "" action_reply_text = ""
for i, result in enumerate(results): for result in results:
if isinstance(result, BaseException): if isinstance(result, BaseException):
logger.error(f"{self.log_prefix} 动作执行异常: {result}") logger.error(f"{self.log_prefix} 动作执行异常: {result}")
continue continue

8
src/chat/heart_flow/heartflow_message_processor.py
View File

@ -1,6 +1,5 @@
import asyncio import asyncio
import re import re
import math
import traceback import traceback
from typing import Tuple, TYPE_CHECKING from typing import Tuple, TYPE_CHECKING
@ -72,16 +71,15 @@ class HeartFCMessageReceiver:
chat = message.chat_stream chat = message.chat_stream
# 2. 兴趣度计算与更新 # 2. 兴趣度计算与更新
interested_rate, keywords = await _calculate_interest(message) _, keywords = await _calculate_interest(message)
await self.storage.store_message(message, chat) await self.storage.store_message(message, chat)
heartflow_chat: HeartFChatting = await heartflow.get_or_create_heartflow_chat(chat.stream_id) # type: ignore heartflow_chat: HeartFChatting = await heartflow.get_or_create_heartflow_chat(chat.stream_id) # type: ignore
# subheartflow.add_message_to_normal_chat_cache(message, interested_rate, is_mentioned)
if global_config.mood.enable_mood: if global_config.mood.enable_mood:
chat_mood = mood_manager.get_mood_by_chat_id(heartflow_chat.stream_id) chat_mood = mood_manager.get_mood_by_chat_id(heartflow_chat.stream_id)
asyncio.create_task(chat_mood.update_mood_by_message(message, interested_rate)) asyncio.create_task(chat_mood.update_mood_by_message(message))
# 3. 日志记录 # 3. 日志记录
mes_name = chat.group_info.group_name if chat.group_info else "私聊" mes_name = chat.group_info.group_name if chat.group_info else "私聊"
@ -109,7 +107,7 @@ class HeartFCMessageReceiver:
replace_bot_name=True, replace_bot_name=True,
) )
logger.info(f"[{mes_name}]{userinfo.user_nickname}:{processed_plain_text}[{interested_rate:.2f}]") # type: ignore logger.info(f"[{mes_name}]{userinfo.user_nickname}:{processed_plain_text}") # type: ignore
_ = Person.register_person( _ = Person.register_person(
platform=message.message_info.platform, # type: ignore platform=message.message_info.platform, # type: ignore

145
src/chat/replyer/default_generator.py
View File

@ -18,7 +18,7 @@ from src.chat.message_receive.chat_stream import ChatStream
from src.chat.message_receive.uni_message_sender import UniversalMessageSender from src.chat.message_receive.uni_message_sender import UniversalMessageSender
from src.chat.utils.timer_calculator import Timer # <--- Import Timer from src.chat.utils.timer_calculator import Timer # <--- Import Timer
from src.chat.utils.utils import get_chat_type_and_target_info from src.chat.utils.utils import get_chat_type_and_target_info
from src.chat.utils.prompt_builder import Prompt, global_prompt_manager from src.chat.utils.prompt_builder import global_prompt_manager
from src.chat.utils.chat_message_builder import ( from src.chat.utils.chat_message_builder import (
build_readable_messages, build_readable_messages,
get_raw_msg_before_timestamp_with_chat, get_raw_msg_before_timestamp_with_chat,
@ -32,108 +32,17 @@ from src.person_info.person_info import Person, is_person_known
from src.plugin_system.base.component_types import ActionInfo, EventType from src.plugin_system.base.component_types import ActionInfo, EventType
from src.plugin_system.apis import llm_api from src.plugin_system.apis import llm_api
from src.chat.replyer.lpmm_prompt import init_lpmm_prompt
from src.chat.replyer.replyer_prompt import init_replyer_prompt
from src.chat.replyer.rewrite_prompt import init_rewrite_prompt
init_lpmm_prompt()
init_replyer_prompt()
init_rewrite_prompt()
logger = get_logger("replyer") logger = get_logger("replyer")
def init_prompt():
Prompt("你正在qq群里聊天，下面是群里在聊的内容：", "chat_target_group1")
Prompt("你正在和{sender_name}聊天，这是你们之前聊的内容：", "chat_target_private1")
Prompt("在群里聊天", "chat_target_group2")
Prompt("和{sender_name}聊天", "chat_target_private2")
Prompt(
"""
{expression_habits_block}
{relation_info_block}
{chat_target}
{time_block}
{chat_info}
{identity}
你现在的心情是：{mood_state}
你正在{chat_target_2},{reply_target_block}
你想要对上述的发言进行回复，回复的具体内容（原句）是：{raw_reply}
原因是：{reason}
现在请你将这条具体内容改写成一条适合在群聊中发送的回复消息。
你需要使用合适的语法和句法，参考聊天内容，组织一条日常且口语化的回复。请你修改你想表达的原句，符合你的表达风格和语言习惯
{reply_style}
你可以完全重组回复，保留最基本的表达含义就好，但重组后保持语意通顺。
{keywords_reaction_prompt}
{moderation_prompt}
不要输出多余内容(包括前后缀，冒号和引号，括号，表情包，emoji,at或 @等 )，只输出一条回复就好。
现在，你说：
""",
"default_expressor_prompt",
)
# s4u 风格的 prompt 模板
Prompt(
"""{identity}
你正在群聊中聊天，你想要回复 {sender_name} 的发言。同时，也有其他用户会参与聊天，你可以参考他们的回复内容，但是你现在想回复{sender_name}的发言。
{time_block}
{background_dialogue_prompt}
{core_dialogue_prompt}
{expression_habits_block}{tool_info_block}
{knowledge_prompt}{relation_info_block}
{extra_info_block}
{reply_target_block}
你的心情：{mood_state}
{reply_style}
注意不要复读你说过的话
{keywords_reaction_prompt}
请注意不要输出多余内容(包括前后缀，冒号和引号，at或 @等 )。只输出回复内容。
{moderation_prompt}
不要输出多余内容(包括前后缀，冒号和引号，括号()，表情包，emoji,at或 @等 )。只输出一条回复就好
现在，你说：""",
"replyer_prompt",
)
Prompt(
"""{identity}
{time_block}
你现在正在一个QQ群里聊天，以下是正在进行的聊天内容：
{background_dialogue_prompt}
{expression_habits_block}{tool_info_block}
{knowledge_prompt}{relation_info_block}
{extra_info_block}
你现在想补充说明你刚刚自己的发言内容：{target}，原因是{reason}
请你根据聊天内容，组织一条新回复。注意，{target} 是刚刚你自己的发言，你要在这基础上进一步发言，请按照你自己的角度来继续进行回复。
注意保持上下文的连贯性。
你现在的心情是：{mood_state}
{reply_style}
{keywords_reaction_prompt}
请注意不要输出多余内容(包括前后缀，冒号和引号，at或 @等 )。只输出回复内容。
{moderation_prompt}
不要输出多余内容(包括前后缀，冒号和引号，括号()，表情包，emoji,at或 @等 )。只输出一条回复就好
现在，你说：
""",
"replyer_self_prompt",
)
Prompt(
"""
你是一个专门获取知识的助手。你的名字是{bot_name}。现在是{time_now}。
群里正在进行的聊天内容：
{chat_history}
现在，{sender}发送了内容:{target_message},你想要回复ta。
请仔细分析聊天内容，考虑以下几点：
1. 内容中是否包含需要查询信息的问题
2. 是否有明确的知识获取指令
If you need to use the search tool, please directly call the function "lpmm_search_knowledge". If you do not need to use any tool, simply output "No tool needed".
""",
name="lpmm_get_knowledge_prompt",
)
class DefaultReplyer: class DefaultReplyer:
def __init__( def __init__(
self, self,
@ -369,7 +278,7 @@ class DefaultReplyer:
expression_habits_title = "" expression_habits_title = ""
if style_habits_str.strip(): if style_habits_str.strip():
expression_habits_title = ( expression_habits_title = (
"你可以参考以下的语言习惯，当情景合适就使用，但不要生硬使用，以合理的方式结合到你的回复中：" "在回复时,你可以参考以下的语言习惯，当情景合适就使用，但不要生硬使用，以合理的方式结合到你的回复中："
) )
expression_habits_block += f"{style_habits_str}\n" expression_habits_block += f"{style_habits_str}\n"
@ -557,18 +466,6 @@ class DefaultReplyer:
except Exception as e: except Exception as e:
logger.error(f"处理消息记录时出错: {msg}, 错误: {e}") logger.error(f"处理消息记录时出错: {msg}, 错误: {e}")
# 构建背景对话 prompt
all_dialogue_prompt = ""
if message_list_before_now:
latest_25_msgs = message_list_before_now[-int(global_config.chat.max_context_size) :]
all_dialogue_prompt_str = build_readable_messages(
latest_25_msgs,
replace_bot_name=True,
timestamp_mode="normal_no_YMD",
truncate=True,
)
all_dialogue_prompt = f"所有用户的发言：\n{all_dialogue_prompt_str}"
# 构建核心对话 prompt # 构建核心对话 prompt
core_dialogue_prompt = "" core_dialogue_prompt = ""
if core_dialogue_list: if core_dialogue_list:
@ -601,6 +498,22 @@ class DefaultReplyer:
-------------------------------- --------------------------------
""" """
# 构建背景对话 prompt
all_dialogue_prompt = ""
if message_list_before_now:
latest_25_msgs = message_list_before_now[-int(global_config.chat.max_context_size) :]
all_dialogue_prompt_str = build_readable_messages(
latest_25_msgs,
replace_bot_name=True,
timestamp_mode="normal_no_YMD",
truncate=True,
)
if core_dialogue_prompt:
all_dialogue_prompt = f"所有用户的发言：\n{all_dialogue_prompt_str}"
else:
all_dialogue_prompt = f"{all_dialogue_prompt_str}"
return core_dialogue_prompt, all_dialogue_prompt return core_dialogue_prompt, all_dialogue_prompt
def build_mai_think_context( def build_mai_think_context(
@ -852,11 +765,11 @@ class DefaultReplyer:
if sender: if sender:
if is_group_chat: if is_group_chat:
reply_target_block = ( reply_target_block = (
f"现在{sender}说的:{target}。引起了你的注意，你想要在群里发言或者回复这条消息。原因是{reply_reason}" f"现在{sender}说的:{target}。引起了你的注意"
) )
else: # private chat else: # private chat
reply_target_block = ( reply_target_block = (
f"现在{sender}说的:{target}。引起了你的注意，针对这条消息回复。原因是{reply_reason}" f"现在{sender}说的:{target}。引起了你的注意"
) )
else: else:
reply_target_block = "" reply_target_block = ""
@ -1148,4 +1061,4 @@ def weighted_sample_no_replacement(items, weights, k) -> list:
return selected return selected
init_prompt()

24
src/chat/replyer/lpmm_prompt.py 100644
View File

@ -0,0 +1,24 @@
from src.chat.utils.prompt_builder import Prompt
# from src.chat.memory_system.memory_activator import MemoryActivator
def init_lpmm_prompt():
Prompt(
"""
你是一个专门获取知识的助手。你的名字是{bot_name}。现在是{time_now}。
群里正在进行的聊天内容：
{chat_history}
现在，{sender}发送了内容:{target_message},你想要回复ta。
请仔细分析聊天内容，考虑以下几点：
1. 内容中是否包含需要查询信息的问题
2. 是否有明确的知识获取指令
If you need to use the search tool, please directly call the function "lpmm_search_knowledge". If you do not need to use any tool, simply output "No tool needed".
""",
name="lpmm_get_knowledge_prompt",
)

55
src/chat/replyer/replyer_prompt.py 100644
View File

@ -0,0 +1,55 @@
from src.chat.utils.prompt_builder import Prompt
# from src.chat.memory_system.memory_activator import MemoryActivator
def init_replyer_prompt():
Prompt("你正在qq群里聊天，下面是群里正在聊的内容:", "chat_target_group1")
Prompt("你正在和{sender_name}聊天，这是你们之前聊的内容：", "chat_target_private1")
Prompt("正在群里聊天", "chat_target_group2")
Prompt("和{sender_name}聊天", "chat_target_private2")
Prompt(
"""{knowledge_prompt}{relation_info_block}{tool_info_block}{extra_info_block}
{expression_habits_block}
你正在qq群里聊天，下面是群里正在聊的内容:
{time_block}
{background_dialogue_prompt}
{core_dialogue_prompt}
{reply_target_block}。
{identity}
你正在群里聊天,现在请你读读之前的聊天记录，然后给出日常且口语化的回复，平淡一些，
尽量简短一些。{keywords_reaction_prompt}请注意把握聊天内容，不要回复的太有条理，可以有个性。
{reply_style}
请注意不要输出多余内容(包括前后缀，冒号和引号，括号，表情等)，只输出回复内容。
{moderation_prompt}不要输出多余内容(包括前后缀，冒号和引号，括号，表情包，at或 @等 )。""",
"replyer_prompt",
)
Prompt(
"""{knowledge_prompt}{relation_info_block}{tool_info_block}{extra_info_block}
{expression_habits_block}
你正在qq群里聊天，下面是群里正在聊的内容:
{time_block}
{background_dialogue_prompt}
你现在想补充说明你刚刚自己的发言内容：{target}，原因是{reason}
请你根据聊天内容，组织一条新回复。注意，{target} 是刚刚你自己的发言，你要在这基础上进一步发言，请按照你自己的角度来继续进行回复。注意保持上下文的连贯性。
{identity}
尽量简短一些。{keywords_reaction_prompt}请注意把握聊天内容，不要回复的太有条理，可以有个性。
{reply_style}
请注意不要输出多余内容(包括前后缀，冒号和引号，括号，表情等)，只输出回复内容。
{moderation_prompt}不要输出多余内容(包括前后缀，冒号和引号，括号，表情包，at或 @等 )。
""",
"replyer_self_prompt",
)

37
src/chat/replyer/rewrite_prompt.py 100644
View File

@ -0,0 +1,37 @@
from src.chat.utils.prompt_builder import Prompt
# from src.chat.memory_system.memory_activator import MemoryActivator
def init_rewrite_prompt():
Prompt("你正在qq群里聊天，下面是群里正在聊的内容:", "chat_target_group1")
Prompt("你正在和{sender_name}聊天，这是你们之前聊的内容：", "chat_target_private1")
Prompt("正在群里聊天", "chat_target_group2")
Prompt("和{sender_name}聊天", "chat_target_private2")
Prompt(
"""
{expression_habits_block}
{relation_info_block}
{chat_target}
{time_block}
{chat_info}
{identity}
你现在的心情是：{mood_state}
你正在{chat_target_2},{reply_target_block}
你想要对上述的发言进行回复，回复的具体内容（原句）是：{raw_reply}
原因是：{reason}
现在请你将这条具体内容改写成一条适合在群聊中发送的回复消息。
你需要使用合适的语法和句法，参考聊天内容，组织一条日常且口语化的回复。请你修改你想表达的原句，符合你的表达风格和语言习惯
{reply_style}
你可以完全重组回复，保留最基本的表达含义就好，但重组后保持语意通顺。
{keywords_reaction_prompt}
{moderation_prompt}
不要输出多余内容(包括前后缀，冒号和引号，括号，表情包，emoji,at或 @等 )，只输出一条回复就好。
现在，你说：
""",
"default_expressor_prompt",
)

3
src/config/api_ada_configs.py
View File

@ -102,9 +102,6 @@ class ModelTaskConfig(ConfigBase):
replyer: TaskConfig replyer: TaskConfig
"""normal_chat首要回复模型模型配置""" """normal_chat首要回复模型模型配置"""
emotion: TaskConfig
"""情绪模型配置"""
vlm: TaskConfig vlm: TaskConfig
"""视觉语言模型配置""" """视觉语言模型配置"""

2
src/config/config.py
View File

@ -54,7 +54,7 @@ TEMPLATE_DIR = os.path.join(PROJECT_ROOT, "template")
# 考虑到，实际上配置文件中的mai_version是不会自动更新的,所以采用硬编码 # 考虑到，实际上配置文件中的mai_version是不会自动更新的,所以采用硬编码
# 对该字段的更新，请严格参照语义化版本规范：https://semver.org/lang/zh-CN/ # 对该字段的更新，请严格参照语义化版本规范：https://semver.org/lang/zh-CN/
MMC_VERSION = "0.10.3-snapshot.3" MMC_VERSION = "0.10.3-snapshot.4"
def get_key_comment(toml_table, key): def get_key_comment(toml_table, key):

13
src/mood/mood_manager.py
View File

@ -62,11 +62,11 @@ class ChatMood:
self.regression_count: int = 0 self.regression_count: int = 0
self.mood_model = LLMRequest(model_set=model_config.model_task_config.emotion, request_type="mood") self.mood_model = LLMRequest(model_set=model_config.model_task_config.utils, request_type="mood")
self.last_change_time: float = 0 self.last_change_time: float = 0
async def update_mood_by_message(self, message: MessageRecv, interested_rate: float): async def update_mood_by_message(self, message: MessageRecv):
self.regression_count = 0 self.regression_count = 0
during_last_time = message.message_info.time - self.last_change_time # type: ignore during_last_time = message.message_info.time - self.last_change_time # type: ignore
@ -74,10 +74,9 @@ class ChatMood:
base_probability = 0.05 base_probability = 0.05
time_multiplier = 4 * (1 - math.exp(-0.01 * during_last_time)) time_multiplier = 4 * (1 - math.exp(-0.01 * during_last_time))
if interested_rate <= 0: # 基于消息长度计算基础兴趣度
interest_multiplier = 0 message_length = len(message.message_content.content or "")
else: interest_multiplier = min(2.0, 1.0 + message_length / 100)
interest_multiplier = 2 * math.pow(interested_rate, 0.25)
logger.debug( logger.debug(
f"base_probability: {base_probability}, time_multiplier: {time_multiplier}, interest_multiplier: {interest_multiplier}" f"base_probability: {base_probability}, time_multiplier: {time_multiplier}, interest_multiplier: {interest_multiplier}"
@ -90,7 +89,7 @@ class ChatMood:
return return
logger.debug( logger.debug(
f"{self.log_prefix} 更新情绪状态，感兴趣度: {interested_rate:.2f}, 更新概率: {update_probability:.2f}" f"{self.log_prefix} 更新情绪状态，更新概率: {update_probability:.2f}"
) )
message_time: float = message.message_info.time # type: ignore message_time: float = message.message_info.time # type: ignore

2
src/plugins/built_in/memory/plugin.py
View File

@ -1,7 +1,7 @@
from typing import List, Tuple, Type from typing import List, Tuple, Type
# 导入新插件系统 # 导入新插件系统
from src.plugin_system import BasePlugin, register_plugin, ComponentInfo from src.plugin_system import BasePlugin, ComponentInfo
from src.plugin_system.base.config_types import ConfigField from src.plugin_system.base.config_types import ConfigField
# 导入依赖的系统组件 # 导入依赖的系统组件

6
template/bot_config_template.toml
View File

@ -1,5 +1,5 @@
[inner] [inner]
version = "6.12.0" version = "6.13.0"
#----以下是给开发人员阅读的，如果你只是部署了麦麦，不需要阅读---- #----以下是给开发人员阅读的，如果你只是部署了麦麦，不需要阅读----
#如果你想要修改配置文件，请递增version的值 #如果你想要修改配置文件，请递增version的值
@ -19,10 +19,10 @@ alias_names = ["麦叠", "牢麦"] # 麦麦的别名
[personality] [personality]
# 建议120字以内，描述人格特质 和 身份特征 # 建议120字以内，描述人格特质 和 身份特征
personality = "是一个女大学生，现在在读大二，会刷贴吧。有时候说话不过脑子,有时候会喜欢说一些奇怪的话。年龄为19岁,有黑色的短发。" personality = "是一个女大学生，现在在读大二，会刷贴吧。"
#アイデンティティがない 生まれないらららら #アイデンティティがない 生まれないらららら
# 描述麦麦说话的表达风格，表达习惯，如要修改，可以酌情新增内容 # 描述麦麦说话的表达风格，表达习惯，如要修改，可以酌情新增内容
reply_style = "回复可以简短一些。可以参考贴吧，知乎和微博的回复风格，回复不要浮夸，不要用夸张修辞，平淡一些。不要浮夸，不要夸张修辞。" reply_style = "请回复的平淡一些，简短一些，说中文，不要刻意突出自身学科背景。可以参考贴吧，知乎和微博的回复风格。"
# 情感特征，影响情绪的变化情况 # 情感特征，影响情绪的变化情况
emotion_style = "情绪较为稳定，但遭遇特定事件的时候起伏较大" emotion_style = "情绪较为稳定，但遭遇特定事件的时候起伏较大"
# 麦麦的兴趣，会影响麦麦对什么话题进行回复 # 麦麦的兴趣，会影响麦麦对什么话题进行回复

45
template/model_config_template.toml
View File

@ -1,5 +1,5 @@
[inner] [inner]
version = "1.6.0" version = "1.7.0"
# 配置文件版本号迭代规则同bot_config.toml # 配置文件版本号迭代规则同bot_config.toml
@ -12,14 +12,14 @@ max_retry = 2 # 最大重试次数（单个模型API
timeout = 30 # API请求超时时间（单位：秒） timeout = 30 # API请求超时时间（单位：秒）
retry_interval = 10 # 重试间隔时间（单位：秒） retry_interval = 10 # 重试间隔时间（单位：秒）
[[api_providers]] # SiliconFlow的API服务商配置 [[api_providers]] # 阿里 百炼 API服务商配置
name = "SiliconFlow" name = "BaiLian"
base_url = "https://api.siliconflow.cn/v1" base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1"
api_key = "your-siliconflow-api-key" api_key = "your-bailian-key"
client_type = "openai" client_type = "openai"
max_retry = 2 max_retry = 2
timeout = 30 timeout = 15
retry_interval = 10 retry_interval = 5
[[api_providers]] # 特殊：Google的Gimini使用特殊API，与OpenAI格式不兼容，需要配置client为"gemini" [[api_providers]] # 特殊：Google的Gimini使用特殊API，与OpenAI格式不兼容，需要配置client为"gemini"
name = "Google" name = "Google"
@ -30,14 +30,14 @@ max_retry = 2
timeout = 30 timeout = 30
retry_interval = 10 retry_interval = 10
[[api_providers]] # 阿里 百炼 API服务商配置 [[api_providers]] # SiliconFlow的API服务商配置
name = "BaiLian" name = "SiliconFlow"
base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1" base_url = "https://api.siliconflow.cn/v1"
api_key = "your-bailian-key" api_key = "your-siliconflow-api-key"
client_type = "openai" client_type = "openai"
max_retry = 2 max_retry = 2
timeout = 15 timeout = 60
retry_interval = 5 retry_interval = 10
[[models]] # 模型（可以配置多个） [[models]] # 模型（可以配置多个）
@ -93,8 +93,8 @@ price_in = 0
price_out = 0 price_out = 0
[model_task_config.utils] # 在麦麦的一些组件中使用的模型，例如表情包模块，取名模块，关系模块，是麦麦必须的模型 [model_task_config.utils] # 在麦麦的一些组件中使用的模型，例如表情包模块，取名模块，关系模块，麦麦的情绪变化等，是麦麦必须的模型
model_list = ["siliconflow-deepseek-v3"] # 使用的模型列表，每个子项对应上面的模型名称(name) model_list = ["siliconflow-deepseek-v3","qwen3-30b"] # 使用的模型列表，每个子项对应上面的模型名称(name)
temperature = 0.2 # 模型温度，新V3建议0.1-0.3 temperature = 0.2 # 模型温度，新V3建议0.1-0.3
max_tokens = 800 # 最大输出token数 max_tokens = 800 # 最大输出token数
@ -103,6 +103,11 @@ model_list = ["qwen3-8b","qwen3-30b"]
temperature = 0.7 temperature = 0.7
max_tokens = 800 max_tokens = 800
[model_task_config.tool_use] #工具调用模型，需要使用支持工具调用的模型
model_list = ["qwen3-30b"]
temperature = 0.7
max_tokens = 800
[model_task_config.replyer] # 首要回复模型，还用于表达器和表达方式学习 [model_task_config.replyer] # 首要回复模型，还用于表达器和表达方式学习
model_list = ["siliconflow-deepseek-v3"] model_list = ["siliconflow-deepseek-v3"]
temperature = 0.3 # 模型温度，新V3建议0.1-0.3 temperature = 0.3 # 模型温度，新V3建议0.1-0.3
@ -113,11 +118,6 @@ model_list = ["siliconflow-deepseek-v3"]
temperature = 0.3 temperature = 0.3
max_tokens = 800 max_tokens = 800
[model_task_config.emotion] #负责麦麦的情绪变化
model_list = ["qwen3-30b"]
temperature = 0.7
max_tokens = 800
[model_task_config.vlm] # 图像识别模型 [model_task_config.vlm] # 图像识别模型
model_list = ["qwen2.5-vl-72b"] model_list = ["qwen2.5-vl-72b"]
max_tokens = 800 max_tokens = 800
@ -125,11 +125,6 @@ max_tokens = 800
[model_task_config.voice] # 语音识别模型 [model_task_config.voice] # 语音识别模型
model_list = ["sensevoice-small"] model_list = ["sensevoice-small"]
[model_task_config.tool_use] #工具调用模型，需要使用支持工具调用的模型
model_list = ["qwen3-30b"]
temperature = 0.7
max_tokens = 800
#嵌入模型 #嵌入模型
[model_task_config.embedding] [model_task_config.embedding]
model_list = ["bge-m3"] model_list = ["bge-m3"]