首页这篇教程带你用 Python 搭一个真正能对话的语音智能体。浏览器负责采集麦克风、加入 RTC/RTM 频道；FastAPI 负责发临时 Token、创建和停止智能体；剩下的交给声网云端——凤鸣做语音识别，DeepSeek 负责对话，MiniMax 把回复念出来。
跑通之后你得到的是一套可以直接改成智能客服、口语陪练、语音助手的后端骨架——中文开场白、多轮对话、双方实时字幕、会话正常释放，整条链路都能走通。
一. 整体架构
浏览器麦克风 ──RTC──→ 声网云端 ─→ 凤鸣 ASR ─→ DeepSeek V4 Flash ─→ MiniMax TTS
↑ │ │
└────智能体音频─────┘ │
└────RTM 字幕与状态────────────────────────────────────────────────┘
FastAPI
├─ GET /get_config：生成频道、UID 和一小时临时 Token
├─ POST /startAgent：创建并启动智能体
└─ POST /stopAgent：停止智能体并释放服务端会话浏览器只能拿到 App ID、频道名、UID 和临时 Token。App Certificate、DeepSeek Key 和 MiniMax Key 永远留在服务端，任何接口都不下发。这条原则贯穿全文。
二. 准备工作
- Python 3.10 或更高版本，推荐 3.11。
- 在声网控制台创建项目，并开通对话式 AI 引擎。
- 备好四个凭证：App ID、App Certificate、DeepSeek API Key、MiniMax API Key。
- 一台带麦克风的电脑。用 Mac mini 的注意：它没有内置麦克风，得插耳机或 USB 麦。
- 前端不用从零写，直接用《Next.js 搭建浏览器端语音智能体》里的完整客户端。两篇教程共用同一份
voice-demo.tsx，省得维护两套 RTC/RTM 实现。
三. 项目结构
python-agent/
├─ .env.local
├─ requirements.txt
├─ agent_service.py
└─ app.py四. 安装依赖
依赖版本都已经固定过，照抄即可，不用自己试哪个组合能装上。创建 requirements.txt:
agora-agents==2.4.1
fastapi==0.116.1
httpx==0.28.1
python-dotenv==1.1.1
uvicorn==0.35.0安装：
python3.11 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt顺带说一句：PyPI 包名就叫 agora-agents，import 时用 agora_agent，这是官方包名，不用怀疑自己装错了。
五. 配置服务端环境变量
创建 .env.local:
AGORA_APP_ID=REPLACE_WITH_APP_ID
AGORA_APP_CERTIFICATE=REPLACE_WITH_APP_CERTIFICATE
DEEPSEEK_API_KEY=REPLACE_WITH_DEEPSEEK_API_KEY
DEEPSEEK_URL=https://api.deepseek.com/chat/completions
DEEPSEEK_MODEL=deepseek-v4-flash
MINIMAX_API_KEY=REPLACE_WITH_MINIMAX_API_KEY
MINIMAX_TTS_MODEL=speech-01-turbo
MINIMAX_VOICE_ID=female-shaonv
AGENT_GREETING=你好，我是你的声网 AI 助手。今天想聊点什么？
WEB_ORIGINS=http://localhost:3000,http://127.0.0.1:3000
PORT=8000.env.local 不要提交到 Git。AGORA_APP_CERTIFICATE、DEEPSEEK_API_KEY、MINIMAX_API_KEY 这三个密钥只能待在服务端，不能加 NEXT_PUBLIC_ 前缀，也不能返回给浏览器。
六. 写智能体服务
agent_service.py 是后端的核心，负责组装 STT、LLM、TTS，并管理智能体会话。创建 agent_service.py:
"""声网对话式智能体公共服务。"""
from __future__ import annotations
import os
from typing import Any
from agora_agent import Area, AsyncAgora
from agora_agent.agentkit import Agent as AgoraAgent
from agora_agent.cn import DeepSeekLLM, FengmingSTT, MiniMaxTTS
class AgentService:
def __init__(self) -> None:
required = (
"AGORA_APP_ID",
"AGORA_APP_CERTIFICATE",
"DEEPSEEK_API_KEY",
"MINIMAX_API_KEY",
)
missing = [name for name in required if not os.getenv(name)]
if missing:
raise RuntimeError(f"缺少服务端环境变量：{', '.join(missing)}")
self.greeting = os.getenv(
"AGENT_GREETING", "你好，我是你的声网 AI 助手。今天想聊点什么？"
)
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) -> AgoraAgent:
llm = DeepSeekLLM(
api_key=os.environ["DEEPSEEK_API_KEY"],
base_url=os.getenv(
"DEEPSEEK_URL", "https://api.deepseek.com/chat/completions"
),
model=os.getenv("DEEPSEEK_MODEL", "deepseek-v4-flash"),
system_messages=[
{
"role": "system",
"content": (
"你是声网对话式 AI 引擎中的中文语音助手。"
"回答自然、简洁；不知道时明确说明，不编造产品信息。"
),
}
],
greeting_message=self.greeting,
failure_message="抱歉，我暂时没有处理成功，请稍后再试。",
max_history=20,
temperature=0.6,
max_tokens=512,
params={"thinking": {"type": "disabled"}},
)
tts = MiniMaxTTS(
key=os.environ["MINIMAX_API_KEY"],
model=os.getenv("MINIMAX_TTS_MODEL", "speech-01-turbo"),
voice_id=os.getenv("MINIMAX_VOICE_ID", "female-shaonv"),
sample_rate=16000,
language_boost="Chinese",
)
return (
AgoraAgent(
client=self.client,
turn_detection={
"language": "zh-CN",
"config": {
"start_of_speech": {
"mode": "vad",
"vad_config": {
"interrupt_duration_ms": 160,
"prefix_padding_ms": 300,
},
},
"end_of_speech": {
"mode": "vad",
"vad_config": {"silence_duration_ms": 480},
},
},
},
advanced_features={"enable_rtm": True, "enable_tools": 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 start(self, channel: str, agent_uid: int, user_uid: int) -> str:
session = self.build_agent().create_async_session(
channel=channel,
agent_uid=str(agent_uid),
remote_uids=[str(user_uid)],
enable_string_uid=False,
idle_timeout=120,
expires_in=3600,
)
agent_id = await session.start()
self.sessions[agent_id] = session
return agent_id
async def stop(self, agent_id: str) -> None:
session = self.sessions.pop(agent_id, None)
if session is not None:
await session.stop()
return
await self.client.stop_agent(agent_id)这段代码里有几个点值得多说两句：
- DeepSeek 这里显式关掉了 thinking（
params={"thinking": {"type": "disabled"}}）。不关的话，模型的推理过程会原样进到字幕和 TTS 里，用户会听到智能体把「思考」念出来，体验相当灾难。 - VAD 的两个参数直接决定对话手感：
prefix_padding_ms管说话开头会不会丢字，silence_duration_ms管智能体多快接话。300 ms 和 480 ms 是中文近场耳机比较舒服的起始值，怎么微调见文末的故障排查。 sessions字典按agent_id记录每个会话。停止时优先走会话自己的stop()，查不到再直接调stop_agent()兜底，避免留下没释放的服务端会话。
七. 写 FastAPI 接口
app.py 只做三件事：发配置和 Token、启动智能体、停止智能体。创建 app.py:
"""最小 FastAPI 后端：Token、启动智能体、停止智能体。"""
from __future__ import annotations
import os
import random
import time
from functools import lru_cache
from agora_agent.agentkit.token import generate_convo_ai_token
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from agent_service import AgentService
load_dotenv(".env.local")
load_dotenv(".env")
app = FastAPI(title="声网对话式智能体最小后端", version="1.0.0")
web_origins = [
origin.strip()
for origin in (
os.getenv("WEB_ORIGINS")
or os.getenv("WEB_ORIGIN")
or "http://localhost:3000,http://127.0.0.1:3000"
).split(",")
if origin.strip()
]
app.add_middleware(
CORSMiddleware,
allow_origins=web_origins,
allow_credentials=True,
allow_methods=["GET", "POST"],
allow_headers=["Content-Type", "Authorization"],
)
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok"}
@lru_cache
def service() -> AgentService:
return AgentService()
class StartRequest(BaseModel):
channelName: str = Field(min_length=1)
rtcUid: int = Field(gt=0)
userUid: int = Field(gt=0)
class StopRequest(BaseModel):
agentId: str = Field(min_length=1)
@app.get("/get_config")
async def get_config() -> dict:
app_id = os.getenv("AGORA_APP_ID")
certificate = os.getenv("AGORA_APP_CERTIFICATE")
if not app_id or not certificate:
raise HTTPException(500, "服务端缺少 AGORA_APP_ID 或 AGORA_APP_CERTIFICATE")
user_uid = random.randint(1000, 9_999_999)
agent_uid = random.randint(10_000_000, 99_999_999)
channel = f"shengwang-ai-{int(time.time())}-{random.randint(100, 999)}"
token = generate_convo_ai_token(
app_id=app_id,
app_certificate=certificate,
channel_name=channel,
uid=user_uid,
token_expire=3600,
)
return {
"code": 0,
"data": {
"app_id": app_id,
"token": token,
"uid": str(user_uid),
"agent_uid": str(agent_uid),
"channel_name": channel,
},
}
@app.post("/startAgent")
async def start_agent(body: StartRequest) -> dict:
try:
agent_id = await service().start(body.channelName, body.rtcUid, body.userUid)
return {"code": 0, "data": {"agent_id": agent_id}}
except Exception as exc:
raise HTTPException(502, f"启动智能体失败：{type(exc).__name__}") from exc
@app.post("/stopAgent")
async def stop_agent(body: StopRequest) -> dict:
try:
await service().stop(body.agentId)
return {"code": 0, "message": "stopped"}
except Exception as exc:
raise HTTPException(502, f"停止智能体失败：{type(exc).__name__}") from exc/get_config 每次都会生成新的频道名、用户 UID、智能体 UID 和一小时有效期的临时 Token，前端不用自己拼这些参数。注意这是演示用的最小实现，上生产前记得补上鉴权、限流和 Token 续期。
八. 接上浏览器客户端
前端直接用下一篇 Next.js 教程的完整客户端，在它的 .env.local 里加一行：
NEXT_PUBLIC_AGENT_API_BASE=http://localhost:8000客户端和这三个接口的对应关系：
| 浏览器动作 | Python 接口 | 请求或返回字段 |
|---|---|---|
| 获取配置 | GET /get_config |
app_id、channel_name、uid、agent_uid、token |
| 启动 | POST /startAgent |
channelName、rtcUid、userUid |
| 停止 | POST /stopAgent |
agentId |
NEXT_PUBLIC_AGENT_API_BASE 只是个公开的服务地址，不含任何密钥。还有，改了任何 NEXT_PUBLIC_ 开头的变量，记得重启或重新构建 Next.js，否则不生效。
九. 跑起来
先启动 Python 服务：
source .venv/bin/activate
uvicorn app:app --host 127.0.0.1 --port 8000确认服务活着：
curl http://127.0.0.1:8000/health再启动前端：
npm install
npm run dev打开 http://localhost:3000，允许麦克风权限，点「开始对话」。
十. 应该看到什么
- 页面状态从「正在获取配置」变成「已连接，请开始说话」。
- 智能体主动播放中文开场白。
- 你说话时，页面上同时出现你和智能体双方的实时字幕。
- 能正常进行多轮中文对话。
- 点「结束对话」后，
/stopAgent返回 200，页面显示「已停止」。
建议连着跑两次——停止之后再启动一遍，确认上一次的会话确实释放干净了。
十一. 故障排查
- 返回
401/403：先确认 App ID 和 App Certificate 来自同一个项目，再检查模型 Key 是否有效、有没有余额。 - 浏览器报跨域：
WEB_ORIGINS里要写浏览器的完整来源，协议、域名、端口一个都不能少；多个来源用英文逗号分隔。 DEVICE_NOT_FOUND：电脑没有可用的输入设备。插上耳机或 USB 麦克风，重新授权。- 智能体起来了但没字幕：检查
enable_rtm=True和data_channel="rtm"，再确认客户端已登录并订阅了 RTM 频道。 - 有字幕但听不到声音：客户端要订阅智能体的远端音频，并调用
audioTrack.play()。 - 回复里混进了思考过程：检查 DeepSeek 参数里的
params={"thinking": {"type": "disabled"}}还在不在。 - 说话开头丢字：把
prefix_padding_ms适当调大。 - 智能体接话太慢：小步调低
silence_duration_ms。一次调太狠，智能体会开始抢话。
十二. 上线前还要做什么
再强调一次，这是最小可运行示例。真正接业务之前，至少把这些补上：接口鉴权、请求限流、Token 续期、日志脱敏、模型调用的超时与重试、进程退出时的会话清理。如果部署多个实例，会话记录也不能留在内存字典里，要放进共享存储。
十三. 下一步
前端别从零写。下一篇给出完整的浏览器客户端，和这套后端接口直接对得上。
《Next.js 搭建浏览器端语音智能体》
