开源一个 PyQt6 AI 写作助手:多智能体协作 + RAG 记忆系统
·
用 Python 做一个 Claude Code 风格的桌面应用:架构设计与实现
项目简介
AI写作助手是一个基于 PyQt6 的本地桌面应用,定位是"小说写作领域的 Claude Code"。
核心特性:
- 多智能体协作:12个专业智能体(大纲师、码字工、督察、评论家等)
- RAG 记忆系统:7个记忆模块 + ChromaDB 向量检索
- 工具调用:16个本地工具 + 6个MCP服务(100工具)
- 本地部署:数据全在本地,不上传云端
开源地址:https://github.com/Aurirlk/AI_novelsTool_obsdian
架构设计
整体架构
核心模块
1. LLM 统一接口
# src/utils/llm.py
class LLMClient:
"""统一LLM接口,支持9家提供商"""
def __init__(self, provider: str = None, model: str = None):
self.provider = provider or self._load_default_provider()
self.model = model or self._load_default_model()
self.client = self._create_client()
def chat(self, message: str, system_prompt: str = None,
history: list = None) -> str:
"""普通对话"""
messages = self._build_messages(message, system_prompt, history)
response = self.client.chat.completions.create(
model=self.model,
messages=messages
)
return response.choices[0].message.content
def chat_stream(self, message: str, system_prompt: str = None,
history: list = None):
"""流式对话"""
messages = self._build_messages(message, system_prompt, history)
stream = self.client.chat.completions.create(
model=self.model,
messages=messages,
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
def chat_with_tools(self, messages: list, tools: list,
handler: Callable) -> str:
"""带工具调用的对话"""
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=tools
)
# 处理工具调用
while response.choices[0].message.tool_calls:
tool_calls = response.choices[0].message.tool_calls
for call in tool_calls:
result = handler(call.function.name, call.function.arguments)
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": result
})
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=tools
)
return response.choices[0].message.content
支持的提供商:
- DeepSeek、智谱GLM、OpenAI、通义千问、Kimi、百川、星火、Ollama、自定义接口
2. 多智能体系统
# src/agents/base.py
class BaseAgent:
"""智能体基类"""
def __init__(self):
self.llm = LLMClient()
self.memory = get_shared_memory()
def run(self, task: str, context: dict = None) -> str:
"""执行任务"""
system_prompt = self._build_system_prompt(context)
return self.llm.chat(task, system_prompt=system_prompt)
def _build_system_prompt(self, context: dict = None) -> str:
"""构建系统提示词"""
parts = [self.base_prompt]
if context:
parts.append(f"上下文:{json.dumps(context, ensure_ascii=False)}")
return "\n\n".join(parts)
12个智能体各有专长:
| 智能体 | 职责 | 核心能力 |
|---|---|---|
| OutlineAgent | 大纲生成 | 黄金三章、爽点设计 |
| WriterAgent | 正文写作 | 续写、扩写、改写 |
| ReviewerAgent | 一致性检查 | 时空/性格/能力/称谓 |
| CriticAgent | 批评分析 | 大纲批评、章节批评 |
| PolisherAgent | 运营优化 | 标题、导语、简介 |
| ReaderSimulator | 读者模拟 | 模拟读者反馈 |
| WritingCoach | 写作教练 | 技巧指导 |
| MaterialSupplier | 素材供给 | 提供创作素材 |
3. RAG 记忆系统
# src/memory/shared_memory.py
class SharedMemory:
"""共享记忆系统"""
def __init__(self):
self.vector_store = get_vector_store() # ChromaDB
self.entity_dict = EntityDictionary()
self.world_rules = WorldRules()
self.psychology = Psychology()
def add_memory(self, content: str, metadata: dict = None):
"""添加记忆(向量化存储)"""
embedding = self._get_embedding(content)
self.vector_store.add(
documents=[content],
embeddings=[embedding],
metadatas=[metadata or {}]
)
def search(self, query: str, n_results: int = 5) -> list:
"""语义检索"""
embedding = self._get_embedding(query)
results = self.vector_store.query(
query_embeddings=[embedding],
n_results=n_results
)
return results["documents"][0]
def get_shared_knowledge(self, context: str = "") -> list:
"""获取共享知识(用于注入system prompt)"""
# 1. 实体词典
entities = self.entity_dict.get_relevant(context)
# 2. 世界规则
rules = self.world_rules.get_relevant(context)
# 3. 心理维度
psych = self.psychology.get_relevant(context)
return entities + rules + psych
7个记忆模块:
| 模块 | 功能 | 存储方式 |
|---|---|---|
| SharedMemory | 共享记忆 | ChromaDB向量 |
| EntityDictionary | 实体词典 | JSON文件 |
| WorldRules | 世界规则 | JSON文件 |
| Psychology | 心理维度 | JSON文件 |
| StyleGuide | 风格指南 | JSON文件 |
| CharacterState | 角色状态 | JSON文件 |
| HookTracker | 钩子追踪 | JSON文件 |
4. 工具系统
# src/tools/project_tools.py
def get_project_tools(groups: list = None) -> list:
"""获取OpenAI格式的工具定义"""
all_tools = [
{
"type": "function",
"function": {
"name": "list_books",
"description": "列出写作空间中的所有书籍",
"parameters": {"type": "object", "properties": {}, "required": []}
}
},
{
"type": "function",
"function": {
"name": "read_chapter",
"description": "读取指定书籍的指定章节全文",
"parameters": {
"type": "object",
"properties": {
"book_name": {"type": "string", "description": "书名"},
"chapter_title": {"type": "string", "description": "章节标题"}
},
"required": ["book_name", "chapter_title"]
}
}
},
# ... 16个工具
]
return all_tools
16个本地工具:
| 分类 | 工具 | 功能 |
|---|---|---|
| 读取 | list_books | 列出所有书 |
| 读取 | read_book | 读章节列表 |
| 读取 | read_chapter | 读章节全文 |
| 读取 | list_characters | 列角色 |
| 读取 | read_character | 读角色详情 |
| 读取 | list_hooks | 列悬念 |
| 读取 | list_events | 列时间线 |
| 写入 | save_character | 保存角色 |
| 写入 | save_hook | 保存悬念 |
| 写入 | save_event | 保存事件 |
| 写入 | write_chapter | 写章节 |
| 写入 | save_outline | 保存大纲 |
| 邮箱 | send_to_editor | 投递到编辑邮箱 |
| 邮箱 | get_delivery_history | 投递记录 |
5. MCP 扩展
# src/mcp/manager.py
class MCPManager:
"""MCP服务器管理器"""
def __init__(self):
self.servers = {}
self._load_config()
def connect(self, name: str, timeout: int = 15):
"""连接MCP服务器"""
config = self.servers[name]
if config.transport == "stdio":
client = StdioClient(config.command, config.args)
elif config.transport == "sse":
client = SSEClient(config.url)
client.connect(timeout=timeout)
self.servers[name].client = client
def to_openai_tools(self) -> list:
"""转换为OpenAI工具格式"""
tools = []
for server in self.servers.values():
if server.client:
for tool in server.client.list_tools():
tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
})
return tools
6个MCP服务:
- novel-workflow:小说生成工作流
- novel-writer:小说写作工具
- book-writer:书籍写作工具
- webnovel:网文平台工具
- fetch:网页抓取工具
- memory:记忆管理工具
关键技术点
1. 上下文注入
当用户在聊天页选择一本书时,自动注入该书的上下文:
def _build_book_context(self) -> str:
"""构建选中书籍的上下文"""
book = self._selected_book
if not book:
return ""
parts = [f"\n# 当前讨论书籍:《{book}》\n"]
# 章节列表
chapters = self._get_chapters(book)
if chapters:
parts.append(f"章节目录:{', '.join(chapters[:30])}")
# 最近一章预览
last_chapter = self._get_last_chapter(book)
if last_chapter:
parts.append(f"最近章节《{last_chapter['name']}》前200字:{last_chapter['content'][:200]}")
# 角色摘要
characters = self._get_characters(book)
if characters:
char_lines = [f" - {c['name']}({c['role_type']},{c['personality'][:20]})"
for c in characters[:10]]
parts.append("角色:\n" + "\n".join(char_lines))
return "\n".join(parts)
2. 正则自动匹配书名
当用户输入中包含 《xxx》 时,自动切换书籍选择器:
def _on_text_changed(self):
text = self.input.toPlainText()
# 正则匹配 《书名》
book_match = re.search(r"《([^》]+)》", text)
if book_match:
book_name = book_match.group(1)
# 在下拉框中查找并切换
for i in range(self.book_combo.count()):
if self.book_combo.itemText(i) == book_name:
self.book_combo.setCurrentIndex(i)
break
3. 记忆自动提取
保存章节后自动提取角色/钩子/事件变更:
# src/data/memory_extractor.py
def extract_memory_changes(chapter_text: str, book_name: str,
chapter_num: int) -> dict:
"""从章节文本中提取记忆变更"""
prompt = f"""分析以下章节内容,提取:
1. 新出现的角色(名称、类型、性格特点)
2. 新的伏笔/悬念(内容、类型)
3. 重要的剧情事件(标题、描述)
章节内容:
{chapter_text[:3000]}
请以JSON格式返回。"""
llm = LLMClient()
result = llm.chat(prompt)
return json.loads(result)
4. 样式表系统
# src/gui/professional_theme.py
class ProfessionalTheme:
"""ChatGPT风格主题"""
DARK = {
"bg_primary": "#212121",
"bg_secondary": "#171717",
"text_primary": "#ececec",
"accent": "#10a37f",
# ...
}
@classmethod
def get_stylesheet(cls, theme_name: str = "dark") -> str:
t = cls.get_theme(theme_name)
return f"""
QMainWindow, QWidget {{
background-color: {t['bg_primary']};
color: {t['text_primary']};
}}
QPushButton {{
background-color: transparent;
border: none;
border-radius: {t['radius_md']};
padding: {t['spacing_sm']} {t['spacing_md']};
}}
QPushButton:hover {{
background-color: {t['bg_hover']};
}}
#btn_primary {{
background-color: {t['accent']};
color: {t['text_inverse']};
}}
"""
项目结构
AI_novelsTool_obsdian/
├── src/
│ ├── core/ # 核心基础设施
│ │ ├── vector_store.py # ChromaDB向量存储
│ │ ├── cache_manager.py # 缓存系统
│ │ └── stream_handler.py # 流式输出
│ ├── agents/ # 智能体(12个)
│ │ ├── outline_agent.py # 大纲师
│ │ ├── writer_agent.py # 码字工
│ │ └── reviewer_agent.py # 督察
│ ├── memory/ # 记忆系统
│ │ ├── memory_manager.py # 核心记忆管理
│ │ ├── shared_memory.py # 共享记忆
│ │ └── entity_dictionary.py # 实体词典
│ ├── data/ # 数据管理
│ │ ├── writing_space.py # 写作空间
│ │ ├── character_store.py # 角色存储
│ │ └── hook_store.py # 钩子存储
│ ├── tools/ # 本地工具
│ │ └── project_tools.py # 项目数据工具
│ ├── gui/ # 界面(PyQt6)
│ │ ├── professional_main_window.py
│ │ ├── professional_theme.py
│ │ └── pages/ # 15个功能页面
│ ├── mcp/ # MCP客户端
│ ├── skills/ # Skills加载
│ └── utils/ # 工具
│ ├── llm.py # LLM客户端
│ └── exporter.py # 导出
├── assets/icons/ # Feather SVG图标
├── skills/ # 技能库(49个)
├── config/ # 全局配置
└── requirements.txt
技术栈
| 组件 | 技术 | 说明 |
|---|---|---|
| UI | PyQt6 | 桌面应用 |
| LLM | OpenAI SDK | 统一接口,支持9家提供商 |
| 向量库 | ChromaDB | 语义检索(RAG) |
| 数据库 | SQLite | 设置/密钥/历史 |
| 工具 | MCP协议 | 扩展工具系统 |
| 图标 | Feather Icons | SVG图标集 |
快速开始
# 克隆仓库
git clone https://github.com/Aurirlk/AI_novelsTool_obsdian.git
cd AI_novelsTool_obsdian
# 创建环境
conda create -n novel python=3.11 -y
conda activate novel
# 安装依赖
pip install -r requirements.txt
# 启动
python src/run_complete.py
首次启动后,进入 设置 → LLM,选择提供商并粘贴你的API密钥。
扩展性
Skills系统
# 安装技能
skills install <github-repo-path>
# 查看已安装技能
skills list
MCP服务器
在 mcp_servers.json 中配置:
{
"servers": [
{
"name": "novel-writer",
"command": "python",
"args": ["tools/novel-writer/server.py"]
}
]
}
总结
这个项目的核心设计思路:
- 本地优先:数据全在本地,不依赖云端
- 模块化:智能体、工具、记忆系统都是独立模块
- 可扩展:通过Skills和MCP协议扩展功能
- 网文适配:专为网文写作场景设计
开源地址:https://github.com/Aurirlk/AI_novelsTool_obsdian
欢迎Star、Fork、Issue。
最后是运行截图展示








更多推荐


所有评论(0)