| """Local tools with explicit source records and no network or model side effects.""" |
| import json |
| from pathlib import Path |
| from .engine import Tool |
| from .schema import Candidate,Measurement,ToolResult,DesignSpec |
| from .metrics import candidate_score,diverse_select |
| from .evidence import retrieval_tool |
| from .structure import af3_input |
|
|
|
|
| def registry_from_config(config): |
| from .engine import CommandTool |
| registry={} |
| for row in config.get('commands',[]): |
| registry[row['name']]=CommandTool(**row) |
| if config.get('evidence_path'): |
| registry['retrieve_evidence']=Tool('retrieve_evidence',retrieval_tool(config['evidence_path']), |
| {'tool_calls':1},'Search the frozen passage corpus by query.',{'query':'string','k':'integer'}) |
| def ingest(a,s): |
| p=Path(a['path']);x=json.loads(p.read_text()) |
| result=ToolResult.model_validate(x) |
| |
| if any(not m.source_id for m in result.measurements):raise ValueError('missing measurement source') |
| return result |
| registry['ingest_results']=Tool('ingest_results',ingest,{'tool_calls':1}, |
| 'Read an operator-prepared ToolResult JSON file; preserve endpoint units and experiment labels.') |
| def rank(a,s): |
| spec=DesignSpec.model_validate(s['spec']);scores={};sequences={} |
| for key,c in s['candidates'].items(): |
| ms=[Measurement.model_validate(m) for m in s['measurements'] if m['candidate_id']==key and m['kind']==a.get('kind','prediction')] |
| try:quality=candidate_score(ms,spec.requirements) |
| except ValueError:continue |
| scores[key]=quality;sequences[key]=c["molecule"]["sequence"] |
| selected=diverse_select(sequences,scores,k=int(a.get('k',12)),penalty=float(a.get('penalty',.2)),max_identity=float(a.get('max_identity',.8))) |
| path=Path(a['output']);path.parent.mkdir(parents=True,exist_ok=True) |
| path.write_text(json.dumps({'selected':selected,'eligible':len(scores)},indent=2)) |
| return ToolResult(artifacts={'selection':str(path.resolve())},message=f'{len(selected)} candidates selected from {len(scores)} with all required endpoints') |
| registry['rank_candidates']=Tool('rank_candidates',rank,{'tool_calls':1},'Rank complete calibrated endpoint vectors and select diverse candidates.') |
| def export(a,s): |
| spec=DesignSpec.model_validate(s['spec']);c=Candidate.model_validate(s['candidates'][a['candidate_id']]) |
| all_targets={t.id:t for t in spec.targets+spec.countertargets} |
| chosen=[all_targets[k] for k in a.get('target_ids',[t.id for t in spec.targets])] |
| result=af3_input(spec.episode_id,chosen,c.molecule,seeds=(spec.seed,spec.seed+1,spec.seed+2)) |
| path=Path(a['output']);path.parent.mkdir(parents=True,exist_ok=True);path.write_text(json.dumps(result,indent=2)) |
| return ToolResult(artifacts={'af3_input':str(path.resolve())}) |
| registry['export_af3']=Tool('export_af3',export,{'tool_calls':1},'Export all target partners and a fixed peptide with explicit PTM chemistry.') |
| def inspect_state(a,s): |
| key=a['collection'] |
| if key not in {'candidates','measurements','evidence','artifacts'}:raise ValueError('invalid state collection') |
| values=s[key] |
| if isinstance(values,dict): |
| selected={k:values[k] for k in a.get('ids',list(values)[-3:])} |
| else: |
| selected=[v for v in values if not a.get('candidate_id') or v['candidate_id']==a['candidate_id']][-10:] |
| return ToolResult(message=json.dumps({'collection':key,'records':selected})) |
| registry['inspect_state']=Tool('inspect_state',inspect_state,{'tool_calls':1},'Retrieve archived candidates, measurements, evidence, or artifacts omitted by deterministic context packing.') |
| return registry |
|
|
