SimplifiedChatEngine/answer.py at main · QQPilotOrganization/SimplifiedChatEngine · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
"""answer.py：simplified_chat_engine 主入口。

GetAnswer(text: list[ChatContent], systemPrompt='auto') -> Optional[str]
    返回回复文本；返回 None 表示"不需要回复"（由调用方决定等待超时或直接跳过）。

流水线（各单元已拆分到独立文件）：
1. eco_reasoning      —— 判断是否需要回复（规则层 + 模型双通道）
2. clumsy_imitation   —— 优化 system 提示词
3. memory_engine      —— 记忆（模型总结关键词 / 检索注入 / 存储）
4. messages           —— ChatContent → OpenAI 兼容 messages（含图片收集）
5. api_client         —— 模型调用（OpenAI 兼容端点 + Ollama 原生 /api/chat）
所有提示词模板集中在 defaultPattern.py。
"""

from pathlib import Path
from typing import List, Optional

from colorama import Fore

import config as engine_config
import defaultPattern
from api_client import _chat
from chatContent import ChatContent
from clumsy_imitation import _improve_system
from eco_reasoning import _should_reply
from memory_engine import InjectMemoryIntoSystem, _store_memory, _summarize
from messages import _build_messages, _collect_image_urls, _group_supports_vision

# 兼容旧版导入方式（参考 QQPilot/unused/answer.py 的用法；输出上限现由 config.xml <parameters> 控制）
MAX_LENGTH = 2048


# ---------- Xreply：正式回复 ----------

def _resolve_system(systemPrompt: str) -> str:
    if systemPrompt != "auto":
        if systemPrompt in ("", "None"):
            return ""
        return systemPrompt
    # auto：优先读 system.txt（与 QQPilot 行为一致）
    p = Path("system.txt")
    if p.exists():
        return p.read_text(encoding="utf-8").strip()
    return defaultPattern.patternXreplyFallback.strip()


def _xreply(contents: List[ChatContent], system: str) -> Optional[str]:
    cfg = engine_config.get_config()
    group = cfg.group(engine_config.GROUP_XREPLY)
    # 只有该组宣称支持图片时才收集图片（避免非视觉模型收到多模态消息被 400）
    image_urls = (_collect_image_urls(contents, cfg.max_image_count)
                  if _group_supports_vision(group) else [])
    messages = _build_messages(contents, system, image_urls)
    print(f"{Fore.LIGHTYELLOW_EX}[Xreply] 开始回复{Fore.RESET}")
    # print(system)
    # print("-")
    return _chat(
        engine_config.GROUP_XREPLY,
        messages,
        max_tokens=cfg.max_reply_tokens,
    )


# ---------- 主入口 ----------

def GetAnswer(text: List[ChatContent], systemPrompt: str = 'auto') -> Optional[str]:
    """聊天引擎主流程。

    Args:
        text: 群聊消息列表（ChatContent）。
        systemPrompt: 'auto' → 读取 system.txt（或内置兜底模板）；其他字符串直接作为
                      system 提示词；'' / 'None' → 无 system。

    Returns:
        需要回复时的回复文本；不需要回复（或出错）返回 None。
    """
    if not text:
        return ""

    # 1. 解析 system
    system = _resolve_system(systemPrompt)

    # 2. EcoReasoning：不需要回复 → None（HTTP 中转站会据此拖到 QQPilot 超时）；
    #    mustAnswer=true 时强制跳过判断，永远进入回复流程
    cfg = engine_config.get_config()
    if not cfg.must_answer and not _should_reply(text):
        return None

    # 3. clumsyImitation：优化 system（失败沿用原 system）
    if system and cfg.group(engine_config.GROUP_CLUMSY_IMITATION):
        improved = _improve_system(system, text)
        if improved:
            system = improved

    # 4. forgetfulMem：模型总结本轮关键词 → 检索记忆注入 system
    keywords = _summarize(text)
    system = InjectMemoryIntoSystem(system, text, keywords)

    # 5. Xreply：生成回复
    reply = _xreply(text, system)

    # 6. forgetfulMem：存储本轮对话与关键词（无论回复成功与否）
    _store_memory(system, text, reply, keywords)

    return reply


def get_answer_as_string(text: str, system_prompt: str) -> Optional[str]:
    """旧版兼容：直接传字符串。"""
    return GetAnswer(
        [ChatContent(username='', imagePaths=[], text=text, time='', ownByMyself=False)],
        system_prompt)


# ---------- 自测 ----------

if __name__ == "__main__":
    import time
    c1 = ChatContent(username='Username1', imagePaths=[], text='你好呀', time='08-10 17:19:57', ownByMyself=False)
    c2 = ChatContent(username='neko', imagePaths=[], text='喵～你好', time='08-10 17:20:00', ownByMyself=True)
    c3 = ChatContent(username='Username2', imagePaths=[], text='现在几点了', time='08-10 17:20:05', ownByMyself=False)
    t0 = time.time()
    ans = GetAnswer([c1, c2, c3], systemPrompt='')
    print(f"\n>>> 回复: {ans!r} (用时 {time.time()-t0:.1f}s)")