add agent_core.py

agent_core.py

@@ -0,0 +1,102 @@
+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: <tool_name>
+            Action Input: <json or plain string>
+        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<tool>\w+)\s*\nAction\s+Input:\s*(?P<input>.+)"
+        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