vent_agent_2.0/scripts/run_main_fan_document.py

#!/usr/bin/env python3
"""默认验证真实 Agent 的沙箱文档解析与报告交付；旧服务须显式选择。"""
import argparse
import asyncio
import hashlib
import json
import os
from pathlib import Path
import shutil
import sys
import uuid

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))


async def run(args, output):
    from agents.workspace import session_root
    from db import chat_store
    from runtime.environment import RuntimeEnvironment
    from extensions.main_fan.service import analyze
    from runtime.execution import execute
    source = args.document.resolve()
    source_hash = hashlib.sha256(source.read_bytes()).hexdigest()
    async with RuntimeEnvironment(prewarm=False, billing_retry=False):
        user = "main-fan-live-document"
        session = chat_store.create_session(user, "full", "原件自动识别实测")
        root = session_root(user, session["id"])
        target = root / "workspace" / "uploads" / source.name
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(source, target)
        def progress(value):
            print(json.dumps(value, ensure_ascii=False), flush=True)
        data = {"mine_code": args.mine_code, "mine_name": args.mine_name,
            "fan_name": args.fan_name, "fan_id": args.fan_name,
            "document_cache_policy": args.cache_policy,
            "materials": [{"path": "/workspace/uploads/"+source.name}]}
        if args.input_json:
            raw = json.loads(args.input_json.read_text(encoding='utf-8'))
            data = {**raw, **data, 'materials': [*raw.get('materials', []), *data['materials']]}
            data['fan_id'] = args.fan_id or raw.get('fan_id', args.fan_name)
        elif args.fan_id:
            data['fan_id'] = args.fan_id
        if args.agent:
            execution = await execute(agent_name="main_fan", user_name=user, session_id=session["id"],
                message="请读取业务 Skill，在沙箱自行编写文档解析、计算和报告生成代码，核对后提交真实脚本输出。完整输入已注入任务配置。",
                graph_config={"configurable": {"business_task_input": data,
                    "document_audit_dir": str(output / "document")}})
            if execution.status != "success" or not execution.structured_output:
                raise RuntimeError("实际智能体执行失败："+execution.error)
            result = execution.structured_output
            audit = result.get('assessment', {}).get('exploration', {}).get('python_runs', [])
            if not audit or result['assessment'].get('mode') != 'sandbox_exploration':
                raise RuntimeError('没有本轮成功沙箱交付记录，不能把旧固定服务作为自主探索实测')
        else:
            result = await analyze(data, user, session["id"], document_audit_dir=output / "document", document_progress=progress)
            audit = result["analysis"]["document_audits"]
            if not audit:
                raise RuntimeError("没有完整自动识别记录，详情见文档调用原始响应")
            if any(a.get("transcript_seed_loaded") or a.get("cache_used") for a in audit) and args.cache_policy != "auto":
                raise RuntimeError("刷新实测不可使用已有缓存或转录")
    if hashlib.sha256(source.read_bytes()).hexdigest() != source_hash:
        raise RuntimeError("原件在实测过程中发生变化")
    summary = {"mode": "live_agent_sandbox" if args.agent else "legacy_model_document_recognition",
               "source": str(source), "source_sha256": source_hash,
               "session_id": session['id'], "workspace": str(root / 'workspace'),
               "transcript_seed_loaded": False, "cache_policy": args.cache_policy,
               "mine_code_note": "本地测试编码，未假定陕西局正式编码", "result": result}
    (output / "result.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2)+"\n", encoding="utf-8")
    print("真实模型研判完成：", output / "result.json", flush=True)
    if args.agent:
        print('成功沙箱交付', len(audit), '结论', result['platform_payload']['judgment']['resultCode'], flush=True)
    else:
        print("完成页数",sum(len(a["pages"]) for a in audit),"模型调用",sum(len(a["calls"]) for a in audit),
              "计算复核",len(result["analysis"]["calculations"]), flush=True)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--document", type=Path, required=True)
    parser.add_argument("--mine-code", default="live-test-unassigned")
    parser.add_argument("--mine-name", required=True, help="任务目标矿井身份，不作为文档模型的预填数据")
    parser.add_argument("--fan-name", required=True)
    parser.add_argument("--fan-id", help="可选的明确设备ID，默认沿用输入JSON中的fan_id")
    parser.add_argument("--output", type=Path)
    parser.add_argument("--input-json", type=Path, help="可选的原始任务JSON，提供时序等明确数据；不注入文档转录结果")
    parser.add_argument("--cache-policy", choices=["refresh", "off", "auto"], default="refresh")
    parser.add_argument("--vision-model", help="仅旧固定服务：覆盖视觉模型；自主模式由主Agent直接看图")
    parser.add_argument("--agent", action="store_true", default=True, help="默认通过真实 Agent 的沙箱自主执行")
    parser.add_argument("--legacy-service", action="store_false", dest="agent", help="显式验证旧固定服务，不验证沙箱自主探索")
    args = parser.parse_args()
    output = (args.output or ROOT/"data"/"main_fan_live"/uuid.uuid4().hex[:12]).resolve()
    output.mkdir(parents=True, exist_ok=True)
    if (output/"business.sqlite").exists():
        raise ValueError("实测目录已有数据库，请使用新目录保留独立识别记录")
    # 沿用项目模型凭证，仅隔离运行数据与网关；不加载任何测试转录 JSON。
    os.environ.update({"DEEP_AGENT_ENV_PRIORITY": "process", "DB_DIALECT": "sqlite",
        "DB_SQLITE_PATH": str(output/"business.sqlite"), "CHECKPOINTER": "sqlite",
        "CHECKPOINT_DB_PATH": str(output/"checkpoints.sqlite"), "SHANXI_GATEWAY_ENABLED": "0",
        "MAIN_FAN_ALLOW_MANUAL_CACHE": "0", "RESULT_PLATFORM_ENABLED": "0", "MAIN_FAN_AUTO_PUBLISH": "0",
        "MAIN_FAN_TARGETS_FILE": "", "LOG_LEVEL": "WARNING"})
    if args.vision_model:
        os.environ["MAIN_FAN_VISION_MODEL"] = args.vision_model
    try:
        asyncio.run(run(args, output))
    except Exception as exc:
        (output/"failure.json").write_text(json.dumps({"type":type(exc).__name__,"message":str(exc),
            "transcript_seed_loaded":False},ensure_ascii=False,indent=2),encoding="utf-8")
        raise


if __name__ == "__main__":
    main()