Setting the file. One moment. Add Capabilities
81
Creating Amazon Aurora Db Cluster With Instances
104
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
def verify_recommendation
— line 171
This file
- Number
- 23.81
- Position
- 81 of 81
- Type
- Python
- Size
- 8 KB
- Lines
- 249
scripts/verify_model_path.py
Python·249 lines·8 KB
14 return datetime.datetime.now(datetime.timezone.utc).isoformat()
15
16
17def _default_mantle_client(region):
18 try:
19 from anthropic import AnthropicBedrockMantle
20 except ImportError as exc:
21 raise RuntimeError(
22 "Mantle verification requires the anthropic package with "
23 "AnthropicBedrockMantle support"
24 ) from exc
25 return AnthropicBedrockMantle(aws_region=region)
26
27
28def _default_runtime_client(region):
29 try:
30 import boto3
31 except ImportError as exc:
32 raise RuntimeError("Runtime verification requires boto3") from exc
33 return boto3.client("bedrock-runtime", region_name=region)
34
35
36def _default_openai_responses_client(region):
37 """Lazily build an OpenAI SDK client pointed at the Bedrock Mantle endpoint.
38
39 Imports are deferred so the offline recommendation/verify code path never
40 requires the openai or token-generator packages.
41 """
42 try:
43 from openai import OpenAI
44 except ImportError as exc:
45 raise RuntimeError(
46 "Mantle Responses verification requires the openai package"
47 ) from exc
48 try:
49 from aws_bedrock_token_generator import provide_token
50 except ImportError as exc:
51 raise RuntimeError(
52 "Mantle Responses verification requires aws-bedrock-token-generator"
53 ) from exc
54 base_url = f"https://bedrock-mantle.{region}.api.aws/openai/v1"
55 return OpenAI(base_url=base_url, api_key=provide_token(region=region))
56
57
58def _probe_openai_responses(client, model_id):
59 return client.responses.create(model=model_id, input=PROMPT)
60
61
62def _response_model_id(response):
63 if isinstance(response, dict):
64 return response.get("model") or response.get("modelId")
65 return getattr(response, "model", None)
66
67
68def _probe_mantle(client, model_id):
69 return client.messages.create(
70 model=model_id,
71 max_tokens=8,
72 messages=[{"role": "user", "content": PROMPT}],
73 )
74
75
76def _probe_converse(client, model_id):
77 return client.converse(
78 modelId=model_id,
79 messages=[{"role": "user", "content": [{"text": PROMPT}]}],
80 inferenceConfig={"maxTokens": 8},
81 )
82
83
84def _probe_invoke(client, model_id):
85 body = {
86 "anthropic_version": "bedrock-2023-05-31",
87 "max_tokens": 8,
88 "messages": [{"role": "user", "content": PROMPT}],
89 }
90 return client.invoke_model(
91 modelId=model_id,
92 contentType="application/json",
93 accept="application/json",
94 body=json.dumps(body).encode("utf-8"),
95 )
96
97
98def _base_result(workload_id, recommendation):
99 identity = recommendation.get("model_identity") or {}
100 verification = recommendation["verification"]
101 return {
102 "workload_id": workload_id,
103 "decision_status": recommendation["decision_status"],
104 "api_path": recommendation.get("api_path"),
105 "path_model_id": identity.get("path_model_id"),
106 "invocation_model_id": recommendation.get("invocation_model_id"),
107 "region": verification["region"],
108 "status": "not_run",
109 "checked_at": None,
110 "response_model_id": None,
111 "error": None,
112 }
113
114
115def verify_workload(
116 workload_id,
117 recommendation,
118 mantle_client_factory=None,
119 runtime_client_factory=None,
120 openai_responses_client_factory=None,
121 now=None,
122):
123 result = _base_result(workload_id, recommendation)
124 if recommendation["decision_status"] != "recommended":
125 result["status"] = "not_applicable"
126 return result
127
128 checked_at = now or _utc_now()
129 result["checked_at"] = checked_at
130 model_id = result["invocation_model_id"]
131 if not model_id:
132 result["status"] = "needs_resolution"
133 result["error"] = {
134 "type": "UnresolvedInvocationModelId",
135 "message": (
136 "Resolve an account-invocable inference profile ID before probing; "
137 "the verifier will not substitute a model ID."
138 ),
139 }
140 return result
141
142 path = result["api_path"]
143 try:
144 if path == "mantle_messages":
145 factory = mantle_client_factory or _default_mantle_client
146 response = _probe_mantle(factory(result["region"]), model_id)
147 elif path == "mantle_openai_responses":
148 factory = openai_responses_client_factory or _default_openai_responses_client
149 response = _probe_openai_responses(factory(result["region"]), model_id)
150 elif path == "runtime_converse":
151 factory = runtime_client_factory or _default_runtime_client
152 response = _probe_converse(factory(result["region"]), model_id)
153 elif path == "runtime_invoke":
154 factory = runtime_client_factory or _default_runtime_client
155 response = _probe_invoke(factory(result["region"]), model_id)
156 else:
157 raise ValueError(f"verification is not implemented for API path: {path}")
158 except Exception as exc:
159 result["status"] = "failed"
160 result["error"] = {
161 "type": type(exc).__name__,
162 "message": str(exc),
163 }
164 return result
165
166 result["status"] = "passed"
167 result["response_model_id"] = _response_model_id(response)
168 return result
169
170
171def verify_recommendation(
172 recommendation,
173 workload_ids=None,
174 mantle_client_factory=None,
175 runtime_client_factory=None,
176 openai_responses_client_factory=None,
177 now=None,
178):
179 selected = set(workload_ids or recommendation["workloads"])
180 unknown = sorted(selected - set(recommendation["workloads"]))
181 if unknown:
182 raise ValueError(f"workload not found in recommendation: {', '.join(unknown)}")
183
184 generated_at = now or _utc_now()
185 workloads = {}
186 for workload_id, workload in recommendation["workloads"].items():
187 if workload_id not in selected:
188 continue
189 workloads[workload_id] = verify_workload(
190 workload_id,
191 workload,
192 mantle_client_factory=mantle_client_factory,
193 runtime_client_factory=runtime_client_factory,
194 openai_responses_client_factory=openai_responses_client_factory,
195 now=generated_at,
196 )
197 return {
198 "schema_version": 1,
199 "recommendation_schema_version": recommendation["schema_version"],
200 "generated_at": generated_at,
201 "workloads": workloads,
202 }
203
204
205def main(argv=None):
206 parser = argparse.ArgumentParser(
207 description="probe agent-advisor Bedrock model/path recommendations"
208 )
209 parser.add_argument("recommendation", type=pathlib.Path)
210 parser.add_argument(
211 "--output",
212 type=pathlib.Path,
213 help="defaults to model-verification.json beside the recommendation",
214 )
215 parser.add_argument(
216 "--workload",
217 action="append",
218 dest="workloads",
219 help="probe only this workload id; repeat to select more than one",
220 )
221 args = parser.parse_args(argv)
222
223 import jsonschema
224
225 recommendation = json.loads(args.recommendation.read_text())
226 jsonschema.validate(
227 recommendation,
228 json.loads((SCHEMAS / "model-recommendation.json").read_text()),
229 )
230 result = verify_recommendation(recommendation, workload_ids=args.workloads)
231 jsonschema.validate(
232 result,
233 json.loads((SCHEMAS / "model-verification.json").read_text()),
234 format_checker=jsonschema.FormatChecker(),
235 )
236 output = args.output or args.recommendation.parent / "model-verification.json"
237 output.write_text(json.dumps(result, indent=2) + "\n")
238
239 statuses = {item["status"] for item in result["workloads"].values()}
240 failed = statuses.intersection({"failed", "needs_resolution"})
241 print(
242 f"RESULT={'failed' if failed else 'ok'} "
243 f"WORKLOADS={len(result['workloads'])}"
244 )
245 return 2 if failed else 0
246
247
248if __name__ == "__main__":
249 raise SystemExit(main())