import os import streamlit as st import openai import pandas as pd from typing import List, Tuple from uuid import uuid4 import time # 1. 🛠️ Configuration Site_Name = '🌟 Self-Taught Reasoner (STaR) App' title = "🤔🌟 STaR: Self-Taught Reasoner - Bootstrapping Reasoning With Reasoning" helpURL = 'https://arxiv.org/abs/2203.14465' bugURL = 'https://arxiv.org/pdf/2203.14465' icons = '🌟🤔' useConfig = True if useConfig: st.set_page_config( page_title=title, page_icon=icons, layout="wide", initial_sidebar_state="auto", menu_items={ 'Get Help': helpURL, 'Report a bug': bugURL, 'About': title } ) # 🔑 Set the OpenAI API key from an environment variable openai.api_key = os.getenv("OPENAI_API_KEY") # 🆔 Function to generate a unique session ID for caching def get_session_id(): if 'session_id' not in st.session_state: st.session_state.session_id = str(uuid4()) return st.session_state.session_id # 🧠 STaR Algorithm Implementation class SelfTaughtReasoner: def __init__(self, model_engine="gpt-3.5-turbo"): self.model_engine = model_engine self.prompt_examples = [] # Initialize with an empty list self.iterations = 0 self.generated_data = pd.DataFrame(columns=['Problem', 'Rationale', 'Answer', 'Is_Correct']) self.rationalized_data = pd.DataFrame(columns=['Problem', 'Rationale', 'Answer', 'Is_Correct']) self.fine_tuned_model = None # 🏗️ Placeholder for fine-tuned model def add_prompt_example(self, problem: str, rationale: str, answer: str): """ ➕ Adds a prompt example to the few-shot examples. """ self.prompt_examples.append({ 'Problem': problem, 'Rationale': rationale, 'Answer': answer }) def construct_prompt(self, problem: str, include_answer: bool = False, answer: str = "") -> List[dict]: """ 📝 Constructs the prompt for the OpenAI API call. Converts examples into the new chat format, where each example is a user message. """ messages = [] for example in self.prompt_examples: messages.append({"role": "system", "content": f"Problem: {example['Problem']}\nRationale: {example['Rationale']}\nAnswer: {example['Answer']}\n"}) messages.append({"role": "user", "content": f"Problem: {problem}\nRationale:"}) if include_answer: messages.append({"role": "system", "content": f"Answer: {answer}"}) return messages def generate_rationale_and_answer(self, problem: str) -> Tuple[str, str]: """ 🤔 Generates a rationale and answer for a given problem using openai.ChatCompletion.create. """ messages = self.construct_prompt(problem) try: response = openai.ChatCompletion.create( model=self.model_engine, messages=messages, max_tokens=150, temperature=0.7 ) rationale = response.choices[0].message['content'].strip() # Now generate the answer using the rationale messages.append({"role": "system", "content": f"Rationale: {rationale}\nAnswer:"}) answer_response = openai.ChatCompletion.create( model=self.model_engine, messages=messages, max_tokens=10, temperature=0 ) answer = answer_response.choices[0].message['content'].strip() return rationale, answer except Exception as e: st.error(f"❌ Error generating rationale and answer: {e}") return "", "" def fine_tune_model(self): """ 🛠️ Fine-tunes the model on the generated rationales. """ time.sleep(1) # ⏳ Simulate time taken for fine-tuning self.fine_tuned_model = f"{self.model_engine}-fine-tuned-{get_session_id()}" st.success(f"✅ Model fine-tuned: {self.fine_tuned_model}") def run_iteration(self, dataset: pd.DataFrame): """ 🔄 Runs one iteration of the STaR process. """ st.write(f"### Iteration {self.iterations + 1}") progress_bar = st.progress(0) total = len(dataset) for idx, row in dataset.iterrows(): problem = row['Problem'] correct_answer = row['Answer'] # 🤖 Generate rationale and answer rationale, answer = self.generate_rationale_and_answer(problem) is_correct = (answer.lower() == correct_answer.lower()) # 📝 Record the generated data self.generated_data = self.generated_data.append({ 'Problem': problem, 'Rationale': rationale, 'Answer': answer, 'Is_Correct': is_correct }, ignore_index=True) # ❌ If incorrect, perform rationalization if not is_correct: rationale, answer = self.rationalize(problem, correct_answer) is_correct = (answer.lower() == correct_answer.lower()) if is_correct: self.rationalized_data = self.rationalized_data.append({ 'Problem': problem, 'Rationale': rationale, 'Answer': answer, 'Is_Correct': is_correct }, ignore_index=True) progress_bar.progress((idx + 1) / total) # 🔧 Fine-tune the model on correct rationales st.write("🔄 Fine-tuning the model on correct rationales...") self.fine_tune_model() self.iterations += 1 # Predefined problem and answer list for dataset EXAMPLE_PROBLEM_ANSWERS = [ {"Problem": "What is deductive reasoning?", "Answer": "It is a logical process that draws specific conclusions from general principles."}, {"Problem": "What is inductive reasoning?", "Answer": "It is reasoning that forms general principles from specific examples."}, {"Problem": "Explain abductive reasoning.", "Answer": "It involves finding the best explanation for incomplete observations."}, {"Problem": "What is the capital of France?", "Answer": "Paris."}, {"Problem": "Who wrote Hamlet?", "Answer": "William Shakespeare."} ] # Additional problem set for testing fine-tuned model TEST_PROBLEM_SET = [ "What is the Pythagorean theorem?", "Who developed the theory of relativity?", "What is the main ingredient in bread?", "Who is the author of 1984?", "What is the boiling point of water?" ] # Convert the example list into 'Problem | Answer' format def format_examples_for_text_area(examples): return '\n'.join([f"{example['Problem']} | {example['Answer']}" for example in examples]) # 🖥️ Streamlit App def main(): st.title("🤖 Self-Taught Reasoners (STaR)") st.markdown(''' # 📄 Papers: 1. 🤫🌟 Quiet-STaR: Language Models Can Teach Themselves to Think 🤔 Before Speaking 🗣️ - 🔗 https://arxiv.org/abs/2403.09629 - 📄 https://arxiv.org/pdf/2403.09629 2. 🌟🤔 STaR: Self-Taught Reasoner - Bootstrapping Reasoning With Reasoning - 🔗 https://arxiv.org/abs/2203.14465 - 📄 https://arxiv.org/pdf/2203.14465 ''') # 🧩 Initialize the Self-Taught Reasoner if 'star' not in st.session_state: st.session_state.star = SelfTaughtReasoner() star = st.session_state.star # Step 1: Few-Shot Prompt Examples st.header("Step 1: Add Few-Shot Prompt Examples") st.write("Choose an example from the dropdown or input your own.") selected_example = st.selectbox( "Select a predefined example", [f"Example {i + 1}: {ex['Problem']}" for i, ex in enumerate(EXAMPLE_PROBLEM_ANSWERS)] ) # Prefill with selected example example_idx = int(selected_example.split(" ")[1].replace(":", "")) - 1 example_problem = EXAMPLE_PROBLEM_ANSWERS[example_idx]['Problem'] example_answer = EXAMPLE_PROBLEM_ANSWERS[example_idx]['Answer'] st.text_area("Problem", value=example_problem, height=50, key="example_problem") st.text_input("Answer", value=example_answer, key="example_answer") if st.button("Add Example"): star.add_prompt_example(st.session_state.example_problem, "Rationale placeholder", st.session_state.example_answer) st.success("Example added successfully!") # Step 2: Input Dataset (Problem | Answer format) st.header("Step 2: Input Dataset") # Provide examples in the format 'Problem | Answer' as a default prefilled_data = format_examples_for_text_area(EXAMPLE_PROBLEM_ANSWERS) dataset_problems = st.text_area( "Enter problems and answers in the format 'Problem | Answer', one per line.", value=prefilled_data, height=200 ) if st.button("Submit Dataset"): dataset = [] lines = dataset_problems.strip().split('\n') for line in lines: if '|' in line: problem, answer = line.split('|', 1) dataset.append({'Problem': problem.strip(), 'Answer': answer.strip()}) st.session_state.dataset = pd.DataFrame(dataset) st.success("Dataset loaded.") if 'dataset' in st.session_state: st.subheader("Current Dataset:") st.dataframe(st.session_state.dataset.head()) # Step 3: Test the Fine-Tuned Model (renamed from Step 4) st.header("Step 3: Test the Fine-Tuned Model") # Add dropdown for selecting a test problem test_problem = st.selectbox( "Select a problem to test the fine-tuned model", TEST_PROBLEM_SET ) if st.button("Solve Problem"): if not test_problem: st.warning("Please enter or select a problem to solve.") else: rationale, answer = star.generate_rationale_and_answer(test_problem) st.subheader("Rationale:") st.write(rationale) st.subheader("Answer:") st.write(answer) # Footer st.write("---") st.write("Developed as a demonstration of the STaR method.") if __name__ == "__main__": main()