aiagent/backend/scripts/create_ai_learning_assistant.py

#!/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())