diff --git a/kb_pipeline/errors.py b/kb_pipeline/errors.py
new file mode 100644
index 0000000..f337fb7
--- /dev/null
+++ b/kb_pipeline/errors.py
@@ -0,0 +1,174 @@
+"""Structured error handling for KB Pipeline
+
+Provides structured error types with context information for debugging.
+All errors include step_name, input_summary, error_type, message, and suggestion.
+"""
+
+from enum import Enum
+from typing import Optional, List, Any
+
+
+class ErrorType(str, Enum):
+    """错误类型分类"""
+    NETWORK = "network"
+    VALIDATION = "validation"
+    SYSTEM = "system"
+    AUTH = "auth"
+
+
+class KBError(Exception):
+    """KB 工作流基础错误
+
+    所有工作流错误的基类，包含丰富的上下文信息便于调试。
+
+    Attributes:
+        message: 错误信息
+        error_type: 错误类型
+        step_name: 发生错误的步骤名称
+        input_summary: 输入数据摘要
+        suggestion: 可操作的建议
+        cause: 原始异常（可选）
+    """
+
+    def __init__(
+        self,
+        message: str,
+        error_type: ErrorType,
+        step_name: str,
+        input_summary: str,
+        suggestion: str,
+        cause: Optional[Exception] = None
+    ):
+        super().__init__(message)
+        self.message = message
+        self.error_type = error_type
+        self.step_name = step_name
+        self.input_summary = input_summary
+        self.suggestion = suggestion
+        self.cause = cause
+
+    def to_dict(self) -> dict:
+        """将错误转换为字典格式用于日志和状态记录
+
+        Returns:
+            包含所有错误上下文信息的字典
+        """
+        return {
+            "error_type": self.error_type.value,
+            "step_name": self.step_name,
+            "input_summary": self.input_summary,
+            "message": self.message,
+            "suggestion": self.suggestion
+        }
+
+
+class NetworkError(KBError):
+    """网络错误 (403/404/500/超时)
+
+    用于表示网络请求失败，包括 HTTP 状态码错误和连接超时。
+
+    Attributes:
+        status_code: HTTP 状态码（可选）
+    """
+
+    def __init__(
+        self,
+        step_name: str,
+        input_summary: str,
+        status_code: Optional[int] = None,
+        cause: Optional[Exception] = None
+    ):
+        if status_code:
+            suggestion = f"检查网络连接，状态码: {status_code}"
+        else:
+            suggestion = "检查网络连接"
+        super().__init__(
+            message=f"网络请求失败: {status_code or 'unknown'}",
+            error_type=ErrorType.NETWORK,
+            step_name=step_name,
+            input_summary=input_summary,
+            suggestion=suggestion,
+            cause=cause
+        )
+        self.status_code = status_code
+
+
+class ValidationError(KBError):
+    """验证错误 (Schema 验证失败)
+
+    用于表示数据验证失败，包含详细的验证错误列表。
+
+    Attributes:
+        validation_errors: 验证错误列表
+    """
+
+    def __init__(
+        self,
+        step_name: str,
+        input_summary: str,
+        validation_errors: Optional[List[str]] = None,
+        cause: Optional[Exception] = None
+    ):
+        super().__init__(
+            message="Schema 验证失败",
+            error_type=ErrorType.VALIDATION,
+            step_name=step_name,
+            input_summary=input_summary,
+            suggestion="检查输入数据格式是否符合要求",
+            cause=cause
+        )
+        self.validation_errors = validation_errors or []
+
+
+class SystemError(KBError):
+    """系统错误 (文件系统/内存/依赖缺失)
+
+    用于表示系统级错误，如文件不存在、内存不足、依赖缺失等。
+
+    Attributes:
+        system_message: 系统错误消息
+    """
+
+    def __init__(
+        self,
+        step_name: str,
+        input_summary: str,
+        system_message: Optional[str] = None,
+        cause: Optional[Exception] = None
+    ):
+        super().__init__(
+            message=f"系统错误: {system_message or 'unknown'}",
+            error_type=ErrorType.SYSTEM,
+            step_name=step_name,
+            input_summary=input_summary,
+            suggestion="检查系统资源和依赖是否满足要求",
+            cause=cause
+        )
+        self.system_message = system_message
+
+
+class AuthError(KBError):
+    """认证/配额错误 (API配额/Cookie过期/Token失效)
+
+    用于表示认证失败或 API 配额超限。
+
+    Attributes:
+        auth_type: 认证类型（如 API Token, Cookie 等）
+    """
+
+    def __init__(
+        self,
+        step_name: str,
+        input_summary: str,
+        auth_type: Optional[str] = None,
+        cause: Optional[Exception] = None
+    ):
+        super().__init__(
+            message=f"认证失败: {auth_type or 'unknown'}",
+            error_type=ErrorType.AUTH,
+            step_name=step_name,
+            input_summary=input_summary,
+            suggestion="检查 API Token 或 Cookie 是否有效",
+            cause=cause
+        )
+        self.auth_type = auth_type