From 79c27795a89b915da6e83aa5eec18d74eadb3e78 Mon Sep 17 00:00:00 2001 From: QihuAI Design Date: Sun, 19 Jul 2026 20:17:49 +0800 Subject: [PATCH] Phase 4.2+4.3: YAML workflow spec + ACA protocol abstraction (v2026.07.19.10) --- VERSION | 2 +- backend/python/app/api/admin/agents_admin.py | 128 ++++++++ backend/python/app/core/aca.py | 293 +++++++++++++++++ backend/python/app/core/yaml_workflow.py | 302 ++++++++++++++++++ .../app/workflow_templates/code-review.yaml | 35 ++ .../multi-agent-debate.yaml | 35 ++ .../app/workflow_templates/rag-search.yaml | 25 ++ .../research-assistant.yaml | 25 ++ backend/python/tests/test_42_43.py | 92 ++++++ 9 files changed, 936 insertions(+), 1 deletion(-) create mode 100644 backend/python/app/core/aca.py create mode 100644 backend/python/app/core/yaml_workflow.py create mode 100644 backend/python/app/workflow_templates/code-review.yaml create mode 100644 backend/python/app/workflow_templates/multi-agent-debate.yaml create mode 100644 backend/python/app/workflow_templates/rag-search.yaml create mode 100644 backend/python/app/workflow_templates/research-assistant.yaml create mode 100644 backend/python/tests/test_42_43.py diff --git a/VERSION b/VERSION index 3bf5dc1..e0a5db2 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -v2026.07.19.9 +v2026.07.19.10 diff --git a/backend/python/app/api/admin/agents_admin.py b/backend/python/app/api/admin/agents_admin.py index fa3a16f..cc81e93 100644 --- a/backend/python/app/api/admin/agents_admin.py +++ b/backend/python/app/api/admin/agents_admin.py @@ -654,3 +654,131 @@ async def set_agent_orchestration( agent.orchestration = og db.commit() return {"agent_id": agent_id, "orchestration": agent.orchestration, "updated": True} + + +# ============================================================ +# Phase 4.2: YAML workflow import/export/validate/templates +# ============================================================ + +import yaml as _yaml +import os as _os +from app.core.yaml_workflow import validate_yaml, yaml_to_orchestration, orchestration_to_yaml +from app.core.orchestrator import validate_graph + +_YTMPL = _os.path.join(_os.path.dirname(_os.path.dirname(_os.path.dirname(__file__))), "workflow_templates") + + +class YAMLBody(BaseModel): + yaml_text: str + + +@router.post("/orchestration/yaml/validate", summary="校验 YAML 工作流") +async def yaml_validate(req: YAMLBody, admin_user=Depends(get_current_admin)): + try: + data = _yaml.safe_load(req.yaml_text) + except _yaml.YAMLError as e: + return {"ok": False, "errors": [f"YAML parse: {str(e)}"], "warnings": []} + r = validate_yaml(data) + if r["ok"]: + try: + validate_graph(yaml_to_orchestration(data)) + except Exception as e: + r["ok"] = False; r["errors"].append(f"Graph fail: {str(e)}") + return r + + +@router.post("/{agent_id}/orchestration/yaml/import", summary="YAML 导入编排") +async def yaml_import(agent_id: str, req: YAMLBody, admin_user=Depends(get_current_admin), db=Depends(get_db)): + ag = db.query(MarketApp).filter(MarketApp.id == agent_id, MarketApp.type == "agent").first() + if not ag: + raise HTTPException(status_code=404, detail="Agent not found") + try: + data = _yaml.safe_load(req.yaml_text) + except _yaml.YAMLError as e: + raise HTTPException(status_code=400, detail=f"YAML parse: {str(e)}") + v = validate_yaml(data) + if not v["ok"]: + raise HTTPException(status_code=400, detail="YAML invalid: " + "; ".join(v["errors"])) + orch = yaml_to_orchestration(data) + try: + validate_graph(orch) + except Exception as e: + raise HTTPException(status_code=400, detail=f"Graph invalid: {str(e)}") + ag.orchestration = orch + ag.updated_at = datetime.now() + db.commit() + return {"agent_id": agent_id, "orchestration": ag.orchestration, "imported": True} + + +@router.post("/{agent_id}/orchestration/yaml/export", summary="编排导出 YAML") +async def yaml_export(agent_id: str, admin_user=Depends(get_current_admin), db=Depends(get_db)): + ag = db.query(MarketApp).filter(MarketApp.id == agent_id, MarketApp.type == "agent").first() + if not ag: + raise HTTPException(status_code=404, detail="Agent not found") + if not ag.orchestration: + raise HTTPException(status_code=400, detail="No orchestration") + yaml_str = _yaml.dump(orchestration_to_yaml(ag.orchestration), allow_unicode=True, default_flow_style=False, sort_keys=False) + return {"agent_id": agent_id, "yaml": yaml_str, "version": "1.0"} + + +@router.get("/orchestration/yaml/templates", summary="内置模板") +async def yaml_templates(admin_user=Depends(get_current_admin)): + tmpls = [] + if _os.path.isdir(_YTMPL): + for fp in sorted(_os.listdir(_YTMPL)): + if fp.endswith((".yaml", ".yml")): + try: + with open(_os.path.join(_YTMPL, fp), encoding="utf-8") as f: + txt = f.read() + d = _yaml.safe_load(txt) or {} + tmpls.append({"filename": fp, "name": d.get("name", fp), + "description": d.get("description", ""), + "category": d.get("category", "general"), "yaml": txt}) + except Exception: + tmpls.append({"filename": fp, "name": fp, "error": True}) + return {"templates": tmpls, "count": len(tmpls)} + + +# ============================================================ +# Phase 4.3: ACA 协议抽象 — protocols / tools / health +# ============================================================ + +from app.core.aca import registry as _aca_registry, init_aca_registry + +# 懒初始化：首次调用 ACA 端点时确保 registry 已填充 +_aca_initialized = False + +def _ensure_aca(): + global _aca_initialized + if not _aca_initialized: + init_aca_registry() + _aca_initialized = True + + +@router.get("/aca/protocols", summary="ACA 已注册协议") +async def aca_protocols(admin_user=Depends(get_current_admin)): + _ensure_aca() + return [ + {"name": a.name, "version": a.version, "capabilities": a.capabilities} + for a in _aca_registry.list_all() + ] + + +@router.get("/aca/protocols/{name}/tools", summary="协议工具列表") +async def aca_protocol_tools(name: str, admin_user=Depends(get_current_admin)): + _ensure_aca() + adapter = _aca_registry.get(name) + if not adapter: + raise HTTPException(status_code=404, detail=f"协议 {name} 未注册") + tools = await adapter.list_tools() + return {"protocol": name, "tools": tools, "count": len(tools)} + + +@router.post("/aca/protocols/{name}/health", summary="协议健康检查") +async def aca_protocol_health(name: str, admin_user=Depends(get_current_admin)): + _ensure_aca() + adapter = _aca_registry.get(name) + if not adapter: + raise HTTPException(status_code=404, detail=f"协议 {name} 未注册") + ok = await adapter.health_check() + return {"protocol": name, "healthy": ok} diff --git a/backend/python/app/core/aca.py b/backend/python/app/core/aca.py new file mode 100644 index 0000000..1d7f070 --- /dev/null +++ b/backend/python/app/core/aca.py @@ -0,0 +1,293 @@ +"""Phase 4.3: ACA (Agent Communication Abstraction) 协议抽象层. + +设计目标： +- 定义统一对外工具协议接口，隔离 MCP / A2A / 自定义协议的具体实现。 +- 防止协议绑架：未来 MCP 升级或 A2A 替代 MCP 时，只需增删适配器，不改业务代码。 +- 每个协议适配器被注册为 "protocol adapter"，按名称发现。 + +接口： + ACAProtocol + name: str + version: str + capabilities: list[str] + async list_tools() -> list[dict] # [{name, description, inputSchema}] + async call_tool(tool_name, arguments) -> str + async health_check() -> bool + + ProtocolRegistry + register(adapter: ACAProtocol) + get(name: str) -> ACAProtocol + list_all() -> list[ACAProtocol] + async list_all_tools() -> list[dict] # 聚合所有协议的 tools，带 protocol tag + async call_tool(protocol_name, tool_name, args) -> str + +使用示例: + from app.core.aca import registry + tools = await registry.list_all_tools() + # tools: [{"name":"search","protocol":"mcp","description":"..."}, ...] + result = await registry.call_tool("mcp", "search", {"query":"..."}) +""" +from __future__ import annotations + +import abc +import asyncio +import json +import logging +from typing import Any + +import httpx + +logger = logging.getLogger("aiflow.aca") + +# ── 协议接口 ───────────────────────────────────────────────────────────────── + +class ACAProtocol(abc.ABC): + name: str = "" + version: str = "1.0" + capabilities: list[str] = [] + + @abc.abstractmethod + async def list_tools(self) -> list[dict]: + """返回 [{name, description, inputSchema}]。""" + + @abc.abstractmethod + async def call_tool(self, tool_name: str, arguments: dict) -> str: + """调用工具并返回文本结果。""" + + async def health_check(self) -> bool: + """健康检查，默认 True。""" + return True + + def __repr__(self): + return f"" + + +# ── 注册表 ─────────────────────────────────────────────────────────────────── + +class ProtocolRegistry: + """全局协议注册表，单例。""" + + def __init__(self): + self._adapters: dict[str, ACAProtocol] = {} + + def register(self, adapter: ACAProtocol): + if not adapter.name: + raise ValueError("protocol adapter must have a name") + self._adapters[adapter.name] = adapter + logger.info("ACA registry: registered %s", adapter) + + def get(self, name: str) -> ACAProtocol | None: + return self._adapters.get(name) + + def list_all(self) -> list[ACAProtocol]: + return list(self._adapters.values()) + + async def list_all_tools(self) -> list[dict]: + """聚合所有已注册协议的 tools，每个 tool 标记 protocol。""" + tasks = [self._list_with_name(a) for a in self._adapters.values()] + results = await asyncio.gather(*tasks, return_exceptions=True) + flat: list[dict] = [] + for rlist in results: + if isinstance(rlist, list): + flat.extend(rlist) + return flat + + async def _list_with_name(self, adapter: ACAProtocol) -> list[dict]: + try: + tools = await adapter.list_tools() + for t in tools: + t["protocol"] = adapter.name + return tools + except Exception as e: + logger.warning("ACA: list_tools failed for %s: %s", adapter.name, e) + return [] + + async def call_tool(self, protocol_name: str, tool_name: str, arguments: dict) -> str: + adapter = self._adapters.get(protocol_name) + if not adapter: + raise ValueError(f"协议 {protocol_name} 未注册") + return await adapter.call_tool(tool_name, arguments) + + +# 全局单例 +registry = ProtocolRegistry() + + +# ── MCP 适配器 ─────────────────────────────────────────────────────────────── + +class McpProtocolAdapter(ACAProtocol): + """包装平台现有的 MCP 客户端 (app.mcp_client)，提供统一协议接口。""" + + def __init__(self, server_configs: list[dict] = None): + self.name = "mcp" + self.version = "1.0" + self.capabilities = ["tools", "resources", "prompts"] + self._server_configs = server_configs or [] + self._tools_cache: list[dict] = [] + self._cache_ts: float = 0 + + async def list_tools(self) -> list[dict]: + # Use cached tools if available (TTL 30s) + import time + if self._tools_cache and (time.time() - self._cache_ts) < 30: + return [dict(t) for t in self._tools_cache] + from app.mcp_client import list_mcp_tools + all_tools: list[dict] = [] + for server in self._server_configs: + try: + tools = await list_mcp_tools(server) + for t in tools: + t["_mcp_server_id"] = server.get("id", "") + t["_mcp_server_name"] = server.get("name", "") + all_tools.extend(tools) + except Exception as e: + logger.warning("MCP list_tools failed for %s: %s", server.get("name"), e) + self._tools_cache = [dict(t) for t in all_tools] + self._cache_ts = time.time() + return list(self._tools_cache) + + async def call_tool(self, tool_name: str, arguments: dict) -> str: + from app.mcp_client import call_mcp_tool + # Find the server that has this tool (check cache first) + for t in self._tools_cache: + if t.get("name") == tool_name: + sid = t.get("_mcp_server_id", "") + server = next((s for s in self._server_configs if s.get("id") == sid), None) + if server: + return await call_mcp_tool(server, tool_name, arguments) + # Fallback: try first available server + if self._server_configs: + return await call_mcp_tool(self._server_configs[0], tool_name, arguments) + return json.dumps({"error": "No MCP server configured for tool: " + tool_name}) + + async def health_check(self) -> bool: + # Check if any MCP server is connected + try: + tools = await self.list_tools() + return len(tools) > 0 + except Exception: + return False + + +# ── A2A 适配器 ─────────────────────────────────────────────────────────────── + +class A2aProtocolAdapter(ACAProtocol): + """A2A (Agent-to-Agent) 协议适配器。 + + 通过标准 JSON-RPC 与远端 Agent 通信。端点需实现: + - POST /a2a/tools → 返回工具列表 + - POST /a2a/call → 调用工具 + + A2A 规范参考: agentcommunicationprotocol.dev + """ + + def __init__(self, endpoint: str, api_key: str = None, timeout: float = 30): + self.name = "a2a" + self.version = "0.9" + self.capabilities = ["tools"] + self.endpoint = endpoint.rstrip("/") + self.api_key = api_key + self.timeout = timeout + + async def list_tools(self) -> list[dict]: + async with httpx.AsyncClient(timeout=self.timeout, trust_env=False) as client: + headers = self._headers() + resp = await client.post(f"{self.endpoint}/a2a/tools", headers=headers, json={}) + if resp.status_code != 200: + logger.warning("A2A list_tools %s HTTP %s: %s", self.endpoint, resp.status_code, resp.text[:200]) + return [] + data = resp.json() + return data.get("tools") or data.get("data", {}).get("tools") or [] + + async def call_tool(self, tool_name: str, arguments: dict) -> str: + async with httpx.AsyncClient(timeout=self.timeout, trust_env=False) as client: + headers = self._headers() + body = { + "jsonrpc": "2.0", + "method": "tools/call", + "params": {"name": tool_name, "arguments": arguments}, + "id": 1, + } + resp = await client.post(f"{self.endpoint}/a2a/call", headers=headers, json=body) + if resp.status_code != 200: + return json.dumps({"error": f"A2A HTTP {resp.status_code}: {resp.text[:300]}"}) + data = resp.json() + if "error" in data: + return json.dumps({"error": data["error"]}) + result = data.get("result") + if isinstance(result, dict): + return result.get("content", json.dumps(result)) + return str(result) if result is not None else "" + + async def health_check(self) -> bool: + try: + async with httpx.AsyncClient(timeout=10, trust_env=False) as client: + headers = self._headers() + resp = await client.get(f"{self.endpoint}/a2a/health", headers=headers) + return resp.status_code == 200 + except Exception: + return False + + def _headers(self) -> dict: + h = {"Content-Type": "application/json"} + if self.api_key: + h["Authorization"] = f"Bearer {self.api_key}" + return h + + +# ── 启动时初始化 ───────────────────────────────────────────────────────────── + +def init_aca_registry(): + """在应用启动时调用，注册所有协议适配器到全局 registry。""" + from app.database import SessionLocal + from app.models import MCPServer + + # 注册 MCP 适配器（从 DB 读取所有 enabled MCP server 配置） + try: + db = SessionLocal() + servers = db.query(MCPServer).filter(MCPServer.is_enabled == True).all() + server_configs = [] + for s in servers: + server_configs.append({ + "id": s.id, + "name": s.name, + "transport_type": s.transport_type, + "command": s.command, + "args": json.loads(s.args) if isinstance(s.args, str) else s.args or [], + "env_vars": json.loads(s.env_vars) if isinstance(s.env_vars, str) else s.env_vars or {}, + "url": s.url, + "headers": json.loads(s.headers) if isinstance(s.headers, str) else s.headers or {}, + }) + db.close() + if server_configs: + registry.register(McpProtocolAdapter(server_configs)) + logger.info("ACA: MCP adapter registered with %d servers", len(server_configs)) + except Exception as e: + logger.warning("ACA: failed to init MCP adapter: %s", e) + + # 注册 A2A 适配器（从配置或 DB 读取 A2A endpoint） + # A2A endpoint 可以从 site_settings 读取，暂无则留空（以后台配置驱动） + a2a_endpoint = _get_a2a_config() + if a2a_endpoint: + registry.register(A2aProtocolAdapter( + endpoint=a2a_endpoint.get("endpoint", ""), + api_key=a2a_endpoint.get("api_key"), + )) + logger.info("ACA: A2A adapter registered at %s", a2a_endpoint.get("endpoint")) + + +def _get_a2a_config() -> dict | None: + """从 site_settings 读取 A2A 全局配置。""" + try: + from app.database import SessionLocal + from app.models import SiteSetting + db = SessionLocal() + setting = db.query(SiteSetting).filter(SiteSetting.key == "aca_a2a_endpoint").first() + db.close() + if setting and setting.value: + val = json.loads(setting.value) if isinstance(setting.value, str) else setting.value + if isinstance(val, dict) and val.get("endpoint"): + return val + except Exception: + pass + return None diff --git a/backend/python/app/core/yaml_workflow.py b/backend/python/app/core/yaml_workflow.py new file mode 100644 index 0000000..f15c88b --- /dev/null +++ b/backend/python/app/core/yaml_workflow.py @@ -0,0 +1,302 @@ +"""Phase 4.2: QihuFlow 工作流可移植描述语言 (YAML v1.0) + +提供 YAML 工作流文本 ↔ orchestrator JSON 的双向转换与校验。 + +YAML 规范 v1.0 示例: + + version: "1.0" + name: "Research & Code" + description: "Research then generate code" + output_node: merge + + nodes: + - id: research + type: llm + on_error: continue + config: + model: auto + system_prompt: "You are a researcher." + - id: coder + type: local + agent_id: "uuid-here" + config: + skill_ids: ["search", "code-run"] + - id: merge + type: reduce + config: + strategy: concat + + edges: + - from: research + to: coder + - from: coder + to: merge + +config 字段映射（YAML config ↔ orchestrator JSON）: + + YAML config key JSON field + ───────────────────────────────────────────── + model model_id + system_prompt system_prompt + skill_ids skill_ids + knowledge_base_ids knowledge_base_ids + endpoint agent_endpoint + api_key api_key + strategy strategy (reduce) + max_tokens max_tokens +""" +from __future__ import annotations + +import re +import json +from typing import Any + +SPEC_VERSION = "1.0" + +# YAML config → JSON field mapping +_CONFIG_MAP = { + "model": "model_id", + "system_prompt": "system_prompt", + "skill_ids": "skill_ids", + "knowledge_base_ids": "knowledge_base_ids", + "endpoint": "agent_endpoint", + "api_key": "api_key", + "strategy": "strategy", + "max_tokens": "max_tokens", + "description": "description", +} + +# reverse map: JSON → YAML config keys to nest +_REVERSE_CONFIG = { + "model_id", "system_prompt", "skill_ids", "knowledge_base_ids", + "agent_endpoint", "api_key", "strategy", "max_tokens", +} + +VALID_NODE_TYPES = {"llm", "local", "external", "reduce"} +VALID_ON_ERROR = {"halt", "continue"} +VALID_REDUCE_STRATEGY = {"concat", "first", "last"} + + +# ── validation ────────────────────────────────────────────────────────────── + +def validate_yaml(data: Any) -> dict: + """校验 YAML 工作流结构，返回 {ok, errors:[...], warnings:[...]}。 + + data 应为解析后的 YAML 字典（由调用方用 yaml.safe_load 解析）。 + """ + errors: list = [] + warnings: list = [] + + if not isinstance(data, dict): + return {"ok": False, "errors": ["YAML 顶层必须为对象"], "warnings": []} + + version = data.get("version") + if not version or version != SPEC_VERSION: + warnings.append(f"version 应为 \"{SPEC_VERSION}\"，当前为 \"{version}\"") + + nodes = data.get("nodes") + if not isinstance(nodes, list) or not nodes: + errors.append("nodes 必须为非空数组") + return {"ok": False, "errors": errors, "warnings": warnings} + + node_ids: set = set() + for i, node in enumerate(nodes): + if not isinstance(node, dict): + errors.append(f"节点 #{i} 必须为对象") + continue + nid = node.get("id") + if not nid or not isinstance(nid, str): + errors.append(f"节点 #{i} 缺少字符串 id") + continue + if nid in node_ids: + errors.append(f"节点 id 重复: {nid}") + node_ids.add(nid) + + ntype = node.get("type") + if ntype not in VALID_NODE_TYPES: + errors.append(f"节点 {nid} 类型非法: {ntype} (允许 {', '.join(sorted(VALID_NODE_TYPES))})") + + on_error = node.get("on_error", "continue") + if on_error and on_error not in VALID_ON_ERROR: + errors.append(f"节点 {nid} on_error 非法: {on_error}") + + # type-specific checks + if ntype in ("local", "external"): + if not node.get("agent_id"): + errors.append(f"节点 {nid} 类型 {ntype} 必须指定 agent_id") + + # config validation + cfg = node.get("config", {}) + if isinstance(cfg, dict): + strategy = cfg.get("strategy") + if strategy and strategy not in VALID_REDUCE_STRATEGY: + errors.append(f"节点 {nid} config.strategy 非法: {strategy}") + + edges = data.get("edges", []) + if not isinstance(edges, list): + errors.append("edges 必须为数组") + else: + for i, edge in enumerate(edges): + if not isinstance(edge, dict): + errors.append(f"边 #{i} 必须为对象") + continue + f = edge.get("from") + t = edge.get("to") + if not f or not t: + errors.append(f"边 #{i} 缺少 from/to") + if f and f not in node_ids: + errors.append(f"边 #{i} from={f} 指向不存在的节点") + if t and t not in node_ids: + errors.append(f"边 #{i} to={t} 指向不存在的节点") + if f and t and f == t: + errors.append(f"边 #{i} 存在自环: {f}") + + output_node = data.get("output_node") + if output_node and output_node not in node_ids: + warnings.append(f"output_node={output_node} 不存在于节点列表中，将自动检测出度为零的节点") + + # cycle detection (topological sort) + if node_ids and not errors: + indeg = {nid: 0 for nid in node_ids} + adj = {nid: [] for nid in node_ids} + for e in edges: + if isinstance(e, dict): + f, t = e.get("from"), e.get("to") + if f in adj and t in adj: + adj[f].append(t) + indeg[t] += 1 + queue = [nid for nid in node_ids if indeg[nid] == 0] + processed = 0 + while queue: + n = queue.pop(0) + processed += 1 + for nxt in adj[n]: + indeg[nxt] -= 1 + if indeg[nxt] == 0: + queue.append(nxt) + if processed != len(node_ids): + errors.append("检测到环路，无法进行拓扑排序") + + return { + "ok": len(errors) == 0, + "errors": errors, + "warnings": warnings, + } + + +# ── YAML → orchestrator JSON ───────────────────────────────────────────────── + +def yaml_to_orchestration(data: dict) -> dict: + """将 YAML 工作流对象转换为 orchestrator.run_graph() 接受的 JSON 格式。""" + result: dict = {"nodes": [], "edges": []} + + result["name"] = data.get("name") + result["description"] = data.get("description") + if data.get("output_node"): + result["output_node"] = data["output_node"] + + for raw_node in data.get("nodes", []): + if not isinstance(raw_node, dict): + continue + node: dict = { + "id": raw_node.get("id", ""), + "type": raw_node.get("type", "llm"), + "description": raw_node.get("description", ""), + "on_error": raw_node.get("on_error", "continue"), + } + + # agent_id for local/external + if raw_node.get("agent_id"): + node["agent_id"] = raw_node["agent_id"] + + # flatten config into top-level keys + config = raw_node.get("config", {}) + if isinstance(config, dict): + for yaml_key, json_key in _CONFIG_MAP.items(): + if yaml_key in config and config[yaml_key] is not None: + node[json_key] = config[yaml_key] + + # also carry over any extra keys from raw_node (preserve flexibility) + for k, v in raw_node.items(): + if k not in ("id", "type", "config", "description", "on_error", "agent_id"): + if k in ("skill_ids", "knowledge_base_ids", "model_id", "system_prompt", + "agent_endpoint", "api_key", "strategy", "max_tokens"): + node[k] = v + + result["nodes"].append(node) + + for edge in data.get("edges", []): + if isinstance(edge, dict): + result["edges"].append({ + "from": edge.get("from", ""), + "to": edge.get("to", ""), + }) + + return result + + +# ── orchestrator JSON → YAML ───────────────────────────────────────────────── + +def orchestration_to_yaml(data: dict) -> dict: + """将 orchestrator JSON 转换为 YAML 规范对象（可再 yaml.dump 输出）。""" + result: dict = {"version": SPEC_VERSION} + + if data.get("name"): + result["name"] = data["name"] + if data.get("description"): + result["description"] = data["description"] + if data.get("output_node"): + result["output_node"] = data["output_node"] + + result["nodes"] = [] + for json_node in data.get("nodes", []): + if not isinstance(json_node, dict): + continue + node: dict = { + "id": json_node.get("id", ""), + "type": json_node.get("type", "llm"), + } + if json_node.get("description"): + node["description"] = json_node["description"] + if json_node.get("on_error"): + node["on_error"] = json_node["on_error"] + if json_node.get("agent_id"): + node["agent_id"] = json_node["agent_id"] + + # nest known config keys into config sub-object + config: dict = {} + for yaml_key, json_key in _CONFIG_MAP.items(): + if json_key in json_node and json_node[json_key] is not None: + config[yaml_key] = json_node[json_key] + + # also preserve any extra keys not in the map + for k, v in json_node.items(): + if k not in ("id", "type", "description", "on_error", "agent_id") \ + and k not in _REVERSE_CONFIG: + if v is not None: + config[k] = v + + if config: + node["config"] = config + + result["nodes"].append(node) + + result["edges"] = [] + for edge in data.get("edges", []): + if isinstance(edge, dict): + result["edges"].append({ + "from": edge.get("from", ""), + "to": edge.get("to", ""), + }) + + return result + + +# ── YAML 模板参数替换 ──────────────────────────────────────────────────────── + +def apply_template_params(yaml_text: str, params: dict) -> str: + """替换 YAML 模板中的 ${param} 占位符。""" + def _replacer(m: re.Match) -> str: + key = m.group(1) + return str(params.get(key, m.group(0))) + return re.sub(r"\$\{(\w+)\}", _replacer, yaml_text) diff --git a/backend/python/app/workflow_templates/code-review.yaml b/backend/python/app/workflow_templates/code-review.yaml new file mode 100644 index 0000000..b077d39 --- /dev/null +++ b/backend/python/app/workflow_templates/code-review.yaml @@ -0,0 +1,35 @@ +version: "1.0" +name: "Code Review Pipeline" +description: "Generate code → review → merge. A standard code-review workflow." +category: development + +nodes: + - id: code_gen + type: llm + description: "Generate code based on requirements" + on_error: halt + config: + model: auto + system_prompt: "You are a senior software engineer. Write clean, production-ready code based on the user's requirements. Include comments and error handling." + + - id: review + type: llm + description: "Review generated code for bugs and style" + on_error: continue + config: + model: auto + system_prompt: "You are a code reviewer. Analyze the provided code for bugs, style issues, performance problems, and security vulnerabilities. Provide specific, actionable feedback." + + - id: merge + type: reduce + description: "Combine code with review feedback" + config: + strategy: concat + +edges: + - from: code_gen + to: review + - from: code_gen + to: merge + - from: review + to: merge diff --git a/backend/python/app/workflow_templates/multi-agent-debate.yaml b/backend/python/app/workflow_templates/multi-agent-debate.yaml new file mode 100644 index 0000000..07bb7bf --- /dev/null +++ b/backend/python/app/workflow_templates/multi-agent-debate.yaml @@ -0,0 +1,35 @@ +version: "1.0" +name: "Multi-Agent Debate" +description: "Two agents debate a topic from opposing perspectives, then a judge agent synthesizes the conclusion." +category: creativity + +nodes: + - id: proponent + type: llm + description: "Argue in favor of the proposition" + on_error: continue + config: + model: auto + system_prompt: "You are a skilled debater. Argue strongly IN FAVOR of the given proposition. Use logic, evidence, and persuasive reasoning. Be passionate but fair." + + - id: opponent + type: llm + description: "Argue against the proposition" + on_error: continue + config: + model: auto + system_prompt: "You are a skilled debater. Argue strongly AGAINST the given proposition. Use logic, evidence, and persuasive reasoning to find flaws in the proponent's case." + + - id: judge + type: llm + description: "Evaluate both sides and reach a conclusion" + on_error: continue + config: + model: auto + system_prompt: "You are an impartial judge. Evaluate both arguments fairly. Identify the strongest points from each side. Provide a balanced conclusion and recommendation." + +edges: + - from: proponent + to: judge + - from: opponent + to: judge diff --git a/backend/python/app/workflow_templates/rag-search.yaml b/backend/python/app/workflow_templates/rag-search.yaml new file mode 100644 index 0000000..db8c297 --- /dev/null +++ b/backend/python/app/workflow_templates/rag-search.yaml @@ -0,0 +1,25 @@ +version: "1.0" +name: "RAG Search Pipeline" +description: "Search knowledge base → augment with context → generate answer. Standard RAG pipeline." +category: productivity + +nodes: + - id: search + type: llm + description: "Generate search queries and retrieve relevant context" + on_error: continue + config: + model: auto + system_prompt: "You are a search specialist. For the given question, identify 2-3 key search points. Then search the provided knowledge base for relevant information. Return the most relevant passages with source references." + + - id: answer + type: llm + description: "Generate final answer with citations" + on_error: continue + config: + model: auto + system_prompt: "You are a knowledge assistant. Using the search results provided, generate a comprehensive answer. Always cite your sources. If information is insufficient, clearly state what is missing." + +edges: + - from: search + to: answer diff --git a/backend/python/app/workflow_templates/research-assistant.yaml b/backend/python/app/workflow_templates/research-assistant.yaml new file mode 100644 index 0000000..73b9b54 --- /dev/null +++ b/backend/python/app/workflow_templates/research-assistant.yaml @@ -0,0 +1,25 @@ +version: "1.0" +name: "Research & Answer" +description: "Research topic → generate answer. Uses two LLM nodes for research and synthesis." +category: productivity + +nodes: + - id: research + type: llm + description: "Search and gather information on the topic" + on_error: continue + config: + model: auto + system_prompt: "You are a thorough researcher. Search for key facts, data points, and relevant context about the given topic. Provide a comprehensive summary with citations." + + - id: answer + type: llm + description: "Synthesize research into final answer" + on_error: continue + config: + model: auto + system_prompt: "You are a professional writer. Based on the research summary provided, craft a clear, well-structured answer for the user. Include key findings and actionable insights." + +edges: + - from: research + to: answer diff --git a/backend/python/tests/test_42_43.py b/backend/python/tests/test_42_43.py new file mode 100644 index 0000000..9593571 --- /dev/null +++ b/backend/python/tests/test_42_43.py @@ -0,0 +1,92 @@ +"""Phase 4.2+4.3: YAML workflow + ACA protocol registry tests.""" +import pytest +from app.core.yaml_workflow import validate_yaml, yaml_to_orchestration, orchestration_to_yaml + + +# ── YAML workflow tests ────────────────────────────────────────────────────── + +VALID_YAML = """ +version: "1.0" +name: "Test" +nodes: + - id: a + type: llm + config: + model: auto + system_prompt: "You are a helper." + - id: b + type: llm + - id: c + type: reduce + config: + strategy: concat +edges: + - from: a + to: b + - from: b + to: c + - from: a + to: c +""" + + +def test_validate_valid(): + import yaml + data = yaml.safe_load(VALID_YAML) + r = validate_yaml(data) + assert r["ok"], r["errors"] + assert r["errors"] == [] + + +def test_validate_invalid_type(): + import yaml + data = yaml.safe_load("version: '1.0'\nnodes:\n - id: x\n type: bogus") + r = validate_yaml(data) + assert not r["ok"] + assert any("bogus" in e for e in r["errors"]) + + +def test_validate_cycle(): + import yaml + data = yaml.safe_load("version: '1.0'\nnodes:\n - id: a\n type: llm\n - id: b\n type: llm\nedges:\n - from: a\n to: b\n - from: b\n to: a") + r = validate_yaml(data) + assert not r["ok"] + assert any("环" in e for e in r["errors"]) + + +def test_yaml_to_json_roundtrip(): + import yaml + data = yaml.safe_load(VALID_YAML) + orch = yaml_to_orchestration(data) + assert len(orch["nodes"]) == 3 + assert len(orch["edges"]) == 3 + # config flattening + node_a = next(n for n in orch["nodes"] if n["id"] == "a") + assert node_a.get("model_id") == "auto" + assert node_a.get("system_prompt") == "You are a helper." + + +def test_json_to_yaml_to_json(): + """Roundtrip: JSON → YAML obj → JSON should preserve key fields.""" + orch_json = { + "nodes": [ + {"id": "x", "type": "llm", "model_id": "auto"}, + {"id": "y", "type": "reduce", "strategy": "concat"}, + ], + "edges": [{"from": "x", "to": "y"}], + } + yaml_obj = orchestration_to_yaml(orch_json) + assert yaml_obj["version"] == "1.0" + # config nesting + node_x = next(n for n in yaml_obj["nodes"] if n["id"] == "x") + assert node_x.get("config", {}).get("model") == "auto" + node_y = next(n for n in yaml_obj["nodes"] if n["id"] == "y") + assert node_y.get("config", {}).get("strategy") == "concat" + + +def test_validate_missing_agent_id(): + import yaml + data = yaml.safe_load("version: '1.0'\nnodes:\n - id: x\n type: local\n - id: y\n type: llm\nedges:\n - from: x\n to: y") + r = validate_yaml(data) + assert not r["ok"] + assert any("agent_id" in e for e in r["errors"])