Spaces:
Sleeping
Sleeping
| 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": "<Reasoning step>", | |
| "action": "<Action taken>", | |
| "observation": "<Result of the action>" | |
| }} | |
| ], | |
| "final_result": "<Final answer or solution>" | |
| }} | |
| 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.") | |