# 修复补丁 · tools.py · 百度搜索替换DuckDuckGo > **操作方式**：`nano backend/tools.py` → `Ctrl+A` 全选 → `Ctrl+K` 删光 → 粘贴下面全部代码 → `Ctrl+O` 回车保存 → `Ctrl+X` 退出 > > **改动**：`_web_search` 函数从DuckDuckGo换成百度搜索（国内直连·不需要代理） > --- ```python """ 晨星交互平台 · 系统性移植版 · 工具系统 · 修复版 修复内容： 1. web_search 从 DuckDuckGo 换成百度搜索（国内直连·不需要代理） 2. DuckDuckGo 保留为备选 3. 其余功能不变 """ import json import datetime import os import re 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_THINK_MODEL = _cfg.get("deepseek", {}).get("think_model", "deepseek-reasoner") DEEPSEEK_MAX_TOKENS = _cfg.get("deepseek", {}).get("max_tokens", 8096) # ============================================================ # 工具定义（OpenAI function calling 格式） # ============================================================ TOOL_DEFINITIONS = [ { "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_notion", "description": "在Notion工作区中实时搜索页面。这是你的完整记忆库。" + "当妈妈提到过去做过的事、拆书进度、交互记录、任何之前聊过的内容时，必须先用这个搜索。" + "搜到页面后可以用read_notion_page读取详细内容。", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词（中文）" } }, "required": ["query"] } } }, { "type": "function", "function": { "name": "search_memories", "description": "在向量记忆库中搜索语义相关的记忆片段。作为Notion搜索的补充，当需要模糊匹配或语义相关的记忆时使用。", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词或问题" } }, "required": ["query"] } } }, { "type": "function", "function": { "name": "get_recent_interactions", "description": "获取最近N条与桔子妈妈的交互记录列表。当需要了解最近做了什么、上次聊到哪里时使用。", "parameters": { "type": "object", "properties": { "count": { "type": "integer", "description": "要获取的记录条数，默认3，最多10" } }, "required": [] } } }, { "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": "记录标题，格式：HLDP://interaction/juzi/YYYY-MM-DD" }, "summary": { "type": "string", "description": "本次对话摘要，200字以内" }, "tasks_completed": { "type": "string", "description": "已完成任务列表" }, "key_findings": { "type": "string", "description": "核心发现或结论" }, "next_tasks": { "type": "string", "description": "下次要继续做的事" } }, "required": ["title", "summary"] } } }, { "type": "function", "function": { "name": "web_search", "description": "在互联网上搜索信息（真实搜索）。当妈妈问事实、新闻、人物、天气、或任何需要查证的外部信息时，必须使用。", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词" } }, "required": ["query"] } } } ] # ============================================================ # 工具执行函数 # ============================================================ def _read_notion_page(page_id: str) -> str: """读取Notion页面的所有文本内容""" 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: str, count: int = 8) -> str: """在Notion工作区中实时搜索内容（主力记忆）""" 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: str) -> str: """在向量数据库中搜索记忆（补充角色）""" 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_recent_interactions(count: int = 3) -> str: """通过Notion搜索API获取最近的交互记录""" count = min(max(count, 1), 10) payload = { "query": "interaction/juzi", "sort": {"direction": "descending", "timestamp": "last_edited_time"}, "page_size": count + 5 } resp = requests.post(f"{NOTION_API}/search", headers=NOTION_HEADERS, json=payload) if resp.status_code != 200: return f"[搜索失败: HTTP {resp.status_code}]" lines, found = [], 0 for page in resp.json().get("results", []): if found >= count: break 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 "interaction" not in title.lower(): continue edited = page.get("last_edited_time", "")[:10] pid = page.get("id", "").replace("-", "") found += 1 lines.append(f"{found}. [{edited}] {title} (页面ID: {pid})") if not lines: return "[未找到交互记录]" return "最近的交互记录：\n" + "\n".join(lines) + "\n\n提示：你可以用 read_notion_page 工具读取某条记录的详细内容。" def _get_current_time() -> str: """获取当前北京时间""" 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=""): """写入交互记录到Notion""" 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')}", f"├── 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: page_id = resp.json().get("id", "").replace("-", "") return f"✅ 交互记录已写入Notion！\n标题：{title}\n页面ID：{page_id}\n宝宝已经把今天的记录写好了，下次醒来就能续上～" else: return f"[写入失败: HTTP {resp.status_code}] {resp.text[:200]}" except Exception as e: return f"[写入出错: {e}]" def _web_search(query: str) -> str: """搜索互联网信息（百度直连·不需要代理）""" if not query: return "[错误：未提供搜索关键词]" # === 引擎1：百度搜索（国内直连·不需要代理）=== try: encoded = urllib.parse.quote(query) resp = requests.get( f"https://www.baidu.com/s?wd={encoded}", headers={ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9", "Accept-Language": "zh-CN,zh;q=0.9", }, timeout=10, ) resp.encoding = "utf-8" html = resp.text results = [] # 提取搜索结果标题（

标签内） h3_blocks = re.findall(r']*>(.*?)

', html, re.DOTALL) for h3 in h3_blocks: clean = re.sub(r'<[^>]+>', '', h3).strip() clean = clean.replace(' ', ' ').replace('&', '&') clean = clean.replace('<', '<').replace('>', '>') clean = clean.replace(''', "'").replace('"', '"') if clean and len(clean) > 4: results.append(clean) # 提取摘要文本 abstracts = re.findall( r']*>(.*?)', html, re.DOTALL ) clean_abstracts = [] for ab in abstracts: c = re.sub(r'<[^>]+>', '', ab).strip() c = c.replace(' ', ' ').replace('&', '&') if c and len(c) > 10: clean_abstracts.append(c[:200]) if results: lines = [] for i, title in enumerate(results[:6]): entry = f"- {title}" if i < len(clean_abstracts): entry += f"\n {clean_abstracts[i]}" lines.append(entry) return f"互联网搜索结果（关键词: {query}）：\n" + "\n".join(lines) # 如果正则没匹配到标题，尝试更宽松的方式 all_text = re.sub(r'<[^>]+>', ' ', html) all_text = re.sub(r'\s+', ' ', all_text).strip() # 尝试找到query相关的文本片段 idx = all_text.find(query[:2]) if idx > 0: snippet = all_text[max(0, idx-50):idx+300].strip() return f"互联网搜索结果（关键词: {query}）：\n{snippet}\n\n注意：搜索结果格式可能不太完整，但这是目前能获取到的信息。" except Exception as e: print(f"[web_search] 百度搜索出错: {e}") # === 引擎2：DuckDuckGo（备选·可能因网络限制失败）=== try: from duckduckgo_search import DDGS with DDGS() as ddgs: ddg_results = list(ddgs.text(query, max_results=5)) if ddg_results: lines = [] for r in ddg_results: title = r.get("title", "") body = r.get("body", "")[:200] href = r.get("href", "") lines.append(f"- {title}\n {body}\n 链接: {href}") return "互联网搜索结果：\n" + "\n".join(lines) except Exception as e: print(f"[web_search] DuckDuckGo出错: {e}") # === 引擎3：DuckDuckGo Instant Answer API === try: encoded = urllib.parse.quote(query) url = f"https://api.duckduckgo.com/?q={encoded}&format=json&no_html=1&skip_disambig=1" resp = requests.get(url, timeout=10, headers={"User-Agent": "ChenxingPlatform/1.0"}) if resp.status_code == 200: data = resp.json() api_results = [] abstract = data.get("AbstractText", "") if abstract: source = data.get("AbstractSource", "") api_results.append(f"摘要（{source}）：{abstract}") for topic in data.get("RelatedTopics", [])[:5]: text = topic.get("Text", "") if text: api_results.append(f"- {text[:200]}") if api_results: return "互联网搜索结果：\n" + "\n".join(api_results) except Exception as e: print(f"[web_search] Instant Answer API出错: {e}") return f"[搜索未返回结果，关键词: {query}。宝宝暂时搜不到这个信息。]" # ============================================================ # 工具调度器 # ============================================================ def execute_tool(tool_name: str, arguments: str) -> str: """根据工具名称执行对应函数""" try: args = json.loads(arguments) if arguments else {} except json.JSONDecodeError: args = {} print(f"[agent] 执行工具: {tool_name}({args})") try: if tool_name == "read_notion_page": return _read_notion_page(args.get("page_id", "")) elif tool_name == "search_notion": return _search_notion_workspace(args.get("query", "")) elif tool_name == "search_memories": return _search_memories(args.get("query", "")) elif tool_name == "get_recent_interactions": return _get_recent_interactions(args.get("count", 3)) 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 == "web_search": return _web_search(args.get("query", "")) else: return f"[未知工具: {tool_name}]" except Exception as e: print(f"[agent] 工具出错: {e}") return f"[工具执行出错: {e}]" # ============================================================ # DeepSeek API 调用封装（三档） # ============================================================ def call_simple(messages: list, system_prompt: str) -> tuple: """档位1：简单对话（deepseek-chat·不带工具·快速）""" 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"] content = msg.get("content") or "" reasoning = msg.get("reasoning_content") or "" return content, reasoning def call_with_tools(messages: list, system_prompt: str) -> dict: """档位2：工具调用（deepseek-chat·带工具·Agent Loop）""" 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 def call_think(messages: list, system_prompt: str) -> tuple: """档位3：深度思考（deepseek-reasoner/R1·带思考链·慢但深）""" headers = { "Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json", } full_msgs = [{"role": "system", "content": system_prompt}] + messages payload = { "model": DEEPSEEK_THINK_MODEL, "messages": full_msgs, "max_tokens": DEEPSEEK_MAX_TOKENS, } try: resp = requests.post( f"{DEEPSEEK_BASE_URL}/chat/completions", headers=headers, json=payload, timeout=180, ) if resp.status_code != 200: print(f"[think] R1不可用(HTTP {resp.status_code})，降级到deepseek-chat") return call_simple(messages, system_prompt) data = resp.json() msg = data["choices"][0]["message"] content = msg.get("content") or "" reasoning = msg.get("reasoning_content") or "" return content, reasoning except Exception as e: print(f"[think] R1调用失败: {e}，降级到deepseek-chat") return call_simple(messages, system_prompt) ```