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.

ExecutionHooks Objects

class ExecutionHooks()

Hooks that can be used to modify or add extra functionality to the run of a plan.

Currently, the only hook is a clarification handler which can be used to handle clarifications that arise during the run of a plan.

__init__

def __init__(
        clarification_handler: ClarificationHandler | None = None) -> None

Initialize ExecutionHooks with default values.

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.

run

def run(query: str,
        tools: list[Tool] | list[str] | None = None,
        example_plans: list[Plan] | 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.

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) -> 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.

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) -> PlanRun

Run a plan.

Arguments:

• plan Plan - The plan to run.

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) -> PlanRun

Create a PlanRun from a Plan.

Arguments:

• plan Plan - The plan to create a plan run from.

Returns:

• PlanRun - The created PlanRun object.