116 lines
3.5 KiB
Python
116 lines
3.5 KiB
Python
 
|
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"]}
|