【人机交互 6-10】第 8/25 课

🤖 第08课：对话管理

📌 对话管理概述

对话管理是语音交互的指挥中心，决定机器人说什么和做什么：

🎯 对话管理三职责

职责描述关键问题
状态跟踪记住对话进度用户说了什么？还缺什么？
行动选择决定下一步追问？确认？执行？
上下文管理维护对话历史"那里"指哪里？

📌 多轮对话管理

class DialogueManager:
    def __init__(self):
        self.state = "idle"  # idle/listening/confirming/executing/followup
        self.context = {"intent": None, "slots": {}, "history": []}
        self.required_slots = {
            "navigate": ["location"],
            "deliver": ["item", "person"],
            "query": ["target"],
        }
        self.responses = {
            "acknowledge": ["好的，","收到，","没问题，"],
            "clarify_location": ["请问您要去哪里？","您想去哪个位置？"],
            "clarify_item": ["请问您要送什么？","需要配送什么物品？"],
            "clarify_person": ["请问送给谁？","需要送给哪位？"],
            "confirm": ["确认一下：{confirm_text}，对吗？"],
            "execute": ["好的，马上为您{action}！"],
            "followup": ["还有其他需要帮助的吗？","还需要什么吗？"],
        }

    def process(self, user_input, nlu_result=None):
        self.context["history"].append({"role": "user", "text": user_input})
        
        if self.state == "idle":
            if nlu_result and nlu_result["intent"] not in [None, "greet"]:
                self.context["intent"] = nlu_result["intent"]
                self.context["slots"].update(nlu_result.get("slots", {}))
                self.state = "confirming"
                return self._check_and_respond()
            return "您好，我是服务机器人，有什么可以帮您？"

        elif self.state == "confirming":
            if "是" in user_input or "对" in user_input or "好的" in user_input:
                self.state = "executing"
                action = self._describe_action()
                resp = f"好的，马上为您{action}！"
                self.context["history"].append({"role": "bot", "text": resp})
                self.state = "followup"
                return resp
            elif "不" in user_input or "改" in user_input:
                self.context["slots"] = {}
                self.state = "confirming"
                return self._check_and_respond()
            else:
                self.context["slots"].update(nlu_result.get("slots", {}) if nlu_result else {})
                return self._check_and_respond()

        elif self.state == "followup":
            if "不" in user_input or "没有" in user_input or "谢谢" in user_input:
                self.state = "idle"
                self.context = {"intent": None, "slots": {}, "history": []}
                return "好的，随时为您服务！"
            else:
                self.state = "idle"
                return self.process(user_input, nlu_result)

    def _check_and_respond(self):
        required = self.required_slots.get(self.context["intent"], [])
        missing = [s for s in required if s not in self.context["slots"]]
        if missing:
            prompts = {"location": "请问您要去哪里？", "item": "请问要送什么？",
                       "person": "请问送给谁？", "target": "请问查询什么？"}
            resp = prompts.get(missing[0], "请提供更多信息")
        else:
            confirm = self._describe_action()
            resp = f"确认：{confirm}，对吗？"
        self.context["history"].append({"role": "bot", "text": resp})
        return resp

    def _describe_action(self):
        s = self.context["slots"]
        intent = self.context["intent"]
        if intent == "navigate":
            floor = f"{s.get('floor','')}楼" if 'floor' in s else ''
            return f"导航到{floor}{s.get('location','目标位置')}"
        elif intent == "deliver":
            return f"送{s.get('item','物品')}给{s.get('person','对方')}"
        elif intent == "query":
            return f"查询{s.get('target','位置')}的信息"
        return "执行操作"

dm = DialogueManager()
print("对话管理模拟")
print("=" * 55)

# 模拟完整对话
conversations = [
    [("你好", None), ("请带我去3楼会议室", {"intent":"navigate","slots":{"floor":3,"location":"会议室"}}),
     ("是的", None), ("没有了，谢谢", None)],
    [("送咖啡给张总", {"intent":"deliver","slots":{"item":"咖啡"}}),
     ("张总", {"intent":"deliver","slots":{"person":"张总"}}),
     ("对的", None), ("没有了", None)],
]

for ci, conv in enumerate(conversations):
    print(f"\n📋 对话{ci+1}:")
    for user_text, nlu in conv:
        resp = dm.process(user_text, nlu)
        print(f"  用户: {user_text}")
        print(f"  机器人: {resp}")

print("\n✅ 对话管理验证通过")
✅ 验证通过 对话管理模拟 ======================================================= 📋 对话1: 用户: 你好 机器人: 您好，我是服务机器人，有什么可以帮您？ 用户: 请带我去3楼会议室 机器人: 确认：导航到3楼会议室，对吗？ 用户: 是的 机器人: 好的，马上为您导航到3楼会议室！ 用户: 没有了，谢谢 机器人: 好的，随时为您服务！ 📋 对话2: 用户: 送咖啡给张总 机器人: 请问送给谁？ 用户: 张总 机器人: 确认：送咖啡给张总，对吗？ 用户: 对的 机器人: 好的，马上为您送咖啡给张总！ 用户: 没有了 机器人: 好的，随时为您服务！ ✅ 对话管理验证通过

📌 状态机对话模型

有限状态机(FSM)是最经典和可靠的对话管理方法：

class StateMachineDialogue:
    """基于有限状态机的对话管理"""
    def __init__(self):
        self.states = {
            "GREET": {"greet": "COLLECT", "navigate": "CONFIRM", "deliver": "COLLECT", "query": "CONFIRM"},
            "COLLECT": {"inform": "CONFIRM", "confirm": "EXECUTE"},
            "CONFIRM": {"affirm": "EXECUTE", "negate": "COLLECT", "inform": "CONFIRM"},
            "EXECUTE": {"done": "FOLLOWUP", "fail": "RECOVER"},
            "FOLLOWUP": {"request": "COLLECT", "negate": "GREET"},
            "RECOVER": {"retry": "COLLECT", "cancel": "GREET"},
        }
        self.current = "GREET"
        self.history = []

    def transition(self, user_action):
        state_transitions = self.states.get(self.current, {})
        new_state = state_transitions.get(user_action, self.current)
        self.history.append((self.current, user_action, new_state))
        self.current = new_state
        return new_state

    def get_prompt(self):
        prompts = {
            "GREET": "您好，请问需要什么帮助？",
            "COLLECT": "请告诉我更多信息。",
            "CONFIRM": "确认以上信息是否正确？",
            "EXECUTE": "正在为您执行...",
            "FOLLOWUP": "还有其他需要帮助的吗？",
            "RECOVER": "出了点问题，要重试还是取消？",
        }
        return prompts.get(self.current, "")

sm = StateMachineDialogue()
print("状态机对话管理")
print("=" * 55)

actions = ["greet", "inform", "confirm", "affirm", "done", "negate"]
for action in actions:
    old = sm.current
    new = sm.transition(action)
    print(f"  {old} --{action}--> {new}: {sm.get_prompt()}")

print("\n✅ 状态机对话验证通过")
✅ 验证通过 状态机对话管理 ======================================================= GREET --greet--> COLLECT: 请告诉我更多信息。 COLLECT --inform--> CONFIRM: 确认以上信息是否正确？ CONFIRM --confirm--> CONFIRM: 确认以上信息是否正确？ CONFIRM --affirm--> EXECUTE: 正在为您执行... EXECUTE --done--> FOLLOWUP: 还有其他需要帮助的吗？ FOLLOWUP --negate--> GREET: 您好，请问需要什么帮助？ ✅ 状态机对话验证通过

📌 对话上下文管理

多轮对话需要上下文记忆和指代消解：

class ContextManager:
    """对话上下文管理 - 多轮对话记忆"""
    def __init__(self, max_turns=10):
        self.max_turns = max_turns
        self.context = {"entities": {}, "last_intent": None, "conversation": [], "global_state": {}}

    def update(self, role, text, nlu=None):
        self.context["conversation"].append({"role": role, "text": text})
        if len(self.context["conversation"]) > self.max_turns * 2:
            self.context["conversation"] = self.context["conversation"][-(self.max_turns*2):]
        if nlu:
            self.context["last_intent"] = nlu.get("intent")
            self.context["entities"].update(nlu.get("slots", {}))

    def resolve_reference(self, text):
        """指代消解"""
        resolved = text
        replacements = {
            "那里": self.context["entities"].get("location"),
            "那个": self.context["entities"].get("item"),
            "他": self.context["entities"].get("person"),
            "那边": self.context["entities"].get("location"),
        }
        for ref, entity in replacements.items():
            if ref in text and entity:
                resolved = text.replace(ref, entity)
        return resolved

    def get_summary(self):
        return {
            "turns": len([m for m in self.context["conversation"] if m["role"]=="user"]),
            "last_intent": self.context["last_intent"],
            "entities": self.context["entities"],
        }

ctx = ContextManager()
ctx.update("user", "请带我去3楼会议室", {"intent":"navigate","slots":{"floor":3,"location":"会议室"}})
ctx.update("bot", "好的，正在为您导航到3楼会议室")
ctx.update("user", "从那里怎么去洗手间？")
resolved = ctx.resolve_reference("从那里怎么去洗手间？")
ctx.update("bot", f"从会议室到洗手间，请左转走50米")

print("对话上下文管理")
print("=" * 55)
print(f"指代消解: '从那里怎么去洗手间？' → '{resolved}'")
print(f"上下文摘要: {ctx.get_summary()}")
print("\n✅ 上下文管理验证通过")
✅ 验证通过 对话上下文管理 ======================================================= 指代消解: '从那里怎么去洗手间？' → '从会议室怎么去洗手间？' 上下文摘要: {'turns': 2, 'last_intent': 'navigate', 'entities': {'floor': 3, 'location': '会议室'}} ✅ 上下文管理验证通过

📌 对话策略

📊 对话策略对比

策略特点优点缺点
状态机预定义转移可控、可解释灵活性差
帧填充槽位驱动结构清晰难以处理复杂逻辑
强化学习学习最优策略自适应训练难、不透明
大模型端到端生成灵活自然不可控、幻觉
💡 生产推荐：状态机主干 + LLM兜底。高频场景用状态机保证可靠性，低频/开放域用LLM增加灵活性。

📌 练习

📝 练习 1

实现打断处理：用户在机器人说话时打断并说'停'，机器人应立即停止当前动作并转入等待状态。

📝 练习 2

设计对话修复机制：当用户说'不对'或'我说错了'时，回退到上一轮对话状态重新收集信息。

📝 练习 3

用强化学习训练对话策略：定义奖励函数（对话轮数最少、任务完成率最高），用Q-Learning学习最优追问策略。

📌 成就

🏆 本课成就

◀ 上一课 📚 目录 下一课 ▶