diff --git a/kb_pipeline/__init__.py b/kb_pipeline/__init__.py
new file mode 100644
index 0000000..889531f
--- /dev/null
+++ b/kb_pipeline/__init__.py
@@ -0,0 +1,26 @@
+"""KB Pipeline - 工作流管道基础设施
+
+提供内容处理工作流的核心 Schema 和工具。
+"""
+
+from .schemas import (
+    StepInput,
+    StepOutput,
+    StepResult,
+    StepStatus,
+    StepRecord,
+    WorkflowState,
+    StepDefinition,
+)
+
+__all__ = [
+    "StepInput",
+    "StepOutput",
+    "StepResult",
+    "StepStatus",
+    "StepRecord",
+    "WorkflowState",
+    "StepDefinition",
+]
+
+__version__ = "0.1.0"
diff --git a/tests/test_pipeline/__init__.py b/tests/test_pipeline/__init__.py
new file mode 100644
index 0000000..c65e052
--- /dev/null
+++ b/tests/test_pipeline/__init__.py
@@ -0,0 +1 @@
+# tests/test_pipeline package
diff --git a/tests/test_pipeline/conftest.py b/tests/test_pipeline/conftest.py
new file mode 100644
index 0000000..14580e5
--- /dev/null
+++ b/tests/test_pipeline/conftest.py
@@ -0,0 +1,59 @@
+"""Pytest 配置和共享 fixtures"""
+
+import pytest
+import tempfile
+from pathlib import Path
+from datetime import datetime
+
+
+@pytest.fixture
+def tmp_content_dir():
+    """临时内容目录 fixture"""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        yield Path(tmpdir)
+
+
+@pytest.fixture
+def sample_step_input_data():
+    """示例步骤输入数据"""
+    return {
+        "content_id": "BV1test123",
+        "platform": "bilibili"
+    }
+
+
+@pytest.fixture
+def sample_step_output_data():
+    """示例步骤输出数据"""
+    return {
+        "success": True,
+        "error_message": None,
+        "completed_at": datetime.now()
+    }
+
+
+@pytest.fixture
+def sample_workflow_state_data():
+    """示例工作流状态数据"""
+    return {
+        "schema_version": "1.0",
+        "content_id": "BV1test123",
+        "platform": "bilibili",
+        "source_url": "https://www.bilibili.com/video/BV1test123",
+        "created_at": datetime.now(),
+        "updated_at": datetime.now(),
+        "steps": {}
+    }
+
+
+@pytest.fixture
+def sample_step_record_data():
+    """示例步骤记录数据"""
+    return {
+        "step_name": "fetch_metadata",
+        "status": "completed",
+        "started_at": datetime.now(),
+        "completed_at": datetime.now(),
+        "error_message": None,
+        "output_summary": "成功获取元数据"
+    }
diff --git a/tests/test_pipeline/test_schemas.py b/tests/test_pipeline/test_schemas.py
new file mode 100644
index 0000000..8d2c1a8
--- /dev/null
+++ b/tests/test_pipeline/test_schemas.py
@@ -0,0 +1,323 @@
+"""Schema 验证测试
+
+测试 kb_pipeline.schemas 模块中的所有核心 Schema。
+"""
+
+import pytest
+from datetime import datetime
+from pydantic import ValidationError
+
+
+class TestStepInput:
+    """测试 StepInput Schema"""
+
+    def test_step_input_requires_content_id(self):
+        """StepInput 必须包含 content_id 字段"""
+        from kb_pipeline.schemas import StepInput
+
+        with pytest.raises(ValidationError) as exc_info:
+            StepInput(platform="bilibili")
+
+        assert "content_id" in str(exc_info.value)
+
+    def test_step_input_requires_platform(self):
+        """StepInput 必须包含 platform 字段"""
+        from kb_pipeline.schemas import StepInput
+
+        with pytest.raises(ValidationError) as exc_info:
+            StepInput(content_id="BV1test123")
+
+        assert "platform" in str(exc_info.value)
+
+    def test_step_input_validates_platform_literal(self):
+        """StepInput platform 必须是 bilibili 或 zhihu"""
+        from kb_pipeline.schemas import StepInput
+
+        # 有效值
+        step_bilibili = StepInput(content_id="BV1test", platform="bilibili")
+        assert step_bilibili.platform == "bilibili"
+
+        step_zhihu = StepInput(content_id="123456", platform="zhihu")
+        assert step_zhihu.platform == "zhihu"
+
+        # 无效值
+        with pytest.raises(ValidationError):
+            StepInput(content_id="BV1test", platform="youtube")
+
+    def test_step_input_content_id_non_empty(self):
+        """StepInput content_id 不能为空字符串"""
+        from kb_pipeline.schemas import StepInput
+
+        with pytest.raises(ValidationError):
+            StepInput(content_id="", platform="bilibili")
+
+
+class TestStepOutput:
+    """测试 StepOutput Schema"""
+
+    def test_step_output_requires_success(self):
+        """StepOutput 必须包含 success 字段"""
+        from kb_pipeline.schemas import StepOutput
+
+        with pytest.raises(ValidationError) as exc_info:
+            StepOutput()
+
+        assert "success" in str(exc_info.value)
+
+    def test_step_output_default_completed_at(self):
+        """StepOutput completed_at 默认为当前时间"""
+        from kb_pipeline.schemas import StepOutput
+
+        before = datetime.now()
+        output = StepOutput(success=True)
+        after = datetime.now()
+
+        assert before <= output.completed_at <= after
+
+    def test_step_output_optional_error_message(self):
+        """StepOutput error_message 是可选的"""
+        from kb_pipeline.schemas import StepOutput
+
+        output = StepOutput(success=True)
+        assert output.error_message is None
+
+        output_with_error = StepOutput(success=False, error_message="测试错误")
+        assert output_with_error.error_message == "测试错误"
+
+
+class TestStepStatus:
+    """测试 StepStatus 枚举"""
+
+    def test_step_status_has_pending(self):
+        """StepStatus 包含 PENDING 状态"""
+        from kb_pipeline.schemas import StepStatus
+
+        assert StepStatus.PENDING.value == "pending"
+
+    def test_step_status_has_running(self):
+        """StepStatus 包含 RUNNING 状态"""
+        from kb_pipeline.schemas import StepStatus
+
+        assert StepStatus.RUNNING.value == "running"
+
+    def test_step_status_has_completed(self):
+        """StepStatus 包含 COMPLETED 状态"""
+        from kb_pipeline.schemas import StepStatus
+
+        assert StepStatus.COMPLETED.value == "completed"
+
+    def test_step_status_has_failed(self):
+        """StepStatus 包含 FAILED 状态"""
+        from kb_pipeline.schemas import StepStatus
+
+        assert StepStatus.FAILED.value == "failed"
+
+    def test_step_status_has_skipped(self):
+        """StepStatus 包含 SKIPPED 状态"""
+        from kb_pipeline.schemas import StepStatus
+
+        assert StepStatus.SKIPPED.value == "skipped"
+
+
+class TestStepRecord:
+    """测试 StepRecord Schema"""
+
+    def test_step_record_requires_step_name(self):
+        """StepRecord 必须包含 step_name"""
+        from kb_pipeline.schemas import StepRecord, StepStatus
+
+        with pytest.raises(ValidationError):
+            StepRecord(status=StepStatus.COMPLETED)
+
+    def test_step_record_requires_status(self):
+        """StepRecord 必须包含 status"""
+        from kb_pipeline.schemas import StepRecord
+
+        with pytest.raises(ValidationError):
+            StepRecord(step_name="fetch_metadata")
+
+    def test_step_record_optional_fields(self):
+        """StepRecord 可选字段正确工作"""
+        from kb_pipeline.schemas import StepRecord, StepStatus
+
+        record = StepRecord(step_name="fetch_metadata", status=StepStatus.COMPLETED)
+        assert record.started_at is None
+        assert record.completed_at is None
+        assert record.error_message is None
+        assert record.output_summary is None
+
+
+class TestWorkflowState:
+    """测试 WorkflowState Schema"""
+
+    def test_workflow_state_requires_content_id(self):
+        """WorkflowState 必须包含 content_id"""
+        from kb_pipeline.schemas import WorkflowState
+
+        with pytest.raises(ValidationError):
+            WorkflowState(platform="bilibili", source_url="https://test.com")
+
+    def test_workflow_state_requires_platform(self):
+        """WorkflowState 必须包含 platform"""
+        from kb_pipeline.schemas import WorkflowState
+
+        with pytest.raises(ValidationError):
+            WorkflowState(content_id="BV1test", source_url="https://test.com")
+
+    def test_workflow_state_requires_source_url(self):
+        """WorkflowState 必须包含 source_url"""
+        from kb_pipeline.schemas import WorkflowState
+
+        with pytest.raises(ValidationError):
+            WorkflowState(content_id="BV1test", platform="bilibili")
+
+    def test_workflow_state_has_steps_dict(self):
+        """WorkflowState 包含 steps 字典"""
+        from kb_pipeline.schemas import WorkflowState
+
+        state = WorkflowState(
+            content_id="BV1test",
+            platform="bilibili",
+            source_url="https://test.com"
+        )
+        assert hasattr(state, "steps")
+        assert isinstance(state.steps, dict)
+
+    def test_workflow_state_default_schema_version(self):
+        """WorkflowState schema_version 默认为 1.0"""
+        from kb_pipeline.schemas import WorkflowState
+
+        state = WorkflowState(
+            content_id="BV1test",
+            platform="bilibili",
+            source_url="https://test.com"
+        )
+        assert state.schema_version == "1.0"
+
+    def test_workflow_state_is_step_completed_method(self):
+        """WorkflowState 有 is_step_completed 方法"""
+        from kb_pipeline.schemas import WorkflowState, StepStatus, StepRecord
+
+        state = WorkflowState(
+            content_id="BV1test",
+            platform="bilibili",
+            source_url="https://test.com"
+        )
+
+        # 未完成的步骤
+        assert state.is_step_completed("fetch_metadata") is False
+
+        # 添加完成的步骤
+        state.steps["fetch_metadata"] = StepRecord(
+            step_name="fetch_metadata",
+            status=StepStatus.COMPLETED
+        )
+        assert state.is_step_completed("fetch_metadata") is True
+
+
+class TestSchemaValidation:
+    """测试 Pydantic 验证"""
+
+    def test_validation_error_on_invalid_input(self):
+        """无效输入抛出 ValidationError"""
+        from kb_pipeline.schemas import StepInput
+
+        with pytest.raises(ValidationError):
+            StepInput(content_id=123, platform="bilibili")  # type: ignore
+
+    def test_model_validate_from_dict(self):
+        """model_validate 从字典正确创建实例"""
+        from kb_pipeline.schemas import StepInput
+
+        data = {"content_id": "BV1test", "platform": "bilibili"}
+        step_input = StepInput.model_validate(data)
+        assert step_input.content_id == "BV1test"
+        assert step_input.platform == "bilibili"
+
+
+class TestJSONSerialization:
+    """测试 JSON 序列化/反序列化"""
+
+    def test_step_input_model_dump_json(self):
+        """StepInput 支持 model_dump_json"""
+        from kb_pipeline.schemas import StepInput
+        import json
+
+        step_input = StepInput(content_id="BV1test", platform="bilibili")
+        json_str = step_input.model_dump_json()
+        data = json.loads(json_str)
+
+        assert data["content_id"] == "BV1test"
+        assert data["platform"] == "bilibili"
+
+    def test_step_output_model_dump_json(self):
+        """StepOutput 支持 model_dump_json"""
+        from kb_pipeline.schemas import StepOutput
+        import json
+
+        output = StepOutput(success=True, error_message=None)
+        json_str = output.model_dump_json()
+        data = json.loads(json_str)
+
+        assert data["success"] is True
+        assert data["error_message"] is None
+
+    def test_workflow_state_model_dump_json(self):
+        """WorkflowState 支持 model_dump_json"""
+        from kb_pipeline.schemas import WorkflowState
+        import json
+
+        state = WorkflowState(
+            content_id="BV1test",
+            platform="bilibili",
+            source_url="https://test.com"
+        )
+        json_str = state.model_dump_json()
+        data = json.loads(json_str)
+
+        assert data["content_id"] == "BV1test"
+        assert data["platform"] == "bilibili"
+        assert data["source_url"] == "https://test.com"
+        assert data["steps"] == {}
+
+    def test_workflow_state_model_validate_from_json(self):
+        """WorkflowState 可从 JSON 恢复"""
+        from kb_pipeline.schemas import WorkflowState
+        import json
+
+        original = WorkflowState(
+            content_id="BV1test",
+            platform="bilibili",
+            source_url="https://test.com"
+        )
+        json_str = original.model_dump_json()
+        restored = WorkflowState.model_validate_json(json_str)
+
+        assert restored.content_id == original.content_id
+        assert restored.platform == original.platform
+        assert restored.source_url == original.source_url
+
+    def test_workflow_state_with_steps_serialization(self):
+        """WorkflowState steps 字典正确序列化/反序列化"""
+        from kb_pipeline.schemas import WorkflowState, StepRecord, StepStatus
+        import json
+
+        original = WorkflowState(
+            content_id="BV1test",
+            platform="bilibili",
+            source_url="https://test.com"
+        )
+        original.steps["fetch_metadata"] = StepRecord(
+            step_name="fetch_metadata",
+            status=StepStatus.COMPLETED
+        )
+
+        json_str = original.model_dump_json()
+        data = json.loads(json_str)
+
+        assert "fetch_metadata" in data["steps"]
+        assert data["steps"]["fetch_metadata"]["status"] == "completed"
+
+        restored = WorkflowState.model_validate_json(json_str)
+        assert "fetch_metadata" in restored.steps
+        assert restored.steps["fetch_metadata"].status == StepStatus.COMPLETED