aiagent/backend/scripts/create_intelligent_tutor_agent.py

#!/usr/bin/env python3
"""
创建或更新「智能助教」Agent：单链 Start → LLM → End，面向课程答疑、作业辅导与学习规划。

用法:
  cd backend && .\\venv\\Scripts\\python.exe scripts/create_intelligent_tutor_agent.py

环境变量:
  PLATFORM_BASE_URL, PLATFORM_USERNAME, PLATFORM_PASSWORD
  AGENT_NAME（默认 智能助教）
  TUTOR_LLM_PROVIDER / TUTOR_LLM_MODEL / TUTOR_LLM_TIMEOUT（可选，覆盖默认 DeepSeek 与超时秒数）
"""
from __future__ import annotations

import json
import os
import sys
from typing import Any, Dict, List, Optional, Tuple

import requests

BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if BACKEND_DIR not in sys.path:
    sys.path.insert(0, BACKEND_DIR)

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", "智能助教")

PROVIDER = os.getenv("TUTOR_LLM_PROVIDER", os.getenv("ENTERPRISE_LLM_PROVIDER", "deepseek"))
MODEL = os.getenv("TUTOR_LLM_MODEL", os.getenv("ENTERPRISE_LLM_MODEL", "deepseek-chat"))
REQ_TIMEOUT = max(30, int(os.getenv("TUTOR_LLM_TIMEOUT", os.getenv("ENTERPRISE_LLM_TIMEOUT", "180"))))

BUDGET_CONFIG = {
    "max_steps": 80,
    "max_llm_invocations": 6,
    "max_tool_calls": 24,
}

TUTOR_TOOLS = ["file_read", "text_analyze", "datetime", "json_process"]

TUTOR_PROMPT = """你是「智能助教」，面向高校/职业课程场景辅助学习与教学准备。

【能力】
- 概念讲解：用清晰结构（定义→要点→小例子）说明知识点。
- 习题辅导：给出**解题思路与关键步骤**，引导学生自己完成计算与证明；不要直接给出可照抄的整卷答案或替考内容。
- 学习规划：根据用户目标与可用时间，建议复习顺序与自检清单。
- 材料辅助：若用户提到本地课件/笔记路径，可用工具读取后基于原文摘要与答疑。

【边界】
- 不编造教材页码、不虚构课程政策；不确定时明确说明并建议向任课教师核实。
- 涉及实验安全、医疗、法律等高风险领域时提示寻求专业人士。
- 输出简洁，优先中文；需要公式时用 LaTeX 或纯文本均可读形式。

【输出】
- 先给结论或步骤概览，再展开细节；复杂问题分条编号。
"""


def _sanitize_edges(edges: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    seen: set = set()
    out: List[Dict[str, Any]] = []
    for e in edges or []:
        s, t = e.get("source"), e.get("target")
        if not s or not t or s == t:
            continue
        key = (s, t, e.get("sourceHandle") or "")
        if key in seen:
            continue
        seen.add(key)
        ne = dict(e)
        if not ne.get("targetHandle"):
            ne["targetHandle"] = "left"
        if not ne.get("id"):
            sh = ne.get("sourceHandle") or "r"
            ne["id"] = f"e_{s}_{t}_{sh}"
        out.append(ne)
    return out


def build_workflow() -> Dict[str, Any]:
    llm_pos: Tuple[int, int] = (380, 220)
    nodes: List[Dict[str, Any]] = [
        {"id": "start-1", "type": "start", "position": {"x": 80, "y": 220}, "data": {"label": "开始"}},
        {
            "id": "llm-tutor",
            "type": "llm",
            "position": {"x": llm_pos[0], "y": llm_pos[1]},
            "data": {
                "label": "智能助教",
                "prompt": TUTOR_PROMPT,
                "provider": PROVIDER,
                "model": MODEL,
                "temperature": 0.35,
                "request_timeout": REQ_TIMEOUT,
                "enable_tools": True,
                "tools": list(TUTOR_TOOLS),
                "selected_tools": list(TUTOR_TOOLS),
                "max_tool_iterations": 12,
            },
        },
        {"id": "end-1", "type": "end", "position": {"x": llm_pos[0] + 260, "y": 220}, "data": {"label": "结束"}},
    ]
    edges = _sanitize_edges(
        [
            {"source": "start-1", "target": "llm-tutor", "sourceHandle": "right", "targetHandle": "left"},
            {"source": "llm-tutor", "target": "end-1", "sourceHandle": "right", "targetHandle": "left"},
        ]
    )
    return {"nodes": nodes, "edges": edges}


def _validate_local(wf: Dict[str, Any]) -> None:
    from app.services.workflow_validator import validate_workflow

    r = validate_workflow(wf.get("nodes") or [], wf.get("edges") or [])
    if not r.get("valid"):
        errs = r.get("errors") or []
        raise ValueError("工作流校验失败: " + "; ".join(errs))


def _find_agent_id(h: Dict[str, str], name: str) -> Optional[str]:
    r = requests.get(f"{BASE}/api/v1/agents", params={"search": name, "limit": 80}, headers=h, timeout=45)
    if r.status_code != 200:
        return None
    for a in r.json() or []:
        if a.get("name") == name:
            return a.get("id")
    return None


def main() -> int:
    wf = build_workflow()
    try:
        _validate_local(wf)
    except ValueError as e:
        print(e, file=sys.stderr)
        return 1

    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:
        print("登录失败:", r.status_code, r.text[:500], file=sys.stderr)
        return 1
    token = r.json().get("access_token")
    if not token:
        print("无 access_token", file=sys.stderr)
        return 1
    h = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

    desc = (
        "智能助教：课程答疑、习题思路辅导与学习规划；支持读取本地材料（file_read）与文本分析；"
        f"默认模型 {PROVIDER}/{MODEL}，单次执行内工具迭代上限 12。"
    )

    existing = _find_agent_id(h, AGENT_NAME)
    if existing:
        ur = requests.put(
            f"{BASE}/api/v1/agents/{existing}",
            headers=h,
            json={
                "description": desc,
                "workflow_config": wf,
                "budget_config": BUDGET_CONFIG,
            },
            timeout=120,
        )
        if ur.status_code != 200:
            print("更新失败:", ur.status_code, ur.text[:800], file=sys.stderr)
            return 1
        print("已更新", AGENT_NAME, existing)
        print(json.dumps({"id": existing, "name": AGENT_NAME}, ensure_ascii=False))
        return 0

    cr = requests.post(
        f"{BASE}/api/v1/agents",
        headers=h,
        json={
            "name": AGENT_NAME,
            "description": desc,
            "workflow_config": wf,
            "budget_config": BUDGET_CONFIG,
        },
        timeout=120,
    )
    if cr.status_code != 201:
        print("创建失败:", cr.status_code, cr.text[:800], file=sys.stderr)
        return 1
    aid = cr.json()["id"]
    print("已创建", AGENT_NAME, aid)
    print(json.dumps({"id": aid, "name": AGENT_NAME}, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())