Cursor Agentandhaijd
85ee9020c6
改为公司定位与 TagPlus-T / BlueTag T10 / GAEA G1000 选型说明，替换原先 Fresh Tag 1 单品样例。 Co-authored-by: haijd <haijd@163.com>
125 lines
4.1 KiB
Python
125 lines
4.1 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"TagPlus|USB|PDF", text, re.I):
|
|
set_slot("tech_category", "TagPlus USB多次记录")
|
|
elif re.search(r"BlueTag|蓝牙|不开箱", text, re.I):
|
|
set_slot("tech_category", "BlueTag蓝牙手机读数")
|
|
elif re.search(r"GAEA|网关|实时", text, re.I):
|
|
set_slot("tech_category", "GAEA网关实时仓储")
|
|
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", "实验室小仓")
|
|
elif any(k in text for k in ("冷库", "仓储", "仓库")):
|
|
set_slot("scenario", "冷库仓储")
|
|
elif any(k in text for k in ("冷链", "药品", "疫苗", "生鲜", "冷冻", "保温箱", "运输")):
|
|
set_slot("scenario", text if len(text) <= 40 else "冷链运输")
|
|
if "Fresh Tag" in text or "fresh tag" in text.lower():
|
|
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"]}
|