Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.
# Creación de una función de Lambda de analizador en Agentes para Amazon Bedrock
Cada plantilla de peticiones incluye una función de Lambda de analizador. Puede escribir su propia función de Lambda de analizador personalizada y especificar las plantillas cuya función de analizador predeterminada desea invalidar. Para escribir una función de Lambda del analizador personalizada, debe comprender el evento de entrada que envía el agente y la respuesta resultante que espera el agente de la función de Lambda. Debe escribir una función controladora para manipular las variables del evento de entrada y devolver la respuesta. Para obtener más información sobre cómo AWS Lambda funciona, consulte la [invocación basada en eventos en la AWS Lambda Guía](https://docs.aws.amazon.com/lambda/latest/dg/lambda-services.html#event-driven-invocation) para desarrolladores.
**Topics**
+ [Evento de entrada de Lambda del analizador](#lambda-parser-input)
+ [Respuesta de Lambda del analizador](#lambda-parser-response)
+ [Ejemplos de Lambda del analizador](#lambda-parser-example)
## Evento de entrada de Lambda del analizador
A continuación presentamos la estructura general del evento de entrada del agente. Utilice los campos para escribir la función de controlador de 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"
}
```
En la siguiente lista se describen los campos del evento de entrada:
+ `messageVersion`: la versión del mensaje que identifica el formato de los datos de evento que se van a pasar a la función de Lambda y el formato que se espera en la respuesta de la función de Lambda. Los agentes de Amazon Bedrock solo admiten la versión 1.0.
+ `agent`: contiene información sobre el nombre, el ID, el alias y la versión del agente a la que pertenecen las peticiones.
+ `invokeModelRawResponse`: la salida del modelo fundacional sin procesar de la petición cuyo resultado se va a analizar.
+ `promptType`: el tipo de petición cuya salida se va a analizar.
+ `overrideType`: los artefactos que sobrescribe esta función de Lambda. Actualmente, solo se admite `OUTPUT_PARSER`, lo que indica que se debe sobrescribir el analizador predeterminado.
## Respuesta de Lambda del analizador
El agente espera una respuesta de la función de Lambda y la usa para realizar otras acciones o para que pueda devolver una respuesta al usuario. El agente ejecuta la siguiente acción recomendada por el modelo del agente. Las siguientes acciones se pueden ejecutar en orden de serie o en paralelo, según el modelo del agente y el momento en que se creó y preparó el agente.
Si ha creado y preparado el agente *antes del 4 de octubre de 2024* y utiliza los modelos Anthropic Claude 3 Sonnet o Anthropic Claude 3.5 Sonnet, la siguiente acción principal recomendada por el modelo del agente se ejecutará en orden de serie de forma predeterminada.
Si ha creado un agente nuevo o ha preparado uno existente *después del 10 de octubre de 2024* y su agente utiliza Anthropic Claude 3 Sonnet, Anthropic Claude 3.5 Sonnet o cualquier modelo non-Anthropic, las siguientes acciones recomendadas por el modelo del agente se ejecutarán en paralelo. Esto significa que varias acciones, por ejemplo, una mezcla de grupos de acciones, funciones y bases de conocimientos, se ejecutarán en paralelo. Esto reduce la cantidad de llamadas realizadas al modelo, lo que, a su vez, reduce la latencia general.
Puede habilitar acciones paralelas para sus agentes creadas y preparadas *antes del 4 de octubre de 2024* llamando a la [PrepareAgent](https://docs.aws.amazon.com//bedrock/latest/APIReference/API_agent_PrepareAgent.html)API o seleccionando **Prepare** en el generador de agentes de su agente en la consola. Cuando el agente esté preparado, verá una plantilla de peticiones actualizada y una nueva versión del esquema de Lambda del analizador.
**Ejemplo de respuesta de Lambda del analizador**
Los siguientes son ejemplos de la estructura general de la respuesta del agente que ejecuta las siguientes acciones recomendadas en orden de serie y del agente que ejecuta las siguientes acciones en paralelo. Utilice los campos de respuesta de la función de Lambda para configurar cómo se devuelve la salida.
**Ejemplo de respuesta de un agente que ejecuta las siguientes acciones recomendadas en orden de serie**
Seleccione la pestaña correspondiente según si ha definido el grupo de acciones con un esquema de OpenAPI o con los detalles de la función:
**nota**
`MessageVersion 1.0` indica que el agente está ejecutando las siguientes acciones principales recomendadas en orden de serie.
------
#### [ 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": {
"": {
"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": {
"": {
"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"
}]
}]
}
}
}
```
------
**Ejemplo de respuesta de un agente que ejecuta las siguientes acciones en paralelo**
Seleccione la pestaña correspondiente según si ha definido el grupo de acciones con un esquema de OpenAPI o con los detalles de la función:
**nota**
`MessageVersion 2.0` indica que el agente está ejecutando las siguientes acciones recomendadas en paralelo.
------
#### [ 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": {
"": {
"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": {
""": {
"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"
}]
}]
}
}
}
```
------
En la siguiente lista se describen los campos de respuesta de Lambda:
+ `messageVersion`: la versión del mensaje que identifica el formato de los datos del evento que se van a pasar a la función de Lambda y el formato previsto de la respuesta de una función de Lambda.
+ `promptType`: el tipo de petición del turno actual.
+ `preProcessingParsedResponse`: la respuesta analizada para el tipo de petición `PRE_PROCESSING`.
+ `orchestrationParsedResponse`: la respuesta analizada para el tipo de petición `ORCHESTRATION`. Consulte a continuación para obtener más detalles.
+ `knowledgeBaseResponseGenerationParsedResponse`: la respuesta analizada para el tipo de petición `KNOWLEDGE_BASE_RESPONSE_GENERATION`.
+ `postProcessingParsedResponse`: la respuesta analizada para el tipo de petición `POST_PROCESSING`.
Para obtener más información sobre las respuestas analizadas de las cuatro plantillas de peticiones, consulte las siguientes pestañas.
------
#### [ preProcessingParsedResponse ]
```
{
"isValidInput": "boolean",
"rationale": "string"
}
```
La `preProcessingParsedResponse` contiene los siguientes campos.
+ `isValidInput`: especifica si la entrada del usuario es válida o no. Puede definir la función para determinar cómo caracterizar la validez de la entrada del usuario.
+ `rationale`: el razonamiento de la categorización de las entradas de los usuarios. Este razonamiento lo proporciona el modelo en la respuesta sin procesar, la función de Lambda la analiza y el agente lo presenta en el seguimiento para su preprocesamiento.
------
#### [ orchestrationResponse ]
El formato de la `orchestrationResponse` depende de si ha definido el grupo de acciones con un esquema de OpenAPI o con detalles de la función:
+ Si ha definido el grupo de acciones con un esquema de OpenAPI, la respuesta debe tener el siguiente formato:
```
{
"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": {
"": {
"value": "string"
},
...
}
},
"agentKnowledgeBase": {
"knowledgeBaseId": "string",
"id": "string",
"searchQuery": {
"value": "string"
}
},
"agentFinalResponse": {
"responseText": "string",
"citations": {
"generatedResponseParts": [
{
"text": "string",
"references": [
{"sourceId": "string"},
...
]
},
...
]
}
},
}
}
```
+ Si ha definido el grupo de acciones con los detalles de la función, la respuesta debe tener el siguiente formato:
```
{
"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": {
"": {
"value": "string"
},
...
}
},
"agentKnowledgeBase": {
"knowledgeBaseId": "string",
"id": "string",
"searchQuery": {
"value": "string"
}
},
"agentFinalResponse": {
"responseText": "string",
"citations": {
"generatedResponseParts": [
{
"text": "string",
"references": [
{"sourceId": "string"},
...
]
},
...
]
}
},
}
}
```
La `orchestrationParsedResponse` contiene los siguientes campos:
+ `rationale`: el razonamiento sobre qué hacer a continuación, en función del resultado del modelo fundacional. Puede definir la función que se analizará a partir de la salida del modelo.
+ `parsingErrorDetails`: contiene la `repromptResponse`, que es el mensaje para volver a solicitar al modelo que actualice su respuesta sin procesar cuando la respuesta del modelo no se puede analizar. Puede definir la función para manipular la forma de volver a solicitar el modelo.
+ `responseDetails`: contiene los detalles sobre cómo gestionar la salida del modelo fundacional. Contiene un `invocationType`, que es el siguiente paso que debe realizar el agente, y un segundo campo que debe coincidir con el `invocationType`. Son posibles los siguientes objetos.
+ `agentAskUser`: compatible con el tipo de invocación `ASK_USER`. Este tipo de invocación finaliza el paso de orquestación. Contiene el `responseText` para solicitar más información al usuario. Puede definir su función para manipular este campo.
+ `actionGroupInvocation`: compatible con el tipo de invocación `ACTION_GROUP`. Puede definir su función de Lambda para determinar los grupos de acciones que se van a invocar y los parámetros que se van a transferir. Contiene los siguientes campos:
+ `actionGroupName`: el grupo de acciones que se va a invocar.
+ Los siguientes campos son obligatorios si ha definido el grupo de acciones con un esquema de OpenAPI:
+ `apiName`: es el nombre de la operación de la API que se debe invocar en el grupo de acciones.
+ `verb`: es el método de la operación de la API que se debe utilizar.
+ Los siguientes campos son obligatorios si ha definido el grupo de acciones con detalles de la función:
+ `functionName`: es el nombre de la función que se debe invocar en el grupo de acciones.
+ `actionGroupInput`: contiene parámetros a especificar en la solicitud de la operación de la API.
+ `agentKnowledgeBase`: compatible con el tipo de invocación `KNOWLEDGE_BASE`. Puede definir su función para determinar cómo consultar las bases de conocimiento. Contiene los siguientes campos:
+ `knowledgeBaseId`: el identificador único de la base de conocimientos.
+ `searchQuery`: contiene la consulta para enviar a la base de conocimientos en el campo `value`.
+ `agentFinalResponse`: compatible con el tipo de invocación `FINISH`. Este tipo de invocación finaliza el paso de orquestación. Contiene la respuesta al usuario en el campo `responseText` y las citas de la respuesta en el objeto `citations`.
------
#### [ knowledgeBaseResponseGenerationParsedResponse ]
```
{
"generatedResponse": {
"generatedResponseParts": [
{
"text": "string",
"references": [
{ "sourceId": "string" },
...
]
},
...
]
}
}
```
La `knowledgeBaseResponseGenerationParsedResponse` contiene el formulario `generatedResponse` de la consulta a la base de conocimientos y las referencias sobre el origen de datos.
------
#### [ postProcessingParsedResponse ]
```
{
"responseText": "string",
"citations": {
"generatedResponseParts": [
{
"text": "string",
"references": [
{ "sourceId": "string" },
...
]
},
...
]
}
}
```
La `postProcessingParsedResponse` contiene los siguientes campos:
+ `responseText`: la respuesta que se debe devolver al usuario final. Puede definir la función para formatear la respuesta.
+ `citations`: contiene una lista de citas de la respuesta. Cada cita muestra el texto citado y las referencias.
------
## Ejemplos de Lambda del analizador
Para ver un ejemplo de eventos de entrada de la función Lambda del analizador y las respuestas, seleccione una de las siguientes pestañas.
------
#### [ Pre-processing ]
**Ejemplo de evento de entrada**
```
{
"agent": {
"alias": "TSTALIASID",
"id": "AGENTID123",
"name": "InsuranceAgent",
"version": "DRAFT"
},
"invokeModelRawResponse": " \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\n\nB",
"messageVersion": "1.0",
"overrideType": "OUTPUT_PARSER",
"promptType": "PRE_PROCESSING"
}
```
**Ejemplo de respuesta**
```
{
"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 ]
**Ejemplo de evento de entrada**
```
{
"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\\n\\nGET::x_amz_knowledgebase_KBID123456::Search(searchQuery=\"What is the phone number I can call?\)",
"messageVersion": "1.0",
"overrideType": "OUTPUT_PARSER",
"promptType": "ORCHESTRATION"
}
```
**Ejemplo de respuesta**
```
{
"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 ]
**Ejemplo de evento de entrada**
```
{
"agent": {
"alias": "TSTALIASID",
"id": "AGENTID123",
"name": "InsuranceAgent",
"version": "DRAFT"
},
"invokeModelRawResponse": "{\"completion\":\" \\\\n\\\\n\\\\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\\\\n\\\\n1234567-1234-1234-1234-123456789abc\\\\n2345678-2345-2345-2345-23456789abcd\\\\n3456789-3456-3456-3456-3456789abcde\\\\n\\\\n\\\\n\",\"stop_reason\":\"stop_sequence\",\"stop\":\"\\\\n\\\\nHuman:\"}",
"messageVersion": "1.0",
"overrideType": "OUTPUT_PARSER",
"promptType": "KNOWLEDGE_BASE_RESPONSE_GENERATION"
}
```
**Ejemplo de respuesta**
```
{
"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 ]
**Ejemplo de evento de entrada**
```
{
"agent": {
"alias": "TSTALIASID",
"id": "AGENTID123",
"name": "InsuranceAgent",
"version": "DRAFT"
},
"invokeModelRawResponse": "\\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",
"messageVersion": "1.0",
"overrideType": "OUTPUT_PARSER",
"promptType": "POST_PROCESSING"
}
```
**Ejemplo de respuesta**
```
{
"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 ]
**Ejemplo de evento de entrada**
```
{
"messageVersion": "1.0",
"promptType": "MEMORY_SUMMARIZATION",
"invokeModelRawResponse": "User initiated the conversation with a greeting."
}
```
**Ejemplo de respuesta**
```
{"topicwiseSummaries": [
{
"topic": "TopicName1",
"summary": "My Topic 1 Summary"
}
...
]
}
```
------
Para ver ejemplos de funciones de Lambda del analizador, expanda la sección para ver los ejemplos de plantillas de peticiones que desee ver. La función `lambda_handler` devuelve la respuesta analizada al agente.
### Preprocesamiento
En el siguiente ejemplo se muestra una función de Lambda de analizador de preprocesamiento escrita en Python.
```
import json
import re
import logging
PRE_PROCESSING_RATIONALE_REGEX = "<thinking>(.*?)</thinking>"
PREPROCESSING_CATEGORY_REGEX = "<category>(.*?)</category>"
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
```
### Orquestación
En el siguiente ejemplo se muestra una función de Lambda de analizador de orquestación escrita en Python.
El código de ejemplo varía en función de si el grupo de acciones se ha definido con un esquema de OpenAPI o con detalles de una función:
1. Para ver ejemplos de un grupo de acciones definido con un esquema de OpenAPI, seleccione la pestaña correspondiente al modelo del que desee ver los ejemplos.
------
#### [ Anthropic Claude 2.0 ]
```
import json
import re
import logging
RATIONALE_REGEX_LIST = [
"(.*?)()",
"(.*?)()"
]
RATIONALE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_REGEX_LIST]
RATIONALE_VALUE_REGEX_LIST = [
"(.*?)()",
"(.*?)()",
"()(.*?)"
]
RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_VALUE_REGEX_LIST]
ANSWER_REGEX = r"(?<=)(.*)"
ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
ANSWER_TAG = ""
FUNCTION_CALL_TAG = ""
ASK_USER_FUNCTION_CALL_REGEX = r"(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"(\w+)::(\w+)::(.+)\((.+)\)"
ANSWER_PART_REGEX = "(.+?)"
ANSWER_TEXT_PART_REGEX = "(.+?)"
ANSWER_REFERENCE_PART_REGEX = "(.+?)"
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: user::askuser(askuser=\"$ASK_USER_INPUT\")."
FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = 'The function call format is incorrect. The format for function calls must be: $FUNCTION_NAME($FUNCTION_ARGUMENT_NAME=""$FUNCTION_ARGUMENT_NAME"").'
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 = [
"(.*?)()",
"(.*?)()"
]
RATIONALE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_REGEX_LIST]
RATIONALE_VALUE_REGEX_LIST = [
"(.*?)()",
"(.*?)()",
"()(.*?)"
]
RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_VALUE_REGEX_LIST]
ANSWER_REGEX = r"(?<=)(.*)"
ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
ANSWER_TAG = ""
FUNCTION_CALL_TAG = ""
ASK_USER_FUNCTION_CALL_REGEX = r"user::askuser"
ASK_USER_FUNCTION_CALL_PATTERN = re.compile(ASK_USER_FUNCTION_CALL_REGEX, re.DOTALL)
ASK_USER_TOOL_NAME_REGEX = r"((.|\n)*?)"
ASK_USER_TOOL_NAME_PATTERN = re.compile(ASK_USER_TOOL_NAME_REGEX, re.DOTALL)
TOOL_PARAMETERS_REGEX = r"((.|\n)*?)"
TOOL_PARAMETERS_PATTERN = re.compile(TOOL_PARAMETERS_REGEX, re.DOTALL)
ASK_USER_TOOL_PARAMETER_REGEX = r"((.|\n)*?)"
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"(?<=)(.*)"
ANSWER_PART_REGEX = "(.+?)"
ANSWER_TEXT_PART_REGEX = "(.+?)"
ANSWER_REFERENCE_PART_REGEX = "(.+?)"
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: user::askuser$QUESTION."
FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls must be: $TOOL_NAME <$PARAMETER_NAME>$PARAMETER_VALUE...."
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("{}".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 = [
"(.*?)()",
"(.*?)()"
]
RATIONALE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_REGEX_LIST]
RATIONALE_VALUE_REGEX_LIST = [
"(.*?)()",
"(.*?)()",
"()(.*?)"
]
RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_VALUE_REGEX_LIST]
ANSWER_REGEX = r"(?<=)(.*)"
ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
ANSWER_TAG = ""
FUNCTION_CALL_TAG = ""
ASK_USER_FUNCTION_CALL_REGEX = r"user::askuser"
ASK_USER_FUNCTION_CALL_PATTERN = re.compile(ASK_USER_FUNCTION_CALL_REGEX, re.DOTALL)
ASK_USER_TOOL_NAME_REGEX = r"((.|\n)*?)"
ASK_USER_TOOL_NAME_PATTERN = re.compile(ASK_USER_TOOL_NAME_REGEX, re.DOTALL)
TOOL_PARAMETERS_REGEX = r"((.|\n)*?)"
TOOL_PARAMETERS_PATTERN = re.compile(TOOL_PARAMETERS_REGEX, re.DOTALL)
ASK_USER_TOOL_PARAMETER_REGEX = r"((.|\n)*?)"
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"(?<=)(.*)"
ANSWER_PART_REGEX = "(.+?)"
ANSWER_TEXT_PART_REGEX = "(.+?)"
ANSWER_REFERENCE_PART_REGEX = "(.+?)"
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: user::askuser$QUESTION."
FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls must be: $TOOL_NAME <$PARAMETER_NAME>$PARAMETER_VALUE...."
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("{}".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 = [
"(.*?)()",
"(.*?)()",
"()(.*?)"
]
RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in
RATIONALE_VALUE_REGEX_LIST]
ANSWER_REGEX = r"(?<=)(.*)"
ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
ANSWER_TAG = ""
ASK_USER = "user__askuser"
KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX = "x_amz_knowledgebase_"
ANSWER_PART_REGEX = "(.+?)"
ANSWER_TEXT_PART_REGEX = "(.+?)"
ANSWER_REFERENCE_PART_REGEX = "(.+?)"
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. Para ver ejemplos de un grupo de acciones definido con detalles de la función, seleccione la pestaña correspondiente al modelo del que desee ver los ejemplos.
------
#### [ Anthropic Claude 2.0 ]
```
import json
import re
import logging
RATIONALE_REGEX_LIST = [
"(.*?)()",
"(.*?)()"
]
RATIONALE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_REGEX_LIST]
RATIONALE_VALUE_REGEX_LIST = [
"(.*?)()",
"(.*?)()",
"()(.*?)"
]
RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_VALUE_REGEX_LIST]
ANSWER_REGEX = r"(?<=)(.*)"
ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
ANSWER_TAG = ""
FUNCTION_CALL_TAG = ""
ASK_USER_FUNCTION_CALL_REGEX = r"(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"(\w+)::(\w+)::(.+)\((.+)\)"
FUNCTION_CALL_REGEX_FUNCTION_SCHEMA = r"(\w+)::(.+)\((.+)\)"
ANSWER_PART_REGEX = "(.+?)"
ANSWER_TEXT_PART_REGEX = "(.+?)"
ANSWER_REFERENCE_PART_REGEX = "(.+?)"
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: user::askuser(askuser=\"$ASK_USER_INPUT\")."
FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = 'The function call format is incorrect. The format for function calls must be: $FUNCTION_NAME($FUNCTION_ARGUMENT_NAME=""$FUNCTION_ARGUMENT_NAME"").'
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 = [
"(.*?)()",
"(.*?)()"
]
RATIONALE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_REGEX_LIST]
RATIONALE_VALUE_REGEX_LIST = [
"(.*?)()",
"(.*?)()",
"()(.*?)"
]
RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_VALUE_REGEX_LIST]
ANSWER_REGEX = r"(?<=)(.*)"
ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
ANSWER_TAG = ""
FUNCTION_CALL_TAG = ""
ASK_USER_FUNCTION_CALL_REGEX = r"user::askuser"
ASK_USER_FUNCTION_CALL_PATTERN = re.compile(ASK_USER_FUNCTION_CALL_REGEX, re.DOTALL)
ASK_USER_TOOL_NAME_REGEX = r"((.|\n)*?)"
ASK_USER_TOOL_NAME_PATTERN = re.compile(ASK_USER_TOOL_NAME_REGEX, re.DOTALL)
TOOL_PARAMETERS_REGEX = r"((.|\n)*?)"
TOOL_PARAMETERS_PATTERN = re.compile(TOOL_PARAMETERS_REGEX, re.DOTALL)
ASK_USER_TOOL_PARAMETER_REGEX = r"((.|\n)*?)"
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"(?<=)(.*)"
ANSWER_PART_REGEX = "(.+?)"
ANSWER_TEXT_PART_REGEX = "(.+?)"
ANSWER_REFERENCE_PART_REGEX = "(.+?)"
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: user::askuser$QUESTION."
FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls must be: $TOOL_NAME <$PARAMETER_NAME>$PARAMETER_VALUE...."
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("{}".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 = [
"(.*?)()",
"(.*?)()"
]
RATIONALE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_REGEX_LIST]
RATIONALE_VALUE_REGEX_LIST = [
"(.*?)()",
"(.*?)()",
"()(.*?)"
]
RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in RATIONALE_VALUE_REGEX_LIST]
ANSWER_REGEX = r"(?<=)(.*)"
ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
ANSWER_TAG = ""
FUNCTION_CALL_TAG = ""
ASK_USER_FUNCTION_CALL_REGEX = r"user::askuser"
ASK_USER_FUNCTION_CALL_PATTERN = re.compile(ASK_USER_FUNCTION_CALL_REGEX, re.DOTALL)
ASK_USER_TOOL_NAME_REGEX = r"((.|\n)*?)"
ASK_USER_TOOL_NAME_PATTERN = re.compile(ASK_USER_TOOL_NAME_REGEX, re.DOTALL)
TOOL_PARAMETERS_REGEX = r"((.|\n)*?)"
TOOL_PARAMETERS_PATTERN = re.compile(TOOL_PARAMETERS_REGEX, re.DOTALL)
ASK_USER_TOOL_PARAMETER_REGEX = r"((.|\n)*?)"
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"(?<=)(.*)"
ANSWER_PART_REGEX = "(.+?)"
ANSWER_TEXT_PART_REGEX = "(.+?)"
ANSWER_REFERENCE_PART_REGEX = "(.+?)"
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: user::askuser$QUESTION."
FUNCTION_CALL_STRUCTURE_REPROMPT_MESSAGE = "The function call format is incorrect. The format for function calls must be: $TOOL_NAME <$PARAMETER_NAME>$PARAMETER_VALUE...."
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("{}".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 = [
"(.*?)()",
"(.*?)()",
"()(.*?)"
]
RATIONALE_VALUE_PATTERNS = [re.compile(regex, re.DOTALL) for regex in
RATIONALE_VALUE_REGEX_LIST]
ANSWER_REGEX = r"(?<=)(.*)"
ANSWER_PATTERN = re.compile(ANSWER_REGEX, re.DOTALL)
ANSWER_TAG = ""
ASK_USER = "user__askuser"
KNOWLEDGE_STORE_SEARCH_ACTION_PREFIX = "x_amz_knowledgebase_"
ANSWER_PART_REGEX = "(.+?)"
ANSWER_TEXT_PART_REGEX = "(.+?)"
ANSWER_REFERENCE_PART_REGEX = "(.+?)"
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
}
```
------
### Generación de respuestas en la base de conocimientos
El siguiente ejemplo muestra una función de Lambda del analizador de generación de respuestas de la base de conocimientos escrita en Python.
```
import json
import re
import logging
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)
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
```
### Posprocesamiento
En el siguiente ejemplo se muestra una función de Lambda de analizador de preprocesamiento escrita en Python.
```
import json
import re
import logging
FINAL_RESPONSE_REGEX = r"<final_response>([\s\S]*?)</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
```
### Resumen de memoria
En el siguiente ejemplo se muestra una función de Lambda de analizador de resumen de memoria escrita en Python.
```
import re
import logging
SUMMARY_TAG_PATTERN = r'(.*?)'
TOPIC_TAG_PATTERN = r'(.+?)'
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
```