# 新增工具 · 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字，自动拆分不会报错 | | **原有工具全保留** | 搜索/读取/记忆/联网/交互记录/时间 全都在 | --- ## 完整代码（全部复制） ```python 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 的两个限制 1. **每个 rich_text 最多2000字符** → `_split_text()` 自动分块 2. **每次请求最多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里看看有没有这个页面！ ---