# -*- coding: utf-8 -*- """Tầng tìm kiếm LAI (hybrid) — nơi phần còn lại của hệ (nao/, llm/) gọi vào. VÌ SAO "lai" chứ không chỉ BM25: khách hỏi bằng cả từ khoá kỹ thuật nguyên văn (GOLDVIP6B, WebRequest) lẫn câu diễn giải tự nhiên ("EA không vào lệnh", "cài sao"). BM25 ăn chắc vế đầu, vector embedding ăn chắc vế sau. Giai đoạn 1 của dự án chưa có vector (xem giao_uoc.py mục 1) nên module này PHẢI chạy tốt chỉ với BM25 — vector chỉ là cộng thêm khi có, không phải điều kiện để hoạt động. """ from __future__ import annotations import os import sys from collections import defaultdict from typing import Any, Callable, Dict, List, Optional, Set, Tuple # --------------------------------------------------------------------- # Nhập giao ước + bm25: chịu được cả chạy như package (nao/ gọi # `from ..tim.lai import ...`) lẫn chạy trực tiếp `python tim/lai.py` # (không có package cha nên relative import vỡ). kho/ đang được viết # song song nên KHÔNG import kho ở đây — chỉ import trễ, bên trong hàm, # lúc thực sự cần nạp dữ liệu. # --------------------------------------------------------------------- try: from ..giao_uoc import Doan, KetQuaTim, LocTim, SO_DOAN_MAC_DINH, SO_DOAN_TOI_DA from .bm25 import BM25, tach_tu except ImportError: # Phải thêm CẢ HAI thư mục, không chỉ thư mục cha. # # Khi ai đó gọi `from tim.lai import lay_bo_tim` thì `tim` là gói cấp # cao nhất, không có gói cha, nên `from ..giao_uoc` gãy và rơi xuống # đây — nhưng lúc đó thư mục `tim/` KHÔNG nằm trong sys.path, nên # `import bm25` trần cũng gãy nốt. Chạy thẳng `python tim/lai.py` thì # lại không lộ ra, vì Python tự đặt `tim/` vào sys.path[0]. # # Nghĩa là lỗi này chỉ hiện khi ghép vào hệ thật. Đã dính một lần. _THU_MUC_NAY = os.path.dirname(os.path.abspath(__file__)) _THU_MUC_CHA = os.path.dirname(_THU_MUC_NAY) for _d in (_THU_MUC_CHA, _THU_MUC_NAY): if _d not in sys.path: sys.path.insert(0, _d) from giao_uoc import Doan, KetQuaTim, LocTim, SO_DOAN_MAC_DINH, SO_DOAN_TOI_DA from bm25 import BM25, tach_tu def _nap_doan_tu_kho(duong_dan: Optional[str] = None) -> List[Doan]: """Nạp toàn bộ Doan từ kho.chi_muc (import TRỄ và có chủ đích). VÌ SAO import trễ: kho/ đang được viết song song bởi người/phiên khác. Nếu import kho ở đầu tệp thì CHỈ import tim/ thôi (vd để chạy test BM25 thuần) cũng vỡ theo, dù không liên quan gì đến SQLite. """ try: from ..kho.chi_muc import doc_tat_ca except ImportError: try: from kho.chi_muc import doc_tat_ca except ImportError as loi: raise RuntimeError( "Chưa có kho/chi_muc.py (hoặc chưa build được), nên " "tim/ không có gì để nạp. Chạy `python kho/chi_muc.py` " "trước rồi thử lại." ) from loi try: return doc_tat_ca(duong_dan) except Exception as loi: # noqa: BLE001 — kho/ có thể ném FileNotFoundError, # sqlite3.OperationalError... đều quy về CÙNG một thông điệp tiếng # Việt dễ hiểu cho người vận hành, thay vì để traceback lạ hoắc. raise RuntimeError( "Không đọc được chỉ mục (kho/chi_muc.sqlite có thể chưa được " "build). Chạy `python kho/chi_muc.py` trước rồi thử lại. " f"Lỗi gốc: {loi!r}" ) from loi # Hạng dùng cho Reciprocal Rank Fusion — hằng số chuẩn trong bài báo gốc # (Cormack, Clarke & Buettcher 2009). k=60 làm phẳng chênh lệch giữa # hạng 1 và hạng 2 của MỘT nguồn, tránh việc một nguồn tự tin thái quá # (khoảng cách điểm hạng 1↔2 rất lớn) lấn át nguồn còn lại chỉ vì thang # điểm nội bộ của nó dốc hơn. _K_RRF = 60 class BoTim: """Tầng tìm kiếm lai — dựng chỉ mục MỘT LẦN, dùng lại nhiều lần. VÌ SAO nạp một lần: dựng BM25 (đếm tần suất + idf trên toàn bộ vài trăm đến vài nghìn đoạn) là phần tốn kém nhất của cả pipeline tìm kiếm. Làm lại việc này ở MỖI câu hỏi của khách thì độ trễ trả lời tăng tuyến tính theo kích thước kho một cách vô ích, trong khi bản thân truy vấn chỉ là vài lần tra dict. Dùng `lay_bo_tim()` để cả tiến trình chỉ trả giá này đúng một lần. """ def __init__(self, cac_doan: Optional[List[Doan]] = None, ham_embed_cau_hoi: Optional[Callable[[str], Any]] = None): self.cac_doan: List[Doan] = ( cac_doan if cac_doan is not None else _nap_doan_tu_kho() ) # Hook để GIAI ĐOẠN SAU (khi llm/ có nhà cung cấp embedding) tiêm # hàm biến câu hỏi -> vector CÙNG không gian với cột `vector` # của đoạn. Giai đoạn 1 không nhà cung cấp nào gắn vào đây nên # mặc định None — nhánh vector im lặng bỏ qua, KHÔNG lỗi. tim/ # cố tình không tự gọi API embedding: gọi nhà cung cấp AI không # phải việc của tầng tìm kiếm (xem giao_uoc.py mục 5), và cũng # tránh việc self-check độc lập của module này phải có mạng/khoá # API mới chạy được. self._ham_embed_cau_hoi = ham_embed_cau_hoi # Tách từ một lần, dùng cho cả BM25 lẫn để giữ song song với # self.cac_doan theo chỉ số (idx) — mọi cấu trúc dưới đây đều # đánh chỉ số theo đúng thứ tự self.cac_doan. tu_vi: List[List[str]] = [] tu_en: List[List[str]] = [] for d in self.cac_doan: tu_vi.append(tach_tu(d.van_ban_vi)) # van_ban_en có thể rỗng (chưa dịch) -> tách ra danh sách # rỗng -> đoạn đó không bao giờ khớp BM25 tiếng Anh. Đây là # hành vi ĐÚNG: thà vắng mặt trong kết quả EN còn hơn lẫn # đoạn tiếng Việt vào câu trả lời tiếng Anh. tu_en.append(tach_tu(d.van_ban_en) if d.van_ban_en else []) self._bm25_vi = BM25(tu_vi) self._bm25_en = BM25(tu_en) # ---- Ma trận vector (tuỳ chọn, numpy import TRỄ) ---- # LƯU Ý: Doan (theo giao_uoc.py) KHÔNG khai báo trường `vector` # chính thức — đó là một cột của bảng SQLite, không phải một # field của dataclass. kho/chi_muc.py (khi build xong) có thể # gắn thuộc tính `.vector` một cách động lên từng Doan (dataclass # thường không cấm setattr ngoài field khai báo). Vì vậy ở đây # dùng getattr phòng thủ: không có -> coi như None, KHÔNG coi là # lỗi. Đây là lựa chọn có chủ đích để giai đoạn 1 (chưa có # vector) chạy đầy đủ mà không cần sửa gì khi kho/ hoàn thiện. self._ma_tran_vector = None # np.ndarray đã chuẩn hoá theo hàng, hoặc None self._idx_co_vector: List[int] = [] self._dung_vector = any(getattr(d, 'vector', None) for d in self.cac_doan) if self._dung_vector: self._dung_ma_tran_vector() # ---- Chỉ mục phụ cho doan_lien_quan (đoạn kề trong cùng tệp) ---- self._theo_ma: Dict[str, Doan] = {d.ma: d for d in self.cac_doan} theo_tep: Dict[str, List[Doan]] = defaultdict(list) for d in self.cac_doan: theo_tep[d.nguon_tep].append(d) # Sắp theo `ma` — mã đoạn có dạng '.../0007' đánh số thứ tự cắt # trong tệp, nên sort theo chuỗi này tương đương thứ tự xuất # hiện trong tài liệu gốc (miễn cùng độ dài đệm số, do kho/ tạo). for tep in theo_tep: theo_tep[tep].sort(key=lambda d: d.ma) self._theo_tep: Dict[str, List[Doan]] = theo_tep # ------------------------------------------------------------------ def _dung_ma_tran_vector(self) -> None: """Dựng ma trận numpy từ các đoạn có vector — numpy import TRỄ. VÌ SAO import trễ ở tận đây: nếu máy chạy (vd VPS nhỏ) không cài numpy và kho chưa có vector nào (đúng trạng thái giai đoạn 1), ta không bao giờ chạm dòng `import numpy` — module vẫn hoạt động 100% bằng BM25 mà không đòi hỏi thêm gì. """ try: import numpy as np except ImportError: self._dung_vector = False return hang: List[Any] = [] idx_co: List[int] = [] for i, d in enumerate(self.cac_doan): vec = getattr(d, 'vector', None) if vec: hang.append(vec) idx_co.append(i) if not hang: self._dung_vector = False return mt = np.asarray(hang, dtype=float) # Chuẩn hoá sẵn từng hàng về độ dài 1 NGAY LÚC DỰNG chỉ mục, để # lúc truy vấn cosine chỉ còn là một phép tích vô hướng (dot), # không phải chia norm lặp lại mỗi câu hỏi. chuan = np.linalg.norm(mt, axis=1, keepdims=True) chuan[chuan == 0] = 1.0 # tránh chia 0 cho vector rỗng bất thường self._ma_tran_vector = mt / chuan self._idx_co_vector = idx_co def _truy_van_vector(self, cau_hoi: str, cho_phep: Set[int], top: int) -> List[Tuple[int, float]]: """Xếp hạng theo cosine similarity — chỉ chạy khi có cả ma trận vector LẪN hàm embedding câu hỏi (hai điều kiện độc lập, thiếu một là bỏ qua nhánh này, không phải lỗi).""" if self._ma_tran_vector is None or self._ham_embed_cau_hoi is None: return [] import numpy as np # đã chắc chắn có numpy vì _ma_tran_vector != None vec = self._ham_embed_cau_hoi(cau_hoi) if vec is None: return [] v = np.asarray(vec, dtype=float) chuan = float(np.linalg.norm(v)) if chuan == 0: return [] v = v / chuan diem = self._ma_tran_vector @ v # cosine, vì cả hai phía đã chuẩn hoá độ dài 1 cap = [ (idx_goc, float(diem[vi_tri])) for vi_tri, idx_goc in enumerate(self._idx_co_vector) if idx_goc in cho_phep ] cap.sort(key=lambda x: x[1], reverse=True) return cap[:top] # ------------------------------------------------------------------ def tim(self, cau_hoi: str, loc: LocTim = None, so_luong: int = SO_DOAN_MAC_DINH) -> List[KetQuaTim]: """Giao diện bắt buộc theo giao_uoc.py — CHỮ KÝ KHÔNG ĐƯỢC ĐỔI.""" if loc is None: loc = LocTim() # Kẹp so_luong về tối đa SO_DOAN_TOI_DA VÔ ĐIỀU KIỆN — kể cả khi # người gọi cố tình xin nhiều hơn. Đây là quyết định CHẤT LƯỢNG # (xem giao_uoc.py): nhồi nhiều đoạn thì mô hình "lạc giữa # dòng" — thông tin đúng lẫn trong nhiễu và bị bỏ qua — không # phải giới hạn kỹ thuật nên không có lý do gì để tầng gọi # bypass được. so_luong = max(1, min(so_luong, SO_DOAN_TOI_DA)) # ---- 1. LỌC san_pham / loai TRƯỚC khi chấm điểm ---- chi_so_cho_phep: Set[int] = set() for i, d in enumerate(self.cac_doan): if (loc.san_pham is not None and d.san_pham is not None and d.san_pham != loc.san_pham): continue # d.san_pham is None -> đoạn DÙNG CHUNG -> luôn giữ lại, đây # là điều khoản NHÂN NHƯỢNG bắt buộc theo đề bài: một đoạn # "cách nạp key" dùng chung cho mọi sản phẩm không được vì # khách hỏi về DCAbot mà biến mất khỏi kết quả. if loc.loai is not None and d.loai != loc.loai: continue chi_so_cho_phep.add(i) if not chi_so_cho_phep: return [] ngon_ngu = (loc.ngon_ngu or 'vi').lower() bm25_dung = self._bm25_en if ngon_ngu == 'en' else self._bm25_vi tu_cau_hoi = tach_tu(cau_hoi) # Lấy TOP RỘNG HƠN so_luong cuối cùng trước khi hợp nhất — nếu # mỗi nguồn chỉ đưa ra đúng so_luong ứng viên, một đoạn hạng 7 # ở BM25 nhưng hạng 1 ở vector (hoặc ngược lại) sẽ bị cắt mất # TRƯỚC khi kịp cộng điểm RRF, làm hỏng cả mục đích hợp nhất. top_mo_rong = max(so_luong * 6, 30) hang_bm25 = bm25_dung.truy_van(tu_cau_hoi, top=top_mo_rong, cho_phep=chi_so_cho_phep) hang_vector = self._truy_van_vector(cau_hoi, chi_so_cho_phep, top_mo_rong) # ---- 2. HỢP NHẤT BẰNG RRF (Reciprocal Rank Fusion) ---- # VÌ SAO không cộng điểm thô: điểm BM25 (tổng idf*tf có trọng # số, biên độ tuỳ kho — có thể là 0.5 hoặc 12) và điểm cosine # (luôn nằm trong [-1, 1]) KHÔNG cùng thang đo. Cộng thẳng hai # con số này lại nghĩa là nguồn nào có biên độ lớn hơn sẽ luôn # thắng bất kể mức độ liên quan thực sự — một chiều, không phải # "kết hợp". RRF bỏ điểm tuyệt đối, chỉ dùng THỨ HẠNG trong danh # sách riêng của từng nguồn (điểm = Σ 1/(k+hạng)), nên một đoạn # đứng #1 ở BM25 và một đoạn đứng #1 ở vector đóng góp NGANG # NHAU vào điểm hợp nhất, dù thang điểm gốc của hai nguồn khác # hẳn nhau. diem_rrf: Dict[int, float] = defaultdict(float) nguon_diem: Dict[int, Dict[str, float]] = defaultdict(dict) for hang, (idx, diem) in enumerate(hang_bm25, start=1): diem_rrf[idx] += 1.0 / (_K_RRF + hang) nguon_diem[idx]['bm25'] = diem for hang, (idx, diem) in enumerate(hang_vector, start=1): diem_rrf[idx] += 1.0 / (_K_RRF + hang) nguon_diem[idx]['vector'] = diem xep_hang = sorted(diem_rrf.items(), key=lambda cap: cap[1], reverse=True) xep_hang = xep_hang[:so_luong] return [ KetQuaTim(doan=self.cac_doan[idx], diem=diem, nguon_diem=dict(nguon_diem[idx])) for idx, diem in xep_hang ] # ------------------------------------------------------------------ def doan_lien_quan(self, ma_doan: str, so_luong: int = 2) -> List[Doan]: """Trả các đoạn KỀ trong cùng tệp nguồn với `ma_doan`. VÌ SAO cần: một đoạn bị cắt riêng lẻ có thể mất mạch — câu trả lời cho "đặt tham số này ở đâu" có khi nằm ở đoạn NGAY TRƯỚC/SAU trong cùng bài hướng dẫn, không phải trong chính đoạn khớp BM25/vector. Tìm lại bằng điểm số không đảm bảo "hàng xóm" được xếp hạng cao (nó có thể ít từ khoá hơn hẳn đoạn khớp chính), nên thay vào đó lấy trực tiếp theo VỊ TRÍ trong tệp gốc. """ goc = self._theo_ma.get(ma_doan) if goc is None: return [] cung_tep = self._theo_tep.get(goc.nguon_tep, []) try: vi_tri = cung_tep.index(goc) except ValueError: return [] # Xen kẽ SAU rồi TRƯỚC, ưu tiên đoạn sau trước (mạch đọc xuôi # thường hữu ích hơn khi tài liệu hướng dẫn theo trình tự bước). ket_qua: List[Doan] = [] sau, truoc = vi_tri + 1, vi_tri - 1 while len(ket_qua) < so_luong and (sau < len(cung_tep) or truoc >= 0): if sau < len(cung_tep): ket_qua.append(cung_tep[sau]) sau += 1 if len(ket_qua) >= so_luong: break if truoc >= 0: ket_qua.append(cung_tep[truoc]) truoc -= 1 return ket_qua # ========================================================================= # SINGLETON — dựng chỉ mục là việc tốn kém, chỉ nên trả giá một lần cho cả # tiến trình (bot chạy dạng service, không phải CLI gọi một phát rồi thoát). # ========================================================================= _bo_tim_theo_shop: Dict[str, BoTim] = {} def lay_bo_tim(ham_embed_cau_hoi: Optional[Callable[[str], Any]] = None, khong_gian: Optional[str] = None) -> BoTim: """BoTim dùng chung cho cả tiến trình, MỘT BỘ CHO MỖI SHOP (tách lõi 24/09/2026) — mỗi shop tìm trong chỉ mục của chính nó, không bao giờ lẫn tài liệu shop khác. Dựng lần đầu, tái sử dụng các lần sau. Shop chưa có tệp chỉ mục (mới khai, chưa dựng) -> bộ rỗng: bot vẫn trả lời bằng kho tư vấn tầng 1 và kiến thức chung, không ngã, không lấy nhầm của ai.""" import shop hs = shop.lay(khong_gian) if hs.ma not in _bo_tim_theo_shop: from kho.chi_muc import tep_chi_muc tep = tep_chi_muc(hs.ma) if hs.ma != shop.MAC_DINH and not os.path.exists(tep): _bo_tim_theo_shop[hs.ma] = BoTim(cac_doan=[], ham_embed_cau_hoi=ham_embed_cau_hoi) else: _bo_tim_theo_shop[hs.ma] = BoTim(cac_doan=_nap_doan_tu_kho(tep), ham_embed_cau_hoi=ham_embed_cau_hoi) return _bo_tim_theo_shop[hs.ma] # ========================================================================= # TỰ KIỂM — chạy `python tim/lai.py` # ========================================================================= if __name__ == '__main__': def _tao_doan_gia_de_tu_kiem() -> List[Doan]: """Bộ Doan BỊA dùng khi kho/chi_muc.py chưa sẵn sàng (đang viết song song). Đủ để chứng minh thuật toán tìm-lai chạy đúng, KHÔNG phản ánh nội dung thật của website.""" tho = [ # (ma, van_ban_vi, van_ban_en, san_pham, loai, tieu_de, tep) ('dcabot/huong_dan/0001', 'Hướng dẫn cài EA DCAbot vào MT5: tải file .ex5, kéo vào thư mục ' 'Experts của MetaTrader 5, khởi động lại nền tảng rồi kéo EA vào biểu đồ.', 'How to install the DCAbot EA into MT5: copy the .ex5 file into the ' 'Experts folder of MetaTrader 5, restart the platform, then drag the EA onto a chart.', 'dcabot', 'huong_dan', 'Cài đặt DCAbot trên MT5', 'guilde/dcabot.php'), ('dcabot/huong_dan/0002', 'Sau khi kéo EA vào biểu đồ, bật AlgoTrading và WebRequest cho domain ' 'license server thì EA mới vào lệnh được.', 'After dragging the EA onto the chart, enable AlgoTrading and allow ' 'WebRequest for the license server domain so the EA can place trades.', 'dcabot', 'huong_dan', 'Bật AlgoTrading và WebRequest', 'guilde/dcabot.php'), ('dcabot/loi/0009', 'Lỗi "Invalid Master Code" xảy ra khi nhập sai mã kích hoạt, hoặc mã ' 'đã được dùng trên tài khoản MT4/MT5 khác. Kiểm tra lại đúng license key GOLDVIP6B.', 'The "Invalid Master Code" error happens when the activation code is ' 'wrong, or already used on another MT4/MT5 account. Double-check the GOLDVIP6B license key.', 'dcabot', 'huong_dan', 'Lỗi Invalid Master Code', 'guilde/dcabot.php'), ('dcabot/tinh_nang/0004', 'MaxLoss là tham số giới hạn lỗ tối đa cho một chu kỳ DCA, đặt trong ' 'phần cấu hình rủi ro của EA, đơn vị theo % vốn hoặc theo tiền tệ.', 'MaxLoss is the parameter that caps the maximum loss for one DCA ' 'cycle, set in the EA risk configuration, in % of equity or currency units.', 'dcabot', 'tinh_nang', 'Tham số MaxLoss', 'intro_sections/dcabot.php'), ('signalbroadcaster/gioi_thieu/0001', 'Signal Copier chỉ sao chép lệnh một chiều từ tài khoản master sang ' 'slave, còn Signal Broadcaster phát lệnh cho nhiều slave cùng lúc qua server trung gian.', 'Signal Copier only copies orders one-way from a master account to a ' 'slave, while Signal Broadcaster pushes orders to many slaves at once through a relay server.', 'signalbroadcaster', 'gioi_thieu', 'So sánh Signal Copier và Broadcaster', 'toolintroduce/signalbroadcaster.php'), ('dungchung/gia/0001', 'Bảng giá các gói: 1 tháng, 3 tháng, 6 tháng và vĩnh viễn, thanh toán ' 'bằng credit, 1 USD quy đổi 1 credit.', 'Pricing tiers: 1 month, 3 months, 6 months and lifetime, paid in ' 'credit, 1 USD equals 1 credit.', None, 'gioi_thieu', 'Bảng giá chung', 'toolintroduce/pricing.php'), ('copytrade/su_dung/0003', 'Copytrade giới hạn số slave mỗi master theo gói đã mua, server không ' 'ép cứng nhưng web bán đúng theo số lượng công bố.', '', # chưa dịch — để kiểm tra hành vi BM25 tiếng Anh với văn bản rỗng 'copytrade', 'su_dung', 'Giới hạn slave copytrade', 'sudung/copytrade.php'), ('tradingview/faq/0001', 'TradingView bot dùng webhook để nhận tín hiệu, không cần WebRequest ' 'như bên MT4/MT5.', 'The TradingView bot uses a webhook to receive signals, no ' 'WebRequest needed unlike the MT4/MT5 side.', 'tradingview', 'faq', 'Webhook TradingView', 'faq/tradingview.php'), ] return [ Doan(ma=ma, van_ban_vi=vi, van_ban_en=en, san_pham=sp, loai=loai, duong_dan='/demo', tieu_de=tieu_de, nguon_tep=tep, so_tu=len(vi.split())) for (ma, vi, en, sp, loai, tieu_de, tep) in tho ] def _in_ket_qua(tieu_de: str, ket_qua: List[KetQuaTim]) -> None: print(f"\n--- {tieu_de} ---") if not ket_qua: print(" (không có kết quả)") for kq in ket_qua[:3]: print(f" [{kq.diem:.4f}] {kq.doan.ma:28} {kq.doan.tieu_de!r} " f"nguồn={kq.nguon_diem}") CAC_CAU_HOI = [ "cài EA vào MT5 thế nào", "Invalid Master Code", "MaxLoss là gì", "giá bao nhiêu", "Signal Copier khác Broadcaster chỗ nào", ] du_lieu_gia = False try: bo = BoTim() print(f"[OK] Đã nạp {len(bo.cac_doan)} đoạn từ chỉ mục thật (kho/chi_muc.sqlite).") except RuntimeError as loi: print(str(loi)) print( "\n[Tự kiểm] kho/chi_muc.py chưa sẵn sàng ở thời điểm chạy thử " "này (module kho/ đang được viết song song). Dựng tạm một " "danh sách Doan GIẢ (bịa) để vẫn chứng minh thuật toán tìm-lai " "chạy đúng. Kết quả bên dưới KHÔNG phản ánh nội dung thật của " "website — chỉ để kiểm chứng mã nguồn." ) du_lieu_gia = True bo = BoTim(cac_doan=_tao_doan_gia_de_tu_kiem()) print(f"\n{'=' * 70}") print(f"THỬ {len(CAC_CAU_HOI)} TRUY VẤN{' (DỮ LIỆU GIẢ)' if du_lieu_gia else ''}") print('=' * 70) for cau in CAC_CAU_HOI: ket_qua = bo.tim(cau, so_luong=5) _in_ket_qua(cau, ket_qua) print(f"\n{'=' * 70}") print(f"KIỂM KẸP so_luong: xin 50, giới hạn hợp đồng SO_DOAN_TOI_DA={SO_DOAN_TOI_DA}") print('=' * 70) ket_qua_50 = bo.tim(CAC_CAU_HOI[0], so_luong=50) dat = len(ket_qua_50) <= SO_DOAN_TOI_DA print(f"[Kho giả {len(bo.cac_doan)} đoạn] Xin so_luong=50 -> thực nhận " f"{len(ket_qua_50)} kết quả -> " f"{'DAT (<=8)' if dat else 'SAI — VỠ HỢP ĐỒNG'}") assert dat, "so_luong không được vượt SO_DOAN_TOI_DA" # Kho 8 đoạn ở trên không đủ để tự nó VƯỢT quá 8 ứng viên khớp — ép # rõ ràng bằng một kho lớn hơn (15 đoạn) nơi TẤT CẢ đều chứa chung # một từ, để chứng minh việc kẹp về đúng 8 THỰC SỰ xảy ra (không # phải "tình cờ" trùng vì corpus nhỏ). doan_kho_lon = [ Doan(ma=f'demo/kho_lon/{i:04d}', van_ban_vi=f'Tài liệu số {i} nói về bot giao dịch TradingAuto và cách vận hành.', van_ban_en='', san_pham=None, loai='faq', duong_dan='/demo', tieu_de=f'Tài liệu {i}', nguon_tep='demo/kho_lon.php') for i in range(15) ] bo_kho_lon = BoTim(cac_doan=doan_kho_lon) ket_qua_kho_lon = bo_kho_lon.tim('bot tradingauto', so_luong=50) dat2 = len(ket_qua_kho_lon) == SO_DOAN_TOI_DA print(f"[Kho giả {len(doan_kho_lon)} đoạn, tất cả đều khớp] Xin so_luong=50 -> " f"thực nhận {len(ket_qua_kho_lon)} kết quả -> " f"{'DAT (kep dung tai '+str(SO_DOAN_TOI_DA)+')' if dat2 else 'SAI'}") assert dat2, "phải kẹp đúng bằng SO_DOAN_TOI_DA khi có nhiều hơn 8 ứng viên khớp" # ---- Demo bổ sung (không bắt buộc): nhánh vector + RRF khi CÓ vector ---- # Chứng minh nhánh numpy không phải code chết — gắn .vector động lên # vài Doan giả (mô phỏng cột `vector` của SQLite) và một hàm embed giả # (không gọi mạng) để RRF có hai nguồn thật sự hợp nhất. print(f"\n{'=' * 70}") print("DEMO BỔ SUNG: nhánh vector (numpy) + hợp nhất RRF hai nguồn") print('=' * 70) doan_co_vector = _tao_doan_gia_de_tu_kiem() # Vector giả 4 chiều, bịa để đoạn 0002 (WebRequest) và 0008 (webhook, # KHÔNG dùng WebRequest) khác hướng nhau rõ rệt trên trục "webrequest". vector_gia = { 'dcabot/huong_dan/0001': [1, 0, 0, 0], 'dcabot/huong_dan/0002': [0.9, 0.1, 0, 0], # gần nhất với câu hỏi demo 'dcabot/loi/0009': [0, 1, 0, 0], 'dcabot/tinh_nang/0004': [0, 0, 1, 0], 'signalbroadcaster/gioi_thieu/0001': [0, 0, 0, 1], 'dungchung/gia/0001': [0.2, 0.2, 0.2, 0.2], 'copytrade/su_dung/0003': [0.3, 0.3, 0, 0.1], 'tradingview/faq/0001': [0.1, 0, 0, 0.9], } for d in doan_co_vector: d.vector = vector_gia.get(d.ma) # gắn ĐỘNG — Doan không khai báo field này def _embed_gia(cau_hoi: str): # Hàm embed GIẢ, không gọi mạng: chỉ để chứng minh cơ chế cosine + # RRF chạy đúng, không nhằm mô phỏng chất lượng embedding thật. if 'webrequest' in cau_hoi.lower() or 'mt5' in cau_hoi.lower(): return [0.9, 0.1, 0, 0] return [0.25, 0.25, 0.25, 0.25] bo_vector = BoTim(cac_doan=doan_co_vector, ham_embed_cau_hoi=_embed_gia) print(f"dung_vector={bo_vector._dung_vector} " f"so_doan_co_vector={len(bo_vector._idx_co_vector)}") ket_qua_vec = bo_vector.tim("cấu hình WebRequest cho MT5", so_luong=3) _in_ket_qua("cấu hình WebRequest cho MT5 (có vector)", ket_qua_vec) # ---- Demo bổ sung: doan_lien_quan ---- print(f"\n{'=' * 70}") print("DEMO BỔ SUNG: doan_lien_quan (đoạn kề cùng tệp)") print('=' * 70) ke = bo.doan_lien_quan('dcabot/huong_dan/0001', so_luong=2) print(f"Đoạn kề với dcabot/huong_dan/0001: {[d.ma for d in ke]}") print("\nTỰ KIỂM lai.py HOÀN TẤT.")