autoplan

Getting Started

AutoPlan is an open-source Python framework that provides a powerful pattern for implementing agentic AI applications, leveraging dynamic plan generation to select and use external tools based on the task's context.

Agentic AI applications are an emerging AI paradigm where LLMs use external tools to accomplish tasks that are beyond their own capabilities while keeping control over how to use those tools so that their abilities are not limited to predefined workflows. Building agentic applications requires a system design that allows plans to be dynamically generated, tools to be efficiently executed, and data flowing between tools to be properly channeled to provide a coherent output. AutoPlan provides just that.

AutoPlan is organized around three core components, each serving a specific purpose to enable dynamic planning, execution, and integration required for building an agentic application:

Tools can be any typed Python function — they can be procedural code, LLM calls, or AutoPlan applications themselves. Tools can be composed from smaller tools.

Planners are LLM-based components that generate the sequence of tools to be executed and the arguments to be passed to each tool to solve a given task.

Composers integrate tool outputs based on the planner’s strategy to produce a final output.

pdoc architecture

Key features

AutoPlan’s core functionality of dynamic planning shares similarities with the function calling feature found in major foundational models like GPT, Claude and Gemini. However, these models only return the functions (i.e., tools) to be called as a JSON object, leaving developers with several open questions: How should the models be prompted to generate an accurate plan that effectively uses the tools? How can the functions be executed efficiently? How should data flow between tools? How should the output of tools be summarized to generate the final output?

AutoPlan fully integrates dynamic planning and execution as a first-class capability, addressing these challenges:

  • Structured Inputs and Outputs: AutoPlan leverages Python type annotations and Pydantic classes to specify tools with structured and typed inputs and outputs. This allows the planner to precisely understand how to interact with external tools and manage data flow effectively.
  • Planner Prompting: AutoPlan provides a structured input and output format for plan generation, making it easy to apply best practices such as chain-of-thought reasoning and tool documentation injection. This ensures the Planner LLM receives high-quality inputs, which are essential for generating effective plans.
  • Efficient Execution: AutoPlan’s orchestration layer automatically captures task interdependencies, identifies opportunities for parallel execution, and optimally executes tasks to minimize overall runtime latency.
  • Dataflow Management: AutoPlan efficiently tracks and manages data dependencies, ensuring proper sequencing of tool executions and efficient transfer of outputs from upstream to downstream tasks.
  • Output composition: AutoPlan supports generating the final output to meet application requirements by providing a pattern that allows developers to define how intermediate outputs from various tools should be composed.

Beyond dynamic planning and execution, AutoPlan provides developers with additional key benefits that improve usability, debugging, and flexibility in building and scaling their applications:

  • Streaming Output: AutoPlan streams incremental results, such as the generated plan and the status of each step’s execution, enabling real-time progress tracking and user interfaces that display results as they are produced.
  • Observability: AutoPlan enhances debugging, auditing, and monitoring through its robust logging capabilities, capturing inputs and outputs for all tools. It also supports multiple observability providers, including Weight & Biases, delivering out-of-the-box observability.
  • Extensibility: Adding new tools is straightforward. Any typed Python function with a well written docstring can be a tool, making it easy to expand and customize your application.
  • LLM Agnostic: AutoPlan integrates with LiteLLM, supporting a diverse range of LLM providers. This flexibility allows developers to experiment with and combine different models, enabling them to identify the best ones for their specific needs.

Installation

AutoGen requires Python 3.12 or later.

Using Pip

We recommend using a virtual environment manager to isolate the dependencies for AutoPlan.

pip install -U autoplan

Using Poetry

Install Poetry if you don't already have it.

Create and activate:

poetry init
poetry shell

poetry add autoplan

Deep dive through an example

To better understand the key features and functionalities of AutoPlan, let's go through a simple example that answers questions about the performance of stocks in the stock market. This is a type of application that an LLM cannot answer on its own because it requires downloading stock data. While incorporating stock data into an LLM application can be achieved with the RAG pattern, the quality of the results can be further improved by using additional tools to compute certain metrics to have a more comprehensive analysis. The adddition of such tools in the answer generation pipeline and the way they should be used in terms of which stock data should be downloaded, which metrics should be computed, and how they should be combined based on the user's specific question calls for a certain level of orchestration where static workflows fall short.

Main components

Let's start by looking at the a simple example where the user asks how Nvidia's stock performs compared to Alphabet's and Amazon's. This question gets fed into the planner along with the list of tools available to the application. Here's is an excerpt from the planner prompt:

Generate a plan to answer the user's query about stock market benchmarks.
...
You will be probided with a list of additional tools to reason abuot the user's query. 
You must use these tools provided to generate the best plan possible to answer the user's query.

The list of tools and how to use them is defined by the json schema below:
{execution_context.plan_class.model_json_schema()}

Apply the following plan to solve for the user's query:
- Download the data for the tickers that are relevant to the user's query.
- Run the calculate statistics tool for every downloaded ticker data.
...
The user's query: " + application_args["question"]

The set of tools available to the planner is defined by an annotation attached to the application as follows:

@with_planning(
    tools=[
        download_ticker,
        calculate_statistics,
    ]
    generate_plan_prompt_generator=generate_plan,
    combine_steps_prompt_generator=combine_steps,
)

Tools are defined with the @tool decorator and are annotated with the expected input and output types. They can optionally declare dependencies on certain data types which can be generated by other tools, such that the planner can generate a plan that uses the tools in the right order. Note that it is very important to add docstrings to the tool methods because these inform the planner on the tool's capabilities and how to use it.

@tool
async def download_ticker(ticker: str) -> TickerData:
    """
    Given a ticker, download the data from yfinance for the past 5 years in monthly frequency.
    Args:
        ticker: The ticker symbol of the stock to download data for.
    Returns:
        TickerData: Object containing the downloaded stock data.
    """
    data = yf.download(ticker, period="5y", interval="1mo")
    return TickerData(
        name=ticker,
        closes=[cell[0] for cell in data["Adj Close"].values.tolist()],
    )


@tool(can_use_prior_results=True)
def calculate_statistics(data: TickerData) -> Statistics:
    """
    Calculate the statistics for the given stock data.

    Args:
        data: TickerData from a prior step containing closing prices

    Returns:
        Statistics: Object containing calculated statistics including:
            - name: Name of the ticker/benchmark
            - one_month_return: Percentage return over last month
            - one_year_return: Percentage return over last year
            - volatility: Standard deviation of the monthly returns
            - sharpe_ratio: Sharpe ratio of the monthly returns
    """
    ...

Finally, you can give a prompt for the composer to combine the results provided by the tools to generate the final answer. There's a lot of flexibility in how you can define this prompt, but in this example, we'll use a simple prompt that combines the results of the tools into a single answer.

Summarize the results coming out of the tool executions in the following way:
1. Answer the user's query, explaining the rationale behind the metrics and data used.
2. Explain the meaning of the metrics and basis for the benchmark comparison.

Running the application

To run the application, follow these steps:

cd examples/stock_market_expert
poetry install
OPENAI_API_KEY=<your-api-key> poetry run python stock_benchmark/app.py

Once you run the application using the above command lines, you will see a Gradio interface that allows you to interact with the application. Now you can type in the question "How does Nvidia's stock perform compared to Alphabet's and Amazon's?" and see its answer.

stock question

Below is a likely plan that would be generated by the planner to answer this question:

stock question

The above question refers to specific stocks, making it an easier task to generate a plan. Since the plan is generated by an LLM, you can also ask questions that require domain and common sense knowledge to answer the user's question. For example, if you ask "Compare Nvidia's performance against the market", you can expect the planner to generate a plan that downloads data from major stock market indices to build up a baseline for comparison.

stock question

Takeaways

From an architectural perspective, this execution pattern allows the decomposition of the problem into smaller, more modular parts. This is a powerful pattern that allows the application to be highly flexible and modular, and to be easily extended with new tools or parts of the application.

From a performance perspective, it is important to note that the tools start executing as soon as the first steps of the plan are generated, without necessarily waiting for the entire plan to be generated. This is achieved by streaming the output of the planner to the executing engine and the execution engine starting taks in an eager manner. Second, the execution engine is able to execute the tools in parallel where possible, such as when the tools are not dependent on the output of each other. These two features combined allows the application to be highly performant and to provide responses as quickly as possible.

From an observability perspective, the application logs all inputs and outputs for all tools, making debugging, auditing, and monitoring straightforward. For example, you can try running the application with an additional environment variable (e.g. WEAVE_PROJECT_ID="Stock") for logging and observing the execution pipeline through Weights & Biases.

stock question

Bootstrapping your own application

AutoPlan has a command-line interface that allows you to create a new application. To create your own application, you can use the following command:

autoplan generate \
  --name "my_app" \
  --description "This is an LLM-powered tool that can generate a research report on a topic of interest based on a user query." \ 
  --outdir .
Note

You must have AutoPlan installed to use this command.

Note that you can also leave the parameters empty and let the CLI prompt you for the information, which will help you get started with bootstrapping your application.

This command line will create a new folder named my_app with the basic structure of an AutoPlan application. You can then go in the application folder and run it with the following command:

cd my_app
poetry install
poetry run python my_app/app.py

This will start a Gradio interface that allows you to interact with the application without any additional effort. You can now use your browser and go to http://localhost:7860 to see the application.

pdoc architecture

Note

By default AutoPlan will use Open AI models and will include a search tool based on you.com, which require API keys. You can set the OPENAI_API_KEY and YDC_API_KEY environment variables to your OpenAI and You API keys to use your own accounts .

Best practices

Here are a few best practices to follow to get the best out of your own AutoPlan applications.

Create reusable tools

One of the key features of AutoPlan is the ability to create tools that can be reused accross different scenarios. When designing your application, think about how to decompose the problem into smaller, more modular parts and how to parameterize so that the planner can use them in different contexts.

Add docstrings to tool methods

The planner needs to know how to use the tools to generate a plan that uses the tools in efficient and effective ways. For this reason, it is very useful to add information about how to use the tools in the planners prompt.

To do this systematically, AutoPlan's execution context provides an interface to generate a JSON schema of the tools (i.e. {execution_context.plan_class.model_json_schema()}), which is highly recommended to be injected into the planner prompt. This interface relies on the docsrtings. It is important to add comprehensive docstrings to the tool methods to make sure that the planner can generate a plan that uses the tools in the right way.

class SearchResult(BaseModel):
    content: str = Field(..., description="The content of the search result.")
    sources: list[str] = Field(..., description="The links to the search result pages.",
    )

Use typed I/O for tools

The composition of an application based on multiple tools requires the data to flow between the tools. To make sure that the data that is produced by one tool can be used by another tool, it is important to type the I/O of the tools. This is achieved by using Pydantic models to define the input and output types of the tools. It is equally important to add field descriptions to the Pydantic models to make sure that the planner can generate a plan that uses the tools in the right way.

@tool
async def download_ticker(ticker: str) -> TickerData:
    """
    Given a ticker, download the data from yfinance for the past 5 years in monthly frequency.
    Args:
        ticker: The ticker symbol of the stock to download data for.
    Returns:
        TickerData: Object containing the downloaded stock data.
    """
    ...

Manage static dependencies

Sometimes the tools need to access to certain parameters that don't need to be generated by the planner. For example, a tool may need a direct access to the user's query. To achieve this, you can add a Dependency object as an argument to the tool method and initialize it in the application.

# Declare a dependency object
from autoplan import Dependency
query_dependency = Dependency()

# Declare a tool that uses the dependency object with a default value
@tool
def analyse_query(user_query:Dependency = query_dependency) -> QueryAnalysis:
    ...

# Set the value of the dependency object in the application code where you have access to the user's input text field
query_dependency.set(input_text.value)

Optimize the order of I/O fields

Chain-of-thought reasoning is a powerful prompt engineering technique that enhances LLM performance by generating intermediate reasoning steps to guide the final output. This approach improves coherence and accuracy by conditioning subsequent outputs on earlier generated tokens.

You can apply this technique effectively by optimizing the order of input/output (I/O) fields in Pydantic models to encourage structured reasoning.

For example, AutoPlan's default data class, which defines the planner's output, is structured as follows:

class Step(BaseModel):
    """
    A step in the plan, using a specific tool.
    """

    tool_call: Tool


class Plan(BaseModel):
    """
    A plan for achieving the application's goal, composed of steps.
    """

    rationale: str = Field(
        description="Layout a detailed rationale for the plan, detailing step by step the tools that should be called, where their inputs should come from, and how the results can be used."
    )
    steps: Sequence[Step]

In this example, while only the steps field is strictly required to execute the plan, including a rationale field before the steps provides additional context. This textual description encourages the planner to produce a well-reasoned and logically sound plan, increasing the likelihood of success.

Decompose tools as AutoPlan applications

AutoPlan allows tools to be defined as AutoPlan applications themselves. Indeed, any AutoPlan application can be used as a tool by other AutoPlan applications. This allows for a high level of modularity and reusability of tools.

As illustrated in the example below, The story_description is an AutoPlan application that uses create_character as a tool. create_character is itself an AutoPlan application that uses you_search as a tool in order to search for information about the character on the web.

@with_planning(
    step_class=CharacterPlanStep,
    plan_class=CharacterPlan,
    tools=[you_search],
    generate_plan_prompt_generator=generate_plan,
    combine_steps_prompt_generator=combine_steps,
)
async def create_character(
    character_request: str,
) -> CharacterOutput:
    ...

with_planning(
    step_class=StoryPlanStep,
    plan_class=StoryPlan,
    tools=[create_character],
    generate_plan_prompt_generator=generate_plan,
    combine_steps_prompt_generator=combine_steps,
)
async def run(
    story_description: str,
) -> StoryOutput:
    pass

Try using different LLMs

You can try using different LLMs by setting the generate_plan_llm_model and combine_steps_llm_model parameters in the with_planning decorator, and/or by setting the model of your choice in your tool implementations.

@with_planning(
    generate_plan_llm_model="claude-3-5-sonnet-latest",
    combine_steps_llm_model="claude-3-5-sonnet-latest",
)

If your application uses other models, you should set the API keys for those models in your environment (e.g. ANTHROPIC_API_KEY = <your-key>) .


 1"""
 2.. include:: ../docs/USER_GUIDE.md
 3
 4"""
 5
 6from autoplan.chain import chain
 7from autoplan.core import with_planning
 8from autoplan.dependency import Dependency
 9from autoplan.models import Plan, Step
10from autoplan.results import (
11    FinalResult,
12    PartialPlanResult,
13    PlanResult,
14    StepResult,
15)
16from autoplan.tool import tool
17from autoplan.trace import WeaveTracer, set_tracer, trace
18
19__all__ = [
20    "Dependency",
21    "FinalResult",
22    "Step",
23    "Plan",
24    "PartialPlanResult",
25    "PlanResult",
26    "StepResult",
27    "tool",
28    "with_planning",
29    "trace",
30    "set_tracer",
31    "WeaveTracer",
32    "chain",
33]
class Dependency:
 5class Dependency:
 6    def __init__(self):
 7        self.item = None
 8
 9    def __str__(self):
10        return f"Dependency with item: {self.item}"
11
12    def set(self, item):
13        self.item = item
14        for name, value in inspect.getmembers(item, callable):
15            if not hasattr(self, name):
16                setattr(self, name, value)
item
def set(self, item):
12    def set(self, item):
13        self.item = item
14        for name, value in inspect.getmembers(item, callable):
15            if not hasattr(self, name):
16                setattr(self, name, value)
class FinalResult(autoplan.results.Result, typing.Generic[Output]):
38class FinalResult[Output: BaseModel](Result):
39    """
40    A final result of the application.
41    """
42
43    result: Output

A final result of the application.

result: Output
class Step(pydantic.main.BaseModel):
 9class Step(BaseModel):
10    """
11    A step in the plan, using a specific tool.
12    """
13
14    tool_call: Tool

A step in the plan, using a specific tool.

tool_call: autoplan.tool.Tool
model_config: ClassVar[pydantic.config.ConfigDict] = {}

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

class Plan(pydantic.main.BaseModel):
17class Plan(BaseModel):
18    """
19    A plan for achieving the application's goal, composed of steps.
20    """
21
22    rationale: str = Field(
23        description="Layout a detailed rationale for the plan, detailing step by step the tools that should be called, where their inputs should come from, and how the results can be used."
24    )
25    steps: Sequence[Step]

A plan for achieving the application's goal, composed of steps.

rationale: str
steps: Sequence[Step]
model_config: ClassVar[pydantic.config.ConfigDict] = {}

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

class PartialPlanResult(autoplan.results.Result, typing.Generic[Plan]):
13class PartialPlanResult[Plan: BaseModel](Result):
14    """
15    An unfinished plan (e.g. a plan that is in the process of being generated).
16    """
17
18    result: Plan

An unfinished plan (e.g. a plan that is in the process of being generated).

result: Plan
class PlanResult(autoplan.results.Result, typing.Generic[Plan]):
21class PlanResult[Plan: BaseModel](Result):
22    """
23    A finished plan.
24    """
25
26    result: Plan

A finished plan.

result: Plan
class StepResult(autoplan.results.Result):
29class StepResult(Result):
30    """
31    A result of a step.
32    """
33
34    step: Step
35    result: BaseModel | str | None

A result of a step.

step: Step
result: pydantic.main.BaseModel | str | None
model_config: ClassVar[pydantic.config.ConfigDict] = {}

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

def tool( f: Optional[Callable[..., Any]] = None, *, can_use_prior_results: bool = False) -> Union[type[autoplan.tool.Tool], Callable[[Callable[..., Any]], type[autoplan.tool.Tool]]]:
 95def tool(
 96    f: Callable[..., Any] | None = None,
 97    *,
 98    can_use_prior_results: bool = False,
 99) -> type[Tool] | Callable[[Callable[..., Any]], type[Tool]]:
100    """
101    Decorator to create a tool from a function.
102
103    Can be called either as:
104    @tool
105    def my_tool(arg: str) -> str:
106        ...
107
108    or as:
109    @tool()
110    def my_tool(arg: str) -> str:
111        ...
112
113
114    if @tool(can_use_prior_results=True)
115    def my_tool(arg: str) -> str:
116        ...
117    then "arg" could be the result of a prior tool, if specified by the plan
118    """
119    if f is None:
120        @wraps(tool)
121        def decorator(func: Callable[..., Any]) -> type[Tool]:
122            return tool(func, can_use_prior_results=can_use_prior_results)
123        return decorator
124    else:
125        if not inspect.iscoroutinefunction(f):
126            raise ValueError("Tool functions must be asynchronous")
127        cls = _function_to_tool_subclass(trace(f), can_use_prior_results)
128        return cls

Decorator to create a tool from a function.

Can be called either as: @tool def my_tool(arg: str) -> str: ...

or as: @tool() def my_tool(arg: str) -> str: ...

if @tool(can_use_prior_results=True) def my_tool(arg: str) -> str: ... then "arg" could be the result of a prior tool, if specified by the plan

def with_planning( step_class: type[Step], plan_class: type[Plan], generate_plan_prompt_generator: Callable[[autoplan.execution_context.ExecutionContext, ~ApplicationArgsVar], list[str]], combine_steps_prompt_generator: Callable[[autoplan.execution_context.ExecutionContext, ~PlanVar, list[str]], list[str]], tools: list[type[autoplan.tool.Tool]], generate_plan_temperature: float = 0.0, combine_steps_temperature: float = 0.0, generate_plan_llm_model: Optional[str] = None, combine_steps_llm_model: Optional[str] = None, generate_plan_llm_args: Optional[dict] = None, combine_steps_llm_args: Optional[dict] = None, can_use_prior_results: bool | None = None):
176def with_planning(
177    step_class: type[Step],
178    plan_class: type[Plan],
179    generate_plan_prompt_generator: GeneratePlanPromptGenerator,
180    combine_steps_prompt_generator: CombineStepsPromptGenerator,
181    tools: list[type[Tool]],
182    generate_plan_temperature: float = 0.0,
183    combine_steps_temperature: float = 0.0,
184    generate_plan_llm_model: Optional[str] = None,
185    combine_steps_llm_model: Optional[str] = None,
186    generate_plan_llm_args: Optional[dict] = None,
187    combine_steps_llm_args: Optional[dict] = None,
188    can_use_prior_results: bool | None = None,
189):
190    """
191    Decorator to add planning to a function.
192
193    step_class: The class for a step in the plan.
194    plan_class: The class for the plan.
195    generate_plan_prompt_generator: A function that generates the prompt for the plan.
196    combine_steps_prompt_generator: A function that generates the prompt for combining the steps.
197    tools: The tools that can be used in the plan.
198    generate_plan_temperature: The temperature for the generate plan prompt.
199    combine_steps_temperature: The temperature for the combine steps prompt.
200    generate_plan_llm_model: The model to use for the generate plan prompt.
201    combine_steps_llm_model: The model to use for the combine steps prompt.
202    generate_plan_llm_args: The arguments to pass to the generate plan prompt.
203    combine_steps_llm_args: The arguments to pass to the combine steps prompt.
204    can_use_prior_results: Whether the tool can use the results of prior steps.
205    """
206
207    def wrapper(func):
208        function_return_type = func.__annotations__.get("return")
209        func_signature = inspect.signature(func)
210
211        updated_tools = [
212            # if the tool has the WITH_PLANNING_ATTR, then we need to wrap it in a from_planned function
213            _from_planned(tool)
214            if hasattr(tool, _WITH_PLANNING_ATTR)
215            # otherwise, we expect it to be a tool
216            else tool
217            for tool in tools
218        ]
219
220        for tool in updated_tools:
221            if not issubclass(tool, Tool):
222                raise ValueError(f"{tool} is not a Tool. Was it decorated with @tool?")
223
224        # These annotations will create a trace whose name and arguments come from the decorated function
225        @functools.wraps(func)
226        async def wrapped(*args, **kwargs):
227            arguments = func_signature.bind(*args, **kwargs).arguments
228
229            queue = asyncio.Queue()
230
231            context = ExecutionContext(
232                plan_class=create_plan_class(step_class, plan_class, updated_tools),
233                tools=tools,
234                output_model=function_return_type,
235                application_args=arguments,
236                generate_plan_llm_model=generate_plan_llm_model or "gpt-4o-mini",
237                generate_plan_llm_args=generate_plan_llm_args or {},
238                combine_steps_llm_model=combine_steps_llm_model or "gpt-4o-mini",
239                combine_steps_llm_args=combine_steps_llm_args or {},
240            )
241
242            # start the execution in the background
243            asyncio.create_task(
244                _execute(
245                    context,
246                    generate_plan_prompt_generator,
247                    combine_steps_prompt_generator,
248                    arguments,
249                    queue,
250                    generate_plan_temperature,
251                    combine_steps_temperature,
252                )
253            )
254
255            # yield each item from the queue as it comes in
256            while True:
257                item = await queue.get()
258                yield item
259
260                if isinstance(item, FinalResult):
261                    return
262
263        # add a marker so we can identify the function as a "with_planning" decorated function
264        setattr(wrapped, _WITH_PLANNING_ATTR, True)
265        return wrapped
266
267    return wrapper

Decorator to add planning to a function.

step_class: The class for a step in the plan. plan_class: The class for the plan. generate_plan_prompt_generator: A function that generates the prompt for the plan. combine_steps_prompt_generator: A function that generates the prompt for combining the steps. tools: The tools that can be used in the plan. generate_plan_temperature: The temperature for the generate plan prompt. combine_steps_temperature: The temperature for the combine steps prompt. generate_plan_llm_model: The model to use for the generate plan prompt. combine_steps_llm_model: The model to use for the combine steps prompt. generate_plan_llm_args: The arguments to pass to the generate plan prompt. combine_steps_llm_args: The arguments to pass to the combine steps prompt. can_use_prior_results: Whether the tool can use the results of prior steps.

def trace(f):
60def trace(f):
61    @wraps(f)
62    def inner(*args, **kwargs):
63        tracer = get_tracer()
64        if tracer:
65            return tracer.trace(f)(*args, **kwargs)
66        else:
67            return f(*args, **kwargs)
68
69    return inner
def set_tracer(tracer: autoplan.trace.Tracer):
51def set_tracer(tracer: Tracer):
52    global _tracer
53    _tracer = tracer
class WeaveTracer(autoplan.trace.Tracer):
28class WeaveTracer(Tracer):
29    def __init__(self, project_id: str):
30        import weave  # pyright: ignore[reportMissingImports]
31
32        self.client = weave.init(project_id)
33
34    def create_call(self, name: str, inputs: dict):
35        call = self.client.create_call(op=name, inputs=inputs)
36        return ManualCall(
37            name=name,
38            inputs=inputs,
39            end=lambda output: self.client.finish_call(call, output=output),
40        )
41
42    def trace(self, f: Callable) -> Callable:
43        import weave  # pyright: ignore[reportMissingImports]
44
45        return weave.op()(f)

Helper class that provides a standard way to create an ABC using inheritance.

WeaveTracer(project_id: str)
29    def __init__(self, project_id: str):
30        import weave  # pyright: ignore[reportMissingImports]
31
32        self.client = weave.init(project_id)
client
def create_call(self, name: str, inputs: dict):
34    def create_call(self, name: str, inputs: dict):
35        call = self.client.create_call(op=name, inputs=inputs)
36        return ManualCall(
37            name=name,
38            inputs=inputs,
39            end=lambda output: self.client.finish_call(call, output=output),
40        )
def trace(self, f: Callable) -> Callable:
42    def trace(self, f: Callable) -> Callable:
43        import weave  # pyright: ignore[reportMissingImports]
44
45        return weave.op()(f)
def chain( tool1: type[autoplan.tool.Tool], tool2: type[autoplan.tool.Tool], name: Optional[str] = None, description: Optional[str] = None) -> type[autoplan.tool.Tool]:
  9def chain(
 10    tool1: type[Tool],
 11    tool2: type[Tool],
 12    name: Optional[str] = None,
 13    description: Optional[str] = None,
 14) -> type[Tool]:
 15    """
 16    Chains two tools together by constructing a new function that calls the first tool with the given parameters,
 17    then passes the result to the second tool.
 18    """
 19
 20    # we need to figure out how the tool2 parameters relate to the tool1 parameters
 21    tool2_parameters = {
 22        k: v.annotation for k, v in tool2.model_fields.items() if k != "type"
 23    }
 24
 25    # tool2 parameters that have the same name as tool1 parameters
 26    tool2_parameters_included_in_tool1 = {
 27        k: v
 28        for k, v in tool1.model_fields.items()
 29        if k in tool2_parameters and k != "type"
 30    }
 31
 32    tool2_parameters_not_included_in_tool1 = {
 33        k: v
 34        for k, v in tool2_parameters.items()
 35        if k not in tool1.model_fields and k != "type"
 36    }
 37
 38    # if there are parameters in tool2 that are not in tool1, we assume the first parameter represents the result of the first tool
 39    if tool2_parameters_not_included_in_tool1:
 40        result_key = next(iter(tool2_parameters_not_included_in_tool1))
 41    # otherwise, we assume the name is reused
 42    else:
 43        result_key = next(iter(tool2_parameters_included_in_tool1))
 44
 45    tool2_parameters_provided_by_tool_1 = [
 46        key for key in tool2_parameters if key != result_key
 47    ]
 48
 49    parameters_from_tool_1 = {
 50        k: v.annotation for k, v in tool1.model_fields.items() if k != "type"
 51    }
 52
 53    parameters_from_tool_2 = {
 54        k: v.annotation
 55        for k, v in tool2.model_fields.items()
 56        if k != "type"
 57        and k in tool2_parameters_not_included_in_tool1
 58        and k != result_key
 59    }
 60    parameters = {
 61        k: v
 62        for k, v in {**parameters_from_tool_1, **parameters_from_tool_2}.items()
 63        if v is not None
 64    }
 65
 66    fn_name = name or tool1.__name__ + "_" + tool2.__name__
 67
 68    async def core(**kwargs):
 69        # construct an instance of the first tool
 70        tool1_instance = tool1(**kwargs)
 71
 72        # call the first tool
 73        tool1_result = await tool1_instance()
 74
 75        tool2_kwargs: dict[str, Any] = {result_key: tool1_result} | {
 76            k: v for k, v in kwargs.items() if k in tool2_parameters_provided_by_tool_1
 77        }
 78
 79        # construct an instance of the second tool
 80        tool2_instance = tool2(**tool2_kwargs)
 81
 82        # call the second tool
 83        return await tool2_instance()
 84
 85    # We need the function to have explicit parameters that match the tool1 parameters
 86    # To do this, we use exec to evaluate code that we templated with the tool1 parameters
 87
 88    # See: https://stackoverflow.com/questions/26987418/programmatically-create-function-specification
 89
 90    # Example:
 91    # async def Double_Triple(x: int):
 92    #     return await core(x=x)
 93
 94    code = f"""
 95async def {fn_name}({', '.join(f'{k}: {v.__name__ if not hasattr(v, "__args__") else v.__args__[0].__name__}' for k, v in parameters.items())}):
 96    return await core({', '.join(f'{k}={k}' for k in parameters)})
 97    """
 98
 99    env = {"core": core}
100
101    namespace = {"Union": Union, "BaseModel": BaseModel}
102    exec(code, env, namespace)
103    f = tool(namespace[fn_name])
104    f.__doc__ = description
105    return f

Chains two tools together by constructing a new function that calls the first tool with the given parameters, then passes the result to the second tool.