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

@@ -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

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({
+        edge_map[src].append({"target": tgt, "sourceHandle": e.get("sourceHandle")})
-            "target": e["target"],
+        if tgt not in reverse_edge_map:
-            "sourceHandle": e.get("sourceHandle")
+            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:
+    if not start_node and nodes:
-        start_node = nodes[0]  # fallback to first node
+        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,72 +585,76 @@ async def execute_workflow(
         "log": [],
     }
 
-    visited = set()
-    queue = [start_node["id"]]
     final_output = ""
+    abort = False
 
-    while queue:
+    # Execute level by level
-        node_id = queue.pop(0)
+    for lvl in range(max_level + 1):
-        if node_id in visited:
+        if abort:
-            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
+        level_nodes = level_groups[lvl]
-            next_edges = edge_map.get(node_id, [])
+        if not level_nodes:
-            if node_type == "condition":
+            continue
-                # 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:
+        # Execute nodes in this level
-                    handle = edge["sourceHandle"]
+        tasks = []
-                    if (cond_result and handle == "true") or (not cond_result and handle == "false") or handle is None:
+        for nid in level_nodes:
-                        queue.append(edge["target"])
+            node = node_map.get(nid)
-            else:
+            if not node:
-                for edge in next_edges:
+                continue
-                    queue.append(edge["target"])
+            task = _execute_node_async(node, context, db, edge_map, node_map, workflow_id)
+            tasks.append(task)
 
-        except Exception as e:
+        # 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": node_id,
+                    "node_id": "unknown",
-                "node_label": node_label,
+                    "node_label": "unknown",
-                "node_type": node_type,
+                    "node_type": "unknown",
                     "success": False,
-                "error": str(e),
+                    "error": str(res),
                 })
-            # Stop execution on error
+                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:
@@ -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

frontend/src/pages/admin/Workflow.tsx

@@ -47,6 +47,7 @@ const NODE_COLORS: Record<string, string> = {
   condition: "#BA7517",
   http: "#D4537E",
   code: "#5F5E5A",
+  agent: "#7C3AED",
   end: "#A32D2D",
 };
 
@@ -58,6 +59,7 @@ const NODE_LABELS: Record<string, string> = {
   condition: "条件分支",
   http: "HTTP 请求",
   code: "代码执行",
+  agent: "Agent 节点",
   end: "结束",
 };
 
@@ -75,6 +77,8 @@ export default function Workflow() {
   const [executeResult, setExecuteResult] = useState<any>(null);
   const [executeInput, setExecuteInput] = useState("");
   const [showExecute, setShowExecute] = useState(false);
+  const [executing, setExecuting] = useState(false);
+  const [nodeStatus, setNodeStatus] = useState<Record<string, string>>({});
   const [stats, setStats] = useState({ total: 0, draft: 0, published: 0, archived: 0 });
   const canvasRef = useRef<HTMLDivElement>(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 && !result) {
+        setExecuteResult(res.data);
+      }
+    } 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) {
+        if (res.code === 0) setExecuteResult(res.data);
-        setExecuteResult(res.data);
+      } catch (e2) { console.error(e2); }
-      } else {
+    } finally {
-        alert("执行失败: " + (res.message || "未知错误"));
+      setExecuting(false);
+      setTimeout(() => {
+        if (ws?.readyState === WebSocket.OPEN) ws.close();
+      }, 1000);
     }
-    } catch (e) { console.error(e); alert("执行失败"); }
   };
 
   // SVG edge path (bezier curve)
@@ -312,7 +392,7 @@ export default function Workflow() {
         <div style={{ display: "flex", justifyContent: "space-between", alignItems: "center", marginBottom: "24px" }}>
           <div>
             <h2 style={{ margin: 0, fontSize: "20px", color: "#fff" }}>Agent 可视化编排</h2>
-            <p style={{ margin: "4px 0 0", color: "#888", fontSize: "13px" }}>拖拽式 DAG 工作流编排引擎 · 支持LLM/知识库/Skill/条件分支/HTTP/代码节点</p>
+            <p style={{ margin: "4px 0 0", color: "#888", fontSize: "13px" }}>拖拽式 DAG 工作流编排引擎 · LLM/知识库/Skill/条件/HTTP/代码/Agent 节点 · 并行多 Agent 协作</p>
           </div>
           <button onClick={handleCreate} style={btnPrimary}>+ 新建工作流</button>
         </div>
@@ -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" }}
               />
             </div>
-            <button onClick={handleExecute} style={btnPrimary}>执行工作流</button>
+            <button onClick={handleExecute} style={{ ...btnPrimary, opacity: executing ? 0.6 : 1 }} disabled={executing}>
+              {executing ? "执行中..." : "执行工作流"}
+            </button>
           </div>
+          {executing && (
+            <div style={{ marginTop: "8px", fontSize: "12px", color: "#534AB7" }}>
+              ⏳ 工作流执行中...
+            </div>
+          )}
           {executeResult && (
             <div style={{ marginTop: "12px" }}>
               <div style={{ fontSize: "12px", color: "#888", marginBottom: "4px" }}>执行日志:</div>
@@ -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 (
               <div
                 key={node.id}
@@ -513,7 +603,7 @@ export default function Workflow() {
                   width: "90px",
                   minHeight: "50px",
                   background: "#1a1a2e",
-                  border: `2px solid ${isSelected ? color : "#333"}`,
+                  border: `${borderWidth}px solid ${color}`,
                   borderRadius: "8px",
                   cursor: draggingNode === node.id ? "grabbing" : "grab",
                   padding: "8px",