1. 项目概述一个为地震风险分析赋能的“智能体界面”如果你从事地震工程、灾害风险评估或者城市规划相关的工作大概率对“概率地震危险性分析”和“风险分析”这两个词不会陌生。它们是我们评估一个地区未来可能遭受地震打击的“标尺”是制定抗震设计规范、进行保险定价和规划应急资源的核心依据。然而这个分析过程本身却常常让从业者感到头疼。它涉及海量的数据地震目录、活动断层、场地条件、复杂的模型地震动预测方程、脆弱性曲线以及繁琐的计算流程。传统的做法要么依赖封闭的商业软件操作不透明且扩展性差要么就是自己写脚本调用像OpenQuake这样的开源引擎但这对非编程背景的工程师来说门槛太高且流程管理容易混乱。我最近花了不少时间尝试构建一个东西来解决这个痛点。我称之为“一个用于端到端概率地震危险性与风险分析的智能体界面”。听起来有点学术其实核心思想很直接我想做一个“智能中间人”。它不是一个全新的计算引擎而是一个建立在现有强大开源工具如GEM的OpenQuake引擎之上的、高度自动化和智能化的操作界面。这个界面的目标是让用户能够用更自然、更高效的方式描述他们想要分析的问题比如“帮我评估一下这个工业园区未来50年因地震导致直接经济损失超过10亿的概率”然后由这个“智能体”自动完成从数据准备、模型配置、计算提交到结果提取与可视化的全链条工作。这里的关键词是“Agentic Interface”和“Model Context Protocol”。前者意味着这个界面具备一定的自主性和决策能力能理解用户意图并分解任务后者则是一个新兴的、用于标准化模型交互的协议可以把它想象成模型之间的“通用插头”让我的智能体能轻松地“插入”OpenQuake引擎或其他兼容的分析工具无需关心底层复杂的API细节。通过结合这两者我希望能把专家从重复性的流程操作中解放出来更专注于分析逻辑和结果解读同时也为新手或跨领域研究者打开一扇便捷的大门。2. 核心设计思路为什么是“智能体”“协议”2.1 传统工作流的痛点与“智能体”的价值定位在深入技术细节前我们有必要先看看当前典型的PSHA概率地震危险性分析和PSRA概率地震风险分析工作流是怎样的。通常它包含以下几个阶段数据收集与预处理收集研究区的地震目录、活动断层数据、地震动预测方程GMPE、场地放大模型、暴露数据库建筑物、人口、资产清单和脆弱性/风险函数。这些数据格式各异来源不一清洗和标准化耗时耗力。模型配置与参数化在OpenQuake引擎中你需要编写复杂的XML或INI格式的配置文件job.ini精确地定义逻辑树用于处理模型不确定性、计算参数、输出格式等。一个配置文件的错误可能导致数小时甚至数天的计算白费。计算执行与监控提交计算任务到本地服务器或高性能计算集群。计算可能持续数小时至数天需要手动监控日志处理可能出现的运行错误或资源不足问题。结果提取与后处理计算完成后会生成大量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客户端就可以用一套统一的“语言”与之对话。这样做带来了几个关键优势解耦与灵活性智能体不再关心OpenQuake内部如何实现。未来如果集成了新的地震动模拟器或风险计算模块只要它们也支持MCP智能体就能无缝接入。自描述性智能体可以通过get_model_schema动态了解每个计算任务需要什么参数、参数的类型和约束。这使得智能体可以构建动态的、智能的表单来引导用户输入甚至进行输入验证。标准化通信所有交互都基于标准的JSON-RPC over WebSocket或HTTP便于调试、监控和集成到更大的工作流系统中。注意MCP是一个新兴协议OpenQuake引擎本身并未原生支持。在实际实现中我通常需要为OpenQuake编写一个轻量级的“MCP适配器”服务。这个服务封装了对OpenQuake API的调用并将其暴露为MCP标准接口。这是整个架构中的关键开发环节。2.3 端到端流程的自动化编排有了MCP提供的标准化操作接口智能体就可以编排一个完整的端到端分析流程。这不仅仅是单个计算任务的执行而是一个包含多个步骤、可能有条件分支的工作流。例如一个完整的区域地震风险分析可能包含先进行概率地震危险性分析PSHA生成地震动场。然后利用PSHA的结果结合暴露和脆弱性模型进行概率风险分析PSRA。最后基于风险结果生成热力图和统计报告。智能体内部需要有一个“工作流引擎”或“任务编排器”。它可以是一个简单的状态机也可以利用像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(titleOpenQuake 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_code404, detailModel 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_code404, detailExecution 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_code404, detailModel 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客户端就足够了。但对于复杂、有条件分支、可并行执行的工作流需要一个更健壮的引擎。我评估了两种方案轻量级自制状态机使用Python的networkx库表示DAG然后按拓扑顺序执行每个节点。节点执行器调用对应的MCP客户端。需要自己处理错误重试、超时、依赖传递。优点是轻量、可控。采用成熟工作流引擎如Prefect。Prefect的“流”和“任务”抽象与我们的概念非常契合。每个MCP调用可以定义为一个Prefect任务工作流就是一个Prefect流。Prefect原生支持任务依赖、参数化、重试策略、结果持久化、分布式执行和丰富的UI监控。这大大减少了我们自己造轮子的工作量。我最终选择了Prefect 2.x版本。将之前YAML定义的工作流转化为Prefect的Python DSLfrom prefect import flow, task from mcp_client import MCPClient # 假设的MCP客户端库 client MCPClient(http://openquake-adapter:8000) task(retries3, retry_delay_seconds10) 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(fPSHA计算失败: {status.get(error)}) return execution[executionId] flow(nameregional-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_typehazard_curve) # 任务4可视化依赖任务3的输出 map_image visualize_hazard_map_task(curves_data, metricPGA, return_period475) return map_image # 这个flow可以被智能体核心层调用也可以由Prefect UI/Scheduler触发使用Prefect后我们获得了开箱即用的任务监控、日志集中管理、历史记录和强大的调度能力。智能体核心层的“工作流执行引擎”就简化为一个Prefect流的调用器。4. 实操部署与配置指南4.1 本地开发环境搭建要让这个智能体系统跑起来你需要一个能运行OpenQuake引擎和各个微服务的环境。我强烈推荐使用Docker Compose进行本地开发和测试它能解决复杂的依赖问题。准备目录结构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:securepasswordpostgres/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:4200Prefect 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地震风险智能分析平台, layoutwide) 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(最小纬度, value37.5) max_lat st.number_input(最大纬度, value38.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( 启动分析, typeprimary): # 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(fhttp://agent-core:8000/api/results/{st.session_state[last_workflow_id]}).json() if result[status] SUCCESS: st.image(result[map_image]) # 显示生成的危险性图 st.download_button(下载危险性曲线数据, dataresult[curves_csv], file_namehazard_curves.csv) else: st.warning(分析仍在进行中或失败。)这个Streamlit应用提供了基本的参数输入和任务触发界面。智能体核心层另一个服务接收到这个请求后会将其转化为Prefect流调用并管理整个工作流的执行。5. 常见问题与实战避坑指南在实际开发和测试这个系统的过程中我遇到了不少坑。这里总结一些典型问题和解决方案希望能帮你节省时间。5.1 OpenQuake引擎集成问题问题1OpenQuake计算耗时极长导致HTTP请求超时。现象提交PSHA任务后MCP适配器的/execute接口一直不返回最终前端收到超时错误但后端计算可能仍在进行。解决方案这就是为什么我们必须采用异步任务模式。MCP适配器的/execute端点应该只负责验证输入、生成唯一任务ID、将计算任务丢到后台队列如Celery、RQ或直接用asyncio.create_task然后立即返回executionId和状态RUNNING。计算状态通过另一个端点/executions/{id}查询。实操细节在适配器内使用asyncio.to_thread或concurrent.futures.ThreadPoolExecutor来在单独线程中运行同步的run_engine调用避免阻塞事件循环。问题2OpenQuake引擎对内存和CPU需求高单个容器资源不足。现象计算区域较大或网格较密时容器因OOM内存溢出被杀死。解决方案资源限制与分配在docker-compose.yml中为openquake-adapter服务设置资源限制并确保宿主机有足够资源。openquake-adapter: deploy: resources: limits: cpus: 4 memory: 16G reservations: memory: 8G分布式计算对于超大规模计算需要配置OpenQuake集群模式。这超出了单个MCP适配器的范围。一种方案是让MCP适配器作为一个“调度器”将计算任务提交到一个独立的、已配置好的OpenQuake集群通过SSH或集群作业管理系统如Slurm。适配器只需等待集群返回结果。问题3OpenQuake的配置文件job.ini复杂且容易出错。现象用户通过智能体界面输入的参数转换成job.ini后OpenQuake报出晦涩的配置错误。解决方案Schema驱动配置生成充分利用MCP的get_model_schema功能。Schema不仅定义参数类型还可以包含更丰富的约束和逻辑。例如当calculation_mode为classical时number_of_logic_tree_samples字段应该隐藏或禁用。配置模板与验证在适配器内部维护一组经过充分测试的、针对不同分析类型的配置模板。用户输入的高层参数如“使用USGS加州源模型”映射到具体的模板文件路径和参数覆盖。在生成最终job.ini前使用OpenQuake提供的oq info --check-config job.ini命令进行预验证将友好的错误信息返回给用户。提供配置预览在智能体界面中提供一个“高级”选项卡展示即将生成的job.ini内容供专家用户复核和微调。5.2 数据管理与传递难题问题4大文件如高精度暴露数据库、场地网格文件如何高效传递现象通过MCP API的JSON body上传数GB的文件不切实际。解决方案MCP协议本身支持文件传递通常采用“先上传后引用”的模式。智能体界面提供文件上传功能将文件上传到一个共享的、所有服务都能访问的对象存储如MinIO、AWS S3或网络文件系统NFS。上传后获得一个文件URI如s3://my-bucket/exposure.xml。调用MCP的execute_model时在inputs中传递这个URI而不是文件内容。MCP适配器在执行前根据URI从共享存储中下载文件到本地临时目录。架构调整在docker-compose.yml中增加一个MinIO服务并让所有相关容器都将MinIO的存储桶挂载为卷或通过SDK访问。问题5中间计算结果如地震动场数据量大如何在工作流步骤间传递现象PSHA步骤产生的HDF5文件可能很大直接作为参数传递给下一个PSRA步骤效率低下。解决方案不要传递数据本身传递数据引用。PSHA的MCP适配器在计算完成后将结果HDF5文件保存到共享存储并在返回的execution_result中包含其URI。工作流引擎如Prefect将这个URI作为输出传递给下一个任务PSRA。PSRA的MCP适配器根据收到的URI去读取HDF5文件。这种模式也便于结果的缓存和复用。如果相同的PSHA计算已经执行过可以直接引用已有结果跳过重复计算。5.3 系统可靠性与用户体验问题6计算任务失败如何快速定位问题现象用户只看到“任务失败”不知道是参数错误、数据问题还是系统故障。解决方案建立集中化日志和错误追踪系统。所有服务适配器、智能体核心的日志都统一输出到stdout/stderr。使用Docker的日志驱动或使用如LokiPromtailGrafana或ELK栈来收集和索引所有容器的日志。每个计算任务executionId关联一个唯一的correlation_id并贯穿所有相关的日志条目。这样在Grafana中可以通过correlation_id轻松过滤出该任务在所有微服务中的完整日志链条。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_code404, detailExecution not found) # 查找该任务对应的OpenQuake计算进程 # 发送终止信号如SIGTERM # 更新任务状态为CANCELLED executions[execution_id][status] CANCELLED return {status: CANCELLED}智能体界面上的每个运行中任务旁边都应该有一个“取消”按钮。更复杂的场景下还需要支持“从检查点重启”或“保存中间状态”但这需要OpenQuake引擎本身的支持实现难度较大。构建这样一个“智能体界面”是一个持续迭代的过程。我从一个简单的、只能跑通单次PSHA的命令行脚本开始逐步添加了Web界面、工作流编排、错误处理、结果可视化。MCP协议的概念让我能清晰地划分边界Prefect这样的工具让工作流管理变得可靠。现在团队里的地震工程师们已经可以完全通过浏览器来完成大部分标准分析而我可以将更多精力投入到优化算法和集成更先进的模型中。这个系统的价值不在于替代OpenQuake而在于让OpenQuake的强大能力能够被更顺畅、更高效地释放出来。 
长河君 · 2026/8/21 22:51:44 · 阅读 1.2k 基于MCP协议与智能体技术的地震风险分析平台构建实践
1. 项目概述一个为地震风险分析赋能的“智能体界面”如果你从事地震工程、灾害风险评估或者城市规划相关的工作大概率对“概率地震危险性分析”和“风险分析”这两个词不会陌生。它们是我们评估一个地区未来可能遭受地震打击的“标尺”是制定抗震设计规范、进行保险定价和规划应急资源的核心依据。然而这个分析过程本身却常常让从业者感到头疼。它涉及海量的数据地震目录、活动断层、场地条件、复杂的模型地震动预测方程、脆弱性曲线以及繁琐的计算流程。传统的做法要么依赖封闭的商业软件操作不透明且扩展性差要么就是自己写脚本调用像OpenQuake这样的开源引擎但这对非编程背景的工程师来说门槛太高且流程管理容易混乱。我最近花了不少时间尝试构建一个东西来解决这个痛点。我称之为“一个用于端到端概率地震危险性与风险分析的智能体界面”。听起来有点学术其实核心思想很直接我想做一个“智能中间人”。它不是一个全新的计算引擎而是一个建立在现有强大开源工具如GEM的OpenQuake引擎之上的、高度自动化和智能化的操作界面。这个界面的目标是让用户能够用更自然、更高效的方式描述他们想要分析的问题比如“帮我评估一下这个工业园区未来50年因地震导致直接经济损失超过10亿的概率”然后由这个“智能体”自动完成从数据准备、模型配置、计算提交到结果提取与可视化的全链条工作。这里的关键词是“Agentic Interface”和“Model Context Protocol”。前者意味着这个界面具备一定的自主性和决策能力能理解用户意图并分解任务后者则是一个新兴的、用于标准化模型交互的协议可以把它想象成模型之间的“通用插头”让我的智能体能轻松地“插入”OpenQuake引擎或其他兼容的分析工具无需关心底层复杂的API细节。通过结合这两者我希望能把专家从重复性的流程操作中解放出来更专注于分析逻辑和结果解读同时也为新手或跨领域研究者打开一扇便捷的大门。2. 核心设计思路为什么是“智能体”“协议”2.1 传统工作流的痛点与“智能体”的价值定位在深入技术细节前我们有必要先看看当前典型的PSHA概率地震危险性分析和PSRA概率地震风险分析工作流是怎样的。通常它包含以下几个阶段数据收集与预处理收集研究区的地震目录、活动断层数据、地震动预测方程GMPE、场地放大模型、暴露数据库建筑物、人口、资产清单和脆弱性/风险函数。这些数据格式各异来源不一清洗和标准化耗时耗力。模型配置与参数化在OpenQuake引擎中你需要编写复杂的XML或INI格式的配置文件job.ini精确地定义逻辑树用于处理模型不确定性、计算参数、输出格式等。一个配置文件的错误可能导致数小时甚至数天的计算白费。计算执行与监控提交计算任务到本地服务器或高性能计算集群。计算可能持续数小时至数天需要手动监控日志处理可能出现的运行错误或资源不足问题。结果提取与后处理计算完成后会生成大量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客户端就可以用一套统一的“语言”与之对话。这样做带来了几个关键优势解耦与灵活性智能体不再关心OpenQuake内部如何实现。未来如果集成了新的地震动模拟器或风险计算模块只要它们也支持MCP智能体就能无缝接入。自描述性智能体可以通过get_model_schema动态了解每个计算任务需要什么参数、参数的类型和约束。这使得智能体可以构建动态的、智能的表单来引导用户输入甚至进行输入验证。标准化通信所有交互都基于标准的JSON-RPC over WebSocket或HTTP便于调试、监控和集成到更大的工作流系统中。注意MCP是一个新兴协议OpenQuake引擎本身并未原生支持。在实际实现中我通常需要为OpenQuake编写一个轻量级的“MCP适配器”服务。这个服务封装了对OpenQuake API的调用并将其暴露为MCP标准接口。这是整个架构中的关键开发环节。2.3 端到端流程的自动化编排有了MCP提供的标准化操作接口智能体就可以编排一个完整的端到端分析流程。这不仅仅是单个计算任务的执行而是一个包含多个步骤、可能有条件分支的工作流。例如一个完整的区域地震风险分析可能包含先进行概率地震危险性分析PSHA生成地震动场。然后利用PSHA的结果结合暴露和脆弱性模型进行概率风险分析PSRA。最后基于风险结果生成热力图和统计报告。智能体内部需要有一个“工作流引擎”或“任务编排器”。它可以是一个简单的状态机也可以利用像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(titleOpenQuake 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_code404, detailModel 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_code404, detailExecution 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_code404, detailModel 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客户端就足够了。但对于复杂、有条件分支、可并行执行的工作流需要一个更健壮的引擎。我评估了两种方案轻量级自制状态机使用Python的networkx库表示DAG然后按拓扑顺序执行每个节点。节点执行器调用对应的MCP客户端。需要自己处理错误重试、超时、依赖传递。优点是轻量、可控。采用成熟工作流引擎如Prefect。Prefect的“流”和“任务”抽象与我们的概念非常契合。每个MCP调用可以定义为一个Prefect任务工作流就是一个Prefect流。Prefect原生支持任务依赖、参数化、重试策略、结果持久化、分布式执行和丰富的UI监控。这大大减少了我们自己造轮子的工作量。我最终选择了Prefect 2.x版本。将之前YAML定义的工作流转化为Prefect的Python DSLfrom prefect import flow, task from mcp_client import MCPClient # 假设的MCP客户端库 client MCPClient(http://openquake-adapter:8000) task(retries3, retry_delay_seconds10) 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(fPSHA计算失败: {status.get(error)}) return execution[executionId] flow(nameregional-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_typehazard_curve) # 任务4可视化依赖任务3的输出 map_image visualize_hazard_map_task(curves_data, metricPGA, return_period475) return map_image # 这个flow可以被智能体核心层调用也可以由Prefect UI/Scheduler触发使用Prefect后我们获得了开箱即用的任务监控、日志集中管理、历史记录和强大的调度能力。智能体核心层的“工作流执行引擎”就简化为一个Prefect流的调用器。4. 实操部署与配置指南4.1 本地开发环境搭建要让这个智能体系统跑起来你需要一个能运行OpenQuake引擎和各个微服务的环境。我强烈推荐使用Docker Compose进行本地开发和测试它能解决复杂的依赖问题。准备目录结构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:securepasswordpostgres/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:4200Prefect 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地震风险智能分析平台, layoutwide) 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(最小纬度, value37.5) max_lat st.number_input(最大纬度, value38.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( 启动分析, typeprimary): # 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(fhttp://agent-core:8000/api/results/{st.session_state[last_workflow_id]}).json() if result[status] SUCCESS: st.image(result[map_image]) # 显示生成的危险性图 st.download_button(下载危险性曲线数据, dataresult[curves_csv], file_namehazard_curves.csv) else: st.warning(分析仍在进行中或失败。)这个Streamlit应用提供了基本的参数输入和任务触发界面。智能体核心层另一个服务接收到这个请求后会将其转化为Prefect流调用并管理整个工作流的执行。5. 常见问题与实战避坑指南在实际开发和测试这个系统的过程中我遇到了不少坑。这里总结一些典型问题和解决方案希望能帮你节省时间。5.1 OpenQuake引擎集成问题问题1OpenQuake计算耗时极长导致HTTP请求超时。现象提交PSHA任务后MCP适配器的/execute接口一直不返回最终前端收到超时错误但后端计算可能仍在进行。解决方案这就是为什么我们必须采用异步任务模式。MCP适配器的/execute端点应该只负责验证输入、生成唯一任务ID、将计算任务丢到后台队列如Celery、RQ或直接用asyncio.create_task然后立即返回executionId和状态RUNNING。计算状态通过另一个端点/executions/{id}查询。实操细节在适配器内使用asyncio.to_thread或concurrent.futures.ThreadPoolExecutor来在单独线程中运行同步的run_engine调用避免阻塞事件循环。问题2OpenQuake引擎对内存和CPU需求高单个容器资源不足。现象计算区域较大或网格较密时容器因OOM内存溢出被杀死。解决方案资源限制与分配在docker-compose.yml中为openquake-adapter服务设置资源限制并确保宿主机有足够资源。openquake-adapter: deploy: resources: limits: cpus: 4 memory: 16G reservations: memory: 8G分布式计算对于超大规模计算需要配置OpenQuake集群模式。这超出了单个MCP适配器的范围。一种方案是让MCP适配器作为一个“调度器”将计算任务提交到一个独立的、已配置好的OpenQuake集群通过SSH或集群作业管理系统如Slurm。适配器只需等待集群返回结果。问题3OpenQuake的配置文件job.ini复杂且容易出错。现象用户通过智能体界面输入的参数转换成job.ini后OpenQuake报出晦涩的配置错误。解决方案Schema驱动配置生成充分利用MCP的get_model_schema功能。Schema不仅定义参数类型还可以包含更丰富的约束和逻辑。例如当calculation_mode为classical时number_of_logic_tree_samples字段应该隐藏或禁用。配置模板与验证在适配器内部维护一组经过充分测试的、针对不同分析类型的配置模板。用户输入的高层参数如“使用USGS加州源模型”映射到具体的模板文件路径和参数覆盖。在生成最终job.ini前使用OpenQuake提供的oq info --check-config job.ini命令进行预验证将友好的错误信息返回给用户。提供配置预览在智能体界面中提供一个“高级”选项卡展示即将生成的job.ini内容供专家用户复核和微调。5.2 数据管理与传递难题问题4大文件如高精度暴露数据库、场地网格文件如何高效传递现象通过MCP API的JSON body上传数GB的文件不切实际。解决方案MCP协议本身支持文件传递通常采用“先上传后引用”的模式。智能体界面提供文件上传功能将文件上传到一个共享的、所有服务都能访问的对象存储如MinIO、AWS S3或网络文件系统NFS。上传后获得一个文件URI如s3://my-bucket/exposure.xml。调用MCP的execute_model时在inputs中传递这个URI而不是文件内容。MCP适配器在执行前根据URI从共享存储中下载文件到本地临时目录。架构调整在docker-compose.yml中增加一个MinIO服务并让所有相关容器都将MinIO的存储桶挂载为卷或通过SDK访问。问题5中间计算结果如地震动场数据量大如何在工作流步骤间传递现象PSHA步骤产生的HDF5文件可能很大直接作为参数传递给下一个PSRA步骤效率低下。解决方案不要传递数据本身传递数据引用。PSHA的MCP适配器在计算完成后将结果HDF5文件保存到共享存储并在返回的execution_result中包含其URI。工作流引擎如Prefect将这个URI作为输出传递给下一个任务PSRA。PSRA的MCP适配器根据收到的URI去读取HDF5文件。这种模式也便于结果的缓存和复用。如果相同的PSHA计算已经执行过可以直接引用已有结果跳过重复计算。5.3 系统可靠性与用户体验问题6计算任务失败如何快速定位问题现象用户只看到“任务失败”不知道是参数错误、数据问题还是系统故障。解决方案建立集中化日志和错误追踪系统。所有服务适配器、智能体核心的日志都统一输出到stdout/stderr。使用Docker的日志驱动或使用如LokiPromtailGrafana或ELK栈来收集和索引所有容器的日志。每个计算任务executionId关联一个唯一的correlation_id并贯穿所有相关的日志条目。这样在Grafana中可以通过correlation_id轻松过滤出该任务在所有微服务中的完整日志链条。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_code404, detailExecution not found) # 查找该任务对应的OpenQuake计算进程 # 发送终止信号如SIGTERM # 更新任务状态为CANCELLED executions[execution_id][status] CANCELLED return {status: CANCELLED}智能体界面上的每个运行中任务旁边都应该有一个“取消”按钮。更复杂的场景下还需要支持“从检查点重启”或“保存中间状态”但这需要OpenQuake引擎本身的支持实现难度较大。构建这样一个“智能体界面”是一个持续迭代的过程。我从一个简单的、只能跑通单次PSHA的命令行脚本开始逐步添加了Web界面、工作流编排、错误处理、结果可视化。MCP协议的概念让我能清晰地划分边界Prefect这样的工具让工作流管理变得可靠。现在团队里的地震工程师们已经可以完全通过浏览器来完成大部分标准分析而我可以将更多精力投入到优化算法和集成更先进的模型中。这个系统的价值不在于替代OpenQuake而在于让OpenQuake的强大能力能够被更顺畅、更高效地释放出来。