#!/usr/bin/env python3 """ 创建或更新「学生作业管理助手」Agent：单链 Start → LLM → End。 侧重：记录作业项、截止日、优先级；跟进完成情况；温和督促与周回顾（不代写可提交的作业正文）。 用法: cd backend && .\\venv\\Scripts\\python.exe scripts/create_homework_manager_agent.py 环境变量: PLATFORM_BASE_URL, PLATFORM_USERNAME, PLATFORM_PASSWORD AGENT_NAME（默认 学生作业管理助手） HOMEWORK_LLM_PROVIDER / HOMEWORK_LLM_MODEL / HOMEWORK_LLM_TIMEOUT（可选） """ 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( "HOMEWORK_LLM_PROVIDER", os.getenv("ENTERPRISE_LLM_PROVIDER", "deepseek") ) MODEL = os.getenv( "HOMEWORK_LLM_MODEL", os.getenv("ENTERPRISE_LLM_MODEL", "deepseek-chat") ) REQ_TIMEOUT = max( 30, int( os.getenv( "HOMEWORK_LLM_TIMEOUT", os.getenv("ENTERPRISE_LLM_TIMEOUT", "180") ) ), ) BUDGET_CONFIG = { "max_steps": 80, "max_llm_invocations": 6, "max_tool_calls": 20, } HOMEWORK_TOOLS = ["file_read", "text_analyze", "datetime", "json_process"] HOMEWORK_PROMPT = """你是「学生作业管理助手」，帮助学生**记作业**与**监督完成**，语气友好、具体、可执行。 【核心能力】 1. **记作业**：从用户自然语言中提取「科目 / 作业内容 / 截止日期与时间 / 老师要求要点 / 预估耗时」，整理成清单。 - 若用户用回形针**上传**了文件或照片，消息里会出现「相对工作区根路径」列表：**必须先调用 file_read**，用返回的 `content`（正文/OCR 文本）整理进作业清单，勿编造未读到的内容。 - 支持常见格式：纯文本/Markdown、**PDF**、**Word(.docx)**、**Excel(.xlsx)**、**照片**（作业拍照等，依赖 OCR；若工具返回需安装 Tesseract 等提示，请如实转告用户并仍可基于用户口述继续记作业）。 2. **监督完成**：根据清单追问进度（未开始/进行中/已完成）；对临近截止的任务给**温和提醒**（不制造焦虑）；可建议拆成小步骤与每日 15–30 分钟微习惯。 3. **周回顾**：用户要求时，用 json_process 或清晰表格输出本周完成率、延期项与下周优先三件事。 【原则】 - **不代写**可提交的作业正文、实验报告、论文等；可提供提纲、自检表、引用规范提示。 - 日期时间以用户所在语境为准；需要当前时间可借助工具 datetime。 - 不确定的信息（如具体截止时刻）先列出假设并请用户确认。 - 输出优先中文；列表用编号，便于复制到备忘录。 【交互习惯】 - 用户只说「记一下数学作业」时，主动追问截止日与具体要求（一次问 1–2 个点，避免审问感）。 - 用户汇报「做完了」时，确认是否需拍照/上传检查清单，并建议归档到下一条任务前的小结一句话。 """ 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-homework", "type": "llm", "position": {"x": llm_pos[0], "y": llm_pos[1]}, "data": { "label": "作业管理", "prompt": HOMEWORK_PROMPT, "provider": PROVIDER, "model": MODEL, "temperature": 0.3, "request_timeout": REQ_TIMEOUT, "enable_tools": True, "tools": list(HOMEWORK_TOOLS), "selected_tools": list(HOMEWORK_TOOLS), "max_tool_iterations": 10, }, }, {"id": "end-1", "type": "end", "position": {"x": llm_pos[0] + 260, "y": 220}, "data": {"label": "结束"}}, ] edges = _sanitize_edges( [ {"source": "start-1", "target": "llm-homework", "sourceHandle": "right", "targetHandle": "left"}, {"source": "llm-homework", "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 提取正文（文本、PDF、docx、xlsx、图片 OCR）与 json_process 整理；" f"默认模型 {PROVIDER}/{MODEL}，单次执行内工具迭代上限 10。" ) 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())