Update tools.py
This commit is contained in:
@@ -1,95 +1,61 @@
|
||||
"""Tool implementations used by the Deep Agent.
|
||||
"""
|
||||
Custom tools for the deep agent.
|
||||
|
||||
The tools are simple wrappers around the virtual file system and a
|
||||
web‑search helper. They expose a ``name`` and ``description`` that are
|
||||
used by the LangChain agent.
|
||||
This module defines two tools:
|
||||
|
||||
* ``search`` – performs a web search using DuckDuckGo and returns a short snippet of the first result.
|
||||
* ``write_file`` – writes content to the in‑memory virtual file system.
|
||||
|
||||
The tools are decorated with ``@tool`` from LangChain so that they can be
|
||||
automatically discovered by the agent.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
from langchain.tools import tool
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
from .vfs import vfs
|
||||
|
||||
from typing import Dict, Any
|
||||
@tool
|
||||
def search(query: str) -> str:
|
||||
"""Search the web for *query* and return a short excerpt.
|
||||
|
||||
from langchain_core.tools import BaseTool
|
||||
from langchain_ollama import OllamaEmbeddings
|
||||
from langchain_community.tools.tavily_search import TavilySearchResults
|
||||
|
||||
from virtual_fs import VirtualFileSystem
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WebSearch tool
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class WebSearch(BaseTool):
|
||||
"""Search the web using Tavily.
|
||||
|
||||
The tool returns a short string containing the top results.
|
||||
The implementation uses DuckDuckGo's HTML interface. It fetches the
|
||||
first result link, downloads the page and extracts the first five
|
||||
paragraphs.
|
||||
"""
|
||||
try:
|
||||
# DuckDuckGo HTML search
|
||||
url = f"https://duckduckgo.com/html/?q={query}"
|
||||
resp = requests.get(url, timeout=10)
|
||||
resp.raise_for_status()
|
||||
soup = BeautifulSoup(resp.text, "html.parser")
|
||||
|
||||
name: str = "web_search"
|
||||
description: str = "Search the web for information. Input should be a query."
|
||||
# Find first result link
|
||||
results = soup.find_all("a", {"class": "result__a"}, limit=1)
|
||||
if not results:
|
||||
return "No results found."
|
||||
link = results[0].get("href")
|
||||
if not link:
|
||||
return "No link found."
|
||||
|
||||
def _run(self, query: str) -> str:
|
||||
tavily = TavilySearchResults(max_results=3)
|
||||
results = tavily.run(query)
|
||||
return "\n".join(f"{i+1}. {r['title']} – {r['url']}" for i, r in enumerate(results))
|
||||
# Fetch the page content
|
||||
content_resp = requests.get(link, timeout=10)
|
||||
content_resp.raise_for_status()
|
||||
content_soup = BeautifulSoup(content_resp.text, "html.parser")
|
||||
|
||||
def _arun(self, query: str) -> str: # pragma: no cover
|
||||
return self._run(query)
|
||||
# Extract text from first few paragraphs
|
||||
paragraphs = content_soup.find_all("p")
|
||||
excerpt = "\n".join(p.get_text() for p in paragraphs[:5])
|
||||
return excerpt.strip() or "No text extracted."
|
||||
except Exception as e:
|
||||
return f"Error during search: {str(e)}"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CreateVirtualFile tool
|
||||
# ---------------------------------------------------------------------------
|
||||
@tool
|
||||
def write_file(file_path: str, content: str) -> str:
|
||||
"""Write *content* to a virtual file named *file_path*.
|
||||
|
||||
class CreateVirtualFile(BaseTool):
|
||||
"""Create or overwrite a virtual file.
|
||||
|
||||
Input format: ``filename:content``. The tool splits on the first
|
||||
colon and stores the content in the shared virtual file system.
|
||||
The file is stored in the in‑memory virtual file system and can be
|
||||
exported to disk later.
|
||||
"""
|
||||
|
||||
name: str = "create_virtual_file"
|
||||
description: str = (
|
||||
"Create or overwrite a virtual file. Input format: 'filename:content'."
|
||||
)
|
||||
|
||||
def __init__(self, vfs: VirtualFileSystem):
|
||||
super().__init__()
|
||||
self.vfs = vfs
|
||||
|
||||
def _run(self, input_text: str) -> str:
|
||||
if ":" not in input_text:
|
||||
raise ValueError("Input must be in the form 'filename:content'")
|
||||
name, content = input_text.split(":", 1)
|
||||
self.vfs.create_file(name.strip(), content.strip())
|
||||
return f"File '{name.strip()}' created with {len(content.strip())} characters."
|
||||
|
||||
def _arun(self, input_text: str) -> str: # pragma: no cover
|
||||
return self._run(input_text)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ExportVirtualFiles tool
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class ExportVirtualFiles(BaseTool):
|
||||
"""Export all virtual files to the real filesystem.
|
||||
|
||||
Input can optionally specify a target directory. If omitted, the
|
||||
current working directory is used.
|
||||
"""
|
||||
|
||||
name: str = "export_virtual_files"
|
||||
description: str = (
|
||||
"Export all virtual files to disk. Input is an optional directory path."
|
||||
)
|
||||
|
||||
def __init__(self, vfs: VirtualFileSystem):
|
||||
super().__init__()
|
||||
self.vfs = vfs
|
||||
|
||||
def _run(self, input_text: str | None = None) -> str:
|
||||
target = input_text.strip() if input_text else "."
|
||||
self.vfs.export_to_disk(target)
|
||||
return f"Exported {len(self.vfs.files)} files to {target}"
|
||||
|
||||
def _arun(self, input_text: str | None = None) -> str: # pragma: no cover
|
||||
return self._run(input_text)
|
||||
vfs.write_file(file_path, content)
|
||||
return f"File {file_path} written."
|
||||
Reference in New Issue
Block a user