首页把几十个业务函数一股脑塞给模型，出问题几乎是必然的：选错工具、参数张冠李戴、执行了本不该执行的操作。工具列表越长，模型判断越容易飘。
更稳的做法是按场景只暴露当下该用的那几个。用智能家居演示最直观：人在客厅时，模型只看得到电视、灯、空调；切到厨房，工具集就换成灯、热水壶、油烟机。
于是「打开电视」在厨房会被直接拒绝——不是靠提示词劝住它，而是那个函数根本不在它的可选范围里。两个会话同时进行时，各自的房间和设备状态也互不干扰。
> 开始之前 先完成《给语音智能体接入自定义 LLM：OpenAI 兼容网关》，工具集的切换逻辑跑在那里面。
一. 架构与准备工作
用户语音 → 凤鸣 → Custom LLM 网关
│ X-Session-ID
├── SQLite 读取当前房间
├── 动态生成本房间工具 schema
├── DeepSeek 选择工具
└── 执行业务函数并流式返回
↓
MiniMax 中文播报需要 App ID、App Certificate、DeepSeek、MiniMax，以及一个云端能访问的 HTTPS Custom LLM 地址。示例用 agora-agents==2.4.1，靠每个会话独立的 X-Session-ID 隔离状态。
二. 环境变量
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
DYNAMIC_TOOLS_DB_PATH=dynamic-tools.db三. 定义动态工具集合
这里是全篇最重要的一条设计原则：业务状态才是工具权限的事实来源，不能靠提示词跟模型说「别调用」。
提示词是建议，模型可能听也可能不听；而不在 tools 列表里的函数，模型压根没法调。下面的 active_tools() 每轮从数据库读当前房间，只生成这个房间设备的 schema。
import sqlite3
ROOMS = {"客厅": "living_room", "卧室": "bedroom", "厨房": "kitchen"}
DEVICES = {
"living_room": {"lamp": "灯", "tv": "电视", "air_conditioner": "空调"},
"bedroom": {"lamp": "灯", "curtain": "窗帘", "air_conditioner": "空调"},
"kitchen": {"light": "灯", "kettle": "热水壶", "hood": "油烟机"},
}
def function_schema(name: str, description: str, properties: dict) -> dict:
return {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": {
"type": "object",
"properties": properties,
"required": list(properties),
"additionalProperties": False,
},
},
}
def active_tools(db: sqlite3.Connection, session_id: str) -> list[dict]:
row = db.execute(
"SELECT room FROM sessions WHERE session_id=?", (session_id,)
).fetchone()
room = row[0] if row else "living_room"
tools = [
function_schema(
"switch_room",
"切换房间；下一轮加载新房间工具",
{"room": {"type": "string", "enum": list(ROOMS)}},
),
function_schema("get_status", "查询设备状态", {}),
]
for device, name in DEVICES[room].items():
tools.append(function_schema(
f"set_{device}",
f"控制当前房间的{name}",
{"state": {
"type": "string",
"enum": ["on", "off", "dim", "open", "closed"],
}},
))
return tools四. 二次校验再执行
模型选了某个函数，不代表它有权执行。服务端拿到 tool_calls 之后，还要根据数据库里的当前房间再校验一次：
def set_device(
db: sqlite3.Connection,
session_id: str,
device: str,
state: str,
) -> str:
row = db.execute(
"SELECT room FROM sessions WHERE session_id=?", (session_id,)
).fetchone()
room = row[0] if row else "living_room"
if device not in DEVICES[room]:
available = "、".join(DEVICES[room].values())
return f"当前房间没有这个设备，可控制{available}。"
if state not in {"on", "off", "dim", "open", "closed"}:
return "设备状态无效。"
db.execute(
"""INSERT INTO devices(session_id, room, device, state)
VALUES (?, ?, ?, ?)
ON CONFLICT(session_id, room, device)
DO UPDATE SET state=excluded.state""",
(session_id, room, device, state),
)
db.commit()
return f"设备 {device} 已设为 {state}。"这是纵深防御：第一层限制模型能看到什么，第二层限制服务端愿意执行什么。少了第二层，一旦工具列表因为并发或缓存问题算错，就直接变成越权执行。
完整网关把 active_tools() 的结果传给 DeepSeek 的 tools，执行返回的 tool_calls，把工具结果作为 role=tool 追加进去，再流式请求最终回答。接口要兼容 OpenAI Chat Completions SSE，并强制校验 Bearer 和会话头：
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-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 X-Session-ID")
return StreamingResponse(
safe_orchestrate(payload, session_id),
media_type="text/event-stream",
)safe_orchestrate()、DeepSeek 的工具调用循环、中文场景和 SQLite 建表代码都在交付目录 code/dynamic-tool-sets/tool_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-dynamic-tools",
headers={"X-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, "enable_tools": True},
parameters={"audio_scenario": "chorus", "data_channel": "rtm"},
)
.with_stt(FengmingSTT())
.with_llm(llm)
.with_tts(tts)
)六. 运行与验证
cd code/dynamic-tool-sets
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env.local
uvicorn tool_service:app --host 0.0.0.0 --port 9200把 9200 端口部署成 CUSTOM_LLM_URL 指向的公网 HTTPS，再启动业务后端：
uvicorn app:app --host 0.0.0.0 --port 8000依次说这四句：「打开电视」「切换到厨房」「打开热水壶」「再打开电视」。
前三句都应该成功，最后一句必须被拒绝，厨房的工具集里没有电视。这一句才是整篇教程要验证的东西。
另外用两个不同的 businessSessionId 各跑一遍，确认房间和设备状态互不干扰。
七. 故障排查
- 工具没切换：
switch_room只更新业务状态，新的工具列表要从**下一轮** LLM 请求才开始生效。这是设计如此，不是 bug。 - 云端收到 401：
CustomLLM.api_key和网关共享密钥必须一致。 - 状态串到别的用户：别用固定的
X-Session-ID，要由服务端为每个授权会话生成随机 ID。 - 模型声称执行成功但数据库没变：播报必须基于工具的真实返回值生成。同时检查业务层有没有二次校验工具名、房间和参数——模型说做了不等于做了。
八. 下一步
工具按场景切换之后，再往上一层就是角色本身按场景切换。
《语音智能体多角色转接：分诊、售前与售后》
