Setting the file. One moment.
Embedding · Foundry Iq · microsoft/azure-skills · Skills Docs
Back to the top of the page File Cu Canary
helpers/ _embedding.py
Python · 128 lines · 7 KB
12 from . import _search_read
13 except ImportError :
14 from _common import HelperFailure, reject_secrets, require_allowed_fields, model_definition
15 import _search_read
16
17
18 MODELS = { "text-embedding-ada-002" : ( 1536 , 1536 ),
19 "text-embedding-3-small" : ( 1 , 1536 ), "text-embedding-3-large" : ( 1 , 3072 )}
20
21
22 def fail (code: str , message: str ) -> HelperFailure:
23 return HelperFailure(code, message, blocked_at = "vector-verification" )
24
25
26 def _text (value: Any, * , maximum: int = 4096 ) -> bool :
27 return isinstance (value, str ) and bool (value.strip()) and len (value) <= maximum
28
29
30 def _json_valid (value: Any) -> None :
31 try :
32 json.dumps(value, ensure_ascii = False , allow_nan = False ).encode( "utf-8" )
33 except ( UnicodeError , TypeError , ValueError ) as exc:
34 raise fail( "input-schema-invalid" , "Evidence must be finite, valid UTF-8 JSON." ) from exc
35
36
37 def validate_choice (value: Any, * , enabled: bool , api_version: str ) -> dict[ str , Any] | None :
38 if not enabled:
39 if value is not None :
40 raise fail( "embedding-choice-conflict" , "Lexical planning must omit embedding configuration." )
41 return None
42 if not isinstance (value, dict ):
43 raise fail( "embedding-prerequisite-missing" , "Vector planning requires resolved embedding choices." )
44 reject_secrets(value)
45 require_allowed_fields(value, {
46 "endpoint" , "deployment" , "model" , "dimensions" , "model_version" , "auth" , "prerequisites" ,
47 }, label = "embedding choices" )
48 if api_version not in { "2026-04-01" , "2026-08-01-preview" }:
49 raise fail( "embedding-version-unsupported" , "Unsupported Search embedding API version." )
50 try :
51 json.dumps(value, ensure_ascii = False , allow_nan = False ).encode( "utf-8" )
52 uri = urlsplit(value.get( "endpoint" ) if isinstance (value.get( "endpoint" ), str ) else "" )
53 valid_endpoint = (
54 uri.scheme == "https" and uri.port is None and not uri.username and not uri.password
55 and not uri.query and not uri.fragment and uri.path in { "" , "/" }
56 and re.fullmatch(
57 r " [ a-z0-9 ][ a-z0-9- ] {0,62} \. ( openai \. azure \. com | services \. ai \. azure \. com | cognitiveservices \. azure \. com ) " ,
58 uri.netloc,
59 )
60 )
61 except ( UnicodeError , ValueError , TypeError ) as exc:
62 raise fail( "embedding-choice-invalid" , "Embedding choices must be valid UTF-8 and an exact HTTPS endpoint." ) from exc
63 if (
64 not valid_endpoint or not _text(value.get( "deployment" ), maximum = 64 )
65 or not re.fullmatch( r " [ A-Za-z0-9 ][ A-Za-z0-9_.- ] * " , value[ "deployment" ])
66 or not isinstance (value.get( "model" ), str ) or value[ "model" ] not in MODELS
67 ):
68 raise fail( "embedding-choice-invalid" , "Select a supported Azure OpenAI endpoint, deployment and embedding model." )
69 if value.get( "dimensions" ) != "service-managed" or value.get( "model_version" ) != "deployment-managed" :
70 raise fail( "embedding-parameter-unsupported" , "Knowledge-source APIs expose neither dimensions nor a model-version pin; custom values are unsupported." )
71 if value.get( "auth" ) != "system-assigned" :
72 raise fail( "embedding-auth-unsupported" , "This planner supports only the existing Search system-assigned identity; no keys or identity changes." )
73 prerequisites = value.get( "prerequisites" )
74 if not isinstance (prerequisites, dict ):
75 raise fail( "embedding-prerequisite-missing" , "Provide current deployment, identity and network evidence references." )
76 require_allowed_fields(prerequisites, { "deployment" , "identity" , "network" }, label = "embedding prerequisites" )
77 if any ( not _text(prerequisites.get(key)) for key in ( "deployment" , "identity" , "network" )):
78 raise fail( "embedding-prerequisite-missing" , "Owner-verified deployment/model, Search identity/RBAC and network evidence is required." )
79 return copy.deepcopy(value)
80
81
82 def verify_model_readback (choice: dict[ str , Any], model: Any) -> None :
83 parameters = model.get( "azureOpenAIParameters" ) if isinstance (model, dict ) else None
84 if not isinstance (parameters, dict ):
85 raise fail( "embedding-readback-invalid" , "Observed embedding parameters must be an object." )
86 # Generic definition normalization omits apiKey; check auth evidence before matching.
87 api_key = parameters.get( "apiKey" )
88 if (
89 api_key is not None and not ( isinstance (api_key, str ) and api_key == "" )
90 or parameters.get( "authIdentity" ) is not None
91 ):
92 raise fail( "embedding-auth-conflict" , "Observed embedding authentication does not prove the selected system-assigned mode; key/identity details withheld." )
93 if not _search_read.definitions_match(model_definition(choice), model):
94 raise fail( "embedding-readback-mismatch" , "Observed embedding endpoint/deployment/model differs from the selected configuration." )
95
96
97 def verify_source_readback (choice: dict[ str , Any] | None , current: Any) -> None :
98 if choice is None or current is None :
99 return
100 kind = current.get( "kind" ) if isinstance (current, dict ) else None
101 if kind not in ( "file" , "azureBlob" ):
102 raise fail( "embedding-readback-invalid" , "Expected a File/Blob embedding source readback." )
103 parameters = current.get( "fileParameters" if kind == "file" else "azureBlobParameters" )
104 ingestion = parameters.get( "ingestionParameters" ) if isinstance (parameters, dict ) else None
105 verify_model_readback(choice, ingestion.get( "embeddingModel" ) if isinstance (ingestion, dict ) else None )
106
107
108 def validate_plan_choice (plan: dict[ str , Any]) -> None :
109 if "embedding" not in plan:
110 return
111 source = plan.get( "source" , {})
112 choice = validate_choice(plan[ "embedding" ], enabled = True , api_version = source.get( "api_version" ))
113 desired = source.get( "desired" , {})
114 parameters = desired.get( "fileParameters" if desired.get( "kind" ) == "file" else "azureBlobParameters" , {})
115 ingestion = parameters.get( "ingestionParameters" , {})
116 if ingestion.get( "contentExtractionMode" ) not in ( "minimal" , "standard" ) or ingestion.get( "embeddingModel" ) != model_definition(choice):
117 raise fail( "embedding-plan-mismatch" , "Embedding choices and the selected extraction definition must match." )
118
119
120 def summary (choice: dict[ str , Any]) -> dict[ str , Any]:
121 return {
122 key: choice[key] for key in ( "endpoint" , "deployment" , "model" , "dimensions" , "model_version" , "auth" )
123 } | {
124 "purpose" : "Source vectors for hybrid/vector search; not CU extraction or KB chat." ,
125 "cost" : "Creating/ingesting this source invokes billable embeddings and moves content to the selected model. Query vectorization needs separate approval." ,
126 "prerequisites" : "Supplied evidence references are owner-verified, not proof of effective access or deployed model version." ,
127 "answer_synthesis" : "Not configured; ingestion dependencies do not configure KB reasoning, reranking or answer synthesis." ,
128 }