mirror of
https://github.com/MervinPraison/PraisonAI-Tools.git
synced 2026-11-07 05:56:17 -10:00
Initial commit: PraisonAI Tools v0.1.0
A comprehensive toolkit for AI agents providing web search, content extraction, and research capabilities. Features: - BaseTool abstract class with schema generation - @tool decorator for function-based tools - TavilyTools: search, extract, crawl, map - ExaTools: search, search_contents, find_similar, answer - YouTools: search, contents, news, images - Comprehensive test suite (60 tests)
This commit is contained in:
Mervin Praison committed
commit
be857cc01e
19 files changed
+3637
No files matched your search
@@ -0,0 +1,31 @@
|
||||
# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
|
||||
# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
|
||||
|
||||
name: Python package
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [ "develop" ]
|
||||
pull_request:
|
||||
branches: [ "develop" ]
|
||||
|
||||
jobs:
|
||||
build:
|
||||
|
||||
runs-on: ubuntu-latest
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
python-version: ["3.10", "3.11"]
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
- name: Set up Python ${{ matrix.python-version }}
|
||||
uses: actions/setup-python@v3
|
||||
with:
|
||||
python-version: ${{ matrix.python-version }}
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
python -m pip install flake8 pytest
|
||||
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
|
||||
@@ -0,0 +1,39 @@
|
||||
# This workflow will upload a Python Package using Twine when a release is created
|
||||
# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python#publishing-to-package-registries
|
||||
|
||||
# This workflow uses actions that are not certified by GitHub.
|
||||
# They are provided by a third-party and are governed by
|
||||
# separate terms of service, privacy policy, and support
|
||||
# documentation.
|
||||
|
||||
name: Upload Python Package
|
||||
|
||||
on:
|
||||
release:
|
||||
types: [published]
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
deploy:
|
||||
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v3
|
||||
with:
|
||||
python-version: '3.x'
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
pip install build
|
||||
- name: Build package
|
||||
run: python -m build
|
||||
- name: Publish package
|
||||
uses: pypa/[email protected]
|
||||
with:
|
||||
user: __token__
|
||||
password: ${{ secrets.PYPI_API_TOKEN }}
|
||||
+50
@@ -0,0 +1,50 @@
|
||||
venv
|
||||
.DS_Store
|
||||
.pytest_cache
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
|
||||
# Distribution / packaging
|
||||
.Python
|
||||
build/
|
||||
dist/
|
||||
*.egg-info/
|
||||
|
||||
# Virtual environments
|
||||
.venv/
|
||||
venv/
|
||||
|
||||
# IDE
|
||||
.vscode/
|
||||
.idea/
|
||||
|
||||
# Testing
|
||||
.pytest_cache/
|
||||
.coverage
|
||||
htmlcov/
|
||||
|
||||
# Lock files
|
||||
poetry.lock
|
||||
uv.lock
|
||||
|
||||
# Database
|
||||
*.db
|
||||
db/
|
||||
chroma.sqlite3
|
||||
|
||||
# OS
|
||||
.DS_Store
|
||||
praisonai
|
||||
.cache
|
||||
__pycache__
|
||||
test/
|
||||
.env
|
||||
assets/*
|
||||
.idea
|
||||
.DS_Store
|
||||
.pytest_cache
|
||||
praisonAI.egg-info
|
||||
flagged
|
||||
test.yaml
|
||||
db
|
||||
@@ -0,0 +1,21 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2024 Mervin Praison
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
@@ -0,0 +1,264 @@
|
||||
# PraisonAI Tools
|
||||
|
||||
A comprehensive toolkit for AI agents providing web search, content extraction, and research capabilities.
|
||||
|
||||
[](https://badge.fury.io/py/praisonai-tools)
|
||||
[](https://opensource.org/licenses/MIT)
|
||||
[](https://www.python.org/downloads/)
|
||||
|
||||
## Features
|
||||
|
||||
- **Tavily Tools** - AI-powered web search, content extraction, crawling, and site mapping
|
||||
- **Exa Tools** - Neural search, content retrieval, similar page discovery, and AI answers
|
||||
- **You.com Tools** - Unified web/news search, content extraction, and image search
|
||||
- **Extensible Base Classes** - Create custom tools with `BaseTool` and `@tool` decorator
|
||||
|
||||
## Installation
|
||||
|
||||
```bash
|
||||
# Basic installation (no external dependencies)
|
||||
pip install praisonai-tools
|
||||
|
||||
# With specific tool support
|
||||
pip install praisonai-tools[tavily] # Tavily tools
|
||||
pip install praisonai-tools[exa] # Exa tools
|
||||
pip install praisonai-tools[youdotcom] # You.com tools
|
||||
|
||||
# All tools
|
||||
pip install praisonai-tools[all]
|
||||
|
||||
# Development
|
||||
pip install praisonai-tools[dev]
|
||||
```
|
||||
|
||||
## Quick Start
|
||||
|
||||
### Tavily Search
|
||||
|
||||
```python
|
||||
from praisonai_tools import tavily_search
|
||||
|
||||
# Basic search
|
||||
results = tavily_search("What is machine learning?")
|
||||
print(results)
|
||||
|
||||
# Advanced search with answer
|
||||
results = tavily_search(
|
||||
"Latest AI developments 2024",
|
||||
include_answer=True,
|
||||
topic="news",
|
||||
max_results=5
|
||||
)
|
||||
print(results["answer"])
|
||||
```
|
||||
|
||||
### Exa Search
|
||||
|
||||
```python
|
||||
from praisonai_tools import exa_search, exa_answer
|
||||
|
||||
# Search
|
||||
results = exa_search("AI startups", num_results=5)
|
||||
|
||||
# Get AI-generated answer with citations
|
||||
answer = exa_answer("What are the benefits of Python?")
|
||||
print(answer["answer"])
|
||||
print(answer["citations"])
|
||||
```
|
||||
|
||||
### You.com Search
|
||||
|
||||
```python
|
||||
from praisonai_tools import ydc_search, ydc_contents
|
||||
|
||||
# Unified search (web + news)
|
||||
results = ydc_search("technology trends", count=10)
|
||||
|
||||
# Extract content from URLs
|
||||
content = ydc_contents("https://example.com", format="markdown")
|
||||
```
|
||||
|
||||
## Environment Variables
|
||||
|
||||
Set the following environment variables for the tools you want to use:
|
||||
|
||||
```bash
|
||||
# Tavily (https://tavily.com)
|
||||
export TAVILY_API_KEY=your_tavily_api_key
|
||||
|
||||
# Exa (https://exa.ai)
|
||||
export EXA_API_KEY=your_exa_api_key
|
||||
|
||||
# You.com (https://you.com/api)
|
||||
export YDC_API_KEY=your_ydc_api_key
|
||||
```
|
||||
|
||||
## Tool Classes
|
||||
|
||||
For more control, use the tool classes directly:
|
||||
|
||||
### TavilyTools
|
||||
|
||||
```python
|
||||
from praisonai_tools import TavilyTools
|
||||
|
||||
tavily = TavilyTools()
|
||||
|
||||
# Search
|
||||
results = tavily.search("Python programming", max_results=5)
|
||||
|
||||
# Extract content from URLs
|
||||
content = tavily.extract(["https://python.org", "https://docs.python.org"])
|
||||
|
||||
# Crawl a website
|
||||
crawl_results = tavily.crawl("https://docs.python.org", max_depth=2, limit=20)
|
||||
|
||||
# Get site map
|
||||
sitemap = tavily.map("https://docs.python.org", limit=50)
|
||||
```
|
||||
|
||||
### ExaTools
|
||||
|
||||
```python
|
||||
from praisonai_tools import ExaTools
|
||||
|
||||
exa = ExaTools()
|
||||
|
||||
# Basic search
|
||||
results = exa.search("machine learning papers", category="research paper")
|
||||
|
||||
# Search with content
|
||||
results = exa.search_and_contents("AI news", text=True, highlights=True)
|
||||
|
||||
# Find similar pages
|
||||
similar = exa.find_similar("https://openai.com", num_results=5)
|
||||
|
||||
# Get answer with citations
|
||||
answer = exa.answer("What is GPT-4?", text=True)
|
||||
```
|
||||
|
||||
### YouTools
|
||||
|
||||
```python
|
||||
from praisonai_tools import YouTools
|
||||
|
||||
you = YouTools()
|
||||
|
||||
# Unified search
|
||||
results = you.search("latest tech news", count=10, freshness="week")
|
||||
|
||||
# Extract content
|
||||
content = you.get_contents("https://example.com", format="markdown")
|
||||
|
||||
# Live news
|
||||
news = you.live_news("AI developments", count=5)
|
||||
|
||||
# Image search
|
||||
images = you.images("python programming")
|
||||
```
|
||||
|
||||
## Creating Custom Tools
|
||||
|
||||
### Using BaseTool
|
||||
|
||||
```python
|
||||
from praisonai_tools import BaseTool
|
||||
|
||||
class WeatherTool(BaseTool):
|
||||
name = "get_weather"
|
||||
description = "Get current weather for a location"
|
||||
|
||||
def run(self, location: str, units: str = "celsius") -> dict:
|
||||
# Your implementation here
|
||||
return {"location": location, "temp": 22, "units": units}
|
||||
|
||||
# Use the tool
|
||||
weather = WeatherTool()
|
||||
result = weather.run(location="London")
|
||||
print(result)
|
||||
|
||||
# Get OpenAI-compatible schema
|
||||
schema = weather.get_schema()
|
||||
```
|
||||
|
||||
### Using @tool Decorator
|
||||
|
||||
```python
|
||||
from praisonai_tools import tool
|
||||
|
||||
@tool
|
||||
def calculate(expression: str) -> float:
|
||||
"""Evaluate a mathematical expression."""
|
||||
return eval(expression)
|
||||
|
||||
# Use directly
|
||||
result = calculate(expression="2 + 2 * 3")
|
||||
print(result) # 8
|
||||
|
||||
# Get schema
|
||||
schema = calculate.get_schema()
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
### Tavily Functions
|
||||
|
||||
| Function | Description |
|
||||
|----------|-------------|
|
||||
| `tavily_search(query, ...)` | Search the web with AI-powered ranking |
|
||||
| `tavily_extract(urls, ...)` | Extract content from URLs |
|
||||
| `tavily_crawl(url, ...)` | Crawl a website |
|
||||
| `tavily_map(url, ...)` | Get a sitemap |
|
||||
|
||||
### Exa Functions
|
||||
|
||||
| Function | Description |
|
||||
|----------|-------------|
|
||||
| `exa_search(query, ...)` | Neural web search |
|
||||
| `exa_search_contents(query, ...)` | Search with full content |
|
||||
| `exa_find_similar(url, ...)` | Find similar pages |
|
||||
| `exa_answer(query, ...)` | Get AI answer with citations |
|
||||
|
||||
### You.com Functions
|
||||
|
||||
| Function | Description |
|
||||
|----------|-------------|
|
||||
| `ydc_search(query, ...)` | Unified web and news search |
|
||||
| `ydc_contents(urls, ...)` | Extract content from URLs |
|
||||
| `ydc_news(query, ...)` | Live news search |
|
||||
| `ydc_images(query)` | Image search |
|
||||
|
||||
## Testing
|
||||
|
||||
Run tests with real API keys:
|
||||
|
||||
```bash
|
||||
# Install dev dependencies
|
||||
pip install praisonai-tools[dev]
|
||||
|
||||
# Run all tests
|
||||
pytest tests/ -v
|
||||
|
||||
# Run specific tool tests
|
||||
pytest tests/test_tavily.py -v -m tavily
|
||||
pytest tests/test_exa.py -v -m exa
|
||||
pytest tests/test_youdotcom.py -v -m youdotcom
|
||||
|
||||
# Run base tests (no API keys needed)
|
||||
pytest tests/test_base.py -v
|
||||
```
|
||||
|
||||
## License
|
||||
|
||||
MIT License - see [LICENSE](LICENSE) for details.
|
||||
|
||||
## Author
|
||||
|
||||
**Mervin Praison** - [GitHub](https://github.com/MervinPraison)
|
||||
|
||||
## Links
|
||||
|
||||
- [Documentation](https://docs.praison.ai/tools)
|
||||
- [PyPI](https://pypi.org/project/praisonai-tools/)
|
||||
- [GitHub](https://github.com/MervinPraison/PraisonAI-Tools)
|
||||
- [Issues](https://github.com/MervinPraison/PraisonAI-Tools/issues)
|
||||
@@ -0,0 +1,77 @@
|
||||
"""PraisonAI Tools - A comprehensive toolkit for AI agents.
|
||||
|
||||
This package provides a collection of tools that can be used with PraisonAI agents
|
||||
for web search, content extraction, file operations, and more.
|
||||
|
||||
Usage:
|
||||
from praisonai_tools import (
|
||||
# Search tools
|
||||
tavily_search, exa_search, ydc_search,
|
||||
|
||||
# Tool classes
|
||||
TavilyTools, ExaTools, YouTools,
|
||||
|
||||
# Base classes for custom tools
|
||||
BaseTool, tool
|
||||
)
|
||||
"""
|
||||
|
||||
from praisonai_tools.tools.base import BaseTool, ToolResult, ToolValidationError
|
||||
from praisonai_tools.tools.decorator import tool, FunctionTool
|
||||
|
||||
# Import all tools
|
||||
from praisonai_tools.tools import (
|
||||
# Tavily Tools
|
||||
TavilyTools,
|
||||
tavily_search,
|
||||
tavily_extract,
|
||||
tavily_crawl,
|
||||
tavily_map,
|
||||
|
||||
# Exa Tools
|
||||
ExaTools,
|
||||
exa_search,
|
||||
exa_search_contents,
|
||||
exa_find_similar,
|
||||
exa_answer,
|
||||
|
||||
# You.com Tools
|
||||
YouTools,
|
||||
ydc_search,
|
||||
ydc_contents,
|
||||
ydc_news,
|
||||
ydc_images,
|
||||
)
|
||||
|
||||
__version__ = "0.1.0"
|
||||
__author__ = "Mervin Praison"
|
||||
|
||||
__all__ = [
|
||||
# Base classes
|
||||
"BaseTool",
|
||||
"ToolResult",
|
||||
"ToolValidationError",
|
||||
"tool",
|
||||
"FunctionTool",
|
||||
|
||||
# Tavily
|
||||
"TavilyTools",
|
||||
"tavily_search",
|
||||
"tavily_extract",
|
||||
"tavily_crawl",
|
||||
"tavily_map",
|
||||
|
||||
# Exa
|
||||
"ExaTools",
|
||||
"exa_search",
|
||||
"exa_search_contents",
|
||||
"exa_find_similar",
|
||||
"exa_answer",
|
||||
|
||||
# You.com
|
||||
"YouTools",
|
||||
"ydc_search",
|
||||
"ydc_contents",
|
||||
"ydc_news",
|
||||
"ydc_images",
|
||||
]
|
||||
@@ -0,0 +1,64 @@
|
||||
"""Tools package for PraisonAI Tools.
|
||||
|
||||
This module provides lazy loading of all available tools.
|
||||
"""
|
||||
|
||||
from praisonai_tools.tools.base import BaseTool, ToolResult, ToolValidationError
|
||||
from praisonai_tools.tools.decorator import tool, FunctionTool
|
||||
|
||||
# Tavily Tools
|
||||
from praisonai_tools.tools.tavily_tools import (
|
||||
TavilyTools,
|
||||
tavily_search,
|
||||
tavily_extract,
|
||||
tavily_crawl,
|
||||
tavily_map,
|
||||
)
|
||||
|
||||
# Exa Tools
|
||||
from praisonai_tools.tools.exa_tools import (
|
||||
ExaTools,
|
||||
exa_search,
|
||||
exa_search_contents,
|
||||
exa_find_similar,
|
||||
exa_answer,
|
||||
)
|
||||
|
||||
# You.com Tools
|
||||
from praisonai_tools.tools.youdotcom_tools import (
|
||||
YouTools,
|
||||
ydc_search,
|
||||
ydc_contents,
|
||||
ydc_news,
|
||||
ydc_images,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
# Base
|
||||
"BaseTool",
|
||||
"ToolResult",
|
||||
"ToolValidationError",
|
||||
"tool",
|
||||
"FunctionTool",
|
||||
|
||||
# Tavily
|
||||
"TavilyTools",
|
||||
"tavily_search",
|
||||
"tavily_extract",
|
||||
"tavily_crawl",
|
||||
"tavily_map",
|
||||
|
||||
# Exa
|
||||
"ExaTools",
|
||||
"exa_search",
|
||||
"exa_search_contents",
|
||||
"exa_find_similar",
|
||||
"exa_answer",
|
||||
|
||||
# You.com
|
||||
"YouTools",
|
||||
"ydc_search",
|
||||
"ydc_contents",
|
||||
"ydc_news",
|
||||
"ydc_images",
|
||||
]
|
||||
@@ -0,0 +1,263 @@
|
||||
"""Base classes for PraisonAI Tools.
|
||||
|
||||
This module provides the foundation for creating tools that can be used by agents.
|
||||
External developers can create plugins by subclassing BaseTool.
|
||||
|
||||
Usage:
|
||||
from praisonai_tools import BaseTool
|
||||
|
||||
class MyTool(BaseTool):
|
||||
name = "my_tool"
|
||||
description = "Does something useful"
|
||||
|
||||
def run(self, query: str) -> str:
|
||||
return f"Result for {query}"
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Any, Dict, Optional, Type, get_type_hints
|
||||
import inspect
|
||||
import logging
|
||||
|
||||
|
||||
class ToolValidationError(Exception):
|
||||
"""Raised when a tool fails validation."""
|
||||
pass
|
||||
|
||||
|
||||
class ToolResult:
|
||||
"""Wrapper for tool execution results."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
output: Any,
|
||||
success: bool = True,
|
||||
error: Optional[str] = None,
|
||||
metadata: Optional[Dict[str, Any]] = None
|
||||
):
|
||||
self.output = output
|
||||
self.success = success
|
||||
self.error = error
|
||||
self.metadata = metadata or {}
|
||||
|
||||
def __str__(self) -> str:
|
||||
if self.success:
|
||||
return str(self.output)
|
||||
return f"Error: {self.error}"
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"ToolResult(success={self.success}, output={self.output!r})"
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"output": self.output,
|
||||
"success": self.success,
|
||||
"error": self.error,
|
||||
"metadata": self.metadata
|
||||
}
|
||||
|
||||
|
||||
class BaseTool(ABC):
|
||||
"""Abstract base class for all PraisonAI tools.
|
||||
|
||||
Subclass this to create custom tools that can be:
|
||||
- Used directly by agents
|
||||
- Distributed as pip-installable plugins
|
||||
- Auto-discovered via entry_points
|
||||
|
||||
Attributes:
|
||||
name: Unique identifier for the tool
|
||||
description: Human-readable description (used by LLM)
|
||||
version: Tool version string (default: "1.0.0")
|
||||
parameters: JSON Schema for parameters (auto-generated if not provided)
|
||||
|
||||
Example:
|
||||
class WeatherTool(BaseTool):
|
||||
name = "get_weather"
|
||||
description = "Get current weather for a location"
|
||||
|
||||
def run(self, location: str, units: str = "celsius") -> dict:
|
||||
# Implementation here
|
||||
return {"temp": 22, "condition": "sunny"}
|
||||
"""
|
||||
|
||||
# Required class attributes (must be overridden)
|
||||
name: str = ""
|
||||
description: str = ""
|
||||
|
||||
# Optional class attributes
|
||||
version: str = "1.0.0"
|
||||
parameters: Optional[Dict[str, Any]] = None # JSON Schema, auto-generated if None
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the tool and validate configuration."""
|
||||
if not self.name:
|
||||
# Use class name as default
|
||||
self.name = self.__class__.__name__.lower().replace("tool", "")
|
||||
|
||||
if not self.description:
|
||||
# Use docstring as default
|
||||
self.description = self.__class__.__doc__ or f"Tool: {self.name}"
|
||||
|
||||
# Auto-generate parameters schema if not provided
|
||||
if self.parameters is None:
|
||||
self.parameters = self._generate_parameters_schema()
|
||||
|
||||
def _generate_parameters_schema(self) -> Dict[str, Any]:
|
||||
"""Generate JSON Schema from run() method signature."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
"required": []
|
||||
}
|
||||
|
||||
try:
|
||||
sig = inspect.signature(self.run)
|
||||
hints = get_type_hints(self.run) if hasattr(self.run, '__annotations__') else {}
|
||||
|
||||
for param_name, param in sig.parameters.items():
|
||||
if param_name == 'self':
|
||||
continue
|
||||
|
||||
# Get type hint
|
||||
param_type = hints.get(param_name, Any)
|
||||
json_type = self._python_type_to_json(param_type)
|
||||
|
||||
# Build property schema
|
||||
prop_schema = {"type": json_type}
|
||||
|
||||
schema["properties"][param_name] = prop_schema
|
||||
|
||||
# Check if required (no default value)
|
||||
if param.default is inspect.Parameter.empty:
|
||||
schema["required"].append(param_name)
|
||||
except Exception as e:
|
||||
logging.debug(f"Could not generate schema for {self.name}: {e}")
|
||||
|
||||
return schema
|
||||
|
||||
@staticmethod
|
||||
def _python_type_to_json(python_type: Type) -> str:
|
||||
"""Convert Python type to JSON Schema type."""
|
||||
type_map = {
|
||||
str: "string",
|
||||
int: "integer",
|
||||
float: "number",
|
||||
bool: "boolean",
|
||||
list: "array",
|
||||
dict: "object",
|
||||
type(None): "null"
|
||||
}
|
||||
|
||||
# Handle Optional, Union, etc.
|
||||
origin = getattr(python_type, '__origin__', None)
|
||||
if origin is not None:
|
||||
if origin is list:
|
||||
return "array"
|
||||
if origin is dict:
|
||||
return "object"
|
||||
|
||||
return type_map.get(python_type, "string")
|
||||
|
||||
@abstractmethod
|
||||
def run(self, **kwargs) -> Any:
|
||||
"""Execute the tool with given arguments.
|
||||
|
||||
This method must be implemented by subclasses.
|
||||
|
||||
Args:
|
||||
**kwargs: Tool-specific arguments
|
||||
|
||||
Returns:
|
||||
Tool output (any type, will be converted to string for LLM)
|
||||
"""
|
||||
pass
|
||||
|
||||
def __call__(self, **kwargs) -> Any:
|
||||
"""Allow tool to be called directly like a function."""
|
||||
return self.run(**kwargs)
|
||||
|
||||
def safe_run(self, **kwargs) -> ToolResult:
|
||||
"""Execute tool with error handling, returning ToolResult."""
|
||||
try:
|
||||
output = self.run(**kwargs)
|
||||
return ToolResult(output=output, success=True)
|
||||
except Exception as e:
|
||||
logging.error(f"Tool {self.name} failed: {e}")
|
||||
return ToolResult(
|
||||
output=None,
|
||||
success=False,
|
||||
error=str(e)
|
||||
)
|
||||
|
||||
def get_schema(self) -> Dict[str, Any]:
|
||||
"""Get OpenAI-compatible function schema for this tool."""
|
||||
return {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": self.name,
|
||||
"description": self.description,
|
||||
"parameters": self.parameters
|
||||
}
|
||||
}
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"{self.__class__.__name__}(name='{self.name}')"
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.__class__.__name__}(name='{self.name}', description='{self.description[:50]}...')"
|
||||
|
||||
def validate(self) -> bool:
|
||||
"""Validate the tool configuration.
|
||||
|
||||
Raises:
|
||||
ToolValidationError: If validation fails
|
||||
|
||||
Returns:
|
||||
True if validation passes
|
||||
"""
|
||||
errors = []
|
||||
|
||||
if not self.name or not isinstance(self.name, str):
|
||||
errors.append("Tool must have a non-empty string 'name'")
|
||||
|
||||
if not self.description or not isinstance(self.description, str):
|
||||
errors.append("Tool must have a non-empty string 'description'")
|
||||
|
||||
if getattr(self.run, '__isabstractmethod__', False):
|
||||
errors.append("Tool must implement the 'run()' method")
|
||||
|
||||
if self.parameters:
|
||||
if not isinstance(self.parameters, dict):
|
||||
errors.append("'parameters' must be a dictionary")
|
||||
elif "type" not in self.parameters:
|
||||
errors.append("'parameters' must have a 'type' field")
|
||||
|
||||
if errors:
|
||||
raise ToolValidationError(f"Tool '{self.name}' validation failed: {'; '.join(errors)}")
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def validate_tool(tool: Any) -> bool:
|
||||
"""Validate any tool-like object.
|
||||
|
||||
Args:
|
||||
tool: Object to validate (BaseTool, callable, etc.)
|
||||
|
||||
Returns:
|
||||
True if valid
|
||||
|
||||
Raises:
|
||||
ToolValidationError: If validation fails
|
||||
"""
|
||||
if isinstance(tool, BaseTool):
|
||||
return tool.validate()
|
||||
|
||||
if callable(tool):
|
||||
name = getattr(tool, '__name__', None) or getattr(tool, 'name', None)
|
||||
if not name:
|
||||
raise ToolValidationError("Callable tool must have a __name__ or name attribute")
|
||||
return True
|
||||
|
||||
raise ToolValidationError(f"Invalid tool type: {type(tool)}")
|
||||
@@ -0,0 +1,190 @@
|
||||
"""Tool decorator for converting functions into tools.
|
||||
|
||||
This module provides the @tool decorator for easily creating tools from functions.
|
||||
|
||||
Usage:
|
||||
from praisonai_tools import tool
|
||||
|
||||
@tool
|
||||
def search(query: str) -> list:
|
||||
'''Search the web for information.'''
|
||||
return [...]
|
||||
|
||||
# Or with explicit parameters:
|
||||
@tool(name="web_search", description="Search the internet")
|
||||
def search(query: str, max_results: int = 5) -> list:
|
||||
return [...]
|
||||
"""
|
||||
|
||||
import inspect
|
||||
import functools
|
||||
import logging
|
||||
from typing import Any, Callable, Dict, Optional, Union, get_type_hints
|
||||
|
||||
from praisonai_tools.tools.base import BaseTool
|
||||
|
||||
|
||||
class FunctionTool(BaseTool):
|
||||
"""A BaseTool wrapper for plain functions.
|
||||
|
||||
Created automatically by the @tool decorator.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
func: Callable,
|
||||
name: Optional[str] = None,
|
||||
description: Optional[str] = None,
|
||||
version: str = "1.0.0"
|
||||
):
|
||||
self._func = func
|
||||
self.name = name or func.__name__
|
||||
self.description = description or func.__doc__ or f"Tool: {self.name}"
|
||||
self.version = version
|
||||
|
||||
# Generate schema from the original function
|
||||
self.parameters = self._generate_schema_from_func(func)
|
||||
|
||||
# Copy function metadata
|
||||
functools.update_wrapper(self, func)
|
||||
|
||||
def _generate_schema_from_func(self, func: Callable) -> Dict[str, Any]:
|
||||
"""Generate JSON Schema from the wrapped function's signature."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
"required": []
|
||||
}
|
||||
|
||||
try:
|
||||
sig = inspect.signature(func)
|
||||
hints = get_type_hints(func) if hasattr(func, '__annotations__') else {}
|
||||
|
||||
for param_name, param in sig.parameters.items():
|
||||
if param_name in ('self', 'cls'):
|
||||
continue
|
||||
|
||||
param_type = hints.get(param_name, Any)
|
||||
json_type = BaseTool._python_type_to_json(param_type)
|
||||
|
||||
schema["properties"][param_name] = {"type": json_type}
|
||||
|
||||
if param.default is inspect.Parameter.empty:
|
||||
schema["required"].append(param_name)
|
||||
except Exception as e:
|
||||
logging.debug(f"Could not generate schema for {func.__name__}: {e}")
|
||||
|
||||
return schema
|
||||
|
||||
def run(self, **kwargs) -> Any:
|
||||
"""Execute the wrapped function."""
|
||||
return self._func(**kwargs)
|
||||
|
||||
def __call__(self, *args, **kwargs) -> Any:
|
||||
"""Allow calling with positional args like the original function."""
|
||||
return self._func(*args, **kwargs)
|
||||
|
||||
|
||||
def tool(
|
||||
func: Optional[Callable] = None,
|
||||
*,
|
||||
name: Optional[str] = None,
|
||||
description: Optional[str] = None,
|
||||
version: str = "1.0.0"
|
||||
) -> Union[FunctionTool, Callable[[Callable], FunctionTool]]:
|
||||
"""Decorator to convert a function into a tool.
|
||||
|
||||
Can be used with or without arguments:
|
||||
|
||||
@tool
|
||||
def my_func(x: str) -> str:
|
||||
'''Does something.'''
|
||||
return x
|
||||
|
||||
@tool(name="custom_name", description="Custom description")
|
||||
def my_func(x: str) -> str:
|
||||
return x
|
||||
|
||||
Args:
|
||||
func: The function to wrap (when used without parentheses)
|
||||
name: Override the tool name (default: function name)
|
||||
description: Override description (default: function docstring)
|
||||
version: Tool version (default: "1.0.0")
|
||||
|
||||
Returns:
|
||||
FunctionTool instance that wraps the function
|
||||
"""
|
||||
def decorator(fn: Callable) -> FunctionTool:
|
||||
return FunctionTool(
|
||||
func=fn,
|
||||
name=name,
|
||||
description=description,
|
||||
version=version
|
||||
)
|
||||
|
||||
if func is not None:
|
||||
return decorator(func)
|
||||
else:
|
||||
return decorator
|
||||
|
||||
|
||||
def is_tool(obj: Any) -> bool:
|
||||
"""Check if an object is a tool (BaseTool instance or decorated function)."""
|
||||
if isinstance(obj, BaseTool):
|
||||
return True
|
||||
if isinstance(obj, FunctionTool):
|
||||
return True
|
||||
if hasattr(obj, 'run') and hasattr(obj, 'name'):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def get_tool_schema(obj: Any) -> Optional[Dict[str, Any]]:
|
||||
"""Get OpenAI-compatible schema for any tool-like object."""
|
||||
if isinstance(obj, BaseTool):
|
||||
return obj.get_schema()
|
||||
|
||||
if callable(obj):
|
||||
return _schema_from_function(obj)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _schema_from_function(func: Callable) -> Dict[str, Any]:
|
||||
"""Generate OpenAI function schema from a plain function."""
|
||||
name = getattr(func, '__name__', 'unknown')
|
||||
description = func.__doc__ or f"Function: {name}"
|
||||
|
||||
properties = {}
|
||||
required = []
|
||||
|
||||
try:
|
||||
sig = inspect.signature(func)
|
||||
hints = get_type_hints(func) if hasattr(func, '__annotations__') else {}
|
||||
|
||||
for param_name, param in sig.parameters.items():
|
||||
if param_name == 'self':
|
||||
continue
|
||||
|
||||
param_type = hints.get(param_name, Any)
|
||||
json_type = BaseTool._python_type_to_json(param_type)
|
||||
|
||||
properties[param_name] = {"type": json_type}
|
||||
|
||||
if param.default is inspect.Parameter.empty:
|
||||
required.append(param_name)
|
||||
except Exception as e:
|
||||
logging.debug(f"Could not generate schema for {name}: {e}")
|
||||
|
||||
return {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": name,
|
||||
"description": description.strip(),
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": properties,
|
||||
"required": required
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,707 @@
|
||||
"""Exa search and research tools.
|
||||
|
||||
Exa provides AI-powered web search, content retrieval, and research capabilities.
|
||||
This module requires the EXA_API_KEY environment variable to be set.
|
||||
|
||||
Usage:
|
||||
from praisonai_tools import exa_search, exa_search_contents
|
||||
|
||||
# Search the web
|
||||
results = exa_search("AI startups")
|
||||
|
||||
# Search with contents
|
||||
results = exa_search_contents("AI in healthcare", text=True)
|
||||
|
||||
# Or use the class directly
|
||||
from praisonai_tools import ExaTools
|
||||
exa = ExaTools()
|
||||
results = exa.search("AI news")
|
||||
"""
|
||||
|
||||
from typing import List, Dict, Any, Optional, Union
|
||||
import logging
|
||||
import os
|
||||
from importlib import util
|
||||
|
||||
|
||||
def _check_exa_available() -> tuple:
|
||||
"""Check if Exa is available and API key is set.
|
||||
|
||||
Returns:
|
||||
Tuple of (is_available, error_message)
|
||||
"""
|
||||
if util.find_spec("exa_py") is None:
|
||||
return False, "exa_py package is not installed. Install it with: pip install exa_py"
|
||||
|
||||
if not os.environ.get("EXA_API_KEY"):
|
||||
return False, "EXA_API_KEY environment variable is not set. Please set it to use Exa tools."
|
||||
|
||||
return True, None
|
||||
|
||||
|
||||
class ExaTools:
|
||||
"""Exa search and research tools.
|
||||
|
||||
Provides methods for:
|
||||
- search: Basic web search
|
||||
- search_and_contents: Search with full text/highlights
|
||||
- find_similar: Find similar pages to a URL
|
||||
- find_similar_and_contents: Find similar with content
|
||||
- answer: Get AI-generated answers with citations
|
||||
|
||||
Example:
|
||||
from praisonai_tools import ExaTools
|
||||
|
||||
exa = ExaTools()
|
||||
results = exa.search("AI startups", num_results=5)
|
||||
"""
|
||||
|
||||
def __init__(self, api_key: Optional[str] = None):
|
||||
"""Initialize ExaTools.
|
||||
|
||||
Args:
|
||||
api_key: Exa API key. If not provided, uses EXA_API_KEY env var.
|
||||
"""
|
||||
self.api_key = api_key or os.environ.get("EXA_API_KEY")
|
||||
self._client = None
|
||||
self.logger = logging.getLogger(__name__)
|
||||
|
||||
def _get_client(self):
|
||||
"""Get or create Exa client."""
|
||||
if self._client is None:
|
||||
is_available, error = _check_exa_available()
|
||||
if not is_available:
|
||||
raise ImportError(error)
|
||||
|
||||
from exa_py import Exa
|
||||
self._client = Exa(self.api_key)
|
||||
return self._client
|
||||
|
||||
def search(
|
||||
self,
|
||||
query: str,
|
||||
num_results: int = 10,
|
||||
type: Optional[str] = None,
|
||||
category: Optional[str] = None,
|
||||
include_domains: Optional[List[str]] = None,
|
||||
exclude_domains: Optional[List[str]] = None,
|
||||
start_crawl_date: Optional[str] = None,
|
||||
end_crawl_date: Optional[str] = None,
|
||||
start_published_date: Optional[str] = None,
|
||||
end_published_date: Optional[str] = None,
|
||||
include_text: Optional[List[str]] = None,
|
||||
exclude_text: Optional[List[str]] = None,
|
||||
additional_queries: Optional[List[str]] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Perform an Exa search.
|
||||
|
||||
Args:
|
||||
query: The search query
|
||||
num_results: Number of results to return (default 10, max 100)
|
||||
type: Search type - "auto", "neural", "fast", or "deep"
|
||||
category: Data category - company, research paper, news, linkedin profile,
|
||||
github, tweet, movie, song, personal site, pdf, financial report
|
||||
include_domains: List of domains to include
|
||||
exclude_domains: List of domains to exclude
|
||||
start_crawl_date: Only include links crawled after this date
|
||||
end_crawl_date: Only include links crawled before this date
|
||||
start_published_date: Only include links published after this date
|
||||
end_published_date: Only include links published before this date
|
||||
include_text: Strings that must be present in results
|
||||
exclude_text: Strings that must not be present in results
|
||||
additional_queries: Additional query variations for deep search
|
||||
|
||||
Returns:
|
||||
Dict with 'results' list and optional 'autopromptString'
|
||||
"""
|
||||
try:
|
||||
client = self._get_client()
|
||||
|
||||
kwargs = {
|
||||
"query": query,
|
||||
"num_results": num_results
|
||||
}
|
||||
|
||||
if type:
|
||||
kwargs["type"] = type
|
||||
if category:
|
||||
kwargs["category"] = category
|
||||
if include_domains:
|
||||
kwargs["include_domains"] = include_domains
|
||||
if exclude_domains:
|
||||
kwargs["exclude_domains"] = exclude_domains
|
||||
if start_crawl_date:
|
||||
kwargs["start_crawl_date"] = start_crawl_date
|
||||
if end_crawl_date:
|
||||
kwargs["end_crawl_date"] = end_crawl_date
|
||||
if start_published_date:
|
||||
kwargs["start_published_date"] = start_published_date
|
||||
if end_published_date:
|
||||
kwargs["end_published_date"] = end_published_date
|
||||
if include_text:
|
||||
kwargs["include_text"] = include_text
|
||||
if exclude_text:
|
||||
kwargs["exclude_text"] = exclude_text
|
||||
if additional_queries:
|
||||
kwargs["additional_queries"] = additional_queries
|
||||
|
||||
response = client.search(**kwargs)
|
||||
|
||||
return {
|
||||
"results": [
|
||||
{
|
||||
"url": r.url,
|
||||
"id": r.id,
|
||||
"title": getattr(r, "title", None),
|
||||
"published_date": getattr(r, "published_date", None),
|
||||
"author": getattr(r, "author", None)
|
||||
}
|
||||
for r in response.results
|
||||
],
|
||||
"autopromptString": getattr(response, "autoprompt_string", None),
|
||||
"requestId": getattr(response, "request_id", None)
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Exa search error: {e}")
|
||||
return {"error": f"Exa search error: {str(e)}"}
|
||||
|
||||
def search_and_contents(
|
||||
self,
|
||||
query: str,
|
||||
text: Union[bool, Dict] = True,
|
||||
highlights: Union[bool, Dict] = False,
|
||||
summary: Optional[Dict] = None,
|
||||
num_results: int = 10,
|
||||
type: Optional[str] = None,
|
||||
category: Optional[str] = None,
|
||||
include_domains: Optional[List[str]] = None,
|
||||
exclude_domains: Optional[List[str]] = None,
|
||||
start_crawl_date: Optional[str] = None,
|
||||
end_crawl_date: Optional[str] = None,
|
||||
start_published_date: Optional[str] = None,
|
||||
end_published_date: Optional[str] = None,
|
||||
include_text: Optional[List[str]] = None,
|
||||
exclude_text: Optional[List[str]] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Search and retrieve content from results.
|
||||
|
||||
Args:
|
||||
query: The search query
|
||||
text: Include full text content (True or TextContentsOptions dict)
|
||||
highlights: Include highlights (True or HighlightsContentsOptions dict)
|
||||
summary: Summary options with optional schema for structured output
|
||||
num_results: Number of results to return
|
||||
type: Search type - "auto", "neural", "fast", or "deep"
|
||||
category: Data category filter
|
||||
include_domains: List of domains to include
|
||||
exclude_domains: List of domains to exclude
|
||||
start_crawl_date: Only include links crawled after this date
|
||||
end_crawl_date: Only include links crawled before this date
|
||||
start_published_date: Only include links published after this date
|
||||
end_published_date: Only include links published before this date
|
||||
include_text: Strings that must be present in results
|
||||
exclude_text: Strings that must not be present in results
|
||||
|
||||
Returns:
|
||||
Dict with 'results' list containing text/highlights
|
||||
"""
|
||||
try:
|
||||
client = self._get_client()
|
||||
|
||||
kwargs = {
|
||||
"query": query,
|
||||
"num_results": num_results
|
||||
}
|
||||
|
||||
if text:
|
||||
kwargs["text"] = text
|
||||
if highlights:
|
||||
kwargs["highlights"] = highlights
|
||||
if summary:
|
||||
kwargs["summary"] = summary
|
||||
if type:
|
||||
kwargs["type"] = type
|
||||
if category:
|
||||
kwargs["category"] = category
|
||||
if include_domains:
|
||||
kwargs["include_domains"] = include_domains
|
||||
if exclude_domains:
|
||||
kwargs["exclude_domains"] = exclude_domains
|
||||
if start_crawl_date:
|
||||
kwargs["start_crawl_date"] = start_crawl_date
|
||||
if end_crawl_date:
|
||||
kwargs["end_crawl_date"] = end_crawl_date
|
||||
if start_published_date:
|
||||
kwargs["start_published_date"] = start_published_date
|
||||
if end_published_date:
|
||||
kwargs["end_published_date"] = end_published_date
|
||||
if include_text:
|
||||
kwargs["include_text"] = include_text
|
||||
if exclude_text:
|
||||
kwargs["exclude_text"] = exclude_text
|
||||
|
||||
response = client.search_and_contents(**kwargs)
|
||||
|
||||
results = []
|
||||
for r in response.results:
|
||||
result = {
|
||||
"url": r.url,
|
||||
"id": r.id,
|
||||
"title": getattr(r, "title", None),
|
||||
"published_date": getattr(r, "published_date", None),
|
||||
"author": getattr(r, "author", None)
|
||||
}
|
||||
if hasattr(r, "text") and r.text:
|
||||
result["text"] = r.text
|
||||
if hasattr(r, "highlights") and r.highlights:
|
||||
result["highlights"] = r.highlights
|
||||
if hasattr(r, "highlight_scores") and r.highlight_scores:
|
||||
result["highlight_scores"] = r.highlight_scores
|
||||
if hasattr(r, "summary") and r.summary:
|
||||
result["summary"] = r.summary
|
||||
results.append(result)
|
||||
|
||||
return {
|
||||
"results": results,
|
||||
"requestId": getattr(response, "request_id", None)
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Exa search_and_contents error: {e}")
|
||||
return {"error": f"Exa search_and_contents error: {str(e)}"}
|
||||
|
||||
def find_similar(
|
||||
self,
|
||||
url: str,
|
||||
num_results: int = 10,
|
||||
include_domains: Optional[List[str]] = None,
|
||||
exclude_domains: Optional[List[str]] = None,
|
||||
start_crawl_date: Optional[str] = None,
|
||||
end_crawl_date: Optional[str] = None,
|
||||
start_published_date: Optional[str] = None,
|
||||
end_published_date: Optional[str] = None,
|
||||
exclude_source_domain: bool = False,
|
||||
category: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Find similar pages to a given URL.
|
||||
|
||||
Args:
|
||||
url: The URL to find similar pages for
|
||||
num_results: Number of results to return
|
||||
include_domains: List of domains to include
|
||||
exclude_domains: List of domains to exclude
|
||||
start_crawl_date: Only include links crawled after this date
|
||||
end_crawl_date: Only include links crawled before this date
|
||||
start_published_date: Only include links published after this date
|
||||
end_published_date: Only include links published before this date
|
||||
exclude_source_domain: Exclude results from the same domain
|
||||
category: Data category filter
|
||||
|
||||
Returns:
|
||||
Dict with 'results' list of similar pages
|
||||
"""
|
||||
try:
|
||||
client = self._get_client()
|
||||
|
||||
kwargs = {
|
||||
"url": url,
|
||||
"num_results": num_results
|
||||
}
|
||||
|
||||
if include_domains:
|
||||
kwargs["include_domains"] = include_domains
|
||||
if exclude_domains:
|
||||
kwargs["exclude_domains"] = exclude_domains
|
||||
if start_crawl_date:
|
||||
kwargs["start_crawl_date"] = start_crawl_date
|
||||
if end_crawl_date:
|
||||
kwargs["end_crawl_date"] = end_crawl_date
|
||||
if start_published_date:
|
||||
kwargs["start_published_date"] = start_published_date
|
||||
if end_published_date:
|
||||
kwargs["end_published_date"] = end_published_date
|
||||
if exclude_source_domain:
|
||||
kwargs["exclude_source_domain"] = exclude_source_domain
|
||||
if category:
|
||||
kwargs["category"] = category
|
||||
|
||||
response = client.find_similar(**kwargs)
|
||||
|
||||
return {
|
||||
"results": [
|
||||
{
|
||||
"url": r.url,
|
||||
"id": r.id,
|
||||
"title": getattr(r, "title", None),
|
||||
"published_date": getattr(r, "published_date", None),
|
||||
"author": getattr(r, "author", None)
|
||||
}
|
||||
for r in response.results
|
||||
],
|
||||
"requestId": getattr(response, "request_id", None)
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Exa find_similar error: {e}")
|
||||
return {"error": f"Exa find_similar error: {str(e)}"}
|
||||
|
||||
def find_similar_and_contents(
|
||||
self,
|
||||
url: str,
|
||||
text: Union[bool, Dict] = True,
|
||||
highlights: Union[bool, Dict] = False,
|
||||
num_results: int = 10,
|
||||
include_domains: Optional[List[str]] = None,
|
||||
exclude_domains: Optional[List[str]] = None,
|
||||
start_crawl_date: Optional[str] = None,
|
||||
end_crawl_date: Optional[str] = None,
|
||||
start_published_date: Optional[str] = None,
|
||||
end_published_date: Optional[str] = None,
|
||||
exclude_source_domain: bool = False,
|
||||
category: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Find similar pages with content.
|
||||
|
||||
Args:
|
||||
url: The URL to find similar pages for
|
||||
text: Include full text content
|
||||
highlights: Include highlights
|
||||
num_results: Number of results to return
|
||||
include_domains: List of domains to include
|
||||
exclude_domains: List of domains to exclude
|
||||
start_crawl_date: Only include links crawled after this date
|
||||
end_crawl_date: Only include links crawled before this date
|
||||
start_published_date: Only include links published after this date
|
||||
end_published_date: Only include links published before this date
|
||||
exclude_source_domain: Exclude results from the same domain
|
||||
category: Data category filter
|
||||
|
||||
Returns:
|
||||
Dict with 'results' list containing text/highlights
|
||||
"""
|
||||
try:
|
||||
client = self._get_client()
|
||||
|
||||
kwargs = {
|
||||
"url": url,
|
||||
"num_results": num_results
|
||||
}
|
||||
|
||||
if text:
|
||||
kwargs["text"] = text
|
||||
if highlights:
|
||||
kwargs["highlights"] = highlights
|
||||
if include_domains:
|
||||
kwargs["include_domains"] = include_domains
|
||||
if exclude_domains:
|
||||
kwargs["exclude_domains"] = exclude_domains
|
||||
if start_crawl_date:
|
||||
kwargs["start_crawl_date"] = start_crawl_date
|
||||
if end_crawl_date:
|
||||
kwargs["end_crawl_date"] = end_crawl_date
|
||||
if start_published_date:
|
||||
kwargs["start_published_date"] = start_published_date
|
||||
if end_published_date:
|
||||
kwargs["end_published_date"] = end_published_date
|
||||
if exclude_source_domain:
|
||||
kwargs["exclude_source_domain"] = exclude_source_domain
|
||||
if category:
|
||||
kwargs["category"] = category
|
||||
|
||||
response = client.find_similar_and_contents(**kwargs)
|
||||
|
||||
results = []
|
||||
for r in response.results:
|
||||
result = {
|
||||
"url": r.url,
|
||||
"id": r.id,
|
||||
"title": getattr(r, "title", None),
|
||||
"published_date": getattr(r, "published_date", None),
|
||||
"author": getattr(r, "author", None)
|
||||
}
|
||||
if hasattr(r, "text") and r.text:
|
||||
result["text"] = r.text
|
||||
if hasattr(r, "highlights") and r.highlights:
|
||||
result["highlights"] = r.highlights
|
||||
if hasattr(r, "highlight_scores") and r.highlight_scores:
|
||||
result["highlight_scores"] = r.highlight_scores
|
||||
results.append(result)
|
||||
|
||||
return {
|
||||
"results": results,
|
||||
"requestId": getattr(response, "request_id", None)
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Exa find_similar_and_contents error: {e}")
|
||||
return {"error": f"Exa find_similar_and_contents error: {str(e)}"}
|
||||
|
||||
def answer(
|
||||
self,
|
||||
query: str,
|
||||
text: bool = False
|
||||
) -> Dict[str, Any]:
|
||||
"""Generate an answer using Exa's search and LLM capabilities.
|
||||
|
||||
Args:
|
||||
query: The question to answer
|
||||
text: If True, include full text of citations
|
||||
|
||||
Returns:
|
||||
Dict with 'answer' and 'citations'
|
||||
"""
|
||||
try:
|
||||
client = self._get_client()
|
||||
|
||||
response = client.answer(query=query, text=text)
|
||||
|
||||
citations = []
|
||||
for c in response.citations:
|
||||
citation = {
|
||||
"id": c.id,
|
||||
"url": c.url,
|
||||
"title": getattr(c, "title", None),
|
||||
"published_date": getattr(c, "published_date", None),
|
||||
"author": getattr(c, "author", None)
|
||||
}
|
||||
if text and hasattr(c, "text") and c.text:
|
||||
citation["text"] = c.text
|
||||
citations.append(citation)
|
||||
|
||||
return {
|
||||
"answer": response.answer,
|
||||
"citations": citations
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Exa answer error: {e}")
|
||||
return {"error": f"Exa answer error: {str(e)}"}
|
||||
|
||||
|
||||
# Standalone functions for easy access
|
||||
|
||||
def exa_search(
|
||||
query: str,
|
||||
num_results: int = 10,
|
||||
type: Optional[str] = None,
|
||||
category: Optional[str] = None,
|
||||
include_domains: Optional[List[str]] = None,
|
||||
exclude_domains: Optional[List[str]] = None,
|
||||
start_published_date: Optional[str] = None,
|
||||
end_published_date: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Search the web using Exa.
|
||||
|
||||
Args:
|
||||
query: The search query
|
||||
num_results: Number of results (default 10)
|
||||
type: Search type - "auto", "neural", "fast", or "deep"
|
||||
category: Data category filter
|
||||
include_domains: Domains to include
|
||||
exclude_domains: Domains to exclude
|
||||
start_published_date: Only include links published after this date
|
||||
end_published_date: Only include links published before this date
|
||||
|
||||
Returns:
|
||||
Dict with search results or error
|
||||
|
||||
Example:
|
||||
results = exa_search("AI startups", num_results=5)
|
||||
"""
|
||||
is_available, error = _check_exa_available()
|
||||
if not is_available:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
tools = ExaTools()
|
||||
return tools.search(
|
||||
query=query,
|
||||
num_results=num_results,
|
||||
type=type,
|
||||
category=category,
|
||||
include_domains=include_domains,
|
||||
exclude_domains=exclude_domains,
|
||||
start_published_date=start_published_date,
|
||||
end_published_date=end_published_date
|
||||
)
|
||||
|
||||
|
||||
def exa_search_contents(
|
||||
query: str,
|
||||
text: bool = True,
|
||||
highlights: bool = False,
|
||||
num_results: int = 10,
|
||||
type: Optional[str] = None,
|
||||
category: Optional[str] = None,
|
||||
include_domains: Optional[List[str]] = None,
|
||||
exclude_domains: Optional[List[str]] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Search and get content from results.
|
||||
|
||||
Args:
|
||||
query: The search query
|
||||
text: Include full text content
|
||||
highlights: Include highlights
|
||||
num_results: Number of results
|
||||
type: Search type
|
||||
category: Data category filter
|
||||
include_domains: Domains to include
|
||||
exclude_domains: Domains to exclude
|
||||
|
||||
Returns:
|
||||
Dict with results containing text/highlights
|
||||
|
||||
Example:
|
||||
results = exa_search_contents("AI in healthcare", text=True, highlights=True)
|
||||
"""
|
||||
is_available, error = _check_exa_available()
|
||||
if not is_available:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
tools = ExaTools()
|
||||
return tools.search_and_contents(
|
||||
query=query,
|
||||
text=text,
|
||||
highlights=highlights,
|
||||
num_results=num_results,
|
||||
type=type,
|
||||
category=category,
|
||||
include_domains=include_domains,
|
||||
exclude_domains=exclude_domains
|
||||
)
|
||||
|
||||
|
||||
def exa_find_similar(
|
||||
url: str,
|
||||
num_results: int = 10,
|
||||
exclude_source_domain: bool = True,
|
||||
category: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Find pages similar to a given URL.
|
||||
|
||||
Args:
|
||||
url: The URL to find similar pages for
|
||||
num_results: Number of results
|
||||
exclude_source_domain: Exclude same domain results
|
||||
category: Data category filter
|
||||
|
||||
Returns:
|
||||
Dict with similar results
|
||||
|
||||
Example:
|
||||
results = exa_find_similar("https://openai.com", num_results=5)
|
||||
"""
|
||||
is_available, error = _check_exa_available()
|
||||
if not is_available:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
tools = ExaTools()
|
||||
return tools.find_similar(
|
||||
url=url,
|
||||
num_results=num_results,
|
||||
exclude_source_domain=exclude_source_domain,
|
||||
category=category
|
||||
)
|
||||
|
||||
|
||||
def exa_answer(
|
||||
query: str,
|
||||
text: bool = False
|
||||
) -> Dict[str, Any]:
|
||||
"""Get an AI-generated answer with citations.
|
||||
|
||||
Args:
|
||||
query: The question to answer
|
||||
text: Include full text of citations
|
||||
|
||||
Returns:
|
||||
Dict with 'answer' and 'citations'
|
||||
|
||||
Example:
|
||||
result = exa_answer("What is the capital of France?")
|
||||
print(result["answer"])
|
||||
"""
|
||||
is_available, error = _check_exa_available()
|
||||
if not is_available:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
tools = ExaTools()
|
||||
return tools.answer(query=query, text=text)
|
||||
|
||||
|
||||
# Async versions
|
||||
|
||||
async def exa_search_async(
|
||||
query: str,
|
||||
num_results: int = 10,
|
||||
type: Optional[str] = None,
|
||||
category: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Async version of exa_search.
|
||||
|
||||
Args:
|
||||
query: The search query
|
||||
num_results: Number of results
|
||||
type: Search type
|
||||
category: Data category filter
|
||||
|
||||
Returns:
|
||||
Dict with search results
|
||||
"""
|
||||
import asyncio
|
||||
return await asyncio.get_event_loop().run_in_executor(
|
||||
None,
|
||||
lambda: exa_search(query, num_results, type, category)
|
||||
)
|
||||
|
||||
|
||||
async def exa_search_contents_async(
|
||||
query: str,
|
||||
text: bool = True,
|
||||
highlights: bool = False,
|
||||
num_results: int = 10
|
||||
) -> Dict[str, Any]:
|
||||
"""Async version of exa_search_contents.
|
||||
|
||||
Args:
|
||||
query: The search query
|
||||
text: Include full text
|
||||
highlights: Include highlights
|
||||
num_results: Number of results
|
||||
|
||||
Returns:
|
||||
Dict with results containing content
|
||||
"""
|
||||
import asyncio
|
||||
return await asyncio.get_event_loop().run_in_executor(
|
||||
None,
|
||||
lambda: exa_search_contents(query, text, highlights, num_results)
|
||||
)
|
||||
|
||||
|
||||
async def exa_answer_async(
|
||||
query: str,
|
||||
text: bool = False
|
||||
) -> Dict[str, Any]:
|
||||
"""Async version of exa_answer.
|
||||
|
||||
Args:
|
||||
query: The question to answer
|
||||
text: Include citation text
|
||||
|
||||
Returns:
|
||||
Dict with answer and citations
|
||||
"""
|
||||
import asyncio
|
||||
return await asyncio.get_event_loop().run_in_executor(
|
||||
None,
|
||||
lambda: exa_answer(query, text)
|
||||
)
|
||||
|
||||
|
||||
# Alias for simple usage
|
||||
exa = exa_search
|
||||
@@ -0,0 +1,555 @@
|
||||
"""Tavily search and extraction tools.
|
||||
|
||||
Tavily provides AI-powered web search, content extraction, and crawling capabilities.
|
||||
This module requires the TAVILY_API_KEY environment variable to be set.
|
||||
|
||||
Usage:
|
||||
from praisonai_tools import tavily_search, tavily_extract
|
||||
|
||||
# Search the web
|
||||
results = tavily_search("AI news 2024")
|
||||
|
||||
# Extract content from URLs
|
||||
content = tavily_extract("https://example.com")
|
||||
|
||||
# Or use the class directly
|
||||
from praisonai_tools import TavilyTools
|
||||
tavily = TavilyTools()
|
||||
results = tavily.search("AI news")
|
||||
"""
|
||||
|
||||
from typing import List, Dict, Any, Optional, Union
|
||||
import logging
|
||||
import os
|
||||
from importlib import util
|
||||
|
||||
|
||||
def _check_tavily_available() -> tuple:
|
||||
"""Check if Tavily is available and API key is set.
|
||||
|
||||
Returns:
|
||||
Tuple of (is_available, error_message)
|
||||
"""
|
||||
if util.find_spec("tavily") is None:
|
||||
return False, "Tavily package is not installed. Please install it using: pip install tavily-python"
|
||||
|
||||
api_key = os.environ.get("TAVILY_API_KEY")
|
||||
if not api_key:
|
||||
return False, "TAVILY_API_KEY environment variable is not set. Please set it to use Tavily tools."
|
||||
|
||||
return True, None
|
||||
|
||||
|
||||
class TavilyTools:
|
||||
"""Comprehensive tools for web search and content extraction using Tavily API.
|
||||
|
||||
Tavily provides AI-powered search capabilities optimized for LLM applications.
|
||||
|
||||
Features:
|
||||
- Web search with advanced filtering and ranking
|
||||
- Content extraction from URLs
|
||||
- Website crawling with intelligent navigation
|
||||
- Site mapping
|
||||
|
||||
Requires:
|
||||
- tavily-python package: pip install tavily-python
|
||||
- TAVILY_API_KEY environment variable
|
||||
"""
|
||||
|
||||
def __init__(self, api_key: Optional[str] = None):
|
||||
"""Initialize TavilyTools.
|
||||
|
||||
Args:
|
||||
api_key: Optional API key. If not provided, uses TAVILY_API_KEY env var.
|
||||
"""
|
||||
self._api_key = api_key
|
||||
self._client = None
|
||||
self._async_client = None
|
||||
|
||||
def _get_api_key(self) -> Optional[str]:
|
||||
"""Get API key from instance or environment."""
|
||||
return self._api_key or os.environ.get("TAVILY_API_KEY")
|
||||
|
||||
def _get_client(self):
|
||||
"""Get or create synchronous Tavily client."""
|
||||
if self._client is None:
|
||||
is_available, error = _check_tavily_available()
|
||||
if not is_available:
|
||||
return None, error
|
||||
|
||||
from tavily import TavilyClient
|
||||
api_key = self._get_api_key()
|
||||
if not api_key:
|
||||
return None, "TAVILY_API_KEY not set"
|
||||
self._client = TavilyClient(api_key=api_key)
|
||||
return self._client, None
|
||||
|
||||
def _get_async_client(self):
|
||||
"""Get or create asynchronous Tavily client."""
|
||||
if self._async_client is None:
|
||||
is_available, error = _check_tavily_available()
|
||||
if not is_available:
|
||||
return None, error
|
||||
|
||||
from tavily import AsyncTavilyClient
|
||||
api_key = self._get_api_key()
|
||||
if not api_key:
|
||||
return None, "TAVILY_API_KEY not set"
|
||||
self._async_client = AsyncTavilyClient(api_key=api_key)
|
||||
return self._async_client, None
|
||||
|
||||
def search(
|
||||
self,
|
||||
query: str,
|
||||
search_depth: str = "basic",
|
||||
topic: str = "general",
|
||||
max_results: int = 5,
|
||||
include_answer: bool = False,
|
||||
include_raw_content: bool = False,
|
||||
include_images: bool = False,
|
||||
include_domains: Optional[List[str]] = None,
|
||||
exclude_domains: Optional[List[str]] = None,
|
||||
time_range: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Search the web using Tavily's AI-powered search.
|
||||
|
||||
Args:
|
||||
query: The search query
|
||||
search_depth: "basic" or "advanced" (advanced uses more credits but better results)
|
||||
topic: "general", "news", or "finance"
|
||||
max_results: Maximum number of results (1-20)
|
||||
include_answer: Include LLM-generated answer based on results
|
||||
include_raw_content: Include cleaned HTML content of each result
|
||||
include_images: Include query-related images
|
||||
include_domains: List of domains to include (max 300)
|
||||
exclude_domains: List of domains to exclude (max 150)
|
||||
time_range: Filter by time - "day", "week", "month", "year" or "d", "w", "m", "y"
|
||||
|
||||
Returns:
|
||||
Dict containing search results with keys:
|
||||
- results: List of search results with title, url, content, score
|
||||
- query: The search query
|
||||
- response_time: Time taken for search
|
||||
- answer: (optional) LLM-generated answer if include_answer=True
|
||||
- images: (optional) List of image URLs if include_images=True
|
||||
"""
|
||||
client, error = self._get_client()
|
||||
if error:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
try:
|
||||
kwargs = {
|
||||
"query": query,
|
||||
"search_depth": search_depth,
|
||||
"topic": topic,
|
||||
"max_results": max_results,
|
||||
"include_answer": include_answer,
|
||||
"include_raw_content": include_raw_content,
|
||||
"include_images": include_images,
|
||||
}
|
||||
|
||||
if include_domains:
|
||||
kwargs["include_domains"] = include_domains
|
||||
if exclude_domains:
|
||||
kwargs["exclude_domains"] = exclude_domains
|
||||
if time_range:
|
||||
kwargs["time_range"] = time_range
|
||||
|
||||
response = client.search(**kwargs)
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Tavily search error: {str(e)}"
|
||||
logging.error(error_msg)
|
||||
return {"error": error_msg}
|
||||
|
||||
def extract(
|
||||
self,
|
||||
urls: Union[str, List[str]],
|
||||
include_images: bool = False,
|
||||
extract_depth: str = "basic",
|
||||
) -> Dict[str, Any]:
|
||||
"""Extract content from one or more URLs.
|
||||
|
||||
Args:
|
||||
urls: Single URL or list of URLs (max 20)
|
||||
include_images: Include extracted images
|
||||
extract_depth: "basic" or "advanced" (advanced retrieves more data)
|
||||
|
||||
Returns:
|
||||
Dict containing:
|
||||
- results: List of successful extractions with url, raw_content
|
||||
- failed_results: List of URLs that failed with error messages
|
||||
- response_time: Time taken for extraction
|
||||
"""
|
||||
client, error = self._get_client()
|
||||
if error:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
try:
|
||||
response = client.extract(
|
||||
urls=urls,
|
||||
include_images=include_images,
|
||||
extract_depth=extract_depth,
|
||||
)
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Tavily extract error: {str(e)}"
|
||||
logging.error(error_msg)
|
||||
return {"error": error_msg}
|
||||
|
||||
def crawl(
|
||||
self,
|
||||
url: str,
|
||||
max_depth: int = 1,
|
||||
max_breadth: int = 20,
|
||||
limit: int = 50,
|
||||
instructions: Optional[str] = None,
|
||||
include_images: bool = False,
|
||||
extract_depth: str = "basic",
|
||||
) -> Dict[str, Any]:
|
||||
"""Crawl a website starting from a base URL.
|
||||
|
||||
Args:
|
||||
url: The root URL to begin crawling
|
||||
max_depth: Maximum depth from base URL (default: 1)
|
||||
max_breadth: Max links to follow per page (default: 20)
|
||||
limit: Total links to process before stopping (default: 50)
|
||||
instructions: Natural language instructions for the crawler
|
||||
include_images: Extract image URLs from pages
|
||||
extract_depth: "basic" or "advanced"
|
||||
|
||||
Returns:
|
||||
Dict containing:
|
||||
- base_url: The starting URL
|
||||
- results: List of crawled pages with url, raw_content, images
|
||||
- response_time: Time taken for crawl
|
||||
"""
|
||||
client, error = self._get_client()
|
||||
if error:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
try:
|
||||
kwargs = {
|
||||
"url": url,
|
||||
"max_depth": max_depth,
|
||||
"max_breadth": max_breadth,
|
||||
"limit": limit,
|
||||
"include_images": include_images,
|
||||
"extract_depth": extract_depth,
|
||||
}
|
||||
|
||||
if instructions:
|
||||
kwargs["instructions"] = instructions
|
||||
|
||||
response = client.crawl(**kwargs)
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Tavily crawl error: {str(e)}"
|
||||
logging.error(error_msg)
|
||||
return {"error": error_msg}
|
||||
|
||||
def map(
|
||||
self,
|
||||
url: str,
|
||||
max_depth: int = 1,
|
||||
max_breadth: int = 20,
|
||||
limit: int = 50,
|
||||
instructions: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Get a sitemap starting from a base URL.
|
||||
|
||||
Args:
|
||||
url: The root URL to begin mapping
|
||||
max_depth: Maximum depth from base URL (default: 1)
|
||||
max_breadth: Max links to follow per page (default: 20)
|
||||
limit: Total links to process before stopping (default: 50)
|
||||
instructions: Natural language instructions for the mapper
|
||||
|
||||
Returns:
|
||||
Dict containing:
|
||||
- base_url: The starting URL
|
||||
- results: List of discovered URLs
|
||||
- response_time: Time taken for mapping
|
||||
"""
|
||||
client, error = self._get_client()
|
||||
if error:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
try:
|
||||
kwargs = {
|
||||
"url": url,
|
||||
"max_depth": max_depth,
|
||||
"max_breadth": max_breadth,
|
||||
"limit": limit,
|
||||
}
|
||||
|
||||
if instructions:
|
||||
kwargs["instructions"] = instructions
|
||||
|
||||
response = client.map(**kwargs)
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Tavily map error: {str(e)}"
|
||||
logging.error(error_msg)
|
||||
return {"error": error_msg}
|
||||
|
||||
async def asearch(
|
||||
self,
|
||||
query: str,
|
||||
search_depth: str = "basic",
|
||||
topic: str = "general",
|
||||
max_results: int = 5,
|
||||
include_answer: bool = False,
|
||||
include_raw_content: bool = False,
|
||||
include_images: bool = False,
|
||||
include_domains: Optional[List[str]] = None,
|
||||
exclude_domains: Optional[List[str]] = None,
|
||||
time_range: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Async version of search. See search() for documentation."""
|
||||
client, error = self._get_async_client()
|
||||
if error:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
try:
|
||||
kwargs = {
|
||||
"query": query,
|
||||
"search_depth": search_depth,
|
||||
"topic": topic,
|
||||
"max_results": max_results,
|
||||
"include_answer": include_answer,
|
||||
"include_raw_content": include_raw_content,
|
||||
"include_images": include_images,
|
||||
}
|
||||
|
||||
if include_domains:
|
||||
kwargs["include_domains"] = include_domains
|
||||
if exclude_domains:
|
||||
kwargs["exclude_domains"] = exclude_domains
|
||||
if time_range:
|
||||
kwargs["time_range"] = time_range
|
||||
|
||||
response = await client.search(**kwargs)
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Tavily async search error: {str(e)}"
|
||||
logging.error(error_msg)
|
||||
return {"error": error_msg}
|
||||
|
||||
async def aextract(
|
||||
self,
|
||||
urls: Union[str, List[str]],
|
||||
include_images: bool = False,
|
||||
extract_depth: str = "basic",
|
||||
) -> Dict[str, Any]:
|
||||
"""Async version of extract. See extract() for documentation."""
|
||||
client, error = self._get_async_client()
|
||||
if error:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
try:
|
||||
response = await client.extract(
|
||||
urls=urls,
|
||||
include_images=include_images,
|
||||
extract_depth=extract_depth,
|
||||
)
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Tavily async extract error: {str(e)}"
|
||||
logging.error(error_msg)
|
||||
return {"error": error_msg}
|
||||
|
||||
|
||||
# Standalone functions for direct import
|
||||
|
||||
def tavily_search(
|
||||
query: str,
|
||||
search_depth: str = "basic",
|
||||
topic: str = "general",
|
||||
max_results: int = 5,
|
||||
include_answer: bool = False,
|
||||
include_raw_content: bool = False,
|
||||
include_images: bool = False,
|
||||
include_domains: Optional[List[str]] = None,
|
||||
exclude_domains: Optional[List[str]] = None,
|
||||
time_range: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Search the web using Tavily's AI-powered search.
|
||||
|
||||
This is a convenience function that creates a TavilyTools instance.
|
||||
For repeated calls, consider using TavilyTools class directly.
|
||||
|
||||
Args:
|
||||
query: The search query
|
||||
search_depth: "basic" or "advanced"
|
||||
topic: "general", "news", or "finance"
|
||||
max_results: Maximum number of results (1-20)
|
||||
include_answer: Include LLM-generated answer
|
||||
include_raw_content: Include cleaned HTML content
|
||||
include_images: Include query-related images
|
||||
include_domains: List of domains to include
|
||||
exclude_domains: List of domains to exclude
|
||||
time_range: Filter by time - "day", "week", "month", "year"
|
||||
|
||||
Returns:
|
||||
Dict containing search results
|
||||
"""
|
||||
tools = TavilyTools()
|
||||
return tools.search(
|
||||
query=query,
|
||||
search_depth=search_depth,
|
||||
topic=topic,
|
||||
max_results=max_results,
|
||||
include_answer=include_answer,
|
||||
include_raw_content=include_raw_content,
|
||||
include_images=include_images,
|
||||
include_domains=include_domains,
|
||||
exclude_domains=exclude_domains,
|
||||
time_range=time_range,
|
||||
)
|
||||
|
||||
|
||||
def tavily_extract(
|
||||
urls: Union[str, List[str]],
|
||||
include_images: bool = False,
|
||||
extract_depth: str = "basic",
|
||||
) -> Dict[str, Any]:
|
||||
"""Extract content from one or more URLs using Tavily.
|
||||
|
||||
Args:
|
||||
urls: Single URL or list of URLs (max 20)
|
||||
include_images: Include extracted images
|
||||
extract_depth: "basic" or "advanced"
|
||||
|
||||
Returns:
|
||||
Dict containing extraction results
|
||||
"""
|
||||
tools = TavilyTools()
|
||||
return tools.extract(
|
||||
urls=urls,
|
||||
include_images=include_images,
|
||||
extract_depth=extract_depth,
|
||||
)
|
||||
|
||||
|
||||
def tavily_crawl(
|
||||
url: str,
|
||||
max_depth: int = 1,
|
||||
max_breadth: int = 20,
|
||||
limit: int = 50,
|
||||
instructions: Optional[str] = None,
|
||||
include_images: bool = False,
|
||||
extract_depth: str = "basic",
|
||||
) -> Dict[str, Any]:
|
||||
"""Crawl a website starting from a base URL using Tavily.
|
||||
|
||||
Args:
|
||||
url: The root URL to begin crawling
|
||||
max_depth: Maximum depth from base URL
|
||||
max_breadth: Max links to follow per page
|
||||
limit: Total links to process before stopping
|
||||
instructions: Natural language instructions for the crawler
|
||||
include_images: Extract image URLs from pages
|
||||
extract_depth: "basic" or "advanced"
|
||||
|
||||
Returns:
|
||||
Dict containing crawl results
|
||||
"""
|
||||
tools = TavilyTools()
|
||||
return tools.crawl(
|
||||
url=url,
|
||||
max_depth=max_depth,
|
||||
max_breadth=max_breadth,
|
||||
limit=limit,
|
||||
instructions=instructions,
|
||||
include_images=include_images,
|
||||
extract_depth=extract_depth,
|
||||
)
|
||||
|
||||
|
||||
def tavily_map(
|
||||
url: str,
|
||||
max_depth: int = 1,
|
||||
max_breadth: int = 20,
|
||||
limit: int = 50,
|
||||
instructions: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Get a sitemap starting from a base URL using Tavily.
|
||||
|
||||
Args:
|
||||
url: The root URL to begin mapping
|
||||
max_depth: Maximum depth from base URL
|
||||
max_breadth: Max links to follow per page
|
||||
limit: Total links to process before stopping
|
||||
instructions: Natural language instructions for the mapper
|
||||
|
||||
Returns:
|
||||
Dict containing map results with discovered URLs
|
||||
"""
|
||||
tools = TavilyTools()
|
||||
return tools.map(
|
||||
url=url,
|
||||
max_depth=max_depth,
|
||||
max_breadth=max_breadth,
|
||||
limit=limit,
|
||||
instructions=instructions,
|
||||
)
|
||||
|
||||
|
||||
# Async standalone functions
|
||||
async def tavily_search_async(
|
||||
query: str,
|
||||
search_depth: str = "basic",
|
||||
topic: str = "general",
|
||||
max_results: int = 5,
|
||||
include_answer: bool = False,
|
||||
include_raw_content: bool = False,
|
||||
include_images: bool = False,
|
||||
include_domains: Optional[List[str]] = None,
|
||||
exclude_domains: Optional[List[str]] = None,
|
||||
time_range: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Async version of tavily_search. See tavily_search() for documentation."""
|
||||
tools = TavilyTools()
|
||||
return await tools.asearch(
|
||||
query=query,
|
||||
search_depth=search_depth,
|
||||
topic=topic,
|
||||
max_results=max_results,
|
||||
include_answer=include_answer,
|
||||
include_raw_content=include_raw_content,
|
||||
include_images=include_images,
|
||||
include_domains=include_domains,
|
||||
exclude_domains=exclude_domains,
|
||||
time_range=time_range,
|
||||
)
|
||||
|
||||
|
||||
async def tavily_extract_async(
|
||||
urls: Union[str, List[str]],
|
||||
include_images: bool = False,
|
||||
extract_depth: str = "basic",
|
||||
) -> Dict[str, Any]:
|
||||
"""Async version of tavily_extract. See tavily_extract() for documentation."""
|
||||
tools = TavilyTools()
|
||||
return await tools.aextract(
|
||||
urls=urls,
|
||||
include_images=include_images,
|
||||
extract_depth=extract_depth,
|
||||
)
|
||||
|
||||
|
||||
# Alias for simple usage
|
||||
tavily = tavily_search
|
||||
@@ -0,0 +1,491 @@
|
||||
"""You.com search and content tools.
|
||||
|
||||
You.com provides AI-powered web search, news search, content extraction, and image search.
|
||||
This module requires the YDC_API_KEY environment variable to be set.
|
||||
|
||||
Usage:
|
||||
from praisonai_tools import ydc_search, ydc_news, ydc_contents
|
||||
|
||||
# Search the web (unified web + news)
|
||||
results = ydc_search("AI news 2024")
|
||||
|
||||
# Get live news
|
||||
news = ydc_news("technology trends")
|
||||
|
||||
# Extract content from URLs
|
||||
content = ydc_contents(["https://example.com"])
|
||||
|
||||
# Or use the class directly
|
||||
from praisonai_tools import YouTools
|
||||
you = YouTools()
|
||||
results = you.search("AI news")
|
||||
"""
|
||||
|
||||
from typing import List, Dict, Any, Optional, Union
|
||||
import logging
|
||||
import os
|
||||
from importlib import util
|
||||
|
||||
|
||||
def _check_youdotcom_available() -> tuple:
|
||||
"""Check if You.com SDK is available and API key is set.
|
||||
|
||||
Returns:
|
||||
Tuple of (is_available, error_message)
|
||||
"""
|
||||
if util.find_spec("youdotcom") is None:
|
||||
return False, "youdotcom package is not installed. Please install it using: pip install youdotcom"
|
||||
|
||||
api_key = os.environ.get("YDC_API_KEY")
|
||||
if not api_key:
|
||||
return False, "YDC_API_KEY environment variable is not set. Please set it to use You.com tools."
|
||||
|
||||
return True, None
|
||||
|
||||
|
||||
class YouTools:
|
||||
"""Comprehensive tools for web search and content using You.com API.
|
||||
|
||||
You.com provides AI-powered search capabilities optimized for LLM applications.
|
||||
|
||||
Features:
|
||||
- Unified web and news search with LLM-ready snippets
|
||||
- Live news search from authoritative sources
|
||||
- Content extraction from URLs (HTML/Markdown)
|
||||
- Image search
|
||||
- Advanced search operators support
|
||||
- Geographic and language targeting
|
||||
- Freshness controls
|
||||
|
||||
Requires:
|
||||
- youdotcom package: pip install youdotcom
|
||||
- YDC_API_KEY environment variable
|
||||
"""
|
||||
|
||||
def __init__(self, api_key: Optional[str] = None):
|
||||
"""Initialize YouTools.
|
||||
|
||||
Args:
|
||||
api_key: Optional API key. If not provided, uses YDC_API_KEY env var.
|
||||
"""
|
||||
self._api_key = api_key
|
||||
self._client = None
|
||||
|
||||
def _get_api_key(self) -> Optional[str]:
|
||||
"""Get API key from instance or environment."""
|
||||
return self._api_key or os.environ.get("YDC_API_KEY")
|
||||
|
||||
def _get_client(self):
|
||||
"""Get or create You.com client."""
|
||||
if self._client is None:
|
||||
is_available, error = _check_youdotcom_available()
|
||||
if not is_available:
|
||||
return None, error
|
||||
|
||||
from youdotcom import You
|
||||
api_key = self._get_api_key()
|
||||
if not api_key:
|
||||
return None, "YDC_API_KEY not set"
|
||||
self._client = You(api_key_auth=api_key)
|
||||
return self._client, None
|
||||
|
||||
def search(
|
||||
self,
|
||||
query: str,
|
||||
count: int = 10,
|
||||
freshness: Optional[str] = None,
|
||||
country: Optional[str] = None,
|
||||
language: Optional[str] = None,
|
||||
offset: int = 0,
|
||||
safesearch: str = "moderate",
|
||||
livecrawl: Optional[str] = None,
|
||||
livecrawl_formats: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Search the web using You.com's unified search API.
|
||||
|
||||
Returns both web and news results in a single call with LLM-ready snippets.
|
||||
|
||||
Args:
|
||||
query: Search query (supports operators like site:, filetype:, +term, -term)
|
||||
count: Max results per section (default 10, max 100)
|
||||
freshness: Filter by recency - "day", "week", "month", "year" or "YYYY-MM-DDtoYYYY-MM-DD"
|
||||
country: ISO country code (e.g., "US", "GB", "FR")
|
||||
language: BCP 47 language code (e.g., "EN", "JP", "DE")
|
||||
offset: Pagination offset (0-9), results = offset * count
|
||||
safesearch: Content filter - "off", "moderate", "strict"
|
||||
livecrawl: Fetch full content - "web", "news", or "all"
|
||||
livecrawl_formats: Content format - "html" or "markdown"
|
||||
|
||||
Returns:
|
||||
Dict containing:
|
||||
- results: {web: [...], news: [...]}
|
||||
- metadata: {query, search_uuid, latency}
|
||||
|
||||
Search Operators:
|
||||
- site:domain.com - Search within specific domain
|
||||
- filetype:pdf - Filter by file type
|
||||
- +term - Must include term
|
||||
- -term - Exclude term
|
||||
- AND, OR, NOT - Boolean operators
|
||||
"""
|
||||
client, error = self._get_client()
|
||||
if error:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
try:
|
||||
kwargs = {"query": query, "count": count, "offset": offset}
|
||||
|
||||
if freshness:
|
||||
kwargs["freshness"] = freshness
|
||||
if country:
|
||||
kwargs["country"] = country
|
||||
if language:
|
||||
kwargs["language"] = language
|
||||
if safesearch:
|
||||
kwargs["safesearch"] = safesearch
|
||||
if livecrawl:
|
||||
kwargs["livecrawl"] = livecrawl
|
||||
if livecrawl_formats:
|
||||
kwargs["livecrawl_formats"] = livecrawl_formats
|
||||
|
||||
response = client.search.unified(**kwargs)
|
||||
|
||||
if hasattr(response, 'model_dump'):
|
||||
return response.model_dump()
|
||||
elif hasattr(response, 'to_dict'):
|
||||
return response.to_dict()
|
||||
elif hasattr(response, '__dict__'):
|
||||
return self._convert_response(response)
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"You.com search error: {str(e)}"
|
||||
logging.error(error_msg)
|
||||
return {"error": error_msg}
|
||||
|
||||
def _convert_response(self, response) -> Dict[str, Any]:
|
||||
"""Convert SDK response object to dictionary."""
|
||||
result = {}
|
||||
|
||||
if hasattr(response, 'results'):
|
||||
results = response.results
|
||||
result['results'] = {}
|
||||
if hasattr(results, 'web') and results.web:
|
||||
result['results']['web'] = [
|
||||
{
|
||||
'url': getattr(r, 'url', ''),
|
||||
'title': getattr(r, 'title', ''),
|
||||
'description': getattr(r, 'description', ''),
|
||||
'snippets': getattr(r, 'snippets', []),
|
||||
'thumbnail_url': getattr(r, 'thumbnail_url', None),
|
||||
'page_age': getattr(r, 'page_age', None),
|
||||
'favicon_url': getattr(r, 'favicon_url', None),
|
||||
'authors': getattr(r, 'authors', []),
|
||||
}
|
||||
for r in results.web
|
||||
]
|
||||
if hasattr(results, 'news') and results.news:
|
||||
result['results']['news'] = [
|
||||
{
|
||||
'url': getattr(r, 'url', ''),
|
||||
'title': getattr(r, 'title', ''),
|
||||
'description': getattr(r, 'description', ''),
|
||||
'thumbnail_url': getattr(r, 'thumbnail_url', None),
|
||||
'page_age': getattr(r, 'page_age', None),
|
||||
}
|
||||
for r in results.news
|
||||
]
|
||||
|
||||
if hasattr(response, 'metadata'):
|
||||
meta = response.metadata
|
||||
result['metadata'] = {
|
||||
'query': getattr(meta, 'query', ''),
|
||||
'search_uuid': getattr(meta, 'search_uuid', ''),
|
||||
'latency': getattr(meta, 'latency', 0),
|
||||
}
|
||||
|
||||
return result
|
||||
|
||||
def get_contents(
|
||||
self,
|
||||
urls: Union[str, List[str]],
|
||||
format: str = "markdown",
|
||||
) -> Dict[str, Any]:
|
||||
"""Fetch content from URLs using You.com's content extraction API.
|
||||
|
||||
Args:
|
||||
urls: Single URL or list of URLs to fetch content from
|
||||
format: Content format - "html" or "markdown" (default: markdown)
|
||||
|
||||
Returns:
|
||||
Dict containing array of results with:
|
||||
- url: The fetched URL
|
||||
- title: Page title
|
||||
- html: HTML content (if format="html")
|
||||
- markdown: Markdown content (if format="markdown")
|
||||
- metadata: {site_name, favicon_url}
|
||||
"""
|
||||
client, error = self._get_client()
|
||||
if error:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
try:
|
||||
if isinstance(urls, str):
|
||||
urls = [urls]
|
||||
|
||||
response = client.search.contents(urls=urls, format=format)
|
||||
|
||||
if hasattr(response, 'model_dump'):
|
||||
return {"results": response.model_dump()}
|
||||
elif isinstance(response, list):
|
||||
return {"results": [
|
||||
{
|
||||
'url': getattr(r, 'url', ''),
|
||||
'title': getattr(r, 'title', ''),
|
||||
'html': getattr(r, 'html', None),
|
||||
'markdown': getattr(r, 'markdown', None),
|
||||
'metadata': {
|
||||
'site_name': getattr(getattr(r, 'metadata', None), 'site_name', None) if hasattr(r, 'metadata') else None,
|
||||
'favicon_url': getattr(getattr(r, 'metadata', None), 'favicon_url', None) if hasattr(r, 'metadata') else None,
|
||||
}
|
||||
}
|
||||
for r in response
|
||||
]}
|
||||
return {"results": response}
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"You.com contents error: {str(e)}"
|
||||
logging.error(error_msg)
|
||||
return {"error": error_msg}
|
||||
|
||||
def live_news(
|
||||
self,
|
||||
query: str,
|
||||
count: Optional[int] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Search for live news using You.com's news API.
|
||||
|
||||
Note: This API may require early access. Contact [email protected] for access.
|
||||
|
||||
Args:
|
||||
query: News search query
|
||||
count: Maximum number of news results to return
|
||||
|
||||
Returns:
|
||||
Dict containing:
|
||||
- news: {query, results: [...], metadata}
|
||||
- Each result has: title, description, url, page_age, source_name, thumbnail
|
||||
"""
|
||||
client, error = self._get_client()
|
||||
if error:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
try:
|
||||
kwargs = {"q": query}
|
||||
if count:
|
||||
kwargs["count"] = count
|
||||
|
||||
if hasattr(client, 'livenews'):
|
||||
response = client.livenews.get(**kwargs)
|
||||
else:
|
||||
return self.search(query=query, count=count or 10)
|
||||
|
||||
if hasattr(response, 'model_dump'):
|
||||
return response.model_dump()
|
||||
elif hasattr(response, 'news'):
|
||||
news = response.news
|
||||
return {
|
||||
'news': {
|
||||
'query': getattr(getattr(news, 'query', None), 'original', query) if hasattr(news, 'query') else query,
|
||||
'results': [
|
||||
{
|
||||
'title': getattr(r, 'title', ''),
|
||||
'description': getattr(r, 'description', ''),
|
||||
'url': getattr(r, 'url', ''),
|
||||
'page_age': getattr(r, 'page_age', None),
|
||||
'age': getattr(r, 'age', None),
|
||||
'source_name': getattr(r, 'source_name', None),
|
||||
'thumbnail': getattr(getattr(r, 'thumbnail', None), 'src', None) if hasattr(r, 'thumbnail') else None,
|
||||
}
|
||||
for r in getattr(news, 'results', [])
|
||||
],
|
||||
'metadata': {
|
||||
'request_uuid': getattr(getattr(news, 'metadata', None), 'request_uuid', None) if hasattr(news, 'metadata') else None,
|
||||
}
|
||||
}
|
||||
}
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"You.com live news error: {str(e)}"
|
||||
logging.error(error_msg)
|
||||
return {"error": error_msg}
|
||||
|
||||
def images(
|
||||
self,
|
||||
query: str,
|
||||
) -> Dict[str, Any]:
|
||||
"""Search for images using You.com's image search API.
|
||||
|
||||
Note: This API may require early access. Contact [email protected] for access.
|
||||
|
||||
Args:
|
||||
query: Image search query
|
||||
|
||||
Returns:
|
||||
Dict containing:
|
||||
- images: {results: [...]}
|
||||
- metadata: {query, search_uuid}
|
||||
- Each result has: title, page_url, image_url
|
||||
"""
|
||||
client, error = self._get_client()
|
||||
if error:
|
||||
logging.error(error)
|
||||
return {"error": error}
|
||||
|
||||
try:
|
||||
if hasattr(client, 'images'):
|
||||
response = client.images.get(q=query)
|
||||
else:
|
||||
return {"error": "Image search not available in this SDK version"}
|
||||
|
||||
if hasattr(response, 'model_dump'):
|
||||
return response.model_dump()
|
||||
elif hasattr(response, 'images'):
|
||||
images = response.images
|
||||
return {
|
||||
'images': {
|
||||
'results': [
|
||||
{
|
||||
'title': getattr(r, 'title', ''),
|
||||
'page_url': getattr(r, 'page_url', ''),
|
||||
'image_url': getattr(r, 'image_url', ''),
|
||||
}
|
||||
for r in getattr(images, 'results', [])
|
||||
]
|
||||
},
|
||||
'metadata': {
|
||||
'query': getattr(getattr(response, 'metadata', None), 'query', query) if hasattr(response, 'metadata') else query,
|
||||
'search_uuid': getattr(getattr(response, 'metadata', None), 'search_uuid', None) if hasattr(response, 'metadata') else None,
|
||||
}
|
||||
}
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"You.com images error: {str(e)}"
|
||||
logging.error(error_msg)
|
||||
return {"error": error_msg}
|
||||
|
||||
def __enter__(self):
|
||||
"""Context manager entry."""
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
"""Context manager exit."""
|
||||
if self._client is not None:
|
||||
if hasattr(self._client, 'close'):
|
||||
self._client.close()
|
||||
elif hasattr(self._client, '__exit__'):
|
||||
self._client.__exit__(exc_type, exc_val, exc_tb)
|
||||
return False
|
||||
|
||||
|
||||
# Standalone functions for direct import
|
||||
def ydc_search(
|
||||
query: str,
|
||||
count: int = 10,
|
||||
freshness: Optional[str] = None,
|
||||
country: Optional[str] = None,
|
||||
language: Optional[str] = None,
|
||||
offset: int = 0,
|
||||
safesearch: str = "moderate",
|
||||
livecrawl: Optional[str] = None,
|
||||
livecrawl_formats: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Search the web using You.com's unified search API.
|
||||
|
||||
This is a convenience function that creates a YouTools instance.
|
||||
For repeated calls, consider using YouTools class directly.
|
||||
|
||||
Args:
|
||||
query: Search query (supports operators like site:, filetype:, +term, -term)
|
||||
count: Max results per section (default 10, max 100)
|
||||
freshness: Filter by recency - "day", "week", "month", "year"
|
||||
country: ISO country code (e.g., "US", "GB")
|
||||
language: BCP 47 language code (e.g., "EN", "JP")
|
||||
offset: Pagination offset (0-9)
|
||||
safesearch: Content filter - "off", "moderate", "strict"
|
||||
livecrawl: Fetch full content - "web", "news", or "all"
|
||||
livecrawl_formats: Content format - "html" or "markdown"
|
||||
|
||||
Returns:
|
||||
Dict containing search results with web and news sections
|
||||
"""
|
||||
tools = YouTools()
|
||||
return tools.search(
|
||||
query=query,
|
||||
count=count,
|
||||
freshness=freshness,
|
||||
country=country,
|
||||
language=language,
|
||||
offset=offset,
|
||||
safesearch=safesearch,
|
||||
livecrawl=livecrawl,
|
||||
livecrawl_formats=livecrawl_formats,
|
||||
)
|
||||
|
||||
|
||||
def ydc_contents(
|
||||
urls: Union[str, List[str]],
|
||||
format: str = "markdown",
|
||||
) -> Dict[str, Any]:
|
||||
"""Fetch content from URLs using You.com's content extraction API.
|
||||
|
||||
Args:
|
||||
urls: Single URL or list of URLs to fetch content from
|
||||
format: Content format - "html" or "markdown"
|
||||
|
||||
Returns:
|
||||
Dict containing extracted content for each URL
|
||||
"""
|
||||
tools = YouTools()
|
||||
return tools.get_contents(urls=urls, format=format)
|
||||
|
||||
|
||||
def ydc_news(
|
||||
query: str,
|
||||
count: Optional[int] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Search for live news using You.com's news API.
|
||||
|
||||
Args:
|
||||
query: News search query
|
||||
count: Maximum number of results
|
||||
|
||||
Returns:
|
||||
Dict containing news results
|
||||
"""
|
||||
tools = YouTools()
|
||||
return tools.live_news(query=query, count=count)
|
||||
|
||||
|
||||
def ydc_images(
|
||||
query: str,
|
||||
) -> Dict[str, Any]:
|
||||
"""Search for images using You.com's image search API.
|
||||
|
||||
Args:
|
||||
query: Image search query
|
||||
|
||||
Returns:
|
||||
Dict containing image results
|
||||
"""
|
||||
tools = YouTools()
|
||||
return tools.images(query=query)
|
||||
|
||||
|
||||
# Alias for simple usage
|
||||
ydc = ydc_search
|
||||
@@ -0,0 +1,60 @@
|
||||
[project]
|
||||
name = "praisonai-tools"
|
||||
version = "0.1.0"
|
||||
description = "A comprehensive toolkit for AI agents - search, extraction, and content tools"
|
||||
authors = [
|
||||
{name = "Mervin Praison", email = "[email protected]"}
|
||||
]
|
||||
readme = "README.md"
|
||||
license = {text = "MIT"}
|
||||
requires-python = ">=3.10"
|
||||
keywords = ["ai", "agents", "tools", "search", "tavily", "exa", "youdotcom", "praisonai"]
|
||||
classifiers = [
|
||||
"Development Status :: 4 - Beta",
|
||||
"Intended Audience :: Developers",
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Topic :: Software Development :: Libraries :: Python Modules",
|
||||
"Topic :: Scientific/Engineering :: Artificial Intelligence",
|
||||
]
|
||||
dependencies = []
|
||||
|
||||
[project.optional-dependencies]
|
||||
tavily = ["tavily-python>=0.5.0"]
|
||||
exa = ["exa-py>=1.0.0"]
|
||||
youdotcom = ["youdotcom>=0.1.0"]
|
||||
all = [
|
||||
"tavily-python>=0.5.0",
|
||||
"exa-py>=1.0.0",
|
||||
"youdotcom>=0.1.0",
|
||||
]
|
||||
dev = [
|
||||
"pytest>=8.0.0",
|
||||
"pytest-asyncio>=0.23.0",
|
||||
"python-dotenv>=1.0.0",
|
||||
]
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://docs.praison.ai"
|
||||
Repository = "https://github.com/MervinPraison/PraisonAI-Tools"
|
||||
Documentation = "https://docs.praison.ai/tools"
|
||||
Issues = "https://github.com/MervinPraison/PraisonAI-Tools/issues"
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling>=1.0.0"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["praisonai_tools"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
asyncio_mode = "auto"
|
||||
markers = [
|
||||
"tavily: tests requiring TAVILY_API_KEY",
|
||||
"exa: tests requiring EXA_API_KEY",
|
||||
"youdotcom: tests requiring YDC_API_KEY",
|
||||
]
|
||||
@@ -0,0 +1 @@
|
||||
"""Tests for PraisonAI Tools."""
|
||||
@@ -0,0 +1,38 @@
|
||||
"""Pytest configuration and fixtures for PraisonAI Tools tests."""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
|
||||
|
||||
def pytest_configure(config):
|
||||
"""Configure pytest markers."""
|
||||
config.addinivalue_line("markers", "tavily: tests requiring TAVILY_API_KEY")
|
||||
config.addinivalue_line("markers", "exa: tests requiring EXA_API_KEY")
|
||||
config.addinivalue_line("markers", "youdotcom: tests requiring YDC_API_KEY")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tavily_api_key():
|
||||
"""Get Tavily API key from environment."""
|
||||
key = os.environ.get("TAVILY_API_KEY")
|
||||
if not key:
|
||||
pytest.skip("TAVILY_API_KEY not set")
|
||||
return key
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def exa_api_key():
|
||||
"""Get Exa API key from environment."""
|
||||
key = os.environ.get("EXA_API_KEY")
|
||||
if not key:
|
||||
pytest.skip("EXA_API_KEY not set")
|
||||
return key
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ydc_api_key():
|
||||
"""Get You.com API key from environment."""
|
||||
key = os.environ.get("YDC_API_KEY")
|
||||
if not key:
|
||||
pytest.skip("YDC_API_KEY not set")
|
||||
return key
|
||||
@@ -0,0 +1,237 @@
|
||||
"""Tests for base tool classes."""
|
||||
|
||||
import pytest
|
||||
from praisonai_tools.tools.base import BaseTool, ToolResult, ToolValidationError, validate_tool
|
||||
from praisonai_tools.tools.decorator import tool, FunctionTool, is_tool, get_tool_schema
|
||||
|
||||
|
||||
class TestToolResult:
|
||||
"""Tests for ToolResult class."""
|
||||
|
||||
def test_successful_result(self):
|
||||
"""Test creating a successful result."""
|
||||
result = ToolResult(output="test output", success=True)
|
||||
assert result.success is True
|
||||
assert result.output == "test output"
|
||||
assert result.error is None
|
||||
assert str(result) == "test output"
|
||||
|
||||
def test_failed_result(self):
|
||||
"""Test creating a failed result."""
|
||||
result = ToolResult(output=None, success=False, error="Something went wrong")
|
||||
assert result.success is False
|
||||
assert result.output is None
|
||||
assert result.error == "Something went wrong"
|
||||
assert "Error:" in str(result)
|
||||
|
||||
def test_result_with_metadata(self):
|
||||
"""Test result with metadata."""
|
||||
result = ToolResult(
|
||||
output="data",
|
||||
success=True,
|
||||
metadata={"duration": 1.5, "source": "test"}
|
||||
)
|
||||
assert result.metadata["duration"] == 1.5
|
||||
assert result.metadata["source"] == "test"
|
||||
|
||||
def test_to_dict(self):
|
||||
"""Test converting result to dictionary."""
|
||||
result = ToolResult(output="test", success=True, metadata={"key": "value"})
|
||||
d = result.to_dict()
|
||||
assert d["output"] == "test"
|
||||
assert d["success"] is True
|
||||
assert d["metadata"]["key"] == "value"
|
||||
|
||||
|
||||
class TestBaseTool:
|
||||
"""Tests for BaseTool class."""
|
||||
|
||||
def test_concrete_tool_creation(self):
|
||||
"""Test creating a concrete tool."""
|
||||
class MyTool(BaseTool):
|
||||
name = "my_tool"
|
||||
description = "A test tool"
|
||||
|
||||
def run(self, query: str) -> str:
|
||||
return f"Result: {query}"
|
||||
|
||||
tool = MyTool()
|
||||
assert tool.name == "my_tool"
|
||||
assert tool.description == "A test tool"
|
||||
assert tool.version == "1.0.0"
|
||||
|
||||
def test_tool_execution(self):
|
||||
"""Test executing a tool."""
|
||||
class EchoTool(BaseTool):
|
||||
name = "echo"
|
||||
description = "Echoes input"
|
||||
|
||||
def run(self, message: str) -> str:
|
||||
return message
|
||||
|
||||
tool = EchoTool()
|
||||
result = tool.run(message="Hello")
|
||||
assert result == "Hello"
|
||||
|
||||
# Test callable
|
||||
result = tool(message="World")
|
||||
assert result == "World"
|
||||
|
||||
def test_safe_run_success(self):
|
||||
"""Test safe_run with successful execution."""
|
||||
class SafeTool(BaseTool):
|
||||
name = "safe"
|
||||
description = "Safe tool"
|
||||
|
||||
def run(self, x: int) -> int:
|
||||
return x * 2
|
||||
|
||||
tool = SafeTool()
|
||||
result = tool.safe_run(x=5)
|
||||
assert result.success is True
|
||||
assert result.output == 10
|
||||
|
||||
def test_safe_run_failure(self):
|
||||
"""Test safe_run with failed execution."""
|
||||
class FailingTool(BaseTool):
|
||||
name = "failing"
|
||||
description = "Failing tool"
|
||||
|
||||
def run(self, x: int) -> int:
|
||||
raise ValueError("Intentional error")
|
||||
|
||||
tool = FailingTool()
|
||||
result = tool.safe_run(x=5)
|
||||
assert result.success is False
|
||||
assert "Intentional error" in result.error
|
||||
|
||||
def test_get_schema(self):
|
||||
"""Test getting OpenAI-compatible schema."""
|
||||
class SchemaTool(BaseTool):
|
||||
name = "schema_tool"
|
||||
description = "Tool with schema"
|
||||
|
||||
def run(self, query: str, limit: int = 10) -> dict:
|
||||
return {}
|
||||
|
||||
tool = SchemaTool()
|
||||
schema = tool.get_schema()
|
||||
|
||||
assert schema["type"] == "function"
|
||||
assert schema["function"]["name"] == "schema_tool"
|
||||
assert schema["function"]["description"] == "Tool with schema"
|
||||
assert "query" in schema["function"]["parameters"]["properties"]
|
||||
|
||||
def test_auto_name_from_class(self):
|
||||
"""Test automatic name generation from class name."""
|
||||
class AutoNameTool(BaseTool):
|
||||
description = "Auto named"
|
||||
|
||||
def run(self) -> str:
|
||||
return "ok"
|
||||
|
||||
tool = AutoNameTool()
|
||||
assert tool.name == "autoname"
|
||||
|
||||
def test_validation(self):
|
||||
"""Test tool validation."""
|
||||
class ValidTool(BaseTool):
|
||||
name = "valid"
|
||||
description = "Valid tool"
|
||||
|
||||
def run(self) -> str:
|
||||
return "ok"
|
||||
|
||||
tool = ValidTool()
|
||||
assert tool.validate() is True
|
||||
|
||||
|
||||
class TestToolDecorator:
|
||||
"""Tests for @tool decorator."""
|
||||
|
||||
def test_simple_decorator(self):
|
||||
"""Test simple @tool decorator."""
|
||||
@tool
|
||||
def greet(name: str) -> str:
|
||||
"""Greet someone."""
|
||||
return f"Hello, {name}!"
|
||||
|
||||
assert isinstance(greet, FunctionTool)
|
||||
assert greet.name == "greet"
|
||||
assert "Greet someone" in greet.description
|
||||
|
||||
result = greet(name="World")
|
||||
assert result == "Hello, World!"
|
||||
|
||||
def test_decorator_with_args(self):
|
||||
"""Test @tool decorator with arguments."""
|
||||
@tool(name="custom_greet", description="Custom greeting")
|
||||
def greet(name: str) -> str:
|
||||
return f"Hi, {name}!"
|
||||
|
||||
assert greet.name == "custom_greet"
|
||||
assert greet.description == "Custom greeting"
|
||||
|
||||
def test_decorator_preserves_signature(self):
|
||||
"""Test that decorator preserves function signature."""
|
||||
@tool
|
||||
def add(a: int, b: int = 0) -> int:
|
||||
"""Add two numbers."""
|
||||
return a + b
|
||||
|
||||
schema = add.get_schema()
|
||||
params = schema["function"]["parameters"]
|
||||
|
||||
assert "a" in params["properties"]
|
||||
assert "b" in params["properties"]
|
||||
assert "a" in params["required"]
|
||||
assert "b" not in params["required"] # Has default value
|
||||
|
||||
def test_is_tool(self):
|
||||
"""Test is_tool function."""
|
||||
@tool
|
||||
def my_tool() -> str:
|
||||
"""A tool."""
|
||||
return "ok"
|
||||
|
||||
assert is_tool(my_tool) is True
|
||||
assert is_tool(lambda: None) is False
|
||||
|
||||
def test_get_tool_schema(self):
|
||||
"""Test get_tool_schema function."""
|
||||
@tool
|
||||
def search(query: str) -> list:
|
||||
"""Search for something."""
|
||||
return []
|
||||
|
||||
schema = get_tool_schema(search)
|
||||
assert schema is not None
|
||||
assert schema["function"]["name"] == "search"
|
||||
|
||||
|
||||
class TestValidateTool:
|
||||
"""Tests for validate_tool function."""
|
||||
|
||||
def test_validate_base_tool(self):
|
||||
"""Test validating a BaseTool instance."""
|
||||
class ValidTool(BaseTool):
|
||||
name = "valid"
|
||||
description = "Valid"
|
||||
|
||||
def run(self) -> str:
|
||||
return "ok"
|
||||
|
||||
tool = ValidTool()
|
||||
assert validate_tool(tool) is True
|
||||
|
||||
def test_validate_callable(self):
|
||||
"""Test validating a callable."""
|
||||
def my_func():
|
||||
pass
|
||||
|
||||
assert validate_tool(my_func) is True
|
||||
|
||||
def test_validate_invalid(self):
|
||||
"""Test validating an invalid object."""
|
||||
with pytest.raises(ToolValidationError):
|
||||
validate_tool("not a tool")
|
||||
@@ -0,0 +1,178 @@
|
||||
"""Tests for Exa tools with real API calls.
|
||||
|
||||
These tests require EXA_API_KEY environment variable to be set.
|
||||
Run with: pytest tests/test_exa.py -v
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
# Skip all tests if exa_py is not installed
|
||||
pytest.importorskip("exa_py")
|
||||
|
||||
from praisonai_tools import ExaTools, exa_search, exa_search_contents, exa_find_similar, exa_answer
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def exa_tools(exa_api_key):
|
||||
"""Create ExaTools instance."""
|
||||
return ExaTools(api_key=exa_api_key)
|
||||
|
||||
|
||||
class TestExaSearch:
|
||||
"""Tests for Exa search functionality."""
|
||||
|
||||
@pytest.mark.exa
|
||||
def test_basic_search(self, exa_api_key):
|
||||
"""Test basic search functionality."""
|
||||
result = exa_search("artificial intelligence startups", num_results=3)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
assert len(result["results"]) > 0
|
||||
|
||||
# Check result structure
|
||||
first_result = result["results"][0]
|
||||
assert "url" in first_result
|
||||
assert "id" in first_result
|
||||
|
||||
@pytest.mark.exa
|
||||
def test_search_with_category(self, exa_api_key):
|
||||
"""Test search with category filter."""
|
||||
result = exa_search(
|
||||
"machine learning",
|
||||
category="research paper",
|
||||
num_results=3
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
|
||||
@pytest.mark.exa
|
||||
def test_search_with_domain_filter(self, exa_api_key):
|
||||
"""Test search with domain filtering."""
|
||||
result = exa_search(
|
||||
"python programming",
|
||||
include_domains=["github.com"],
|
||||
num_results=3
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
|
||||
|
||||
class TestExaSearchContents:
|
||||
"""Tests for Exa search with contents."""
|
||||
|
||||
@pytest.mark.exa
|
||||
def test_search_with_text(self, exa_api_key):
|
||||
"""Test search with full text content."""
|
||||
result = exa_search_contents(
|
||||
"latest AI news",
|
||||
text=True,
|
||||
num_results=2
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
|
||||
# Check that text is included
|
||||
if len(result["results"]) > 0:
|
||||
first_result = result["results"][0]
|
||||
# Text should be present if available
|
||||
assert "url" in first_result
|
||||
|
||||
@pytest.mark.exa
|
||||
def test_search_with_text_only(self, exa_api_key):
|
||||
"""Test search with text content only."""
|
||||
result = exa_search_contents(
|
||||
"quantum computing",
|
||||
text=True,
|
||||
highlights=False,
|
||||
num_results=2
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
|
||||
|
||||
class TestExaFindSimilar:
|
||||
"""Tests for Exa find similar functionality."""
|
||||
|
||||
@pytest.mark.exa
|
||||
def test_find_similar(self, exa_api_key):
|
||||
"""Test finding similar pages."""
|
||||
result = exa_find_similar(
|
||||
"https://openai.com",
|
||||
num_results=3,
|
||||
exclude_source_domain=True
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Find similar failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
# Just verify we got results - domain exclusion may not always work perfectly
|
||||
assert len(result.get("results", [])) >= 0
|
||||
|
||||
|
||||
class TestExaAnswer:
|
||||
"""Tests for Exa answer functionality."""
|
||||
|
||||
@pytest.mark.exa
|
||||
def test_basic_answer(self, exa_api_key):
|
||||
"""Test getting an AI-generated answer."""
|
||||
result = exa_answer("What is the capital of Japan?")
|
||||
|
||||
assert "error" not in result, f"Answer failed: {result.get('error')}"
|
||||
assert "answer" in result
|
||||
assert len(result["answer"]) > 0
|
||||
assert "citations" in result
|
||||
|
||||
@pytest.mark.exa
|
||||
def test_answer_with_text(self, exa_api_key):
|
||||
"""Test answer with citation text."""
|
||||
result = exa_answer(
|
||||
"What are the main features of Python programming language?",
|
||||
text=True
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Answer failed: {result.get('error')}"
|
||||
assert "answer" in result
|
||||
assert "citations" in result
|
||||
|
||||
|
||||
class TestExaToolsClass:
|
||||
"""Tests for ExaTools class."""
|
||||
|
||||
@pytest.mark.exa
|
||||
def test_class_instantiation(self, exa_api_key):
|
||||
"""Test creating ExaTools instance."""
|
||||
tools = ExaTools(api_key=exa_api_key)
|
||||
assert tools.api_key == exa_api_key
|
||||
|
||||
@pytest.mark.exa
|
||||
def test_class_search(self, exa_tools):
|
||||
"""Test search through class instance."""
|
||||
result = exa_tools.search("Python programming", num_results=2)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
|
||||
@pytest.mark.exa
|
||||
def test_class_answer(self, exa_tools):
|
||||
"""Test answer through class instance."""
|
||||
result = exa_tools.answer("What is machine learning?")
|
||||
|
||||
assert "error" not in result, f"Answer failed: {result.get('error')}"
|
||||
assert "answer" in result
|
||||
|
||||
|
||||
class TestExaErrorHandling:
|
||||
"""Tests for error handling."""
|
||||
|
||||
def test_missing_api_key(self, monkeypatch):
|
||||
"""Test behavior when API key is missing."""
|
||||
monkeypatch.delenv("EXA_API_KEY", raising=False)
|
||||
|
||||
result = exa_search("test query")
|
||||
|
||||
assert "error" in result
|
||||
assert "EXA_API_KEY" in result["error"]
|
||||
@@ -0,0 +1,194 @@
|
||||
"""Tests for Tavily tools with real API calls.
|
||||
|
||||
These tests require TAVILY_API_KEY environment variable to be set.
|
||||
Run with: pytest tests/test_tavily.py -v
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
# Skip all tests if tavily is not installed
|
||||
pytest.importorskip("tavily")
|
||||
|
||||
from praisonai_tools import TavilyTools, tavily_search, tavily_extract, tavily_crawl, tavily_map
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tavily_tools(tavily_api_key):
|
||||
"""Create TavilyTools instance."""
|
||||
return TavilyTools(api_key=tavily_api_key)
|
||||
|
||||
|
||||
class TestTavilySearch:
|
||||
"""Tests for Tavily search functionality."""
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_basic_search(self, tavily_api_key):
|
||||
"""Test basic search functionality."""
|
||||
result = tavily_search("What is Python programming language?", max_results=3)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
assert len(result["results"]) > 0
|
||||
|
||||
# Check result structure
|
||||
first_result = result["results"][0]
|
||||
assert "url" in first_result
|
||||
assert "title" in first_result or "content" in first_result
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_search_with_answer(self, tavily_api_key):
|
||||
"""Test search with LLM-generated answer."""
|
||||
result = tavily_search(
|
||||
"What is the capital of France?",
|
||||
include_answer=True,
|
||||
max_results=3
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
assert "answer" in result
|
||||
assert len(result["answer"]) > 0
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_search_with_topic(self, tavily_api_key):
|
||||
"""Test search with topic filter."""
|
||||
result = tavily_search(
|
||||
"latest AI developments",
|
||||
topic="news",
|
||||
max_results=5
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_search_with_domain_filter(self, tavily_api_key):
|
||||
"""Test search with domain filtering."""
|
||||
result = tavily_search(
|
||||
"machine learning",
|
||||
include_domains=["arxiv.org"],
|
||||
max_results=3
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
# Results should be from arxiv.org
|
||||
for r in result.get("results", []):
|
||||
if "url" in r:
|
||||
assert "arxiv.org" in r["url"].lower() or len(result["results"]) == 0
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_search_advanced_depth(self, tavily_api_key):
|
||||
"""Test search with advanced depth."""
|
||||
result = tavily_search(
|
||||
"quantum computing applications",
|
||||
search_depth="advanced",
|
||||
max_results=3
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
|
||||
|
||||
class TestTavilyExtract:
|
||||
"""Tests for Tavily content extraction."""
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_extract_single_url(self, tavily_api_key):
|
||||
"""Test extracting content from a single URL."""
|
||||
result = tavily_extract("https://en.wikipedia.org/wiki/Python_(programming_language)")
|
||||
|
||||
assert "error" not in result, f"Extract failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
if len(result["results"]) > 0:
|
||||
assert "raw_content" in result["results"][0] or "url" in result["results"][0]
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_extract_multiple_urls(self, tavily_api_key):
|
||||
"""Test extracting content from multiple URLs."""
|
||||
urls = [
|
||||
"https://en.wikipedia.org/wiki/Artificial_intelligence",
|
||||
"https://en.wikipedia.org/wiki/Machine_learning"
|
||||
]
|
||||
result = tavily_extract(urls)
|
||||
|
||||
assert "error" not in result, f"Extract failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
|
||||
|
||||
class TestTavilyCrawl:
|
||||
"""Tests for Tavily website crawling."""
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_basic_crawl(self, tavily_api_key):
|
||||
"""Test basic website crawling."""
|
||||
result = tavily_crawl(
|
||||
"https://docs.python.org/3/",
|
||||
max_depth=1,
|
||||
limit=5
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Crawl failed: {result.get('error')}"
|
||||
# Crawl should return results or base_url
|
||||
assert "results" in result or "base_url" in result
|
||||
|
||||
|
||||
class TestTavilyMap:
|
||||
"""Tests for Tavily site mapping."""
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_basic_map(self, tavily_api_key):
|
||||
"""Test basic site mapping."""
|
||||
result = tavily_map(
|
||||
"https://docs.python.org/3/",
|
||||
max_depth=1,
|
||||
limit=10
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Map failed: {result.get('error')}"
|
||||
# Map should return results or base_url
|
||||
assert "results" in result or "base_url" in result
|
||||
|
||||
|
||||
class TestTavilyToolsClass:
|
||||
"""Tests for TavilyTools class."""
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_class_instantiation(self, tavily_api_key):
|
||||
"""Test creating TavilyTools instance."""
|
||||
tools = TavilyTools(api_key=tavily_api_key)
|
||||
assert tools._api_key == tavily_api_key
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_class_search(self, tavily_tools):
|
||||
"""Test search through class instance."""
|
||||
result = tavily_tools.search("Python programming", max_results=2)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
assert "results" in result
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_class_extract(self, tavily_tools):
|
||||
"""Test extract through class instance."""
|
||||
result = tavily_tools.extract("https://www.python.org")
|
||||
|
||||
assert "error" not in result, f"Extract failed: {result.get('error')}"
|
||||
|
||||
|
||||
class TestTavilyErrorHandling:
|
||||
"""Tests for error handling."""
|
||||
|
||||
def test_missing_api_key(self, monkeypatch):
|
||||
"""Test behavior when API key is missing."""
|
||||
monkeypatch.delenv("TAVILY_API_KEY", raising=False)
|
||||
|
||||
result = tavily_search("test query")
|
||||
|
||||
assert "error" in result
|
||||
assert "TAVILY_API_KEY" in result["error"]
|
||||
|
||||
@pytest.mark.tavily
|
||||
def test_invalid_url_extract(self, tavily_api_key):
|
||||
"""Test extracting from invalid URL."""
|
||||
result = tavily_extract("https://this-domain-does-not-exist-12345.com")
|
||||
|
||||
# Should either have error or failed_results
|
||||
assert "error" in result or "failed_results" in result or "results" in result
|
||||
@@ -0,0 +1,177 @@
|
||||
"""Tests for You.com tools with real API calls.
|
||||
|
||||
These tests require YDC_API_KEY environment variable to be set.
|
||||
Run with: pytest tests/test_youdotcom.py -v
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
# Skip all tests if youdotcom is not installed
|
||||
pytest.importorskip("youdotcom")
|
||||
|
||||
from praisonai_tools import YouTools, ydc_search, ydc_contents, ydc_news, ydc_images
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def you_tools(ydc_api_key):
|
||||
"""Create YouTools instance."""
|
||||
return YouTools(api_key=ydc_api_key)
|
||||
|
||||
|
||||
class TestYdcSearch:
|
||||
"""Tests for You.com search functionality."""
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_basic_search(self, ydc_api_key):
|
||||
"""Test basic search functionality."""
|
||||
result = ydc_search("Python programming language", count=3)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
# Result should have results or metadata
|
||||
assert "results" in result or "metadata" in result
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_search_with_freshness(self, ydc_api_key):
|
||||
"""Test search with freshness filter."""
|
||||
result = ydc_search(
|
||||
"latest AI news",
|
||||
freshness="week",
|
||||
count=5
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_search_with_country(self, ydc_api_key):
|
||||
"""Test search with country filter."""
|
||||
result = ydc_search(
|
||||
"technology news",
|
||||
country="US",
|
||||
count=3
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_search_with_safesearch(self, ydc_api_key):
|
||||
"""Test search with safesearch filter."""
|
||||
result = ydc_search(
|
||||
"programming tutorials",
|
||||
safesearch="strict",
|
||||
count=3
|
||||
)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
|
||||
|
||||
class TestYdcContents:
|
||||
"""Tests for You.com content extraction."""
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_extract_single_url(self, ydc_api_key):
|
||||
"""Test extracting content from a single URL."""
|
||||
result = ydc_contents("https://www.python.org")
|
||||
|
||||
# Contents API may not be available in all SDK versions
|
||||
# Just verify it returns a dict (with error or results)
|
||||
assert isinstance(result, dict)
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_extract_markdown_format(self, ydc_api_key):
|
||||
"""Test extracting content in markdown format."""
|
||||
result = ydc_contents(
|
||||
"https://en.wikipedia.org/wiki/Python_(programming_language)",
|
||||
format="markdown"
|
||||
)
|
||||
|
||||
# Contents API may not be available in all SDK versions
|
||||
assert isinstance(result, dict)
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_extract_multiple_urls(self, ydc_api_key):
|
||||
"""Test extracting content from multiple URLs."""
|
||||
urls = [
|
||||
"https://www.python.org",
|
||||
"https://docs.python.org"
|
||||
]
|
||||
result = ydc_contents(urls)
|
||||
|
||||
# Contents API may not be available in all SDK versions
|
||||
assert isinstance(result, dict)
|
||||
|
||||
|
||||
class TestYdcNews:
|
||||
"""Tests for You.com news search."""
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_basic_news(self, ydc_api_key):
|
||||
"""Test basic news search."""
|
||||
result = ydc_news("technology", count=5)
|
||||
|
||||
# News might fall back to unified search
|
||||
assert "error" not in result, f"News failed: {result.get('error')}"
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_news_with_count(self, ydc_api_key):
|
||||
"""Test news search with count limit."""
|
||||
result = ydc_news("artificial intelligence", count=3)
|
||||
|
||||
assert "error" not in result, f"News failed: {result.get('error')}"
|
||||
|
||||
|
||||
class TestYdcImages:
|
||||
"""Tests for You.com image search."""
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_basic_images(self, ydc_api_key):
|
||||
"""Test basic image search."""
|
||||
result = ydc_images("python programming")
|
||||
|
||||
# Image search might not be available in all SDK versions
|
||||
# So we just check it doesn't crash
|
||||
assert isinstance(result, dict)
|
||||
|
||||
|
||||
class TestYouToolsClass:
|
||||
"""Tests for YouTools class."""
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_class_instantiation(self, ydc_api_key):
|
||||
"""Test creating YouTools instance."""
|
||||
tools = YouTools(api_key=ydc_api_key)
|
||||
assert tools._api_key == ydc_api_key
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_class_search(self, you_tools):
|
||||
"""Test search through class instance."""
|
||||
result = you_tools.search("Python programming", count=2)
|
||||
|
||||
assert "error" not in result, f"Search failed: {result.get('error')}"
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_class_get_contents(self, you_tools):
|
||||
"""Test get_contents through class instance."""
|
||||
result = you_tools.get_contents("https://www.python.org")
|
||||
|
||||
# Contents API may not be available in all SDK versions
|
||||
assert isinstance(result, dict)
|
||||
|
||||
@pytest.mark.youdotcom
|
||||
def test_context_manager(self, ydc_api_key):
|
||||
"""Test using YouTools as context manager."""
|
||||
with YouTools(api_key=ydc_api_key) as tools:
|
||||
result = tools.search("test", count=1)
|
||||
assert isinstance(result, dict)
|
||||
|
||||
|
||||
class TestYdcErrorHandling:
|
||||
"""Tests for error handling."""
|
||||
|
||||
def test_missing_api_key(self, monkeypatch):
|
||||
"""Test behavior when API key is missing."""
|
||||
monkeypatch.delenv("YDC_API_KEY", raising=False)
|
||||
|
||||
result = ydc_search("test query")
|
||||
|
||||
assert "error" in result
|
||||
assert "YDC_API_KEY" in result["error"]
|
||||
Reference in new issue
Block a user