diff --git a/kb_pipeline/idempotency.py b/kb_pipeline/idempotency.py
new file mode 100644
index 0000000..d1ce2fe
--- /dev/null
+++ b/kb_pipeline/idempotency.py
@@ -0,0 +1,80 @@
+"""Idempotency Checker module
+
+Provides file existence checking to avoid reprocessing content.
+Checks for metadata JSON and content Markdown files.
+"""
+
+from pathlib import Path
+from typing import Tuple, List
+
+
+class IdempotencyChecker:
+    """Checks if content has already been processed
+
+    Determines if the output files for a given content already exist,
+    enabling idempotent processing to avoid duplicate work.
+
+    File naming convention:
+        - Metadata: {platform}_{content_id}_metadata.json
+        - Content: {platform}_{content_id}_content.md
+    """
+
+    def check_exists(
+        self,
+        output_dir: Path,
+        content_id: str,
+        platform: str
+    ) -> Tuple[bool, List[Path]]:
+        """Check if output files already exist for this content
+
+        Args:
+            output_dir: Directory where output files are stored
+            content_id: Unique content identifier (e.g., BV1GJ411x7Qw)
+            platform: Platform name (e.g., bilibili, zhihu)
+
+        Returns:
+            Tuple of (all_exist, existing_files):
+                - all_exist: True if both metadata and content files exist
+                - existing_files: List of Path objects for existing files
+        """
+        existing_files: List[Path] = []
+
+        if not content_id or not platform:
+            return (False, existing_files)
+
+        # Define expected file paths
+        metadata_file = output_dir / f"{platform}_{content_id}_metadata.json"
+        content_file = output_dir / f"{platform}_{content_id}_content.md"
+
+        # Check if files exist
+        if metadata_file.exists():
+            existing_files.append(metadata_file)
+
+        if content_file.exists():
+            existing_files.append(content_file)
+
+        # Return True only if both files exist
+        all_exist = len(existing_files) == 2
+
+        return (all_exist, existing_files)
+
+    def get_expected_files(
+        self,
+        output_dir: Path,
+        content_id: str,
+        platform: str
+    ) -> Tuple[Path, Path]:
+        """Get the expected file paths for a content
+
+        Args:
+            output_dir: Directory where output files are stored
+            content_id: Unique content identifier
+            platform: Platform name
+
+        Returns:
+            Tuple of (metadata_path, content_path)
+        """
+        metadata_path = output_dir / f"{platform}_{content_id}_metadata.json"
+        content_path = output_dir / f"{platform}_{content_id}_content.md"
+
+        return (metadata_path, content_path)
diff --git a/tests/test_pipeline/test_idempotency.py b/tests/test_pipeline/test_idempotency.py
new file mode 100644
index 0000000..8b39fdf
--- /dev/null
+++ b/tests/test_pipeline/test_idempotency.py
@@ -0,0 +1,183 @@
+"""Tests for IdempotencyChecker module
+
+Tests file existence checking for idempotent processing:
+- File exists scenario
+- File not exists scenario
+- Partial files exist scenario
+"""
+
+import pytest
+from pathlib import Path
+
+
+class TestIdempotencyChecker:
+    """Tests for IdempotencyChecker class"""
+
+    def test_check_exists_returns_true_when_files_exist(self, tmp_path):
+        """Test 1: 文件存在时 check_exists 返回 (True, [file_list])"""
+        from kb_pipeline.idempotency import IdempotencyChecker
+
+        # Create output directory with expected files
+        output_dir = tmp_path / "bilibili"
+        output_dir.mkdir()
+
+        content_id = "BV1GJ411x7Qw"
+        platform = "bilibili"
+
+        # Create expected files
+        metadata_file = output_dir / f"{platform}_{content_id}_metadata.json"
+        content_file = output_dir / f"{platform}_{content_id}_content.md"
+
+        metadata_file.write_text('{"title": "test"}')
+        content_file.write_text("# Test content")
+
+        checker = IdempotencyChecker()
+        exists, files = checker.check_exists(output_dir, content_id, platform)
+
+        assert exists is True
+        assert len(files) == 2
+        assert metadata_file in files
+        assert content_file in files
+
+    def test_check_exists_returns_false_when_files_not_exist(self, tmp_path):
+        """Test 2: 文件不存在时 check_exists 返回 (False, [])"""
+        from kb_pipeline.idempotency import IdempotencyChecker
+
+        output_dir = tmp_path / "bilibili"
+        output_dir.mkdir()
+
+        content_id = "BV1GJ411x7Qw"
+        platform = "bilibili"
+
+        checker = IdempotencyChecker()
+        exists, files = checker.check_exists(output_dir, content_id, platform)
+
+        assert exists is False
+        assert len(files) == 0
+
+    def test_check_exists_returns_partial_files_when_some_exist(self, tmp_path):
+        """Test 3: 部分文件存在时返回正确的存在文件列表"""
+        from kb_pipeline.idempotency import IdempotencyChecker
+
+        output_dir = tmp_path / "zhihu"
+        output_dir.mkdir()
+
+        content_id = "12345678"
+        platform = "zhihu"
+
+        # Create only metadata file, not content file
+        metadata_file = output_dir / f"{platform}_{content_id}_metadata.json"
+        metadata_file.write_text('{"title": "test"}')
+
+        checker = IdempotencyChecker()
+        exists, files = checker.check_exists(output_dir, content_id, platform)
+
+        # Should return False because not all files exist
+        assert exists is False
+        # But should list the existing file
+        assert len(files) == 1
+        assert metadata_file in files
+
+    def test_check_exists_creates_directory_if_not_exists(self, tmp_path):
+        """如果输出目录不存在，check_exists 应该返回 False"""
+        from kb_pipeline.idempotency import IdempotencyChecker
+
+        output_dir = tmp_path / "new_platform"  # Directory doesn't exist
+        content_id = "BV1test123"
+        platform = "bilibili"
+
+        checker = IdempotencyChecker()
+        exists, files = checker.check_exists(output_dir, content_id, platform)
+
+        assert exists is False
+        assert len(files) == 0
+
+
+class TestIdempotencyCheckerFileNaming:
+    """Tests for file naming conventions"""
+
+    def test_bilibili_file_naming(self, tmp_path):
+        """测试B站文件的命名格式"""
+        from kb_pipeline.idempotency import IdempotencyChecker
+
+        output_dir = tmp_path / "bilibili"
+        output_dir.mkdir()
+
+        content_id = "BV1GJ411x7Qw"
+        platform = "bilibili"
+
+        # Create files with correct naming
+        metadata_file = output_dir / "bilibili_BV1GJ411x7Qw_metadata.json"
+        content_file = output_dir / "bilibili_BV1GJ411x7Qw_content.md"
+
+        metadata_file.write_text('{}')
+        content_file.write_text('# content')
+
+        checker = IdempotencyChecker()
+        exists, files = checker.check_exists(output_dir, content_id, platform)
+
+        assert exists is True
+        assert len(files) == 2
+
+    def test_zhihu_file_naming(self, tmp_path):
+        """测试知乎文件的命名格式"""
+        from kb_pipeline.idempotency import IdempotencyChecker
+
+        output_dir = tmp_path / "zhihu"
+        output_dir.mkdir()
+
+        content_id = "87654321"
+        platform = "zhihu"
+
+        # Create files with correct naming
+        metadata_file = output_dir / "zhihu_87654321_metadata.json"
+        content_file = output_dir / "zhihu_87654321_content.md"
+
+        metadata_file.write_text('{}')
+        content_file.write_text('# content')
+
+        checker = IdempotencyChecker()
+        exists, files = checker.check_exists(output_dir, content_id, platform)
+
+        assert exists is True
+        assert len(files) == 2
+
+
+class TestIdempotencyCheckerEdgeCases:
+    """Tests for edge cases"""
+
+    def test_check_exists_with_empty_content_id(self, tmp_path):
+        """空content_id应该返回False"""
+        from kb_pipeline.idempotency import IdempotencyChecker
+
+        output_dir = tmp_path / "bilibili"
+        output_dir.mkdir()
+
+        checker = IdempotencyChecker()
+        exists, files = checker.check_exists(output_dir, "", "bilibili")
+
+        assert exists is False
+        assert len(files) == 0
+
+    def test_check_exists_with_special_characters_in_id(self, tmp_path):
+        """ID中包含特殊字符时的处理"""
+        from kb_pipeline.idempotency import IdempotencyChecker
+
+        output_dir = tmp_path / "bilibili"
+        output_dir.mkdir()
+
+        content_id = "BV1_test-123"
+        platform = "bilibili"
+
+        # Create files with special characters
+        metadata_file = output_dir / f"{platform}_{content_id}_metadata.json"
+        content_file = output_dir / f"{platform}_{content_id}_content.md"
+
+        metadata_file.write_text('{}')
+        content_file.write_text('# content')
+
+        checker = IdempotencyChecker()
+        exists, files = checker.check_exists(output_dir, content_id, platform)
+
+        assert exists is True
+        assert len(files) == 2