| """
|
| core/decide.py
|
| ==============
|
| ทาง จ ส่วนกลาง — ตัดสินในโค้ดว่าจะ "ตอบ / ถามกลับ / ปฏิเสธ / ส่งฝ่ายขาย" ก่อนถึง LLM
|
| ย้ายมาจาก DialogueManager.decide ชั้น 4 · ส่วนที่ผูกกับแบรนด์ย้ายไป brands/<key>/plugin.py
|
|
|
| ลำดับการตัดสินในหนึ่งเทิร์น (เท่าเดิม)
|
| รูปที่ระบบปฏิเสธ → reject_image
|
| คำตอบของการถามกลับ → รวมกับคำถามเดิม (ถ้าถามแบรนด์ไว้ ดูว่าตอบแบรนด์ไหน)
|
| ถามแบรนด์เพราะรูป (core/image_policy) → ใช้ผลรูปที่เก็บไว้กับแบรนด์ที่ตอบ
|
| ทักทาย/ขอบคุณ → smalltalk
|
| ไม่เกี่ยวกับสินค้า → refuse_offtopic
|
| ขอช่องทางติดต่ออย่างเดียว → escalate (ไม่ต้องรู้แบรนด์)
|
| สัญญาณแข็งหลายแบรนด์ → ask ยืนยันแบรนด์
|
| สแตนเลสไม่ระบุแบรนด์ → cross_brand (ทุกแบรนด์ตอบส่วนของตัวเอง)
|
| ยังไม่รู้แบรนด์ → ask แบรนด์
|
| รู้แบรนด์แล้ว → plugin.decide()
|
|
|
| ไม่ import torch / qdrant / genai — เทสต์ด้วย python ล้วน
|
| """
|
| from __future__ import annotations
|
|
|
| import copy
|
| import re
|
| from typing import Optional, Sequence
|
|
|
| from core import router as _router
|
| from core import templates as T
|
| from core.protocol import BrandPlugin, DecideInput, Decision, ImageScore
|
| from core.image_policy import ImageCarry
|
| from core.sales import escalation_gate
|
| from core.text import any_word, has
|
|
|
|
|
|
|
| DOMAIN_WORDS = [
|
| "สายพาน", "โซ่", "สเตอร์", "เฟือง", "เพลา", "พิน", "ก้าน", "ลำเลียง", "คอนเวเยอร์",
|
| "belt", "chain", "sprocket", "conveyor",
|
| "กว้าง", "ความยาว", "หนา", "workload", "โหลด", "รับน้ำหนัก", "แรงดึง", "น้ำหนัก", "แข็งแรง",
|
| "วัสดุ", "ทำจาก", "เลี้ยว", "โค้ง", "วิ่งตรง", "รัศมี", "ผลิต", "ทน", "อุณหภูมิ", "องศา",
|
| "ร้อน", "เย็น", "แช่แข็ง", "เปียก", "แห้ง", "drawing", "datasheet", "ดรออิ้ง", "แบบ", "สเปก", "spec",
|
| "ราคา", "สต็อก", "สต๊อก", "ส่งของ", "อะไหล่", "ผิว", "ไลน์", "รุ่น", "series", "ซีรีส์",
|
| "สเปค", "เทียบ",
|
| "ฟัน", "pitch", "พิทช์", "เบอร์", "ขนาด",
|
|
|
|
|
| "นิ้ว", "inch", "มิลลิเมตร", "เซนติเมตร", "มม", "ซม", "ฟุต", "feet", "หน่วย", "เมตร",
|
|
|
|
|
|
|
| "สั่งซื้อ", "ซื้อ", "สั่ง", "ออเดอร์", "order", "ใบเสนอราคา", "quotation",
|
| "ติดต่อ", "ฝ่ายขาย", "เซลล์", "อีเมล", "email", "โทร", "ไลน์",
|
| ]
|
|
|
|
|
|
|
|
|
|
|
| FOLLOWUP_MAX_CHARS = 40
|
| THANKS_WORDS = ["ขอบคุณ", "ขอบใจ", "thank"]
|
| THANKS_MAX_CHARS = 25
|
| SMALLTALK_WORDS = ["สวัสดี", "ขอบคุณ", "ขอบใจ", "hello", "hi ", "thank", "thanks", "ดีครับ", "ดีค่ะ"]
|
| GREETING_WORDS = ["สวัสดี", "hello", "hi ", "ดีครับ", "ดีค่ะ"]
|
|
|
|
|
| CONTINUE_WORDS = ["ปรึกษา", "คุยกับ", "หัวหน้า", "เดี๋ยว", "ไว้ก่อน", "เข้าใจแล้ว", "ถามซ้ำ", "บอกไปแล้ว",
|
| "ไม่รู้เรื่อง", "ไม่เข้าใจ", "ทางตรง", "ทางโค้ง"]
|
| REFERENCE_WORDS = ["ตัวนั้น", "ตัวนี้", "อันนั้น", "อันนี้", "รุ่นนั้น", "รุ่นนี้", "ตัวที่", "อันที่",
|
| "ตัวแรก", "ตัวหลัง", "เมื่อกี้", "ข้างบน", "ตัวเดิม"]
|
|
|
| STAINLESS_RX = r"สแตนเลส|stainless|(?<![A-Za-z])SS(?![A-Za-z])"
|
| CROSS_BRAND_STAINLESS = "stainless"
|
|
|
|
|
| BRAND_CONFIRM = "brand:"
|
| _CONFIRM_NO = re.compile(r"ไม่ใช่|ไม่ได้เป็น|^\s*ไม่\s*(?:ครับ|ค่ะ|คะ)?\s*$|\bno\b|\bnope\b|\bnot\b", re.I)
|
| _CONFIRM_YES = re.compile(r"ใช่|ถูกต้อง|ถูกแล้ว|โอเค|\bok\b|\byes\b|^\s*(?:ครับ|ค่ะ|คะ|จ้า)\s*$", re.I)
|
|
|
|
|
| def brand_pending(p) -> bool:
|
| """กำลังรอคำตอบเรื่องแบรนด์ — เมนูหลายแบรนด์ ("brand") หรือถามยืนยันแบรนด์เดียว ("brand:<key>")"""
|
| return p == "brand" or (isinstance(p, str) and p.startswith(BRAND_CONFIRM))
|
|
|
|
|
|
|
|
|
|
|
| GENERIC_REJECT_IMAGE_TH = ("รูปนี้อาจถ่ายไกลเกินไป หรืออาจไม่ใช่โซ่หรือสายพานครับ "
|
| "หากเป็นโซ่หรือสายพาน กรุณาถ่ายใกล้กว่านี้ให้ตัวสินค้าเต็มกรอบและเห็นผิวด้านบนชัดเจน "
|
| "หรือแจ้งรหัสรุ่นที่ปั๊มบนตัวสินค้าครับ")
|
|
|
|
|
| class Dialogue:
|
| """
|
| plugins เรียงตาม info.order (ลำดับนี้คือลำดับในข้อความและการจับคำตอบเลือกแบรนด์)
|
| state_cls คลาสจาก core.dialogue.state_class ที่ผูก schema ของแบรนด์ชุดเดียวกัน
|
| image_brands แบรนด์ที่รูปในเทิร์นนี้อาจเป็นของมัน (ImageRouter ตัดสินก่อนส่งมา)
|
| """
|
|
|
| def __init__(self, plugins: Sequence[BrandPlugin], state_cls: type,
|
| image_brands: Sequence[str] = ()):
|
| self.plugins = {p.info.key: p for p in sorted(plugins, key=lambda p: p.info.order)}
|
| self.infos = [p.info for p in self.plugins.values()]
|
| self.state_cls = state_cls
|
| self.image_brands = tuple(image_brands)
|
| self.ask_brand_th = T.ask_brand(self.infos)
|
| self.offtopic_th = T.offtopic(self.infos)
|
| self.smalltalk_th = T.smalltalk(self.infos)
|
| self.greeting_th = T.greeting(self.infos)
|
| self._domain_extra = [w for p in self.plugins.values() for w in p.domain_terms()]
|
|
|
|
|
|
|
| def decide(self, text: str, has_image: bool = False, image: Optional[ImageScore] = None,
|
| image_rejected: bool = False, state=None,
|
| image_carry: Optional[ImageCarry] = None, image_confirm: Optional[str] = None) -> Decision:
|
| """image_confirm = แบรนด์ที่รูปน่าจะเป็น แต่ไม่มั่นใจพอ (core/image_policy › confirm) → ถาม "ใช่ไหม" """
|
| st = copy.deepcopy(state) if state is not None else self.state_cls()
|
| text = (text or "").strip()
|
|
|
| if image_rejected:
|
| key = image.brand if image is not None else (
|
| self.image_brands[0] if len(self.image_brands) == 1 else None)
|
| reply = (self.plugins[key].image_reject_th if key in self.plugins else "") or GENERIC_REJECT_IMAGE_TH
|
| return Decision("reject_image", key, reply, "ViT/SigLIP ไม่พบสินค้า",
|
| effective_text=text, state=st)
|
|
|
| if image_confirm in self.plugins and has_image:
|
| st.pending, st.pending_question = BRAND_CONFIRM + image_confirm, text
|
| return Decision("ask", None, T.confirm_brand(self.plugins[image_confirm].info),
|
| f"รูปน่าจะเป็น {image_confirm} แต่ไม่มั่นใจ → ถามยืนยัน", topic="brand",
|
| effective_text=text, state=st)
|
|
|
|
|
|
|
| if st.pending and any_word(THANKS_WORDS, text) and not any_word(DOMAIN_WORDS, text) \
|
| and len(text) <= THANKS_MAX_CHARS:
|
| st.pending, st.pending_question = None, ""
|
| return Decision("smalltalk", st.brand, self.smalltalk_th, "ขอบคุณ (ปิดคำถามที่ค้าง)",
|
| effective_text=text, state=st)
|
|
|
|
|
|
|
|
|
|
|
| if st.pending and not brand_pending(st.pending) and st.brand:
|
| sig_now = {k: p.detect(text).strength for k, p in self.plugins.items()}
|
| other = [k for k, v in sig_now.items() if k != st.brand and v in ("strong", "weak")]
|
| if other and (sig_now.get(st.brand) in (None, "none")
|
| or (sig_now.get(st.brand) != "strong" and any(sig_now[k] == "strong" for k in other))):
|
| st.pending, st.pending_question = None, ""
|
|
|
|
|
| effective = text
|
| answered_brand = None
|
| carried = None
|
|
|
| confirmed = None
|
| if st.pending and (st.pending_question or brand_pending(st.pending)):
|
| if brand_pending(st.pending):
|
| for key, p in self.plugins.items():
|
| if any(has(pat, text) for pat in T.brand_answer_patterns(p.info)):
|
| st.brand = answered_brand = key
|
| break
|
| ask_key = st.pending[len(BRAND_CONFIRM):] if st.pending.startswith(BRAND_CONFIRM) else None
|
| if ask_key in self.plugins and answered_brand is None:
|
| others = [k for k in self.plugins if k != ask_key]
|
| if _CONFIRM_NO.search(text):
|
| if len(others) != 1:
|
| st.pending = "brand"
|
| return Decision("ask", None, self.ask_brand_th, "ตอบไม่ใช่ · เหลือหลายแบรนด์",
|
| topic="brand", effective_text=st.pending_question, state=st)
|
| st.brand = answered_brand = confirmed = others[0]
|
| elif _CONFIRM_YES.search(text):
|
| st.brand = answered_brand = confirmed = ask_key
|
|
|
|
|
| brand_only = confirmed is not None or (
|
| answered_brand is not None and len(text) <= 20
|
| and not any(p.detect(text).has_code for p in self.plugins.values()))
|
| if brand_only and not st.pending_question and not has_image:
|
| sc = image_carry.best if image_carry is not None else None
|
| if not (sc is not None and sc.brand == answered_brand):
|
|
|
|
|
| info = self.plugins[answered_brand].info
|
| st.pending, st.pending_question = None, ""
|
| return Decision("ask", answered_brand, T.image_need_question(info),
|
| f"ตอบแบรนด์ {answered_brand} หลังรูปเปล่า · ไม่มีคำถาม", topic="image",
|
| effective_text="", state=st)
|
| if image_carry is not None and st.brand is not None and not has_image:
|
| sc = image_carry.best
|
| if sc is not None and sc.brand == st.brand:
|
| if image_carry.rejected:
|
| reply = self.plugins[st.brand].image_reject_th or GENERIC_REJECT_IMAGE_TH
|
| st.pending, st.pending_question = None, ""
|
| return Decision("reject_image", st.brand, reply,
|
| f"ตอบ {st.brand} · รูปเทิร์นก่อนถูกปฏิเสธ",
|
| effective_text=text, state=st,
|
| payload={"image_carried": True})
|
| carried = sc
|
|
|
| effective = st.pending_question.strip() if confirmed else f"{st.pending_question} {text}".strip()
|
| pending_q_before = st.pending_question
|
| st.pending, pending_before, st.pending_question = None, st.pending, ""
|
|
|
|
|
| signals = {k: p.detect(effective) for k, p in self.plugins.items()}
|
| route = _router.choose(signals, prev_brand=st.brand, has_image=has_image,
|
| image_brands=self.image_brands)
|
| if answered_brand and route.kind in ("conflict_strong", "conflict_weak"):
|
|
|
| route = _router.RouteResult(answered_brand, "strong", [], {answered_brand: "ลูกค้าตอบแบรนด์"},
|
| route.signals)
|
| has_code = any(s.has_code for s in signals.values())
|
|
|
|
|
|
|
| in_focus = bool(st.brand) and any(p.has_focus(st.brands[k]) for k, p in self.plugins.items())
|
| followup = in_focus and len(text) <= FOLLOWUP_MAX_CHARS
|
| domain = has_code or has_image or bool(any_word(DOMAIN_WORDS, effective)) \
|
| or bool(any_word(self._domain_extra, effective)) or pending_before is not None \
|
| or has(STAINLESS_RX, effective) or followup
|
| if not domain:
|
| if any_word(SMALLTALK_WORDS, text):
|
| greet = any_word(GREETING_WORDS, text) and not any_word(["ขอบคุณ", "ขอบใจ", "thank"], text)
|
| return Decision("smalltalk", st.brand, self.greeting_th if greet else self.smalltalk_th, "ทักทาย",
|
| effective_text=effective, state=st)
|
| if not (st.brand and (any_word(REFERENCE_WORDS, text) or any_word(CONTINUE_WORDS, text))):
|
| return Decision("refuse_offtopic", st.brand, self.offtopic_th, "ไม่มีคำเกี่ยวกับสินค้า",
|
| effective_text=effective, state=st)
|
|
|
|
|
|
|
|
|
|
|
|
|
| _esc = escalation_gate(effective)
|
| if _esc and _esc.topics == ["ช่องทางติดต่อ"] and not _esc.has_spec_question:
|
|
|
|
|
|
|
| reask = getattr(self.plugins.get(st.brand), "pending_reask", None) \
|
| if pending_before is not None and not brand_pending(pending_before) and st.brand else None
|
| q = reask(st.brands[st.brand]) if reask else None
|
| if q:
|
| st.pending, st.pending_question = pending_before, pending_q_before
|
| return Decision("ask", st.brand, f"{_esc.reply_th}\n\n{q}", "ขอช่องทางติดต่อกลางการถาม → ถามข้อเดิมต่อ",
|
| topic="escalate", effective_text=effective, state=st)
|
| return Decision("escalate", st.brand, _esc.reply_th, "ขอช่องทางติดต่อ",
|
| topic="escalate", effective_text=effective, state=st)
|
|
|
|
|
| if route.kind == "conflict_strong":
|
| st.pending, st.pending_question = "brand", effective
|
| conflict = [self.plugins[k].info for k in route.conflict]
|
| only_image = all(v == _router.IMAGE_EVIDENCE for v in route.evidence.values())
|
| why = ("รูปไม่ระบุแบรนด์ (บางแบรนด์ยังไม่มีตัวให้คะแนนรูป)" if only_image
|
| else f"เจอรหัสของ{T.all_brands_th(len(conflict))}")
|
| return Decision("ask", None, T.clarify_conflict(conflict), why,
|
| topic="brand", effective_text=effective, state=st)
|
| brand = route.brand if route.brand else st.brand
|
| if route.brand and st.brand and route.brand != st.brand:
|
|
|
| for k, p in self.plugins.items():
|
| st.brands[k] = p.on_brand_switch(st.brands[k])
|
| if brand is None and has(STAINLESS_RX, effective):
|
|
|
| n = len(self.plugins)
|
|
|
|
|
| st.pending, st.pending_question = "brand", effective
|
| return Decision("cross_brand", None, self.cross_brand_th(CROSS_BRAND_STAINLESS),
|
| f"SS ไม่ระบุแบรนด์ → ตอบ{T.all_brands_th(n)}",
|
| topic="material", effective_text=effective, state=st)
|
| if brand is None:
|
| st.pending, st.pending_question = "brand", effective
|
| return Decision("ask", None, self.ask_brand_th, "ยังไม่รู้แบรนด์", topic="brand",
|
| effective_text=effective, state=st)
|
| st.brand = brand
|
|
|
| plugin = self.plugins[brand]
|
| img = image if (has_image and image is not None and image.brand == brand) else None
|
| use_image = has_image
|
| if carried is not None and carried.brand == brand:
|
| img, use_image = carried, True
|
| d = plugin.decide(DecideInput(text=effective, signal=signals[brand], has_image=use_image,
|
| image=img, answer=text if pending_before else ""), st.brands[brand])
|
| if d.pending:
|
|
|
|
|
|
|
| st.pending, st.pending_question = d.pending, (d.effective_text or effective)
|
| if carried is not None and carried.brand == brand:
|
| d.payload = {**(d.payload or {}), "image_carried": True}
|
| d.state = st
|
| return d
|
|
|
|
|
|
|
| def cross_brand_th(self, topic: str) -> str:
|
| """สร้างจากข้อมูลจริงของทุกแบรนด์ทุกครั้ง — แต่ละแบรนด์เขียนส่วนของตัวเอง"""
|
| sections = [(p.info, p.cross_brand_section(topic)) for p in self.plugins.values()]
|
| sections = [(i, s) for i, s in sections if s is not None]
|
| lines = [f"สินค้าที่เกี่ยวกับสแตนเลส (SS) ของ{T.all_brands_th(len(self.plugins))}ครับ", ""]
|
| for n, (info, sec) in enumerate(sections):
|
| if n:
|
| lines.append("")
|
| lines.append(f"**{info.name} ({info.kind_th})**")
|
| lines += sec
|
| lines += ["", "สนใจแบรนด์ไหนหรือรุ่นไหน บอกได้เลยครับ จะให้รายละเอียดสเปกเพิ่ม"]
|
| return "\n".join(lines)
|
|
|
