Skip to main content

portia.portia

Portia classes that plan and execute runs for queries.

This module contains the core classes responsible for generating, managing, and executing plans in response to queries. The Portia class serves as the main entry point, orchestrating the planning and execution process. It uses various agents and tools to carry out tasks step by step, saving the state of the run at each stage. It also handles error cases, clarification requests, and run state transitions.

The Portia class provides methods to:

  • Generate a plan for executing a query.
  • Create and manage runs.
  • Execute runs step by step, using agents to handle the execution of tasks.
  • Resolve clarifications required during the execution of runs.
  • Wait for runs to reach a state where they can be resumed.

Modules in this file work with different storage backends (memory, disk, cloud) and can handle complex queries using various planning and execution agent configurations.

Portia Objects​

class Portia()

Portia client is the top level abstraction and entrypoint for most programs using the SDK.

It is responsible for intermediating planning via PlanningAgents and execution via ExecutionAgents.

__init__​

def __init__(config: Config | None = None,
             tools: ToolRegistry | list[Tool] | None = None,
             execution_hooks: ExecutionHooks | None = None) -> None

Initialize storage and tools.

Arguments:

  • config Config - The configuration to initialize the Portia client. If not provided, the default configuration will be used.
  • tools ToolRegistry | list[Tool] - The registry or list of tools to use. If not provided, the open source tool registry will be used, alongside the default tools from Portia cloud if a Portia API key is set.
  • execution_hooks ExecutionHooks | None - Hooks that can be used to modify or add extra functionality to the run of a plan.

initialize_end_user​

def initialize_end_user(end_user: str | EndUser | None = None) -> EndUser

Handle initializing the end_user based on the provided type.

run​

def run(
    query: str,
    tools: list[Tool] | list[str] | None = None,
    example_plans: list[Plan] | None = None,
    end_user: str | EndUser | None = None,
    plan_run_inputs: list[PlanInput] | list[dict[str, str]] | dict[str, str]
    | None = None
) -> PlanRun

End-to-end function to generate a plan and then execute it.

This is the simplest way to plan and execute a query using the SDK.

Arguments:

  • query str - The query to be executed.
  • tools list[Tool] | list[str] | None - List of tools to use for the query. If not provided all tools in the registry will be used.
  • example_plans list[Plan] | None - Optional list of example plans. If not provide a default set of example plans will be used.
  • end_user str | EndUser | None = None - The end user for this plan run. plan_run_inputs (list[PlanInput] | list[dict[str, str]] | dict[str, str] | None): Provides input values for the run. This can be a list of PlanInput objects, a list of dicts with keys "name", "description" (optional) and "value", or a dict of plan run input name to value.

Returns:

  • PlanRun - The run resulting from executing the query.

plan​

def plan(
    query: str,
    tools: list[Tool] | list[str] | None = None,
    example_plans: list[Plan] | None = None,
    end_user: str | EndUser | None = None,
    plan_inputs: list[PlanInput] | list[dict[str, str]] | list[str]
    | None = None
) -> Plan

Plans how to do the query given the set of tools and any examples.

Arguments:

  • query str - The query to generate the plan for.
  • tools list[Tool] | list[str] | None - List of tools to use for the query. If not provided all tools in the registry will be used.
  • example_plans list[Plan] | None - Optional list of example plans. If not provide a default set of example plans will be used.
  • end_user str | EndUser | None = None - The optional end user for this plan.
  • plan_inputs list[PlanInput] | None - Optional list of inputs required for the plan. This can be a list of Planinput objects, a list of dicts with keys "name" and "description" (optional), or a list of plan run input names. If a value is provided with a PlanInput object or in a dictionary, it will be ignored as values are only used when running the plan.

Returns:

  • Plan - The plan for executing the query.

Raises:

  • PlanError - If there is an error while generating the plan.

run_plan​

def run_plan(
    plan: Plan | PlanUUID | UUID,
    end_user: str | EndUser | None = None,
    plan_run_inputs: list[PlanInput]
    | list[dict[str, Serializable]]
    | dict[str, Serializable]
    | None = None
) -> PlanRun

Run a plan.

Arguments:

  • plan Plan | PlanUUID | UUID - The plan to run, or the ID of the plan to load from storage.
  • end_user str | EndUser | None = None - The end user to use. plan_run_inputs (list[PlanInput] | list[dict[str, Serializable]] | dict[str, Serializable] | None): Provides input values for the run. This can be a list of PlanInput objects, a list of dicts with keys "name", "description" (optional) and "value", or a dict of plan run input name to value.

Returns:

  • PlanRun - The resulting PlanRun object.

resume​

def resume(plan_run: PlanRun | None = None,
           plan_run_id: PlanRunUUID | str | None = None) -> PlanRun

Resume a PlanRun.

If a clarification handler was provided as part of the execution hooks, it will be used to handle any clarifications that are raised during the execution of the plan run. If no clarification handler was provided and a clarification is raised, the run will be returned in the NEED_CLARIFICATION state. The clarification will then need to be handled by the caller before the plan run is resumed.

Arguments:

  • plan_run PlanRun | None - The PlanRun to resume. Defaults to None.
  • plan_run_id RunUUID | str | None - The ID of the PlanRun to resume. Defaults to None.

Returns:

  • PlanRun - The resulting PlanRun after execution.

Raises:

  • ValueError - If neither plan_run nor plan_run_id is provided.
  • InvalidPlanRunStateError - If the plan run is not in a valid state to be resumed.

execute_plan_run_and_handle_clarifications​

def execute_plan_run_and_handle_clarifications(plan: Plan,
                                               plan_run: PlanRun) -> PlanRun

Execute a plan run and handle any clarifications that are raised.

resolve_clarification​

def resolve_clarification(clarification: Clarification,
                          response: object,
                          plan_run: PlanRun | None = None) -> PlanRun

Resolve a clarification updating the run state as needed.

Arguments:

  • clarification Clarification - The clarification to resolve.
  • response object - The response to the clarification.
  • plan_run PlanRun | None - Optional - the plan run being updated.

Returns:

  • PlanRun - The updated PlanRun.

error_clarification​

def error_clarification(clarification: Clarification,
                        error: object,
                        plan_run: PlanRun | None = None) -> PlanRun

Mark that there was an error handling the clarification.

wait_for_ready​

def wait_for_ready(plan_run: PlanRun,
                   max_retries: int = 6,
                   backoff_start_time_seconds: int = 7 * 60,
                   backoff_time_seconds: int = 2) -> PlanRun

Wait for the run to be in a state that it can be re-plan_run.

This is generally because there are outstanding clarifications that need to be resolved.

Arguments:

  • plan_run PlanRun - The PlanRun to wait for.
  • max_retries int - The maximum number of retries to wait for the run to be ready after the backoff period starts.
  • backoff_start_time_seconds int - The time after which the backoff period starts.
  • backoff_time_seconds int - The time to wait between retries after the backoff period starts.

Returns:

  • PlanRun - The updated PlanRun once it is ready to be re-plan_run.

Raises:

  • InvalidRunStateError - If the run cannot be waited for.

create_plan_run​

def create_plan_run(plan: Plan,
                    end_user: str | EndUser | None = None,
                    plan_run_inputs: list[PlanInput] | None = None) -> PlanRun

Create a PlanRun from a Plan.

Arguments:

  • plan Plan - The plan to create a plan run from.
  • end_user str | EndUser | None = None - The end user this plan run is for.
  • plan_run_inputs list[PlanInput] | None = None - The plan inputs for the plan run with their values.

Returns:

  • PlanRun - The created PlanRun object.