agentscope.plan

The plan module in AgentScope.

class SubTask

Bases: BaseModel

The subtask model used in the plan module.

name: str

description: str

expected_outcome: str

outcome: str | None

state: Literal['todo', 'in_progress', 'done', 'abandoned']

created_at: str

finished_at: str | None

finish(outcome)

Finish the subtask with the actual outcome.

Parameters:

outcome (str)

Return type:

None

to_oneline_markdown()

Convert the subtask to MarkDown format.

Return type:

str

to_markdown(detailed=False)

Convert the subtask to MarkDown format.

Parameters:

detailed (bool, defaults to False) – Whether to include detailed information about the subtask.

Return type:

str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class Plan

Bases: BaseModel

The plan model used in the plan module, contains a list of subtasks.

id: str

name: str

description: str

expected_outcome: str

subtasks: list[SubTask]

created_at: str

state: Literal['todo', 'in_progress', 'done', 'abandoned']

finished_at: str | None

outcome: str | None

finish(state, outcome)

Finish the plan.

Parameters:

• state (Literal['done', 'abandoned'])

• outcome (str)

Return type:

None

to_markdown(detailed=False)

Convert the plan to MarkDown format.

Parameters:

detailed (bool)

Return type:

str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class DefaultPlanToHint

Bases: object

The default function to generate the hint message based on the current plan to guide the agent on next steps.

hint_prefix: str = '<system-hint>'

hint_suffix: str = '</system-hint>'

no_plan: str = "If the user's query is complex (e.g. programming a website, game or app), or requires a long chain of steps to complete (e.g. conduct research on a certain topic from different sources), you NEED to create a plan first by calling 'create_plan'. Otherwise, you can directly execute the user's query without planning."

at_the_beginning: str = "The current plan:\n```\n{plan}\n```\nYour options include:\n- Mark the first subtask as 'in_progress' by calling 'update_subtask_state' with subtask_idx=0 and state='in_progress', and start executing it.\n- If the first subtask is not executable, analyze why and what you can do to advance the plan, e.g. ask user for more information, revise the plan by calling 'revise_current_plan'.\n- If the user asks you to do something unrelated to the plan, prioritize the completion of user's query first, and then return to the plan afterward.\n- If the user no longer wants to perform the current plan, confirm with the user and call the 'finish_plan' function.\n"

when_a_subtask_in_progress: str = "The current plan:\n```\n{plan}\n```\nNow the subtask at index {subtask_idx}, named '{subtask_name}', is 'in_progress'. Its details are as follows:\n```\n{subtask}\n```\nYour options include:\n- Go on execute the subtask and get the outcome.\n- Call 'finish_subtask' with the specific outcome if the subtask is finished.\n- Ask the user for more information if you need.\n- Revise the plan by calling 'revise_current_plan' if necessary.\n- If the user asks you to do something unrelated to the plan, prioritize the completion of user's query first, and then return to the plan afterward."

when_no_subtask_in_progress: str = "The current plan:\n```\n{plan}\n```\nThe first {index} subtasks are done, and there is no subtask 'in_progress'. Now Your options include:\n- Mark the next subtask as 'in_progress' by calling 'update_subtask_state', and start executing it.\n- Ask the user for more information if you need.\n- Revise the plan by calling 'revise_current_plan' if necessary.\n- If the user asks you to do something unrelated to the plan, prioritize the completion of user's query first, and then return to the plan afterward."

at_the_end: str = "The current plan:\n```\n{plan}\n```\nAll the subtasks are done. Now your options are:\n- Finish the plan by calling 'finish_plan' with the specific outcome, and summarize the whole process and outcome to the user.\n- Revise the plan by calling 'revise_current_plan' if necessary.\n- If the user asks you to do something unrelated to the plan, prioritize the completion of user's query first, and then return to the plan afterward."

__call__(plan)

Generate the hint message based on the input plan to guide the agent on next steps.

Parameters:

plan (Plan | None) – The current plan, used to generate the hint message.

Returns:

The generated hint message, or None if the plan is None or there is no relevant hint.

Return type:

str | None

class PlanNotebook

Bases: StateModule

The plan notebook to manage the plan, providing hints and plan related tool functions to the agent.

description: str = "The plan-related tools. Activate this tool when you need to execute complex task, e.g. building a website or a game. Once activated, you'll enter the plan mode, where you will be guided to complete the given query by creating and following a plan, and hint message wrapped by <system-hint></system-hint> will guide you to complete the task. If you think the user no longer wants to perform the current task, you need to confirm with the user and call the 'finish_plan' function."

__init__(max_subtasks=None, plan_to_hint=None, storage=None)

Initialize the plan notebook.

Parameters:

• max_subtasks (int | None, optional) – The maximum number of subtasks in a plan.

• plan_to_hint (Callable[[Plan | None], str | None] | None, optional) – The function to generate the hint message based on the current plan. If not provided, a default DefaultPlanToHint object will be used.

• storage (PlanStorageBase | None, optional) – The plan storage. If not provided, an in-memory storage will be used.

Return type:

None

async create_plan(name, description, expected_outcome, subtasks)

Create a plan by given name and sub-tasks.

Parameters:

• name (str) – The plan name, should be concise, descriptive and not exceed 10 words.

• description (str) – The plan description, including the constraints, target and outcome to be achieved. The description should be clear, specific and concise, and all the constraints, target and outcome should be specific and measurable.

• expected_outcome (str) – The expected outcome of the plan, which should be specific, concrete and measurable.

• subtasks (list[SubTask]) – A list of sequential sub-tasks that make up the plan.

Returns:

The response of the tool call.

Return type:

ToolResponse

async revise_current_plan(subtask_idx, action, subtask=None)

Revise the current plan by adding, revising or deleting a sub-task.

Parameters:

• subtask_idx (int) – The index of the sub-task to be revised, starting from 0.

• action (Literal[“add”, “revise”, “delete”]) – The action to be performed on the sub-task. If “add”, the sub-task will be inserted before the given index. If “revise”, the sub-task at the given index will be revised. If “delete”, the sub-task at the given index will be deleted.

• subtask (SubTask | None, optional) – The sub-task to be added or revised. Required if action is “add” or “revise”.

Raises:

ValueError – If the current plan is None, ValueError will be raised.

Returns:

The response of the tool call.

Return type:

ToolResponse

async update_subtask_state(subtask_idx, state)

Update the state of a subtask by given index and state. Note if you want to mark a subtask as done, you SHOULD call finish_subtask instead with the specific outcome.

Parameters:

• subtask_idx (int) – The index of the subtask to be updated, starting from 0.

• state (Literal[“todo”, “in_progress”, “abandoned”]) – The new state of the subtask. If you want to mark a subtask as done, you SHOULD call finish_subtask instead with the specific outcome.

Return type:

ToolResponse

async finish_subtask(subtask_idx, subtask_outcome)

Label the subtask as done by given index and outcome.

Args:

subtask_idx (int):

The index of the sub-task to be marked as done, starting from 0.

subtask_outcome (str):

The specific outcome of the sub-task, should exactly match the expected outcome in the sub-task description. SHOULDN’t be what you did or general description, e.g. “I have searched xxx”, “I have written the code for xxx”, etc. It SHOULD be the specific data, information, or path to the file, e.g. “There are 5 articles about xxx, they are

• xxx

• xxx

…”

Parameters:

• subtask_idx (int)

• subtask_outcome (str)

Return type:

ToolResponse

async view_subtasks(subtask_idx)

View the details of the sub-tasks by given indexes.

Parameters:

subtask_idx (list[int]) – The indexes of the sub-tasks to be viewed, starting from 0.

Return type:

ToolResponse

async finish_plan(state, outcome)

Finish the current plan by given outcome, or abandon it with the given reason if the user no longer wants to perform it. Note that you SHOULD confirm with the user before abandoning the plan.

Parameters:

• state (Literal[“done”, “abandoned”]) – The state to finish the plan. If “done”, the plan will be marked as done with the given outcome. If “abandoned”, the plan will be abandoned with the given reason.

• outcome (str) – The specific outcome of the plan if state is “done”, or the reason for abandoning the plan if state is “abandoned”.

Return type:

ToolResponse

async view_historical_plans()

View the historical plans.

Return type:

ToolResponse

async recover_historical_plan(plan_id)

Recover a historical plan by given plan ID, the plan ID can be obtained by calling view_historical_plans. Note the recover operation will override the current plan if exists.

Parameters:

plan_id (str) – The ID of the historical plan to be recovered.

Return type:

ToolResponse

list_tools()

List all tool functions provided to agent

Returns:

A list of all tool functions provided by the plan notebook to the agent.

Return type:

list[Callable[…, ToolResponse]]

async get_current_hint()

Get the hint message based on the current plan and subtasks states. This function will call the plan_to_hint function to generate the hint message.

Returns:

The hint message wrapped by <system-hint></system-hint>, or None if there is no relevant hint.

Return type:

Msg | None

register_plan_change_hook(hook_name, hook)

Register a plan hook that will be triggered when the plan is changed.

Parameters:

• hook_name (str) – The name of the hook, should be unique.

• hook (Callable[[Plan], None]) – The hook function, which takes the current plan as input and returns nothing.

Return type:

None

remove_plan_change_hook(hook_name)

Remove a plan change hook by given name.

Parameters:

hook_name (str) – The name of the hook to be removed.

Return type:

None

class PlanStorageBase

Bases: StateModule

The base class for plan storage.

abstract async add_plan(plan)

Add a plan to the storage.

Parameters:

plan (Plan)

Return type:

None

abstract async delete_plan(plan_id)

Delete a plan from the storage.

Parameters:

plan_id (str)

Return type:

None

abstract async get_plans()

Get all plans from the storage.

Return type:

list[Plan]

abstract async get_plan(plan_id)

Get a plan by its ID.

Parameters:

plan_id (str)

Return type:

Plan | None

class InMemoryPlanStorage

Bases: PlanStorageBase

In-memory plan storage.

__init__()

Initialize the in-memory plan storage.

Return type:

None

async add_plan(plan, override=True)

Add a plan to the storage.

Parameters:

• plan (Plan) – The plan to be added.

• override (bool, defaults to True) – Whether to override the existing plan with the same ID.

Return type:

None

async delete_plan(plan_id)

Delete a plan from the storage.

Parameters:

plan_id (str) – The ID of the plan to be deleted.

Return type:

None

async get_plans()

Get all plans from the storage.

Returns:

A list of all plans in the storage.

Return type:

list[Plan]

async get_plan(plan_id)

Get a plan by its ID.

Parameters:

plan_id (str) – The ID of the plan to be retrieved.

Returns:

The plan with the specified ID, or None if not found.

Return type:

Plan | None