vent_agent_2.0/scripts/demo_main_fan.py

#!/usr/bin/env python3
"""历史固定服务的离线演示：队列、网关、独立结果平台、SQLite；不验证自主 Agent。"""
import argparse
import asyncio
import json
import os
from pathlib import Path
import socket
import sys
import uuid

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


def local_orchestrator():
    """旧工具协议的离线模型，保留给历史调用方；当前 Agent 已使用 Skill 工具集。"""
    from langchain_core.language_models.chat_models import BaseChatModel
    from langchain_core.messages import AIMessage, AIMessageChunk, ToolMessage, HumanMessage
    from langchain_core.outputs import ChatResult, ChatGeneration, ChatGenerationChunk

    class OfflineModel(BaseChatModel):
        @property
        def _llm_type(self):
            return "offline-main-fan-orchestrator"

        def bind_tools(self, tools, **kwargs):
            return self

        def response(self, messages):
            last = messages[-1]
            if isinstance(last, ToolMessage) and last.name == "analyze_main_fan":
                value = json.loads(last.content)
                return AIMessage(content="", tool_calls=[{"name": "MainFanResult", "args": value, "id": uuid.uuid4().hex}])
            text = next(m.content for m in reversed(messages) if isinstance(m, HumanMessage))
            text = text.split("输入：", 1)[1]
            value, _ = json.JSONDecoder().raw_decode(text.strip())
            return AIMessage(content="", tool_calls=[{"name": "analyze_main_fan", "args": {"data": value}, "id": uuid.uuid4().hex}])

        def _generate(self, messages, stop=None, run_manager=None, **kwargs):
            return ChatResult(generations=[ChatGeneration(message=self.response(messages))])

        async def _astream(self, messages, **kwargs):
            message = self.response(messages)
            yield ChatGenerationChunk(message=AIMessageChunk(content=message.content, tool_calls=message.tool_calls))

    return OfflineModel()


async def run(args, output, sock, result_sock):
    import uvicorn
    from db import business_store, db_conn
    from mock_gateway.main_fan import MockState, create_app, task_input
    from mock_result_platform.app import ResultStore, create_app as create_result_app
    from runtime import business
    from runtime.business_tasks import submit, get_controller
    from runtime.environment import RuntimeEnvironment
    state = MockState(args.data_dir.resolve())
    server = uvicorn.Server(uvicorn.Config(create_app(state), log_level="warning", access_log=False))
    server_task = asyncio.create_task(server.serve(sockets=[sock]))
    received = ResultStore(output / "received_results.sqlite")
    result_server = uvicorn.Server(uvicorn.Config(create_result_app(received), log_level="warning", access_log=False))
    result_task = asyncio.create_task(result_server.serve(sockets=[result_sock]))
    try:
        for _ in range(100):
            if server.started and result_server.started:
                break
            if server_task.done():
                await server_task
                raise RuntimeError("模拟服务器未启动")
            if result_task.done():
                await result_task
                raise RuntimeError("结果接收平台未启动")
            await asyncio.sleep(.05)
        results = []
        management = []
        async with RuntimeEnvironment(prewarm=False, billing_retry=False, business_jobs=True):
            # 显式使用旧服务做离线网关联调，不能把它当成自主探索模式的模型验收。
            from dataclasses import replace
            from extensions.main_fan.service import analyze
            controller = get_controller()

            async def offline_executor(*, graph_config, **kwargs):
                configured = graph_config['configurable']
                job = await asyncio.to_thread(business_store.get, configured['business_task_id'])
                legacy = replace(controller.catalog.business('main_fan'), test_runner=analyze)
                return await controller._run_test_pipeline(legacy, job, configured['business_task_input'])

            controller.executor = offline_executor
            state.seed_reviewed_cache()
            for scenario in (args.scenario,) if args.scenario != "all" else ("original", "report", "synthetic"):
                created = await asyncio.to_thread(submit, "main_fan", task_input(scenario),
                    user_name="demo", request_id="demo-"+scenario, publish=True)
                for _ in range(160):
                    job = await asyncio.to_thread(business_store.get, created["id"])
                    if job["push_status"] == "pushed" or job["status"] in {"failed", "needs_materials", "cancelled"} or job["push_status"] == "failed":
                        break
                    await asyncio.sleep(.25)
                if job["push_status"] != "pushed":
                    raise RuntimeError(f"演示任务未成功推送：{(job['status'], job['push_status'], job['error'])}")
                results.append({"scenario": scenario, "job_id": job["id"], "status": job["status"],
                    "push_status": job["push_status"], "result": json.loads(job["result_json"]),
                    "receipt": json.loads(job["receipt_json"])})
            for action, online in (("stop", False), ("start", True)):
                await (business.stop("main_fan") if action == "stop" else business.start("main_fan"))
                current = await asyncio.to_thread(business_store.state, "main_fan")
                assert bool(current["local_enabled"]) == online
                management.append({"action": action, "local_enabled": online})
            metrics = await asyncio.to_thread(business_store.metrics, "main_fan")
        assert not any(call["operation"] in {"register", "push", "upload"} for call in state.calls)
        value = {"mode": "legacy_offline_pipeline_gateway_and_separate_results", "facts_source": "manual_transcription_verified_by_sha256",
                 "results": results, "gateway_calls": state.calls, "uploads": state.uploads,
                 "management": management, "business_metrics": metrics,
                 "received_results": received.rows(),
                 "database": str(output / "demo.sqlite")}
        (output / "result.json").write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        print("演示完成：", output / "result.json")
        for result in results:
            print(result["scenario"], result["status"], result["push_status"],
                  "关联", len(result["result"]["analysis"]["associations"]),
                  "计算", len(result["result"]["analysis"]["calculations"]))
    finally:
        server.should_exit = True
        result_server.should_exit = True
        await asyncio.gather(server_task, result_task)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--data-dir", type=Path, required=True)
    parser.add_argument("--scenario", choices=["all", "original", "report", "synthetic"], default="all")
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()
    output = (args.output or ROOT / "data" / "main_fan_demo" / uuid.uuid4().hex[:12]).resolve()
    output.mkdir(parents=True, exist_ok=True)
    if (output / "demo.sqlite").exists():
        raise ValueError("演示输出目录已有数据库，请使用新目录避免覆盖")
    sock = socket.socket()
    sock.bind(("127.0.0.1", 0))
    sock.setblocking(False)
    port = sock.getsockname()[1]
    result_sock = socket.socket()
    result_sock.bind(("127.0.0.1", 0))
    result_sock.setblocking(False)
    result_port = result_sock.getsockname()[1]
    os.environ.update({"DEEP_AGENT_ENV_PRIORITY": "none", "DB_DIALECT": "sqlite",
        "DB_SQLITE_PATH": str(output / "demo.sqlite"), "CHECKPOINTER": "sqlite",
        "CHECKPOINT_DB_PATH": str(output / "checkpoints.sqlite"), "CTX_COMPACT_ENABLED": "0",
        "CONTEXT_DEMO_ENABLED": "0", "MAIN_FAN_ENABLED": "1", "MAIN_FAN_AGENT_CODE": "mock-main-fan",
        "MAIN_FAN_CHECK_TYPE_CODE": "mock-fan-screening", "MAIN_FAN_TOPICS": "main_fan.judgment",
        "SHANXI_GATEWAY_ENABLED": "1", "SHANXI_GATEWAY_URL": f"http://127.0.0.1:{port}",
        "SHANXI_FILE_BASE_URL": f"http://127.0.0.1:{port}", "SHANXI_GATEWAY_TOKEN": "mock-internal-token",
        "RESULT_PLATFORM_ENABLED": "1", "RESULT_PLATFORM_URL": f"http://127.0.0.1:{result_port}/api/judgments",
        "RESULT_PLATFORM_TOKEN": "local-result-platform-token", "MAIN_FAN_AUTO_PUBLISH": "0",
        "MAIN_FAN_DEVICES_PATH": "/mock-platform/main-fan/devices", "MAIN_FAN_POINTS_PATH": "/mock-platform/main-fan/points",
        "MAIN_FAN_REPORTS_PATH": "/mock-platform/main-fan/reports", "MAIN_FAN_VISION_MODEL": "",
        "MAIN_FAN_ALLOW_MANUAL_CACHE": "1",
        "TFAGENTS_DAILY_TOKEN_LIMIT": "0", "BILLING_ALLOW_UNPRICED": "1", "LOG_LEVEL": "WARNING"})
    try:
        asyncio.run(run(args, output, sock, result_sock))
    finally:
        sock.close()
        result_sock.close()


if __name__ == "__main__":
    main()