|
1 | 1 | import os |
2 | 2 | import boto3 |
3 | 3 | from aws_lambda_powertools import Logger |
4 | | -from botocore.exceptions import ClientError |
5 | 4 |
|
6 | 5 | logger = Logger(service="promptLoader") |
7 | 6 |
|
8 | 7 |
|
9 | | -def _test_list_prompts(client): |
10 | | - """Test listing prompts.""" |
11 | | - try: |
12 | | - response = client.list_prompts(maxResults=10) |
13 | | - logger.info(f"✅ ListPrompts SUCCEEDED: Found {len(response.get('promptSummaries', []))} prompts") |
14 | | - for prompt in response.get("promptSummaries", []): |
15 | | - logger.info(f" - Prompt: {prompt.get('name')} (ID: {prompt.get('id')})") |
16 | | - except ClientError as e: |
17 | | - logger.error(f"❌ ListPrompts FAILED: {e}") |
18 | | - |
19 | | - |
20 | | -def _test_get_prompt(client, prompt_name): |
21 | | - """Test getting a specific prompt.""" |
22 | | - try: |
23 | | - response = client.get_prompt(promptIdentifier=prompt_name, promptVersion="$LATEST") |
24 | | - logger.info(f"✅ GetPrompt SUCCEEDED for {prompt_name}: {len(response.get('variants', []))} variants") |
25 | | - except ClientError as e: |
26 | | - error_code = e.response.get("Error", {}).get("Code", "Unknown") |
27 | | - error_message = e.response.get("Error", {}).get("Message", str(e)) |
28 | | - logger.error(f"❌ GetPrompt FAILED for {prompt_name}: {error_code} - {error_message}") |
29 | | - _test_alternative_prompts(client) |
30 | | - |
31 | | - |
32 | | -def _test_alternative_prompts(client): |
33 | | - """Test getting prompts with different identifiers.""" |
34 | | - list_response = client.list_prompts(maxResults=10) |
35 | | - for prompt in list_response.get("promptSummaries", []): |
36 | | - try: |
37 | | - prompt_id = prompt.get("id") |
38 | | - test_response = client.get_prompt(promptIdentifier=prompt_id, promptVersion="$LATEST") |
39 | | - logger.info( |
40 | | - f"✅ GetPrompt SUCCEEDED for prompt ID {prompt_id}: {len(test_response.get('variants', []))} variants" |
41 | | - ) |
42 | | - break |
43 | | - except ClientError as e2: |
44 | | - logger.error(f"❌ GetPrompt FAILED for prompt ID {prompt_id}: {e2}") |
45 | | - |
46 | | - |
47 | | -def _analyze_iam_permissions(): |
48 | | - """Analyze IAM permissions for the current role.""" |
49 | | - try: |
50 | | - sts = boto3.client("sts") |
51 | | - identity = sts.get_caller_identity() |
52 | | - logger.info(f"Lambda running as: {identity.get('Arn')}") |
53 | | - |
54 | | - role_arn = identity.get("Arn") |
55 | | - if "assumed-role" in role_arn: |
56 | | - role_name = role_arn.split("/")[-2] |
57 | | - iam = boto3.client("iam") |
58 | | - |
59 | | - try: |
60 | | - attached_policies = iam.list_attached_role_policies(RoleName=role_name) |
61 | | - logger.info("Attached managed policies:") |
62 | | - for policy in attached_policies.get("AttachedPolicies", []): |
63 | | - logger.info(f" - {policy.get('PolicyName')} ({policy.get('PolicyArn')})") |
64 | | - except Exception as e: |
65 | | - logger.warning(f"Could not list attached policies: {e}") |
66 | | - except Exception as e: |
67 | | - logger.warning(f"Could not analyze IAM permissions: {e}") |
68 | | - |
69 | | - |
70 | | -def debug_bedrock_permissions(): |
71 | | - """Debug Bedrock operations and permissions.""" |
72 | | - try: |
73 | | - logger.info(f"AWS_REGION: {os.environ.get('AWS_REGION')}") |
74 | | - logger.info(f"QUERY_REFORMULATION_PROMPT_NAME: {os.environ.get('QUERY_REFORMULATION_PROMPT_NAME')}") |
75 | | - |
76 | | - client = boto3.client("bedrock-agent", region_name=os.environ["AWS_REGION"]) |
77 | | - logger.info("Testing Bedrock Agent permissions...") |
78 | | - |
79 | | - _test_list_prompts(client) |
80 | | - prompt_name = os.environ.get("QUERY_REFORMULATION_PROMPT_NAME", "epsam-pr-27-queryReformulation") |
81 | | - _test_get_prompt(client, prompt_name) |
82 | | - _analyze_iam_permissions() |
83 | | - |
84 | | - except Exception as e: |
85 | | - logger.error(f"Debug permissions test failed: {e}") |
86 | | - |
87 | | - |
88 | 8 | def load_prompt(prompt_name: str, version: str = "$LATEST") -> str: |
89 | 9 | """ |
90 | 10 | Load a prompt template from Amazon Bedrock Prompt Management. |
91 | | - Includes debugging to help understand permission issues. |
92 | 11 | """ |
93 | | - |
94 | | - # Run debug test on first call |
95 | | - if not hasattr(load_prompt, "_debug_run"): |
96 | | - debug_bedrock_permissions() |
97 | | - load_prompt._debug_run = True |
98 | | - |
99 | 12 | try: |
100 | 13 | client = boto3.client("bedrock-agent", region_name=os.environ["AWS_REGION"]) |
101 | | - response = client.get_prompt(promptIdentifier=prompt_name, promptVersion=version) |
102 | | - prompt_text = response["variants"][0]["templateConfiguration"]["text"]["text"] |
103 | 14 |
|
104 | | - logger.info(f"Successfully loaded prompt '{prompt_name}' version '{version}'") |
105 | | - return prompt_text |
106 | | - |
107 | | - except ClientError as e: |
108 | | - error_code = e.response.get("Error", {}).get("Code", "Unknown") |
109 | | - error_message = e.response.get("Error", {}).get("Message", str(e)) |
110 | | - |
111 | | - logger.error(f"Failed to load prompt '{prompt_name}': {error_code} - {error_message}") |
112 | | - logger.info("This appears to be the known AWS Bedrock Prompt Management issue affecting multiple users") |
| 15 | + response = client.get_prompt(promptIdentifier=prompt_name, promptVersion=version) |
113 | 16 |
|
114 | | - raise Exception(f"Failed to load prompt '{prompt_name}': {error_code} - {error_message}") |
| 17 | + return response["variants"][0]["templateConfiguration"]["text"]["text"] |
115 | 18 |
|
116 | 19 | except Exception as e: |
117 | | - logger.error(f"Unexpected error loading prompt '{prompt_name}': {e}") |
118 | | - raise Exception(f"Unexpected error loading prompt '{prompt_name}': {e}") |
| 20 | + logger.error(f"Error loading prompt {prompt_name}: {e}") |
| 21 | + raise |
0 commit comments