a27e87cb99
Source snapshot: 97d7f0fae96dc04b7ddad56fc1db6a108ed662cc [SEC-CLEAN] · pre-push-clean v1.0 · 109处敏感信息已自动转乱码
639 lines
24 KiB
Markdown
639 lines
24 KiB
Markdown
# 修复补丁 · 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> 标签内）
|
||
h3_blocks = re.findall(r'<h3[^>]*>(.*?)</h3>', 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'<span class="[^"]*content[_-][^"]*"[^>]*>(.*?)</span>', 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)
|
||
``` |