在线咨询

专属客服在线解答，提供专业解决方案

工单支持

专业技术支持团队，随时响应服务需求

首页博客正文

在不停机的情况下更新语音智能体的角色和提示词

2026-07-27

开发者教程

客服场景经常要中途换角色：用户升级成 VIP、流程走到售后阶段、过了工作时间要切成留言模式。如果每次改提示词都得停掉智能体重建一个，用户会听到一段明显的断线和重连——这个体验损失，比换角色带来的收益大得多。

对话式 AI 引擎支持两件事：创建时注入模板变量，运行中更新 system_messages。两条路下面都走一遍。

效果是：调用更新接口的那一刻通话毫无察觉，用户还在说话，下一轮回复就已经换了个角色。

> 开始之前　这篇在《Python + FastAPI 搭建中文语音智能体》基础上加一个更新接口，先把它跑起来。

一. 架构与准备工作

业务系统 ── 用户/场景变量 ──→ 创建智能体
   │                           │
   └── 新提示词 ──→ update ───┤（RTC 会话保持连接）
                               ↓
用户语音 → 凤鸣 → DeepSeek V4 Flash → MiniMax → 用户

本文在《Python + FastAPI 搭建中文语音智能体》的基础上增加一个 /updateInstructions 接口，需要 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
WEB_ORIGINS=http://localhost:3000,http://127.0.0.1:3000

所有凭证只留在服务端。还有一条更要紧的规矩。前端只能提交业务预先授权过的角色选项，绝不能让用户直接覆盖系统提示词。否则就是把提示词注入的大门敞开了。

三. 构造模板变量和更新请求

from datetime import date

from agora_agent.agents.types.update_agents_request_properties import (
    UpdateAgentsRequestProperties,
)
from agora_agent.agents.types.update_agents_request_properties_llm import (
    UpdateAgentsRequestPropertiesLlm,
)

SYSTEM_PROMPT = (
    "你是 {{assistant_name}}，一名中文语音助手。今天是 {{today}}。"
    "当前服务场景是 {{scene}}。回答自然、简短，不确定时明确说明。"
)

def template_variables(assistant_name: str, scene: str) -> dict[str, str]:
    return {
        "assistant_name": assistant_name.strip() or "小声",
        "scene": scene.strip() or "通用咨询",
        "today": date.today().isoformat(),
    }

def update_properties(instructions: str) -> UpdateAgentsRequestProperties:
    cleaned = instructions.strip()
    if not cleaned:
        raise ValueError("instructions 不能为空")
    return UpdateAgentsRequestProperties(
        llm=UpdateAgentsRequestPropertiesLlm(
            system_messages=[{"role": "system", "content": cleaned}]
        )
    )

模板变量可以出现在 system_messages、问候语、失败提示和静默提示里。但变量值不会被递归解析，所以别在变量值里再塞一个 {{variable}}，它不会展开。

四. 创建并更新智能体

import os
from typing import Any

from agora_agent import Area, AsyncAgora
from agora_agent.agentkit import Agent
from agora_agent.cn import DeepSeekLLM, FengmingSTT, MiniMaxTTS

class DynamicAgentService:
    def __init__(self):
        self.client = AsyncAgora(
            area=Area.CN,
            app_id=os.environ["AGORA_APP_ID"],
            app_certificate=os.environ["AGORA_APP_CERTIFICATE"],
        )
        self.sessions: dict[str, Any] = {}

    def build_agent(self, assistant_name: str, scene: str) -> Agent:
        llm = DeepSeekLLM(
            api_key=os.environ["DEEPSEEK_API_KEY"],
            base_url="https://api.deepseek.com/chat/completions",
            model="deepseek-v4-flash",
            system_messages=[{"role": "system", "content": SYSTEM_PROMPT}],
            template_variables=template_variables(assistant_name, scene),
            greeting_message=f"你好，我是{assistant_name}，很高兴为你服务。",
            failure_message="抱歉，服务暂时繁忙，请稍后再试。",
            max_history=20,
            max_tokens=512,
            temperature=0.5,
            params={"thinking": {"type": "disabled"}},
        )
        tts = MiniMaxTTS(
            key=os.environ["MINIMAX_API_KEY"],
            model="speech-01-turbo", voice_id="female-shaonv",
            sample_rate=16000, language_boost="Chinese",
        )
        return (Agent(
            client=self.client,
            turn_detection={"language": "zh-CN"},
            advanced_features={"enable_rtm": True},
            parameters={"audio_scenario": "chorus", "data_channel": "rtm",
                        "enable_metrics": True, "enable_error_message": True},
        ).with_stt(FengmingSTT()).with_llm(llm).with_tts(tts))

    async def update(self, agent_id: str, instructions: str) -> None:
        session = self.sessions.get(agent_id)
        if session is None:
            raise ValueError("找不到运行中的智能体会话")
        await session.update(update_properties(instructions))

创建会话后一定要按 agent_id 把 session 存下来。session.update() 只能更新还在运行的会话，找不到 session 就更新不了。

五. 暴露更新接口

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI()
service = DynamicAgentService()

class UpdateRequest(BaseModel):
    agentId: str = Field(min_length=1)
    instructions: str = Field(min_length=1, max_length=4000)

@app.post("/updateInstructions")
async def update_instructions(body: UpdateRequest):
    try:
        await service.update(body.agentId, body.instructions)
        return {"code": 0, "message": "updated"}
    except ValueError as exc:
        raise HTTPException(404, str(exc)) from exc

运行完整交付代码：

cd code/dynamic-instructions
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env.local
uvicorn app:app --host 0.0.0.0 --port 8000

启动智能体之后，换个角色试试：

curl -X POST http://localhost:8000/updateInstructions \
  -H 'Content-Type: application/json' \
  -d '{"agentId":"REPLACE_WITH_RUNNING_AGENT_ID","instructions":"你是售后顾问，只回答退款和物流问题，每次不超过两句话。"}'

六. 应该看到什么

接口返回 {"code":0,"message":"updated"}，RTC 会话不中断，下一轮回复就开始遵守新角色。

验收时一起确认这几样：旧提示词确实不再生效、字幕连续没断、agentId 保持不变（说明是更新不是重建）、越权的角色请求被业务层挡住了。

七. 故障排查

更新成功但回答没变：确认更新的是 system_messages，而且要等下一轮用户输入之后再看，已经在生成的那条回复不会被中途改写。
返回找不到会话：服务重启后内存里的 session 映射就没了。生产环境要么做实例粘滞，要么直接调 REST 的 update 接口。
防提示词注入：不要把用户原话直接当 system prompt 用。正确做法是服务端维护一份白名单模板，加上业务权限校验，用户只能在允许的选项里选。

八. 下一步

能换角色了，下一步是让它自己判断该换成哪个角色。

《语音智能体多角色转接：分诊、售前与售后》

在声网，连接无限可能

想进一步了解「对话式 AI 与实时互动」？欢迎注册，开启探索之旅。

注册体验

本博客为技术交流与平台行业信息分享平台，内容仅供交流参考，文章内容不代表本公司立场和观点，亦不构成任何出版或销售行为。