This commit is contained in:
João Moura
2025-06-01 03:08:25 -07:00
parent 7c12aeaa0c
commit 6ecb30ee87
8 changed files with 1067 additions and 33 deletions

View File

@@ -1 +1,6 @@
"""Agent tools for crewAI."""
from .agent_tools import AgentTools
from .scratchpad_tool import ScratchpadTool
__all__ = ["AgentTools", "ScratchpadTool"]

View File

@@ -0,0 +1,145 @@
"""Tool for accessing data stored in the agent's scratchpad during reasoning."""
from typing import Any, Dict, Optional, Type, Union
from pydantic import BaseModel, Field
from crewai.tools import BaseTool
class ScratchpadToolSchema(BaseModel):
"""Input schema for ScratchpadTool."""
key: str = Field(
...,
description=(
"The key name to retrieve data from the scratchpad. "
"Must be one of the available keys shown in the tool description. "
"Example: if 'email_data' is listed as available, use {\"key\": \"email_data\"}"
)
)
class ScratchpadTool(BaseTool):
"""Tool that allows agents to access data stored in their scratchpad during task execution.
This tool's description is dynamically updated to show all available keys,
making it easy for agents to know what data they can retrieve.
"""
name: str = "Access Scratchpad Memory"
description: str = "Access data stored in your scratchpad memory during task execution."
args_schema: Type[BaseModel] = ScratchpadToolSchema
scratchpad_data: Dict[str, Any] = Field(default_factory=dict)
def __init__(self, scratchpad_data: Optional[Dict[str, Any]] = None, **kwargs):
"""Initialize the scratchpad tool with optional initial data.
Args:
scratchpad_data: Initial scratchpad data (usually from agent state)
"""
super().__init__(**kwargs)
if scratchpad_data:
self.scratchpad_data = scratchpad_data
self._update_description()
def _run(
self,
key: str,
**kwargs: Any,
) -> Union[str, Dict[str, Any], Any]:
"""Retrieve data from the scratchpad using the specified key.
Args:
key: The key to look up in the scratchpad
Returns:
The value associated with the key, or an error message if not found
"""
if not self.scratchpad_data:
return (
"❌ SCRATCHPAD IS EMPTY\n\n"
"The scratchpad does not contain any data yet.\n"
"Data will be automatically stored here as you use other tools.\n"
"Try executing other tools first to gather information."
)
if key not in self.scratchpad_data:
available_keys = list(self.scratchpad_data.keys())
keys_formatted = "\n".join(f" - '{k}'" for k in available_keys)
return (
f"❌ KEY NOT FOUND: '{key}'\n\n"
f"The key '{key}' does not exist in the scratchpad.\n\n"
f"Available keys:\n{keys_formatted}\n\n"
f"To retrieve data, use the EXACT key name from the list above.\n"
f"Example Action Input: {{\"key\": \"{available_keys[0] if available_keys else 'example_key'}\"}}\n\n"
f"Remember: Keys are case-sensitive and must match exactly!"
)
value = self.scratchpad_data[key]
# Format the output nicely based on the type
if isinstance(value, dict):
import json
return json.dumps(value, indent=2)
elif isinstance(value, list):
import json
return json.dumps(value, indent=2)
else:
return str(value)
def update_scratchpad(self, new_data: Dict[str, Any]) -> None:
"""Update the scratchpad data and refresh the tool description.
Args:
new_data: The new complete scratchpad data
"""
self.scratchpad_data = new_data
self._update_description()
def _update_description(self) -> None:
"""Update the tool description to include all available keys."""
base_description = (
"Access data stored in your scratchpad memory during task execution.\n\n"
"HOW TO USE THIS TOOL:\n"
"Provide a JSON object with a 'key' field containing the exact name of the data you want to retrieve.\n"
"Example: {\"key\": \"email_data\"}"
)
if not self.scratchpad_data:
self.description = (
f"{base_description}\n\n"
"📝 STATUS: Scratchpad is currently empty.\n"
"Data will be automatically stored here as you use other tools."
)
return
# Build a description of available keys with a preview of their contents
key_descriptions = []
example_key = None
for key, value in self.scratchpad_data.items():
if not example_key:
example_key = key
# Create a brief description of what's stored
if isinstance(value, dict):
preview = f"dict with {len(value)} items"
if 'data' in value and isinstance(value['data'], list):
preview = f"list of {len(value['data'])} items"
elif isinstance(value, list):
preview = f"list of {len(value)} items"
elif isinstance(value, str):
preview = f"string ({len(value)} chars)"
else:
preview = type(value).__name__
key_descriptions.append(f" 📌 '{key}': {preview}")
available_keys_text = "\n".join(key_descriptions)
self.description = (
f"{base_description}\n\n"
f"📦 AVAILABLE DATA IN SCRATCHPAD:\n{available_keys_text}\n\n"
f"💡 EXAMPLE USAGE:\n"
f"To retrieve the '{example_key}' data, use:\n"
f"Action Input: {{\"key\": \"{example_key}\"}}"
)