""" 数字员工工厂 — 完整使用演示 用法: python scripts/demo_digital_employee.py 流程: 1. 登录获取 Token 2. 创建 Main Agent（数字员工 PM） 3. 升级为 Main Agent（预置系统提示词 + 工具） 4. 创建目标（Goal） 5. 分解目标为子任务（decompose） 6. 异步启动执行 7. 查看任务树和进度 """ import urllib.request import json import time import sys BASE = "http://localhost:8037" # ── 登录 ── def login(): req = urllib.request.Request( f"{BASE}/api/v1/auth/login", data=b"username=admin&password=123456", headers={"Content-Type": "application/x-www-form-urlencoded"}, ) return json.loads(urllib.request.urlopen(req).read())["access_token"] # ── API 助手 ── def api(method, path, data=None, token=""): url = f"{BASE}{path}" body = json.dumps(data).encode("utf-8") if data else None req = urllib.request.Request(url, data=body, method=method, headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}) try: r = urllib.request.urlopen(req, timeout=30) if r.status == 204: return None return json.loads(r.read()) except urllib.error.HTTPError as e: try: return json.loads(e.read()) except: return {"detail": str(e), "status": e.code} # ── 主流程 ── def main(): token = login() print("=" * 60) print(" 数字员工工厂 — 完整使用演示") print("=" * 60) # ── Step 1: 创建 Main Agent ── print("\n[1] 创建 Main Agent（数字员工 PM）...") agent = api("POST", "/api/v1/agents", token=token, data={ "name": "Demo 数字员工PM", "description": "我是数字员工项目经理，负责接收目标、分解任务、调度执行、追踪进度", "workflow_config": { "nodes": [ {"id": "start-1", "data": {"label": "开始"}, "type": "start", "position": {"x": 80, "y": 200}}, {"id": "agent-1", "data": { "label": "项目经理", "system_prompt": ( "你是一个数字员工项目经理。当用户提出目标时：\n" "1. 理解目标的核心意图\n" "2. 将目标分解为 2-5 个具体子任务\n" "3. 调用 create_task 创建子任务\n" "4. 调用 assign_task 分配 Agent\n" "5. 调用 check_progress 追踪进度\n" "6. 遇到问题调用 notify_user 通知用户" ), "model": "deepseek-v4-flash", "provider": "deepseek", "temperature": 0.7, "max_iterations": 15, "tools": ["web_search", "knowledge_graph_search"], "memory": True, }, "type": "agent", "position": {"x": 350, "y": 200}}, {"id": "end-1", "data": {"label": "结束"}, "type": "end", "position": {"x": 620, "y": 200}}, ], "edges": [ {"id": "e_start", "source": "start-1", "target": "agent-1", "sourceHandle": "right", "targetHandle": "left"}, {"id": "e_end", "source": "agent-1", "target": "end-1", "sourceHandle": "right", "targetHandle": "left"}, ], }, }) agent_id = agent.get("id") if not agent_id: print(f" [FAIL] 创建失败: {agent}") sys.exit(1) print(f" [OK] Agent 已创建: {agent_id}") # ── Step 2: 升级为 Main Agent ── print("\n[2] 升级为 Main Agent（预置项目经理提示词 + 工具）...") try: upgraded = api("POST", f"/api/v1/agents/{agent_id}/create-main-agent", token=token) print(f" [OK] Agent Type = {upgraded.get('agent_type', '?')}") except Exception as e: print(f" [WARN] 升级失败 (后端未重启?): {e}") print(" → 手动设置 agent_type=main") try: api("PUT", f"/api/v1/agents/{agent_id}", token=token, data={ "workflow_config": { "nodes": [ {"id": "start-1", "data": {"label": "开始"}, "type": "start", "position": {"x": 80, "y": 200}}, {"id": "agent-1", "data": { "label": "项目经理", "system_prompt": ( "你是一个数字员工项目经理。当用户提出目标时：\n" "1. 理解目标的核心意图\n" "2. 将目标分解为 2-5 个具体子任务\n" "3. 调用 create_task 创建子任务\n" "4. 调用 assign_task 分配 Agent\n" "5. 调用 check_progress 追踪进度\n" "6. 遇到问题调用 notify_user 通知用户" ), "model": "deepseek-v4-flash", "provider": "deepseek", "temperature": 0.7, "max_iterations": 15, "tools": ["web_search", "knowledge_graph_search"], "memory": True, }, "type": "agent", "position": {"x": 350, "y": 200}}, {"id": "end-1", "data": {"label": "结束"}, "type": "end", "position": {"x": 620, "y": 200}}, ], "edges": [ {"id": "e_start", "source": "start-1", "target": "agent-1", "sourceHandle": "right", "targetHandle": "left"}, {"id": "e_end", "source": "agent-1", "target": "end-1", "sourceHandle": "right", "targetHandle": "left"}, ], }, "description": "数字员工项目经理（Main Agent）", }) print(" [OK] workflow_config 已更新") except Exception as e2: print(f" [WARN] 手动更新也失败: {e2}") # ── Step 3: 创建目标 ── print("\n[3] 创建目标: 调研主流 AI Agent 平台...") goal = api("POST", "/api/v1/goals", token=token, data={ "title": "调研主流 AI Agent 平台功能差异", "description": ( "调研 Langchain、AutoGPT、CrewAI、Dify、Coze、字节方舟 " "等主流 AI Agent 平台的核心功能、优劣势、适用场景，" "输出一份对比分析报告" ), "priority": 3, "main_agent_id": agent_id, "autonomy_config": { "check_interval_minutes": 5, "auto_replan": True, "notify_on_progress": True, }, }) goal_id = goal.get("id") if not goal_id: print(f" [FAIL] 创建目标失败: {goal}") sys.exit(1) print(f" [OK] Goal 已创建: {goal_id}") print(f" Title: {goal['title']}") print(f" Status: {goal['status']}") print(f" Priority: P{goal['priority']}") # ── Step 4: 分解目标 ── print("\n[4] 分解目标为子任务 (decompose)...") try: decomp = api("POST", f"/api/v1/goals/{goal_id}/decompose", token=token) if decomp: print(f" [OK] 分解完成: {decomp.get('tasks_count', '?')} 个任务") except Exception as e: print(f" [INFO] 同步分解不可用 (可能 LLM 未配置): {e}") print(" → 任务将由 execute-async 异步创建") # ── Step 5: 启动异步执行 ── print("\n[5] 启动异步执行 (execute-async)...") try: exec_result = api("POST", f"/api/v1/goals/{goal_id}/execute-async", token=token) print(f" [OK] {exec_result.get('message', '?')}") except Exception as e: print(f" [WARN] 执行投递失败 (Celery Worker 未启动?): {e}") # ── Step 6: 等待并查看结果 ── print("\n[6] 等待 3 秒后查看任务树和进度...") time.sleep(3) tree = api("GET", f"/api/v1/goals/{goal_id}/tasks", token=token) tasks = tree.get("tasks", []) if tree else [] print(f" ┌─ 任务列表 ({len(tasks)} 个任务) " + "─" * 30) for t in tasks: deps = len(t.get("depends_on") or []) dep_str = f" [依赖 {deps} 项]" if deps else "" print(f" │ [{t['status']:12s}] {t['title'][:55]}{dep_str} P{t['priority']}") print(f" └" + "─" * 50) goal_status = api("GET", f"/api/v1/goals/{goal_id}", token=token) print(f" Status: {goal_status.get('status', '?')}") print(f" Progress: {int(goal_status.get('progress', 0) * 100)}%") # ── Step 7: 总结 ── print("\n" + "=" * 60) print(" 演示完成！") print() print(" 前端页面:") print(f" 目标列表: http://localhost:3001/digital-employees") print(f" 目标详情: http://localhost:3001/goals/{goal_id}") print(f" Swagger: http://localhost:8037/docs") print() print(" Goal ID:", goal_id) print(" Agent ID:", agent_id) print("=" * 60) if __name__ == "__main__": main()