1. 项目背景与核心价值
作为一名同时需要写代码和写内容的技术创作者，我发现自己每天有超过60%的时间都消耗在重复性工作上：技术文档阅读、自媒体文案撰写、灵感记录整理。直到去年接触了几个AI工具后，突然意识到——这些机械劳动完全可以用Python自动化。
经过三个月的迭代开发，这个个人AI效率系统已经成为我的"第二大脑"。它最核心的价值在于：
- 写作环节：根据平台特性自动生成不同风格的初稿（头条的短平快、知乎的深度解析、小红书的种草体）
- 阅读环节：快速提取技术文档/论文的核心观点，支持追问细节
- 记录环节：自动将碎片灵感结构化存储到Notion知识库
实测下来，原本需要4小时完成的日常工作，现在1小时内就能搞定。下面我就从技术实现角度，完整拆解这个系统的搭建过程。
提示：本文所有代码均基于Python 3.11开发，建议使用虚拟环境运行。关键API涉及敏感信息，请勿直接提交到公开仓库。
2. 系统架构设计
2.1 核心工作流设计
系统采用"输入-处理-输出"的管道模式，三个核心组件通过标准化接口通信：
code复制[写作请求] -> [DeepSeek文案生成] -> [人工润色] -> [发布]
[文档上传] -> [Kimi摘要提取] -> [重点标注] -> [归档]
[语音/文本] -> [Notion分类存储] -> [知识图谱构建]
2.2 技术选型解析
选择当前技术栈的深层考量：
- Python 3.11：新版本的模式匹配语法（PEP 634）非常适合处理不同平台的内容格式要求
- OpenAI/DeepSeek：相比直接使用ChatGPT，DeepSeek的API对中文长文本生成更稳定（实测生成2000字文章时断句更合理）
- Notion官方SDK：比第三方库更及时支持最新功能块（如公式数据库）
- 本地JSON缓存：防止因网络问题导致数据丢失，同时作为开发时的测试用例库
2.3 文件结构规划
建议按以下结构组织项目代码：
code复制ai_assistant/
├── core/
│ ├── writer.py # 文案生成组件
│ ├── reader.py # 文档处理组件
│ └── recorder.py # 知识管理组件
├── configs/
│ └── platforms.json # 各平台文案风格配置
├── storage/
│ ├── cache/ # 本地JSON缓存
│ └── templates/ # Notion模板
└── main.py # 系统入口
3. DeepSeek文案生成器实现
3.1 初始化配置
首先需要处理API密钥的安全存储问题。我推荐使用python-dotenv+对称加密的方案：
python复制# secure_config.py
from cryptography.fernet import Fernet
import os
from dotenv import load_dotenv
class ConfigManager:
def __init__(self):
load_dotenv()
self.cipher = Fernet(os.getenv('ENCRYPT_KEY'))
def get_api_key(self):
encrypted = os.getenv('DEEPSEEK_KEY')
return self.cipher.decrypt(encrypted.encode()).decode()
在.env文件中存储加密后的密钥：
code复制ENCRYPT_KEY=你的加密密钥
DEEPSEEK_KEY=加密后的API密钥
3.2 多平台内容生成
核心生成器的完整实现包含以下关键功能：
python复制# writer.py
import re
from typing import Literal
from pathlib import Path
PLATFORMS = Literal["toutiao", "zhihu", "xiaohongshu", "csdn"]
class DeepSeekWriter:
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.deepseek.com/v1"
)
self.style_cache = self._load_platform_styles()
def _load_platform_styles(self) -> dict:
"""加载各平台预置风格"""
style_file = Path(__file__).parent.parent / "configs/platforms.json"
return json.loads(style_file.read_text())
def generate(
self,
topic: str,
platform: PLATFORMS = "toutiao",
word_count: int = 1500
) -> str:
"""生成符合平台特性的文案
Args:
topic: 文章主题
platform: 发布平台类型
word_count: 目标字数
Returns:
生成的文章内容（含Markdown格式）
"""
style = self.style_cache[platform]
prompt = (
f"请以{style['tone']}的风格，"
f"使用{style['paragraph']}的段落结构，"
f"为《{topic}》创作一篇{word_count}字左右的{style['name']}。"
f"特别注意：{style['special_notes']}"
)
response = self.client.chat.completions.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": style["system_role"]},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=word_count * 2,
stop=style.get("stop_sequences", [])
)
return self._post_process(response.choices[0].message.content)
def _post_process(self, content: str) -> str:
"""后处理：去AI痕迹+段落优化"""
# 替换常见AI用语
replacements = {
"首先": "第一", "其次": "第二",
"值得注意的是": "需要说明的是",
"综上所述": "总的来说"
}
for k, v in replacements.items():
content = content.replace(k, v)
# 智能分段：长段落拆解
return "\n\n".join(self._split_paragraphs(content))
def _split_paragraphs(self, text: str) -> list[str]:
"""将超过100字的段落按标点拆分"""
result = []
for para in text.split("\n\n"):
if len(para) <= 100:
result.append(para)
continue
# 按句子分割
sentences = re.split(r'(?<=[。！？])', para)
current = ""
for sent in sentences:
if len(current) + len(sent) > 80:
result.append(current.strip())
current = sent
else:
current += sent
if current:
result.append(current.strip())
return result
3.3 平台风格配置
在configs/platforms.json中预置各平台特性：
json复制{
"toutiao": {
"name": "头条文章",
"tone": "口语化、接地气",
"paragraph": "3-4行短段落",
"special_notes": "使用设问句开头，每段包含情绪词",
"system_role": "你是资深自媒体运营，擅长制造爆款标题",
"stop_sequences": ["### 相关推荐"]
},
"zhihu": {
"name": "知乎回答",
"tone": "专业但有亲和力",
"paragraph": "5-8行中段落",
"special_notes": "需要数据支撑观点，适当使用列表",
"system_role": "你是行业专家，回答要体现深度思考"
}
}
4. Kimi文档阅读器实现
4.1 文件预处理模块
python复制# reader.py
from typing import Union
import io
import magic
class KimiReader:
SUPPORTED_TYPES = {
'application/pdf': 'pdf',
'text/plain': 'txt',
'application/vnd.openxmlformats-officedocument.wordprocessingml.document': 'docx'
}
def __init__(self, api_key: str):
self.client = openai.OpenAI(api_key=api_key)
def _detect_file_type(self, file: Union[str, io.BytesIO]) -> str:
"""自动检测文件类型"""
mime = magic.Magic(mime=True)
if isinstance(file, str):
with open(file, 'rb') as f:
file_type = mime.from_buffer(f.read(1024))
else:
file_type = mime.from_buffer(file.getvalue())
return self.SUPPORTED_TYPES.get(file_type)
4.2 智能摘要生成
python复制 def generate_summary(
self,
file_path: str,
focus_areas: list[str] = None,
summary_length: Literal["short", "medium", "long"] = "medium"
) -> dict:
"""生成结构化摘要
Args:
file_path: 文档路径
focus_areas: 重点关注领域列表
summary_length: 摘要长度
Returns:
{
"overview": "文档整体概述",
"key_points": ["核心论点1", "核心论点2"],
"action_items": ["待办事项1", "待办事项2"]
}
"""
file_type = self._detect_file_type(file_path)
if not file_type:
raise ValueError("Unsupported file type")
with open(file_path, 'rb') as f:
response = self.client.chat.completions.create(
model="kimi",
messages=[
{
"role": "system",
"content": (
"你是一个专业文档分析师，需要提取以下内容：\n"
"1. 用50字概括文档核心价值\n"
"2. 列出3-5个关键论点\n"
"3. 提取读者需要采取的行动项\n"
f"重点关注：{focus_areas or '全部内容'}"
)
},
{
"role": "user",
"content": [
{
"type": "file",
"file": f,
"file_name": file_path.split("/")[-1]
}
]
}
],
temperature=0.3
)
return self._parse_summary(response.choices[0].message.content)
5. Notion知识管理实现
5.1 数据库连接配置
python复制# recorder.py
from notion_client import Client
from datetime import datetime
class NotionRecorder:
def __init__(self, api_key: str, database_id: str):
self.notion = Client(auth=api_key)
self.database_id = database_id
self._verify_connection()
def _verify_connection(self):
"""验证数据库结构是否符合预期"""
try:
db = self.notion.databases.retrieve(self.database_id)
required_props = {"Title", "Type", "Status", "Tags"}
if not required_props.issubset(db["properties"].keys()):
raise ValueError("Database missing required properties")
except Exception as e:
raise ConnectionError(f"Notion连接失败: {str(e)}")
5.2 智能分类存储
python复制 def create_page(
self,
title: str,
content: str,
source_type: Literal["idea", "article", "code"] = "idea",
auto_tag: bool = True
) -> str:
"""创建智能分类的知识条目
Args:
title: 条目标题
content: 内容（支持Markdown）
source_type: 内容类型
auto_tag: 是否自动生成标签
Returns:
创建的页面ID
"""
properties = {
"Title": {"title": [{"text": {"content": title}}]},
"Type": {"select": {"name": source_type}},
"Status": {"status": {"name": "未处理"}},
"Created": {"date": {"start": datetime.now().isoformat()}}
}
if auto_tag:
tags = self._generate_tags(content)
properties["Tags"] = {"multi_select": [{"name": tag} for tag in tags]}
children = self._parse_content_blocks(content)
return self.notion.pages.create(
parent={"database_id": self.database_id},
properties=properties,
children=children
).get("id")
def _generate_tags(self, text: str) -> list[str]:
"""基于内容生成标签"""
# 调用本地关键词提取模型（简化版）
from collections import Counter
import jieba
words = [w for w in jieba.cut(text) if len(w) > 1 and w not in self.STOP_WORDS]
return [w for w, _ in Counter(words).most_common(3)]
6. 系统集成与优化
6.1 主控程序实现
python复制# main.py
from core.writer import DeepSeekWriter
from core.reader import KimiReader
from core.recorder import NotionRecorder
import click
@click.group()
def cli():
pass
@cli.command()
@click.option("--topic", required=True, help="文章主题")
@click.option("--platform", default="toutiao", help="目标平台")
def write(topic, platform):
"""生成平台适配文案"""
writer = DeepSeekWriter.load_from_env()
article = writer.generate(topic, platform)
print(f"生成完成！字符数：{len(article)}")
@cli.command()
@click.argument("file_path")
def read(file_path):
"""解析文档并生成摘要"""
reader = KimiReader.load_from_env()
summary = reader.generate_summary(file_path)
print(f"文档摘要：\n{summary['overview']}")
if __name__ == "__main__":
cli()
6.2 性能优化技巧
-
缓存策略：
- 对相同主题的文案生成结果进行MD5缓存
- 使用LRU机制管理缓存大小
python复制from functools import lru_cache @lru_cache(maxsize=100) def cached_generate(topic: str, platform: str) -> str: return self.generate(topic, platform) -
异步处理：
python复制import asyncio async def async_generate(self, topics: list[str]): semaphore = asyncio.Semaphore(3) # 限制并发数 async with semaphore: return await asyncio.gather( *[self._async_generate_one(t) for t in topics] ) -
错误重试机制：
python复制from tenacity import retry, stop_after_attempt, wait_exponential @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10) ) def safe_api_call(self): # API调用代码
7. 常见问题与解决方案
7.1 内容质量问题
问题1：生成的文案过于模板化
- 解决方案：
- 在prompt中添加负面示例："避免使用'首先、其次、最后'这样的过渡词"
- 使用更精细的温度参数（creative内容用0.7，技术文档用0.3）
- 添加个性化示例："参考作者XXX的写作风格"
问题2：技术文档摘要不准确
- 解决步骤：
python复制def improve_tech_summary(self): # 添加领域术语表 with open("glossary.txt") as f: glossary = f.read() # 修改系统指令 system_msg = f""" 你是一个技术文档专家，请特别注意以下术语： {glossary} """
7.2 技术实现问题
问题3：Notion API速率限制
- 优化方案：
- 实现请求队列：使用
asyncio.Queue控制并发 - 错误处理代码示例：
python复制async def safe_notion_call(self, func, *args): try: return await func(*args) except Exception as e: if "rate_limited" in str(e): await asyncio.sleep(5) return await self.safe_notion_call(func, *args) raise - 实现请求队列：使用
问题4：长文档处理超时
- 处理流程：
- 使用
python-magic识别文档类型 - 对PDF/DOCX先做文本提取预处理
- 超过1万字的文档自动分块处理
- 使用
8. 实际应用案例
8.1 技术博客创作流程
-
生成初稿：
bash复制python main.py write --topic "Python异步编程实战" --platform csdn -
插入代码片段：
python复制def insert_code_example(self, article: str, code: str): """在文章适当位置插入代码示例""" # 查找技术概念出现的位置 positions = [m.start() for m in re.finditer(r"异步|await|asyncio", article)] if not positions: return article + f"\n\n代码示例：\n```python\n{code}\n```" insert_at = positions[len(positions)//2] return (article[:insert_at] + "\n\n代码示例：\n```python\n" + code + "\n```\n\n" + article[insert_at:]) -
自动发布到WordPress：
python复制import xmlrpc.client def publish_to_blog(self, title, content): wp = xmlrpc.client.ServerProxy('https://your-site.com/xmlrpc.php') post_id = wp.metaWeblog.newPost( blog_id=1, username='admin', password='xxx', content={ 'title': title, 'description': content, 'post_status': 'publish' }, publish=True ) return post_id
这个系统我已经稳定使用8个月，最大的体会是：不要追求一步到位的完美自动化，而要在关键环节实现人机协作。比如文案生成后我通常会花10-15分钟做人工润色，但这比从零开始写作已经节省了80%的时间。