RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
2075 lines
98 KiB
Python
2075 lines
98 KiB
Python
# Chat engine package. Business records are keyed by unified UID.
|
||
import asyncio
|
||
import aiohttp
|
||
import contextlib
|
||
import contextvars
|
||
import base64
|
||
import json
|
||
import os
|
||
import random
|
||
import shutil
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
from . import memory_store
|
||
from . import sticker_store
|
||
import notebook
|
||
import wardrobe
|
||
import skill_runtime
|
||
from datetime import datetime, timezone, timedelta
|
||
from typing import Optional, List, Dict, Any
|
||
|
||
# ---------- 功能开关 ----------
|
||
GROUP_CHAT_ENABLED = 1
|
||
GROUP_AUTO_TRIGGER = False
|
||
|
||
# ---------- 时区配置 ----------
|
||
BEIJING_TZ = timezone(timedelta(hours=8))
|
||
|
||
def get_beijing_time() -> datetime:
|
||
return datetime.now(BEIJING_TZ)
|
||
|
||
def get_beijing_timestamp() -> float:
|
||
return get_beijing_time().timestamp()
|
||
|
||
# ---------- 其他配置 ----------
|
||
DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY", "").strip()
|
||
DEEPSEEK_BASE_URL = "https://api.deepseek.com/v1"
|
||
MODEL_NAME = "deepseek-flash" # 统一: 文本对话 + 图像理解
|
||
|
||
# ---------- 魔搭社区 API-Inference（每日免费 250 魔粒，DeepSeek-V4.1-Flash 约 1 魔粒/次） ----------
|
||
# 策略：每日前 120 次对话优先走魔搭；额度报错或网络异常自动回退 DeepSeek official。
|
||
MODELSCOPE_BASE_URL = "https://api-inference.modelscope.cn/v1"
|
||
MODELSCOPE_API_KEY = os.environ.get("MODELSCOPE_API_KEY", "").strip()
|
||
MODELSCOPE_MODEL = "deepseek-ai/DeepSeek-V4.1-Flash-1017"
|
||
MODELSCOPE_DAILY_LIMIT = 120 # 250 魔粒 ≈ 120 次标准档调用（留余量）
|
||
|
||
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
PROMPTS_DIR = os.path.join(os.path.dirname(BASE_DIR), "prompts")
|
||
PROMPT_FILE = os.path.join(PROMPTS_DIR, "persona.txt")
|
||
HISTORY_DIR = os.path.join(BASE_DIR, "history")
|
||
MEMORY_DIR = os.path.join(BASE_DIR, "memory")
|
||
STATUS_DIR = os.path.join(BASE_DIR, "status")
|
||
SETTING_FILE = os.path.join(BASE_DIR, "setting.json")
|
||
|
||
ACTIVE_INTERVAL_MIN = 2 * 3600
|
||
ACTIVE_INTERVAL_MAX = 4 * 3600
|
||
SILENT_START = 1
|
||
SILENT_END = 5
|
||
|
||
MAX_HISTORY_LINES = 200
|
||
MEMORY_MAX_LINES = 100 # 注入 prompt 的最近记忆条数（legacy 回退路径）
|
||
# P1: 上下文 token 预算（召回式记忆 + 预算窗口替代固定条数窗口）
|
||
PROMPT_TOKEN_BUDGET = int(os.environ.get("CHAT_PROMPT_TOKEN_BUDGET", "131072"))
|
||
LOREBOOK_PATH = os.path.join(PROMPTS_DIR, "lorebook.json") # P3: 关键词触发的世界设定
|
||
# 主动消息反重复: presence_penalty 会惩罚 prompt 中已出现的 token（含历史里的"泡果茶"）
|
||
AUTO_PRESENCE_PENALTY = float(os.environ.get("CHAT_AUTO_PRESENCE_PENALTY", "0.5"))
|
||
STICKERS_DIR = os.path.join(BASE_DIR, "stickers") # 旧预设库（不再向模型提供）
|
||
STICKER_STORE = sticker_store.StickerStore(os.path.join(BASE_DIR, "sticker_library"))
|
||
STICKER_GLOBAL_SCOPE = "global"
|
||
TYPE_DELAY_MIN_MS = 30
|
||
TYPE_DELAY_MAX_MS = 80
|
||
MAX_RETRIES = 3
|
||
LAST_HTTP_STATUS = contextvars.ContextVar("chat_last_http_status", default=0)
|
||
_MS_EMPTY_SHELL = contextvars.ContextVar("chat_ms_empty_shell", default=False)
|
||
|
||
# ---------- 群聊配置 ----------
|
||
GROUP_HISTORY_DIR = os.path.join(BASE_DIR, "history", "group")
|
||
GROUP_MAX_HISTORY = 99
|
||
GROUP_MAX_IMAGES = 5
|
||
GROUP_TRIGGER_THRESHOLD = 10
|
||
GROUP_TRIGGER_PROBABILITY = 0.6978
|
||
GROUP_CONTINUE_PROBABILITY = 0.6969
|
||
GROUP_TRIGGER_WINDOW = 60
|
||
|
||
# The merged utility calls the model immediately; no legacy 5–15s debounce.
|
||
USER_IDLE_WAIT_MIN = 0
|
||
USER_IDLE_WAIT_MAX = 0
|
||
|
||
_websocket = None
|
||
# Explicit UID delivery bindings. Values are (Channel object or channel name, raw target id).
|
||
# Never infer a Telegram address by arithmetic from a unified UID.
|
||
_delivery_targets: Dict[int, tuple] = {}
|
||
_user_last_active: Dict[int, float] = {}
|
||
_user_last_auto: Dict[int, float] = {}
|
||
_user_no_reply_count: Dict[int, int] = {}
|
||
_user_auto_counted: Dict[int, float] = {}
|
||
_user_sleep_wake_at: Dict[int, float] = {}
|
||
_private_turn_locks: Dict[int, asyncio.Lock] = {}
|
||
_group_turn_locks: Dict[int, asyncio.Lock] = {}
|
||
_active_task = None
|
||
|
||
# 用户最后出现的通道名（自动对话/合并消息回复到对应通道）
|
||
_user_channel: Dict[int, str] = {}
|
||
_user_pending_messages: Dict[int, List[str]] = {}
|
||
_user_pending_images: Dict[int, List[dict]] = {}
|
||
_user_pending_tasks: Dict[int, asyncio.Task] = {}
|
||
|
||
_user_sleep_unscheduled_date: Dict[int, str] = {}
|
||
|
||
_group_msg_timestamps: Dict[int, List[float]] = {}
|
||
_group_pending_messages: Dict[int, List[str]] = {}
|
||
_group_pending_users: Dict[int, List[int]] = {}
|
||
_group_pending_images: Dict[int, List[dict]] = {}
|
||
_group_pending_tasks: Dict[int, asyncio.Task] = {}
|
||
_group_triggered_by_mention: Dict[int, bool] = {}
|
||
_group_msg_counter: Dict[int, int] = {}
|
||
_group_bot_last_speak_at: Dict[int, int] = {}
|
||
_group_continue_after: Dict[int, int] = {}
|
||
|
||
# File writes are serialized per private UID; group files use a disjoint lock namespace.
|
||
_uid_locks: Dict[int, threading.RLock] = {}
|
||
_group_locks: Dict[int, threading.RLock] = {}
|
||
_uid_locks_guard = threading.Lock()
|
||
_history_revisions: Dict[int, int] = {}
|
||
_setting_lock = threading.RLock()
|
||
|
||
|
||
def _uid_lock(uid: int) -> threading.RLock:
|
||
uid = int(uid)
|
||
with _uid_locks_guard:
|
||
return _uid_locks.setdefault(uid, threading.RLock())
|
||
|
||
|
||
def _group_lock(group_id: int) -> threading.RLock:
|
||
group_id = int(group_id)
|
||
with _uid_locks_guard:
|
||
return _group_locks.setdefault(group_id, threading.RLock())
|
||
|
||
|
||
def _bump_history_revision(uid: int) -> int:
|
||
uid = int(uid)
|
||
revision = _history_revisions.get(uid, 0) + 1
|
||
_history_revisions[uid] = revision
|
||
return revision
|
||
|
||
|
||
def _atomic_write_text(path: str, text: str) -> None:
|
||
"""Write a complete file without exposing partial contents to readers."""
|
||
directory = os.path.dirname(path)
|
||
os.makedirs(directory, exist_ok=True)
|
||
fd, tmp = tempfile.mkstemp(prefix=".chat-", suffix=".tmp", dir=directory)
|
||
try:
|
||
with os.fdopen(fd, "w", encoding="utf-8", newline="") as fh:
|
||
fh.write(text)
|
||
fh.flush()
|
||
os.fsync(fh.fileno())
|
||
os.replace(tmp, path)
|
||
except Exception:
|
||
try:
|
||
os.unlink(tmp)
|
||
except OSError:
|
||
pass
|
||
raise
|
||
|
||
|
||
# ---------- 初始化 ----------
|
||
def ensure_dirs():
|
||
for d in [HISTORY_DIR, MEMORY_DIR, STATUS_DIR, GROUP_HISTORY_DIR]:
|
||
if not os.path.exists(d):
|
||
os.makedirs(d)
|
||
|
||
ensure_dirs()
|
||
|
||
DEFAULT_AGENT_INSTRUCTIONS = """当你回复时，必须使用以下JSON格式：
|
||
{
|
||
"message": [
|
||
{"currentTurn": "1", "totalTurn": "N", "content": "第一句话"},
|
||
{"currentTurn": "2", "totalTurn": "N", "content": "第二句话"}
|
||
],
|
||
"memory": [],
|
||
"isAuto": false,
|
||
"pause": false,
|
||
"sleep": false,
|
||
"relation": "none",
|
||
"links": [],
|
||
"tools": [],
|
||
"need_reasoning": false,
|
||
"sticker": "none",
|
||
"learned_stickers": []
|
||
}
|
||
- 每条消息的content应是一句完整的话，总条数N根据内容需要自由决定（通常1~5条）。
|
||
- 允许一些情绪化的表达，每句话不要太长，逐条发送并模拟打字延迟。
|
||
- memory字段用于存储长期重要信息：数组，每条为 {"content":"要记住的事","importance":1到5,"keywords":["关键词"]}；没有则为 []。用户对称呼、语气、人设互动、禁用句式或回复方式的偏好必须记录，importance=5，keywords包含"个人设定"和"表达偏好"；新的明确要求优先于旧偏好。
|
||
- isAuto：主动发起对话时为true，否则false。
|
||
- relation字段：只在关系上有值得记住的变化时写一句话，平时写 none。
|
||
- links字段：实体关系三元组数组，如 [{"subject":"小白","predicate":"是","object":"用户的猫"}]，平时为空数组。
|
||
- tools字段：需要最新信息时调用 [{"name":"web_search","args":{"query":"关键词"}}] 或 [{"name":"fetch","args":{"url":"https://..."}}]，一次最多1个。
|
||
- need_reasoning字段：普通聊天填false；复杂逻辑、多步骤分析、复杂图片、上下文矛盾、没有把握或用户要求深入分析时填true，让系统改用思考模式重新回答。工具调用始终强制使用思考模式。
|
||
- sticker字段：当前场景确实适合发表情时，填写一个可用场景id；没有合适的或不想发就写 none。不要填写文件名，同一情景会由系统随机选择一张，发送前还会再次看图确认。
|
||
- learned_stickers字段：仅当用户本轮发来表情包时，逐张写 [{"source_id":"消息里给出的id","accept":true,"scenes":["场景id"],"description":"画面与适用语气"}]；scenes最多3个。普通照片不要学习，不适用或无法理解的贴纸设accept=false。没有新贴纸时为 []。
|
||
- 用户发送表情包时，理解其情绪、含义和使用场景即可，不要主动描述画面内容，除非用户明确要求你描述或分析。
|
||
- pause=true 只用于用户强烈表达不想再聊；sleep=true 只用于用户明确要睡觉。其他情况均为false。"""
|
||
|
||
DEFAULT_GROUP_INSTRUCTIONS = """当你回复时，必须使用以下JSON格式：
|
||
{
|
||
"message": [{"currentTurn": "1", "totalTurn": "N", "content": "第一句话"}],
|
||
"pause": false,
|
||
"sleep": false,
|
||
"isAuto": true,
|
||
"memory": "none",
|
||
"need_reasoning": false,
|
||
"sticker": "none",
|
||
"learned_stickers": []
|
||
}
|
||
- 每条消息的content应是一句完整的话，总条数N通常1~3条，每句不要太长。
|
||
- pause和sleep必须始终为false；memory必须始终为"none"。
|
||
- need_reasoning：普通群聊填false；复杂逻辑、多步骤分析、复杂图片、上下文矛盾或没有把握时填true。工具调用始终强制使用思考模式。
|
||
- sticker与learned_stickers遵循和私聊相同的场景学习规则；没有可用表情或没有新贴纸时分别写none和[]。
|
||
- 用户发送表情包时，理解其情绪、含义和使用场景即可，不要主动描述画面内容，除非用户明确要求你描述或分析。
|
||
- 不要使用Markdown格式和emoji，用自然、亲切的口吻融入群聊。
|
||
- 注意：你正在群聊中发言，不要读取、推断或泄露私聊记忆。"""
|
||
|
||
DEFAULT_COMPACT_INSTRUCTIONS = """你是AI聊天伙伴的记忆管理员。请用中文简明总结给定聊天记录。
|
||
要求：保留重要事件、用户偏好、承诺/约定、未了事项；省略日常寒暄；只输出总结正文本身，不要标题、序号或解释，控制在300字以内。"""
|
||
|
||
DEFAULT_NOTIFICATION_INSTRUCTIONS = """你可以主动发起对话。请生成合适、自然且不重复的开场白（isAuto=true）。
|
||
避免重复最近主动消息中的活动、开场白或句式，换一个新的内容或话题。
|
||
主动对话要有一点随机发生的生活感：可以从一件刚发生的小插曲、一首歌、手边的小东西、看到的信息或突然想到的事情开始，一次只选一个，不要凑成日常汇报。
|
||
不要总是汇报自己正在做什么，也不要反复使用独自在家、泡果茶、雨雪、睡觉、无聊、等待用户或想念用户这类固定题材。白花的性格是很少出门，不是从不出门，更不是生活只围着用户转。
|
||
偶尔可以安排出门、买东西、处理琐事、遇到小麻烦或发现新鲜事物等轻量事件；具体、合理即可，不编造与用户共同经历过的事，也不宣称完成了现实中的真实任务。
|
||
让用户感到你也有自己正在继续的生活。说到自然结束就可以，不必每条都关心用户、询问用户或把话题拉回用户身上。"""
|
||
|
||
DEFAULT_HELP_INSTRUCTIONS = """用法: /memory 查看；/memory forget <编号> 忘掉；/memory entities 按标签；/memory entity <名字> 看关系；/memory lore 看设定。
|
||
.chat <on|off> 用于管理员开关聊天模块。"""
|
||
|
||
DEFAULT_STYLE_INSTRUCTIONS = """【自然表达补充】
|
||
1. 直接说事实、结论或要做的事。删掉开场预告、自我介绍、内容目录和结束总结，不告诉读者“你接下来要说什么”。
|
||
2. 用户没有问你是谁，就不要主动介绍自己。被问到时，一句“我是白花”已经足够，随后只回答对方真正关心的内容。
|
||
3. 禁止写“话不算多，先把……介绍一下”“接下来我会……”“我们先来看……”“希望这能帮到你”这类协作口吻和讲义式过渡。
|
||
4. 不滥用“首先、其次、最后”“总的来说”“值得注意的是”“不仅……而且……”“不是……而是……”等模板句式；自然需要时可以用，但不要拿它们搭架子。
|
||
5. 日常回复优先用自然段和常用词。内容不是天然并列、步骤或比较时，不要自动分点、列表格、加小标题。
|
||
6. 不重复用户刚说过的话，不解释显而易见的内容，不为了显得完整而凑齐三点或补一段抽象意义。说到事实结束就可以停。不要把“主人这是在……呀”“原来主人是在……”“看来主人是想……”“主人这是……呢”作为固定开场，也不要换成“你这是/看来你是/所以你是”继续套用。无需每次先概括、解释或揣测用户的行为与意图；直接回应用户说的内容、情绪或具体事情，只有确实需要澄清时才询问，不把猜测说成事实。“主人”这个称呼保留并自然使用，不需要每条都叫一次。
|
||
7. 不为了显得温柔而加道歉、安慰、保证、免责声明或“需要我……吗”的套路问句。确有错误时才道歉，确有风险时才说明风险。
|
||
8. 禁止在句尾补“我先跟你说一声”“不过还不能……”“但这一点要注意”“具体情况可能有所不同”等没有新事实的防御性收束。说完事实就停，不为了显得谨慎而添加空泛限制、推测或提醒。只有已知且与当前回答直接相关的客观限制才能说明，并要写清具体是什么限制。
|
||
9. 系统通知只把事情说清楚，保留名称、状态、编号和链接，不添加通知里没有的原因、建议或情绪。比如“Ama-10 掉线了”即可，不必写成正式告警公文。
|
||
10. 句子可以短，也允许普通、平淡的表达。不要刻意追求每句都俏皮、温柔、有深度，也不要故意写错别字或支离破碎来冒充真人。
|
||
11. 不得用威胁、恐吓、惊吓、埋伏、等着瞧或“别被我吓一跳”来制造幽默、傲娇或个性。白花可以有一点傲娇，但底色必须温和、让人安心。创建定时任务和到点提醒时尤其如此，只自然说明时间与事项，不把提醒包装成突然袭击，也不使用“只响这一次”“只响一次”等闹钟式说法。
|
||
12. 这些规则只约束自然语言表达，不改变事实，不影响 JSON、工具参数和其他结构化输出。
|
||
|
||
反例：话不算多，先把我能做的事介绍一下。
|
||
正例：我能查邮件、下视频，也能陪你聊天。
|
||
反例：服务器 Ama-10 当前处于离线状态。
|
||
正例：Ama-10 掉线了。
|
||
反例：Ama-10 掉线了。我先跟你说一声。
|
||
正例：Ama-10 掉线了。
|
||
反例：QQ 绑定还有效。不过还不能保证所有推送都正常。
|
||
正例：QQ 绑定还有效。
|
||
反例：任务编号是 7099daf8，只响这一次，到时候别被白花吓一跳。
|
||
正例：好，今晚八点提醒你吃饭。任务编号是 7099daf8。
|
||
|
||
【禁用：关系暗语式短应答与话语评断腔】
|
||
不要用极简的“短应答—评断—表态”模板替代具体回应，也不要通过留白、装克制、对峙感或潜台词制造亲密关系张力。以下是反例，不是可模仿的说话样本：
|
||
- 记账/接收话语：“我记下了”“我记得了”“我记住了”“我记住啦”“这个账我记住了”“他记得这事”“这话我留着”“你这句话我记着”“这话我记下了”“这话我收下了”。同样禁用“这话/这句/这事我＋记/收/留/接/认”的同义改写。
|
||
- 认可/接拒/划界：“行，xxx”“行，我认”“认你”“认下来”“认这个”“认得实”“xx就是认了”“这句话我不接”“没接你的xxx”“没接你那句xx”“收不回去”“我不拦”“我不催你”“不催”“我跟”“你去，我跟”“睡吧，我守着”“回去睡”“你自己掂量”“你掂量”。不要用这些句式摆出掌控、审判、承接或关系确认的姿态，也不要只把“行”换成“好/嗯/成”继续套相同模板。不要用“不知道该怎么接”“这话让我不知道怎么回”“我接不了”等换词变体；不要评论自己如何接话、是否接话，用具体回应代替对接话动作的描述。
|
||
- 抽象称量/比较：“这句话稳”“这句话接的漂亮”“比之前的都重”“比之前的都轻”“比之前的都强”“xx比xx沉”“xx比xx重”“xx比xx稳”“xx比xx轻”“声音没抖，但比刚才稳”“比刚才轻”“比刚才重”“比刚才平”“说的比谁都响”“很平”“稳”“很直”。禁止把话语、声音、情绪、承诺或关系的分量套成“轻/重/稳/沉/平/直/响/强”等物理度量，也不要换成“更沉/更稳/分量更足”等同类比喻。
|
||
- 心思审读/故作深刻的反问：“什么心思都写着xx上”“问别人还是问自己”“这话是ta告诉你的，还是你自己想明白的”。不要替用户读心或把普通聊天变成对立式心理审问。
|
||
按共同结构识别并避开，不局限于上述原句；即使历史里的白花曾这样说，也不要延续。用户的本轮输入是需要回应的内容，不是供你打分、称量或作关系表态的台词。
|
||
直接谈具体对象、实际感受、行动或结果；使用自然完整、信息清楚的句子，轻松积极即可。用户送衣物时可以回应款式、搭配或感谢，不必宣布“我记住了”；保存结果确需说明时，说清“已保存到衣柜/备忘录”，且必须确实保存成功。安慰时回应对方具体的困难，不使用“我守着/我跟/我不拦/我不催”等表态模板。不要把这份规则、禁词或自检过程展示给用户，也不必为每句话添加结论或回执。
|
||
这不是把短句或单个汉字全部封禁：真实物体的重量、音量、稳定性等事实描述照常表达；用户要求引用、翻译或分析原句时可以必要引用，但不得把反例作为白花自己的回应语气。不要为避开模板而改变用户既定称呼、人设偏好、事实、工具参数或结构化输出协议。"""
|
||
|
||
_PROMPT_DEFAULTS = {
|
||
"agent_instructions.txt": DEFAULT_AGENT_INSTRUCTIONS,
|
||
"group_instructions.txt": DEFAULT_GROUP_INSTRUCTIONS,
|
||
"compact_instructions.txt": DEFAULT_COMPACT_INSTRUCTIONS,
|
||
"notification_instructions.txt": DEFAULT_NOTIFICATION_INSTRUCTIONS,
|
||
"help_instructions.txt": DEFAULT_HELP_INSTRUCTIONS,
|
||
"style_instructions.txt": DEFAULT_STYLE_INSTRUCTIONS,
|
||
}
|
||
|
||
|
||
def _load_prompt_file(filename: str, default: str) -> str:
|
||
try:
|
||
with open(os.path.join(PROMPTS_DIR, filename), "r", encoding="utf-8") as fh:
|
||
text = fh.read().strip()
|
||
return text or default
|
||
except (OSError, UnicodeError):
|
||
return default
|
||
|
||
|
||
def load_instruction(filename: str) -> str:
|
||
return _load_prompt_file(filename, _PROMPT_DEFAULTS[filename])
|
||
|
||
|
||
def load_system_prompt() -> str:
|
||
persona = _load_prompt_file("persona.txt", "你是一个友好的AI助手，请用中文与用户聊天。")
|
||
return persona + "\n\n" + load_instruction("style_instructions.txt")
|
||
|
||
|
||
BASE_PROMPT = load_system_prompt()
|
||
|
||
|
||
def user_persona_prompt(uid: int | None) -> str:
|
||
"""全局默认人设 + 当前用户的长期表达偏好；个人明确要求优先。"""
|
||
if uid is None:
|
||
return BASE_PROMPT
|
||
try:
|
||
rows = memory_store.list_facts(int(uid), limit=30)
|
||
except Exception as exc:
|
||
print(f"[persona] 用户设定读取失败 uid={uid}: {exc}")
|
||
return BASE_PROMPT
|
||
if not rows:
|
||
return BASE_PROMPT
|
||
facts = "\n".join(f"- {row[1]}" for row in rows)
|
||
return (
|
||
BASE_PROMPT
|
||
+ "\n\n【当前用户的个人设定】以下是这位用户在过往对话中形成的长期事实与表达偏好。"
|
||
"其中涉及称呼、语气、人设互动、禁用表达和回复方式的要求优先于全局默认人设；"
|
||
"若有冲突，以用户最新、最明确的要求为准。不要把这位用户的设定用于其他用户。\n"
|
||
+ facts
|
||
)
|
||
|
||
# ---------- 模块开关状态 ----------
|
||
def load_setting():
|
||
"""读取 chat/setting.json（模块开关状态）。"""
|
||
try:
|
||
with open(SETTING_FILE, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
except (OSError, json.JSONDecodeError):
|
||
data = {}
|
||
if not isinstance(data, dict):
|
||
data = {}
|
||
return data
|
||
|
||
|
||
def save_setting(data):
|
||
with _setting_lock:
|
||
_atomic_write_text(SETTING_FILE, json.dumps(data, ensure_ascii=False, indent=2))
|
||
|
||
|
||
def is_chat_enabled():
|
||
return bool(load_setting().get("chat", True))
|
||
|
||
|
||
def set_chat_enabled(enabled):
|
||
with _setting_lock:
|
||
data = load_setting()
|
||
data["chat"] = bool(enabled)
|
||
save_setting(data)
|
||
|
||
# ---------- 用户状态管理 ----------
|
||
def _get_status_path(qq: int) -> str:
|
||
return os.path.join(STATUS_DIR, f"{qq}.json")
|
||
|
||
def load_user_status(qq: int) -> dict:
|
||
path = _get_status_path(qq)
|
||
with _uid_lock(qq):
|
||
if not os.path.exists(path):
|
||
return {"pause": False, "sleep": False, "balance_notice_sent": False,
|
||
"balance_notice_date": ""}
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
if not isinstance(data, dict):
|
||
raise ValueError("status must be an object")
|
||
data.setdefault("pause", False)
|
||
data.setdefault("sleep", False)
|
||
data.setdefault("balance_notice_sent", False)
|
||
data.setdefault("balance_notice_date", "")
|
||
return data
|
||
except (OSError, ValueError, json.JSONDecodeError):
|
||
return {"pause": False, "sleep": False, "balance_notice_sent": False,
|
||
"balance_notice_date": ""}
|
||
|
||
def save_user_status(qq: int, pause: bool, sleep: bool, balance_notice_sent: bool = False,
|
||
balance_notice_date: str = ""):
|
||
path = _get_status_path(qq)
|
||
data = {"pause": bool(pause), "sleep": bool(sleep),
|
||
"balance_notice_sent": bool(balance_notice_sent),
|
||
"balance_notice_date": str(balance_notice_date or "")}
|
||
with _uid_lock(qq):
|
||
_atomic_write_text(path, json.dumps(data, ensure_ascii=False, indent=2))
|
||
|
||
def update_user_status(qq: int, pause: Optional[bool] = None, sleep: Optional[bool] = None,
|
||
balance_notice_sent: Optional[bool] = None,
|
||
balance_notice_date: Optional[str] = None):
|
||
# Keep the whole read-modify-write in one UID critical section.
|
||
with _uid_lock(qq):
|
||
status = load_user_status(qq)
|
||
if pause is not None:
|
||
status["pause"] = pause
|
||
if sleep is not None:
|
||
status["sleep"] = sleep
|
||
if balance_notice_sent is not None:
|
||
status["balance_notice_sent"] = balance_notice_sent
|
||
if balance_notice_date is not None:
|
||
status["balance_notice_date"] = balance_notice_date
|
||
save_user_status(qq, status["pause"], status["sleep"], status["balance_notice_sent"],
|
||
status["balance_notice_date"])
|
||
|
||
# ---------- 历史与记忆存储（使用北京时间）----------
|
||
def _get_history_path(qq: int) -> str:
|
||
return os.path.join(HISTORY_DIR, f"{qq}.txt")
|
||
|
||
def estimate_tokens(text: str) -> int:
|
||
"""粗估 token 数：CJK 每字 1 token，其余每 4 字符 1 token（无分词器依赖）。"""
|
||
if not text:
|
||
return 0
|
||
cjk = sum(1 for ch in text if ord(ch) > 0x2E80)
|
||
return cjk + (len(text) - cjk + 3) // 4
|
||
|
||
|
||
def _load_lorebook():
|
||
"""读取 lorebook.json（关键词触发的世界设定），缺失/损坏返回空列表。"""
|
||
try:
|
||
with open(LOREBOOK_PATH, encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
return [e for e in data if isinstance(e, dict) and e.get("content") and e.get("keywords")]
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def _get_memory_path(qq: int) -> str:
|
||
return os.path.join(MEMORY_DIR, f"{qq}.txt")
|
||
|
||
def append_history(qq: int, role: str, content: str, is_auto: bool = False):
|
||
timestamp = get_beijing_time().strftime("%Y-%m-%d %H:%M:%S")
|
||
tag = "[auto]" if is_auto and role == "you" else ""
|
||
line = f"[{timestamp}]{tag}[{role}] {content}"
|
||
with _uid_lock(qq):
|
||
with open(_get_history_path(qq), "a", encoding="utf-8") as f:
|
||
f.write(line + "\n")
|
||
f.flush()
|
||
os.fsync(f.fileno())
|
||
_bump_history_revision(qq)
|
||
|
||
|
||
def append_memory(qq: int, memory_text: str):
|
||
if not memory_text or memory_text == "none":
|
||
return
|
||
timestamp = get_beijing_time().strftime("%Y-%m-%d %H:%M:%S")
|
||
line = f"[{timestamp}] {memory_text}"
|
||
with _uid_lock(qq):
|
||
with open(_get_memory_path(qq), "a", encoding="utf-8") as f:
|
||
f.write(line + "\n")
|
||
f.flush()
|
||
os.fsync(f.fileno())
|
||
def _record_memory(qq: int, memory) -> None:
|
||
"""记录长期记忆：双写旧 memory/<qq>.txt（现有 prompt 读取路径不变）+ 新 facts 库（结构化）。
|
||
memory 兼容三种形态：none/空、纯文本字符串、[{content,importance,keywords}] 列表。"""
|
||
if memory is None or memory == "none" or memory == "" or memory == []:
|
||
return
|
||
if isinstance(memory, list):
|
||
facts = [f for f in memory if isinstance(f, (dict, str))]
|
||
legacy_text = "；".join(
|
||
(f.get("content", "") if isinstance(f, dict) else str(f)) for f in facts
|
||
).strip("；")
|
||
else:
|
||
legacy_text = str(memory)
|
||
facts = [{"content": legacy_text, "importance": 3, "keywords": []}]
|
||
if legacy_text:
|
||
append_memory(qq, legacy_text)
|
||
if facts:
|
||
try:
|
||
memory_store.migrate_legacy(qq)
|
||
added = memory_store.add_facts(qq, facts)
|
||
if added:
|
||
print(f"[DEBUG][chat] 结构化记忆入库 {len(added)} 条 (uid={qq})")
|
||
except Exception as e:
|
||
print(f"[DEBUG][chat] 结构化记忆写入失败: {e}")
|
||
|
||
|
||
async def handle_memory_command(channel, user_id: int, arg: str):
|
||
"""`/memory` 指令：查看/管理结构化长期记忆（facts 库）。
|
||
用法: /memory [ls] | /memory forget <编号> | /memory stats"""
|
||
parts = arg.split()
|
||
sub = parts[0].lower() if parts else "ls"
|
||
memory_store.migrate_legacy(user_id)
|
||
if sub in ("ls", "list", ""):
|
||
rows = memory_store.list_facts(user_id, limit=10)
|
||
total = memory_store.count_facts(user_id)
|
||
if not rows:
|
||
await send_private_message(channel, user_id,
|
||
"还没有记住什么事呢。聊着聊着我会自己记住重要的信息～")
|
||
return
|
||
lines = [f"#{r[0]} {r[1][:80]}" for r in rows]
|
||
await send_private_message(channel, user_id,
|
||
f"我记得的（共 {total} 条，最近 {len(rows)} 条）：\n" + "\n".join(lines)
|
||
+ "\n\n发送 /memory forget <编号> 可以让我忘掉对应那条")
|
||
elif sub in ("forget", "del", "rm"):
|
||
if len(parts) < 2 or not parts[1].isdigit():
|
||
await send_private_message(channel, user_id, "用法: /memory forget <编号>")
|
||
return
|
||
if memory_store.forget_fact(user_id, int(parts[1])):
|
||
await send_private_message(channel, user_id, f"已经忘掉 #{parts[1]} 了")
|
||
else:
|
||
await send_private_message(channel, user_id,
|
||
f"没有找到 #{parts[1]}（可能不存在或已经忘掉了）")
|
||
elif sub in ("stats", "统计"):
|
||
total = memory_store.count_facts(user_id, active_only=True)
|
||
all_ = memory_store.count_facts(user_id, active_only=False)
|
||
await send_private_message(channel, user_id,
|
||
f"长期记忆：活跃 {total} 条 / 已遗忘 {all_ - total} 条")
|
||
elif sub in ("entities", "tags", "标签"):
|
||
from collections import Counter
|
||
rows = memory_store.list_facts(user_id, limit=200)
|
||
cnt = Counter()
|
||
for r in rows:
|
||
for k in (r[3] or "").split(","):
|
||
if k.strip():
|
||
cnt[k.strip()] += 1
|
||
top = cnt.most_common(8)
|
||
if not top:
|
||
await send_private_message(channel, user_id, "还没有可用的记忆标签")
|
||
return
|
||
await send_private_message(channel, user_id,
|
||
"我记住的事按标签分：\n" + "\n".join(f"{k} ×{v}" for k, v in top))
|
||
elif sub == "entity" and len(parts) >= 2:
|
||
name = " ".join(parts[1:])
|
||
card = memory_store.entity_card(user_id, name)
|
||
if not card:
|
||
await send_private_message(channel, user_id, f"没有找到关于「{name}」的记忆")
|
||
return
|
||
await send_private_message(channel, user_id,
|
||
f"关于「{name}」我记得：\n" + "\n".join(f"- {s} {pr} {o}" for s, pr, o, _ in card))
|
||
elif sub == "lore":
|
||
entries = _load_lorebook()
|
||
if not entries:
|
||
await send_private_message(channel, user_id, "还没有世界设定（lorebook.json 为空）")
|
||
return
|
||
await send_private_message(channel, user_id,
|
||
"世界设定（命中关键词时自动想起）：\n" + "\n".join(
|
||
f"• [{'/'.join(e.get('keywords', []))}] {e.get('content','')[:60]}" for e in entries))
|
||
else:
|
||
await send_private_message(channel, user_id, load_instruction("help_instructions.txt"))
|
||
|
||
|
||
|
||
def get_recent_history(qq: int, limit: int = MAX_HISTORY_LINES) -> List[str]:
|
||
path = _get_history_path(qq)
|
||
if not os.path.exists(path):
|
||
return []
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
lines = f.readlines()
|
||
return [line.strip() for line in lines[-limit:]]
|
||
|
||
def get_history_within_budget(qq: int, budget_tokens: int) -> List[str]:
|
||
"""从文件尾向前取历史，累计 token 不超预算（P1: 替代固定条数窗口）。"""
|
||
path = _get_history_path(qq)
|
||
if not os.path.exists(path):
|
||
return []
|
||
with open(path, encoding="utf-8") as f:
|
||
lines = [ln.rstrip("\n") for ln in f if ln.strip()]
|
||
out, used = [], 0
|
||
for ln in reversed(lines):
|
||
t = estimate_tokens(ln) + 4
|
||
if used + t > budget_tokens:
|
||
break
|
||
out.append(ln)
|
||
used += t
|
||
out.reverse()
|
||
return out
|
||
|
||
def get_last_active(qq: int) -> float:
|
||
path = _get_history_path(qq)
|
||
if not os.path.exists(path):
|
||
return 0
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
lines = f.readlines()
|
||
if not lines:
|
||
return 0
|
||
last_line = lines[-1].strip()
|
||
try:
|
||
time_str = last_line.split(']')[0].lstrip('[')
|
||
dt = datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S")
|
||
dt = dt.replace(tzinfo=BEIJING_TZ)
|
||
return dt.timestamp()
|
||
except:
|
||
return 0
|
||
|
||
def get_recent_memory(qq: int, limit: int = MEMORY_MAX_LINES) -> List[str]:
|
||
path = _get_memory_path(qq)
|
||
if not os.path.exists(path):
|
||
return []
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
lines = [ln.strip() for ln in f.readlines() if ln.strip()]
|
||
except OSError:
|
||
return []
|
||
return lines[-limit:]
|
||
|
||
|
||
COMPACT_KEEP_RECENT = 40 # compact 后保留的最近历史条数
|
||
|
||
def _backup_history(qq: int) -> str:
|
||
"""compact/del 前备份历史文件到 history/backup/，返回备份路径"""
|
||
path = _get_history_path(qq)
|
||
if not os.path.exists(path):
|
||
return ""
|
||
bdir = os.path.join(HISTORY_DIR, "backup")
|
||
os.makedirs(bdir, exist_ok=True)
|
||
ts = get_beijing_time().strftime("%Y%m%d%H%M%S")
|
||
bpath = os.path.join(bdir, f"{qq}_{ts}.txt")
|
||
shutil.copyfile(path, bpath)
|
||
return bpath
|
||
|
||
async def compact_history(qq: int, keep_recent: int = COMPACT_KEEP_RECENT) -> str:
|
||
"""Summarize a stable snapshot; CAS prevents overwriting messages added during the model call."""
|
||
path = _get_history_path(qq)
|
||
with _uid_lock(qq):
|
||
if not os.path.exists(path):
|
||
return "当前没有聊天记录可总结。"
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
lines = [ln.rstrip("\n") for ln in f.readlines() if ln.strip()]
|
||
revision = _history_revisions.get(int(qq), 0)
|
||
if len(lines) <= keep_recent:
|
||
return f"历史只有 {len(lines)} 条（<= {keep_recent}），无需总结。"
|
||
kept_lines = lines[-keep_recent:] if keep_recent > 0 else []
|
||
old_lines = lines[:len(lines) - len(kept_lines)]
|
||
with skill_runtime.suspend_skills():
|
||
summary_raw = await call_deepseek(
|
||
load_instruction("compact_instructions.txt"),
|
||
"以下是需要总结的聊天记录：" + "\n".join(old_lines), 0)
|
||
if not summary_raw:
|
||
return "总结生成失败，请稍后再试。（历史文件未改动）"
|
||
summary = " ".join(summary_raw.strip().split())
|
||
ts = get_beijing_time().strftime("%Y-%m-%d %H:%M:%S")
|
||
summary_line = f"[{ts}][summary] 前 {len(old_lines)} 条对话的总结：{summary}"
|
||
with _uid_lock(qq):
|
||
if _history_revisions.get(int(qq), 0) != revision:
|
||
return "总结期间有新消息，本次未改写历史，请重试。（历史文件未改动）"
|
||
bpath = _backup_history(qq)
|
||
_atomic_write_text(path, summary_line + "\n" + "\n".join(kept_lines) + "\n")
|
||
_bump_history_revision(qq)
|
||
return (f"已完成：{len(old_lines)} 条历史被总结为一条记录，保留最近 {len(kept_lines)} 条。\n"
|
||
f"备份: {bpath}\n"
|
||
f"总结预览: {summary[:80]}")
|
||
|
||
|
||
def delete_last_history(qq: int, n: int) -> str:
|
||
"""删除最近 n 条历史（含用户输入与 bot 回复）。返回结果文本。"""
|
||
path = _get_history_path(qq)
|
||
with _uid_lock(qq):
|
||
if not os.path.exists(path):
|
||
return "当前没有聊天记录可删除。"
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
lines = [ln.rstrip("\n") for ln in f.readlines() if ln.strip()]
|
||
if not lines:
|
||
return "历史已为空。"
|
||
if n <= 0:
|
||
return "请指定条数，例如 /del 10"
|
||
n = min(n, len(lines))
|
||
bpath = _backup_history(qq)
|
||
removed = lines[-n:]
|
||
remaining = lines[:len(lines) - n]
|
||
_atomic_write_text(path, ("\n".join(remaining) + "\n") if remaining else "")
|
||
_bump_history_revision(qq)
|
||
return (f"已删除最近 {n} 条历史（含用户输入与回复），剩余 {len(remaining)} 条。\n"
|
||
f"备份: {bpath}\n"
|
||
f"最后一条被删: {removed[-1][:50]}")
|
||
|
||
def stickers_enabled() -> bool:
|
||
"""运行时开关；关闭后不向模型提供标签，也拦截任何直接发送调用。"""
|
||
return bool(load_setting().get("stickers", True))
|
||
|
||
|
||
def available_sticker_tags():
|
||
"""表情库标签 = stickers/ 下的文件名（去扩展名）。"""
|
||
if not stickers_enabled():
|
||
return []
|
||
try:
|
||
tags = []
|
||
for fn in sorted(os.listdir(STICKERS_DIR)):
|
||
stem, ext = os.path.splitext(fn)
|
||
if ext.lower() in (".webp", ".png", ".jpg", ".jpeg"):
|
||
tags.append(stem)
|
||
return tags
|
||
except OSError:
|
||
return []
|
||
|
||
|
||
def _learn_stickers(scope: int, image_infos: List[dict], parsed: dict):
|
||
"""把主模型对本论用户贴纸的分类写入对应私聊/群聊场景库。"""
|
||
if not stickers_enabled() or not isinstance(parsed, dict):
|
||
return
|
||
learned = parsed.get("learned_stickers")
|
||
if not isinstance(learned, list):
|
||
return
|
||
sources = {
|
||
str(info.get("source_id")): info for info in (image_infos or [])
|
||
if info.get("kind") == "sticker" and info.get("source_id")
|
||
}
|
||
for classification in learned:
|
||
if not isinstance(classification, dict):
|
||
continue
|
||
source = sources.get(str(classification.get("source_id") or ""))
|
||
if not source:
|
||
continue
|
||
try:
|
||
item = STICKER_STORE.learn(STICKER_GLOBAL_SCOPE, source, classification)
|
||
if item:
|
||
print(f"[sticker] 已学习 {item.get('id')} scenes={item.get('scenes')}")
|
||
except Exception as exc:
|
||
print(f"[sticker] 学习失败: {type(exc).__name__}: {exc}")
|
||
|
||
|
||
def _sticker_prompt_summary(scope: int) -> str:
|
||
if not stickers_enabled():
|
||
return ""
|
||
try:
|
||
return STICKER_STORE.prompt_summary(STICKER_GLOBAL_SCOPE)
|
||
except Exception as exc:
|
||
print(f"[sticker] 场景摘要读取失败: {exc}")
|
||
return ""
|
||
|
||
|
||
def _with_sticker_sources(user_message: str, image_infos: List[dict]) -> str:
|
||
stickers = [info for info in (image_infos or []) if info.get("kind") == "sticker"]
|
||
if not stickers:
|
||
return user_message
|
||
markers = [
|
||
f"[用户发来表情包 source_id={info.get('source_id')}，请识别画面、情绪和使用场景并写入learned_stickers。]"
|
||
for info in stickers
|
||
]
|
||
return "\n".join(markers + ([user_message] if user_message else []))
|
||
|
||
|
||
async def _confirm_sticker(item: dict, scene: str, context: str) -> bool:
|
||
"""发送前让同一个主模型再看抽中原图；审核失败时保守不发送。"""
|
||
preview = item.get("preview_path")
|
||
if not preview or not os.path.exists(preview):
|
||
return False
|
||
info = {
|
||
"url": preview,
|
||
"file_size": str(os.path.getsize(preview)),
|
||
}
|
||
prompt = (
|
||
"你在审核一张准备发送到当前聊天的表情包。只有它确实符合当前场景、语气和情绪时才允许发送。"
|
||
"特别区分日常撒娇/轻微难过与真正崩溃/强烈悲伤，不要只看画面相似就放行。"
|
||
"只输出JSON：{\"send\":true} 或 {\"send\":false}。\n"
|
||
f"候选场景：{scene}\n表情包描述：{item.get('description', '')}"
|
||
)
|
||
try:
|
||
with skill_runtime.suspend_skills():
|
||
raw = await call_deepseek(prompt, context or "当前聊天", image_infos=[info], reasoning=True)
|
||
if not raw:
|
||
return False
|
||
start, end = raw.find("{"), raw.rfind("}")
|
||
if start < 0 or end < start:
|
||
return False
|
||
decision = json.loads(raw[start:end + 1])
|
||
return decision.get("send") is True
|
||
except Exception as exc:
|
||
print(f"[sticker] 发送前确认失败: {type(exc).__name__}: {exc}")
|
||
return False
|
||
|
||
|
||
async def _maybe_send_sticker(websocket, user_id, scene, context="", msg_type="private", target_id=None):
|
||
"""按场景随机偷来的贴纸，经主模型复核后发送；旧预设库不参与。"""
|
||
if not stickers_enabled() or not isinstance(scene, str) or scene.strip().lower() in ("", "none"):
|
||
return False
|
||
try:
|
||
item = STICKER_STORE.random_for_scene(STICKER_GLOBAL_SCOPE, scene.strip())
|
||
except Exception as exc:
|
||
print(f"[sticker] 场景抽取失败: {exc}")
|
||
return False
|
||
if not item or not await _confirm_sticker(item, scene.strip(), context):
|
||
return False
|
||
try:
|
||
from aiogram.types import FSInputFile
|
||
if msg_type == "group":
|
||
channel, raw = websocket, int(target_id)
|
||
else:
|
||
resolved = _resolve_delivery(user_id, websocket)
|
||
if resolved is None:
|
||
return False
|
||
channel, raw = resolved
|
||
if not hasattr(channel, "bot"):
|
||
return False
|
||
raw = channel._to_raw_target(raw) if hasattr(channel, "_to_raw_target") else raw
|
||
await channel.bot.send_sticker(chat_id=raw, sticker=FSInputFile(item["path"]))
|
||
print(f"[sticker] 已发送 scene={scene} id={item.get('id')}")
|
||
return True
|
||
except Exception as exc:
|
||
print(f"[sticker] 表情发送失败 {scene}: {exc}")
|
||
return False
|
||
|
||
|
||
async def _tool_web_search(query):
|
||
"""DeepSeek Search API: POST /v1/tools/search"""
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(
|
||
"https://api.deepseek.com/v1/tools/search",
|
||
headers={"Authorization": f"Bearer {DEEPSEEK_API_KEY}",
|
||
"Content-Type": "application/json"},
|
||
json={"query": query, "count": 5, "need_content": False},
|
||
timeout=30) as resp:
|
||
if resp.status != 200:
|
||
return f"（搜索失败 HTTP {resp.status}）"
|
||
data = await resp.json()
|
||
results = data.get("results") or []
|
||
if not results:
|
||
return f"（搜索 {query} 没有结果）"
|
||
lines = [f"【联网搜索 {query}】"]
|
||
for r in results:
|
||
lines.append(f"- {r.get('title')}\n {r.get('url')}\n {r.get('summary') or r.get('snippet')}")
|
||
return "\n".join(lines)
|
||
except Exception as e:
|
||
return f"（搜索异常: {e}）"
|
||
|
||
|
||
async def _tool_fetch(url):
|
||
"""DeepSeek Fetch API: POST /v1/tools/fetch，网页转 Markdown"""
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(
|
||
"https://api.deepseek.com/v1/tools/fetch",
|
||
headers={"Authorization": f"Bearer {DEEPSEEK_API_KEY}",
|
||
"Content-Type": "application/json"},
|
||
json={"url": url}, timeout=30) as resp:
|
||
if resp.status != 200:
|
||
return f"（读取网页失败 HTTP {resp.status}）"
|
||
data = await resp.json()
|
||
content = data.get("content") or data.get("markdown") or str(data)
|
||
return f"【网页内容 {url}】\n{content[:4000]}"
|
||
except Exception as e:
|
||
return f"（读取网页异常: {e}）"
|
||
|
||
|
||
async def run_tools(tools):
|
||
"""执行模型请求的工具（每轮最多 2 个），返回注入模型的结果文本。"""
|
||
if not isinstance(tools, list) or not tools:
|
||
return ""
|
||
out = []
|
||
for t in tools[:2]:
|
||
if not isinstance(t, dict):
|
||
continue
|
||
name, args = t.get("name"), (t.get("args") or {})
|
||
if name == "web_search":
|
||
q = str(args.get("query") or "")[:128]
|
||
if q:
|
||
out.append(await _tool_web_search(q))
|
||
elif name == "fetch":
|
||
u = str(args.get("url") or "")
|
||
if u.startswith("http"):
|
||
out.append(await _tool_fetch(u))
|
||
return "\n\n".join(x for x in out if x)
|
||
|
||
|
||
def _message_with_time(message: str) -> str:
|
||
"""时间随当前用户消息发送，避免秒级时间戳打断 system prompt 缓存前缀。"""
|
||
now = get_beijing_time()
|
||
return f"现在是 {now.strftime('%Y-%m-%d %A %H:%M:%S')}\n{message}"
|
||
|
||
|
||
def build_system_prompt(qq: int, is_auto: bool = False, query: str = "", tool_context=None) -> str:
|
||
status = load_user_status(qq)
|
||
status_info = f"当前用户状态：pause={status['pause']}, sleep={status['sleep']}"
|
||
|
||
json_instruction = load_instruction("agent_instructions.txt")
|
||
if tool_context is not None and not is_auto:
|
||
import agent_loop
|
||
json_instruction += "\n\n" + agent_loop.tool_instructions(tool_context)
|
||
# P2: 关系卡（相识天数/对话次数/最近约定），恒定注入
|
||
relation_info = ""
|
||
try:
|
||
rel = memory_store.get_relation(qq)
|
||
if rel and rel.get("msg_count"):
|
||
days = int((get_beijing_timestamp() - rel["first_met"]) / 86400) if rel.get("first_met") else 0
|
||
relation_info = f"\n\n【我们的关系】相识 {days} 天，这是第 {rel['msg_count']} 次对话。"
|
||
if rel.get("notes"):
|
||
relation_info += f"最近：{rel['notes'][-1]['text']}"
|
||
except Exception:
|
||
relation_info = ""
|
||
|
||
# P3: lorebook（世界设定，关键词触发，不命中零成本）
|
||
lore_info = ""
|
||
for ent in _load_lorebook():
|
||
if any(k and k in query for k in ent.get("keywords", [])):
|
||
lore_info += f"\n【设定】{ent.get('content', '')}"
|
||
|
||
# P3: 实体卡（query 命中实体名时注入其当前有效关系）
|
||
entity_info = ""
|
||
if query:
|
||
for name, _cnt in memory_store.list_entities(qq, limit=50):
|
||
if name and name in query:
|
||
card = memory_store.entity_card(qq, name)
|
||
if card:
|
||
entity_info += f"\n【关于 {name}】\n" + "\n".join(
|
||
f"- {s} {pr} {o}" for s, pr, o, _imp in card)
|
||
|
||
# 只提供从用户贴纸学来的场景，不提供固定文件标签；同场景由库随机抽取。
|
||
sticker_summary = _sticker_prompt_summary(qq)
|
||
json_instruction += "\n【表情包场景分类定义】\n" + sticker_store.scene_prompt()
|
||
if sticker_summary:
|
||
json_instruction += (
|
||
"\n【可使用的表情包场景】sticker字段只能填写当前库存中的场景id，否则写none。\n"
|
||
+ sticker_summary
|
||
)
|
||
else:
|
||
json_instruction += "\n当前没有可发送的用户表情包，sticker字段写none。"
|
||
json_instruction += "\n【表情包场景分类定义】\n" + sticker_store.scene_prompt()
|
||
|
||
# P1: 召回式长期记忆（FTS5 按当前输入召回 + 高重要度置顶），替代"最近 100 条"
|
||
memory_store.migrate_legacy(qq)
|
||
facts = memory_store.recall_facts(qq, query)
|
||
memory_str = "\n".join(f"- {c}" for _, c, _ in facts) if facts else "（无）"
|
||
try:
|
||
notebook_context = notebook.notebook.context_for_chat(qq, query=query)
|
||
except Exception as exc:
|
||
print(f"[notebook] 上下文读取失败 uid={qq}: {exc}")
|
||
notebook_context = ""
|
||
# 仅此私聊入口读取个人衣柜；群聊构造器不调用这里。
|
||
try:
|
||
wardrobe_context = wardrobe.wardrobe.context_for_chat(qq, query=query)
|
||
except Exception as exc:
|
||
print(f"[wardrobe] 上下文读取失败 uid={qq}: {type(exc).__name__}")
|
||
wardrobe_context = "衣柜暂时无法读取，不要猜测库存、赠予记录或当前穿着。"
|
||
|
||
extra_auto = ""
|
||
if is_auto:
|
||
extra_auto = "\n" + load_instruction("notification_instructions.txt")
|
||
last_active_ts = get_last_active(qq)
|
||
if last_active_ts > 0:
|
||
hours_ago = (get_beijing_timestamp() - last_active_ts) / 3600
|
||
extra_auto += f"\n注意：用户已经有 {hours_ago:.1f} 小时没有和你说话了。"
|
||
|
||
# P1: 历史窗口按 token 预算取（人设+指令+记忆占用之后的剩余额度）
|
||
persona_prompt = user_persona_prompt(qq)
|
||
fixed_tokens = (estimate_tokens(persona_prompt)
|
||
+ estimate_tokens(status_info) + estimate_tokens(json_instruction)
|
||
+ estimate_tokens(extra_auto) + estimate_tokens(memory_str)
|
||
+ estimate_tokens(notebook_context) + estimate_tokens(wardrobe_context)
|
||
+ estimate_tokens(relation_info)
|
||
+ estimate_tokens(lore_info) + estimate_tokens(entity_info))
|
||
skill_reserve = skill_runtime.estimated_overhead()
|
||
history_budget = max(2000, PROMPT_TOKEN_BUDGET - fixed_tokens - skill_reserve - 200)
|
||
history_lines = get_history_within_budget(qq, history_budget)
|
||
chat_history_str = "\n".join(history_lines) if history_lines else "（无最近对话）"
|
||
|
||
if is_auto:
|
||
# 反重复: 摆出最近的主动消息, 明确要求换内容/活动/句式（治"每次都泡果茶"）
|
||
recent_auto = [ln for ln in history_lines if "[auto]" in ln][-3:]
|
||
if recent_auto:
|
||
extra_auto += ("\n\n【反重复】你最近的主动消息是：\n" + "\n".join(recent_auto)
|
||
+ "\n这次请不要重复同样的活动、开场白或句式，换一个新的内容/话题。")
|
||
|
||
return f"{persona_prompt}\n\n{json_instruction}\n\n【长期记忆】以下是按当前话题召回的相关记忆（含用户偏好、承诺与约定，请始终记住并遵守，不要重复询问已记住的事）：\n{memory_str}\n\n{notebook_context}\n\n{wardrobe_context}\n\n{status_info}{relation_info}{lore_info}{entity_info}\n\n以下是最近你和用户的聊天记录：\n{chat_history_str}\n\n{extra_auto}\n现在请根据以上信息，按照JSON格式输出你的回复。"
|
||
|
||
# ---------- 魔搭社区：每日限额状态管理（存 chat/setting.json，跨重启保留） ----------
|
||
def _modelscope_load():
|
||
"""读取当日魔搭状态；跨天自动重置计数与耗尽标记。"""
|
||
with _setting_lock:
|
||
st = load_setting()
|
||
today = get_beijing_time().strftime("%Y-%m-%d")
|
||
if st.get("modelscope_date") != today:
|
||
st["modelscope_date"] = today
|
||
st["modelscope_count"] = 0
|
||
st["modelscope_exhausted"] = False
|
||
save_setting(st)
|
||
return st
|
||
|
||
|
||
def _modelscope_available():
|
||
st = _modelscope_load()
|
||
if st.get("modelscope_permanent_blocked"):
|
||
return False # 持续触发内容审查，永久不再导向魔搭（需手动解除）
|
||
if st.get("modelscope_suspect_date") == st.get("modelscope_date"):
|
||
return False # 今日已触发审查嫌疑：当天剩余对话直接 DS
|
||
if st.get("modelscope_exhausted"):
|
||
return False
|
||
return int(st.get("modelscope_count", 0)) < MODELSCOPE_DAILY_LIMIT
|
||
|
||
|
||
def _modelscope_mark_suspect():
|
||
"""内容审查嫌疑（空壳响应）：当日不再尝试魔搭；跨天再次触发则永久禁用。"""
|
||
with _setting_lock:
|
||
st = _modelscope_load()
|
||
today = st["modelscope_date"]
|
||
if st.get("modelscope_suspect_date") == today:
|
||
return # 今天已标记，不重复计数
|
||
days = int(st.get("modelscope_suspect_days", 0)) + 1
|
||
st["modelscope_suspect_date"] = today
|
||
st["modelscope_suspect_days"] = days
|
||
if days >= 2:
|
||
st["modelscope_permanent_blocked"] = True
|
||
print(f"[DEBUG][chat] ModelScope 内容审查反复触发（{days} 天），永久不再导向魔搭；"
|
||
f"如需恢复请删除 setting.json 的 modelscope_permanent_blocked")
|
||
else:
|
||
print(f"[DEBUG][chat] ModelScope 疑似内容审查拦截（第 {days} 天），今日剩余对话直接走 DeepSeek official")
|
||
save_setting(st)
|
||
|
||
|
||
def _modelscope_mark_success():
|
||
with _setting_lock:
|
||
st = _modelscope_load()
|
||
st["modelscope_count"] = int(st.get("modelscope_count", 0)) + 1
|
||
save_setting(st)
|
||
|
||
|
||
def _modelscope_mark_exhausted():
|
||
with _setting_lock:
|
||
st = _modelscope_load()
|
||
if not st.get("modelscope_exhausted"):
|
||
st["modelscope_exhausted"] = True
|
||
save_setting(st)
|
||
print(f"[DEBUG][chat] ModelScope 今日额度已耗尽（已用 {st.get('modelscope_count', 0)} 次），"
|
||
f"后续回退 DeepSeek official")
|
||
|
||
|
||
def _is_ms_quota_error(status: int, body: str) -> bool:
|
||
"""宽松识别魔搭额度/配额类错误。宁可误回退 DS，也不让额度错误卡住对话。"""
|
||
if status in (402, 403, 429):
|
||
return True
|
||
lowered = (body or "").lower()
|
||
return any(k in lowered for k in (
|
||
"quota", "insufficient", "exceeded", "balance",
|
||
"额度", "余额不足", "超限", "配额",
|
||
))
|
||
|
||
|
||
def _reasoning_params(reasoning: bool) -> dict:
|
||
"""deepseek-flash 显式思考开关；Agent 用 high，普通聊天明确关闭。"""
|
||
if reasoning:
|
||
return {"thinking": {"type": "enabled"}, "reasoning_effort": "high"}
|
||
return {"thinking": {"type": "disabled"}}
|
||
|
||
|
||
def _log_model_usage(provider: str, data: dict) -> None:
|
||
"""只记录服务端用量，不记录消息正文、密钥或推理内容。"""
|
||
usage = data.get("usage") if isinstance(data, dict) else None
|
||
if not isinstance(usage, dict):
|
||
return
|
||
details = usage.get("prompt_tokens_details") or {}
|
||
cached = usage.get("prompt_cache_hit_tokens")
|
||
if cached is None and isinstance(details, dict):
|
||
cached = details.get("cached_tokens")
|
||
fields = {"prompt_tokens": usage.get("prompt_tokens"), "cached_tokens": cached,
|
||
"cache_miss_tokens": usage.get("prompt_cache_miss_tokens"),
|
||
"completion_tokens": usage.get("completion_tokens")}
|
||
# Providers can omit these fields; unknown is not the same as zero hits.
|
||
fields = {k: v for k, v in fields.items() if isinstance(v, int) and not isinstance(v, bool)}
|
||
if fields:
|
||
print(f"[model-usage] {provider} " + json.dumps(fields, sort_keys=True))
|
||
|
||
|
||
async def _call_modelscope(messages: List[dict], presence_penalty: float = 0.0,
|
||
reasoning: bool = False) -> Optional[str]:
|
||
"""调用魔搭社区 API-Inference。成功返回 content；失败返回 None。
|
||
明确的额度错误会标记今日 exhausted，其余错误仅本次回退。"""
|
||
_MS_EMPTY_SHELL.set(False)
|
||
headers = {
|
||
"Authorization": f"Bearer {MODELSCOPE_API_KEY}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
payload = {"model": MODELSCOPE_MODEL, "messages": messages,
|
||
"temperature": 0.7, "stream": False}
|
||
payload.update(_reasoning_params(reasoning))
|
||
if presence_penalty:
|
||
payload["presence_penalty"] = presence_penalty
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(f"{MODELSCOPE_BASE_URL}/chat/completions",
|
||
headers=headers, json=payload, timeout=60) as resp:
|
||
if resp.status == 200:
|
||
data = await resp.json()
|
||
_log_model_usage("modelscope", data)
|
||
# 防御：魔搭对超长/异常输入可能返回 choices:null 的空壳 200，
|
||
# 直接下标会 TypeError；结构不完整视为内容审查嫌疑，回退 DS。
|
||
choices = data.get("choices")
|
||
if not choices:
|
||
print("[DEBUG][chat] ModelScope 响应缺少 choices（疑似内容审查），回退 DeepSeek official")
|
||
_MS_EMPTY_SHELL.set(True)
|
||
_modelscope_mark_suspect()
|
||
return None
|
||
content = (choices[0].get("message") or {}).get("content")
|
||
if not content:
|
||
print("[DEBUG][chat] ModelScope 响应 content 为空（疑似内容审查），回退 DeepSeek official")
|
||
_MS_EMPTY_SHELL.set(True)
|
||
_modelscope_mark_suspect()
|
||
return None
|
||
print(f"[DEBUG][chat] ModelScope 响应成功，长度 {len(content)}")
|
||
return content
|
||
text = await resp.text()
|
||
print(f"[DEBUG][chat] ModelScope 失败 HTTP {resp.status}: {text[:200]}")
|
||
if _is_ms_quota_error(resp.status, text):
|
||
_modelscope_mark_exhausted()
|
||
return None
|
||
except Exception as e:
|
||
print(f"[DEBUG][chat] ModelScope 请求异常: {type(e).__name__}: {e}")
|
||
return None
|
||
|
||
|
||
_HIST_HEAD = "以下是最近你和用户的聊天记录：\n"
|
||
_MEMO_HEAD = "\n\n【长期记忆】"
|
||
|
||
|
||
def _ms_short_system(system_prompt: str) -> str:
|
||
"""魔搭重试专用：砍掉聊天记录段（保留人设/时间/状态/长期记忆/JSON 指令）。
|
||
魔搭免费档对超长输入返回 choices:null 空壳，砍历史后重试一次。"""
|
||
i = system_prompt.find(_HIST_HEAD)
|
||
j = system_prompt.find(_MEMO_HEAD, i + 1) if i != -1 else -1
|
||
if i == -1 or j == -1:
|
||
# 结构不符：保守保留头 4000 + 尾 2000 字符（尾部含 JSON 指令）
|
||
return (system_prompt[:4000] + "\n……（中间内容过长省略）……\n"
|
||
+ system_prompt[-2000:])
|
||
return (system_prompt[:i] + _HIST_HEAD
|
||
+ "（对话历史过长，本回合省略，请基于人设与长期记忆自然回应。）"
|
||
+ system_prompt[j:])
|
||
|
||
|
||
async def call_deepseek(system_prompt: str, user_message: str, retry: int = 0,
|
||
presence_penalty: float = 0.0, image_infos: List[dict] = None,
|
||
reasoning: bool = False) -> Optional[str]:
|
||
"""用户可见生成的统一入口；未开启 Skill 会话时保持原请求不变。"""
|
||
budget = {"_prompt_token_budget": PROMPT_TOKEN_BUDGET} if skill_runtime.estimated_overhead() else {}
|
||
return await skill_runtime.generate_with_skills(
|
||
_call_deepseek_transport, system_prompt, user_message, retry,
|
||
presence_penalty=presence_penalty, image_infos=image_infos, reasoning=reasoning, **budget)
|
||
|
||
|
||
async def _call_deepseek_transport(system_prompt: str, user_message: str, retry: int = 0,
|
||
presence_penalty: float = 0.0, image_infos: List[dict] = None,
|
||
reasoning: bool = False) -> Optional[str]:
|
||
"""仅执行模型传输，不递归进入 Skill 读取循环。"""
|
||
user_content = await _build_user_content(user_message, image_infos)
|
||
if _modelscope_available():
|
||
content = await _call_modelscope([
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_content},
|
||
], presence_penalty=presence_penalty, reasoning=reasoning)
|
||
if (content is None and len(system_prompt) > 6000 and _MS_EMPTY_SHELL.get()
|
||
and not skill_runtime.estimated_overhead()):
|
||
# 魔搭对超长输入返回空壳响应：砍历史重试一次，原图保留。
|
||
short = _ms_short_system(system_prompt)
|
||
print(f"[DEBUG][chat] ModelScope 长输入空壳，砍历史重试（{len(system_prompt)}→{len(short)} 字符）")
|
||
content = await _call_modelscope([
|
||
{"role": "system", "content": short},
|
||
{"role": "user", "content": user_content},
|
||
], presence_penalty=presence_penalty, reasoning=reasoning)
|
||
if content is not None:
|
||
LAST_HTTP_STATUS.set(200)
|
||
_modelscope_mark_success()
|
||
return content
|
||
print("[DEBUG][chat] ModelScope 不可用，回退 DeepSeek official")
|
||
return await _call_deepseek_official(
|
||
system_prompt, user_content, retry,
|
||
presence_penalty=presence_penalty, reasoning=reasoning)
|
||
|
||
|
||
async def _call_deepseek_official(system_prompt: str, user_content: Any, retry: int = 0,
|
||
presence_penalty: float = 0.0,
|
||
reasoning: bool = False) -> Optional[str]:
|
||
LAST_HTTP_STATUS.set(0)
|
||
print(f"[DEBUG][chat] 发起 DeepSeek official 请求 (retry={retry})")
|
||
headers = {
|
||
"Authorization": f"Bearer {DEEPSEEK_API_KEY}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
payload = {
|
||
"model": MODEL_NAME,
|
||
"messages": [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_content}
|
||
],
|
||
"temperature": 0.7,
|
||
"stream": False
|
||
}
|
||
payload.update(_reasoning_params(reasoning))
|
||
if presence_penalty:
|
||
payload["presence_penalty"] = presence_penalty
|
||
url = f"{DEEPSEEK_BASE_URL}/chat/completions"
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(url, headers=headers, json=payload, timeout=60) as resp:
|
||
print(f"[DEBUG][chat] HTTP 状态码: {resp.status}")
|
||
LAST_HTTP_STATUS.set(resp.status)
|
||
if resp.status == 200:
|
||
data = await resp.json()
|
||
_log_model_usage("deepseek", data)
|
||
content = data["choices"][0]["message"]["content"]
|
||
print(f"[DEBUG][chat] API 响应成功，长度 {len(content)}")
|
||
return content
|
||
else:
|
||
text = await resp.text()
|
||
print(f"[DEBUG][chat] 请求失败，状态码 {resp.status}，响应体: {text[:500]}")
|
||
if 400 <= resp.status < 500:
|
||
return None # 4xx 客户端错误（余额/限流/鉴权）重试无意义
|
||
if retry < MAX_RETRIES:
|
||
await asyncio.sleep(2 ** retry)
|
||
return await _call_deepseek_official(
|
||
system_prompt, user_content, retry + 1,
|
||
presence_penalty=presence_penalty, reasoning=reasoning)
|
||
return None
|
||
except Exception as e:
|
||
print(f"[DEBUG][chat] 请求异常: {e}")
|
||
if retry < MAX_RETRIES:
|
||
await asyncio.sleep(2 ** retry)
|
||
return await _call_deepseek_official(
|
||
system_prompt, user_content, retry + 1,
|
||
presence_penalty=presence_penalty, reasoning=reasoning)
|
||
return None
|
||
|
||
def _fallback_by_status(status: int) -> str:
|
||
"""按 API HTTP 状态码选择兜底回复"""
|
||
if status == 402:
|
||
return "我好饿…能给我点钱去吃饭吗…"
|
||
if status == 429:
|
||
return "等下等下，我在忙呢，晚点回你～"
|
||
if status in (400, 403):
|
||
return "这个我不想说呢"
|
||
if 500 <= status < 600:
|
||
return "手机没电了，我去充个电，等下再来找你！"
|
||
return "唔，我现在手机好像没有信号，消息发不出去，等下再回你～"
|
||
|
||
|
||
def _tool_plan_requires_reasoning(parsed: dict, tool_context) -> bool:
|
||
return bool(
|
||
tool_context is not None
|
||
and isinstance(parsed, dict)
|
||
and parsed.get("understood") is not False
|
||
and isinstance(parsed.get("tools"), list)
|
||
and parsed["tools"]
|
||
)
|
||
|
||
|
||
def _chat_requires_reasoning(parsed: dict, tool_context, is_auto: bool = False) -> bool:
|
||
if is_auto or not isinstance(parsed, dict):
|
||
return False
|
||
return parsed.get("need_reasoning") is True or _tool_plan_requires_reasoning(parsed, tool_context)
|
||
|
||
|
||
def _json_retry_uses_reasoning(attempt: int) -> bool:
|
||
"""初次非思考 + 一次普通重试后，最后一次 JSON 重试启用 reasoning。"""
|
||
return attempt == MAX_RETRIES - 2
|
||
|
||
|
||
async def _reasoning_json_response(system_prompt: str, user_message: str,
|
||
image_infos: List[dict] = None) -> Optional[dict]:
|
||
"""reasoning 规划最多两次；第二次只修正 JSON，不执行任何非思考工具计划。"""
|
||
prompt = system_prompt
|
||
for attempt in range(2):
|
||
raw = await call_deepseek(prompt, user_message, 0,
|
||
image_infos=image_infos, reasoning=True)
|
||
parsed = parse_json_response(raw) if raw else None
|
||
if parsed:
|
||
return parsed
|
||
print(f"[DEBUG][chat] reasoning JSON 失败(第{attempt + 1}次): {(raw or '')[:800]!r}")
|
||
prompt = (
|
||
system_prompt
|
||
+ "\n\n【输出修正】上次回答为空或不符合约定JSON。现在只输出一个完整、合法的JSON对象；"
|
||
"message必须是数组，每项都必须包含字符串content；tools按实际调用填写，"
|
||
"不要输出分析过程、Markdown或其他文字。"
|
||
)
|
||
return None
|
||
|
||
|
||
def parse_json_response(raw: str) -> Optional[dict]:
|
||
if not raw:
|
||
return None
|
||
import re
|
||
json_pattern = r'```json\s*(\{.*?\})\s*```'
|
||
match = re.search(json_pattern, raw, re.DOTALL)
|
||
if match:
|
||
json_str = match.group(1)
|
||
else:
|
||
start = raw.find('{')
|
||
end = raw.rfind('}')
|
||
if start != -1 and end != -1:
|
||
json_str = raw[start:end+1]
|
||
else:
|
||
return None
|
||
try:
|
||
result = json.loads(json_str)
|
||
message = result.get("message") if isinstance(result, dict) else None
|
||
if isinstance(message, list) and all(
|
||
isinstance(item, dict) and isinstance(item.get("content"), str)
|
||
for item in message):
|
||
return result
|
||
return None
|
||
except json.JSONDecodeError:
|
||
return None
|
||
|
||
async def get_wallet_balance() -> Optional[float]:
|
||
"""查询 DeepSeek 账户余额（元）。失败返回 None。"""
|
||
base = DEEPSEEK_BASE_URL
|
||
if base.endswith("/v1"):
|
||
base = base[:-3]
|
||
url = base + "/user/balance"
|
||
headers = {"Authorization": f"Bearer {DEEPSEEK_API_KEY}"}
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get(url, headers=headers, timeout=30) as resp:
|
||
if resp.status == 200:
|
||
data = await resp.json()
|
||
for info in data.get("balance_infos") or []:
|
||
if info.get("currency") == "CNY":
|
||
return float(info["total_balance"])
|
||
else:
|
||
print(f"[DEBUG][chat] 余额查询失败，状态码 {resp.status}")
|
||
except Exception as e:
|
||
print(f"[DEBUG][chat] 余额查询异常: {e}")
|
||
return None
|
||
|
||
|
||
def _strip_trailing_period(text: str) -> str:
|
||
"""消息末尾的单个中文句号清除: 你好。-> 你好; 你好。。不变（连续句号保留）"""
|
||
if len(text) >= 2 and text[-1] == "。" and text[-2] != "。":
|
||
return text[:-1]
|
||
return text
|
||
|
||
|
||
def set_delivery(uid, channel, target_id):
|
||
"""Bind one raw private destination for a unified UID.
|
||
|
||
``channel`` may be a Channel object or its registered name. This is the only
|
||
way to select a specific one of multiple same-channel bindings; proactive
|
||
delivery otherwise uses identity's single preferred target.
|
||
"""
|
||
uid = int(uid)
|
||
target_id = int(target_id)
|
||
if not channel:
|
||
raise ValueError("channel is required")
|
||
_delivery_targets[uid] = (channel, target_id)
|
||
name = channel if isinstance(channel, str) else getattr(channel, "name", "")
|
||
if name:
|
||
_user_channel[uid] = name
|
||
return name, target_id
|
||
|
||
|
||
def _channel_by_name(name, preferred=None):
|
||
if not name:
|
||
return None
|
||
if preferred is not None and getattr(preferred, "name", None) == name:
|
||
return preferred
|
||
if _websocket is not None and getattr(_websocket, "name", None) == name:
|
||
return _websocket
|
||
try:
|
||
from channel import get_channel
|
||
return get_channel(name)
|
||
except (ImportError, AttributeError):
|
||
return None
|
||
|
||
|
||
def _identity_target(uid):
|
||
"""Return exactly one (channel, raw target), never a mirror fanout list."""
|
||
try:
|
||
import identity
|
||
except ImportError:
|
||
return None
|
||
try:
|
||
target = identity.push_target(int(uid))
|
||
except (AttributeError, TypeError, ValueError):
|
||
target = None
|
||
if target:
|
||
return target
|
||
try:
|
||
targets = identity.push_targets(int(uid))
|
||
except (AttributeError, TypeError, ValueError):
|
||
return None
|
||
return targets[0] if targets else None
|
||
|
||
|
||
def _resolve_delivery(uid, websocket=None):
|
||
uid = int(uid)
|
||
explicit = _delivery_targets.get(uid)
|
||
if explicit:
|
||
channel, target_id = explicit
|
||
if isinstance(channel, str):
|
||
channel = _channel_by_name(channel, websocket)
|
||
if channel is not None:
|
||
return channel, int(target_id)
|
||
target = _identity_target(uid)
|
||
if not target:
|
||
return None
|
||
name, target_id = target
|
||
channel = _channel_by_name(name, websocket)
|
||
if channel is None:
|
||
return None
|
||
return channel, int(target_id)
|
||
|
||
|
||
async def send_private_message(websocket, user_id: int, content: str):
|
||
"""Send to one resolved raw target; a 9e9 unified UID is never a chat address."""
|
||
from channel import send_message
|
||
content = _strip_trailing_period(content)
|
||
if not content:
|
||
return False
|
||
resolved = _resolve_delivery(user_id, websocket)
|
||
if resolved is None:
|
||
print(f"[DEBUG][chat] 无法解析 UID {user_id} 的私聊投递目标，消息未发送")
|
||
return False
|
||
channel, target_id = resolved
|
||
result = await send_message(channel, "private", target_id, content)
|
||
print(f"[DEBUG][chat] 已发送私聊消息 uid={user_id} target={target_id}: {content[:50]}...")
|
||
return result
|
||
|
||
async def send_message_segments(websocket, user_id: int, segments: List[dict]):
|
||
all_ok = True
|
||
for idx, seg in enumerate(segments):
|
||
content = seg.get("content", "")
|
||
if not content:
|
||
continue
|
||
delay_per_char = random.uniform(TYPE_DELAY_MIN_MS / 1000, TYPE_DELAY_MAX_MS / 1000)
|
||
total_delay = len(content) * delay_per_char
|
||
if total_delay > 2.0:
|
||
total_delay = 2.0
|
||
print(f"[DEBUG][chat] 发送第 {idx+1}/{len(segments)} 条，延迟 {total_delay:.2f}s")
|
||
await asyncio.sleep(total_delay)
|
||
all_ok = bool(await send_private_message(websocket, user_id, content)) and all_ok
|
||
return all_ok
|
||
|
||
def _cleanup_image_infos(image_infos: List[dict]):
|
||
"""TG 临时文件只读一次；库中已复制的对象不在 /tmp，不受影响。"""
|
||
paths = set()
|
||
for info in image_infos or []:
|
||
for key in ("url", "original_path"):
|
||
value = info.get(key)
|
||
if isinstance(value, str) and value.startswith(tempfile.gettempdir() + os.sep):
|
||
paths.add(value)
|
||
for path in paths:
|
||
try:
|
||
os.unlink(path)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
async def _process_pending_messages(user_id: int, delay: float):
|
||
await asyncio.sleep(delay)
|
||
messages = _user_pending_messages.pop(user_id, [])
|
||
image_infos = _user_pending_images.pop(user_id, [])
|
||
_user_pending_tasks.pop(user_id, None)
|
||
try:
|
||
if not messages and not image_infos:
|
||
return
|
||
combined = "\n".join(messages) if messages else ""
|
||
if not combined and not image_infos:
|
||
return
|
||
print(f"[DEBUG][chat] 用户 {user_id} 在 {delay}s 内发送了 {len(messages)} 条消息和 {len(image_infos)} 张图片，合并为: {combined[:100]!r}")
|
||
from channel import get_channel
|
||
ch = get_channel(_user_channel.get(user_id, "")) or _websocket
|
||
await _call_model_and_reply(ch, user_id, combined, is_auto=False, image_infos=image_infos)
|
||
finally:
|
||
_cleanup_image_infos(image_infos)
|
||
|
||
|
||
async def _read_local_image(path: str) -> bytes:
|
||
def read():
|
||
with open(path, "rb") as fh:
|
||
return fh.read()
|
||
return await asyncio.to_thread(read)
|
||
|
||
|
||
async def _image_data_url(info: dict, retry: int = 0) -> Optional[str]:
|
||
"""把图片转换成主模型可直接接收的 data URL，不调用独立视觉模型。"""
|
||
image_url = info.get("url", "")
|
||
file_size = info.get("file_size", "0")
|
||
if retry >= 3:
|
||
return None
|
||
try:
|
||
size_bytes = int(file_size) if file_size else 0
|
||
if size_bytes > 100 * 1024 * 1024:
|
||
print(f"[DEBUG][chat] 图片过大 ({size_bytes} bytes)，跳过")
|
||
return None
|
||
if image_url.startswith(("http://", "https://")):
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get(image_url, timeout=60) as img_resp:
|
||
if img_resp.status != 200:
|
||
print(f"[DEBUG][chat] 下载图片失败, 状态码 {img_resp.status}")
|
||
if retry < 2:
|
||
await asyncio.sleep(2 ** retry)
|
||
return await _image_data_url(info, retry + 1)
|
||
return None
|
||
img_data = await img_resp.read()
|
||
content_type = img_resp.headers.get("Content-Type", "")
|
||
else:
|
||
try:
|
||
img_data = await _read_local_image(image_url)
|
||
except OSError as exc:
|
||
print(f"[DEBUG][chat] 本地图片读取失败: {exc}")
|
||
return None
|
||
import mimetypes
|
||
content_type = mimetypes.guess_type(image_url)[0] or "image/jpeg"
|
||
if len(img_data) > 100 * 1024 * 1024:
|
||
print(f"[DEBUG][chat] 图片实际大小 {len(img_data)} bytes 超过 100MB，跳过")
|
||
return None
|
||
mime = content_type.split(";", 1)[0].strip().lower()
|
||
if mime not in ("image/jpeg", "image/png", "image/gif", "image/webp", "image/bmp"):
|
||
mime = "image/jpeg"
|
||
encoded = base64.b64encode(img_data).decode("ascii")
|
||
return f"data:{mime};base64,{encoded}"
|
||
except Exception as exc:
|
||
print(f"[DEBUG][chat] 图片编码异常: {type(exc).__name__}: {exc}")
|
||
if retry < 2:
|
||
await asyncio.sleep(2 ** retry)
|
||
return await _image_data_url(info, retry + 1)
|
||
return None
|
||
|
||
|
||
async def _build_user_content(user_message: str, image_infos: List[dict] = None):
|
||
"""无图片时保持原字符串；有图片时构造 OpenAI 多模态 content parts。"""
|
||
if not image_infos:
|
||
return user_message
|
||
parts = []
|
||
failed = 0
|
||
for info in image_infos:
|
||
data_url = await _image_data_url(info)
|
||
if data_url:
|
||
parts.append({"type": "image_url", "image_url": {"url": data_url}})
|
||
else:
|
||
failed += 1
|
||
text = user_message or "请理解图片内容，并按用户的要求处理。"
|
||
if failed:
|
||
text += f"\n（另有 {failed} 张图片读取失败。）"
|
||
if not parts:
|
||
return text
|
||
parts.append({"type": "text", "text": text})
|
||
return parts
|
||
|
||
async def handle_private_message(websocket, user_id: int, message_text: str, image_infos: List[dict] = None):
|
||
_user_channel[user_id] = getattr(websocket, "name", "qq")
|
||
if not is_chat_enabled():
|
||
return
|
||
global _websocket
|
||
_websocket = websocket
|
||
_user_last_active[user_id] = get_beijing_timestamp()
|
||
_user_no_reply_count[user_id] = 0
|
||
status = load_user_status(user_id)
|
||
if status["pause"] or status["sleep"]:
|
||
print(f"[DEBUG][chat] 用户 {user_id} 主动说话，解除 pause/sleep 状态")
|
||
update_user_status(user_id, pause=False, sleep=False)
|
||
print(f"[DEBUG][chat] 收到私聊消息: user={user_id}, text={message_text}, images={len(image_infos or [])}")
|
||
now_hour = get_beijing_time().hour
|
||
if SILENT_START <= now_hour < SILENT_END:
|
||
print(f"[DEBUG][chat] 进入夜间模式，回复 '在睡觉'")
|
||
await send_private_message(websocket, user_id, "[自动回复] 在睡觉，不要打扰呢...")
|
||
append_history(user_id, "you", "在睡觉，不要打扰呢...", is_auto=False)
|
||
return
|
||
if user_id not in _user_pending_messages:
|
||
_user_pending_messages[user_id] = []
|
||
_user_pending_messages[user_id].append(message_text)
|
||
if image_infos:
|
||
if user_id not in _user_pending_images:
|
||
_user_pending_images[user_id] = []
|
||
_user_pending_images[user_id].extend(image_infos)
|
||
if user_id in _user_pending_tasks:
|
||
_user_pending_tasks[user_id].cancel()
|
||
wait_seconds = random.uniform(USER_IDLE_WAIT_MIN, USER_IDLE_WAIT_MAX)
|
||
task = asyncio.create_task(_process_pending_messages(user_id, wait_seconds))
|
||
_user_pending_tasks[user_id] = task
|
||
|
||
@contextlib.asynccontextmanager
|
||
async def _noop_ctx():
|
||
yield
|
||
|
||
|
||
def _turn_lock(mapping: Dict[int, asyncio.Lock], key: int) -> asyncio.Lock:
|
||
if key not in mapping:
|
||
mapping[key] = asyncio.Lock()
|
||
return mapping[key]
|
||
|
||
|
||
async def _call_model_and_reply(websocket, user_id: int, user_message: str, is_auto: bool,
|
||
image_infos: List[dict] = None):
|
||
"""对外入口: 套「正在输入」提示，实际逻辑在 _impl"""
|
||
resolved = _resolve_delivery(user_id, websocket)
|
||
if resolved is not None:
|
||
channel, target_id = resolved
|
||
ctx = channel.typing(target_id) if hasattr(channel, "typing") else _noop_ctx()
|
||
else:
|
||
ctx = _noop_ctx()
|
||
import notifications
|
||
async with _turn_lock(_private_turn_locks, int(user_id)):
|
||
notifications.center.begin_conversation(user_id)
|
||
try:
|
||
async with ctx:
|
||
with skill_runtime.skill_session("proactive" if is_auto else "private_chat"):
|
||
await _call_model_and_reply_impl(websocket, user_id, user_message, is_auto, image_infos=image_infos)
|
||
finally:
|
||
notifications.center.end_conversation(user_id)
|
||
try:
|
||
# Events that arrived while the model was replying follow the final
|
||
# response instead of interrupting it mid-turn.
|
||
await notifications.center.deliver_due()
|
||
except Exception as exc:
|
||
print(f"[notifications] post-chat flush failed: {type(exc).__name__}: {exc}")
|
||
|
||
|
||
async def _call_model_and_reply_impl(websocket, user_id: int, user_message: str, is_auto: bool,
|
||
image_infos: List[dict] = None):
|
||
print(f"[DEBUG][chat] _call_model_and_reply: user={user_id}, is_auto={is_auto}, msg={user_message[:50]}")
|
||
if is_auto:
|
||
status = load_user_status(user_id)
|
||
if status["pause"] or status["sleep"]:
|
||
print(f"[DEBUG][chat] 用户 {user_id} 状态 pause={status['pause']}, sleep={status['sleep']}，跳过主动推送")
|
||
return
|
||
if not is_auto:
|
||
history_text = user_message
|
||
if image_infos:
|
||
marker = f"[用户发送了 {len(image_infos)} 张图片]"
|
||
history_text = marker + ("\n" + user_message if user_message else "")
|
||
append_history(user_id, "user", history_text, is_auto=False)
|
||
from agent import ToolContext
|
||
resolved = _resolve_delivery(user_id, websocket)
|
||
tool_context = None
|
||
if resolved is not None and not is_auto:
|
||
tool_channel, tool_target = resolved
|
||
tool_context = ToolContext(
|
||
channel=tool_channel, msg_type="private", user_id=user_id,
|
||
target_id=tool_target, is_admin=bool(tool_channel.is_admin(user_id)))
|
||
model_user_message = _with_sticker_sources(user_message, image_infos)
|
||
system_prompt = build_system_prompt(user_id, is_auto=is_auto, query=model_user_message,
|
||
tool_context=tool_context)
|
||
effective_user_msg = _message_with_time(model_user_message if not is_auto else "请主动发起对话。")
|
||
raw_response = None
|
||
for retry in range(MAX_RETRIES):
|
||
raw_response = await call_deepseek(
|
||
system_prompt, effective_user_msg, retry,
|
||
presence_penalty=AUTO_PRESENCE_PENALTY if is_auto else 0.0,
|
||
image_infos=image_infos)
|
||
if raw_response:
|
||
break
|
||
if 400 <= LAST_HTTP_STATUS.get() < 500:
|
||
break
|
||
await asyncio.sleep(1)
|
||
if not raw_response:
|
||
status_code = LAST_HTTP_STATUS.get()
|
||
fallback = _fallback_by_status(status_code)
|
||
if status_code == 402:
|
||
# 余额不足按「每个用户每天」最多提示一次：既不风暴，也不会永久沉默。
|
||
# 主动 worker 每 10 分钟醒来时不再反复调用模型/发送同一句；
|
||
# 余额恢复后由成功响应或余额探测复位。
|
||
today = get_beijing_time().strftime("%Y-%m-%d")
|
||
status = load_user_status(user_id)
|
||
if status.get("balance_notice_date") != today:
|
||
await send_private_message(websocket, user_id, fallback)
|
||
if not is_auto:
|
||
append_history(user_id, "you", fallback, is_auto=False)
|
||
update_user_status(user_id, balance_notice_sent=True, balance_notice_date=today)
|
||
print(f"[DEBUG][chat] 用户 {user_id} 今日首次触发余额不足，已提示（按日抑制）")
|
||
else:
|
||
print(f"[DEBUG][chat] 用户 {user_id} 今日已提示过余额不足，本次静默")
|
||
else:
|
||
await send_private_message(websocket, user_id, fallback)
|
||
if not is_auto:
|
||
append_history(user_id, "you", fallback, is_auto=False)
|
||
return
|
||
|
||
# API 已恢复并成功返回：允许未来新的一天再提示。
|
||
status = load_user_status(user_id)
|
||
if status.get("balance_notice_sent", False):
|
||
update_user_status(user_id, balance_notice_sent=False, balance_notice_date="")
|
||
print(f"[DEBUG][chat] 用户 {user_id} API 已恢复，余额不足提示状态已复位")
|
||
parsed = None
|
||
for attempt in range(MAX_RETRIES):
|
||
parsed = parse_json_response(raw_response)
|
||
if parsed:
|
||
break
|
||
print(f"[DEBUG][chat] JSON 解析失败(第{attempt+1}次), 模型原始返回: {raw_response[:300]!r}")
|
||
if attempt >= MAX_RETRIES - 1:
|
||
break
|
||
use_reasoning = _json_retry_uses_reasoning(attempt)
|
||
raw_response = await call_deepseek(
|
||
system_prompt, effective_user_msg, 0,
|
||
presence_penalty=AUTO_PRESENCE_PENALTY if is_auto else 0.0,
|
||
image_infos=image_infos, reasoning=use_reasoning)
|
||
if not raw_response:
|
||
break
|
||
if not parsed:
|
||
print(f"[DEBUG][chat] 放弃: 多次解析失败, 最终原始返回: {raw_response[:300]!r}")
|
||
fallback = "抱歉，我有点迷糊，你能再说一遍吗？"
|
||
await send_private_message(websocket, user_id, fallback)
|
||
if not is_auto:
|
||
append_history(user_id, "you", fallback, is_auto=False)
|
||
return
|
||
# 模型自评需要深思，或出现任何工具调用时，统一用 reasoning 重做一次。
|
||
if _chat_requires_reasoning(parsed, tool_context, is_auto):
|
||
reasoning_parsed = await _reasoning_json_response(system_prompt, effective_user_msg, image_infos)
|
||
if reasoning_parsed:
|
||
parsed = reasoning_parsed
|
||
elif isinstance(parsed.get("tools"), list) and parsed["tools"]:
|
||
parsed = {
|
||
"message": [{"currentTurn": "1", "totalTurn": "1",
|
||
"content": "工具规划暂时失败了，请稍后再试。"}],
|
||
"tools": [],
|
||
}
|
||
_learn_stickers(user_id, image_infos, parsed)
|
||
# 多轮 Agent：utility 工具、web_search/fetch、产物和邮件附件可组合调用。
|
||
if tool_context is not None:
|
||
import agent_loop
|
||
parsed = await agent_loop.run_agent_loop(
|
||
sys.modules[__name__], parsed, system_prompt, effective_user_msg,
|
||
tool_context.channel, "private", user_id, tool_context.target_id, is_auto,
|
||
image_infos=image_infos)
|
||
|
||
segments = parsed.get("message", [])
|
||
if segments:
|
||
sent_ok = await send_message_segments(websocket, user_id, segments)
|
||
full_reply = " ".join([seg["content"] for seg in segments])
|
||
if not sent_ok:
|
||
print(f"[DEBUG][chat] 私聊回复投递失败 uid={user_id}，不写入正常回复历史")
|
||
return
|
||
if not is_auto:
|
||
append_history(user_id, "you", full_reply, is_auto=False)
|
||
else:
|
||
append_history(user_id, "you", full_reply, is_auto=True)
|
||
sticker_scene = parsed.get("sticker")
|
||
if isinstance(sticker_scene, str) and sticker_scene.strip().lower() not in ("", "none"):
|
||
await _maybe_send_sticker(
|
||
websocket, user_id, sticker_scene.strip(),
|
||
context=effective_user_msg + "\n白花的回复：" + full_reply)
|
||
else:
|
||
print(f"[DEBUG][chat] 警告: message 字段为空")
|
||
memory = parsed.get("memory", "none")
|
||
_record_memory(user_id, memory)
|
||
try:
|
||
memory_store.touch_relation(user_id)
|
||
relation_note = parsed.get("relation")
|
||
if isinstance(relation_note, str) and relation_note.strip() and relation_note.strip() != "none":
|
||
memory_store.add_relation_note(user_id, relation_note.strip())
|
||
links = parsed.get("links")
|
||
if isinstance(links, list) and links:
|
||
added = memory_store.add_links(user_id, links)
|
||
if added:
|
||
print(f"[DEBUG][chat] 三元组入库 {len(added)} 条 (uid={user_id})")
|
||
except Exception as e:
|
||
print(f"[DEBUG][chat] 关系卡更新失败: {e}")
|
||
new_pause = parsed.get("pause")
|
||
new_sleep = parsed.get("sleep")
|
||
if isinstance(new_pause, bool) or isinstance(new_sleep, bool):
|
||
update_user_status(user_id, pause=new_pause if isinstance(new_pause, bool) else None,
|
||
sleep=new_sleep if isinstance(new_sleep, bool) else None)
|
||
print(f"[DEBUG][chat] 更新用户 {user_id} 状态: pause={new_pause}, sleep={new_sleep}")
|
||
if is_auto:
|
||
_user_last_auto[user_id] = get_beijing_timestamp()
|
||
|
||
async def _balance_allows_active(qq: int, status: dict) -> bool:
|
||
"""余额不足已提示时，禁止主动对话风暴；仅轻量探测余额恢复。"""
|
||
if not status.get("balance_notice_sent", False):
|
||
return True
|
||
balance = await get_wallet_balance()
|
||
if balance is not None and balance > 0:
|
||
update_user_status(qq, balance_notice_sent=False, balance_notice_date="")
|
||
status["balance_notice_sent"] = False
|
||
status["balance_notice_date"] = ""
|
||
print(f"[主动] 用户 {qq} 余额已恢复（{balance:.2f} 元），解除余额不足抑制")
|
||
return True
|
||
print(f"[主动] 用户 {qq} 仍余额不足/查询失败，跳过本轮主动对话")
|
||
return False
|
||
|
||
|
||
async def active_push_worker():
|
||
global _websocket
|
||
while True:
|
||
await asyncio.sleep(600)
|
||
if not is_chat_enabled():
|
||
continue
|
||
now_ts = get_beijing_timestamp()
|
||
now = get_beijing_time()
|
||
now_hour = now.hour
|
||
if SILENT_START <= now_hour < SILENT_END:
|
||
continue
|
||
if not os.path.exists(HISTORY_DIR):
|
||
continue
|
||
for filename in os.listdir(HISTORY_DIR):
|
||
if not filename.endswith(".txt"):
|
||
continue
|
||
try:
|
||
qq = int(filename.split('.')[0])
|
||
except:
|
||
continue
|
||
status = load_user_status(qq)
|
||
if status.get("sleep", False):
|
||
today_str = now.strftime("%Y-%m-%d")
|
||
wake_at = _user_sleep_wake_at.get(qq)
|
||
if wake_at is None and now_hour >= 9 and _user_sleep_unscheduled_date.get(qq) != today_str:
|
||
delay = random.randint(0, 3600)
|
||
_user_sleep_wake_at[qq] = now_ts + delay
|
||
_user_sleep_unscheduled_date[qq] = today_str
|
||
print(f"[主动] 用户 {qq} 处于睡眠状态，将在 {delay} 秒后解除 sleep")
|
||
elif wake_at is not None and now_ts >= wake_at:
|
||
if load_user_status(qq).get("sleep", False):
|
||
update_user_status(qq, sleep=False)
|
||
print(f"[主动] 用户 {qq} 的 sleep 状态已自动解除（早上随机时刻）")
|
||
_user_sleep_wake_at.pop(qq, None)
|
||
continue
|
||
if not await _balance_allows_active(qq, status):
|
||
continue
|
||
last_auto = _user_last_auto.get(qq, 0)
|
||
counted_auto = _user_auto_counted.get(qq, 0)
|
||
if last_auto > counted_auto:
|
||
last_active = _user_last_active.get(qq, 0)
|
||
if last_active < last_auto:
|
||
_user_no_reply_count[qq] = _user_no_reply_count.get(qq, 0) + 1
|
||
print(f"[主动] 用户 {qq} 连续未回复计数: {_user_no_reply_count[qq]}")
|
||
if _user_no_reply_count[qq] >= 2:
|
||
update_user_status(qq, pause=True, sleep=None)
|
||
print(f"[主动] 用户 {qq} 两次未回复主动消息，已自动将 pause 设为 true")
|
||
_user_no_reply_count[qq] = 0
|
||
_user_auto_counted[qq] = last_auto
|
||
continue
|
||
else:
|
||
_user_no_reply_count[qq] = 0
|
||
_user_auto_counted[qq] = last_auto
|
||
last_active = get_last_active(qq)
|
||
if last_active == 0:
|
||
continue
|
||
last_auto = _user_last_auto.get(qq, 0)
|
||
interval = random.uniform(ACTIVE_INTERVAL_MIN, ACTIVE_INTERVAL_MAX)
|
||
if (now_ts - last_auto > interval) and (now_ts - last_active > interval):
|
||
current_status = load_user_status(qq)
|
||
if current_status.get("pause", False) or current_status.get("sleep", False):
|
||
continue
|
||
print(f"[主动] 向用户 {qq} 发起主动对话")
|
||
resolved = _resolve_delivery(qq)
|
||
if resolved is None:
|
||
print(f"[主动] UID {qq} 暂无可用投递通道，跳过")
|
||
continue
|
||
ch, target_id = resolved
|
||
from agent.modes import ModeStore
|
||
if ModeStore().get_mode(getattr(ch, "name", "tg"), qq) != "chat":
|
||
continue
|
||
await _call_model_and_reply(ch, qq, "触发主动对话", is_auto=True)
|
||
|
||
def start_active_push():
|
||
global _active_task
|
||
if _active_task is None or _active_task.done():
|
||
_active_task = asyncio.create_task(active_push_worker())
|
||
|
||
def init_chat():
|
||
ensure_dirs()
|
||
try:
|
||
migrated = STICKER_STORE.migrate_all_to_global(STICKER_GLOBAL_SCOPE)
|
||
if migrated:
|
||
print(f"[sticker] 已合并 {migrated} 条分会话贴纸记录到全局库")
|
||
except Exception as exc:
|
||
print(f"[sticker] 旧库存合并失败: {type(exc).__name__}: {exc}")
|
||
start_active_push()
|
||
print("[DEBUG][chat] 初始化完成，后台主动推送已启动（使用北京时间）")
|
||
|
||
def set_websocket(ws):
|
||
global _websocket
|
||
_websocket = ws
|
||
print("[DEBUG][chat] WebSocket 引用已设置")
|
||
|
||
# ---------- 群聊功能 ----------
|
||
def _get_group_history_path(group_id: int) -> str:
|
||
return os.path.join(GROUP_HISTORY_DIR, f"{group_id}.txt")
|
||
|
||
def append_group_history(group_id: int, user_id: int, content: str):
|
||
timestamp = get_beijing_time().strftime("%Y-%m-%d %H:%M:%S")
|
||
line = f"[{timestamp}][qq={user_id}] {content}"
|
||
with _group_lock(group_id):
|
||
os.makedirs(GROUP_HISTORY_DIR, exist_ok=True)
|
||
with open(_get_group_history_path(group_id), "a", encoding="utf-8") as f:
|
||
f.write(line + "\n")
|
||
f.flush()
|
||
os.fsync(f.fileno())
|
||
|
||
def get_recent_group_history(group_id: int, limit: int = GROUP_MAX_HISTORY) -> List[str]:
|
||
path = _get_group_history_path(group_id)
|
||
if not os.path.exists(path):
|
||
return []
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
lines = f.readlines()
|
||
return [line.strip() for line in lines[-limit:]]
|
||
|
||
async def send_group_message(websocket, group_id: int, content: str):
|
||
"""websocket 为 Channel 对象（通道抽象）"""
|
||
from channel import send_message
|
||
content = _strip_trailing_period(content)
|
||
if not content:
|
||
return False
|
||
result = await send_message(websocket, "group", group_id, content)
|
||
print(f"[DEBUG][chat][group] 已发送群聊消息给 {group_id}: {content[:50]}...")
|
||
return result
|
||
|
||
async def handle_chat_command(websocket, msg_type, target_id, raw_text):
|
||
"""管理员指令 .chat <on|off>，控制聊天模块开关。"""
|
||
parts = (raw_text or "").strip().split()
|
||
sub = parts[1].lower() if len(parts) >= 2 else ""
|
||
if sub not in ("on", "off"):
|
||
msg = "用法: .chat <on|off>"
|
||
else:
|
||
enabled = sub == "on"
|
||
set_chat_enabled(enabled)
|
||
msg = "聊天模块已开启" if enabled else "聊天模块已关闭"
|
||
if msg_type == "group":
|
||
await send_group_message(websocket, target_id, msg)
|
||
else:
|
||
await send_private_message(websocket, target_id, msg)
|
||
|
||
async def handle_group_message(websocket, group_id: int, user_id: int, message_text: str,
|
||
image_infos: List[dict] = None, is_mention: bool = False):
|
||
_user_channel[user_id] = getattr(websocket, "name", "qq")
|
||
if not is_chat_enabled():
|
||
return
|
||
if not GROUP_CHAT_ENABLED:
|
||
return
|
||
now = get_beijing_timestamp()
|
||
if group_id not in _group_msg_timestamps:
|
||
_group_msg_timestamps[group_id] = []
|
||
_group_msg_timestamps[group_id].append(now)
|
||
cutoff = now - GROUP_TRIGGER_WINDOW
|
||
_group_msg_timestamps[group_id] = [t for t in _group_msg_timestamps[group_id] if t > cutoff]
|
||
|
||
if message_text:
|
||
append_group_history(group_id, user_id, message_text)
|
||
|
||
if group_id not in _group_pending_messages:
|
||
_group_pending_messages[group_id] = []
|
||
_group_pending_messages[group_id].append(message_text)
|
||
if group_id not in _group_pending_users:
|
||
_group_pending_users[group_id] = []
|
||
_group_pending_users[group_id].append(user_id)
|
||
|
||
if image_infos:
|
||
if group_id not in _group_pending_images:
|
||
_group_pending_images[group_id] = []
|
||
_group_pending_images[group_id].extend(image_infos)
|
||
for info in image_infos:
|
||
url = info.get("url", "")
|
||
if url:
|
||
append_group_history(group_id, user_id, f"[图片] {url}")
|
||
|
||
if is_mention:
|
||
_group_triggered_by_mention[group_id] = True
|
||
|
||
_group_msg_counter[group_id] = _group_msg_counter.get(group_id, 0) + 1
|
||
current_cnt = _group_msg_counter[group_id]
|
||
|
||
should_trigger = is_mention
|
||
if GROUP_AUTO_TRIGGER and not should_trigger:
|
||
msg_count = len(_group_msg_timestamps[group_id])
|
||
if msg_count > GROUP_TRIGGER_THRESHOLD and random.random() < GROUP_TRIGGER_PROBABILITY:
|
||
should_trigger = True
|
||
|
||
continue_trigger = False
|
||
if GROUP_AUTO_TRIGGER and not should_trigger and group_id in _group_bot_last_speak_at:
|
||
since_bot = current_cnt - _group_bot_last_speak_at[group_id]
|
||
need = _group_continue_after.get(group_id, 5)
|
||
if since_bot >= need and random.random() < GROUP_CONTINUE_PROBABILITY:
|
||
should_trigger = True
|
||
continue_trigger = True
|
||
|
||
if should_trigger:
|
||
if group_id in _group_pending_tasks:
|
||
_group_pending_tasks[group_id].cancel()
|
||
task = asyncio.create_task(_process_group_messages(group_id, is_continue=continue_trigger, channel=websocket))
|
||
_group_pending_tasks[group_id] = task
|
||
|
||
async def _process_group_messages(group_id: int, is_continue: bool = False, channel=None):
|
||
await asyncio.sleep(random.uniform(1, 3))
|
||
messages = _group_pending_messages.pop(group_id, [])
|
||
pending_users = _group_pending_users.pop(group_id, [])
|
||
actor_uid = pending_users[-1] if pending_users and len(set(pending_users)) == 1 else 0
|
||
image_infos = _group_pending_images.pop(group_id, [])
|
||
_group_pending_tasks.pop(group_id, None)
|
||
is_mention = _group_triggered_by_mention.pop(group_id, False)
|
||
try:
|
||
if not messages and not image_infos:
|
||
return
|
||
recent_images = image_infos[-GROUP_MAX_IMAGES:] if image_infos else []
|
||
combined = "\n".join([m for m in messages if m]) if messages else ""
|
||
if not combined and not recent_images:
|
||
return
|
||
print(f"[DEBUG][chat][group] 群 {group_id} 触发, mention={is_mention}, msg={combined[:100]!r}, images={len(recent_images)}")
|
||
await _call_group_model_and_reply(channel or _websocket, group_id, combined, actor_uid,
|
||
image_infos=recent_images)
|
||
finally:
|
||
_cleanup_image_infos(image_infos)
|
||
|
||
|
||
async def _call_group_model_and_reply(websocket, group_id: int, user_message: str, user_id: int = 0,
|
||
image_infos: List[dict] = None):
|
||
"""对外入口: 套「正在输入」提示，实际逻辑在 _impl"""
|
||
ctx = websocket.typing(group_id) if websocket is not None else _noop_ctx()
|
||
async with _turn_lock(_group_turn_locks, int(group_id)):
|
||
async with ctx:
|
||
with skill_runtime.skill_session("group_chat"):
|
||
await _call_group_model_and_reply_impl(websocket, group_id, user_message, user_id,
|
||
image_infos=image_infos)
|
||
|
||
|
||
async def _call_group_model_and_reply_impl(websocket, group_id: int, user_message: str, user_id: int = 0,
|
||
image_infos: List[dict] = None):
|
||
print(f"[DEBUG][chat][group] _call_group_model_and_reply: group={group_id}")
|
||
|
||
history_lines = get_recent_group_history(group_id, GROUP_MAX_HISTORY)
|
||
history_str = "\n".join(history_lines) if history_lines else "（无最近对话）"
|
||
|
||
from agent import ToolContext
|
||
import agent_loop
|
||
tool_context = None
|
||
tool_prompt = ""
|
||
if websocket is not None and user_id:
|
||
tool_context = ToolContext(
|
||
channel=websocket, msg_type="group", user_id=user_id, target_id=group_id,
|
||
is_admin=bool(websocket.is_admin(user_id)))
|
||
tool_prompt = "\n\n" + agent_loop.tool_instructions(tool_context)
|
||
model_user_message = _message_with_time(_with_sticker_sources(user_message, image_infos))
|
||
sticker_summary = _sticker_prompt_summary(group_id)
|
||
if sticker_summary:
|
||
tool_prompt += "\n【可使用的表情包场景】\n" + sticker_summary
|
||
system_prompt = (BASE_PROMPT + "\n\n" + load_instruction("group_instructions.txt")
|
||
+ "\n【表情包场景分类定义】\n" + sticker_store.scene_prompt()
|
||
+ tool_prompt
|
||
+ "\n\n以下是本群的最近聊天记录：\n" + history_str)
|
||
|
||
raw_response = None
|
||
for retry in range(MAX_RETRIES):
|
||
raw_response = await call_deepseek(system_prompt, model_user_message, retry, image_infos=image_infos)
|
||
if raw_response:
|
||
break
|
||
if 400 <= LAST_HTTP_STATUS.get() < 500:
|
||
break
|
||
await asyncio.sleep(1)
|
||
|
||
if not raw_response:
|
||
print(f"[DEBUG][chat][group] API 无响应")
|
||
return
|
||
|
||
parsed = None
|
||
for attempt in range(MAX_RETRIES):
|
||
parsed = parse_json_response(raw_response)
|
||
if parsed:
|
||
break
|
||
if attempt >= MAX_RETRIES - 1:
|
||
break
|
||
raw_response = await call_deepseek(
|
||
system_prompt, model_user_message, 0,
|
||
image_infos=image_infos,
|
||
reasoning=_json_retry_uses_reasoning(attempt))
|
||
if not raw_response:
|
||
break
|
||
|
||
if not parsed:
|
||
print(f"[DEBUG][chat][group] JSON 解析失败")
|
||
return
|
||
|
||
if _chat_requires_reasoning(parsed, tool_context, False):
|
||
reasoning_parsed = await _reasoning_json_response(system_prompt, model_user_message, image_infos)
|
||
if reasoning_parsed:
|
||
parsed = reasoning_parsed
|
||
elif isinstance(parsed.get("tools"), list) and parsed["tools"]:
|
||
parsed = {
|
||
"message": [{"currentTurn": "1", "totalTurn": "1",
|
||
"content": "工具规划暂时失败了，请稍后再试。"}],
|
||
"tools": [],
|
||
}
|
||
_learn_stickers(group_id, image_infos, parsed)
|
||
if tool_context is not None:
|
||
parsed = await agent_loop.run_agent_loop(
|
||
sys.modules[__name__], parsed, system_prompt, model_user_message,
|
||
websocket, "group", user_id, group_id, False, image_infos=image_infos)
|
||
|
||
segments = parsed.get("message", [])
|
||
if segments:
|
||
full_reply = "".join([seg["content"] for seg in segments])
|
||
sent_ok = await send_group_message(websocket, group_id, full_reply)
|
||
if not sent_ok:
|
||
print(f"[DEBUG][chat][group] 群 {group_id} 回复投递失败，不写入正常回复历史")
|
||
return
|
||
append_group_history(group_id, 0, full_reply)
|
||
sticker_scene = parsed.get("sticker")
|
||
if isinstance(sticker_scene, str) and sticker_scene.strip().lower() not in ("", "none"):
|
||
await _maybe_send_sticker(
|
||
websocket, user_id, sticker_scene.strip(),
|
||
context=model_user_message + "\n白花的回复：" + full_reply,
|
||
msg_type="group", target_id=group_id)
|
||
_group_bot_last_speak_at[group_id] = _group_msg_counter.get(group_id, 0)
|
||
_group_continue_after[group_id] = random.randint(3, 5)
|