#!/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()