《【AI实战】用Python搭建个人知识库：3种方案对比+完整代码》

weixin_41015620

468人浏览 · 2026-03-31 10:52:48

weixin_41015620 · 2026-03-31 10:52:48 发布

📄 文章1：《【AI实战】用Python搭建个人知识库：3种方案对比+完整代码》
markdown

复制代码

【AI实战】用Python搭建个人知识库：3种方案对比+完整代码

信息太多记不住？我用Python搭建了个人知识库，3种方案实测对比。代码可直接运行，帮你找到最适合自己的方案。

为什么需要个人知识库？

我每天要处理的信息：

技术文章：10-20篇
代码片段：5-10个
灵感想法：3-5个
项目资料：若干

以前：收藏夹吃灰，笔记软件混乱，要用的时候找不到
现在：3秒定位任何信息，效率提升10倍

方案一：本地方案（SQLite + Python）

适合人群

注重隐私，数据不想上云
技术能力较强
主要在单台电脑使用

核心代码

```python
import sqlite3
from datetime import datetime
import json

class LocalKnowledgeBase:
"""本地知识库 - SQLite版"""
  
    def __init__(self, db_path="knowledge.db"):
        self.db_path = db_path
self.init_db()
  
def init_db(self):
"""初始化数据库"""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS knowledge (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
content TEXT,
category TEXT,
tags TEXT,
source TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_category ON knowledge(category)
            """)
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_tags ON knowledge(tags)
            """)
  
    def add(self, title, content, category="笔记", tags="", source=""):
"""添加知识条目"""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT INTO knowledge (title, content, category, tags, source, updated_at)
VALUES (?, ?, ?, ?, ?, ?)
""", (title, content, category, tags, source, datetime.now()))
conn.commit()
        print(f"✅ 已添加：{title}")
  
    def search(self, keyword, category=None):
"""搜索知识"""
with sqlite3.connect(self.db_path) as conn:
            if category:
                cursor = conn.execute("""
SELECT * FROM knowledge 
WHERE (title LIKE ? OR content LIKE ? OR tags LIKE ?)
                    AND category = ?
ORDER BY updated_at DESC
                """, (f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', category))
else:
                cursor = conn.execute("""
SELECT * FROM knowledge 
WHERE title LIKE ? OR content LIKE ? OR tags LIKE ?
ORDER BY updated_at DESC
                """, (f'%{keyword}%', f'%{keyword}%', f'%{keyword}%'))
          
            return cursor.fetchall()
  
def get_by_category(self, category):
"""按分类获取"""
with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("""
                SELECT * FROM knowledge WHERE category = ? ORDER BY updated_at DESC
""", (category,))
            return cursor.fetchall()
  
def stats(self):
"""统计信息"""
with sqlite3.connect(self.db_path) as conn:
            total = conn.execute("SELECT COUNT(*) FROM knowledge").fetchone()[0]
            categories = conn.execute("""
SELECT category, COUNT(*) FROM knowledge GROUP BY category
""").fetchall()
            return {"total": total, "categories": categories}

# 使用示例
if __name__ == "__main__":
    kb = LocalKnowledgeBase()
  
# 添加知识
kb.add(
        title="Python列表推导式",
        content="[x for x in range(10) if x % 2 == 0]",
        category="代码片段",
        tags="python,语法",
        source="CSDN"
    )
  
kb.add(
        title="AI副业变现思路",
        content="1. 代写服务 2. 技术文章 3. 教程账号",
        category="灵感",
        tags="ai,副业",
        source="思考"
    )
  
# 搜索
    results = kb.search("Python")
    print(f"\n🔍 找到 {len(results)} 条相关记录")
  
# 统计
    stats = kb.stats()
    print(f"\n📊 知识库统计：共 {stats['total']} 条")
    for cat, count in stats['categories']:
        print(f"  {cat}: {count} 条")

优缺点

在这里插入图片描述

方案二：云端方案（Notion API）

适合人群

多设备使用（手机/电脑/平板）
不想折腾技术
核心代码

核心代码

import requests
from datetime import datetime

class NotionKnowledgeBase:
"""云端知识库 - Notion版"""
  
def __init__(self, token, database_id):
        self.token = token
        self.database_id = database_id
        self.headers = {
            "Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"Notion-Version": "2022-06-28"
        }
  
    def add(self, title, content, category="笔记", tags=None):
"""添加知识条目"""
        url = "https://api.notion.com/v1/pages"
      
        data = {
            "parent": {"database_id": self.database_id},
            "properties": {
                "标题": {"title": [{"text": {"content": title}}]},
                "内容": {"rich_text": [{"text": {"content": content}}]},
                "分类": {"select": {"name": category}},
                "标签": {"multi_select": [{"name": tag} for tag in (tags or [])]},
                "创建时间": {"date": {"start": datetime.now().isoformat()}}
            }
        }
      
        response = requests.post(url, headers=self.headers, json=data)
        if response.status_code == 200:
            print(f"✅ 已同步到Notion：{title}")
            return response.json()
else:
            print(f"❌ 同步失败：{response.text}")
            return None
  
def search(self, keyword):
"""搜索知识"""
        url = f"https://api.notion.com/v1/databases/{self.database_id}/query"
      
        data = {
            "filter": {
                "or": [
                    {"property": "标题", "title": {"contains": keyword}},
                    {"property": "内容", "rich_text": {"contains": keyword}}
                ]
            }
        }
      
        response = requests.post(url, headers=self.headers, json=data)
        if response.status_code == 200:
            results = response.json()["results"]
            print(f"🔍 找到 {len(results)} 条相关记录")
            return results
        return []
  
def get_by_category(self, category):
"""按分类获取"""
        url = f"https://api.notion.com/v1/databases/{self.database_id}/query"
      
        data = {
            "filter": {
"property": "分类",
                "select": {"equals": category}
            }
        }
      
        response = requests.post(url, headers=self.headers, json=data)
        if response.status_code == 200:
            return response.json()["results"]
        return []

# 使用示例
if __name__ == "__main__":
# 替换为你的Notion Token和Database ID
    TOKEN = "your_notion_integration_token"
    DATABASE_ID = "your_database_id"
  
    kb = NotionKnowledgeBase(TOKEN, DATABASE_ID)
  
# 添加知识
kb.add(
        title="Cursor使用技巧",
        content="Cmd+K打开AI对话框，Cmd+L查看历史",
        category="工具",
        tags=["cursor", "ai", "效率"]
    )

优缺点

在这里插入图片描述

方案三：混合方案（本地+云端同步）

适合人群

既要本地速度，又要云端备份
数据安全意识强
愿意折腾技术

核心代码

import sqlite3
import requests
import json
from datetime import datetime

class HybridKnowledgeBase:
"""混合知识库 - 本地+云端"""
  
    def __init__(self, db_path="knowledge.db", notion_token=None, database_id=None):
        self.local = LocalKnowledgeBase(db_path)
        self.notion = NotionKnowledgeBase(notion_token, database_id) if notion_token else None
        self.sync_log = []
  
    def add(self, title, content, category="笔记", tags="", source="", sync_to_cloud=True):
"""添加知识（本地+可选云端）"""
# 先存本地
self.local.add(title, content, category, tags, source)
      
# 同步到云端
        if sync_to_cloud and self.notion:
            try:
                tag_list = [t.strip() for t in tags.split(",") if t.strip()]
self.notion.add(title, content, category, tag_list)
                self.sync_log.append({
"time": datetime.now(),
"title": title,
"status": "success"
                })
except Exception as e:
                print(f"⚠️ 云端同步失败：{e}")
                self.sync_log.append({
"time": datetime.now(),
"title": title,
"status": "failed",
"error": str(e)
                })
  
    def search(self, keyword, category=None):
"""搜索（优先本地）"""
        return self.local.search(keyword, category)
  
def sync_status(self):
"""查看同步状态"""
        total = len(self.sync_log)
        success = len([s for s in self.sync_log if s["status"] == "success"])
        failed = total - success
      
print(f"\n🔄 同步状态：")
        print(f"  总计：{total} 条")
        print(f"  成功：{success} 条")
        print(f"  失败：{failed} 条")
      
        if failed > 0:
print("\n❌ 失败记录：")
            for log in self.sync_log[-5:]:
                if log["status"] == "failed":
                    print(f"  - {log['title']}: {log.get('error', '未知错误')}")

# 使用示例
if __name__ == "__main__":
    kb = HybridKnowledgeBase(
        db_path="my_knowledge.db",
        notion_token="your_token",  # 可选
        database_id="your_db_id"    # 可选
    )
  
# 添加知识（同步到云端）
kb.add(
        title="混合方案优势",
        content="本地速度快+云端可备份",
        category="技术方案",
        tags="python,架构",
        sync_to_cloud=True
    )
  
# 查看同步状态
kb.sync_status()