本章系统介绍 GeoPipeAgent 的数据模型（models/）和错误体系（errors.py），这是理解框架内部机制和编写可靠代码的基础。
GeoPipeAgent 使用 Python dataclass 定义数据模型，位于 src/geopipe_agent/models/ 目录。
PipelineDefinition描述一个已解析的流水线，是 YAML 文件在内存中的完整表示。
@dataclass
class PipelineDefinition:
name: str # 流水线名称（必填）
steps: list[StepDefinition] # 步骤列表
description: str = "" # 流水线描述
crs: str | None = None # 全局默认 CRS
variables: dict = {} # 变量字典
outputs: dict = {} # 输出声明
何时使用：通过 parse_yaml() 获得；运行时通过修改 variables 覆盖变量值。
StepDefinition描述流水线中的一个步骤。
@dataclass
class StepDefinition:
id: str # 步骤唯一 ID
use: str # 步骤注册 ID（如 "vector.buffer"）
params: dict = {} # 步骤参数（含变量/引用占位符）
when: str | None = None # 条件执行表达式
on_error: str = "fail" # 错误策略："fail" | "skip" | "retry"
backend: str | None = None # 指定后端
注意：params 中的值在执行时才会解析，此时仍可能含有 ${var} 和 $step-id 占位符。
StepResult步骤执行结果，是步骤间数据传递的核心载体。
@dataclass
class StepResult:
output: Any = None # 主输出（GeoDataFrame、str、raster_info 等）
stats: dict = {} # 统计信息（feature_count、crs 等）
metadata: dict = {} # 元数据（issues_gdf、transform 等）
issues: list[QcIssue] = [] # QC 问题列表（仅 QC 步骤填充）
__getattr__ 机制：StepResult 实现了 __getattr__，允许通过属性语法访问 stats 和 metadata 中的键：
result.feature_count # → result.stats["feature_count"]
result.issues_gdf # → result.metadata["issues_gdf"]
result.crs # → result.stats["crs"] 或 result.metadata["crs"]
这个机制使得 $step.feature_count 等引用语法成为可能。
summary() 方法：为 JSON 报告生成可序列化的摘要：
def summary(self) -> dict:
summary = {}
if self.output is not None:
summary["feature_count"] = len(self.output)
summary["crs"] = str(self.output.crs)
summary["geometry_types"] = list(self.output.geometry.geom_type.unique())
summary.update(self.stats)
if self.issues:
summary["issues_count"] = len(self.issues)
summary["issues_by_severity"] = Counter(i.severity for i in self.issues)
return summary
QcIssue描述一个数据质量问题，由 QC 步骤生成。
@dataclass
class QcIssue:
rule_id: str # 规则 ID（如 "geometry_validity"）
severity: str # 级别："error" | "warning" | "info"
feature_index: int | None # 问题要素的 GeoDataFrame 行索引
message: str # 人类可读的问题描述
geometry: Any # 问题位置几何（可选，用于可视化）
details: dict # 规则特定详情
def to_dict(self) -> dict: # JSON 可序列化字典
QcIssue 列表存储在 StepResult.issues 中，同时通过 metadata["issues_gdf"] 提供 GeoDataFrame 视图。
GeoPipeAgent 定义了一个层次化的错误体系，所有异常都继承自 GeopipeAgentError：
GeopipeAgentError（基类）
├── PipelineParseError # YAML 解析失败
├── PipelineValidationError # 校验失败（ID 格式、引用不存在等）
├── StepExecutionError # 步骤执行失败（含 step_id、建议、原因）
├── BackendNotAvailableError # 后端不可用
├── StepNotFoundError # 步骤 ID 未注册
└── VariableResolutionError # 变量或引用解析失败
PipelineParseError：
# 触发条件
parse_yaml("""
name: "no pipeline key" # 缺少顶层 pipeline:
steps: []
""")
# 错误: PipelineParseError("Missing 'pipeline' key at the top level.")
PipelineValidationError：
# 触发条件
validate_pipeline(pipeline)
# 步骤 ID 不合法、引用不存在的步骤、使用未注册的步骤类型等
# 错误: PipelineValidationError("Step id 'my.step' is invalid. step_id must match [a-z0-9_-]")
StepExecutionError（最重要）：
@dataclass
class StepExecutionError(GeopipeAgentError):
step_id: str # 失败的步骤 ID
suggestion: str # AI 友好的修复建议（可能为 None）
cause: Exception # 原始异常
def to_dict(self) -> dict:
return {
"error": "StepExecutionError",
"step_id": self.step_id,
"message": str(self),
"suggestion": self.suggestion,
"cause": str(self.cause)
}
JSON 报告中的错误示例：
{
"error": "StepExecutionError",
"step_id": "buffer-roads",
"message": "Geometry must be a Point or LineString geometry type if operations are in degrees.",
"suggestion": "Add a vector.reproject step before this step to convert to a projected CRS.",
"cause": "Geometry must be a Point or ..."
}
VariableResolutionError：
# 触发条件：引用不存在的步骤或变量
# ${undefined_var} 在执行时（非校验时）
# $nonexistent-step 在执行时（如果校验被绕过）
from geopipe_agent.errors import (
GeopipeAgentError,
PipelineParseError,
PipelineValidationError,
StepExecutionError,
VariableResolutionError,
)
try:
pipeline = parse_yaml("pipeline.yaml")
warnings = validate_pipeline(pipeline)
report = execute_pipeline(pipeline)
except PipelineParseError as e:
print(f"YAML 格式错误: {e}")
except PipelineValidationError as e:
print(f"流水线校验失败: {e}")
except StepExecutionError as e:
print(f"步骤 '{e.step_id}' 执行失败: {e}")
if e.suggestion:
print(f"建议: {e.suggestion}")
except GeopipeAgentError as e:
print(f"框架错误: {e}")
运行时报错
├── 格式错误（如缺少 pipeline:）
│ → PipelineParseError → 检查 YAML 格式
├── 校验错误（如步骤 ID 含点号）
│ → PipelineValidationError → 检查步骤定义
├── 步骤执行错误（如文件不存在）
│ → StepExecutionError → 查看 suggestion 字段
│ → step_id 指示哪个步骤失败
└── 变量解析错误（如引用未定义变量）
→ VariableResolutionError → 检查 variables 和引用拼写
StepInfo：步骤注册信息StepInfo 是步骤注册表中的元数据对象，由 @step 装饰器创建：
@dataclass
class StepInfo:
id: str # 注册 ID（如 "vector.buffer"）
func: Callable # 步骤函数
name: str # 显示名称
description: str # 步骤描述
category: str # 类别（io/vector/raster/analysis/network/qc）
params: dict # 参数规范（含 type/required/default/description）
outputs: dict # 输出规范
backends: list[str] # 支持的后端列表
examples: list[dict] # 示例（用于 Skill 文档生成）
def to_dict(self) -> dict # 序列化（用于 describe 命令和 Skill 生成）
查询步骤信息：
# 查看步骤详情
geopipe-agent describe vector.buffer
# 输出（JSON 格式）
{
"id": "vector.buffer",
"name": "矢量缓冲区分析",
"description": "对输入的矢量数据生成指定距离的缓冲区",
"category": "vector",
"params": {
"input": {"type": "geodataframe", "required": true, "description": "输入矢量数据"},
"distance": {"type": "number", "required": true, "description": "缓冲区距离"},
"cap_style": {"type": "string", "required": false, "default": "round", ...}
},
"outputs": {
"output": {"type": "geodataframe"},
"stats": {"type": "dict"}
},
"backends": ["native_python", "qgis_process"]
}
本章系统介绍了 GeoPipeAgent 的数据模型和错误体系：
PipelineDefinition / StepDefinition：内存中的流水线表示StepResult：步骤间数据传递的核心，__getattr__ 机制支持点语法访问 stats/metadataQcIssue：质检问题记录，含规则 ID、严重级别、要素索引、描述和几何GeopipeAgentError 基类 → 5 个子类，各有特定触发场景StepExecutionError：含 step_id 和 suggestion 字段，AI 友好此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。