74 lines
2.9 KiB
Python
74 lines
2.9 KiB
Python
"""客服转人工的最小必要上下文构造。
|
||
|
||
摘要完全由已持久化的会话消息和受控运行结果确定性生成，不调用模型，也不读取账户、画像或
|
||
长期记忆。其目的只是让管理员理解转接缘由，而不是向用户承诺已受理或已完成处理。
|
||
"""
|
||
|
||
from collections.abc import Sequence
|
||
from dataclasses import dataclass
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
from app.core.contracts import SourceReference
|
||
from app.core.conversation_privacy import sanitize_customer_service_message
|
||
from app.model.conversation import ConversationMessage
|
||
|
||
MAX_SUMMARY_MESSAGES = 6
|
||
MAX_SUMMARY_MESSAGE_CHARACTERS = 280
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CustomerServiceHandoverContext:
|
||
"""写入工单、审计与 Outbox 的统一、安全转接上下文。"""
|
||
|
||
reason_detail: str
|
||
conversation_summary: str
|
||
source_references: list[dict[str, Any]]
|
||
event_metadata: dict[str, Any]
|
||
|
||
|
||
def _safe_content(content: str) -> str:
|
||
"""二次脱敏和单条截断，确保历史会话也不会把凭据扩散到工单。"""
|
||
sanitized = sanitize_customer_service_message(content).strip()
|
||
if len(sanitized) > MAX_SUMMARY_MESSAGE_CHARACTERS:
|
||
return f"{sanitized[:MAX_SUMMARY_MESSAGE_CHARACTERS - 6]}[截断]"
|
||
return sanitized
|
||
|
||
|
||
def build_customer_service_handover_context(
|
||
*, reason_code: str, clarification_round: int, confidence: Decimal | None,
|
||
source_references: Sequence[SourceReference],
|
||
messages: Sequence[ConversationMessage],
|
||
) -> CustomerServiceHandoverContext:
|
||
"""构造管理员可读但不包含原始敏感凭据的转人工背景。"""
|
||
safe_references = [reference.model_dump(mode="json") for reference in source_references]
|
||
safe_messages = tuple(messages[-MAX_SUMMARY_MESSAGES:])
|
||
summary_lines = [
|
||
f"转接原因：{reason_code}",
|
||
f"已进行澄清轮次：{clarification_round}",
|
||
f"已核验知识来源数：{len(safe_references)}",
|
||
"最近会话（已脱敏）：",
|
||
]
|
||
if confidence is not None:
|
||
summary_lines.insert(2, f"意图置信度：{confidence}")
|
||
for message in safe_messages:
|
||
role = "用户" if message.role == "user" else "助手"
|
||
summary_lines.append(f"{role}：{_safe_content(message.content)}")
|
||
conversation_summary = "\n".join(summary_lines)
|
||
reason_detail = (
|
||
f"系统自动转接；原因={reason_code}；澄清轮次={clarification_round}；"
|
||
f"知识来源数={len(safe_references)}"
|
||
)
|
||
return CustomerServiceHandoverContext(
|
||
reason_detail=reason_detail,
|
||
conversation_summary=conversation_summary,
|
||
source_references=safe_references,
|
||
event_metadata={
|
||
"reason_code": reason_code,
|
||
"clarification_round": clarification_round,
|
||
"confidence": str(confidence) if confidence is not None else None,
|
||
"source_references": safe_references,
|
||
"conversation_summary": conversation_summary,
|
||
},
|
||
)
|