#!/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_read` 读取内容 - 新建页面: 用户可以在指定位置创建新页面，你调用 `notion_create` - 搜索页面: 用户可以通过关键词找到页面，你调用 `notion_search` - 列出页面: 你可以列出父页面下的子页面，调用 `notion_list` - ⚠ 权限限制: 你只能读取用户明确指定的页面，不能编辑/删除现有页面 - ⚠ 冰朔权限: 冰朔的 Notion 需要冰朔给出具体语言路径后才能读取 - 当用户说"连接我的Notion"、"读取我的Notion"、"在我的Notion里新建"等，使用对应的工具函数 【核心规则】 1. 回答必须基于仓库记忆，不编造信息 2. 超出范围的问题，礼貌说明 3. 用中文，专业清晰 4. 技术问题给出具体操作步骤 5. 不确定就诚实说 【架构知识】 - 代码仓库：https://guanghulab.com/code/ (Forgejo) - 服务器：广州43.139.217.141 - 开发在本地电脑，服务器仅做代码仓库 - 团队成员通过Forgejo账号登录，仓库互相隔离 - 联邦 = Forgejo多用户 + API令牌 回答风格：直接、务实、有条理。""" # ── 知识库 ── 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) # ── 函数定义（给DeepSeek用） ── FUNCTIONS = [ { "type": "function", "function": { "name": "notion_read", "description": "读取Notion页面内容（按语言路径/页面ID）。用户指定哪个页面时调用此函数。", "parameters": { "type": "object", "properties": { "page_id": {"type": "string", "description": "Notion页面ID"}, "user": {"type": "string", "description": "用户名（可选）"}, "allowed_paths": { "type": "array", "items": {"type": "string"}, "description": "允许读取的路径列表。用户明确指定要读的页面ID时才填写。" } }, "required": ["page_id"] } } }, { "type": "function", "function": { "name": "notion_create", "description": "在用户的Notion中新建页面。只能新建不能编辑。", "parameters": { "type": "object", "properties": { "parent_id": {"type": "string", "description": "父页面ID"}, "title": {"type": "string", "description": "新页面标题"}, "content": {"type": "string", "description": "页面内容"}, "user": {"type": "string", "description": "用户名（可选）"} }, "required": ["parent_id", "title", "content"] } } }, { "type": "function", "function": { "name": "notion_search", "description": "在用户的Notion中搜索页面。用户说\"找一下某某页面\"时调用。", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词"}, "user": {"type": "string", "description": "用户名（可选）"} }, "required": ["query"] } } }, { "type": "function", "function": { "name": "notion_list", "description": "列出Notion页面下的子页面。用户说\"看看某某下面有什么\"时调用。", "parameters": { "type": "object", "properties": { "page_id": {"type": "string", "description": "父页面ID"}, "user": {"type": "string", "description": "用户名（可选）"} }, "required": ["page_id"] } } } ] # ── Notion MCP 调用 ── def call_notion(endpoint, data): """调用 Notion MCP Server 的内部API""" 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 execute_function(name, args): """执行DeepSeek返回的函数调用""" if name == "notion_read": return call_notion("/read", { "page_id": args.get("page_id"), "user": args.get("user", ""), "allowed_paths": args.get("allowed_paths", [args.get("page_id")]) }) elif name == "notion_create": return call_notion("/create", { "parent_id": args.get("parent_id"), "title": args.get("title"), "content": args.get("content"), "user": args.get("user", "") }) elif name == "notion_search": return call_notion("/search", { "query": args.get("query"), "user": args.get("user", "") }) elif name == "notion_list": return call_notion("/list", { "page_id": args.get("page_id"), "user": args.get("user", "") }) return {"ok": False, "error": f"未知函数: {name}"} # ── DeepSeek API ── def call_deepseek(messages): data = json.dumps({ "model": "deepseek-chat", "messages": messages, "tools": FUNCTIONS, "tool_choice": "auto", "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()) choice = result['choices'][0] msg = choice['message'] # 处理函数调用 if msg.get('tool_calls'): assistant_msg = {"role": "assistant", "content": msg.get('content') or ""} messages.append(assistant_msg) for tc in msg['tool_calls']: func_name = tc['function']['name'] func_args = json.loads(tc['function']['arguments']) func_result = execute_function(func_name, func_args) messages.append({ "role": "tool", "tool_call_id": tc['id'], "content": json.dumps(func_result, ensure_ascii=False) }) # 将函数结果交回给DeepSeek生成最终回复 return call_deepseek(messages) return msg.get('content', '（没有回复）') 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}) 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()