1229 lines
42 KiB
Python
1229 lines
42 KiB
Python
"""
|
|
Agent 可视化编排 API（管理员）
|
|
Phase 3.6: Agent Visual Orchestration - DAG Workflow Editor
|
|
"""
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Body, WebSocket, WebSocketDisconnect
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import or_, func
|
|
from pydantic import BaseModel
|
|
from typing import Optional, List, Any
|
|
import json
|
|
import uuid
|
|
import asyncio
|
|
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 / agent / 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 变量"}
|
|
}
|
|
},
|
|
"agent": {
|
|
"type": "agent",
|
|
"label": "Agent 节点",
|
|
"icon": "bot",
|
|
"color": "#7C3AED",
|
|
"description": "调用平台 AI Agent — 支持并行多 Agent 协作",
|
|
"config_schema": {
|
|
"agent_id": {"type": "select", "label": "选择 Agent", "options_from": "agents"},
|
|
"prompt": {"type": "textarea", "label": "提示词模板", "default": "{input}", "help": "可用变量: {input}, {prev_output}, {context} 中的变量"},
|
|
"system_prompt": {"type": "textarea", "label": "系统提示词 (覆盖Agent默认)", "default": "", "help": "留空则使用Agent默认系统提示词"},
|
|
"temperature": {"type": "number", "label": "温度 (覆盖默认)", "default": 0.7, "min": 0, "max": 2},
|
|
"max_tokens": {"type": "number", "label": "最大Token", "default": 4096},
|
|
"output_variable": {"type": "string", "label": "输出变量名", "default": "", "help": "将Agent输出存入context.{变量名}，供后续节点使用"}
|
|
}
|
|
},
|
|
"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).filter(KnowledgeBase.status == "active").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]
|
|
|
|
# Agents (for multi-agent orchestration - Phase G3)
|
|
agents = db.query(MarketApp).filter(MarketApp.type == "agent", MarketApp.status == "published").all()
|
|
options["agents"] = [{"value": a.id, "label": a.name} for a in agents]
|
|
|
|
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 拓扑顺序处理节点，同层节点并行执行 (Phase G3)"""
|
|
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 structures
|
|
node_map = {n["id"]: n for n in nodes}
|
|
edge_map = {}
|
|
reverse_edge_map = {} # target -> [sources]
|
|
for e in edges:
|
|
src = e["source"]
|
|
tgt = e["target"]
|
|
if src not in edge_map:
|
|
edge_map[src] = []
|
|
edge_map[src].append({"target": tgt, "sourceHandle": e.get("sourceHandle")})
|
|
if tgt not in reverse_edge_map:
|
|
reverse_edge_map[tgt] = []
|
|
reverse_edge_map[tgt].append(src)
|
|
|
|
# Find start nodes
|
|
start_node = None
|
|
for n in nodes:
|
|
if n["type"] == "start":
|
|
start_node = n["id"]
|
|
break
|
|
if not start_node and nodes:
|
|
start_node = nodes[0]["id"]
|
|
|
|
# Topological sort to assign levels
|
|
in_degree = {n["id"]: len(reverse_edge_map.get(n["id"], [])) for n in nodes}
|
|
levels = {}
|
|
queue = [start_node]
|
|
levels[start_node] = 0
|
|
|
|
while queue:
|
|
nid = queue.pop(0)
|
|
level = levels[nid]
|
|
for edge in edge_map.get(nid, []):
|
|
tgt = edge["target"]
|
|
if tgt in node_map:
|
|
new_level = level + 1
|
|
if tgt not in levels or new_level > levels[tgt]:
|
|
levels[tgt] = new_level
|
|
in_degree[tgt] -= 1
|
|
if in_degree[tgt] <= 0:
|
|
queue.append(tgt)
|
|
|
|
# Group nodes by level
|
|
max_level = max(levels.values()) if levels else 0
|
|
level_groups = {i: [] for i in range(max_level + 1)}
|
|
for nid, lvl in levels.items():
|
|
level_groups[lvl].append(nid)
|
|
|
|
# Context
|
|
context = {
|
|
"input": req.input_text,
|
|
"variables": req.variables or {},
|
|
"prev_output": "",
|
|
"results": {},
|
|
"log": [],
|
|
}
|
|
|
|
final_output = ""
|
|
abort = False
|
|
|
|
# Execute level by level
|
|
for lvl in range(max_level + 1):
|
|
if abort:
|
|
break
|
|
|
|
level_nodes = level_groups[lvl]
|
|
if not level_nodes:
|
|
continue
|
|
|
|
# Execute nodes in this level
|
|
tasks = []
|
|
for nid in level_nodes:
|
|
node = node_map.get(nid)
|
|
if not node:
|
|
continue
|
|
task = _execute_node_async(node, context, db, edge_map, node_map, workflow_id)
|
|
tasks.append(task)
|
|
|
|
# Parallel execution for same-level nodes
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
|
|
for res in results:
|
|
if isinstance(res, Exception):
|
|
context["log"].append({
|
|
"node_id": "unknown",
|
|
"node_label": "unknown",
|
|
"node_type": "unknown",
|
|
"success": False,
|
|
"error": str(res),
|
|
})
|
|
abort = True
|
|
continue
|
|
|
|
if res is None:
|
|
continue
|
|
|
|
nid = res["node_id"]
|
|
context["results"][nid] = res["result"]
|
|
|
|
if res["success"]:
|
|
context["prev_output"] = res["result"].get("output", "")
|
|
context["log"].append({
|
|
"node_id": nid,
|
|
"node_label": res["node_label"],
|
|
"node_type": res["node_type"],
|
|
"success": True,
|
|
"output_preview": str(res["result"].get("output", ""))[:200],
|
|
})
|
|
|
|
if res["node_type"] == "end":
|
|
final_output = res["result"].get("output", "")
|
|
abort = True
|
|
break
|
|
|
|
# Handle condition node branching — update context for next level
|
|
if res["node_type"] == "condition" and "next_nodes" in res:
|
|
# The condition routing is applied by _execute_node_async
|
|
pass
|
|
else:
|
|
context["log"].append({
|
|
"node_id": nid,
|
|
"node_label": res["node_label"],
|
|
"node_type": res["node_type"],
|
|
"success": False,
|
|
"error": res.get("error", "未知错误"),
|
|
})
|
|
abort = True
|
|
break
|
|
|
|
if not final_output:
|
|
final_output = context["prev_output"]
|
|
|
|
return {
|
|
"code": 0,
|
|
"data": {
|
|
"output": final_output,
|
|
"log": context["log"],
|
|
"results": context["results"],
|
|
}
|
|
}
|
|
|
|
|
|
@router.websocket("/{workflow_id}/execute-stream")
|
|
async def execute_workflow_stream(
|
|
websocket: WebSocket,
|
|
workflow_id: str,
|
|
):
|
|
"""WebSocket 端点 — 实时流式推送工作流执行进度 (Phase G3)"""
|
|
await websocket.accept()
|
|
|
|
try:
|
|
# Receive execution params
|
|
data = await websocket.receive_json()
|
|
input_text = data.get("input_text", "")
|
|
|
|
# Create DB session
|
|
db = SessionLocal()
|
|
try:
|
|
rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id})
|
|
if not rows:
|
|
await websocket.send_json({"type": "workflow_error", "data": {"output": "工作流不存在"}})
|
|
await websocket.close()
|
|
return
|
|
|
|
wf = workflow_to_dict(rows[0])
|
|
nodes = wf["nodes"]
|
|
edges = wf["edges"]
|
|
|
|
if not nodes:
|
|
await websocket.send_json({"type": "workflow_error", "data": {"output": "工作流没有节点"}})
|
|
await websocket.close()
|
|
return
|
|
|
|
# Build structures
|
|
node_map = {n["id"]: n for n in nodes}
|
|
edge_map = {}
|
|
reverse_edge_map = {}
|
|
for e in edges:
|
|
src = e["source"]
|
|
tgt = e["target"]
|
|
if src not in edge_map:
|
|
edge_map[src] = []
|
|
edge_map[src].append({"target": tgt, "sourceHandle": e.get("sourceHandle")})
|
|
if tgt not in reverse_edge_map:
|
|
reverse_edge_map[tgt] = []
|
|
reverse_edge_map[tgt].append(src)
|
|
|
|
# Find start
|
|
start_node = None
|
|
for n in nodes:
|
|
if n["type"] == "start":
|
|
start_node = n["id"]
|
|
break
|
|
if not start_node and nodes:
|
|
start_node = nodes[0]["id"]
|
|
|
|
# Topo sort
|
|
in_degree = {n["id"]: len(reverse_edge_map.get(n["id"], [])) for n in nodes}
|
|
levels = {start_node: 0}
|
|
queue = [start_node]
|
|
while queue:
|
|
nid = queue.pop(0)
|
|
lvl = levels[nid]
|
|
for edge in edge_map.get(nid, []):
|
|
tgt = edge["target"]
|
|
if tgt in node_map:
|
|
nl = lvl + 1
|
|
if tgt not in levels or nl > levels[tgt]:
|
|
levels[tgt] = nl
|
|
in_degree[tgt] -= 1
|
|
if in_degree[tgt] <= 0:
|
|
queue.append(tgt)
|
|
|
|
max_level = max(levels.values()) if levels else 0
|
|
level_groups = {i: [] for i in range(max_level + 1)}
|
|
for nid, lvl in levels.items():
|
|
level_groups[lvl].append(nid)
|
|
|
|
context = {
|
|
"input": input_text,
|
|
"variables": {},
|
|
"prev_output": "",
|
|
"results": {},
|
|
"log": [],
|
|
}
|
|
|
|
final_output = ""
|
|
abort = False
|
|
|
|
for lvl in range(max_level + 1):
|
|
if abort:
|
|
break
|
|
|
|
level_nodes = level_groups[lvl]
|
|
if not level_nodes:
|
|
continue
|
|
|
|
tasks_list = []
|
|
for nid in level_nodes:
|
|
node = node_map.get(nid)
|
|
if not node:
|
|
continue
|
|
|
|
# Send node_start
|
|
await websocket.send_json({
|
|
"type": "node_start",
|
|
"node_id": nid,
|
|
"node_type": node["type"],
|
|
"node_label": node.get("data", {}).get("label", node["type"]),
|
|
})
|
|
|
|
task = _execute_node_async(node, context, db, edge_map, node_map, workflow_id)
|
|
tasks_list.append(task)
|
|
|
|
results = await asyncio.gather(*tasks_list, return_exceptions=True)
|
|
|
|
for res in results:
|
|
if isinstance(res, Exception):
|
|
await websocket.send_json({
|
|
"type": "node_error",
|
|
"node_id": "unknown",
|
|
"node_label": "unknown",
|
|
"node_type": "unknown",
|
|
"error": str(res),
|
|
})
|
|
abort = True
|
|
continue
|
|
|
|
if res is None:
|
|
continue
|
|
|
|
nid = res["node_id"]
|
|
context["results"][nid] = res["result"]
|
|
|
|
if res["success"]:
|
|
context["prev_output"] = res["result"].get("output", "")
|
|
|
|
await websocket.send_json({
|
|
"type": "node_complete",
|
|
"node_id": nid,
|
|
"node_label": res["node_label"],
|
|
"node_type": res["node_type"],
|
|
"output_preview": str(res["result"].get("output", ""))[:200],
|
|
})
|
|
|
|
if res["node_type"] == "end":
|
|
final_output = res["result"].get("output", "")
|
|
abort = True
|
|
break
|
|
else:
|
|
await websocket.send_json({
|
|
"type": "node_error",
|
|
"node_id": nid,
|
|
"node_label": res["node_label"],
|
|
"node_type": res["node_type"],
|
|
"error": res.get("error", "未知错误"),
|
|
})
|
|
abort = True
|
|
break
|
|
|
|
if not final_output:
|
|
final_output = context["prev_output"]
|
|
|
|
await websocket.send_json({
|
|
"type": "workflow_complete",
|
|
"data": {
|
|
"output": final_output,
|
|
"log": context["log"],
|
|
}
|
|
})
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
except WebSocketDisconnect:
|
|
pass
|
|
except Exception as e:
|
|
try:
|
|
await websocket.send_json({
|
|
"type": "workflow_error",
|
|
"data": {"output": f"执行失败: {str(e)}"}
|
|
})
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
try:
|
|
await websocket.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
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 == "agent":
|
|
return _execute_agent_node(config, context, db)
|
|
|
|
elif node_type == "end":
|
|
var_name = config.get("output_variable", "output")
|
|
return {"output": context.get(var_name, context.get("prev_output", ""))}
|
|
|
|
return {"output": ""}
|
|
|
|
|
|
async def _execute_node_async(node: dict, context: dict, db: Session, edge_map: dict, node_map: dict, workflow_id: str) -> Optional[dict]:
|
|
"""Execute a single node async, returning structured result for parallel orchestration"""
|
|
nid = node["id"]
|
|
node_type = node["type"]
|
|
node_config = node.get("data", {}).get("config", {})
|
|
node_label = node.get("data", {}).get("label", node_type)
|
|
|
|
try:
|
|
# Publish node start event
|
|
from app.api.admin.acp_gateway import publish_workflow_event
|
|
publish_workflow_event(workflow_id, nid, {
|
|
"event": "node_start",
|
|
"node_type": node_type,
|
|
"node_label": node_label,
|
|
})
|
|
|
|
if node_type == "agent":
|
|
# Async agent invocation — may take time
|
|
result = await _execute_agent_node_async(node_config, context, db, workflow_id, nid)
|
|
|
|
# Store output in context variable if specified
|
|
output_var = node_config.get("output_variable", "")
|
|
if output_var and result.get("output"):
|
|
context[output_var] = result["output"]
|
|
|
|
publish_workflow_event(workflow_id, nid, {
|
|
"event": "node_complete",
|
|
"node_type": node_type,
|
|
"node_label": node_label,
|
|
"output_preview": str(result.get("output", ""))[:200],
|
|
})
|
|
else:
|
|
# Sync node execution (run in thread)
|
|
result = await asyncio.to_thread(_execute_node, node_type, node_config, context, db)
|
|
publish_workflow_event(workflow_id, nid, {
|
|
"event": "node_complete",
|
|
"node_type": node_type,
|
|
"node_label": node_label,
|
|
"output_preview": str(result.get("output", ""))[:200],
|
|
})
|
|
|
|
return {
|
|
"node_id": nid,
|
|
"node_label": node_label,
|
|
"node_type": node_type,
|
|
"success": True,
|
|
"result": result,
|
|
}
|
|
|
|
except Exception as e:
|
|
from app.api.admin.acp_gateway import publish_workflow_event
|
|
publish_workflow_event(workflow_id, nid, {
|
|
"event": "node_error",
|
|
"node_type": node_type,
|
|
"node_label": node_label,
|
|
"error": str(e),
|
|
})
|
|
return {
|
|
"node_id": nid,
|
|
"node_label": node_label,
|
|
"node_type": node_type,
|
|
"success": False,
|
|
"error": str(e),
|
|
}
|
|
|
|
|
|
async def _execute_agent_node_async(config: dict, context: dict, db: Session, workflow_id: str, node_id: str) -> dict:
|
|
"""Execute agent node via ACP Gateway — async"""
|
|
from app.api.admin.acp_gateway import get_acp_gateway
|
|
|
|
agent_id = config.get("agent_id")
|
|
if not agent_id:
|
|
return {"output": "[Agent] 未选择 Agent", "error": "no agent_id"}
|
|
|
|
# Build prompt from template
|
|
prompt = config.get("prompt", "{input}")
|
|
prompt = prompt.replace("{input}", str(context.get("input", "")))
|
|
prompt = prompt.replace("{prev_output}", str(context.get("prev_output", "")))
|
|
# Substitute context variables
|
|
for key, val in context.items():
|
|
if isinstance(val, str):
|
|
prompt = prompt.replace("{" + key + "}", val)
|
|
|
|
system_prompt = config.get("system_prompt", "")
|
|
temperature = float(config.get("temperature", 0.7))
|
|
max_tokens = int(config.get("max_tokens", 4096))
|
|
|
|
gateway = get_acp_gateway()
|
|
|
|
# Set notification handler for Redis EventBus
|
|
from app.api.admin.acp_gateway import publish_workflow_event
|
|
gateway.set_notification_handler(
|
|
lambda data: publish_workflow_event(workflow_id, node_id, data)
|
|
)
|
|
|
|
result = await gateway.invoke_agent(
|
|
agent_id=agent_id,
|
|
prompt=prompt,
|
|
system_prompt=system_prompt,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
context=context,
|
|
db_session=db,
|
|
)
|
|
|
|
return result
|
|
|
|
|
|
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_agent_node(config: dict, context: dict, db: Session) -> dict:
|
|
"""Execute agent node (sync wrapper) — calls the async version via asyncio.run"""
|
|
agent_id = config.get("agent_id")
|
|
if not agent_id:
|
|
return {"output": "[Agent] 未选择 Agent", "error": "no agent_id"}
|
|
|
|
prompt = config.get("prompt", "{input}")
|
|
prompt = prompt.replace("{input}", str(context.get("input", "")))
|
|
prompt = prompt.replace("{prev_output}", str(context.get("prev_output", "")))
|
|
|
|
system_prompt = config.get("system_prompt", "")
|
|
temperature = float(config.get("temperature", 0.7))
|
|
max_tokens = int(config.get("max_tokens", 4096))
|
|
|
|
from .acp_gateway import get_acp_gateway
|
|
|
|
gateway = get_acp_gateway()
|
|
|
|
try:
|
|
import asyncio
|
|
loop = asyncio.get_event_loop()
|
|
if loop.is_running():
|
|
# In async context — create new loop
|
|
new_loop = asyncio.new_event_loop()
|
|
result = new_loop.run_until_complete(
|
|
gateway.invoke_agent(
|
|
agent_id=agent_id,
|
|
prompt=prompt,
|
|
system_prompt=system_prompt,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
context=context,
|
|
db_session=db,
|
|
)
|
|
)
|
|
new_loop.close()
|
|
else:
|
|
result = loop.run_until_complete(
|
|
gateway.invoke_agent(
|
|
agent_id=agent_id,
|
|
prompt=prompt,
|
|
system_prompt=system_prompt,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
context=context,
|
|
db_session=db,
|
|
)
|
|
)
|
|
return result
|
|
except Exception as e:
|
|
return {"output": f"[Agent] 调用失败: {str(e)}", "error": str(e)}
|
|
|
|
|
|
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)}
|