首页用一个通用提示词同时干分诊、报价、下单、售后，结果通常是四件事都干得含糊——售前的话术混进售后，退款问题被当成销售机会。
更清楚的做法是拆角色：分诊客服先识别需求，售前顾问回答套餐和购买问题，创建订单后转给售后专员。每个角色规则独立，但用户这边 RTC 音频和字幕全程不断，听不出中间换过人。
这里的 Handoff 是同一个语音智能体内部的业务角色路由，不是转人工坐席，也不是在频道里同时开三个智能体。真要转人工，还得接呼叫中心队列、坐席状态和 RTC 用户管理，那是另一个话题。
用户那边的体感是：三个角色随对话意图自动切换，用户全程听不出中间换过人；而不同用户的订单状态互相隔离，谁也查不到谁的。
> 开始之前 先完成《给语音智能体接入自定义 LLM：OpenAI 兼容网关》，角色路由就写在那个网关里。
一. 架构与准备工作
用户语音 → 凤鸣 → Custom LLM 网关 ── X-Business-Session-ID
│
├── 意图路由：分诊 / 售前 / 售后
├── SQLite：角色 + 演示订单
└── 角色提示词 → DeepSeek 流式回答
↓
MiniMax 中文播报需要 App ID、App Certificate、DeepSeek、MiniMax，以及云端能访问的 HTTPS Custom LLM。本文使用 agora-agents==2.4.1。
二. 环境变量
AGORA_APP_ID=REPLACE_WITH_SHENGWANG_APP_ID
AGORA_APP_CERTIFICATE=REPLACE_WITH_PRIMARY_CERTIFICATE
DEEPSEEK_API_KEY=REPLACE_WITH_DEEPSEEK_KEY
MINIMAX_API_KEY=REPLACE_WITH_MINIMAX_KEY
CUSTOM_LLM_URL=https://REPLACE_WITH_PUBLIC_HOST/chat/completions
CUSTOM_LLM_SHARED_SECRET=REPLACE_WITH_RANDOM_SHARED_SECRET
HANDOFF_DB_PATH=handoff.db三. 持久化角色并路由
路由有两条规则。明确的售后意图优先于售前；没有新意图时，保持当前角色。
第二条尤其重要。如果每轮都重新判断一次，用户说一句「嗯」或者「那个」，角色就会跳回分诊，对话直接错乱。
import sqlite3
import time
ROLE_NAMES = {
"triage": "分诊客服",
"sales": "售前顾问",
"after_sales": "售后专员",
}
SALES = ("价格", "套餐", "购买", "试用", "下单", "报价", "续费")
AFTER_SALES = ("售后", "退款", "发票", "订单", "故障", "无法使用", "取消")
def connect(path: str = "handoff.db") -> sqlite3.Connection:
db = sqlite3.connect(path)
db.execute("""
CREATE TABLE IF NOT EXISTS sessions (
session_id TEXT PRIMARY KEY,
role TEXT NOT NULL,
updated_at REAL NOT NULL
)
""")
db.commit()
return db
def current_role(db: sqlite3.Connection, session_id: str) -> str:
db.execute(
"""INSERT OR IGNORE INTO sessions(session_id, role, updated_at)
VALUES (?, 'triage', ?)""",
(session_id, time.time()),
)
db.commit()
return db.execute(
"SELECT role FROM sessions WHERE session_id=?", (session_id,)
).fetchone()[0]
def set_role(db: sqlite3.Connection, session_id: str, role: str) -> str:
if role not in ROLE_NAMES:
raise ValueError("invalid role")
db.execute(
"""INSERT INTO sessions(session_id, role, updated_at)
VALUES (?, ?, ?)
ON CONFLICT(session_id)
DO UPDATE SET role=excluded.role, updated_at=excluded.updated_at""",
(session_id, role, time.time()),
)
db.commit()
return role
def route_role(
db: sqlite3.Connection,
session_id: str,
text: str,
) -> str:
if any(word in text for word in AFTER_SALES):
return set_role(db, session_id, "after_sales")
if any(word in text for word in SALES):
return set_role(db, session_id, "sales")
return current_role(db, session_id)四. 给每个角色单独的提示词
def role_prompt(role: str, context: str) -> str:
prompts = {
"triage": (
"你是分诊客服。先用一个简短问题判断用户需要售前还是售后；"
"不能假装已经完成购买、退款或人工转接。"
),
"sales": (
"你是售前顾问。介绍专业版和企业版，回答价格、试用和购买问题；"
"用户明确说确认下单时，业务网关会创建演示订单。"
),
"after_sales": (
"你是售后专员。根据订单上下文处理状态、发票、故障和退款咨询；"
"不能编造订单，也不能承诺真实退款已经到账。"
),
}
return (
f"{prompts[role]}回答自然、简短，适合语音播报。"
f"当前业务上下文：{context}"
)网关每收到一轮 /chat/completions，先读最后一条用户消息、更新角色，再把对应的提示词和业务上下文发给 DeepSeek。
用户明确说「确认下单」时，教程代码会建一个 DEMO- 开头的 SQLite 演示订单并转到售后；说「退款」只会把演示状态改成 refund_pending。这两个操作都不碰真实资金，教程代码不应该有真的支付能力。
五. 保护 Custom LLM 接口
import os
import re
from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import StreamingResponse
app = FastAPI()
SESSION_ID = re.compile(r"^[A-Za-z0-9_-]{1,80}$")
@app.post("/chat/completions")
async def chat(
payload: dict,
authorization: str | None = Header(None),
session_id: str | None = Header(
None, alias="X-Business-Session-ID"
),
):
secret = os.getenv("CUSTOM_LLM_SHARED_SECRET")
if not secret or authorization != f"Bearer {secret}":
raise HTTPException(401, "invalid credential")
if not session_id or not SESSION_ID.fullmatch(session_id):
raise HTTPException(400, "invalid business session")
return StreamingResponse(
safe_orchestrate(payload, session_id),
media_type="text/event-stream",
)完整的 DeepSeek 流式代理、SSE 降级、演示订单和状态隔离实现，都在交付目录 code/agent-handoff/handoff_service.py。
六. 配置智能体
import os
from agora_agent import Area, AsyncAgora
from agora_agent.agentkit import Agent, CustomLLM
from agora_agent.cn import FengmingSTT, MiniMaxTTS
client = AsyncAgora(
area=Area.CN,
app_id=os.environ["AGORA_APP_ID"],
app_certificate=os.environ["AGORA_APP_CERTIFICATE"],
)
llm = CustomLLM(
api_key=os.environ["CUSTOM_LLM_SHARED_SECRET"],
base_url=os.environ["CUSTOM_LLM_URL"],
model="shengwang-agent-handoff",
headers={
"X-Business-Session-ID": "REPLACE_WITH_BUSINESS_SESSION_ID"
},
system_messages=[{"role": "system", "content":
"你是中文客服入口，业务网关会为每轮选择正确角色。"}],
greeting_message="你好，请告诉我你需要售前咨询、购买还是售后服务。",
failure_message="客服服务暂时不可用，请稍后再试。",
max_history=20,
)
tts = MiniMaxTTS(
key=os.environ["MINIMAX_API_KEY"],
model="speech-01-turbo",
voice_id="female-shaonv",
sample_rate=16000,
language_boost="Chinese",
)
agent = (
Agent(
client=client,
turn_detection={"language": "zh-CN"},
advanced_features={"enable_rtm": True},
parameters={"audio_scenario": "chorus", "data_channel": "rtm"},
)
.with_stt(FengmingSTT())
.with_llm(llm)
.with_tts(tts)
.with_labels({
"recipe": "agent-handoff",
"business_session_id": "REPLACE_WITH_BUSINESS_SESSION_ID",
})
)business_session_id 必须由服务端生成，或者从已认证的业务会话里取。绝对不能接受用户任意传入——否则用户改个 ID 就能读到别人的订单。
七. 运行与验证
cd code/agent-handoff
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env.local
uvicorn handoff_service:app --host 0.0.0.0 --port 9300把 9300 端口部署成公网 HTTPS，再启动智能体业务接口：
uvicorn app:app --host 0.0.0.0 --port 8000按这个脚本对一遍：
用户：你们有什么套餐？ → 售前顾问
用户：我确认下单专业版。 → 创建演示订单并转售后
用户：查询我的订单。 → 售后专员读取同一会话订单
用户：我要退款。 → 标记 refund_pending然后换一个 businessSessionId 再问订单，预期回答「没有演示订单」。这一步验证的是状态隔离，别跳过。
验收还要覆盖：模糊意图、来回反复切换、DeepSeek 超时、数据库故障、越权的会话 ID，以及切换角色时字幕是否连续。
八. 故障排查
- 每轮都回到分诊：确认角色按
business_session_id持久化了，并且 Custom LLM 的请求头没变。 - 用户状态串线：业务会话 ID 必须随机、不可猜、和已认证用户绑定，查询时还要做租户隔离。
- 退款被误执行：高风险操作绝不能交给模型判断。放到确定性的业务层，加上权限校验、幂等、二次确认和审计。模型可以决定「用户想退款」，但不能决定「执行退款」。
- 用户要转人工：返回一个明确的状态给业务后端去调坐席系统。不能让模型说一句「已为你转接」就完事——那边根本没人接。
九. 下一步
角色能切了，每个角色的提示词也可以不停机更新。
《不停机更新语音智能体的角色和提示词》
