- Create a new directory with these three files (requirements.txt, main.py, README.md)
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt
python main.py
- Update
main()
to run the example prompt chains
Created
March 31, 2024 14:34
-
-
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)
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
llm | |
openai | |
python-dotenv | |
llm-claude-3 |
Very useful. Thanks a lot. I learn a lot from your videos.
Thank you for sharing these.
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.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
You da man, Dan. 🙏