36 lines
2.1 KiB
Python
36 lines
2.1 KiB
Python
"""Adapter for the inference(prompt, image, task=...) API shown in module DR.
|
|
|
|
loader is a deployment-pinned callable returning the already loaded model.
|
|
No guessed vendor import, checkpoint download, remote code execution flag or CUDA map.
|
|
"""
|
|
import importlib
|
|
import re
|
|
|
|
def create(config):
|
|
if not isinstance(config,dict) or not isinstance(config.get('loader'),str) or not re.fullmatch(r'[A-Za-z_]\w*(?:\.[A-Za-z_]\w*)*:[A-Za-z_]\w*',config['loader']):
|
|
raise ValueError('loader must name a deployment-provided module:callable')
|
|
mapping=config.get('task_mapping')
|
|
if not isinstance(mapping,dict) or not mapping or any(not isinstance(value,str) or not value.strip() for value in mapping.values()):
|
|
raise ValueError('task_mapping must contain deployment-validated model tasks')
|
|
if 'dense_feedback' in mapping:
|
|
raise ValueError('dense feedback requires a deployment-provided multi-frame adapter; this adapter accepts a single image')
|
|
if set(mapping)-{'plan','shelf','localize3d'}:
|
|
raise ValueError('unsupported capability in task_mapping')
|
|
if not isinstance(config.get('model'),dict):
|
|
raise ValueError('model must contain deployment-provided loading configuration')
|
|
module,name=config['loader'].split(':',1)
|
|
try:loader=getattr(importlib.import_module(module),name)
|
|
except (ImportError,AttributeError) as ex:
|
|
raise ValueError('deployment model loader unavailable: '+config['loader']) from ex
|
|
if not callable(loader):raise ValueError('deployment model loader is not callable')
|
|
model=loader(config['model'])
|
|
if not callable(getattr(model,'inference',None)):
|
|
raise ValueError('deployment model must expose inference(prompt, image, task=...)')
|
|
def infer(request):
|
|
capability=request['capability']
|
|
task=config.get('task_mapping',{}).get(capability)
|
|
if task is None:raise ValueError('capability has no validated model task mapping')
|
|
image=request.get('observation',{}).get('image_path')
|
|
return model.inference(request['prompt'],image,task=task,do_sample=False)
|
|
return infer
|