A Python micro framework for creating LLM-driven agents
# Configuration file for the Sphinx documentation builder.
#
# For the full list of built-in configuration values, see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html

# -- Project information -----------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information

project = "MakeAgents"
copyright = "2023, Sidney Radcliffe"
author = "Sidney Radcliffe"

# -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

extensions = ["sphinx.ext.autodoc", "sphinx.ext.napoleon"]

templates_path = ["_templates"]
exclude_patterns = []


# -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output

html_theme = "alabaster"
html_static_path = ["_static"]

import make_agents

version = make_agents.__version__  # access this in index.rst using {version}
print("version", version)

# Source: https://stackoverflow.com/a/69211912
variables_to_export = [
    "version",
]
frozen_locals = dict(locals())
rst_epilog = "\n".join(
    map(lambda x: f".. |{x}| replace:: {frozen_locals[x]}", variables_to_export)
)
del frozen_locals

docs_creator/source/conf.py
sradc/MakeAgents
from pydantic import BaseModel, Field

import make_agents as ma


@ma.action
def get_task_instructions():
    return "Get the users email address, and validate it."


class MessageUserArg(BaseModel):
    message: str = Field(description="Message to send user")


@ma.action
def message_user(arg: MessageUserArg):
    """Send the user a message, and get their response."""
    response = ""
    while response == "":
        response = input(arg.message + "\n").strip()
    return response


class SendValidationEmailArg(BaseModel):
    users_email_address: str = Field(description="The users email address")


@ma.action
def send_validation_email(arg: SendValidationEmailArg):
    """Send the user a validation email."""
    if not arg.users_email_address.endswith(".com"):
        return {"status": "error", "description": "Email address must end with `.com`"}
    else:
        return {"status": "success", "description": "Validation code sent"}


class CheckValidationCodeArg(BaseModel):
    validation_code: str = Field(description="The validation code (6 digits)")


@ma.action
def check_validation_code(arg: CheckValidationCodeArg):
    """Send the user a validation email."""
    if len(arg.validation_code) != 6:
        return {"status": "error", "description": "Validation code must be 6 digits"}
    elif arg.validation_code == "123456":
        return {"status": "success", "description": "Validation code correct"}
    else:
        return {"status": "error", "description": "Validation code incorrect"}


def action_graph(current_action: callable, current_action_result: dict) -> list[callable]:
    """Return the next action(s) to run, given the current action and its result."""
    if current_action == ma.Start:
        return [get_task_instructions]
    elif current_action == get_task_instructions:
        return [message_user]
    elif current_action == message_user:
        return [message_user, send_validation_email, check_validation_code]
    elif current_action == send_validation_email:
        if current_action_result["status"] == "success":
            return [message_user]
        else:
            return [message_user, send_validation_email]
    elif current_action == check_validation_code:
        if current_action_result["status"] == "success":
            return [ma.End]
        else:
            return [message_user, check_validation_code]
    else:
        raise ValueError(f"Unknown action: {current_action}")


# Run the agent
print("Starting agent...\n\n")
for messages in ma.run_agent(action_graph):
    ma.bonus.pretty_print(messages[-1])

examples/email_verification_example.py
sradc/MakeAgents
# Copyright 2023 Sidney Radcliffe

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from importlib_metadata import version

__version__ = version(__package__)

# Expose the objects that are part of the API
import make_agents.bonus as bonus  # noqa: F401
import make_agents.gpt as gpt  # noqa: F401
from make_agents.make_agents import End, Start, action, run_agent  # noqa: F401

make_agents/__init__.py
sradc/MakeAgents
# Copyright 2023 Sidney Radcliffe

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Bonus content.
Not properly documented or tested, and dependencies not captured in Poetry.
Use at your own risk.
"""


def draw_graph(agent_graph: dict[callable, list[callable]]):
    import io
    import warnings

    try:
        import graphviz
        from PIL import Image

        dot = graphviz.Digraph(comment="graph", format="png", graph_attr={"dpi": "80"})
        for node in agent_graph:
            dot.node(node.__name__, node.__name__)
        for node, children in agent_graph.items():
            if isinstance(children, list):
                for child in children:
                    dot.edge(node.__name__, child.__name__)
            else:
                dot.edge(node.__name__, children.__name__)
        gvz_graph = dot.pipe(format="png")
        image = Image.open(io.BytesIO(gvz_graph), mode="r", formats=["png"]).convert(
            "RGB"
        )
        return image
    except ImportError:
        warnings.warn(
            "Could not import graphviz, so could not draw graph. (Note that make_agents does not install this package)."
        )
        return None


def pretty_print(message: dict):
    # Originally based on: https://github.com/openai/openai-cookbook/blob/f52ffdaca42073066f8f43f7d65a59dcc01c9349/examples/How_to_call_functions_with_chat_models.ipynb
    if message["role"] == "system":
        print(f"system message: {message['content']}\n")
    elif message["role"] == "user":
        print(f"user message: {message['content']}\n")
    elif message["role"] == "assistant" and message.get("function_call"):
        arguments = message["function_call"]["arguments"]
        arguments = "<no arguments>" if arguments == "null" else arguments
        print(f"call `{message['function_call']['name']}`: {arguments}\n")
    elif message["role"] == "assistant" and not message.get("function_call"):
        print(f"assistant message: {message['content']}\n")
    elif message["role"] == "function":
        print(f"`{message['name']}` result: {message['content']}\n")

make_agents/bonus.py
sradc/MakeAgents
# Copyright 2023 Sidney Radcliffe

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import openai
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_random_exponential,
)


def get_completion_func(model: str = "gpt-4", **kwargs) -> callable:
    """Returns a function for getting completions from OpenAI.
    Can specify more parameters, e.g. temperature, etc. via kwargs, see:
    https://platform.openai.com/docs/api-reference/introduction?lang=python

    Parameters
    ----------
    model : str, optional
        The chat model to use, by default "gpt-4".

    Returns
    -------
    callable
        A function that is used to get completions from OpenAI, to drive agents.
    """

    @retry(
        retry=retry_if_exception_type(
            (openai.error.Timeout, openai.error.RateLimitError)
        ),
        wait=wait_random_exponential(min=0, max=60),
        stop=stop_after_attempt(6),
    )
    def completion(**kwargs2):
        return openai.ChatCompletion.create(model=model, **kwargs, **kwargs2)

    return completion

make_agents/gpt.py
sradc/MakeAgents
# Copyright 2023 Sidney Radcliffe

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import inspect
import json
from copy import deepcopy
from enum import Enum
from typing import Iterator, Optional, Union

from pydantic import BaseModel, Field

from make_agents.gpt import get_completion_func

default_completion = get_completion_func()

default_system_prompt = """You are a helpful assistant. You will be given tasks, via function calls. You will be given the ability to run different functions at different times. Please use them to complete the most recent task you have been given."""


def action(func: callable) -> callable:
    """A decorator to create *action functions* — functions to be used by the agent.
    An action function must have *at most* one parameter, which must be annotated with a Pydantic model.

    Note that the following should be considered part of the "prompt" for the agent:

    - The name of the function

    - The Pydantic model, if the function has a parameter

    - The function's docstring (don't annotate the parameter in the docstring, use the Pydantic model for this)

    Parameters
    ----------
    func : callable
        The function to be decorated.

    Returns
    -------
    callable
        The same function, with metadata attached.

    Raises
    ------
    ValueError
        If the function has more than one parameter, or if the parameter is not annotated with a Pydantic model.
    """
    parameters = inspect.signature(func).parameters
    if len(parameters) == 0:
        func.description_for_llm = {
            "name": func.__name__,
            "description": func.__doc__,
            "parameters": None,
        }
    elif len(parameters) == 1:
        func.description_for_llm = {
            "name": func.__name__,
            "description": func.__doc__,
            "parameters": get_pydantic_model_from_action_func(func).model_json_schema(),
        }
    else:
        raise ValueError(f"Function {func.__name__} must have at most one parameter.")
    return func


def get_pydantic_model_from_action_func(func: callable) -> BaseModel:
    (arg,) = inspect.signature(func).parameters.values()
    try:
        arg.annotation.model_json_schema
    except AttributeError:
        raise ValueError(
            f"The parameter of {func.__name__} must be annotated with a pydantic model."
        )
    return arg.annotation


def select_next_action_factory(options: list[callable]) -> callable:
    names = [description(x)["name"] for x in options]
    if len(names) != len(set(names)):
        raise ValueError(f"Duplicate function names: {names}")

    class SelectNextFuncArg(BaseModel):
        thought_process: str = Field(
            ...,
            description="Describe your thought process for selecting the next function in a few words.",
        )
        next_function: Enum(
            "function_names",
            {description(x)["name"]: description(x)["name"] for x in options},
        ) = Field(..., description="Name of the function to call next")

    def select_next_func(arg: SelectNextFuncArg):
        return arg.next_function.value

    select_next_func.__doc__ = (
        "Given the following functions, choose the one that will most help you achieve"
        " your goal: " + ", ".join([json.dumps(description(x)) for x in options])
    )
    return action(select_next_func)


def description(action_func: callable) -> dict:
    try:
        return action_func.description_for_llm
    except AttributeError:
        raise ValueError(
            f"Missing metadata. Has function {action_func.__name__} been decorated with"
            f" `{action.__name__}`?"
        )


def get_func_input_from_llm(messages: list[dict], func: callable, completion: callable):
    response = completion(
        messages=messages,
        functions=[description(func)],
        function_call={
            "name": description(func)["name"]
        },  # force the function to be called
    )
    # Validate the arg
    pydantic_model = get_pydantic_model_from_action_func(func)
    func_arg = pydantic_model(
        **json.loads(response.choices[0].message.function_call.arguments)
    )
    # If the above didn't raise an error, we can assume the arg is valid
    func_arg_message = json.loads(
        json.dumps(response.choices[0].message)
    )  # make a clean dict
    return func_arg_message, func_arg


def run_func_for_llm(func: callable, arg: Optional[BaseModel]):
    func_result = func(arg) if arg else func()
    func_result_message = {
        "role": "function",
        "name": func.description_for_llm["name"],
        "content": json.dumps(func_result),
    }
    return func_result_message, func_result


class Start:
    """Used to mark the start of the action graph."""


class End:
    """Can be used to end the action graph."""

    description_for_llm = {
        "name": "End",
        "description": "End your assistance with immediate effect.",
        "parameters": None,
    }


def identity(x):
    return x


def run_agent(
    action_graph: Union[dict, callable],
    messages_init: Optional[list[dict]] = None,
    completion: Optional[callable] = default_completion,
    pre_llm_callback: Optional[callable] = identity,
) -> Iterator[list[dict[str, str]]]:
    """Run an agent. This is a generator that yields the list of messages after each step.
    Be mindful that the yielded messages are mutable, allowing them to be modified in place,
    (make copies if you want to avoid this).

    Parameters
    ----------
    action_graph : Union[dict[callable, list[callable]], callable]
        The graph of actions that the agent can take. Can either be a dictionary
        or a callable. Use a callable to create a dynamic action graph. (See examples in the README)
    messages_init : Optional[list[dict]], optional
        Optionally initialise the list of messages, e.g. to specify a custom system prompt.
        If not provided, the default system prompt will be used.
    completion : Optional[callable], optional
        The function that will be used to get completions from the LLM.
    pre_llm_callback : Optional[callable], optional
        This function is called before any LLM calls.
        It will be passed the list of messages, and can modify it in place.
        Can be used for, e.g. reducing the list of messages to only the most recent ones,
        or reducing the list by summarising, etc.

    Yields
    ------
    Iterator[list[dict[str, str]]]
        At each step, the list of messages is yielded,
        i.e. the same list that was yielded in the previous step, with one more message appended.
    """
    if isinstance(action_graph, dict):
        action_graph = dict_to_action_graph_func(action_graph)
    messages = (
        deepcopy(messages_init)
        if messages_init
        else [{"role": "system", "content": default_system_prompt}]
    )
    current_action = Start
    current_action_result = None
    while True:
        next_action_options = action_graph(
            current_action=current_action, current_action_result=current_action_result
        )
        if not next_action_options:
            break
        # DECIDE NEXT ACTION
        if len(next_action_options) == 1:
            current_action = next_action_options[0]
        else:
            pre_llm_callback(messages)
            select_next_action: callable = select_next_action_factory(next_action_options)
            func_arg_message, func_arg = get_func_input_from_llm(
                messages, select_next_action, completion
            )
            messages.append(func_arg_message)
            yield deepcopy(messages)
            pre_llm_callback(messages)
            func_result_message, func_result = run_func_for_llm(
                select_next_action, func_arg
            )
            messages.append(func_result_message)
            yield deepcopy(messages)
            current_action = next(
                x for x in next_action_options if description(x)["name"] == func_result
            )
        if current_action == End:
            break
        # RUN THE ACTION
        if description(current_action)["parameters"]:
            pre_llm_callback(messages)
            func_arg_message, func_arg = get_func_input_from_llm(
                messages, current_action, completion
            )
        else:
            func_arg_message = {
                "role": "assistant",
                "content": None,
                "function_call": {
                    "name": description(current_action)["name"],
                    "arguments": "null",
                },
            }
            func_arg = None
        messages.append(func_arg_message)
        yield deepcopy(messages)
        pre_llm_callback(messages)
        func_result_message, func_result = run_func_for_llm(current_action, func_arg)
        messages.append(func_result_message)
        yield deepcopy(messages)
        current_action_result = func_result


def dict_to_action_graph_func(action_graph: dict) -> callable:
    def action_graph_func(
        current_action: callable, current_action_result: Union[dict, None]
    ):
        return action_graph.get(current_action, None)

    return action_graph_func

make_agents/make_agents.py
sradc/MakeAgents
from pathlib import Path

import make_agents

if __name__ == "__main__":
    main_page_path = Path("docs/index.html")
    if main_page_path.read_text().find(make_agents.__version__) == -1:
        raise RuntimeError(
            f"Version mismatch between docs and package: {main_page_path} does not contain {make_agents.__version__}"
        )

scripts/assert_docs_up_to_date.py
sradc/MakeAgents
"""Replace relative links with absolute links in README.md, so images also work on PyPI.
"""
import re

if __name__ == "__main__":
    FILE_PATH = "README.md"
    GITHUB_PATH = "sradc/MakeAgents/main"

    with open(FILE_PATH, "r") as file:
        data = file.read()

    # Find links referencing README_files/ and replace them with absolute links
    pattern = r"(README_files/.*\.(png|jpg))"
    replacement = r"https://raw.githubusercontent.com/" + GITHUB_PATH + r"/\1"

    # Substitute the pattern with the replacement
    new_data = re.sub(pattern, replacement, data)

    with open(FILE_PATH, "w") as file:
        file.write(new_data)

scripts/replace_readme_image_links.py
sradc/MakeAgents
import pytest
from pydantic import BaseModel, Field

from make_agents.make_agents import action


def test_action_decorator():
    class ExampleFuncArg(BaseModel):
        example_field: str = Field(description="This is an example field.")

    @action
    def example_func(arg: ExampleFuncArg):
        """This is an example docstring"""
        return arg.example_field

    assert example_func.description_for_llm == {
        "name": "example_func",
        "description": "This is an example docstring",
        "parameters": {
            "properties": {
                "example_field": {
                    "description": "This is an example field.",
                    "title": "Example Field",
                    "type": "string",
                }
            },
            "required": ["example_field"],
            "title": "ExampleFuncArg",
            "type": "object",
        },
    }, "Did not get expected metadata."

    assert (
        example_func(ExampleFuncArg(example_field="example")) == "example"
    ), "Expected to be able to use function as normal."

    # Check that the function raises an error if there are multiple parameters
    with pytest.raises(ValueError):

        @action
        def example_func(arg: ExampleFuncArg, arg2: ExampleFuncArg):
            pass

    # Check that the function doesn't raise an error if there are no parameters
    @action
    def example_func():
        pass

    # Check that the function raises an error if type hint not pydantic
    with pytest.raises(ValueError):

        @action
        def example_func(arg: str):
            pass

    # Check that the function raises an error if no type hint
    with pytest.raises(ValueError):

        @action
        def example_func(arg):
            pass

tests/test_make_agents.py
sradc/MakeAgents