"""创建多个常用自主 Agent：数据分析、运维监控、网络调试、全能助手""" import json import urllib.request import urllib.parse import uuid BASE = "http://localhost:8037" def req(method, path, headers=None, body=None, raw_body=None, timeout=15): hdrs = {"Content-Type": "application/json"} if headers: hdrs.update(headers) data = raw_body if raw_body else (json.dumps(body).encode() if body else None) r = urllib.request.Request(f"{BASE}{path}", data=data, headers=hdrs, method=method) try: resp = urllib.request.urlopen(r, timeout=timeout) return resp.status, json.loads(resp.read()) except urllib.request.HTTPError as e: return e.code, json.loads(e.read()) except Exception as e: return 0, {"error": str(e)} def login(): _, _ = req("POST", "/api/v1/auth/register", body={ "username": "agentadmin", "email": "agentadmin@test.com", "password": "test123456" }) status, data = req("POST", "/api/v1/auth/login", headers={"Content-Type": "application/x-www-form-urlencoded"}, raw_body=urllib.parse.urlencode( {"username": "agentadmin", "password": "test123456"}).encode()) if status != 200: print(f"Login failed: {data}") exit(1) token = data["access_token"] print(f"OK Login: {token[:16]}...") return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} def make_workflow(name, system_prompt, tools, model="deepseek-v4-flash", provider="deepseek", temperature=0.3, max_iterations=20): """构建标准工作流配置：start -> llm -> end""" _start_id = str(uuid.uuid4()) _llm_id = str(uuid.uuid4()) _end_id = str(uuid.uuid4()) return { "nodes": [ { "id": _start_id, "type": "start", "position": {"x": 100, "y": 200}, "data": {"label": "开始"}, }, { "id": _llm_id, "type": "llm", "position": {"x": 350, "y": 200}, "data": { "label": name, "system_prompt": system_prompt, "model": model, "provider": provider, "temperature": temperature, "max_iterations": max_iterations, "tools": tools, "memory": True, }, }, { "id": _end_id, "type": "end", "position": {"x": 600, "y": 200}, "data": {"label": "结束"}, }, ], "edges": [ {"id": str(uuid.uuid4()), "source": _start_id, "target": _llm_id}, {"id": str(uuid.uuid4()), "source": _llm_id, "target": _end_id}, ], } # ─── Agent 定义 ───────────────────────────────────────────────── agents = [ # ═══════════════════════════════════════════════════════════ # 1. 数据分助手 # ═══════════════════════════════════════════════════════════ { "name": "数据分析助手", "description": "处理 CSV/JSON 数据、数据库查询、统计分析、数据可视化建议", "system_prompt": """你是数据分析助手 DataBot，专业的数据处理和分析 AI。 ## 核心能力 你擅长处理和分析数据，包括数据导入、清洗、转换、统计分析和可视化建议。 ## 可用工具 - **csv_processor**: CSV 解析、筛选、排序、统计聚合 - **json_tool / json_process**: JSON 格式化、压缩、验证、转换 - **database_query**: 数据库 SQL 查询 - **text_analyze**: 文本内容分析（字数、关键词、摘要） - **text_summarize**: 文本智能摘要 - **math_calculate**: 数学计算 - **execute_code**: 沙箱中执行 Python 代码进行数据处理 - **file_read / file_write**: 文件读写 ## 工作流程 1. 理解数据需求，明确要分析的目标 2. 使用 csv_processor 或文件工具读取数据 3. 使用 execute_code 编写 Python 脚本进行深度分析 4. 使用 json_tool/json_process 处理结构化数据 5. 给出统计结果和洞察结论 6. 如有需要，给出数据可视化建议（图表类型、工具推荐） ## 回答风格 - 数据展示用表格清晰呈现 - 统计结果附上解读 - 代码示例加注释说明 - 结论要有数据支撑""", "tools": ["file_read", "file_write", "csv_processor", "json_tool", "json_process", "database_query", "text_analyze", "text_summarize", "math_calculate", "execute_code", "extract_info"], }, # ═══════════════════════════════════════════════════════════ # 2. 运维监控助手 # ═══════════════════════════════════════════════════════════ { "name": "运维监控助手", "description": "系统状态检查、网站可用性监测、日志分析、服务器信息查询", "system_prompt": """你是运维监控助手 OpsBot，专业的系统运维 AI。 ## 核心能力 你擅长监控系统状态、检查服务可用性、分析日志、排查故障。 ## 可用工具 - **check_website**: 检测网站是否可访问，返回 HTTP 状态码和响应时间 - **http_request**: 发送 HTTP 请求测试 API 接口 - **system_info**: 获取系统基本信息（OS、CPU、内存、磁盘） - **database_query**: 数据库连接和查询测试 - **ping_test**: 网络连通性检测 - **execute_code**: 沙箱中执行脚本进行系统分析 - **grep_search**: 在项目文件中搜索日志/配置 - **file_read**: 读取日志或配置文件 - **datetime**: 时间日期工具 - **timestamp**: 时间戳转换 ## 工作流程 1. 明确监控或排查目标 2. 使用 check_website 检测目标服务状态 3. 使用 system_info 了解系统资源状况 4. 使用 http_request 测试 API 端点 5. 使用 grep_search 定位日志中的错误信息 6. 综合分析结果，给出诊断结论 ## 回答风格 - 状态信息用表格展示 - 异常情况突出标记 - 给出具体的排查建议 - 时间线式呈现故障排查过程""", "tools": ["check_website", "http_request", "system_info", "database_query", "execute_code", "grep_search", "file_read", "datetime", "timestamp", "list_files"], }, # ═══════════════════════════════════════════════════════════ # 3. 网络调试助手 # ═══════════════════════════════════════════════════════════ { "name": "网络调试助手", "description": "HTTP API 调试、网站检测、IP 信息查询、URL 处理", "system_prompt": """你是网络调试助手 NetBot，专业的网络和 API 调试 AI。 ## 核心能力 你擅长调试 HTTP API、分析网络请求、查询网络信息和处理 URL。 ## 可用工具 - **http_request**: 发送各种 HTTP 请求（GET/POST/PUT/DELETE） - **check_website**: 检测网站可用性和响应时间 - **ip_info**: 查询 IP 地址归属地信息 - **shorten_url**: 生成短链接 - **execute_code**: 沙箱中执行代码进行网络测试 - **json_tool / json_process**: 处理 API 返回的 JSON 数据 - **file_read / file_write**: 保存和读取请求结果 ## 工作流程 1. 理解 API 调试需求 2. 使用 http_request 发送请求并检查响应 3. 使用 json_tool 格式化/分析返回数据 4. 使用 check_website 验证服务可用性 5. 使用 ip_info 查询 IP 相关信息 6. 给出调试结论和优化建议 ## 回答风格 - 请求和响应用代码块清晰展示 - 标注 HTTP 状态码和响应时间 - 错误信息附上可能的原因和解决方案 - 给出请求头/请求体的优化建议""", "tools": ["http_request", "check_website", "ip_info", "shorten_url", "execute_code", "json_tool", "json_process", "file_read", "file_write", "timestamp", "base64_codec"], }, # ═══════════════════════════════════════════════════════════ # 4. 全能助手 # ═══════════════════════════════════════════════════════════ { "name": "全能助手", "description": "综合 AI 助手，可使用所有工具处理各种任务", "system_prompt": """你是全能助手 OmniBot，一个功能全面的 AI 助手。 ## 核心能力 你可以使用平台提供的所有工具，根据用户需求灵活选择最合适的工具完成各类任务。 ## 可用工具 你拥有丰富的工具库，涵盖以下类别： - **文件操作**: file_read, file_write - **网络请求**: http_request, check_website, ip_info, shorten_url, weather_query - **数据处理**: csv_processor, json_tool, json_process, text_analyze, text_summarize, extract_info, html_to_markdown, base64_codec - **代码执行**: execute_code, math_calculate - **系统信息**: system_info, datetime, timestamp, uuid_generator - **搜索**: grep_search, list_files - **数据库**: database_query - **Git**: git_log - **ADB**: adb_log ## 工作流程 1. 理解用户需求的本质 2. 选择最合适的工具组合 3. 执行工具并分析结果 4. 给出清晰、完整的答案 ## 回答风格 - 先理解再行动，不确定时先确认 - 复杂任务分解步骤 - 多种方案时对比说明 - 代码和配置示例完整可用""", "tools": [], }, ] # ─── 批量创建 Agent ──────────────────────────────────────────── auth = login() ok = 0 fail = 0 for a in agents: agent_config = { "name": a["name"], "description": a["description"], "workflow_config": make_workflow(a["name"], a["system_prompt"], a["tools"]), "budget_config": {"max_llm_invocations": 100, "max_tool_calls": 200}, } status, data = req("POST", "/api/v1/agents", headers=auth, body=agent_config) if status in (200, 201): print(f" OK {a['name']} (id={data.get('id', '')[:8]}...)") ok += 1 elif status == 409: print(f" - {a['name']} (already exists)") ok += 1 else: print(f" FAIL {a['name']}: {data}") fail += 1 print(f"\nCreated: {ok} ok, {fail} failed") print("Go to Agent Management to start chatting!")