103 lines
3.8 KiB
Python
103 lines
3.8 KiB
Python
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
|