Skip to content

Instantly share code, notes, and snippets.

@disler
Created March 31, 2024 14:34
Show Gist options
  • Save disler/409d9685c8b251ed723a7aca43cc4b9b to your computer and use it in GitHub Desktop.
Save disler/409d9685c8b251ed723a7aca43cc4b9b to your computer and use it in GitHub Desktop.
Use these Prompt Chains to build HIGH QUALITY AI Agents (Agentic Building Blocks)

Setup

  1. Create a new directory with these three files (requirements.txt, main.py, README.md)
  2. python -m venv venv
  3. source venv/bin/activate
  4. pip install -r requirements.txt
  5. python main.py
  6. Update main() to run the example prompt chains
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()
llm
openai
python-dotenv
llm-claude-3
@itsbrex
Copy link

itsbrex commented Apr 6, 2024

You da man, Dan. 🙏

@alexsohr
Copy link

alexsohr commented Apr 9, 2024

Very useful. Thanks a lot. I learn a lot from your videos.

@ElliotWood
Copy link

Thank you for sharing these.

@code1zero
Copy link

Thank you

@piovis2023
Copy link

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

@alloke613
Copy link

Awesome, thanks for sharing.

@bleszily
Copy link

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment