96 lines
2.7 KiB
Python
96 lines
2.7 KiB
Python
"""将洛书规划结果映射为通用 Agent 任务，并写入 Run 事件。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from app.modules.agent import AgentTask, Plan, PlanStep
|
|
from app.modules.planner import DeliberativePlan
|
|
|
|
|
|
def task_from_deliberative_plan(
|
|
plan: DeliberativePlan,
|
|
*,
|
|
run_id: str,
|
|
agent_name: str,
|
|
objective: str,
|
|
) -> AgentTask:
|
|
"""把一次规划结果转换为仅在当前 Run 存活的 AgentTask。"""
|
|
return AgentTask(
|
|
id=f"task_{run_id}",
|
|
objective=objective,
|
|
agent_name=agent_name,
|
|
run_id=run_id,
|
|
status="running",
|
|
plan=Plan(
|
|
steps=tuple(
|
|
PlanStep(
|
|
id=str(step.id),
|
|
title=step.title,
|
|
description=step.description,
|
|
tool=step.tool,
|
|
risk_level=step.risk_level,
|
|
)
|
|
for step in plan.steps
|
|
),
|
|
assumptions=tuple(plan.assumptions),
|
|
risks=tuple(plan.risks),
|
|
status="running",
|
|
),
|
|
)
|
|
|
|
|
|
def next_step_for_tool(task: AgentTask, tool_name: str, completed_step_ids: set[str]) -> PlanStep | None:
|
|
"""返回尚未处理且明确绑定到工具的首个计划步骤。"""
|
|
if task.plan is None:
|
|
return None
|
|
return next(
|
|
(
|
|
step
|
|
for step in task.plan.steps
|
|
if step.id not in completed_step_ids and step.tool == tool_name
|
|
),
|
|
None,
|
|
)
|
|
|
|
|
|
def step_event_payload(
|
|
task: AgentTask,
|
|
step: PlanStep,
|
|
*,
|
|
status: str,
|
|
result_summary: str | None = None,
|
|
) -> dict[str, object]:
|
|
"""生成计划步骤生命周期事件 payload。"""
|
|
return {
|
|
"task_id": task.id,
|
|
"step_id": step.id,
|
|
"title": step.title,
|
|
"tool": step.tool,
|
|
"risk_level": step.risk_level,
|
|
"status": status,
|
|
**({"result_summary": result_summary} if result_summary else {}),
|
|
}
|
|
|
|
|
|
def task_plan_payload(task: AgentTask) -> dict[str, object]:
|
|
"""生成安全且可直接展示的计划事件 payload。"""
|
|
assert task.plan is not None
|
|
return {
|
|
"task_id": task.id,
|
|
"agent_name": task.agent_name,
|
|
"objective": task.objective,
|
|
"status": task.plan.status,
|
|
"assumptions": list(task.plan.assumptions),
|
|
"risks": list(task.plan.risks),
|
|
"steps": [
|
|
{
|
|
"id": step.id,
|
|
"title": step.title,
|
|
"description": step.description,
|
|
"tool": step.tool,
|
|
"risk_level": step.risk_level,
|
|
"status": step.status,
|
|
}
|
|
for step in task.plan.steps
|
|
],
|
|
}
|