big refactory, support per folder locali, workflow per interazione con sorgenti differnti e analisi dati

This commit is contained in:
Alfredo Oliviero 2025-09-04 12:27:01 +02:00
parent e05a4f7b35
commit 39b8951e96
32 changed files with 1046 additions and 411 deletions

12
.gitignore vendored
View File

@ -1,6 +1,12 @@
long-term-memory
d4s_credentials.json
d4s_config.json
mnt/long-term-memory
mnt/tmp_file_storage
mnt/shared
config/d4s_credentials.json
config/d4s_config.json
config/dave_config.json
adk-samples
.DS_Store
# ---> Python

108
README.md
View File

@ -2,12 +2,7 @@
DAVE is an AI agent designed to act as a knowledgeable and helpful assistant for researchers using the D4Science platform. It helps users find information, understand services, and navigate the D4Science ecosystem by leveraging a long-term memory system powered by a vector database.
This project is built using the Google Agent Development Kit (ADK) and is designed to run entirely within Docker containers.
## Prerequisites
- Docker
- Docker Compose
This project is built using the Google Agent Development Kit (ADK).
## Project Structure
@ -25,11 +20,13 @@ This project is built using the Google Agent Development Kit (ADK) and is design
└── .env.sample # Sample environment variables
```
## Configuration
## Setup and Configuration
Configuration is managed through a combination of a `config` directory and an `.env` file.
This project can be run in two ways: via Docker (recommended for consistency) or with a local Python virtual environment.
### 1. Environment Variables (.env)
### 1. Common Configuration (Required for Both Setups)
#### Environment Variables (.env)
The application uses an `.env` file to manage environment-specific variables, such as ports and container names.
@ -40,11 +37,11 @@ The application uses an `.env` file to manage environment-specific variables, su
```
2. **Edit the `.env` file**:
Review the variables in the `.env` file and adjust them if necessary (e.g., if the default ports conflict with other services on your machine).
Review the variables in the `.env` file and adjust them if necessary.
### 2. D4Science Configuration (config/ directory)
#### D4Science Configuration (config/ directory)
The agent requires specific configuration and credential files to interact with D4Science services. These files are located in the `config/` directory and are mounted into the container at `/etc/d4science` at runtime.
The agent requires specific configuration and credential files to interact with D4Science services.
1. **Copy the Sample Files**:
Navigate to the `config` directory, copy the sample files, and remove the `.sample` extension.
@ -52,17 +49,19 @@ The agent requires specific configuration and credential files to interact with
cd config
cp d4s_config.sample.json d4s_config.json
cp d4s_credentials.json.sample d4s_credentials.json
cp dave_config.sample.json dave_config.json
cd ..
```
2. **Edit the JSON Files**:
Fill in the required values in both `d4s_config.json` and `d4s_credentials.json` with your specific D4Science environment details and credentials.
- Fill in the required values in both `d4s_config.json` and `d4s_credentials.json` with your specific D4Science environment details and credentials.
- Review and update `dave_config.json` to configure the agent's behavior, such as setting the vector store path.
## Running the Agent
### 2. Docker Setup (Recommended)
The application can be run in two modes: production and development.
The application is designed to run with Docker and Docker Compose. Ensure Docker is running on your machine.
### Production Mode
#### Production Mode
This mode runs the optimized production-ready container.
@ -74,12 +73,11 @@ This mode runs the optimized production-ready container.
2. **Access the Web UI**:
Open your browser to `http://127.0.0.1:8000` (or the `DAVE_AGENT_PORT` you configured in `.env`).
### Development Mode (with Debugging)
#### Development Mode (with Debugging)
This mode is for active development. It mounts the local source code into the container, enabling hot-reloading on code changes. It also exposes a debug port.
This mode is for active development. It mounts the local source code into the container, enabling hot-reloading on code changes, and exposes a debug port.
1. **Build and Run the Containers**:
Use the `debug.yaml` file to start the services.
```bash
docker compose -f debug.yaml up --build
```
@ -91,4 +89,74 @@ This mode is for active development. It mounts the local source code into the co
- Open the "Run and Debug" panel in VS Code (Ctrl+Shift+D).
- Select the **"Python: Attach to Docker"** configuration from the dropdown menu.
- Press F5 to start debugging.
- VS Code will attach to the `debugpy` server running inside the container, allowing you to set breakpoints and inspect the code as it runs.
## Development Scripts
### Documentation Scraper (`devel/scrape_docs.py`)
This script is a utility to download the official ADK documentation as HTML files and convert them to Markdown. This can be useful for local reference or potentially for feeding the documentation into the agent's knowledge base in the future.
**Prerequisites:**
- You must have a local virtual environment set up.
- You must have installed the `[tools]` dependencies (`uv pip install -e ".[dev,tools]"`).
**Usage:**
1. Ensure your virtual environment is activated (`source .venv/bin/activate`).
2. Run the script from the project root directory:
```bash
python devel/scrape_docs.py
```
The script will process the files and place the Markdown output in the `docs/adk-docs/markdown` directory.
### 3. Local Virtual Environment Setup (with uv)
For development outside of Docker, you can use `uv` to manage your virtual environment and dependencies.
#### 1. Install uv
If you don't have it already, install `uv`.
```bash
pip install uv
```
#### 2. Create and Activate Virtual Environment
From the project root, create and activate a Python virtual environment. `uv` will create and manage a `.venv` directory.
- **Create**:
```bash
uv venv
```
- **Activate (macOS/Linux)**:
```bash
source .venv/bin/activate
```
- **Activate (Windows)**:
```bash
.venv\Scripts\activate
```
#### 3. Install Dependencies
Install the project dependencies from `pyproject.toml` in editable mode. The `[dev]` extra includes packages for debugging, and `[tools]` includes packages for running development scripts.
```bash
uv pip install -e ".[dev,tools]"
```
##### Updating Dependencies
If you modify the `pyproject.toml` file, re-run the installation command to sync your virtual environment. `uv` is fast, so a fresh install is often as quick as an upgrade.
```bash
uv pip install -e ".[dev,tools]"
```
#### 4. Running the Agent Locally
After setup, you can run the agent directly using the ADK CLI. Ensure your virtual environment is activated.
- **To run with the interactive web UI**:
```bash
adk web src
```
- **To run on the command line**:
```bash
adk run src

View File

@ -8,8 +8,10 @@ services:
- "${DAVE_AGENT_PORT}:8000"
volumes:
- ./mnt/tmp_file_storage:/app/tmp
- ./config/dave_config.json:/etc/dave_config.json
- ./config/d4s_config.json:${D4S_CONFIG_DIR}/${D4S_CONFIG_FILENAME}
- ./config/d4s_credentials.json:/root/.d4science/auth.${D4S_ENV}.json
- ./mnt/local_content:/app/local_content
env_file:
- ./.env
depends_on:

View File

@ -0,0 +1,14 @@
{
"local_file": {
"mappings": [
{
"alias": "local",
"path": "/app/shared/local_content"
},
{
"alias": "shared",
"path": "/app/shared/shared_content"
}
]
}
}

@ -1 +1 @@
Subproject commit 5976ecb432cdcdd2f274b9eca8b46416db530ed8
Subproject commit 4d5b313ad15ae8ede27c9167b95ce94715df0c56

View File

@ -12,7 +12,9 @@ services:
- ./d4science_lib:/app/d4science_lib
- ./mnt/tmp_file_storage:/app/tmp
- ./config/d4s_config.json:${D4S_CONFIG_DIR}/${D4S_CONFIG_FILENAME}
- ./config/dave_config.json:/etc/dave_config.json
- ./config/d4s_credentials.json:/root/.d4science/auth.${D4S_ENV}.json
- ./mnt/shared:/app/shared
env_file:
- ./.env
command:

1
docs/adk-docs/.gitignore vendored Normal file
View File

@ -0,0 +1 @@
html

View File

@ -0,0 +1 @@
empty placeholder

View File

@ -0,0 +1 @@
empty placeholder

View File

@ -0,0 +1 @@
empty placeholder

View File

@ -16,8 +16,8 @@ dependencies = [
"sentence-transformers",
"langchain-core",
"langchain-community",
"pdfminer.six",
"beautifulsoup4",
"pdfminer.six"
]
[tool.setuptools.packages.find]
@ -26,5 +26,7 @@ where = ["src"]
[project.optional-dependencies]
dev = [
"pytest",
"debugpy"
"debugpy",
"beautifulsoup4",
"markdownify"
]

View File

@ -1 +1,2 @@
debugpy
debugpy
BeautifulSoup

133
src/config_manager.py Normal file
View File

@ -0,0 +1,133 @@
import json
import os
from typing import Any, Dict
class ConfigManager:
"""
A class to manage configurations from a JSON file.
It delegates the responsibility of loading, accessing, and saving
configuration data, abstracting away the file system management.
"""
def __init__(self, config_path: str = None):
"""
Initializes the configuration manager.
Args:
config_path (str): The path to the JSON configuration file.
If it doesn't exist, it will be created on the first save.
"""
if config_path is None:
config_path = os.environ.get('DAVE_CONFIG', '/etc/dave_config.json')
self.config_path = config_path
self._config_data = self._load_config()
def _load_config(self) -> Dict[str, Any]:
"""
Private method to load the configuration file.
Handles cases where the file does not exist or is malformed.
"""
if not os.path.exists(self.config_path):
return {} # Return an empty configuration if the file does not exist
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
# Handle empty file case
content = f.read()
if not content:
return {}
return json.loads(content)
except (json.JSONDecodeError, FileNotFoundError):
# If the file is corrupt or not found, start with an empty dictionary
return {}
def get(self, key: str, default: Any = None) -> Any:
"""
Retrieves a configuration value for a given key.
Supports nested keys using dot notation (e.g., 'database.host').
Args:
key (str): The key of the value to retrieve.
default (Any, optional): The value to return if the key is not found.
Defaults to None.
Returns:
Any: The value associated with the key, or the default value.
"""
keys = key.split('.')
value = self._config_data
try:
for k in keys:
value = value[k]
return value
except (TypeError, KeyError):
return default
def set(self, key: str, value: Any) -> None:
"""
Sets a configuration value for a given key.
Supports nested keys using dot notation (e.g., 'database.host').
Intermediate dictionaries will be created if they don't exist.
To make the change persistent, you must call the save() method.
Args:
key (str): The key for which to set the value.
value (Any): The value to store.
"""
keys = key.split('.')
current_level = self._config_data
for i, k in enumerate(keys[:-1]):
if k not in current_level or not isinstance(current_level.get(k), dict):
current_level[k] = {}
current_level = current_level[k]
current_level[keys[-1]] = value
def save(self) -> None:
"""
Saves the current state of the configuration to the JSON file.
Creates necessary directories if they do not exist.
"""
# Ensure the directory exists before writing the file
os.makedirs(os.path.dirname(self.config_path) or '.', exist_ok=True)
with open(self.config_path, 'w', encoding='utf-8') as f:
json.dump(self._config_data, f, indent=4, ensure_ascii=False)
def __str__(self) -> str:
"""String representation of the object."""
return f"ConfigManager(path='{self.config_path}', data={self._config_data})"
# Example usage:
if __name__ == '__main__':
# 1. Create an instance, specifying a path for the configuration.
config = ConfigManager('config/settings.json')
# 2. Set new values, including nested ones.
print("Setting new and nested values...")
config.set('api_key', 'abcdef-123456-ghijkl')
config.set('database.host', 'localhost')
config.set('database.port', 5432)
config.set('database.user', 'admin')
config.set('timeout', 30)
# 3. Save the changes to the file.
config.save()
print(f"Configuration saved to '{config.config_path}'")
# 4. Create a new instance to demonstrate data persistence.
print("\nReloading configuration from file...")
new_config_instance = ConfigManager('config/settings.json')
# 5. Retrieve the saved values using dot notation.
api_key = new_config_instance.get('api_key')
db_host = new_config_instance.get('database.host')
db_port = new_config_instance.get('database.port')
non_existent = new_config_instance.get('user.profile.theme', 'dark') # Example with default
print(f"Reloaded API Key: {api_key}")
print(f"Reloaded Database Host: {db_host}")
print(f"Reloaded Database Port: {db_port}")
print(f"Non-existent value with default: {non_existent}")

View File

@ -1,8 +1,7 @@
from google.adk.agents import LlmAgent
from .prompts import get_agent_instructions
from .sub_agents.workspace.agent import workspace_agent
from .sub_agents.rag.agent import rag_agent
from .sub_agents.document_processor.agent import document_processor_agent
from .sub_agents.ingestion_workflow.agent import ingestion_workflow
# Declare the root RouterAgent
root_agent = LlmAgent(
@ -10,8 +9,7 @@ root_agent = LlmAgent(
model="gemini-2.5-pro",
instruction=get_agent_instructions(),
sub_agents=[
workspace_agent,
rag_agent,
document_processor_agent,
ingestion_workflow,
],
)

View File

@ -2,12 +2,8 @@ def get_agent_instructions() -> str:
"""Returns the instructional prompt for the RouterAgent."""
return """
You are the central router for DAVE, the D4Science AI Virtual Assistant.
Your primary role is to understand the user's request and delegate it to the appropriate specialist sub-agent.
Your primary role is to understand the user's request and delegate it to the most appropriate specialist sub-agent based on its description.
You have the following sub-agents available:
- **WorkspaceAgent**: Use this agent for any tasks related to the D4Science Workspace, such as listing files or analyzing a file's content.
- **RagAgent**: Use this agent for any tasks related to the agent's long-term memory, such as answering questions based on stored knowledge, searching for information, or storing new facts.
- **DocumentProcessorAgent**: Use this agent to analyze the content of a document to extract metadata, a summary, and key topics.
Analyze the user's query and route it to the correct sub-agent based on its capabilities.
"""
Analyze the user's query and choose the best sub-agent to handle the task.
Do not attempt to answer the user's question or perform the task yourself; your sole responsibility is to route the request to the correct sub-agent.
"""

View File

@ -2,6 +2,7 @@ from google.adk.agents import LlmAgent
from pydantic import BaseModel, Field
from typing import List
from .prompts import get_document_processor_instructions
from .tools import process_and_memorize_file
class DocumentAnalysis(BaseModel):
"""A model to hold the structured analysis of a document."""
@ -12,8 +13,9 @@ class DocumentAnalysis(BaseModel):
document_processor_agent = LlmAgent(
name="DocumentProcessorAgent",
description="Analyzes the content of a document to extract metadata, a summary, and key topics.",
description="Processes a file artifact from the session state, parsing its content, splitting it into chunks, and storing it in long-term memory.",
instruction=get_document_processor_instructions(),
model="gemini-2.5-pro",
tools=[process_and_memorize_file],
output_schema=DocumentAnalysis,
)

View File

@ -0,0 +1,95 @@
import logging
import traceback
from typing import List
from google.adk.tools import ToolContext
from langchain_core.documents import Document
from langchain_core.documents.base import Blob
from langchain_community.document_loaders.parsers.generic import MimeTypeBasedParser
from langchain_community.document_loaders.parsers.html.bs4 import BS4HTMLParser
from langchain_community.document_loaders.parsers.pdf import PDFMinerParser
from langchain_community.document_loaders.parsers.txt import TextParser
from langchain_text_splitters import RecursiveCharacterTextSplitter
# Import shared components
from ...tools.common import rag
logging.basicConfig(level=logging.INFO)
# --- Document Parsing and Splitting ---
class ParserFactory:
"""A static factory for creating a configured document parser."""
_mime_based_parser = None
@staticmethod
def get_parser():
if ParserFactory._mime_based_parser is None:
handlers = {
"application/pdf": PDFMinerParser(),
"text/plain": TextParser(),
"text/html": BS4HTMLParser(),
}
ParserFactory._mime_based_parser = MimeTypeBasedParser(handlers=handlers, fallback_parser=None)
return ParserFactory._mime_based_parser
def parse_document_blob(blob: Blob) -> List[Document]:
"""Parses a blob of data into a list of documents."""
parser = ParserFactory.get_parser()
return parser.parse(blob)
def split_documents(documents: List[Document]) -> List[Document]:
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
logging.info(f"Splitting {len(documents)} document(s) into smaller chunks...")
split_docs = text_splitter.split_documents(documents)
logging.info(f"Successfully created {len(split_docs)} document chunks.")
return split_docs
def process_and_memorize_file(tool_context: ToolContext) -> str:
"""
Processes a file stored as a session artifact and adds its content to long-term memory.
It reads the artifact name from the session state.
Args:
tool_context: The context containing the session state and artifact service.
Returns:
A confirmation message.
"""
artifact_name = tool_context.state.get('temp:artifact_name')
item_name = tool_context.state.get('temp:item_name')
if not artifact_name or not item_name:
return "Error: Could not find artifact name or item name in session state."
try:
logging.info(f"Processing and memorizing artifact: {artifact_name} (original name: {item_name})")
# 1. Load the artifact
artifact_part = tool_context.load_artifact(artifact_name)
if not artifact_part or not artifact_part.inline_data:
return f"Error: Could not load artifact '{artifact_name}' or it has no content."
# 2. Parse the content from the blob
blob = artifact_part.inline_data
raw_documents = parse_document_blob(blob)
if not raw_documents:
return f"Could not parse document {item_name}. It might be empty or an unsupported format."
# 3. Add metadata and split
for doc in raw_documents:
doc.metadata["source"] = item_name
chunked_documents = split_documents(raw_documents)
# 4. Add to RAG memory
rag.add_documents(chunked_documents)
return f"Successfully processed and memorized file: {item_name}"
except Exception as e:
logging.error(f"An error occurred in the process_and_memorize_file tool: {e}")
logging.error(traceback.format_exc())
return f"Error processing artifact {artifact_name}: {e}"

View File

@ -0,0 +1,12 @@
from google.adk.agents import SequentialAgent
from .router import data_source_router_agent
from ..document_processor.agent import document_processor_agent
ingestion_workflow = SequentialAgent(
name="IngestionWorkflow",
description="Handles all interactions with file sources, including the D4Science Workspace and the local filesystem. Use this for tasks like listing files and folders, reading file content, or starting a file ingestion and processing workflow.",
sub_agents=[
data_source_router_agent,
document_processor_agent
]
)

View File

@ -0,0 +1,18 @@
from google.adk.agents import LlmAgent
from ..workspace.agent import workspace_agent
from ..local_file_agent.agent import local_file_agent
data_source_router_agent = LlmAgent(
name="DataSourceRouterAgent",
description="First step in an ingestion workflow. Determines the source of a file (e.g., D4Science Workspace, local filesystem) and delegates the file retrieval to the appropriate sub-agent. The sub-agent will save the file as a session artifact.",
instruction="""
You are a router responsible for determining the source of a file.
Based on the user's request and the descriptions of the available sub-agents, delegate the task to the appropriate agent to retrieve the file.
Your sole responsibility is to route the request; do not attempt to retrieve the file yourself.
""",
model="gemini-2.5-pro",
sub_agents=[
workspace_agent,
local_file_agent,
],
)

View File

@ -0,0 +1,18 @@
from google.adk.agents import LlmAgent
from .tools import list_local_files, read_local_file, list_shared_folders
local_file_agent = LlmAgent(
name="LocalFileAgent",
description="Handles interactions with a secure, aliased local filesystem. It can discover shared folders, and then list and read files from them.",
instruction="Your task is to interact with a local filesystem using a secure, two-step process. \
First, you MUST call the `list_shared_folders` tool to discover the available top-level folder aliases (e.g., 'local', 'shared'). \
Second, once you have the aliases, you can use the `list_local_files` tool with a virtual path (e.g., 'local/' or 'shared/subfolder') to see the contents of a specific folder. \
You can also read files using the `read_local_file` tool with a full virtual path (e.g., 'shared/report.txt'). \
Do not invent paths; always start by discovering the available aliases.",
model="gemini-2.5-pro",
tools=[
list_shared_folders,
list_local_files,
read_local_file,
],
)

View File

@ -0,0 +1,147 @@
import json
import logging
import os
import traceback
from typing import List, Dict, Any
from google.adk.tools import ToolContext
from google.genai import types
# Assuming ConfigManager is in a reachable path, e.g., src.config_manager
# Adjust the import path according to your project structure.
from src.config_manager import ConfigManager
logging.basicConfig(level=logging.INFO)
# --- Security Helper ---
def _get_path_mappings() -> Dict[str, str]:
"""Loads path mappings from config, returning a dictionary of alias -> real_path."""
config = ConfigManager()
mappings = config.get("local_file.mappings", [])
return {item['alias']: item['path'] for item in mappings if 'alias' in item and 'path' in item}
def _resolve_virtual_path(virtual_path: str) -> str | None:
"""
Translates a virtual path (e.g., 'local/file.txt') to a secure, absolute filesystem path.
Prevents directory traversal attacks.
Returns:
The absolute, verified path or None if access is denied.
"""
mappings = _get_path_mappings()
try:
# Normalize path to prevent issues with leading slashes
virtual_path = virtual_path.lstrip('/')
parts = virtual_path.split('/', 1)
alias = parts[0]
sub_path = parts[1] if len(parts) > 1 else ''
if alias not in mappings:
return None # Alias not found
base_path = os.path.realpath(mappings[alias])
# Securely join the base path with the sub-path
target_path = os.path.realpath(os.path.join(base_path, sub_path))
# Security check: ensure the final path is still within the intended base directory
if os.path.commonpath([target_path, base_path]) == base_path:
return target_path
except Exception:
return None
return None
# --- Agent Tools ---
def list_local_files(path: str) -> Dict[str, Any]:
"""
Lists files and directories using a virtual path (e.g., 'local/my_folder').
Access is restricted to pre-configured aliases.
Args:
path (str): The virtual directory path to list (e.g., 'local/').
Returns:
A dictionary containing the list of files and directories or an error.
"""
real_path = _resolve_virtual_path(path)
if not real_path:
return {"error": f"Access denied. The path '{path}' is not valid or not allowed."}
try:
if not os.path.isdir(real_path):
return {"error": f"The path '{path}' is not a valid directory."}
entries = os.listdir(real_path)
files = [e for e in entries if os.path.isfile(os.path.join(real_path, e))]
directories = [e for e in entries if os.path.isdir(os.path.join(real_path, e))]
return {
"path": path,
"directories": directories,
"files": files
}
except Exception as e:
logging.error(f"Error listing local files at virtual path '{path}' (real: '{real_path}'): {e}")
return {"error": str(e)}
def read_local_file(path: str, tool_context: ToolContext) -> str:
"""
Reads a file from the local filesystem using a virtual path (e.g., 'local/my_file.txt').
The file content is saved as a session artifact for processing.
Args:
path (str): The virtual path of the file to read.
Returns:
A confirmation message or an error.
"""
real_path = _resolve_virtual_path(path)
if not real_path:
return f"Access denied. The path '{path}' is not valid or not allowed."
try:
if not os.path.isfile(real_path):
return f"Error: The path '{path}' is not a valid file."
item_name = os.path.basename(real_path)
artifact_name = f"localfile_{item_name}"
logging.info(f"Reading file from virtual path '{path}' (real: '{real_path}') to save as artifact '{artifact_name}'.")
with open(real_path, "rb") as f:
file_bytes = f.read()
# Save content as an artifact
artifact_part = types.Part(inline_data=types.Blob(mime_type="application/octet-stream", data=file_bytes))
tool_context.save_artifact(artifact_name, artifact_part)
# Save artifact info to state for the ingestion workflow
tool_context.state['temp:artifact_name'] = artifact_name
tool_context.state['temp:item_name'] = item_name
return f"Successfully read file '{item_name}' and stored it as a session artifact for processing."
except Exception as e:
logging.error(f"An error occurred in the read_local_file tool: {e}")
logging.error(traceback.format_exc())
return f"Error executing read_local_file: {e}"
def list_shared_folders() -> Dict[str, Any]:
"""
Lists the available top-level shared folders (aliases) that can be accessed.
Returns:
A dictionary containing a list of available folder aliases.
"""
try:
mappings = _get_path_mappings()
aliases = list(mappings.keys())
return {"shared_folders": aliases}
except Exception as e:
logging.error(f"Error listing shared folders: {e}")
return {"error": "Could not retrieve the list of shared folders."}

View File

@ -1,5 +1,5 @@
from google.adk.agents import LlmAgent
from ...tools.d4science_tools import (
from .tools import (
search_memory,
store_in_memory,
forget_source,

View File

@ -3,5 +3,7 @@ def get_rag_agent_instructions() -> str:
return """
You are a specialist agent responsible for managing and querying the long-term memory system.
Your capabilities are focused on searching for information, storing new facts, and managing memory sources.
Use your tools to answer user questions based on the knowledge stored in the vector database.
If you cannot find an answer to a question, you must clearly state that the information is not available in your knowledge base. Do not attempt to answer questions that are outside the scope of your stored knowledge.
"""

View File

@ -0,0 +1,66 @@
import logging
from typing import Dict, Any, Optional
# Import the shared RAG instance
from ...tools.common import rag
logging.basicConfig(level=logging.INFO)
def search_memory(query: str, source: Optional[str] = None, max_results: int = 5, min_score: float = 0.5) -> Dict[str, Any]:
"""Search the long-term memory (vector database) for relevant information."""
try:
logging.info(f"Searching memory for query: '{query}' with source: '{source or ''}'")
results = rag.search(query, source=source, limit=max_results, score_threshold=min_score)
if not results:
return {"status": "success", "message": "No relevant information found in memory", "query": query, "results": []}
return {"status": "success", "query": query, "results": results, "total_found": len(results)}
except Exception as e:
return {"status": "error", "message": f"Error searching memory: {str(e)}", "results": []}
def store_in_memory(content: str, source: str = "user_input", category: str = "general") -> Dict[str, Any]:
"""Store a simple string of information in long-term memory."""
try:
from langchain_core.documents import Document
doc = Document(page_content=content, metadata={"source": source, "category": category})
success = rag.add_documents([doc])
if success:
return {"status": "success", "message": f"Content stored in memory under source '{source}'", "source": source, "category": category}
else:
return {"status": "error", "message": "Failed to store content in memory"}
except Exception as e:
return {"status": "error", "message": f"Error storing in memory: {str(e)}"}
def forget_source(source: str) -> Dict[str, Any]:
"""Remove all information from a specific source in long-term memory."""
try:
success = rag.delete_by_source(source)
if success:
return {"status": "success", "message": f"All content from source '{source}' has been removed from memory"}
else:
return {"status": "error", "message": f"Failed to remove content from source '{source}'"}
except Exception as e:
return {"status": "error", "message": f"Error removing from memory: {str(e)}"}
def memory_status() -> Dict[str, Any]:
"""Get status information about the long-term memory system."""
try:
info = rag.get_collection_info()
return {"status": "success", "memory_info": info}
except Exception as e:
return {"status": "error", "message": f"Error getting memory status: {str(e)}"}
def list_memory_entries(metadata_filter: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""List the titles of files in memory, optionally filtering by metadata."""
try:
summary = rag.summarize_by_source(metadata_filter=metadata_filter)
if not summary:
return {"status": "success", "message": "No entries found matching the specified criteria.", "files": []}
files = [{"title": source, "chunks": count} for source, count in summary.items()]
return {"status": "success", "files": files, "total_files": len(files)}
except Exception as e:
return {"status": "error", "message": f"Error listing memory entries: {str(e)}"}

View File

@ -1,14 +1,14 @@
from google.adk.agents import LlmAgent
from ...tools.d4science_tools import list_files, analyze_file
from .tools import list_files, download_file
from .prompts import get_workspace_agent_instructions
workspace_agent = LlmAgent(
name="WorkspaceAgent",
description="Handles all interactions with the D4Science Workspace, such as listing and analyzing files.",
description="Handles interactions with the D4Science Workspace, such as listing and downloading files.",
instruction=get_workspace_agent_instructions(),
model="gemini-2.5-pro",
tools=[
list_files,
analyze_file,
download_file,
],
)

View File

@ -2,6 +2,8 @@ def get_workspace_agent_instructions() -> str:
"""Returns the instructional prompt for the WorkspaceAgent."""
return """
You are a specialist agent responsible for interacting with the D4Science Workspace.
Your capabilities are focused on listing files and analyzing them to be stored in memory.
Use your tools to fulfill user requests related to the workspace.
Your capabilities are strictly limited to the functions provided by your tools, which include listing and retrieving files.
Analyze the user's request and use the appropriate tool to fulfill it.
Do not attempt to perform any actions that are not supported by your tools, such as creating, modifying, or deleting files.
"""

View File

@ -0,0 +1,117 @@
import logging
import traceback
import os
from google.adk.tools import ToolContext
from google.genai.types import Part
# Import shared components
from ...tools.common import get_storagehub_client
logging.basicConfig(level=logging.INFO)
def list_files(path: str, tool_context: ToolContext) -> str:
"""
Lists all files at the specified D4Science Workspace path.
Args:
path (str): The directory path to list files from.
Returns:
str: JSON-formatted list of files and folders with their metadata.
"""
try:
session_id = tool_context._invocation_context.session.id
storagehub_client = get_storagehub_client(session_id)
root_folder = storagehub_client.workspace_manager.getWorkspace(as_object=False)
root_id = root_folder['id']
if not path or path == "/" or path == "":
target_id = root_id
else:
clean_path = path.strip("/")
if not clean_path:
target_id = root_id
else:
try:
item = storagehub_client.items_manager.getByRelativePath(
parent_id=root_id,
relative_path=clean_path,
as_object=False
)
target_id = item['id']
except Exception:
target_id = path
files = storagehub_client.items_manager.listById(
item_id=target_id,
exclude=["hl:accounting"],
as_object=False
)
return str(files)
except Exception as e:
logging.error(f"An error occurred in the list_files tool: {e}")
logging.error(traceback.format_exc())
return f"Error executing list_files: {e}"
def download_file(path: str, tool_context: ToolContext) -> str:
"""
Downloads a file from the D4Science Workspace and saves it as a session artifact.
Args:
path (str): The workspace path of the file to download.
Returns:
str: A confirmation message.
"""
temp_path = None
try:
session_id = tool_context._invocation_context.session.id
storagehub_client = get_storagehub_client(session_id)
root_folder = storagehub_client.workspace_manager.getWorkspace(as_object=False)
root_id = root_folder['id']
target_id = path # Default to path if resolution fails
if not path or path == "/" or path == "":
target_id = root_id
else:
clean_path = path.strip("/")
if clean_path:
try:
item = storagehub_client.items_manager.getByRelativePath(
parent_id=root_id, relative_path=clean_path, as_object=False
)
target_id = item['id']
except Exception:
logging.warning(f"Could not resolve path '{clean_path}', assuming it's an ID.")
item = storagehub_client.items_manager.get(item_id=target_id, as_object=False)
item_name = item["name"]
artifact_name = f"downloaded_{target_id}_{item_name}"
logging.info(f"Downloading file '{item_name}' (ID: {target_id}) to save as artifact '{artifact_name}'.")
# Download to a temporary path to read its bytes
temp_path = f'/tmp/{artifact_name}'
storagehub_client.items_manager.download(item_id=target_id, destination=temp_path)
with open(temp_path, "rb") as f:
file_bytes = f.read()
# Save content as an artifact
artifact_part = Part(inline_data=types.Blob(mime_type="application/octet-stream", data=file_bytes))
tool_context.save_artifact(artifact_name, artifact_part)
# Save artifact info to state for the next step
tool_context.state['temp:artifact_name'] = artifact_name
tool_context.state['temp:item_name'] = item_name
return f"Successfully downloaded file '{item_name}' and stored it as a session artifact."
except Exception as e:
logging.error(f"An error occurred in the download_file tool: {e}")
logging.error(traceback.format_exc())
return f"Error executing download_file: {e}"
finally:
# Clean up the intermediate temporary file immediately
if temp_path and os.path.exists(temp_path):
os.remove(temp_path)

View File

@ -0,0 +1,176 @@
import hashlib
import logging
import os
from typing import Dict, List, Any, Optional
from d4science.core.config import load_configuration
from d4science.auth.auth import get_auth_token
from d4science.auth.clients import create_oauth_client
from d4science.storagehub.client import StorageHubClient
from langchain_core.documents import Document
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue
from sentence_transformers import SentenceTransformer
logging.basicConfig(level=logging.INFO)
# --- D4Science Client Management ---
STORAGEHUB_CLIENT_CACHE: Dict[str, StorageHubClient] = {}
def create_new_storagehub_client() -> StorageHubClient:
"""Factory function to create a new, authenticated StorageHubClient."""
logging.info("Creating a new authenticated StorageHub client...")
config, _ = load_configuration(
env=os.getenv("D4S_ENV") or 'prod',
cli_config_dir=os.getenv("D4S_CONFIG_DIR") or '/etc/d4science',
cli_config_filename=os.getenv("D4S_CONFIG_FILENAME") or 'd4s_config.json'
)
auth_token = get_auth_token(configuration=config)
oauth_client = create_oauth_client(auth_token)
client = StorageHubClient(client=oauth_client, configuration=config)
logging.info("Client created successfully.")
return client
def get_storagehub_client(session_id: str) -> StorageHubClient:
"""Retrieve the StorageHubClient for the given session ID from the cache."""
if session_id not in STORAGEHUB_CLIENT_CACHE:
STORAGEHUB_CLIENT_CACHE[session_id] = create_new_storagehub_client()
return STORAGEHUB_CLIENT_CACHE[session_id]
# --- RAG System (Qdrant) ---
class QdrantRAG:
def __init__(self, host: str = None, port: int = None):
self.host = host or os.getenv('QDRANT_HOST', 'long-term-memory')
self.port = port or int(os.getenv('QDRANT_PORT', '6333'))
self.encoder_model = os.getenv('ENCODER_MODEL', 'all-MiniLM-L6-v2')
self.vector_size = int(os.getenv('VECTOR_SIZE', '384'))
self.collection_name = os.getenv('COLLECTION_NAME', 'dave_knowledge')
self.client = QdrantClient(host=self.host, port=self.port)
self.encoder = SentenceTransformer(self.encoder_model)
print(f"RAG initialized with encoder: {self.encoder_model} (vector size: {self.vector_size})")
self._ensure_collection()
def _ensure_collection(self):
try:
collections = self.client.get_collections().collections
collection_names = [col.name for col in collections]
if self.collection_name not in collection_names:
self.client.create_collection(
collection_name=self.collection_name,
vectors_config=VectorParams(size=self.vector_size, distance=Distance.COSINE)
)
print(f"Created collection: {self.collection_name}")
else:
print(f"Collection {self.collection_name} already exists")
except Exception as e:
print(f"Error ensuring collection: {e}")
def _generate_id(self, text: str, source: str) -> str:
content = f"{source}:{text}"
return hashlib.md5(content.encode()).hexdigest()
def add_documents(self, documents: List[Document]) -> bool:
try:
points = []
for doc in documents:
embedding = self.encoder.encode(doc.page_content).tolist()
payload = doc.metadata
payload["text"] = doc.page_content
source = payload.get("source", "unknown")
doc_id = self._generate_id(doc.page_content, source)
point = PointStruct(id=doc_id, vector=embedding, payload=payload)
points.append(point)
if not points:
print("Warning: No documents provided to add.")
return True
self.client.upsert(collection_name=self.collection_name, points=points, wait=True)
print(f"Successfully added {len(points)} document chunks to memory.")
return True
except Exception as e:
print(f"Error adding documents: {e}")
return False
def search(self, query: str, limit: int = 5, score_threshold: float = 0.5, source: Optional[str] = None) -> List[Dict[str, Any]]:
try:
query_embedding = self.encoder.encode(query).tolist()
search_filter = None
if source:
search_filter = Filter(must=[FieldCondition(key="source", match=MatchValue(value=source))])
search_results = self.client.search(
collection_name=self.collection_name,
query_vector=query_embedding,
query_filter=search_filter,
limit=limit,
score_threshold=score_threshold
)
results = []
for result in search_results:
results.append({
"text": result.payload.get("text", ""),
"source": result.payload.get("source", "unknown"),
"score": result.score,
"chunk_index": result.payload.get("chunk_index", 0),
"metadata": {k: v for k, v in result.payload.items() if k not in ["text", "source", "chunk_index", "total_chunks"]}
})
return results
except Exception as e:
print(f"Error searching: {e}")
return []
def delete_by_source(self, source: str) -> bool:
try:
filter_condition = Filter(must=[FieldCondition(key="source", match=MatchValue(value=source))])
self.client.delete(collection_name=self.collection_name, points_selector=filter_condition)
return True
except Exception as e:
print(f"Error deleting documents: {e}")
return False
def get_collection_info(self) -> Dict[str, Any]:
try:
info = self.client.get_collection(self.collection_name)
return {
"name": info.config.name,
"vector_size": info.config.params.vectors.size,
"distance": info.config.params.vectors.distance,
"points_count": info.points_count,
"status": info.status
}
except Exception as e:
return {"error": str(e)}
def summarize_by_source(self, metadata_filter: Optional[Dict[str, Any]] = None) -> Dict[str, int]:
try:
filter_conditions = Filter(must=[])
if metadata_filter:
for key, value in metadata_filter.items():
condition = FieldCondition(key=key, match=MatchValue(value=value))
filter_conditions.must.append(condition)
if not filter_conditions.must:
filter_conditions = None
source_counts = {}
next_offset = None
while True:
scroll_response, next_offset = self.client.scroll(
collection_name=self.collection_name,
scroll_filter=filter_conditions,
limit=1000,
offset=next_offset,
with_payload=["source"]
)
points = scroll_response
if not points:
break
for point in points:
source = point.payload.get("source", "unknown")
source_counts[source] = source_counts.get(source, 0) + 1
return source_counts
except Exception as e:
print(f"Error summarizing by source: {e}")
return {}
# Initialize RAG system
rag = QdrantRAG()

View File

@ -1,365 +0,0 @@
import hashlib
import logging
import mimetypes
import os
import traceback
from typing import Dict, List, Any, Optional
from d4science.core.config import load_configuration
from d4science.auth.auth import get_auth_token
from d4science.auth.clients import create_oauth_client
from d4science.storagehub.client import StorageHubClient
from google.adk.tools import ToolContext
from langchain_community.document_loaders.blob_loaders import FileSystemBlobLoader
from langchain_community.document_loaders.parsers.generic import MimeTypeBasedParser
from langchain_community.document_loaders.parsers.html.bs4 import BS4HTMLParser
from langchain_community.document_loaders.parsers.pdf import PDFMinerParser
from langchain_community.document_loaders.parsers.txt import TextParser
from langchain_core.documents import Document
from langchain_core.documents.base import Blob
from langchain_text_splitters import RecursiveCharacterTextSplitter
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue
from sentence_transformers import SentenceTransformer
logging.basicConfig(level=logging.INFO)
# --- D4Science Client Management ---
STORAGEHUB_CLIENT_CACHE: Dict[str, StorageHubClient] = {}
def create_new_storagehub_client() -> StorageHubClient:
"""Factory function to create a new, authenticated StorageHubClient."""
logging.info("Creating a new authenticated StorageHub client...")
config, _ = load_configuration(
env=os.getenv("D4S_ENV") or 'prod',
cli_config_dir=os.getenv("D4S_CONFIG_DIR") or '/etc/d4science',
cli_config_filename=os.getenv("D4S_CONFIG_FILENAME") or 'd4s_config.json'
)
auth_token = get_auth_token(configuration=config)
oauth_client = create_oauth_client(auth_token)
client = StorageHubClient(client=oauth_client, configuration=config)
logging.info("Client created successfully.")
return client
def get_storagehub_client(session_id: str) -> StorageHubClient:
"""Retrieve the StorageHubClient for the given session ID from the cache."""
if session_id not in STORAGEHUB_CLIENT_CACHE:
STORAGEHUB_CLIENT_CACHE[session_id] = create_new_storagehub_client()
return STORAGEHUB_CLIENT_CACHE[session_id]
# --- Document Parsing and Splitting ---
class ParserFactory:
"""A static factory for creating a configured document parser."""
_mime_based_parser = None
@staticmethod
def get_parser():
if ParserFactory._mime_based_parser is None:
handlers = {
"application/pdf": PDFMinerParser(),
"text/plain": TextParser(),
"text/html": BS4HTMLParser(),
}
ParserFactory._mime_based_parser = MimeTypeBasedParser(handlers=handlers, fallback_parser=None)
return ParserFactory._mime_based_parser
def parse_document(file_path):
loader = FileSystemBlobLoader(file_path)
parser = ParserFactory.get_parser()
for blob in loader.yield_blobs():
return parser.parse(blob)
def split_documents(documents: List[Document]) -> List[Document]:
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
logging.info(f"Splitting {len(documents)} document(s) into smaller chunks...")
split_docs = text_splitter.split_documents(documents)
logging.info(f"Successfully created {len(split_docs)} document chunks.")
return split_docs
# --- RAG System (Qdrant) ---
class QdrantRAG:
def __init__(self, host: str = None, port: int = None):
self.host = host or os.getenv('QDRANT_HOST', 'long-term-memory')
self.port = port or int(os.getenv('QDRANT_PORT', '6333'))
self.encoder_model = os.getenv('ENCODER_MODEL', 'all-MiniLM-L6-v2')
self.vector_size = int(os.getenv('VECTOR_SIZE', '384'))
self.collection_name = os.getenv('COLLECTION_NAME', 'dave_knowledge')
self.client = QdrantClient(host=self.host, port=self.port)
self.encoder = SentenceTransformer(self.encoder_model)
print(f"RAG initialized with encoder: {self.encoder_model} (vector size: {self.vector_size})")
self._ensure_collection()
def _ensure_collection(self):
try:
collections = self.client.get_collections().collections
collection_names = [col.name for col in collections]
if self.collection_name not in collection_names:
self.client.create_collection(
collection_name=self.collection_name,
vectors_config=VectorParams(size=self.vector_size, distance=Distance.COSINE)
)
print(f"Created collection: {self.collection_name}")
else:
print(f"Collection {self.collection_name} already exists")
except Exception as e:
print(f"Error ensuring collection: {e}")
def _generate_id(self, text: str, source: str) -> str:
content = f"{source}:{text}"
return hashlib.md5(content.encode()).hexdigest()
def add_documents(self, documents: List[Document]) -> bool:
try:
points = []
for doc in documents:
embedding = self.encoder.encode(doc.page_content).tolist()
payload = doc.metadata
payload["text"] = doc.page_content
source = payload.get("source", "unknown")
doc_id = self._generate_id(doc.page_content, source)
point = PointStruct(id=doc_id, vector=embedding, payload=payload)
points.append(point)
if not points:
print("Warning: No documents provided to add.")
return True
self.client.upsert(collection_name=self.collection_name, points=points, wait=True)
print(f"Successfully added {len(points)} document chunks to memory.")
return True
except Exception as e:
print(f"Error adding documents: {e}")
return False
def search(self, query: str, limit: int = 5, score_threshold: float = 0.5, source: Optional[str] = None) -> List[Dict[str, Any]]:
try:
query_embedding = self.encoder.encode(query).tolist()
search_filter = None
if source:
search_filter = Filter(must=[FieldCondition(key="source", match=MatchValue(value=source))])
search_results = self.client.search(
collection_name=self.collection_name,
query_vector=query_embedding,
query_filter=search_filter,
limit=limit,
score_threshold=score_threshold
)
results = []
for result in search_results:
results.append({
"text": result.payload.get("text", ""),
"source": result.payload.get("source", "unknown"),
"score": result.score,
"chunk_index": result.payload.get("chunk_index", 0),
"metadata": {k: v for k, v in result.payload.items() if k not in ["text", "source", "chunk_index", "total_chunks"]}
})
return results
except Exception as e:
print(f"Error searching: {e}")
return []
def delete_by_source(self, source: str) -> bool:
try:
filter_condition = Filter(must=[FieldCondition(key="source", match=MatchValue(value=source))])
self.client.delete(collection_name=self.collection_name, points_selector=filter_condition)
return True
except Exception as e:
print(f"Error deleting documents: {e}")
return False
def get_collection_info(self) -> Dict[str, Any]:
try:
info = self.client.get_collection(self.collection_name)
return {
"name": info.config.name,
"vector_size": info.config.params.vectors.size,
"distance": info.config.params.vectors.distance,
"points_count": info.points_count,
"status": info.status
}
except Exception as e:
return {"error": str(e)}
def summarize_by_source(self, metadata_filter: Optional[Dict[str, Any]] = None) -> Dict[str, int]:
try:
filter_conditions = Filter(must=[])
if metadata_filter:
for key, value in metadata_filter.items():
condition = FieldCondition(key=key, match=MatchValue(value=value))
filter_conditions.must.append(condition)
if not filter_conditions.must:
filter_conditions = None
source_counts = {}
next_offset = None
while True:
scroll_response, next_offset = self.client.scroll(
collection_name=self.collection_name,
scroll_filter=filter_conditions,
limit=1000,
offset=next_offset,
with_payload=["source"]
)
points = scroll_response
if not points:
break
for point in points:
source = point.payload.get("source", "unknown")
source_counts[source] = source_counts.get(source, 0) + 1
return source_counts
except Exception as e:
print(f"Error summarizing by source: {e}")
return {}
# Initialize RAG system
rag = QdrantRAG()
# --- Agent Tools ---
def list_files(path: str, tool_context: ToolContext) -> str:
"""
Lists all files at the specified D4Science Workspace path.
Args:
path (str): The directory path to list files from.
Returns:
str: JSON-formatted list of files and folders with their metadata.
"""
try:
session_id = tool_context._invocation_context.session.id
storagehub_client = get_storagehub_client(session_id)
root_folder = storagehub_client.workspace_manager.getWorkspace(as_object=False)
root_id = root_folder['id']
if not path or path == "/" or path == "":
target_id = root_id
else:
clean_path = path.strip("/")
if not clean_path:
target_id = root_id
else:
try:
item = storagehub_client.items_manager.getByRelativePath(
parent_id=root_id,
relative_path=clean_path,
as_object=False
)
target_id = item['id']
except Exception:
target_id = path
files = storagehub_client.items_manager.listById(
item_id=target_id,
exclude=["hl:accounting"],
as_object=False
)
return str(files)
except Exception as e:
logging.error(f"An error occurred in the list_files tool: {e}")
logging.error(traceback.format_exc())
return f"Error executing list_files: {e}"
def analyze_file(path: str, tool_context: ToolContext) -> str:
"""
Memorizes and analyzes a file downloaded from the specified D4Science Workspace path.
Its content will remain stored in the agent's long-term memory.
Args:
path (str): The file path to download.
"""
stored_file_path = None
try:
session_id = tool_context._invocation_context.session.id
storagehub_client = get_storagehub_client(session_id)
root_folder = storagehub_client.workspace_manager.getWorkspace(as_object=False)
root_id = root_folder['id']
if not path or path == "/" or path == "":
target_id = root_id
else:
clean_path = path.strip("/")
if not clean_path:
target_id = root_id
else:
try:
item = storagehub_client.items_manager.getByRelativePath(
parent_id=root_id,
relative_path=clean_path,
as_object=False
)
target_id = item['id']
except Exception:
target_id = path
item = storagehub_client.items_manager.get(item_id=target_id, as_object=False) # Get item metadata to have the name
logging.info(f"Downloading file with ID: {target_id}")
stored_file_path = storagehub_client.items_manager.download(item_id=target_id, destination=f'/tmp/__tmp_file__{session_id}_{target_id}_{item["name"]}')
raw_documents = parse_document(stored_file_path)
for doc in raw_documents:
doc.metadata["source"] = item.get("name", "unknown_file")
chunked_documents = split_documents(raw_documents)
rag.add_documents(chunked_documents)
return f"Successfully downloaded, processed, and memorized file: {item['name']}"
except Exception as e:
logging.error(f"An error occurred in the analyze_file tool: {e}")
logging.error(traceback.format_exc())
return f"Error executing analyze_file: {e}"
finally:
if stored_file_path and os.path.exists(stored_file_path):
try:
os.remove(stored_file_path)
logging.info(f"Cleaned up temporary file: {stored_file_path}")
except Exception as cleanup_error:
logging.warning(f"Failed to clean up temporary file {stored_file_path}: {cleanup_error}")
def search_memory(query: str, source: Optional[str] = None, max_results: int = 5, min_score: float = 0.5) -> Dict[str, Any]:
"""Search the long-term memory (vector database) for relevant information."""
try:
logging.info(f"Searching memory for query: '{query}' with source: '{source or ''}'")
results = rag.search(query, source=source, limit=max_results, score_threshold=min_score)
if not results:
return {"status": "success", "message": "No relevant information found in memory", "query": query, "results": []}
return {"status": "success", "query": query, "results": results, "total_found": len(results)}
except Exception as e:
return {"status": "error", "message": f"Error searching memory: {str(e)}", "results": []}
def store_in_memory(content: str, source: str = "user_input", category: str = "general") -> Dict[str, Any]:
"""Store a simple string of information in long-term memory."""
try:
doc = Document(page_content=content, metadata={"source": source, "category": category})
success = rag.add_documents([doc])
if success:
return {"status": "success", "message": f"Content stored in memory under source '{source}'", "source": source, "category": category}
else:
return {"status": "error", "message": "Failed to store content in memory"}
except Exception as e:
return {"status": "error", "message": f"Error storing in memory: {str(e)}"}
def forget_source(source: str) -> Dict[str, Any]:
"""Remove all information from a specific source in long-term memory."""
try:
success = rag.delete_by_source(source)
if success:
return {"status": "success", "message": f"All content from source '{source}' has been removed from memory"}
else:
return {"status": "error", "message": f"Failed to remove content from source '{source}'"}
except Exception as e:
return {"status": "error", "message": f"Error removing from memory: {str(e)}"}
def memory_status() -> Dict[str, Any]:
"""Get status information about the long-term memory system."""
try:
info = rag.get_collection_info()
return {"status": "success", "memory_info": info}
except Exception as e:
return {"status": "error", "message": f"Error getting memory status: {str(e)}"}
def list_memory_entries(metadata_filter: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""List the titles of files in memory, optionally filtering by metadata."""
try:
summary = rag.summarize_by_source(metadata_filter=metadata_filter)
if not summary:
return {"status": "success", "message": "No entries found matching the specified criteria.", "files": []}
files = [{"title": source, "chunks": count} for source, count in summary.items()]
return {"status": "success", "files": files, "total_files": len(files)}
except Exception as e:
return {"status": "error", "message": f"Error listing memory entries: {str(e)}"}

119
tools/scrape_docs.py Normal file
View File

@ -0,0 +1,119 @@
import os
import requests
import re
import shutil
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse
from markdownify import markdownify as md
def scrape_docs(base_url, output_dir_html):
"""
Scrapes a website and saves the HTML pages locally.
"""
print("--- Starting HTML Scraping ---")
os.makedirs(output_dir_html, exist_ok=True)
try:
response = requests.get(base_url)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print(f"Error fetching the main page: {e}")
return
soup = BeautifulSoup(response.content, "html.parser")
links = set([base_url]) # Add the base url to the set of links to scrape
base_url_parsed = urlparse(base_url)
for a_tag in soup.find_all("a", href=True):
href = a_tag["href"]
full_url = urljoin(base_url, href)
full_url_parsed = urlparse(full_url)
# Only scrape links from the same domain
if (full_url_parsed.scheme == base_url_parsed.scheme and
full_url_parsed.netloc == base_url_parsed.netloc):
links.add(full_url)
for link in sorted(list(links)):
try:
page_response = requests.get(link)
page_response.raise_for_status()
html_content = page_response.text
parsed_url = urlparse(link)
relative_path = parsed_url.path.lstrip('/')
if not relative_path or relative_path.endswith('/'):
file_name = os.path.join(relative_path, 'index.html')
elif '.' not in os.path.basename(relative_path):
file_name = relative_path + '/index.html'
else:
file_name = relative_path
html_file_path = os.path.join(output_dir_html, file_name)
os.makedirs(os.path.dirname(html_file_path), exist_ok=True)
with open(html_file_path, "w", encoding="utf-8") as f:
f.write(html_content)
print(f"Downloaded HTML: {link} to {html_file_path}")
except requests.exceptions.RequestException as e:
print(f"Error processing {link}: {e}")
print("--- HTML Scraping Finished ---")
def convert_html_to_markdown(input_dir_html, output_dir_md):
"""
Converts previously downloaded HTML files to Markdown.
"""
print("\n--- Starting Markdown Conversion ---")
# Clean the output directory before writing new files
if os.path.exists(output_dir_md):
shutil.rmtree(output_dir_md)
os.makedirs(output_dir_md, exist_ok=True)
for root, _, files in os.walk(input_dir_html):
for file in files:
if file.endswith(".html"):
html_file_path = os.path.join(root, file)
with open(html_file_path, "r", encoding="utf-8") as f:
html_content = f.read()
soup = BeautifulSoup(html_content, "html.parser")
content_div = soup.find("div", class_="md-content")
if not content_div:
print(f"Warning: 'md-content' div not found in {html_file_path}. Skipping.")
continue
cleaned_html = str(content_div)
markdown_content = md(cleaned_html, heading_style="ATX")
# Adjust links
markdown_content = re.sub(r'\((?!https?:\/\/)([^)]*)/\)', r'(\1/index.md)', markdown_content)
markdown_content = re.sub(r'\.html(?=[)#])', '.md', markdown_content)
# Determine output path
relative_path = os.path.relpath(html_file_path, input_dir_html)
md_relative_path = relative_path.replace(".html", ".md")
md_file_path = os.path.join(output_dir_md, md_relative_path)
os.makedirs(os.path.dirname(md_file_path), exist_ok=True)
with open(md_file_path, "w", encoding="utf-8") as f:
f.write(markdown_content)
print(f"Converted: {html_file_path} -> {md_file_path}")
print("--- Markdown Conversion Finished ---")
if __name__ == "__main__":
BASE_URL = "https://google.github.io/adk-docs/"
HTML_DIR = "docs/adk-docs/html"
MARKDOWN_DIR = "docs/adk-docs/markdown"
# If HTML directory doesn't exist, scrape the docs first
if not os.path.exists(HTML_DIR):
print(f"HTML directory not found at '{HTML_DIR}'. Starting scrape.")
scrape_docs(BASE_URL, HTML_DIR)
else:
print(f"HTML directory found at '{HTML_DIR}'. Skipping scrape.")
# Always run the conversion
convert_html_to_markdown(HTML_DIR, MARKDOWN_DIR)