feat: Phase 2 - Orchestrator workflow node + tool-level human approval

2.1 Orchestrator in workflow:
- New run_orchestrator_node() in workflow_integration.py loads agents from DB,
  supports route/sequential/debate/pipeline modes
- New 'orchestrator' node type in workflow_engine.py execute_node dispatch

2.2 Tool-level human approval:
- AgentToolConfig extended with require_approval, approval_timeout_ms,
  approval_default fields
- New ApprovalManager (approval_manager.py) with asyncio.Event-based
  create/wait_for_decision/resolve pattern
- AgentRuntime run() and run_stream() intercept tool execution,
  wait for approval decision before executing
- New POST /api/v1/approval/{id}/resolve REST endpoint
- Frontend: approval_required SSE event handling, approval dialog UI
  with approve/deny/skip buttons

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
renjianbo
2026-05-04 23:17:59 +08:00
parent d895922438
commit f3cb35c460
8 changed files with 437 additions and 1 deletions

View File

@@ -129,3 +129,163 @@ async def run_agent_node(
"status": "error",
"error": result.error,
}
async def run_orchestrator_node(
node_data: Dict[str, Any],
input_data: Dict[str, Any],
execution_logger: Optional[Any] = None,
user_id: Optional[str] = None,
on_tool_executed: Optional[Any] = None,
on_llm_invocation: Optional[Any] = None,
) -> Dict[str, Any]:
"""
在工作流中执行多 Agent 编排节点。
node_data 支持的字段:
mode — "route" | "sequential" | "debate" | "pipeline"
agents — Agent ID 列表(必填,至少 2 个)
routing_prompt — route 模式的路由指令(可选)
aggregation_prompt— debate 模式的汇总指令(可选)
model — 覆盖各 Agent 的模型(可选)
provider — 覆盖各 Agent 的提供商(可选)
temperature — 覆盖各 Agent 的温度(可选)
max_iterations — 覆盖各 Agent 的最大步数(可选)
input_data 中的 "query""input" 字段作为用户输入。
"""
from sqlalchemy.orm import Session
from app.core.database import SessionLocal
from app.models.agent import Agent
from app.agent_runtime.orchestrator import (
AgentOrchestrator,
OrchestratorAgentConfig,
)
# 1. 解析输入
query = (
input_data.get("query")
or input_data.get("input")
or input_data.get("text", "")
)
if not isinstance(query, str):
query = str(query) if query else ""
if not query:
return {"output": "错误Orchestrator 节点未收到用户输入", "status": "error"}
# 2. 解析编排模式
mode = node_data.get("mode", "debate").lower()
if mode not in ("route", "sequential", "debate", "pipeline"):
return {"output": f"错误:不支持的编排模式 '{mode}',可选: route, sequential, debate, pipeline", "status": "error"}
# 3. 解析 Agent 列表
agent_ids = node_data.get("agents", [])
if not agent_ids or not isinstance(agent_ids, list) or len(agent_ids) < 1:
return {"output": "错误Orchestrator 节点需要至少 1 个 Agent", "status": "error"}
# 4. 从 DB 加载 Agent 配置
db: Optional[Session] = None
try:
db = SessionLocal()
# 覆盖配置(可选)
override_model = node_data.get("model")
override_provider = node_data.get("provider")
override_temperature = node_data.get("temperature")
override_max_iterations = node_data.get("max_iterations")
agent_configs: list = []
for agent_id in agent_ids:
agent = db.query(Agent).filter(Agent.id == agent_id).first()
if not agent:
logger.warning("Orchestrator: Agent %s 不存在,跳过", agent_id)
continue
# 从 workflow_config 提取 Agent 的 LLM 配置
wc = agent.workflow_config or {}
nodes = wc.get("nodes", [])
system_prompt = agent.description or ""
model = override_model or "deepseek-v4-flash"
provider = override_provider or "deepseek"
temperature = float(override_temperature) if override_temperature else 0.7
max_iterations = int(override_max_iterations) if override_max_iterations else 10
tools_whitelist: list = []
for n in nodes:
if n.get("type") not in ("agent", "llm", "template"):
continue
cfg = n.get("data", {}) if isinstance(n, dict) else getattr(n, "data", {})
system_prompt = cfg.get("system_prompt", "") or system_prompt
if not override_model:
model = cfg.get("model", model)
if not override_provider:
provider = cfg.get("provider", provider)
if not override_temperature:
temperature = float(cfg.get("temperature", temperature))
if not override_max_iterations:
max_iterations = int(cfg.get("max_iterations", max_iterations))
tools_whitelist = cfg.get("tools", tools_whitelist)
break
agent_configs.append(OrchestratorAgentConfig(
id=agent.id,
name=agent.name or "Agent",
system_prompt=system_prompt,
model=model,
provider=provider,
temperature=temperature,
max_iterations=max_iterations,
tools=tools_whitelist,
description=agent.description or "",
))
if not agent_configs:
return {"output": "错误:没有找到可用的 Agent", "status": "error"}
# 5. 创建 Orchestrator 并执行
orchestrator = AgentOrchestrator(
default_llm_config=AgentLLMConfig(
model=override_model or "deepseek-v4-flash",
temperature=0.3,
),
)
result = await orchestrator.run(
mode=mode,
question=query,
agents=agent_configs,
on_llm_call=on_llm_invocation,
)
# 6. 返回结构化结果
return {
"output": result.final_answer,
"status": "success",
"orchestrator_meta": {
"mode": result.mode,
"agent_count": len(agent_configs),
"steps": [
{
"agent_id": s.agent_id,
"agent_name": s.agent_name,
"input": s.input[:200] if s.input else "",
"output": s.output[:500] if s.output else "",
"iterations_used": s.iterations_used,
"tool_calls_made": s.tool_calls_made,
"error": s.error,
}
for s in result.steps
],
},
}
except Exception as e:
logger.error("Orchestrator 节点执行失败: %s", e, exc_info=True)
return {
"output": None,
"status": "failed",
"error": f"Orchestrator 执行失败: {e}",
}
finally:
if db:
db.close()