AWS Lex链式回退机器人触发异常及Lambda响应报错排查

问题背景 #

通过回退意图串联多个AWS Lex机器人，按内容类型拆分处理：Bot_A含自定义问询意图，Bot_B、Bot_C仅返回单句答复并关闭会话。新增带自定义问询意图的机器人（或新建），要求用户输入无法匹配前3个机器人时触发该机器人的Lambda问询对话，但通过回退Lambda触发槽位引导（Elicit Slot）时，Lex抛出错误：An error has occurred: The server encountered an error processing the Lambda response，已按规范格式返回响应仍报错。

错误原因 #

当前代码中handle_other_chatbot函数强制将子机器人的所有响应以Close动作返回给主机器人，忽略了子机器人返回的dialogState类型。当子机器人需要触发ElicitSlot引导槽位时，主机器人返回的Close动作与子机器人的对话状态冲突，导致Lex无法处理该响应。

修正方案 #

修改handle_other_chatbot函数，根据子机器人返回的dialogState动态生成对应的dialogAction，而非统一返回Close动作：

import logging
import boto3

logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

lex_runtime = boto3.client('lex-runtime')

# --- Helpers that build all of the responses to Lex ---

def elicit_slot(session_attributes, intent_name, slots, slot_to_elicit, message):
    return {
        'sessionAttributes': session_attributes,
        'dialogAction': {
            'type': 'ElicitSlot',
            'intentName': intent_name,
            'slots': slots,
            'slotToElicit': slot_to_elicit,
            'message': message
        }
    }

def confirm_intent(session_attributes, intent_name, slots, message):
    return {
        'sessionAttributes': session_attributes,
        'dialogAction': {
            'type': 'ConfirmIntent',
            'intentName': intent_name,
            'slots': slots,
            'message': message
        }
    }

def close(session_attributes, fulfillment_state, message):
    response = {
        'sessionAttributes': session_attributes,
        'dialogAction': {
            'type': 'Close',
            'fulfillmentState': fulfillment_state,
            'message': message
        }
    }

    return response

def delegate(session_attributes, slots):
    return {
        'sessionAttributes': session_attributes,
        'dialogAction': {
            'type': 'Delegate',
            'slots': slots
        }
    }
    
# --- Helper Functions ---

def get_session_attributes(intent_request):
    """
    Get session attributes from intent request
    """
    return intent_request['sessionAttributes'] if 'sessionAttributes' in intent_request and intent_request['sessionAttributes'] is not None else {}
    
def get_user_id(intent_request):
    """
    Get user id from intent request
    """
    return intent_request.get('userId')
    
def get_input_transcript(intent_request):
    """
    Get input transcript from intent request
    """
    return intent_request['inputTranscript']
    
# --- Intent handler ---

def handle_other_chatbot(intent_request, bot_name, bot_alias):
    userId = get_user_id(intent_request)
    session_attributes = get_session_attributes(intent_request)
    input_text = get_input_transcript(intent_request)
    
    response = lex_runtime.post_text(
        botName=bot_name,
        botAlias=bot_alias,
        userId=userId,
        sessionAttributes=session_attributes,
        inputText=input_text
    )
    
    session_attributes = response['sessionAttributes']
    message = {
        "contentType": response['messageFormat'],
        "content": response['message']
    }
    
    # 根据子机器人的dialogState返回对应动作
    dialog_state = response['dialogState']
    if dialog_state == 'ElicitSlot':
        return elicit_slot(
            session_attributes,
            response['intentName'],
            response['slots'],
            response['slotToElicit'],
            message
        )
    elif dialog_state == 'ConfirmIntent':
        return confirm_intent(
            session_attributes,
            response['intentName'],
            response['slots'],
            message
        )
    elif dialog_state == 'Delegate':
        return delegate(session_attributes, response['slots'])
    elif dialog_state in ['Fulfilled', 'Failed']:
        return close(session_attributes, dialog_state, message)
    else:
        # 处理未知状态，返回默认关闭动作
        return close(session_attributes, 'Failed', {
            'contentType': 'PlainText',
            'content': '对话状态异常，请重试'
        })

# --- Intents ---

def dispatch(intent_request):
    logger.debug('dispatch userId={}, botName={}'.format(intent_request.get('userId', ""), intent_request['bot']['name']))
    
    if intent_request['bot']['name'] == "BOT_A":
        return handle_other_chatbot(intent_request, "BOT_B", "AliasA")
    elif intent_request['bot']['name'] == "BOT_B":
        return handle_other_chatbot(intent_request, "BOT_C", "AliasA")
    else:
        # 新增逻辑：调用带自定义问询的机器人，比如BOT_D
        return handle_other_chatbot(intent_request, "BOT_D", "AliasA")

def lambda_handler(event, context):
    logger.debug('event.bot.name={}'.format(event['bot']['name']))
    return dispatch(event)

关键修改点 #

1. 新增根据子机器人dialogState分支处理逻辑，匹配ElicitSlot、ConfirmIntent、Delegate、Close等所有对话状态
2. 补充未知状态的兜底处理，避免异常
3. 在dispatch函数中添加对新增机器人（如BOT_D）的调用逻辑

验证步骤 #

1. 确保Lambda角色拥有lex-runtime:PostText权限，能调用所有串联的机器人
2. 单独测试新增的机器人，确认其能正常触发ElicitSlot流程
3. 测试完整串联流程，验证用户输入无法匹配前3个机器人时，新增机器人的槽位引导能正常触发

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