Use these Prompt Chains to build HIGH QUALITY AI Agents (Agentic Building Blocks)

README.md

Setup

0. Create a new directory with these three files (requirements.txt, main.py, README.md)
1. python -m venv venv
2. source venv/bin/activate
3. pip install -r requirements.txt
4. python main.py
5. Update main() to run the example prompt chains

main.py

import json
import os
import random
from dotenv import load_dotenv
import llm
import time

load_dotenv()

# ------ Helpers Methods


def build_models():

    ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")

    haiku_model = llm.get_model("claude-3-haiku")
    haiku_model.key = ANTHROPIC_API_KEY

    sonnet_model = llm.get_model("claude-3-sonnet")
    sonnet_model.key = ANTHROPIC_API_KEY

    opus_model = llm.get_model("claude-3-opus")
    opus_model.key = ANTHROPIC_API_KEY

    return haiku_model, sonnet_model, opus_model


def coin_flip():
    return random.randint(0, 1)


# ------ Prompt Chains


def prompt_chain_snowball(haiku_model):
    """
    Snowball prompt - start with a little information, that is developed over each prompt.

    Use Case
    - Blogs
    - Newsletters
    - Research
    - Summaries

    Mermaid Diagram
    A[Start]
    B{Base Information}
    C[Snowball Prompt 1]
    D[Snowball Prompt 2]
    E[Snowball Prompt 3]
    F[Summary/Format Prompt]
    G[End]

    A --> B --> C --> D --> E --> F --> G
    """

    base_information = "3 Unusual Use Cases for LLMs"

    snowball_prompt_response_1 = haiku_model.prompt(
        f"Generate a clickworthy title about this topic: '{base_information}'. Respond in JSON format {{title: 'title', topic: '{base_information}'}}"
    )

    print("Snowball #1: ", snowball_prompt_response_1.text())

    snowball_prompt_response_2 = haiku_model.prompt(
        f"Generate a compelling 3 section outline given this information: '{snowball_prompt_response_1.text()}'. Respond in JSON format {{title: '<title>', topic: '<topic>', sections: ['<section1>', '<section2>', '<section3>']}}"
    )

    print("Snowball #2: ", snowball_prompt_response_2.text())

    snowball_prompt_response_3 = haiku_model.prompt(
        f"Generate 1 paragraph of content for each section outline given this information: '{snowball_prompt_response_2.text()}'. Respond in JSON format {{title: '<title>', topic: '<topic>', sections: ['<section1>', '<section2>', '<section3>'], content: ['<content1>', '<content2>', '<content3>']}}"
    )

    print("Snowball #3: ", snowball_prompt_response_3.text())

    snowball_markdown_prompt = haiku_model.prompt(
        f"Generate a markdown formatted blog post given this information: '{snowball_prompt_response_3.text()}'. Respond in JSON format {{markdown_blog: '<entire markdown blog post>'}}"
    )

    print("Final Snowball: ", snowball_markdown_prompt.text())

    with open("snowball_prompt_chain.txt", "w") as file:
        file.write(snowball_markdown_prompt.text())

    pass


def prompt_chain_workers(haiku_model, sonnet_model, opus_model):
    """
    Delegate different parts of your workload to individual prompt workers.

    Use Case
    - Research
    - Parallelization
    - Autocomplete
    - Divide and Conquer
    - Similar Tasks More Scalable

    Mermaid Diagram
    A[Start]
    B[Plan Prompt]
    C[Worker Prompt 1]
    D[Worker Prompt 2]
    E[Worker Prompt 3]
    F[Summary/Format Prompt]
    G[End]

    A --> B --> C
    B --> D
    B --> E
    E --> F
    C --> F
    D --> F
    F --> G
    """

    print("Generating function stubs...")

    code_planner_prompt_response = opus_model.prompt(
        '''Create the function stubs with detailed comments of how to write the code to build write_json_file, write_yml_file, write_toml_file. 
Example Function Stub:
def read_toml_file(file_path: str) -> str:
    """
    Read the file at the given file path and return the contents as a string.

    Use a toml parser to parse the file.

    Usage:
    read_toml_file("path/to/file.toml")
    """
    pass

Respond in json format {function_stubs: ["def function_stub1...", "def function_stub2...", ...]}'''
    )
    code_planner_result = code_planner_prompt_response.text()

    print(code_planner_result)

    function_stubs = json.loads(code_planner_result)["function_stubs"]

    function_stub_raw_results = ""

    for stub in function_stubs:

        time.sleep(0.5)

        print(f"\n\nImplementing function stub in worker prompt...\n\n")

        code_executor_prompt_response = opus_model.prompt(
            f"Implement this function stub in python: {stub}. Assume all imports are already installed. Respond exclusively with runnable code in json format {{code: '<code>'}}"
        )
        print(code_executor_prompt_response.text())

        function_stub_raw_results += code_executor_prompt_response.text()

    print(function_stub_raw_results)

    final_module = opus_model.prompt(
        f"Clean up and combine the following python code and combine every code stub into a final python file that can be executed. Code: {function_stub_raw_results}. Respond exclusively with the finalized code in json format {{code: '<code>'}}"
    )

    print(final_module.text())

    # write to python file
    with open("files.py", "w") as file:
        file.write(final_module.text())

    pass


def prompt_chain_fallback(haiku_model, sonnet_model, opus_model):
    """
    Fallback Prompt Chain

    Use Case
    - Save Money (Cheap LLMs first)
    - Save Time (Fast LLMs first)
    - Generating Content
    - Specific Formats
    - Last Resort

    Mermaid Diagram
    A[Start]
    B[Top Priority Prompt/Model]
    C[Run Process]
    D[Secondary Fallback Prompt/Model]
    E[Run Process]
    F[Final Fallback Prompt/Model]
    G[End]

    A --> B --> C --> D --> E --> F --> G

    C --> G
    E --> G
    """

    def run_fallback_flow(evaluator_function, fallback_functions):
        for fallback_function, model_name in fallback_functions:
            response = fallback_function()
            print(f"{model_name} Response: ", response.text())

            success = evaluator_function(response.text())

            if success:
                print(f"{model_name} Success - Returning")
                return True
            else:
                print(f"{model_name} Failed - Fallback")

        print("All Fallback Functions Failed")
        return False

    def run_code(code):
        """
        Fake run code that returns a random boolean.
        50% chance of returning True.
        """
        return coin_flip()

    function_generation_prompt = f"Generate the solution in python given this function definition: 'def text_to_speech(text) -> Bytes'. Respond in JSON format {{python_code: '<python code>'}}"

    fallback_functions = [
        (
            lambda: haiku_model.prompt(function_generation_prompt),
            "(Haiku) Cheap, Fast Top Priority Prompt/Model",
        ),
        (
            lambda: sonnet_model.prompt(function_generation_prompt),
            "(Sonnet) Cheap, Moderate Secondary Fallback Prompt/Model",
        ),
        (
            lambda: opus_model.prompt(function_generation_prompt),
            "(Opus) Expensive, Slow, Accurate Final Fallback Prompt/Model",
        ),
    ]

    success = run_fallback_flow(run_code, fallback_functions)

    print(f"Fallback Flow was {'✅ Successful' if success else '❌ Unsuccessful'}")

    pass


def prompt_chain_decision_maker(haiku_model):
    """
    Based on a decision from a prompt, run a different prompt chain.

    Use Case
    - Creative Direction
    - Dictate Flow Control
    - Decision Making
    - Dynamic Prompting
    - Multiple Prompts

    Mermaid Diagram
    A[Start]
    B[Decision Prompt]
    C[Prompt Chain 1]
    D[Prompt Chain 2]
    E[Prompt Chain 3]
    F[End]

    A --> B
    B --IF--> C --> F
    B --IF--> D --> F
    B --IF--> E --> F
    """

    mock_live_feed = [
        "Our revenue exceeded expectations this quarter, driven by strong sales in our core product lines.",
        "We experienced some supply chain disruptions that impacted our ability to meet customer demand in certain regions.",
        "Our new product launch has been well-received by customers and is contributing significantly to our growth.",
        "We incurred higher than expected costs related to our expansion efforts, which put pressure on our margins.",
        "We are seeing positive trends in customer retention and loyalty, with many customers increasing their spend with us.",
        "The competitive landscape remains challenging, with some competitors engaging in aggressive pricing strategies.",
        "We made significant progress on our sustainability initiatives this quarter, reducing our carbon footprint and waste.",
        "We had to write off some inventory due to changing consumer preferences, which negatively impacted our bottom line.",
    ]

    live_feed = random.choice(mock_live_feed)

    print(f"Analyzing Sentiment Of Latest Audio Clip: '{live_feed}'")

    sentiment_analysis_prompt_response = haiku_model.prompt(
        f"Analyze the sentiment of the following text as either positive or negative: '{live_feed}'. Respond in JSON format {{sentiment: 'positive' | 'negative'}}."
    )

    sentiment = json.loads(sentiment_analysis_prompt_response.text())["sentiment"]

    def positive_sentiment_action(live_feed):
        positive_sentiment_thesis_prompt_response = haiku_model.prompt(
            f"The following text has a positive sentiment: '{live_feed}'. Generate a short thesis statement about why the sentiment is positive."
        )
        print(
            f"\n\nPositive Sentiment Thesis: {positive_sentiment_thesis_prompt_response.text()}"
        )

    def negative_sentiment_action(live_feed):
        negative_sentiment_thesis_prompt_response = haiku_model.prompt(
            f"The following text has a negative sentiment: '{live_feed}'. Generate a short thesis statement about why the sentiment is negative."
        )
        print(
            f"\n\nNegative Sentiment Thesis: {negative_sentiment_thesis_prompt_response.text()}\n\n"
        )

    def unknown_sentiment_action(_):
        print(
            f"Could not determine sentiment. Raw response: {sentiment_analysis_prompt_response.text()}"
        )

    sentiment_actions = {
        "positive": positive_sentiment_action,
        "negative": negative_sentiment_action,
    }

    sentiment_action = sentiment_actions.get(sentiment, unknown_sentiment_action)

    sentiment_action(live_feed)

    pass


def prompt_chain_plan_execute(haiku_model):
    """
    Plan Execute Prompt Chain

    Use Case
    - Tasks
    - Projects
    - Research
    - Coding

    Mermaid Diagram
    A[Start]
    B<Plan Prompt>
    C<Execute Prompt>
    D<End>

    A --> B --> C --> D
    """

    task = "Design the software architecture for an AI assistant that uses tts, llms, local sqlite. "

    plan_prompt_response = haiku_model.prompt(
        f"Let's think step by step about how we would accomplish this task: '{task}'. Write all the steps, ideas, variables, mermaid diagrams, use cases, and examples concisely in markdown format. Respond in JSON format {{markdown_plan: '<plan>'}}"
    )

    print(plan_prompt_response.text())

    execute_prompt_response = haiku_model.prompt(
        f"Create a detailed architecture document on how to execute this task '{task}' given this detailed plan {plan_prompt_response.text()}.  Respond in JSON format {{architecture_document: '<document>'}}"
    )

    print(execute_prompt_response.text())

    # write the plan and execute to a file
    with open("plan_execute_prompt_chain.txt", "w") as file:
        file.write(execute_prompt_response.text())


def prompt_chain_human_in_the_loop(haiku_model, sonnet_model):
    """
    Human In The Loop Prompt Chain

    Use Case
    - Human In The Loop
    - Validation
    - Content Creation
    - Coding
    - Chat

    Mermaid Diagram
    A[Start]
    B[Initial Prompt]
    C[Human Feedback]
    D[Iterative Prompt]
    E[End]

    A --> B --> C --> D
    D --> C
    D --> E
    """
    topic = "Personal AI Assistants"
    prompt = f"Generate 5 ideas surrounding this topic: '{topic}'"
    result = haiku_model.prompt(prompt).text()
    print(result)

    while True:
        user_input = input("Iterate on result or type 'done' to finish: ")
        if user_input.lower() == "done":
            break
        else:
            prompt += f"\n\n----------------\n\nPrevious result: {result}\n\n----------------\n\nIterate on the previous result and generate 5 more ideas based on this feedback: {user_input}"
            result = sonnet_model.prompt(prompt).text()
            print(result + "\n\n----------------\n\n")

    pass


def prompt_chain_self_correct(haiku_model):
    """
    Self correct/review the output of a prompt.

    Use Case
    - Coding
    - Execution
    - Self Correct
    - Review
    - Iterate
    - Improve

    Mermaid Diagram
    A[Start]
    B[Prompt]
    C[Execute Output]
    D[Self Correct]
    E[End]

    A --> B --> C --> D --> E
    C --> E
    """

    def run_bash(command):
        print(f"Running command: {command}")
        if coin_flip() == 0:
            return "Mock error: command failed to execute properly"
        else:
            return "Command executed successfully"

    outcome = "list all files in the current directory"

    initial_response = haiku_model.prompt(
        f"Generate a bash command that enables us to {outcome}. Respond with only the command."
    )
    print(f"Initial response: {initial_response.text()}")

    # Run the generated command and check for errors
    result = run_bash(initial_response.text())

    if "error" in result.lower():

        print("Received error, running self-correction prompt")

        # If error, run self-correction prompt
        self_correct_response = haiku_model.prompt(
            f"The following bash command was generated to {outcome}, but encountered an error when run:\n\nCommand: {initial_response.text()}\nError: {result}\n\nPlease provide an updated bash command that will successfully {outcome}. Respond with only the updated command in JSON format {{command: '<command>'}}."
        )
        print(f"Self-corrected response: {self_correct_response.text()}")

        # Run the self-corrected command
        run_bash(self_correct_response.text())

    else:
        print(f"Original command executed successfully: {result}")


def main():

    haiku_model, sonnet_model, opus_model = build_models()

    prompt_chain_snowball(haiku_model)

    # prompt_chain_workers(haiku_model, sonnet_model, opus_model)

    # prompt_chain_fallback(haiku_model, sonnet_model, opus_model)

    # prompt_chain_decision_maker(haiku_model)

    # prompt_chain_plan_execute(haiku_model)

    # prompt_chain_human_in_the_loop(haiku_model, sonnet_model)

    # prompt_chain_self_correct(haiku_model)


if __name__ == "__main__":
    main()

requirements.txt

llm
openai
python-dotenv
llm-claude-3

Thank you

@disler Excellent video (as always). Can you share some code where you've made one of these chains do an if then tool execution please?

For example,
If you find data numerical data, save it in googlesheets.
If you find non-numerical data (instructions/ comments) save it to googledocs.

This would be incredible!
Thanks in advance.

Awesome, thanks for sharing.

Great content. Thanks for sharing this. Building UC to help me address some challenges at work.