Voice Agent 怎么安全转人工？意图阈值、工单、上下文交接与 CRM 回流实战
·
Voice Agent 的转人工不是调用一次
transfer_to_human()，而是一条需要保证状态一致性的业务链路。一个相对安全的实现，至少要同时解决五件事：什么时候转、AI 什么时候闭嘴、交给人工什么信息、工单如何可靠落库、CRM 超时或重复请求时怎么办。
本文用 FastAPI、SQLite 和事务 Outbox 搭建一个可以复制运行的最小 Demo，包含：
- 显式转人工、低置信度、敏感意图和连续失败等触发条件；
- 可校准的风险评分与硬规则；
AI_ACTIVE → HANDOFF_PENDING → TICKET_CREATED → HUMAN_ACTIVE状态机；- 可审计的上下文快照；
- 工单与 Outbox 同事务写入；
- CRM 重试和幂等去重；
- 转人工后阻止旧的 AI 音频继续输出；
- 人工接通、工单创建和转接失败的独立状态。
先说明边界：文中的 0.65 阈值及各项权重只用于演示代码，不是行业标准，也不能直接复制到生产环境。真实阈值必须使用业务标注集、误转成本和漏转成本重新校准。
一、为什么“创建一张工单”不等于转人工成功
很多实现把流程写成：
if need_human:
create_ticket()
return "正在为您转接人工"
这段代码至少有四个问题。
1. 工单创建成功，但没有人工接起
CRM 返回了一个工单编号，只能说明记录已经进入系统，不能说明：
- 已经分配坐席；
- 坐席已经看到上下文；
- 坐席接受了会话；
- 用户已经和人工建立连接。
因此下面几个状态不能合并：
HANDOFF_PENDING 已进入转人工流程
TICKET_CREATED CRM 工单已创建
HUMAN_ASSIGNED 已分配人工
HUMAN_ACTIVE 人工已接管
HANDOFF_FAILED 转接失败
2. AI 仍在继续输出
转人工触发时，LLM、TTS 或客户端音频队列可能还有内容。
如果只创建工单，不冻结 AI 输出，就可能出现：
AI：我正在为您查询退款规则……
人工：您好，我来处理。
AI：根据平台规定，您的订单暂不支持退款。
这不是简单的体验问题。涉及退款、投诉、合同或账户操作时，AI 和人工同时给出不同结论，会直接产生业务风险。
3. CRM 超时后重复创建工单
客户端调用 CRM 的 POST /tickets，连接在返回前断开。
此时调用方并不知道：
- CRM 根本没收到请求；
- CRM 收到但还没处理；
- CRM 已创建工单，只是响应丢了。
如果不做幂等保护，自动重试可能创建两张甚至多张工单。
HTTP 规范只把 PUT、DELETE 和安全方法定义为幂等方法，POST 本身并不天然幂等，因此工单创建接口需要业务侧的去重键。RFC 9110：Idempotent Methods
4. 人工看到了一大段原始对话，却不知道要做什么
把最近 50 轮聊天原样塞给坐席，看起来信息很全，实际经常增加处理时间。
人工真正需要的是：
- 为什么转人工；
- 用户当前要解决什么问题；
- 已经确认了哪些身份和业务字段；
- AI 查过什么、做过什么；
- 哪些动作已经产生副作用；
- 哪些动作尚未完成；
- 用户最后一句话是什么；
- 当前风险和权限边界是什么。
所以转人工交接的核心不是“多传文本”，而是生成一份可验证的业务快照。
二、安全转人工的完整链路
用户输入 / ASR final
|
v
规则与风险判断
|
+---- 不转人工 ----> AI 继续处理
|
v
生成 handoff_id
|
v
冻结 AI 输出 + generation 失效
|
v
生成结构化上下文快照
|
v
同一数据库事务
├── 保存 handoff 记录
├── 更新会话状态
├── 写入 outbox 事件
└── 写入审计日志
|
v
Outbox Worker
|
v
CRM / 工单 API
|
+---- 超时/失败 ---> 重试、退避、死信
|
v
TICKET_CREATED
|
v
分配人工
|
v
HUMAN_ACTIVE
|
v
人工处理结果回流 CRM / 会话系统
这里最关键的是：
转人工决策、会话冻结和 Outbox 事件必须形成一个一致的状态变化。
如果数据库里显示“正在转人工”，但工单事件没有写入，后续系统就没有机会补偿。
三、哪些情况应该触发转人工
我会把触发条件分成两层。
3.1 硬规则
满足后直接进入转人工，不再依赖综合分数。
典型情况包括：
- 用户明确要求人工；
- 命中法律、欺诈、支付争议等敏感流程；
- 当前操作超出 AI 权限；
- 安全策略禁止 AI 继续回答；
- 连续多次工具调用失败；
- 身份验证不满足后续操作要求；
- 已产生业务冲突，必须由人工判断。
示例：
if signal.explicit_request:
should_handoff = True
if signal.sensitive_intent:
should_handoff = True
if signal.policy_blocked:
should_handoff = True
if signal.repeated_failures >= 3:
should_handoff = True
“用户明确要人工”不应该再被一个模型置信度拦截。
3.2 软评分
没有命中硬规则时，再综合多个弱信号：
低意图置信度
用户情绪风险
连续失败次数
工具/API 异常
上下文缺失
重复澄清
本文 Demo 使用：
risk_score =
0.45 × (1 - intent_confidence)
+ 0.25 × emotion_risk
+ 0.15 × min(repeated_failures / 3, 1)
+ 0.15 × tool_failed
当：
risk_score >= 0.65
时触发转人工。
例如：
intent_confidence = 0.42
emotion_risk = 0.70
repeated_failures = 2
tool_failed = True
得到：
0.45 × 0.58
+ 0.25 × 0.70
+ 0.15 × 2/3
+ 0.15 × 1
= 0.686
因此进入转人工。
但下面这组信号：
intent_confidence = 0.90
emotion_risk = 0.20
repeated_failures = 0
tool_failed = False
得分只有：
0.095
不应该仅凭轻微负面情绪就转人工。
3.3 情绪不能单独决定转人工
情绪识别容易受噪声、方言、语速和表达习惯影响。
生产环境更稳妥的做法是：
- 情绪只作为一项信号；
- 连续多个窗口升高后再触发；
- 设置进入阈值和退出阈值，避免频繁抖动；
- 投诉、支付争议等业务意图优先于情绪模型；
- 保留用户主动要求人工的最高优先级。
四、状态机：AI 什么时候必须闭嘴
推荐的最小状态机：
AI_ACTIVE
|
| 触发转人工
v
HANDOFF_PENDING
|
| CRM 工单创建成功
v
TICKET_CREATED
|
| 坐席分配
v
HUMAN_ASSIGNED
|
| 坐席接受会话
v
HUMAN_ACTIVE
|
| 问题解决
v
CLOSED
失败分支：
HANDOFF_PENDING
|
| CRM/API 多次失败
v
HANDOFF_FAILED
一旦从 AI_ACTIVE 进入 HANDOFF_PENDING，AI 就不应该继续产生新的业务结论。
我通常会给每轮生成任务带一个 generation：
会话当前 generation = 8
LLM 任务 generation = 8
转人工时：
会话 generation = 9
state = HANDOFF_PENDING
旧任务即使晚到，也会因为 generation 不一致而被丢弃：
can_emit = (
session.state == "AI_ACTIVE"
and task_generation == session.generation
)
这与上一篇音频队列里的 turn_id 思路相同：取消任务用于停止继续生产，generation 用于阻止迟到结果重新进入输出链路。
五、上下文交接应该传什么
5.1 建议传递的字段
{
"handoff_id": "ho_xxx",
"session_id": "session_001",
"reason_codes": [
"risk_score_threshold",
"tool_failed"
],
"decision_score": 0.686,
"case_overview": "用户申请退款，订单状态查询失败，需要人工核验",
"latest_user_message": "那你帮我转人工吧",
"intent": "refund_request",
"known_facts": {
"order_id": "ORDER-1008",
"issue_type": "refund"
},
"completed_actions": [
"已完成基础身份校验"
],
"pending_actions": [
"核验退款资格",
"确认原路退款账户"
],
"auth_context": {
"authenticated": true,
"auth_level": "strong",
"consent_to_share": true
}
}
5.2 不建议默认传递的内容
- 全量原始录音；
- 完整银行卡号；
- 身份证原文；
- 密码、验证码、Token；
- 与当前问题无关的历史会话；
- 未经过滤的模型内部提示词；
- 第三方 API 返回的全部原始对象。
OWASP 的日志安全建议也明确提醒，访问令牌、密码、连接字符串、加密密钥及敏感个人信息不应直接写入日志，而应删除、脱敏、哈希或加密。OWASP Logging Cheat Sheet
5.3 摘要不要只交给 LLM 自由发挥
如果让模型把整段聊天“总结一下”，可能出现：
- 把用户的猜测写成事实；
- 漏掉已执行动作；
- 把未确认身份写成已验证；
- 混淆订单、金额和时间；
- 忽略失败的工具调用。
更安全的方式是：
确定性字段 + 可选的 LLM 摘要
其中这些字段必须来自业务系统：
身份验证状态
订单号
已执行动作
待执行动作
权限状态
工具调用结果
转人工原因码
LLM 只负责生成便于人工阅读的 case_overview，输出后还要经过结构校验和字段白名单。
六、为什么需要事务 Outbox
最危险的写法是：
save_handoff_to_database()
call_crm_api()
可能出现：
数据库成功，CRM 失败
系统认为已转人工，但 CRM 根本没有工单。
反过来：
call_crm_api()
save_handoff_to_database()
又可能出现：
CRM 成功，数据库回滚
CRM 有一张无人追踪的工单，本地系统却认为没有转接。
事务 Outbox 的思路是：
同一个本地数据库事务：
1. 保存 handoff
2. 更新 session 状态
3. 保存 outbox event
4. 提交事务
独立 Worker 再读取 Outbox，把事件发送给 CRM。
AWS 的事务 Outbox 指南指出，这个模式用于解决“数据库写入 + 外部事件通知”的双写一致性问题；同时也提醒，下游可能收到重复消息，因此消费端仍需具备幂等能力。AWS Transactional Outbox Pattern
七、运行环境与目录
Python 3.12
FastAPI 0.115.12
Uvicorn 0.34.3
SQLite
目录：
voice-agent-handoff/
├── app.py
└── requirements.txt
requirements.txt：
fastapi==0.115.12
uvicorn[standard]==0.34.3
启动：
python3.12 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
uvicorn app:app --reload
八、完整代码：FastAPI + SQLite + Outbox
from __future__ import annotations
import json
import re
import sqlite3
import uuid
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import StrEnum
from typing import Annotated, Literal
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field
DB_PATH = "handoff_demo.db"
SOFT_THRESHOLD = 0.65
ALLOWED_SLOTS = {
"order_id",
"product",
"issue_type",
"preferred_callback_window",
}
ACTIVE_HANDOFF_STATES = (
"HANDOFF_PENDING",
"TICKET_CREATED",
"HUMAN_ASSIGNED",
"HUMAN_ACTIVE",
)
class SessionState(StrEnum):
AI_ACTIVE = "AI_ACTIVE"
HANDOFF_PENDING = "HANDOFF_PENDING"
TICKET_CREATED = "TICKET_CREATED"
HUMAN_ASSIGNED = "HUMAN_ASSIGNED"
HUMAN_ACTIVE = "HUMAN_ACTIVE"
HANDOFF_FAILED = "HANDOFF_FAILED"
CLOSED = "CLOSED"
class AuthContext(BaseModel):
authenticated: bool = False
consent_to_share: bool = False
auth_level: Literal["none", "weak", "strong"] = "none"
class HandoffSignal(BaseModel):
intent: str = Field(min_length=1, max_length=64)
intent_confidence: float = Field(ge=0, le=1)
emotion_risk: float = Field(ge=0, le=1)
repeated_failures: int = Field(default=0, ge=0, le=10)
explicit_request: bool = False
sensitive_intent: bool = False
policy_blocked: bool = False
tool_failed: bool = False
latest_user_text: str = Field(default="", max_length=500)
safe_slots: dict[str, str] = Field(default_factory=dict)
completed_actions: list[str] = Field(
default_factory=list,
max_length=20,
)
pending_actions: list[str] = Field(
default_factory=list,
max_length=20,
)
auth_context: AuthContext = Field(
default_factory=AuthContext,
)
class HumanAcceptRequest(BaseModel):
agent_id: str = Field(min_length=2, max_length=64)
@dataclass(frozen=True)
class Decision:
should_handoff: bool
score: float
reasons: tuple[str, ...]
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def connect() -> sqlite3.Connection:
conn = sqlite3.connect(DB_PATH, timeout=5)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
return conn
def init_db() -> None:
with connect() as conn:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS sessions (
session_id TEXT PRIMARY KEY,
state TEXT NOT NULL,
generation INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS handoffs (
handoff_id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
event_key TEXT NOT NULL,
status TEXT NOT NULL,
score REAL NOT NULL,
reasons_json TEXT NOT NULL,
summary_json TEXT NOT NULL,
ticket_id TEXT,
agent_id TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE(session_id, event_key),
FOREIGN KEY(session_id)
REFERENCES sessions(session_id)
);
CREATE TABLE IF NOT EXISTS outbox (
outbox_id TEXT PRIMARY KEY,
event_key TEXT NOT NULL UNIQUE,
event_type TEXT NOT NULL,
payload_json TEXT NOT NULL,
status TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS crm_tickets (
ticket_id TEXT PRIMARY KEY,
handoff_id TEXT NOT NULL UNIQUE,
payload_json TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS audit_logs (
audit_id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
handoff_id TEXT,
action TEXT NOT NULL,
details_json TEXT NOT NULL,
created_at TEXT NOT NULL
);
"""
)
@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
init_db()
yield
app = FastAPI(
title="Voice Agent Safe Handoff Demo",
lifespan=lifespan,
)
def mask_text(text: str) -> str:
text = text.replace("\r", " ").replace("\n", " ")
# 手机号
text = re.sub(
r"(?<!\d)1[3-9]\d{9}(?!\d)",
"[MOBILE]",
text,
)
# 身份证、银行卡等长数字串
text = re.sub(
r"(?<!\d)\d{13,19}[0-9Xx]?(?!\d)",
"[SENSITIVE_NUMBER]",
text,
)
return text[:500]
def sanitize_slots(slots: dict[str, str]) -> dict[str, str]:
return {
key: mask_text(str(value))
for key, value in slots.items()
if key in ALLOWED_SLOTS
}
def decide_handoff(signal: HandoffSignal) -> Decision:
reasons: list[str] = []
if signal.explicit_request:
reasons.append("explicit_handoff_request")
if signal.sensitive_intent:
reasons.append("sensitive_intent")
if signal.policy_blocked:
reasons.append("policy_blocked")
if signal.repeated_failures >= 3:
reasons.append("repeated_failures")
score = (
0.45 * (1 - signal.intent_confidence)
+ 0.25 * signal.emotion_risk
+ 0.15 * min(signal.repeated_failures / 3, 1)
+ 0.15 * int(signal.tool_failed)
)
score = round(score, 3)
hard_triggered = bool(reasons)
soft_triggered = score >= SOFT_THRESHOLD
if soft_triggered:
reasons.append("risk_score_threshold")
return Decision(
should_handoff=hard_triggered or soft_triggered,
score=score,
reasons=tuple(dict.fromkeys(reasons)),
)
def build_summary(
session_id: str,
signal: HandoffSignal,
decision: Decision,
) -> dict:
return {
"session_id": session_id,
"reason_codes": list(decision.reasons),
"decision_score": decision.score,
"intent": signal.intent,
"latest_user_message": mask_text(
signal.latest_user_text
),
"known_facts": sanitize_slots(signal.safe_slots),
"completed_actions": [
mask_text(item)
for item in signal.completed_actions
],
"pending_actions": [
mask_text(item)
for item in signal.pending_actions
],
"auth_context": signal.auth_context.model_dump(),
}
def audit(
conn: sqlite3.Connection,
*,
session_id: str,
handoff_id: str | None,
action: str,
details: dict,
) -> None:
conn.execute(
"""
INSERT INTO audit_logs(
session_id,
handoff_id,
action,
details_json,
created_at
)
VALUES (?, ?, ?, ?, ?)
""",
(
session_id,
handoff_id,
action,
json.dumps(details, ensure_ascii=False),
now_iso(),
),
)
def ensure_session(
conn: sqlite3.Connection,
session_id: str,
) -> None:
conn.execute(
"""
INSERT OR IGNORE INTO sessions(
session_id,
state,
generation,
updated_at
)
VALUES (?, ?, 0, ?)
""",
(
session_id,
SessionState.AI_ACTIVE,
now_iso(),
),
)
def serialize_handoff(row: sqlite3.Row) -> dict:
return {
"handoff_id": row["handoff_id"],
"session_id": row["session_id"],
"status": row["status"],
"score": row["score"],
"reason_codes": json.loads(row["reasons_json"]),
"summary": json.loads(row["summary_json"]),
"ticket_id": row["ticket_id"],
"agent_id": row["agent_id"],
"created_at": row["created_at"],
"updated_at": row["updated_at"],
}
def trigger_handoff(
*,
session_id: str,
event_key: str,
signal: HandoffSignal,
) -> dict:
decision = decide_handoff(signal)
conn = connect()
try:
conn.execute("BEGIN IMMEDIATE")
ensure_session(conn, session_id)
duplicate = conn.execute(
"""
SELECT *
FROM handoffs
WHERE session_id = ? AND event_key = ?
""",
(session_id, event_key),
).fetchone()
if duplicate:
conn.commit()
return {
"should_handoff": True,
"duplicate": True,
"handoff": serialize_handoff(duplicate),
}
active = conn.execute(
f"""
SELECT *
FROM handoffs
WHERE session_id = ?
AND status IN ({",".join("?" for _ in ACTIVE_HANDOFF_STATES)})
ORDER BY created_at DESC
LIMIT 1
""",
(session_id, *ACTIVE_HANDOFF_STATES),
).fetchone()
if active:
conn.commit()
return {
"should_handoff": True,
"duplicate": True,
"reused_active_handoff": True,
"handoff": serialize_handoff(active),
}
if not decision.should_handoff:
audit(
conn,
session_id=session_id,
handoff_id=None,
action="handoff_not_triggered",
details={
"score": decision.score,
"reasons": list(decision.reasons),
},
)
conn.commit()
return {
"should_handoff": False,
"score": decision.score,
"reason_codes": list(decision.reasons),
}
session = conn.execute(
"""
SELECT state, generation
FROM sessions
WHERE session_id = ?
""",
(session_id,),
).fetchone()
if session["state"] != SessionState.AI_ACTIVE:
raise HTTPException(
status_code=409,
detail=f"当前状态不允许转人工：{session['state']}",
)
handoff_id = f"ho_{uuid.uuid4().hex}"
outbox_id = f"ob_{uuid.uuid4().hex}"
timestamp = now_iso()
summary = build_summary(
session_id,
signal,
decision,
)
conn.execute(
"""
UPDATE sessions
SET state = ?,
generation = generation + 1,
updated_at = ?
WHERE session_id = ?
""",
(
SessionState.HANDOFF_PENDING,
timestamp,
session_id,
),
)
conn.execute(
"""
INSERT INTO handoffs(
handoff_id,
session_id,
event_key,
status,
score,
reasons_json,
summary_json,
created_at,
updated_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
handoff_id,
session_id,
event_key,
SessionState.HANDOFF_PENDING,
decision.score,
json.dumps(
decision.reasons,
ensure_ascii=False,
),
json.dumps(summary, ensure_ascii=False),
timestamp,
timestamp,
),
)
conn.execute(
"""
INSERT INTO outbox(
outbox_id,
event_key,
event_type,
payload_json,
status,
created_at,
updated_at
)
VALUES (?, ?, ?, ?, 'PENDING', ?, ?)
""",
(
outbox_id,
f"create-ticket:{handoff_id}",
"CREATE_CRM_TICKET",
json.dumps(
{
"handoff_id": handoff_id,
"session_id": session_id,
"summary": summary,
},
ensure_ascii=False,
),
timestamp,
timestamp,
),
)
audit(
conn,
session_id=session_id,
handoff_id=handoff_id,
action="handoff_started",
details={
"score": decision.score,
"reasons": list(decision.reasons),
"old_generation": session["generation"],
"new_generation": session["generation"] + 1,
},
)
conn.commit()
row = conn.execute(
"""
SELECT *
FROM handoffs
WHERE handoff_id = ?
""",
(handoff_id,),
).fetchone()
return {
"should_handoff": True,
"duplicate": False,
"ai_output_frozen": True,
"handoff": serialize_handoff(row),
}
except Exception:
conn.rollback()
raise
finally:
conn.close()
def dispatch_outbox_once() -> dict:
conn = connect()
try:
event = conn.execute(
"""
SELECT *
FROM outbox
WHERE status = 'PENDING'
ORDER BY created_at
LIMIT 1
"""
).fetchone()
if not event:
return {"dispatched": False}
payload = json.loads(event["payload_json"])
handoff_id = payload["handoff_id"]
# Demo：用本地表模拟 CRM。
# 真实环境应调用 CRM API，并把 handoff_id
# 作为对方的幂等键。
ticket_id = f"CRM-{handoff_id[3:11].upper()}"
timestamp = now_iso()
conn.execute("BEGIN IMMEDIATE")
conn.execute(
"""
INSERT OR IGNORE INTO crm_tickets(
ticket_id,
handoff_id,
payload_json,
created_at
)
VALUES (?, ?, ?, ?)
""",
(
ticket_id,
handoff_id,
event["payload_json"],
timestamp,
),
)
conn.execute(
"""
UPDATE handoffs
SET status = ?,
ticket_id = ?,
updated_at = ?
WHERE handoff_id = ?
""",
(
SessionState.TICKET_CREATED,
ticket_id,
timestamp,
handoff_id,
),
)
conn.execute(
"""
UPDATE sessions
SET state = ?, updated_at = ?
WHERE session_id = ?
""",
(
SessionState.TICKET_CREATED,
timestamp,
payload["session_id"],
),
)
conn.execute(
"""
UPDATE outbox
SET status = 'SENT',
attempts = attempts + 1,
last_error = NULL,
updated_at = ?
WHERE outbox_id = ?
""",
(timestamp, event["outbox_id"]),
)
audit(
conn,
session_id=payload["session_id"],
handoff_id=handoff_id,
action="crm_ticket_created",
details={"ticket_id": ticket_id},
)
conn.commit()
return {
"dispatched": True,
"handoff_id": handoff_id,
"ticket_id": ticket_id,
"status": SessionState.TICKET_CREATED,
}
except Exception as exc:
conn.rollback()
conn.execute(
"""
UPDATE outbox
SET attempts = attempts + 1,
last_error = ?,
updated_at = ?
WHERE outbox_id = ?
""",
(
str(exc)[:300],
now_iso(),
event["outbox_id"],
),
)
conn.commit()
raise
finally:
conn.close()
@app.post("/sessions/{session_id}/signals")
def receive_signal(
session_id: str,
signal: HandoffSignal,
idempotency_key: Annotated[
str,
Header(
alias="Idempotency-Key",
min_length=8,
max_length=64,
),
],
) -> dict:
return trigger_handoff(
session_id=session_id,
event_key=idempotency_key,
signal=signal,
)
@app.post("/workers/outbox/run-once")
def run_outbox_once() -> dict:
# Demo 为方便测试直接暴露。
# 生产环境必须改成受保护的内部 Worker。
return dispatch_outbox_once()
@app.post("/handoffs/{handoff_id}/accept")
def accept_handoff(
handoff_id: str,
request: HumanAcceptRequest,
) -> dict:
conn = connect()
try:
conn.execute("BEGIN IMMEDIATE")
handoff = conn.execute(
"""
SELECT *
FROM handoffs
WHERE handoff_id = ?
""",
(handoff_id,),
).fetchone()
if not handoff:
raise HTTPException(
status_code=404,
detail="handoff 不存在",
)
if handoff["status"] not in (
SessionState.TICKET_CREATED,
SessionState.HUMAN_ASSIGNED,
):
raise HTTPException(
status_code=409,
detail=f"当前状态不能接管：{handoff['status']}",
)
timestamp = now_iso()
conn.execute(
"""
UPDATE handoffs
SET status = ?,
agent_id = ?,
updated_at = ?
WHERE handoff_id = ?
""",
(
SessionState.HUMAN_ACTIVE,
request.agent_id,
timestamp,
handoff_id,
),
)
conn.execute(
"""
UPDATE sessions
SET state = ?, updated_at = ?
WHERE session_id = ?
""",
(
SessionState.HUMAN_ACTIVE,
timestamp,
handoff["session_id"],
),
)
audit(
conn,
session_id=handoff["session_id"],
handoff_id=handoff_id,
action="human_accepted",
details={"agent_id": request.agent_id},
)
conn.commit()
return {
"handoff_id": handoff_id,
"status": SessionState.HUMAN_ACTIVE,
"agent_id": request.agent_id,
}
except Exception:
conn.rollback()
raise
finally:
conn.close()
@app.get("/sessions/{session_id}/can-emit")
def can_emit(
session_id: str,
task_generation: int,
) -> dict:
with connect() as conn:
session = conn.execute(
"""
SELECT state, generation
FROM sessions
WHERE session_id = ?
""",
(session_id,),
).fetchone()
if not session:
raise HTTPException(
status_code=404,
detail="session 不存在",
)
allowed = (
session["state"] == SessionState.AI_ACTIVE
and task_generation == session["generation"]
)
return {
"allowed": allowed,
"session_state": session["state"],
"current_generation": session["generation"],
"task_generation": task_generation,
}
@app.get("/handoffs/{handoff_id}")
def get_handoff(handoff_id: str) -> dict:
with connect() as conn:
row = conn.execute(
"""
SELECT *
FROM handoffs
WHERE handoff_id = ?
""",
(handoff_id,),
).fetchone()
if not row:
raise HTTPException(
status_code=404,
detail="handoff 不存在",
)
return serialize_handoff(row)
FastAPI 可以使用 Pydantic 嵌套模型对复杂 JSON 请求进行类型和字段验证，适合把身份状态、业务槽位和转人工信号分层建模。FastAPI Nested Models
九、完整测试步骤
9.1 发送一次低置信度转人工信号
curl -X POST \
'http://127.0.0.1:8000/sessions/session-001/signals' \
-H 'Content-Type: application/json' \
-H 'Idempotency-Key: turn-000001' \
-d '{
"intent": "refund_request",
"intent_confidence": 0.42,
"emotion_risk": 0.70,
"repeated_failures": 2,
"explicit_request": false,
"sensitive_intent": false,
"policy_blocked": false,
"tool_failed": true,
"latest_user_text": "一直查不到，那你帮我转人工吧",
"safe_slots": {
"order_id": "ORDER-1008",
"issue_type": "refund",
"bank_card": "6222000000000000"
},
"completed_actions": [
"已完成基础身份校验"
],
"pending_actions": [
"核验退款资格"
],
"auth_context": {
"authenticated": true,
"consent_to_share": true,
"auth_level": "strong"
}
}'
应看到类似状态：
{
"should_handoff": true,
"duplicate": false,
"ai_output_frozen": true,
"handoff": {
"status": "HANDOFF_PENDING",
"score": 0.686,
"reason_codes": [
"risk_score_threshold"
]
}
}
注意请求里的 bank_card 不在 ALLOWED_SLOTS 中，因此不会进入人工交接快照。
9.2 使用同一个幂等键重试
再次执行相同请求：
{
"should_handoff": true,
"duplicate": true,
"handoff": {
"handoff_id": "ho_原来的ID"
}
}
不会生成第二条 handoff。
9.3 运行一次 Outbox Worker
curl -X POST \
'http://127.0.0.1:8000/workers/outbox/run-once'
返回：
{
"dispatched": true,
"handoff_id": "ho_xxx",
"ticket_id": "CRM-XXXXXXXX",
"status": "TICKET_CREATED"
}
此时只表示工单已经创建，不能向用户宣称“人工已经接通”。
9.4 检查旧 AI 任务还能否输出
转人工前的生成任务假设使用：
task_generation = 0
调用：
curl \
'http://127.0.0.1:8000/sessions/session-001/can-emit?task_generation=0'
返回：
{
"allowed": false,
"session_state": "TICKET_CREATED",
"current_generation": 1,
"task_generation": 0
}
旧 LLM 文本和 TTS 音频都应该被丢弃。
9.5 人工接受会话
curl -X POST \
'http://127.0.0.1:8000/handoffs/ho_xxx/accept' \
-H 'Content-Type: application/json' \
-d '{
"agent_id": "agent-007"
}'
返回：
{
"handoff_id": "ho_xxx",
"status": "HUMAN_ACTIVE",
"agent_id": "agent-007"
}
只有进入 HUMAN_ACTIVE 后，系统才能确认人工真正接管。
十、CRM/API 回流不能只考虑“成功或失败”
真实 CRM 调用至少要区分：
| 类型 | 示例 | 处理方式 |
|---|---|---|
| 参数错误 | 400、422 | 不盲目重试，进入人工排查 |
| 未认证 | 401 | 刷新凭证或报警 |
| 无权限 | 403 | 停止重试，检查权限配置 |
| 对象不存在 | 404 | 检查客户或订单映射 |
| 冲突/重复 | 409 | 查询幂等键对应的已有工单 |
| 限流 | 429 | 按 Retry-After 退避 |
| 服务异常 | 500、502、503 | 有上限地指数退避 |
| 网络超时 | timeout | 查询后重试，不能直接创建新工单 |
第三方 API 的响应也不能默认可信。
OWASP API Security Top 10 将“不安全地消费第三方 API”单列为风险，建议对下游返回内容进行验证和清洗，设置超时，并避免盲目跟随重定向。OWASP API10:2023 Unsafe Consumption of APIs
生产 Worker 建议具备：
最大重试次数
指数退避
随机抖动
请求超时
幂等键
死信队列
人工补偿入口
状态查询
审计日志
十一、转人工时怎样对用户说
工单已创建、人工尚未接通
我已经记录了您的问题，并正在为您安排人工处理。
当前还没有人工接入，请稍候。
没有空闲坐席
当前人工坐席较忙。我可以为您保留工单，
由客服在可用时回电，或者您也可以继续等待。
已经分配人工
已经为您匹配到客服，正在同步本次沟通记录。
人工真正接管
人工客服已接入，接下来由客服继续为您处理。
不要在 TICKET_CREATED 状态就说“人工已接入”。
十二、权限和隐私边界
人工接管后并不代表坐席可以看到所有数据。
至少需要检查：
这个坐席是否属于正确业务组
是否有权查看该客户
是否有权查看订单、退款或投诉信息
是否有权执行当前操作
是否需要二次身份校验
是否需要用户授权共享某些信息
建议按字段做白名单：
ALLOWED_SLOTS = {
"order_id",
"product",
"issue_type",
"preferred_callback_window",
}
并遵循：
最小权限
默认拒绝
按对象授权
按动作授权
关键操作二次校验
全过程留痕
OWASP 的授权建议强调最小权限和默认拒绝：用户已完成身份认证，不等于有权访问所有对象或执行所有动作。OWASP Authorization Cheat Sheet
十三、常见错误与排查
13.1 用户说“转人工”，系统还在继续劝说
错误逻辑：
if explicit_request and score >= threshold:
handoff()
显式请求不应该再依赖软评分：
if explicit_request:
handoff()
13.2 工单创建了两次
检查：
- 是否给每次业务事件生成稳定的幂等键；
- CRM 是否把
handoff_id设置为唯一键； - 超时重试前是否查询已有结果；
- Outbox Worker 是否可能重复消费；
- 消费端是否支持幂等写入。
13.3 人工看到的摘要与事实不一致
检查摘要中的每个字段来自哪里：
LLM 推断
ASR 原文
数据库
工具调用
用户确认
人工修改
身份状态、订单状态和已执行动作不能只靠 LLM 推断。
13.4 人工已经接管，AI 又说了一句话
检查：
LLM task 是否取消
TTS task 是否取消
客户端音频队列是否清空
generation 是否递增
迟到音频是否按 turn_id 丢弃
13.5 CRM 超时后流程永久卡在 HANDOFF_PENDING
需要：
- 独立 Worker；
- 超时重试；
- 最大尝试次数；
next_retry_at；- 死信状态；
- 运维告警；
- 人工补偿按钮。
不要把关键工单投递只放在 Web 进程的临时后台任务里。
13.6 人工重复询问用户已经说过的问题
这通常不是摘要不够长，而是缺少结构化事实：
用户诉求
业务对象
已验证字段
已执行动作
失败原因
下一步待办
十四、生产环境建议监控的指标
转人工决策
handoff_trigger_total
handoff_trigger_by_reason
missed_handoff_rate
unnecessary_handoff_rate
explicit_request_blocked_total
状态流转
handoff_pending_duration
ticket_create_success_rate
ticket_create_retry_count
human_assign_latency
human_accept_latency
handoff_failed_total
上下文质量
summary_validation_failed_total
missing_required_field_rate
human_reask_rate
summary_manual_correction_rate
状态一致性
duplicate_ticket_total
outbox_pending_age
ai_output_after_handoff_total
handoff_state_conflict_total
其中我最关注：
ai_output_after_handoff_total
这个指标应该尽量保持为 0。一旦非零，说明会话冻结、任务取消或客户端队列至少有一层没有生效。
十五、阈值应该怎么校准
不要凭感觉决定 0.65。
先准备真实标注集：
应该转人工
可以由 AI 继续处理
必须立即转人工
可以澄清一次再决定
然后同时看：
漏转率：本该转人工但没有转
误转率：AI 能处理却转给人工
人工接管后的解决率
用户重复表达率
投诉升级率
每千次会话人工成本
不同场景要使用不同成本：
| 场景 | 漏转成本 | 误转成本 |
|---|---|---|
| 普通知识咨询 | 低 | 中 |
| 退款争议 | 高 | 中 |
| 账户安全 | 极高 | 低 |
| 法律投诉 | 极高 | 低 |
| 商品推荐 | 低 | 高 |
因此更合理的方式是：
全局硬规则
+ 场景阈值
+ 用户显式请求
+ 连续信号
+ 人工反馈校准
而不是全业务共用一个分数。
十六、FAQ
Q1：创建工单后就能把会话状态设为 HUMAN_ACTIVE 吗？
不能。工单创建、人工分配和人工接管是三个不同事件。
Q2：用户主动要求人工，还需要计算意图置信度吗？
不需要用置信度阻止转人工。可以继续记录置信度用于分析，但显式请求应走硬规则。
Q3：应该把多少轮对话交给人工？
优先传结构化摘要，再附最近少量必要原文。不要默认发送无限历史。
Q4：摘要能不能完全由大模型生成？
不建议。大模型可以负责可读描述，但身份、订单、动作结果和权限状态必须来自确定性系统。
Q5：为什么已有 Outbox 还需要幂等？
因为 Outbox 或消息系统可能至少投递一次。Worker 在成功后崩溃，也可能导致同一个事件再次发送。
Q6：没有人工在线怎么办？
不要假装正在实时转接。进入排队、预约回电或离线工单流程，并向用户说明当前状态。
Q7：转人工后还能让 AI 辅助坐席吗？
可以，但角色要改变。AI 可以给坐席生成建议或检索资料，不应绕过人工直接对用户执行高风险动作。
Q8：CRM 返回 200 就一定成功吗？
不一定。还要校验响应结构、业务状态、工单 ID 和幂等键是否一致。
Q9：转人工接口需要哪些权限保护？
至少需要坐席身份、对象级授权、动作级授权、最小权限和审计。内部接口也不能默认可信。
十七、这条链路是否合格，我会检查八件事
- 用户明确要求人工时是否立即进入硬规则；
- 转人工后 AI 文本和音频是否真正停止；
- 工单、Outbox 和会话状态是否同事务提交；
- 相同幂等键是否只产生一个 handoff；
- CRM 重试是否只产生一张工单；
- 摘要是否区分事实、推断、已执行和待执行；
- 工单创建与人工接管是否使用不同状态；
- 整个过程是否有权限校验、脱敏和审计记录。
安全转人工的核心不是“把电话交给另一个人”，而是：
在正确的时机停止 AI，
把可信的业务状态交给正确的人，
并确保失败以后仍然能够恢复。
系列阅读
上一篇：
Voice Agent 到底慢在哪？用 t0-t8 埋点拆解 ASR、LLM、TTS 全链路延迟
下一篇计划：
真实电话环境下测 ASR：数字、地址、噪声与业务热词
将继续拆测试集设计、数字与地址错误分类、WER/CER，以及怎样避免用“安静房间里的几句录音”冒充真实业务效果。
官方资料
火山引擎视频云技术社区，是面向 AI 音视频开发者的技术交流平台。这里汇聚源自抖音、豆包等亿级 DAU 产品的 RTC、直播、点播、AI 媒体处理、音视频互动技术，提供接入指南、最佳实践、性能调优、场景案例、Demo 代码、开源项目、白皮书和 API 文档。社区汇聚官方工程师与一线开发者，为 AI 视频通话、数字人、AI 视频处理等应用的开发与落地提供技术支持。
更多推荐
9
0- 0
已为社区贡献27条内容
所有评论(0)