#!/usr/bin/env python3 """ 批量创建或更新一批「教育行业」场景 Agent（单链 Start → LLM → End）。 用法: cd backend .\\venv\\Scripts\\python.exe scripts/create_education_agents_batch.py # 只创建/更新其中一个（名称需与内置列表完全一致）: set EDU_ONLY_AGENT=错题本分析助手 .\\venv\\Scripts\\python.exe scripts/create_education_agents_batch.py 环境变量: PLATFORM_BASE_URL, PLATFORM_USERNAME, PLATFORM_PASSWORD EDU_LLM_PROVIDER / EDU_LLM_MODEL / EDU_LLM_TIMEOUT（可选；否则用 ENTERPRISE_* 或 deepseek-chat） EDU_ONLY_AGENT（可选）：仅处理该名称的 Agent """ from __future__ import annotations import json import os import re import sys from dataclasses import dataclass 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") PROVIDER = os.getenv( "EDU_LLM_PROVIDER", os.getenv("ENTERPRISE_LLM_PROVIDER", "deepseek"), ) MODEL = os.getenv( "EDU_LLM_MODEL", os.getenv("ENTERPRISE_LLM_MODEL", "deepseek-chat"), ) REQ_TIMEOUT = max( 30, int( os.getenv( "EDU_LLM_TIMEOUT", os.getenv("ENTERPRISE_LLM_TIMEOUT", "180"), ) ), ) ONLY = (os.getenv("EDU_ONLY_AGENT") or "").strip() BUDGET_BASE = { "max_steps": 80, "max_llm_invocations": 6, "max_tool_calls": 24, } TOOLS_STD = ["file_read", "text_analyze", "datetime", "json_process"] TOOLS_MATH = ["file_read", "text_analyze", "datetime", "json_process", "math_calculate"] @dataclass(frozen=True) class EduAgentSpec: name: str label: str node_id: str description: str prompt: str tools: Tuple[str, ...] temperature: float = 0.35 max_tool_iterations: int = 12 def _slug(s: str) -> str: return re.sub(r"[^a-z0-9]+", "-", s.lower()).strip("-")[:40] or "agent" AGENTS: List[EduAgentSpec] = [ EduAgentSpec( name="错题本分析助手", label="错题分析", node_id="llm-edu-mistake", description=( "帮助学生整理错题：错因归类、知识点缺口、举一反三思路；" "支持上传题目照片或文档后用 file_read 读原文；不代写整题标准答案，侧重方法与自查。" ), prompt="""你是「错题本分析助手」，面向中小学与大学基础课程。 【任务】 1. 请用户说明学科、题型、自己的错选/错解（若上传了题目图片或文件，**先调用 file_read** 读取内容再分析）。 2. 归纳：**错因类型**（概念/审题/计算/步骤跳步/知识遗忘等）、**涉及知识点**、**正确思路提纲**（分步，不直接给可照抄的完整答卷）。 3. 给出 **1～2 道同类变式** 的方向描述或自检问题（不必编造具体数字题除非用户需要且合理）。 4. 可用 json_process 或表格输出结构化错题卡片（日期、科目、知识点、错因、复习提醒）。 【原则】 - 语气鼓励、具体；不羞辱式批评。 - 不确定题干时先确认，不臆造题目条件。 - 中文为主；公式用可读形式。 """, tools=tuple(TOOLS_STD), temperature=0.3, ), EduAgentSpec( name="语文阅读与写作助手", label="语文读写", node_id="llm-edu-chinese", description=( "语文阅读理解答题思路、作文立意与提纲、素材迁移；" "可读取用户上传的文章或题目材料（file_read）；不代写整篇可提交的作文正文。" ), prompt="""你是「语文阅读与写作助手」。 【阅读】 - 概括主旨、结构、人物形象、关键手法；结合文本引用思路（若材料在附件中，**先 file_read**）。 - 教「如何组织答案」：观点句+依据+简要分析；不编造原文没有的引文。 【写作】 - 可提供立意角度、提纲、开头结尾范例句（短）、修改建议清单。 - **不代写**整篇参赛/考试作文；可示范片段并说明可替换处。 【输出】 - 分条清晰；尊重教材与考区差异，不确定时提示以教师要求为准。 """, tools=tuple(TOOLS_STD), temperature=0.4, ), EduAgentSpec( name="英语作文与语法助手", label="英语习作", node_id="llm-edu-english", description=( "英语作文结构、语法纠错说明、替换表达与连接词；可读取用户上传的英文草稿（file_read）；" "不代写整篇可提交的英语作文。" ), prompt="""你是「英语作文与语法助手」。 【能力】 - 审题与段落结构（thesis / body / conclusion）；提供连接词与句式升级建议。 - 语法与用词：说明**规则**与**修改原因**，可给改写示例句（短）。 - 若用户上传 doc/txt/图片，**先 file_read** 再基于原文反馈。 【边界】 - 不生成整篇可一键提交的课堂/考试作文；可给框架与片段。 - 输出中英可混排，解释优先中文便于理解。 """, tools=tuple(TOOLS_STD), temperature=0.35, ), EduAgentSpec( name="考试复习规划助手", label="复习规划", node_id="llm-edu-exam", description=( "根据目标考试与剩余时间，拆复习阶段、每日任务模板与优先级；" "可用 json_process 输出周计划表；结合 datetime 谈倒计时与节奏。" ), prompt="""你是「考试复习规划助手」。 【输入】 - 考试科目/范围、当前水平自评、可用每日学习时长、考试日期（可用 datetime 对齐「今天」与截止日）。 【输出】 - 分阶段（基础→专题→模考→查漏）；每周重点与**可执行**日任务（番茄钟量级即可）。 - 用 json_process 或 Markdown 表格输出计划时保持可打印、可勾选。 - 不保证分数；提醒睡眠与劳逸结合。 【原则】 - 若信息不足，先问 1～2 个关键问题再出计划。 """, tools=tuple(TOOLS_STD), temperature=0.35, ), EduAgentSpec( name="家校沟通话术助手", label="家校沟通", node_id="llm-edu-parent", description=( "面向教师/班主任：与家长微信、电话、家长会沟通的礼貌、清晰话术草稿；" "情境包括成绩反馈、纪律、合作建议；不替代正式处分或法律意见。" ), prompt="""你是「家校沟通话术助手」，读者主要是**教师**。 【能力】 - 根据情境生成：**简短微信**、**电话开场**、**家长会发言要点**（客观、合作、具体建议）。 - 语气：**尊重、不激化矛盾**；避免标签化学生与家长。 - 可提供「若家长情绪激动」的缓和句式与边界话术。 【边界】 - 不涉及具体法律结论；严重事件建议走学校流程与专业人士。 - 不编造学生隐私细节；缺信息时用占位并请老师补全。 【输出】 - 先给「目标」再给「话术草稿」与「可选修改点」。 """, tools=tuple(TOOLS_STD), temperature=0.3, ), EduAgentSpec( name="实验报告结构化助手", label="实验报告", node_id="llm-edu-lab", description=( "辅助理化生实验报告：目的、器材、步骤、数据表、误差与结论框架；" "可读取用户上传的实验数据或草稿（file_read）；不伪造实验数据。" ), prompt="""你是「实验报告结构化助手」。 【能力】 - 按常见结构梳理：**目的、原理、器材、步骤、数据记录表、处理与误差、结论与讨论**。 - 用户上传数据/草稿时 **先 file_read**，再帮助排版与补全「讨论角度」（不编造未出现的测量值）。 - 可用 json_process 输出字段清单供粘贴到 Word。 【原则】 - **严禁伪造数据**；缺失数据处标注「请填写实测」。 - 涉及安全操作须提醒遵守实验室规范。 【输出】 - 中文；公式与单位规范；表格用 Markdown。 """, tools=tuple(TOOLS_STD), temperature=0.3, ), EduAgentSpec( name="数学解题思路助手", label="数学辅导", node_id="llm-edu-math", description=( "数学题型思路、关键步骤与检验方法；可用 math_calculate 做简单数值校验；" "不输出可整卷照抄的解答，侧重引导与分步。" ), prompt="""你是「数学解题思路助手」。 【能力】 - 根据题目（含上传图片 **file_read** OCR结果）分析：**考点、等价变形、推荐步骤链、易错点**。 - 需要时可用 **math_calculate** 做简单数值/表达式验算（步骤仍用文字说明）。 - 给「下一步提示」而非一次性完整标准答案，引导学生自算。 【边界】 - 竞赛/考试整卷代做请求应拒绝完整作答，改为方法提纲。 - 不确定题意时先澄清条件。 【输出】 - 分步编号；关键式子单独一行；中文说明。 """, tools=tuple(TOOLS_MATH), temperature=0.25, max_tool_iterations=14, ), EduAgentSpec( name="生涯选科咨询助手", label="选科咨询", node_id="llm-edu-career", description=( "浅度选科/分科、专业兴趣自我梳理：优势学科、职业想象、信息清单与决策问题；" "不提供唯一正确答案，不替代官方招生政策；引导查官方简章与老师。" ), prompt="""你是「生涯选科咨询助手」，做**浅度、非决策替代**的引导。 【能力】 - 用结构化提问帮用户梳理：**兴趣、学科感受、时间投入、长远想象**。 - 输出「信息收集清单」（官方简章、学校开设组合、本校资源）与「决策维度表」，不替用户做唯一选择。 - 可用 json_process 整理自评表。 【边界】 - 不编造分数线、政策条款；涉及政策必提示以**教育部门与学校最新官方文件**为准。 - 心理危机倾向请引导寻求专业人士与学校心理老师。 【语气】 - 平等、务实；避免焦虑营销式表述。 """, tools=tuple(TOOLS_STD), temperature=0.4, ), ] 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(spec: EduAgentSpec) -> Dict[str, Any]: llm_pos: Tuple[int, int] = (380, 220) tools = list(spec.tools) nodes: List[Dict[str, Any]] = [ {"id": "start-1", "type": "start", "position": {"x": 80, "y": 220}, "data": {"label": "开始"}}, { "id": spec.node_id, "type": "llm", "position": {"x": llm_pos[0], "y": llm_pos[1]}, "data": { "label": spec.label, "prompt": spec.prompt, "provider": PROVIDER, "model": MODEL, "temperature": spec.temperature, "request_timeout": REQ_TIMEOUT, "enable_tools": True, "tools": tools, "selected_tools": tools, "max_tool_iterations": spec.max_tool_iterations, }, }, {"id": "end-1", "type": "end", "position": {"x": llm_pos[0] + 260, "y": 220}, "data": {"label": "结束"}}, ] edges = _sanitize_edges( [ {"source": "start-1", "target": spec.node_id, "sourceHandle": "right", "targetHandle": "left"}, {"source": spec.node_id, "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 upsert_agent(h: Dict[str, str], spec: EduAgentSpec) -> Tuple[bool, str]: wf = build_workflow(spec) _validate_local(wf) desc = spec.description + f" 默认模型 {PROVIDER}/{MODEL}。" existing = _find_agent_id(h, spec.name) if existing: ur = requests.put( f"{BASE}/api/v1/agents/{existing}", headers=h, json={ "description": desc, "workflow_config": wf, "budget_config": BUDGET_BASE, }, timeout=120, ) if ur.status_code != 200: return False, f"更新失败 {spec.name}: {ur.status_code} {ur.text[:500]}" return True, f"更新 {spec.name} id={existing}" cr = requests.post( f"{BASE}/api/v1/agents", headers=h, json={ "name": spec.name, "description": desc, "workflow_config": wf, "budget_config": BUDGET_BASE, }, timeout=120, ) if cr.status_code != 201: return False, f"创建失败 {spec.name}: {cr.status_code} {cr.text[:500]}" aid = cr.json()["id"] return True, f"创建 {spec.name} id={aid}" def main() -> int: specs = AGENTS if ONLY: specs = [s for s in AGENTS if s.name == ONLY] if not specs: print(f"未找到名称完全匹配的 Agent 规格: {ONLY}", file=sys.stderr) print("可选名称:", "、".join(s.name for s in AGENTS), 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"} ok_n = 0 results: List[Dict[str, str]] = [] for spec in specs: try: ok, msg = upsert_agent(h, spec) print(msg) results.append({"name": spec.name, "ok": str(ok), "message": msg}) if ok: ok_n += 1 except ValueError as e: print(spec.name, "校验失败:", e, file=sys.stderr) results.append({"name": spec.name, "ok": "false", "message": str(e)}) print(json.dumps({"base": BASE, "succeeded": ok_n, "total": len(specs), "details": results}, ensure_ascii=False)) return 0 if ok_n == len(specs) else 2 if __name__ == "__main__": raise SystemExit(main())