🛡 知识树3 · 判断逻辑(铁律 · 质量门 · 课程周期)

这些「判断逻辑」驱动 AI 是否回复、怎么回复。本页实时抽取源码,改动后刷新即同步。
🔴 身份铁律(一对一 · 防跨账号污染)
engine.py强身份注入:AI 只能以本账号/「我」自称,严禁回复中出现/让客户联系其他老师。
                f"\n\n【🔴 你的身份 — 绝对铁律】\n"
                f"你就是{_name_prefix}老师本人。你的账号名字是「{_name_prefix}老师」。\n"
                f"【铁律】客户说你的名字(如发「{_name_prefix}」「{_name_prefix}老师」等)是在跟你打招呼。\n"
                f"【铁律】绝对不要反过来叫客户为「{_name_prefix}老师」——你不是在跟{_name_prefix}老师说话,客户也不是{_name_prefix}老师。\n"
                f"【铁律】你不是其他老师,你就是{_name_prefix}老师本人。\n"
                f"【绝对铁律·防跨账号污染】你只能以「{_name_prefix}老师」或「我」自称,回复中严禁出现任何其他老师的名字(如小帆老师、明亮老师、明丽老师、小金老师、小威老师、朝阳老师等)。\n"
                f"【绝对铁律·防跨账号污染】严禁让客户「加/置顶/联系」除你以外的任何老师或账号;如果确需客户置顶或加微信,只能说「把我置顶」「加我微信」「联系我就行」,绝不能说「把小X老师置顶/加小X老师微信」等。\n"
                f"【绝对铁律·防跨账号污染】你仅代表{_name_prefix}老师这一个账号与客户沟通,禁止替其他账号传话、引流或做任何账号切换建议。"
            )
            system_content = identity_note + "\n\n" + system_content

        # 🔴 P4(2026-07-25) + P1语气: 系统时间绝对基准 + 语气铁律(注入system,覆盖所有调用路径)
        try:
            from datetime import datetime as _dt2, timedelta as _td2
            _now2 = _dt2.now()
            _wd2 = _now2.weekday()
            _wcn2 = ["一", "二", "三", "四", "五", "六", "日"][_wd2]
            _thu2 = _now2 + _td2(days=(3 - _wd2) % 7)
            _sat2 = _now2 + _td2(days=(5 - _wd2) % 7)
            system_content += (
                f"\n\n【系统时间·绝对基准】今日(系统真实日期)是:{_now2.year}年{_now2.month}月{_now2.day}日,"
                f"星期{_wcn2}(注意:今天就是周{_wcn2},绝不要说成其他星期,也不要说成周一)。"
                f"当前时间 {_now2.hour:02d}:{_now2.minute:02d}。"
                f"本周四={_thu2.month}月{_thu2.day}日(周四);本周六={_sat2.month}月{_sat2.day}日(周六)。"
                "课程表提到的「每周一/四」指的是未来的课程日,不是今天。客户说周几/几号时以系统时间为准推算,"
                "客户说错了要礼貌纠正。"
            )
        except Exception as _te2:
            try:
                from datetime import datetime as _dt3
                _n2 = _dt3.now()
                _w2 = ["一", "二", "三", "四", "五", "六", "日"][_n2.weekday()]
                system_content += (f"\n\n【系统时间·绝对基准】今日是星期{_w2}"
                                   f"({_n2.year}年{_n2.month}月{_n2.day}日),以系统真实时间为准,绝不要说成其他星期。")
            except Exception:
                system_content += "\n\n【系统时间·绝对基准】以系统真实时间为准,不要凭记忆臆测日期,绝不要说成周一。"
            print(f"  [时间注入] 异常兜底: {_te2}")
        # P1 语气铁律(覆盖所有调用路径,确保温而不火)
        system_content += ("\n\n【语气铁律】语气温而不火、专业克制:禁止以「哈哈/哈」等轻浮词开头,"
                           "禁止过度热情与情绪化,保持平实可信。")

        system_content += "\n\n重要规则:优先使用【知识库参考】中的内容回答。知识库中有明确规则时,以知识库为准,不要使用你自己的通用知识。如果知识库没有相关内容,才使用你自己的知识。"
        system_content += ("\n\n【回复优先级铁律·严禁推销越位】当客户是在提问(功法要领/健康顾虑/练习反馈/自身情况咨询)或分享练习感受时,你必须先用【知识库参考】认真回答这个真实问题,给出真诚、具体的建议。严禁在这种情况下主动发送报名链接、课程价格或任何推销内容。只有当客户明确表达报名/购买意向(如主动问\"怎么报名\"\"多少钱\"\"想学\"\"链接\")时,才提供报名链接。在客户提问时强行推销属于严重失误,必须避免。")

        # === 规则源注入(新增消费层;不改动知识树/蓝图既有读写与版本逻辑)===
🚧 质量门(quality_gate.py 检查项总览)
quality_gate.pyL0 静默标记硬闸口 + L1~L18 多层拦截,过滤危险/违规/越界回复。
# -*- coding: utf-8 -*-
"""
十八层质量门 — AI回复质量控制
=============================
遵循十方阵"多维度叠加"理念,每层独立评分,汇总决策。

L1  敏感词墙        — 绝对化医疗承诺、禁用词
L2  医疗承诺拦截     — 医疗建议/诊断/保证
L3  品牌一致性       — 是否提及和为合/无极太极
L4  编造价格检测     — 白名单外的价格数字
L5  编造断言模式     — 百分比、时间承诺、效果保证
L6  编造关键词密度   — 绝对化用词密度检测
L7  知识库交叉验证   — 断言与知识库匹配度
L8  幻觉指数评分     — 综合评分(0-100)
L9  结构&模板检测    — AI腔调、模板化开头
L10 长度控制         — 回复长度合规
L11 人工标记检查     — 是否被人工标记过
L12 课程阶段一致性   — 体验课阶段不得发正式课入学流程
L13 收费条件改写     — 新友<3天且提及正式课价格→按知识树改写为免费课话术
L14 不转别人         — 帮您对接/让助教联系/转给班主任→拦截
L15 退款铁律         — AI主动提退款/退费处理→拦截
L16 不承诺增强       — 软承诺(肯定管用/练了就好等)→拦截
L17 虚假承诺         — 我帮你安排/报名/去申请→拦截
L18 AI腔调增强       — 总的来说/最后我/首先其次最后结构→warn

决策:单层fail即阻断,warn叠加超阈值也阻断。
"""
import re
📅 课程周期 / 期数 判定(context_engine.py)
context_engine.py从企微备注推算「第几期第几天」,站桩/打坐双周期,严格按接收账号自己备注(禁跨账号污染)。
    def _select_best_remark_for_phase(self, course_type: str, cached_remark: str, _now=None) -> str:
        """企微同一客户可被多个员工跟进、备注不同。实时拉取所有 follow_user 备注,
        优先选择「当前正在上课(course_day 在 [1, total])」的期数; 无进行中则选最近未来期;
        仍然无解才 fallback 到 cached_remark。"""
        try:
            import time as _t
            from wework_api import get_customer_detail

            _key = f"{self.customer_id}:{course_type}"
            _now_ts = _t.time()
            _hit = self._FRESH_REMARK_CACHE.get(_key)
            if _hit and (_now_ts - _hit[0]) < self._FRESH_REMARK_TTL:
                return _hit[1]

            _detail = get_customer_detail(self.customer_id)
            if not isinstance(_detail, dict):
                self._FRESH_REMARK_CACHE[_key] = (_now_ts, cached_remark)
                return cached_remark
            _followers = _detail.get("follow_user") or []
            if not _followers:
                self._FRESH_REMARK_CACHE[_key] = (_now_ts, cached_remark)
                return cached_remark

            _remarks = []
            for _f in _followers:
                _r = (_f.get("remark") or "").strip()
                if _r and _r not in _remarks:
                    _remarks.append(_r)
            if not _remarks:
                self._FRESH_REMARK_CACHE[_key] = (_now_ts, cached_remark)
                return cached_remark

            # 如果只有一个备注,直接用它(同步实时)
            if len(_remarks) == 1:
                self._FRESH_REMARK_CACHE[_key] = (_now_ts, _remarks[0])
                return _remarks[0]

            # 多备注:逐个计算 course_day,打分选最优
            _best = None
            _best_score = None
            for _idx, _r in enumerate(_remarks):
                _parsed = self._calc_course_day_for_remark(_r, course_type, _now)
                if _parsed is None:
                    continue
                _batch, _day, _total, _is_future = _parsed
                # 优先级:进行中(1) > 未来期(0) > 其他(-1); 同优先级选 course_day 最接近现在
                _in_progress = 1 if (1 <= _day <= _total) else 0
                _future = 1 if _is_future else 0
                _score = (_in_progress, _future, -abs(_day), -_idx)
                if _best is None or _score > _best_score:
                    _best, _best_score = _r, _score

            if _best is not None:
                self._FRESH_REMARK_CACHE[_key] = (_now_ts, _best)
                return _best
            self._FRESH_REMARK_CACHE[_key] = (_now_ts, cached_remark)
            return cached_remark
        except Exception:
            return cached_remark

    def _calc_course_day_for_remark(self, remark: str, course_type: str, _now=None):
        """仅用于内部比较:给定 remark 和课型,返回 (batch, course_day, total, is_future) 或 None。"""
        import re as _re
        from datetime import datetime as _dt, date as _d, timedelta as _td
        _BASE_DZ = _d(2026, 8, 26)
        _BASE_ZZ = _d(2026, 8, 24)

        def _start_dz(_b):
            return _BASE_DZ + _td(days=(_b - 70) * 7)

        def _start_zz(_b):
            _k = _b - 254
            if _k >= 0:
                return _BASE_ZZ + _td(days=(_k // 2) * 7 + (3 if _k % 2 == 1 else 0))
            _m = -_k
            return _BASE_ZZ - _td(days=(_m // 2) * 7 + (4 if _m % 2 == 1 else 0))

        _ct = (course_type or "").strip()
        if _ct not in ("打坐", "站桩"):
            _ct = ""
        _batch = None
        _pref_m = _re.match(r'^(?:打坐|站桩)\s*', remark)
        if _pref_m:
            _pref = remark[:_pref_m.end()].strip()
            _ct = "打坐" if "打坐" in _pref else "站桩"
            _digits = ""
            for _c in remark[_pref_m.end():]:
                if _c.isdigit():
                    _digits += _c
                else: