bd6a7ffa32
Backend:
- Channel/initiation now driven by a SCENARIO chosen at chat start, not baked into the
persona: social (customer opens), f2f_call (seller must open, proactive), recontact
(customer re-contacts after 1-3 months).
- /chat/start accepts {scenario}; session stores scenario + internal{turns,score}.
- persona_reply takes scenario + adapts tone; system-role transcript entries are fed to
the persona as hidden scene notes.
- JUDGE updated for realism: good response can WIN even in hard/tough-text scenarios;
long/no-close chats (turns >~12) lose; pushy/ignoring-need loses. Efficiency rewarded.
Frontend:
- Scenario picker before chat (choose Social / Face-to-face-call / Re-contact).
- Chat thread renders role=system as a centered time-lapse/scene note.
- Choose-scenario i18n (EN+TH).
Rebuilt dist.
241 lines
9.4 KiB
Python
241 lines
9.4 KiB
Python
"""Chat/session API: start a one-shot session, send messages, finish + debrief."""
|
|
from __future__ import annotations
|
|
|
|
from flask import Blueprint, jsonify, request
|
|
|
|
from ..llm import LLMError
|
|
from ..services.simulator import Simulator
|
|
from .helpers import ApiError, current_user, require_auth, require_roles
|
|
|
|
chat_bp = Blueprint("chat", __name__)
|
|
|
|
|
|
def _stores():
|
|
from flask import current_app
|
|
|
|
return {
|
|
"groups": current_app.extensions["group_store"],
|
|
"sessions": current_app.extensions["session_store"],
|
|
"llm": current_app.extensions["llm"],
|
|
}
|
|
|
|
|
|
def _sim(group, persona):
|
|
llm = _stores()["llm"]
|
|
if not llm:
|
|
raise ApiError("LLM not configured", 500)
|
|
return Simulator(llm)
|
|
|
|
|
|
SCENARIOS = {
|
|
"social": {
|
|
"label": "Social Media",
|
|
"init": "customer",
|
|
"system_preamble": "💬 ช่องทาง ข้อความโซเชียล — ลูกค้าทักมาหาคุณก่อน (โทนสั้น ทักๆ ตามสไตล์แชท)",
|
|
"adapt": "You are chatting on a social-messaging app (LINE-style). Keep replies SHORT, casual, and quick — one line to a few lines. The customer opened the chat.",
|
|
},
|
|
"f2f_call": {
|
|
"label": "พบหน้า / โทรศัพท์",
|
|
"init": "seller",
|
|
"system_preamble": "📞 สถานการณ์ พบหน้าหรือโทรศัพท์ — คุณต้องเป็นฝ่ายเปิดการสนทนาเชิงรุกกับลีด (lead)",
|
|
"adapt": "This is a face-to-face or phone sales situation. You must sound natural and conversational like a live talk. The seller will open — you are the lead they are reaching out to.",
|
|
},
|
|
"recontact": {
|
|
"label": "ลูกค้ากลับมาติดต่อ (เคยได้ข้อมูล 1-3 เดือน)",
|
|
"init": "customer",
|
|
"system_preamble": "⏳ 1-3 เดือนผ่านไป... ลูกค้าคนนี้เคยได้รับข้อมูลสินค้าไปแล้ว ตอนนี้กลับมาติดต่อคุณอีกครั้ง (พร้อมตัดสินใจมากขึ้น)",
|
|
"adapt": "This customer researched you 1-3 months ago. They have already read about the product and are now more ready to decide. They re-contacted you on a messaging channel. Keep replies natural and fairly concise.",
|
|
},
|
|
}
|
|
|
|
|
|
def _scenario_config(scenario: str, persona: dict):
|
|
cfg = SCENARIOS.get(scenario, SCENARIOS["social"])
|
|
# OR with persona's own initiation preference: f2f/call forces seller-first unless persona
|
|
# is strongly customer-initiated (we let scenario win, but keep persona special handling).
|
|
return cfg, cfg["init"]
|
|
|
|
|
|
def _get_ready_group(s, gid: str) -> dict:
|
|
"""Org-scoped group access for trainees + require ready status (IDOR defense)."""
|
|
group = s["groups"].get_or_none(gid)
|
|
if not group or group.get("status") != "ready":
|
|
raise ApiError("group not ready", 404)
|
|
actor = current_user()
|
|
# super_admin can access any; otherwise owner (for personal groups) + same org.
|
|
owner = group.get("owner_user_id")
|
|
if actor.get("role") != "super_admin":
|
|
if owner and owner != actor["id"]:
|
|
raise ApiError("permission denied", 403)
|
|
if group.get("org_id") != actor.get("org_id"):
|
|
raise ApiError("permission denied", 403)
|
|
return group
|
|
|
|
|
|
@chat_bp.post("/<gid>/personas/<pid>/chat/start")
|
|
@require_auth
|
|
@require_roles("user")
|
|
def start_session(gid: str, pid: str):
|
|
s = _stores()
|
|
group = _get_ready_group(s, gid)
|
|
persona = s["groups"].get_persona(gid, pid)
|
|
if not persona:
|
|
raise ApiError("persona not found", 404)
|
|
actor = current_user()
|
|
# Scenario chosen by the trainee at chat start (not baked into the persona).
|
|
body = request.get_json(silent=True) or {}
|
|
scenario = (body.get("scenario") or "social").strip().lower()
|
|
if scenario not in ("social", "f2f_call", "recontact"):
|
|
scenario = "social"
|
|
scenario_meta, init_mode = _scenario_config(scenario, persona)
|
|
# One-shot: reject if already finished this persona
|
|
try:
|
|
session = s["sessions"].create(
|
|
user_id=actor["id"], group_id=gid, persona_id=pid,
|
|
persona_name=persona.get("name", "?"),
|
|
persona_meta={
|
|
"tier": persona.get("tier"),
|
|
"initiation_mode": init_mode,
|
|
"channel": persona.get("channel"),
|
|
"scenario": scenario,
|
|
},
|
|
)
|
|
except ValueError as exc:
|
|
raise ApiError(str(exc), 400)
|
|
|
|
s["sessions"].update(
|
|
session["id"],
|
|
scenario=scenario,
|
|
internal={**(session.get("internal") or {}), "turns": 0, "score": 50, "signals": []},
|
|
)
|
|
sim = _sim(group, persona)
|
|
# Seed messages: system note(s) then customer opener for customer-first scenarios.
|
|
seeded = list(session.get("messages", []))
|
|
if scenario_meta.get("system_preamble"):
|
|
seeded.append({"role": "system", "text": scenario_meta["system_preamble"]})
|
|
if init_mode == "customer":
|
|
opener = persona.get("opener") or "Hi, I saw your product and had a question."
|
|
seeded.append({"role": "customer", "text": opener})
|
|
s["sessions"].update(session["id"], messages=seeded)
|
|
sess = s["sessions"].get(session["id"])
|
|
return jsonify({
|
|
"session": sess,
|
|
"initiation_mode": init_mode,
|
|
"scenario": scenario,
|
|
"scenario_meta": scenario_meta,
|
|
})
|
|
|
|
|
|
@chat_bp.post("/<gid>/personas/<pid>/chat/send")
|
|
@require_auth
|
|
@require_roles("user")
|
|
def send_message(gid: str, pid: str):
|
|
s = _stores()
|
|
actor = current_user()
|
|
session = s["sessions"].active_for_persona(actor["id"], pid)
|
|
if not session or session.get("group_id") != gid:
|
|
raise ApiError("no active session for this persona", 404)
|
|
|
|
data = request.get_json(silent=True) or {}
|
|
text = (data.get("text") or "").strip()
|
|
if not text:
|
|
raise ApiError("message is empty")
|
|
if len(text) > 2000:
|
|
raise ApiError("message too long")
|
|
|
|
group = s["groups"].get_or_none(gid)
|
|
persona = s["groups"].get_persona(gid, pid) if group else None
|
|
if not group or not persona:
|
|
raise ApiError("session context missing", 404)
|
|
messages = list(session.get("messages", []))
|
|
messages.append({"role": "seller", "text": text})
|
|
scenario = session.get("scenario", "social") or "social"
|
|
adapt = SCENARIOS.get(scenario, SCENARIOS["social"]).get("adapt", "")
|
|
|
|
sim = _sim(group, persona)
|
|
try:
|
|
reply = sim.persona_reply(
|
|
persona=persona,
|
|
sales_kit=group.get("sales_kit") or {},
|
|
messages=messages,
|
|
internal=session.get("internal", {}),
|
|
scenario=scenario,
|
|
scenario_adapt=adapt,
|
|
)
|
|
except LLMError as exc:
|
|
raise ApiError(f"LLM error: {exc}", 500)
|
|
messages.append({"role": "customer", "text": reply})
|
|
|
|
# Per-turn tracking: count exchanged turns + nudge score down as the chat drags.
|
|
internal = session.get("internal", {}) or {}
|
|
internal.setdefault("turns", 0)
|
|
internal["turns"] = internal.get("turns", 0) + 1
|
|
|
|
s["sessions"].update(session["id"], messages=messages, internal=internal)
|
|
return jsonify({"reply": reply, "messages": messages})
|
|
|
|
|
|
@chat_bp.post("/<gid>/personas/<pid>/chat/finish")
|
|
@require_auth
|
|
@require_roles("user")
|
|
def finish_session(gid: str, pid: str):
|
|
"""End the chat and produce the debrief via the judge-LLM (reveals latent fields)."""
|
|
s = _stores()
|
|
actor = current_user()
|
|
session = s["sessions"].active_for_persona(actor["id"], pid)
|
|
if not session or session.get("group_id") != gid:
|
|
raise ApiError("no active session for this persona", 404)
|
|
group = s["groups"].get_or_none(gid)
|
|
persona = s["groups"].get_persona(gid, pid)
|
|
|
|
sim = _sim(group, persona)
|
|
messages = session.get("messages", [])
|
|
try:
|
|
verdict = sim.judge(
|
|
persona=persona, messages=messages, internal=session.get("internal", {})
|
|
)
|
|
except LLMError as exc:
|
|
raise ApiError(f"LLM error: {exc}", 500)
|
|
|
|
outcome = "won" if verdict.get("outcome") == "won" else "lost"
|
|
debrief = {
|
|
**verdict,
|
|
"revealed_persona": {
|
|
"pains": persona.get("pains", []),
|
|
"income": persona.get("income", ""),
|
|
"personality": persona.get("personality", ""),
|
|
"budget": persona.get("budget", ""),
|
|
"negotiation_levers": persona.get("negotiation_levers", []),
|
|
"opener": persona.get("opener", ""),
|
|
"background": persona.get("background", ""),
|
|
},
|
|
}
|
|
s["sessions"].update(
|
|
session["id"],
|
|
status="finished",
|
|
outcome=outcome,
|
|
debrief=debrief,
|
|
internal=session.get("internal", {}),
|
|
)
|
|
return jsonify({"session": s["sessions"].get(session["id"]), "debrief": debrief})
|
|
|
|
|
|
@chat_bp.get("/sessions")
|
|
@require_auth
|
|
def my_sessions():
|
|
s = _stores()
|
|
uid = current_user()["id"]
|
|
sessions = s["sessions"].list_for_user(uid)
|
|
return jsonify({"sessions": sessions})
|
|
|
|
|
|
@chat_bp.get("/sessions/<sid>")
|
|
@require_auth
|
|
@require_roles("user")
|
|
def get_session(sid: str):
|
|
s = _stores()
|
|
session = s["sessions"].get_or_none(sid)
|
|
if not session or session.get("user_id") != current_user()["id"]:
|
|
raise ApiError("session not found", 404)
|
|
return jsonify({"session": session})
|