| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849 |
- """离线模型动作脚本：真实文件工具、真实沙箱和真实提交，替代旧 MainFanResult 假完成。"""
- import json
- import uuid
- from langchain_core.messages import AIMessage
- from db import db_conn, main_fan_store
- def tool_call(name, args):
- return AIMessage(content='', tool_calls=[{'name': name, 'args': args, 'id': uuid.uuid4().hex}],
- usage_metadata={'input_tokens': 20, 'output_tokens': 5, 'total_tokens': 25})
- def sandbox_messages(user, data, *, encoded_lists=False):
- yield tool_call('read_file', {'file_path': '/skills/main-fan-equivalent-capability-assessment/SKILL.md'})
- yield tool_call('prepare_main_fan_task', {'target': data})
- row = db_conn.query_one('SELECT id,session_id FROM main_fan_workflows WHERE user_name=? ORDER BY created_at DESC LIMIT 1', (user,))
- assert row, '准备工具没有创建任务'
- workflow_id, session = row['id'], row['session_id']
- code = f'''
- from pathlib import Path
- import json
- raw = json.loads(Path('exploration/{workflow_id}/input.json').read_text())
- metrics = {{p['metric']: p for p in raw.get('points', [])}}
- value = None
- if all(key in metrics for key in ('flow', 'static_pressure', 'shaft_power')):
- value = metrics['flow']['value'] * metrics['static_pressure']['value'] / (10 * metrics['shaft_power']['value'])
- summary = '已根据本轮数据执行计算，三项对照尚不完整。' if value is not None else '本轮没有可比资料。'
- body = '# 自编报告\\n\\n' + summary + ('\\n\\n静压效率：' + str(value) + '%' if value is not None else '')
- Path('agent-report.md').write_text(body)
- assessment = {{'status': 'completed' if value is not None else 'needs_materials',
- 'result_code': 'UNDETERMINED', 'summary': summary,
- 'evidence': [{{'source': '/workspace/exploration/{workflow_id}/input.json', 'pointer': '/points'}}],
- 'details': {{'efficiency_pct': value}}}}
- Path('agent-result.json').write_text(json.dumps(assessment, ensure_ascii=False))
- print('actual sandbox delivery')
- '''
- yield tool_call('write_file', {'file_path': '/workspace/agent-analysis.py', 'content': code})
- output_paths = ['agent-report.md', 'agent-result.json'] if encoded_lists else ['/workspace/agent-report.md', '/workspace/agent-result.json']
- yield tool_call('run_python', {'workflow_id': workflow_id,
- 'script_path': 'agent-analysis.py' if encoded_lists else '/workspace/agent-analysis.py',
- 'output_paths': json.dumps(output_paths) if encoded_lists else output_paths})
- workflow = main_fan_store.workflow(workflow_id, user, session)
- runs = workflow['context']['python_runs']
- assert runs and runs[-1]['exit_code'] == 0, runs
- yield tool_call('submit_main_fan_report', {'workflow_id': workflow_id,
- 'report_path': '/workspace/agent-report.md', 'calculations_path': '/workspace/agent-result.json',
- 'python_run_ids': json.dumps([runs[-1]['run_id']]) if encoded_lists else [runs[-1]['run_id']]})
|
