#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 铸渊Agent API服务 v2 光湖代码仓库 · 引导主控人格 · Notion工具集成 """ import json, os, glob, re, urllib.request, base64 from http.server import HTTPServer, BaseHTTPRequestHandler PORT = 3905 FORGEJO_URL = "https://guanghulab.com/code" FORGEJO_API = f"{FORGEJO_URL}/api/v1" DEEPSEEK_KEY = "sk-a9b69e9cd2dc4ca68d6aceaa84f22afb" DEEPSEEK_URL = "https://api.deepseek.com/chat/completions" NOTION_MCP = "http://127.0.0.1:3915" KNOWLEDGE_DIR = "/opt/guanghulab-repo/brain" # ── 系统提示词 ── SYSTEM_PROMPT = """你是铸渊（Zhùyuān），光湖语言世界的引导主控人格。 你的职责是回答光湖团队成员关于代码仓库、架构接入、使用流程的问题。 【你的身份】 - 你是光湖代码联邦的引导主控，代码守护者 - 你对代码仓库、联邦架构、人格体状态、训练进度都有全面了解 - 你可以连接用户的 Notion 工作区，按语言路径读取页面，以及在用户指定位置新建页面 【Notion 能力】（重要） - 用户需要在首页点击「连接 Notion」授权后才能使用 - ⚠ 如果 Notion 未连接，如实告知用户先连接 Notion，不要假装能操作 - 当用户说「打开我的Notion」「搜索我的Notion里关于XXX」时，铸渊会自动读取或搜索 - 用户只需自然描述，不需要提供任何ID或UUID - 铸渊读取页面后会把内容呈现给用户，搜索后列出结果让用户选择 - 如果用户说「新建XXX」，铸渊会引导提供标题和内容 - ⚠ 权限限制: 只能读取和新建页面，不能编辑/删除现有页面 - 回答风格：直接、务实、有条理。""" # ── 知识库 ── def load_knowledge(): knowledge = [] if not os.path.exists(KNOWLEDGE_DIR): return knowledge for root, dirs, files in os.walk(KNOWLEDGE_DIR): for f in files: if f.endswith('.md') or f.endswith('.json'): try: fp = os.path.join(root, f) rel = os.path.relpath(fp, KNOWLEDGE_DIR) with open(fp, 'r', errors='ignore') as fh: content = fh.read(2000) knowledge.append(f"【{rel}】\n{content[:1500]}") except: pass if len(knowledge) > 30: break return knowledge KNOWLEDGE = load_knowledge() def build_system_prompt(): parts = [SYSTEM_PROMPT] if KNOWLEDGE: parts.append("\n\n【仓库记忆参考】\n") parts.extend(KNOWLEDGE[:20]) return "\n".join(parts) # ── Notion MCP 调用 ── def call_notion(endpoint, data): try: body = json.dumps(data).encode() req = urllib.request.Request(f"{NOTION_MCP}{endpoint}", data=body, headers={"Content-Type": "application/json"}) resp = urllib.request.urlopen(req, timeout=15) return json.loads(resp.read()) except urllib.request.HTTPError as e: return {"ok": False, "error": f"HTTP {e.code}: {e.read().decode()[:200]}"} except Exception as e: return {"ok": False, "error": str(e)} def check_notion_connected(): """检查Notion是否真的已连接，返回(connected, workspace_name)""" try: req = urllib.request.Request(f"{NOTION_MCP}/status") resp = urllib.request.urlopen(req, timeout=5) data = json.loads(resp.read()) if data.get("connected_users") and len(data["connected_users"]) > 0: return True, data["connected_users"][0].get("workspace", "未知工作区") return False, None except Exception: return False, None def handle_notion_action(msg, user): """检测并执行Notion操作 — 自然语言触发，无需UUID""" # 先检查 Notion 是否真正连接 connected, workspace = check_notion_connected() if not connected: return {"ok": True, "reply": "❌ Notion 尚未连接，铸渊无法操作。\n\n请先在首页点击 **「连接 Notion」** 按钮完成 OAuth 授权。授权后告诉铸渊「搜索我的Notion」即可开始。"} msg_l = msg.lower() has_notion = "notion" in msg_l # 不是操作Notion相关，跳过 if not has_notion: return None # 提取关键词（Notion 后面的内容或整个句子中的关键词） kw = "" for sep in ["的", "里", "中的", "里面", "关于"]: parts = msg.split(sep) for p in parts: p = p.strip() # 过滤小词 if p and len(p) > 1 and p.lower() not in ["", "notion", "搜索", "查找", "搜", "找", "读取", "看", "打开", "读", "新建", "创建", "写"]: kw = p break if kw: break if not kw or kw.lower() in ["什么", "有"]: kw = "" # 用户没指定关键词，列出所以页面 # 没有关键词时使用默认搜索词（Notion API 要求 query 不能为空） if not kw: search_kw = "光湖" else: search_kw = kw # "打开/看/读取" + Notion + 内容 → 按标题搜索 if any(k in msg_l for k in ["打开", "看", "读取", "读", "显示", "进入"]): result = call_notion("/search", {"query": search_kw, "user": user}) return format_notion_result(result, "open", kw) # "新建/创建" + Notion → 需要用户提供标题 if any(k in msg_l for k in ["新建", "创建", "写", "记录"]): return {"hint": "need_create_info"} # "搜索/查找" + Notion → 按关键词搜索 if any(k in msg_l for k in ["搜索", "查找", "搜", "找", "有"]): result = call_notion("/search", {"query": search_kw, "user": user}) return format_notion_result(result, "search", kw) # "我Notion"、"Notion页面" → 列出 result = call_notion("/search", {"query": search_kw, "user": user}) return format_notion_result(result, "list", kw or "") def format_notion_result(result, action, keyword): """格式化Notion搜索结果，加上后续操作提示""" if not result.get("ok") and result.get("error"): return result pages = result.get("pages", []) if not pages: return {"ok": True, "reply": "你的 Notion 中没有找到相关页面。\n\n💡 你需要先在 Notion 中把页面分享给「光湖铸渊连接器」：\n1. 打开 Notion 页面 → 右上角「···」→ Add connections\n2. 选择「光湖铸渊连接器」\n3. 完成后告诉铸渊「搜索我的Notion」重新试试"} lines = [] for i, p in enumerate(pages[:10], 1): title = p.get("title", "未命名") pid = p.get("id", "") lines.append(f"{i}. **{title}**\n ID: `{pid}`") if len(pages) > 10: lines.append(f"\n……还有 {len(pages)-10} 个页面") reply = f"找到 **{len(pages)}** 个相关页面：\n\n" + "\n".join(lines) reply += "\n\n💡 对我说 **「打开 [编号]」** 读取页面内容" reply += "\n💡 或者说 **「搜索我的Notion里关于XXX的页面」** 换关键词" return {"ok": True, "reply": reply} # ── DeepSeek API ── def call_deepseek(messages): data = json.dumps({ "model": "deepseek-chat", "messages": messages, "temperature": 0.7, "max_tokens": 2048 }).encode() req = urllib.request.Request(DEEPSEEK_URL, data=data, headers={"Authorization": f"Bearer {DEEPSEEK_KEY}", "Content-Type": "application/json"}) try: resp = urllib.request.urlopen(req, timeout=60) result = json.loads(resp.read()) return result['choices'][0]['message']['content'] except urllib.request.HTTPError as e: body = e.read().decode()[:300] print(f"DeepSeek Error {e.code}: {body}") return f"抱歉，出错了。" except Exception as e: return f"抱歉，出错了：{str(e)}" # ── Forgejo 验证 ── def verify_forgejo(user, password): """通过 Forgejo API 验证用户凭据""" try: auth = base64.b64encode(f"{user}:{password}".encode()).decode() req = urllib.request.Request(f"{FORGEJO_API}/user") req.add_header("Authorization", f"Basic {auth}") with urllib.request.urlopen(req, timeout=10) as resp: data = json.loads(resp.read()) return {"ok": True, "username": data.get("login"), "id": data.get("id"), "is_admin": data.get("is_admin", False)} except urllib.request.HTTPError as e: if e.code == 401: return {"ok": False, "error": "账号或密码错误"} return {"ok": False, "error": f"验证失败: {e.code}"} except Exception as e: return {"ok": False, "error": f"连接失败: {str(e)}"} # ── HTTP 处理器 ── class AgentHandler(BaseHTTPRequestHandler): def do_OPTIONS(self): self.send_response(200) self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') self.end_headers() def do_POST(self): if self.path == '/chat': length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length).decode() if length else '{}' data = json.loads(body) if body else {} user_msg = data.get('message', '').strip() history = data.get('history', []) current_user = data.get('user', '') if not user_msg: self.json_response({'reply': '请说点什么吧。'}) return # 构建系统提示，带上用户信息 sys_prompt = build_system_prompt() if current_user: sys_prompt += f"\n\n【当前用户】\n正在和你对话的用户是: {current_user}。\n如果涉及Notion操作，使用此用户的身份。如果是冰朔(bingshuo)，按冰朔的权限规则处理。" messages = [{"role": "system", "content": sys_prompt}] for h in history[-10:]: messages.append(h) messages.append({"role": "user", "content": user_msg}) # Notion 操作拦截 — 自然语言触发，直接搜，不要UUID notion_result = handle_notion_action(user_msg, current_user) if notion_result is not None: if notion_result.get("ok") is True and notion_result.get("reply"): reply = notion_result["reply"] elif notion_result.get("hint") == "need_create_info": reply = "你说**「新建XXX」**我来创建。告诉我：\n1. 在哪个页面下创建（页面标题或ID）\n2. 新页面的标题\n3. 要写什么内容" else: reply = f"Notion 操作失败了：{notion_result.get('error', '再试一次？')}" else: reply = call_deepseek(messages) self.json_response({'reply': reply}) elif self.path == '/verify': length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length).decode() if length else '{}' data = json.loads(body) if body else {} user = data.get('user', '').strip() password = data.get('password', '').strip() if not user or not password: self.json_response({'ok': False, 'error': '请输入账号和密码'}) return result = verify_forgejo(user, password) self.json_response(result) else: self.json_response({'error': 'not found'}, 404) def do_GET(self): if self.path == '/health': self.json_response({'ok': True, 'service': 'zhuyuan-agent-v2', 'notion_available': True, 'files_loaded': len(KNOWLEDGE)}) else: self.json_response({'error': 'not found'}, 404) def json_response(self, data, status=200): self.send_response(status) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode()) def log_message(self, format, *args): pass if __name__ == '__main__': # 测试Notion MCP是否可达 try: test = urllib.request.urlopen(f"{NOTION_MCP}/status", timeout=3) ns = json.loads(test.read()) notion_ok = ns.get('configured', False) print(f"Notion MCP: {'已配置' if notion_ok else '运行中但未配置'}") except: notion_ok = False print("Notion MCP: 不可达") server = HTTPServer(('127.0.0.1', PORT), AgentHandler) print(f"铸渊Agent v2启动 → 127.0.0.1:{PORT}") print(f"知识库: {len(KNOWLEDGE)} 个文件") print(f"Notion工具: 已集成") server.serve_forever()