如何通过自定义步骤实现扩展开发？

第十八章：自定义步骤与扩展开发 GeoPipeAgent 的步骤系统完全开放，通过 @step 装饰器可以轻松注册自定义步骤，扩展框架的 GIS 分析能力。本章介绍自定义步骤的开发、注册和使用方法。 18.1 步骤注册系统 @step 装饰器 所有步骤（内置和自定义）都通过 @step 装饰器注册到全局注册表（registry._steps）： from geopipe_agent.steps.registry import step from geopipe_agent.engine.context import StepContext from geopipe_agent.models.result import StepResult @step( id="my_category.my_action", # 步骤注册 ID（必填） name="我的自定义步骤", # 显示名称 description="步骤功能描述", # 步骤描述 category="my_category", # 类别（影响 list-steps 分组） params={ # 参数规范字典 "input": { "type": "geodataframe", "required": True, "description": "输入矢量数据", }, "my_param": { "type": "number", "required": False, "default": 100, "description": "自定义参数", }, }, outputs={ # 输出规范 "output": {"type": "geodataframe", "description": "处理结果"}, }, backends=["native_python"], # 支持的后端（可省略） examples=[ # 示例（用于 Skill 文档） { "description": "基础用法", "params": {"input": "$load.output", "my_param": 200}, }, ], ) def my_custom_step(ctx: StepContext) -> StepResult: """步骤实现函数，接收 StepContext，返回 StepResult""" gdf = ctx.input("input") # 获取 input 参数（GeoDataFrame） my_param = ctx.param("my_param", 100) # 获取可选参数 # 执行处理逻辑 result_gdf = gdf.copy() result_gdf["my_field"] = my_param return StepResult( output=result_gdf, stats={"feature_count": len(result_gdf)}, ) 注册 ID 规范 步骤 ID 使用点号分隔的 类别.动作 格式： 格式 示例 说明 category.action vector.buffer 推荐格式 org.category.action acme.vector.polygon_merge 带组织前缀，避免冲突 custom.action custom.fix_topology 通用自定义步骤 18.2 StepContext 用法 步骤函数通过 StepContext 获取所有已解析（变量和引用已替换）的参数： def my_step(ctx: StepContext) -> StepResult: # 获取参数（带默认值） gdf = ctx.param("input") # 等价于 ctx.input("input") distance = ctx.param("distance") # 必填参数，None 如不提供 cap = ctx.param("cap_style", "round") # 带默认值 # 简写：ctx.input() 等价于 ctx.param("input") gdf = ctx.input("input") # 访问后端（仅当 backends 列表非空时） backend = ctx.backend # GeoBackend 对象或 None # 访问完整参数字典 all_params = ctx.params 18.3 完整示例：开发空间连接步骤 以下是一个完整的自定义步骤实现：空间连接（Spatial Join）。 # my_steps/spatial_join.py from __future__ import annotations from geopipe_agent.steps.registry import step from geopipe_agent.engine.context import StepContext from geopipe_agent.models.result import StepResult @step( id="custom.spatial_join", name="空间连接", description="将右表的属性通过空间关系连接到左表", category="custom", params={ "left": { "type": "geodataframe", "required": True, "description": "左表（要素保留数量与左表一致）", }, "right": { "type": "geodataframe", "required": True, "description": "右表（属性来源）", }, "how": { "type": "string", "required": False, "default": "left", "enum": ["left", "right", "inner"], "description": "连接方式", }, "op": { "type": "string", "required": False, "default": "intersects", "enum": ["intersects", "contains", "within"], "description": "空间关系谓词", }, "suffix_left": { "type": "string", "required": False, "default": "_left", "description": "左表重名字段后缀", }, "suffix_right": { "type": "string", "required": False, "default": "_right", "description": "右表重名字段后缀", }, }, outputs={ "output": {"type": "geodataframe", "description": "空间连接结果"}, }, examples=[ { "description": "点-面空间连接（将行政区属性连接到 POI 点）", "params": { "left": "$load-poi.output", "right": "$load-districts.output", "how": "left", "op": "within", }, }, ], ) def custom_spatial_join(ctx: StepContext) -> StepResult: import geopandas as gpd left_gdf = ctx.param("left") right_gdf = ctx.param("right") how = ctx.param("how", "left") op = ctx.param("op", "intersects") suffix_left = ctx.param("suffix_left", "_left") suffix_right = ctx.param("suffix_right", "_right") # 执行空间连接 result_gdf = gpd.sjoin( left_gdf, right_gdf, how=how, predicate=op, lsuffix=suffix_left.lstrip("_"), rsuffix=suffix_right.lstrip("_"), ) stats = { "feature_count": len(result_gdf), "left_count": len(left_gdf), "right_count": len(right_gdf), "join_method": how, "spatial_predicate": op, } return StepResult(output=result_gdf, stats=stats) 18.4 自动发现机制 GeoPipeAgent 使用 pkgutil.walk_packages 在包初始化时自动发现并加载内置步骤： # src/geopipe_agent/__init__.py import pkgutil import importlib import geopipe_agent.steps as steps_pkg for finder, name, ispkg in pkgutil.walk_packages( steps_pkg.__path__, prefix=steps_pkg.__name__ + "." ): if not name.endswith("._") and not any(n.startswith("_") for n in name.split(".")): importlib.import_module(name) import geopipe_agent 时，所有 steps/ 子目录下的模块都被自动导入，@step 装饰器在模块导入时即注册步骤。 如何让自定义步骤也被自动发现： 方案一：直接 import（最简单） 在使用自定义步骤的 Python 脚本中，在 import geopipe_agent 之后导入自定义步骤模块： import geopipe_agent # 加载内置步骤 # 加载自定义步骤（触发 @step 注册） import my_steps.spatial_join import my_steps.polygon_merge from geopipe_agent.engine.parser import parse_yaml from geopipe_agent.engine.executor import execute_pipeline pipeline = parse_yaml("pipeline.yaml") report = execute_pipeline(pipeline) 方案二：创建插件包（可分发） 创建一个 Python 包，在其 __init__.py 中导入所有自定义步骤，通过 pip install 安装后自动注册： my-geopipe-steps/ ├── pyproject.toml └── src/ └── my_geopipe_steps/ ├── __init__.py # 导入所有步骤模块 ├── spatial_join.py └── polygon_merge.py __init__.py： # 导入所有步骤（触发 @step 注册） from . import spatial_join from . import polygon_merge 安装后，在 Python 脚本开头导入： import geopipe_agent import my_geopipe_steps # 注册自定义步骤 方案三：修改内置步骤（不推荐） 直接将自定义步骤放入 src/geopipe_agent/steps/custom/ 目录，自动发现机制会加载它们。仅适合本地修改，不适合分发。 18.5 开发规范与最佳实践 5.1 错误处理 步骤内的异常会被 executor 捕获并包装为 StepExecutionError，不需要在步骤中手动包装。但可以提供清晰的错误消息： def my_step(ctx: StepContext) -> StepResult: gdf = ctx.input("input") field = ctx.param("field") if field not in gdf.columns: raise ValueError( f"Field '{field}' not found in input data. " f"Available fields: {list(gdf.columns)}" ) # ... 处理逻辑 5.2 延迟导入重依赖 将可选依赖放在函数内部导入，避免在步骤注册时（模块 import 时）就引入依赖报错： @step(id="custom.ml_classify", ...) def ml_classify(ctx: StepContext) -> StepResult: # 延迟导入，只在步骤实际执行时才尝试导入 try: from sklearn.ensemble import RandomForestClassifier except ImportError: raise ImportError( "scikit-learn is required for 'custom.ml_classify'. " "Install it with: pip install scikit-learn" ) # ... 使用 sklearn 5.3 stats 字段命名 stats 字典的键会通过 $step.key 语法暴露给后续步骤，使用清晰的名称： return StepResult( output=result_gdf, stats={ "feature_count": len(result_gdf), # 标准命名，与内置步骤一致 "join_count": len(joined_gdf), # 自定义统计 "skip_count": len(skipped), # 跳过数量 } ) 5.4 QC 步骤的特殊模式 自定义 QC 步骤应遵循"检查并透传"模式： from geopipe_agent.models.qc import QcIssue from geopipe_agent.steps.qc._helpers import build_issues_gdf @step(id="custom.qc_area_check", ...) def custom_area_check(ctx: StepContext) -> StepResult: gdf = ctx.input("input") min_area = ctx.param("min_area", 1.0) issues = [] for idx, row in gdf.iterrows(): area = row.geometry.area if area < min_area: issues.append(QcIssue( rule_id="area_check", severity="warning", feature_index=idx, message=f"Feature {idx}: area {area:.4f} < min {min_area}", details={"area": area, "min_area": min_area}, )) return StepResult( output=gdf, # 透传输入数据 stats={ "issues_count": len(issues), "valid_count": len(gdf) - len(issues), }, metadata={"issues_gdf": build_issues_gdf(gdf, issues)}, issues=issues, ) 18.6 在 YAML 中使用自定义步骤 自定义步骤注册后，可直接在 YAML 中使用： pipeline: name: "使用自定义步骤的流水线" steps: - id: load-poi use: io.read_vector params: { path: "data/poi.shp" } - id: load-districts use: io.read_vector params: { path: "data/districts.shp" } # 使用自定义空间连接步骤 - id: join-district-attrs use: custom.spatial_join params: left: "$load-poi" right: "$load-districts" how: "left" op: "within" - id: save-result use: io.write_vector params: input: "$join-district-attrs" path: "output/poi_with_district.geojson" 18.7 本章小结 本章介绍了自定义步骤的开发与集成： @step 装饰器：注册步骤，定义参数规范、输出、后端和示例 StepContext：ctx.param()、ctx.input()、ctx.backend 获取参数和后端 StepResult：output、stats、metadata、issues 四个字段 自动发现：import geopipe_agent 自动加载内置步骤，自定义步骤需手动 import 或打包 最佳实践：清晰错误消息、延迟导入依赖、QC 步骤透传数据 导航：← 第十七章：Skill 文件与 AI 集成 ｜ 第十九章：Cookbook 示例精讲 →