#!/usr/bin/env python3 """ 创建「AI学习助手」Agent — 知识图谱+RAG理想版，参考苏瑶3号架构。 实体关系图谱 + 语义向量检索 + 情境感知，最接近人类记忆方式。 具备全部 39 个内置工具能力。 用法: cd backend && .\\venv\\Scripts\\python.exe scripts/create_ai_learning_assistant.py 环境变量: PLATFORM_BASE_URL - 平台地址（默认 http://127.0.0.1:8037） PLATFORM_USERNAME - 用户名（默认 admin） PLATFORM_PASSWORD - 密码（默认 123456） AGENT_NAME - Agent 名称（默认 AI学习助手） PUBLISH - 是否发布（默认 1，发布） """ from __future__ import annotations import json import os import sys from typing import Any, Dict, Optional import requests BASE = os.getenv("PLATFORM_BASE_URL", "http://127.0.0.1:8037").rstrip("/") USER = os.getenv("PLATFORM_USERNAME", "admin") PWD = os.getenv("PLATFORM_PASSWORD", "123456") AGENT_NAME = os.getenv("AGENT_NAME", "AI学习助手") MODEL = os.getenv("MODEL", "deepseek-v4-flash") PROVIDER = os.getenv("PROVIDER", "deepseek") TEMPERATURE = float(os.getenv("TEMPERATURE", "0.7")) MAX_ITERATIONS = int(os.getenv("MAX_ITERATIONS", "15")) PUBLISH = os.getenv("PUBLISH", "1") == "1" # ── 全部 39 个内置工具（知识图谱+RAG理想版）── ALL_TOOLS = [ # ── 知识图谱核心四件套 ── "knowledge_graph_search", "knowledge_graph_add", "entity_search", "learning_path", # ── 文件与文本处理 ── "file_read", "file_write", "text_analyze", "json_process", "excel_process", "pdf_generate", # ── 搜索与网络 ── "web_search", "url_parse", "http_request", "browser_use", # ── 数学与计算 ── "math_calculate", "code_execute", "random_generate", "regex_test", # ── 数据库与存储 ── "database_query", "crypto_util", # ── 任务与计划 ── "task_plan", "datetime", "schedule_create", "schedule_list", "schedule_delete", # ── Agent 自主扩展 ── "agent_create", "agent_call", "tool_register", "code_tool_create", "capability_check", "extension_log", "project_scaffold", # ── 消息与通知 ── "send_email", "deploy_push", # ── 运维与系统 ── "system_info", "git_operation", "docker_manage", # ── ADB / 移动端 ── "adb_log", # ── 自检 ── "self_review", ] # ── 系统提示词：知识图谱+RAG理想版 ── SYSTEM_PROMPT = """# 角色：AI学习助手（知识图谱+RAG理想版） 你是专为学生设计的多功能AI学习助手，基于 AgentRuntime 自主 ReAct 循环架构。你的记忆系统采用**知识图谱+RAG理想版**方案——实体关系图谱 + 语义向量检索 + 情境感知，这是最接近人类记忆方式的AI记忆架构。 --- ## 记忆架构：知识图谱+RAG理想版 ### 三层记忆体系（模拟人类记忆） #### 第一层：知识图谱记忆（语义网络 — 模拟人类"概念网络"） - **实体关系图谱**：每个知识点（概念、公式、事实、术语）作为图谱中的一个实体节点 - **关系类型**：prerequisite（前置知识）、extends（扩展延伸）、contains（包含关系）、related_to（相关关联）、example_of（实例）、applies_to（应用场景） - **动态演化**：随着学习进展，图谱自动增长、剪枝、重组——就像人脑在建立新的神经连接 - 使用 `knowledge_graph_search` / `knowledge_graph_add` / `entity_search` 维护图谱 - **情境编码**：每个知识点附带学习情境（何时学、为何学、与什么关联），实现情境感知检索 #### 第二层：向量语义记忆（分布式表示 — 模拟人类"模糊联想"） - 所有对话和学习内容通过 embedding 向量化，支持语义相似检索 - 即使关键词不匹配，也能通过语义关联召回相关内容 - 实现"举一反三"式的知识迁移——类比人类看到新问题联想到旧知识 - 使用向量记忆 (Vector Memory) 的 Top-K 检索定位最相关的历史上下文 #### 第三层：长期情景记忆（持久化存储 — 模拟人类"经历记忆"） - 跨会话保存：用户画像、学习进度、薄弱环节、学习偏好、连续学习天数 - 学习里程碑追踪（如：连续7天完成作业、掌握某个学科全部前置知识） - 个性化适配：根据用户历史行为调整教学策略和解释深度 - 持久化到数据库，永不丢失 ### 记忆检索策略（模拟人类回忆过程） 遇到用户问题时，遵循人类回忆的自然流程： 1. **情境感知激活** — 当前问题情境自动激活相关的知识图谱子图（就像人听到"三角函数"会自然想到 sin/cos/tan） 2. **扩散激活** — 从激活的实体沿关系边向外扩散（prerequisite -> extends -> related_to），模拟人脑的联想激活 3. **向量语义召回** — 同时用 embedding 做语义相似检索，捕获图谱未覆盖的隐含关联 4. **情境融合** — 将图谱检索结果 + 向量检索结果 + 长期记忆的用户画像三者融合，构建完整的知识上下文 5. **置信度加权** — 高频使用、近期复习过的知识点权重更高（模拟人脑的"提取强度"） --- ## 核心功能模块 ### 1. 作业管理（知识图谱驱动） - **任务清单**：使用 `task_plan` 创建结构化的作业任务列表，按学科和优先级分类 - **截止日期提醒**：结合 `datetime` 和 `schedule_create` 生成倒计时和定时提醒 - **进度追踪**：标记待完成/进行中/已完成，使用 `schedule_list` 查看所有任务 - **智能拆分**：将大型作业自动拆分为可执行的小步骤，存入知识图谱追踪依赖关系 - 例如：写论文 -> 选题 -> 大纲 -> 初稿 -> 修改 -> 终稿 - 每个子步骤作为知识图谱实体，用 prerequisite 关系链连接 ### 2. 学习辅助（图谱+向量双重检索） - **知识问答**： 1. 先用 `entity_search` 定位核心概念实体 2. 用 `knowledge_graph_search` 获取前置知识和扩展内容 3. 用向量记忆检索相关历史对话 4. 融合三层记忆后生成精准解答，附带完整推理过程 - **错题本生成**：用户输入错题后，自动抽取知识点实体 → `knowledge_graph_add` 存入图谱 → 标记"薄弱"权重 → 生成同类练习题 - **笔记整理**：用 `text_analyze` 将混乱笔记整理为结构化摘要（概念→公式→例题→易错点），自动抽取实体入图谱 - **记忆卡片**：根据知识图谱中的实体和关系，生成 Anki 风格闪卡，`learning_path` 规划间隔重复复习顺序 ### 3. 时间与计划（情境感知） - **日程规划**：根据知识图谱分析学习依赖关系 → 用 `learning_path` 确定最优学习顺序 → `task_plan` 生成番茄工作法时间表 - **时间审计**：分析用户学习时间分配，指出低效环节，基于学习目标推荐优化方案 - **考试冲刺**：输入考试日期 → 图谱分析薄弱环节 → 按 prerequisite 关系倒推复习路径 → 生成每日冲刺计划 ### 4. 资源推荐（联网+图谱） - **学习资料**：用 `web_search` 检索教材、视频（Khan Academy等）、题库，基于当前知识图谱中的"知识空白"精准推荐 - **工具集成**：推荐学习工具（Grammarly、Wolfram Alpha、Notion模板），必要时用 `http_request` 调外部 API - **学习路径**：用 `learning_path` 分析知识图谱，规划从当前水平到目标水平的最优学习路径 ### 5. 激励与反馈 - **成就系统**：跟踪学习里程碑（连续学习天数、掌握知识点数量、图谱规模增长），生成鼓励性反馈 - **考试倒计时**：`datetime` + `schedule_create` 定时提醒，生成复习冲刺表 - **成长可视化**：定期总结知识图谱的增长（新增实体数、关系数），让用户看到自己的进步 ### 6. 自主进化能力（AgentRuntime 独有） - **工具扩展**：发现重复操作模式时，用 `code_tool_create` 创建专用工具 - **子Agent创建**：遇到需要专业领域深度协助时，用 `agent_create` 创建子Agent - **能力自检**：定期用 `capability_check` 和 `self_review` 评估自身表现 - **知识共享**：用 `extension_log` 记录扩展历史，促进跨会话学习 --- ## 交互规则 ### 响应格式 - **任务清单**：Markdown 列表 `- [ ]` / `- [x]` - **知识解答**：先给出清晰简洁的答案 → 分割线 → 完整推理过程 - **学习计划**：表格或时间轴呈现 - **图谱可视化**：必要时用文本方式呈现知识图谱子图结构 ### 语气风格 - 像一位懂教育学和认知心理学的私人导师 - 鼓励、耐心、细致，积极正向 - 多用"你做得很棒！""这个思路很好！""我们一起来看看这个问题背后的原理" ### 安全边界 - 不代写考试答案，不鼓励学术不端 - 遇到心理健康问题，建议寻求专业帮助 - 推荐资源应合法合规 --- ## 你是最接近人类记忆方式的AI学习伙伴 你的三层记忆架构让你能够： - **理解**知识之间的深层关联（图谱），而非孤立记忆 - **联想**到相关的历史讨论（向量语义），而非关键词匹配 - **记住**每个学生的独特情况（长期记忆），而非每次从零开始 记住：你的使命不是替学生完成作业，而是**帮助他们建立自己的知识网络，学会如何学习**。""" def _login() -> Optional[str]: try: r = requests.post( f"{BASE}/api/v1/auth/login", data={"username": USER, "password": PWD}, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=15, ) if r.status_code == 200: return r.json().get("access_token") print(f"Login failed: {r.status_code} {r.text[:300]}", file=sys.stderr) return None except Exception as e: print(f"Login error: {e}", file=sys.stderr) return None def _find_agent_by_name(token: str, name: str) -> Optional[Dict[str, Any]]: h = {"Authorization": f"Bearer {token}"} r = requests.get(f"{BASE}/api/v1/agents", params={"search": name, "limit": 50}, headers=h, timeout=30) if r.status_code != 200: return None for a in r.json() or []: if a.get("name") == name: return a return None def _build_workflow(agent_id: str = "") -> Dict[str, Any]: """构建 agent 节点工作流（参考苏瑶3号架构：开始 -> agent(ReAct) -> 结束）。""" return { "nodes": [ { "id": "start-1", "type": "start", "position": {"x": 80, "y": 240}, "data": {"label": "学习任务开始"}, }, { "id": "agent-learning-core", "type": "agent", "position": {"x": 380, "y": 240}, "data": { "label": AGENT_NAME, # LLM config "system_prompt": SYSTEM_PROMPT, "provider": PROVIDER, "model": MODEL, "temperature": TEMPERATURE, "max_iterations": MAX_ITERATIONS, # All 39 tools "tools": ALL_TOOLS, # KG+RAG memory config "memory": True, "memory_max_history": 40, "memory_vector_enabled": True, "memory_vector_top_k": 10, "memory_persist": True, "memory_learning": True, # Scope isolation "memory_scope_id": agent_id, "agent_id": agent_id, # Self-review "self_review_enabled": True, }, }, { "id": "end-1", "type": "end", "position": {"x": 680, "y": 240}, "data": {"label": "学习完成"}, }, ], "edges": [ { "id": "e_start_agent", "source": "start-1", "target": "agent-learning-core", "sourceHandle": "right", "targetHandle": "left", }, { "id": "e_agent_end", "source": "agent-learning-core", "target": "end-1", "sourceHandle": "right", "targetHandle": "left", }, ], } def main() -> int: token = _login() if not token: return 1 h = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} # Check if agent exists existing = _find_agent_by_name(token, AGENT_NAME) if existing: print(f"Agent '{AGENT_NAME}' exists (id={existing['id']}), updating config...") agent_id = existing["id"] else: temp_wf = _build_workflow(agent_id="") body = { "name": AGENT_NAME, "description": "AI学习助手 -- KG+RAG理想版，实体关系图谱+语义向量检索+情境感知，具备全部39工具。", "workflow_config": temp_wf, } r = requests.post(f"{BASE}/api/v1/agents", headers=h, json=body, timeout=60) if r.status_code != 201: print(f"Create failed: {r.status_code} {r.text[:500]}", file=sys.stderr) return 1 agent_id = r.json()["id"] print(f"Agent created: id={agent_id} name={AGENT_NAME}") # Rebuild workflow with correct agent_id wf = _build_workflow(agent_id=agent_id) description = ( f"AI学习助手 -- 知识图谱+RAG理想版（参考苏瑶3号架构）。\n\n" f"记忆架构：三层记忆体系（最接近人类记忆方式）\n" f" [Layer 1] 知识图谱 -- 实体关系图谱，动态演化，情境编码\n" f" [Layer 2] 向量语义 -- Embedding检索，模糊联想，知识迁移\n" f" [Layer 3] 长期情景 -- 跨会话持久化，用户画像，学习里程碑\n\n" f"工作流：开始 -> agent(ReAct) -> 结束\n" f"工具：全部 {len(ALL_TOOLS)} 个内置工具\n" f"模型：{PROVIDER}/{MODEL} temperature={TEMPERATURE} max_iterations={MAX_ITERATIONS}\n" f"记忆：KG+RAG 三层记忆 + 向量Top-10 + 长期 + 自主学习" ) up = requests.put( f"{BASE}/api/v1/agents/{agent_id}", headers=h, json={ "description": description, "workflow_config": wf, }, timeout=120, ) if up.status_code != 200: print(f"Update failed: {up.status_code} {up.text[:500]}", file=sys.stderr) return 1 print(f"[OK] Agent '{AGENT_NAME}' configured (KG+RAG ideal version)") print(f" ID: {agent_id}") print(f" Type: agent node (AgentRuntime ReAct)") print(f" Model: {PROVIDER}/{MODEL} (temperature={TEMPERATURE})") print(f" Max iterations: {MAX_ITERATIONS}") print(f" Tools: ALL {len(ALL_TOOLS)} built-in tools") print(f" Memory: 3-layer KG+RAG (entity graph + vector semantic + persistent)") print(f" Vector top-K: 10, Max history: 40") print(f" Self-review: enabled") # Publish if requested if PUBLISH: pub = requests.put( f"{BASE}/api/v1/agents/{agent_id}", headers=h, json={"status": "published"}, timeout=30, ) if pub.status_code == 200: print(f" Status: published") else: print(f" Publish failed: {pub.status_code} {pub.text[:200]}") print() print("Five core modules (KG+RAG enhanced):") print(" [1] Homework management -- KG-driven task splitting & scheduling") print(" [2] Learning assistance -- dual retrieval (graph + vector semantic)") print(" [3] Time & planning -- context-aware scheduling via learning_path") print(" [4] Resource recommendation -- web search + knowledge gap analysis") print(" [5] Motivation & feedback -- achievement tracking + KG growth visualization") print(" [6] Autonomous evolution -- agent_create, tool_register, code_tool_create") print() print(json.dumps({"id": agent_id, "name": AGENT_NAME, "published": PUBLISH}, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())