|
|
from pydantic import Field |
|
|
from typing import Optional, List |
|
|
|
|
|
from ..core.logging import logger |
|
|
from ..models.base_model import BaseLLM |
|
|
from .action import Action, ActionInput, ActionOutput |
|
|
from ..prompts.task_planner import TASK_PLANNING_ACTION |
|
|
from ..workflow.workflow_graph import WorkFlowNode |
|
|
|
|
|
|
|
|
class TaskPlanningInput(ActionInput): |
|
|
""" |
|
|
Input specification for the task planning action. |
|
|
""" |
|
|
goal: str = Field(description="A clear and detailed description of the user's goal, specifying what needs to be achieved.") |
|
|
history: Optional[str] = Field(default=None, description="Optional field containing previously generated task plan.") |
|
|
suggestion: Optional[str] = Field(default=None, description="Optional suggestions or ideas to guide the planning process.") |
|
|
|
|
|
|
|
|
class TaskPlanningOutput(ActionOutput): |
|
|
""" |
|
|
Output structure for the task planning action. |
|
|
""" |
|
|
sub_tasks: List[WorkFlowNode] = Field(description="A list of sub-tasks that collectively achieve user's goal.") |
|
|
|
|
|
|
|
|
class TaskPlanning(Action): |
|
|
""" |
|
|
Action for planning a series of tasks to achieve a goal. |
|
|
""" |
|
|
|
|
|
def __init__(self, **kwargs): |
|
|
|
|
|
name = kwargs.pop("name") if "name" in kwargs else TASK_PLANNING_ACTION["name"] |
|
|
description = kwargs.pop("description") if "description" in kwargs else TASK_PLANNING_ACTION["description"] |
|
|
prompt = kwargs.pop("prompt") if "prompt" in kwargs else TASK_PLANNING_ACTION["prompt"] |
|
|
|
|
|
|
|
|
inputs_format = kwargs.pop("inputs_format", None) or TaskPlanningInput |
|
|
outputs_format = kwargs.pop("outputs_format", None) or TaskPlanningOutput |
|
|
super().__init__(name=name, description=description, prompt=prompt, inputs_format=inputs_format, outputs_format=outputs_format, **kwargs) |
|
|
|
|
|
def execute(self, llm: Optional[BaseLLM] = None, inputs: Optional[dict] = None, sys_msg: Optional[str]=None, return_prompt: bool = False, **kwargs) -> TaskPlanningOutput: |
|
|
"""Execute the task planning process. |
|
|
|
|
|
This method uses the provided language model to generate a structured |
|
|
plan of sub-tasks based on the user's goal and any additional context. |
|
|
|
|
|
Args: |
|
|
llm: The language model to use for planning. |
|
|
inputs: Input data containing the goal and optional context. |
|
|
sys_msg: Optional system message for the language model. |
|
|
return_prompt: Whether to return both the task plan and the prompt used. |
|
|
**kwargs: Additional keyword arguments. |
|
|
|
|
|
Returns: |
|
|
If return_prompt is False (default): The generated task plan. |
|
|
If return_prompt is True: A tuple of (task plan, prompt used). |
|
|
|
|
|
Raises: |
|
|
ValueError: If the inputs are None or empty. |
|
|
""" |
|
|
if not inputs: |
|
|
logger.error("TaskPlanning action received invalid `inputs`: None or empty.") |
|
|
raise ValueError('The `inputs` to TaskPlanning action is None or empty.') |
|
|
|
|
|
prompt_params_names = ["goal", "history", "suggestion"] |
|
|
prompt_params_values = {param: inputs.get(param, "") for param in prompt_params_names} |
|
|
prompt = self.prompt.format(**prompt_params_values) |
|
|
task_plan = llm.generate( |
|
|
prompt = prompt, |
|
|
system_message = sys_msg, |
|
|
parser=self.outputs_format, |
|
|
parse_mode="json" |
|
|
) |
|
|
|
|
|
if return_prompt: |
|
|
return task_plan, prompt |
|
|
|
|
|
return task_plan |
