vent_agent_2.0/tests/sandbox_agent_helpers.py

"""离线模型动作脚本：真实文件工具、真实沙箱和真实提交，替代旧 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']]})