如何在单个Alexa Skill Intent中实现多轮问答并存储数据（附Python代码需求）

问题核心 #

你已创建QuestionIntent并触发第一个问题，但用户回复时会误触发其他Intent——这是因为Alexa默认匹配置信度最高的Intent，未通过会话上下文锁定流程范围。要实现单Intent内完成4-5轮问答，核心是用会话属性跟踪流程状态+请求拦截强制路由。

核心解决思路 #

• 用**会话属性（Session Attributes）**保存当前问题索引、已收集的回答，维持问答流程上下文
• 添加请求拦截器：如果会话中存在问答流程标记，强制将所有用户请求路由到QuestionIntent，避免误触发其他Intent
• 自定义槽位选用AMAZON.SearchQuery（支持自由文本输入），确保用户任意回答都能被捕获

完整Python实现代码 #

from ask_sdk_core.skill_builder import SkillBuilder
from ask_sdk_core.dispatch_components import AbstractRequestInterceptor, AbstractRequestHandler
from ask_sdk_core.utils import is_intent_name, get_slot_value
from ask_sdk_core.handler_input import HandlerInput
from ask_sdk_model import Response
import sqlite3
from typing import Dict, Any

# 初始化SQLite数据库（示例，可替换为MySQL/PostgreSQL等）
def init_db():
    conn = sqlite3.connect('alexa_responses.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS user_responses
                 (user_id TEXT, question1 TEXT, question2 TEXT, question3 TEXT, question4 TEXT, question5 TEXT)''')
    conn.commit()
    conn.close()

init_db()

# 模拟问题列表（实际可替换为API调用获取）
QUESTIONS = [
    "请问你的年龄是多少？",
    "你最喜欢的颜色是什么？",
    "你平时的业余爱好是什么？",
    "你最喜欢的食物是什么？",
    "你居住在哪个城市？"
]

# 请求拦截器：强制路由到QuestionIntent如果处于问答流程中
class QaFlowInterceptor(AbstractRequestInterceptor):
    def process(self, handler_input: HandlerInput) -> None:
        session_attr = handler_input.attributes_manager.session_attributes
        # 检查会话中是否存在当前问题索引，判定处于问答流程
        if "current_question_idx" in session_attr:
            # 修改请求的Intent为QuestionIntent
            handler_input.request_envelope.request.intent.name = "QuestionIntent"

# 问答流程核心Handler
class QuestionIntentHandler(AbstractRequestHandler):
    def can_handle(self, handler_input: HandlerInput) -> bool:
        return is_intent_name("QuestionIntent")(handler_input)

    def handle(self, handler_input: HandlerInput) -> Response:
        session_attr = handler_input.attributes_manager.session_attributes
        user_id = handler_input.request_envelope.context.system.user.user_id
        speech_text = ""

        # 初始化流程或继续流程
        if "current_question_idx" not in session_attr:
            # 首次触发，启动第一个问题
            session_attr["current_question_idx"] = 0
            session_attr["collected_responses"] = []
            speech_text = QUESTIONS[0]
        else:
            # 收集当前回答
            current_idx = session_attr["current_question_idx"]
            user_response = get_slot_value(handler_input=handler_input, slot_name="UserAnswer")
            session_attr["collected_responses"].append(user_response)

            # 判断是否完成所有问题
            if current_idx + 1 < len(QUESTIONS):
                # 进入下一个问题
                session_attr["current_question_idx"] = current_idx + 1
                speech_text = QUESTIONS[current_idx + 1]
            else:
                # 所有问题完成，存储到数据库
                responses = session_attr["collected_responses"]
                # 补全不足5个的回答（适配固定表结构）
                while len(responses) < 5:
                    responses.append(None)
                
                conn = sqlite3.connect('alexa_responses.db')
                c = conn.cursor()
                c.execute('''INSERT INTO user_responses 
                             (user_id, question1, question2, question3, question4, question5)
                             VALUES (?, ?, ?, ?, ?, ?)''', 
                          (user_id, responses[0], responses[1], responses[2], responses[3], responses[4]))
                conn.commit()
                conn.close()

                # 结束流程，清除会话属性
                session_attr.pop("current_question_idx")
                session_attr.pop("collected_responses")
                speech_text = "所有问题已完成，你的回答已保存，谢谢！"

        return (
            handler_input.response_builder
                .speak(speech_text)
                .ask(speech_text if "current_question_idx" in session_attr else None)
                .response
        )

# 帮助Intent示例（按需扩展其他Intent）
class HelpIntentHandler(AbstractRequestHandler):
    def can_handle(self, handler_input: HandlerInput) -> bool:
        return is_intent_name("AMAZON.HelpIntent")(handler_input)

    def handle(self, handler_input: HandlerInput) -> Response:
        speech_text = "我会依次问你几个问题，请直接回答即可。"
        return (
            handler_input.response_builder
                .speak(speech_text)
                .ask(speech_text)
                .response
        )

# 构建技能
sb = SkillBuilder()
sb.add_request_interceptor(QaFlowInterceptor())
sb.add_request_handler(QuestionIntentHandler())
sb.add_request_handler(HelpIntentHandler())

lambda_handler = sb.lambda_handler()

关键说明 #

1. 会话属性管理：通过current_question_idx跟踪当前问题位置，collected_responses存储已回答内容，确保流程上下文不中断
2. 请求拦截器：QaFlowInterceptor会检查会话属性，只要处于问答流程中，就强制将用户请求路由到QuestionIntent，彻底避免误触发其他Intent
3. 槽位配置：在Alexa开发者控制台中，给QuestionIntent添加名为UserAnswer的槽位，类型选择AMAZON.SearchQuery，支持自由文本输入
4. 数据库适配：示例用SQLite，实际可替换为其他数据库，只需修改连接和操作代码
5. 问题来源：示例中QUESTIONS是静态列表，可替换为API调用逻辑，在流程初始化时动态获取问题

内容的提问来源于stack exchange，提问作者shubham salokhe