Fix workflow scheduling for Skill Individuals and LeftPanel polling (#57)
- Update `workflow_runner.py` to retrieve `Skill Individual` items with their real `agent_id` rather than names, preventing assignment mismatches when routing via `WorkerCluster`. - Update `ConsciousnessNode` prompts to strictly instruct the LLM to output real `agent_id`s in `agent_id` workflow fields. - Update `WorkStep` schemas to reflect the requirement of `agent_id`. - Add a 2-second polling interval in the frontend `LeftPanel.tsx` to automatically update the workflow list state rather than fetching only once upon tab selection. Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com> Co-authored-by: zhaoxi826 <198742034+zhaoxi826@users.noreply.github.com>
This commit is contained in:
GitHub
parent
b20beb23c0
commit
d713bd1b30
|
|
@ -40,9 +40,10 @@ export function LeftPanel({ activeTab, setActiveTab, selectedWorkflow, setSelect
|
|||
const memPercent = totalMemory > 0 ? (usedMemory / totalMemory) * 100 : 0;
|
||||
|
||||
useEffect(() => {
|
||||
if (activeTab === 'workflows') {
|
||||
const fetchWorkflows = async () => {
|
||||
setLoadingWorkflows(true);
|
||||
let intervalId: ReturnType<typeof setInterval>;
|
||||
|
||||
const fetchWorkflows = async (isInitial = false) => {
|
||||
if (isInitial) setLoadingWorkflows(true);
|
||||
try {
|
||||
const response = await apiClient.get('/api/v1/workflow/list');
|
||||
// Fallback parsing just in case it returns an object or array
|
||||
|
|
@ -59,11 +60,18 @@ export function LeftPanel({ activeTab, setActiveTab, selectedWorkflow, setSelect
|
|||
console.error("Failed to fetch workflows", error);
|
||||
setWorkflows([]);
|
||||
} finally {
|
||||
setLoadingWorkflows(false);
|
||||
if (isInitial) setLoadingWorkflows(false);
|
||||
}
|
||||
};
|
||||
fetchWorkflows();
|
||||
|
||||
if (activeTab === 'workflows') {
|
||||
fetchWorkflows(true);
|
||||
intervalId = setInterval(() => fetchWorkflows(false), 2000);
|
||||
}
|
||||
|
||||
return () => {
|
||||
if (intervalId) clearInterval(intervalId);
|
||||
};
|
||||
}, [activeTab]);
|
||||
|
||||
return (
|
||||
|
|
|
|||
|
|
@ -80,9 +80,9 @@ class ConsciousnessNode:
|
|||
prompt += f"- 选定工作流模板 (Workflow Template): {ctx.deps.workflow_template}\n"
|
||||
if ctx.deps.available_skills:
|
||||
prompt += "\n=== 当前可用 Skill Individual ===\n"
|
||||
prompt += "你可以直接将以下 Skill Individual 安排进工作流的步骤中（设置 node 为 skill_individual，并将 agent_id 设置为对应的 Skill Individual 名称），作为可调用的工具。\n"
|
||||
prompt += "你可以直接将以下 Skill Individual 安排进工作流的步骤中（设置 node 为 skill_individual，并将 agent_id 设置为对应 Skill Individual 的真实 agent_id，不要用名称！），作为可调用的工具。\n"
|
||||
for skill in ctx.deps.available_skills:
|
||||
prompt += f"- 名称: {skill['name']}\n 描述: {skill['description']}\n"
|
||||
prompt += f"- 真实 agent_id: {skill.get('agent_id')}\n 名称: {skill['name']}\n 描述: {skill['description']}\n"
|
||||
|
||||
return prompt
|
||||
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ class WorkStep(BaseModel):
|
|||
desc: str = Field(..., description="动作细节的自然语言描述，包含人工规范指导")
|
||||
inputs: Optional[Union[str, List[str]]] = Field(default=None, description="前置依赖输出")
|
||||
outputs: Optional[str] = Field(default=None, description="当前步骤产出物变量名")
|
||||
agent_id: Optional[str] = Field(default=None, description="分配给 skill_individual 的 Skill Individual 名称")
|
||||
agent_id: Optional[str] = Field(default=None, description="分配给 skill_individual 的 Skill Individual 真实 agent_id，不可用名称代替")
|
||||
logic_gate: Optional[LogicGate] = Field(default=None, description="逻辑跳转控制")
|
||||
status: Literal["waiting", "running", "completed", "failed"] = Field(
|
||||
default="waiting",
|
||||
|
|
|
|||
|
|
@ -316,13 +316,17 @@ class WorkflowRunningEngine:
|
|||
available_skills = None
|
||||
if self.global_state_machine:
|
||||
try:
|
||||
raw_skills = await self.global_state_machine.get_skill_list.remote()
|
||||
available_skills = [
|
||||
{"name": name, "description": details[0], "instructions": details[1]}
|
||||
for name, details in raw_skills.items()
|
||||
]
|
||||
all_individuals = await self.global_state_machine.list_individuals.remote()
|
||||
available_skills = []
|
||||
for agent_id, config in all_individuals.items():
|
||||
if config.get("agent_type") == "skill_individual" or config.get("type") == "skill_individual":
|
||||
available_skills.append({
|
||||
"agent_id": agent_id,
|
||||
"name": config.get("agent_name", "Unknown"),
|
||||
"description": config.get("description", "")
|
||||
})
|
||||
except Exception as e:
|
||||
self.logger.warning(f"获取技能列表失败: {e}")
|
||||
self.logger.warning(f"获取Skill Individual列表失败: {e}")
|
||||
|
||||
payload = ForWorkflowEngineInput(
|
||||
original_command=event.message,
|
||||
|
|
|
|||
|
|
@ -60,6 +60,8 @@ async def test_workflow_engine_run():
|
|||
step1.step = 1
|
||||
step1.status = "waiting"
|
||||
step1.node = "control_node"
|
||||
step1.name = "mock_name"
|
||||
step1.desc = "mock_desc"
|
||||
step1.action = "mock_action"
|
||||
step1.inputs = []
|
||||
step1.outputs = "res"
|
||||
|
|
|
|||
|
|
@ -1,11 +1,5 @@
|
|||
import pytest
|
||||
from pretor.core.workflow.workflow import WorkerGroup, WorkStep, PretorWorkflow, WorkflowStatus, LogicGate
|
||||
|
||||
def test_worker_group():
|
||||
wg = WorkerGroup(name="group1", primary_individual={"coder": 1}, composite_individual={"tester": 1})
|
||||
assert wg.name == "group1"
|
||||
assert wg.primary_individual == {"coder": 1}
|
||||
assert wg.composite_individual == {"tester": 1}
|
||||
from pretor.core.workflow.workflow import WorkStep, PretorWorkflow, WorkflowStatus, LogicGate
|
||||
|
||||
def test_work_step():
|
||||
ws = WorkStep(
|
||||
|
|
@ -25,31 +19,27 @@ def test_work_step():
|
|||
def test_pretor_workflow_validation_success():
|
||||
ws1 = WorkStep(step=1, name="s1", node="control_node", action="a1", desc="d1")
|
||||
ws2 = WorkStep(step=2, name="s2", node="supervisory_node", action="a2", desc="d2")
|
||||
wg = WorkerGroup(name="g1", primary_individual={"coder": 1}, composite_individual={})
|
||||
wf = PretorWorkflow(title="wf1", workgroup_list=[wg], work_link=[ws1, ws2], trace_id="t", event_info={"platform":"a", "user_name":"b"})
|
||||
wf = PretorWorkflow(title="wf1", work_link=[ws1, ws2], trace_id="t", event_info={"platform":"a", "user_name":"b"})
|
||||
assert wf.title == "wf1"
|
||||
|
||||
def test_pretor_workflow_validation_error_step_discontinuous():
|
||||
ws1 = WorkStep(step=1, name="s1", node="control_node", action="a1", desc="d1")
|
||||
ws2 = WorkStep(step=3, name="s3", node="supervisory_node", action="a2", desc="d2")
|
||||
wg = WorkerGroup(name="g1", primary_individual={}, composite_individual={})
|
||||
with pytest.raises(ValueError, match="工作链步数不连续"):
|
||||
PretorWorkflow(title="wf1", workgroup_list=[wg], work_link=[ws1, ws2], trace_id="t", event_info={"platform":"a", "user_name":"b"})
|
||||
PretorWorkflow(title="wf1", work_link=[ws1, ws2], trace_id="t", event_info={"platform":"a", "user_name":"b"})
|
||||
|
||||
def test_pretor_workflow_validation_error_jump_out_of_bounds():
|
||||
lg = LogicGate(if_fail="jump_to_step_3", if_pass="continue")
|
||||
ws1 = WorkStep(step=1, name="s1", node="control_node", action="a1", desc="d1", logic_gate=lg)
|
||||
ws2 = WorkStep(step=2, name="s2", node="supervisory_node", action="a2", desc="d2")
|
||||
wg = WorkerGroup(name="g1", primary_individual={}, composite_individual={})
|
||||
with pytest.raises(ValueError, match="跳转目标 Step 3 越界了"):
|
||||
PretorWorkflow(title="wf1", workgroup_list=[wg], work_link=[ws1, ws2], trace_id="t", event_info={"platform":"a", "user_name":"b"})
|
||||
PretorWorkflow(title="wf1", work_link=[ws1, ws2], trace_id="t", event_info={"platform":"a", "user_name":"b"})
|
||||
|
||||
def test_pretor_workflow_validation_error_jump_format_error():
|
||||
lg = LogicGate(if_fail="jump_to_step_invalid", if_pass="continue")
|
||||
ws1 = WorkStep(step=1, name="s1", node="control_node", action="a1", desc="d1", logic_gate=lg)
|
||||
wg = WorkerGroup(name="g1", primary_individual={}, composite_individual={})
|
||||
with pytest.raises(ValueError, match="LogicGate 格式错误"):
|
||||
PretorWorkflow(title="wf1", workgroup_list=[wg], work_link=[ws1], trace_id="t", event_info={"platform":"a", "user_name":"b"})
|
||||
PretorWorkflow(title="wf1", work_link=[ws1], trace_id="t", event_info={"platform":"a", "user_name":"b"})
|
||||
|
||||
def test_workflow_status():
|
||||
status = WorkflowStatus()
|
||||
|
|
|
|||
Loading…
Reference in New Issue