用语音模型加决策模型搭建带状态跟踪的语音检查助手

真实的口语对话是混乱的。对方可能把细节说乱顺序，可能漏掉一项，也可能说完一句之后又回头更正。如果要做的是一个语音助手，应用必须清楚自己到底掌握了什么、还缺什么，并据此决定下一句该问什么。这部分「判断」工作可以交给决策模型：给它上下文和一组问题，它从你预先定义好的选项里挑一个。它专门为这类任务训练，因此响应快、成本低，应用拿到返回值就能直接执行。

本文把工作拆成两半：语音模型负责听和说，决策模型放在它后面，随着对话推进读取内容，告诉应用哪些已经完成、哪些还缺证据。示例场景是一个保洁员在客人入住前检查出租公寓，助手通过追问补齐证据，最终对四项要求给出评估。整套流程用 Python 串起来，涉及的两个模型分别是 OpenAI 的实时语音模型 GPT-Live 和微软的决策模型 Decision-1。文中出现的模型名称、接口路径、参数与可用性均可能变化，实际接入前请以各厂商官网当前信息为准。

为什么需要决策模型

先看一个具体情形。用户说「我修好了」，能不能直接把任务标记为完成？通常不行。如果完成的标准里包含「测试通过」，那么正确的下一步不是标记完成，而是追问对方有没有测试、结果如何。

假设用户确认测试通过，任务标记完成。紧接着用户补了一句：「不过那是在我最后一次改动之前。」这时先前的测试结果已经不能证明当前状态，那项检查需要重新打开；同时，其他仍然成立的信息必须保留下来。

这类判断正是决策模型的用武之地。做法是：对每一项要求，先定义什么样的证据才算数；然后让模型读对话，返回当前状态——要求是否已满足、用户是否确认了问题、还是仍然需要证据。无论模型选了什么，应用都知道下一步该做什么：更新状态、请用户验证某一项，或者直接进入下一项要求。

把判断逻辑从语音模型里剥离出来还有两个实际好处。第一，判断标准写在配置里，改标准不用改对话逻辑；第二，一次请求可以同时回答多个问题，减少往返次数。

准备工作

动手之前需要确认以下几项。

• 两个模型的访问权限：一个实时语音模型（示例使用 GPT-Live，通过 Azure 端点访问），一个决策模型（示例使用 Decision-1，同样通过 Azure 端点访问）。两者都需要对应的 API 密钥与端点地址。
• Python 运行环境：需要能发起 HTTP 请求，并能建立 WebSocket 长连接。示例使用 websockets 库的异步客户端，以及一个支持 await http.post(...) 的 HTTP 客户端。
• 端点与密钥的存放方式：示例中以 AZURE_ENDPOINT、LIVE_URL、API_KEY 三个变量表示，实际使用时从环境变量或密钥管理服务读取，不要写死在代码里。
• 业务侧的要求清单：也就是助手要逐项确认的内容。示例是四项检查，每项都要能用一两句话说清「什么算通过」。

决策模型并非只有一种选择。Jev 或 OpenAI 的 Decisions API 也可以承担同样的角色，整体模式不变，只是决策函数内部的请求写法不同。选型时请以各家官网当前公布的能力、计费与区域可用性为准。

操作步骤

第一步：定义四项检查与证据标准

示例场景是保洁员在客人入住前检查一套出租公寓。助手需要判断四项要求：

• 入户门：当前有效的客人门禁码能否打开门。
• 卧室：是否换上了干净的床单和枕套，床是否铺好。
• 卫生间洗手池：水能否正常排走，开着水龙头时池子下方是否保持干燥。
• 厨房：台面是否干净，冰箱里是否没有上一位客人留下的食物。

一段真实对话可能长这样：

保洁员：「门禁码能开门。床已经铺好，床单枕套都是新的。厨房和洗手池还没看。」

助手：「能不能打开水龙头，看看排水是否正常、池子下面是否保持干燥？」

保洁员：「水要到下午才通，不过厨房现在可以看。」

助手：「好，那我们先看厨房。请检查台面和冰箱。」

注意助手在这里没有机械地按房间顺序追问，而是根据「水没通」这个现实约束，先推进当前可做的检查。这一点需要在决策模型的指令里明确写出来。

第二步：为每项要求定义三个状态

每项要求都从三个选项里选一个：

• ready：根据现有证据，该项要求已满足。
• issue：存在尚未解决的问题。
• needs_evidence：证据不足，既不能判定满足，也不能判定有问题。

这三个状态的定义在四项要求之间共用，写一次即可。以卫生间洗手池为例，问题条目的结构如下（指令文字在示例中做了压缩，实际使用时写完整）：

QUESTIONS["bathroom"] = {
    "type": "choice",
    "instructions": (
        "Assess whether the sink drains normally and its underside "
        "stays dry while water is running. "
        "Use the cleaner's reported observations as evidence. "
        "Interpret replies in context, apply corrections to the facts "
        "they concern, and preserve unrelated valid observations. "
        "Earlier evidence must still apply to the current condition. "
        "Plans and general assurances do not establish a result."
    ),
    "criteria": {
        "needs_evidence": (
            "Current reported evidence is insufficient to establish "
            "every part of the requirement, and does not establish "
            "a current failure."
        ),
        "ready": (
            "Current reported evidence establishes every part "
            "of the requirement."
        ),
        "issue": (
            "Current reported evidence establishes an unresolved "
            "failure of at least one part of the requirement."
        ),
    },
}

这里遵循的是 Decision-1 的 choice 格式：instructions 告诉模型评估什么、如何解读对话；criteria 定义可选结果以及每个结果的含义。其余三项要求结构相同，共用同一套状态定义。

指令里有几句值得留意，它们直接决定了判断质量：

• 「在上下文中解读回复，把更正应用到它所涉及的事实上，并保留无关的有效观察」：这是处理「不过那是在我最后一次改动之前」这类句子的关键。
• 「先前的证据必须仍然适用于当前状况」：防止把已经失效的旧结论继续当作有效证据。
• 「计划和一般性保证不能确立结果」：用户说「我待会儿会去测」不算证据。

第三步：定义下一步动作

除了四项状态，还要再问一个问题：根据目前掌握的信息和保洁员现在能做的事，助手下一句该问什么或说什么？这个问题的选项有六个：

• ask_entrance：追问入户门方面缺失的证据。
• ask_bedroom：追问床品方面缺失的证据。
• ask_bathroom：追问洗手池方面缺失的证据。
• ask_kitchen：追问厨房方面缺失的证据。
• defer：说明剩余检查被什么卡住，然后等待新信息。
• finish：汇总评估结果，包括已确认的问题。

这里有一个容易误解的地方：finish 表示四项要求都已有足够证据可以评估，并不表示全部通过。确认存在的问题同样会出现在汇总里。

该问题的结构如下（指令文字有压缩）：

QUESTIONS["next_action"] = {
    "type": "choice",
    "instructions": (
        "Choose the next useful conversational step. "
        "Use the cleaner's observations, corrections, and the "
        "requirements below to determine what still needs evidence. "
        "Ask for evidence the cleaner can collect now, considering "
        "access, timing, and available resources. "
        "There is no fixed room order. "
        "Choose defer if evidence is missing but no remaining check "
        "can be advanced now. "
        "Choose finish when every requirement has a determined "
        "outcome, including any confirmed problems."
        "\nRequirements: " + json.dumps(CHECKS)
    ),
    "criteria": {
        "ask_entrance": (
            "Collect missing evidence that the current guest code "
            "successfully opens the entrance door."
        ),
        "ask_bedroom": (
            "Collect missing evidence that the bed is made "
            "with clean sheets and clean pillowcases."
        ),
        "ask_bathroom": (
            "Collect missing evidence that the sink drains normally "
            "and stays dry underneath while water runs."
        ),
        "ask_kitchen": (
            "Collect missing evidence that the counters are clean "
            "and the fridge has no previous guests' food."
        ),
        "defer": (
            "Explain what is blocking the remaining checks "
            "and wait for new information."
        ),
        "finish": (
            "Summarize the assessment, including any confirmed problems."
        ),
    },
}

其中 CHECKS 是包含四项要求描述的字典。把它拼进指令，是为了让模型能对照整份检查清单判断还缺什么。指令里「没有固定的房间顺序」和「考虑能否进入、时间安排和可用资源」这两句，正是让助手在「水下午才通」时先转去厨房的依据。

第四步：编写决策函数

把五个问题组装成一个字典，然后写一个函数，输入是到目前为止的对话，输出是各项状态和下一步动作。

QUESTIONS = {
    "entrance": {...},
    "bedroom": {...},
    "bathroom": {...},
    "kitchen": {...},
    "next_action": {...},
}

函数本身很短：

async def assess_inspection(conversation):
    response = await http.post(
        f"{AZURE_ENDPOINT}/providers/microsoft/v1/systemone",
        json={
            "model": "Decision-1",
            "state": {"conversation": conversation},
            "questions": QUESTIONS,
        },
    )
    response.raise_for_status()
    return {
        name: answer["choice"]
        for name, answer in response.json()["answers"].items()
    }

请求里有两个主要成分：state 提供累积的对话内容，questions 指定要做的判断。函数直接把模型选中的标签返回给调用方。

一个关键细节：五个问题放在同一个请求里发送。Decision-1 的接口支持针对同一份上下文同时回答多个问题，因此不需要为每项要求各发一次请求。这既减少了延迟，也保证了五项判断看到的是完全一致的对话快照。

第五步：建立实时语音会话

决策函数需要对话内容，也需要把结果变成口语回复，这两件事由实时语音模型完成。在示例中，GPT-Live 承担两个职责：

• 听保洁员说话，并生成语音回复。
• 随着对话推进，把双方的发言都转写下来。

转写结果以事件形式通过同一条连接送达，因此不需要额外的语音转文字步骤。应用把这些事件收集成一份持续增长的对话记录，正好就是决策函数的输入。

这里需要换一种思路。使用常规大模型或决策模型时，是发一次请求、拿一次响应，然后就结束了。使用实时语音模型时，是打开一个会话并让它持续运行：保洁员说话、助手回复、保洁员再说，全部发生在同一条连接上。WebSocket 连接可以维持这种持续交互。

import json
from websockets.asyncio.client import connect

async with connect(
    LIVE_URL,
    additional_headers={"api-key": API_KEY},
) as ws:
    await ws.send(json.dumps({
        "type": "session.start",
        "session": {
            "model": "gpt-live-1",
            "instructions": LIVE_INSTRUCTIONS,
            "audio": {
                "output": {"voice": "marin"},
            },
            "delegation": {"type": "client"},
        },
    }))
    # Audio streaming and event handling continue here.

几个参数的含义：

• LIVE_URL：实时语音模型的端点地址，示例中通过 Azure 访问。
• LIVE_INSTRUCTIONS：告诉语音模型它扮演什么角色、该怎么说话。
• audio.output.voice：输出音色，示例使用 marin。
• delegation.type：设为 client，表示允许语音模型在对话过程中把工作交给我们的 Python 应用。「client」在这里指的就是这个 Python 应用。

第六步：把决策结果接回语音回复

当语音模型需要一次评估时，它会通过 WebSocket 发来一个 delegation 事件。应用收到后调用 assess_inspection，把决策模型返回的五个选项发回给语音模型，由它组织成对保洁员说的话。

整个循环可以概括为：

1. 保洁员说一段话，语音模型转写并通过事件送达应用。
2. 应用把转写追加到对话记录。
3. 语音模型发出 delegation 事件，请求评估。
4. 应用调用决策函数，一次拿到四项状态加一个下一步动作。
5. 应用把结果回传给语音模型。
6. 语音模型据此生成口语回复。
7. 保洁员再次回应，回到第 1 步。

应用侧不需要自己决定「该问哪一项」，这个判断已经包含在 next_action 的返回值里；应用也不需要自己组织话术，措辞交给语音模型。应用真正要做的是维护对话记录、转发事件、以及在状态变化时更新自己的业务数据。

一个完整示例

下面把各部分串起来，走一遍最小可运行的流程。为便于阅读，省略了音频流的收发细节，只保留与决策相关的骨架。

import json
from websockets.asyncio.client import connect

conversation = []

async def assess_inspection(conversation):
    response = await http.post(
        f"{AZURE_ENDPOINT}/providers/microsoft/v1/systemone",
        json={
            "model": "Decision-1",
            "state": {"conversation": conversation},
            "questions": QUESTIONS,
        },
    )
    response.raise_for_status()
    return {
        name: answer["choice"]
        for name, answer in response.json()["answers"].items()
    }

async def run():
    async with connect(
        LIVE_URL,
        additional_headers={"api-key": API_KEY},
    ) as ws:
        await ws.send(json.dumps({
            "type": "session.start",
            "session": {
                "model": "gpt-live-1",
                "instructions": LIVE_INSTRUCTIONS,
                "audio": {"output": {"voice": "marin"}},
                "delegation": {"type": "client"},
            },
        }))

        async for raw in ws:
            event = json.loads(raw)

            if event["type"] == "transcript":
                conversation.append(event["text"])

            elif event["type"] == "delegation":
                decisions = await assess_inspection(conversation)
                await ws.send(json.dumps({
                    "type": "delegation.result",
                    "result": decisions,
                }))

用前面那段对话跑一遍，预期会看到这样的推进过程：

1. 保洁员报告门禁码可用、床已铺好、厨房和洗手池还没看。此时 entrance 与 bedroom 应为 ready，bathroom 与 kitchen 应为 needs_evidence，next_action 在 ask_bathroom 与 ask_kitchen 之间择一。
2. 助手追问洗手池。保洁员说明水要到下午才通，但厨房现在可以看。此时 bathroom 仍为 needs_evidence，next_action 应转为 ask_kitchen，而不是继续追问洗手池。
3. 保洁员报告厨房情况后，若四项都已有结论，next_action 变为 finish，助手汇总结果，包括任何已确认的问题。

如果中途出现「不过那是在我最后一次改动之前」这类更正，决策模型应当把相关项从 ready 退回 needs_evidence，同时保留其他不受影响的结论。这正是把证据标准写进 instructions 的意义所在。

注意事项

状态定义要共用，不要各写一套。 四项要求共用同一组 ready / issue / needs_evidence 定义，可以避免同一段对话在不同项上被解读成互相矛盾的结果。

指令里的措辞直接影响判断。 「计划和一般性保证不能确立结果」「先前的证据必须仍然适用于当前状况」这两句不是装饰，去掉之后模型很容易把「我待会儿去测」当成证据，或者把已经失效的旧结论继续沿用。

finish 不等于全部通过。 它只表示四项都已有足够证据可以评估。应用在收到 finish 后仍要读取各项状态，把 issue 单独呈现出来。

五个问题必须一次发出。 分开发送会让五项判断基于不同的对话快照，可能出现前后不一致。Decision-1 的接口支持针对同一上下文同时回答多个问题，应利用这一点。

对话记录要完整。 决策函数的输入是累积的对话，而不是最近一句。只传最后一句会丢失上下文，更正类语句尤其无法正确处理。

delegation 需要显式开启。 会话配置里 delegation.type 设为 client，语音模型才会在需要评估时把工作交回应用。漏掉这一项，决策函数不会被调用。

模型与接口信息以官网为准。 本文涉及的模型名称、端点路径、请求字段、音色标识、计费方式与区域可用性都可能调整，接入前请查阅对应厂商当前公布的文档。决策模型也可以替换为其他同类服务，替换时只需改动决策函数内部的请求写法，整体流程不变。