78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from runtime.executor import Executor, ExecutorConfig, StepResult, ToolCallable
|
|
from runtime.planner import PlannedStep
|
|
from skills_learning.models import FlowTemplateSkill
|
|
|
|
|
|
class SkillExecutionError(Exception):
|
|
pass
|
|
|
|
|
|
def validate_skill_args(skill: FlowTemplateSkill, args: dict[str, Any]) -> None:
|
|
if skill.metadata.kind != "flow_template":
|
|
raise SkillExecutionError(f"skill {skill.id} is not a flow_template skill")
|
|
required = set(skill.parameters)
|
|
provided = set(args)
|
|
missing = sorted(required - provided)
|
|
extra = sorted(provided - required)
|
|
errors: list[str] = []
|
|
if missing:
|
|
errors.append(f"missing required parameters: {', '.join(missing)}")
|
|
if extra:
|
|
errors.append(f"unrecognized parameters: {', '.join(extra)}")
|
|
if errors:
|
|
raise SkillExecutionError("; ".join(errors))
|
|
|
|
|
|
def resolve_skill_steps(
|
|
skill: FlowTemplateSkill,
|
|
args: dict[str, Any],
|
|
) -> list[dict[str, Any]]:
|
|
validate_skill_args(skill, args)
|
|
return [
|
|
{
|
|
"tool_name": step.tool_name,
|
|
"args": _resolve_value(step.args, args),
|
|
}
|
|
for step in skill.steps
|
|
]
|
|
|
|
|
|
def run_flow_template_skill(
|
|
skill: FlowTemplateSkill,
|
|
args: dict[str, Any],
|
|
*,
|
|
tools: dict[str, ToolCallable] | None = None,
|
|
) -> list[StepResult]:
|
|
resolved_steps = resolve_skill_steps(skill, args)
|
|
executor = Executor(
|
|
tools=tools,
|
|
config=ExecutorConfig(max_retries=1, backoff_seconds=0),
|
|
)
|
|
results: list[StepResult] = []
|
|
for index, resolved in enumerate(resolved_steps, start=1):
|
|
step = PlannedStep(
|
|
action=resolved["tool_name"],
|
|
description=f"Run skill {skill.name} step {index}",
|
|
args=resolved["args"],
|
|
)
|
|
result = executor.execute(step)
|
|
results.append(result)
|
|
if not result.success:
|
|
break
|
|
return results
|
|
|
|
|
|
def _resolve_value(value: Any, args: dict[str, Any]) -> Any:
|
|
if isinstance(value, str) and value.startswith("{") and value.endswith("}"):
|
|
name = value[1:-1]
|
|
return args[name]
|
|
if isinstance(value, dict):
|
|
return {key: _resolve_value(nested, args) for key, nested in value.items()}
|
|
if isinstance(value, list):
|
|
return [_resolve_value(item, args) for item in value]
|
|
return value
|