1
llm:
2
  model_path: "./models/qwen/qwen2.5-1.5b-instruct-q4_k_m.gguf"
3
  backend: "llama.cpp"
4
  context_size: 2048        # 上下文窗口
5
  n_threads: 4              # GPU 模式下的 CPU 调度线程
6
  n_gpu_layers: -1          # -1 = 全部层放 GPU
7
  n_batch: 512
8
  temperature: 0.6
9
  top_p: 0.9
10
  max_tokens: 80            # 刻意压低：限制长度，避免过长回复和重复
11

12
asr:
13
  model_size: "small"       # tiny / small / base；small 的中文识别更稳
14
  language: "zh"
15
  audio_sample_rate: 16000
16

17
tts:
18
  voice: "huayan"
19
  sample_rate: 22050

1
def run(self):
2
    while self.running:
3
        # 1. 音频采集与 ASR
4
        asr_result = self.asr_service.listen_and_transcribe()
5
        if not asr_result or not asr_result.get("text"):
6
            continue
7

8
        user_text = asr_result.get("text", "")
9

10
        # 2. 检查唤醒词——没叫我就不处理
11
        if not self._check_wake_word(user_text):
12
            continue
13

14
        # 移除唤醒词，只把有效内容交给下游
15
        user_text_clean = self._remove_wake_word(user_text)
16
        if not user_text_clean.strip():
17
            self.tts_service.speak("我在，请说")
18
            continue
19

20
        asr_result["text"] = user_text_clean
21

22
        # 3. 对话处理
23
        result = self.dialogue_manager.process_input(asr_result)
24

25
        # 4. TTS 输出
26
        if result.get("need_tts") and result.get("response"):
27
            self.tts_service.speak(result["response"])

1
def _check_wake_word(self, text: str) -> bool:
2
    wake_variants = [self.wake_word.lower(), "maya", "玛雅", "麦娅", "may a"]
3
    return any(variant in text.lower() for variant in wake_variants)

1
def route_intent(self, asr_text) -> IntentObject:
2
    text = asr_text.get("text", "").strip() if isinstance(asr_text, dict) else str(asr_text).strip()
3

4
    if not text:
5
        return IntentObject(self.FALLBACK_TO_LLM, 0.0, {}, text)
6

7
    # 1. 唤醒词
8
    if self._check_wake_up(text):
9
        return IntentObject(self.WAKE_UP, 0.95, {}, text)
10

11
    # 2. 设备控制
12
    device_control = self._check_device_control(text)
13
    if device_control:
14
        return device_control
15

16
    # 3. 系统控制（先排除「进入 ROOT 模式」这类请求，避免误分类）
17
    root_keywords = ["进入root", "开启root", "激活root", "root模式", "管理员模式"]
18
    is_root_request = any(kw in text.lower() for kw in root_keywords)
19
    if not is_root_request:
20
        system_control = self._check_system_control(text)
21
        if system_control:
22
            return system_control
23

24
    # 4~7：设备查询 → 记忆写入 → 记忆查询 → 普通对话
25
    ...

1
def _extract_device_slots(self, text: str) -> Dict[str, Any]:
2
    slots = {"room": None, "device": None, "action": None, "value": None}
3

4
    for room in self.ROOMS:              # 客厅 / 卧室 / 厨房 / 卫生间 / 阳台 / 书房
5
        if room in text:
6
            slots["room"] = room
7
            break
8

9
    for device in self.DEVICE_NAMES:     # 灯 / 风扇 / 空调 / 窗帘 / 插座 / 电视 / 音响
10
        if device in text:
11
            slots["device"] = device
12
            break
13

14
    if "打开" in text or "开启" in text:
15
        slots["action"] = "ON"
16
    elif "关闭" in text:
17
        slots["action"] = "OFF"
18
    elif "调高" in text:
19
        slots["action"] = "UP"
20
    elif "调低" in text:
21
        slots["action"] = "DOWN"
22

23
    m = re.search(r'(\d+)', text)        # 「调到 26 度」
24
    if m:
25
        slots["value"] = int(m.group(1))
26

27
    return slots

1
CREATE TABLE memory (
2
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
3
    memory_type TEXT NOT NULL,          -- USER_HABIT / DEVICE_INFO / USER_STATE / SYSTEM_EVENT
4
    content     TEXT NOT NULL,          -- 标准化后的记忆文本
5
    importance  INTEGER DEFAULT 1,      -- 1~5，越大越重要
6
    embedding   BLOB,                   -- 向量（可选，用于语义检索）
7
    created_at  DATETIME DEFAULT CURRENT_TIMESTAMP,
8
    last_access DATETIME
9
);

1
SELECT content FROM memory
2
WHERE memory_type IN ('USER_HABIT', 'USER_STATE')
3
ORDER BY importance DESC, last_access DESC
4
LIMIT 5;

1
# 【优先级1】检查是否需要 ROOT 模式（必须在意图路由之前）
2
needs_privacy = self.root_mode_manager.check_needs_privacy(user_text)
3
requests_root = self.root_mode_manager.check_request_root_mode(user_text)
4

5
# 需要隐私但 ROOT 未激活 → 先去要口令
6
if (needs_privacy or requests_root) and not self.root_mode_manager.is_active():
7
    return self._handle_root_mode_request(user_text, needs_privacy, requests_root)
8

9
# ROOT 已激活且需要隐私 → 强制走普通对话分支（那里会读隐私并注入 Prompt）
10
if self.root_mode_manager.is_active() and needs_privacy:
11
    intent = IntentObject(intent_type=self.intent_router.GENERAL_CHAT, ...)
12
    return self._handle_general_chat(intent, user_text)

1
class RootModeManager:
2
    # ROOT 模式口令（真实值在源码常量里，这里以占位符示意）
3
    ROOT_PASSWORD = "••••"
4

5
    def verify_password(self, password: str) -> bool:
6
        """支持多种格式：纯数字、中文数字、以及 ASR 听起来像的变体"""
7
        password_clean = password.strip().replace(" ", "").replace("，", "").replace(",", "")
8

9
        if password_clean == self.ROOT_PASSWORD:
10
            return True
11

12
        # 「二零三九」这类中文念法要能转回数字——语音输入经常吐这种
13
        chinese_to_digit = {"零": "0", "一": "1", "二": "2", "三": "3", ...}
14
        if all(c in chinese_to_digit for c in password_clean):
15
            converted = "".join(chinese_to_digit.get(c, c) for c in password_clean)
16
            if converted == self.ROOT_PASSWORD:
17
                return True
18

19
        return False