import os import json from typing import List, Tuple, Optional class DeepAgent: """Simple ReAct style agent implemented from scratch. Parameters ---------- llm: callable A callable that accepts a list of messages and returns a dict with a ``content`` field containing the LLM response. tools: list List of tool callables. Each tool must be a function that accepts a single string argument and returns a string. """ def __init__(self, llm, tools: List): self.llm = llm self.tools = {t.__name__: t for t in tools} self.history: List[dict] = [] def _format_tools_prompt(self) -> str: """Return a human‑readable description of available tools. The format is used in the system prompt so the LLM knows what it can call. Each tool is described by its name and the first line of its docstring. """ lines = ["Available tools:"] for name, func in self.tools.items(): doc = func.__doc__ or "No description" first_line = doc.strip().split("\n")[0] lines.append(f"- {name}: {first_line}") return "\n".join(lines) def _parse_action(self, text: str) -> Optional[Tuple[str, str]]: """Parse an Action and Action Input from a ReAct response. Expected format (case‑insensitive): Action: Action Input: The method returns a tuple ``(tool_name, action_input)`` or ``None`` if the pattern is not found. """ import re pattern = r"(?i)Action:\s*(?P\w+)\s*\nAction\s+Input:\s*(?P.+)" match = re.search(pattern, text, re.DOTALL) if not match: return None tool = match.group("tool").strip() action_input = match.group("input").strip() return tool, action_input def run(self, query: str) -> str: """Execute a ReAct loop until the LLM returns a FINAL ANSWER. The method returns the final answer string. """ # System prompt with tool descriptions system_prompt = ( "You are a helpful assistant that can use the following tools. " "When you need to perform an action, output the tool name and the input. " "When you are finished, output FINAL ANSWER." f"\n\n{self._format_tools_prompt()}" ) self.history = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": query}, ] while True: # Ask LLM for next step response = self.llm(self.history) content = response.get("content", "") # Append LLM output to history self.history.append({"role": "assistant", "content": content}) # Check for FINAL ANSWER if "FINAL ANSWER:" in content.upper(): # Extract everything after FINAL ANSWER: final = content.split("FINAL ANSWER:", 1)[1].strip() return final # Try to parse an action parsed = self._parse_action(content) if not parsed: # If no action found, continue loop (LLM may just think) continue tool_name, action_input = parsed tool = self.tools.get(tool_name) if not tool: observation = f"Error: unknown tool {tool_name}" else: try: observation = tool(action_input) except Exception as e: observation = f"Error executing {tool_name}: {e}" # Append observation self.history.append({"role": "assistant", "content": f"Observation: {observation}"}) # End of DeepAgent class