Compare commits

...
12 Commits
Author SHA1 Message Date
Mervin Praison 69740baa7c Release v2.3.7 2025-12-16 07:46:40 +00:00
Mervin Praison 5035978a29 fix: resolve Mintlify parsing errors
- Fix corrupted/duplicated database service in docker-compose.dev.yml
- Fix MDX parsing error in tests/README.md (<1s -> under 1s)
2025-12-15 23:11:04 +00:00
Mervin Praison 4ddb043324 Release v2.3.7 2025-12-15 22:49:12 +00:00
Mervin Praison 337014f4db planning example 2025-12-15 22:25:09 +00:00
Mervin Praison 35c371b25e update 2025-12-15 22:22:42 +00:00
Mervin Praison bba75d9fdd Release v2.3.6 2025-12-15 22:16:57 +00:00
Mervin Praison bf8cd488da feat: add planning tools and reasoning mode for research-enabled plan creation
- Add planning_tools and planning_reasoning parameters to PraisonAIAgents
- Add RESEARCH_TOOLS list for safe planning research tools (web_search, read_file, etc.)
- Add tools, reasoning, and max_reasoning_steps parameters to PlanningAgent
- Add _get_agent() method to PlanningAgent for tool-enabled planning with Agent.chat()
- Update plan creation to use Agent with tools when available, fallback to bare LLM
- Add Planning
2025-12-15 22:07:59 +00:00
Mervin Praison 28905db46d g 2025-12-15 21:27:32 +00:00
Mervin Praison 3e7aa44dfa docs: simplify feature descriptions by removing tool comparisons
- Remove references to Cursor, Windsurf, Claude Code, Gemini CLI, Codex CLI, and Cascade from feature descriptions
- Keep feature descriptions focused on PraisonAI's own capabilities
- Update Planning Mode, Agent Memory, Rules & Instructions, Auto-Generated Memories, Workflows, and Hooks sections
2025-12-15 21:26:04 +00:00
Mervin Praison 2f7a65ab92 update 2025-12-15 21:25:41 +00:00
Mervin Praison e6e2379b02 update 2025-12-15 21:25:29 +00:00
Mervin Praison f31f9a04aa update 2025-12-15 21:23:46 +00:00
36 changed files with 4058 additions and 85 deletions

No files matched your search

+137 -12
View File
@@ -60,7 +60,10 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
- 💾 **Prompt Caching** (Reduce costs & latency - OpenAI, Anthropic, Bedrock, Deepseek)
- 🧠 **Claude Memory Tool** (Persistent cross-conversation memory - Anthropic Beta)
- 💾 **File-Based Memory** (Zero-dependency persistent memory for all agents)
- 📋 **Planning Mode** (Plan before execution like Cursor, Windsurf, Claude Code)
- 🔍 **Built-in Search Tools** (Tavily, You.com, Exa - web search, news, content extraction)
- 📋 **Planning Mode** (Plan before execution - Agent & Multi-Agent)
- 🔧 **Planning Tools** (Research with tools during planning)
- 🧠 **Planning Reasoning** (Chain-of-thought planning)
## Using Python Code
@@ -104,7 +107,34 @@ Run:
python app.py
```
### 3. Deep Research Agent
### 3. Agent with Planning Mode
Enable planning for any agent - the agent creates a plan, then executes step by step:
```python
from praisonaiagents import Agent
def search_web(query: str) -> str:
return f"Search results for: {query}"
agent = Agent(
name="AI Assistant",
instructions="Research and write about topics",
planning=True, # Enable planning mode
planning_tools=[search_web], # Tools for planning research
planning_reasoning=True # Chain-of-thought reasoning
)
result = agent.start("Research AI trends in 2025 and write a summary")
```
**What happens:**
1. 📋 Agent creates a multi-step plan
2. 🚀 Executes each step sequentially
3. 📊 Shows progress with context passing
4. ✅ Returns final result
### 4. Deep Research Agent
Automated research with real-time streaming, web search, and citations using OpenAI or Gemini Deep Research APIs.
@@ -142,7 +172,7 @@ print(result.report)
- 🛠️ Built-in tools: web search, code interpreter, MCP, file search
- 🔄 Automatic provider detection from model name
### 4. Query Rewriter Agent
### 5. Query Rewriter Agent
Transform user queries to improve RAG retrieval quality using multiple strategies.
@@ -177,9 +207,9 @@ result = agent.rewrite("What about cost?", chat_history=[...])
- **CONTEXTUAL**: Resolve references using conversation history
- **AUTO**: Automatically detect best strategy
### 5. Agent Memory (Zero Dependencies)
### 6. Agent Memory (Zero Dependencies)
Enable persistent memory for agents - works out of the box without any extra packages. Features inspired by Claude, Gemini CLI, Codex CLI, Cursor, and Windsurf.
Enable persistent memory for agents - works out of the box without any extra packages.
```python
from praisonaiagents import Agent
@@ -204,7 +234,7 @@ result = agent.start("My name is John and I prefer Python")
- **Entity**: People, places, organizations with attributes
- **Episodic**: Date-based interaction history
**Advanced Features (like Gemini CLI):**
**Advanced Features:**
```python
from praisonaiagents.memory import FileMemory
@@ -234,7 +264,7 @@ memory.handle_command("/memory save my_session")
| `memory="sqlite"` | Built-in | SQLite with indexing |
| `memory="chromadb"` | chromadb | Vector/semantic search |
### 6. Rules & Instructions (like Cursor/Windsurf)
### 6. Rules & Instructions
PraisonAI auto-discovers instruction files from your project root and git root:
@@ -262,7 +292,7 @@ agent = Agent(name="Assistant", instructions="You are helpful.")
# Rules are injected into system prompt automatically
```
**@Import Syntax (like Claude Code):**
**@Import Syntax:**
```markdown
# CLAUDE.md
See @README for project overview
@@ -283,7 +313,7 @@ activation: always # always, glob, manual, ai_decision
- Follow PEP 8
```
### 7. Auto-Generated Memories (like Windsurf Cascade)
### 7. Auto-Generated Memories
```python
from praisonaiagents.memory import FileMemory, AutoMemory
@@ -298,7 +328,7 @@ memories = auto.process_interaction(
# Extracts: name="John", preference="Python for backend"
```
### 8. Workflows (like Windsurf)
### 8. Workflows
Create reusable multi-step workflows in `.praison/workflows/`:
@@ -315,7 +345,7 @@ result = manager.execute(
)
```
### 9. Hooks (like Windsurf Cascade Hooks)
### 9. Hooks
Configure in `.praison/hooks.json`:
@@ -385,6 +415,101 @@ praisonai research --query-rewrite --tools tools.py --save "Your research query"
praisonai research -v "Your research query"
```
### Planning Mode CLI:
```bash
# Enable planning mode - agent creates a plan before execution
praisonai "Research AI trends and write a summary" --planning
# Planning with tools for research
praisonai "Analyze market trends" --planning --planning-tools tools.py
# Planning with chain-of-thought reasoning
praisonai "Complex analysis task" --planning --planning-reasoning
# Auto-approve plans without confirmation
praisonai "Task" --planning --auto-approve-plan
```
### Memory CLI:
```bash
# Enable memory for agent (persists across sessions)
praisonai "My name is John" --memory
# Memory with user isolation
praisonai "Remember my preferences" --memory --user-id user123
# Memory management commands
praisonai memory show # Show memory statistics
praisonai memory add "User prefers Python" # Add to long-term memory
praisonai memory search "Python" # Search memories
praisonai memory clear # Clear short-term memory
praisonai memory clear all # Clear all memory
praisonai memory save my_session # Save session
praisonai memory resume my_session # Resume session
praisonai memory sessions # List saved sessions
praisonai memory checkpoint # Create checkpoint
praisonai memory restore <checkpoint_id> # Restore checkpoint
praisonai memory checkpoints # List checkpoints
praisonai memory help # Show all commands
```
### Rules CLI:
```bash
# List all loaded rules (from PRAISON.md, CLAUDE.md, etc.)
praisonai rules list
# Show specific rule details
praisonai rules show <rule_name>
# Create a new rule
praisonai rules create my_rule "Always use type hints"
# Delete a rule
praisonai rules delete my_rule
# Show rules statistics
praisonai rules stats
# Include manual rules with prompts
praisonai "Task" --include-rules security,testing
```
### Workflow CLI:
```bash
# List available workflows
praisonai workflow list
# Execute a workflow
praisonai workflow run deploy
# Execute with variables
praisonai workflow run deploy --workflow-var environment=staging --workflow-var branch=main
# Show workflow details
praisonai workflow show deploy
# Create a new workflow template
praisonai workflow create my_workflow
```
### Hooks CLI:
```bash
# List configured hooks
praisonai hooks list
# Show hooks statistics
praisonai hooks stats
# Create hooks.json template
praisonai hooks init
```
### Claude Memory Tool CLI:
```bash
# Enable Claude Memory Tool (Anthropic models only)
praisonai "Research and remember findings" --claude-memory --llm anthropic/claude-sonnet-4-20250514
```
## Using JavaScript Code
```bash
@@ -1016,7 +1141,7 @@ cd src/praisonai && uv publish
- 🎨 Interactive UIs
- 📄 YAML-based Configuration
- 🛠️ Custom Tool Integration
- 🔍 Internet Search Capability (using Crawl4AI and Tavily)
- 🔍 Internet Search Capability (Tavily, You.com, Exa, DuckDuckGo, Crawl4AI)
- 🖼️ Vision Language Model (VLM) Support
- 🎙️ Real-time Voice Interaction
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
flask \
"praisonai>=2.3.5" \
"praisonai>=2.3.7" \
"praisonai[api]" \
gunicorn \
markdown
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.5" \
"praisonai>=2.3.7" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.5" \
"praisonai>=2.3.7" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.5" \
"praisonai>=2.3.7" \
"praisonai[ui]" \
"praisonai[crewai]"
+4 -15
View File
@@ -137,22 +137,11 @@ services:
POSTGRES_USER: praisonai
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-praisonai_dev_pass}
volumes:
database:
image: postgres:15-alpine
environment:
POSTGRES_DB: praisonai_dev
POSTGRES_USER: praisonai
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-praisonai_dev_pass}
volumes:
- postgres_data:/var/lib/postgresql/data
read_only: true
tmpfs:
- /tmp
- /var/tmp
ports:
- "5432:5432"
networks:
- postgres_data:/var/lib/postgresql/data
read_only: true
tmpfs:
- /tmp
- /var/tmp
ports:
- "5432:5432"
networks:
+1
View File
@@ -86,3 +86,4 @@ build
graph.py
chroma_db/
.qodo
*_COMPARISON_ANALYSIS.md
+41 -15
View File
@@ -86,7 +86,30 @@ result = agents.start()
### Planning Mode 🆕
Plan before execution, like Cursor, Windsurf, and Claude Code:
#### Single Agent Planning (Simplest)
Enable planning for any single agent:
```python
from praisonaiagents import Agent
def search_web(query: str) -> str:
return f"Search results for: {query}"
agent = Agent(
name="AI Assistant",
instructions="Research and write about topics",
planning=True, # Enable planning mode
planning_tools=[search_web], # Tools for planning research
planning_reasoning=True # Chain-of-thought reasoning
)
result = agent.start("Research AI trends in 2025 and write a summary")
```
#### Multi-Agent Planning
Plan before execution with multiple agents:
```python
from praisonaiagents import Agent, Task, PraisonAIAgents
@@ -97,20 +120,23 @@ writer = Agent(name="Writer", role="Content Writer")
research_task = Task(description="Research AI benefits", agent=researcher)
write_task = Task(description="Write summary", agent=writer)
# Enable planning mode
agents = PraisonAIAgents(
agents=[researcher, writer],
tasks=[research_task, write_task],
planning=True, # Enable planning
planning_llm="gpt-4o-mini", # Fast LLM for planning
auto_approve_plan=True # Auto-approve plans
planning_tools=[search_web], # Tools for planning research
planning_reasoning=True, # Chain-of-thought reasoning
auto_approve_plan=True # Auto-approve plans
)
result = agents.start()
```
**Features:**
- **Plan Creation**: LLM creates step-by-step implementation plans
- **Single Agent Planning**: `Agent(planning=True)` - simplest API
- **Multi-Agent Planning**: `PraisonAIAgents(planning=True)` - orchestrated
- **Planning Tools**: Research with tools before creating plans
- **Reasoning Mode**: Chain-of-thought planning with tool use
- **Todo Lists**: Auto-generated from plans with progress tracking
- **Read-Only Mode**: `plan_mode=True` restricts agents to safe tools
- **Approval Flow**: Review and approve plans before execution
@@ -248,7 +274,7 @@ router = RouterAgent(
## Memory
PraisonAI provides a comprehensive memory system with features inspired by Claude, Gemini CLI, Codex CLI, Cursor, and Windsurf.
PraisonAI provides a comprehensive memory system.
### Quick Start (Zero Dependencies)
@@ -293,7 +319,7 @@ agent = Agent(name="Memory Agent", memory=memory)
| **Episodic** | Date-based interaction memories | JSON files |
| **Graph** | Relationship-based memory | Mem0/Neo4j |
### Session Save/Resume (like Gemini CLI)
### Session Save/Resume
```python
from praisonaiagents.memory import FileMemory
@@ -314,7 +340,7 @@ session_data = memory.resume_session("ml_project")
sessions = memory.list_sessions()
```
### Context Compression (like Gemini CLI)
### Context Compression
```python
# Auto-compress when memory gets full
@@ -327,7 +353,7 @@ def llm_summarize(prompt):
summary = memory.compress(llm_func=llm_summarize, max_items=10)
```
### Checkpointing (like Gemini CLI)
### Checkpointing
```python
# Create checkpoint before risky operations
@@ -369,9 +395,9 @@ result = memory.handle_command("/memory help")
---
## Rules & Instructions (like Cursor/Windsurf)
## Rules & Instructions
PraisonAI automatically discovers and applies rules from multiple sources, similar to Cursor, Windsurf, Claude Code, and Codex CLI.
PraisonAI automatically discovers and applies rules from multiple sources.
### Supported Instruction Files
@@ -494,7 +520,7 @@ project/
└── global.md
```
### @Import Syntax (like Claude Code)
### @Import Syntax
Include other files in your rules:
@@ -505,7 +531,7 @@ See @docs/architecture.md for system design
@~/.praison/my-preferences.md
```
### Auto-Generated Memories (like Windsurf Cascade)
### Auto-Generated Memories
```python
from praisonaiagents.memory import FileMemory, AutoMemory
@@ -520,7 +546,7 @@ memories = auto.process_interaction(
# Extracts: name="John", preference="Python for backend"
```
### Workflows (like Windsurf)
### Workflows
Create reusable multi-step workflows:
@@ -537,7 +563,7 @@ result = manager.execute(
)
```
### Hooks (like Windsurf Cascade Hooks)
### Hooks
```python
from praisonaiagents.memory import HooksManager
@@ -239,7 +239,10 @@ class Agent:
web_fetch: Optional[Union[bool, Dict[str, Any]]] = None,
prompt_caching: Optional[bool] = None,
claude_memory: Optional[Union[bool, Any]] = None,
plan_mode: bool = False
plan_mode: bool = False,
planning: bool = False,
planning_tools: Optional[List[Any]] = None,
planning_reasoning: bool = False
):
"""Initialize an Agent instance.
@@ -526,6 +529,10 @@ Your Goal: {self.goal}
self.user_id = user_id or "praison"
self.reasoning_steps = reasoning_steps
self.plan_mode = plan_mode # Read-only mode for planning
self.planning = planning # Enable planning mode
self.planning_tools = planning_tools # Tools for planning phase
self.planning_reasoning = planning_reasoning # Enable reasoning during planning
self._planning_agent = None # Lazy loaded PlanningAgent
self.web_search = web_search
self.web_fetch = web_fetch
self.prompt_caching = prompt_caching
@@ -2492,10 +2499,88 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
def run(self, prompt: str, **kwargs):
"""Alias for start() method"""
return self.start(prompt, **kwargs)
return self.start(prompt, **kwargs)
def _get_planning_agent(self):
"""Lazy load PlanningAgent for planning mode."""
if self._planning_agent is None and self.planning:
from ..planning import PlanningAgent
self._planning_agent = PlanningAgent(
llm=self.llm if hasattr(self, 'llm') else (self.llm_instance.model if hasattr(self, 'llm_instance') else "gpt-4o-mini"),
tools=self.planning_tools,
reasoning=self.planning_reasoning,
verbose=1 if self.verbose else 0
)
return self._planning_agent
def _start_with_planning(self, prompt: str, **kwargs):
"""Execute with planning mode - creates plan then executes each step."""
from rich.console import Console
from rich.panel import Panel
from rich.markdown import Markdown
console = Console()
# Step 1: Create the plan
console.print("\n[bold blue]📋 PLANNING PHASE[/bold blue]")
console.print("[dim]Creating implementation plan...[/dim]\n")
planner = self._get_planning_agent()
plan = planner.create_plan_sync(request=prompt, agents=[self])
if not plan or not plan.steps:
console.print("[yellow]⚠️ Planning failed, falling back to direct execution[/yellow]")
return self.chat(prompt, **kwargs)
# Display the plan
console.print(Panel(
Markdown(plan.to_markdown()),
title="[bold green]Generated Plan[/bold green]",
border_style="green"
))
# Step 2: Execute each step
console.print("\n[bold blue]🚀 EXECUTION PHASE[/bold blue]\n")
results = []
context = ""
for i, step in enumerate(plan.steps):
progress = (i + 1) / len(plan.steps)
bar_length = 30
filled = int(bar_length * progress)
bar = "█" * filled + "░" * (bar_length - filled)
console.print(f"[dim]Progress: [{bar}] {progress * 100:.0f}%[/dim]")
console.print(f"\n[bold]📌 Step {i + 1}/{len(plan.steps)}:[/bold] {step.description[:60]}...")
# Build prompt with context from previous steps
step_prompt = step.description
if context:
step_prompt = f"{step.description}\n\nContext from previous steps:\n{context}"
# Execute the step
result = self.chat(step_prompt, **kwargs)
results.append(result)
# Update context for next step
context += f"\n\nStep {i + 1} result: {result[:500] if result else 'No result'}"
console.print(f" [green]✅ Completed[/green]")
console.print(f"\n[bold green]🎉 EXECUTION COMPLETE[/bold green]")
console.print(f"[dim]Progress: [{'█' * bar_length}] 100%[/dim]")
console.print(f"Completed {len(plan.steps)}/{len(plan.steps)} steps!\n")
# Return the final result
return results[-1] if results else None
def start(self, prompt: str, **kwargs):
"""Start the agent with a prompt. This is a convenience method that wraps chat()."""
# Check if planning mode is enabled
if self.planning:
return self._start_with_planning(prompt, **kwargs)
# Check if streaming is enabled (either from kwargs or agent's stream attribute)
stream_enabled = kwargs.get('stream', getattr(self, 'stream', False))
@@ -127,7 +127,7 @@ def process_task_context(context_item, verbose=0, user_id=None):
return str(context_item) # Fallback for unknown types
class PraisonAIAgents:
def __init__(self, agents, tasks=None, verbose=0, completion_checker=None, max_retries=5, process="sequential", manager_llm=None, memory=False, memory_config=None, embedder=None, user_id=None, max_iter=10, stream=True, name: Optional[str] = None, planning: bool = False, planning_llm: Optional[str] = None, auto_approve_plan: bool = False):
def __init__(self, agents, tasks=None, verbose=0, completion_checker=None, max_retries=5, process="sequential", manager_llm=None, memory=False, memory_config=None, embedder=None, user_id=None, max_iter=10, stream=True, name: Optional[str] = None, planning: bool = False, planning_llm: Optional[str] = None, auto_approve_plan: bool = False, planning_tools: Optional[List] = None, planning_reasoning: bool = False):
# Add check at the start if memory is requested
if memory:
try:
@@ -217,6 +217,8 @@ class PraisonAIAgents:
self.planning = planning
self.planning_llm = planning_llm or "gpt-4o-mini"
self.auto_approve_plan = auto_approve_plan
self.planning_tools = planning_tools
self.planning_reasoning = planning_reasoning
self._current_plan = None
self._todo_list = None
self._planning_agent = None
@@ -1566,7 +1568,9 @@ Context:
self._planning_agent = PlanningAgent(
llm=self.planning_llm,
read_only=True,
verbose=self.verbose
verbose=self.verbose,
tools=self.planning_tools,
reasoning=self.planning_reasoning
)
return self._planning_agent
@@ -73,6 +73,23 @@ RESTRICTED_TOOLS = [
"pip_install",
]
# Research tools that are safe for planning research
RESEARCH_TOOLS = [
"web_search",
"search_web",
"duckduckgo_search",
"tavily_search",
"brave_search",
"google_search",
"read_url",
"fetch_url",
"read_file",
"list_directory",
"search_codebase",
"grep_search",
"find_files",
]
__all__ = [
"Plan",
"PlanStep",
@@ -83,4 +100,5 @@ __all__ = [
"ApprovalCallback",
"READ_ONLY_TOOLS",
"RESTRICTED_TOOLS",
"RESEARCH_TOOLS",
]
@@ -139,7 +139,10 @@ Provide a concise analysis that will help in creating an implementation plan.'''
self,
llm: str = "gpt-4o-mini",
read_only: bool = True,
verbose: int = 0
verbose: int = 0,
tools: Optional[List] = None,
reasoning: bool = False,
max_reasoning_steps: int = 5
):
"""
Initialize PlanningAgent.
@@ -148,12 +151,19 @@ Provide a concise analysis that will help in creating an implementation plan.'''
llm: LLM model to use for planning
read_only: Whether to restrict to read-only tools
verbose: Verbosity level
tools: Optional list of tools for research during planning
reasoning: Whether to enable reasoning mode with chain-of-thought
max_reasoning_steps: Maximum reasoning iterations (default 5)
"""
self.llm_model = llm
self.read_only = read_only
self.verbose = verbose
self.tools = tools or []
self.reasoning = reasoning
self.max_reasoning_steps = max_reasoning_steps
self.allowed_tools = READ_ONLY_TOOLS.copy() if read_only else []
self._llm = None
self._agent = None
def _get_llm(self):
"""Lazy load LLM instance."""
@@ -162,6 +172,28 @@ Provide a concise analysis that will help in creating an implementation plan.'''
self._llm = LLM(model=self.llm_model)
return self._llm
def _get_agent(self):
"""
Lazy load Agent instance for tool-enabled planning.
Returns:
Agent instance if tools are provided, None otherwise
"""
if not self.tools:
return None
if self._agent is None:
from ..agent.agent import Agent
self._agent = Agent(
name="Planning Research Agent",
role="Research and Planning Specialist",
goal="Research information and create detailed implementation plans",
llm=self.llm_model,
tools=self.tools,
verbose=self.verbose > 0
)
return self._agent
def _format_agents_info(self, agents: List['Agent']) -> str:
"""
Format agent information for the prompt.
@@ -321,8 +353,6 @@ Provide a concise analysis that will help in creating an implementation plan.'''
Returns:
Plan instance with steps
"""
llm = self._get_llm()
# Format the prompt
prompt = self.PLANNING_PROMPT.format(
request=request,
@@ -333,9 +363,16 @@ Provide a concise analysis that will help in creating an implementation plan.'''
if self.verbose >= 2:
logger.info(f"Creating plan for: {request[:100]}...")
# Get LLM response
response = llm.get_response(prompt)
# Use Agent with tools if available, otherwise use bare LLM
agent = self._get_agent()
if agent is not None:
# Use Agent.chat() for tool-enabled planning
response = agent.chat(prompt)
else:
# Use bare LLM for simple planning
llm = self._get_llm()
response = llm.get_response(prompt)
# Parse response into Plan
plan = self._parse_plan_response(response)
@@ -176,6 +176,38 @@ TOOL_MAPPINGS = {
# Claude Memory Tool (Anthropic Beta)
'ClaudeMemoryTool': ('.claude_memory_tool', 'ClaudeMemoryTool'),
# Tavily Tools (requires TAVILY_API_KEY)
'tavily': ('.tavily_tools', None),
'tavily_search': ('.tavily_tools', None),
'tavily_extract': ('.tavily_tools', None),
'tavily_crawl': ('.tavily_tools', None),
'tavily_map': ('.tavily_tools', None),
'tavily_search_async': ('.tavily_tools', None),
'tavily_extract_async': ('.tavily_tools', None),
'TavilyTools': ('.tavily_tools', 'TavilyTools'),
'tavily_tools': ('.tavily_tools', None),
# You.com Tools (requires YDC_API_KEY)
'ydc': ('.youdotcom_tools', None),
'ydc_search': ('.youdotcom_tools', None),
'ydc_contents': ('.youdotcom_tools', None),
'ydc_news': ('.youdotcom_tools', None),
'ydc_images': ('.youdotcom_tools', None),
'YouTools': ('.youdotcom_tools', 'YouTools'),
'youdotcom_tools': ('.youdotcom_tools', None),
# Exa Tools (requires EXA_API_KEY)
'exa': ('.exa_tools', None),
'exa_search': ('.exa_tools', None),
'exa_search_contents': ('.exa_tools', None),
'exa_find_similar': ('.exa_tools', None),
'exa_answer': ('.exa_tools', None),
'exa_search_async': ('.exa_tools', None),
'exa_search_contents_async': ('.exa_tools', None),
'exa_answer_async': ('.exa_tools', None),
'ExaTools': ('.exa_tools', 'ExaTools'),
'exa_tools': ('.exa_tools', None),
}
_instances = {} # Cache for class instances
@@ -187,8 +219,8 @@ def __getattr__(name: str) -> Any:
module_path, class_name = TOOL_MAPPINGS[name]
# Return class itself (not instance) for ClaudeMemoryTool
if name == 'ClaudeMemoryTool':
# Return class itself (not instance) for ClaudeMemoryTool, TavilyTools, YouTools, ExaTools
if name in ('ClaudeMemoryTool', 'TavilyTools', 'YouTools', 'ExaTools'):
module = import_module(module_path, __package__)
return getattr(module, class_name)
@@ -205,11 +237,16 @@ def __getattr__(name: str) -> Any:
'insert_document', 'insert_documents', 'find_documents', 'update_document', 'delete_document',
'create_vector_index', 'vector_search', 'store_with_embedding', 'text_search', 'get_stats', 'connect_mongodb',
'execute_command', 'list_processes', 'kill_process', 'get_system_info',
'evaluate', 'solve_equation', 'convert_units', 'calculate_statistics', 'calculate_financial'
'evaluate', 'solve_equation', 'convert_units', 'calculate_statistics', 'calculate_financial',
'tavily', 'tavily_search', 'tavily_extract', 'tavily_crawl', 'tavily_map',
'tavily_search_async', 'tavily_extract_async',
'ydc', 'ydc_search', 'ydc_contents', 'ydc_news', 'ydc_images',
'exa', 'exa_search', 'exa_search_contents', 'exa_find_similar', 'exa_answer',
'exa_search_async', 'exa_search_contents_async', 'exa_answer_async'
]:
return getattr(module, name)
if name in ['file_tools', 'pandas_tools', 'wikipedia_tools',
'newspaper_tools', 'arxiv_tools', 'spider_tools', 'duckdb_tools', 'mongodb_tools', 'csv_tools', 'json_tools', 'excel_tools', 'xml_tools', 'yaml_tools', 'calculator_tools', 'python_tools', 'shell_tools', 'cot_tools']:
'newspaper_tools', 'arxiv_tools', 'spider_tools', 'duckdb_tools', 'mongodb_tools', 'csv_tools', 'json_tools', 'excel_tools', 'xml_tools', 'yaml_tools', 'calculator_tools', 'python_tools', 'shell_tools', 'cot_tools', 'tavily_tools', 'youdotcom_tools', 'exa_tools']:
return module # Returns the callable module
return getattr(module, name)
else:
@@ -0,0 +1,741 @@
"""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 praisonaiagents.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 praisonaiagents.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[bool, Optional[str]]:
"""Check if Exa is available and API key is set.
Returns:
Tuple of (is_available, error_message)
"""
# Check if exa_py package is installed
if util.find_spec("exa_py") is None:
return False, "exa_py package is not installed. Install it with: pip install exa_py"
# Check if API key is set
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 praisonaiagents.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
if __name__ == "__main__":
# Example usage
print("=" * 60)
print("Exa Tools - Example Usage")
print("=" * 60)
is_available, error = _check_exa_available()
if not is_available:
print(f"\nExa is not available: {error}")
print("\nTo use Exa tools:")
print("1. Install the package: pip install exa_py")
print("2. Set environment variable: export EXA_API_KEY=your_api_key")
else:
print("Exa SDK is available!")
# Example search
print("\n1. Basic Search")
print("-" * 30)
results = exa_search("Latest AI developments 2024", num_results=3)
if "error" not in results:
print(f"Found {len(results.get('results', []))} results")
for r in results.get("results", [])[:3]:
print(f" - {r.get('title', 'No title')[:50]}...")
print(f" URL: {r.get('url', 'No URL')}")
else:
print(f"Error: {results['error']}")
print("\n" + "=" * 60)
print("Demonstration Complete")
print("=" * 60)
@@ -0,0 +1,591 @@
"""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 praisonaiagents.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 praisonaiagents.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[bool, Optional[str]]:
"""Check if Tavily is available and API key is set.
Returns:
Tuple of (is_available, error_message)
"""
# Check if tavily package is installed
if util.find_spec("tavily") is None:
return False, "Tavily package is not installed. Please install it using: pip install tavily-python"
# Check if API key is set
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 methods
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,
)
if __name__ == "__main__":
# Example usage
print("\n==================================================")
print("Tavily Tools Demonstration")
print("==================================================\n")
# Check if API key is available
is_available, error = _check_tavily_available()
if not is_available:
print(f"Error: {error}")
print("\nTo use Tavily tools:")
print("1. Install: pip install tavily-python")
print("2. Set environment variable: export TAVILY_API_KEY=your_api_key")
else:
print("Tavily is available!")
# Example search
print("\n1. Basic Search")
print("-" * 30)
results = tavily_search("Latest AI developments 2024", max_results=3)
if "error" not in results:
print(f"Found {len(results.get('results', []))} results")
for r in results.get("results", [])[:3]:
print(f" - {r.get('title', 'No title')}")
print(f" URL: {r.get('url', 'No URL')}")
else:
print(f"Error: {results['error']}")
print("\n==================================================")
print("Demonstration Complete")
print("==================================================")
# Alias for simple usage: from praisonaiagents.tools import tavily
tavily = tavily_search
@@ -0,0 +1,533 @@
"""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 praisonaiagents.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 praisonaiagents.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[bool, Optional[str]]:
"""Check if You.com SDK is available and API key is set.
Returns:
Tuple of (is_available, error_message)
"""
# Check if youdotcom package is installed
if util.find_spec("youdotcom") is None:
return False, "youdotcom package is not installed. Please install it using: pip install youdotcom"
# Check if API key is set
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)
# Convert response to dict if it's a model object
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]
# Use the contents endpoint
response = client.search.contents(urls=urls, format=format)
# Convert response
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
# Try to access live news endpoint
if hasattr(client, 'livenews'):
response = client.livenews.get(**kwargs)
else:
# Fallback: use unified search with news focus
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:
# Close client if it has a close method
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)
if __name__ == "__main__":
# Example usage
print("\n" + "=" * 60)
print("You.com Tools Demonstration")
print("=" * 60 + "\n")
# Check if API key is available
is_available, error = _check_youdotcom_available()
if not is_available:
print(f"Error: {error}")
print("\nTo use You.com tools:")
print("1. Install: pip install youdotcom")
print("2. Set environment variable: export YDC_API_KEY=your_api_key")
else:
print("You.com SDK is available!")
# Example search
print("\n1. Basic Search")
print("-" * 30)
results = ydc_search("Latest AI developments 2024", count=3)
if "error" not in results:
web_results = results.get('results', {}).get('web', [])
print(f"Found {len(web_results)} web results")
for r in web_results[:3]:
print(f" - {r.get('title', 'No title')[:50]}...")
print(f" URL: {r.get('url', 'No URL')}")
else:
print(f"Error: {results['error']}")
print("\n" + "=" * 60)
print("Demonstration Complete")
print("=" * 60)
# Alias for simple usage: from praisonaiagents.tools import ydc
ydc = ydc_search
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.0.183"
version = "0.0.185"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
requires-python = ">=3.10"
authors = [
@@ -0,0 +1,16 @@
from praisonaiagents import Agent
def search_web(query: str) -> str:
return "AI trends: LLMs, multimodal AI, autonomous agents"
agent = Agent(
name="AI Assistant",
role="Research and Writing Specialist",
llm="gpt-4o-mini",
planning=True,
planning_tools=[search_web],
planning_reasoning=True
)
result = agent.start("Research AI trends in 2025 and write a 2 sentence summary")
print(result)
@@ -0,0 +1,18 @@
from praisonaiagents import Agent
def search_web(query: str) -> str:
return "AI trends: LLMs, multimodal AI, autonomous agents"
agent = Agent(
name="AI Assistant",
role="Research and Writing Specialist",
llm="gpt-4o-mini",
planning=True,
planning_tools=[search_web],
planning_reasoning=True,
verbose=False # Quiet mode
)
result = agent.start("Research AI trends in 2025 and write a 2 sentence summary")
print("\n=== FINAL RESULT ===")
print(result)
@@ -0,0 +1,62 @@
from praisonaiagents import Agent
def search_web(query: str) -> str:
return "AI trends: LLMs, multimodal AI, autonomous agents, edge AI"
print("=" * 60)
print("TEST 1: Planning ONLY (no tools, no reasoning)")
print("=" * 60)
agent1 = Agent(
name="Assistant",
llm="gpt-4o-mini",
planning=True,
verbose=False
)
result1 = agent1.start("Write 2 sentences about AI in healthcare")
print(f"\nResult: {result1[:200]}...")
print("\n" + "=" * 60)
print("TEST 2: Planning + Tools (no reasoning)")
print("=" * 60)
agent2 = Agent(
name="Assistant",
llm="gpt-4o-mini",
planning=True,
planning_tools=[search_web],
verbose=False
)
result2 = agent2.start("Write 2 sentences about AI trends")
print(f"\nResult: {result2[:200]}...")
print("\n" + "=" * 60)
print("TEST 3: Planning + Tools + Reasoning")
print("=" * 60)
agent3 = Agent(
name="Assistant",
llm="gpt-4o-mini",
planning=True,
planning_tools=[search_web],
planning_reasoning=True,
verbose=False
)
result3 = agent3.start("Write 2 sentences about AI trends")
print(f"\nResult: {result3[:200]}...")
print("\n" + "=" * 60)
print("TEST 4: No Planning (direct execution)")
print("=" * 60)
agent4 = Agent(
name="Assistant",
llm="gpt-4o-mini",
verbose=False
)
result4 = agent4.start("Write 2 sentences about AI trends")
print(f"\nResult: {result4[:200]}...")
print("\n" + "=" * 60)
print("COMPARISON COMPLETE")
print("=" * 60)
@@ -0,0 +1,22 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
def search_web(query: str) -> str:
return f"Results: AI trends include LLMs, multimodal AI, autonomous agents"
researcher = Agent(name="Researcher", role="Research Analyst", llm="gpt-4o-mini")
writer = Agent(name="Writer", role="Content Writer", llm="gpt-4o-mini")
task1 = Task(description="Research AI trends in 2025", agent=researcher)
task2 = Task(description="Write a 2 sentence summary", agent=writer)
agents = PraisonAIAgents(
agents=[researcher, writer],
tasks=[task1, task2],
planning=True,
planning_tools=[search_web],
planning_reasoning=True,
auto_approve_plan=True
)
result = agents.start()
print(result)
@@ -0,0 +1,28 @@
from praisonaiagents import Agent, Task, PraisonAIAgents
def search_web(query: str) -> str:
return "AI trends: LLMs, multimodal AI, autonomous agents"
agent = Agent(
name="AI Assistant",
role="Research and Writing Specialist",
llm="gpt-4o-mini",
instructions="Research AI trends in 2025 and write a 2 sentence summary"
)
task = Task(
description="Research AI trends in 2025 and write a 2 sentence summary",
agent=agent
)
agents = PraisonAIAgents(
agents=[agent],
tasks=[task],
planning=True,
planning_tools=[search_web],
planning_reasoning=True,
auto_approve_plan=True
)
result = agents.start()
print(result)
@@ -0,0 +1,34 @@
from praisonaiagents import Agent
from praisonaiagents.planning import PlanningAgent
def search_web(query: str) -> str:
return f"Results for '{query}': LLMs, multimodal AI, autonomous agents"
researcher = Agent(name="Researcher", role="Research Analyst", llm="gpt-4o-mini")
writer = Agent(name="Writer", role="Content Writer", llm="gpt-4o-mini")
print("\n=== TEST 1: Basic Planning (NO tools) ===")
planner1 = PlanningAgent(llm="gpt-4o-mini")
plan1 = planner1.create_plan_sync(
request="Write about AI trends 2025",
agents=[researcher, writer]
)
print(f"Plan: {plan1.name}, Steps: {len(plan1.steps)}")
print("\n=== TEST 2: Planning WITH tools ===")
planner2 = PlanningAgent(llm="gpt-4o-mini", tools=[search_web])
plan2 = planner2.create_plan_sync(
request="Write about AI trends 2025",
agents=[researcher, writer]
)
print(f"Plan: {plan2.name}, Steps: {len(plan2.steps)}")
print("\n=== TEST 3: Planning WITH tools + reasoning ===")
planner3 = PlanningAgent(llm="gpt-4o-mini", tools=[search_web], reasoning=True)
plan3 = planner3.create_plan_sync(
request="Write about AI trends 2025",
agents=[researcher, writer]
)
print(f"Plan: {plan3.name}, Steps: {len(plan3.steps)}")
print("\n✅ Done!")
@@ -0,0 +1,407 @@
"""
Test-Driven Development tests for Planning Mode with Tools Support.
This module contains tests for:
- PlanningAgent with tools parameter
- PlanningAgent with reasoning mode
- PraisonAIAgents with planning_tools parameter
- Research-enabled planning
"""
import pytest
import json
from unittest.mock import Mock, MagicMock, patch, AsyncMock
# =============================================================================
# SECTION 1: PlanningAgent Tools Parameter Tests
# =============================================================================
class TestPlanningAgentToolsParameter:
"""Tests for PlanningAgent tools parameter."""
def test_planning_agent_accepts_tools_parameter(self):
"""Test that PlanningAgent accepts tools parameter."""
from praisonaiagents.planning import PlanningAgent
def search_tool(query: str) -> str:
"""Search the web."""
return f"Results for {query}"
planner = PlanningAgent(tools=[search_tool])
assert planner.tools is not None
assert len(planner.tools) == 1
def test_planning_agent_tools_default_empty(self):
"""Test that tools defaults to empty list."""
from praisonaiagents.planning import PlanningAgent
planner = PlanningAgent()
assert planner.tools == []
def test_planning_agent_with_multiple_tools(self):
"""Test PlanningAgent with multiple tools."""
from praisonaiagents.planning import PlanningAgent
def tool1(x: str) -> str:
return x
def tool2(x: str) -> str:
return x
def tool3(x: str) -> str:
return x
planner = PlanningAgent(tools=[tool1, tool2, tool3])
assert len(planner.tools) == 3
# =============================================================================
# SECTION 2: PlanningAgent Internal Agent Tests
# =============================================================================
class TestPlanningAgentInternalAgent:
"""Tests for PlanningAgent internal Agent creation."""
def test_planning_agent_creates_internal_agent_when_tools_provided(self):
"""Test that internal Agent is created when tools are provided."""
from praisonaiagents.planning import PlanningAgent
def search_tool(query: str) -> str:
"""Search the web."""
return f"Results for {query}"
planner = PlanningAgent(tools=[search_tool])
# Internal agent should be created lazily
agent = planner._get_agent()
assert agent is not None
assert agent.tools is not None
def test_planning_agent_no_internal_agent_without_tools(self):
"""Test that no internal Agent is created without tools."""
from praisonaiagents.planning import PlanningAgent
planner = PlanningAgent()
agent = planner._get_agent()
assert agent is None
def test_planning_agent_internal_agent_uses_planning_llm(self):
"""Test that internal Agent uses the planning LLM."""
from praisonaiagents.planning import PlanningAgent
def search_tool(query: str) -> str:
return "results"
planner = PlanningAgent(llm="gpt-4o", tools=[search_tool])
agent = planner._get_agent()
assert agent is not None
# The agent should use the same LLM model
assert agent.llm == "gpt-4o"
# =============================================================================
# SECTION 3: PlanningAgent Create Plan with Tools Tests
# =============================================================================
class TestPlanningAgentCreatePlanWithTools:
"""Tests for create_plan_sync with tools."""
def test_create_plan_sync_uses_agent_when_tools_provided(self):
"""Test that create_plan_sync uses Agent.chat when tools provided."""
from praisonaiagents.planning import PlanningAgent
from praisonaiagents import Agent
def search_tool(query: str) -> str:
"""Search the web for information."""
return f"Search results for: {query}"
planner = PlanningAgent(tools=[search_tool])
# Mock the internal agent's chat method
mock_agent = MagicMock()
mock_agent.chat = MagicMock(return_value=json.dumps({
"name": "Research Plan",
"description": "Plan created with research",
"steps": [
{"description": "Research topic", "agent": "Researcher"},
{"description": "Write content", "agent": "Writer"}
]
}))
mock_agent.tools = [search_tool]
mock_agent.llm = "gpt-4o-mini"
planner._agent = mock_agent
agents = [Agent(name="Researcher", role="Researcher")]
plan = planner.create_plan_sync(
request="Research AI trends",
agents=agents
)
# Verify agent.chat was called (not LLM.get_response)
mock_agent.chat.assert_called_once()
assert plan.name == "Research Plan"
def test_create_plan_sync_uses_llm_when_no_tools(self):
"""Test that create_plan_sync uses LLM when no tools provided."""
from praisonaiagents.planning import PlanningAgent
from praisonaiagents import Agent
planner = PlanningAgent()
# Mock the LLM
mock_llm = MagicMock()
mock_llm.get_response = MagicMock(return_value=json.dumps({
"name": "Simple Plan",
"description": "Plan without research",
"steps": [{"description": "Step 1"}]
}))
planner._llm = mock_llm
agents = [Agent(name="Worker", role="Worker")]
plan = planner.create_plan_sync(
request="Simple task",
agents=agents
)
# Verify LLM.get_response was called
mock_llm.get_response.assert_called_once()
assert plan.name == "Simple Plan"
# =============================================================================
# SECTION 4: PraisonAIAgents planning_tools Parameter Tests
# =============================================================================
class TestPraisonAIAgentsPlanningTools:
"""Tests for PraisonAIAgents planning_tools parameter."""
def test_praisonaiagents_accepts_planning_tools(self):
"""Test that PraisonAIAgents accepts planning_tools parameter."""
from praisonaiagents import Agent, Task, PraisonAIAgents
def search_tool(query: str) -> str:
return "results"
agent = Agent(name="Test", role="Tester")
task = Task(description="Test task", agent=agent)
agents = PraisonAIAgents(
agents=[agent],
tasks=[task],
planning=True,
planning_tools=[search_tool]
)
assert agents.planning_tools is not None
assert len(agents.planning_tools) == 1
def test_praisonaiagents_planning_tools_default_none(self):
"""Test that planning_tools defaults to None."""
from praisonaiagents import Agent, Task, PraisonAIAgents
agent = Agent(name="Test", role="Tester")
task = Task(description="Test task", agent=agent)
agents = PraisonAIAgents(
agents=[agent],
tasks=[task],
planning=True
)
assert agents.planning_tools is None
def test_praisonaiagents_passes_tools_to_planning_agent(self):
"""Test that planning_tools are passed to PlanningAgent."""
from praisonaiagents import Agent, Task, PraisonAIAgents
def search_tool(query: str) -> str:
return "results"
agent = Agent(name="Test", role="Tester")
task = Task(description="Test task", agent=agent)
agents = PraisonAIAgents(
agents=[agent],
tasks=[task],
planning=True,
planning_tools=[search_tool]
)
# Get the planning agent
planning_agent = agents._get_planning_agent()
assert planning_agent is not None
assert planning_agent.tools is not None
assert len(planning_agent.tools) == 1
# =============================================================================
# SECTION 5: PlanningAgent Reasoning Mode Tests
# =============================================================================
class TestPlanningAgentReasoningMode:
"""Tests for PlanningAgent reasoning mode."""
def test_planning_agent_accepts_reasoning_parameter(self):
"""Test that PlanningAgent accepts reasoning parameter."""
from praisonaiagents.planning import PlanningAgent
planner = PlanningAgent(reasoning=True)
assert planner.reasoning is True
def test_planning_agent_reasoning_default_false(self):
"""Test that reasoning defaults to False."""
from praisonaiagents.planning import PlanningAgent
planner = PlanningAgent()
assert planner.reasoning is False
def test_planning_agent_accepts_max_reasoning_steps(self):
"""Test that PlanningAgent accepts max_reasoning_steps parameter."""
from praisonaiagents.planning import PlanningAgent
planner = PlanningAgent(reasoning=True, max_reasoning_steps=10)
assert planner.max_reasoning_steps == 10
def test_planning_agent_max_reasoning_steps_default(self):
"""Test that max_reasoning_steps defaults to 5."""
from praisonaiagents.planning import PlanningAgent
planner = PlanningAgent(reasoning=True)
assert planner.max_reasoning_steps == 5
# =============================================================================
# SECTION 6: PraisonAIAgents planning_reasoning Parameter Tests
# =============================================================================
class TestPraisonAIAgentsPlanningReasoning:
"""Tests for PraisonAIAgents planning_reasoning parameter."""
def test_praisonaiagents_accepts_planning_reasoning(self):
"""Test that PraisonAIAgents accepts planning_reasoning parameter."""
from praisonaiagents import Agent, Task, PraisonAIAgents
agent = Agent(name="Test", role="Tester")
task = Task(description="Test task", agent=agent)
agents = PraisonAIAgents(
agents=[agent],
tasks=[task],
planning=True,
planning_reasoning=True
)
assert agents.planning_reasoning is True
def test_praisonaiagents_planning_reasoning_default_false(self):
"""Test that planning_reasoning defaults to False."""
from praisonaiagents import Agent, Task, PraisonAIAgents
agent = Agent(name="Test", role="Tester")
task = Task(description="Test task", agent=agent)
agents = PraisonAIAgents(
agents=[agent],
tasks=[task],
planning=True
)
assert agents.planning_reasoning is False
def test_praisonaiagents_passes_reasoning_to_planning_agent(self):
"""Test that planning_reasoning is passed to PlanningAgent."""
from praisonaiagents import Agent, Task, PraisonAIAgents
agent = Agent(name="Test", role="Tester")
task = Task(description="Test task", agent=agent)
agents = PraisonAIAgents(
agents=[agent],
tasks=[task],
planning=True,
planning_reasoning=True
)
planning_agent = agents._get_planning_agent()
assert planning_agent.reasoning is True
# =============================================================================
# SECTION 7: Integration Tests
# =============================================================================
class TestPlanningToolsIntegration:
"""Integration tests for planning with tools."""
def test_planning_with_tools_and_reasoning(self):
"""Test planning with both tools and reasoning enabled."""
from praisonaiagents import Agent, Task, PraisonAIAgents
def search_tool(query: str) -> str:
return "search results"
agent = Agent(name="Researcher", role="Research Analyst")
task = Task(description="Research AI trends", agent=agent)
agents = PraisonAIAgents(
agents=[agent],
tasks=[task],
planning=True,
planning_tools=[search_tool],
planning_reasoning=True,
auto_approve_plan=True
)
assert agents.planning is True
assert agents.planning_tools is not None
assert agents.planning_reasoning is True
planning_agent = agents._get_planning_agent()
assert planning_agent.tools is not None
assert planning_agent.reasoning is True
# =============================================================================
# SECTION 8: RESEARCH_TOOLS List Tests
# =============================================================================
class TestResearchToolsList:
"""Tests for RESEARCH_TOOLS list in planning module."""
def test_research_tools_list_exists(self):
"""Test that RESEARCH_TOOLS list exists."""
from praisonaiagents.planning import RESEARCH_TOOLS
assert RESEARCH_TOOLS is not None
assert isinstance(RESEARCH_TOOLS, list)
def test_research_tools_contains_search_tools(self):
"""Test that RESEARCH_TOOLS contains common search tools."""
from praisonaiagents.planning import RESEARCH_TOOLS
expected_tools = [
"web_search",
"search_web",
"duckduckgo_search",
]
for tool in expected_tools:
assert tool in RESEARCH_TOOLS, f"{tool} should be in RESEARCH_TOOLS"
if __name__ == "__main__":
pytest.main([__file__, "-v"])
@@ -0,0 +1,142 @@
"""Test Tavily tools integration."""
import os
import sys
# Test 1: Import test - verify lazy loading works
print("=" * 60)
print("Test 1: Import Test (Lazy Loading)")
print("=" * 60)
try:
from praisonaiagents.tools import tavily_search, tavily_extract, tavily_crawl, tavily_map
print("✓ Successfully imported tavily_search, tavily_extract, tavily_crawl, tavily_map")
except ImportError as e:
print(f"✗ Import failed: {e}")
sys.exit(1)
try:
from praisonaiagents.tools import TavilyTools
print("✓ Successfully imported TavilyTools class")
except ImportError as e:
print(f"✗ TavilyTools import failed: {e}")
sys.exit(1)
try:
from praisonaiagents.tools import tavily_search_async, tavily_extract_async
print("✓ Successfully imported async functions")
except ImportError as e:
print(f"✗ Async import failed: {e}")
sys.exit(1)
# Test 2: Check API key detection
print("\n" + "=" * 60)
print("Test 2: API Key Detection")
print("=" * 60)
api_key = os.environ.get("TAVILY_API_KEY")
if api_key:
print(f"✓ TAVILY_API_KEY is set (length: {len(api_key)})")
else:
print("✗ TAVILY_API_KEY is not set")
print(" To run full tests, set: export TAVILY_API_KEY=your_api_key")
# Test 3: Error handling without API key
print("\n" + "=" * 60)
print("Test 3: Error Handling (No API Key)")
print("=" * 60)
# Temporarily unset API key to test error handling
original_key = os.environ.pop("TAVILY_API_KEY", None)
try:
result = tavily_search("test query")
if "error" in result:
print(f"✓ Proper error returned when no API key: {result['error'][:50]}...")
else:
print("✗ Expected error but got results")
except Exception as e:
print(f"✗ Unexpected exception: {e}")
# Restore API key
if original_key:
os.environ["TAVILY_API_KEY"] = original_key
# Test 4: TavilyTools class instantiation
print("\n" + "=" * 60)
print("Test 4: TavilyTools Class")
print("=" * 60)
try:
tools = TavilyTools()
print("✓ TavilyTools instantiated successfully")
# Check methods exist
methods = ['search', 'extract', 'crawl', 'map', 'asearch', 'aextract']
for method in methods:
if hasattr(tools, method):
print(f" ✓ Method '{method}' exists")
else:
print(f" ✗ Method '{method}' missing")
except Exception as e:
print(f"✗ TavilyTools instantiation failed: {e}")
# Test 5: Full integration test (only if API key is available)
print("\n" + "=" * 60)
print("Test 5: Full Integration Test")
print("=" * 60)
if os.environ.get("TAVILY_API_KEY"):
print("Running search test...")
try:
result = tavily_search("Python programming", max_results=2)
if "error" not in result:
print(f"✓ Search successful!")
print(f" Query: {result.get('query', 'N/A')}")
print(f" Results: {len(result.get('results', []))} items")
if result.get('results'):
print(f" First result: {result['results'][0].get('title', 'N/A')[:50]}...")
else:
print(f"✗ Search returned error: {result['error']}")
except Exception as e:
print(f"✗ Search failed with exception: {e}")
else:
print("⊘ Skipped (TAVILY_API_KEY not set)")
# Test 6: Agent integration test
print("\n" + "=" * 60)
print("Test 6: Agent Integration Test")
print("=" * 60)
if os.environ.get("TAVILY_API_KEY") and os.environ.get("OPENAI_API_KEY"):
try:
from praisonaiagents import Agent
agent = Agent(
name="SearchAgent",
role="Web Researcher",
goal="Search the web for information",
tools=[tavily_search]
)
print("✓ Agent created with tavily_search tool")
# Check tool is properly registered
if agent.tools and tavily_search in agent.tools:
print("✓ tavily_search is in agent's tools list")
else:
print("⊘ Tool registration check inconclusive")
except Exception as e:
print(f"✗ Agent integration failed: {e}")
else:
missing = []
if not os.environ.get("TAVILY_API_KEY"):
missing.append("TAVILY_API_KEY")
if not os.environ.get("OPENAI_API_KEY"):
missing.append("OPENAI_API_KEY")
print(f"⊘ Skipped (missing: {', '.join(missing)})")
print("\n" + "=" * 60)
print("Test Summary")
print("=" * 60)
print("All import and basic tests passed!")
print("For full functionality, ensure TAVILY_API_KEY is set.")
@@ -0,0 +1,135 @@
"""Test You.com tools integration."""
import os
import sys
# Test 1: Import test - verify lazy loading works
print("=" * 60)
print("Test 1: Import Test (Lazy Loading)")
print("=" * 60)
try:
from praisonaiagents.tools import ydc_search, ydc_contents, ydc_news, ydc_images
print("✓ Successfully imported ydc_search, ydc_contents, ydc_news, ydc_images")
except ImportError as e:
print(f"✗ Import failed: {e}")
sys.exit(1)
try:
from praisonaiagents.tools import YouTools
print("✓ Successfully imported YouTools class")
except ImportError as e:
print(f"✗ YouTools import failed: {e}")
sys.exit(1)
# Test 2: Check API key detection
print("\n" + "=" * 60)
print("Test 2: API Key Detection")
print("=" * 60)
api_key = os.environ.get("YDC_API_KEY")
if api_key:
print(f"✓ YDC_API_KEY is set (length: {len(api_key)})")
else:
print("✗ YDC_API_KEY is not set")
print(" To run full tests, set: export YDC_API_KEY=your_api_key")
# Test 3: Error handling without API key
print("\n" + "=" * 60)
print("Test 3: Error Handling (No API Key)")
print("=" * 60)
# Temporarily unset API key to test error handling
original_key = os.environ.pop("YDC_API_KEY", None)
try:
result = ydc_search("test query")
if "error" in result:
print(f"✓ Proper error returned when no API key: {result['error'][:50]}...")
else:
print("✗ Expected error but got results")
except Exception as e:
print(f"✗ Unexpected exception: {e}")
# Restore API key
if original_key:
os.environ["YDC_API_KEY"] = original_key
# Test 4: YouTools class instantiation
print("\n" + "=" * 60)
print("Test 4: YouTools Class")
print("=" * 60)
try:
tools = YouTools()
print("✓ YouTools instantiated successfully")
# Check methods exist
methods = ['search', 'get_contents', 'live_news', 'images']
for method in methods:
if hasattr(tools, method):
print(f" ✓ Method '{method}' exists")
else:
print(f" ✗ Method '{method}' missing")
except Exception as e:
print(f"✗ YouTools instantiation failed: {e}")
# Test 5: Full integration test (only if API key is available)
print("\n" + "=" * 60)
print("Test 5: Full Integration Test")
print("=" * 60)
if os.environ.get("YDC_API_KEY"):
print("Running search test...")
try:
result = ydc_search("Python programming", count=2)
if "error" not in result:
print(f"✓ Search successful!")
web_results = result.get('results', {}).get('web', [])
print(f" Web results: {len(web_results)} items")
if web_results:
print(f" First result: {web_results[0].get('title', 'N/A')[:50]}...")
else:
print(f"✗ Search returned error: {result['error']}")
except Exception as e:
print(f"✗ Search failed with exception: {e}")
else:
print("⊘ Skipped (YDC_API_KEY not set)")
# Test 6: Agent integration test
print("\n" + "=" * 60)
print("Test 6: Agent Integration Test")
print("=" * 60)
if os.environ.get("YDC_API_KEY") and os.environ.get("OPENAI_API_KEY"):
try:
from praisonaiagents import Agent
agent = Agent(
name="SearchAgent",
role="Web Researcher",
goal="Search the web for information",
tools=[ydc_search]
)
print("✓ Agent created with ydc_search tool")
# Check tool is properly registered
if agent.tools and ydc_search in agent.tools:
print("✓ ydc_search is in agent's tools list")
else:
print("⊘ Tool registration check inconclusive")
except Exception as e:
print(f"✗ Agent integration failed: {e}")
else:
missing = []
if not os.environ.get("YDC_API_KEY"):
missing.append("YDC_API_KEY")
if not os.environ.get("OPENAI_API_KEY"):
missing.append("OPENAI_API_KEY")
print(f"⊘ Skipped (missing: {', '.join(missing)})")
print("\n" + "=" * 60)
print("Test Summary")
print("=" * 60)
print("All import and basic tests passed!")
print("For full functionality, ensure YDC_API_KEY is set.")
+1 -1
View File
@@ -2422,7 +2422,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.0.183"
version = "0.0.185"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
+137 -12
View File
@@ -60,7 +60,10 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
- 💾 **Prompt Caching** (Reduce costs & latency - OpenAI, Anthropic, Bedrock, Deepseek)
- 🧠 **Claude Memory Tool** (Persistent cross-conversation memory - Anthropic Beta)
- 💾 **File-Based Memory** (Zero-dependency persistent memory for all agents)
- 📋 **Planning Mode** (Plan before execution like Cursor, Windsurf, Claude Code)
- 🔍 **Built-in Search Tools** (Tavily, You.com, Exa - web search, news, content extraction)
- 📋 **Planning Mode** (Plan before execution - Agent & Multi-Agent)
- 🔧 **Planning Tools** (Research with tools during planning)
- 🧠 **Planning Reasoning** (Chain-of-thought planning)
## Using Python Code
@@ -104,7 +107,34 @@ Run:
python app.py
```
### 3. Deep Research Agent
### 3. Agent with Planning Mode
Enable planning for any agent - the agent creates a plan, then executes step by step:
```python
from praisonaiagents import Agent
def search_web(query: str) -> str:
return f"Search results for: {query}"
agent = Agent(
name="AI Assistant",
instructions="Research and write about topics",
planning=True, # Enable planning mode
planning_tools=[search_web], # Tools for planning research
planning_reasoning=True # Chain-of-thought reasoning
)
result = agent.start("Research AI trends in 2025 and write a summary")
```
**What happens:**
1. 📋 Agent creates a multi-step plan
2. 🚀 Executes each step sequentially
3. 📊 Shows progress with context passing
4. ✅ Returns final result
### 4. Deep Research Agent
Automated research with real-time streaming, web search, and citations using OpenAI or Gemini Deep Research APIs.
@@ -142,7 +172,7 @@ print(result.report)
- 🛠️ Built-in tools: web search, code interpreter, MCP, file search
- 🔄 Automatic provider detection from model name
### 4. Query Rewriter Agent
### 5. Query Rewriter Agent
Transform user queries to improve RAG retrieval quality using multiple strategies.
@@ -177,9 +207,9 @@ result = agent.rewrite("What about cost?", chat_history=[...])
- **CONTEXTUAL**: Resolve references using conversation history
- **AUTO**: Automatically detect best strategy
### 5. Agent Memory (Zero Dependencies)
### 6. Agent Memory (Zero Dependencies)
Enable persistent memory for agents - works out of the box without any extra packages. Features inspired by Claude, Gemini CLI, Codex CLI, Cursor, and Windsurf.
Enable persistent memory for agents - works out of the box without any extra packages.
```python
from praisonaiagents import Agent
@@ -204,7 +234,7 @@ result = agent.start("My name is John and I prefer Python")
- **Entity**: People, places, organizations with attributes
- **Episodic**: Date-based interaction history
**Advanced Features (like Gemini CLI):**
**Advanced Features:**
```python
from praisonaiagents.memory import FileMemory
@@ -234,7 +264,7 @@ memory.handle_command("/memory save my_session")
| `memory="sqlite"` | Built-in | SQLite with indexing |
| `memory="chromadb"` | chromadb | Vector/semantic search |
### 6. Rules & Instructions (like Cursor/Windsurf)
### 6. Rules & Instructions
PraisonAI auto-discovers instruction files from your project root and git root:
@@ -262,7 +292,7 @@ agent = Agent(name="Assistant", instructions="You are helpful.")
# Rules are injected into system prompt automatically
```
**@Import Syntax (like Claude Code):**
**@Import Syntax:**
```markdown
# CLAUDE.md
See @README for project overview
@@ -283,7 +313,7 @@ activation: always # always, glob, manual, ai_decision
- Follow PEP 8
```
### 7. Auto-Generated Memories (like Windsurf Cascade)
### 7. Auto-Generated Memories
```python
from praisonaiagents.memory import FileMemory, AutoMemory
@@ -298,7 +328,7 @@ memories = auto.process_interaction(
# Extracts: name="John", preference="Python for backend"
```
### 8. Workflows (like Windsurf)
### 8. Workflows
Create reusable multi-step workflows in `.praison/workflows/`:
@@ -315,7 +345,7 @@ result = manager.execute(
)
```
### 9. Hooks (like Windsurf Cascade Hooks)
### 9. Hooks
Configure in `.praison/hooks.json`:
@@ -385,6 +415,101 @@ praisonai research --query-rewrite --tools tools.py --save "Your research query"
praisonai research -v "Your research query"
```
### Planning Mode CLI:
```bash
# Enable planning mode - agent creates a plan before execution
praisonai "Research AI trends and write a summary" --planning
# Planning with tools for research
praisonai "Analyze market trends" --planning --planning-tools tools.py
# Planning with chain-of-thought reasoning
praisonai "Complex analysis task" --planning --planning-reasoning
# Auto-approve plans without confirmation
praisonai "Task" --planning --auto-approve-plan
```
### Memory CLI:
```bash
# Enable memory for agent (persists across sessions)
praisonai "My name is John" --memory
# Memory with user isolation
praisonai "Remember my preferences" --memory --user-id user123
# Memory management commands
praisonai memory show # Show memory statistics
praisonai memory add "User prefers Python" # Add to long-term memory
praisonai memory search "Python" # Search memories
praisonai memory clear # Clear short-term memory
praisonai memory clear all # Clear all memory
praisonai memory save my_session # Save session
praisonai memory resume my_session # Resume session
praisonai memory sessions # List saved sessions
praisonai memory checkpoint # Create checkpoint
praisonai memory restore <checkpoint_id> # Restore checkpoint
praisonai memory checkpoints # List checkpoints
praisonai memory help # Show all commands
```
### Rules CLI:
```bash
# List all loaded rules (from PRAISON.md, CLAUDE.md, etc.)
praisonai rules list
# Show specific rule details
praisonai rules show <rule_name>
# Create a new rule
praisonai rules create my_rule "Always use type hints"
# Delete a rule
praisonai rules delete my_rule
# Show rules statistics
praisonai rules stats
# Include manual rules with prompts
praisonai "Task" --include-rules security,testing
```
### Workflow CLI:
```bash
# List available workflows
praisonai workflow list
# Execute a workflow
praisonai workflow run deploy
# Execute with variables
praisonai workflow run deploy --workflow-var environment=staging --workflow-var branch=main
# Show workflow details
praisonai workflow show deploy
# Create a new workflow template
praisonai workflow create my_workflow
```
### Hooks CLI:
```bash
# List configured hooks
praisonai hooks list
# Show hooks statistics
praisonai hooks stats
# Create hooks.json template
praisonai hooks init
```
### Claude Memory Tool CLI:
```bash
# Enable Claude Memory Tool (Anthropic models only)
praisonai "Research and remember findings" --claude-memory --llm anthropic/claude-sonnet-4-20250514
```
## Using JavaScript Code
```bash
@@ -1016,7 +1141,7 @@ cd src/praisonai && uv publish
- 🎨 Interactive UIs
- 📄 YAML-based Configuration
- 🛠️ Custom Tool Integration
- 🔍 Internet Search Capability (using Crawl4AI and Tavily)
- 🔍 Internet Search Capability (Tavily, You.com, Exa, DuckDuckGo, Crawl4AI)
- 🖼️ Vision Language Model (VLM) Support
- 🎙️ Real-time Voice Interaction
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.5.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.5.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.7.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.7.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+619 -1
View File
@@ -549,7 +549,7 @@ class PraisonAI:
return default_args
# Define special commands
special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research']
special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks']
parser = argparse.ArgumentParser(prog="praisonai", description="praisonAI command-line interface")
parser.add_argument("--framework", choices=["crewai", "autogen", "praisonai"], help="Specify the framework")
@@ -588,6 +588,25 @@ class PraisonAI:
parser.add_argument("--web-fetch", action="store_true", help="Enable web fetch to retrieve URL content (Anthropic only)")
parser.add_argument("--prompt-caching", action="store_true", help="Enable prompt caching to reduce costs (OpenAI, Anthropic, Bedrock, Deepseek)")
# Planning Mode arguments
parser.add_argument("--planning", action="store_true", help="Enable planning mode - create plan before execution")
parser.add_argument("--planning-tools", type=str, help="Tools for planning research (path to tools.py or comma-separated tool names)")
parser.add_argument("--planning-reasoning", action="store_true", help="Enable chain-of-thought reasoning in planning")
parser.add_argument("--auto-approve-plan", action="store_true", help="Auto-approve generated plans without user confirmation")
# Memory arguments
parser.add_argument("--memory", action="store_true", help="Enable file-based memory for agent")
parser.add_argument("--user-id", type=str, help="User ID for memory isolation")
# Rules arguments
parser.add_argument("--include-rules", type=str, help="Include manual rules by name (comma-separated)")
# Workflow arguments
parser.add_argument("--workflow-var", action="append", help="Workflow variable in key=value format (can be used multiple times)")
# Claude Memory Tool arguments
parser.add_argument("--claude-memory", action="store_true", help="Enable Claude Memory Tool (Anthropic models only)")
# If we're in a test environment, parse with empty args to avoid pytest interference
if in_test_env:
args, unknown_args = parser.parse_known_args([])
@@ -734,6 +753,60 @@ class PraisonAI:
self.handle_research_command(research_query, research_model, verbose, save, query_rewrite, tools_path, rewrite_tools)
sys.exit(0)
elif args.command == 'memory':
if not PRAISONAI_AVAILABLE:
print("[red]ERROR: PraisonAI Agents is not installed. Install with:[/red]")
print("\npip install praisonaiagents\n")
sys.exit(1)
# Get action and arguments from remaining args
action = unknown_args[0] if unknown_args else 'help'
action_args = unknown_args[1:] if len(unknown_args) > 1 else []
user_id = getattr(args, 'user_id', None)
self.handle_memory_command(action, action_args, user_id)
sys.exit(0)
elif args.command == 'rules':
if not PRAISONAI_AVAILABLE:
print("[red]ERROR: PraisonAI Agents is not installed. Install with:[/red]")
print("\npip install praisonaiagents\n")
sys.exit(1)
# Get action and arguments from remaining args
action = unknown_args[0] if unknown_args else 'list'
action_args = unknown_args[1:] if len(unknown_args) > 1 else []
self.handle_rules_command(action, action_args)
sys.exit(0)
elif args.command == 'workflow':
if not PRAISONAI_AVAILABLE:
print("[red]ERROR: PraisonAI Agents is not installed. Install with:[/red]")
print("\npip install praisonaiagents\n")
sys.exit(1)
# Get action and arguments from remaining args
action = unknown_args[0] if unknown_args else 'list'
action_args = unknown_args[1:] if len(unknown_args) > 1 else []
workflow_vars = {}
if getattr(args, 'workflow_var', None):
for var in args.workflow_var:
if '=' in var:
key, value = var.split('=', 1)
workflow_vars[key] = value
self.handle_workflow_command(action, action_args, workflow_vars)
sys.exit(0)
elif args.command == 'hooks':
if not PRAISONAI_AVAILABLE:
print("[red]ERROR: PraisonAI Agents is not installed. Install with:[/red]")
print("\npip install praisonaiagents\n")
sys.exit(1)
# Get action from remaining args
action = unknown_args[0] if unknown_args else 'list'
self.handle_hooks_command(action)
sys.exit(0)
# Only check framework availability for agent-related operations
if not args.command and (args.init or args.auto or args.framework):
if not CREWAI_AVAILABLE and not AUTOGEN_AVAILABLE and not PRAISONAI_AVAILABLE:
@@ -936,6 +1009,520 @@ class PraisonAI:
verbose = getattr(self.args, 'verbose', False)
return self._expand_prompt(prompt, expand_tools, verbose)
def _load_tools(self, tools_path: str) -> list:
"""
Load tools from a file path or comma-separated tool names.
Args:
tools_path: Path to tools.py file or comma-separated tool names
Returns:
List of tool functions
"""
tools_list = []
if not tools_path:
return tools_list
if os.path.isfile(tools_path):
# Load from file
try:
import inspect
spec = importlib.util.spec_from_file_location("tools_module", tools_path)
if spec and spec.loader:
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
for name, obj in inspect.getmembers(module):
if inspect.isfunction(obj) and not name.startswith('_'):
tools_list.append(obj)
if tools_list:
print(f"[cyan]Loaded {len(tools_list)} tools from {tools_path}[/cyan]")
except Exception as e:
print(f"[yellow]Warning: Failed to load tools from {tools_path}: {e}[/yellow]")
else:
# Treat as comma-separated tool names
try:
from praisonaiagents.tools import TOOL_MAPPINGS
import praisonaiagents.tools as tools_module
tool_names = [t.strip() for t in tools_path.split(',')]
for tool_name in tool_names:
if tool_name in TOOL_MAPPINGS:
try:
tool = getattr(tools_module, tool_name)
tools_list.append(tool)
except Exception as e:
print(f"[yellow]Warning: Failed to load tool '{tool_name}': {e}[/yellow]")
else:
print(f"[yellow]Warning: Unknown tool '{tool_name}'[/yellow]")
if tools_list:
print(f"[cyan]Loaded {len(tools_list)} built-in tools[/cyan]")
except ImportError:
print("[yellow]Warning: Could not import tools module[/yellow]")
return tools_list
def handle_memory_command(self, action: str, action_args: list, user_id: str = None):
"""
Handle memory subcommand actions.
Args:
action: The memory action (show, add, search, clear, save, resume, etc.)
action_args: Additional arguments for the action
user_id: User ID for memory isolation
"""
try:
from praisonaiagents.memory import FileMemory
from rich import print
from rich.table import Table
from rich.console import Console
console = Console()
memory = FileMemory(user_id=user_id or "default")
if action == 'show':
stats = memory.get_stats()
table = Table(title="Memory Statistics")
table.add_column("Property", style="cyan")
table.add_column("Value", style="green")
for key, value in stats.items():
table.add_row(str(key), str(value))
console.print(table)
# Show recent memories
print("\n[bold]Recent Short-term Memories:[/bold]")
short_term = memory.get_short_term(limit=5)
for i, item in enumerate(short_term, 1):
content = item.get('content', str(item))[:100]
print(f" {i}. {content}")
elif action == 'add':
if not action_args:
print("[red]ERROR: Content required. Usage: praisonai memory add \"Your memory content\"[/red]")
return
content = ' '.join(action_args)
memory.add_long_term(content, importance=0.8)
print(f"[green]✅ Added to long-term memory: {content[:50]}...[/green]")
elif action == 'search':
if not action_args:
print("[red]ERROR: Query required. Usage: praisonai memory search \"query\"[/red]")
return
query = ' '.join(action_args)
results = memory.search(query, limit=10)
print(f"[bold]Search results for '{query}':[/bold]")
for i, result in enumerate(results, 1):
content = result.get('content', str(result))[:100]
print(f" {i}. {content}")
elif action == 'clear':
target = action_args[0] if action_args else 'short'
if target == 'all':
memory.clear_all()
print("[green]✅ All memory cleared[/green]")
else:
memory.clear_short_term()
print("[green]✅ Short-term memory cleared[/green]")
elif action == 'save':
if not action_args:
print("[red]ERROR: Session name required. Usage: praisonai memory save <session_name>[/red]")
return
session_name = action_args[0]
memory.save_session(session_name)
print(f"[green]✅ Session saved: {session_name}[/green]")
elif action == 'resume':
if not action_args:
print("[red]ERROR: Session name required. Usage: praisonai memory resume <session_name>[/red]")
return
session_name = action_args[0]
memory.resume_session(session_name)
print(f"[green]✅ Session resumed: {session_name}[/green]")
elif action == 'sessions':
sessions = memory.list_sessions()
if sessions:
print("[bold]Saved Sessions:[/bold]")
for s in sessions:
print(f" - {s.get('name', 'Unknown')} (saved: {s.get('saved_at', 'Unknown')})")
else:
print("[yellow]No saved sessions found[/yellow]")
elif action == 'compress':
print("[cyan]Compressing short-term memory...[/cyan]")
# Note: compress requires an LLM function, so we'll just show stats
stats = memory.get_stats()
print(f"[green]Short-term items: {stats.get('short_term_count', 0)}[/green]")
print("[yellow]Note: Full compression requires an LLM. Use programmatically with memory.compress(llm_func=...)[/yellow]")
elif action == 'checkpoint':
name = action_args[0] if action_args else None
checkpoint_id = memory.create_checkpoint(name)
print(f"[green]✅ Checkpoint created: {checkpoint_id}[/green]")
elif action == 'restore':
if not action_args:
print("[red]ERROR: Checkpoint ID required. Usage: praisonai memory restore <checkpoint_id>[/red]")
return
checkpoint_id = action_args[0]
memory.restore_checkpoint(checkpoint_id)
print(f"[green]✅ Checkpoint restored: {checkpoint_id}[/green]")
elif action == 'checkpoints':
checkpoints = memory.list_checkpoints()
if checkpoints:
print("[bold]Checkpoints:[/bold]")
for cp in checkpoints:
print(f" - {cp.get('id', 'Unknown')} ({cp.get('name', 'Unnamed')})")
else:
print("[yellow]No checkpoints found[/yellow]")
elif action == 'help' or action == '--help':
print("[bold]Memory Commands:[/bold]")
print(" praisonai memory show - Show memory statistics")
print(" praisonai memory add <content> - Add to long-term memory")
print(" praisonai memory search <query> - Search memories")
print(" praisonai memory clear [short|all] - Clear memory")
print(" praisonai memory save <session_name> - Save session")
print(" praisonai memory resume <session_name> - Resume session")
print(" praisonai memory sessions - List saved sessions")
print(" praisonai memory compress - Compress short-term memory")
print(" praisonai memory checkpoint [name] - Create checkpoint")
print(" praisonai memory restore <checkpoint_id> - Restore checkpoint")
print(" praisonai memory checkpoints - List checkpoints")
print("\n[bold]Options:[/bold]")
print(" --user-id <id> - User ID for memory isolation")
else:
print(f"[red]Unknown memory action: {action}[/red]")
print("Use 'praisonai memory help' for available commands")
except ImportError as e:
print(f"[red]ERROR: Failed to import memory module: {e}[/red]")
print("Make sure praisonaiagents is installed: pip install praisonaiagents")
except Exception as e:
print(f"[red]ERROR: Memory command failed: {e}[/red]")
def handle_rules_command(self, action: str, action_args: list):
"""
Handle rules subcommand actions.
Args:
action: The rules action (list, show, create, delete, stats)
action_args: Additional arguments for the action
"""
try:
from praisonaiagents.memory import RulesManager
from rich import print
from rich.table import Table
from rich.console import Console
console = Console()
rules = RulesManager(workspace_path=os.getcwd())
if action == 'list':
all_rules = rules.get_all_rules()
if all_rules:
table = Table(title="Loaded Rules")
table.add_column("Name", style="cyan")
table.add_column("Description", style="white")
table.add_column("Activation", style="green")
table.add_column("Priority", style="yellow")
for rule in all_rules:
table.add_row(
rule.name,
rule.description[:50] + "..." if len(rule.description) > 50 else rule.description,
rule.activation,
str(rule.priority)
)
console.print(table)
else:
print("[yellow]No rules found. Create PRAISON.md, CLAUDE.md, or files in .praison/rules/[/yellow]")
elif action == 'show':
if not action_args:
print("[red]ERROR: Rule name required. Usage: praisonai rules show <name>[/red]")
return
rule_name = action_args[0]
rule = rules.get_rule_by_name(rule_name)
if rule:
print(f"[bold cyan]Rule: {rule.name}[/bold cyan]")
print(f"[bold]Description:[/bold] {rule.description}")
print(f"[bold]Activation:[/bold] {rule.activation}")
print(f"[bold]Priority:[/bold] {rule.priority}")
if rule.globs:
print(f"[bold]Globs:[/bold] {', '.join(rule.globs)}")
print(f"\n[bold]Content:[/bold]\n{rule.content}")
else:
print(f"[red]Rule not found: {rule_name}[/red]")
elif action == 'create':
if len(action_args) < 2:
print("[red]ERROR: Name and content required. Usage: praisonai rules create <name> <content>[/red]")
return
rule_name = action_args[0]
content = ' '.join(action_args[1:])
rules.create_rule(
name=rule_name,
content=content,
description=f"Rule created via CLI: {rule_name}",
activation="always",
scope="workspace"
)
print(f"[green]✅ Rule created: {rule_name}[/green]")
elif action == 'delete':
if not action_args:
print("[red]ERROR: Rule name required. Usage: praisonai rules delete <name>[/red]")
return
rule_name = action_args[0]
rules.delete_rule(rule_name)
print(f"[green]✅ Rule deleted: {rule_name}[/green]")
elif action == 'stats':
stats = rules.get_stats()
table = Table(title="Rules Statistics")
table.add_column("Property", style="cyan")
table.add_column("Value", style="green")
for key, value in stats.items():
if isinstance(value, dict):
table.add_row(str(key), str(value))
else:
table.add_row(str(key), str(value))
console.print(table)
elif action == 'help' or action == '--help':
print("[bold]Rules Commands:[/bold]")
print(" praisonai rules list - List all loaded rules")
print(" praisonai rules show <name> - Show specific rule details")
print(" praisonai rules create <name> <content> - Create a new rule")
print(" praisonai rules delete <name> - Delete a rule")
print(" praisonai rules stats - Show rules statistics")
print("\n[bold]Supported Rule Files:[/bold]")
print(" PRAISON.md, CLAUDE.md, AGENTS.md, GEMINI.md")
print(" .cursorrules, .windsurfrules")
print(" .praison/rules/*.md, ~/.praison/rules/*.md")
else:
print(f"[red]Unknown rules action: {action}[/red]")
print("Use 'praisonai rules help' for available commands")
except ImportError as e:
print(f"[red]ERROR: Failed to import rules module: {e}[/red]")
print("Make sure praisonaiagents is installed: pip install praisonaiagents")
except Exception as e:
print(f"[red]ERROR: Rules command failed: {e}[/red]")
def handle_workflow_command(self, action: str, action_args: list, variables: dict = None):
"""
Handle workflow subcommand actions.
Args:
action: The workflow action (list, run, create, show)
action_args: Additional arguments for the action
variables: Workflow variables for substitution
"""
try:
from praisonaiagents.memory import WorkflowManager
from rich import print
from rich.table import Table
from rich.console import Console
console = Console()
manager = WorkflowManager(workspace_path=os.getcwd())
if action == 'list':
workflows = manager.list_workflows()
if workflows:
table = Table(title="Available Workflows")
table.add_column("Name", style="cyan")
table.add_column("Description", style="white")
table.add_column("Steps", style="green")
for workflow in workflows:
table.add_row(
workflow.name,
workflow.description[:50] + "..." if len(workflow.description) > 50 else workflow.description,
str(len(workflow.steps))
)
console.print(table)
else:
print("[yellow]No workflows found. Create files in .praison/workflows/[/yellow]")
elif action == 'run':
if not action_args:
print("[red]ERROR: Workflow name required. Usage: praisonai workflow run <name>[/red]")
return
workflow_name = action_args[0]
# Create a simple executor using PraisonAgent
def executor(prompt):
agent = PraisonAgent(
name="WorkflowExecutor",
role="Task Executor",
goal="Execute workflow steps"
)
return agent.chat(prompt)
print(f"[bold cyan]Running workflow: {workflow_name}[/bold cyan]")
result = manager.execute(
workflow_name,
executor=executor,
variables=variables or {}
)
if result.get("success"):
print("[green]✅ Workflow completed successfully![/green]")
for step_result in result.get("results", []):
status = "✅" if step_result.get("status") == "completed" else "❌"
print(f" {status} {step_result.get('step', 'Unknown step')}")
else:
print(f"[red]❌ Workflow failed: {result.get('error', 'Unknown error')}[/red]")
elif action == 'show':
if not action_args:
print("[red]ERROR: Workflow name required. Usage: praisonai workflow show <name>[/red]")
return
workflow_name = action_args[0]
workflow = manager.get_workflow(workflow_name)
if workflow:
print(f"[bold cyan]Workflow: {workflow.name}[/bold cyan]")
print(f"[bold]Description:[/bold] {workflow.description}")
print(f"\n[bold]Steps ({len(workflow.steps)}):[/bold]")
for i, step in enumerate(workflow.steps, 1):
print(f" {i}. {step.name}: {step.action[:80]}...")
if workflow.variables:
print(f"\n[bold]Variables:[/bold] {workflow.variables}")
else:
print(f"[red]Workflow not found: {workflow_name}[/red]")
elif action == 'create':
if not action_args:
print("[red]ERROR: Workflow name required. Usage: praisonai workflow create <name>[/red]")
return
workflow_name = action_args[0]
# Create a simple template workflow
manager.create_workflow(
name=workflow_name,
description=f"Workflow created via CLI: {workflow_name}",
steps=[
{"name": "Step 1", "action": "First step - edit this in .praison/workflows/"},
{"name": "Step 2", "action": "Second step - edit this in .praison/workflows/"}
]
)
print(f"[green]✅ Workflow created: {workflow_name}[/green]")
print(f"[cyan]Edit the workflow in .praison/workflows/{workflow_name}.md[/cyan]")
elif action == 'help' or action == '--help':
print("[bold]Workflow Commands:[/bold]")
print(" praisonai workflow list - List available workflows")
print(" praisonai workflow run <name> - Execute a workflow")
print(" praisonai workflow show <name> - Show workflow details")
print(" praisonai workflow create <name> - Create a new workflow")
print("\n[bold]Options:[/bold]")
print(" --workflow-var key=value - Set workflow variable (can be repeated)")
print("\n[bold]Example:[/bold]")
print(" praisonai workflow run deploy --workflow-var environment=staging")
else:
print(f"[red]Unknown workflow action: {action}[/red]")
print("Use 'praisonai workflow help' for available commands")
except ImportError as e:
print(f"[red]ERROR: Failed to import workflow module: {e}[/red]")
print("Make sure praisonaiagents is installed: pip install praisonaiagents")
except Exception as e:
print(f"[red]ERROR: Workflow command failed: {e}[/red]")
def handle_hooks_command(self, action: str):
"""
Handle hooks subcommand actions.
Args:
action: The hooks action (list, stats, init)
"""
try:
from praisonaiagents.memory import HooksManager
from rich import print
from rich.table import Table
from rich.console import Console
import json
console = Console()
hooks = HooksManager(workspace_path=os.getcwd())
if action == 'list':
stats = hooks.get_stats()
if stats.get('total_hooks', 0) > 0:
print("[bold]Configured Hooks:[/bold]")
for event in stats.get('events', []):
print(f" - {event}")
else:
print("[yellow]No hooks configured. Create .praison/hooks.json[/yellow]")
elif action == 'stats':
stats = hooks.get_stats()
table = Table(title="Hooks Statistics")
table.add_column("Property", style="cyan")
table.add_column("Value", style="green")
for key, value in stats.items():
table.add_row(str(key), str(value))
console.print(table)
elif action == 'init':
hooks_dir = os.path.join(os.getcwd(), ".praison")
os.makedirs(hooks_dir, exist_ok=True)
hooks_file = os.path.join(hooks_dir, "hooks.json")
if os.path.exists(hooks_file):
print(f"[yellow]hooks.json already exists at {hooks_file}[/yellow]")
else:
template = {
"enabled": True,
"timeout": 30,
"hooks": {
"pre_write_code": "./scripts/lint.sh",
"post_write_code": "./scripts/format.sh",
"pre_run_command": {
"command": "./scripts/validate.sh",
"timeout": 60,
"block_on_failure": True
}
}
}
with open(hooks_file, 'w') as f:
json.dump(template, f, indent=2)
print(f"[green]✅ Created hooks.json at {hooks_file}[/green]")
print("[cyan]Edit the file to configure your hooks[/cyan]")
elif action == 'help' or action == '--help':
print("[bold]Hooks Commands:[/bold]")
print(" praisonai hooks list - List configured hooks")
print(" praisonai hooks stats - Show hooks statistics")
print(" praisonai hooks init - Create hooks.json template")
print("\n[bold]Hook Events:[/bold]")
print(" pre_read_code, post_read_code")
print(" pre_write_code, post_write_code")
print(" pre_run_command, post_run_command")
print(" pre_user_prompt, post_user_prompt")
print(" pre_mcp_tool_use, post_mcp_tool_use")
else:
print(f"[red]Unknown hooks action: {action}[/red]")
print("Use 'praisonai hooks help' for available commands")
except ImportError as e:
print(f"[red]ERROR: Failed to import hooks module: {e}[/red]")
print("Make sure praisonaiagents is installed: pip install praisonaiagents")
except Exception as e:
print(f"[red]ERROR: Hooks command failed: {e}[/red]")
def _save_output(self, prompt: str, result: str):
"""
Save output to output/prompts/ folder.
@@ -994,6 +1581,37 @@ class PraisonAI:
agent_config["web_fetch"] = True
if getattr(self.args, 'prompt_caching', False):
agent_config["prompt_caching"] = True
# Planning Mode
if getattr(self.args, 'planning', False):
agent_config["planning"] = True
print("[bold cyan]Planning mode enabled - agent will create a plan before execution[/bold cyan]")
# Load planning tools if specified
if getattr(self.args, 'planning_tools', None):
planning_tools_list = self._load_tools(self.args.planning_tools)
if planning_tools_list:
agent_config["planning_tools"] = planning_tools_list
if getattr(self.args, 'planning_reasoning', False):
agent_config["planning_reasoning"] = True
# Memory
if getattr(self.args, 'memory', False):
agent_config["memory"] = True
print("[bold cyan]Memory enabled - agent will remember context across sessions[/bold cyan]")
if getattr(self.args, 'user_id', None):
agent_config["user_id"] = self.args.user_id
# Claude Memory Tool (Anthropic only)
if getattr(self.args, 'claude_memory', False):
llm = getattr(self.args, 'llm', '')
if llm and 'anthropic' in llm.lower():
agent_config["claude_memory"] = True
print("[bold cyan]Claude Memory Tool enabled - Claude will autonomously manage memories[/bold cyan]")
else:
print("[yellow]Warning: --claude-memory requires an Anthropic model (--llm anthropic/...)[/yellow]")
agent = PraisonAgent(**agent_config)
result = agent.start(prompt)
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==2.3.5 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==2.3.7 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+1 -1
View File
@@ -1 +1 @@
__version__ = "2.3.5"
__version__ = "2.3.7"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.0.183",
"praisonaiagents>=0.0.185",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
+1 -1
View File
@@ -365,7 +365,7 @@ This is **normal and expected** during development. The key metrics are:
2. **Mocking**: Mock external dependencies and APIs
3. **Naming**: Clear, descriptive test names
4. **Documentation**: Document complex test scenarios
5. **Performance**: Keep unit tests fast (<1s each)
5. **Performance**: Keep unit tests fast (under 1s each)
6. **Coverage**: Aim for high coverage of critical paths
7. **Maintainability**: Regular test maintenance and updates
+179
View File
@@ -0,0 +1,179 @@
#!/usr/bin/env python3
"""
Tests for new CLI features (fast, direct import, no subprocess):
- Planning Mode
- Memory CLI
- Rules CLI
- Workflow CLI
- Hooks CLI
- Claude Memory Tool
"""
import argparse
class TestCLIArgumentParsing:
"""Test CLI argument parsing directly (fast, no subprocess)."""
def get_parser(self):
"""Create argument parser matching CLI."""
parser = argparse.ArgumentParser(description="praisonAI CLI")
# Planning args
parser.add_argument("--planning", action="store_true")
parser.add_argument("--planning-tools")
parser.add_argument("--planning-reasoning", action="store_true")
parser.add_argument("--auto-approve-plan", action="store_true")
# Memory args
parser.add_argument("--memory", action="store_true")
parser.add_argument("--user-id")
# Rules args
parser.add_argument("--include-rules")
# Workflow args
parser.add_argument("--workflow-var", action="append")
# Claude memory
parser.add_argument("--claude-memory", action="store_true")
parser.add_argument("command", nargs="?")
return parser
def test_planning_args_parsed(self):
"""Test planning arguments are parsed correctly."""
parser = self.get_parser()
args = parser.parse_args(["--planning", "--planning-reasoning", "test"])
assert args.planning is True
assert args.planning_reasoning is True
assert args.command == "test"
def test_planning_tools_parsed(self):
"""Test planning-tools argument is parsed."""
parser = self.get_parser()
args = parser.parse_args(["--planning", "--planning-tools", "tools.py", "test"])
assert args.planning is True
assert args.planning_tools == "tools.py"
def test_memory_args_parsed(self):
"""Test memory arguments are parsed correctly."""
parser = self.get_parser()
args = parser.parse_args(["--memory", "--user-id", "user123", "test"])
assert args.memory is True
assert args.user_id == "user123"
def test_rules_args_parsed(self):
"""Test rules arguments are parsed correctly."""
parser = self.get_parser()
args = parser.parse_args(["--include-rules", "security,testing", "test"])
assert args.include_rules == "security,testing"
def test_workflow_vars_parsed(self):
"""Test workflow-var arguments are parsed correctly."""
parser = self.get_parser()
args = parser.parse_args([
"--workflow-var", "env=prod",
"--workflow-var", "branch=main",
"test"
])
assert args.workflow_var == ["env=prod", "branch=main"]
def test_claude_memory_parsed(self):
"""Test claude-memory argument is parsed correctly."""
parser = self.get_parser()
args = parser.parse_args(["--claude-memory", "test"])
assert args.claude_memory is True
def test_auto_approve_plan_parsed(self):
"""Test auto-approve-plan argument is parsed correctly."""
parser = self.get_parser()
args = parser.parse_args(["--planning", "--auto-approve-plan", "test"])
assert args.auto_approve_plan is True
class TestMemoryModule:
"""Test FileMemory module directly."""
def test_file_memory_import(self):
"""Test FileMemory can be imported."""
try:
from praisonaiagents.memory import FileMemory
assert FileMemory is not None
except ImportError:
pass # Skip if not installed
def test_file_memory_basic(self, tmp_path):
"""Test FileMemory basic operations."""
try:
from praisonaiagents.memory import FileMemory
memory = FileMemory(user_id="test_user")
# Test add
memory.add_long_term("Test fact")
# Test stats
stats = memory.get_stats()
assert stats["user_id"] == "test_user"
assert stats["long_term_count"] >= 1
except (ImportError, TypeError):
pass # Skip if not installed or API changed
class TestRulesModule:
"""Test RulesManager module directly."""
def test_rules_manager_import(self):
"""Test RulesManager can be imported."""
try:
from praisonaiagents.memory import RulesManager
assert RulesManager is not None
except ImportError:
pass # Skip if not installed
def test_rules_manager_basic(self):
"""Test RulesManager basic operations."""
try:
from praisonaiagents.memory import RulesManager
manager = RulesManager()
# Test stats
stats = manager.get_stats()
assert "total_rules" in stats
except ImportError:
pass # Skip if not installed
class TestWorkflowModule:
"""Test WorkflowManager module directly."""
def test_workflow_manager_import(self):
"""Test WorkflowManager can be imported."""
try:
from praisonaiagents.memory import WorkflowManager
assert WorkflowManager is not None
except ImportError:
pass # Skip if not installed
class TestHooksModule:
"""Test HooksManager module directly."""
def test_hooks_manager_import(self):
"""Test HooksManager can be imported."""
try:
from praisonaiagents.memory import HooksManager
assert HooksManager is not None
except ImportError:
pass # Skip if not installed
def test_hooks_manager_basic(self):
"""Test HooksManager basic operations."""
try:
from praisonaiagents.memory import HooksManager
hooks = HooksManager()
# Test stats
stats = hooks.get_stats()
assert "total_hooks" in stats
except ImportError:
pass # Skip if not installed
if __name__ == "__main__":
import pytest
pytest.main([__file__, "-v"])
+4 -4
View File
@@ -4197,7 +4197,7 @@ requires-dist = [
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.22" },
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.0.22" },
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.22" },
{ name = "praisonaiagents", specifier = ">=0.0.183" },
{ name = "praisonaiagents", specifier = ">=0.0.185" },
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
{ name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
{ name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4252,7 +4252,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.0.183"
version = "0.0.185"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -4261,9 +4261,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/ba/fb/0971e0a371e5f8636ce34fb3fb556c8438db352f866a646df197998c7465/praisonaiagents-0.0.183.tar.gz", hash = "sha256:fccb8e97b43cda260ecd1f470eeb25abfa8c3f2610995bdede89a337054a4eab", size = 409014, upload-time = "2025-12-15T21:04:24.104Z" }
sdist = { url = "https://files.pythonhosted.org/packages/eb/b9/e04ca0b88ccd779820f895c773afdb124a01242fe3e4f4831f24eea951de/praisonaiagents-0.0.185.tar.gz", hash = "sha256:1213893e3a690e64c5ef3f0991cd230ec94b052114340ebcf0b84a6c3797a41a", size = 425649, upload-time = "2025-12-16T07:44:28.169Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2a/e8/df5b543876604c2ea93e630dfb8482742a479c06d422b840f1f0d4430802/praisonaiagents-0.0.183-py3-none-any.whl", hash = "sha256:d5c4796dc74d639321e65aa09b9f15a74848a4a969c844260a7097801e76bda3", size = 397016, upload-time = "2025-12-15T21:04:21.085Z" },
{ url = "https://files.pythonhosted.org/packages/ca/04/9f60d36df1b3340847fd7b930c99f101678d7b8a0c301024d7940e771aa8/praisonaiagents-0.0.185-py3-none-any.whl", hash = "sha256:bad408a451b8e26957fa444caa305ff8a784461a67dd8a71e0507af533fa7972", size = 410732, upload-time = "2025-12-16T07:44:26.525Z" },
]
[[package]]