# -*- coding: utf-8 -*- """COMPACT.PY — tóm gọn cuộc tư vấn GIỮA phiên, rồi tư vấn tiếp trên bản tóm gọn. VÌ SAO ------ Chủ dự án (23/09/2026): "sau mỗi khoảng hội thoại, cần gọi để tóm tắt lại phiên như dạng compact lại, sau đó lại tiếp tục dùng compact này và các hội thoại mới để truyền và tư vấn tiếp. Như thế có thể sẽ hiệu quả về tư duy hơn." Trước đây mô hình nhận N lượt nguyên văn gần nhất. Lượt cũ hơn thì rơi mất, lượt còn lại thì đầy câu trả lời dài của chính bot. Người tư vấn thật không nhớ nguyên văn — họ nhớ TÌNH TRẠNG: khách muốn gì, đã làm tới đâu, còn kẹt gì. Bản tóm gọn ghi đúng cái đó. CÁCH CHẠY --------- - Sau mỗi lượt, nếu đã có >= MOI_LUOT lượt chưa nằm trong bản tóm gọn, bật một luồng nền viết lại bản tóm gọn (bản cũ + các lượt mới). Khách KHÔNG phải chờ — câu trả lời đã gửi đi trước đó. - Lượt sau, mô hình nhận: bản tóm gọn + các lượt sau lần tóm gọn (tối thiểu GIU_NGUYEN_VAN lượt gần nhất nguyên văn, để giữ giọng hội thoại). - Tóm gọn hỏng (mạng, mô hình, CSDL) thì chỉ là chưa có bản mới; lượt sau vẫn chạy bằng lịch sử nguyên văn như cũ. Chạy tự kiểm (không cần CSDL, không gọi mạng): python nao/compact.py """ from __future__ import annotations import logging import os import sys import threading from typing import Any, Callable, Dict, List, Optional, Tuple try: from giao_uoc import GIU_LICH_SU from congcu.so_tay import thanh_van_ban, che_du_lieu_nhay_cam from congcu.phien import luu_compact except ImportError: _THU_MUC_CHA = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if _THU_MUC_CHA not in sys.path: sys.path.insert(0, _THU_MUC_CHA) from giao_uoc import GIU_LICH_SU # type: ignore from congcu.so_tay import thanh_van_ban, che_du_lieu_nhay_cam # type: ignore from congcu.phien import luu_compact # type: ignore _log = logging.getLogger('chatbot.nao.compact') MOI_LUOT = int(os.environ.get('CHATBOT_COMPACT_MOI_LUOT', '4') or 4) GIU_NGUYEN_VAN = 2 # ===================================================================== # LỊCH SỬ ĐƯA VÀO PROMPT # ===================================================================== def lich_su_cho_prompt(lich_su_day_du: List[Dict[str, str]], so_luot: int, compact_den_luot: int, compact: str) -> List[Dict[str, str]]: """Chưa có bản tóm gọn: GIU_LICH_SU lượt gần nhất (như trước). Có bản tóm gọn: các lượt SAU lần tóm gọn, tối thiểu GIU_NGUYEN_VAN lượt. `so_luot` = số lượt đã lưu TRƯỚC lượt hiện tại.""" ls = list(lich_su_day_du or []) if not compact: return ls[-(GIU_LICH_SU * 2):] moi = max(so_luot - compact_den_luot, GIU_NGUYEN_VAN) return ls[-(moi * 2):] def can_tom_gon(so_luot_sau: int, compact_den_luot: int) -> bool: return so_luot_sau - compact_den_luot >= MOI_LUOT # ===================================================================== # PROMPT TÓM GỌN # ===================================================================== _CAU_DAN = ( "Em là chuyên gia tư vấn của {shop}, đang giữa một cuộc tư vấn. " "Em tự ghi lại TÌNH TRẠNG cuộc tư vấn để tư vấn tiếp mà không phải đọc lại từng câu. " "Viết lại bản tóm gọn: gộp bản cũ (nếu có) với các lượt mới, điều mới thay điều cũ nếu " "mâu thuẫn. Đủ chi tiết để một chuyên gia khác đọc là tư vấn tiếp được ngay.\n" "\n" "Đúng các mục sau, mục nào chưa có thì ghi \"—\":\n" "MỤC TIÊU CỦA KHÁCH: khách đang muốn đạt được gì.\n" "ĐÃ BIẾT VỀ KHÁCH: sản phẩm, MT4/MT5, sàn, loại tài khoản, kinh nghiệm, vốn… — chỉ điều " "khách đã nói.\n" "ĐÃ TƯ VẤN: đã hướng dẫn những gì, khách đã làm tới bước nào, kết quả ra sao.\n" "VẤN ĐỀ CÒN MỞ: điều chưa giải quyết, lỗi còn đó, băn khoăn của khách.\n" "EM ĐANG CHỜ KHÁCH: câu em đã hỏi mà khách chưa trả lời (nếu có).\n" "BƯỚC TIẾP THEO: em nên làm gì ở lượt sau.\n" "\n" "Luật: chỉ ghi điều có trong hội thoại, không đoán. KHÔNG ghi key, mật khẩu, email, số " "điện thoại, số tài khoản ('[ẩn]' thì bỏ qua). KHÔNG ghi giá. Viết tiếng Việt, gạch ý ngắn." ) def dung_prompt(compact_cu: str, cac_luot_moi: List[Dict[str, str]], so_tay: Optional[Dict[str, Any]], khong_gian: Optional[str] = None) -> Tuple[str, str]: phan = [ 'BẢN TÓM GỌN CŨ:\n' + (compact_cu.strip() if compact_cu and compact_cu.strip() else '(chưa có)') ] st = thanh_van_ban(so_tay) if st: phan.append('SỔ TAY TỰ ĐỘNG:\n' + st) dong = [] for l in cac_luot_moi: vai = 'Khách' if l.get('vai') == 'khach' else 'Em' dong.append(f"{vai}: {che_du_lieu_nhay_cam(l.get('noi_dung') or '')}") phan.append('CÁC LƯỢT MỚI:\n' + '\n'.join(dong)) return _CAU_DAN.replace('{shop}', _ho_so(khong_gian).vai_ghi_chep or _ho_so(khong_gian).ten), '\n\n'.join(phan) def _ho_so(khong_gian): """Hồ sơ shop (shop/) — phần riêng của từng shop.""" import shop return shop.lay(khong_gian) # ===================================================================== # CHẠY # ===================================================================== def _goi_mo_hinh(he_thong: str, nguoi_dung: str, vai: str = 'tong_hop') -> Optional[str]: try: from nao.tra_loi import _lay_nha_cung_cap_dung_chung ncc = _lay_nha_cung_cap_dung_chung() if ncc is None or getattr(ncc, 'ten', '') == 'rong': return None from congcu.cau_hinh_ai import mo_hinh van_ban = ncc.tom_tat(he_thong, nguoi_dung, mo_hinh=mo_hinh(vai) or None) try: from bao_ve.ngan_sach import ghi_nhan tv = getattr(ncc, 'token_vao_gan_nhat', None) or 0 tr = getattr(ncc, 'token_ra_gan_nhat', None) or 0 if tv or tr: ghi_nhan(int(tv), int(tr), getattr(ncc, 'model_da_gui_gan_nhat', None) or '') except Exception: # noqa: BLE001 _log.exception('Không ghi được chi phí tóm gọn vào ngân sách.') return van_ban except Exception as loi: # noqa: BLE001 _log.warning('Tóm gọn phiên lỗi: %s', loi) return None def tom_gon( phien: str, compact_cu: str, cac_luot_moi: List[Dict[str, str]], so_tay: Optional[Dict[str, Any]], den_luot: int, *, khong_gian: Optional[str] = None, _goi: Optional[Callable[[str, str], Optional[str]]] = None, _luu: Optional[Callable[..., bool]] = None, ) -> Optional[str]: """Viết bản tóm gọn mới và lưu. Trả bản mới, hoặc None nếu không làm được.""" if not cac_luot_moi: return None he_thong, nguoi_dung = dung_prompt(compact_cu, cac_luot_moi, so_tay, khong_gian) moi = ((_goi or _goi_mo_hinh)(he_thong, nguoi_dung) or '').strip() if not moi: return None moi = che_du_lieu_nhay_cam(moi) (_luu or luu_compact)(phien, moi, den_luot) _log.info('Đã tóm gọn phiên %s tới lượt %d (%d ký tự).', phien, den_luot, len(moi)) return moi _dang_chay: set = set() _khoa = threading.Lock() def kich_hoat_nen(phien: str, compact_cu: str, lich_su_day_du: List[Dict[str, str]], so_tay: Optional[Dict[str, Any]], so_luot_sau: int, compact_den_luot: int, khong_gian: Optional[str] = None) -> bool: """Nếu đủ lượt mới thì tóm gọn trong luồng nền. True nếu đã bật luồng. Mỗi phiên chỉ một luồng tóm gọn chạy cùng lúc; luồng thứ hai bị bỏ qua (lượt sau sẽ kích hoạt lại, và luu_compact không cho bản cũ đè bản mới).""" if os.environ.get('CHATBOT_COMPACT', '1').strip() == '0': return False if not phien or not can_tom_gon(so_luot_sau, compact_den_luot): return False so_moi = so_luot_sau - compact_den_luot cac_luot_moi = list(lich_su_day_du or [])[-(so_moi * 2):] with _khoa: if phien in _dang_chay: return False _dang_chay.add(phien) def _chay(): try: tom_gon(phien, compact_cu, cac_luot_moi, so_tay, so_luot_sau, khong_gian=khong_gian) except Exception: # noqa: BLE001 - luồng nền không bao giờ được làm sập cổng _log.exception('Tóm gọn phiên %s lỗi bất ngờ.', phien) finally: with _khoa: _dang_chay.discard(phien) threading.Thread(target=_chay, name=f'compact-{phien[:16]}', daemon=True).start() return True # ===================================================================== # TỰ KIỂM # ===================================================================== if __name__ == '__main__': def _ls(n): out = [] for i in range(1, n + 1): out += [{'vai': 'khach', 'noi_dung': f'hỏi {i}'}, {'vai': 'bot', 'noi_dung': f'đáp {i}'}] return out print('=== 1) lich_su_cho_prompt ===') ls10 = _ls(10) r = lich_su_cho_prompt(ls10, 10, 0, '') assert len(r) == GIU_LICH_SU * 2 and r[-1]['noi_dung'] == 'đáp 10', r r = lich_su_cho_prompt(ls10, 10, 7, 'tóm gọn') assert [x['noi_dung'] for x in r if x['vai'] == 'khach'] == ['hỏi 8', 'hỏi 9', 'hỏi 10'], r r = lich_su_cho_prompt(ls10, 10, 10, 'tóm gọn') assert [x['noi_dung'] for x in r if x['vai'] == 'khach'] == ['hỏi 9', 'hỏi 10'], \ 'vừa tóm gọn xong vẫn phải giữ GIU_NGUYEN_VAN lượt nguyên văn' print(' OK') print('\n=== 2) can_tom_gon ===') assert not can_tom_gon(3, 0) and can_tom_gon(4, 0) and not can_tom_gon(7, 4) and can_tom_gon(8, 4) print(f' OK — mỗi {MOI_LUOT} lượt') print('\n=== 3) tom_gon: gửi bản cũ + lượt mới, che dữ liệu, lưu đúng mốc ===') gui, luu = {}, {} def _goi_gia(h, n): gui['n'] = n return 'MỤC TIÊU CỦA KHÁCH: chạy copytrade\nĐÃ BIẾT VỀ KHÁCH: MT4, email a@b.com' kq = tom_gon('p1', 'bản cũ', [{'vai': 'khach', 'noi_dung': 'sdt 0912345678'}], {'nen_tang': 'MT4'}, 8, _goi=_goi_gia, _luu=lambda p, c, d: luu.update(p=p, c=c, d=d) or True) assert 'bản cũ' in gui['n'] and '0912345678' not in gui['n'], gui['n'] assert luu['d'] == 8 and 'a@b.com' not in luu['c'] and 'copytrade' in luu['c'], luu print(' OK') print('\n=== 4) mô hình hỏng -> None, không lưu ===') luu.clear() assert tom_gon('p1', '', [{'vai': 'khach', 'noi_dung': 'x'}], None, 4, _goi=lambda h, n: None, _luu=lambda *a: luu.update(x=1)) is None and not luu assert tom_gon('p1', '', [], None, 4, _goi=_goi_gia, _luu=lambda *a: True) is None print(' OK') print('\nTỰ KIỂM compact.py QUA.')