Setting the file. One moment.
Model Recommendation · Agent Advisor · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page Add Capabilities
81
Creating Amazon Aurora Db Cluster With Instances
104
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
scripts/ model_recommendation.py
Python · 207 lines · 8 KB
15
import
openai_model_recommendation
16
17
18 SKILL_DIR = pathlib.Path( __file__ ).parent.parent
19 MODELS_DIR = SKILL_DIR / "references" / "models"
20 DEFAULT_CATALOG = MODELS_DIR / "anthropic-bedrock-2026-07-21.json"
21 OPENAI_CATALOG = MODELS_DIR / "openai-bedrock-2026-08-21.json"
22
23 # Which provider module owns a source. This is a TWO-way decision, not a per-provider table: OpenAI
24 # has its own module, and everything else goes to the Anthropic one — including `none`/`unknown` (no
25 # detected provider means the Bedrock-native pool) and the providers with no module yet
26 # (azure_openai, google_genai, bedrock). Routing those to the Anthropic module is deliberate rather
27 # than a silent default: that module classifies the source against its own ANTHROPIC_POOL and attaches
28 # a `provider_module_pending` [BLOCKS] finding to anything outside it, so the recommendation is
29 # produced but stays provisional. Enumerating the fall-through providers here would duplicate that
30 # classification in a second place and let the two drift.
31 def _provider_module (provider):
32 return "openai" if provider == "openai" else "anthropic"
33
34
35
36 def load_catalog (path = DEFAULT_CATALOG ):
37 catalog_path = pathlib.Path(path)
38 try :
39 catalog = json.loads(catalog_path.read_text())
40 except json.JSONDecodeError as exc:
41 raise ValueError ( f " { catalog_path } : invalid JSON ( { exc } )" ) from exc
42 required = {
43 "schema_version" ,
44 "provider" ,
45 "verified_at" ,
46 "verified_region" ,
47 "paths" ,
48 "models" ,
49 }
50 missing = sorted (required - set (catalog))
51 if missing:
52 raise ValueError ( f " { catalog_path } : missing required keys { missing } " )
53 for model_key, model in catalog[ "models" ].items():
54 for field in (
55 "display_name" ,
56 "family" ,
57 "version" ,
58 "context_window" ,
59 "output_token_ceiling" ,
60 "capabilities" ,
61 "paths" ,
62 ):
63 if field not in model:
64 raise ValueError ( f " { catalog_path } : { model_key } missing { field } " )
65 return catalog
66
67
68 def load_openai_catalog (path = OPENAI_CATALOG ):
69 """Load and validate the dated OpenAI/Bedrock path catalog.
70
71 Unlike the Anthropic catalog, numeric limits may be the string "unknown"
72 (the reference proves paths, not context/output ceilings), and each model
73 carries a `generation` and per-path `evidence` citation.
74 """
75 catalog_path = pathlib.Path(path)
76 try :
77 catalog = json.loads(catalog_path.read_text())
78 except json.JSONDecodeError as exc:
79 raise ValueError ( f " { catalog_path } : invalid JSON ( { exc } )" ) from exc
80 required = {
81 "schema_version" ,
82 "provider" ,
83 "verified_at" ,
84 "verified_region" ,
85 "paths" ,
86 "models" ,
87 }
88 missing = sorted (required - set (catalog))
89 if missing:
90 raise ValueError ( f " { catalog_path } : missing required keys { missing } " )
91 for model_key, model in catalog[ "models" ].items():
92 for field in (
93 "display_name" ,
94 "family" ,
95 "generation" ,
96 "version" ,
97 "context_window" ,
98 "output_token_ceiling" ,
99 "capabilities" ,
100 "paths" ,
101 ):
102 if field not in model:
103 raise ValueError ( f " { catalog_path } : { model_key } missing { field } " )
104 for limit in ( "context_window" , "output_token_ceiling" ):
105 value = model[limit]
106 if value != "unknown" and not ( isinstance (value, int ) and value > 0 ):
107 raise ValueError (
108 f " { catalog_path } : { model_key } { limit } must be a positive int or "
109 f '"unknown", got { value !r} '
110 )
111 return catalog
112
113
114
115
116 def _catalog_provenance (catalog):
117 return {
118 "provider" : catalog[ "provider" ],
119 "verified_at" : catalog[ "verified_at" ],
120 "verified_region" : catalog[ "verified_region" ],
121 "source" : catalog.get( "source" ),
122 }
123
124
125 def recommend (input_data, catalog = None , openai_catalog = None ):
126 """Dispatch each workload to its provider module.
127
128 Anthropic (and none/unknown/generic) use `catalog`; OpenAI uses
129 `openai_catalog`. Every workload records the provenance of the catalog that
130 produced it so a mixed-provider run never mislabels a source.
131 """
132 if catalog is None :
133 catalog = load_catalog()
134 workloads = {}
135 provenance = {}
136 catalogs_used = {}
137 for workload in input_data[ "workloads" ]:
138 workload_id = workload[ "workload_id" ]
139 if workload_id in workloads:
140 raise ValueError ( f "duplicate workload_id: { workload_id } " )
141 provider = workload[ "source" ][ "provider" ]
142 module = _provider_module(provider)
143 if module == "openai" :
144 if openai_catalog is None :
145 openai_catalog = load_openai_catalog()
146 workloads[workload_id] = openai_model_recommendation.recommend_openai_workload(
147 workload, input_data[ "region" ], openai_catalog
148 )
149 provenance[workload_id] = _catalog_provenance(openai_catalog)
150 catalogs_used[ "openai" ] = openai_catalog
151 else :
152 workloads[workload_id] = anthropic_model_recommendation.recommend_anthropic_workload(
153 workload, input_data[ "region" ], catalog
154 )
155 provenance[workload_id] = _catalog_provenance(catalog)
156 catalogs_used[ "anthropic" ] = catalog
157 primary_unit = input_data[ "primary_unit" ]
158 if primary_unit not in workloads:
159 raise ValueError ( f "primary_unit not found in workloads: { primary_unit } " )
160 # Backward-compatible top-level `catalog`: the primary unit's catalog, plus
161 # explicit per-workload provenance so a mixed run keeps correct attribution.
162 return {
163 "schema_version" : 2 ,
164 "catalog" : provenance[primary_unit],
165 "catalog_provenance" : provenance,
166 "primary_unit" : primary_unit,
167 "workloads" : workloads,
168 }
169
170
171 def main (argv = None ):
172 parser = argparse.ArgumentParser(
173 description = "agent-advisor Bedrock model recommendation"
174 )
175 parser.add_argument( "input" , type = pathlib.Path)
176 parser.add_argument(
177 "--output" ,
178 type = pathlib.Path,
179 help = "defaults to model-recommendation.json beside the input" ,
180 )
181 parser.add_argument( "--catalog" , type = pathlib.Path, default = DEFAULT_CATALOG )
182 args = parser.parse_args(argv)
183
184 try :
185 import jsonschema
186 except ImportError :
187 jsonschema = None
188
189 def validate (instance, schema_name):
190 if jsonschema is None :
191 return
192 schemas = pathlib.Path( __file__ ).parent / "schemas"
193 jsonschema.validate(instance, json.loads((schemas / schema_name).read_text()))
194
195 input_data = json.loads(args.input.read_text())
196 validate(input_data, "model-recommendation-input.json" )
197 result = recommend(input_data, load_catalog(args.catalog))
198 validate(result, "model-recommendation.json" )
199 output = args.output or args.input.parent / "model-recommendation.json"
200 output.write_text(json.dumps(result, indent = 2 ) + " \n " )
201 validated = "no" if jsonschema is None else "yes"
202 print ( f "RESULT=ok WORKLOADS= { len (result[ 'workloads' ]) } SCHEMA_VALIDATED= { validated } " )
203 return 0
204
205
206 if __name__ == "__main__" :
207 raise SystemExit (main())