Ares-Lab / ares /sft.py
Ares Publisher
Add resumable training metrics and role-specific SFT preparation
4f53876
Raw
History Blame Contribute Delete
1.63 kB
"""Convert reviewed JSONL examples to explicit Ares/Xiphos chat-section training text.
Input is local, human-curated JSONL. This module never downloads data or auto-labels it.
"""
import argparse,json
from pathlib import Path
REQUIRED={'ares':('user','assistant'),'xiphos':('request','plan')}
def format_row(role,row):
if role=='ares':
user,assistant=(str(row[x]).strip() for x in REQUIRED[role])
if not user or not assistant:raise ValueError('empty user or assistant')
system=str(row.get('system','You are Ares, a helpful and honest assistant.')).strip()
return f'<system>{system}</system><user>{user}</user><assistant>{assistant}</assistant>'
request,plan=(str(row[x]).strip() for x in REQUIRED[role])
if not request or not plan:raise ValueError('empty request or plan')
system='You are Xiphos. Draft reviewable plans; do not execute tools or code.'
return f'<system>{system}</system><user>{request}</user><assistant><plan>{plan}</plan></assistant>'
def main():
p=argparse.ArgumentParser();p.add_argument('--role',choices=('ares','xiphos'),required=True);p.add_argument('--input',required=True);p.add_argument('--out',required=True);a=p.parse_args();written=0
Path(a.out).parent.mkdir(parents=True,exist_ok=True)
with open(a.out,'w',encoding='utf8') as out:
for line_no,line in enumerate(open(a.input,encoding='utf8'),1):
try:out.write(format_row(a.role,json.loads(line))+'\n');written+=1
except (json.JSONDecodeError,KeyError,ValueError) as e:raise SystemExit(f'Invalid reviewed example at line {line_no}: {e}')
print({'role':a.role,'examples':written,'out':a.out})
if __name__=='__main__':main()