Tools - Extending AI Capabilities
Learn to create custom tools and integrate external services, APIs, and systems to dramatically expand what your AI agents can accomplish
🛠️ Understanding Tools in LangChain
Tools are the bridge between AI reasoning and real-world actions. They enable agents to interact with external systems, perform calculations, access databases, call APIs, and execute code - transforming language models from text generators into capable problem-solving systems.
🧠 The Tool Paradigm
text
🧠 FROM LANGUAGE TO ACTION 🧠
(Tools enable real capabilities)
┌─────────────────────────────────────────────────────────────────┐
│ TRADITIONAL LLM │
│ (Text in, text out) │
│ │
│ Input: "What's the weather in New York?" │
│ Output: "I can't check real-time weather..." │
│ │
│ ❌ No external data access │
│ ❌ Can't perform actions │
│ ❌ Limited to training knowledge │
└─────────────────────┬───────────────────────────────────────────┘
│
▼ ADD TOOLS
┌─────────────────────────────────────────────────────────────────┐
│ TOOL-ENABLED AGENT │
│ (Text + actions + data) │
│ │
│ Input: "What's the weather in New York?" │
│ 🔧 Uses weather_api tool │
│ Output: "Current weather in NYC: 72°F, sunny..." │
│ │
│ ✅ Real-time data access │
│ ✅ Performs external actions │
│ ✅ Unlimited capabilities through tools │
└─────────────────────────────────────────────────────────────────┘🏗️ Building Custom Tools
🔧 Basic Tool Creation
python
from langchain.tools import Tool, StructuredTool
from langchain_core.tools import BaseTool
from langchain.pydantic_v1 import BaseModel, Field
from typing import Optional, Type
import requests
import json
import sqlite3
from datetime import datetime
import os
# Simple function-based tool
def calculate_area(shape: str, **kwargs) -> str:
"""
Calculate the area of geometric shapes.
Args:
shape: Type of shape (rectangle, circle, triangle)
**kwargs: Shape-specific parameters
Returns:
Area calculation result
"""
try:
if shape.lower() == "rectangle":
length = kwargs.get("length", 0)
width = kwargs.get("width", 0)
if length <= 0 or width <= 0:
return "Error: Length and width must be positive numbers"
area = length * width
return f"Rectangle area: {area} square units (length: {length}, width: {width})"
elif shape.lower() == "circle":
radius = kwargs.get("radius", 0)
if radius <= 0:
return "Error: Radius must be a positive number"
area = 3.14159 * radius * radius
return f"Circle area: {area:.2f} square units (radius: {radius})"
elif shape.lower() == "triangle":
base = kwargs.get("base", 0)
height = kwargs.get("height", 0)
if base <= 0 or height <= 0:
return "Error: Base and height must be positive numbers"
area = 0.5 * base * height
return f"Triangle area: {area} square units (base: {base}, height: {height})"
else:
return f"Error: Unsupported shape '{shape}'. Supported: rectangle, circle, triangle"
except Exception as e:
return f"Error calculating area: {str(e)}"
# Create tool from function
area_tool = Tool(
name="area_calculator",
description="Calculate the area of geometric shapes. Input format: 'shape|param1:value1|param2:value2' (e.g., 'rectangle|length:5|width:3')",
func=lambda input_str: calculate_area(*_parse_tool_input(input_str))
)
def _parse_tool_input(input_str: str):
"""Parse tool input string into shape and parameters"""
parts = input_str.split('|')
if not parts:
return "rectangle", {}
shape = parts[0]
kwargs = {}
for part in parts[1:]:
if ':' in part:
key, value = part.split(':', 1)
try:
kwargs[key] = float(value)
except ValueError:
kwargs[key] = value
return shape, kwargs
# Test the basic tool
print("🔧 Testing basic area calculator tool:")
test_inputs = [
"rectangle|length:5|width:3",
"circle|radius:4",
"triangle|base:6|height:8"
]
for test_input in test_inputs:
result = area_tool.func(test_input)
print(f"Input: {test_input}")
print(f"Result: {result}\n")🎯 Structured Tools with Pydantic
python
# Define input schema for structured tools
class WeatherInput(BaseModel):
"""Input for weather lookup tool"""
city: str = Field(description="City name to get weather for")
country_code: Optional[str] = Field(default="US", description="Country code (e.g., 'US', 'UK')")
units: Optional[str] = Field(default="metric", description="Temperature units ('metric', 'imperial')")
class DatabaseQueryInput(BaseModel):
"""Input for database query tool"""
query: str = Field(description="SQL query to execute")
database: str = Field(description="Database name to query")
class FileOperationInput(BaseModel):
"""Input for file operations"""
operation: str = Field(description="Operation type: 'read', 'write', 'delete', 'list'")
file_path: str = Field(description="Path to the file")
content: Optional[str] = Field(default=None, description="Content for write operation")
# Structured weather tool
class WeatherTool(BaseTool):
"""Tool for getting weather information"""
name: str = "weather_lookup"
description: str = "Get current weather information for a specific city"
args_schema: Type[BaseModel] = WeatherInput
def _run(self, city: str, country_code: str = "US", units: str = "metric") -> str:
"""Execute weather lookup"""
try:
# In practice, you'd use a real weather API like OpenWeatherMap
# This is a simulation
weather_data = {
"new york": {"temp": 22, "condition": "sunny", "humidity": 65},
"london": {"temp": 15, "condition": "cloudy", "humidity": 80},
"tokyo": {"temp": 28, "condition": "partly cloudy", "humidity": 70},
"sydney": {"temp": 18, "condition": "rainy", "humidity": 85}
}
city_key = city.lower()
if city_key in weather_data:
data = weather_data[city_key]
temp_unit = "°C" if units == "metric" else "°F"
# Convert temperature if needed
temp = data["temp"]
if units == "imperial":
temp = (temp * 9/5) + 32
return f"""Weather in {city}, {country_code}:
Temperature: {temp:.1f}{temp_unit}
Condition: {data['condition']}
Humidity: {data['humidity']}%
Last updated: {datetime.now().strftime('%Y-%m-%d %H:%M')}"""
else:
return f"Weather data not available for {city}. Try: New York, London, Tokyo, or Sydney."
except Exception as e:
return f"Error getting weather for {city}: {str(e)}"
async def _arun(self, city: str, country_code: str = "US", units: str = "metric") -> str:
"""Async version of weather lookup"""
return self._run(city, country_code, units)
# Database query tool
class DatabaseTool(BaseTool):
"""Tool for querying databases"""
name: str = "database_query"
description: str = "Execute SQL queries on databases. Use with caution - only SELECT queries recommended."
args_schema: Type[BaseModel] = DatabaseQueryInput
def _run(self, query: str, database: str) -> str:
"""Execute database query"""
try:
# Create sample in-memory database for demonstration
conn = sqlite3.connect(":memory:")
cursor = conn.cursor()
# Create sample tables
cursor.execute("""
CREATE TABLE employees (
id INTEGER PRIMARY KEY,
name TEXT,
department TEXT,
salary INTEGER
)
""")
cursor.execute("""
CREATE TABLE products (
id INTEGER PRIMARY KEY,
name TEXT,
price REAL,
category TEXT
)
""")
# Insert sample data
employees_data = [
(1, "Alice Johnson", "Engineering", 85000),
(2, "Bob Smith", "Marketing", 65000),
(3, "Carol Davis", "Engineering", 92000),
(4, "David Wilson", "Sales", 58000)
]
products_data = [
(1, "Laptop", 1200.00, "Electronics"),
(2, "Phone", 800.00, "Electronics"),
(3, "Desk", 300.00, "Furniture"),
(4, "Chair", 150.00, "Furniture")
]
cursor.executemany("INSERT INTO employees VALUES (?, ?, ?, ?)", employees_data)
cursor.executemany("INSERT INTO products VALUES (?, ?, ?, ?)", products_data)
# Execute user query (with safety checks)
query_upper = query.upper().strip()
# Basic safety check - only allow SELECT queries
if not query_upper.startswith("SELECT"):
return "Error: Only SELECT queries are allowed for safety"
# Prevent dangerous operations
dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE"]
if any(keyword in query_upper for keyword in dangerous_keywords):
return "Error: Query contains potentially dangerous operations"
cursor.execute(query)
results = cursor.fetchall()
column_names = [description[0] for description in cursor.description]
if not results:
return "Query executed successfully but returned no results"
# Format results as table
formatted_results = []
formatted_results.append(" | ".join(column_names))
formatted_results.append("-" * len(formatted_results[0]))
for row in results:
formatted_results.append(" | ".join(str(value) for value in row))
conn.close()
return f"Query: {query}\n\nResults:\n" + "\n".join(formatted_results)
except Exception as e:
return f"Database error: {str(e)}"
async def _arun(self, query: str, database: str) -> str:
"""Async version of database query"""
return self._run(query, database)
# File operations tool
class FileOperationsTool(BaseTool):
"""Tool for file system operations"""
name: str = "file_operations"
description: str = "Perform file operations: read, write, delete, or list files"
args_schema: Type[BaseModel] = FileOperationInput
def _run(self, operation: str, file_path: str, content: Optional[str] = None) -> str:
"""Execute file operation"""
try:
# Security check - restrict to current directory and subdirectories
if ".." in file_path or file_path.startswith("/"):
return "Error: File path not allowed for security reasons"
if operation == "read":
if not os.path.exists(file_path):
return f"Error: File {file_path} does not exist"
with open(file_path, 'r', encoding='utf-8') as f:
file_content = f.read()
# Limit output length
if len(file_content) > 2000:
file_content = file_content[:2000] + "\n... (truncated)"
return f"Content of {file_path}:\n\n{file_content}"
elif operation == "write":
if content is None:
return "Error: Content is required for write operation"
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
return f"Successfully wrote {len(content)} characters to {file_path}"
elif operation == "delete":
if not os.path.exists(file_path):
return f"Error: File {file_path} does not exist"
os.remove(file_path)
return f"Successfully deleted {file_path}"
elif operation == "list":
if os.path.isfile(file_path):
return f"{file_path} is a file"
elif os.path.isdir(file_path):
files = os.listdir(file_path)
if not files:
return f"Directory {file_path} is empty"
# Limit number of files shown
if len(files) > 20:
shown_files = files[:20]
file_list = "\n".join(shown_files)
return f"Contents of {file_path} (showing first 20 of {len(files)}):\n{file_list}"
else:
file_list = "\n".join(files)
return f"Contents of {file_path}:\n{file_list}"
else:
return f"Error: {file_path} does not exist"
else:
return f"Error: Unknown operation '{operation}'. Use: read, write, delete, or list"
except Exception as e:
return f"File operation error: {str(e)}"
async def _arun(self, operation: str, file_path: str, content: Optional[str] = None) -> str:
"""Async version of file operations"""
return self._run(operation, file_path, content)
# Create structured tools
weather_tool = WeatherTool()
database_tool = DatabaseTool()
file_tool = FileOperationsTool()
print("🎯 Testing structured tools:")
# Test weather tool
weather_result = weather_tool._run("New York", "US", "metric")
print(f"Weather Tool Result:\n{weather_result}\n")
# Test database tool
db_result = database_tool._run("SELECT name, department FROM employees WHERE salary > 70000", "demo")
print(f"Database Tool Result:\n{db_result}\n")
# Test file tool
file_result = file_tool._run("list", ".", None)
print(f"File Tool Result:\n{file_result}")🌐 API Integration Tools
python
import requests
from urllib.parse import quote
import json
class WebSearchTool(BaseTool):
"""Tool for web search using DuckDuckGo API"""
name: str = "web_search"
description: str = "Search the web for current information"
def _run(self, query: str) -> str:
"""Perform web search"""
try:
# Using DuckDuckGo Instant Answer API (free, no key required)
encoded_query = quote(query)
url = f"https://api.duckduckgo.com/?q={encoded_query}&format=json&no_html=1&skip_disambig=1"
response = requests.get(url, timeout=10)
response.raise_for_status()
data = response.json()
# Extract relevant information
results = []
# Abstract (main answer)
if data.get("Abstract"):
results.append(f"Summary: {data['Abstract']}")
if data.get("AbstractURL"):
results.append(f"Source: {data['AbstractURL']}")
# Definition
if data.get("Definition"):
results.append(f"Definition: {data['Definition']}")
if data.get("DefinitionURL"):
results.append(f"Source: {data['DefinitionURL']}")
# Related topics
if data.get("RelatedTopics"):
related = [topic.get("Text", "") for topic in data["RelatedTopics"][:3]]
if related:
results.append(f"Related: {'; '.join(related)}")
# Answer (for calculations, conversions, etc.)
if data.get("Answer"):
results.append(f"Answer: {data['Answer']}")
if results:
return f"Search results for '{query}':\n\n" + "\n\n".join(results)
else:
return f"No specific information found for '{query}'. Try rephrasing your search."
except requests.RequestException as e:
return f"Web search error: {str(e)}"
except Exception as e:
return f"Search processing error: {str(e)}"
class NewsAPITool(BaseTool):
"""Tool for getting news using a news API"""
name: str = "news_search"
description: str = "Get recent news articles about a topic"
def _run(self, topic: str, limit: int = 5) -> str:
"""Get news articles"""
try:
# For demonstration - using a mock news response
# In practice, you'd use NewsAPI, Google News API, etc.
mock_articles = [
{
"title": f"Latest developments in {topic}",
"description": f"Recent news about {topic} shows significant progress...",
"url": "https://example.com/news1",
"published": "2024-01-15T10:30:00Z"
},
{
"title": f"{topic} industry updates",
"description": f"Market analysis reveals new trends in {topic}...",
"url": "https://example.com/news2",
"published": "2024-01-14T15:45:00Z"
}
]
if not mock_articles:
return f"No recent news found for '{topic}'"
results = [f"Recent news about '{topic}':\n"]
for i, article in enumerate(mock_articles[:limit], 1):
results.append(f"{i}. {article['title']}")
results.append(f" {article['description']}")
results.append(f" Published: {article['published']}")
results.append(f" URL: {article['url']}\n")
return "\n".join(results)
except Exception as e:
return f"News search error: {str(e)}"
class TranslationTool(BaseTool):
"""Tool for text translation"""
name: str = "translate_text"
description: str = "Translate text between languages"
def _run(self, text: str, target_language: str = "english", source_language: str = "auto") -> str:
"""Translate text"""
try:
# Mock translation for demonstration
# In practice, you'd use Google Translate API, DeepL, etc.
translations = {
"hola": {"english": "hello", "french": "bonjour"},
"bonjour": {"english": "hello", "spanish": "hola"},
"hello": {"spanish": "hola", "french": "bonjour"},
"gracias": {"english": "thank you", "french": "merci"},
"merci": {"english": "thank you", "spanish": "gracias"}
}
text_lower = text.lower().strip()
target_lower = target_language.lower()
# Simple mock translation
if text_lower in translations and target_lower in translations[text_lower]:
translated = translations[text_lower][target_lower]
return f"Translation from {source_language} to {target_language}:\n'{text}' → '{translated}'"
else:
return f"Translation not available for '{text}' to {target_language}. This is a demo translation tool."
except Exception as e:
return f"Translation error: {str(e)}"
class CalculatorTool(BaseTool):
"""Advanced calculator tool"""
name: str = "calculator"
description: str = "Perform mathematical calculations including basic math, trigonometry, and unit conversions"
def _run(self, expression: str) -> str:
"""Perform calculation"""
try:
import math
# Define safe functions for evaluation
safe_functions = {
'sin': math.sin, 'cos': math.cos, 'tan': math.tan,
'asin': math.asin, 'acos': math.acos, 'atan': math.atan,
'sqrt': math.sqrt, 'log': math.log, 'log10': math.log10,
'exp': math.exp, 'abs': abs, 'round': round,
'pi': math.pi, 'e': math.e
}
# Clean expression
expression = expression.strip()
# Handle unit conversions
if "to" in expression.lower():
return self._handle_unit_conversion(expression)
# Validate expression (only allow safe characters)
allowed_chars = set('0123456789+-*/().,e pi sincotaglrqubdxp ')
if not all(c.lower() in allowed_chars for c in expression):
return f"Error: Expression contains invalid characters"
# Replace function names with safe versions
safe_expression = expression
for func_name, func in safe_functions.items():
if func_name in safe_expression:
if callable(func):
# For functions, we need special handling
continue
else:
safe_expression = safe_expression.replace(func_name, str(func))
# Evaluate safely
result = eval(safe_expression, {"__builtins__": {}}, safe_functions)
return f"Calculation: {expression} = {result}"
except Exception as e:
return f"Calculation error: {str(e)}"
def _handle_unit_conversion(self, expression: str) -> str:
"""Handle unit conversions"""
conversions = {
# Temperature
"celsius to fahrenheit": lambda c: (c * 9/5) + 32,
"fahrenheit to celsius": lambda f: (f - 32) * 5/9,
# Length
"meters to feet": lambda m: m * 3.28084,
"feet to meters": lambda f: f / 3.28084,
"kilometers to miles": lambda km: km * 0.621371,
"miles to kilometers": lambda mi: mi / 0.621371,
# Weight
"kg to pounds": lambda kg: kg * 2.20462,
"pounds to kg": lambda lb: lb / 2.20462,
}
try:
# Parse expression like "25 celsius to fahrenheit"
parts = expression.lower().split()
if len(parts) >= 4 and "to" in parts:
value = float(parts[0])
unit_from = parts[1]
unit_to = parts[3]
conversion_key = f"{unit_from} to {unit_to}"
if conversion_key in conversions:
result = conversions[conversion_key](value)
return f"Conversion: {value} {unit_from} = {result:.2f} {unit_to}"
else:
available = list(conversions.keys())
return f"Conversion not available. Available conversions: {', '.join(available)}"
else:
return "Invalid conversion format. Use: 'value unit to unit' (e.g., '25 celsius to fahrenheit')"
except ValueError:
return "Error: Invalid number in conversion"
except Exception as e:
return f"Conversion error: {str(e)}"
# Create API tools
web_search_tool = WebSearchTool()
news_tool = NewsAPITool()
translation_tool = TranslationTool()
calculator_tool = CalculatorTool()
# Tool collection
api_tools = [web_search_tool, news_tool, translation_tool, calculator_tool]
print("🌐 Testing API integration tools:")
# Test web search
search_result = web_search_tool._run("Python programming language")
print(f"Web Search Result:\n{search_result}\n")
# Test calculator
calc_result = calculator_tool._run("25 celsius to fahrenheit")
print(f"Calculator Result:\n{calc_result}\n")
# Test translation
trans_result = translation_tool._run("hello", "spanish")
print(f"Translation Result:\n{trans_result}")🔧 Advanced Tool Patterns
🎯 Tool Composition and Chaining
python
class CompositeToolSystem:
"""System for combining multiple tools into complex workflows"""
def __init__(self, tools: list):
self.tools = {tool.name: tool for tool in tools}
self.workflow_history = []
def create_workflow(self, workflow_name: str, steps: list):
"""Create a multi-step workflow"""
class WorkflowTool(BaseTool):
name: str = workflow_name
description: str = f"Execute {workflow_name} workflow with multiple steps"
def __init__(self, parent_system, workflow_steps):
super().__init__()
self.parent_system = parent_system
self.steps = workflow_steps
def _run(self, input_data: str) -> str:
"""Execute workflow steps"""
results = []
current_data = input_data
for step in self.steps:
tool_name = step["tool"]
operation = step["operation"]
if tool_name not in self.parent_system.tools:
return f"Error: Tool '{tool_name}' not found"
tool = self.parent_system.tools[tool_name]
try:
# Execute tool with current data
if hasattr(tool, '_run'):
if tool_name == "calculator":
result = tool._run(operation.format(data=current_data))
elif tool_name == "web_search":
result = tool._run(operation.format(query=current_data))
else:
result = tool._run(current_data)
else:
result = tool.func(current_data)
results.append(f"Step {len(results)+1} ({tool_name}): {result}")
current_data = result
except Exception as e:
return f"Workflow failed at step {len(results)+1}: {str(e)}"
return f"Workflow '{workflow_name}' completed:\n\n" + "\n\n".join(results)
workflow_tool = WorkflowTool(self, steps)
self.tools[workflow_name] = workflow_tool
return workflow_tool
def execute_tool_chain(self, chain_config: dict) -> str:
"""Execute a chain of tools with data flow"""
results = []
current_output = chain_config.get("initial_input", "")
for step in chain_config["steps"]:
tool_name = step["tool"]
if tool_name not in self.tools:
return f"Error: Tool '{tool_name}' not found in chain"
tool = self.tools[tool_name]
# Process current output as input for next tool
try:
if hasattr(tool, '_run'):
result = tool._run(current_output)
else:
result = tool.func(current_output)
results.append({
"tool": tool_name,
"input": current_output,
"output": result
})
# Extract relevant data for next step
current_output = step.get("output_transform", lambda x: x)(result)
except Exception as e:
return f"Chain failed at tool '{tool_name}': {str(e)}"
return self._format_chain_results(results)
def _format_chain_results(self, results: list) -> str:
"""Format chain execution results"""
formatted = ["Tool Chain Execution Results:\n"]
for i, result in enumerate(results, 1):
formatted.append(f"Step {i}: {result['tool']}")
formatted.append(f"Input: {result['input'][:100]}...")
formatted.append(f"Output: {result['output'][:200]}...")
formatted.append("---")
return "\n".join(formatted)
# Create composite tool system
composite_system = CompositeToolSystem([
weather_tool, calculator_tool, web_search_tool, file_tool
])
# Create a research workflow
research_workflow = composite_system.create_workflow(
"research_and_calculate",
[
{"tool": "web_search", "operation": "{query}"},
{"tool": "calculator", "operation": "25 + 30"}, # Example calculation
{"tool": "file_operations", "operation": "write summary to file"}
]
)
print("🎯 Tool Composition System Created")
# Test tool chain
chain_config = {
"initial_input": "artificial intelligence",
"steps": [
{
"tool": "web_search",
"output_transform": lambda x: "AI research trends"
},
{
"tool": "calculator",
"output_transform": lambda x: "50 + 25"
}
]
}
chain_result = composite_system.execute_tool_chain(chain_config)
print(f"Tool Chain Result:\n{chain_result}")🔄 Dynamic Tool Discovery
python
class DynamicToolRegistry:
"""Registry for dynamic tool discovery and management"""
def __init__(self):
self.tools = {}
self.tool_categories = {}
self.usage_stats = {}
def register_tool(self, tool: BaseTool, category: str = "general", metadata: dict = None):
"""Register a tool with metadata"""
tool_info = {
"tool": tool,
"category": category,
"metadata": metadata or {},
"registered_at": datetime.now(),
"usage_count": 0
}
self.tools[tool.name] = tool_info
# Add to category
if category not in self.tool_categories:
self.tool_categories[category] = []
self.tool_categories[category].append(tool.name)
def discover_tools_for_task(self, task_description: str, max_tools: int = 5) -> list:
"""Discover relevant tools for a task"""
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(temperature=0)
# Get tool descriptions
tool_descriptions = []
for name, info in self.tools.items():
tool = info["tool"]
tool_descriptions.append(f"- {name}: {tool.description}")
discovery_prompt = f"""
Given this task: "{task_description}"
Available tools:
{chr(10).join(tool_descriptions)}
Select the most relevant tools for this task (maximum {max_tools}).
Respond with just the tool names, one per line.
"""
response = llm.invoke(discovery_prompt).content
# Parse response
suggested_tools = []
for line in response.strip().split('\n'):
tool_name = line.strip().replace('-', '').replace('*', '').strip()
if tool_name in self.tools:
suggested_tools.append(self.tools[tool_name]["tool"])
return suggested_tools[:max_tools]
def get_tools_by_category(self, category: str) -> list:
"""Get all tools in a category"""
if category not in self.tool_categories:
return []
return [self.tools[name]["tool"] for name in self.tool_categories[category]]
def get_tool_usage_stats(self) -> dict:
"""Get usage statistics for all tools"""
stats = {}
for name, info in self.tools.items():
stats[name] = {
"usage_count": info["usage_count"],
"category": info["category"],
"last_used": info.get("last_used", "Never")
}
return stats
def execute_tool(self, tool_name: str, *args, **kwargs) -> str:
"""Execute a tool and track usage"""
if tool_name not in self.tools:
return f"Error: Tool '{tool_name}' not found"
tool_info = self.tools[tool_name]
tool = tool_info["tool"]
try:
# Execute tool
if hasattr(tool, '_run'):
result = tool._run(*args, **kwargs)
else:
result = tool.func(*args, **kwargs)
# Update usage stats
tool_info["usage_count"] += 1
tool_info["last_used"] = datetime.now()
return result
except Exception as e:
return f"Error executing {tool_name}: {str(e)}"
def create_tool_manifest(self) -> dict:
"""Create a manifest of all available tools"""
manifest = {
"total_tools": len(self.tools),
"categories": {},
"tools": {}
}
# Category summary
for category, tool_names in self.tool_categories.items():
manifest["categories"][category] = {
"count": len(tool_names),
"tools": tool_names
}
# Tool details
for name, info in self.tools.items():
tool = info["tool"]
manifest["tools"][name] = {
"description": tool.description,
"category": info["category"],
"usage_count": info["usage_count"],
"metadata": info["metadata"]
}
return manifest
# Create dynamic tool registry
tool_registry = DynamicToolRegistry()
# Register tools with categories
tool_registry.register_tool(weather_tool, "data", {"data_source": "api", "real_time": True})
tool_registry.register_tool(calculator_tool, "computation", {"capabilities": ["math", "conversions"]})
tool_registry.register_tool(web_search_tool, "information", {"scope": "web", "real_time": True})
tool_registry.register_tool(file_tool, "system", {"permissions": ["read", "write"]})
tool_registry.register_tool(database_tool, "data", {"query_language": "sql"})
print("🔄 Dynamic Tool Registry Created")
# Test tool discovery
task = "I need to find current weather information and perform some calculations"
discovered_tools = tool_registry.discover_tools_for_task(task)
print(f"Tools discovered for task: {[tool.name for tool in discovered_tools]}")
# Get tool manifest
manifest = tool_registry.create_tool_manifest()
print(f"Tool manifest: {json.dumps(manifest, indent=2, default=str)}")🧠 Intelligent Tool Selection
python
class IntelligentToolSelector:
"""AI-powered tool selection and orchestration"""
def __init__(self, tool_registry: DynamicToolRegistry, llm):
self.registry = tool_registry
self.llm = llm
self.selection_history = []
def analyze_task_requirements(self, task: str) -> dict:
"""Analyze what capabilities are needed for a task"""
analysis_prompt = f"""
Analyze this task and determine what capabilities are needed:
Task: {task}
Provide analysis in the following format:
Primary Goal: [main objective]
Required Capabilities: [list of needed capabilities]
Data Sources: [what data sources might be needed]
Output Format: [how results should be presented]
Complexity: [simple/moderate/complex]
Analysis:"""
response = self.llm.invoke(analysis_prompt).content
# Parse response (simplified)
analysis = {
"task": task,
"analysis": response,
"timestamp": datetime.now()
}
return analysis
def select_optimal_tools(self, task: str, max_tools: int = 3) -> dict:
"""Select optimal tools for a task with reasoning"""
# Get task analysis
analysis = self.analyze_task_requirements(task)
# Get available tools
available_tools = list(self.registry.tools.keys())
tool_descriptions = []
for name in available_tools:
tool_info = self.registry.tools[name]
tool = tool_info["tool"]
category = tool_info["category"]
usage_count = tool_info["usage_count"]
tool_descriptions.append(
f"- {name} ({category}): {tool.description} [Used {usage_count} times]"
)
selection_prompt = f"""
Task Analysis:
{analysis['analysis']}
Available Tools:
{chr(10).join(tool_descriptions)}
Select the best tools for this task (maximum {max_tools}).
Consider:
1. Tool relevance to task requirements
2. Tool reliability (usage history)
3. Potential tool combinations
Provide your selection in this format:
Selected Tools: [tool1, tool2, tool3]
Reasoning: [why these tools were chosen]
Execution Order: [suggested order of tool usage]
Selection:"""
response = self.llm.invoke(selection_prompt).content
# Parse tool selection
selected_tools = []
reasoning = ""
execution_order = ""
lines = response.split('\n')
for line in lines:
if line.startswith("Selected Tools:"):
# Extract tool names
tools_text = line.replace("Selected Tools:", "").strip()
tools_text = tools_text.replace("[", "").replace("]", "")
tool_names = [name.strip() for name in tools_text.split(",")]
for name in tool_names:
if name in self.registry.tools:
selected_tools.append(self.registry.tools[name]["tool"])
elif line.startswith("Reasoning:"):
reasoning = line.replace("Reasoning:", "").strip()
elif line.startswith("Execution Order:"):
execution_order = line.replace("Execution Order:", "").strip()
selection_result = {
"task": task,
"analysis": analysis,
"selected_tools": selected_tools,
"reasoning": reasoning,
"execution_order": execution_order,
"selection_time": datetime.now()
}
self.selection_history.append(selection_result)
return selection_result
def create_execution_plan(self, selection_result: dict) -> dict:
"""Create detailed execution plan"""
tools = selection_result["selected_tools"]
task = selection_result["task"]
plan_prompt = f"""
Create an execution plan for this task using the selected tools:
Task: {task}
Selected Tools: {[tool.name for tool in tools]}
Create a step-by-step plan:
1. What inputs are needed?
2. In what order should tools be used?
3. How should outputs be processed?
4. What is the expected final result?
Execution Plan:"""
plan = self.llm.invoke(plan_prompt).content
return {
"task": task,
"tools": tools,
"plan": plan,
"created_at": datetime.now()
}
def execute_plan(self, execution_plan: dict) -> dict:
"""Execute the planned tool sequence"""
task = execution_plan["task"]
tools = execution_plan["tools"]
plan = execution_plan["plan"]
execution_results = {
"task": task,
"plan": plan,
"steps": [],
"success": False,
"final_result": "",
"execution_time": 0
}
start_time = datetime.now()
try:
# For demonstration, execute tools in order
for i, tool in enumerate(tools):
step_start = datetime.now()
# Create appropriate input for each tool
if tool.name == "weather_lookup":
result = tool._run("New York")
elif tool.name == "calculator":
result = tool._run("25 + 30")
elif tool.name == "web_search":
result = tool._run(task)
else:
result = f"Tool {tool.name} executed for task: {task}"
step_time = (datetime.now() - step_start).total_seconds()
execution_results["steps"].append({
"step": i + 1,
"tool": tool.name,
"result": result,
"execution_time": step_time
})
# Combine results
final_result = self._combine_tool_results(execution_results["steps"], task)
execution_results["final_result"] = final_result
execution_results["success"] = True
except Exception as e:
execution_results["error"] = str(e)
execution_results["execution_time"] = (datetime.now() - start_time).total_seconds()
return execution_results
def _combine_tool_results(self, steps: list, task: str) -> str:
"""Combine tool results into final answer"""
results_text = []
for step in steps:
results_text.append(f"{step['tool']}: {step['result']}")
combination_prompt = f"""
Original Task: {task}
Tool Results:
{chr(10).join(results_text)}
Combine these results into a comprehensive answer for the original task:
"""
combined_result = self.llm.invoke(combination_prompt).content
return combined_result
# Create intelligent tool selector
intelligent_selector = IntelligentToolSelector(tool_registry, ChatOpenAI())
print("🧠 Intelligent Tool Selector Created")
# Test intelligent tool selection
test_task = "I need to check the weather in New York and calculate how much I'd save on heating costs if the temperature rises by 10 degrees"
selection_result = intelligent_selector.select_optimal_tools(test_task)
print(f"Tool Selection Result:")
print(f"Selected Tools: {[tool.name for tool in selection_result['selected_tools']]}")
print(f"Reasoning: {selection_result['reasoning']}")
# Create and execute plan
execution_plan = intelligent_selector.create_execution_plan(selection_result)
print(f"\nExecution Plan:\n{execution_plan['plan']}")
execution_result = intelligent_selector.execute_plan(execution_plan)
print(f"\nExecution Result:\n{execution_result['final_result']}")🔗 Tool Integration with Agents
🤖 Agent-Tool Integration
python
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
class ToolAwareAgent:
"""Agent with advanced tool awareness and selection"""
def __init__(self, tool_registry: DynamicToolRegistry):
self.registry = tool_registry
self.llm = ChatOpenAI(temperature=0.1)
self.agent_executor = self._create_agent()
def _create_agent(self):
"""Create agent with all registered tools"""
# Get all tools from registry
all_tools = [info["tool"] for info in self.registry.tools.values()]
# Create tool-aware prompt
prompt = ChatPromptTemplate.from_messages([
("system", """You are an intelligent assistant with access to various tools.
Available tool categories:
- Data: Access weather, databases, and real-time information
- Computation: Perform calculations and conversions
- Information: Search web and get current news
- System: File operations and system tasks
Instructions:
1. Analyze each request to understand what capabilities are needed
2. Choose the most appropriate tools for the task
3. Use tools in the optimal order
4. Combine results to provide comprehensive answers
5. Explain your tool choices when relevant
Always think step by step and use tools effectively."""),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
# Create agent
agent = create_openai_functions_agent(
llm=self.llm,
tools=all_tools,
prompt=prompt
)
return AgentExecutor(
agent=agent,
tools=all_tools,
verbose=True,
handle_parsing_errors=True,
max_iterations=10
)
def query(self, user_input: str) -> dict:
"""Process user query with tool usage tracking"""
start_time = datetime.now()
try:
# Execute agent
result = self.agent_executor.invoke({"input": user_input})
# Track tool usage
used_tools = self._extract_used_tools(result)
return {
"input": user_input,
"output": result["output"],
"tools_used": used_tools,
"success": True,
"execution_time": (datetime.now() - start_time).total_seconds()
}
except Exception as e:
return {
"input": user_input,
"error": str(e),
"success": False,
"execution_time": (datetime.now() - start_time).total_seconds()
}
def _extract_used_tools(self, result: dict) -> list:
"""Extract which tools were used during execution"""
# This is a simplified extraction
# In practice, you'd implement more sophisticated tracking
used_tools = []
if "intermediate_steps" in result:
for step in result["intermediate_steps"]:
if hasattr(step[0], 'tool'):
used_tools.append(step[0].tool)
return used_tools
def get_tool_recommendations(self, task: str) -> dict:
"""Get tool recommendations without executing"""
# Use intelligent selector for recommendations
selector = IntelligentToolSelector(self.registry, self.llm)
selection_result = selector.select_optimal_tools(task)
return {
"task": task,
"recommended_tools": [tool.name for tool in selection_result["selected_tools"]],
"reasoning": selection_result["reasoning"],
"execution_order": selection_result["execution_order"]
}
# Create tool-aware agent
tool_agent = ToolAwareAgent(tool_registry)
print("🤖 Tool-Aware Agent Created")
# Test complex queries
test_queries = [
"What's the weather in London and how many degrees Celsius is 68 Fahrenheit?",
"Search for information about machine learning and save a summary to a file",
"Check the weather in Tokyo and find recent news about Japan"
]
for query in test_queries:
print(f"\n--- Query: {query} ---")
# Get recommendations first
recommendations = tool_agent.get_tool_recommendations(query)
print(f"Recommended tools: {recommendations['recommended_tools']}")
print(f"Reasoning: {recommendations['reasoning']}")
# Execute query
result = tool_agent.query(query)
if result["success"]:
print(f"Result: {result['output']}")
print(f"Tools used: {result['tools_used']}")
print(f"Execution time: {result['execution_time']:.2f}s")
else:
print(f"Error: {result['error']}")🔗 Next Steps
Ready to dive deeper into AI systems? Continue with:
- Memory Basics - Add conversation memory and context
- Callbacks and Monitoring - Monitor and debug tool usage
- Advanced RAG - Combine tools with retrieval systems
Key Tool Takeaways:
- Tools extend AI capabilities beyond text generation to real-world actions
- Structured tools with Pydantic schemas provide better validation and documentation
- Tool composition enables complex workflows through tool chaining
- Dynamic discovery helps agents select optimal tools for tasks
- API integration opens unlimited possibilities for external service access
- Intelligent selection improves tool usage efficiency and effectiveness
- Agent integration creates powerful autonomous systems with tool access