# -*- coding: utf-8 -*- """THU_DU_PHONG.PY — ép nhánh dự phòng của lớp AI chạy qua lỗi THẬT. VÌ SAO TỆP NÀY TỒN TẠI ---------------------- `llm/tuongthich_openai.py` có một nhánh chỉ chạy khi cổng AI hỏng: đổi sang `LLM_MODEL_DU_PHONG` rồi thử lại đúng một lần. Nhánh đó KHÔNG BAO GIỜ chạy trong lúc mọi thứ bình thường — nghĩa là nó có thể sai suốt nhiều tháng mà không ai biết, rồi hỏng đúng vào lúc cổng chết, tức là đúng lúc cần nó nhất. Cách duy nhất kiểm được mà không tốn tiền và không phụ thuộc mạng: thay `httpx.post` bằng một hàm giả lập trả về đúng mã lỗi ta muốn. Bộ này kiểm CẢ HAI CHIỀU, và chiều thứ hai mới là chiều quan trọng: - có thử lại khi hỏng HẠ TẦNG (403 / 429 / 5xx / timeout / trả rỗng); - TUYỆT ĐỐI KHÔNG thử lại vì bất cứ lý do nào khác. Thử lại cho tới khi có câu vừa ý là "mua câu trả lời". Trên một hệ bán hàng thật, đó là cách tự lừa nguy hiểm nhất — nên ca 401 và ca "gọi phát ăn ngay" phải chứng minh được là CHỈ gọi một lượt. Chạy: python thu/thu_du_phong.py (từ thư mục gốc của module chatbot) """ from __future__ import annotations import logging import os import sys _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) def chay() -> int: """Trả về số ca SAI. 0 nghĩa là đạt.""" os.environ.update( LLM_GOC='https://vidu.test/v1', # cố tình kèm /v1 để kiểm chuẩn hoá # Khoá GIẢ, không bao giờ rời khỏi tiến trình này (httpx.post đã bị thay # bằng hàm giả lập). ĐỪNG dán khoá thật vào đây dù chỉ để thử: tệp này # nằm trong git, và một khoá thật trong mã nguồn thì coi như đã lộ. LLM_KHOA='sk-KHOA-GIA-CHI-DE-TU-KIEM-0000', LLM_MODEL_DIEN_GIAI='model-chinh', LLM_MODEL_DU_PHONG='model-du-phong', ) import httpx from llm.giao_dien import LoiNhaCungCap from llm.tuongthich_openai import NhaCungCapTuongThichOpenAI nhat_ky_goi = [] def gia_lap(ke_hoach): """ke_hoach[i] quyết định lượt gọi thứ i. Lượt thừa lặp lại phần tử cuối.""" def post(url, json=None, headers=None, timeout=None, **kw): n = len(nhat_ky_goi) nhat_ky_goi.append({'model': (json or {}).get('model'), 'ua': (headers or {}).get('User-Agent'), 'url': url}) return ke_hoach[min(n, len(ke_hoach) - 1)]() return post def tra_ve_ok(model='model-du-phong', noi_dung='Dạ, em trả lời đây ạ.'): return lambda: httpx.Response( 200, json={'model': model, 'choices': [{'message': {'content': noi_dung}}], 'usage': {'prompt_tokens': 123, 'completion_tokens': 45}}, request=httpx.Request('POST', 'https://vidu.test')) def tra_ve_cat(): """200 nhưng BỊ CẮT vì chạm trần max_tokens (finish_reason='length').""" return lambda: httpx.Response( 200, json={'model': 'model-chinh', 'choices': [{'message': {'content': 'Cau tra loi dut giua chu'}, 'finish_reason': 'length'}], 'usage': {'prompt_tokens': 2000, 'completion_tokens': 700}}, request=httpx.Request('POST', 'https://vidu.test')) def tra_ve_ma(code, than='{"error":"loi gia lap"}'): return lambda: httpx.Response(code, text=than, request=httpx.Request('POST', 'https://vidu.test')) def tra_ve_treo(): def f(): raise httpx.TimeoutException('gia lap treo') return f # (tên ca, kịch bản từng lượt, CÓ ĐƯỢC PHÉP thử lại không) CAC_CA = [ ('403 -> phải sang dự phòng', [tra_ve_ma(403), tra_ve_ok()], True), ('500 -> phải sang dự phòng', [tra_ve_ma(500), tra_ve_ok()], True), ('429 -> phải sang dự phòng', [tra_ve_ma(429), tra_ve_ok()], True), ('timeout -> phải sang dự phòng', [tra_ve_treo(), tra_ve_ok()], True), ('200 rỗng -> phải sang dự phòng', [tra_ve_ok(noi_dung=' '), tra_ve_ok()], True), ('401 -> KHÔNG được thử lại', [tra_ve_ma(401), tra_ve_ok()], False), ('200 tốt ngay -> KHÔNG thử lại', [tra_ve_ok('model-chinh')], False), ('cả hai hỏng -> ném lỗi', [tra_ve_ma(500), tra_ve_ma(500)], True), # Bị cắt KHÔNG phải hỏng hạ tầng: mô hình đã trả lời, chỉ là dài quá # trần. Đổi sang model khác cũng bị cắt y hệt, chỉ tốn thêm tiền. ('bị cắt -> KHÔNG được thử lại', [tra_ve_cat(), tra_ve_ok()], False), ] post_that = httpx.post so_sai = 0 print('=' * 92) print('NHÁNH DỰ PHÒNG CỦA llm/tuongthich_openai.py — ép chạy qua lỗi giả lập') print('=' * 92) for ten, ke_hoach, duoc_thu_lai in CAC_CA: nhat_ky_goi.clear() httpx.post = gia_lap(ke_hoach) try: ncc = NhaCungCapTuongThichOpenAI() try: ncc._goi('model-chinh', 'câu dẫn hệ thống', 'câu người dùng', 100) ket = 'trả lời được' except LoiNhaCungCap as loi: ket = 'ném lỗi: ' + str(loi)[:40] finally: httpx.post = post_that da_thu_lai = len(nhat_ky_goi) > 1 dung = (da_thu_lai == duoc_thu_lai) so_sai += 0 if dung else 1 print('%-34s lượt=%d %-32s %-22s %s' % (ten, len(nhat_ky_goi), [g['model'] for g in nhat_ky_goi], ket[:22], 'đạt' if dung else '<== SAI')) # Bị cắt PHẢI ghi cảnh báo — cắt im lặng là kiểu hỏng tệ nhất: khách nhận # một câu đứt giữa chừng, còn log chỉ thấy một lượt gọi thành công. bat = [] class _Bat(logging.Handler): def emit(self, ban_ghi): bat.append(ban_ghi.getMessage()) _lg = logging.getLogger('chatbot.llm.tuongthich_openai') _lg.addHandler(_Bat()) _lg.setLevel(logging.WARNING) logging.disable(logging.NOTSET) httpx.post = gia_lap([tra_ve_cat()]) try: NhaCungCapTuongThichOpenAI()._goi('model-chinh', 'hệ thống', 'người dùng', 700) except Exception: # noqa: BLE001 pass finally: httpx.post = post_that logging.disable(logging.CRITICAL) co_canh_bao = any('BỊ CẮT' in m for m in bat) so_sai += 0 if co_canh_bao else 1 print('-' * 92) print('%-52s %-40s %s' % ('bị cắt -> phải ghi cảnh báo', 'co_canh_bao=%s' % co_canh_bao, 'đạt' if co_canh_bao else '<== SAI')) # Ba tính chất phụ, cũng chỉ lộ ra khi có lượt gọi thật đi qua print('-' * 92) g = nhat_ky_goi[0] kiem_phu = [ ('User-Agent riêng (Cloudflare chặn UA mặc định)', g['ua'] == 'tradingauto-chatbot/1.0', g['ua']), ('URL không lặp /v1 dù LLM_GOC có sẵn /v1', g['url'] == 'https://vidu.test/v1/chat/completions', g['url']), ] for ten, dat, thuc in kiem_phu: so_sai += 0 if dat else 1 print('%-52s %-40s %s' % (ten, thuc, 'đạt' if dat else '<== SAI')) print('=' * 92) print('SỐ CA SAI: %d' % so_sai) return so_sai if __name__ == '__main__': logging.disable(logging.CRITICAL) # tắt log để bảng kết quả đọc được sys.exit(1 if chay() else 0)