feat: 修复了一些bug

This commit is contained in:
朝夕 2026-04-28 21:04:40 +08:00
parent 689b7dc7f1
commit 478b0813c1
5 changed files with 26 additions and 14 deletions

View File

@ -20,6 +20,12 @@ import asyncio
workflow_router = APIRouter(prefix="/api/v1/workflow", tags=["workflow"])
@workflow_router.get("/list")
async def get_workflow_list():
global_state_machine = ray_actor_hook("global_state_machine").global_state_machine
events = await global_state_machine.list_events.remote()
return events
@workflow_router.get("/sse/{trace_id}")
async def get_workflow_sse(trace_id: str, request: Request):
global_state_machine = ray_actor_hook("global_state_machine").global_state_machine

View File

@ -139,6 +139,20 @@ class GlobalStateMachine:
def get_workflow(self, trace_id: str) -> PretorWorkflow:
return self.event_dict[trace_id].workflow
def list_events(self) -> list[dict]:
result = []
for trace_id, event in self.event_dict.items():
workflow_title = event.workflow.title if event.workflow else None
workflow_status = event.workflow.status.status if event.workflow and event.workflow.status else None
result.append({
"event_id": trace_id,
"workflow_title": workflow_title,
"status": workflow_status,
"user_name": event.user_name,
"message": event.message,
})
return result
async def put_pending(self, trace_id, item) -> None:
await self.event_dict[trace_id].pending_queue.put(item)

View File

@ -80,7 +80,7 @@ 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 为 composite_individual 或 primary_individual并将 agent_id 设置为对应的 Skill Individual 名称),作为可调用的工具。\n"
prompt += "你可以直接将以下 Skill Individual 安排进工作流的步骤中(设置 node 为 skill_individual并将 agent_id 设置为对应的 Skill Individual 名称),作为可调用的工具。\n"
for skill in ctx.deps.available_skills:
prompt += f"- 名称: {skill['name']}\n 描述: {skill['description']}\n"

View File

@ -18,8 +18,7 @@ from pydantic import BaseModel, Field, model_validator
from pretor.utils.logger import get_logger
logger = get_logger('workflow')
NodeType = Literal[
"consciousness_node", "control_node", "supervisory_node",
"composite_individual", "primary_individual"
"consciousness_node", "control_node", "supervisory_node", "skill_individual"
]
class EventInfo(BaseModel):
@ -38,17 +37,13 @@ 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 名称")
logic_gate: Optional[LogicGate] = Field(default=None, description="逻辑跳转控制")
status: Literal["waiting", "running", "completed", "failed"] = Field(
default="waiting",
description="执行状态 (LLM建议保留默认值)"
)
class WorkerGroup(BaseModel):
name: str = Field(..., description="工作组名称,如 'coding_squad'")
primary_individual: Dict[str, int] = Field(..., description="基础子个体配置,例如 {'coder': 2, 'tester': 1}")
composite_individual: Dict[str, int] = Field(..., description="复合子个体配置,例如 {'code_reviewer': 1}")
class WorkflowStatus(BaseModel):
step: int = Field(default=1, gt=0, description="当前运行到的工作流步数")
@ -59,7 +54,6 @@ class WorkflowStatus(BaseModel):
class PretorWorkflow(BaseModel):
title: str = Field(..., description="工作流的标题")
workgroup_list: List[WorkerGroup] = Field(..., description="工作组资源编排列表")
work_link: List[WorkStep] = Field(..., description="工作链逻辑定义")
# ---------------- 以下为系统级管控字段LLM 无需关心 ---------------- #
trace_id: str | None = Field(description="系统自动生成的追溯ID")

View File

@ -199,15 +199,13 @@ class WorkflowEngine:
return result_obj.output, True
return result_obj, True
elif step.node in ["primary_individual", "composite_individual"]:
self.logger.info(f"正在通过 WorkerCluster 调度 {step.node} {step.action} 动作")
elif step.node == "skill_individual":
self.logger.info(f"正在通过 WorkerCluster 调度 skill_individual 执行 {step.action}")
try:
from pretor.utils.ray_hook import ray_actor_hook
worker_cluster = ray_actor_hook("worker_cluster").worker_cluster
task_id = f"{self.workflow.trace_id}_step_{step.step}"
agent_id = getattr(step, 'agent_id', f"default_{step.node}")
if isinstance(input_data, dict) and "agent_id" in input_data:
agent_id = input_data.get("agent_id")
agent_id = step.agent_id or f"default_{step.node}"
task_event = {
"action": step.action,
"description": step.desc,