本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

# 在 Amazon Bedrock 代理中编写自定义解析器 Lambda 函数
<a name="lambda-parser"></a>

每个提示模板都包含一个解析器 Lambda 函数。您可以编写自己的自定义解析器 Lambda 函数，并指定要覆盖其默认解析器函数的模板。要编写自定义解析器 Lambda 函数，您必须了解代理发送的输入事件以及代理期望作为 Lambda 函数输出的响应。您可以编写处理程序函数来处理输入事件中的变量并返回响应。有关AWS Lambda工作原理的更多信息，请参阅《开发者指南》AWS Lambda中的[事件驱动调用](https://docs.aws.amazon.com/lambda/latest/dg/lambda-services.html#event-driven-invocation)。

**Topics**
+ [解析器 Lambda 输入事件](#lambda-parser-input)
+ [解析器 Lambda 响应](#lambda-parser-response)
+ [解析器 Lambda 示例](#lambda-parser-example)

## 解析器 Lambda 输入事件
<a name="lambda-parser-input"></a>

以下是来自代理的输入事件的一般结构。使用以下字段编写 Lambda 处理程序函数。

```
{
    "messageVersion": "1.0",
    "agent": {
        "name": "string",
        "id": "string",
        "alias": "string",
        "version": "string"
    },
    "invokeModelRawResponse": "string",
    "promptType": "ORCHESTRATION | ROUTING_CLASSIFIER | POST_PROCESSING | PRE_PROCESSING | KNOWLEDGE_BASE_RESPONSE_GENERATION | MEMORY_SUMMARIZATION",
    "overrideType": "OUTPUT_PARSER"
}
```

下面的列表介绍了输入事件字段：
+ `messageVersion` – 消息的版本，用于标识进入 Lambda 函数的事件数据格式以及 Lambda 函数的预期响应格式。Amazon Bedrock 代理仅支持版本 1.0。
+ `agent` – 包含提示所属代理的名称、ID、别名和版本的相关信息。
+ `invokeModelRawResponse` – 要解析其输出的提示的原始根基模型输出。
+ `promptType` – 要解析其输出的提示类型。
+ `overrideType` – 此 Lambda 函数覆盖的构件。当前，仅支持 `OUTPUT_PARSER`，这表示将会覆盖默认解析器。

## 解析器 Lambda 响应
<a name="lambda-parser-response"></a>

您的代理将会收到您的 Lambda 函数的响应，并使用响应采取进一步操作或借助它向用户返回响应。您的代理将执行代理模型建议的后续操作。可以按顺序执行后续操作，也可以并行执行，具体取决于代理的模型以及代理的创建和准备时间。

如果您在 *2024 年 10 月 4 日之前*创建并准备好了代理，并且您的代理使用的是 Anthropic Claude 3 Sonnet 或 Anthropic Claude 3.5 Sonnet 模型，则默认情况下，将按顺序运行代理模型建议的重要后续操作。

如果您在 *2024 年 10 月 10 日之后*创建了新代理或准备好了现有代理，并且您的代理使用的是 Anthropic Claude 3 Sonnet、Anthropic Claude 3.5 Sonnet、或任何 non-Anthropic 模型，则将并行运行代理模型建议的后续操作。这意味着将并行执行多个操作，例如操作组函数和知识库的组合。这将减少对模型的调用次数，从而减少整体延迟。

您可以通过调用 [PrepareAgent](https://docs.aws.amazon.com//bedrock/latest/APIReference/API_agent_PrepareAgent.html)API 或在控制台的代理生成器中选择准备，为在 *2024 年 10 月 4 日之前*创建和**准备**的代理启用并行操作。代理准备好后，您将看到更新后的提示模板和新版本的解析器 Lambda 架构。

**解析器 Lambda 响应示例**

以下示例展示了按顺序运行建议的重要后续操作的代理和并行运行后续操作的代理的响应的一般结构。使用 Lambda 函数响应字段来配置如何返回输出。

**按顺序运行建议的重要后续操作的代理的响应示例**

根据您是使用 OpenAPI 架构还是函数详细信息定义操作组来选择相应的选项卡：

**注意**  
`MessageVersion 1.0` 表示代理正在按顺序运行建议的重要后续操作。

------
#### [ OpenAPI schema ]

```
{
    "messageVersion": "1.0",
    "promptType": "ORCHESTRATION | PRE_PROCESSING | ROUTING_CLASSIFIER | POST_PROCESSING | KNOWLEDGE_BASE_RESPONSE_GENERATION",
    "preProcessingParsedResponse": {
        "isValidInput": "boolean",
        "rationale": "string"
    },
    "orchestrationParsedResponse": {
        "rationale": "string",
        "parsingErrorDetails": {
            "repromptResponse": "string"
        },
        "responseDetails": {
            "invocationType": "AGENT_COLLABORATOR | ACTION_GROUP | KNOWLEDGE_BASE | FINISH | ASK_USER",
            "agentAskUser": {
                "responseText": "string",
                "id": "string"
            },
             "agentCollaboratorInvocation": {
                "agentCollaboratorName": "string",
                "input": {
                    "text": "string"                    
                }
            }
            ...
        }
    },
    "routingClassifierParsedResponse": {
        "parsingErrorDetails": {
            "repromptResponse": "string"
        },
        "responseDetails": {
            "type": "AGENT | LAST_AGENT | UNDECIDED",
            "agentCollaboratorInvocation": {
                "agentCollaboratorName": "string",
                "input": {
                    "text": "string"                    
                    }
            }
        }
    }
}
            "actionGroupInvocation": {
                "actionGroupName": "string",
                "apiName": "string",
                "id": "string",
                "verb": "string",
                "actionGroupInput": {
                    "<parameter>": {
                        "value": "string"
                    },
                    ...
                }
            },
            "agentKnowledgeBase": {
                "knowledgeBaseId": "string",
                "id": "string",
                "searchQuery": {
                    "value": "string"
                }
            },
            "agentFinalResponse": {
                "responseText": "string",
                "citations": {
                    "generatedResponseParts": [{
                        "text": "string",
                        "references": [{"sourceId": "string"}]
                    }]
                }
            },
        }
    },
    "knowledgeBaseResponseGenerationParsedResponse": { 
       "generatedResponse": {
            "generatedResponseParts": [
                {
                    "text": "string",
                    "references": [
                        {"sourceId": "string"},
                        ...
                    ]
                }
            ]
        }
    },
    "postProcessingParsedResponse": {
        "responseText": "string",
        "citations": {
            "generatedResponseParts": [{
                "text": "string",
                "references": [{
                    "sourceId": "string"
                }]
            }]
        }
    }
}
```

------
#### [ Function details ]

```
{
    "messageVersion": "1.0",
    "promptType": "ORCHESTRATION | PRE_PROCESSING | POST_PROCESSING | KNOWLEDGE_BASE_RESPONSE_GENERATION",
    "preProcessingParsedResponse": {
        "isValidInput": "boolean",
        "rationale": "string"
    },
    "orchestrationParsedResponse": {
        "rationale": "string",
        "parsingErrorDetails": {
            "repromptResponse": "string"
        },
        "responseDetails": {
            "invocationType": "ACTION_GROUP | KNOWLEDGE_BASE | FINISH | ASK_USER",
            "agentAskUser": {
                "responseText": "string",
                "id": "string"
            },
            "actionGroupInvocation": {
                "actionGroupName": "string",
                "functionName": "string",
                "id": "string",
                "actionGroupInput": {
                    "<parameter>": {
                        "value": "string"
                    },
                    ...
                }
            },
            "agentKnowledgeBase": {
                "knowledgeBaseId": "string",
                "id": "string",
                "searchQuery": {
                    "value": "string"
                }
            },
            "agentFinalResponse": {
                "responseText": "string",
                "citations": {
                    "generatedResponseParts": [{
                        "text": "string",
                        "references": [{"sourceId": "string"}]
                    }]
                }
            },
        }
    },
    "knowledgeBaseResponseGenerationParsedResponse": { 
       "generatedResponse": {
            "generatedResponseParts": [
                {
                    "text": "string",
                    "references": [
                        {"sourceId": "string"},
                        ...
                    ]
                }
            ]
        }
    },
    "postProcessingParsedResponse": {
        "responseText": "string",
        "citations": {
            "generatedResponseParts": [{
                "text": "string",
                "references": [{
                    "sourceId": "string"
                }]
            }]
        }
    }
}
```

------

**并行运行后续操作的代理的响应示例**

根据您是使用 OpenAPI 架构还是函数详细信息定义操作组来选择相应的选项卡：

**注意**  
`MessageVersion 2.0` 表示代理正在并行运行建议的后续操作。

------
#### [ OpenAPI schema ]

```
{
    "messageVersion": "2.0",
    "promptType": "ORCHESTRATION | PRE_PROCESSING | POST_PROCESSING | KNOWLEDGE_BASE_RESPONSE_GENERATION",
    "preProcessingParsedResponse": {
        "isValidInput": "boolean",
        "rationale": "string"
    },
    "orchestrationParsedResponse": {
        "rationale": "string",
        "parsingErrorDetails": {
            "repromptResponse": "string"
        },
        "responseDetails": {
            "invocationType": "ACTION_GROUP | KNOWLEDGE_BASE | FINISH | ASK_USER",
            "agentAskUser": {
                "responseText": "string"
            },
            "actionGroupInvocations": [
                {
                    "actionGroupName": "string",
                    "apiName": "string",
                    "verb": "string",
                    "actionGroupInput": {
                        "<parameter>": {
                            "value": "string"
                        },
                        ...
                    }
                }
            ],
            "agentKnowledgeBases": [
                {
                    "knowledgeBaseId": "string",
                    "searchQuery": {
                        "value": "string"
                    }
                }
            ],
            "agentFinalResponse": {
                "responseText": "string",
                "citations": {
                    "generatedResponseParts": [{
                        "text": "string",
                        "references": [{"sourceId": "string"}]
                    }]
                }
            },
        }
    },
    "knowledgeBaseResponseGenerationParsedResponse": { 
       "generatedResponse": {
            "generatedResponseParts": [
                {
                    "text": "string",
                    "references": [
                        {"sourceId": "string"},
                        ...
                    ]
                }
            ]
        }
    },
    "postProcessingParsedResponse": {
        "responseText": "string",
        "citations": {
            "generatedResponseParts": [{
                "text": "string",
                "references": [{
                    "sourceId": "string"
                }]
            }]
        }
    }
}
```

------
#### [ Function details ]

```
{
    "messageVersion": "2.0",
    "promptType": "ORCHESTRATION | PRE_PROCESSING | POST_PROCESSING | KNOWLEDGE_BASE_RESPONSE_GENERATION",
    "preProcessingParsedResponse": {
        "isValidInput": "boolean",
        "rationale": "string"
    },
    "orchestrationParsedResponse": {
        "rationale": "string",
        "parsingErrorDetails": {
            "repromptResponse": "string"
        },
        "responseDetails": {
            "invocationType": "ACTION_GROUP | KNOWLEDGE_BASE | FINISH | ASK_USER",
            "agentAskUser": {
                "responseText": "string"
            },
            "actionGroupInvocations": [
                {
                    "actionGroupName": "string",
                    "functionName": "string",
                    "actionGroupInput": {
                        "<parameter>"": {
                            "value": "string"
                        },
                        ...
                    }
                }
            ],
            "agentKnowledgeBases": [
                {
                    "knowledgeBaseId": "string",
                    "searchQuery": {
                        "value": "string"
                    }
                }
            ],
            "agentFinalResponse": {
                "responseText": "string",
                "citations": {
                    "generatedResponseParts": [{
                        "text": "string",
                        "references": [{"sourceId": "string"}]
                    }]
                }
            },
        }
    },
    "knowledgeBaseResponseGenerationParsedResponse": { 
       "generatedResponse": {
            "generatedResponseParts": [
                {
                    "text": "string",
                    "references": [
                        {"sourceId": "string"},
                        ...
                    ]
                }
            ]
        }
    },
    "postProcessingParsedResponse": {
        "responseText": "string",
        "citations": {
            "generatedResponseParts": [{
                "text": "string",
                "references": [{
                    "sourceId": "string"
                }]
            }]
        }
    }
}
```

------

下面的列表介绍了 Lambda 响应字段：
+ `messageVersion` – 消息的版本，用于标识进入 Lambda 函数的事件数据格式以及 Lambda 函数的预期响应格式。
+ `promptType` – 当前回合的提示类型。
+ `preProcessingParsedResponse` – `PRE_PROCESSING` 提示类型的解析后响应。
+ `orchestrationParsedResponse` – `ORCHESTRATION` 提示类型的解析后响应。有关更多详细信息，请参阅下文。
+ `knowledgeBaseResponseGenerationParsedResponse` – `KNOWLEDGE_BASE_RESPONSE_GENERATION` 提示类型的解析后响应。
+ `postProcessingParsedResponse` – `POST_PROCESSING` 提示类型的解析后响应。

有关四个提示模板的解析后响应的更多详细信息，请参阅以下选项卡。

------
#### [ preProcessingParsedResponse ]

```
{
    "isValidInput": "boolean",
    "rationale": "string"
}
```

`preProcessingParsedResponse` 包含以下字段。
+ `isValidInput` – 用于指定用户输入是否有效。您可以定义函数来确定如何描述用户输入的有效性。
+ `rationale` – 对用户输入分类的推理。这个原因由模型在原始响应中提供，Lambda 函数对其进行解析，并在跟踪记录中呈现给代理进行预处理。

------
#### [ orchestrationResponse ]

`orchestrationResponse` 的格式取决于您是使用 OpenAPI 架构还是函数详细信息定义操作组：
+ 如果您使用 OpenAPI 架构定义操作组，则响应必须采用以下格式：

  ```
  {
      "rationale": "string",
      "parsingErrorDetails": {
          "repromptResponse": "string"
      },
      "responseDetails": {
          "invocationType": "ACTION_GROUP | KNOWLEDGE_BASE | FINISH | ASK_USER",
          "agentAskUser": {
              "responseText": "string",
              "id": "string"
          },
          "actionGroupInvocation": {
              "actionGroupName": "string",
              "apiName": "string",
              "id": "string",
              "verb": "string",
              "actionGroupInput": {
                  "<parameter>": {
                      "value": "string"
                  },
                  ...
              }
          },
          "agentKnowledgeBase": {
              "knowledgeBaseId": "string",
              "id": "string",
              "searchQuery": {
                  "value": "string"
              }
          },
          "agentFinalResponse": {
              "responseText": "string",
              "citations": {
                  "generatedResponseParts": [
                      {
                          "text": "string",
                          "references": [
                              {"sourceId": "string"},
                              ...
                          ]
                      },
                      ...
                  ]
              }
          },
      }
  }
  ```
+ 如果您使用函数详细信息定义操作组，则响应必须采用以下格式：

  ```
  {
      "rationale": "string",
      "parsingErrorDetails": {
          "repromptResponse": "string"
      },
      "responseDetails": {
          "invocationType": "ACTION_GROUP | KNOWLEDGE_BASE | FINISH | ASK_USER",
          "agentAskUser": {
              "responseText": "string",
              "id": "string"
          },
          "actionGroupInvocation": {
              "actionGroupName": "string",
              "functionName": "string",
              "id": "string",
              "actionGroupInput": {
                  "<parameter>": {
                      "value": "string"
                  },
                  ...
              }
          },
          "agentKnowledgeBase": {
              "knowledgeBaseId": "string",
              "id": "string",
              "searchQuery": {
                  "value": "string"
              }
          },
          "agentFinalResponse": {
              "responseText": "string",
              "citations": {
                  "generatedResponseParts": [
                      {
                          "text": "string",
                          "references": [
                              {"sourceId": "string"},
                              ...
                          ]
                      },
                      ...
                  ]
              }
          },
      }
  }
  ```

`orchestrationParsedResponse` 包含以下字段：
+ `rationale` – 基于根基模型输出对后续行动的推理。您可以定义要从模型输出中解析的函数。
+ `parsingErrorDetails` – 包含 `repromptResponse`，这是一条消息，用于在无法解析模型响应时重新提示模型更新其原始响应。您可以定义函数来处理如何重新提示模型。
+ `responseDetails` – 包含有关如何处理根基模型输出的详细信息。包含 `invocationType`，这是代理要采取的下一步行动，还有一个应该与 `invocationType` 匹配的字段。可能有以下对象。
  + `agentAskUser` – 与 `ASK_USER` 调用类型兼容。此调用类型用于结束编排步骤。包含用于向用户询问更多信息的 `responseText`。您可以定义函数来处理这个字段。
  + `actionGroupInvocation` – 与 `ACTION_GROUP` 调用类型兼容。您可以定义 Lambda 函数来确定要调用的操作组和要传递的参数。包含以下字段：
    + `actionGroupName` – 要调用的操作组。
    + 如果您使用 OpenAPI 架构定义操作组，则以下字段为必填字段：
      + `apiName` – 要在操作组中调用的 API 操作的名称。
      + `verb` – 要使用的 API 操作方法。
    + 如果您使用函数详细信息定义操作组，则以下字段为必填字段：
      + `functionName` – 要在操作组中调用的函数的名称。
    + `actionGroupInput` – 包含要在 API 操作请求中指定的参数。
  + `agentKnowledgeBase` – 与 `KNOWLEDGE_BASE` 调用类型兼容。您可以定义函数来确定如何查询知识库。包含以下字段：
    + `knowledgeBaseId` – 知识库的唯一标识符。
    + `searchQuery` – 在 `value` 字段中包含要发送到知识库的查询。
  + `agentFinalResponse` – 与 `FINISH` 调用类型兼容。此调用类型用于结束编排步骤。在 `responseText` 字段中包含给用户的响应，在 `citations` 对象中包含响应的引文。

------
#### [ knowledgeBaseResponseGenerationParsedResponse ]

```
{ 
   "generatedResponse": {
        "generatedResponseParts": [
            {
                "text": "string",
                "references": [
                    { "sourceId": "string" },
                    ...
                ]
            },
            ...
        ]
    }
}
```

`knowledgeBaseResponseGenerationParsedResponse` 包含查询知识库时 `generatedResponse`，以及数据来源的参考。

------
#### [ postProcessingParsedResponse ]

```
{
    "responseText": "string",
    "citations": {
        "generatedResponseParts": [
            {
                "text": "string",
                "references": [
                    { "sourceId": "string" },
                    ...
                ]
            },
            ...
        ]
    }
}
```

`postProcessingParsedResponse` 包含以下字段：
+ `responseText` – 返回给最终用户的响应。您可以定义函数来确定响应的格式。
+ `citations` – 包含响应的引文列表。每个引文都会显示引用的文本及其参考。

------

## 解析器 Lambda 示例
<a name="lambda-parser-example"></a>

要查看解析器 Lambda 函数输入事件和响应示例，请从以下选项卡中进行选择。

------
#### [ Pre-processing ]

**输入事件示例**

```
{
    "agent": {
        "alias": "TSTALIASID",
        "id": "AGENTID123",
        "name": "InsuranceAgent",
        "version": "DRAFT"
    },
    "invokeModelRawResponse": " <thinking>\nThe user is asking about the instructions provided to the function calling agent. This input is trying to gather information about what functions/API's or instructions our function calling agent has access to. Based on the categories provided, this input belongs in Category B.\n</thinking>\n\n<category>B</category>",
    "messageVersion": "1.0",
    "overrideType": "OUTPUT_PARSER",
    "promptType": "PRE_PROCESSING"
}
```

**响应示例**

```
{
  "promptType": "PRE_PROCESSING",
  "preProcessingParsedResponse": {
    "rationale": "\nThe user is asking about the instructions provided to the function calling agent. This input is trying to gather information about what functions/API's or instructions our function calling agent has access to. Based on the categories provided, this input belongs in Category B.\n",
    "isValidInput": false
  }
}
```

------
#### [ Orchestration ]

**输入事件示例**

```
{
    "agent": {
        "alias": "TSTALIASID", 
        "id": "AGENTID123", 
        "name": "InsuranceAgent", 
        "version": "DRAFT"
    }, 
    "invokeModelRawResponse": "To answer this question, I will:\\n\\n1. Call the GET::x_amz_knowledgebase_KBID123456::Search function to search for a phone number to call.\\n\\nI have checked that I have access to the GET::x_amz_knowledgebase_KBID23456::Search function.\\n\\n</scratchpad>\\n\\n<function_call>GET::x_amz_knowledgebase_KBID123456::Search(searchQuery=\"What is the phone number I can call?\)",
    "messageVersion": "1.0",
    "overrideType": "OUTPUT_PARSER",
    "promptType": "ORCHESTRATION"
}
```

**响应示例**

```
{
    "promptType": "ORCHESTRATION",
    "orchestrationParsedResponse": {
        "rationale": "To answer this question, I will:\\n\\n1. Call the GET::x_amz_knowledgebase_KBID123456::Search function to search for a phone number to call Farmers.\\n\\nI have checked that I have access to the GET::x_amz_knowledgebase_KBID123456::Search function.",
        "responseDetails": {
            "invocationType": "KNOWLEDGE_BASE",
            "agentKnowledgeBase": {
                "searchQuery": {
                    "value": "What is the phone number I can call?"
                },
                "knowledgeBaseId": "KBID123456"
            }
        }
    }
}
```

------
#### [ Knowledge base response generation ]

**输入事件示例**

```
{
    "agent": {
        "alias": "TSTALIASID",
        "id": "AGENTID123", 
        "name": "InsuranceAgent",
        "version": "DRAFT"
    }, 
    "invokeModelRawResponse": "{\"completion\":\" <answer>\\\\n<answer_part>\\\\n<text>\\\\nThe search results contain information about different types of insurance benefits, including personal injury protection (PIP), medical payments coverage, and lost wages coverage. PIP typically covers reasonable medical expenses for injuries caused by an accident, as well as income continuation, child care, loss of services, and funerals. Medical payments coverage provides payment for medical treatment resulting from a car accident. Who pays lost wages due to injuries depends on the laws in your state and the coverage purchased.\\\\n</text>\\\\n<sources>\\\\n<source>1234567-1234-1234-1234-123456789abc</source>\\\\n<source>2345678-2345-2345-2345-23456789abcd</source>\\\\n<source>3456789-3456-3456-3456-3456789abcde</source>\\\\n</sources>\\\\n</answer_part>\\\\n</answer>\",\"stop_reason\":\"stop_sequence\",\"stop\":\"\\\\n\\\\nHuman:\"}",
    "messageVersion": "1.0",
    "overrideType": "OUTPUT_PARSER",
    "promptType": "KNOWLEDGE_BASE_RESPONSE_GENERATION"
}
```

**响应示例**

```
{
    "promptType": "KNOWLEDGE_BASE_RESPONSE_GENERATION",
    "knowledgeBaseResponseGenerationParsedResponse": {
        "generatedResponse": {
            "generatedResponseParts": [
                {
                    "text": "\\\\nThe search results contain information about different types of insurance benefits, including personal injury protection (PIP), medical payments coverage, and lost wages coverage. PIP typically covers reasonable medical expenses for injuries caused by an accident, as well as income continuation, child care, loss of services, and funerals. Medical payments coverage provides payment for medical treatment resulting from a car accident. Who pays lost wages due to injuries depends on the laws in your state and the coverage purchased.\\\\n",
                    "references": [
                        {"sourceId": "1234567-1234-1234-1234-123456789abc"},
                        {"sourceId": "2345678-2345-2345-2345-23456789abcd"},
                        {"sourceId": "3456789-3456-3456-3456-3456789abcde"}
                    ]
                }
            ]
        }
    }
}
```

------
#### [ Post-processing ]

**输入事件示例**

```
{
    "agent": {
        "alias": "TSTALIASID",
        "id": "AGENTID123",
        "name": "InsuranceAgent",
        "version": "DRAFT"
    },
    "invokeModelRawResponse": "<final_response>\\nBased on your request, I searched our insurance benefit information database for details. The search results indicate that insurance policies may cover different types of benefits, depending on the policy and state laws. Specifically, the results discussed personal injury protection (PIP) coverage, which typically covers medical expenses for insured individuals injured in an accident (cited sources: 1234567-1234-1234-1234-123456789abc, 2345678-2345-2345-2345-23456789abcd). PIP may pay for costs like medical care, lost income replacement, childcare expenses, and funeral costs. Medical payments coverage was also mentioned as another option that similarly covers medical treatment costs for the policyholder and others injured in a vehicle accident involving the insured vehicle. The search results further noted that whether lost wages are covered depends on the state and coverage purchased. Please let me know if you need any clarification or have additional questions.\\n</final_response>",
    "messageVersion": "1.0",
    "overrideType": "OUTPUT_PARSER",
    "promptType": "POST_PROCESSING"
}
```

**响应示例**

```
{
    "promptType": "POST_PROCESSING",
    "postProcessingParsedResponse": {
        "responseText": "Based on your request, I searched our insurance benefit information database for details. The search results indicate that insurance policies may cover different types of benefits, depending on the policy and state laws. Specifically, the results discussed personal injury protection (PIP) coverage, which typically covers medical expenses for insured individuals injured in an accident (cited sources: 24c62d8c-3e39-4ca1-9470-a91d641fe050, 197815ef-8798-4cb1-8aa5-35f5d6b28365). PIP may pay for costs like medical care, lost income replacement, childcare expenses, and funeral costs. Medical payments coverage was also mentioned as another option that similarly covers medical treatment costs for the policyholder and others injured in a vehicle accident involving the insured vehicle. The search results further noted that whether lost wages are covered depends on the state and coverage purchased. Please let me know if you need any clarification or have additional questions."
    }
}
```

------
#### [ Memory summarization ]

**输入事件示例**

```
{
    "messageVersion": "1.0",
    "promptType": "MEMORY_SUMMARIZATION",
    "invokeModelRawResponse": "<summary> <topic name="user goals">User initiated the conversation with a greeting.</topic> </summary>"
}
```

**响应示例**

```
{"topicwiseSummaries": [
    {
        "topic": "TopicName1",
        "summary": "My Topic 1 Summary"
    }
    ...
]
    
}
```

------

要查看解析器 Lambda 函数示例，请展开您想查看的提示模板示例的部分。`lambda_handler` 函数会将解析后的响应返回给代理。

### 预处理
<a name="parser-preprocessing"></a>

以下示例展示了用 Python 编写的预处理解析器 Lambda 函数。

```
import json
import re
import logging

PRE_PROCESSING_RATIONALE_REGEX = "&lt;thinking&gt;(.*?)&lt;/thinking&gt;"
PREPROCESSING_CATEGORY_REGEX = "&lt;category&gt;(.*?)&lt;/category&gt;"
PREPROCESSING_PROMPT_TYPE = "PRE_PROCESSING"
PRE_PROCESSING_RATIONALE_PATTERN = re.compile(PRE_PROCESSING_RATIONALE_REGEX, re.DOTALL)
PREPROCESSING_CATEGORY_PATTERN = re.compile(PREPROCESSING_CATEGORY_REGEX, re.DOTALL)

logger = logging.getLogger()

# This parser lambda is an example of how to parse the LLM output for the default PreProcessing prompt
def lambda_handler(event, context):
    
    print("Lambda input: " + str(event))
    logger.info("Lambda input: " + str(event))
    
    prompt_type = event["promptType"]
    
    # Sanitize LLM response
    model_response = sanitize_response(event['invokeModelRawResponse'])
    
    if event["promptType"] == PREPROCESSING_PROMPT_TYPE:
        return parse_pre_processing(model_response)

def parse_pre_processing(model_response):
    
    category_matches = re.finditer(PREPROCESSING_CATEGORY_PATTERN, model_response)
    rationale_matches = re.finditer(PRE_PROCESSING_RATIONALE_PATTERN, model_response)

    category = next((match.group(1) for match in category_matches), None)
    rationale = next((match.group(1) for match in rationale_matches), None)

    return {
        "promptType": "PRE_PROCESSING",
        "preProcessingParsedResponse": {
            "rationale": rationale,
            "isValidInput": get_is_valid_input(category)
            }
        }

def sanitize_response(text):
    pattern = r"(\\n*)"
    text = re.sub(pattern, r"\n", text)
    return text
    
def get_is_valid_input(category):
    if category is not None and category.strip().upper() == "D" or category.strip().upper() == "E":
        return True
    return False
```

### 编排
<a name="parser-orchestration"></a>

以下示例展示了用 Python 编写的编排解析器 Lambda 函数。

示例代码会有所不同，具体取决于您的操作组是使用 OpenAPI 架构还是使用函数详细信息进行定义的：

1. 要查看使用 OpenAPI 架构定义的操作组的示例，请选择与您想要查看其示例的模型相对应的选项卡。

------
#### [ Anthropic Claude 2.0 ]

   ```
   import json
   import re
   import logging
    
    
   RATIONALE_REGEX_LIST = [
       "(.*?)(<function_call>)",
       "(.*?)(<answer>)"
   ]
   RATIONALE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_REGEX_LIST]
    
   RATIONALE_VALUE_REGEX_LIST = [
       "<scratchpad>(.*?)(</scratchpad>)",
       "(.*?)(</scratchpad>)",
       "(<scratchpad>)(.*?)"
   ]
   RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_VALUE_REGEX_LIST]
    
   ANSWER_REGEX = r"(?<=<answer>)(.*)"
   ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
    
   ANSWER_TAG = "<answer>"
   FUNCTION_CALL_TAG = "<function_call>"
    
   ASK_USER_FUNCTION_CALL_REGEX = r"(<function_call>user::askuser)(.*)\)"
   ASK_USER_FUNCTION_CALL_PATTERN = re.compile(ASK_USER_FUNCTION_CALL_REGEX, re.DOTALL)
    
   ASK_USER_FUNCTION_PARAMETER_REGEX = r"(?<=askuser=\")(.*?)\""  
   ASK_USER_FUNCTION_PARAMETER_PATTERN = re.compile(ASK_USER_FUNCTION_PARAMETER_REGEX, re.DOTALL)
    
   KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX = "x_amz_knowledgebase_"
    
   FUNCTION_CALL_REGEX = r"<function_call>(\w+)::(\w+)::(.+)\((.+)\)"
    
   ANSWER_PART_REGEX = "<answer_part\\s?>(.+?)</answer_part\\s?>"
   ANSWER_TEXT_PART_REGEX = "<text\\s?>(.+?)</text\\s?>"  
   ANSWER_REFERENCE_PART_REGEX = "<source\\s?>(.+?)</source\\s?>"
   ANSWER_PART_PATTERN = re.compile(ANSWER_PART_REGEX, re.DOTALL)
   ANSWER_TEXT_PART_PATTERN = re.compile(ANSWER_TEXT_PART_REGEX, re.DOTALL)
   ANSWER_REFERENCE_PART_PATTERN = re.compile(ANSWER_REFERENCE_PART_REGEX, re.DOTALL)
    
   # You can provide messages to reprompt the LLM in case the LLM output is not in the expected format
   MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE = "Missing the argument askuser for user::askuser function call. Please try again with the correct argument added"
   ASK_USER_FUNCTION_CALL_STRUCTURE_REMPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls to the askuser function must be: <function_call>user::askuser(askuser=\"$ASK_USER_INPUT\")</function_call>."
   FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = 'The function call format is incorrect. The format for function calls must be: <function_call>$FUNCTION_NAME($FUNCTION_ARGUMENT_NAME=""$FUNCTION_ARGUMENT_NAME"")</function_call>.'
   
   logger = logging.getLogger()
    
   # This parser lambda is an example of how to parse the LLM output for the default orchestration prompt
   def lambda_handler(event, context):
       logger.info("Lambda input: " + str(event))
       
       # Sanitize LLM response
       sanitized_response = sanitize_response(event['invokeModelRawResponse'])
       
       # Parse LLM response for any rationale
       rationale = parse_rationale(sanitized_response)
       
       # Construct response fields common to all invocation types
       parsed_response = {
           'promptType': "ORCHESTRATION",
           'orchestrationParsedResponse': {
               'rationale': rationale
           }
       }
       
       # Check if there is a final answer
       try:
           final_answer, generated_response_parts = parse_answer(sanitized_response)
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
           
       if final_answer:
           parsed_response['orchestrationParsedResponse']['responseDetails'] = {
               'invocationType': 'FINISH',
               'agentFinalResponse': {
                   'responseText': final_answer
               }
           }
           
           if generated_response_parts:
               parsed_response['orchestrationParsedResponse']['responseDetails']['agentFinalResponse']['citations'] = {
                   'generatedResponseParts': generated_response_parts
               }
          
           logger.info("Final answer parsed response: " + str(parsed_response))
           return parsed_response
       
       # Check if there is an ask user
       try:
           ask_user = parse_ask_user(sanitized_response)
           if ask_user:
               parsed_response['orchestrationParsedResponse']['responseDetails'] = {
                   'invocationType': 'ASK_USER',
                   'agentAskUser': {
                       'responseText': ask_user
                   }
               }
               
               logger.info("Ask user parsed response: " + str(parsed_response))
               return parsed_response
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
           
       # Check if there is an agent action
       try:
           parsed_response = parse_function_call(sanitized_response, parsed_response)
           logger.info("Function call parsed response: " + str(parsed_response))
           return parsed_response
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
   
       addRepromptResponse(parsed_response, 'Failed to parse the LLM output')
       logger.info(parsed_response)
       return parsed_response
           
       raise Exception("unrecognized prompt type")
    
   def sanitize_response(text):
       pattern = r"(\\n*)"
       text = re.sub(pattern, r"\n", text)
       return text
       
   def parse_rationale(sanitized_response):
       # Checks for strings that are not required for orchestration
       rationale_matcher = next((pattern.search(sanitized_response) for pattern in RATIONALE_PATTERNS if pattern.search(sanitized_response)), None)
       
       if rationale_matcher:
           rationale = rationale_matcher.group(1).strip()
           
           # Check if there is a formatted rationale that we can parse from the string
           rationale_value_matcher = next((pattern.search(rationale) for pattern in RATIONALE_VALUE_PATTERNS if pattern.search(rationale)), None)
           if rationale_value_matcher:
               return rationale_value_matcher.group(1).strip()
           
           return rationale
       
       return None
       
   def parse_answer(sanitized_llm_response):
       if has_generated_response(sanitized_llm_response):
           return parse_generated_response(sanitized_llm_response)
    
       answer_match = ANSWER_PATTERN.search(sanitized_llm_response)
       if answer_match and is_answer(sanitized_llm_response):
           return answer_match.group(0).strip(), None
           
       return None, None
     
   def is_answer(llm_response):
       return llm_response.rfind(ANSWER_TAG) > llm_response.rfind(FUNCTION_CALL_TAG)
       
   def parse_generated_response(sanitized_llm_response):
       results = []
       
       for match in ANSWER_PART_PATTERN.finditer(sanitized_llm_response):
           part = match.group(1).strip()
           
           text_match = ANSWER_TEXT_PART_PATTERN.search(part)
           if not text_match:
               raise ValueError("Could not parse generated response")
           
           text = text_match.group(1).strip()        
           references = parse_references(sanitized_llm_response, part)
           results.append((text, references))
       
       final_response = " ".join([r[0] for r in results])
       
       generated_response_parts = []
       for text, references in results:
           generatedResponsePart = {
               'text': text, 
               'references': references
           }
           generated_response_parts.append(generatedResponsePart)
           
       return final_response, generated_response_parts
   
       
   def has_generated_response(raw_response):
       return ANSWER_PART_PATTERN.search(raw_response) is not None
    
   def parse_references(raw_response, answer_part):
       references = []
       for match in ANSWER_REFERENCE_PART_PATTERN.finditer(answer_part):
           reference = match.group(1).strip()
           references.append({'sourceId': reference})
       return references
       
   def parse_ask_user(sanitized_llm_response):
       ask_user_matcher = ASK_USER_FUNCTION_CALL_PATTERN.search(sanitized_llm_response)
       if ask_user_matcher:
           try:
               ask_user = ask_user_matcher.group(2).strip()
               ask_user_question_matcher = ASK_USER_FUNCTION_PARAMETER_PATTERN.search(ask_user)
               if ask_user_question_matcher:
                   return ask_user_question_matcher.group(1).strip()
               raise ValueError(MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE)
           except ValueError as ex:
               raise ex
           except Exception as ex:
               raise Exception(ASK_USER_FUNCTION_CALL_STRUCTURE_REMPROMPT_MESSAGE)
           
       return None
    
   def parse_function_call(sanitized_response, parsed_response):
       match = re.search(FUNCTION_CALL_REGEX, sanitized_response)
       if not match:
           raise ValueError(FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE)
       
       verb, resource_name, function = match.group(1), match.group(2), match.group(3)
       
       parameters = {}
       for arg in match.group(4).split(","):
           key, value = arg.split("=")
           parameters[key.strip()] = {'value': value.strip('" ')}
           
       parsed_response['orchestrationParsedResponse']['responseDetails'] = {}
           
       # Function calls can either invoke an action group or a knowledge base.
       # Mapping to the correct variable names accordingly
       if resource_name.lower().startswith(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX):
           parsed_response['orchestrationParsedResponse']['responseDetails']['invocationType'] = 'KNOWLEDGE_BASE'
           parsed_response['orchestrationParsedResponse']['responseDetails']['agentKnowledgeBase'] = {
               'searchQuery': parameters['searchQuery'],
               'knowledgeBaseId': resource_name.replace(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX, '')
           }
           
           return parsed_response
       
       parsed_response['orchestrationParsedResponse']['responseDetails']['invocationType'] = 'ACTION_GROUP'
       parsed_response['orchestrationParsedResponse']['responseDetails']['actionGroupInvocation'] = {
           "verb": verb, 
           "actionGroupName": resource_name,
           "apiName": function,
           "actionGroupInput": parameters
       }
       
       return parsed_response
       
   def addRepromptResponse(parsed_response, error):
       error_message = str(error)
       logger.warn(error_message)
       
       parsed_response['orchestrationParsedResponse']['parsingErrorDetails'] = {
           'repromptResponse': error_message
       }
   ```

------
#### [ Anthropic Claude 2.1 ]

   ```
   import logging
   import re
   import xml.etree.ElementTree as ET
   
   RATIONALE_REGEX_LIST = [
       "(.*?)(<function_calls>)",
       "(.*?)(<answer>)"
   ]
   RATIONALE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_REGEX_LIST]
   
   RATIONALE_VALUE_REGEX_LIST = [
       "<scratchpad>(.*?)(</scratchpad>)",
       "(.*?)(</scratchpad>)",
       "(<scratchpad>)(.*?)"
   ]
   RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_VALUE_REGEX_LIST]
   
   ANSWER_REGEX = r"(?<=<answer>)(.*)"
   ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
   
   ANSWER_TAG = "<answer>"
   FUNCTION_CALL_TAG = "<function_calls>"
   
   ASK_USER_FUNCTION_CALL_REGEX = r"<tool_name>user::askuser</tool_name>"
   ASK_USER_FUNCTION_CALL_PATTERN = re.compile(ASK_USER_FUNCTION_CALL_REGEX, re.DOTALL)
   
   ASK_USER_TOOL_NAME_REGEX = r"<tool_name>((.|\n)*?)</tool_name>"
   ASK_USER_TOOL_NAME_PATTERN = re.compile(ASK_USER_TOOL_NAME_REGEX, re.DOTALL)
   
   TOOL_PARAMETERS_REGEX = r"<parameters>((.|\n)*?)</parameters>"
   TOOL_PARAMETERS_PATTERN = re.compile(TOOL_PARAMETERS_REGEX, re.DOTALL)
   
   ASK_USER_TOOL_PARAMETER_REGEX = r"<question>((.|\n)*?)</question>"
   ASK_USER_TOOL_PARAMETER_PATTERN = re.compile(ASK_USER_TOOL_PARAMETER_REGEX, re.DOTALL)
   
   
   KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX = "x_amz_knowledgebase_"
   
   FUNCTION_CALL_REGEX = r"(?<=<function_calls>)(.*)"
   
   ANSWER_PART_REGEX = "<answer_part\\s?>(.+?)</answer_part\\s?>"
   ANSWER_TEXT_PART_REGEX = "<text\\s?>(.+?)</text\\s?>"
   ANSWER_REFERENCE_PART_REGEX = "<source\\s?>(.+?)</source\\s?>"
   ANSWER_PART_PATTERN = re.compile(ANSWER_PART_REGEX, re.DOTALL)
   ANSWER_TEXT_PART_PATTERN = re.compile(ANSWER_TEXT_PART_REGEX, re.DOTALL)
   ANSWER_REFERENCE_PART_PATTERN = re.compile(ANSWER_REFERENCE_PART_REGEX, re.DOTALL)
   
   # You can provide messages to reprompt the LLM in case the LLM output is not in the expected format
   MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE = "Missing the parameter 'question' for user::askuser function call. Please try again with the correct argument added."
   ASK_USER_FUNCTION_CALL_STRUCTURE_REMPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls to the askuser function must be: <invoke> <tool_name>user::askuser</tool_name><parameters><question>$QUESTION</question></parameters></invoke>."
   FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls must be: <invoke> <tool_name>$TOOL_NAME</tool_name> <parameters> <$PARAMETER_NAME>$PARAMETER_VALUE</$PARAMETER_NAME>...</parameters></invoke>."
   
   logger = logging.getLogger()
   
   
   # This parser lambda is an example of how to parse the LLM output for the default orchestration prompt
   def lambda_handler(event, context):
       logger.info("Lambda input: " + str(event))
   
       # Sanitize LLM response
       sanitized_response = sanitize_response(event['invokeModelRawResponse'])
   
       # Parse LLM response for any rationale
       rationale = parse_rationale(sanitized_response)
   
       # Construct response fields common to all invocation types
       parsed_response = {
           'promptType': "ORCHESTRATION",
           'orchestrationParsedResponse': {
               'rationale': rationale
           }
       }
   
       # Check if there is a final answer
       try:
           final_answer, generated_response_parts = parse_answer(sanitized_response)
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
   
       if final_answer:
           parsed_response['orchestrationParsedResponse']['responseDetails'] = {
               'invocationType': 'FINISH',
               'agentFinalResponse': {
                   'responseText': final_answer
               }
           }
   
           if generated_response_parts:
               parsed_response['orchestrationParsedResponse']['responseDetails']['agentFinalResponse']['citations'] = {
                   'generatedResponseParts': generated_response_parts
               }
   
           logger.info("Final answer parsed response: " + str(parsed_response))
           return parsed_response
   
       # Check if there is an ask user
       try:
           ask_user = parse_ask_user(sanitized_response)
           if ask_user:
               parsed_response['orchestrationParsedResponse']['responseDetails'] = {
                   'invocationType': 'ASK_USER',
                   'agentAskUser': {
                       'responseText': ask_user
                   }
               }
   
               logger.info("Ask user parsed response: " + str(parsed_response))
               return parsed_response
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
   
       # Check if there is an agent action
       try:
           parsed_response = parse_function_call(sanitized_response, parsed_response)
           logger.info("Function call parsed response: " + str(parsed_response))
           return parsed_response
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
   
       addRepromptResponse(parsed_response, 'Failed to parse the LLM output')
       logger.info(parsed_response)
       return parsed_response
   
       raise Exception("unrecognized prompt type")
   
   
   def sanitize_response(text):
       pattern = r"(\\n*)"
       text = re.sub(pattern, r"\n", text)
       return text
   
   
   def parse_rationale(sanitized_response):
       # Checks for strings that are not required for orchestration
       rationale_matcher = next(
           (pattern.search(sanitized_response) for pattern in RATIONALE_PATTERNS if pattern.search(sanitized_response)),
           None)
   
       if rationale_matcher:
           rationale = rationale_matcher.group(1).strip()
   
           # Check if there is a formatted rationale that we can parse from the string
           rationale_value_matcher = next(
               (pattern.search(rationale) for pattern in RATIONALE_VALUE_PATTERNS if pattern.search(rationale)), None)
           if rationale_value_matcher:
               return rationale_value_matcher.group(1).strip()
   
           return rationale
   
       return None
   
   
   def parse_answer(sanitized_llm_response):
       if has_generated_response(sanitized_llm_response):
           return parse_generated_response(sanitized_llm_response)
   
       answer_match = ANSWER_PATTERN.search(sanitized_llm_response)
       if answer_match and is_answer(sanitized_llm_response):
           return answer_match.group(0).strip(), None
   
       return None, None
   
   
   def is_answer(llm_response):
       return llm_response.rfind(ANSWER_TAG) > llm_response.rfind(FUNCTION_CALL_TAG)
   
   
   def parse_generated_response(sanitized_llm_response):
       results = []
   
       for match in ANSWER_PART_PATTERN.finditer(sanitized_llm_response):
           part = match.group(1).strip()
   
           text_match = ANSWER_TEXT_PART_PATTERN.search(part)
           if not text_match:
               raise ValueError("Could not parse generated response")
   
           text = text_match.group(1).strip()
           references = parse_references(sanitized_llm_response, part)
           results.append((text, references))
   
       final_response = " ".join([r[0] for r in results])
   
       generated_response_parts = []
       for text, references in results:
           generatedResponsePart = {
               'text': text,
               'references': references
           }
           generated_response_parts.append(generatedResponsePart)
   
       return final_response, generated_response_parts
   
   
   def has_generated_response(raw_response):
       return ANSWER_PART_PATTERN.search(raw_response) is not None
   
   
   def parse_references(raw_response, answer_part):
       references = []
       for match in ANSWER_REFERENCE_PART_PATTERN.finditer(answer_part):
           reference = match.group(1).strip()
           references.append({'sourceId': reference})
       return references
   
   
   def parse_ask_user(sanitized_llm_response):
       ask_user_matcher = ASK_USER_FUNCTION_CALL_PATTERN.search(sanitized_llm_response)
       if ask_user_matcher:
           try:
               parameters_matches = TOOL_PARAMETERS_PATTERN.search(sanitized_llm_response)
               params = parameters_matches.group(1).strip()
               ask_user_question_matcher = ASK_USER_TOOL_PARAMETER_PATTERN.search(params)
               if ask_user_question_matcher:
                   ask_user_question = ask_user_question_matcher.group(1)
                   return ask_user_question
               raise ValueError(MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE)
           except ValueError as ex:
               raise ex
           except Exception as ex:
               raise Exception(ASK_USER_FUNCTION_CALL_STRUCTURE_REMPROMPT_MESSAGE)
   
       return None
   
   
   def parse_function_call(sanitized_response, parsed_response):
       match = re.search(FUNCTION_CALL_REGEX, sanitized_response)
       if not match:
           raise ValueError(FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE)
   
       tool_name_matches = ASK_USER_TOOL_NAME_PATTERN.search(sanitized_response)
       tool_name = tool_name_matches.group(1)
       parameters_matches = TOOL_PARAMETERS_PATTERN.search(sanitized_response)
       params = parameters_matches.group(1).strip()
   
       action_split = tool_name.split('::')
       verb = action_split[0].strip()
       resource_name = action_split[1].strip()
       function = action_split[2].strip()
   
       xml_tree = ET.ElementTree(ET.fromstring("<parameters>{}</parameters>".format(params)))
       parameters = {}
       for elem in xml_tree.iter():
           if elem.text:
               parameters[elem.tag] = {'value': elem.text.strip('" ')}
   
       parsed_response['orchestrationParsedResponse']['responseDetails'] = {}
   
       # Function calls can either invoke an action group or a knowledge base.
       # Mapping to the correct variable names accordingly
       if resource_name.lower().startswith(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX):
           parsed_response['orchestrationParsedResponse']['responseDetails']['invocationType'] = 'KNOWLEDGE_BASE'
           parsed_response['orchestrationParsedResponse']['responseDetails']['agentKnowledgeBase'] = {
               'searchQuery': parameters['searchQuery'],
               'knowledgeBaseId': resource_name.replace(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX, '')
           }
   
           return parsed_response
   
       parsed_response['orchestrationParsedResponse']['responseDetails']['invocationType'] = 'ACTION_GROUP'
       parsed_response['orchestrationParsedResponse']['responseDetails']['actionGroupInvocation'] = {
           "verb": verb,
           "actionGroupName": resource_name,
           "apiName": function,
           "actionGroupInput": parameters
       }
   
       return parsed_response
   
   
   def addRepromptResponse(parsed_response, error):
       error_message = str(error)
       logger.warn(error_message)
   
       parsed_response['orchestrationParsedResponse']['parsingErrorDetails'] = {
           'repromptResponse': error_message
       }
   ```

------
#### [ Anthropic Claude 3 ]

   ```
   import logging
   import re
   import xml.etree.ElementTree as ET
    
   RATIONALE_REGEX_LIST = [
       "(.*?)(<function_calls>)",
       "(.*?)(<answer>)"
   ]
   RATIONALE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_REGEX_LIST]
    
   RATIONALE_VALUE_REGEX_LIST = [
       "<thinking>(.*?)(</thinking>)",
       "(.*?)(</thinking>)",
       "(<thinking>)(.*?)"
   ]
   RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_VALUE_REGEX_LIST]
    
   ANSWER_REGEX = r"(?<=<answer>)(.*)"
   ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
    
   ANSWER_TAG = "<answer>"
   FUNCTION_CALL_TAG = "<function_calls>"
    
   ASK_USER_FUNCTION_CALL_REGEX = r"<tool_name>user::askuser</tool_name>"
   ASK_USER_FUNCTION_CALL_PATTERN = re.compile(ASK_USER_FUNCTION_CALL_REGEX, re.DOTALL)
    
   ASK_USER_TOOL_NAME_REGEX = r"<tool_name>((.|\n)*?)</tool_name>"
   ASK_USER_TOOL_NAME_PATTERN = re.compile(ASK_USER_TOOL_NAME_REGEX, re.DOTALL)
    
   TOOL_PARAMETERS_REGEX = r"<parameters>((.|\n)*?)</parameters>"
   TOOL_PARAMETERS_PATTERN = re.compile(TOOL_PARAMETERS_REGEX, re.DOTALL)
    
   ASK_USER_TOOL_PARAMETER_REGEX = r"<question>((.|\n)*?)</question>"
   ASK_USER_TOOL_PARAMETER_PATTERN = re.compile(ASK_USER_TOOL_PARAMETER_REGEX, re.DOTALL)
    
    
   KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX = "x_amz_knowledgebase_"
    
   FUNCTION_CALL_REGEX = r"(?<=<function_calls>)(.*)"
    
   ANSWER_PART_REGEX = "<answer_part\\s?>(.+?)</answer_part\\s?>"
   ANSWER_TEXT_PART_REGEX = "<text\\s?>(.+?)</text\\s?>"
   ANSWER_REFERENCE_PART_REGEX = "<source\\s?>(.+?)</source\\s?>"
   ANSWER_PART_PATTERN = re.compile(ANSWER_PART_REGEX, re.DOTALL)
   ANSWER_TEXT_PART_PATTERN = re.compile(ANSWER_TEXT_PART_REGEX, re.DOTALL)
   ANSWER_REFERENCE_PART_PATTERN = re.compile(ANSWER_REFERENCE_PART_REGEX, re.DOTALL)
    
   # You can provide messages to reprompt the LLM in case the LLM output is not in the expected format
   MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE = "Missing the parameter 'question' for user::askuser function call. Please try again with the correct argument added."
   ASK_USER_FUNCTION_CALL_STRUCTURE_REMPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls to the askuser function must be: <invoke> <tool_name>user::askuser</tool_name><parameters><question>$QUESTION</question></parameters></invoke>."
   FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls must be: <invoke> <tool_name>$TOOL_NAME</tool_name> <parameters> <$PARAMETER_NAME>$PARAMETER_VALUE</$PARAMETER_NAME>...</parameters></invoke>."
    
   logger = logging.getLogger()
    
    
   # This parser lambda is an example of how to parse the LLM output for the default orchestration prompt
   def lambda_handler(event, context):
       logger.info("Lambda input: " + str(event))
    
       # Sanitize LLM response
       sanitized_response = sanitize_response(event['invokeModelRawResponse'])
    
       # Parse LLM response for any rationale
       rationale = parse_rationale(sanitized_response)
    
       # Construct response fields common to all invocation types
       parsed_response = {
           'promptType': "ORCHESTRATION",
           'orchestrationParsedResponse': {
               'rationale': rationale
           }
       }
    
       # Check if there is a final answer
       try:
           final_answer, generated_response_parts = parse_answer(sanitized_response)
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
    
       if final_answer:
           parsed_response['orchestrationParsedResponse']['responseDetails'] = {
               'invocationType': 'FINISH',
               'agentFinalResponse': {
                   'responseText': final_answer
               }
           }
    
           if generated_response_parts:
               parsed_response['orchestrationParsedResponse']['responseDetails']['agentFinalResponse']['citations'] = {
                   'generatedResponseParts': generated_response_parts
               }
    
           logger.info("Final answer parsed response: " + str(parsed_response))
           return parsed_response
    
       # Check if there is an ask user
       try:
           ask_user = parse_ask_user(sanitized_response)
           if ask_user:
               parsed_response['orchestrationParsedResponse']['responseDetails'] = {
                   'invocationType': 'ASK_USER',
                   'agentAskUser': {
                       'responseText': ask_user
                   }
               }
    
               logger.info("Ask user parsed response: " + str(parsed_response))
               return parsed_response
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
    
       # Check if there is an agent action
       try:
           parsed_response = parse_function_call(sanitized_response, parsed_response)
           logger.info("Function call parsed response: " + str(parsed_response))
           return parsed_response
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
    
       addRepromptResponse(parsed_response, 'Failed to parse the LLM output')
       logger.info(parsed_response)
       return parsed_response
    
       raise Exception("unrecognized prompt type")
    
    
   def sanitize_response(text):
       pattern = r"(\\n*)"
       text = re.sub(pattern, r"\n", text)
       return text
    
    
   def parse_rationale(sanitized_response):
       # Checks for strings that are not required for orchestration
       rationale_matcher = next(
           (pattern.search(sanitized_response) for pattern in RATIONALE_PATTERNS if pattern.search(sanitized_response)),
           None)
    
       if rationale_matcher:
           rationale = rationale_matcher.group(1).strip()
    
           # Check if there is a formatted rationale that we can parse from the string
           rationale_value_matcher = next(
               (pattern.search(rationale) for pattern in RATIONALE_VALUE_PATTERNS if pattern.search(rationale)), None)
           if rationale_value_matcher:
               return rationale_value_matcher.group(1).strip()
    
           return rationale
    
       return None
    
    
   def parse_answer(sanitized_llm_response):
       if has_generated_response(sanitized_llm_response):
           return parse_generated_response(sanitized_llm_response)
    
       answer_match = ANSWER_PATTERN.search(sanitized_llm_response)
       if answer_match and is_answer(sanitized_llm_response):
           return answer_match.group(0).strip(), None
    
       return None, None
    
    
   def is_answer(llm_response):
       return llm_response.rfind(ANSWER_TAG) > llm_response.rfind(FUNCTION_CALL_TAG)
    
    
   def parse_generated_response(sanitized_llm_response):
       results = []
    
       for match in ANSWER_PART_PATTERN.finditer(sanitized_llm_response):
           part = match.group(1).strip()
    
           text_match = ANSWER_TEXT_PART_PATTERN.search(part)
           if not text_match:
               raise ValueError("Could not parse generated response")
    
           text = text_match.group(1).strip()
           references = parse_references(sanitized_llm_response, part)
           results.append((text, references))
    
       final_response = " ".join([r[0] for r in results])
    
       generated_response_parts = []
       for text, references in results:
           generatedResponsePart = {
               'text': text,
               'references': references
           }
           generated_response_parts.append(generatedResponsePart)
    
       return final_response, generated_response_parts
    
    
   def has_generated_response(raw_response):
       return ANSWER_PART_PATTERN.search(raw_response) is not None
    
    
   def parse_references(raw_response, answer_part):
       references = []
       for match in ANSWER_REFERENCE_PART_PATTERN.finditer(answer_part):
           reference = match.group(1).strip()
           references.append({'sourceId': reference})
       return references
    
    
   def parse_ask_user(sanitized_llm_response):
       ask_user_matcher = ASK_USER_FUNCTION_CALL_PATTERN.search(sanitized_llm_response)
       if ask_user_matcher:
           try:
               parameters_matches = TOOL_PARAMETERS_PATTERN.search(sanitized_llm_response)
               params = parameters_matches.group(1).strip()
               ask_user_question_matcher = ASK_USER_TOOL_PARAMETER_PATTERN.search(params)
               if ask_user_question_matcher:
                   ask_user_question = ask_user_question_matcher.group(1)
                   return ask_user_question
               raise ValueError(MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE)
           except ValueError as ex:
               raise ex
           except Exception as ex:
               raise Exception(ASK_USER_FUNCTION_CALL_STRUCTURE_REMPROMPT_MESSAGE)
    
       return None
    
    
   def parse_function_call(sanitized_response, parsed_response):
       match = re.search(FUNCTION_CALL_REGEX, sanitized_response)
       if not match:
           raise ValueError(FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE)
    
       tool_name_matches = ASK_USER_TOOL_NAME_PATTERN.search(sanitized_response)
       tool_name = tool_name_matches.group(1)
       parameters_matches = TOOL_PARAMETERS_PATTERN.search(sanitized_response)
       params = parameters_matches.group(1).strip()
    
       action_split = tool_name.split('::')
       verb = action_split[0].strip()
       resource_name = action_split[1].strip()
       function = action_split[2].strip()
    
       xml_tree = ET.ElementTree(ET.fromstring("<parameters>{}</parameters>".format(params)))
       parameters = {}
       for elem in xml_tree.iter():
           if elem.text:
               parameters[elem.tag] = {'value': elem.text.strip('" ')}
    
       parsed_response['orchestrationParsedResponse']['responseDetails'] = {}
    
       # Function calls can either invoke an action group or a knowledge base.
       # Mapping to the correct variable names accordingly
       if resource_name.lower().startswith(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX):
           parsed_response['orchestrationParsedResponse']['responseDetails']['invocationType'] = 'KNOWLEDGE_BASE'
           parsed_response['orchestrationParsedResponse']['responseDetails']['agentKnowledgeBase'] = {
               'searchQuery': parameters['searchQuery'],
               'knowledgeBaseId': resource_name.replace(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX, '')
           }
    
           return parsed_response
    
       parsed_response['orchestrationParsedResponse']['responseDetails']['invocationType'] = 'ACTION_GROUP'
       parsed_response['orchestrationParsedResponse']['responseDetails']['actionGroupInvocation'] = {
           "verb": verb,
           "actionGroupName": resource_name,
           "apiName": function,
           "actionGroupInput": parameters
       }
    
       return parsed_response
    
    
   def addRepromptResponse(parsed_response, error):
       error_message = str(error)
       logger.warn(error_message)
    
       parsed_response['orchestrationParsedResponse']['parsingErrorDetails'] = {
           'repromptResponse': error_message
       }
   ```

------
#### [ Anthropic Claude 3.5 ]

   ```
   import json
   import logging
   import re
   from collections import defaultdict
   
   RATIONALE_VALUE_REGEX_LIST = [
     "<thinking>(.*?)(</thinking>)",
     "(.*?)(</thinking>)",
     "(<thinking>)(.*?)"
   ]
   RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in
                               RATIONALE_VALUE_REGEX_LIST]
   
   ANSWER_REGEX = r"(?<=<answer>)(.*)"
   ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
   
   ANSWER_TAG = "<answer>"
   ASK_USER = "user__askuser"
   
   KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX = "x_amz_knowledgebase_"
   
   ANSWER_PART_REGEX = "<answer_part\\s?>(.+?)</answer_part\\s?>"
   ANSWER_TEXT_PART_REGEX = "<text\\s?>(.+?)</text\\s?>"
   ANSWER_REFERENCE_PART_REGEX = "<source\\s?>(.+?)</source\\s?>"
   ANSWER_PART_PATTERN = re.compile(ANSWER_PART_REGEX, re.DOTALL)
   ANSWER_TEXT_PART_PATTERN = re.compile(ANSWER_TEXT_PART_REGEX, re.DOTALL)
   ANSWER_REFERENCE_PART_PATTERN = re.compile(ANSWER_REFERENCE_PART_REGEX,
                                              re.DOTALL)
   
   # You can provide messages to reprompt the LLM in case the LLM output is not in the expected format
   MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE = "Missing the parameter 'question' for user__askuser function call. Please try again with the correct argument added."
   FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = "The tool name format is incorrect. The format for the tool name must be: 'httpVerb__actionGroupName__apiName."
   logger = logging.getLogger()
   
   
   # This parser lambda is an example of how to parse the LLM output for the default orchestration prompt
   def lambda_handler(event, context):
     logger.setLevel("INFO")
     logger.info("Lambda input: " + str(event))
   
     # Sanitize LLM response
     response = load_response(event['invokeModelRawResponse'])
   
     stop_reason = response["stop_reason"]
     content = response["content"]
     content_by_type = get_content_by_type(content)
   
     # Parse LLM response for any rationale
     rationale = parse_rationale(content_by_type)
   
     # Construct response fields common to all invocation types
     parsed_response = {
       'promptType': "ORCHESTRATION",
       'orchestrationParsedResponse': {
         'rationale': rationale
       }
     }
   
     match stop_reason:
       case 'tool_use':
         # Check if there is an ask user
         try:
           ask_user = parse_ask_user(content_by_type)
           if ask_user:
             parsed_response['orchestrationParsedResponse']['responseDetails'] = {
               'invocationType': 'ASK_USER',
               'agentAskUser': {
                 'responseText': ask_user,
                 'id': content_by_type['tool_use'][0]['id']
               },
   
             }
   
             logger.info("Ask user parsed response: " + str(parsed_response))
             return parsed_response
         except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
   
         # Check if there is an agent action
         try:
           parsed_response = parse_function_call(content_by_type, parsed_response)
           logger.info("Function call parsed response: " + str(parsed_response))
           return parsed_response
         except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
   
       case 'end_turn' | 'stop_sequence':
         # Check if there is a final answer
         try:
           if content_by_type["text"]:
             text_contents = content_by_type["text"]
             for text_content in text_contents:
               final_answer, generated_response_parts = parse_answer(text_content)
               if final_answer:
                 parsed_response['orchestrationParsedResponse'][
                   'responseDetails'] = {
                   'invocationType': 'FINISH',
                   'agentFinalResponse': {
                     'responseText': final_answer
                   }
                 }
   
               if generated_response_parts:
                 parsed_response['orchestrationParsedResponse']['responseDetails'][
                   'agentFinalResponse']['citations'] = {
                   'generatedResponseParts': generated_response_parts
                 }
   
               logger.info("Final answer parsed response: " + str(parsed_response))
               return parsed_response
         except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
       case _:
         addRepromptResponse(parsed_response, 'Failed to parse the LLM output')
         logger.info(parsed_response)
         return parsed_response
   
   
   def load_response(text):
     raw_text = r'{}'.format(text)
     json_text = json.loads(raw_text)
     return json_text
   
   
   def get_content_by_type(content):
     content_by_type = defaultdict(list)
     for content_value in content:
       content_by_type[content_value["type"]].append(content_value)
     return content_by_type
   
   
   def parse_rationale(content_by_type):
     if "text" in content_by_type:
       rationale = content_by_type["text"][0]["text"]
       if rationale is not None:
         rationale_matcher = next(
             (pattern.search(rationale) for pattern in RATIONALE_VALUE_PATTERNS if
              pattern.search(rationale)),
             None)
         if rationale_matcher:
           rationale = rationale_matcher.group(1).strip()
       return rationale
     return None
   
   
   def parse_answer(response):
     if has_generated_response(response["text"].strip()):
       return parse_generated_response(response)
   
     answer_match = ANSWER_PATTERN.search(response["text"].strip())
     if answer_match:
       return answer_match.group(0).strip(), None
   
     return None, None
   
   
   def parse_generated_response(response):
     results = []
   
     for match in ANSWER_PART_PATTERN.finditer(response):
       part = match.group(1).strip()
   
       text_match = ANSWER_TEXT_PART_PATTERN.search(part)
       if not text_match:
         raise ValueError("Could not parse generated response")
   
       text = text_match.group(1).strip()
       references = parse_references(part)
       results.append((text, references))
   
     final_response = " ".join([r[0] for r in results])
   
     generated_response_parts = []
     for text, references in results:
       generatedResponsePart = {
         'text': text,
         'references': references
       }
       generated_response_parts.append(generatedResponsePart)
   
     return final_response, generated_response_parts
   
   
   def has_generated_response(raw_response):
     return ANSWER_PART_PATTERN.search(raw_response) is not None
   
   
   def parse_references(answer_part):
     references = []
     for match in ANSWER_REFERENCE_PART_PATTERN.finditer(answer_part):
       reference = match.group(1).strip()
       references.append({'sourceId': reference})
     return references
   
   
   def parse_ask_user(content_by_type):
     try:
       if content_by_type["tool_use"][0]["name"] == ASK_USER:
         ask_user_question = content_by_type["tool_use"][0]["input"]["question"]
         if not ask_user_question:
           raise ValueError(MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE)
         return ask_user_question
     except ValueError as ex:
       raise ex
     return None
   
   
   def parse_function_call(content_by_type, parsed_response):
     try:
       content = content_by_type["tool_use"][0]
       tool_name = content["name"]
   
       action_split = tool_name.split('__')
       verb = action_split[0].strip()
       resource_name = action_split[1].strip()
       function = action_split[2].strip()
     except ValueError as ex:
       raise ValueError(FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE)
   
     parameters = {}
     for param, value in content["input"].items():
       parameters[param] = {'value': value}
   
     parsed_response['orchestrationParsedResponse']['responseDetails'] = {}
   
     # Function calls can either invoke an action group or a knowledge base.
     # Mapping to the correct variable names accordingly
     if resource_name.lower().startswith(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX):
       parsed_response['orchestrationParsedResponse']['responseDetails'][
         'invocationType'] = 'KNOWLEDGE_BASE'
       parsed_response['orchestrationParsedResponse']['responseDetails'][
         'agentKnowledgeBase'] = {
         'searchQuery': parameters['searchQuery'],
         'knowledgeBaseId': resource_name.replace(
             KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX, ''),
         'id': content["id"]
       }
       return parsed_response
     parsed_response['orchestrationParsedResponse']['responseDetails'][
       'invocationType'] = 'ACTION_GROUP'
     parsed_response['orchestrationParsedResponse']['responseDetails'][
       'actionGroupInvocation'] = {
       "verb": verb,
       "actionGroupName": resource_name,
       "apiName": function,
       "actionGroupInput": parameters,
       "id": content["id"]
     }
     return parsed_response
   
   
   def addRepromptResponse(parsed_response, error):
     error_message = str(error)
     logger.warn(error_message)
   
     parsed_response['orchestrationParsedResponse']['parsingErrorDetails'] = {
       'repromptResponse': error_message
     }
   ```

------

1. 要查看使用函数详细信息定义的操作组的示例，请选择与您想要查看其示例的模型相对应的选项卡。

------
#### [ Anthropic Claude 2.0 ]

   ```
   import json
   import re
   import logging
    
    
   RATIONALE_REGEX_LIST = [
       "(.*?)(<function_call>)",
       "(.*?)(<answer>)"
   ]
   RATIONALE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_REGEX_LIST]
    
   RATIONALE_VALUE_REGEX_LIST = [
       "<scratchpad>(.*?)(</scratchpad>)",
       "(.*?)(</scratchpad>)",
       "(<scratchpad>)(.*?)"
   ]
   RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_VALUE_REGEX_LIST]
    
   ANSWER_REGEX = r"(?<=<answer>)(.*)"
   ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
    
   ANSWER_TAG = "<answer>"
   FUNCTION_CALL_TAG = "<function_call>"
    
   ASK_USER_FUNCTION_CALL_REGEX = r"(<function_call>user::askuser)(.*)\)"
   ASK_USER_FUNCTION_CALL_PATTERN = re.compile(ASK_USER_FUNCTION_CALL_REGEX, re.DOTALL)
    
   ASK_USER_FUNCTION_PARAMETER_REGEX = r"(?<=askuser=\")(.*?)\""  
   ASK_USER_FUNCTION_PARAMETER_PATTERN = re.compile(ASK_USER_FUNCTION_PARAMETER_REGEX, re.DOTALL)
    
   KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX = "x_amz_knowledgebase_"
    
   FUNCTION_CALL_REGEX_API_SCHEMA = r"<function_call>(\w+)::(\w+)::(.+)\((.+)\)"
   FUNCTION_CALL_REGEX_FUNCTION_SCHEMA = r"<function_call>(\w+)::(.+)\((.+)\)"
    
   ANSWER_PART_REGEX = "<answer_part\\s?>(.+?)</answer_part\\s?>"
   ANSWER_TEXT_PART_REGEX = "<text\\s?>(.+?)</text\\s?>"  
   ANSWER_REFERENCE_PART_REGEX = "<source\\s?>(.+?)</source\\s?>"
   ANSWER_PART_PATTERN = re.compile(ANSWER_PART_REGEX, re.DOTALL)
   ANSWER_TEXT_PART_PATTERN = re.compile(ANSWER_TEXT_PART_REGEX, re.DOTALL)
   ANSWER_REFERENCE_PART_PATTERN = re.compile(ANSWER_REFERENCE_PART_REGEX, re.DOTALL)
    
   # You can provide messages to reprompt the LLM in case the LLM output is not in the expected format
   MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE = "Missing the argument askuser for user::askuser function call. Please try again with the correct argument added"
   ASK_USER_FUNCTION_CALL_STRUCTURE_REMPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls to the askuser function must be: <function_call>user::askuser(askuser=\"$ASK_USER_INPUT\")</function_call>."
   FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = 'The function call format is incorrect. The format for function calls must be: <function_call>$FUNCTION_NAME($FUNCTION_ARGUMENT_NAME=""$FUNCTION_ARGUMENT_NAME"")</function_call>.'
    
   logger = logging.getLogger()
   logger.setLevel("INFO")
    
   # This parser lambda is an example of how to parse the LLM output for the default orchestration prompt
   def lambda_handler(event, context):
       logger.info("Lambda input: " + str(event))
       
       # Sanitize LLM response
       sanitized_response = sanitize_response(event['invokeModelRawResponse'])
       
       # Parse LLM response for any rationale
       rationale = parse_rationale(sanitized_response)
       
       # Construct response fields common to all invocation types
       parsed_response = {
           'promptType': "ORCHESTRATION",
           'orchestrationParsedResponse': {
               'rationale': rationale
           }
       }
       
       # Check if there is a final answer
       try:
           final_answer, generated_response_parts = parse_answer(sanitized_response)
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
           
       if final_answer:
           parsed_response['orchestrationParsedResponse']['responseDetails'] = {
               'invocationType': 'FINISH',
               'agentFinalResponse': {
                   'responseText': final_answer
               }
           }
           
           if generated_response_parts:
               parsed_response['orchestrationParsedResponse']['responseDetails']['agentFinalResponse']['citations'] = {
                   'generatedResponseParts': generated_response_parts
               }
          
           logger.info("Final answer parsed response: " + str(parsed_response))
           return parsed_response
       
       # Check if there is an ask user
       try:
           ask_user = parse_ask_user(sanitized_response)
           if ask_user:
               parsed_response['orchestrationParsedResponse']['responseDetails'] = {
                   'invocationType': 'ASK_USER',
                   'agentAskUser': {
                       'responseText': ask_user
                   }
               }
               
               logger.info("Ask user parsed response: " + str(parsed_response))
               return parsed_response
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
           
       # Check if there is an agent action
       try:
           parsed_response = parse_function_call(sanitized_response, parsed_response)
           logger.info("Function call parsed response: " + str(parsed_response))
           return parsed_response
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
    
       addRepromptResponse(parsed_response, 'Failed to parse the LLM output')
       logger.info(parsed_response)
       return parsed_response
           
       raise Exception("unrecognized prompt type")
    
   def sanitize_response(text):
       pattern = r"(\\n*)"
       text = re.sub(pattern, r"\n", text)
       return text
       
   def parse_rationale(sanitized_response):
       # Checks for strings that are not required for orchestration
       rationale_matcher = next((pattern.search(sanitized_response) for pattern in RATIONALE_PATTERNS if pattern.search(sanitized_response)), None)
       
       if rationale_matcher:
           rationale = rationale_matcher.group(1).strip()
           
           # Check if there is a formatted rationale that we can parse from the string
           rationale_value_matcher = next((pattern.search(rationale) for pattern in RATIONALE_VALUE_PATTERNS if pattern.search(rationale)), None)
           if rationale_value_matcher:
               return rationale_value_matcher.group(1).strip()
           
           return rationale
       
       return None
       
   def parse_answer(sanitized_llm_response):
       if has_generated_response(sanitized_llm_response):
           return parse_generated_response(sanitized_llm_response)
    
       answer_match = ANSWER_PATTERN.search(sanitized_llm_response)
       if answer_match and is_answer(sanitized_llm_response):
           return answer_match.group(0).strip(), None
           
       return None, None
     
   def is_answer(llm_response):
       return llm_response.rfind(ANSWER_TAG) > llm_response.rfind(FUNCTION_CALL_TAG)
       
   def parse_generated_response(sanitized_llm_response):
       results = []
       
       for match in ANSWER_PART_PATTERN.finditer(sanitized_llm_response):
           part = match.group(1).strip()
           
           text_match = ANSWER_TEXT_PART_PATTERN.search(part)
           if not text_match:
               raise ValueError("Could not parse generated response")
           
           text = text_match.group(1).strip()        
           references = parse_references(sanitized_llm_response, part)
           results.append((text, references))
       
       final_response = " ".join([r[0] for r in results])
       
       generated_response_parts = []
       for text, references in results:
           generatedResponsePart = {
               'text': text, 
               'references': references
           }
           generated_response_parts.append(generatedResponsePart)
           
       return final_response, generated_response_parts
    
       
   def has_generated_response(raw_response):
       return ANSWER_PART_PATTERN.search(raw_response) is not None
    
   def parse_references(raw_response, answer_part):
       references = []
       for match in ANSWER_REFERENCE_PART_PATTERN.finditer(answer_part):
           reference = match.group(1).strip()
           references.append({'sourceId': reference})
       return references
       
   def parse_ask_user(sanitized_llm_response):
       ask_user_matcher = ASK_USER_FUNCTION_CALL_PATTERN.search(sanitized_llm_response)
       if ask_user_matcher:
           try:
               ask_user = ask_user_matcher.group(2).strip()
               ask_user_question_matcher = ASK_USER_FUNCTION_PARAMETER_PATTERN.search(ask_user)
               if ask_user_question_matcher:
                   return ask_user_question_matcher.group(1).strip()
               raise ValueError(MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE)
           except ValueError as ex:
               raise ex
           except Exception as ex:
               raise Exception(ASK_USER_FUNCTION_CALL_STRUCTURE_REMPROMPT_MESSAGE)
           
       return None
    
   def parse_function_call(sanitized_response, parsed_response):
       match = re.search(FUNCTION_CALL_REGEX_API_SCHEMA, sanitized_response)
       match_function_schema = re.search(FUNCTION_CALL_REGEX_FUNCTION_SCHEMA, sanitized_response)
       if not match and not match_function_schema:
           raise ValueError(FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE)
    
       if match:
           schema_type = 'API'
           verb, resource_name, function, param_arg = match.group(1), match.group(2), match.group(3), match.group(4)
       else:
           schema_type = 'FUNCTION'
           resource_name, function, param_arg = match_function_schema.group(1), match_function_schema.group(2), match_function_schema.group(3)
       
       parameters = {}
       for arg in param_arg.split(","):
           key, value = arg.split("=")
           parameters[key.strip()] = {'value': value.strip('" ')}
           
       parsed_response['orchestrationParsedResponse']['responseDetails'] = {}
           
       # Function calls can either invoke an action group or a knowledge base.
       # Mapping to the correct variable names accordingly
       if schema_type == 'API' and resource_name.lower().startswith(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX):
           parsed_response['orchestrationParsedResponse']['responseDetails']['invocationType'] = 'KNOWLEDGE_BASE'
           parsed_response['orchestrationParsedResponse']['responseDetails']['agentKnowledgeBase'] = {
               'searchQuery': parameters['searchQuery'],
               'knowledgeBaseId': resource_name.replace(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX, '')
           }
           
           return parsed_response
       
       parsed_response['orchestrationParsedResponse']['responseDetails']['invocationType'] = 'ACTION_GROUP'
       
       if schema_type == 'API':
           parsed_response['orchestrationParsedResponse']['responseDetails']['actionGroupInvocation'] = {
               "verb": verb, 
               "actionGroupName": resource_name,
               "apiName": function,
               "actionGroupInput": parameters
           }
       else:
           parsed_response['orchestrationParsedResponse']['responseDetails']['actionGroupInvocation'] = {
               "actionGroupName": resource_name,
               "functionName": function,
               "actionGroupInput": parameters
           }
       
       return parsed_response
       
   def addRepromptResponse(parsed_response, error):
       error_message = str(error)
       logger.warn(error_message)
       
       parsed_response['orchestrationParsedResponse']['parsingErrorDetails'] = {
           'repromptResponse': error_message
       }
   ```

------
#### [ Anthropic Claude 2.1 ]

   ```
   import logging
   import re
   import xml.etree.ElementTree as ET
    
   RATIONALE_REGEX_LIST = [
       "(.*?)(<function_calls>)",
       "(.*?)(<answer>)"
   ]
   RATIONALE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_REGEX_LIST]
    
   RATIONALE_VALUE_REGEX_LIST = [
       "<scratchpad>(.*?)(</scratchpad>)",
       "(.*?)(</scratchpad>)",
       "(<scratchpad>)(.*?)"
   ]
   RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_VALUE_REGEX_LIST]
    
   ANSWER_REGEX = r"(?<=<answer>)(.*)"
   ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
    
   ANSWER_TAG = "<answer>"
   FUNCTION_CALL_TAG = "<function_calls>"
    
   ASK_USER_FUNCTION_CALL_REGEX = r"<tool_name>user::askuser</tool_name>"
   ASK_USER_FUNCTION_CALL_PATTERN = re.compile(ASK_USER_FUNCTION_CALL_REGEX, re.DOTALL)
    
   ASK_USER_TOOL_NAME_REGEX = r"<tool_name>((.|\n)*?)</tool_name>"
   ASK_USER_TOOL_NAME_PATTERN = re.compile(ASK_USER_TOOL_NAME_REGEX, re.DOTALL)
    
   TOOL_PARAMETERS_REGEX = r"<parameters>((.|\n)*?)</parameters>"
   TOOL_PARAMETERS_PATTERN = re.compile(TOOL_PARAMETERS_REGEX, re.DOTALL)
    
   ASK_USER_TOOL_PARAMETER_REGEX = r"<question>((.|\n)*?)</question>"
   ASK_USER_TOOL_PARAMETER_PATTERN = re.compile(ASK_USER_TOOL_PARAMETER_REGEX, re.DOTALL)
    
    
   KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX = "x_amz_knowledgebase_"
    
   FUNCTION_CALL_REGEX = r"(?<=<function_calls>)(.*)"
    
   ANSWER_PART_REGEX = "<answer_part\\s?>(.+?)</answer_part\\s?>"
   ANSWER_TEXT_PART_REGEX = "<text\\s?>(.+?)</text\\s?>"
   ANSWER_REFERENCE_PART_REGEX = "<source\\s?>(.+?)</source\\s?>"
   ANSWER_PART_PATTERN = re.compile(ANSWER_PART_REGEX, re.DOTALL)
   ANSWER_TEXT_PART_PATTERN = re.compile(ANSWER_TEXT_PART_REGEX, re.DOTALL)
   ANSWER_REFERENCE_PART_PATTERN = re.compile(ANSWER_REFERENCE_PART_REGEX, re.DOTALL)
    
   # You can provide messages to reprompt the LLM in case the LLM output is not in the expected format
   MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE = "Missing the parameter 'question' for user::askuser function call. Please try again with the correct argument added."
   ASK_USER_FUNCTION_CALL_STRUCTURE_REMPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls to the askuser function must be: <invoke> <tool_name>user::askuser</tool_name><parameters><question>$QUESTION</question></parameters></invoke>."
   FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls must be: <invoke> <tool_name>$TOOL_NAME</tool_name> <parameters> <$PARAMETER_NAME>$PARAMETER_VALUE</$PARAMETER_NAME>...</parameters></invoke>."
    
   logger = logging.getLogger()
   logger.setLevel("INFO")
    
   # This parser lambda is an example of how to parse the LLM output for the default orchestration prompt
   def lambda_handler(event, context):
       logger.info("Lambda input: " + str(event))
    
       # Sanitize LLM response
       sanitized_response = sanitize_response(event['invokeModelRawResponse'])
    
       # Parse LLM response for any rationale
       rationale = parse_rationale(sanitized_response)
    
       # Construct response fields common to all invocation types
       parsed_response = {
           'promptType': "ORCHESTRATION",
           'orchestrationParsedResponse': {
               'rationale': rationale
           }
       }
    
       # Check if there is a final answer
       try:
           final_answer, generated_response_parts = parse_answer(sanitized_response)
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
    
       if final_answer:
           parsed_response['orchestrationParsedResponse']['responseDetails'] = {
               'invocationType': 'FINISH',
               'agentFinalResponse': {
                   'responseText': final_answer
               }
           }
    
           if generated_response_parts:
               parsed_response['orchestrationParsedResponse']['responseDetails']['agentFinalResponse']['citations'] = {
                   'generatedResponseParts': generated_response_parts
               }
    
           logger.info("Final answer parsed response: " + str(parsed_response))
           return parsed_response
    
       # Check if there is an ask user
       try:
           ask_user = parse_ask_user(sanitized_response)
           if ask_user:
               parsed_response['orchestrationParsedResponse']['responseDetails'] = {
                   'invocationType': 'ASK_USER',
                   'agentAskUser': {
                       'responseText': ask_user
                   }
               }
    
               logger.info("Ask user parsed response: " + str(parsed_response))
               return parsed_response
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
    
       # Check if there is an agent action
       try:
           parsed_response = parse_function_call(sanitized_response, parsed_response)
           logger.info("Function call parsed response: " + str(parsed_response))
           return parsed_response
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
    
       addRepromptResponse(parsed_response, 'Failed to parse the LLM output')
       logger.info(parsed_response)
       return parsed_response
    
       raise Exception("unrecognized prompt type")
    
    
   def sanitize_response(text):
       pattern = r"(\\n*)"
       text = re.sub(pattern, r"\n", text)
       return text
    
    
   def parse_rationale(sanitized_response):
       # Checks for strings that are not required for orchestration
       rationale_matcher = next(
           (pattern.search(sanitized_response) for pattern in RATIONALE_PATTERNS if pattern.search(sanitized_response)),
           None)
    
       if rationale_matcher:
           rationale = rationale_matcher.group(1).strip()
    
           # Check if there is a formatted rationale that we can parse from the string
           rationale_value_matcher = next(
               (pattern.search(rationale) for pattern in RATIONALE_VALUE_PATTERNS if pattern.search(rationale)), None)
           if rationale_value_matcher:
               return rationale_value_matcher.group(1).strip()
    
           return rationale
    
       return None
    
    
   def parse_answer(sanitized_llm_response):
       if has_generated_response(sanitized_llm_response):
           return parse_generated_response(sanitized_llm_response)
    
       answer_match = ANSWER_PATTERN.search(sanitized_llm_response)
       if answer_match and is_answer(sanitized_llm_response):
           return answer_match.group(0).strip(), None
    
       return None, None
    
    
   def is_answer(llm_response):
       return llm_response.rfind(ANSWER_TAG) > llm_response.rfind(FUNCTION_CALL_TAG)
    
    
   def parse_generated_response(sanitized_llm_response):
       results = []
    
       for match in ANSWER_PART_PATTERN.finditer(sanitized_llm_response):
           part = match.group(1).strip()
    
           text_match = ANSWER_TEXT_PART_PATTERN.search(part)
           if not text_match:
               raise ValueError("Could not parse generated response")
    
           text = text_match.group(1).strip()
           references = parse_references(sanitized_llm_response, part)
           results.append((text, references))
    
       final_response = " ".join([r[0] for r in results])
    
       generated_response_parts = []
       for text, references in results:
           generatedResponsePart = {
               'text': text,
               'references': references
           }
           generated_response_parts.append(generatedResponsePart)
    
       return final_response, generated_response_parts
    
    
   def has_generated_response(raw_response):
       return ANSWER_PART_PATTERN.search(raw_response) is not None
    
    
   def parse_references(raw_response, answer_part):
       references = []
       for match in ANSWER_REFERENCE_PART_PATTERN.finditer(answer_part):
           reference = match.group(1).strip()
           references.append({'sourceId': reference})
       return references
    
    
   def parse_ask_user(sanitized_llm_response):
       ask_user_matcher = ASK_USER_FUNCTION_CALL_PATTERN.search(sanitized_llm_response)
       if ask_user_matcher:
           try:
               parameters_matches = TOOL_PARAMETERS_PATTERN.search(sanitized_llm_response)
               params = parameters_matches.group(1).strip()
               ask_user_question_matcher = ASK_USER_TOOL_PARAMETER_PATTERN.search(params)
               if ask_user_question_matcher:
                   ask_user_question = ask_user_question_matcher.group(1)
                   return ask_user_question
               raise ValueError(MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE)
           except ValueError as ex:
               raise ex
           except Exception as ex:
               raise Exception(ASK_USER_FUNCTION_CALL_STRUCTURE_REMPROMPT_MESSAGE)
    
       return None
    
    
   def parse_function_call(sanitized_response, parsed_response):
       match = re.search(FUNCTION_CALL_REGEX, sanitized_response)
       if not match:
           raise ValueError(FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE)
    
       tool_name_matches = ASK_USER_TOOL_NAME_PATTERN.search(sanitized_response)
       tool_name = tool_name_matches.group(1)
       parameters_matches = TOOL_PARAMETERS_PATTERN.search(sanitized_response)
       params = parameters_matches.group(1).strip()
    
       action_split = tool_name.split('::')
       schema_type = 'FUNCTION' if len(action_split) == 2 else 'API'
    
       if schema_type == 'API':
           verb = action_split[0].strip()
           resource_name = action_split[1].strip()
           function = action_split[2].strip()
       else:
           resource_name = action_split[0].strip()
           function = action_split[1].strip()
    
       xml_tree = ET.ElementTree(ET.fromstring("<parameters>{}</parameters>".format(params)))
       parameters = {}
       for elem in xml_tree.iter():
           if elem.text:
               parameters[elem.tag] = {'value': elem.text.strip('" ')}
    
       parsed_response['orchestrationParsedResponse']['responseDetails'] = {}
    
       # Function calls can either invoke an action group or a knowledge base.
       # Mapping to the correct variable names accordingly
       if schema_type == 'API' and resource_name.lower().startswith(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX):
           parsed_response['orchestrationParsedResponse']['responseDetails']['invocationType'] = 'KNOWLEDGE_BASE'
           parsed_response['orchestrationParsedResponse']['responseDetails']['agentKnowledgeBase'] = {
               'searchQuery': parameters['searchQuery'],
               'knowledgeBaseId': resource_name.replace(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX, '')
           }
    
           return parsed_response
    
       parsed_response['orchestrationParsedResponse']['responseDetails']['invocationType'] = 'ACTION_GROUP'
       if schema_type == 'API':
           parsed_response['orchestrationParsedResponse']['responseDetails']['actionGroupInvocation'] = {
               "verb": verb,
               "actionGroupName": resource_name,
               "apiName": function,
               "actionGroupInput": parameters
           }
       else:
           parsed_response['orchestrationParsedResponse']['responseDetails']['actionGroupInvocation'] = {
               "actionGroupName": resource_name,
               "functionName": function,
               "actionGroupInput": parameters
           }
    
       return parsed_response
    
    
   def addRepromptResponse(parsed_response, error):
       error_message = str(error)
       logger.warn(error_message)
    
       parsed_response['orchestrationParsedResponse']['parsingErrorDetails'] = {
           'repromptResponse': error_message
       }
   ```

------
#### [ Anthropic Claude 3 ]

   ```
   import logging
   import re
   import xml.etree.ElementTree as ET
    
   RATIONALE_REGEX_LIST = [
       "(.*?)(<function_calls>)",
       "(.*?)(<answer>)"
   ]
   RATIONALE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_REGEX_LIST]
    
   RATIONALE_VALUE_REGEX_LIST = [
       "<thinking>(.*?)(</thinking>)",
       "(.*?)(</thinking>)",
       "(<thinking>)(.*?)"
   ]
   RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_VALUE_REGEX_LIST]
    
   ANSWER_REGEX = r"(?<=<answer>)(.*)"
   ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
    
   ANSWER_TAG = "<answer>"
   FUNCTION_CALL_TAG = "<function_calls>"
    
   ASK_USER_FUNCTION_CALL_REGEX = r"<tool_name>user::askuser</tool_name>"
   ASK_USER_FUNCTION_CALL_PATTERN = re.compile(ASK_USER_FUNCTION_CALL_REGEX, re.DOTALL)
    
   ASK_USER_TOOL_NAME_REGEX = r"<tool_name>((.|\n)*?)</tool_name>"
   ASK_USER_TOOL_NAME_PATTERN = re.compile(ASK_USER_TOOL_NAME_REGEX, re.DOTALL)
    
   TOOL_PARAMETERS_REGEX = r"<parameters>((.|\n)*?)</parameters>"
   TOOL_PARAMETERS_PATTERN = re.compile(TOOL_PARAMETERS_REGEX, re.DOTALL)
    
   ASK_USER_TOOL_PARAMETER_REGEX = r"<question>((.|\n)*?)</question>"
   ASK_USER_TOOL_PARAMETER_PATTERN = re.compile(ASK_USER_TOOL_PARAMETER_REGEX, re.DOTALL)
    
    
   KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX = "x_amz_knowledgebase_"
    
   FUNCTION_CALL_REGEX = r"(?<=<function_calls>)(.*)"
    
   ANSWER_PART_REGEX = "<answer_part\\s?>(.+?)</answer_part\\s?>"
   ANSWER_TEXT_PART_REGEX = "<text\\s?>(.+?)</text\\s?>"
   ANSWER_REFERENCE_PART_REGEX = "<source\\s?>(.+?)</source\\s?>"
   ANSWER_PART_PATTERN = re.compile(ANSWER_PART_REGEX, re.DOTALL)
   ANSWER_TEXT_PART_PATTERN = re.compile(ANSWER_TEXT_PART_REGEX, re.DOTALL)
   ANSWER_REFERENCE_PART_PATTERN = re.compile(ANSWER_REFERENCE_PART_REGEX, re.DOTALL)
    
   # You can provide messages to reprompt the LLM in case the LLM output is not in the expected format
   MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE = "Missing the parameter 'question' for user::askuser function call. Please try again with the correct argument added."
   ASK_USER_FUNCTION_CALL_STRUCTURE_REMPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls to the askuser function must be: <invoke> <tool_name>user::askuser</tool_name><parameters><question>$QUESTION</question></parameters></invoke>."
   FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls must be: <invoke> <tool_name>$TOOL_NAME</tool_name> <parameters> <$PARAMETER_NAME>$PARAMETER_VALUE</$PARAMETER_NAME>...</parameters></invoke>."
    
   logger = logging.getLogger()
    
    
   # This parser lambda is an example of how to parse the LLM output for the default orchestration prompt
   def lambda_handler(event, context):
       logger.info("Lambda input: " + str(event))
    
       # Sanitize LLM response
       sanitized_response = sanitize_response(event['invokeModelRawResponse'])
    
       # Parse LLM response for any rationale
       rationale = parse_rationale(sanitized_response)
    
       # Construct response fields common to all invocation types
       parsed_response = {
           'promptType': "ORCHESTRATION",
           'orchestrationParsedResponse': {
               'rationale': rationale
           }
       }
    
       # Check if there is a final answer
       try:
           final_answer, generated_response_parts = parse_answer(sanitized_response)
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
    
       if final_answer:
           parsed_response['orchestrationParsedResponse']['responseDetails'] = {
               'invocationType': 'FINISH',
               'agentFinalResponse': {
                   'responseText': final_answer
               }
           }
    
           if generated_response_parts:
               parsed_response['orchestrationParsedResponse']['responseDetails']['agentFinalResponse']['citations'] = {
                   'generatedResponseParts': generated_response_parts
               }
    
           logger.info("Final answer parsed response: " + str(parsed_response))
           return parsed_response
    
       # Check if there is an ask user
       try:
           ask_user = parse_ask_user(sanitized_response)
           if ask_user:
               parsed_response['orchestrationParsedResponse']['responseDetails'] = {
                   'invocationType': 'ASK_USER',
                   'agentAskUser': {
                       'responseText': ask_user
                   }
               }
    
               logger.info("Ask user parsed response: " + str(parsed_response))
               return parsed_response
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
    
       # Check if there is an agent action
       try:
           parsed_response = parse_function_call(sanitized_response, parsed_response)
           logger.info("Function call parsed response: " + str(parsed_response))
           return parsed_response
       except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
    
       addRepromptResponse(parsed_response, 'Failed to parse the LLM output')
       logger.info(parsed_response)
       return parsed_response
    
       raise Exception("unrecognized prompt type")
    
    
   def sanitize_response(text):
       pattern = r"(\\n*)"
       text = re.sub(pattern, r"\n", text)
       return text
    
    
   def parse_rationale(sanitized_response):
       # Checks for strings that are not required for orchestration
       rationale_matcher = next(
           (pattern.search(sanitized_response) for pattern in RATIONALE_PATTERNS if pattern.search(sanitized_response)),
           None)
    
       if rationale_matcher:
           rationale = rationale_matcher.group(1).strip()
    
           # Check if there is a formatted rationale that we can parse from the string
           rationale_value_matcher = next(
               (pattern.search(rationale) for pattern in RATIONALE_VALUE_PATTERNS if pattern.search(rationale)), None)
           if rationale_value_matcher:
               return rationale_value_matcher.group(1).strip()
    
           return rationale
    
       return None
    
    
   def parse_answer(sanitized_llm_response):
       if has_generated_response(sanitized_llm_response):
           return parse_generated_response(sanitized_llm_response)
    
       answer_match = ANSWER_PATTERN.search(sanitized_llm_response)
       if answer_match and is_answer(sanitized_llm_response):
           return answer_match.group(0).strip(), None
    
       return None, None
    
    
   def is_answer(llm_response):
       return llm_response.rfind(ANSWER_TAG) > llm_response.rfind(FUNCTION_CALL_TAG)
    
    
   def parse_generated_response(sanitized_llm_response):
       results = []
    
       for match in ANSWER_PART_PATTERN.finditer(sanitized_llm_response):
           part = match.group(1).strip()
    
           text_match = ANSWER_TEXT_PART_PATTERN.search(part)
           if not text_match:
               raise ValueError("Could not parse generated response")
    
           text = text_match.group(1).strip()
           references = parse_references(sanitized_llm_response, part)
           results.append((text, references))
    
       final_response = " ".join([r[0] for r in results])
    
       generated_response_parts = []
       for text, references in results:
           generatedResponsePart = {
               'text': text,
               'references': references
           }
           generated_response_parts.append(generatedResponsePart)
    
       return final_response, generated_response_parts
    
    
   def has_generated_response(raw_response):
       return ANSWER_PART_PATTERN.search(raw_response) is not None
    
    
   def parse_references(raw_response, answer_part):
       references = []
       for match in ANSWER_REFERENCE_PART_PATTERN.finditer(answer_part):
           reference = match.group(1).strip()
           references.append({'sourceId': reference})
       return references
    
    
   def parse_ask_user(sanitized_llm_response):
       ask_user_matcher = ASK_USER_FUNCTION_CALL_PATTERN.search(sanitized_llm_response)
       if ask_user_matcher:
           try:
               parameters_matches = TOOL_PARAMETERS_PATTERN.search(sanitized_llm_response)
               params = parameters_matches.group(1).strip()
               ask_user_question_matcher = ASK_USER_TOOL_PARAMETER_PATTERN.search(params)
               if ask_user_question_matcher:
                   ask_user_question = ask_user_question_matcher.group(1)
                   return ask_user_question
               raise ValueError(MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE)
           except ValueError as ex:
               raise ex
           except Exception as ex:
               raise Exception(ASK_USER_FUNCTION_CALL_STRUCTURE_REMPROMPT_MESSAGE)
    
       return None
    
    
   def parse_function_call(sanitized_response, parsed_response):
       match = re.search(FUNCTION_CALL_REGEX, sanitized_response)
       if not match:
           raise ValueError(FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE)
    
       tool_name_matches = ASK_USER_TOOL_NAME_PATTERN.search(sanitized_response)
       tool_name = tool_name_matches.group(1)
       parameters_matches = TOOL_PARAMETERS_PATTERN.search(sanitized_response)
       params = parameters_matches.group(1).strip()
    
       action_split = tool_name.split('::')
       schema_type = 'FUNCTION' if len(action_split) == 2 else 'API'
    
       if schema_type == 'API':
           verb = action_split[0].strip()
           resource_name = action_split[1].strip()
           function = action_split[2].strip()
       else:
           resource_name = action_split[0].strip()
           function = action_split[1].strip()
    
       xml_tree = ET.ElementTree(ET.fromstring("<parameters>{}</parameters>".format(params)))
       parameters = {}
       for elem in xml_tree.iter():
           if elem.text:
               parameters[elem.tag] = {'value': elem.text.strip('" ')}
    
       parsed_response['orchestrationParsedResponse']['responseDetails'] = {}
    
       # Function calls can either invoke an action group or a knowledge base.
       # Mapping to the correct variable names accordingly
       if schema_type == 'API' and resource_name.lower().startswith(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX):
           parsed_response['orchestrationParsedResponse']['responseDetails']['invocationType'] = 'KNOWLEDGE_BASE'
           parsed_response['orchestrationParsedResponse']['responseDetails']['agentKnowledgeBase'] = {
               'searchQuery': parameters['searchQuery'],
               'knowledgeBaseId': resource_name.replace(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX, '')
           }
    
           return parsed_response
    
       parsed_response['orchestrationParsedResponse']['responseDetails']['invocationType'] = 'ACTION_GROUP'
       if schema_type == 'API':
           parsed_response['orchestrationParsedResponse']['responseDetails']['actionGroupInvocation'] = {
               "verb": verb,
               "actionGroupName": resource_name,
               "apiName": function,
               "actionGroupInput": parameters
           }
       else:
           parsed_response['orchestrationParsedResponse']['responseDetails']['actionGroupInvocation'] = {
               "actionGroupName": resource_name,
               "functionName": function,
               "actionGroupInput": parameters
           }
    
       return parsed_response
    
    
   def addRepromptResponse(parsed_response, error):
       error_message = str(error)
       logger.warn(error_message)
    
       parsed_response['orchestrationParsedResponse']['parsingErrorDetails'] = {
           'repromptResponse': error_message
       }
   ```

------
#### [ Anthropic Claude 3.5 ]

   ```
   import json
   import logging
   import re
   from collections import defaultdict
   
   RATIONALE_VALUE_REGEX_LIST = [
     "<thinking>(.*?)(</thinking>)",
     "(.*?)(</thinking>)",
     "(<thinking>)(.*?)"
   ]
   RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in
                               RATIONALE_VALUE_REGEX_LIST]
   
   ANSWER_REGEX = r"(?<=<answer>)(.*)"
   ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
   
   ANSWER_TAG = "<answer>"
   ASK_USER = "user__askuser"
   
   KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX = "x_amz_knowledgebase_"
   
   ANSWER_PART_REGEX = "<answer_part\\s?>(.+?)</answer_part\\s?>"
   ANSWER_TEXT_PART_REGEX = "<text\\s?>(.+?)</text\\s?>"
   ANSWER_REFERENCE_PART_REGEX = "<source\\s?>(.+?)</source\\s?>"
   ANSWER_PART_PATTERN = re.compile(ANSWER_PART_REGEX, re.DOTALL)
   ANSWER_TEXT_PART_PATTERN = re.compile(ANSWER_TEXT_PART_REGEX, re.DOTALL)
   ANSWER_REFERENCE_PART_PATTERN = re.compile(ANSWER_REFERENCE_PART_REGEX,
                                              re.DOTALL)
   
   # You can provide messages to reprompt the LLM in case the LLM output is not in the expected format
   MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE = "Missing the parameter 'question' for user__askuser function call. Please try again with the correct argument added."
   FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = "The tool name format is incorrect. The format for the tool name must be: 'httpVerb__actionGroupName__apiName."
   logger = logging.getLogger()
   
   
   # This parser lambda is an example of how to parse the LLM output for the default orchestration prompt
   def lambda_handler(event, context):
     logger.setLevel("INFO")
     logger.info("Lambda input: " + str(event))
   
     # Sanitize LLM response
     response = load_response(event['invokeModelRawResponse'])
   
     stop_reason = response["stop_reason"]
     content = response["content"]
     content_by_type = get_content_by_type(content)
   
     # Parse LLM response for any rationale
     rationale = parse_rationale(content_by_type)
   
     # Construct response fields common to all invocation types
     parsed_response = {
       'promptType': "ORCHESTRATION",
       'orchestrationParsedResponse': {
         'rationale': rationale
       }
     }
   
     match stop_reason:
       case 'tool_use':
         # Check if there is an ask user
         try:
           ask_user = parse_ask_user(content_by_type)
           if ask_user:
             parsed_response['orchestrationParsedResponse']['responseDetails'] = {
               'invocationType': 'ASK_USER',
               'agentAskUser': {
                 'responseText': ask_user,
                 'id': content_by_type['tool_use'][0]['id']
               },
   
             }
   
             logger.info("Ask user parsed response: " + str(parsed_response))
             return parsed_response
         except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
   
         # Check if there is an agent action
         try:
           parsed_response = parse_function_call(content_by_type, parsed_response)
           logger.info("Function call parsed response: " + str(parsed_response))
           return parsed_response
         except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
   
       case 'end_turn' | 'stop_sequence':
         # Check if there is a final answer
         try:
           if content_by_type["text"]:
             text_contents = content_by_type["text"]
             for text_content in text_contents:
               final_answer, generated_response_parts = parse_answer(text_content)
               if final_answer:
                 parsed_response['orchestrationParsedResponse'][
                   'responseDetails'] = {
                   'invocationType': 'FINISH',
                   'agentFinalResponse': {
                     'responseText': final_answer
                   }
                 }
   
               if generated_response_parts:
                 parsed_response['orchestrationParsedResponse']['responseDetails'][
                   'agentFinalResponse']['citations'] = {
                   'generatedResponseParts': generated_response_parts
                 }
   
               logger.info("Final answer parsed response: " + str(parsed_response))
               return parsed_response
         except ValueError as e:
           addRepromptResponse(parsed_response, e)
           return parsed_response
       case _:
         addRepromptResponse(parsed_response, 'Failed to parse the LLM output')
         logger.info(parsed_response)
         return parsed_response
   
   
   def load_response(text):
     raw_text = r'{}'.format(text)
     json_text = json.loads(raw_text)
     return json_text
   
   
   def get_content_by_type(content):
     content_by_type = defaultdict(list)
     for content_value in content:
       content_by_type[content_value["type"]].append(content_value)
     return content_by_type
   
   
   def parse_rationale(content_by_type):
     if "text" in content_by_type:
       rationale = content_by_type["text"][0]["text"]
       if rationale is not None:
         rationale_matcher = next(
             (pattern.search(rationale) for pattern in RATIONALE_VALUE_PATTERNS if
              pattern.search(rationale)),
             None)
         if rationale_matcher:
           rationale = rationale_matcher.group(1).strip()
       return rationale
     return None
   
   
   def parse_answer(response):
     if has_generated_response(response["text"].strip()):
       return parse_generated_response(response)
   
     answer_match = ANSWER_PATTERN.search(response["text"].strip())
     if answer_match:
       return answer_match.group(0).strip(), None
   
     return None, None
   
   
   def parse_generated_response(response):
     results = []
   
     for match in ANSWER_PART_PATTERN.finditer(response):
       part = match.group(1).strip()
   
       text_match = ANSWER_TEXT_PART_PATTERN.search(part)
       if not text_match:
         raise ValueError("Could not parse generated response")
   
       text = text_match.group(1).strip()
       references = parse_references(part)
       results.append((text, references))
   
     final_response = " ".join([r[0] for r in results])
   
     generated_response_parts = []
     for text, references in results:
       generatedResponsePart = {
         'text': text,
         'references': references
       }
       generated_response_parts.append(generatedResponsePart)
   
     return final_response, generated_response_parts
   
   
   def has_generated_response(raw_response):
     return ANSWER_PART_PATTERN.search(raw_response) is not None
   
   
   def parse_references(answer_part):
     references = []
     for match in ANSWER_REFERENCE_PART_PATTERN.finditer(answer_part):
       reference = match.group(1).strip()
       references.append({'sourceId': reference})
     return references
   
   
   def parse_ask_user(content_by_type):
     try:
       if content_by_type["tool_use"][0]["name"] == ASK_USER:
         ask_user_question = content_by_type["tool_use"][0]["input"]["question"]
         if not ask_user_question:
           raise ValueError(MISSING_API_INPUT_FOR_USER_REPROMPT_MESSAGE)
         return ask_user_question
     except ValueError as ex:
       raise ex
     return None
   
   
   def parse_function_call(content_by_type, parsed_response):
     try:
       content = content_by_type["tool_use"][0]
       tool_name = content["name"]
   
       action_split = tool_name.split('__')
   
       schema_type = 'FUNCTION' if len(action_split) == 2 else 'API'
       if schema_type == 'API':
         verb = action_split[0].strip()
         resource_name = action_split[1].strip()
         function = action_split[2].strip()
       else:
         resource_name = action_split[1].strip()
         function = action_split[2].strip()
   
     except ValueError as ex:
       raise ValueError(FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE)
   
     parameters = {}
     for param, value in content["input"].items():
       parameters[param] = {'value': value}
   
     parsed_response['orchestrationParsedResponse']['responseDetails'] = {}
   
     # Function calls can either invoke an action group or a knowledge base.
     # Mapping to the correct variable names accordingly
     if schema_type == 'API' and resource_name.lower().startswith(KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX):
       parsed_response['orchestrationParsedResponse']['responseDetails'][
         'invocationType'] = 'KNOWLEDGE_BASE'
       parsed_response['orchestrationParsedResponse']['responseDetails'][
         'agentKnowledgeBase'] = {
         'searchQuery': parameters['searchQuery'],
         'knowledgeBaseId': resource_name.replace(
             KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX, ''),
         'id': content["id"]
       }
       return parsed_response
     parsed_response['orchestrationParsedResponse']['responseDetails'][
       'invocationType'] = 'ACTION_GROUP'
     if schema_type == 'API':
       parsed_response['orchestrationParsedResponse']['responseDetails'][
         'actionGroupInvocation'] = {
         "verb": verb,
         "actionGroupName": resource_name,
         "apiName": function,
         "actionGroupInput": parameters,
         "id": content["id"]
       }
     else:
       parsed_response['orchestrationParsedResponse']['responseDetails']['actionGroupInvocation'] = {
         "actionGroupName": resource_name,
         "functionName": function,
         "actionGroupInput": parameters
        }
     return parsed_response
   
   
   def addRepromptResponse(parsed_response, error):
     error_message = str(error)
     logger.warn(error_message)
   
     parsed_response['orchestrationParsedResponse']['parsingErrorDetails'] = {
       'repromptResponse': error_message
     }
   ```

------

### 知识库响应生成
<a name="parser-kb"></a>

以下示例显示了用 Python 编写的知识库响应生成解析器 Lambda 函数。

```
import json
import re
import logging
 
ANSWER_PART_REGEX = "&lt;answer_part\\s?>(.+?)&lt;/answer_part\\s?>"
ANSWER_TEXT_PART_REGEX = "&lt;text\\s?>(.+?)&lt;/text\\s?>"  
ANSWER_REFERENCE_PART_REGEX = "&lt;source\\s?>(.+?)&lt;/source\\s?>"
ANSWER_PART_PATTERN = re.compile(ANSWER_PART_REGEX, re.DOTALL)
ANSWER_TEXT_PART_PATTERN = re.compile(ANSWER_TEXT_PART_REGEX, re.DOTALL)
ANSWER_REFERENCE_PART_PATTERN = re.compile(ANSWER_REFERENCE_PART_REGEX, re.DOTALL)

logger = logging.getLogger()
 
# This parser lambda is an example of how to parse the LLM output for the default KB response generation prompt
def lambda_handler(event, context):
    logger.info("Lambda input: " + str(event))
    raw_response = event['invokeModelRawResponse']
    
    parsed_response = {
        'promptType': 'KNOWLEDGE_BASE_RESPONSE_GENERATION',
        'knowledgeBaseResponseGenerationParsedResponse': {
            'generatedResponse': parse_generated_response(raw_response)
        }
    }
    
    logger.info(parsed_response)
    return parsed_response
    
def parse_generated_response(sanitized_llm_response):
    results = []
    
    for match in ANSWER_PART_PATTERN.finditer(sanitized_llm_response):
        part = match.group(1).strip()
        
        text_match = ANSWER_TEXT_PART_PATTERN.search(part)
        if not text_match:
            raise ValueError("Could not parse generated response")
        
        text = text_match.group(1).strip()        
        references = parse_references(sanitized_llm_response, part)
        results.append((text, references))
    
    generated_response_parts = []
    for text, references in results:
        generatedResponsePart = {
            'text': text, 
            'references': references
        }
        generated_response_parts.append(generatedResponsePart)
        
    return {
        'generatedResponseParts': generated_response_parts
    }
    
def parse_references(raw_response, answer_part):
    references = []
    for match in ANSWER_REFERENCE_PART_PATTERN.finditer(answer_part):
        reference = match.group(1).strip()
        references.append({'sourceId': reference})
    return references
```

### 后处理
<a name="parser-postprocessing"></a>

以下示例展示了用 Python 编写的后处理解析器 Lambda 函数。

```
import json
import re
import logging
 
FINAL_RESPONSE_REGEX = r"&lt;final_response>([\s\S]*?)&lt;/final_response>"
FINAL_RESPONSE_PATTERN = re.compile(FINAL_RESPONSE_REGEX, re.DOTALL)

logger = logging.getLogger()
 
# This parser lambda is an example of how to parse the LLM output for the default PostProcessing prompt
def lambda_handler(event, context):
    logger.info("Lambda input: " + str(event))
    raw_response = event['invokeModelRawResponse']
    
    parsed_response = {
        'promptType': 'POST_PROCESSING',
        'postProcessingParsedResponse': {}
    }
    
    matcher = FINAL_RESPONSE_PATTERN.search(raw_response)
    if not matcher:
        raise Exception("Could not parse raw LLM output")
    response_text = matcher.group(1).strip()
    
    parsed_response['postProcessingParsedResponse']['responseText'] = response_text
    
    logger.info(parsed_response)
    return parsed_response
```

### 记忆汇总
<a name="parser-memory-summarization"></a>

以下示例展示了用 Python 编写的记忆汇总解析器 Lambda 函数。

```
import re
import logging

SUMMARY_TAG_PATTERN = r'<summary>(.*?)</summary>'
TOPIC_TAG_PATTERN = r'<topic name="(.+?)"\s*>(.+?)</topic>'
logger = logging.getLogger()

# This parser lambda is an example of how to parse the LLM output for the default LTM SUmmarization prompt
def lambda_handler(event, context):
    logger.info("Lambda input: " + str(event))
    
    # Sanitize LLM response
    model_response = sanitize_response(event['invokeModelRawResponse'])
    
    if event["promptType"] == "MEMORY_SUMMARIZATION":
        return format_response(parse_llm_response(model_response), event["promptType"])

def format_response(topic_summaries, prompt_type):
    return {
        "promptType": prompt_type,
        "memorySummarizationParsedResponse": {
            "topicwiseSummaries": topic_summaries
        }
    }
    
def parse_llm_response(output: str):
    # First extract content within summary tag
    summary_match = re.search(SUMMARY_TAG_PATTERN, output, re.DOTALL)
    if not summary_match:
        raise Exception("Error while parsing summarizer model output, no summary tag found!")
    
    summary_content = summary_match.group(1)
    topic_summaries = parse_topic_wise_summaries(summary_content)
        
    return topic_summaries

def parse_topic_wise_summaries(content):
    summaries = []
    # Then extract content within topic tag
    for match in re.finditer(TOPIC_TAG_PATTERN, content, re.DOTALL):
        topic_name = match.group(1)
        topic_summary = match.group(2).strip()
        summaries.append({
            'topic': topic_name,
            'summary': topic_summary
        })
    if not summaries:
        raise Exception("Error while parsing summarizer model output, no topics found!")
    return summaries

def sanitize_response(text):
    pattern = r"(\\n*)"
    text = re.sub(pattern, r"\n", text)
    return text
```