import streamlit as st import os from groq import Groq from dotenv import load_dotenv import logging import json import re # Configure logging logging.basicConfig( filename="app.log", level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s", ) logging.debug("Logging is configured correctly.") # Load environment variables load_dotenv() reflection_cycles = 2 # Define the Groq API key and initialize the client GROQ_API_KEY = os.environ.get("GROQ_API_KEY") if not GROQ_API_KEY: raise ValueError("API Key is not set. Please check your environment variables or .env file.") client = Groq(api_key=GROQ_API_KEY) # Define the ReAct system prompt template SYSTEM_PROMPT_TEMPLATE = """ You are an advanced AI agent using the ReAct (Reasoning + Action) framework to solve complex tasks. Follow these steps iteratively: 1. Generate a "Thought" based on the current input or observations. 2. Decide on an "Action" (e.g., search, calculation, etc.) to take. 3. Return an "Observation" after the action to guide the next step. Continue this loop until the task is solved or no further actions are needed. Return the result in this JSON format: {{ "thoughts": [ {{ "thought": "", "action": "", "observation": "" }} ], "final_result": "" }} Previous Context: {history_context} Input: {user_input} """ # Initialize Streamlit app st.title("ReAct AI Chatbot") # Initialize session state if "messages" not in st.session_state: st.session_state.messages = [] if "react_history" not in st.session_state: st.session_state.react_history = [] def sanitize_json(json_str): json_str = re.sub(r"[\x00-\x1F\x7F]", "", json_str) # Remove control characters return json_str def generate_react_response(user_input, react_history): """ Generate a ReAct-based response for the given input. """ try: # Combine history context MAX_HISTORY = 5 history_context = "\n".join(react_history[-MAX_HISTORY:]) logging.debug(f"History Context: {history_context}") # Ensure the user_input is sanitized user_input = sanitize_json(user_input) logging.debug(f"Sanitized User Input: {user_input}") # Format the system prompt formatted_prompt = SYSTEM_PROMPT_TEMPLATE.format_map({ "user_input": user_input, "history_context": history_context or "No context available." }) logging.debug(f"Formatted Prompt: {formatted_prompt}") # Send the request to the Groq API chat_completion = client.chat.completions.create( model="llama3-8b-8192", messages=[ {"role": "system", "content": formatted_prompt}, {"role": "user", "content": user_input}, ], max_tokens=2048, temperature=0.7, top_p=0.9, ) logging.debug(f"Raw API Response: {chat_completion}") # Extract content from the response content = chat_completion.choices[0].message.content if not content: logging.warning("Received empty content in API response.") return None # Updated regex patterns to capture full content including recipe details thought_match = re.search(r'\*\*Thought:?\*\*:?\s*"?(.*?)(?="?\s*\*\*Action|\n\n|$)', content, re.DOTALL | re.IGNORECASE) action_match = re.search(r'\*\*Action:?\*\*:?\s*"?(.*?)(?="?\s*\*\*Observation|\n\n|$)', content, re.DOTALL | re.IGNORECASE) observation_match = re.search(r'\*\*Observation:?\*\*:?\s*"?(.*?)(?=\n\n\*\*Thought|\Z)', content, re.DOTALL | re.IGNORECASE) # Extract and clean the matches thought = thought_match.group(1).strip(' "') if thought_match else "No thought provided" action = action_match.group(1).strip(' "') if action_match else "No action provided" observation = observation_match.group(1).strip(' "') if observation_match else "No observation provided" # Check if observation contains a recipe (indicated by "Ingredients:" or "Instructions:") if "Ingredients:" in observation or "Instructions:" in observation: final_result = observation # Use the full recipe text as the final result else: final_result = observation if observation != "No observation provided" else "Ready to provide assistance once preferences are specified." parsed_response = { "thoughts": [{ "thought": thought, "action": action, "observation": observation }], "final_result": final_result } logging.debug(f"Parsed Response: {parsed_response}") return parsed_response except Exception as e: logging.error(f"Error generating ReAct response: {e}", exc_info=True) return { "thoughts": [], "final_result": f"An error occurred: {str(e)}", } # Display chat messages from history for message in st.session_state.messages: with st.chat_message(message["role"]): st.markdown(message["content"]) # Accept user input user_input = st.chat_input("Enter your query:") if user_input: # Display user message st.chat_message("user").markdown(user_input) st.session_state.messages.append({"role": "user", "content": user_input}) # Generate ReAct-based response with st.spinner("Thinking..."): response = generate_react_response(user_input, st.session_state.react_history) if response: try: # Process thoughts and actions thoughts = response.get("thoughts", []) for step in thoughts: thought = step.get("thought", "No thought provided.") action = step.get("action", "No action taken.") observation = step.get("observation", "No observation.") st.chat_message("assistant").markdown( f"**Thought:** {thought}\n\n**Action:** {action}\n\n**Observation:** {observation}" ) st.session_state.messages.append( { "role": "assistant", "content": f"**Thought:** {thought}\n\n**Action:** {action}\n\n**Observation:** {observation}", } ) # Final result final_result = response.get("final_result", "No final result.") st.chat_message("assistant").markdown(f"**Final Result:** {final_result}") st.session_state.messages.append({"role": "assistant", "content": f"**Final Result:** {final_result}"}) # Update history st.session_state.react_history.append(f"User: {user_input}\nAI: {final_result}") except Exception as e: logging.error(f"Error processing ReAct response: {e}") st.error("Failed to process the ReAct response.")