Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| import re | |
| from duckduckgo_search import DDGS | |
| from langchain_groq import ChatGroq | |
| from langchain_core.output_parsers import StrOutputParser | |
| # --- Tool Definitions --- | |
| def search_tool(query: str): | |
| """A web search tool that uses DuckDuckGo to find information.""" | |
| st.write(f"๐ Searching the web for: `{query}`") | |
| try: | |
| with DDGS() as ddgs: | |
| results = [r["body"] for r in ddgs.text(query, max_results=5)] | |
| return "\n".join(results) if results else "No results found." | |
| except Exception as e: | |
| return f"An error occurred during the search: {e}" | |
| def calculator_tool(expression: str): | |
| """A calculator tool that evaluates a mathematical expression.""" | |
| st.write(f"๐งฎ Calculating: `{expression}`") | |
| try: | |
| if not re.match(r"^[0-9+\-*/.() \t]+$", expression): | |
| return "Invalid characters detected. Calculation aborted." | |
| result = eval(expression) | |
| return f"Result: {result}" | |
| except Exception as e: | |
| return f"Calculation error: {e}" | |
| # --- LLM Call Function (Groq) --- | |
| def call_llm(prompt, api_key): | |
| try: | |
| chat = ChatGroq( | |
| groq_api_key=api_key, | |
| model_name="llama3-8b-8192" | |
| ) | |
| output_parser = StrOutputParser() | |
| result = output_parser.invoke(chat.invoke(prompt)) | |
| cleaned = re.sub(r"[*`]", "", result) | |
| cleaned = re.sub(r"[โโ]", '"', cleaned) | |
| return cleaned.strip() | |
| except Exception as e: | |
| st.error(f"Error during Groq API call: {e}") | |
| return None | |
| # --- Agent Reasoning Loop --- | |
| def run_agent_loop(goal, api_key): | |
| system_prompt_template = f""" | |
| You are an AI agent that thinks step-by-step to accomplish a goal using available tools. | |
| At every step, you must generate a **Thought** about the goal and then choose an **Action** to execute that thought. | |
| You can use the following tools: | |
| 1. `search(query)`: For searching information on the web. Example: `search("capital of Turkey")` | |
| 2. `calculator(expression)`: For performing a mathematical calculation. Example: `calculator(100 * (3 + 5))` | |
| 3. `finish(answer)`: When you reach the final answer and want to complete the task. Example: `finish("Ankara is the capital of Turkey.")` | |
| โ ๏ธ You must **always** reply in the following strict format: | |
| Thought: [your reasoning in one sentence] | |
| Action: search("...") or calculator("...") or finish("...") | |
| โ ๏ธ Do **not** explain the tools, do not use markdown, do not write extra comments. | |
| Respond with only **two lines**: one Thought and one Action. | |
| Your goal is: "{goal}" | |
| Now, give your first Thought and Action. | |
| """ | |
| history = [system_prompt_template] | |
| max_steps = 10 | |
| for step in range(max_steps): | |
| st.write("---") | |
| st.info(f"๐ Step {step + 1}") | |
| prompt = "\n".join(history) | |
| with st.spinner("Thinking..."): | |
| response = call_llm(prompt, api_key) | |
| if not response: | |
| st.error("The agent failed to produce a response.") | |
| break | |
| thought_match = re.search(r"Thought(?: \(.*?\))?:\s*(.*)", response) | |
| action_match = re.search(r"Action(?: \(.*?\))?:\s*(.*)", response) | |
| if not thought_match or not action_match: | |
| st.error("The agent did not produce a valid 'Thought' or 'Action' format.") | |
| st.write(response) | |
| break | |
| thought = thought_match.group(1).strip() | |
| action_full = action_match.group(1).strip() | |
| with st.expander("๐ค Agent's Thought Process", expanded=True): | |
| st.markdown(f"**Thought:** {thought}") | |
| st.markdown(f"**Action:** `{action_full}`") | |
| action_name_match = re.match(r"(\w+)\((.*)\)", action_full) | |
| if not action_name_match: | |
| st.error(f"Invalid action format: {action_full}") | |
| break | |
| action_name = action_name_match.group(1) | |
| action_input = action_name_match.group(2).strip(' "\'') | |
| if action_name == "search": | |
| observation = search_tool(action_input) | |
| elif action_name == "calculator": | |
| observation = calculator_tool(action_input) | |
| elif action_name == "finish": | |
| st.success("โ The agent has completed the task!") | |
| st.balloons() | |
| st.markdown("### Final Answer:") | |
| st.write(action_input) | |
| break | |
| else: | |
| observation = f"Unknown action: {action_name}" | |
| st.warning(f"**Observation:**\n{observation}") | |
| history.append(response) | |
| history.append(f"Observation: {observation}") | |
| else: | |
| st.error("The agent reached the maximum number of steps without completing the task.") | |
| # --- Streamlit UI --- | |
| st.set_page_config(page_title="AI Agentic App", page_icon="๐ค") | |
| st.title("๐ค AI Agentic App") | |
| st.markdown(""" | |
| This application is an AI agent that operates using the **ReAct (Reason + Act)** paradigm. | |
| The agent thinks, takes actions (like search or calculator), and observes the results to achieve a defined goal. | |
| """) | |
| st.sidebar.header("Configuration") | |
| api_key = st.sidebar.text_input("Your Groq API Key", type="password", help="Your API key is never stored.") | |
| st.header("Define the Agent's Goal") | |
| goal = st.text_input("Goal:", placeholder="e.g. 'What is the current age of the Matrix movie's lead actor, and what is the square root of that?'") | |
| if st.button("Start Agent", disabled=(not api_key or not goal)): | |
| run_agent_loop(goal, api_key) | |