| """
|
| core/protocol.py
|
| ================
|
| สัญญาที่ปลั๊กอินแบรนด์ต้องตอบ — แกนกลางรู้จักแบรนด์ผ่านไฟล์นี้เท่านั้น
|
|
|
| หนึ่งแบรนด์ = หนึ่งโฟลเดอร์ brands/<key>/ ที่มี brand.yaml + plugin.py
|
| plugin.py ต้องมีฟังก์ชัน create(ctx: PluginContext) -> BrandPlugin
|
|
|
| ข้อห้ามของไฟล์นี้และทุกไฟล์ใน core/
|
| · ไม่มีชื่อแบรนด์สักคำ (ตรวจด้วย grep ใน training/run_gates.py)
|
| · ไม่ import torch / qdrant / genai — เทสต์ด้วย python ล้วนได้
|
|
|
| คำถามหกข้อจาก HANDOFF ข้อ 3 → เมธอดที่ตอบ
|
| 1. รู้จักรหัสในข้อความนี้ไหม มั่นใจแค่ไหน detect · domain_terms
|
| 2. ค้นข้อมูลอย่างไร build_turn (ปลั๊กอินเลือกวิธีค้นเอง)
|
| 3. มีเกตอะไร เรียงลำดับไหน gate_chain · decide
|
| 4. ใส่กฎอะไรลง prompt build_turn → LLMTurn.system_prompt
|
| 5. เก็บสถานะอะไรข้ามเทิร์น empty_state · has_focus · on_brand_switch · sanitize_state
|
| 6. มีโมเดลรูปไหม collection ชื่ออะไร brand.yaml › image + image_scorer
|
| """
|
| from __future__ import annotations
|
|
|
| from dataclasses import dataclass, field
|
| from typing import Any, Callable, Literal, Optional, Protocol, Sequence, runtime_checkable
|
|
|
| PROTOCOL_VERSION = 1
|
|
|
| Strength = Literal["none", "weak", "strong"]
|
| Action = Literal["reject_image", "smalltalk", "refuse_offtopic", "ask",
|
| "escalate", "cross_brand", "search", "answer"]
|
|
|
|
|
|
|
|
|
|
|
|
|
| @dataclass(frozen=True)
|
| class BrandInfo:
|
| """ข้อความที่แกนกลางใช้ประกอบเทมเพลตรวมหลายแบรนด์ ทุกช่องมาจาก brand.yaml
|
|
|
| ตัวอย่างการใช้ในแกนกลาง (สองแบรนด์ต้องได้ข้อความเดิมทุกตัวอักษร)
|
| ถามแบรนด์ "{order}. {menu_line_th}" ทีละบรรทัด
|
| สัญญาณชนกัน "{name} ({kind_th})" ต่อกันด้วย " หรือ "
|
| นอกเรื่อง "{offtopic_th}" ต่อกันด้วย " และ" (ไม่มีช่องว่างหลัง และ — ตามข้อความเดิม)
|
| ทักทาย "{category_th}" ต่อกันด้วย "หรือ"
|
| """
|
| key: str
|
| name: str
|
| order: int
|
| kind_th: str
|
| menu_line_th: str
|
| offtopic_th: str
|
| category_th: str
|
| answer_patterns: tuple[str, ...] = ()
|
|
|
|
|
|
|
|
|
|
|
|
|
| @dataclass
|
| class BrandSignal:
|
| """ผลของ detect() — แกนกลางเอาไปเทียบข้ามแบรนด์ใน core/router.py
|
|
|
| strength strong = รหัส/ชื่อแบรนด์/รูปที่โมเดลของแบรนด์นี้รับ · weak = คำอุตสาหกรรม/งาน · none
|
| has_code เจอรหัสสินค้าจริงในข้อความ (ใช้ตัดสินว่าเป็นคำถามเรื่องสินค้า แยกจาก strength
|
| เพราะชื่อแบรนด์เป็น strong แต่ไม่ใช่รหัส)
|
| parsed ของที่ปลั๊กอินแยกได้ แกนกลางไม่เปิดดู ส่งคืนเข้า decide() ตัวเดิม
|
| ไม่ต้องแยกรหัสซ้ำสองรอบ และไม่มีชนิดข้อมูลของแบรนด์รั่วเข้าแกน
|
| """
|
| strength: Strength = "none"
|
| evidence: str = ""
|
| has_code: bool = False
|
| parsed: Any = None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| @dataclass
|
| class DecideInput:
|
| """สิ่งที่แกนกลางส่งให้ decide() หลังเลือกแบรนด์แล้ว"""
|
| text: str
|
| signal: BrandSignal
|
| has_image: bool = False
|
| image: Optional["ImageScore"] = None
|
| answer: str = ""
|
|
|
|
|
| @dataclass
|
| class Decision:
|
| """ผลการตัดสิน — สร้างโดยแกนกลาง (ก่อนเลือกแบรนด์) หรือโดยปลั๊กอิน (หลังเลือก)
|
|
|
| payload ข้อมูลของแบรนด์ที่ตัดสิน (รหัส slot คำค้น) ปลั๊กอินอ่านกลับใน build_turn
|
| reply_th มีค่าเมื่อเป็นเทมเพลต — ตอบได้โดยไม่เรียก LLM
|
| """
|
| action: Action
|
| brand: Optional[str]
|
| reply_th: str = ""
|
| reason: str = ""
|
| topic: Optional[str] = None
|
| effective_text: str = ""
|
| payload: dict = field(default_factory=dict)
|
| state: Any = None
|
| sales_handoff_th: str = ""
|
| sales_topics: list = field(default_factory=list)
|
| pending: Optional[str] = None
|
|
|
| @property
|
| def needs_llm(self) -> bool:
|
| return self.action in ("answer", "search")
|
|
|
|
|
|
|
|
|
|
|
|
|
| @dataclass
|
| class TurnInput:
|
| """หนึ่งเทิร์นที่แกนกลางส่งเข้าปลั๊กอินตอนต้องเรียก LLM
|
|
|
| ของที่ฝากข้ามเทิร์นมีสามช่อง (ตำแหน่งเดิมใน tuple ของ chat_interaction) แกนกลางไม่เปิดดู
|
| last_focus รหัสที่พูดถึงในเทิร์น LLM ล่าสุด (list)
|
| last_image สัญญาณรูปล่าสุดที่ปลั๊กอินฝากไว้
|
| last_context บริบทค้น/กราฟล่าสุดที่ปลั๊กอินฝากไว้
|
| """
|
| user_text: str
|
| image: Any
|
| history: list
|
| session_id: str
|
| effective_text: str = ""
|
| followup: Any = False
|
| last_focus: list = field(default_factory=list)
|
| last_image: Any = None
|
| last_context: Any = None
|
| image_score: Optional["ImageScore"] = None
|
| log: Any = None
|
|
|
|
|
| @dataclass
|
| class LLMTurn:
|
| """สิ่งที่ปลั๊กอินขอให้แกนกลางส่งเข้า LLM — แกนกลางเป็นคนเดียวที่คุยกับ LLM"""
|
| system_prompt: str
|
| user_parts: list
|
| tools: Sequence[Callable] = ()
|
| debug: list = field(default_factory=list)
|
| state: Any = None
|
| direct_reply: Optional[str] = None
|
|
|
|
|
| @dataclass
|
| class TurnOutput:
|
| reply: str
|
| extra_messages: list = field(default_factory=list)
|
| debug: list = field(default_factory=list)
|
|
|
|
|
| carry_fn: Optional[Callable[[str], tuple]] = None
|
|
|
| state_patch: dict = field(default_factory=dict)
|
|
|
|
|
|
|
|
|
|
|
|
|
| @dataclass
|
| class ImageScore:
|
| """คะแนนจากโมเดลรูปของหนึ่งแบรนด์ — core/image_router.py เทียบข้ามแบรนด์
|
|
|
| แกนกลางใช้แค่ confidence กับ accepted ตัดสิน ที่เหลือส่งคืนปลั๊กอิน
|
| accepted = ระบบรูปของแบรนด์นี้หาสินค้าเจอ (ไม่เจอ = ถือว่ารูปไม่ใช่ของแบรนด์นี้)
|
| เกณฑ์ความมั่นใจระดับรุ่น/ซีรีส์ ปลั๊กอินใช้เองใน decide (thresholds.json)
|
| ห้ามใส่เกรดวัสดุหรือสภาวะเปียก/แห้งในนี้ — อนุมานจากรูปไม่ได้ (HANDOFF ข้อ 4)
|
| """
|
| brand: str
|
| confidence: float
|
| accepted: bool
|
| label: Optional[str] = None
|
| group: Optional[str] = None
|
| dominant: bool = False
|
| product_ref: str = ""
|
| raw: Any = None
|
|
|
|
|
| @runtime_checkable
|
| class ImageScorer(Protocol):
|
| def score(self, image: Any, text: str) -> ImageScore: ...
|
|
|
|
|
|
|
|
|
|
|
|
|
| @dataclass
|
| class PluginContext:
|
| """registry สร้างแล้วส่งให้ create() ของปลั๊กอิน
|
|
|
| shared ของที่โหลดครั้งเดียวใช้ร่วมกันทุกแบรนด์ (embedding model, device, logger)
|
| ใส่โดยตัวเริ่มระบบ ไม่ใช่โดยแกนกลาง — เทสต์ส่ง {} ได้ ปลั๊กอินต้องทำงานได้
|
| ในโหมดไม่มีโมเดล (ตัดสินอย่างเดียว) ไม่งั้นชุดเกตรันไม่ได้
|
| """
|
| brand_dir: str
|
| config: dict
|
| image_model: Any = None
|
| shared: dict = field(default_factory=dict)
|
|
|
|
|
|
|
|
|
|
|
|
|
| @runtime_checkable
|
| class BrandPlugin(Protocol):
|
| info: BrandInfo
|
| gate_chain: tuple[str, ...]
|
| image_reject_th: str
|
|
|
|
|
| def detect(self, text: str) -> BrandSignal:
|
| """อ่านข้อความอย่างเดียว ไม่ดูรูป (รูปผ่าน image_scorer) · ห้ามมี side effect"""
|
| ...
|
|
|
| def domain_terms(self) -> Sequence[str]:
|
| """คำที่แปลว่าเป็นคำถามเรื่องสินค้าของแบรนด์นี้ (รวมกับคำกลางใน core เพื่อกันนอกเรื่อง)"""
|
| ...
|
|
|
|
|
| def empty_state(self) -> dict: ...
|
|
|
| def has_focus(self, st: dict) -> bool:
|
| """มีรุ่นค้างอยู่ในบทสนทนาไหม — ใช้ตัดสินว่าข้อความสั้นเป็นคำถามต่อเนื่อง"""
|
| ...
|
|
|
| def on_brand_switch(self, st: dict) -> dict:
|
| """ลูกค้าเปลี่ยนแบรนด์กลางบทสนทนา — คืนสถานะที่ล้างแล้ว (ปลั๊กอินเลือกว่าล้างอะไร)"""
|
| ...
|
|
|
| def sanitize_state(self, raw: dict) -> dict:
|
| """กรองสถานะที่เบราว์เซอร์ส่งกลับมา เก็บเฉพาะรหัสที่มีจริงในฐานข้อมูลของแบรนด์นี้
|
| raw คือฟิลด์ของแบรนด์นี้ที่ตัด prefix แล้ว (webapi ตัดความยาว list/dict ให้ก่อน)
|
| ค่าที่ไม่รู้จักต้องทิ้ง ไม่ใช่ raise"""
|
| ...
|
|
|
| def resumable(self, st: dict) -> bool:
|
| """สถานะนี้มีของพอให้กู้บทสนทนาไหม (ใช้ตอนเบราว์เซอร์ส่งประวัติกลับมา)"""
|
| ...
|
|
|
|
|
| def decide(self, inp: DecideInput, st: dict) -> Decision:
|
| """ตัดสินหลังเลือกแบรนด์แล้ว: ask / escalate / search / answer / reject_image
|
| แก้ st ได้ (แกนกลางส่งสำเนามา) · ห้ามเรียก LLM · ห้ามเข้าเน็ต
|
| ถามกลับแล้วรอคำตอบ → ตั้ง Decision.pending (แกนกลางเก็บคำถามเดิมให้เอง)"""
|
| ...
|
|
|
|
|
| def build_turn(self, decision: Decision, turn: TurnInput) -> LLMTurn:
|
| """เรียกเฉพาะเมื่อ decision.needs_llm · ค้นข้อมูลด้วยวิธีของแบรนด์ แล้วคืน prompt + tools"""
|
| ...
|
|
|
| def finish_turn(self, decision: Decision, turn: TurnInput, llm: LLMTurn, result: Any) -> TurnOutput:
|
| """หลัง LLM ตอบ — result คือ core.llm.LLMResult (response · session · tool_calls)
|
| ดึงข้อความ ต่อของที่โค้ดต้องเขียนเอง (รูป datasheet) และคืน carry_fn ของเทิร์นหน้า"""
|
| ...
|
|
|
|
|
| def cross_brand_section(self, topic: str) -> Optional[list[str]]:
|
| """ส่วนของแบรนด์นี้ในคำตอบรวมทุกแบรนด์ (เช่น "สแตนเลส" ไม่ระบุแบรนด์)
|
| คืน None = แบรนด์นี้ไม่ตอบหัวข้อนี้ · สร้างจากข้อมูลจริงทุกครั้ง ไม่ฮาร์ดโค้ดรายชื่อรุ่น"""
|
| ...
|
|
|
|
|
| def image_scorer(self) -> Optional[ImageScorer]:
|
| """None = แบรนด์นี้ไม่รับรูป (หรือ enabled: auto แล้วยังไม่มีโมเดล/collection)"""
|
| ...
|
|
|
