feat: 可配置方案引擎通用原型（Mock 检索）

需求未定稿时可先跑通追问/判定/出方案；槽位与判定树外置 YAML，后续可接 Ragflow。

Co-authored-by: haijd <haijd@163.com>

engine/app/session_store.py

@@ -0,0 +1,115 @@
+from __future__ import annotations
+
+import re
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+@dataclass
+class Session:
+    id: str
+    slots: dict[str, str] = field(default_factory=dict)
+    messages: list[dict[str, Any]] = field(default_factory=list)
+    phase: str = "collecting"  # collecting | scheme_ready | handoff
+    last_scheme: dict[str, Any] | None = None
+    created_at: str = field(default_factory=_now)
+    updated_at: str = field(default_factory=_now)
+
+
+class SessionStore:
+    def __init__(self) -> None:
+        self._sessions: dict[str, Session] = {}
+
+    def create(self) -> Session:
+        session = Session(id=str(uuid.uuid4()))
+        self._sessions[session.id] = session
+        return session
+
+    def get(self, session_id: str) -> Session | None:
+        return self._sessions.get(session_id)
+
+    def touch(self, session: Session) -> None:
+        session.updated_at = _now()
+
+
+store = SessionStore()
+
+
+def looks_like_handoff(text: str, phrases: list[str]) -> bool:
+    t = text.strip().lower()
+    return any(p.lower() in t for p in phrases)
+
+
+def naive_fill_slots(message: str, slots_meta: list[dict[str, Any]], current: dict[str, str]) -> dict[str, str]:
+    """每轮通常只推进一个槽：先做明确关键词，再把整句写入当前优先空槽。"""
+    updated = dict(current)
+    text = message.strip()
+    if not text:
+        return updated
+
+    touched = False
+
+    def set_slot(slot_id: str, value: str) -> None:
+        nonlocal touched
+        if not updated.get(slot_id):
+            updated[slot_id] = value
+            touched = True
+
+    if re.search(r"A类|传感", text, re.I):
+        set_slot("tech_category", "A类传感")
+    elif re.search(r"B类|控制", text, re.I):
+        set_slot("tech_category", "B类控制")
+    elif "不确定" in text:
+        set_slot("tech_category", "不确定")
+
+    if "产线" in text:
+        set_slot("scenario", "产线")
+    elif "户外" in text:
+        set_slot("scenario", "户外监测")
+    elif "实验室" in text:
+        set_slot("scenario", "实验室")
+
+    # 本轮若已通过关键词写入至少一个槽，就不再把整句灌进其它空槽
+    if touched:
+        return updated
+
+    pending = sorted(
+        [s for s in slots_meta if s.get("required") and not updated.get(s["id"])],
+        key=lambda s: s.get("priority", 100),
+    )
+    if pending:
+        updated[pending[0]["id"]] = text
+        return updated
+
+    optional = sorted(
+        [s for s in slots_meta if not s.get("required") and not updated.get(s["id"])],
+        key=lambda s: s.get("priority", 100),
+    )
+    if optional:
+        updated[optional[0]["id"]] = text
+    return updated
+
+
+def next_slot_prompt(slots_meta: list[dict[str, Any]], filled: dict[str, str]) -> dict[str, str] | None:
+    pending = sorted(
+        [s for s in slots_meta if s.get("required") and not (filled.get(s["id"]) or "").strip()],
+        key=lambda s: s.get("priority", 100),
+    )
+    if not pending:
+        # optional slots: ask once if empty
+        optional = sorted(
+            [s for s in slots_meta if not s.get("required") and not (filled.get(s["id"]) or "").strip()],
+            key=lambda s: s.get("priority", 100),
+        )
+        if optional:
+            s = optional[0]
+            return {"id": s["id"], "label": s["label"], "prompt": s["prompt"]}
+        return None
+    s = pending[0]
+    return {"id": s["id"], "label": s["label"], "prompt": s["prompt"]}