diff --git a/.gitignore b/.gitignore index d257c3689..b9e101e40 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,7 @@ queue_update.txt memory_graph.gml .env .env.* +.cursor config/bot_config_dev.toml config/bot_config.toml config/bot_config.toml.bak diff --git a/Dockerfile b/Dockerfile index 6c6041ff3..fe96ac033 100644 --- a/Dockerfile +++ b/Dockerfile @@ -17,4 +17,5 @@ RUN pip install --upgrade -r requirements.txt COPY . . EXPOSE 8000 + ENTRYPOINT [ "python","bot.py" ] \ No newline at end of file diff --git a/changelogs/changelog_dev.md b/changelogs/changelog_dev.md index c88422815..ab211c4b9 100644 --- a/changelogs/changelog_dev.md +++ b/changelogs/changelog_dev.md @@ -1,5 +1,9 @@ 这里放置了测试版本的细节更新 +## [test-0.6.0-snapshot-7] - 2025-4-2 +- 修改版本号命名：test-前缀为测试版，无前缀为正式版 +- 提供私聊的PFC模式 + ## [0.6.0-mmc-4] - 2025-4-1 - 提供两种聊天逻辑，思维流聊天（ThinkFlowChat 和 推理聊天（ReasoningChat） - 从结构上可支持多种回复消息逻辑 \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index 7b4fcd2d3..367d28cdd 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,14 +1,15 @@ services: adapters: container_name: maim-bot-adapters - image: sengokucola/maimbot-adapter:latest + image: maple127667/maimbot-adapter:latest + # image: infinitycat/maimbot-adapter:latest environment: - TZ=Asia/Shanghai ports: - "18002:18002" volumes: - - ./adapters/plugins:/adapters/src/plugins # 持久化adapters插件 - - ./adapters/.env:/adapters/.env # 持久化adapters配置文件 + - ./docker-config/adapters/plugins:/adapters/src/plugins # 持久化adapters + - ./docker-config/adapters/.env:/adapters/.env # 持久化adapters配置文件 - ./data/qq:/app/.config/QQ # 持久化QQ本体并同步qq表情和图片到adapters restart: always depends_on: @@ -18,6 +19,7 @@ services: core: container_name: maim-bot-core image: sengokucola/maimbot:refactor + # image: infinitycat/maimbot:refactor environment: - TZ=Asia/Shanghai # - EULA_AGREE=35362b6ea30f12891d46ef545122e84a # 同意EULA @@ -25,9 +27,8 @@ services: ports: - "8000:8000" volumes: - - ./mmc-data:/MaiMBot/data - - ./mmc-config/.env:/MaiMBot/.env # 持久化bot配置文件 - - ./mmc-config/bot_config.toml:/MaiMBot/config/bot_config.toml # 持久化bot配置文件 + - ./docker-config/mmc/.env:/MaiMBot/.env # 持久化env配置文件 + - ./docker-config/mmc:/MaiMBot/config # 持久化bot配置文件 - ./data/MaiMBot:/MaiMBot/data # NapCat 和 NoneBot 共享此卷，否则发送图片会有问题 restart: always depends_on: @@ -58,7 +59,7 @@ services: - "6099:6099" - "8095:8095" volumes: - - ./napcat-config:/app/napcat/config # 持久化napcat配置文件 + - ./docker-config/napcat:/app/napcat/config # 持久化napcat配置文件 - ./data/qq:/app/.config/QQ # 持久化QQ本体并同步qq表情和图片到adapters - ./data/MaiMBot:/MaiMBot/data # NapCat 和 NoneBot 共享此卷，否则发送图片会有问题 container_name: maim-bot-napcat diff --git a/scripts/run.sh b/scripts/run.sh new file mode 100644 index 000000000..1f7fba1ce --- /dev/null +++ b/scripts/run.sh @@ -0,0 +1,613 @@ +#!/bin/bash + +# MaiCore & Nonebot adapter一键安装脚本 by Cookie_987 +# 适用于Arch/Ubuntu 24.10/Debian 12/CentOS 9 +# 请小心使用任何一键脚本！ + +INSTALLER_VERSION="0.0.1-refactor" +LANG=C.UTF-8 + +# 如无法访问GitHub请修改此处镜像地址 +GITHUB_REPO="https://ghfast.top/https://github.com" + +# 颜色输出 +GREEN="\e[32m" +RED="\e[31m" +RESET="\e[0m" + +# 需要的基本软件包 + +declare -A REQUIRED_PACKAGES=( + ["common"]="git sudo python3 curl gnupg" + ["debian"]="python3-venv python3-pip" + ["ubuntu"]="python3-venv python3-pip" + ["centos"]="python3-pip" + ["arch"]="python-virtualenv python-pip" +) + +# 默认项目目录 +DEFAULT_INSTALL_DIR="/opt/maicore" + +# 服务名称 +SERVICE_NAME="maicore" +SERVICE_NAME_WEB="maicore-web" +SERVICE_NAME_NBADAPTER="maicore-nonebot-adapter" + +IS_INSTALL_MONGODB=false +IS_INSTALL_NAPCAT=false +IS_INSTALL_DEPENDENCIES=false + +# 检查是否已安装 +check_installed() { + [[ -f /etc/systemd/system/${SERVICE_NAME}.service ]] +} + +# 加载安装信息 +load_install_info() { + if [[ -f /etc/maicore_install.conf ]]; then + source /etc/maicore_install.conf + else + INSTALL_DIR="$DEFAULT_INSTALL_DIR" + BRANCH="refactor" + fi +} + +# 显示管理菜单 +show_menu() { + while true; do + choice=$(whiptail --title "MaiCore管理菜单" --menu "请选择要执行的操作：" 15 60 7 \ + "1" "启动MaiCore" \ + "2" "停止MaiCore" \ + "3" "重启MaiCore" \ + "4" "启动Nonebot adapter" \ + "5" "停止Nonebot adapter" \ + "6" "重启Nonebot adapter" \ + "7" "更新MaiCore及其依赖" \ + "8" "切换分支" \ + "9" "退出" 3>&1 1>&2 2>&3) + + [[ $? -ne 0 ]] && exit 0 + + case "$choice" in + 1) + systemctl start ${SERVICE_NAME} + whiptail --msgbox "✅MaiCore已启动" 10 60 + ;; + 2) + systemctl stop ${SERVICE_NAME} + whiptail --msgbox "🛑MaiCore已停止" 10 60 + ;; + 3) + systemctl restart ${SERVICE_NAME} + whiptail --msgbox "🔄MaiCore已重启" 10 60 + ;; + 4) + systemctl start ${SERVICE_NAME_NBADAPTER} + whiptail --msgbox "✅Nonebot adapter已启动" 10 60 + ;; + 5) + systemctl stop ${SERVICE_NAME_NBADAPTER} + whiptail --msgbox "🛑Nonebot adapter已停止" 10 60 + ;; + 6) + systemctl restart ${SERVICE_NAME_NBADAPTER} + whiptail --msgbox "🔄Nonebot adapter已重启" 10 60 + ;; + 7) + update_dependencies + ;; + 8) + switch_branch + ;; + 9) + exit 0 + ;; + *) + whiptail --msgbox "无效选项！" 10 60 + ;; + esac + done +} + +# 更新依赖 +update_dependencies() { + cd "${INSTALL_DIR}/MaiBot" || { + whiptail --msgbox "🚫 无法进入安装目录！" 10 60 + return 1 + } + if ! git pull origin "${BRANCH}"; then + whiptail --msgbox "🚫 代码更新失败！" 10 60 + return 1 + fi + source "${INSTALL_DIR}/venv/bin/activate" + if ! pip install -r requirements.txt; then + whiptail --msgbox "🚫 依赖安装失败！" 10 60 + deactivate + return 1 + fi + deactivate + systemctl restart ${SERVICE_NAME} + whiptail --msgbox "✅ 依赖已更新并重启服务！" 10 60 +} + +# 切换分支 +switch_branch() { + new_branch=$(whiptail --inputbox "请输入要切换的分支名称：" 10 60 "${BRANCH}" 3>&1 1>&2 2>&3) + [[ -z "$new_branch" ]] && { + whiptail --msgbox "🚫 分支名称不能为空！" 10 60 + return 1 + } + + cd "${INSTALL_DIR}/MaiBot" || { + whiptail --msgbox "🚫 无法进入安装目录！" 10 60 + return 1 + } + + if ! git ls-remote --exit-code --heads origin "${new_branch}" >/dev/null 2>&1; then + whiptail --msgbox "🚫 分支 ${new_branch} 不存在！" 10 60 + return 1 + fi + + if ! git checkout "${new_branch}"; then + whiptail --msgbox "🚫 分支切换失败！" 10 60 + return 1 + fi + + if ! git pull origin "${new_branch}"; then + whiptail --msgbox "🚫 代码拉取失败！" 10 60 + return 1 + fi + + source "${INSTALL_DIR}/venv/bin/activate" + pip install -r requirements.txt + deactivate + + sed -i "s/^BRANCH=.*/BRANCH=${new_branch}/" /etc/maicore_install.conf + BRANCH="${new_branch}" + check_eula + systemctl restart ${SERVICE_NAME} + whiptail --msgbox "✅ 已切换到分支 ${new_branch} 并重启服务！" 10 60 +} + +check_eula() { + # 首先计算当前EULA的MD5值 + current_md5=$(md5sum "${INSTALL_DIR}/MaiBot/EULA.md" | awk '{print $1}') + + # 首先计算当前隐私条款文件的哈希值 + current_md5_privacy=$(md5sum "${INSTALL_DIR}/MaiBot/PRIVACY.md" | awk '{print $1}') + + # 如果当前的md5值为空，则直接返回 + if [[ -z $current_md5 || -z $current_md5_privacy ]]; then + whiptail --msgbox "🚫 未找到使用协议\n 请检查PRIVACY.md和EULA.md是否存在" 10 60 + fi + + # 检查eula.confirmed文件是否存在 + if [[ -f ${INSTALL_DIR}/MaiBot/eula.confirmed ]]; then + # 如果存在则检查其中包含的md5与current_md5是否一致 + confirmed_md5=$(cat ${INSTALL_DIR}/MaiBot/eula.confirmed) + else + confirmed_md5="" + fi + + # 检查privacy.confirmed文件是否存在 + if [[ -f ${INSTALL_DIR}/MaiBot/privacy.confirmed ]]; then + # 如果存在则检查其中包含的md5与current_md5是否一致 + confirmed_md5_privacy=$(cat ${INSTALL_DIR}/MaiBot/privacy.confirmed) + else + confirmed_md5_privacy="" + fi + + # 如果EULA或隐私条款有更新，提示用户重新确认 + if [[ $current_md5 != $confirmed_md5 || $current_md5_privacy != $confirmed_md5_privacy ]]; then + whiptail --title "📜 使用协议更新" --yesno "检测到MaiCore EULA或隐私条款已更新。\nhttps://github.com/MaiM-with-u/MaiBot/blob/refactor/EULA.md\nhttps://github.com/MaiM-with-u/MaiBot/blob/refactor/PRIVACY.md\n\n您是否同意上述协议？ \n\n " 12 70 + if [[ $? -eq 0 ]]; then + echo -n $current_md5 > ${INSTALL_DIR}/MaiBot/eula.confirmed + echo -n $current_md5_privacy > ${INSTALL_DIR}/MaiBot/privacy.confirmed + else + exit 1 + fi + fi + +} + +# ----------- 主安装流程 ----------- +run_installation() { + # 1/6: 检测是否安装 whiptail + if ! command -v whiptail &>/dev/null; then + echo -e "${RED}[1/6] whiptail 未安装，正在安装...${RESET}" + + if command -v apt-get &>/dev/null; then + apt-get update && apt-get install -y whiptail + elif command -v pacman &>/dev/null; then + pacman -Syu --noconfirm whiptail + elif command -v yum &>/dev/null; then + yum install -y whiptail + else + echo -e "${RED}[Error] 无受支持的包管理器，无法安装 whiptail!${RESET}" + exit 1 + fi + fi + + # 协议确认 + if ! (whiptail --title "ℹ️ [1/6] 使用协议" --yes-button "我同意" --no-button "我拒绝" --yesno "使用MaiCore及此脚本前请先阅读EULA协议及隐私协议\nhttps://github.com/MaiM-with-u/MaiBot/blob/refactor/EULA.md\nhttps://github.com/MaiM-with-u/MaiBot/blob/refactor/PRIVACY.md\n\n您是否同意上述协议？" 12 70); then + exit 1 + fi + + # 欢迎信息 + whiptail --title "[2/6] 欢迎使用MaiCore一键安装脚本 by Cookie987" --msgbox "检测到您未安装MaiCore，将自动进入安装流程，安装完成后再次运行此脚本即可进入管理菜单。\n\n项目处于活跃开发阶段，代码可能随时更改\n文档未完善，有问题可以提交 Issue 或者 Discussion\nQQ机器人存在被限制风险，请自行了解，谨慎使用\n由于持续迭代，可能存在一些已知或未知的bug\n由于开发中，可能消耗较多token\n\n本脚本可能更新不及时，如遇到bug请优先尝试手动部署以确定是否为脚本问题" 17 60 + + # 系统检查 + check_system() { + if [[ "$(id -u)" -ne 0 ]]; then + whiptail --title "🚫 权限不足" --msgbox "请使用 root 用户运行此脚本！\n执行方式: sudo bash $0" 10 60 + exit 1 + fi + + if [[ -f /etc/os-release ]]; then + source /etc/os-release + if [[ "$ID" == "debian" && "$VERSION_ID" == "12" ]]; then + return + elif [[ "$ID" == "ubuntu" && "$VERSION_ID" == "24.10" ]]; then + return + elif [[ "$ID" == "centos" && "$VERSION_ID" == "9" ]]; then + return + elif [[ "$ID" == "arch" ]]; then + whiptail --title "⚠️ 兼容性警告" --msgbox "NapCat无可用的 Arch Linux 官方安装方法，将无法自动安装NapCat。\n\n您可尝试在AUR中搜索相关包。" 10 60 + whiptail --title "⚠️ 兼容性警告" --msgbox "MongoDB无可用的 Arch Linux 官方安装方法，将无法自动安装MongoDB。\n\n您可尝试在AUR中搜索相关包。" 10 60 + return + else + whiptail --title "🚫 不支持的系统" --msgbox "此脚本仅支持 Arch/Debian 12 (Bookworm)/Ubuntu 24.10 (Oracular Oriole)/CentOS9！\n当前系统: $PRETTY_NAME\n安装已终止。" 10 60 + exit 1 + fi + else + whiptail --title "⚠️ 无法检测系统" --msgbox "无法识别系统版本，安装已终止。" 10 60 + exit 1 + fi + } + check_system + + # 设置包管理器 + case "$ID" in + debian|ubuntu) + PKG_MANAGER="apt" + ;; + centos) + PKG_MANAGER="yum" + ;; + arch) + # 添加arch包管理器 + PKG_MANAGER="pacman" + ;; + esac + + # 检查MongoDB + check_mongodb() { + if command -v mongod &>/dev/null; then + MONGO_INSTALLED=true + else + MONGO_INSTALLED=false + fi + } + check_mongodb + + # 检查NapCat + check_napcat() { + if command -v napcat &>/dev/null; then + NAPCAT_INSTALLED=true + else + NAPCAT_INSTALLED=false + fi + } + check_napcat + + # 安装必要软件包 + install_packages() { + missing_packages=() + # 检查 common 及当前系统专属依赖 + for package in ${REQUIRED_PACKAGES["common"]} ${REQUIRED_PACKAGES["$ID"]}; do + case "$PKG_MANAGER" in + apt) + dpkg -s "$package" &>/dev/null || missing_packages+=("$package") + ;; + yum) + rpm -q "$package" &>/dev/null || missing_packages+=("$package") + ;; + pacman) + pacman -Qi "$package" &>/dev/null || missing_packages+=("$package") + ;; + esac + done + + if [[ ${#missing_packages[@]} -gt 0 ]]; then + whiptail --title "📦 [3/6] 依赖检查" --yesno "以下软件包缺失:\n${missing_packages[*]}\n\n是否自动安装？" 10 60 + if [[ $? -eq 0 ]]; then + IS_INSTALL_DEPENDENCIES=true + else + whiptail --title "⚠️ 注意" --yesno "未安装某些依赖，可能影响运行！\n是否继续？" 10 60 || exit 1 + fi + fi + } + install_packages + + # 安装MongoDB + install_mongodb() { + [[ $MONGO_INSTALLED == true ]] && return + whiptail --title "📦 [3/6] 软件包检查" --yesno "检测到未安装MongoDB，是否安装？\n如果您想使用远程数据库，请跳过此步。" 10 60 && { + IS_INSTALL_MONGODB=true + } + } + + # 仅在非Arch系统上安装MongoDB + [[ "$ID" != "arch" ]] && install_mongodb + + + # 安装NapCat + install_napcat() { + [[ $NAPCAT_INSTALLED == true ]] && return + whiptail --title "📦 [3/6] 软件包检查" --yesno "检测到未安装NapCat，是否安装？\n如果您想使用远程NapCat，请跳过此步。" 10 60 && { + IS_INSTALL_NAPCAT=true + } + } + + # 仅在非Arch系统上安装NapCat + [[ "$ID" != "arch" ]] && install_napcat + + # Python版本检查 + check_python() { + PYTHON_VERSION=$(python3 -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")') + if ! python3 -c "import sys; exit(0) if sys.version_info >= (3,9) else exit(1)"; then + whiptail --title "⚠️ [4/6] Python 版本过低" --msgbox "检测到 Python 版本为 $PYTHON_VERSION，需要 3.9 或以上！\n请升级 Python 后重新运行本脚本。" 10 60 + exit 1 + fi + } + + # 如果没安装python则不检查python版本 + if command -v python3 &>/dev/null; then + check_python + fi + + + # 选择分支 + choose_branch() { + BRANCH=refactor + } + choose_branch + + # 选择安装路径 + choose_install_dir() { + INSTALL_DIR=$(whiptail --title "📂 [6/6] 选择安装路径" --inputbox "请输入MaiCore的安装目录：" 10 60 "$DEFAULT_INSTALL_DIR" 3>&1 1>&2 2>&3) + [[ -z "$INSTALL_DIR" ]] && { + whiptail --title "⚠️ 取消输入" --yesno "未输入安装路径，是否退出安装？" 10 60 && exit 1 + INSTALL_DIR="$DEFAULT_INSTALL_DIR" + } + } + choose_install_dir + + # 确认安装 + confirm_install() { + local confirm_msg="请确认以下更改：\n\n" + confirm_msg+="📂 安装MaiCore、Nonebot Adapter到: $INSTALL_DIR\n" + confirm_msg+="🔀 分支: $BRANCH\n" + [[ $IS_INSTALL_DEPENDENCIES == true ]] && confirm_msg+="📦 安装依赖：${missing_packages[@]}\n" + [[ $IS_INSTALL_MONGODB == true || $IS_INSTALL_NAPCAT == true ]] && confirm_msg+="📦 安装额外组件：\n" + + [[ $IS_INSTALL_MONGODB == true ]] && confirm_msg+=" - MongoDB\n" + [[ $IS_INSTALL_NAPCAT == true ]] && confirm_msg+=" - NapCat\n" + confirm_msg+="\n注意：本脚本默认使用ghfast.top为GitHub进行加速，如不想使用请手动修改脚本开头的GITHUB_REPO变量。" + + whiptail --title "🔧 安装确认" --yesno "$confirm_msg" 20 60 || exit 1 + } + confirm_install + + # 开始安装 + echo -e "${GREEN}安装${missing_packages[@]}...${RESET}" + + if [[ $IS_INSTALL_DEPENDENCIES == true ]]; then + case "$PKG_MANAGER" in + apt) + apt update && apt install -y "${missing_packages[@]}" + ;; + yum) + yum install -y "${missing_packages[@]}" --nobest + ;; + pacman) + pacman -S --noconfirm "${missing_packages[@]}" + ;; + esac + fi + + if [[ $IS_INSTALL_MONGODB == true ]]; then + echo -e "${GREEN}安装 MongoDB...${RESET}" + case "$ID" in + debian) + curl -fsSL https://www.mongodb.org/static/pgp/server-8.0.asc | gpg -o /usr/share/keyrings/mongodb-server-8.0.gpg --dearmor + echo "deb [ signed-by=/usr/share/keyrings/mongodb-server-8.0.gpg ] http://repo.mongodb.org/apt/debian bookworm/mongodb-org/8.0 main" | tee /etc/apt/sources.list.d/mongodb-org-8.0.list + apt update + apt install -y mongodb-org + systemctl enable --now mongod + ;; + ubuntu) + curl -fsSL https://www.mongodb.org/static/pgp/server-8.0.asc | gpg -o /usr/share/keyrings/mongodb-server-8.0.gpg --dearmor + echo "deb [ signed-by=/usr/share/keyrings/mongodb-server-8.0.gpg ] http://repo.mongodb.org/apt/debian bookworm/mongodb-org/8.0 main" | tee /etc/apt/sources.list.d/mongodb-org-8.0.list + apt update + apt install -y mongodb-org + systemctl enable --now mongod + ;; + centos) + cat > /etc/yum.repos.d/mongodb-org-8.0.repo < pyproject.toml < README.md + mkdir src + cp -r ../../nonebot-plugin-maibot-adapters/nonebot_plugin_maibot_adapters src/plugins/nonebot_plugin_maibot_adapters + cd .. + cd .. + + + echo -e "${GREEN}同意协议...${RESET}" + + # 首先计算当前EULA的MD5值 + current_md5=$(md5sum "MaiBot/EULA.md" | awk '{print $1}') + + # 首先计算当前隐私条款文件的哈希值 + current_md5_privacy=$(md5sum "MaiBot/PRIVACY.md" | awk '{print $1}') + + echo -n $current_md5 > MaiBot/eula.confirmed + echo -n $current_md5_privacy > MaiBot/privacy.confirmed + + echo -e "${GREEN}创建系统服务...${RESET}" + cat > /etc/systemd/system/${SERVICE_NAME}.service < /etc/systemd/system/${SERVICE_NAME_WEB}.service < /etc/systemd/system/${SERVICE_NAME_NBADAPTER}.service < /etc/maicore_install.conf + echo "INSTALL_DIR=${INSTALL_DIR}" >> /etc/maicore_install.conf + echo "BRANCH=${BRANCH}" >> /etc/maicore_install.conf + + whiptail --title "🎉 安装完成" --msgbox "MaiCore安装完成！\n已创建系统服务：${SERVICE_NAME}、${SERVICE_NAME_WEB}、${SERVICE_NAME_NBADAPTER}\n\n使用以下命令管理服务：\n启动服务：systemctl start ${SERVICE_NAME}\n查看状态：systemctl status ${SERVICE_NAME}" 14 60 +} + +# ----------- 主执行流程 ----------- +# 检查root权限 +[[ $(id -u) -ne 0 ]] && { + echo -e "${RED}请使用root用户运行此脚本！${RESET}" + exit 1 +} + +# 如果已安装显示菜单，并检查协议是否更新 +if check_installed; then + load_install_info + check_eula + show_menu +else + run_installation + # 安装完成后询问是否启动 + if whiptail --title "安装完成" --yesno "是否立即启动MaiCore服务？" 10 60; then + systemctl start ${SERVICE_NAME} + whiptail --msgbox "✅ 服务已启动！\n使用 systemctl status ${SERVICE_NAME} 查看状态" 10 60 + fi +fi diff --git a/src/heart_flow/heartflow.py b/src/heart_flow/heartflow.py index c34def599..2d0326384 100644 --- a/src/heart_flow/heartflow.py +++ b/src/heart_flow/heartflow.py @@ -144,23 +144,28 @@ class Heartflow: 添加一个SubHeartflow实例到self._subheartflows字典中 并根据subheartflow_id为子心流创建一个观察对象 """ - if subheartflow_id not in self._subheartflows: - logger.debug(f"创建 subheartflow: {subheartflow_id}") - subheartflow = SubHeartflow(subheartflow_id) - # 创建一个观察对象，目前只可以用chat_id创建观察对象 - logger.debug(f"创建 observation: {subheartflow_id}") - observation = ChattingObservation(subheartflow_id) + + try: + if subheartflow_id not in self._subheartflows: + logger.debug(f"创建 subheartflow: {subheartflow_id}") + subheartflow = SubHeartflow(subheartflow_id) + # 创建一个观察对象，目前只可以用chat_id创建观察对象 + logger.debug(f"创建 observation: {subheartflow_id}") + observation = ChattingObservation(subheartflow_id) - logger.debug("添加 observation ") - subheartflow.add_observation(observation) - logger.debug("添加 observation 成功") - # 创建异步任务 - logger.debug("创建异步任务") - asyncio.create_task(subheartflow.subheartflow_start_working()) - logger.debug("创建异步任务 成功") - self._subheartflows[subheartflow_id] = subheartflow - logger.info("添加 subheartflow 成功") - return self._subheartflows[subheartflow_id] + logger.debug("添加 observation ") + subheartflow.add_observation(observation) + logger.debug("添加 observation 成功") + # 创建异步任务 + logger.debug("创建异步任务") + asyncio.create_task(subheartflow.subheartflow_start_working()) + logger.debug("创建异步任务 成功") + self._subheartflows[subheartflow_id] = subheartflow + logger.info("添加 subheartflow 成功") + return self._subheartflows[subheartflow_id] + except Exception as e: + logger.error(f"创建 subheartflow 失败: {e}") + return None def get_subheartflow(self, observe_chat_id): """获取指定ID的SubHeartflow实例""" diff --git a/src/plugins/P.F.C/pfc.py b/src/plugins/P.F.C/pfc.py deleted file mode 100644 index 9b83bce40..000000000 --- a/src/plugins/P.F.C/pfc.py +++ /dev/null @@ -1,3 +0,0 @@ -#Programmable Friendly Conversationalist -#Prefrontal cortex - diff --git a/src/plugins/PFC/chat_observer.py b/src/plugins/PFC/chat_observer.py new file mode 100644 index 000000000..4fa6951e2 --- /dev/null +++ b/src/plugins/PFC/chat_observer.py @@ -0,0 +1,292 @@ +import time +import asyncio +from typing import Optional, Dict, Any, List +from src.common.logger import get_module_logger +from src.common.database import db +from ..message.message_base import UserInfo +from ..config.config import global_config + +logger = get_module_logger("chat_observer") + +class ChatObserver: + """聊天状态观察器""" + + # 类级别的实例管理 + _instances: Dict[str, 'ChatObserver'] = {} + + @classmethod + def get_instance(cls, stream_id: str) -> 'ChatObserver': + """获取或创建观察器实例 + + Args: + stream_id: 聊天流ID + + Returns: + ChatObserver: 观察器实例 + """ + if stream_id not in cls._instances: + cls._instances[stream_id] = cls(stream_id) + return cls._instances[stream_id] + + def __init__(self, stream_id: str): + """初始化观察器 + + Args: + stream_id: 聊天流ID + """ + if stream_id in self._instances: + raise RuntimeError(f"ChatObserver for {stream_id} already exists. Use get_instance() instead.") + + self.stream_id = stream_id + self.last_user_speak_time: Optional[float] = None # 对方上次发言时间 + self.last_bot_speak_time: Optional[float] = None # 机器人上次发言时间 + self.last_check_time: float = time.time() # 上次查看聊天记录时间 + self.last_message_read: Optional[str] = None # 最后读取的消息ID + self.last_message_time: Optional[float] = None # 最后一条消息的时间戳 + + self.waiting_start_time: Optional[float] = None # 等待开始时间 + + # 消息历史记录 + self.message_history: List[Dict[str, Any]] = [] # 所有消息历史 + self.last_message_id: Optional[str] = None # 最后一条消息的ID + self.message_count: int = 0 # 消息计数 + + # 运行状态 + self._running: bool = False + self._task: Optional[asyncio.Task] = None + self._update_event = asyncio.Event() # 触发更新的事件 + self._update_complete = asyncio.Event() # 更新完成的事件 + + def new_message_after(self, time_point: float) -> bool: + """判断是否在指定时间点后有新消息 + + Args: + time_point: 时间戳 + + Returns: + bool: 是否有新消息 + """ + return self.last_message_time is None or self.last_message_time > time_point + + def _add_message_to_history(self, message: Dict[str, Any]): + """添加消息到历史记录 + + Args: + message: 消息数据 + """ + self.message_history.append(message) + self.last_message_id = message["message_id"] + self.last_message_time = message["time"] # 更新最后消息时间 + self.message_count += 1 + + # 更新说话时间 + user_info = UserInfo.from_dict(message.get("user_info", {})) + if user_info.user_id == global_config.BOT_QQ: + self.last_bot_speak_time = message["time"] + else: + self.last_user_speak_time = message["time"] + + def get_message_history( + self, + start_time: Optional[float] = None, + end_time: Optional[float] = None, + limit: Optional[int] = None, + user_id: Optional[str] = None + ) -> List[Dict[str, Any]]: + """获取消息历史 + + Args: + start_time: 开始时间戳 + end_time: 结束时间戳 + limit: 限制返回消息数量 + user_id: 指定用户ID + + Returns: + List[Dict[str, Any]]: 消息列表 + """ + filtered_messages = self.message_history + + if start_time is not None: + filtered_messages = [m for m in filtered_messages if m["time"] >= start_time] + + if end_time is not None: + filtered_messages = [m for m in filtered_messages if m["time"] <= end_time] + + if user_id is not None: + filtered_messages = [ + m for m in filtered_messages + if UserInfo.from_dict(m.get("user_info", {})).user_id == user_id + ] + + if limit is not None: + filtered_messages = filtered_messages[-limit:] + + return filtered_messages + + async def _fetch_new_messages(self) -> List[Dict[str, Any]]: + """获取新消息 + + Returns: + List[Dict[str, Any]]: 新消息列表 + """ + query = {"chat_id": self.stream_id} + if self.last_message_read: + # 获取ID大于last_message_read的消息 + last_message = db.messages.find_one({"message_id": self.last_message_read}) + if last_message: + query["time"] = {"$gt": last_message["time"]} + + new_messages = list( + db.messages.find(query).sort("time", 1) + ) + + if new_messages: + self.last_message_read = new_messages[-1]["message_id"] + + return new_messages + + async def _fetch_new_messages_before(self, time_point: float) -> List[Dict[str, Any]]: + """获取指定时间点之前的消息 + + Args: + time_point: 时间戳 + + Returns: + List[Dict[str, Any]]: 最多5条消息 + """ + query = { + "chat_id": self.stream_id, + "time": {"$lt": time_point} + } + + new_messages = list( + db.messages.find(query).sort("time", -1).limit(5) # 倒序获取5条 + ) + + # 将消息按时间正序排列 + new_messages.reverse() + + if new_messages: + self.last_message_read = new_messages[-1]["message_id"] + + return new_messages + + async def _update_loop(self): + """更新循环""" + try: + start_time = time.time() + messages = await self._fetch_new_messages_before(start_time) + for message in messages: + self._add_message_to_history(message) + except Exception as e: + logger.error(f"缓冲消息出错: {e}") + + while self._running: + try: + # 等待事件或超时（1秒） + try: + await asyncio.wait_for(self._update_event.wait(), timeout=1) + except asyncio.TimeoutError: + pass # 超时后也执行一次检查 + + self._update_event.clear() # 重置触发事件 + self._update_complete.clear() # 重置完成事件 + + # 获取新消息 + new_messages = await self._fetch_new_messages() + + if new_messages: + # 处理新消息 + for message in new_messages: + self._add_message_to_history(message) + + # 设置完成事件 + self._update_complete.set() + + except Exception as e: + logger.error(f"更新循环出错: {e}") + self._update_complete.set() # 即使出错也要设置完成事件 + + def trigger_update(self): + """触发一次立即更新""" + self._update_event.set() + + async def wait_for_update(self, timeout: float = 5.0) -> bool: + """等待更新完成 + + Args: + timeout: 超时时间（秒） + + Returns: + bool: 是否成功完成更新（False表示超时） + """ + try: + await asyncio.wait_for(self._update_complete.wait(), timeout=timeout) + return True + except asyncio.TimeoutError: + logger.warning(f"等待更新完成超时（{timeout}秒）") + return False + + def start(self): + """启动观察器""" + if self._running: + return + + self._running = True + self._task = asyncio.create_task(self._update_loop()) + logger.info(f"ChatObserver for {self.stream_id} started") + + def stop(self): + """停止观察器""" + self._running = False + self._update_event.set() # 设置事件以解除等待 + self._update_complete.set() # 设置完成事件以解除等待 + if self._task: + self._task.cancel() + logger.info(f"ChatObserver for {self.stream_id} stopped") + + async def process_chat_history(self, messages: list): + """处理聊天历史 + + Args: + messages: 消息列表 + """ + self.update_check_time() + + for msg in messages: + try: + user_info = UserInfo.from_dict(msg.get("user_info", {})) + if user_info.user_id == global_config.BOT_QQ: + self.update_bot_speak_time(msg["time"]) + else: + self.update_user_speak_time(msg["time"]) + except Exception as e: + logger.warning(f"处理消息时间时出错: {e}") + continue + + def update_check_time(self): + """更新查看时间""" + self.last_check_time = time.time() + + def update_bot_speak_time(self, speak_time: Optional[float] = None): + """更新机器人说话时间""" + self.last_bot_speak_time = speak_time or time.time() + + def update_user_speak_time(self, speak_time: Optional[float] = None): + """更新用户说话时间""" + self.last_user_speak_time = speak_time or time.time() + + def get_time_info(self) -> str: + """获取时间信息文本""" + current_time = time.time() + time_info = "" + + if self.last_bot_speak_time: + bot_speak_ago = current_time - self.last_bot_speak_time + time_info += f"\n距离你上次发言已经过去了{int(bot_speak_ago)}秒" + + if self.last_user_speak_time: + user_speak_ago = current_time - self.last_user_speak_time + time_info += f"\n距离对方上次发言已经过去了{int(user_speak_ago)}秒" + + return time_info diff --git a/src/plugins/PFC/pfc.py b/src/plugins/PFC/pfc.py new file mode 100644 index 000000000..667a6f035 --- /dev/null +++ b/src/plugins/PFC/pfc.py @@ -0,0 +1,834 @@ +#Programmable Friendly Conversationalist +#Prefrontal cortex +import datetime +import asyncio +from typing import List, Optional, Dict, Any, Tuple, Literal +from enum import Enum +from src.common.logger import get_module_logger +from ..chat.chat_stream import ChatStream +from ..message.message_base import UserInfo, Seg +from ..chat.message import Message +from ..models.utils_model import LLM_request +from ..config.config import global_config +from src.plugins.chat.message import MessageSending +from src.plugins.chat.chat_stream import chat_manager +from ..message.api import global_api +from ..storage.storage import MessageStorage +from .chat_observer import ChatObserver +from .pfc_KnowledgeFetcher import KnowledgeFetcher +from .reply_checker import ReplyChecker +import json +import time + +logger = get_module_logger("pfc") + + +class ConversationState(Enum): + """对话状态""" + INIT = "初始化" + RETHINKING = "重新思考" + ANALYZING = "分析历史" + PLANNING = "规划目标" + GENERATING = "生成回复" + CHECKING = "检查回复" + SENDING = "发送消息" + WAITING = "等待" + LISTENING = "倾听" + ENDED = "结束" + JUDGING = "判断" + + +ActionType = Literal["direct_reply", "fetch_knowledge", "wait"] + + +class ActionPlanner: + """行动规划器""" + + def __init__(self, stream_id: str): + self.llm = LLM_request( + model=global_config.llm_normal, + temperature=0.7, + max_tokens=1000, + request_type="action_planning" + ) + self.personality_info = " ".join(global_config.PROMPT_PERSONALITY) + self.name = global_config.BOT_NICKNAME + self.chat_observer = ChatObserver.get_instance(stream_id) + + async def plan( + self, + goal: str, + method: str, + reasoning: str, + action_history: List[Dict[str, str]] = None, + chat_observer: Optional[ChatObserver] = None, # 添加chat_observer参数 + ) -> Tuple[str, str]: + """规划下一步行动 + + Args: + goal: 对话目标 + method: 实现方式 + reasoning: 目标原因 + action_history: 行动历史记录 + + Returns: + Tuple[str, str]: (行动类型, 行动原因) + """ + # 构建提示词 + # 获取最近20条消息 + self.chat_observer.waiting_start_time = time.time() + + messages = self.chat_observer.get_message_history(limit=20) + chat_history_text = "" + for msg in messages: + time_str = datetime.datetime.fromtimestamp(msg["time"]).strftime("%H:%M:%S") + user_info = UserInfo.from_dict(msg.get("user_info", {})) + sender = user_info.user_nickname or f"用户{user_info.user_id}" + if sender == self.name: + sender = "你说" + chat_history_text += f"{time_str},{sender}:{msg.get('processed_plain_text', '')}\n" + + personality_text = f"你的名字是{self.name}，{self.personality_info}" + + # 构建action历史文本 + action_history_text = "" + if action_history: + if action_history[-1]['action'] == "direct_reply": + action_history_text = "你刚刚发言回复了对方" + + # 获取时间信息 + time_info = self.chat_observer.get_time_info() + + prompt = f"""现在你在参与一场QQ聊天，请分析以下内容，根据信息决定下一步行动： +{personality_text} +当前对话目标：{goal} +实现该对话目标的方式：{method} +产生该对话目标的原因：{reasoning} +{time_info} +最近的对话记录： +{chat_history_text} +{action_history_text} +请你接下去想想要你要做什么，可以发言，可以等待，可以倾听，可以调取知识。注意不同行动类型的要求，不要重复发言： +行动类型： +fetch_knowledge: 需要调取知识，当需要专业知识或特定信息时选择 +wait: 当你做出了发言,对方尚未回复时等待对方的回复 +listening: 倾听对方发言，当你认为对方发言尚未结束时采用 +direct_reply: 不符合上述情况，回复对方，注意不要过多或者重复发言 +rethink_goal: 重新思考对话目标，当发现对话目标不合适时选择，会重新思考对话目标 +judge_conversation: 判断对话是否结束，当发现对话目标已经达到或者希望停止对话时选择，会判断对话是否结束 + +请以JSON格式输出，包含以下字段： +1. action: 行动类型，注意你之前的行为 +2. reason: 选择该行动的原因，注意你之前的行为（简要解释） + +注意：请严格按照JSON格式输出，不要包含任何其他内容。""" + + logger.debug(f"发送到LLM的提示词: {prompt}") + try: + content, _ = await self.llm.generate_response_async(prompt) + logger.debug(f"LLM原始返回内容: {content}") + + # 清理内容，尝试提取JSON部分 + content = content.strip() + try: + # 尝试直接解析 + result = json.loads(content) + except json.JSONDecodeError: + # 如果直接解析失败，尝试查找和提取JSON部分 + import re + json_pattern = r'\{[^{}]*\}' + json_match = re.search(json_pattern, content) + if json_match: + try: + result = json.loads(json_match.group()) + except json.JSONDecodeError: + logger.error("提取的JSON内容解析失败，返回默认行动") + return "direct_reply", "JSON解析失败，选择直接回复" + else: + # 如果找不到JSON，尝试从文本中提取行动和原因 + if "direct_reply" in content.lower(): + return "direct_reply", "从文本中提取的行动" + elif "fetch_knowledge" in content.lower(): + return "fetch_knowledge", "从文本中提取的行动" + elif "wait" in content.lower(): + return "wait", "从文本中提取的行动" + elif "listening" in content.lower(): + return "listening", "从文本中提取的行动" + elif "rethink_goal" in content.lower(): + return "rethink_goal", "从文本中提取的行动" + elif "judge_conversation" in content.lower(): + return "judge_conversation", "从文本中提取的行动" + else: + logger.error("无法从返回内容中提取行动类型") + return "direct_reply", "无法解析响应，选择直接回复" + + # 验证JSON字段 + action = result.get("action", "direct_reply") + reason = result.get("reason", "默认原因") + + # 验证action类型 + if action not in ["direct_reply", "fetch_knowledge", "wait", "listening", "rethink_goal", "judge_conversation"]: + logger.warning(f"未知的行动类型: {action}，默认使用listening") + action = "listening" + + logger.info(f"规划的行动: {action}") + logger.info(f"行动原因: {reason}") + return action, reason + + except Exception as e: + logger.error(f"规划行动时出错: {str(e)}") + return "direct_reply", "发生错误，选择直接回复" + + +class GoalAnalyzer: + """对话目标分析器""" + + def __init__(self, stream_id: str): + self.llm = LLM_request( + model=global_config.llm_normal, + temperature=0.7, + max_tokens=1000, + request_type="conversation_goal" + ) + + self.personality_info = " ".join(global_config.PROMPT_PERSONALITY) + self.name = global_config.BOT_NICKNAME + self.nick_name = global_config.BOT_ALIAS_NAMES + self.chat_observer = ChatObserver.get_instance(stream_id) + + async def analyze_goal(self) -> Tuple[str, str, str]: + """分析对话历史并设定目标 + + Args: + chat_history: 聊天历史记录列表 + + Returns: + Tuple[str, str, str]: (目标, 方法, 原因) + """ + max_retries = 3 + for retry in range(max_retries): + try: + # 构建提示词 + messages = self.chat_observer.get_message_history(limit=20) + chat_history_text = "" + for msg in messages: + time_str = datetime.datetime.fromtimestamp(msg["time"]).strftime("%H:%M:%S") + user_info = UserInfo.from_dict(msg.get("user_info", {})) + sender = user_info.user_nickname or f"用户{user_info.user_id}" + if sender == self.name: + sender = "你说" + chat_history_text += f"{time_str},{sender}:{msg.get('processed_plain_text', '')}\n" + + personality_text = f"你的名字是{self.name}，{self.personality_info}" + + prompt = f"""{personality_text}。现在你在参与一场QQ聊天，请分析以下聊天记录，并根据你的性格特征确定一个明确的对话目标。 +这个目标应该反映出对话的意图和期望的结果。 +聊天记录： +{chat_history_text} +请以JSON格式输出，包含以下字段： +1. goal: 对话目标（简短的一句话） +2. reasoning: 对话原因，为什么设定这个目标（简要解释） + +输出格式示例： +{{ + "goal": "回答用户关于Python编程的具体问题", + "reasoning": "用户提出了关于Python的技术问题，需要专业且准确的解答" +}}""" + + logger.debug(f"发送到LLM的提示词: {prompt}") + content, _ = await self.llm.generate_response_async(prompt) + logger.debug(f"LLM原始返回内容: {content}") + + # 清理和验证返回内容 + if not content or not isinstance(content, str): + logger.error("LLM返回内容为空或格式不正确") + continue + + # 尝试提取JSON部分 + content = content.strip() + try: + # 尝试直接解析 + result = json.loads(content) + except json.JSONDecodeError: + # 如果直接解析失败，尝试查找和提取JSON部分 + import re + json_pattern = r'\{[^{}]*\}' + json_match = re.search(json_pattern, content) + if json_match: + try: + result = json.loads(json_match.group()) + except json.JSONDecodeError: + logger.error(f"提取的JSON内容解析失败，重试第{retry + 1}次") + continue + else: + logger.error(f"无法在返回内容中找到有效的JSON，重试第{retry + 1}次") + continue + + # 验证JSON字段 + if not all(key in result for key in ["goal", "reasoning"]): + logger.error(f"JSON缺少必要字段，实际内容: {result}，重试第{retry + 1}次") + continue + + goal = result["goal"] + reasoning = result["reasoning"] + + # 验证字段内容 + if not isinstance(goal, str) or not isinstance(reasoning, str): + logger.error(f"JSON字段类型错误，goal和reasoning必须是字符串，重试第{retry + 1}次") + continue + + if not goal.strip() or not reasoning.strip(): + logger.error(f"JSON字段内容为空，重试第{retry + 1}次") + continue + + # 使用默认的方法 + method = "以友好的态度回应" + return goal, method, reasoning + + except Exception as e: + logger.error(f"分析对话目标时出错: {str(e)}，重试第{retry + 1}次") + if retry == max_retries - 1: + return "保持友好的对话", "以友好的态度回应", "确保对话顺利进行" + continue + + # 所有重试都失败后的默认返回 + return "保持友好的对话", "以友好的态度回应", "确保对话顺利进行" + + async def analyze_conversation(self,goal,reasoning): + messages = self.chat_observer.get_message_history() + chat_history_text = "" + for msg in messages: + time_str = datetime.datetime.fromtimestamp(msg["time"]).strftime("%H:%M:%S") + user_info = UserInfo.from_dict(msg.get("user_info", {})) + sender = user_info.user_nickname or f"用户{user_info.user_id}" + if sender == self.name: + sender = "你说" + chat_history_text += f"{time_str},{sender}:{msg.get('processed_plain_text', '')}\n" + + personality_text = f"你的名字是{self.name}，{self.personality_info}" + + prompt = f"""{personality_text}。现在你在参与一场QQ聊天， + 当前对话目标：{goal} + 产生该对话目标的原因：{reasoning} + + 请分析以下聊天记录，并根据你的性格特征评估该目标是否已经达到，或者你是否希望停止该次对话。 + 聊天记录： + {chat_history_text} + 请以JSON格式输出，包含以下字段： + 1. goal_achieved: 对话目标是否已经达到（true/false） + 2. stop_conversation: 是否希望停止该次对话（true/false） + 3. reason: 为什么希望停止该次对话（简要解释） + +输出格式示例： +{{ + "goal_achieved": true, + "stop_conversation": false, + "reason": "用户已经得到了满意的回答，但我仍希望继续聊天" +}}""" + logger.debug(f"发送到LLM的提示词: {prompt}") + try: + content, _ = await self.llm.generate_response_async(prompt) + logger.debug(f"LLM原始返回内容: {content}") + + # 清理和验证返回内容 + if not content or not isinstance(content, str): + logger.error("LLM返回内容为空或格式不正确") + return False, False, "确保对话顺利进行" + + # 尝试提取JSON部分 + content = content.strip() + try: + # 尝试直接解析 + result = json.loads(content) + except json.JSONDecodeError: + # 如果直接解析失败，尝试查找和提取JSON部分 + import re + json_pattern = r'\{[^{}]*\}' + json_match = re.search(json_pattern, content) + if json_match: + try: + result = json.loads(json_match.group()) + except json.JSONDecodeError as e: + logger.error(f"提取的JSON内容解析失败: {e}") + return False, False, "确保对话顺利进行" + else: + logger.error("无法在返回内容中找到有效的JSON") + return False, False, "确保对话顺利进行" + + # 验证JSON字段 + if not all(key in result for key in ["goal_achieved", "stop_conversation", "reason"]): + logger.error(f"JSON缺少必要字段，实际内容: {result}") + return False, False, "确保对话顺利进行" + + goal_achieved = result["goal_achieved"] + stop_conversation = result["stop_conversation"] + reason = result["reason"] + + # 验证字段类型 + if not isinstance(goal_achieved, bool): + logger.error("goal_achieved 必须是布尔值") + return False, False, "确保对话顺利进行" + + if not isinstance(stop_conversation, bool): + logger.error("stop_conversation 必须是布尔值") + return False, False, "确保对话顺利进行" + + if not isinstance(reason, str): + logger.error("reason 必须是字符串") + return False, False, "确保对话顺利进行" + + if not reason.strip(): + logger.error("reason 不能为空") + return False, False, "确保对话顺利进行" + + return goal_achieved, stop_conversation, reason + + except Exception as e: + logger.error(f"分析对话目标时出错: {str(e)}") + return False, False, "确保对话顺利进行" + + +class Waiter: + """快 速 等 待""" + def __init__(self, stream_id: str): + self.chat_observer = ChatObserver.get_instance(stream_id) + self.personality_info = " ".join(global_config.PROMPT_PERSONALITY) + self.name = global_config.BOT_NICKNAME + + async def wait(self) -> bool: + """等待 + + Returns: + bool: 是否超时（True表示超时） + """ + wait_start_time = self.chat_observer.waiting_start_time + while not self.chat_observer.new_message_after(wait_start_time): + await asyncio.sleep(1) + logger.info("等待中...") + # 检查是否超过60秒 + if time.time() - wait_start_time > 60: + logger.info("等待超过60秒，结束对话") + return True + logger.info("等待结束") + return False + + +class ReplyGenerator: + """回复生成器""" + + def __init__(self, stream_id: str): + self.llm = LLM_request( + model=global_config.llm_normal, + temperature=0.7, + max_tokens=300, + request_type="reply_generation" + ) + self.personality_info = " ".join(global_config.PROMPT_PERSONALITY) + self.name = global_config.BOT_NICKNAME + self.chat_observer = ChatObserver.get_instance(stream_id) + self.reply_checker = ReplyChecker(stream_id) + + async def generate( + self, + goal: str, + chat_history: List[Message], + knowledge_cache: Dict[str, str], + previous_reply: Optional[str] = None, + retry_count: int = 0 + ) -> Tuple[str, bool]: + """生成回复 + + Args: + goal: 对话目标 + method: 实现方式 + chat_history: 聊天历史 + knowledge_cache: 知识缓存 + previous_reply: 上一次生成的回复（如果有） + retry_count: 当前重试次数 + + Returns: + Tuple[str, bool]: (生成的回复, 是否需要重新规划) + """ + # 构建提示词 + logger.debug(f"开始生成回复：当前目标: {goal}") + self.chat_observer.trigger_update() # 触发立即更新 + if not await self.chat_observer.wait_for_update(): + logger.warning("等待消息更新超时") + + messages = self.chat_observer.get_message_history(limit=20) + chat_history_text = "" + for msg in messages: + time_str = datetime.datetime.fromtimestamp(msg["time"]).strftime("%H:%M:%S") + user_info = UserInfo.from_dict(msg.get("user_info", {})) + sender = user_info.user_nickname or f"用户{user_info.user_id}" + if sender == self.name: + sender = "你说" + chat_history_text += f"{time_str},{sender}:{msg.get('processed_plain_text', '')}\n" + + # 整理知识缓存 + knowledge_text = "" + if knowledge_cache: + knowledge_text = "\n相关知识：" + if isinstance(knowledge_cache, dict): + for _source, content in knowledge_cache.items(): + knowledge_text += f"\n{content}" + elif isinstance(knowledge_cache, list): + for item in knowledge_cache: + knowledge_text += f"\n{item}" + + # 添加上一次生成的回复信息 + previous_reply_text = "" + if previous_reply: + previous_reply_text = f"\n上一次生成的回复（需要改进）：\n{previous_reply}" + + personality_text = f"你的名字是{self.name}，{self.personality_info}" + + prompt = f"""{personality_text}。现在你在参与一场QQ聊天，请根据以下信息生成回复： + +当前对话目标：{goal} +{knowledge_text} +{previous_reply_text} +最近的聊天记录： +{chat_history_text} + +请根据上述信息，以你的性格特征生成一个自然、得体的回复。回复应该： +1. 符合对话目标，以"你"的角度发言 +2. 体现你的性格特征 +3. 自然流畅，像正常聊天一样，简短 +4. 适当利用相关知识，但不要生硬引用 +{'5. 改进上一次回复中的问题' if previous_reply else ''} + +请注意把握聊天内容，不要回复的太有条理，可以有个性。请分清"你"和对方说的话，不要把"你"说的话当做对方说的话，这是你自己说的话。 +请你回复的平淡一些，简短一些，说中文，不要刻意突出自身学科背景，尽量不要说你说过的话 +请你注意不要输出多余内容(包括前后缀，冒号和引号，括号，表情等)，只输出回复内容。 +不要输出多余内容(包括前后缀，冒号和引号，括号，表情包，at或 @等 )。 + +请直接输出回复内容，不需要任何额外格式。""" + + try: + content, _ = await self.llm.generate_response_async(prompt) + logger.info(f"生成的回复: {content}") + + # 检查生成的回复是否合适 + is_suitable, reason, need_replan = await self.reply_checker.check( + content, goal, retry_count + ) + + if not is_suitable: + logger.warning(f"生成的回复不合适，原因: {reason}") + if need_replan: + logger.info("需要重新规划对话目标") + return "让我重新思考一下...", True + else: + # 递归调用，将当前回复作为previous_reply传入 + return await self.generate( + goal, chat_history, knowledge_cache, + content, retry_count + 1 + ) + + return content, False + + except Exception as e: + logger.error(f"生成回复时出错: {e}") + return "抱歉，我现在有点混乱，让我重新思考一下...", True + + +class Conversation: + # 类级别的实例管理 + _instances: Dict[str, 'Conversation'] = {} + + @classmethod + def get_instance(cls, stream_id: str) -> 'Conversation': + """获取或创建对话实例""" + if stream_id not in cls._instances: + cls._instances[stream_id] = cls(stream_id) + logger.info(f"创建新的对话实例: {stream_id}") + return cls._instances[stream_id] + + @classmethod + def remove_instance(cls, stream_id: str): + """删除对话实例""" + if stream_id in cls._instances: + # 停止相关组件 + instance = cls._instances[stream_id] + instance.chat_observer.stop() + # 删除实例 + del cls._instances[stream_id] + logger.info(f"已删除对话实例 {stream_id}") + + def __init__(self, stream_id: str): + """初始化对话系统""" + self.stream_id = stream_id + self.state = ConversationState.INIT + self.current_goal: Optional[str] = None + self.current_method: Optional[str] = None + self.goal_reasoning: Optional[str] = None + self.generated_reply: Optional[str] = None + self.should_continue = True + + # 初始化聊天观察器 + self.chat_observer = ChatObserver.get_instance(stream_id) + + # 添加action历史记录 + self.action_history: List[Dict[str, str]] = [] + + # 知识缓存 + self.knowledge_cache: Dict[str, str] = {} # 确保初始化为字典 + + # 初始化各个组件 + self.goal_analyzer = GoalAnalyzer(self.stream_id) + self.action_planner = ActionPlanner(self.stream_id) + self.reply_generator = ReplyGenerator(self.stream_id) + self.knowledge_fetcher = KnowledgeFetcher() + self.direct_sender = DirectMessageSender() + self.waiter = Waiter(self.stream_id) + + # 创建聊天流 + self.chat_stream = chat_manager.get_stream(self.stream_id) + + def _clear_knowledge_cache(self): + """清空知识缓存""" + self.knowledge_cache.clear() # 使用clear方法清空字典 + + async def start(self): + """开始对话流程""" + logger.info("对话系统启动") + self.should_continue = True + self.chat_observer.start() # 启动观察器 + await asyncio.sleep(1) + # 启动对话循环 + await self._conversation_loop() + + async def _conversation_loop(self): + """对话循环""" + # 获取最近的消息历史 + self.current_goal, self.current_method, self.goal_reasoning = await self.goal_analyzer.analyze_goal() + + while self.should_continue: + # 执行行动 + self.chat_observer.trigger_update() # 触发立即更新 + if not await self.chat_observer.wait_for_update(): + logger.warning("等待消息更新超时") + + action, reason = await self.action_planner.plan( + self.current_goal, + self.current_method, + self.goal_reasoning, + self.action_history, # 传入action历史 + self.chat_observer # 传入chat_observer + ) + + # 执行行动 + await self._handle_action(action, reason) + + def _convert_to_message(self, msg_dict: Dict[str, Any]) -> Message: + """将消息字典转换为Message对象""" + try: + chat_info = msg_dict.get("chat_info", {}) + chat_stream = ChatStream.from_dict(chat_info) + user_info = UserInfo.from_dict(msg_dict.get("user_info", {})) + + return Message( + message_id=msg_dict["message_id"], + chat_stream=chat_stream, + time=msg_dict["time"], + user_info=user_info, + processed_plain_text=msg_dict.get("processed_plain_text", ""), + detailed_plain_text=msg_dict.get("detailed_plain_text", "") + ) + except Exception as e: + logger.warning(f"转换消息时出错: {e}") + raise + + async def _handle_action(self, action: str, reason: str): + """处理规划的行动""" + logger.info(f"执行行动: {action}, 原因: {reason}") + + # 记录action历史 + self.action_history.append({ + "action": action, + "reason": reason, + "time": datetime.datetime.now().strftime("%H:%M:%S") + }) + + # 只保留最近的10条记录 + if len(self.action_history) > 10: + self.action_history = self.action_history[-10:] + + if action == "direct_reply": + self.state = ConversationState.GENERATING + messages = self.chat_observer.get_message_history(limit=30) + self.generated_reply, need_replan = await self.reply_generator.generate( + self.current_goal, + self.current_method, + [self._convert_to_message(msg) for msg in messages], + self.knowledge_cache + ) + if need_replan: + self.state = ConversationState.RETHINKING + self.current_goal, self.current_method, self.goal_reasoning = await self.goal_analyzer.analyze_goal() + else: + await self._send_reply() + + elif action == "fetch_knowledge": + self.state = ConversationState.GENERATING + messages = self.chat_observer.get_message_history(limit=30) + knowledge, sources = await self.knowledge_fetcher.fetch( + self.current_goal, + [self._convert_to_message(msg) for msg in messages] + ) + logger.info(f"获取到知识，来源: {sources}") + + if knowledge != "未找到相关知识": + self.knowledge_cache[sources] = knowledge + + self.generated_reply, need_replan = await self.reply_generator.generate( + self.current_goal, + self.current_method, + [self._convert_to_message(msg) for msg in messages], + self.knowledge_cache + ) + if need_replan: + self.state = ConversationState.RETHINKING + self.current_goal, self.current_method, self.goal_reasoning = await self.goal_analyzer.analyze_goal() + else: + await self._send_reply() + + elif action == "rethink_goal": + self.state = ConversationState.RETHINKING + self.current_goal, self.current_method, self.goal_reasoning = await self.goal_analyzer.analyze_goal() + + elif action == "judge_conversation": + self.state = ConversationState.JUDGING + self.goal_achieved, self.stop_conversation, self.reason = await self.goal_analyzer.analyze_conversation(self.current_goal, self.goal_reasoning) + if self.stop_conversation: + await self._stop_conversation() + + elif action == "listening": + self.state = ConversationState.LISTENING + logger.info("倾听对方发言...") + if await self.waiter.wait(): # 如果返回True表示超时 + await self._send_timeout_message() + await self._stop_conversation() + + else: # wait + self.state = ConversationState.WAITING + logger.info("等待更多信息...") + if await self.waiter.wait(): # 如果返回True表示超时 + await self._send_timeout_message() + await self._stop_conversation() + + async def _stop_conversation(self): + """完全停止对话""" + logger.info("停止对话") + self.should_continue = False + self.state = ConversationState.ENDED + # 删除实例（这会同时停止chat_observer） + self.remove_instance(self.stream_id) + + async def _send_timeout_message(self): + """发送超时结束消息""" + try: + messages = self.chat_observer.get_message_history(limit=1) + if not messages: + return + + latest_message = self._convert_to_message(messages[0]) + await self.direct_sender.send_message( + chat_stream=self.chat_stream, + content="抱歉，由于等待时间过长，我需要先去忙别的了。下次再聊吧~", + reply_to_message=latest_message + ) + except Exception as e: + logger.error(f"发送超时消息失败: {str(e)}") + + async def _send_reply(self): + """发送回复""" + if not self.generated_reply: + logger.warning("没有生成回复") + return + + messages = self.chat_observer.get_message_history(limit=1) + if not messages: + logger.warning("没有最近的消息可以回复") + return + + latest_message = self._convert_to_message(messages[0]) + try: + await self.direct_sender.send_message( + chat_stream=self.chat_stream, + content=self.generated_reply, + reply_to_message=latest_message + ) + self.chat_observer.trigger_update() # 触发立即更新 + if not await self.chat_observer.wait_for_update(): + logger.warning("等待消息更新超时") + + self.state = ConversationState.ANALYZING + except Exception as e: + logger.error(f"发送消息失败: {str(e)}") + self.state = ConversationState.ANALYZING + + +class DirectMessageSender: + """直接发送消息到平台的发送器""" + + def __init__(self): + self.logger = get_module_logger("direct_sender") + self.storage = MessageStorage() + + async def send_message( + self, + chat_stream: ChatStream, + content: str, + reply_to_message: Optional[Message] = None, + ) -> None: + """直接发送消息到平台 + + Args: + chat_stream: 聊天流 + content: 消息内容 + reply_to_message: 要回复的消息 + """ + # 构建消息对象 + message_segment = Seg(type="text", data=content) + bot_user_info = UserInfo( + user_id=global_config.BOT_QQ, + user_nickname=global_config.BOT_NICKNAME, + platform=chat_stream.platform, + ) + + message = MessageSending( + message_id=f"dm{round(time.time(), 2)}", + chat_stream=chat_stream, + bot_user_info=bot_user_info, + sender_info=reply_to_message.message_info.user_info if reply_to_message else None, + message_segment=message_segment, + reply=reply_to_message, + is_head=True, + is_emoji=False, + thinking_start_time=time.time(), + ) + + # 处理消息 + await message.process() + + # 发送消息 + try: + message_json = message.to_dict() + end_point = global_config.api_urls.get(chat_stream.platform, None) + + if not end_point: + raise ValueError(f"未找到平台：{chat_stream.platform} 的url配置") + + await global_api.send_message(end_point, message_json) + + # 存储消息 + await self.storage.store_message(message, message.chat_stream) + + self.logger.info(f"直接发送消息成功: {content[:30]}...") + + except Exception as e: + self.logger.error(f"直接发送消息失败: {str(e)}") + raise + diff --git a/src/plugins/PFC/pfc_KnowledgeFetcher.py b/src/plugins/PFC/pfc_KnowledgeFetcher.py new file mode 100644 index 000000000..560283f25 --- /dev/null +++ b/src/plugins/PFC/pfc_KnowledgeFetcher.py @@ -0,0 +1,54 @@ +from typing import List, Tuple +from src.common.logger import get_module_logger +from src.plugins.memory_system.Hippocampus import HippocampusManager +from ..models.utils_model import LLM_request +from ..config.config import global_config +from ..chat.message import Message + +logger = get_module_logger("knowledge_fetcher") + +class KnowledgeFetcher: + """知识调取器""" + + def __init__(self): + self.llm = LLM_request( + model=global_config.llm_normal, + temperature=0.7, + max_tokens=1000, + request_type="knowledge_fetch" + ) + + async def fetch(self, query: str, chat_history: List[Message]) -> Tuple[str, str]: + """获取相关知识 + + Args: + query: 查询内容 + chat_history: 聊天历史 + + Returns: + Tuple[str, str]: (获取的知识, 知识来源) + """ + # 构建查询上下文 + chat_history_text = "" + for msg in chat_history: + # sender = msg.message_info.user_info.user_nickname or f"用户{msg.message_info.user_info.user_id}" + chat_history_text += f"{msg.detailed_plain_text}\n" + + # 从记忆中获取相关知识 + related_memory = await HippocampusManager.get_instance().get_memory_from_text( + text=f"{query}\n{chat_history_text}", + max_memory_num=3, + max_memory_length=2, + max_depth=3, + fast_retrieval=False + ) + + if related_memory: + knowledge = "" + sources = [] + for memory in related_memory: + knowledge += memory[1] + "\n" + sources.append(f"记忆片段{memory[0]}") + return knowledge.strip(), "，".join(sources) + + return "未找到相关知识", "无记忆匹配" \ No newline at end of file diff --git a/src/plugins/PFC/reply_checker.py b/src/plugins/PFC/reply_checker.py new file mode 100644 index 000000000..3d8c743f2 --- /dev/null +++ b/src/plugins/PFC/reply_checker.py @@ -0,0 +1,141 @@ +import json +import datetime +from typing import Tuple +from src.common.logger import get_module_logger +from ..models.utils_model import LLM_request +from ..config.config import global_config +from .chat_observer import ChatObserver +from ..message.message_base import UserInfo + +logger = get_module_logger("reply_checker") + +class ReplyChecker: + """回复检查器""" + + def __init__(self, stream_id: str): + self.llm = LLM_request( + model=global_config.llm_normal, + temperature=0.7, + max_tokens=1000, + request_type="reply_check" + ) + self.name = global_config.BOT_NICKNAME + self.chat_observer = ChatObserver.get_instance(stream_id) + self.max_retries = 2 # 最大重试次数 + + async def check( + self, + reply: str, + goal: str, + retry_count: int = 0 + ) -> Tuple[bool, str, bool]: + """检查生成的回复是否合适 + + Args: + reply: 生成的回复 + goal: 对话目标 + retry_count: 当前重试次数 + + Returns: + Tuple[bool, str, bool]: (是否合适, 原因, 是否需要重新规划) + """ + # 获取最新的消息记录 + messages = self.chat_observer.get_message_history(limit=5) + chat_history_text = "" + for msg in messages: + time_str = datetime.datetime.fromtimestamp(msg["time"]).strftime("%H:%M:%S") + user_info = UserInfo.from_dict(msg.get("user_info", {})) + sender = user_info.user_nickname or f"用户{user_info.user_id}" + if sender == self.name: + sender = "你说" + chat_history_text += f"{time_str},{sender}:{msg.get('processed_plain_text', '')}\n" + + prompt = f"""请检查以下回复是否合适： + +当前对话目标：{goal} +最新的对话记录： +{chat_history_text} + +待检查的回复： +{reply} + +请检查以下几点： +1. 回复是否依然符合当前对话目标和实现方式 +2. 回复是否与最新的对话记录保持一致性 +3. 回复是否重复发言，重复表达 +4. 回复是否包含违法违规内容（政治敏感、暴力等） +5. 回复是否以你的角度发言,不要把"你"说的话当做对方说的话，这是你自己说的话 + +请以JSON格式输出，包含以下字段： +1. suitable: 是否合适 (true/false) +2. reason: 原因说明 +3. need_replan: 是否需要重新规划对话目标 (true/false)，当发现当前对话目标不再适合时设为true + +输出格式示例： +{{ + "suitable": true, + "reason": "回复符合要求，内容得体", + "need_replan": false +}} + +注意：请严格按照JSON格式输出，不要包含任何其他内容。""" + + try: + content, _ = await self.llm.generate_response_async(prompt) + logger.debug(f"检查回复的原始返回: {content}") + + # 清理内容，尝试提取JSON部分 + content = content.strip() + try: + # 尝试直接解析 + result = json.loads(content) + except json.JSONDecodeError: + # 如果直接解析失败，尝试查找和提取JSON部分 + import re + json_pattern = r'\{[^{}]*\}' + json_match = re.search(json_pattern, content) + if json_match: + try: + result = json.loads(json_match.group()) + except json.JSONDecodeError: + # 如果JSON解析失败，尝试从文本中提取结果 + is_suitable = "不合适" not in content.lower() and "违规" not in content.lower() + reason = content[:100] if content else "无法解析响应" + need_replan = "重新规划" in content.lower() or "目标不适合" in content.lower() + return is_suitable, reason, need_replan + else: + # 如果找不到JSON，从文本中判断 + is_suitable = "不合适" not in content.lower() and "违规" not in content.lower() + reason = content[:100] if content else "无法解析响应" + need_replan = "重新规划" in content.lower() or "目标不适合" in content.lower() + return is_suitable, reason, need_replan + + # 验证JSON字段 + suitable = result.get("suitable", None) + reason = result.get("reason", "未提供原因") + need_replan = result.get("need_replan", False) + + # 如果suitable字段是字符串，转换为布尔值 + if isinstance(suitable, str): + suitable = suitable.lower() == "true" + + # 如果suitable字段不存在或不是布尔值，从reason中判断 + if suitable is None: + suitable = "不合适" not in reason.lower() and "违规" not in reason.lower() + + # 如果不合适且未达到最大重试次数，返回需要重试 + if not suitable and retry_count < self.max_retries: + return False, reason, False + + # 如果不合适且已达到最大重试次数，返回需要重新规划 + if not suitable and retry_count >= self.max_retries: + return False, f"多次重试后仍不合适: {reason}", True + + return suitable, reason, need_replan + + except Exception as e: + logger.error(f"检查回复时出错: {e}") + # 如果出错且已达到最大重试次数，建议重新规划 + if retry_count >= self.max_retries: + return False, "多次检查失败，建议重新规划", True + return False, f"检查过程出错，建议重试: {str(e)}", False \ No newline at end of file diff --git a/src/plugins/chat/bot.py b/src/plugins/chat/bot.py index 53047f31e..9046198c9 100644 --- a/src/plugins/chat/bot.py +++ b/src/plugins/chat/bot.py @@ -1,14 +1,14 @@ - from ..moods.moods import MoodManager # 导入情绪管理器 from ..config.config import global_config -from ..chat_module.reasoning_chat.reasoning_generator import ResponseGenerator - - -from ..storage.storage import MessageStorage # 修改导入路径 +from .message import MessageRecv +from ..PFC.pfc import Conversation, ConversationState +from .chat_stream import chat_manager +from ..chat_module.only_process.only_message_process import MessageProcessor from src.common.logger import get_module_logger, CHAT_STYLE_CONFIG, LogConfig from ..chat_module.think_flow_chat.think_flow_chat import ThinkFlowChat from ..chat_module.reasoning_chat.reasoning_chat import ReasoningChat +import asyncio # 定义日志配置 chat_config = LogConfig( @@ -23,20 +23,33 @@ logger = get_module_logger("chat_bot", config=chat_config) class ChatBot: def __init__(self): - self.storage = MessageStorage() - self.gpt = ResponseGenerator() self.bot = None # bot 实例引用 self._started = False self.mood_manager = MoodManager.get_instance() # 获取情绪管理器单例 self.mood_manager.start_mood_update() # 启动情绪更新 self.think_flow_chat = ThinkFlowChat() self.reasoning_chat = ReasoningChat() + self.only_process_chat = MessageProcessor() async def _ensure_started(self): """确保所有任务已启动""" if not self._started: self._started = True + async def _create_PFC_chat(self, message: MessageRecv): + try: + chat_id = str(message.chat_stream.stream_id) + + if global_config.enable_pfc_chatting: + # 获取或创建对话实例 + conversation = Conversation.get_instance(chat_id) + # 如果是新创建的实例，启动对话系统 + if conversation.state == ConversationState.INIT: + asyncio.create_task(conversation.start()) + logger.info(f"为聊天 {chat_id} 创建新的对话实例") + except Exception as e: + logger.error(f"创建PFC聊天流失败: {e}") + async def message_process(self, message_data: str) -> None: """处理转化后的统一格式消息 根据global_config.response_mode选择不同的回复模式： @@ -50,7 +63,11 @@ class ChatBot: - 没有思维流相关的状态管理 - 更简单直接的回复逻辑 - 两种模式都包含： + 3. pfc_chatting模式：仅进行消息处理 + - 不进行任何回复 + - 只处理和存储消息 + + 所有模式都包含： - 消息过滤 - 记忆激活 - 意愿计算 @@ -58,13 +75,52 @@ class ChatBot: - 表情包处理 - 性能计时 """ + + message = MessageRecv(message_data) + groupinfo = message.message_info.group_info - if global_config.response_mode == "heart_flow": - await self.think_flow_chat.process_message(message_data) - elif global_config.response_mode == "reasoning": - await self.reasoning_chat.process_message(message_data) + if global_config.enable_pfc_chatting: + try: + if groupinfo is None and global_config.enable_friend_chat: + userinfo = message.message_info.user_info + messageinfo = message.message_info + # 创建聊天流 + chat = await chat_manager.get_or_create_stream( + platform=messageinfo.platform, + user_info=userinfo, + group_info=groupinfo, + ) + message.update_chat_stream(chat) + await self.only_process_chat.process_message(message) + await self._create_PFC_chat(message) + else: + if groupinfo.group_id in global_config.talk_allowed_groups: + if global_config.response_mode == "heart_flow": + await self.think_flow_chat.process_message(message_data) + elif global_config.response_mode == "reasoning": + await self.reasoning_chat.process_message(message_data) + else: + logger.error(f"未知的回复模式，请检查配置文件！！: {global_config.response_mode}") + except Exception as e: + logger.error(f"处理PFC消息失败: {e}") else: - logger.error(f"未知的回复模式，请检查配置文件！！: {global_config.response_mode}") + if groupinfo is None and global_config.enable_friend_chat: + # 私聊处理流程 + # await self._handle_private_chat(message) + if global_config.response_mode == "heart_flow": + await self.think_flow_chat.process_message(message_data) + elif global_config.response_mode == "reasoning": + await self.reasoning_chat.process_message(message_data) + else: + logger.error(f"未知的回复模式，请检查配置文件！！: {global_config.response_mode}") + else: # 群聊处理 + if groupinfo.group_id in global_config.talk_allowed_groups: + if global_config.response_mode == "heart_flow": + await self.think_flow_chat.process_message(message_data) + elif global_config.response_mode == "reasoning": + await self.reasoning_chat.process_message(message_data) + else: + logger.error(f"未知的回复模式，请检查配置文件！！: {global_config.response_mode}") # 创建全局ChatBot实例 diff --git a/src/plugins/chat/chat_stream.py b/src/plugins/chat/chat_stream.py index 32994ec48..8cddb9376 100644 --- a/src/plugins/chat/chat_stream.py +++ b/src/plugins/chat/chat_stream.py @@ -137,36 +137,40 @@ class ChatManager: ChatStream: 聊天流对象 """ # 生成stream_id - stream_id = self._generate_stream_id(platform, user_info, group_info) + try: + stream_id = self._generate_stream_id(platform, user_info, group_info) - # 检查内存中是否存在 - if stream_id in self.streams: - stream = self.streams[stream_id] - # 更新用户信息和群组信息 - stream.update_active_time() - stream = copy.deepcopy(stream) - stream.user_info = user_info - if group_info: - stream.group_info = group_info - return stream + # 检查内存中是否存在 + if stream_id in self.streams: + stream = self.streams[stream_id] + # 更新用户信息和群组信息 + stream.update_active_time() + stream = copy.deepcopy(stream) + stream.user_info = user_info + if group_info: + stream.group_info = group_info + return stream - # 检查数据库中是否存在 - data = db.chat_streams.find_one({"stream_id": stream_id}) - if data: - stream = ChatStream.from_dict(data) - # 更新用户信息和群组信息 - stream.user_info = user_info - if group_info: - stream.group_info = group_info - stream.update_active_time() - else: - # 创建新的聊天流 - stream = ChatStream( - stream_id=stream_id, - platform=platform, - user_info=user_info, - group_info=group_info, - ) + # 检查数据库中是否存在 + data = db.chat_streams.find_one({"stream_id": stream_id}) + if data: + stream = ChatStream.from_dict(data) + # 更新用户信息和群组信息 + stream.user_info = user_info + if group_info: + stream.group_info = group_info + stream.update_active_time() + else: + # 创建新的聊天流 + stream = ChatStream( + stream_id=stream_id, + platform=platform, + user_info=user_info, + group_info=group_info, + ) + except Exception as e: + logger.error(f"创建聊天流失败: {e}") + raise e # 保存到内存和数据库 self.streams[stream_id] = stream diff --git a/src/plugins/chat/utils_image.py b/src/plugins/chat/utils_image.py index 729c8e1f8..f19fedfdd 100644 --- a/src/plugins/chat/utils_image.py +++ b/src/plugins/chat/utils_image.py @@ -166,7 +166,7 @@ class ImageManager: # 查询缓存的描述 cached_description = self._get_description_from_db(image_hash, "image") if cached_description: - logger.info(f"图片描述缓存中 {cached_description}") + logger.debug(f"图片描述缓存中 {cached_description}") return f"[图片：{cached_description}]" # 调用AI获取描述 diff --git a/src/plugins/chat_module/only_process/only_message_process.py b/src/plugins/chat_module/only_process/only_message_process.py new file mode 100644 index 000000000..4c1e7d5e1 --- /dev/null +++ b/src/plugins/chat_module/only_process/only_message_process.py @@ -0,0 +1,66 @@ +from src.common.logger import get_module_logger +from src.plugins.chat.message import MessageRecv +from src.plugins.storage.storage import MessageStorage +from src.plugins.config.config import global_config +import re +from datetime import datetime + +logger = get_module_logger("pfc_message_processor") + +class MessageProcessor: + """消息处理器，负责处理接收到的消息并存储""" + + def __init__(self): + self.storage = MessageStorage() + + def _check_ban_words(self, text: str, chat, userinfo) -> bool: + """检查消息中是否包含过滤词""" + for word in global_config.ban_words: + if word in text: + logger.info( + f"[{chat.group_info.group_name if chat.group_info else '私聊'}]{userinfo.user_nickname}:{text}" + ) + logger.info(f"[过滤词识别]消息中含有{word}，filtered") + return True + return False + + def _check_ban_regex(self, text: str, chat, userinfo) -> bool: + """检查消息是否匹配过滤正则表达式""" + for pattern in global_config.ban_msgs_regex: + if re.search(pattern, text): + logger.info( + f"[{chat.group_info.group_name if chat.group_info else '私聊'}]{userinfo.user_nickname}:{text}" + ) + logger.info(f"[正则表达式过滤]消息匹配到{pattern}，filtered") + return True + return False + + async def process_message(self, message: MessageRecv) -> None: + """处理消息并存储 + + Args: + message: 消息对象 + """ + userinfo = message.message_info.user_info + chat = message.chat_stream + + # 处理消息 + await message.process() + + # 过滤词/正则表达式过滤 + if self._check_ban_words(message.processed_plain_text, chat, userinfo) or self._check_ban_regex( + message.raw_message, chat, userinfo + ): + return + + # 存储消息 + await self.storage.store_message(message, chat) + + # 打印消息信息 + mes_name = chat.group_info.group_name if chat.group_info else "私聊" + # 将时间戳转换为datetime对象 + current_time = datetime.fromtimestamp(message.message_info.time).strftime("%H:%M:%S") + logger.info( + f"[{current_time}][{mes_name}]" + f"{chat.user_info.user_nickname}: {message.processed_plain_text}" + ) \ No newline at end of file diff --git a/src/plugins/chat_module/reasoning_chat/reasoning_chat.py b/src/plugins/chat_module/reasoning_chat/reasoning_chat.py index 2a6561d83..0163a306e 100644 --- a/src/plugins/chat_module/reasoning_chat/reasoning_chat.py +++ b/src/plugins/chat_module/reasoning_chat/reasoning_chat.py @@ -143,11 +143,6 @@ class ReasoningChat: userinfo = message.message_info.user_info messageinfo = message.message_info - - if groupinfo == None and global_config.enable_friend_chat:#如果是私聊 - pass - elif groupinfo.group_id not in global_config.talk_allowed_groups: - return # logger.info("使用推理聊天模式") diff --git a/src/plugins/chat_module/think_flow_chat/think_flow_chat.py b/src/plugins/chat_module/think_flow_chat/think_flow_chat.py index 6cbde1bac..8197af0b1 100644 --- a/src/plugins/chat_module/think_flow_chat/think_flow_chat.py +++ b/src/plugins/chat_module/think_flow_chat/think_flow_chat.py @@ -155,10 +155,6 @@ class ThinkFlowChat: userinfo = message.message_info.user_info messageinfo = message.message_info - if groupinfo == None and global_config.enable_friend_chat:#如果是私聊 - pass - elif groupinfo.group_id not in global_config.talk_allowed_groups: - return # 创建聊天流 chat = await chat_manager.get_or_create_stream( @@ -188,16 +184,15 @@ class ThinkFlowChat: ) timer2 = time.time() timing_results["记忆激活"] = timer2 - timer1 + logger.debug(f"记忆激活: {interested_rate}") is_mentioned = is_mentioned_bot_in_message(message) # 计算回复意愿 - if global_config.enable_think_flow: - current_willing_old = willing_manager.get_willing(chat_stream=chat) - current_willing_new = (heartflow.get_subheartflow(chat.stream_id).current_state.willing - 5) / 4 - current_willing = (current_willing_old + current_willing_new) / 2 - else: - current_willing = willing_manager.get_willing(chat_stream=chat) + current_willing_old = willing_manager.get_willing(chat_stream=chat) + current_willing_new = (heartflow.get_subheartflow(chat.stream_id).current_state.willing - 5) / 4 + current_willing = (current_willing_old + current_willing_new) / 2 + willing_manager.set_willing(chat.stream_id, current_willing) @@ -213,6 +208,7 @@ class ThinkFlowChat: ) timer2 = time.time() timing_results["意愿激活"] = timer2 - timer1 + logger.debug(f"意愿激活: {reply_probability}") # 打印消息信息 mes_name = chat.group_info.group_name if chat.group_info else "私聊" diff --git a/src/plugins/config/config.py b/src/plugins/config/config.py index 338c140c2..6db225a4b 100644 --- a/src/plugins/config/config.py +++ b/src/plugins/config/config.py @@ -24,8 +24,8 @@ config_config = LogConfig( logger = get_module_logger("config", config=config_config) #考虑到，实际上配置文件中的mai_version是不会自动更新的,所以采用硬编码 -mai_version_main = "0.6.0" -mai_version_fix = "mmc-4" +mai_version_main = "test-0.6.0" +mai_version_fix = "snapshot-7" mai_version = f"{mai_version_main}-{mai_version_fix}" def update_config(): @@ -230,7 +230,8 @@ class BotConfig: # experimental enable_friend_chat: bool = False # 是否启用好友聊天 - enable_think_flow: bool = False # 是否启用思考流程 + # enable_think_flow: bool = False # 是否启用思考流程 + enable_pfc_chatting: bool = False # 是否启用PFC聊天 # 模型配置 llm_reasoning: Dict[str, str] = field(default_factory=lambda: {}) @@ -333,7 +334,7 @@ class BotConfig: personality_config = parent["personality"] personality = personality_config.get("prompt_personality") if len(personality) >= 2: - logger.debug(f"载入自定义人格:{personality}") + logger.info(f"载入自定义人格:{personality}") config.PROMPT_PERSONALITY = personality_config.get("prompt_personality", config.PROMPT_PERSONALITY) config.PERSONALITY_1 = personality_config.get("personality_1_probability", config.PERSONALITY_1) @@ -563,7 +564,9 @@ class BotConfig: def experimental(parent: dict): experimental_config = parent["experimental"] config.enable_friend_chat = experimental_config.get("enable_friend_chat", config.enable_friend_chat) - config.enable_think_flow = experimental_config.get("enable_think_flow", config.enable_think_flow) + # config.enable_think_flow = experimental_config.get("enable_think_flow", config.enable_think_flow) + if config.INNER_VERSION in SpecifierSet(">=1.1.0"): + config.enable_pfc_chatting = experimental_config.get("pfc_chatting", config.enable_pfc_chatting) # 版本表达式：>=1.0.0,<2.0.0 # 允许字段：func: method, support: str, notice: str, necessary: bool diff --git a/src/plugins/models/utils_model.py b/src/plugins/models/utils_model.py index 263e11618..260c5f5a6 100644 --- a/src/plugins/models/utils_model.py +++ b/src/plugins/models/utils_model.py @@ -198,156 +198,195 @@ class LLM_request: headers["Accept"] = "text/event-stream" async with aiohttp.ClientSession() as session: - async with session.post(api_url, headers=headers, json=payload) as response: - # 处理需要重试的状态码 - if response.status in policy["retry_codes"]: - wait_time = policy["base_wait"] * (2**retry) - logger.warning(f"错误码: {response.status}, 等待 {wait_time}秒后重试") - if response.status == 413: - logger.warning("请求体过大，尝试压缩...") - image_base64 = compress_base64_image_by_scale(image_base64) - payload = await self._build_payload(prompt, image_base64, image_format) - elif response.status in [500, 503]: - logger.error(f"错误码: {response.status} - {error_code_mapping.get(response.status)}") - raise RuntimeError("服务器负载过高，模型恢复失败QAQ") - else: - logger.warning(f"请求限制(429)，等待{wait_time}秒后重试...") - - await asyncio.sleep(wait_time) - continue - elif response.status in policy["abort_codes"]: - logger.error(f"错误码: {response.status} - {error_code_mapping.get(response.status)}") - # 尝试获取并记录服务器返回的详细错误信息 - try: - error_json = await response.json() - if error_json and isinstance(error_json, list) and len(error_json) > 0: - for error_item in error_json: - if "error" in error_item and isinstance(error_item["error"], dict): - error_obj = error_item["error"] - error_code = error_obj.get("code") - error_message = error_obj.get("message") - error_status = error_obj.get("status") - logger.error( - f"服务器错误详情: 代码={error_code}, 状态={error_status}, " - f"消息={error_message}" - ) - elif isinstance(error_json, dict) and "error" in error_json: - # 处理单个错误对象的情况 - error_obj = error_json.get("error", {}) - error_code = error_obj.get("code") - error_message = error_obj.get("message") - error_status = error_obj.get("status") - logger.error( - f"服务器错误详情: 代码={error_code}, 状态={error_status}, 消息={error_message}" - ) + try: + async with session.post(api_url, headers=headers, json=payload) as response: + # 处理需要重试的状态码 + if response.status in policy["retry_codes"]: + wait_time = policy["base_wait"] * (2**retry) + logger.warning(f"错误码: {response.status}, 等待 {wait_time}秒后重试") + if response.status == 413: + logger.warning("请求体过大，尝试压缩...") + image_base64 = compress_base64_image_by_scale(image_base64) + payload = await self._build_payload(prompt, image_base64, image_format) + elif response.status in [500, 503]: + logger.error(f"错误码: {response.status} - {error_code_mapping.get(response.status)}") + raise RuntimeError("服务器负载过高，模型恢复失败QAQ") else: - # 记录原始错误响应内容 - logger.error(f"服务器错误响应: {error_json}") - except Exception as e: - logger.warning(f"无法解析服务器错误响应: {str(e)}") + logger.warning(f"请求限制(429)，等待{wait_time}秒后重试...") - if response.status == 403: - # 只针对硅基流动的V3和R1进行降级处理 - if ( - self.model_name.startswith("Pro/deepseek-ai") - and self.base_url == "https://api.siliconflow.cn/v1/" - ): - old_model_name = self.model_name - self.model_name = self.model_name[4:] # 移除"Pro/"前缀 - logger.warning(f"检测到403错误，模型从 {old_model_name} 降级为 {self.model_name}") - - # 对全局配置进行更新 - if global_config.llm_normal.get("name") == old_model_name: - global_config.llm_normal["name"] = self.model_name - logger.warning(f"将全局配置中的 llm_normal 模型临时降级至{self.model_name}") - - if global_config.llm_reasoning.get("name") == old_model_name: - global_config.llm_reasoning["name"] = self.model_name - logger.warning(f"将全局配置中的 llm_reasoning 模型临时降级至{self.model_name}") - - # 更新payload中的模型名 - if payload and "model" in payload: - payload["model"] = self.model_name - - # 重新尝试请求 - retry -= 1 # 不计入重试次数 - continue - - raise RuntimeError(f"请求被拒绝: {error_code_mapping.get(response.status)}") - - response.raise_for_status() - reasoning_content = "" - - # 将流式输出转化为非流式输出 - if stream_mode: - flag_delta_content_finished = False - accumulated_content = "" - usage = None # 初始化usage变量，避免未定义错误 - - async for line_bytes in response.content: + await asyncio.sleep(wait_time) + continue + elif response.status in policy["abort_codes"]: + logger.error(f"错误码: {response.status} - {error_code_mapping.get(response.status)}") + # 尝试获取并记录服务器返回的详细错误信息 try: - line = line_bytes.decode("utf-8").strip() - if not line: + error_json = await response.json() + if error_json and isinstance(error_json, list) and len(error_json) > 0: + for error_item in error_json: + if "error" in error_item and isinstance(error_item["error"], dict): + error_obj = error_item["error"] + error_code = error_obj.get("code") + error_message = error_obj.get("message") + error_status = error_obj.get("status") + logger.error( + f"服务器错误详情: 代码={error_code}, 状态={error_status}, " + f"消息={error_message}" + ) + elif isinstance(error_json, dict) and "error" in error_json: + # 处理单个错误对象的情况 + error_obj = error_json.get("error", {}) + error_code = error_obj.get("code") + error_message = error_obj.get("message") + error_status = error_obj.get("status") + logger.error( + f"服务器错误详情: 代码={error_code}, 状态={error_status}, 消息={error_message}" + ) + else: + # 记录原始错误响应内容 + logger.error(f"服务器错误响应: {error_json}") + except Exception as e: + logger.warning(f"无法解析服务器错误响应: {str(e)}") + + if response.status == 403: + # 只针对硅基流动的V3和R1进行降级处理 + if ( + self.model_name.startswith("Pro/deepseek-ai") + and self.base_url == "https://api.siliconflow.cn/v1/" + ): + old_model_name = self.model_name + self.model_name = self.model_name[4:] # 移除"Pro/"前缀 + logger.warning(f"检测到403错误，模型从 {old_model_name} 降级为 {self.model_name}") + + # 对全局配置进行更新 + if global_config.llm_normal.get("name") == old_model_name: + global_config.llm_normal["name"] = self.model_name + logger.warning(f"将全局配置中的 llm_normal 模型临时降级至{self.model_name}") + + if global_config.llm_reasoning.get("name") == old_model_name: + global_config.llm_reasoning["name"] = self.model_name + logger.warning(f"将全局配置中的 llm_reasoning 模型临时降级至{self.model_name}") + + # 更新payload中的模型名 + if payload and "model" in payload: + payload["model"] = self.model_name + + # 重新尝试请求 + retry -= 1 # 不计入重试次数 continue - if line.startswith("data:"): - data_str = line[5:].strip() - if data_str == "[DONE]": - break - try: - chunk = json.loads(data_str) - if flag_delta_content_finished: - chunk_usage = chunk.get("usage", None) - if chunk_usage: - usage = chunk_usage # 获取token用量 - else: - delta = chunk["choices"][0]["delta"] - delta_content = delta.get("content") - if delta_content is None: - delta_content = "" - accumulated_content += delta_content - # 检测流式输出文本是否结束 - finish_reason = chunk["choices"][0].get("finish_reason") - if delta.get("reasoning_content", None): - reasoning_content += delta["reasoning_content"] - if finish_reason == "stop": + + raise RuntimeError(f"请求被拒绝: {error_code_mapping.get(response.status)}") + + response.raise_for_status() + reasoning_content = "" + + # 将流式输出转化为非流式输出 + if stream_mode: + flag_delta_content_finished = False + accumulated_content = "" + usage = None # 初始化usage变量，避免未定义错误 + + async for line_bytes in response.content: + try: + line = line_bytes.decode("utf-8").strip() + if not line: + continue + if line.startswith("data:"): + data_str = line[5:].strip() + if data_str == "[DONE]": + break + try: + chunk = json.loads(data_str) + if flag_delta_content_finished: chunk_usage = chunk.get("usage", None) if chunk_usage: - usage = chunk_usage - break - # 部分平台在文本输出结束前不会返回token用量，此时需要再获取一次chunk - flag_delta_content_finished = True + usage = chunk_usage # 获取token用量 + else: + delta = chunk["choices"][0]["delta"] + delta_content = delta.get("content") + if delta_content is None: + delta_content = "" + accumulated_content += delta_content + # 检测流式输出文本是否结束 + finish_reason = chunk["choices"][0].get("finish_reason") + if delta.get("reasoning_content", None): + reasoning_content += delta["reasoning_content"] + if finish_reason == "stop": + chunk_usage = chunk.get("usage", None) + if chunk_usage: + usage = chunk_usage + break + # 部分平台在文本输出结束前不会返回token用量，此时需要再获取一次chunk + flag_delta_content_finished = True - except Exception as e: - logger.exception(f"解析流式输出错误: {str(e)}") - except GeneratorExit: - logger.warning("流式输出被中断") - break - except Exception as e: - logger.error(f"处理流式输出时发生错误: {str(e)}") - break - content = accumulated_content - think_match = re.search(r"(.*?)", content, re.DOTALL) - if think_match: - reasoning_content = think_match.group(1).strip() - content = re.sub(r".*?", "", content, flags=re.DOTALL).strip() - # 构造一个伪result以便调用自定义响应处理器或默认处理器 - result = { - "choices": [{"message": {"content": content, "reasoning_content": reasoning_content}}], - "usage": usage, - } - return ( - response_handler(result) - if response_handler - else self._default_response_handler(result, user_id, request_type, endpoint) - ) + except Exception as e: + logger.exception(f"解析流式输出错误: {str(e)}") + except GeneratorExit: + logger.warning("流式输出被中断，正在清理资源...") + # 确保资源被正确清理 + await response.release() + # 返回已经累积的内容 + result = { + "choices": [{"message": {"content": accumulated_content, "reasoning_content": reasoning_content}}], + "usage": usage, + } + return ( + response_handler(result) + if response_handler + else self._default_response_handler(result, user_id, request_type, endpoint) + ) + except Exception as e: + logger.error(f"处理流式输出时发生错误: {str(e)}") + # 确保在发生错误时也能正确清理资源 + try: + await response.release() + except Exception as cleanup_error: + logger.error(f"清理资源时发生错误: {cleanup_error}") + # 返回已经累积的内容 + result = { + "choices": [{"message": {"content": accumulated_content, "reasoning_content": reasoning_content}}], + "usage": usage, + } + return ( + response_handler(result) + if response_handler + else self._default_response_handler(result, user_id, request_type, endpoint) + ) + content = accumulated_content + think_match = re.search(r"(.*?)", content, re.DOTALL) + if think_match: + reasoning_content = think_match.group(1).strip() + content = re.sub(r".*?", "", content, flags=re.DOTALL).strip() + # 构造一个伪result以便调用自定义响应处理器或默认处理器 + result = { + "choices": [{"message": {"content": content, "reasoning_content": reasoning_content}}], + "usage": usage, + } + return ( + response_handler(result) + if response_handler + else self._default_response_handler(result, user_id, request_type, endpoint) + ) + else: + result = await response.json() + # 使用自定义处理器或默认处理 + return ( + response_handler(result) + if response_handler + else self._default_response_handler(result, user_id, request_type, endpoint) + ) + + except (aiohttp.ClientError, asyncio.TimeoutError) as e: + if retry < policy["max_retries"] - 1: + wait_time = policy["base_wait"] * (2**retry) + logger.error(f"网络错误，等待{wait_time}秒后重试... 错误: {str(e)}") + await asyncio.sleep(wait_time) + continue else: - result = await response.json() - # 使用自定义处理器或默认处理 - return ( - response_handler(result) - if response_handler - else self._default_response_handler(result, user_id, request_type, endpoint) - ) + logger.critical(f"网络错误达到最大重试次数: {str(e)}") + raise RuntimeError(f"网络请求失败: {str(e)}") from e + except Exception as e: + logger.critical(f"未预期的错误: {str(e)}") + raise RuntimeError(f"请求过程中发生错误: {str(e)}") from e except aiohttp.ClientResponseError as e: # 处理aiohttp抛出的响应错误 diff --git a/template/bot_config_template.toml b/template/bot_config_template.toml index b9d39c682..2372b10b1 100644 --- a/template/bot_config_template.toml +++ b/template/bot_config_template.toml @@ -1,5 +1,5 @@ [inner] -version = "1.0.4" +version = "1.1.0" #以下是给开发人员阅读的，一般用户不需要阅读 @@ -149,6 +149,7 @@ enable = true [experimental] enable_friend_chat = false # 是否启用好友聊天 +pfc_chatting = false # 是否启用PFC聊天 #下面的模型若使用硅基流动则不需要更改，使用ds官方则改成.env自定义的宏，使用自定义模型则选择定位相似的模型自己填写 #推理模型