""" Agent 可视化编排 API（管理员） Phase 3.6: Agent Visual Orchestration - DAG Workflow Editor """ from fastapi import APIRouter, Depends, HTTPException, Query, Body from sqlalchemy.orm import Session from sqlalchemy import or_, func from pydantic import BaseModel from typing import Optional, List, Any import json import uuid from datetime import datetime from ...database import get_db, SessionLocal from ...models import MarketApp, EvolutionSkill, AIModel, KnowledgeBase from ...auth import get_current_admin from ...middleware.log_service import LogService router = APIRouter() # ============================================================ # Pydantic models # ============================================================ class WorkflowNode(BaseModel): id: str type: str # start / llm / knowledge / skill / condition / http / code / end position: dict # {x, y} data: dict # {label, config: {...}} class WorkflowEdge(BaseModel): id: str source: str target: str sourceHandle: Optional[str] = None targetHandle: Optional[str] = None class WorkflowCreate(BaseModel): name: str description: Optional[str] = None agent_id: Optional[str] = None nodes: List[WorkflowNode] = [] edges: List[WorkflowEdge] = [] variables: Optional[dict] = None class WorkflowUpdate(BaseModel): name: Optional[str] = None description: Optional[str] = None agent_id: Optional[str] = None nodes: Optional[List[WorkflowNode]] = None edges: Optional[List[WorkflowEdge]] = None variables: Optional[dict] = None status: Optional[str] = None # draft / published / archived class WorkflowExecuteRequest(BaseModel): input_text: str = "" variables: Optional[dict] = None # ============================================================ # Node type definitions (templates) # ============================================================ NODE_TEMPLATES = { "start": { "type": "start", "label": "开始", "icon": "play", "color": "#3B6D11", "description": "工作流入口节点", "config_schema": { "input_variable": {"type": "string", "label": "输入变量名", "default": "input"} } }, "llm": { "type": "llm", "label": "LLM 调用", "icon": "brain", "color": "#534AB7", "description": "调用大语言模型生成回复", "config_schema": { "model_id": {"type": "select", "label": "模型", "options_from": "models"}, "system_prompt": {"type": "textarea", "label": "系统提示词", "default": "你是一个有用的AI助手。"}, "user_prompt": {"type": "textarea", "label": "用户提示词模板", "default": "{input}", "help": "可用变量: {input}, {prev_output}"}, "temperature": {"type": "number", "label": "温度", "default": 0.7, "min": 0, "max": 2}, "max_tokens": {"type": "number", "label": "最大Token", "default": 2048} } }, "knowledge": { "type": "knowledge", "label": "知识库检索", "icon": "book", "color": "#185FA5", "description": "从知识库检索相关文档", "config_schema": { "kb_id": {"type": "select", "label": "知识库", "options_from": "knowledge_bases"}, "query_template": {"type": "textarea", "label": "查询模板", "default": "{input}"}, "top_k": {"type": "number", "label": "返回数量", "default": 3} } }, "skill": { "type": "skill", "label": "Skill 调用", "icon": "zap", "color": "#0F6E56", "description": "执行已注册的 Skill", "config_schema": { "skill_id": {"type": "select", "label": "Skill", "options_from": "skills"}, "params_template": {"type": "textarea", "label": "参数模板 (JSON)", "default": "{}"} } }, "condition": { "type": "condition", "label": "条件分支", "icon": "git-branch", "color": "#BA7517", "description": "根据条件选择分支", "config_schema": { "condition_expr": {"type": "textarea", "label": "条件表达式", "default": "len(input) > 0", "help": "Python表达式, 返回True走sourceHandle=true, False走false"} } }, "http": { "type": "http", "label": "HTTP 请求", "icon": "globe", "color": "#D4537E", "description": "发送 HTTP 请求", "config_schema": { "url": {"type": "string", "label": "URL", "default": "https://api.example.com"}, "method": {"type": "select", "label": "方法", "options": ["GET", "POST", "PUT", "DELETE"], "default": "GET"}, "headers": {"type": "textarea", "label": "请求头 (JSON)", "default": "{}"}, "body_template": {"type": "textarea", "label": "请求体模板", "default": ""} } }, "code": { "type": "code", "label": "代码执行", "icon": "terminal", "color": "#5F5E5A", "description": "执行 Python 代码片段", "config_schema": { "code": {"type": "textarea", "label": "Python 代码", "default": "result = input\noutput = result", "help": "可用变量: input, prev_output, context. 必须设置 output 变量"} } }, "end": { "type": "end", "label": "结束", "icon": "stop", "color": "#A32D2D", "description": "工作流出口节点", "config_schema": { "output_variable": {"type": "string", "label": "输出变量名", "default": "output"} } } } # ============================================================ # Helpers # ============================================================ def _get_workflow_table(): """Get the agent_workflows table via raw SQL (avoid model changes)""" from sqlalchemy import text return text def workflow_to_dict(row) -> dict: """Convert a workflow DB row (dict) to response dict""" return { "id": row.get("id"), "name": row.get("name"), "description": row.get("description"), "agent_id": row.get("agent_id"), "nodes": json.loads(row["nodes"]) if isinstance(row.get("nodes"), str) else (row.get("nodes") or []), "edges": json.loads(row["edges"]) if isinstance(row.get("edges"), str) else (row.get("edges") or []), "variables": json.loads(row["variables"]) if isinstance(row.get("variables"), str) else (row.get("variables") or {}), "status": row.get("status", "draft"), "version": row.get("version", 1), "created_by": row.get("created_by"), "created_at": row["created_at"].isoformat() if row.get("created_at") else None, "updated_at": row["updated_at"].isoformat() if row.get("updated_at") else None, } def _execute_raw(db, sql, params=None): """Execute raw SQL and return rows as dicts""" from sqlalchemy import text result = db.execute(text(sql), params or {}) if result.returns_rows: cols = result.keys() return [dict(zip(cols, row)) for row in result.fetchall()] return [] # ============================================================ # Endpoints # ============================================================ @router.get("/templates", summary="获取节点模板") async def get_node_templates( admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """获取所有可用的节点类型模板和配置 schema""" # Fetch options for select fields options = {} # Models models = db.query(AIModel).filter(AIModel.is_enabled == True).all() options["models"] = [{"value": m.id, "label": f"{getattr(m, 'display_name', None) or m.name}"} for m in models] # Knowledge bases kbs = db.query(KnowledgeBase).all() options["knowledge_bases"] = [{"value": kb.id, "label": kb.name} for kb in kbs] # Skills skills = db.query(EvolutionSkill).filter(EvolutionSkill.state == "active").all() options["skills"] = [{"value": s.id, "label": s.skill_name} for s in skills] return { "code": 0, "data": { "node_templates": NODE_TEMPLATES, "options": options, } } @router.get("/", summary="获取工作流列表") async def list_workflows( keyword: str = Query(None), status: str = Query(None), page: int = Query(1, ge=1), page_size: int = Query(15, ge=1, le=100), admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """获取所有工作流列表（分页）""" conditions = [] params = {} if keyword: conditions.append("(name LIKE :kw OR description LIKE :kw)") params["kw"] = f"%{keyword}%" if status: conditions.append("status = :status") params["status"] = status where_clause = " AND ".join(conditions) if conditions else "1=1" # Count count_sql = f"SELECT COUNT(*) as total FROM agent_workflows WHERE {where_clause}" count_rows = _execute_raw(db, count_sql, params) total = count_rows[0]["total"] if count_rows else 0 # Query offset = (page - 1) * page_size query_sql = f""" SELECT * FROM agent_workflows WHERE {where_clause} ORDER BY updated_at DESC LIMIT :limit OFFSET :offset """ params["limit"] = page_size params["offset"] = offset rows = _execute_raw(db, query_sql, params) return { "code": 0, "data": { "total": total, "page": page, "page_size": page_size, "items": [workflow_to_dict(r) for r in rows], } } @router.get("/stats", summary="工作流统计") async def get_workflow_stats( admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """获取工作流统计数据""" rows = _execute_raw(db, "SELECT status, COUNT(*) as cnt FROM agent_workflows GROUP BY status") stats = {r["status"]: r["cnt"] for r in rows} total = sum(stats.values()) return { "code": 0, "data": { "total": total, "draft": stats.get("draft", 0), "published": stats.get("published", 0), "archived": stats.get("archived", 0), } } @router.get("/{workflow_id}", summary="获取工作流详情") async def get_workflow( workflow_id: str, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """获取单个工作流详情""" rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id}) if not rows: raise HTTPException(status_code=404, detail="工作流不存在") result = workflow_to_dict(rows[0]) # Get agent info if linked if result.get("agent_id"): agent = db.query(MarketApp).filter(MarketApp.id == result["agent_id"]).first() if agent: result["agent_name"] = agent.name return {"code": 0, "data": result} @router.post("/", summary="创建工作流") async def create_workflow( req: WorkflowCreate, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """创建新的工作流""" from sqlalchemy import text wf_id = str(uuid.uuid4()) now = datetime.now() # Validate: must have exactly one start node start_nodes = [n for n in req.nodes if n.type == "start"] if len(start_nodes) == 0 and len(req.nodes) > 0: raise HTTPException(status_code=400, detail="工作流必须包含一个开始节点") if len(start_nodes) > 1: raise HTTPException(status_code=400, detail="工作流只能有一个开始节点") sql = text(""" INSERT INTO agent_workflows (id, name, description, agent_id, nodes, edges, variables, status, version, created_by, created_at, updated_at) VALUES (:id, :name, :description, :agent_id, :nodes, :edges, :variables, 'draft', 1, :created_by, :created_at, :updated_at) """) db.execute(sql, { "id": wf_id, "name": req.name, "description": req.description or "", "agent_id": req.agent_id, "nodes": json.dumps([n.dict() for n in req.nodes], ensure_ascii=False), "edges": json.dumps([e.dict() for e in req.edges], ensure_ascii=False), "variables": json.dumps(req.variables or {}, ensure_ascii=False), "created_by": admin_user.id, "created_at": now, "updated_at": now, }) db.commit() LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="create_workflow", target_type="workflow", target_id=wf_id, detail=f"创建工作流: {req.name}", ) # Return created rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": wf_id}) return {"code": 0, "message": "工作流创建成功", "data": workflow_to_dict(rows[0])} @router.put("/{workflow_id}", summary="更新工作流") async def update_workflow( workflow_id: str, req: WorkflowUpdate, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """更新工作流""" from sqlalchemy import text rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id}) if not rows: raise HTTPException(status_code=404, detail="工作流不存在") existing = rows[0] sets = [] params = {"id": workflow_id, "updated_at": datetime.now()} if req.name is not None: sets.append("name = :name") params["name"] = req.name if req.description is not None: sets.append("description = :description") params["description"] = req.description if req.agent_id is not None: sets.append("agent_id = :agent_id") params["agent_id"] = req.agent_id if req.nodes is not None: sets.append("nodes = :nodes") params["nodes"] = json.dumps([n.dict() for n in req.nodes], ensure_ascii=False) if req.edges is not None: sets.append("edges = :edges") params["edges"] = json.dumps([e.dict() for e in req.edges], ensure_ascii=False) if req.variables is not None: sets.append("variables = :variables") params["variables"] = json.dumps(req.variables, ensure_ascii=False) if req.status is not None: sets.append("status = :status") params["status"] = req.status if sets: sets.append("version = version + 1") sql = text(f"UPDATE agent_workflows SET {', '.join(sets)} WHERE id = :id") db.execute(sql, params) db.commit() LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="update_workflow", target_type="workflow", target_id=workflow_id, detail=f"更新工作流: {existing.get('name', '')}", ) rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id}) return {"code": 0, "message": "工作流已更新", "data": workflow_to_dict(rows[0])} @router.delete("/{workflow_id}", summary="删除工作流") async def delete_workflow( workflow_id: str, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """删除工作流""" from sqlalchemy import text rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id}) if not rows: raise HTTPException(status_code=404, detail="工作流不存在") wf_name = rows[0].get("name", "") db.execute(text("DELETE FROM agent_workflows WHERE id = :id"), {"id": workflow_id}) db.commit() LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="delete_workflow", target_type="workflow", target_id=workflow_id, detail=f"删除工作流: {wf_name}", ) return {"code": 0, "message": "工作流已删除"} @router.put("/{workflow_id}/toggle", summary="发布/归档工作流") async def toggle_workflow( workflow_id: str, is_active: bool = Query(..., description="True=发布, False=归档"), admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """发布或归档工作流""" from sqlalchemy import text rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id}) if not rows: raise HTTPException(status_code=404, detail="工作流不存在") new_status = "published" if is_active else "archived" db.execute( text("UPDATE agent_workflows SET status = :status, updated_at = :now WHERE id = :id"), {"status": new_status, "now": datetime.now(), "id": workflow_id} ) db.commit() return {"code": 0, "message": f"工作流已{'发布' if is_active else '归档'}"} @router.post("/{workflow_id}/execute", summary="执行工作流") async def execute_workflow( workflow_id: str, req: WorkflowExecuteRequest, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """执行工作流 — 按 DAG 拓扑顺序处理节点""" rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id}) if not rows: raise HTTPException(status_code=404, detail="工作流不存在") wf = workflow_to_dict(rows[0]) nodes = wf["nodes"] edges = wf["edges"] if not nodes: raise HTTPException(status_code=400, detail="工作流没有节点") # Build adjacency map node_map = {n["id"]: n for n in nodes} edge_map = {} # source_id -> [{target, handle}] for e in edges: src = e["source"] if src not in edge_map: edge_map[src] = [] edge_map[src].append({ "target": e["target"], "sourceHandle": e.get("sourceHandle") }) # Find start node start_node = None for n in nodes: if n["type"] == "start": start_node = n break if not start_node: start_node = nodes[0] # fallback to first node # Execute context = { "input": req.input_text, "variables": req.variables or {}, "prev_output": "", "results": {}, "log": [], } visited = set() queue = [start_node["id"]] final_output = "" while queue: node_id = queue.pop(0) if node_id in visited: continue visited.add(node_id) node = node_map.get(node_id) if not node: continue node_type = node["type"] node_config = node.get("data", {}).get("config", {}) node_label = node.get("data", {}).get("label", node_type) try: result = _execute_node(node_type, node_config, context, db) context["results"][node_id] = result context["prev_output"] = result.get("output", "") context["log"].append({ "node_id": node_id, "node_label": node_label, "node_type": node_type, "success": True, "output_preview": str(result.get("output", ""))[:200], }) if node_type == "end": final_output = result.get("output", "") break # Determine next nodes next_edges = edge_map.get(node_id, []) if node_type == "condition": # Evaluate condition cond_expr = node_config.get("condition_expr", "True") try: cond_result = eval(cond_expr, {"__builtins__": {}}, { "input": context["input"], "prev_output": context["prev_output"], "len": len, "str": str, }) except Exception: cond_result = True for edge in next_edges: handle = edge["sourceHandle"] if (cond_result and handle == "true") or (not cond_result and handle == "false") or handle is None: queue.append(edge["target"]) else: for edge in next_edges: queue.append(edge["target"]) except Exception as e: context["log"].append({ "node_id": node_id, "node_label": node_label, "node_type": node_type, "success": False, "error": str(e), }) # Stop execution on error break if not final_output: final_output = context["prev_output"] return { "code": 0, "data": { "output": final_output, "log": context["log"], "results": context["results"], } } def _execute_node(node_type: str, config: dict, context: dict, db: Session) -> dict: """Execute a single node and return result""" if node_type == "start": return {"output": context["input"]} elif node_type == "llm": return _execute_llm_node(config, context, db) elif node_type == "knowledge": return _execute_knowledge_node(config, context, db) elif node_type == "skill": return _execute_skill_node(config, context, db) elif node_type == "condition": return {"output": context["prev_output"]} elif node_type == "http": return _execute_http_node(config, context) elif node_type == "code": return _execute_code_node(config, context) elif node_type == "end": var_name = config.get("output_variable", "output") return {"output": context.get(var_name, context.get("prev_output", ""))} return {"output": ""} def _execute_llm_node(config: dict, context: dict, db: Session) -> dict: """Execute LLM call""" import httpx model_id = config.get("model_id") if not model_id: return {"output": "[LLM] 未配置模型", "error": "no model_id"} model = db.query(AIModel).filter(AIModel.id == model_id).first() if not model: return {"output": "[LLM] 模型不存在", "error": "model not found"} system_prompt = config.get("system_prompt", "你是一个有用的AI助手。") user_prompt = config.get("user_prompt", "{input}") # Template substitution user_prompt = user_prompt.replace("{input}", str(context.get("input", ""))) user_prompt = user_prompt.replace("{prev_output}", str(context.get("prev_output", ""))) api_key = model.api_key or "" base_url = model.api_base or "https://api.siliconflow.cn/v1" model_name = model.model_id or model.name headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } payload = { "model": model_name, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], "temperature": float(config.get("temperature", 0.7)), "max_tokens": int(config.get("max_tokens", 2048)), } try: with httpx.Client(timeout=60) as client: resp = client.post(f"{base_url}/chat/completions", headers=headers, json=payload) resp.raise_for_status() data = resp.json() output = data["choices"][0]["message"]["content"] return {"output": output, "usage": data.get("usage", {})} except Exception as e: return {"output": f"[LLM] 调用失败: {str(e)}", "error": str(e)} def _execute_knowledge_node(config: dict, context: dict, db: Session) -> dict: """Execute knowledge base search""" kb_id = config.get("kb_id") if not kb_id: return {"output": "[KB] 未配置知识库", "error": "no kb_id"} query = config.get("query_template", "{input}") query = query.replace("{input}", str(context.get("input", ""))) query = query.replace("{prev_output}", str(context.get("prev_output", ""))) top_k = int(config.get("top_k", 3)) # Use BM25 search (same as chat.py) try: from ...models import KnowledgeDocumentChunk chunks = db.query(KnowledgeDocumentChunk).filter( KnowledgeDocumentChunk.knowledge_base_id == kb_id ).limit(100).all() if not chunks: return {"output": "[KB] 知识库无文档", "results": []} # Simple keyword matching (BM25 simplified) query_words = set(query.lower().split()) scored = [] for chunk in chunks: content = (chunk.content or "").lower() score = sum(1 for w in query_words if w in content) if score > 0: scored.append((score, chunk.content[:500])) scored.sort(key=lambda x: -x[0]) results = [s[1] for s in scored[:top_k]] if not results: results = [c.content[:500] for c in chunks[:top_k]] combined = "\n\n---\n\n".join(results) return {"output": combined, "results": results} except Exception as e: return {"output": f"[KB] 检索失败: {str(e)}", "error": str(e)} def _execute_skill_node(config: dict, context: dict, db: Session) -> dict: """Execute skill call""" skill_id = config.get("skill_id") if not skill_id: return {"output": "[Skill] 未配置 Skill", "error": "no skill_id"} skill = db.query(EvolutionSkill).filter(EvolutionSkill.id == skill_id).first() if not skill: return {"output": "[Skill] Skill 不存在", "error": "skill not found"} # For now, return skill info (actual execution would call the skill handler) return { "output": f"[Skill] {skill.skill_name} 已调用", "skill_id": skill_id, "skill_name": skill.skill_name, } def _execute_http_node(config: dict, context: dict) -> dict: """Execute HTTP request""" import httpx url = config.get("url", "") method = config.get("method", "GET") headers_str = config.get("headers", "{}") body_template = config.get("body_template", "") # Template substitution url = url.replace("{input}", str(context.get("input", ""))) body_template = body_template.replace("{input}", str(context.get("input", ""))) body_template = body_template.replace("{prev_output}", str(context.get("prev_output", ""))) try: headers = json.loads(headers_str) if headers_str else {} except Exception: headers = {} try: with httpx.Client(timeout=30) as client: if method == "GET": resp = client.get(url, headers=headers) elif method == "POST": resp = client.post(url, headers=headers, content=body_template) elif method == "PUT": resp = client.put(url, headers=headers, content=body_template) elif method == "DELETE": resp = client.delete(url, headers=headers) else: return {"output": f"[HTTP] 不支持的方法: {method}"} return {"output": resp.text[:2000], "status_code": resp.status_code} except Exception as e: return {"output": f"[HTTP] 请求失败: {str(e)}", "error": str(e)} def _execute_code_node(config: dict, context: dict) -> dict: """Execute Python code (sandboxed)""" code = config.get("code", "output = input") # Very basic sandbox — restrict builtins safe_builtins = { "len": len, "str": str, "int": int, "float": float, "bool": bool, "list": list, "dict": dict, "set": set, "range": range, "enumerate": enumerate, "zip": zip, "min": min, "max": max, "sum": sum, "abs": abs, "round": round, "sorted": sorted, "reversed": reversed, "print": print, "json": json, "True": True, "False": False, "None": None, } local_vars = { "input": context.get("input", ""), "prev_output": context.get("prev_output", ""), "context": context, "output": "", } try: exec(code, {"__builtins__": safe_builtins}, local_vars) return {"output": str(local_vars.get("output", ""))} except Exception as e: return {"output": f"[Code] 执行失败: {str(e)}", "error": str(e)}