Data & research · SKILL.md

geopipe-agent

geopipe-agent

Use when building AI-driven GIS data pipelines with YAML-defined steps — format conversion, spatial validation, QC reporting, PostGIS loading, WMS publishing. GeoPipe Agent: YAML-driven GIS ETL pipeline agent with quality control.

Independent GitHub snapshot · Automated classification · Original documentation governs

Zznlgis / opengis-skillsMIT

Skill documentation

GitHub

name: geopipe-agent description: "Use when building AI-driven GIS data pipelines with YAML-defined steps — format conversion, spatial validation, QC reporting, PostGIS loading, WMS publishing. GeoPipe Agent: YAML-driven GIS ETL pipeline agent with quality control." tags:

gis
agent
pipeline
yaml
qc
spatial-analysis
vector
raster
geoprocessing
python

项目地址： https://github.com/znlgis/GeoPipeAgent（发布包名 geopipe-agent，截至 2026 年尚未上传 PyPI，需从源码安装）

许可证： MIT

概述

GeoPipeAgent 是一个 AI 原生的 GIS 分析流水线引擎。核心理念是：你（AI）生成 YAML 管道配置，框架负责执行并返回结构化结果。

核心能力：

声明式流水线：YAML 定义分析步骤，AI 友好
丰富步骤库：矢量分析、栅格分析、网络分析、空间聚类、数据质检
多后端支持：native_python（基于 geopandas/shapely/rasterio）和 qgis_process
质检框架：10 种 QC 检查，支持错误/警告/信息三级
结构化报告：JSON 输出，含统计信息和 QC 汇总
步骤引用：$step_id.output 语法实现步骤间数据传递

快速开始

安装

git clone https://github.com/znlgis/GeoPipeAgent.git
cd GeoPipeAgent
pip install -e ".[analysis,network]"   # 按需启用分析/网络分析可选依赖

基本用法

# 列出所有可用步骤（可按类别过滤）
geopipe-agent list-steps --category qc

# 查看步骤详情
geopipe-agent describe vector.buffer

# 校验 YAML 流水线（不执行）
geopipe-agent validate pipeline.yaml

# 执行流水线
geopipe-agent run pipeline.yaml

# 查看可用后端
geopipe-agent backends

最小示例

pipeline:
  name: "快速缓冲区"
  steps:
    - id: read
      use: io.read_vector
      params:
        path: "data/roads.shp"
    - id: buffer
      use: vector.buffer
      params:
        input: "$read"
        distance: 500
    - id: save
      use: io.write_vector
      params:
        input: "$buffer"
        path: "output/result.geojson"
  outputs:
    result: "$save"

核心概念

步骤引用语法

语法	说明	示例
`$step_id`	等同于 `$step_id.output`	`$buffer`
`$step_id.output`	引用步骤输出	`$buffer.output`
`$step_id.stats`	引用步骤统计信息	`$buffer.stats`
`$step_id.issues`	引用 QC 步骤问题列表	`$check.issues`
`$step_id.issues_count`	QC 问题数量	`$check.issues_count`
`${var_name}`	变量替换	`${input_path}`

条件执行 (`when`)

- id: fix-geometries
  use: qc.geometry_validity
  params:
    input: "$data"
    auto_fix: true
  when: "$check.issues_count > 0"

支持比较运算符（==, !=, >, <, >=, <=）、布尔运算符（and, or, not）。

错误处理 (`on_error`)

值	行为
`fail`（默认）	停止管道执行
`skip`	跳过当前步骤，继续下一步
`retry`	重试最多 3 次（带退避）

步骤分类

IO 步骤（io.*）

步骤	说明	关键参数
`io.read_vector`	读取矢量数据（SHP/GeoJSON/GPKG）	`path`, `layer`, `encoding`
`io.read_raster`	读取栅格数据（GeoTIFF 等）	`path`
`io.write_vector`	写入矢量数据	`input`, `path`, `format`, `encoding`
`io.write_raster`	写入栅格数据	`input`, `path`

矢量分析步骤（vector.*）

步骤	说明	关键参数	后端
`vector.buffer`	缓冲区分析	`input`, `distance`, `cap_style`	native_python, qgis_process
`vector.clip`	矢量裁剪	`input`, `clip_geometry`	native_python, qgis_process
`vector.reproject`	投影转换	`input`, `target_crs`	native_python, qgis_process
`vector.dissolve`	按字段融合	`input`, `by`, `aggfunc`	native_python
`vector.overlay`	叠加分析	`input`, `overlay_layer`, `how`	native_python
`vector.query`	属性查询	`input`, `expression`	—
`vector.simplify`	几何简化	`input`, `tolerance`, `preserve_topology`	native_python

叠加方式 (vector.overlay 的 how 参数)：intersection、union、difference、symmetric_difference、identity

栅格分析步骤（raster.*）

步骤	说明	关键参数
`raster.reproject`	栅格投影转换	`input`, `target_crs`, `resampling`
`raster.clip`	栅格裁剪	`input`, `mask`, `crop`, `nodata`
`raster.calc`	栅格波段计算	`input`, `expression`（如 `'(B4-B3)/(B4+B3)'`）
`raster.contour`	等值线提取	`input`, `interval`, `band`, `base`
`raster.stats`	栅格统计	`input`, `band`

重采样方法 (raster.reproject 的 resampling)：nearest、bilinear、cubic、lanczos

高级分析步骤（analysis.*）

步骤	说明	关键参数
`analysis.voronoi`	泰森多边形	`input`, `envelope`
`analysis.heatmap`	热力图（KDE）	`input`, `resolution`, `bandwidth`
`analysis.interpolate`	空间插值	`input`, `value_field`, `method`, `resolution`
`analysis.cluster`	空间聚类	`input`, `method`（dbscan/kmeans）, `eps`, `n_clusters`

插值方法 (analysis.interpolate 的 method)：linear、nearest、cubic、idw

网络分析步骤（network.*）

步骤	说明	关键参数
`network.shortest_path`	最短路径	`input`, `origin`, `destination`, `weight`
`network.service_area`	服务区分析	`input`, `center`, `cost_limit`, `weight`
`network.geocode`	地理编码	`addresses`, `provider`, `user_agent`

质检步骤（qc.*）

QC 步骤遵循 「检查即透传」 模式：输入数据原样通过为 output，问题单独收集在 $step.issues 和 $step.issues_gdf 中。

步骤	检查内容	关键参数
`qc.geometry_validity`	几何有效性（自相交、空几何等）	`input`, `auto_fix`, `severity`
`qc.topology`	拓扑关系（重叠、缝隙、悬挂线）	`input`, `rules`, `tolerance`, `severity`
`qc.attribute_completeness`	必填字段完整性	`input`, `required_fields`, `allow_empty`, `severity`
`qc.attribute_domain`	属性值域合规性	`input`, `field`, `allowed_values`, `pattern`, `severity`
`qc.value_range`	数值字段范围	`input`, `field`, `min`, `max`, `severity`
`qc.duplicate_check`	重复要素检测	`input`, `check_geometry`, `check_fields`, `tolerance`, `severity`
`qc.crs_check`	坐标参考系验证	`input`, `expected_crs`, `severity`
`qc.raster_nodata`	NoData 一致性	`input`, `expected_nodata`, `max_nodata_ratio`, `severity`
`qc.raster_value_range`	栅格值域检查	`input`, `min`, `max`, `band`, `severity`
`qc.raster_resolution`	分辨率一致性	`input`, `expected_x`, `expected_y`, `tolerance`, `severity`

严重级别 (severity)：error、warning、info

拓扑规则 (qc.topology 的 rules)：no_overlaps、no_gaps、no_dangles

典型工作流

工作流 1：数据质检 + 修复流水线

pipeline:
  name: "数据质检修复"
  steps:
    - id: read
      use: io.read_vector
      params:
        path: "data/buildings.shp"

    # 第一步：检查几何有效性
    - id: check_valid
      use: qc.geometry_validity
      params:
        input: "$read"
        severity: error

    # 第二步：条件修复（仅当发现问题时执行）
    - id: fix_valid
      use: qc.geometry_validity
      params:
        input: "$check_valid"
        auto_fix: true
      when: "$check_valid.issues_count > 0"

    # 第三步：属性完整性检查
    - id: check_attr
      use: qc.attribute_completeness
      params:
        input: "$fix_valid"
        required_fields: ["name", "height", "type"]
        severity: warning

    # 第四步：写入结果
    - id: save
      use: io.write_vector
      params:
        input: "$check_attr"
        path: "output/buildings_checked.gpkg"
        format: GPKG

  outputs:
    result: "$save"
    issues: "$check_attr.issues"

工作流 2：缓冲区 + 叠加分析

pipeline:
  name: "道路影响范围分析"
  steps:
    - id: roads
      use: io.read_vector
      params:
        path: "data/roads.shp"
    - id: parcels
      use: io.read_vector
      params:
        path: "data/parcels.shp"

    # 500 米道路缓冲区
    - id: buffer
      use: vector.buffer
      params:
        input: "$roads"
        distance: 500

    # 缓冲区与地块求交集
    - id: overlay
      use: vector.overlay
      params:
        input: "$parcels"
        overlay_layer: "$buffer"
        how: intersection

    - id: save
      use: io.write_vector
      params:
        input: "$overlay"
        path: "output/affected_parcels.geojson"

  outputs:
    result: "$save"

工作流 3：DEM 分析 + 等值线

pipeline:
  name: "DEM 地形分析"
  steps:
    - id: read_dem
      use: io.read_raster
      params:
        path: "data/dem.tif"

    # 栅格 QC：检查 NoData 和分辨率
    - id: check_nodata
      use: qc.raster_nodata
      params:
        input: "$read_dem"
        max_nodata_ratio: 0.1

    - id: check_resolution
      use: qc.raster_resolution
      params:
        input: "$check_nodata"
        expected_x: 30
        expected_y: 30

    # 生成等高线（50 米间距）
    - id: contour
      use: raster.contour
      params:
        input: "$check_resolution"
        interval: 50

    # 输出矢量等值线
    - id: save_contour
      use: io.write_vector
      params:
        input: "$contour"
        path: "output/contours_50m.gpkg"
        format: GPKG

  outputs:
    contour: "$save_contour"

工作流 4：网络分析

pipeline:
  name: "紧急响应路径分析"
  steps:
    - id: roads
      use: io.read_vector
      params:
        path: "data/road_network.shp"

    # 最短路径
    - id: route
      use: network.shortest_path
      params:
        input: "$roads"
        origin: [116.3, 39.9]
        destination: [116.5, 40.0]

    # 服务区（5000 米）
    - id: service
      use: network.service_area
      params:
        input: "$roads"
        center: [116.4, 39.95]
        cost_limit: 5000

    - id: save_route
      use: io.write_vector
      params:
        input: "$route"
        path: "output/route.geojson"

    - id: save_service
      use: io.write_vector
      params:
        input: "$service"
        path: "output/service_area.geojson"

  outputs:
    route: "$save_route"
    service_area: "$save_service"

工作流 5：空间聚类 + 热力图

pipeline:
  name: "犯罪热点分析"
  steps:
    - id: points
      use: io.read_vector
      params:
        path: "data/crime_points.shp"

    # DBSCAN 聚类
    - id: cluster
      use: analysis.cluster
      params:
        input: "$points"
        method: dbscan
        eps: 0.01
        min_samples: 3

    # 热力图
    - id: heatmap
      use: analysis.heatmap
      params:
        input: "$points"
        resolution: 200

    - id: save_clusters
      use: io.write_vector
      params:
        input: "$cluster"
        path: "output/clusters.geojson"

    - id: save_heatmap
      use: io.write_raster
      params:
        input: "$heatmap"
        path: "output/heatmap.tif"

  outputs:
    clusters: "$save_clusters"
    heatmap: "$save_heatmap"

AI 使用建议

关键注意事项

步骤 ID 规则：只能用小写字母、数字、下划线和连字符，不能含点号（. 保留用于属性引用）
引用链完整性：确保每个 $step_id 引用的步骤在当前步骤之前定义
QC 链式调用：QC 步骤可串联（qc.check1 → qc.check2 → qc.check3），问题汇总到管道报告的 qc_summary 中
后端选择：简单分析用 native_python（无外部依赖），复杂分析使用 qgis_process 后端
路径使用绝对路径：避免工作目录问题
大数据注意：管道在内存中运行，超大栅格建议先裁剪

与其它工具对比

工具	适用场景	GeoPipeAgent 优势
GDAL CLI	手动命令行操作	AI 可编程，结构化输出
qgis_process	复杂空间分析	多步流水线编排，条件执行
PyQGIS 脚本	Python 编程	声明式配置，无需写 Python
Python 脚本	自定义分析	模板化，AI 直接生成配置

常见问题

问题	解决
步骤引用解析失败	检查步骤 ID 是否含点号，引用目标是否在当前步骤之前定义
管线执行超时	对大范围栅格先裁剪，减少数据量
QC 步骤无输出	QC 步骤透传输入数据，问题在 `$step.issues` 中，`output` 即输入
条件步骤未触发	检查 `when` 表达式语法，确保引用的 `$step.attr` 存在
后端不兼容	`native_python` 和 `qgis_process` 后端参数可能不同，查阅 `list-steps` 确认

参考资源

步骤参考：reference/steps-reference.md
管道 Schema：reference/pipeline-schema.md
中文教程（znlgis）：https://znlgis.github.io/gis/tutorial/GeoPipeAgent/

MAKE IT PART OF YOUR WORKFLOW

Install skill

TERMINAL

npx skills add znlgis/opengis-skills --skill geopipe-agent

Run the command in your terminal. Review the code and license before installing.

ZIP · Complete skill directory with resources and licenses

SKILL.md · Instructions file only

Repository stars: Not synced
License: MIT
Version / branch: HEAD · c721cedc
Repository updated: Not synced

Review scripts and permissions before installing.

Source code

geopipe-agent

Skill documentation

概述

快速开始

安装

基本用法

最小示例

核心概念

步骤引用语法

条件执行 (`when`)

错误处理 (`on_error`)

步骤分类

IO 步骤（io.*）

矢量分析步骤（vector.*）

栅格分析步骤（raster.*）

高级分析步骤（analysis.*）

网络分析步骤（network.*）

质检步骤（qc.*）

典型工作流

工作流 1：数据质检 + 修复流水线

工作流 2：缓冲区 + 叠加分析

工作流 3：DEM 分析 + 等值线

工作流 4：网络分析

工作流 5：空间聚类 + 热力图

AI 使用建议

推荐工作流

关键注意事项

与其它工具对比

常见问题

相关技能

参考资源

geopipe-agent

Skill documentation

概述

快速开始

安装

基本用法

最小示例

核心概念

步骤引用语法

条件执行 (when)

错误处理 (on_error)

步骤分类

IO 步骤（io.*）

矢量分析步骤（vector.*）

栅格分析步骤（raster.*）

高级分析步骤（analysis.*）

网络分析步骤（network.*）

质检步骤（qc.*）

典型工作流

工作流 1：数据质检 + 修复流水线

工作流 2：缓冲区 + 叠加分析

工作流 3：DEM 分析 + 等值线

工作流 4：网络分析

工作流 5：空间聚类 + 热力图

AI 使用建议

推荐工作流

关键注意事项

与其它工具对比

常见问题

相关技能

参考资源

More skills to explore

skill-creator

vercel-react-best-practices

migrate-to-shoehorn

条件执行 (`when`)

错误处理 (`on_error`)