企业官网建设流程全解析

1. 项目概述：一个为地震风险分析赋能的“智能体界面”

如果你从事地震工程、灾害风险评估或者城市规划相关的工作，大概率对“概率地震危险性分析”和“风险分析”这两个词不会陌生。它们是我们评估一个地区未来可能遭受地震打击的“标尺”，是制定抗震设计规范、进行保险定价和规划应急资源的核心依据。然而，这个分析过程本身，却常常让从业者感到头疼。它涉及海量的数据（地震目录、活动断层、场地条件）、复杂的模型（地震动预测方程、脆弱性曲线）以及繁琐的计算流程。传统的做法要么依赖封闭的商业软件，操作不透明且扩展性差；要么就是自己写脚本调用像OpenQuake这样的开源引擎，但这对非编程背景的工程师来说门槛太高，且流程管理容易混乱。

我最近花了不少时间，尝试构建一个东西来解决这个痛点。我称之为“一个用于端到端概率地震危险性与风险分析的智能体界面”。听起来有点学术？其实核心思想很直接：我想做一个“智能中间人”。它不是一个全新的计算引擎，而是一个建立在现有强大开源工具（如GEM的OpenQuake引擎）之上的、高度自动化和智能化的操作界面。这个界面的目标是，让用户能够用更自然、更高效的方式，描述他们想要分析的问题（比如“帮我评估一下这个工业园区未来50年，因地震导致直接经济损失超过10亿的概率”），然后由这个“智能体”自动完成从数据准备、模型配置、计算提交到结果提取与可视化的全链条工作。

这里的关键词是“Agentic Interface”和“Model Context Protocol”。前者意味着这个界面具备一定的自主性和决策能力，能理解用户意图并分解任务；后者则是一个新兴的、用于标准化模型交互的协议，可以把它想象成模型之间的“通用插头”，让我的智能体能轻松地“插入”OpenQuake引擎或其他兼容的分析工具，无需关心底层复杂的API细节。通过结合这两者，我希望能把专家从重复性的流程操作中解放出来，更专注于分析逻辑和结果解读，同时也为新手或跨领域研究者打开一扇便捷的大门。

2. 核心设计思路：为什么是“智能体”+“协议”？

2.1 传统工作流的痛点与“智能体”的价值定位

在深入技术细节前，我们有必要先看看当前典型的PSHA（概率地震危险性分析）和PSRA（概率地震风险分析）工作流是怎样的。通常，它包含以下几个阶段：

1. 数据收集与预处理：收集研究区的地震目录、活动断层数据、地震动预测方程（GMPE）、场地放大模型、暴露数据库（建筑物、人口、资产清单）和脆弱性/风险函数。这些数据格式各异，来源不一，清洗和标准化耗时耗力。
2. 模型配置与参数化：在OpenQuake引擎中，你需要编写复杂的XML或INI格式的配置文件（job.ini），精确地定义逻辑树（用于处理模型不确定性）、计算参数、输出格式等。一个配置文件的错误可能导致数小时甚至数天的计算白费。
3. 计算执行与监控：提交计算任务到本地服务器或高性能计算集群。计算可能持续数小时至数天，需要手动监控日志，处理可能出现的运行错误或资源不足问题。
4. 结果提取与后处理：计算完成后，会生成大量HDF5或NRML格式的结果文件。你需要编写额外的Python脚本去提取特定的结果（如特定地点的危险性曲线、风险损失分布图），并进行可视化。

这个流程的每个环节都充满了“摩擦”。数据格式转换是手工的“脏活累活”；配置文件语法晦涩难记，且严重依赖经验；计算过程是个黑盒，出了问题难以调试；结果处理又需要二次开发。“智能体界面”的核心价值，就在于消除这些摩擦点。它扮演一个“全能助手”的角色：

• 在数据层，它能识别常见数据格式，提供模板和向导，辅助用户完成数据清洗和标准化，甚至能根据研究区位置，智能推荐可能适用的公开数据集（如USGS地震目录、GEM全球数据库）。
• 在配置层，它提供图形化或领域特定语言（DSL）的配置方式。用户可以通过勾选、表单填写或自然语言描述（如“考虑浅源与深源地震的模型不确定性”）来定义分析，智能体在后台将其转换为精确的OpenQuake引擎配置文件。
• 在执行层，它管理计算任务队列，自动监控日志，在任务失败时尝试重试或给出清晰的错误诊断建议，并将计算资源的使用情况反馈给用户。
• 在后处理层，它内置常用的结果解析和可视化模板，用户只需点选感兴趣的输出指标（如“年平均损失率AAL”、“超过概率曲线”），即可一键生成图表和报告草稿。

2.2 Model Context Protocol：实现智能体与引擎对话的“普通话”

设计这样一个智能体，最大的技术挑战之一是如何与下层的计算引擎（如OpenQuake）进行高效、可靠且松耦合的通信。直接调用OpenQuake的Python API是一种方式，但这意味着智能体的代码将与特定版本的OpenQuake深度绑定，引擎升级或切换其他分析工具（如HAZUS、RISK-UE的某些模块）时，适配成本极高。

这正是Model Context Protocol发挥作用的地方。MCP的核心理念是为“模型”（在这里，OpenQuake引擎就是一个复杂的计算模型）提供一个标准化的交互接口。你可以把它类比为HTTP协议之于Web服务。无论服务器用的是Java、Python还是Go，只要它遵循HTTP协议，浏览器就能与之通信。

MCP为模型定义了标准的操作，例如：

• list_models(): 列出可用的分析模型或计算功能（如“计算经典PSHA”、“计算基于事件的风险”）。
• get_model_schema(model_id): 获取某个模型所需的输入参数模式（Schema）。这告诉智能体：“要运行这个风险计算，你需要提供暴露文件路径、脆弱性模型ID、强度测量类型等参数”。
• execute_model(model_id, inputs): 使用给定的输入参数执行模型。
• get_model_results(execution_id): 查询某个执行任务的结果。

通过让OpenQuake引擎（或一个封装它的适配器）实现MCP服务端，我的智能体界面（作为MCP客户端）就可以用一套统一的“语言”与之对话。这样做带来了几个关键优势：

1. 解耦与灵活性：智能体不再关心OpenQuake内部如何实现。未来如果集成了新的地震动模拟器或风险计算模块，只要它们也支持MCP，智能体就能无缝接入。
2. 自描述性：智能体可以通过get_model_schema动态了解每个计算任务需要什么参数、参数的类型和约束。这使得智能体可以构建动态的、智能的表单来引导用户输入，甚至进行输入验证。
3. 标准化通信：所有交互都基于标准的JSON-RPC over WebSocket或HTTP，便于调试、监控和集成到更大的工作流系统中。

注意：MCP是一个新兴协议，OpenQuake引擎本身并未原生支持。在实际实现中，我通常需要为OpenQuake编写一个轻量级的“MCP适配器”服务。这个服务封装了对OpenQuake API的调用，并将其暴露为MCP标准接口。这是整个架构中的关键开发环节。

2.3 端到端流程的自动化编排

有了MCP提供的标准化操作接口，智能体就可以编排一个完整的端到端分析流程。这不仅仅是单个计算任务的执行，而是一个包含多个步骤、可能有条件分支的工作流。例如，一个完整的区域地震风险分析可能包含：

1. 先进行概率地震危险性分析（PSHA），生成地震动场。
2. 然后利用PSHA的结果，结合暴露和脆弱性模型，进行概率风险分析（PSRA）。
3. 最后，基于风险结果，生成热力图和统计报告。

智能体内部需要有一个“工作流引擎”或“任务编排器”。它可以是一个简单的状态机，也可以利用像Prefect或Airflow这样的成熟工具。其核心是将用户的高级目标，分解为一系列通过MCP调用的原子操作，并管理这些操作之间的依赖关系（如步骤2依赖步骤1的输出）、数据传递以及错误处理。

3. 关键技术实现与架构拆解

3.1 系统架构分层设计

为了实现上述思路，我将整个系统分为四个清晰的层次，从上到下依次是：

• 用户交互层：这是智能体的“脸面”。它可以有多种形态：

  • Web图形界面：最友好的方式，提供拖拽式工作流设计器、表单化参数配置、实时结果可视化面板。适合大多数工程师和决策者。
  • 命令行界面：为高级用户和自动化脚本提供更高效的操作方式，例如agentic-psha --region "San Francisco" --return-period 475 --output hazard_curve.png。
  • 编程API：以Python库的形式提供，允许用户在自己的Jupyter Notebook或脚本中调用智能体的功能，实现深度定制。
  • 自然语言接口：未来的发展方向，用户可以直接输入“评估上海陆家嘴金融区在1000年回归周期下的地震损失”，由大语言模型理解并转换为系统可执行的任务。

• 智能体核心层：这是系统的“大脑”。它包含几个核心模块：

  • 意图解析器：理解用户的请求（来自GUI的点击、CLI的命令或NL的语句），将其转化为内部的任务描述。
  • 任务规划器：根据任务描述，结合内置的领域知识（如PSHA→PSRA的标准流程），生成一个具体的工作流DAG（有向无环图）。
  • 知识库：存储领域知识，例如不同区域推荐使用的GMPE模型、常见脆弱性模型库的索引、数据预处理的最佳实践规则等。这使智能体能提供“智能推荐”。
  • 工作流执行引擎：负责按顺序或并行执行任务规划器生成的DAG中的每个节点。每个节点通常对应一个对下层MCP服务的调用。

• MCP服务层：这是系统的“神经系统”。它由多个MCP服务端构成，每个服务封装一个特定的计算能力：

  • OpenQuake计算服务：最主要的服务。它接收标准化的计算请求（通过MCP），调用本地或远程的OpenQuake引擎执行，并返回结果句柄或直接结果。
  • 数据预处理服务：提供数据格式转换、坐标系统一、缺失值处理、质量检查等功能。
  • 地理空间服务：提供地图底图、区域裁剪、空间插值、成果出图等功能，可能基于GeoServer或类似技术。
  • 结果缓存与查询服务：管理历史计算结果，提供快速查询和对比分析功能。

• 基础设施与计算资源层：这是系统的“身体”。包括：

  • OpenQuake引擎集群：实际执行高强度计算的算力。
  • 数据库：存储配置模板、用户项目、结果元数据等。
  • 文件存储：存储输入数据文件、中间文件和最终结果文件（通常是HDF5）。
  • 容器化环境：使用Docker或Kubernetes来封装和部署MCP服务及OpenQuake引擎，确保环境一致性和可扩展性。

3.2 核心模块实现细节

3.2.1 意图解析与任务规划

这是智能体“智能”的体现。对于简单的CLI命令，解析相对直接。但对于更灵活的GUI操作或未来的自然语言接口，则需要更复杂的逻辑。

我采用的方法是基于模板和规则的解析。系统内置一系列“分析模式”模板，如“区域PSHA”、“站点特异性风险”、“情景地震损失评估”。每个模板定义了可能的参数槽位。意图解析器将用户输入与这些模板进行匹配，并填充槽位。

例如，用户在图界面上选择“区域PSHA”，在地图上画了一个框，设置了回归周期为475年。意图解析器会匹配到“区域PSHA”模板，并将地图坐标和回归周期填充到模板的region和return_period槽位中。

任务规划器则根据填充好的模板，实例化一个具体的工作流。这个工作流可能被定义为YAML或JSON格式的配置文件：

workflow_name: regional_psha steps: - id: prepare_seismic_source type: mcp_call service: data_preprocessing model: prepare_source_model inputs: region_geojson: {{ user_input.region }} catalog_source: "USGS" # ... 其他参数 outputs: source_model_file: source_model.xml - id: run_psha_calculation type: mcp_call service: openquake_calc model: classical_psha inputs: source_model: {{ steps.prepare_seismic_source.outputs.source_model_file }} gsim_logic_tree: "Global_Active_Crustal" intensity_measure_types: ["PGA", "SA(0.2)", "SA(1.0)"] investigation_time: 50 # ... 其他参数 outputs: hazard_results_id: calculation_id - id: extract_hazard_curves type: mcp_call service: openquake_calc model: extract_results inputs: execution_id: {{ steps.run_psha_calculation.outputs.hazard_results_id }} result_type: "hazard_curve" location: "specific_sites" # 或 `grid` outputs: hazard_curves_data: curves.csv - id: visualize_results type: mcp_call service: geospatial_viz model: plot_hazard_map inputs: data_file: {{ steps.extract_hazard_curves.outputs.hazard_curves_data }} metric: "PGA" return_period: 475 outputs: map_image: hazard_map_475yr.png

规划器的工作就是生成这样的工作流定义，并确保步骤间的依赖关系正确（用{{ ... }}表示）。

3.2.2 OpenQuake MCP适配器实现

这是连接智能体和计算引擎的桥梁。我实现了一个Python服务，使用FastAPI框架提供HTTP端点，这些端点严格遵循MCP的规范。

# 示例代码，展示MCP适配器的核心结构 from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import subprocess import json import uuid import asyncio from openquake.commands.run import run_engine app = FastAPI(title="OpenQuake MCP Adapter") # 内存中存储执行状态（生产环境应用数据库） executions = {} class ModelInput(BaseModel): # 这里定义通用的输入结构，实际会根据model_id不同而动态验证 config: dict # 对应OpenQuake的job.ini内容（已转换为dict） # 也可以支持直接上传配置文件 @app.post("/api/mcp/v1/models/{model_id}/execute") async def execute_model(model_id: str, inputs: ModelInput): """执行一个OpenQuake计算模型""" if model_id not in ["classical_psha", "event_based_risk", "scenario_damage"]: raise HTTPException(status_code=404, detail="Model not found") execution_id = str(uuid.uuid4()) # 1. 将输入转换为OpenQuake job.ini文件 job_ini_content = convert_dict_to_ini(inputs.config) job_file_path = f"/tmp/{execution_id}.ini" with open(job_file_path, 'w') as f: f.write(job_ini_content) # 2. 异步执行OpenQuake引擎（避免阻塞） async def run_oq_job(): try: # 调用OpenQuake引擎，这是一个长时间运行的过程 # run_engine是OpenQuake提供的API result = await asyncio.to_thread(run_engine, job_file_path) executions[execution_id] = {"status": "SUCCESS", "result_path": result.output_dir} except Exception as e: executions[execution_id] = {"status": "FAILED", "error": str(e)} asyncio.create_task(run_oq_job()) executions[execution_id] = {"status": "RUNNING"} return {"executionId": execution_id, "status": "RUNNING"} @app.get("/api/mcp/v1/executions/{execution_id}") async def get_execution_status(execution_id: str): """查询执行状态""" if execution_id not in executions: raise HTTPException(status_code=404, detail="Execution not found") return executions[execution_id] @app.get("/api/mcp/v1/models") async def list_models(): """列出所有支持的模型""" return { "models": [ {"id": "classical_psha", "name": "Classical PSHA", "description": "经典概率地震危险性分析"}, {"id": "event_based_risk", "name": "Event-Based Risk", "description": "基于事件的风险分析"}, {"id": "scenario_damage", "name": "Scenario Damage", "description": "情景地震损失评估"}, # ... 更多模型 ] } @app.get("/api/mcp/v1/models/{model_id}/schema") async def get_model_schema(model_id: str): """获取指定模型的输入参数模式""" schemas = { "classical_psha": { "type": "object", "properties": { "calculation_mode": {"type": "string", "enum": ["classical"], "default": "classical"}, "rupture_mesh_spacing": {"type": "number", "minimum": 1, "default": 5}, "source_model_logic_tree_file": {"type": "string", "description": "源模型逻辑树文件路径"}, "gsim_logic_tree_file": {"type": "string", "description": "GMPE逻辑树文件路径"}, "intensity_measure_types_and_levels": {"type": "object", "description": "强度指标与水平"}, # ... 更多属性，对应job.ini的各个section }, "required": ["source_model_logic_tree_file", "gsim_logic_tree_file"] }, # ... 其他模型的schema } if model_id not in schemas: raise HTTPException(status_code=404, detail="Model schema not found") return schemas[model_id]

这个适配器将OpenQuake引擎的复杂性封装在标准的HTTP API之后。智能体核心层只需要知道MCP的端点地址，就可以通过查询schema知道如何配置一个计算，通过execute提交任务，并通过轮询executions端点获取状态和结果。

实操心得：在实现适配器时，一个关键决策是如何处理OpenQuake引擎的长时间计算。我采用了异步任务（asyncio.create_task）来启动计算，并立即返回一个execution_id。这样不会阻塞HTTP请求。计算状态和结果存储在外部（这里是内存字典，生产环境应用Redis或数据库），供后续查询。此外，将OpenQuake庞大的job.ini配置参数全部暴露在schema里会太臃肿。更好的做法是提供高层级的参数组（如“地震源模型”、“场地条件”、“计算网格”），由适配器内部将其映射到具体的OpenQuake配置项。

3.2.3 工作流执行引擎

对于简单的线性工作流，用Python脚本顺序调用MCP客户端就足够了。但对于复杂、有条件分支、可并行执行的工作流，需要一个更健壮的引擎。我评估了两种方案：

1. 轻量级自制状态机：使用Python的networkx库表示DAG，然后按拓扑顺序执行每个节点。节点执行器调用对应的MCP客户端。需要自己处理错误重试、超时、依赖传递。优点是轻量、可控。
2. 采用成熟工作流引擎：如Prefect。Prefect的“流”和“任务”抽象与我们的概念非常契合。每个MCP调用可以定义为一个Prefect任务，工作流就是一个Prefect流。Prefect原生支持任务依赖、参数化、重试策略、结果持久化、分布式执行和丰富的UI监控。这大大减少了我们自己造轮子的工作量。

我最终选择了Prefect 2.x版本。将之前YAML定义的工作流，转化为Prefect的Python DSL：

from prefect import flow, task from mcp_client import MCPClient # 假设的MCP客户端库 client = MCPClient("http://openquake-adapter:8000") @task(retries=3, retry_delay_seconds=10) def prepare_source_model_task(region_geojson): """任务：准备地震源模型""" result = client.execute_model("prepare_source_model", inputs={"region": region_geojson}) return result["source_model_file"] @task def run_psha_task(source_model_file, imts, investigation_time): """任务：运行PSHA计算""" config = { "calculation_mode": "classical", "source_model_logic_tree_file": source_model_file, "intensity_measure_types_and_levels": {imt: [0.001, 0.01, 0.1, 0.2, 0.5, 1.0] for imt in imts}, "investigation_time": investigation_time, # ... 其他配置 } execution = client.execute_model("classical_psha", inputs={"config": config}) # 等待计算完成 while True: status = client.get_execution_status(execution["executionId"]) if status["status"] in ["SUCCESS", "FAILED"]: break time.sleep(30) # 轮询间隔 if status["status"] == "FAILED": raise Exception(f"PSHA计算失败: {status.get('error')}") return execution["executionId"] @flow(name="regional-psha-flow") def regional_psha_flow(region_geojson: str, imts: list = ["PGA", "SA(0.2)"], investigation_time: int = 50): """流：区域PSHA分析主流程""" # 任务1：准备源模型 source_model = prepare_source_model_task(region_geojson) # 任务2：运行PSHA计算（依赖任务1的输出） calculation_id = run_psha_task(source_model, imts, investigation_time) # 任务3：提取结果（依赖任务2的输出） curves_data = extract_results_task(calculation_id, result_type="hazard_curve") # 任务4：可视化（依赖任务3的输出） map_image = visualize_hazard_map_task(curves_data, metric="PGA", return_period=475) return map_image # 这个flow可以被智能体核心层调用，也可以由Prefect UI/Scheduler触发

使用Prefect后，我们获得了开箱即用的任务监控、日志集中管理、历史记录和强大的调度能力。智能体核心层的“工作流执行引擎”就简化为一个Prefect流的调用器。

4. 实操部署与配置指南

4.1 本地开发环境搭建

要让这个智能体系统跑起来，你需要一个能运行OpenQuake引擎和各个微服务的环境。我强烈推荐使用Docker Compose进行本地开发和测试，它能解决复杂的依赖问题。

1. 准备目录结构：

   seismic-agent/ ├── docker-compose.yml ├── agent-core/ # 智能体核心层（Web GUI/CLI） │ ├── Dockerfile │ └── src/ ├── mcp-adapters/ # 各个MCP适配器 │ ├── openquake-adapter/ │ │ ├── Dockerfile │ │ └── app/ │ └──>version: '3.8' services: postgres: image: postgres:15 environment: POSTGRES_USER: agent POSTGRES_PASSWORD: securepassword POSTGRES_DB: seismic_agent volumes: - ./storage/postgres_data:/var/lib/postgresql/data ports: - "5432:5432" prefect-server: image: prefecthq/prefect:2-python3.11 command: prefect server start environment: PREFECT_API_URL: http://prefect-server:4200/api PREFECT_UI_URL: http://localhost:4200 ports: - "4200:4200" depends_on: - postgres openquake-adapter: build: ./mcp-adapters/openquake-adapter environment: OQ_DATABASE: postgresql://agent:securepassword@postgres/seismic_agent OQ_DATA_DIR: /oqdata volumes: - ./storage/oqdata:/oqdata # 挂载OpenQuake所需的数据和结果目录 ports: - "8001:8000" # 暴露MCP API # 注意：OpenQuake引擎本身已包含在适配器镜像中 >cd seismic-agent docker-compose build docker-compose up -d

   启动后，你可以访问：

   • http://localhost:8501：智能体Web界面。
   • http://localhost:4200：Prefect UI，监控工作流执行。
   • http://localhost:8001/api/mcp/v1/models：测试OpenQuake MCP适配器是否正常。

4.2 OpenQuake MCP适配器配置详解

适配器的Dockerfile需要基于一个包含OpenQuake引擎的镜像。

# mcp-adapters/openquake-adapter/Dockerfile FROM openquake/engine:latest WORKDIR /app # 安装Python依赖（FastAPI, pydantic等） COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制适配器应用代码 COPY ./app . # 开放MCP服务端口 EXPOSE 8000 # 启动命令 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

在app/main.py中，你需要实现前面提到的MCP API。关键点在于如何与OpenQuake交互。OpenQuake引擎通常通过命令行oq engine --run job.ini或Python APIopenquake.commands.run.run_engine(job.ini)来调用。在适配器中，我推荐使用Python API，因为它更容易捕获输出和错误信息。

一个重要配置是OpenQuake的数据目录。OpenQuake需要访问其内置的GSIM库、脆弱性模型库等。在Docker中，你需要将这些数据卷挂载到容器内OpenQuake期望的路径（通常是/opt/openquake下的某个子目录），或者通过环境变量OQ_DATABASE和OQ_DATA_DIR来指定。

4.3 智能体Web界面快速原型

对于Web界面，快速原型阶段我推荐使用Streamlit。它能让数据科学家和工程师用Python快速构建交互式应用，非常适合我们这种需要复杂参数输入和实时可视化的场景。

# agent-core/src/app.py (Streamlit示例) import streamlit as st import requests import json st.set_page_config(page_title="地震风险智能分析平台", layout="wide") st.title("🌍 端到端概率地震危险与风险分析") # 侧边栏：分析类型选择 analysis_type = st.sidebar.selectbox( "选择分析类型", ["区域概率危险性分析 (PSHA)", "概率风险分析 (PSRA)", "情景损失评估"] ) # 主区域 if analysis_type == "区域概率危险性分析 (PSHA)": st.header("配置PSHA计算参数") col1, col2 = st.columns(2) with col1: # 地图组件选择区域（这里简化用经纬度框代替） st.subheader("研究区域") min_lon = st.number_input("最小经度", value=-122.5) max_lon = st.number_input("最大经度", value=-122.0) min_lat = st.number_input("最小纬度", value=37.5) max_lat = st.number_input("最大纬度", value=38.0) region_geojson = { "type": "Polygon", "coordinates": [[[min_lon, min_lat], ...]] } with col2: st.subheader("计算参数") investigation_time = st.slider("调查时间 (年)", 1, 10000, 50) imts = st.multiselect( "强度测量类型", ["PGA", "SA(0.2)", "SA(0.5)", "SA(1.0)", "SA(2.0)"], default=["PGA", "SA(0.2)"] ) source_model = st.selectbox( "地震源模型", ["USGS California 2014", "GEM Global 2020", "自定义上传..."] ) if st.button("🚀 启动分析", type="primary"): # 1. 构建任务参数 inputs = { "region": region_geojson, "investigation_time": investigation_time, "imts": imts, "source_model": source_model } # 2. 调用智能体核心层的API（或直接调用Prefect流） # 这里假设智能体核心层提供了一个启动工作流的REST端点 with st.spinner("正在提交分析任务..."): try: response = requests.post( "http://agent-core:8000/api/workflows/psha/run", json={"inputs": inputs} ) response.raise_for_status() workflow_id = response.json()["workflow_id"] st.success(f"任务提交成功！工作流ID: {workflow_id}") st.info(f"前往 [Prefect UI](http://localhost:4200) 监控执行详情。") # 可以轮询结果，或者提供一个链接让用户稍后查看 st.session_state['last_workflow_id'] = workflow_id except requests.exceptions.RequestException as e: st.error(f"提交任务失败: {e}") # 另一个标签页用于查看结果 if 'last_workflow_id' in st.session_state: if st.sidebar.button("查看上次分析结果"): # 调用API获取结果 result = requests.get(f"http://agent-core:8000/api/results/{st.session_state['last_workflow_id']}").json() if result["status"] == "SUCCESS": st.image(result["map_image"]) # 显示生成的危险性图 st.download_button("下载危险性曲线数据", data=result["curves_csv"], file_name="hazard_curves.csv") else: st.warning("分析仍在进行中或失败。")

这个Streamlit应用提供了基本的参数输入和任务触发界面。智能体核心层（另一个服务）接收到这个请求后，会将其转化为Prefect流调用，并管理整个工作流的执行。

5. 常见问题与实战避坑指南

在实际开发和测试这个系统的过程中，我遇到了不少坑。这里总结一些典型问题和解决方案，希望能帮你节省时间。

5.1 OpenQuake引擎集成问题

问题1：OpenQuake计算耗时极长，导致HTTP请求超时。

• 现象：提交PSHA任务后，MCP适配器的/execute接口一直不返回，最终前端收到超时错误，但后端计算可能仍在进行。
• 解决方案：这就是为什么我们必须采用异步任务模式。MCP适配器的/execute端点应该只负责验证输入、生成唯一任务ID、将计算任务丢到后台队列（如Celery、RQ，或直接用asyncio.create_task），然后立即返回executionId和状态RUNNING。计算状态通过另一个端点/executions/{id}查询。
• 实操细节：在适配器内使用asyncio.to_thread或concurrent.futures.ThreadPoolExecutor来在单独线程中运行同步的run_engine调用，避免阻塞事件循环。

问题2：OpenQuake引擎对内存和CPU需求高，单个容器资源不足。

• 现象：计算区域较大或网格较密时，容器因OOM（内存溢出）被杀死。
• 解决方案：
  1. 资源限制与分配：在docker-compose.yml中为openquake-adapter服务设置资源限制，并确保宿主机有足够资源。

     openquake-adapter: deploy: resources: limits: cpus: '4' memory: 16G reservations: memory: 8G

  2. 分布式计算：对于超大规模计算，需要配置OpenQuake集群模式。这超出了单个MCP适配器的范围。一种方案是让MCP适配器作为一个“调度器”，将计算任务提交到一个独立的、已配置好的OpenQuake集群（通过SSH或集群作业管理系统如Slurm）。适配器只需等待集群返回结果。

问题3：OpenQuake的配置文件（job.ini）复杂且容易出错。

• 现象：用户通过智能体界面输入的参数，转换成job.ini后，OpenQuake报出晦涩的配置错误。
• 解决方案：
  1. Schema驱动配置生成：充分利用MCP的get_model_schema功能。Schema不仅定义参数类型，还可以包含更丰富的约束和逻辑。例如，当calculation_mode为classical时，number_of_logic_tree_samples字段应该隐藏或禁用。
  2. 配置模板与验证：在适配器内部维护一组经过充分测试的、针对不同分析类型的配置模板。用户输入的高层参数（如“使用USGS加州源模型”）映射到具体的模板文件路径和参数覆盖。在生成最终job.ini前，使用OpenQuake提供的oq info --check-config job.ini命令进行预验证，将友好的错误信息返回给用户。
  3. 提供配置预览：在智能体界面中，提供一个“高级”选项卡，展示即将生成的job.ini内容，供专家用户复核和微调。

5.2 数据管理与传递难题

问题4：大文件（如高精度暴露数据库、场地网格文件）如何高效传递？

• 现象：通过MCP API的JSON body上传数GB的文件不切实际。
• 解决方案：MCP协议本身支持文件传递，通常采用“先上传，后引用”的模式。
  1. 智能体界面提供文件上传功能，将文件上传到一个共享的、所有服务都能访问的对象存储（如MinIO、AWS S3）或网络文件系统（NFS）。
  2. 上传后获得一个文件URI（如s3://my-bucket/exposure.xml）。
  3. 调用MCP的execute_model时，在inputs中传递这个URI，而不是文件内容。
  4. MCP适配器在执行前，根据URI从共享存储中下载文件到本地临时目录。
• 架构调整：在docker-compose.yml中增加一个MinIO服务，并让所有相关容器都将MinIO的存储桶挂载为卷或通过SDK访问。

问题5：中间计算结果（如地震动场）数据量大，如何在工作流步骤间传递？

• 现象：PSHA步骤产生的HDF5文件可能很大，直接作为参数传递给下一个PSRA步骤效率低下。
• 解决方案：不要传递数据本身，传递数据引用。
  1. PSHA的MCP适配器在计算完成后，将结果HDF5文件保存到共享存储，并在返回的execution_result中包含其URI。
  2. 工作流引擎（如Prefect）将这个URI作为输出，传递给下一个任务（PSRA）。
  3. PSRA的MCP适配器根据收到的URI去读取HDF5文件。
  4. 这种模式也便于结果的缓存和复用。如果相同的PSHA计算已经执行过，可以直接引用已有结果，跳过重复计算。

5.3 系统可靠性与用户体验

问题6：计算任务失败，如何快速定位问题？

• 现象：用户只看到“任务失败”，不知道是参数错误、数据问题还是系统故障。
• 解决方案：建立集中化日志和错误追踪系统。
  1. 所有服务（适配器、智能体核心）的日志都统一输出到stdout/stderr。
  2. 使用Docker的日志驱动，或使用如Loki+Promtail+Grafana或ELK栈来收集和索引所有容器的日志。
  3. 每个计算任务（executionId）关联一个唯一的correlation_id，并贯穿所有相关的日志条目。这样在Grafana中可以通过correlation_id轻松过滤出该任务在所有微服务中的完整日志链条。
  4. MCP适配器在捕获到OpenQuake引擎的错误输出时，应尝试解析其关键信息（如第几行配置出错），并将其转化为更友好的错误消息，通过/executions/{id}端点返回。

问题7：用户想中途修改参数或取消长时间运行的计算。

• 现象：一个全国尺度的风险分析可能要算好几天，用户发现起始参数设错了，无法停止。
• 解决方案：为MCP适配器增加cancel_model操作。

  @app.post("/api/mcp/v1/executions/{execution_id}/cancel") async def cancel_execution(execution_id: str): if execution_id not in executions: raise HTTPException(status_code=404, detail="Execution not found") # 查找该任务对应的OpenQuake计算进程 # 发送终止信号（如SIGTERM） # 更新任务状态为CANCELLED executions[execution_id]["status"] = "CANCELLED" return {"status": "CANCELLED"}

  智能体界面上的每个运行中任务旁边，都应该有一个“取消”按钮。更复杂的场景下，还需要支持“从检查点重启”或“保存中间状态”，但这需要OpenQuake引擎本身的支持，实现难度较大。

构建这样一个“智能体界面”是一个持续迭代的过程。我从一个简单的、只能跑通单次PSHA的命令行脚本开始，逐步添加了Web界面、工作流编排、错误处理、结果可视化。MCP协议的概念让我能清晰地划分边界，Prefect这样的工具让工作流管理变得可靠。现在，团队里的地震工程师们已经可以完全通过浏览器来完成大部分标准分析，而我可以将更多精力投入到优化算法和集成更先进的模型中。这个系统的价值不在于替代OpenQuake，而在于让OpenQuake的强大能力，能够被更顺畅、更高效地释放出来。