From d2e3147fb8f808a2973b2d2e67e832ed3687522d Mon Sep 17 00:00:00 2001 From: qyyfw Date: Mon, 6 Jul 2026 15:48:37 +0800 Subject: [PATCH] Phase G3: ACP Gateway + DAG Multi-Agent - acp_gateway.py (Agent communication layer, 350L), workflow_admin.py upgraded (async parallel execution + agent_node), Workflow.tsx updated (agent node + real-time streaming + WebSocket) --- backend/python/app/api/admin/acp_gateway.py | 491 +++++++++++++++ .../python/app/api/admin/workflow_admin.py | 562 +++++++++++++++--- frontend/src/pages/admin/Workflow.tsx | 108 +++- 3 files changed, 1073 insertions(+), 88 deletions(-) create mode 100644 backend/python/app/api/admin/acp_gateway.py diff --git a/backend/python/app/api/admin/acp_gateway.py b/backend/python/app/api/admin/acp_gateway.py new file mode 100644 index 0000000..7b0f172 --- /dev/null +++ b/backend/python/app/api/admin/acp_gateway.py @@ -0,0 +1,491 @@ +""" +ACP Gateway - Agent Communication Protocol 内部实现 +Phase G3: 多 Agent 协作编排层 + +将平台已有的 Agent（MarketApp type='agent'）封装为可编排调用的"Agent 节点"。 +每个 Agent 节点执行时，通过本 Gateway 创建会话、发送提示、接收流式响应。 +支持通过 Redis EventBus 发布节点进度通知，供 WebSocket 实时推送到前端。 +""" +import json +import uuid +import time +import logging +import asyncio +import threading +from typing import Optional, Dict, Any, Callable, AsyncIterator +from dataclasses import dataclass, field + +logger = logging.getLogger(__name__) + +# ============================================================ +# Data Models +# ============================================================ + +@dataclass +class AcpSessionState: + """ACP 会话状态""" + session_id: str + agent_id: str # MarketApp id + agent_name: str = "" + status: str = "idle" # idle / running / completed / failed / cancelled + prompt: str = "" + created_at: float = field(default_factory=time.time) + completed_at: Optional[float] = None + output: str = "" + error: Optional[str] = None + tool_calls: list = field(default_factory=list) + chain_of_thought: list = field(default_factory=list) # 思考过程 + + def to_dict(self) -> dict: + return { + "session_id": self.session_id, + "agent_id": self.agent_id, + "agent_name": self.agent_name, + "status": self.status, + "prompt": self.prompt[:200], + "created_at": self.created_at, + "completed_at": self.completed_at, + "output": self.output[:500] if self.output else "", + "error": self.error, + "tool_calls": self.tool_calls, + "chain_of_thought": self.chain_of_thought, + } + + +@dataclass +class AgentConfig: + """从 MarketApp.skill_content 解析出的 Agent 配置""" + agent_id: str = "" + agent_name: str = "" + system_prompt: str = "" + instructions: str = "" + model_id: Optional[str] = None + temperature: float = 0.7 + max_tokens: int = 4096 + skills: list = field(default_factory=list) + knowledge_base_ids: list = field(default_factory=list) + tools: list = field(default_factory=list) + + +# ============================================================ +# Agent Resolver +# ============================================================ + +def resolve_agent_config(db, agent_id: str) -> Optional[AgentConfig]: + """从 MarketApp 解析 Agent 配置""" + from app.models import MarketApp + + agent = db.query(MarketApp).filter( + MarketApp.id == agent_id, + MarketApp.type == "agent" + ).first() + + if not agent: + return None + + config_json = {} + if agent.skill_content: + try: + config_json = json.loads(agent.skill_content) if isinstance(agent.skill_content, str) else agent.skill_content + except Exception: + pass + + return AgentConfig( + agent_id=agent.id, + agent_name=agent.name, + system_prompt=config_json.get("system_prompt", config_json.get("role", agent.description or "AI助手")), + instructions=config_json.get("instructions", ""), + model_id=config_json.get("model_id"), + temperature=float(config_json.get("temperature", 0.7)), + max_tokens=int(config_json.get("max_tokens", 4096)), + skills=config_json.get("skills", []), + knowledge_base_ids=config_json.get("knowledge_base_ids", []), + tools=config_json.get("tools", []), + ) + + +def resolve_model_config(db, model_id: str) -> dict: + """获取模型的 API 配置""" + from app.models import AIModel + + model = db.query(AIModel).filter(AIModel.id == model_id).first() + if not model: + # Fallback to default model + default_model = db.query(AIModel).filter(AIModel.is_enabled == True).first() + if default_model: + model = default_model + else: + return { + "api_key": "", + "api_base": "https://api.siliconflow.cn/v1", + "model_name": "deepseek-ai/DeepSeek-V4-Flash", + } + + return { + "api_key": model.api_key or "", + "api_base": model.api_base or "https://api.siliconflow.cn/v1", + "model_name": model.model_id or model.name, + } + + +# ============================================================ +# ACP Gateway +# ============================================================ + +class AcpGateway: + """Agent 通信网关 — 平台内部多 Agent 编排核心""" + + def __init__(self): + self._sessions: Dict[str, AcpSessionState] = {} + self._lock = threading.Lock() + self._on_notification: Optional[Callable] = None + + def set_notification_handler(self, handler: Callable[[dict], None]): + """设置通知回调 — 用于发布到 Redis EventBus""" + self._on_notification = handler + + def _notify(self, event_type: str, data: dict): + """发送通知""" + if self._on_notification: + try: + self._on_notification({ + "type": event_type, + "timestamp": time.time(), + **data, + }) + except Exception as e: + logger.error(f"[AcpGateway] 通知回调异常: {e}") + + # ---- Session Management ---- + + def create_session(self, agent_id: str, agent_name: str = "") -> str: + """创建新的 Agent 会话""" + session_id = f"acp-{uuid.uuid4().hex[:12]}" + with self._lock: + self._sessions[session_id] = AcpSessionState( + session_id=session_id, + agent_id=agent_id, + agent_name=agent_name or agent_id, + status="idle", + ) + logger.info(f"[AcpGateway] 创建会话: {session_id} (agent={agent_name or agent_id})") + return session_id + + def get_session(self, session_id: str) -> Optional[AcpSessionState]: + """获取会话""" + with self._lock: + return self._sessions.get(session_id) + + def cancel_session(self, session_id: str) -> bool: + """取消会话""" + with self._lock: + sess = self._sessions.get(session_id) + if sess and sess.status == "running": + sess.status = "cancelled" + sess.completed_at = time.time() + self._notify("agent_cancelled", sess.to_dict()) + return True + return False + + # ---- Agent Invocation ---- + + async def invoke_agent( + self, + agent_id: str, + prompt: str, + system_prompt: str = "", + model_id: Optional[str] = None, + temperature: float = 0.7, + max_tokens: int = 4096, + context: Optional[dict] = None, + db_session=None, + ) -> dict: + """调用 Agent — 创建会话、发送提示、等待完成 + + 这是一个高级封装，内部自动创建和管理会话生命周期。 + 返回完整结果字典，包含 output, tool_calls, chain_of_thought 等。 + """ + import httpx + + # Resolve agent config from DB + agent_name = agent_id + if db_session: + config = resolve_agent_config(db_session, agent_id) + if config: + agent_name = config.agent_name + system_prompt = system_prompt or config.system_prompt + model_id = model_id or config.model_id + temperature = config.temperature + max_tokens = config.max_tokens + + # Create session + session_id = self.create_session(agent_id, agent_name) + sess = self.get_session(session_id) + + # Resolve model + if db_session and model_id: + model_cfg = resolve_model_config(db_session, model_id) + else: + model_cfg = {"api_key": "", "api_base": "https://api.siliconflow.cn/v1", "model_name": "deepseek-ai/DeepSeek-V4-Flash"} + + # Update status + with self._lock: + sess.status = "running" + sess.prompt = prompt + + self._notify("agent_running", {"session_id": session_id, "agent_name": agent_name, "prompt": prompt[:200]}) + + try: + # Build messages + messages = [] + if system_prompt: + messages.append({"role": "system", "content": system_prompt}) + messages.append({"role": "user", "content": prompt}) + + # Call LLM + headers = { + "Authorization": f"Bearer {model_cfg['api_key']}", + "Content-Type": "application/json", + } + payload = { + "model": model_cfg["model_name"], + "messages": messages, + "temperature": temperature, + "max_tokens": max_tokens, + "stream": False, + } + + async with httpx.AsyncClient(timeout=120.0) as client: + resp = await client.post( + f"{model_cfg['api_base']}/chat/completions", + headers=headers, + json=payload, + ) + resp.raise_for_status() + data = resp.json() + + output = data["choices"][0]["message"]["content"] + usage = data.get("usage", {}) + + with self._lock: + sess.status = "completed" + sess.output = output + sess.completed_at = time.time() + + result = { + "session_id": session_id, + "status": "completed", + "output": output, + "agent_name": agent_name, + "usage": usage, + "tool_calls": sess.tool_calls, + "chain_of_thought": sess.chain_of_thought, + } + self._notify("agent_completed", {"session_id": session_id, "output_preview": output[:200]}) + return result + + except Exception as e: + error_msg = str(e) + with self._lock: + sess.status = "failed" + sess.error = error_msg + sess.completed_at = time.time() + + result = { + "session_id": session_id, + "status": "failed", + "output": f"[Agent] 调用失败: {error_msg}", + "agent_name": agent_name, + "error": error_msg, + } + self._notify("agent_failed", {"session_id": session_id, "error": error_msg}) + return result + + async def invoke_agent_streaming( + self, + agent_id: str, + prompt: str, + system_prompt: str = "", + model_id: Optional[str] = None, + temperature: float = 0.7, + max_tokens: int = 4096, + context: Optional[dict] = None, + db_session=None, + ) -> AsyncIterator[dict]: + """流式调用 Agent — 逐步返回 LLM 生成的 token chunks + + 每个 yield 的 dict 格式: + {"type": "token"|"done"|"error", "content": str, ...} + """ + import httpx + + # Resolve agent config + agent_name = agent_id + if db_session: + config = resolve_agent_config(db_session, agent_id) + if config: + agent_name = config.agent_name + system_prompt = system_prompt or config.system_prompt + model_id = model_id or config.model_id + temperature = config.temperature + max_tokens = config.max_tokens + + session_id = self.create_session(agent_id, agent_name) + sess = self.get_session(session_id) + + if db_session and model_id: + model_cfg = resolve_model_config(db_session, model_id) + else: + model_cfg = {"api_key": "", "api_base": "https://api.siliconflow.cn/v1", "model_name": "deepseek-ai/DeepSeek-V4-Flash"} + + with self._lock: + sess.status = "running" + sess.prompt = prompt + + self._notify("agent_running", {"session_id": session_id, "agent_name": agent_name}) + + try: + messages = [] + if system_prompt: + messages.append({"role": "system", "content": system_prompt}) + messages.append({"role": "user", "content": prompt}) + + headers = { + "Authorization": f"Bearer {model_cfg['api_key']}", + "Content-Type": "application/json", + } + payload = { + "model": model_cfg["model_name"], + "messages": messages, + "temperature": temperature, + "max_tokens": max_tokens, + "stream": True, + } + + collected = [] + async with httpx.AsyncClient(timeout=120.0) as client: + async with client.stream("POST", f"{model_cfg['api_base']}/chat/completions", + headers=headers, json=payload) as resp: + resp.raise_for_status() + async for line in resp.aiter_lines(): + if line.startswith("data: "): + data_str = line[6:] + if data_str.strip() == "[DONE]": + break + try: + chunk = json.loads(data_str) + delta = chunk.get("choices", [{}])[0].get("delta", {}) + content = delta.get("content", "") + if content: + collected.append(content) + yield {"type": "token", "content": content, "session_id": session_id} + # Notify periodically (every ~20 tokens) + if len(collected) % 20 == 0: + self._notify("agent_token", { + "session_id": session_id, + "partial": "".join(collected[-20:]), + }) + except json.JSONDecodeError: + continue + + full_output = "".join(collected) + with self._lock: + sess.status = "completed" + sess.output = full_output + sess.completed_at = time.time() + + self._notify("agent_completed", {"session_id": session_id, "output_preview": full_output[:200]}) + yield {"type": "done", "session_id": session_id, "output": full_output, "status": "completed"} + + except Exception as e: + error_msg = str(e) + with self._lock: + sess.status = "failed" + sess.error = error_msg + sess.completed_at = time.time() + + self._notify("agent_failed", {"session_id": session_id, "error": error_msg}) + yield {"type": "error", "session_id": session_id, "error": error_msg, "status": "failed"} + + # ---- Multi-Agent Parallel Execution ---- + + async def invoke_agents_parallel( + self, + tasks: list, + db_session=None, + ) -> list: + """并行调用多个 Agent + + tasks = [ + {"agent_id": "xxx", "prompt": "...", "system_prompt": "..."}, + {"agent_id": "yyy", "prompt": "...", "system_prompt": "..."}, + ] + + Returns list of results in same order. + """ + async def _invoke_one(task: dict) -> dict: + try: + return await self.invoke_agent( + agent_id=task["agent_id"], + prompt=task.get("prompt", ""), + system_prompt=task.get("system_prompt", ""), + model_id=task.get("model_id"), + temperature=task.get("temperature", 0.7), + max_tokens=task.get("max_tokens", 4096), + context=task.get("context"), + db_session=db_session, + ) + except Exception as e: + return { + "status": "failed", + "output": f"[Agent {task.get('agent_id', 'unknown')}] 执行异常: {str(e)}", + "error": str(e), + } + + return await asyncio.gather(*[_invoke_one(t) for t in tasks]) + + +# ============================================================ +# Global Singleton +# ============================================================ + +_acp_gateway_instance: Optional[AcpGateway] = None +_lock_gateway = threading.Lock() + + +def get_acp_gateway() -> AcpGateway: + """获取全局 ACP Gateway 单例""" + global _acp_gateway_instance + if _acp_gateway_instance is None: + with _lock_gateway: + if _acp_gateway_instance is None: + _acp_gateway_instance = AcpGateway() + return _acp_gateway_instance + + +# ============================================================ +# Redis Notification Helper +# ============================================================ + +def publish_workflow_event(workflow_id: str, node_id: str, event: dict): + """发布工作流执行事件到 Redis EventBus""" + try: + from app.event_bus import get_event_bus, MessageEvent + + bus = get_event_bus() + msg = MessageEvent( + platform="workflow", + channel_id=f"{workflow_id}:{node_id}", + content=json.dumps(event, ensure_ascii=False), + content_type="workflow_event", + raw={ + "workflow_id": workflow_id, + "node_id": node_id, + **event, + }, + ) + bus.publish(msg) + return True + except Exception as e: + logger.error(f"[AcpGateway] Redis 发布失败: {e}") + return False diff --git a/backend/python/app/api/admin/workflow_admin.py b/backend/python/app/api/admin/workflow_admin.py index b710e3a..0423dc3 100644 --- a/backend/python/app/api/admin/workflow_admin.py +++ b/backend/python/app/api/admin/workflow_admin.py @@ -2,13 +2,14 @@ Agent 可视化编排 API（管理员） Phase 3.6: Agent Visual Orchestration - DAG Workflow Editor """ -from fastapi import APIRouter, Depends, HTTPException, Query, Body +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 @@ -25,7 +26,7 @@ router = APIRouter() class WorkflowNode(BaseModel): id: str - type: str # start / llm / knowledge / skill / condition / http / code / end + type: str # start / llm / knowledge / skill / condition / http / code / agent / end position: dict # {x, y} data: dict # {label, config: {...}} @@ -147,6 +148,21 @@ NODE_TEMPLATES = { "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": "结束", @@ -216,13 +232,17 @@ async def get_node_templates( 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() + 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": { @@ -496,7 +516,7 @@ async def execute_workflow( admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): - """执行工作流 — 按 DAG 拓扑顺序处理节点""" + """执行工作流 — 按 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="工作流不存在") @@ -508,28 +528,55 @@ async def execute_workflow( if not nodes: raise HTTPException(status_code=400, detail="工作流没有节点") - # Build adjacency map + # Build structures node_map = {n["id"]: n for n in nodes} - edge_map = {} # source_id -> [{target, handle}] + 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": e["target"], - "sourceHandle": e.get("sourceHandle") - }) + 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 node + # Find start nodes start_node = None for n in nodes: if n["type"] == "start": - start_node = n + start_node = n["id"] break - if not start_node: - start_node = nodes[0] # fallback to first node + if not start_node and nodes: + start_node = nodes[0]["id"] - # Execute + # 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 {}, @@ -538,74 +585,78 @@ async def execute_workflow( "log": [], } - visited = set() - queue = [start_node["id"]] final_output = "" + abort = False - 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 + # 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"] @@ -619,6 +670,195 @@ async def execute_workflow( } +@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""" @@ -643,6 +883,9 @@ def _execute_node(node_type: str, config: dict, context: dict, db: Session) -> d 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", ""))} @@ -650,6 +893,114 @@ def _execute_node(node_type: str, config: dict, context: dict, db: Session) -> d 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 @@ -759,6 +1110,59 @@ def _execute_skill_node(config: dict, context: dict, db: Session) -> dict: } +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 diff --git a/frontend/src/pages/admin/Workflow.tsx b/frontend/src/pages/admin/Workflow.tsx index ea0e394..d59eb51 100644 --- a/frontend/src/pages/admin/Workflow.tsx +++ b/frontend/src/pages/admin/Workflow.tsx @@ -47,6 +47,7 @@ const NODE_COLORS: Record = { condition: "#BA7517", http: "#D4537E", code: "#5F5E5A", + agent: "#7C3AED", end: "#A32D2D", }; @@ -58,6 +59,7 @@ const NODE_LABELS: Record = { condition: "条件分支", http: "HTTP 请求", code: "代码执行", + agent: "Agent 节点", end: "结束", }; @@ -75,6 +77,8 @@ export default function Workflow() { const [executeResult, setExecuteResult] = useState(null); const [executeInput, setExecuteInput] = useState(""); const [showExecute, setShowExecute] = useState(false); + const [executing, setExecuting] = useState(false); + const [nodeStatus, setNodeStatus] = useState>({}); const [stats, setStats] = useState({ total: 0, draft: 0, published: 0, archived: 0 }); const canvasRef = useRef(null); const dragOffset = useRef({ x: 0, y: 0 }); @@ -278,19 +282,95 @@ export default function Workflow() { setSelectedNode(updatedNodes.find(n => n.id === selectedNode.id) || null); }; - // Execute workflow + // Execute workflow (supports streaming via WebSocket when available) const handleExecute = async () => { if (!currentWf) return; + setExecuting(true); + setExecuteResult(null); + setNodeStatus({}); + + // Highlight running nodes in real-time + const wsProtocol = window.location.protocol === "https:" ? "wss:" : "ws:"; + const wsHost = window.location.host; + const wsUrl = `${wsProtocol}//${wsHost}/app/api/admin/workflows/${currentWf.id}/execute-stream`; + + let ws: WebSocket | null = null; + let result: any = null; + try { + // Connect WebSocket + ws = new WebSocket(wsUrl); + ws.onmessage = (event) => { + try { + const msg = JSON.parse(event.data); + if (msg.type === "node_start") { + setNodeStatus(prev => ({ ...prev, [msg.node_id]: "running" })); + // Add to log + setExecuteResult((prev: any) => ({ + ...prev, + log: [...(prev?.log || []), { + node_id: msg.node_id, + node_label: msg.node_label, + node_type: msg.node_type, + success: null, + }], + })); + } else if (msg.type === "node_complete") { + setNodeStatus(prev => ({ ...prev, [msg.node_id]: "completed" })); + // Update log entry + setExecuteResult((prev: any) => { + const log = (prev?.log || []).map((l: any) => + l.node_id === msg.node_id ? { ...l, success: true, output_preview: msg.output_preview } : l + ); + return { ...prev, log }; + }); + } else if (msg.type === "node_error") { + setNodeStatus(prev => ({ ...prev, [msg.node_id]: "failed" })); + setExecuteResult((prev: any) => { + const log = (prev?.log || []).map((l: any) => + l.node_id === msg.node_id ? { ...l, success: false, error: msg.error } : l + ); + return { ...prev, log }; + }); + } else if (msg.type === "workflow_complete") { + result = msg.data; + setExecuteResult(msg.data); + } else if (msg.type === "workflow_error") { + result = msg.data; + setExecuteResult(msg.data); + } + } catch (e) { /* ignore parse errors */ } + }; + + ws.onopen = () => { + ws?.send(JSON.stringify({ + input_text: executeInput, + token: (window as any).api?.getToken?.() || "", + })); + }; + + // Also call REST API as fallback const res = await api.post(`/admin/workflows/${currentWf.id}/execute`, { input_text: executeInput, }); - if (res.code === 0) { + if (res.code === 0 && !result) { setExecuteResult(res.data); - } else { - alert("执行失败: " + (res.message || "未知错误")); } - } catch (e) { console.error(e); alert("执行失败"); } + } catch (e) { + console.error(e); + // Try REST API fallback + try { + const res = await api.post(`/admin/workflows/${currentWf.id}/execute`, { + input_text: executeInput, + }); + if (res.code === 0) setExecuteResult(res.data); + } catch (e2) { console.error(e2); } + } finally { + setExecuting(false); + setTimeout(() => { + if (ws?.readyState === WebSocket.OPEN) ws.close(); + }, 1000); + } }; // SVG edge path (bezier curve) @@ -312,7 +392,7 @@ export default function Workflow() {

Agent 可视化编排

-

拖拽式 DAG 工作流编排引擎 · 支持LLM/知识库/Skill/条件分支/HTTP/代码节点

+

拖拽式 DAG 工作流编排引擎 · LLM/知识库/Skill/条件/HTTP/代码/Agent 节点 · 并行多 Agent 协作

@@ -408,8 +488,15 @@ export default function Workflow() { style={{ width: "100%", background: "#0d0d1a", border: "1px solid #333", borderRadius: "6px", padding: "8px", color: "#fff", fontSize: "13px", minHeight: "40px", resize: "vertical" }} /> - + + {executing && ( +
+ ⏳ 工作流执行中... +
+ )} {executeResult && (
执行日志:
@@ -500,7 +587,10 @@ export default function Workflow() { {/* Nodes */} {currentWf.nodes.map(node => { const isSelected = selectedNode?.id === node.id; - const color = NODE_COLORS[node.type] || "#5F5E5A"; + const status = nodeStatus[node.id]; + const baseColor = NODE_COLORS[node.type] || "#5F5E5A"; + const color = status === "running" ? "#58a6ff" : status === "completed" ? "#3fb950" : status === "failed" ? "#f85149" : baseColor; + const borderWidth = isSelected || status ? 2 : 1; return (