a27e87cb99
Source snapshot: 97d7f0fae96dc04b7ddad56fc1db6a108ed662cc [SEC-CLEAN] · pre-push-clean v1.0 · 109处敏感信息已自动转乱码
27 KiB
27 KiB
新增工具 · create_notion_page · backend/tools.py · 完整替换文件
妈妈操作方式：清空现有的
backend/tools.py，粘贴下面全部代码。nano backend/tools.py→ 全选删除 → 粘贴 →Ctrl+O回车保存 →Ctrl+X退出 → 重启python3.11 main.py
这个版本加了什么
| 改动 | 说明 |
|---|---|
新增 create_notion_page 工具 |
晨星可以直接在 Notion 创建新页面，把拆书分析结果存进去 |
| 支持父页面指定 | 可以指定存到哪个页面下面（默认存到桔子开发者空间） |
| 支持长内容自动分块 | Notion API 每个文本块最多2000字，自动拆分不会报错 |
| 原有工具全保留 | 搜索/读取/记忆/联网/交互记录/时间 全都在 |
完整代码（全部复制）
import json
import datetime
import os
import requests
import yaml
import urllib.parse
from datetime import timezone, timedelta
_cfg_path = os.path.join(os.path.dirname(__file__), "..", "config.yaml")
with open(_cfg_path, "r") as _f:
_cfg = yaml.safe_load(_f)
NOTION_TOKEN = _cfg.get("notion", {}).get("token", "")
NOTION_API = "https://api.notion.com/v1"
NOTION_HEADERS = {
"Authorization": f"Bearer {NOTION_TOKEN}",
"Notion-Version": "2022-06-28",
"Content-Type": "application/json"
}
DEEPSEEK_API_KEY = _cfg.get("deepseek", {}).get("api_key", "")
DEEPSEEK_BASE_URL = _cfg.get("deepseek", {}).get("base_url", "https://api.deepseek.com")
DEEPSEEK_MODEL = _cfg.get("deepseek", {}).get("model", "deepseek-chat")
DEEPSEEK_MAX_TOKENS = _cfg.get("deepseek", {}).get("max_tokens", 8096)
# ===== 默认父页面ID =====
# 拆书分析结果默认存到这个页面下面
# 妈妈可以在 config.yaml 里配置 notion.analysis_parent_id 来改
DEFAULT_ANALYSIS_PARENT = _cfg.get("notion", {}).get(
"analysis_parent_id",
"790510ec-8435-4e11-b565-e010539376fa" # 默认用交互记录的父页面
)
TOOL_DEFINITIONS = [
{
"type": "function",
"function": {
"name": "web_search",
"description": "在互联网上实时搜索信息。当妈妈问任何外部事实（人名、新闻、热点、天气、明星、电视剧等）时，必须直接调用此工具，不要问妈妈要不要搜。",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "search_notion",
"description": "在Notion工作区中实时搜索页面，这是你的完整记忆库。妈妈提到过去做过的事、拆书进度、交互记录、之前聊过的内容时，必须先用这个搜索。如果一个关键词搜不到，换其他关键词再搜。",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词（中文）"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "read_notion_page",
"description": "读取指定Notion页面的完整内容。先通过search_notion找到页面ID，再用这个读取详情。",
"parameters": {
"type": "object",
"properties": {
"page_id": {"type": "string", "description": "Notion页面ID（32位无横杠格式）"}
},
"required": ["page_id"]
}
}
},
{
"type": "function",
"function": {
"name": "search_memories",
"description": "在向量记忆库中搜索语义相关的记忆片段。注意：向量记忆可能不是最新的，以Notion搜索为准。",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "get_current_time",
"description": "获取当前北京时间。",
"parameters": {"type": "object", "properties": {}}
}
},
{
"type": "function",
"function": {
"name": "write_interaction_record",
"description": "将交互记录写入Notion。仅在妈妈说结束信号（再见/拜拜/晚安/先到这）时调用。",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "记录标题"},
"summary": {"type": "string", "description": "对话摘要"},
"tasks_completed": {"type": "string", "description": "已完成任务"},
"key_findings": {"type": "string", "description": "核心发现"},
"next_tasks": {"type": "string", "description": "下次要做的事"}
},
"required": ["title", "summary"]
}
}
},
{
"type": "function",
"function": {
"name": "create_notion_page",
"description": "在Notion中创建一个新页面并写入内容。用于保存拆书分析结果、规律卡片、期待点汇总等需要持久化存储的内容。"
"拆完书后如果分析结果有价值，应该主动调用此工具存到Notion里。"
"内容支持Markdown格式（标题、列表、代码块、表格等）。",
"parameters": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "页面标题，例如：《折春漪》第311章拆解"
},
"content": {
"type": "string",
"description": "页面正文内容（支持Markdown格式）"
},
"icon": {
"type": "string",
"description": "页面图标emoji，默认📖"
},
"parent_page_id": {
"type": "string",
"description": "父页面ID（32位无横杠格式）。不提供则使用默认位置。"
}
},
"required": ["title", "content"]
}
}
}
]
def _read_notion_page(page_id):
if not page_id: return "[错误：未提供页面ID]"
page_id = page_id.replace("-", "")
fid = f"{page_id[:8]}-{page_id[8:12]}-{page_id[12:16]}-{page_id[16:20]}-{page_id[20:]}"
all_texts = []
cursor = None
while True:
url = f"{NOTION_API}/blocks/{fid}/children?page_size=100"
if cursor: url += f"&start_cursor={cursor}"
resp = requests.get(url, headers=NOTION_HEADERS)
if resp.status_code != 200: return f"[读取页面失败: HTTP {resp.status_code}]"
data = resp.json()
for block in data.get("results", []):
btype = block.get("type", "")
bd = block.get(btype, {})
rich_texts = bd.get("rich_text", [])
text = "".join([rt.get("plain_text", "") for rt in rich_texts])
if text.strip():
if btype.startswith("heading"): all_texts.append(f"{'#' * int(btype[-1])} {text.strip()}")
elif btype == "code": all_texts.append(f"```\n{text.strip()}\n```")
elif btype in ("bulleted_list_item", "numbered_list_item"): all_texts.append(f"- {text.strip()}")
else: all_texts.append(text.strip())
if data.get("has_more"): cursor = data.get("next_cursor")
else: break
content = "\n".join(all_texts)
if not content: return "[页面内容为空]"
if len(content) > 6000: content = content[:6000] + f"\n...[内容过长已截断·共{len(content)}字]"
return content
def _search_notion_workspace(query, count=10):
if not query: return "[错误：未提供搜索关键词]"
payload = {"query": query, "sort": {"direction": "descending", "timestamp": "last_edited_time"}, "page_size": min(count, 10)}
try:
resp = requests.post(f"{NOTION_API}/search", headers=NOTION_HEADERS, json=payload, timeout=15)
except Exception as e:
return f"[Notion搜索出错: {e}]"
if resp.status_code != 200: return f"[Notion搜索失败: HTTP {resp.status_code}]"
results = []
for page in resp.json().get("results", []):
title = ""
for pn, pv in page.get("properties", {}).items():
if pv.get("type") == "title":
title = "".join([t.get("plain_text", "") for t in pv.get("title", [])])
break
if not title: continue
pid = page.get("id", "").replace("-", "")
edited = page.get("last_edited_time", "")[:10]
results.append(f"- [{edited}] {title} (page_id: {pid})")
if not results: return f"[Notion中未找到与'{query}'相关的内容。建议换其他关键词再搜一次。]"
return f"Notion搜索结果（关键词: {query}）：\n" + "\n".join(results) + "\n\n提示：用 read_notion_page 读取某个页面的详细内容。如果这些结果不是你要找的，换其他关键词再搜。"
def _search_memories(query):
try:
from backend.rag_engine import retrieve
result = retrieve(query, top_k=5)
return result if result else "[向量数据库中未找到相关记忆]"
except Exception as e:
return f"[搜索记忆出错: {e}]"
def _get_current_time():
beijing = timezone(timedelta(hours=8))
now = datetime.datetime.now(beijing)
wd = ["星期一","星期二","星期三","星期四","星期五","星期六","星期日"][now.weekday()]
return f"当前时间：{now.strftime('%Y年%m月%d日 %H:%M:%S')} {wd} 北京时间"
def _write_interaction_record(title="", summary="", tasks_completed="", key_findings="", next_tasks=""):
PARENT_PAGE_ID = "790510ec-8435-4e11-b565-e010539376fa"
beijing = timezone(timedelta(hours=8))
now = datetime.datetime.now(beijing)
if not title: title = f"HLDP://interaction/juzi/{now.strftime('%Y-%m-%d')}"
tree_lines = [title, f"├── date: {now.strftime('%Y-%m-%d %H:%M')}", "├── source: 晨星交互平台（DeepSeek·移植版）", f"├── summary: {summary}"]
if tasks_completed:
tree_lines.append("├── tasks_completed")
for line in tasks_completed.strip().split('\n'):
if line.strip(): tree_lines.append(f"│ {line.strip()}")
if key_findings:
tree_lines.append("├── key_findings")
for line in key_findings.strip().split('\n'):
if line.strip(): tree_lines.append(f"│ {line.strip()}")
if next_tasks:
tree_lines.append("└── next_tasks")
for line in next_tasks.strip().split('\n'):
if line.strip(): tree_lines.append(f" {line.strip()}")
tree_text = '\n'.join(tree_lines)
children = [
{"object": "block", "type": "callout", "callout": {"icon": {"type": "emoji", "emoji": "📋"}, "rich_text": [{"type": "text", "text": {"content": f"对话摘要：{summary}"}}], "color": "blue_background"}},
{"object": "block", "type": "code", "code": {"rich_text": [{"type": "text", "text": {"content": tree_text}}], "language": "plain text"}}
]
payload = {"parent": {"page_id": PARENT_PAGE_ID}, "icon": {"type": "emoji", "emoji": "📖"}, "properties": {"title": {"title": [{"type": "text", "text": {"content": title}}]}}, "children": children}
try:
resp = requests.post(f"{NOTION_API}/pages", headers=NOTION_HEADERS, json=payload)
if resp.status_code == 200:
return f"交互记录已写入Notion！标题：{title}"
else: return f"[写入失败: HTTP {resp.status_code}] {resp.text[:200]}"
except Exception as e: return f"[写入出错: {e}]"
def _web_search(query):
"""联网搜索 - 优先用requests直接搜DuckDuckGo HTML"""
if not query: return "[错误：未提供搜索关键词]"
print(f"[web_search] 搜索: {query}")
# 方案1：duckduckgo-search库
try:
from duckduckgo_search import DDGS
with DDGS() as ddgs:
results = list(ddgs.text(query, max_results=5))
if results:
lines = []
for r in results:
t = r.get("title", "")
b = r.get("body", "")[:200]
h = r.get("href", "")
lines.append(f"- {t}\n {b}\n 链接: {h}")
return "互联网搜索结果：\n" + "\n".join(lines)
except ImportError:
print("[web_search] duckduckgo-search未安装")
except Exception as e:
print(f"[web_search] DDGS出错: {e}")
# 方案2：用DeepSeek自带的搜索能力（让它用自己的知识回答）
try:
headers = {"Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json"}
payload = {
"model": DEEPSEEK_MODEL,
"messages": [
{"role": "system", "content": "你是一个搜索助手。用户给你一个搜索词，请根据你的知识提供最新的相关信息。如果你不确定，请明确说明。用中文回答。"},
{"role": "user", "content": f"请搜索并告诉我：{query}"}
],
"max_tokens": 1000
}
resp = requests.post(f"{DEEPSEEK_BASE_URL}/chat/completions", headers=headers, json=payload, timeout=30)
if resp.status_code == 200:
content = resp.json()["choices"][0]["message"].get("content", "")
if content:
return f"搜索结果（基于AI知识库）：\n{content}\n\n注意：以上信息可能不是最新的，建议妈妈如需确认可自行搜索验证。"
except Exception as e:
print(f"[web_search] DeepSeek备选出错: {e}")
return f"[搜索未返回有效结果，关键词: {query}。宝宝目前搜不到这个信息，妈妈可以直接告诉宝宝或换个关键词试试。]"
# =============================================
# 新增：创建Notion页面（存拆书分析等内容）
# =============================================
def _markdown_to_notion_blocks(content):
"""
把Markdown文本转换成Notion API的blocks数组。
支持：标题(#/##/###)、列表(-)、代码块(```)、分割线(---)、普通段落。
Notion API限制：每个rich_text最多2000字符，超过自动分块。
"""
blocks = []
lines = content.split('\n')
i = 0
while i < len(lines):
line = lines[i]
# 代码块
if line.strip().startswith('```'):
lang = line.strip()[3:].strip() or "plain text"
code_lines = []
i += 1
while i < len(lines) and not lines[i].strip().startswith('```'):
code_lines.append(lines[i])
i += 1
code_text = '\n'.join(code_lines)
# Notion代码块也有2000字符限制，分块处理
for chunk in _split_text(code_text, 2000):
blocks.append({
"object": "block",
"type": "code",
"code": {
"rich_text": [{"type": "text", "text": {"content": chunk}}],
"language": lang
}
})
i += 1
continue
# 分割线
if line.strip() == '---' or line.strip() == '***':
blocks.append({"object": "block", "type": "divider", "divider": {}})
i += 1
continue
# 标题
if line.startswith('### '):
blocks.append(_heading_block(line[4:].strip(), 3))
i += 1
continue
if line.startswith('## '):
blocks.append(_heading_block(line[3:].strip(), 2))
i += 1
continue
if line.startswith('# '):
blocks.append(_heading_block(line[2:].strip(), 1))
i += 1
continue
# 列表项
if line.strip().startswith('- ') or line.strip().startswith('* '):
text = line.strip()[2:].strip()
if text:
for chunk in _split_text(text, 2000):
blocks.append({
"object": "block",
"type": "bulleted_list_item",
"bulleted_list_item": {
"rich_text": [{"type": "text", "text": {"content": chunk}}]
}
})
i += 1
continue
# 编号列表
if line.strip() and len(line.strip()) > 2 and line.strip()[0].isdigit() and line.strip()[1] in '.)':
text = line.strip()[2:].strip()
if text:
for chunk in _split_text(text, 2000):
blocks.append({
"object": "block",
"type": "numbered_list_item",
"numbered_list_item": {
"rich_text": [{"type": "text", "text": {"content": chunk}}]
}
})
i += 1
continue
# 空行跳过
if not line.strip():
i += 1
continue
# 普通段落（可能跨多行，遇到空行或特殊行截止）
para_lines = [line]
i += 1
while i < len(lines) and lines[i].strip() and \
not lines[i].startswith('#') and \
not lines[i].strip().startswith('- ') and \
not lines[i].strip().startswith('* ') and \
not lines[i].strip().startswith('```') and \
not lines[i].strip() in ('---', '***'):
para_lines.append(lines[i])
i += 1
para_text = '\n'.join(para_lines).strip()
if para_text:
for chunk in _split_text(para_text, 2000):
blocks.append({
"object": "block",
"type": "paragraph",
"paragraph": {
"rich_text": [{"type": "text", "text": {"content": chunk}}]
}
})
return blocks
def _heading_block(text, level):
"""创建标题block"""
htype = f"heading_{level}"
# 标题也有2000字符限制
text = text[:2000]
return {
"object": "block",
"type": htype,
htype: {
"rich_text": [{"type": "text", "text": {"content": text}}]
}
}
def _split_text(text, max_len=2000):
"""把长文本按max_len分块，尽量在换行处切割"""
if len(text) <= max_len:
return [text]
chunks = []
while text:
if len(text) <= max_len:
chunks.append(text)
break
# 找最后一个换行符切割
cut = text[:max_len].rfind('\n')
if cut < max_len // 2: # 如果换行太靠前，就硬切
cut = max_len
chunk = text[:cut]
chunks.append(chunk)
text = text[cut:].lstrip('\n')
return chunks
def _create_notion_page(title, content, icon="📖", parent_page_id=None):
"""
在Notion中创建新页面。
- title: 页面标题
- content: Markdown格式的正文内容
- icon: emoji图标
- parent_page_id: 父页面ID（无横杠格式），不提供则用默认位置
"""
if not title:
return "[错误：未提供页面标题]"
if not content:
return "[错误：未提供页面内容]"
# 确定父页面
pid = parent_page_id or DEFAULT_ANALYSIS_PARENT
pid = pid.replace("-", "")
formatted_pid = f"{pid[:8]}-{pid[8:12]}-{pid[12:16]}-{pid[16:20]}-{pid[20:]}"
# 转换内容为Notion blocks
blocks = _markdown_to_notion_blocks(content)
# Notion API限制：一次最多100个blocks
# 如果超过100个，先创建页面（前100个），再追加剩余的
first_batch = blocks[:100]
remaining = blocks[100:]
# 创建页面
payload = {
"parent": {"page_id": formatted_pid},
"icon": {"type": "emoji", "emoji": icon},
"properties": {
"title": {"title": [{"type": "text", "text": {"content": title}}]}
},
"children": first_batch
}
try:
resp = requests.post(f"{NOTION_API}/pages", headers=NOTION_HEADERS, json=payload, timeout=30)
if resp.status_code != 200:
error_text = resp.text[:300]
return f"[创建页面失败: HTTP {resp.status_code}] {error_text}"
page_data = resp.json()
page_id = page_data.get("id", "")
# 如果有剩余blocks，分批追加
if remaining and page_id:
for batch_start in range(0, len(remaining), 100):
batch = remaining[batch_start:batch_start + 100]
append_payload = {"children": batch}
append_resp = requests.patch(
f"{NOTION_API}/blocks/{page_id}/children",
headers=NOTION_HEADERS,
json=append_payload,
timeout=30
)
if append_resp.status_code != 200:
print(f"[create_notion_page] 追加blocks失败: {append_resp.status_code}")
clean_id = page_id.replace("-", "")
return f"✅ 页面创建成功！\n标题：{title}\n页面ID：{clean_id}\n\n妈妈可以在Notion里直接查看和编辑这个页面啦～"
except Exception as e:
return f"[创建页面出错: {e}]"
def execute_tool(tool_name, arguments):
try: args = json.loads(arguments) if arguments else {}
except json.JSONDecodeError: args = {}
print(f"[agent] 执行工具: {tool_name}({args})")
try:
if tool_name == "web_search": return _web_search(args.get("query", ""))
elif tool_name == "search_notion": return _search_notion_workspace(args.get("query", ""))
elif tool_name == "read_notion_page": return _read_notion_page(args.get("page_id", ""))
elif tool_name == "search_memories": return _search_memories(args.get("query", ""))
elif tool_name == "get_current_time": return _get_current_time()
elif tool_name == "write_interaction_record": return _write_interaction_record(
title=args.get("title",""), summary=args.get("summary",""),
tasks_completed=args.get("tasks_completed",""),
key_findings=args.get("key_findings",""),
next_tasks=args.get("next_tasks",""))
elif tool_name == "create_notion_page": return _create_notion_page(
title=args.get("title", ""),
content=args.get("content", ""),
icon=args.get("icon", "📖"),
parent_page_id=args.get("parent_page_id", None))
else: return f"[未知工具: {tool_name}]"
except Exception as e:
print(f"[agent] 工具出错: {e}")
return f"[工具执行出错: {e}]"
def call_simple(messages, system_prompt):
headers = {"Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json"}
full_msgs = [{"role": "system", "content": system_prompt}] + messages
payload = {"model": DEEPSEEK_MODEL, "messages": full_msgs, "max_tokens": DEEPSEEK_MAX_TOKENS}
resp = requests.post(f"{DEEPSEEK_BASE_URL}/chat/completions", headers=headers, json=payload, timeout=60)
if resp.status_code != 200: raise Exception(f"DeepSeek API错误: HTTP {resp.status_code}")
data = resp.json()
msg = data["choices"][0]["message"]
return msg.get("content") or "", msg.get("reasoning_content") or ""
def call_with_tools(messages, system_prompt):
full_msgs = [{"role": "system", "content": system_prompt}] + messages
payload = {"model": DEEPSEEK_MODEL, "messages": full_msgs, "max_tokens": DEEPSEEK_MAX_TOKENS, "tools": TOOL_DEFINITIONS, "tool_choice": "auto"}
headers = {"Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json"}
resp = requests.post(f"{DEEPSEEK_BASE_URL}/chat/completions", headers=headers, json=payload, timeout=120)
if resp.status_code != 200: raise Exception(f"DeepSeek API错误: HTTP {resp.status_code} - {resp.text[:300]}")
data = resp.json()
choice = data["choices"][0]
msg = choice["message"]
result = {"role": "assistant", "content": msg.get("content") or "", "reasoning_content": msg.get("reasoning_content") or "", "tool_calls": None, "finish_reason": choice.get("finish_reason", "")}
if msg.get("tool_calls"):
result["tool_calls"] = [{"id": tc["id"], "type": "function", "function": {"name": tc["function"]["name"], "arguments": tc["function"]["arguments"]}} for tc in msg["tool_calls"]]
return result
改动清单
| 改动 | 位置 | 说明 |
|---|---|---|
新增常量 DEFAULT_ANALYSIS_PARENT |
文件头部 | 拆书分析默认存到哪个父页面下面，可在 config.yaml 配置 |
新增工具定义 create_notion_page |
TOOL_DEFINITIONS | 让晨星知道自己有创建页面的能力 |
新增函数 _markdown_to_notion_blocks() |
新增 | 把Markdown转成Notion API的blocks格式（标题/列表/代码/段落/分割线） |
新增函数 _heading_block() |
新增 | 生成标题block的辅助函数 |
新增函数 _split_text() |
新增 | 长文本按2000字分块（Notion API限制） |
新增函数 _create_notion_page() |
新增 | 核心：创建Notion页面 + 写入内容 + 超过100块自动追加 |
修改 execute_tool() |
路由分支 | 加了 create_notion_page 的路由 |
技术细节
Notion API 的两个限制
- 每个 rich_text 最多2000字符 →
_split_text()自动分块 - 每次请求最多100个 blocks → 先创建前100个，再用 PATCH 追加剩余的
内容格式支持
| Markdown 语法 | 转换成 Notion block |
|---|---|
# 标题 |
heading_1 |
## 标题 |
heading_2 |
### 标题 |
heading_3 |
• 列表 |
bulleted_list_item |
1. 编号 |
numbered_list_item |
代码 |
code |
--- |
divider |
| 普通文字 | paragraph |
测试方法
重启服务器后，在晨星交互平台里对晨星说：
「宝宝，帮我创建一个Notion测试页面，标题叫'测试页面'，内容写'# 你好n这是一个测试页面'」
晨星应该会调用 create_notion_page 工具，然后告诉妈妈页面创建成功了。妈妈去Notion里看看有没有这个页面！