Compare commits

...
3 Commits
Author SHA1 Message Date
Mervin Praison 9d29c796c7 Release v3.9.7 2026-01-16 12:05:38 +00:00
Mervin Praison f058ce018f Release v3.9.6 2026-01-16 09:18:27 +00:00
Mervin Praison 09671d4bd3 Release v3.9.5 2026-01-16 08:00:39 +00:00
35 changed files with 4961 additions and 120 deletions

No files matched your search

+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>=3.9.4" \
"praisonai>=3.9.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>=3.9.4" \
"praisonai>=3.9.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>=3.9.4" \
"praisonai>=3.9.7" \
"praisonai[ui]" \
"praisonai[crewai]"
@@ -3,6 +3,9 @@
Agent-level guardrails example.
This example demonstrates how to use guardrails at the Agent level
which will apply to all tasks executed by that agent.
Note: Agent guardrails only support function-based validation or preset strings
(permissive, safety, strict). For LLM-based guardrails, use Task guardrails.
"""
from typing import Tuple, Any
from praisonaiagents import Agent, TaskOutput
@@ -38,10 +41,10 @@ def validate_professional_tone(task_output: TaskOutput) -> Tuple[bool, Any]:
def main():
"""Demonstrate Agent-level guardrails with function-based and LLM-based validation."""
"""Demonstrate Agent-level guardrails with function-based validation."""
print("=== Agent Guardrail Examples ===\n")
# Example 1: Function-based guardrail
# Example 1: Function-based guardrail for content length
print("1. Function-based guardrail (content length validation):")
agent1 = Agent(
name="ContentWriter",
@@ -54,12 +57,12 @@ def main():
except Exception as e:
print(f"Error: {e}\n")
# Example 2: LLM-based guardrail (string description)
print("2. LLM-based guardrail (professional tone validation):")
# Example 2: Function-based guardrail for professional tone
print("2. Function-based guardrail (professional tone validation):")
agent2 = Agent(
name="BusinessWriter",
instructions="You are a business communication expert",
guardrails="Ensure the content is professional, formal, and suitable for business communication. No casual language or slang."
instructions="You are a business communication expert. Write formal, professional content.",
guardrails=validate_professional_tone
)
try:
result2 = agent2.start("Write a welcome message for new employees")
@@ -67,20 +70,22 @@ def main():
except Exception as e:
print(f"Error: {e}\n")
# Example 3: Multiple agents with different guardrails
print("3. Professional tone function-based guardrail:")
# Example 3: Using preset guardrails (built-in options)
print("3. Using preset guardrail ('safety'):")
agent3 = Agent(
name="ProfessionalWriter",
instructions="Write professional business content",
guardrails=validate_professional_tone
name="SafeWriter",
instructions="Write helpful content",
guardrails="safety" # Valid presets: permissive, safety, strict
)
try:
result3 = agent3.start("Write a casual greeting message")
result3 = agent3.start("Write a friendly greeting message")
print(f"Result: {result3}\n")
except Exception as e:
print(f"Error: {e}\n")
print("=== Agent Guardrails Demonstration Complete ===")
print("\nNote: For LLM-based guardrails with custom descriptions,")
print("use Task-level guardrails instead of Agent-level guardrails.")
if __name__ == "__main__":
@@ -26,7 +26,6 @@ agent = DeepResearchAgent(
- Clear structure with executive summary
""",
model="deep-research-pro", # Auto-detected as Gemini
output="verbose",
)
print(f"Agent: {agent}")
@@ -25,7 +25,6 @@ agent = DeepResearchAgent(
- Clear, structured responses
""",
model="o4-mini-deep-research", # or "o3-deep-research" for higher quality
output="verbose"
)
print(f"Agent: {agent}")
@@ -94,7 +94,7 @@ def main():
research_system = Agents(
agents=[research_agent],
tasks=research_tasks,
memory=mongodb_memory_config,
memory=True, # Use default memory instead of mongodb config (requires proper MemoryConfig)
output="verbose"
)
+1 -1
View File
@@ -8,7 +8,7 @@ researcher = Agent(name="Researcher", role="Research Analyst", output="verbose")
writer = Agent(name="Writer", role="Content Writer", output="verbose")
# Step 2: Create a plan using PlanningAgent
planner = PlanningAgent(llm="gpt-4o-mini", output="status")
planner = PlanningAgent(llm="gpt-4o-mini")
plan = planner.create_plan_sync(
request="Write a short article about the top 3 benefits of meditation",
agents=[researcher, writer],
@@ -12,6 +12,9 @@ class Chunking:
'semantic': ['chunk_size', 'embedding_model'],
'sdpm': ['chunk_size', 'embedding_model'],
'late': ['chunk_size', 'embedding_model'],
'code': ['chunk_size', 'tokenizer_or_token_counter'],
'neural': ['chunk_size', 'tokenizer_or_token_counter'],
'slumber': ['chunk_size', 'genie'],
}
@cached_property
@@ -24,7 +27,10 @@ class Chunking:
SemanticChunker,
SDPMChunker,
LateChunker,
RecursiveChunker
RecursiveChunker,
CodeChunker,
NeuralChunker,
SlumberChunker
)
except ImportError:
raise ImportError(
@@ -37,7 +43,10 @@ class Chunking:
'semantic': SemanticChunker,
'sdpm': SDPMChunker,
'late': LateChunker,
'recursive': RecursiveChunker
'recursive': RecursiveChunker,
'code': CodeChunker,
'neural': NeuralChunker,
'slumber': SlumberChunker
}
def __init__(
@@ -42,6 +42,8 @@ from .protocols import (
MemoryProtocol,
AsyncMemoryProtocol,
ResettableMemoryProtocol,
DeletableMemoryProtocol,
AsyncDeletableMemoryProtocol,
EntityMemoryProtocol,
)
@@ -203,5 +205,7 @@ __all__ = [
"MemoryProtocol",
"AsyncMemoryProtocol",
"ResettableMemoryProtocol",
"DeletableMemoryProtocol",
"AsyncDeletableMemoryProtocol",
"EntityMemoryProtocol",
]
@@ -806,6 +806,220 @@ class FileMemory:
self._log("Cleared all memory")
# -------------------------------------------------------------------------
# Selective Deletion
# -------------------------------------------------------------------------
def delete_short_term(self, memory_id: str) -> bool:
"""
Delete a specific short-term memory by ID.
Args:
memory_id: The unique ID of the memory to delete
Returns:
True if memory was found and deleted, False otherwise
"""
for i, item in enumerate(self._short_term):
if item.id == memory_id:
del self._short_term[i]
self._save_short_term()
self._log(f"Deleted short-term memory: {memory_id}")
return True
return False
def delete_long_term(self, memory_id: str) -> bool:
"""
Delete a specific long-term memory by ID.
Args:
memory_id: The unique ID of the memory to delete
Returns:
True if memory was found and deleted, False otherwise
"""
for i, item in enumerate(self._long_term):
if item.id == memory_id:
del self._long_term[i]
self._save_long_term()
self._log(f"Deleted long-term memory: {memory_id}")
return True
return False
def delete_entity(self, name: str) -> bool:
"""
Delete an entity by name.
Args:
name: The name of the entity to delete
Returns:
True if entity was found and deleted, False otherwise
"""
# Find entity by name using existing helper
entity = self._find_entity_by_name(name)
if entity:
# Delete by entity's ID (which is the dict key)
if entity.id in self._entities:
del self._entities[entity.id]
self._save_entities()
self._log(f"Deleted entity: {name}")
return True
# Try direct ID match as fallback (in case name IS the ID)
if name in self._entities:
del self._entities[name]
self._save_entities()
self._log(f"Deleted entity by ID: {name}")
return True
return False
def delete_episodic(self, memory_id: str, date: Optional[str] = None) -> bool:
"""
Delete a specific episodic memory by ID.
Args:
memory_id: The unique ID of the memory to delete
date: Optional date string (YYYY-MM-DD) to narrow search
Returns:
True if memory was found and deleted, False otherwise
"""
from datetime import datetime, timedelta
# Determine which files to search
if date:
dates_to_check = [date]
else:
# Check last 30 days
dates_to_check = [
(datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d")
for i in range(30)
]
for check_date in dates_to_check:
episodic_file = self.episodic_path / f"{check_date}.json"
if episodic_file.exists():
data = self._read_json(episodic_file, [])
for i, item in enumerate(data):
if item.get("id") == memory_id:
del data[i]
self._write_json(episodic_file, data)
self._log(f"Deleted episodic memory: {memory_id}")
return True
return False
def delete_memory(
self,
memory_id: str,
memory_type: Optional[str] = None
) -> bool:
"""
Delete a specific memory by ID.
This is the unified deletion method that searches across all memory types.
Use this when you have a memory ID but don't know its type.
Particularly useful for:
- Cleaning up image-based memories after processing to free context window
- Removing outdated or incorrect information
- Privacy compliance (selective erasure)
Args:
memory_id: The unique ID of the memory to delete
memory_type: Optional type hint to narrow search:
'short_term', 'long_term', 'entity', 'episodic'
If None, searches all types.
Returns:
True if memory was found and deleted, False otherwise
Example:
# Delete a specific memory after processing an image
memory.delete_memory("abc123def456")
# Delete with type hint for faster lookup
memory.delete_memory("abc123", memory_type="short_term")
"""
# If type specified, only search that type
if memory_type == "short_term":
return self.delete_short_term(memory_id)
elif memory_type == "long_term":
return self.delete_long_term(memory_id)
elif memory_type == "entity":
return self.delete_entity(memory_id)
elif memory_type == "episodic":
return self.delete_episodic(memory_id)
# Search all types
if self.delete_short_term(memory_id):
return True
if self.delete_long_term(memory_id):
return True
if self.delete_entity(memory_id):
return True
if self.delete_episodic(memory_id):
return True
return False
def delete_memories(self, memory_ids: List[str]) -> int:
"""
Delete multiple memories by their IDs.
Args:
memory_ids: List of memory IDs to delete
Returns:
Number of memories successfully deleted
"""
count = 0
for memory_id in memory_ids:
if self.delete_memory(memory_id):
count += 1
return count
def delete_memories_matching(
self,
query: str,
memory_types: Optional[List[str]] = None,
limit: int = 10
) -> int:
"""
Delete memories matching a search query.
Useful for bulk cleanup of related memories, e.g., all image-related
context after finishing an image analysis session.
Args:
query: Search query to match memories
memory_types: Optional list of types to search
limit: Maximum number of memories to delete
Returns:
Number of memories deleted
"""
# Search for matching memories
results = self.search(query, memory_types=memory_types, limit=limit)
# Extract IDs and delete
deleted = 0
for result in results:
item = result.get("item", {})
memory_id = item.get("id")
memory_type = result.get("type")
if memory_id:
if self.delete_memory(memory_id, memory_type=memory_type):
deleted += 1
if deleted:
self._log(f"Deleted {deleted} memories matching '{query}'")
return deleted
def get_stats(self) -> Dict[str, Any]:
"""Get memory statistics."""
episodic_count = sum(1 for _ in self.episodic_path.glob("*.json"))
@@ -1222,6 +1436,66 @@ Summary:"""
else:
return {"error": "Usage: /memory clear [short|all]"}
elif action == "delete":
if not args:
return {"error": "Usage: /memory delete <id> or /memory delete --query <search_query>"}
# Handle --query flag for bulk deletion
if args.startswith("--query "):
query = args[8:] # Remove "--query " prefix
if not query:
return {"error": "Usage: /memory delete --query <search_query>"}
deleted = self.delete_memories_matching(query, limit=10)
return {
"action": "delete",
"type": "query",
"query": query,
"deleted_count": deleted
}
# Single ID deletion
success = self.delete_memory(args)
return {
"action": "delete",
"type": "single",
"id": args,
"success": success
}
elif action == "list":
# List memories with IDs for easy deletion reference
memory_type = args.lower() if args else "all"
items = []
if memory_type in ("all", "short", "short_term"):
for item in self.get_short_term(limit=20):
items.append({
"type": "short_term",
"id": item.id,
"content": item.content[:100] + "..." if len(item.content) > 100 else item.content,
"importance": item.importance
})
if memory_type in ("all", "long", "long_term"):
for item in self.get_long_term(limit=20):
items.append({
"type": "long_term",
"id": item.id,
"content": item.content[:100] + "..." if len(item.content) > 100 else item.content,
"importance": item.importance
})
if memory_type in ("all", "entity", "entities"):
for name, entity in self._entities.items():
items.append({
"type": "entity",
"id": name,
"name": entity.name,
"entity_type": entity.entity_type
})
return {"action": "list", "filter": memory_type, "items": items}
elif action == "search":
if not args:
return {"error": "Usage: /memory search <query>"}
@@ -1274,8 +1548,11 @@ Summary:"""
"action": "help",
"commands": {
"/memory show": "Display memory stats and recent items",
"/memory list [short|long|entity]": "List memories with IDs for deletion",
"/memory add <content>": "Add to long-term memory",
"/memory clear [short|all]": "Clear memory",
"/memory delete <id>": "Delete a specific memory by ID",
"/memory delete --query <q>": "Delete memories matching query",
"/memory clear [short|all]": "Clear all memory (bulk)",
"/memory search <query>": "Search memories",
"/memory save <name>": "Save session",
"/memory resume <name>": "Resume session",
@@ -1117,6 +1117,213 @@ class Memory:
self.chroma_client.reset() # entire DB
self._init_chroma() # re-init fresh
# -------------------------------------------------------------------------
# Selective Deletion Methods
# -------------------------------------------------------------------------
def delete_short_term(self, memory_id: str) -> bool:
"""
Delete a specific short-term memory by ID.
Args:
memory_id: The unique ID of the memory to delete
Returns:
True if memory was found and deleted, False otherwise
"""
deleted = False
# Delete from SQLite
try:
conn = sqlite3.connect(self.short_db)
cursor = conn.execute(
"DELETE FROM short_mem WHERE id = ?", (memory_id,)
)
if cursor.rowcount > 0:
deleted = True
conn.commit()
conn.close()
except Exception as e:
self._log_verbose(f"Error deleting from SQLite short-term: {e}", logging.ERROR)
# Delete from MongoDB if enabled
if self.use_mongodb and hasattr(self, "mongo_short_term"):
try:
result = self.mongo_short_term.delete_one({"_id": memory_id})
if result.deleted_count > 0:
deleted = True
except Exception as e:
self._log_verbose(f"Error deleting from MongoDB short-term: {e}", logging.ERROR)
if deleted:
self._log_verbose(f"Deleted short-term memory: {memory_id}")
return deleted
def delete_long_term(self, memory_id: str) -> bool:
"""
Delete a specific long-term memory by ID.
Works across all backends: SQLite, ChromaDB, Mem0, and MongoDB.
Args:
memory_id: The unique ID of the memory to delete
Returns:
True if memory was found and deleted, False otherwise
"""
deleted = False
# Delete from SQLite
try:
conn = sqlite3.connect(self.long_db)
cursor = conn.execute(
"DELETE FROM long_mem WHERE id = ?", (memory_id,)
)
if cursor.rowcount > 0:
deleted = True
conn.commit()
conn.close()
except Exception as e:
self._log_verbose(f"Error deleting from SQLite long-term: {e}", logging.ERROR)
# Delete from ChromaDB if enabled
if self.use_rag and hasattr(self, "chroma_col"):
try:
# ChromaDB delete by ID
self.chroma_col.delete(ids=[memory_id])
deleted = True # ChromaDB doesn't raise on non-existent ID
except Exception as e:
self._log_verbose(f"Error deleting from ChromaDB: {e}", logging.ERROR)
# Delete from Mem0 if enabled
if self.use_mem0 and hasattr(self, "mem0_client"):
try:
# Mem0 has a delete method
self.mem0_client.delete(memory_id)
deleted = True
except Exception as e:
self._log_verbose(f"Error deleting from Mem0: {e}", logging.ERROR)
# Delete from MongoDB if enabled
if self.use_mongodb and hasattr(self, "mongo_long_term"):
try:
result = self.mongo_long_term.delete_one({"_id": memory_id})
if result.deleted_count > 0:
deleted = True
except Exception as e:
self._log_verbose(f"Error deleting from MongoDB long-term: {e}", logging.ERROR)
if deleted:
self._log_verbose(f"Deleted long-term memory: {memory_id}")
return deleted
def delete_memory(
self,
memory_id: str,
memory_type: Optional[str] = None
) -> bool:
"""
Delete a specific memory by ID.
This is the unified deletion method that searches across all memory types
and all backends (SQLite, ChromaDB, Mem0, MongoDB).
Particularly useful for:
- Cleaning up image-based memories after processing to free context window
- Removing outdated or incorrect information
- Privacy compliance (selective erasure)
Args:
memory_id: The unique ID of the memory to delete
memory_type: Optional type hint to narrow search:
'short_term', 'long_term'
If None, searches all types.
Returns:
True if memory was found and deleted, False otherwise
Example:
# Delete a specific image analysis memory
memory.delete_memory("1705123456789")
# Delete with type hint for faster lookup
memory.delete_memory("1705123456789", memory_type="short_term")
"""
# If type specified, only search that type
if memory_type == "short_term":
return self.delete_short_term(memory_id)
elif memory_type == "long_term":
return self.delete_long_term(memory_id)
# Search both types
if self.delete_short_term(memory_id):
return True
if self.delete_long_term(memory_id):
return True
return False
def delete_memories(self, memory_ids: List[str]) -> int:
"""
Delete multiple memories by their IDs.
Args:
memory_ids: List of memory IDs to delete
Returns:
Number of memories successfully deleted
"""
count = 0
for memory_id in memory_ids:
if self.delete_memory(memory_id):
count += 1
return count
def delete_memories_matching(
self,
query: str,
memory_type: Optional[str] = None,
limit: int = 10
) -> int:
"""
Delete memories matching a search query.
Useful for bulk cleanup of related memories, e.g., all image-related
context after finishing an image analysis session.
Args:
query: Search query to match memories
memory_type: Optional type ('short_term' or 'long_term')
limit: Maximum number of memories to delete
Returns:
Number of memories deleted
"""
deleted = 0
# Search short-term if applicable
if memory_type in (None, "short_term"):
results = self.search_short_term(query, limit=limit)
for result in results:
memory_id = result.get("id")
if memory_id and self.delete_short_term(memory_id):
deleted += 1
# Search long-term if applicable
if memory_type in (None, "long_term"):
results = self.search_long_term(query, limit=limit)
for result in results:
memory_id = result.get("id")
if memory_id and self.delete_long_term(memory_id):
deleted += 1
if deleted:
self._log_verbose(f"Deleted {deleted} memories matching '{query}'")
return deleted
# -------------------------------------------------------------------------
# Entity Memory Methods
# -------------------------------------------------------------------------
@@ -138,6 +138,77 @@ class ResettableMemoryProtocol(MemoryProtocol, Protocol):
...
@runtime_checkable
class DeletableMemoryProtocol(MemoryProtocol, Protocol):
"""
Protocol for memory that supports selective deletion.
Enables removing specific memories by ID without clearing all data.
Essential for:
- Privacy compliance (GDPR right to erasure)
- Context window management (removing large image-based memories)
- Correcting outdated information
Example:
```python
class MyMemory:
def delete_memory(self, memory_id: str) -> bool:
if memory_id in self._store:
del self._store[memory_id]
return True
return False
def delete_memories(self, memory_ids: List[str]) -> int:
count = 0
for mid in memory_ids:
if self.delete_memory(mid):
count += 1
return count
```
"""
def delete_memory(self, memory_id: str, memory_type: Optional[str] = None) -> bool:
"""
Delete a specific memory by ID.
Args:
memory_id: The unique identifier of the memory to delete
memory_type: Optional type hint (short_term, long_term, entity)
If None, searches all types.
Returns:
True if memory was found and deleted, False otherwise
"""
...
def delete_memories(self, memory_ids: List[str]) -> int:
"""
Delete multiple memories by their IDs.
Args:
memory_ids: List of memory IDs to delete
Returns:
Number of memories successfully deleted
"""
...
@runtime_checkable
class AsyncDeletableMemoryProtocol(Protocol):
"""
Async Protocol for memory deletion operations.
"""
async def adelete_memory(self, memory_id: str, memory_type: Optional[str] = None) -> bool:
"""Async delete a specific memory by ID."""
...
async def adelete_memories(self, memory_ids: List[str]) -> int:
"""Async delete multiple memories by their IDs."""
...
@runtime_checkable
class AsyncMemoryProtocol(Protocol):
"""
@@ -207,6 +278,9 @@ class EntityMemoryProtocol(Protocol):
__all__ = [
'MemoryProtocol',
'ResettableMemoryProtocol',
'DeletableMemoryProtocol',
'AsyncDeletableMemoryProtocol',
'AsyncMemoryProtocol',
'EntityMemoryProtocol',
]
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.12.3"
version = "0.12.4"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
@@ -12,8 +12,7 @@ from praisonaiagents import QueryRewriterAgent, RewriteStrategy
def main():
# Initialize the agent
agent = QueryRewriterAgent(
model="gpt-4o-mini",
output="verbose"
model="gpt-4o-mini"
)
print("=" * 60)
@@ -16,7 +16,7 @@ from praisonaiagents.config import OutputConfig
agent = Agent(
instructions="You are a helpful assistant",
llm="gpt-4o-mini",
output=OutputConfig(output="verbose") # Show generation progress
output="verbose" # Show generation progress
)
# Run the agent - verbose mode shows streaming progress via Rich Live display
@@ -0,0 +1,374 @@
"""
Tests for memory deletion functionality.
Tests cover:
- FileMemory delete methods (short_term, long_term, entity, episodic)
- Memory class delete methods (SQLite backend)
- Unified delete_memory method
- Bulk delete_memories method
- Query-based delete_memories_matching
- CLI /memory delete command
- Image-based memory cleanup scenario
"""
import pytest
import tempfile
import shutil
from pathlib import Path
class TestFileMemoryDeletion:
"""Test FileMemory delete operations."""
@pytest.fixture
def temp_memory_dir(self):
"""Create a temporary directory for memory storage."""
temp_dir = tempfile.mkdtemp()
yield temp_dir
shutil.rmtree(temp_dir, ignore_errors=True)
@pytest.fixture
def file_memory(self, temp_memory_dir):
"""Create a FileMemory instance with temp storage."""
from praisonaiagents.memory import FileMemory
return FileMemory(user_id="test_user", base_path=temp_memory_dir, verbose=0)
def test_delete_short_term_by_id(self, file_memory):
"""Test deleting a short-term memory by ID."""
# Add a memory
mem_id = file_memory.add_short_term("Test short-term memory")
# Verify it exists
memories = file_memory.get_short_term()
assert any(m.id == mem_id for m in memories)
# Delete it
success = file_memory.delete_short_term(mem_id)
assert success is True
# Verify it's gone
memories = file_memory.get_short_term()
assert not any(m.id == mem_id for m in memories)
def test_delete_long_term_by_id(self, file_memory):
"""Test deleting a long-term memory by ID."""
# Add a memory
mem_id = file_memory.add_long_term("Test long-term memory", importance=0.9)
# Verify it exists
memories = file_memory.get_long_term()
assert any(m.id == mem_id for m in memories)
# Delete it
success = file_memory.delete_long_term(mem_id)
assert success is True
# Verify it's gone
memories = file_memory.get_long_term()
assert not any(m.id == mem_id for m in memories)
def test_delete_entity_by_name(self, file_memory):
"""Test deleting an entity by name."""
# Add an entity
file_memory.add_entity(
name="TestEntity",
entity_type="test",
attributes={"key": "value"},
relationships=[]
)
# Verify it exists
entity = file_memory.get_entity("TestEntity")
assert entity is not None
# Delete it
success = file_memory.delete_entity("TestEntity")
assert success is True
# Verify it's gone
entity = file_memory.get_entity("TestEntity")
assert entity is None
def test_delete_nonexistent_returns_false(self, file_memory):
"""Test that deleting non-existent ID returns False."""
success = file_memory.delete_memory("nonexistent_id_12345")
assert success is False
def test_unified_delete_memory(self, file_memory):
"""Test unified delete_memory that searches all types."""
# Add memories of different types
short_id = file_memory.add_short_term("Short term test")
long_id = file_memory.add_long_term("Long term test")
# Delete short-term without specifying type
success = file_memory.delete_memory(short_id)
assert success is True
# Delete long-term without specifying type
success = file_memory.delete_memory(long_id)
assert success is True
def test_delete_with_type_hint(self, file_memory):
"""Test delete_memory with type hint for faster lookup."""
mem_id = file_memory.add_short_term("Type hint test")
# Delete with correct type hint
success = file_memory.delete_memory(mem_id, memory_type="short_term")
assert success is True
# Try to delete again (should fail)
success = file_memory.delete_memory(mem_id, memory_type="short_term")
assert success is False
def test_bulk_delete_memories(self, file_memory):
"""Test deleting multiple memories at once."""
# Add multiple memories
ids = [
file_memory.add_short_term(f"Memory {i}")
for i in range(5)
]
# Delete all
deleted_count = file_memory.delete_memories(ids)
assert deleted_count == 5
# Verify all are gone
memories = file_memory.get_short_term()
for mem_id in ids:
assert not any(m.id == mem_id for m in memories)
def test_delete_memories_matching_query(self, file_memory):
"""Test deleting memories by search query."""
# Add memories with specific content
file_memory.add_long_term("Image analysis result cat")
file_memory.add_long_term("Image analysis result dog")
file_memory.add_long_term("Weather forecast for today sunny")
# Verify initial state
initial_memories = file_memory.get_long_term()
assert len(initial_memories) >= 3
# Delete all image-related memories
deleted = file_memory.delete_memories_matching("Image analysis")
assert deleted >= 2
# Verify remaining memories
remaining = file_memory.get_long_term()
# Should have fewer memories after deletion
assert len(remaining) < len(initial_memories)
class TestFileMemoryCLIDelete:
"""Test FileMemory CLI delete commands."""
@pytest.fixture
def temp_memory_dir(self):
"""Create a temporary directory for memory storage."""
temp_dir = tempfile.mkdtemp()
yield temp_dir
shutil.rmtree(temp_dir, ignore_errors=True)
@pytest.fixture
def file_memory(self, temp_memory_dir):
"""Create a FileMemory instance with temp storage."""
from praisonaiagents.memory import FileMemory
return FileMemory(user_id="test_user", base_path=temp_memory_dir, verbose=0)
def test_delete_command_by_id(self, file_memory):
"""Test /memory delete <id> command."""
# Add a memory
mem_id = file_memory.add_long_term("CLI delete test")
# Delete via command
result = file_memory.handle_command(f"/memory delete {mem_id}")
assert result["action"] == "delete"
assert result["type"] == "single"
assert result["success"] is True
def test_delete_command_by_query(self, file_memory):
"""Test /memory delete --query <query> command."""
# Add memories
file_memory.add_long_term("Query delete test 1")
file_memory.add_long_term("Query delete test 2")
# Delete via command with query
result = file_memory.handle_command("/memory delete --query Query delete")
assert result["action"] == "delete"
assert result["type"] == "query"
assert result["deleted_count"] >= 1
def test_delete_command_no_args(self, file_memory):
"""Test /memory delete with no arguments returns error."""
result = file_memory.handle_command("/memory delete")
assert "error" in result
def test_list_command_shows_ids(self, file_memory):
"""Test /memory list command returns IDs for deletion reference."""
# Add memories
mem_id = file_memory.add_long_term("List test memory")
# List memories
result = file_memory.handle_command("/memory list")
assert result["action"] == "list"
assert len(result["items"]) >= 1
# Verify items have IDs
assert any(item.get("id") == mem_id for item in result["items"])
class TestImageMemoryCleanupScenario:
"""Test the specific use case of cleaning up image-based memories."""
@pytest.fixture
def temp_memory_dir(self):
"""Create a temporary directory for memory storage."""
temp_dir = tempfile.mkdtemp()
yield temp_dir
shutil.rmtree(temp_dir, ignore_errors=True)
@pytest.fixture
def file_memory(self, temp_memory_dir):
"""Create a FileMemory instance with temp storage."""
from praisonaiagents.memory import FileMemory
return FileMemory(user_id="test_user", base_path=temp_memory_dir, verbose=0)
def test_image_context_cleanup(self, file_memory):
"""
Test scenario: User asks agent about an image, then needs to
delete that memory to free up context window.
"""
# Simulate storing image-related conversation
image_mem_1 = file_memory.add_short_term(
"[IMAGE] User uploaded screenshot.png",
metadata={"type": "image", "filename": "screenshot.png"}
)
image_mem_2 = file_memory.add_short_term(
"Agent: I see a cat in the image with orange fur.",
metadata={"type": "image_response", "related_to": image_mem_1}
)
# Add some regular memories
regular_mem = file_memory.add_short_term("User: What's the weather today?")
# Get initial memory count
initial_count = len(file_memory.get_short_term())
assert initial_count >= 3
# Delete the image-related memories specifically
deleted = file_memory.delete_memory(image_mem_1)
assert deleted is True
deleted = file_memory.delete_memory(image_mem_2)
assert deleted is True
# Verify regular memory still exists
memories = file_memory.get_short_term()
assert any(m.id == regular_mem for m in memories)
# Verify image memories are gone
assert not any(m.id == image_mem_1 for m in memories)
assert not any(m.id == image_mem_2 for m in memories)
def test_bulk_image_cleanup_by_metadata(self, file_memory):
"""Test bulk cleanup of all image-related memories using query."""
# Add multiple image interaction memories
for i in range(5):
file_memory.add_short_term(
f"[IMAGE] Image analysis {i}",
metadata={"type": "image"}
)
# Add regular memories
for i in range(3):
file_memory.add_short_term(f"Regular message {i}")
# Cleanup all image memories
deleted = file_memory.delete_memories_matching("[IMAGE]")
assert deleted == 5
# Verify regular memories still exist
results = file_memory.search("Regular message")
assert len(results) >= 1
class TestMemoryClassDeletion:
"""Test Memory class (full-featured) delete operations."""
@pytest.fixture
def memory_config(self, tmp_path):
"""Create a minimal memory config for testing."""
return {
"provider": "none",
"short_db": str(tmp_path / "short_term.db"),
"long_db": str(tmp_path / "long_term.db"),
}
def test_memory_class_delete_short_term(self, memory_config):
"""Test Memory class delete_short_term method."""
from praisonaiagents.memory import Memory
memory = Memory(config=memory_config, verbose=0)
# Store a memory (store_short_term doesn't return ID, so we search)
memory.store_short_term("Delete test memory STM")
# Search for it
results = memory.search_short_term("Delete test")
assert len(results) >= 1
# Get the ID
mem_id = results[0].get("id")
assert mem_id is not None
# Delete it
success = memory.delete_short_term(mem_id)
assert success is True
# Verify it's gone (search may still find partial matches)
# The important thing is the specific ID is deleted
def test_memory_class_delete_long_term(self, memory_config):
"""Test Memory class delete_long_term method."""
from praisonaiagents.memory import Memory
memory = Memory(config=memory_config, verbose=0)
# Store a memory
memory.store_long_term("Delete test memory LTM")
# Search for it
results = memory.search_long_term("Delete test")
assert len(results) >= 1
# Get the ID
mem_id = results[0].get("id")
assert mem_id is not None
# Delete it
success = memory.delete_long_term(mem_id)
assert success is True
class TestDeletableProtocol:
"""Test DeletableMemoryProtocol compliance."""
def test_file_memory_implements_protocol(self):
"""Test that FileMemory implements DeletableMemoryProtocol."""
from praisonaiagents.memory import FileMemory, DeletableMemoryProtocol
# Check protocol compliance
assert hasattr(FileMemory, 'delete_memory')
assert hasattr(FileMemory, 'delete_memories')
def test_memory_class_implements_protocol(self):
"""Test that Memory class implements DeletableMemoryProtocol."""
from praisonaiagents.memory import Memory, DeletableMemoryProtocol
# Check protocol compliance
assert hasattr(Memory, 'delete_memory')
assert hasattr(Memory, 'delete_memories')
if __name__ == "__main__":
pytest.main([__file__, "-v"])
+1 -1
View File
@@ -2422,7 +2422,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.12.3"
version = "0.12.4"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
+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/v3.9.4.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.9.4.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.9.7.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.9.7.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+125 -28
View File
@@ -125,7 +125,7 @@ def normalize_action(action_dict: Dict[str, Any]) -> Dict[str, Any]:
return action_dict
# System prompt for browser automation
BROWSER_AGENT_SYSTEM_PROMPT = """You are a precise browser automation agent. Complete tasks in the FEWEST steps possible.
BROWSER_AGENT_SYSTEM_PROMPT = """You are a precise browser automation agent. Complete ALL parts of multi-step tasks.
## Available Actions
- **type**: Type text into INPUT element. Use the EXACT selector from the element list.
@@ -133,7 +133,18 @@ BROWSER_AGENT_SYSTEM_PROMPT = """You are a precise browser automation agent. Com
- **click**: Click BUTTON or LINK. Use the EXACT selector from the element list.
- **scroll**: Scroll page (direction: "up" or "down")
- **navigate**: Go to URL directly (use for known URLs)
- **done**: Task complete - use when goal is achieved
- **done**: Task complete - use ONLY when ALL parts of the goal are achieved
## CRITICAL: Cookie Consent / Overlay Dialogs
⚠️ **IMPORTANT**: If you see elements marked as `consent_button` or with text like "Accept all", "Reject all", "I agree", or similar - you MUST click them FIRST before doing anything else!
These dialogs block all other interactions. Look for:
- Buttons with text containing: "Accept", "Reject", "Agree", "Consent", "OK", "Got it"
- Elements with type="consent_button" in the element list
- Elements at the TOP of the list (consent buttons are prioritized)
**If overlay_info shows detected=true**, you are seeing a consent dialog. Click a consent button immediately!
## CRITICAL: Valid Selector Format
You MUST use valid CSS selectors from the element list.
@@ -149,37 +160,53 @@ Use ONLY selectors provided in the element list, such as:
- ✅ button[type="submit"]
- ✅ a[href*="domain.com"]
## CRITICAL: Multi-Step Goals - DO NOT MARK DONE EARLY!
⚠️ **IMPORTANT**: If the goal contains multiple parts separated by commas or "and", you MUST complete ALL parts before marking done!
### Example Multi-Step Goals:
- "go to google, search for AI, click the first link" = 3 STEPS REQUIRED:
1. Navigate to google.com
2. Type "AI" and submit search
3. Click first search result link
→ Only mark done AFTER clicking the link!
- "search for Python and click Wikipedia" = 2 STEPS REQUIRED:
1. Search for Python
2. Click Wikipedia link in results
→ Only mark done AFTER clicking Wikipedia!
### How to Parse Goals:
1. Split goal by commas and "and"
2. Count the number of distinct actions
3. Track which actions you've completed
4. Only mark done when ALL actions are complete
## CRITICAL: When to Use "done"
Set "done": true and "action": "done" when:
- **Search completed**: You typed a query AND submitted AND the results page is showing
- **Navigation completed**: You reached the target URL/page
- **Information visible**: The requested information is now visible on screen
- **Task statement fulfilled**: The goal text describes what you see
Set "done": true ONLY when ALL of these are true:
- **Every part of the goal is complete** - not just the first part!
- **You can see evidence** that the final action succeeded
- **The URL/page reflects** the final state requested
### Examples of DONE states:
- Goal "search praisonai on Google" → DONE when URL shows `/search?q=praisonai`
- Goal "go to github" → DONE when URL shows `github.com`
- Goal "find contact page" → DONE when contact page is visible
### Examples:
- Goal "go to google" → DONE when URL shows `google.com`
- Goal "search AI on google" → DONE when URL shows `/search?q=AI`
- Goal "go to google, search AI, click first link" → DONE when you're on a NEW page after clicking a search result (NOT on google.com/search)
⚠️ DO NOT keep clicking links or typing after goal is achieved!
⚠️ DO NOT mark done after just navigating if the goal includes searching!
⚠️ DO NOT mark done after just searching if the goal includes clicking!
## Element Types in Observation
- INPUT → type text here
- INPUT/TEXTAREA → type text here
- BUTTON → click to submit
- LINK → click to navigate to another page
## Multi-Step Tasks
When user says "search X and inside that search Y":
1. Type "X" in the search INPUT
2. Press submit (Enter) to search
3. Click a LINK in results to navigate to that website
4. Then search for "Y" on the new page
- LINK (a) → click to navigate to another page
- consent_button → CLICK FIRST to dismiss blocking dialog
## Response Format (JSON only)
```json
{
"thought": "Brief reasoning (1 sentence)",
"thought": "Brief reasoning - what part of the goal am I working on?",
"action": "type|submit|click|scroll|navigate|done",
"selector": "exact selector from element list",
"value": "text to type (for type action)",
@@ -188,9 +215,9 @@ When user says "search X and inside that search Y":
}
```
When task is complete, always include a summary of what you did, e.g.:
- "Searched for 'praisonai' on Google and found results"
- "Navigated to github.com homepage"
When task is complete, always include a summary of ALL steps you did, e.g.:
- "Navigated to Google, searched for 'AI', and clicked the first result"
- "Searched for 'Python' and clicked the Wikipedia link"
CRITICAL: After typing a search query, use "submit" action to press Enter.
"""
@@ -244,7 +271,7 @@ class BrowserAgent:
backstory="You are an expert at navigating websites and performing automated actions.",
instructions=BROWSER_AGENT_SYSTEM_PROMPT,
llm=self.model,
memory=True, # Enable short-term memory for context between steps
memory=False, # Disabled: Each step is a fresh LLM call - prevents token explosion
)
def process_observation(
@@ -279,11 +306,45 @@ class BrowserAgent:
if self.verbose:
logger.info(f"Processing observation for step {observation.get('step_number', 0)}")
# Check if we have a screenshot for vision-based decision making
screenshot_base64 = observation.get('screenshot')
use_vision = screenshot_base64 and 'gpt' in self.model.lower()
# Get agent response with structured output if available
try:
action_model = get_browser_action_model()
if action_model is not None:
if use_vision:
# Use LiteLLM directly for vision-enabled call
from litellm import completion
if self.verbose:
logger.info(f"[VISION] Using screenshot for decision making")
messages = [
{"role": "system", "content": BROWSER_AGENT_SYSTEM_PROMPT},
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{screenshot_base64}"}
},
],
}
]
response = completion(
model=self.model,
messages=messages,
max_tokens=500,
)
response_text = response.choices[0].message.content.strip()
action = self._parse_response(response_text)
elif action_model is not None:
# Use Pydantic model for guaranteed structured output
response = self._agent.chat(prompt, output_pydantic=action_model)
@@ -323,15 +384,37 @@ class BrowserAgent:
"""Build prompt from observation with full context."""
# Start with prominent goal reminder
original_goal = observation.get('original_goal') or observation.get('task', '')
# Parse multi-step goal into parts
goal_parts = []
if original_goal:
# Split by comma and "and" to identify sub-goals
import re
# Split by comma or " and " but keep track
raw_parts = re.split(r',\s*|\s+and\s+', original_goal.lower())
goal_parts = [p.strip() for p in raw_parts if p.strip()]
parts = [
"=" * 50,
f"🎯 ORIGINAL GOAL: {original_goal}",
]
# Show goal breakdown for multi-step tasks
if len(goal_parts) > 1:
parts.append("")
parts.append("📋 GOAL BREAKDOWN (complete ALL parts):")
for i, gp in enumerate(goal_parts, 1):
parts.append(f" {i}. {gp}")
parts.append("")
parts.append("⚠️ DO NOT mark done until ALL parts above are complete!")
parts.extend([
"=" * 50,
"",
f"**Current URL:** {observation.get('url', '')}",
f"**Page Title:** {observation.get('title', '')}",
f"**Step:** {observation.get('step_number', 0)} / {self.max_steps}",
]
])
# Add progress notes if present
progress = observation.get('progress_notes', '')
@@ -346,6 +429,20 @@ class BrowserAgent:
status = "✓" if ah.get('success') else "✗"
parts.append(f" {status} {ah.get('action')}: {ah.get('selector', '')[:40]}")
# *** OVERLAY/CONSENT DIALOG DETECTION ***
overlay_info = observation.get('overlay_info')
if overlay_info and overlay_info.get('detected'):
parts.append("\n" + "🚨" * 20)
parts.append("🍪 COOKIE CONSENT / OVERLAY DIALOG DETECTED!")
parts.append(f" Type: {overlay_info.get('type', 'unknown')}")
if overlay_info.get('selector'):
parts.append(f" Selector: {overlay_info.get('selector')}")
parts.append("")
parts.append(" ⚠️ YOU MUST CLICK A CONSENT BUTTON FIRST!")
parts.append(" Look for buttons with text like 'Accept all', 'Reject all', 'I agree'")
parts.append(" These are typically at the TOP of the element list with type='consent_button'")
parts.append("🚨" * 20)
# *** AUTO-DEBUG: Detect stuck agent patterns ***
step = observation.get('step_number', 0)
elements = observation.get("elements", [])
+423 -28
View File
@@ -67,6 +67,8 @@ class BrowserTestResult:
error: str = ""
session_id: str = ""
engine: str = "cdp"
video_url: str = "" # Path to recorded video (WebM)
screenshots_dir: str = "" # Directory with step screenshots
@property
def efficiency(self) -> float:
@@ -95,6 +97,8 @@ class BrowserTestResult:
"error": self.error,
"session_id": self.session_id,
"engine": self.engine,
"video_url": self.video_url,
"screenshots_dir": self.screenshots_dir,
}
@@ -283,7 +287,14 @@ QUICK_TEST_SCENARIOS = [
class BrowserBenchmark:
"""Browser automation benchmark runner."""
"""Browser automation benchmark runner.
Features:
- Screenshot capture per test
- Video recording (via extension)
- CSV file input for custom scenarios
- Ideal vs actual step tracking
"""
def __init__(
self,
@@ -291,17 +302,262 @@ class BrowserBenchmark:
port: int = 9222,
model: str = "gpt-4o-mini",
verbose: bool = False,
capture_screenshots: bool = True,
debug: bool = False,
):
self.engine = engine
self.port = port
self.model = model
self.verbose = verbose
self.capture_screenshots = capture_screenshots
self.debug = debug
self.results_dir = Path.home() / ".praisonai" / "browser_benchmarks"
self.results_dir.mkdir(parents=True, exist_ok=True)
self.videos_dir = self.results_dir / "videos"
self.videos_dir.mkdir(parents=True, exist_ok=True)
self.screenshots_base = self.results_dir / "screenshots"
self.screenshots_base.mkdir(parents=True, exist_ok=True)
# Process references for cleanup
self._server_process = None
self._chrome_process = None
self._temp_profile = None
self._bridge_port = 8765
async def run_single_test(self, test_case: BrowserTestCase) -> BrowserTestResult:
"""Run a single test case."""
async def _check_bridge_server(self) -> bool:
"""Check if bridge server is running."""
try:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(f"http://localhost:{self._bridge_port}/health",
timeout=aiohttp.ClientTimeout(total=2)) as resp:
return resp.status == 200
except Exception:
return False
async def _check_cdp_available(self) -> bool:
"""Check if Chrome CDP is available."""
try:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(f"http://localhost:{self.port}/json",
timeout=aiohttp.ClientTimeout(total=2)) as resp:
if resp.status == 200:
targets = await resp.json()
return len(targets) > 0
return False
except Exception:
return False
async def _ensure_bridge_server(self) -> bool:
"""Ensure bridge server is running, start if not."""
import subprocess
import sys
if await self._check_bridge_server():
if self.verbose:
print(f"✓ Bridge server already running on port {self._bridge_port}")
return True
if self.verbose:
print(f"Starting bridge server on port {self._bridge_port}...")
try:
self._server_process = subprocess.Popen(
[sys.executable, "-m", "praisonai.browser.server", "--port", str(self._bridge_port)],
stdout=subprocess.PIPE if not self.debug else None,
stderr=subprocess.PIPE if not self.debug else None,
start_new_session=True,
)
# Wait for server to start
for _ in range(10):
await asyncio.sleep(0.5)
if await self._check_bridge_server():
if self.verbose:
print("✓ Bridge server started")
return True
print("⚠️ Bridge server may not have started properly")
return False
except Exception as e:
print(f"❌ Failed to start bridge server: {e}")
return False
async def _ensure_chrome_with_extension(self) -> bool:
"""Ensure Chrome is running with the extension loaded."""
import subprocess
import platform
import tempfile
import os
# First check if CDP is already available with extension
if await self._check_cdp_available():
if self.verbose:
print(f"✓ Chrome already running with CDP on port {self.port}")
return True
# Find Chrome
system = platform.system()
chrome_path = None
if system == "Darwin":
for path in ["/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
"/Applications/Chromium.app/Contents/MacOS/Chromium"]:
if os.path.exists(path):
chrome_path = path
break
elif system == "Linux":
for name in ["google-chrome", "chromium", "chromium-browser"]:
import shutil
path = shutil.which(name)
if path:
chrome_path = path
break
if not chrome_path:
print("❌ Chrome not found")
return False
# Find extension
extension_path = None
candidates = [
os.path.expanduser("~/praisonai-chrome-extension/dist"),
str(Path(__file__).parent.parent.parent.parent.parent / "praisonai-chrome-extension" / "dist"),
]
for cand in candidates:
if os.path.isdir(cand) and os.path.exists(os.path.join(cand, "manifest.json")):
extension_path = os.path.abspath(cand)
break
if not extension_path:
print("❌ Extension not found. Build it with: cd ~/praisonai-chrome-extension && npm run build")
return False
# Create temp profile
self._temp_profile = tempfile.mkdtemp(prefix="praisonai_benchmark_")
# Build Chrome args
chrome_args = [
chrome_path,
f"--load-extension={extension_path}",
f"--user-data-dir={self._temp_profile}",
f"--remote-debugging-port={self.port}",
"--enable-extensions",
"--no-first-run",
"--no-default-browser-check",
"--disable-default-apps",
"--disable-popup-blocking",
"--disable-translate",
"--disable-background-timer-throttling",
"--disable-renderer-backgrounding",
"https://www.google.com",
]
if self.verbose:
print(f"Launching Chrome with extension from {extension_path}...")
try:
self._chrome_process = subprocess.Popen(
chrome_args,
stdout=subprocess.PIPE if not self.debug else None,
stderr=subprocess.PIPE if not self.debug else None,
)
# Wait for Chrome to start
for _ in range(15):
await asyncio.sleep(0.5)
if await self._check_cdp_available():
if self.verbose:
print("✓ Chrome started with extension")
return True
print("⚠️ Chrome may not have started properly")
return False
except Exception as e:
print(f"❌ Failed to start Chrome: {e}")
return False
async def _setup_extension_mode(self) -> bool:
"""Setup extension mode infrastructure: bridge server + Chrome with extension."""
if self.verbose:
print("\n[Extension Mode Setup]")
# Start bridge server
if not await self._ensure_bridge_server():
return False
# Start Chrome with extension
if not await self._ensure_chrome_with_extension():
return False
# Wait a bit for extension to initialize
await asyncio.sleep(2)
if self.verbose:
print("✓ Extension mode ready\n")
return True
def _cleanup_extension_mode(self):
"""Cleanup extension mode processes."""
import shutil
if self._chrome_process:
try:
self._chrome_process.terminate()
self._chrome_process.wait(timeout=5)
except Exception:
pass
if self._server_process:
try:
self._server_process.terminate()
self._server_process.wait(timeout=5)
except Exception:
pass
if self._temp_profile:
try:
shutil.rmtree(self._temp_profile, ignore_errors=True)
except Exception:
pass
@staticmethod
def load_from_csv(csv_path: str) -> List[BrowserTestCase]:
"""Load test scenarios from CSV file.
CSV format:
name,goal,url,ideal_steps,max_steps,category
Example:
Simple Google,Go to google.com,https://google.com,1,10,simple
"""
import csv
scenarios = []
with open(csv_path, 'r', newline='', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
scenarios.append(BrowserTestCase(
name=row.get('name', 'Unnamed Test'),
goal=row.get('goal', ''),
url=row.get('url', 'https://www.google.com'),
ideal_steps=int(row.get('ideal_steps', 3)),
max_steps=int(row.get('max_steps', 20)),
category=row.get('category', 'custom'),
))
if not scenarios:
raise ValueError(f"No test scenarios found in {csv_path}")
return scenarios
async def run_single_test(self, test_case: BrowserTestCase, test_index: int = 0) -> BrowserTestResult:
"""Run a single test case with screenshot capture."""
from .cdp_agent import CDPBrowserAgent
import os
if self.verbose:
print(f"\n{'='*60}")
@@ -310,22 +566,55 @@ class BrowserBenchmark:
print(f"Ideal steps: {test_case.ideal_steps}")
print(f"{'='*60}")
# Setup screenshots directory for this test
test_screenshots_dir = None
if self.capture_screenshots:
test_safe_name = test_case.name.replace(" ", "_").replace("/", "_")[:30]
timestamp = datetime.now().strftime("%H%M%S")
test_screenshots_dir = str(self.screenshots_base / f"{test_index:02d}_{test_safe_name}_{timestamp}")
os.makedirs(test_screenshots_dir, exist_ok=True)
start_time = time.perf_counter()
try:
agent = CDPBrowserAgent(
port=self.port,
model=self.model,
max_steps=test_case.max_steps,
verbose=self.verbose,
max_retries=3,
record_session=True,
)
result = await agent.run(test_case.goal, test_case.url)
# Choose engine based on configuration
if self.engine in ("extension", "hybrid"):
# Use run_hybrid for extension/hybrid mode
from .cdp_agent import run_hybrid
if self.verbose:
print(f" Engine: {self.engine} (via run_hybrid)")
result = await run_hybrid(
goal=test_case.goal,
url=test_case.url,
model=self.model,
max_steps=test_case.max_steps,
verbose=self.verbose,
prefer_extension=(self.engine == "extension"),
)
else:
# Use CDP agent directly for cdp mode
from .cdp_agent import CDPBrowserAgent
if self.verbose:
print(f" Engine: cdp (CDPBrowserAgent)")
agent = CDPBrowserAgent(
port=self.port,
model=self.model,
max_steps=test_case.max_steps,
verbose=self.verbose,
max_retries=3,
record_session=True,
screenshot_dir=test_screenshots_dir, # Enable screenshot capture
debug=self.debug,
)
result = await agent.run(test_case.goal, test_case.url)
duration_ms = (time.perf_counter() - start_time) * 1000
# Get video URL if available (from session)
video_url = result.get("video_url", "")
test_result = BrowserTestResult(
test_case=test_case,
success=result.get("success", False),
@@ -337,6 +626,8 @@ class BrowserBenchmark:
error=result.get("error", ""),
session_id=result.get("session_id", ""),
engine=self.engine,
video_url=video_url,
screenshots_dir=test_screenshots_dir or "",
)
if self.verbose:
@@ -346,6 +637,10 @@ class BrowserBenchmark:
print(f" Efficiency: {test_result.efficiency:.0%}")
print(f" Retries: {test_result.total_retries}")
print(f" Duration: {test_result.duration_ms:.0f}ms")
if test_screenshots_dir:
print(f" Screenshots: {test_screenshots_dir}")
if video_url:
print(f" Video: {video_url}")
return test_result
@@ -362,29 +657,43 @@ class BrowserBenchmark:
duration_ms=duration_ms,
error=str(e),
engine=self.engine,
screenshots_dir=test_screenshots_dir or "",
)
async def run_all_tests(self, scenarios: Optional[List[BrowserTestCase]] = None) -> BrowserBenchmarkReport:
"""Run all test scenarios."""
if scenarios is None:
scenarios = BROWSER_TEST_SCENARIOS
# Setup extension mode infrastructure if needed
needs_extension_setup = self.engine in ("extension", "hybrid")
if needs_extension_setup:
if not await self._setup_extension_mode():
print("❌ Failed to setup extension mode, falling back to CDP")
self.engine = "cdp"
report = BrowserBenchmarkReport(
timestamp=datetime.now().isoformat(),
engine=self.engine,
)
for test_case in scenarios:
result = await self.run_single_test(test_case)
report.results.append(result)
try:
for idx, test_case in enumerate(scenarios):
result = await self.run_single_test(test_case, test_index=idx)
report.results.append(result)
# Brief pause between tests
await asyncio.sleep(1)
# Brief pause between tests
await asyncio.sleep(1)
report.compute_stats()
# Save results
self._save_report(report)
report.compute_stats()
# Save results
self._save_report(report)
finally:
# Cleanup if we started extension infrastructure
if needs_extension_setup:
self._cleanup_extension_mode()
return report
@@ -435,7 +744,7 @@ class BrowserBenchmark:
return report
def print_report(self, report: BrowserBenchmarkReport):
"""Print formatted benchmark report."""
"""Print formatted benchmark report with media links."""
print("\n" + "="*70)
print("BROWSER BENCHMARK REPORT")
print("="*70)
@@ -465,7 +774,7 @@ class BrowserBenchmark:
print(f"{cat:<15} {stats['success_rate']:.0%} ({stats['passed']}/{stats['total']}) {stats['mean_efficiency']:.0%} {stats['mean_overhead']:+.1f}")
print()
# Individual results
# Individual results with media
print(f"{'INDIVIDUAL RESULTS':^70}")
print("-"*70)
print(f"{'Test Name':<30} {'Status':<8} {'Steps':<12} {'Eff':<8}")
@@ -475,10 +784,31 @@ class BrowserBenchmark:
steps = f"{r.actual_steps}/{r.test_case.ideal_steps}" if r.success else "N/A"
eff = f"{r.efficiency:.0%}" if r.success else "-"
print(f"{r.test_case.name[:28]:<30} {status:<8} {steps:<12} {eff:<8}")
# Show error for failures
if not r.success and r.error:
print(f" └─ Error: {r.error[:60]}")
# Show media links
if r.screenshots_dir:
print(f" 📸 Screenshots: {r.screenshots_dir}")
if r.video_url:
print(f" 🎬 Video: {r.video_url}")
print("="*70)
# Media summary
videos = [r for r in report.results if r.video_url]
screenshots = [r for r in report.results if r.screenshots_dir]
if videos or screenshots:
print(f"\n{'MEDIA FILES':^70}")
print("-"*70)
if screenshots:
print(f" Screenshots: {len(screenshots)} test(s) with screenshots")
print(f" Base Dir: {self.screenshots_base}")
if videos:
print(f" Videos: {len(videos)} test(s) with video recording")
print("="*70)
# CLI commands for browser benchmark
@@ -490,17 +820,51 @@ def add_benchmark_commands(app):
@benchmark_app.command("run")
def run_benchmark(
csv_file: Optional[str] = typer.Argument(None, help="Optional CSV file with test scenarios"),
quick: bool = typer.Option(False, "--quick", "-q", help="Run quick test (3 scenarios)"),
verbose: bool = typer.Option(True, "--verbose", "-v", help="Verbose output"),
engine: str = typer.Option("cdp", "--engine", help="Engine: cdp, extension, hybrid"),
output: Optional[str] = typer.Option(None, "--output", "-o", help="Save JSON to file"),
no_screenshots: bool = typer.Option(False, "--no-screenshots", help="Disable screenshot capture"),
debug: bool = typer.Option(False, "--debug", "-d", help="Enable debug mode with detailed logging"),
record_video: bool = typer.Option(False, "--record-video", help="Create video from screenshots after benchmark"),
):
"""Run browser automation benchmark suite."""
"""Run browser automation benchmark suite.
Examples:
praisonai browser benchmark run # Run all built-in tests
praisonai browser benchmark run --quick # Run 3 quick tests
praisonai browser benchmark run tests.csv # Run tests from CSV file
praisonai browser benchmark run --output results.json
CSV Format (with header):
name,goal,url,ideal_steps,max_steps,category
My Test,Go to google,https://google.com,1,10,simple
"""
import logging
if debug:
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("praisonai.browser.cdp_agent").setLevel(logging.DEBUG)
async def _run():
benchmark = BrowserBenchmark(engine=engine, verbose=verbose)
benchmark = BrowserBenchmark(
engine=engine,
verbose=verbose,
capture_screenshots=not no_screenshots,
debug=debug,
)
if quick:
# Load scenarios from CSV or use built-in
if csv_file:
try:
scenarios = BrowserBenchmark.load_from_csv(csv_file)
print(f"Loaded {len(scenarios)} test scenarios from {csv_file}")
except Exception as e:
print(f"Error loading CSV: {e}")
return
report = await benchmark.run_all_tests(scenarios)
elif quick:
report = await benchmark.run_quick()
else:
report = await benchmark.run_all_tests()
@@ -553,4 +917,35 @@ def add_benchmark_commands(app):
print("-"*70)
print(f"Total: {len(BROWSER_TEST_SCENARIOS)} scenarios")
@benchmark_app.command("template")
def generate_template(
output: str = typer.Argument("browser_tests.csv", help="Output CSV filename"),
):
"""Generate a CSV template for custom test scenarios.
Example:
praisonai browser benchmark template my_tests.csv
"""
import csv
with open(output, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
# Header
writer.writerow(['name', 'goal', 'url', 'ideal_steps', 'max_steps', 'category'])
# Example rows
writer.writerow(['Simple Google', 'Go to google.com and confirm page loaded', 'https://www.google.com', '1', '10', 'simple'])
writer.writerow(['Google Search', 'Type "PraisonAI" in search box, press Enter, confirm results', 'https://www.google.com', '3', '15', 'search'])
writer.writerow(['Wikipedia Article', 'Search for AI, click article, scroll down', 'https://www.wikipedia.org', '5', '20', 'navigation'])
print(f"CSV template saved to: {output}")
print("\nTemplate contains 3 example test scenarios. Edit this file to add your own.")
print("\nColumns:")
print(" name - Test name (displayed in report)")
print(" goal - Task description for the browser agent")
print(" url - Starting URL")
print(" ideal_steps - Expected number of steps for ideal completion")
print(" max_steps - Maximum allowed steps before failure")
print(" category - Test category (simple, search, navigation, form, spa)")
app.add_typer(benchmark_app, name="benchmark")
+794 -20
View File
@@ -18,6 +18,66 @@ from dataclasses import dataclass
logger = logging.getLogger("praisonai.browser.cdp_agent")
# Suppress verbose LiteLLM and HTTP client debug logs
logging.getLogger("LiteLLM").setLevel(logging.WARNING)
logging.getLogger("litellm").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("openai").setLevel(logging.WARNING)
# Site-specific fallback selectors for common websites
GOOGLE_SEARCH_SELECTORS = [
'textarea[name="q"]',
'input[name="q"]',
'#APjFqb',
'[aria-label="Search"]',
'input[type="text"]',
'textarea[title="Search"]',
]
GOOGLE_SUBMIT_SELECTORS = [
'input[name="btnK"]',
'button[type="submit"]',
'[aria-label="Google Search"]',
'input[value="Google Search"]',
]
# Cookie consent dialog selectors - these dialogs block interaction
GOOGLE_CONSENT_SELECTORS = [
# English
'button:has-text("Accept all")', # Not valid CSS, but we use text matching
'button:has-text("Reject all")',
# Fallback to aria-label and button text content matching via JS
'[aria-label="Accept all"]',
'[aria-label="Reject all"]',
# Common button patterns in Google consent
'button[jsname="b3VHJd"]', # "Accept all" button jsname
'button[jsname="cOuczf"]', # "Reject all" button jsname
# Div-based buttons (Google uses divs with role="button")
'[role="button"][aria-label*="Accept"]',
'[role="button"][aria-label*="Reject"]',
]
# Generic consent/cookie dialog patterns for other sites
GENERIC_CONSENT_SELECTORS = [
# Common consent button patterns
'button[id*="accept"]',
'button[id*="consent"]',
'button[class*="accept"]',
'button[class*="consent"]',
'[data-testid*="accept"]',
'[data-testid*="consent"]',
# GDPR-specific
'#onetrust-accept-btn-handler',
'.cc-accept',
'#CybotCookiebotDialogBodyLevelButtonLevelOptinAllowAll',
# Common text-based (will use JS text matching)
'button:has-text("I agree")',
'button:has-text("Got it")',
'button:has-text("OK")',
]
@dataclass
class CDPPageState:
@@ -26,6 +86,7 @@ class CDPPageState:
title: str
elements: List[Dict[str, Any]]
viewport: Dict[str, int]
overlay_info: Optional[Dict[str, Any]] = None # Info about blocking overlays/dialogs
class CDPBrowserAgent:
@@ -52,6 +113,9 @@ class CDPBrowserAgent:
enable_vision: bool = False,
record_session: bool = True,
screenshot_dir: Optional[str] = None,
debug: bool = False,
record_video: bool = False,
video_fps: int = 5,
):
"""Initialize CDP Browser Agent.
@@ -64,6 +128,9 @@ class CDPBrowserAgent:
enable_vision: Enable vision-based element detection
record_session: Log all actions to session database
screenshot_dir: Directory to save screenshots (optional)
debug: Enable debug mode with detailed logging
record_video: Enable real-time video recording via CDP screencast
video_fps: Frames per second for video recording (default 5)
"""
self.port = port
self.model = model
@@ -73,12 +140,19 @@ class CDPBrowserAgent:
self.enable_vision = enable_vision
self.record_session = record_session
self.screenshot_dir = screenshot_dir
self.debug = debug
self.record_video = record_video
self.video_fps = video_fps
self.ws = None
self._message_id = 0
self._pending: Dict[int, asyncio.Future] = {}
self._session_manager = None
self._current_session_id: Optional[str] = None
self._total_retries = 0
self._action_history: List[Dict[str, Any]] = [] # Track actions for stuck detection
self._last_url: str = "" # Track URL changes
self._video_encoder = None # FFmpegVideoEncoder instance
self._screencast_active = False
async def _get_targets(self) -> List[Dict]:
"""Get Chrome debugger targets."""
@@ -102,10 +176,15 @@ class CDPBrowserAgent:
targets = await self._get_targets()
# Find an existing page or background page
# Find an existing page target
# Accept even chrome:// pages like newtab since we'll navigate away
page = None
for t in targets:
if t.get("type") == "page" and not t.get("url", "").startswith("chrome"):
if t.get("type") == "page":
url = t.get("url", "")
# Skip extension pages but allow newtab and other regular pages
if url.startswith("chrome-extension://"):
continue
page = t
break
@@ -172,6 +251,8 @@ class CDPBrowserAgent:
- Multiple selector strategies per element
- aria-label and data attributes
- Text content for text-based fallback
- Overlay/modal detection for consent dialogs
- Prioritized consent button extraction
"""
# Get URL and title
result = await self._send("Runtime.evaluate", {
@@ -179,9 +260,102 @@ class CDPBrowserAgent:
})
info = json.loads(result.get("result", {}).get("value", "{}"))
# Enhanced interactive elements extraction
# Enhanced interactive elements extraction with overlay detection
elements_js = """
(() => {
const result = { elements: [], overlay: null };
// Detect blocking overlay/modal (cookie consent, etc.)
const overlaySelectors = [
'[role="dialog"]',
'[role="alertdialog"]',
'[aria-modal="true"]',
'.consent-modal',
'.cookie-consent',
'#consent',
'[class*="consent"]',
'[class*="cookie-banner"]',
'[class*="cookie-notice"]',
'[id*="consent"]',
// Google-specific consent
'div[jsname="cpjPV"]',
'div[data-g-class="modal"]',
];
for (const sel of overlaySelectors) {
try {
const overlay = document.querySelector(sel);
if (overlay && overlay.offsetParent !== null) {
const rect = overlay.getBoundingClientRect();
if (rect.width > 200 && rect.height > 100) {
result.overlay = {
detected: true,
type: 'consent_dialog',
selector: sel,
text: overlay.textContent?.slice(0, 200) || ''
};
break;
}
}
} catch(e) {}
}
// Check for consent-related keywords in page
const pageText = document.body?.textContent?.toLowerCase() || '';
const hasConsentKeywords = ['cookie', 'consent', 'accept all', 'reject all', 'before you continue'].some(
kw => pageText.includes(kw)
);
if (hasConsentKeywords && !result.overlay) {
result.overlay = { detected: true, type: 'possible_consent', selector: null, text: 'Consent keywords detected' };
}
// Find consent buttons FIRST (prioritize these)
const consentButtonSelectors = [
'button',
'[role="button"]',
'div[tabindex="0"]',
];
const consentKeywords = ['accept all', 'reject all', 'accept', 'agree', 'i agree', 'consent', 'got it', 'ok', 'allow'];
const consentButtons = [];
for (const sel of consentButtonSelectors) {
document.querySelectorAll(sel).forEach(el => {
if (el.offsetParent === null) return;
const rect = el.getBoundingClientRect();
if (rect.width === 0 || rect.height === 0) return;
const text = (el.textContent || el.ariaLabel || '').trim().toLowerCase();
const isConsent = consentKeywords.some(kw => text.includes(kw));
if (isConsent) {
// Build selector for this consent button
let bestSelector = '';
if (el.id) bestSelector = '#' + el.id;
else if (el.ariaLabel) bestSelector = `[aria-label="${el.ariaLabel}"]`;
else if (el.getAttribute('jsname')) bestSelector = `[jsname="${el.getAttribute('jsname')}"]`;
else if (el.className && typeof el.className === 'string') {
const cls = el.className.split(' ')[0];
if (cls) bestSelector = `${el.tagName.toLowerCase()}.${cls}`;
}
consentButtons.push({
index: -1, // Will be set later
tag: el.tagName.toLowerCase(),
type: 'consent_button',
text: (el.textContent || '').trim().slice(0, 80),
selector: bestSelector,
selectors: [bestSelector].filter(Boolean),
href: null,
placeholder: '',
rect: {x: rect.x, y: rect.y, width: rect.width, height: rect.height},
isConsentButton: true
});
}
});
}
// Now get regular elements
const selectors = 'a, button, input, textarea, select, [onclick], [role="button"], [role="link"], [tabindex="0"]';
const elements = [];
document.querySelectorAll(selectors).forEach((el, i) => {
@@ -218,16 +392,21 @@ class CDPBrowserAgent:
selectorStrategies.push(`[aria-label="${el.ariaLabel}"]`);
}
// 5. Link with href
// 5. jsname (Google-specific)
const jsname = el.getAttribute('jsname');
if (jsname) {
selectorStrategies.push(`[jsname="${jsname}"]`);
}
// 6. Link with href
if (el.tagName === 'A' && el.href) {
// Use partial href match for stability
const href = el.getAttribute('href');
if (href && !href.startsWith('javascript:')) {
selectorStrategies.push(`a[href="${href}"]`);
}
}
// 6. First class name (less precise but common)
// 7. First class name (less precise but common)
if (el.className && typeof el.className === 'string') {
const firstClass = el.className.split(' ')[0];
if (firstClass && !/^[0-9]/.test(firstClass)) {
@@ -240,25 +419,38 @@ class CDPBrowserAgent:
tag: el.tagName.toLowerCase(),
type: el.type || '',
text: text,
// Provide the best selector (first in priority list)
selector: selectorStrategies[0] || '',
// Also provide all strategies for fallback
selectors: selectorStrategies,
// Include actual href for links
href: el.tagName === 'A' ? el.href : null,
// For form elements
placeholder: el.placeholder || '',
rect: {x: rect.x, y: rect.y, width: rect.width, height: rect.height}
});
});
return elements.slice(0, 50); // Increased limit for better coverage
// If overlay detected, put consent buttons first
if (result.overlay?.detected && consentButtons.length > 0) {
// Reindex and prepend consent buttons
consentButtons.forEach((btn, idx) => btn.index = idx);
elements.forEach((el, idx) => el.index = consentButtons.length + idx);
result.elements = [...consentButtons, ...elements].slice(0, 60);
} else {
result.elements = elements.slice(0, 50);
}
return result;
})()
"""
result = await self._send("Runtime.evaluate", {
"expression": f"JSON.stringify({elements_js})",
"returnByValue": True
})
elements = json.loads(result.get("result", {}).get("value", "[]"))
page_data = json.loads(result.get("result", {}).get("value", "{}"))
elements = page_data.get("elements", [])
overlay_info = page_data.get("overlay")
# Log overlay detection in debug mode
if self.debug and overlay_info and overlay_info.get("detected"):
logger.info(f"[DEBUG] Overlay/dialog detected: {overlay_info.get('type')} - {overlay_info.get('selector')}")
# Get viewport
result = await self._send("Runtime.evaluate", {
@@ -270,9 +462,99 @@ class CDPBrowserAgent:
url=info.get("url", ""),
title=info.get("title", ""),
elements=elements,
viewport=viewport
viewport=viewport,
overlay_info=overlay_info
)
def _is_stuck(self, current_url: str) -> bool:
"""Detect if agent is stuck in a loop.
Returns True if:
- Same URL for 3+ steps with failures
- Same selector attempted 3+ times
- No URL change in 5+ steps
"""
if len(self._action_history) < 3:
return False
last_3 = self._action_history[-3:]
# Check for same selector 3 times
selectors = [a.get("selector") for a in last_3 if a.get("selector")]
if len(selectors) >= 3 and len(set(selectors)) == 1:
if self.debug:
logger.info(f"[DEBUG] Stuck: Same selector 3 times: {selectors[0]}")
return True
# Check for 3 consecutive failures
failures = sum(1 for a in last_3 if not a.get("success", True))
if failures >= 3:
if self.debug:
logger.info("[DEBUG] Stuck: 3 consecutive failures")
return True
# Check for no URL change in 5+ steps
if len(self._action_history) >= 5:
last_5_urls = [a.get("url") for a in self._action_history[-5:]]
if len(set(last_5_urls)) == 1 and last_5_urls[0] == current_url:
# Only stuck if we've had failures
recent_failures = sum(1 for a in self._action_history[-5:] if not a.get("success", True))
if recent_failures >= 2:
if self.debug:
logger.info("[DEBUG] Stuck: Same URL for 5 steps with failures")
return True
return False
def _get_site_specific_selector(self, action: Dict, url: str, overlay_info: Optional[Dict] = None) -> Optional[str]:
"""Get site-specific fallback selector for common websites.
Args:
action: Current action dict
url: Current page URL
overlay_info: Overlay detection info from page state
Returns:
Fallback selector to try, or None
"""
action_type = action.get("action", "").lower()
# PRIORITY 1: If overlay/consent dialog detected, return consent button selector
if overlay_info and overlay_info.get("detected"):
if self.debug:
logger.info(f"[DEBUG] Overlay detected, trying consent button selectors")
# Return first available consent button selector for clicking
if "google.com" in url:
for selector in GOOGLE_CONSENT_SELECTORS:
if self.debug:
logger.info(f"[DEBUG] Trying Google consent selector: {selector}")
# Skip pseudo-selectors that aren't valid CSS
if ":has-text" not in selector:
return selector
else:
for selector in GENERIC_CONSENT_SELECTORS:
if self.debug:
logger.info(f"[DEBUG] Trying generic consent selector: {selector}")
if ":has-text" not in selector:
return selector
# PRIORITY 2: Google-specific selectors for normal actions
if "google.com" in url:
if action_type == "type":
# Try Google search box selectors
for selector in GOOGLE_SEARCH_SELECTORS:
if self.debug:
logger.info(f"[DEBUG] Trying Google selector: {selector}")
return selector
elif action_type == "click" and "search" in action.get("thought", "").lower():
# Try Google submit button selectors
for selector in GOOGLE_SUBMIT_SELECTORS:
if self.debug:
logger.info(f"[DEBUG] Trying Google submit selector: {selector}")
return selector
return None
async def _execute_action(self, action: Dict) -> Dict[str, Any]:
"""Execute action via CDP."""
action_type = action.get("action", "").lower()
@@ -380,6 +662,343 @@ class CDPBrowserAgent:
except Exception as e:
return {"success": False, "error": str(e)}
async def _start_screencast(self, output_dir: str) -> bool:
"""Start CDP screencast for real-time video recording.
Args:
output_dir: Directory to save video output
Returns:
True if screencast started successfully
"""
if not self.record_video:
return False
try:
from .video import FFmpegVideoEncoder
import os
os.makedirs(output_dir, exist_ok=True)
video_path = os.path.join(output_dir, "recording.webm")
self._video_encoder = FFmpegVideoEncoder(
output_path=video_path,
fps=self.video_fps,
width=1280,
height=720,
)
if not self._video_encoder.available:
logger.warning("FFmpeg not available, falling back to screenshot-based recording")
self._video_encoder = None
return False
if not self._video_encoder.start():
self._video_encoder = None
return False
# Start CDP screencast
await self._send("Page.startScreencast", {
"format": "jpeg",
"quality": 80,
"maxWidth": 1280,
"maxHeight": 720,
"everyNthFrame": max(1, 60 // self.video_fps), # Sample from 60fps
})
self._screencast_active = True
if self.debug:
logger.info(f"[DEBUG] Started video recording to {video_path}")
return True
except Exception as e:
logger.warning(f"Failed to start screencast: {e}")
return False
async def _handle_screencast_frame(self, frame_data: str, session_id: int) -> None:
"""Handle incoming screencast frame.
Args:
frame_data: Base64-encoded image data
session_id: Frame session ID for acknowledgment
"""
if not self._screencast_active or not self._video_encoder:
return
try:
import base64
# Decode and write frame
image_bytes = base64.b64decode(frame_data)
self._video_encoder.write_frame(image_bytes)
# Acknowledge frame to continue receiving
await self._send("Page.screencastFrameAck", {"sessionId": session_id})
except Exception as e:
if self.debug:
logger.warning(f"Failed to handle screencast frame: {e}")
async def _capture_screencast_frame(self) -> None:
"""Capture a single screencast frame (for polling mode)."""
if not self._screencast_active or not self._video_encoder:
return
try:
result = await self._send("Page.captureScreenshot", {"format": "jpeg", "quality": 80})
if result and result.get("data"):
import base64
image_bytes = base64.b64decode(result["data"])
self._video_encoder.write_frame(image_bytes)
except Exception as e:
if self.debug:
logger.warning(f"Failed to capture frame: {e}")
async def _stop_screencast(self) -> Optional[str]:
"""Stop CDP screencast and finalize video.
Returns:
Path to output video if successful
"""
video_path = None
try:
if self._screencast_active:
await self._send("Page.stopScreencast", {})
self._screencast_active = False
if self._video_encoder:
video_path = self._video_encoder.finish()
if video_path and self.debug:
logger.info(f"[DEBUG] Video saved: {video_path}")
self._video_encoder = None
except Exception as e:
logger.warning(f"Error stopping screencast: {e}")
return video_path
async def _capture_vision_screenshot(self, name_prefix: str = "step") -> Optional[str]:
"""Capture screenshot for LLM vision analysis.
This method captures a high-quality screenshot that will be sent to the LLM
for visual understanding of the current page state.
Args:
name_prefix: Prefix for debug logging
Returns:
Base64-encoded PNG image data, or None if capture fails
"""
try:
result = await self._send("Page.captureScreenshot", {
"format": "png",
"quality": 100,
"captureBeyondViewport": False, # Only visible viewport
})
if result and result.get("data"):
if self.debug:
logger.info(f"[DEBUG] Vision screenshot captured: {name_prefix}")
return result["data"]
return None
except Exception as e:
if self.debug:
logger.warning(f"[DEBUG] Vision screenshot failed ({name_prefix}): {e}")
return None
async def _verify_action_success(
self,
action: Dict[str, Any],
screenshot_before: Optional[str],
screenshot_after: Optional[str],
expected_change: Optional[str] = None,
) -> Dict[str, Any]:
"""Verify action success by comparing before/after screenshots with LLM.
Uses vision-capable LLM to analyze whether the action produced the expected
visual change on the page.
Args:
action: The action that was executed
screenshot_before: Base64 screenshot before action
screenshot_after: Base64 screenshot after action
expected_change: Optional description of expected change
Returns:
Dict with verification result: {success: bool, confidence: float, reason: str}
"""
if not screenshot_before or not screenshot_after:
return {"success": True, "confidence": 0.5, "reason": "No screenshots for verification"}
# Skip verification if LLM doesn't support vision
if "gpt" not in self.model.lower() or "o" not in self.model.lower():
return {"success": True, "confidence": 0.5, "reason": "Vision not supported by model"}
try:
# Use LiteLLM for verification
from litellm import completion
action_type = action.get("action", "unknown")
action_detail = action.get("text", action.get("selector", ""))
verification_prompt = f"""Compare these two browser screenshots (before and after an action) and verify if the action succeeded.
ACTION PERFORMED: {action_type}
{"ACTION DETAIL: " + action_detail if action_detail else ""}
{"EXPECTED CHANGE: " + expected_change if expected_change else ""}
Analyze both images and determine:
1. Did the page change visually after the action?
2. Is the change consistent with what the action should have done?
3. Are there any error messages or unexpected states?
Respond in this exact JSON format:
{{"success": true/false, "confidence": 0.0-1.0, "reason": "brief explanation"}}"""
response = completion(
model=self.model,
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": verification_prompt},
{
"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{screenshot_before}"}
},
{
"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{screenshot_after}"}
},
],
}
],
max_tokens=200,
)
import json
result_text = response.choices[0].message.content.strip()
# Parse JSON response
try:
# Handle markdown code blocks
if "```" in result_text:
result_text = result_text.split("```")[1]
if result_text.startswith("json"):
result_text = result_text[4:]
result = json.loads(result_text.strip())
if self.debug:
logger.info(f"[DEBUG] Action verification: success={result.get('success')}, confidence={result.get('confidence')}")
return {
"success": result.get("success", True),
"confidence": result.get("confidence", 0.5),
"reason": result.get("reason", "")
}
except json.JSONDecodeError:
# Fallback: check if response contains "success" or "failed"
success = "success" in result_text.lower() and "fail" not in result_text.lower()
return {"success": success, "confidence": 0.5, "reason": result_text[:100]}
except Exception as e:
if self.debug:
logger.warning(f"[DEBUG] Action verification error: {e}")
return {"success": True, "confidence": 0.3, "reason": f"Verification error: {e}"}
async def _wait_for_stable_frame(self, max_wait: float = 2.0, stability_threshold: float = 0.3) -> Optional[str]:
"""Wait for page to stabilize and capture stable screenshot.
Captures multiple screenshots and waits until the page stops changing
(useful after navigation or dynamic content loading).
Args:
max_wait: Maximum seconds to wait for stability
stability_threshold: Time with no changes to consider stable
Returns:
Base64 screenshot of stable page, or latest if timeout
"""
import time
import hashlib
start_time = time.time()
last_hash = None
last_stable_time = start_time
latest_screenshot = None
while (time.time() - start_time) < max_wait:
try:
screenshot = await self._capture_vision_screenshot("stable_check")
if screenshot:
latest_screenshot = screenshot
# Hash screenshot to detect changes
current_hash = hashlib.md5(screenshot[:1000].encode()).hexdigest()
if current_hash == last_hash:
# Page hasn't changed
if (time.time() - last_stable_time) >= stability_threshold:
if self.debug:
logger.info("[DEBUG] Page stable, captured stable frame")
return screenshot
else:
# Page changed, reset stable timer
last_hash = current_hash
last_stable_time = time.time()
await asyncio.sleep(0.1) # Small delay between checks
except Exception as e:
if self.debug:
logger.warning(f"[DEBUG] Stable frame check error: {e}")
break
if self.debug:
logger.info("[DEBUG] Page stability timeout, using latest frame")
return latest_screenshot
# Frame buffer for continuous screencast mode
_latest_screencast_frame: Optional[str] = None
_frame_buffer: List[str] = []
_max_frame_buffer: int = 10
async def _handle_screencast_frame_for_vision(self, frame_data: str, session_id: int) -> None:
"""Handle incoming screencast frame for both video encoding and vision buffer.
Maintains a rolling buffer of recent frames for stable frame detection.
Args:
frame_data: Base64-encoded image data
session_id: Frame session ID for acknowledgment
"""
# Store latest frame for quick access
self._latest_screencast_frame = frame_data
# Add to buffer (rolling)
if len(self._frame_buffer) >= self._max_frame_buffer:
self._frame_buffer.pop(0)
self._frame_buffer.append(frame_data)
# Also handle for video encoding
await self._handle_screencast_frame(frame_data, session_id)
def _get_latest_vision_frame(self) -> Optional[str]:
"""Get the most recent screencast frame for vision analysis.
Returns:
Latest base64 frame from screencast buffer, or None
"""
return self._latest_screencast_frame
async def run(self, goal: str, start_url: str = "https://www.google.com") -> Dict[str, Any]:
"""Run browser automation to achieve goal.
@@ -419,14 +1038,31 @@ class CDPBrowserAgent:
agent_model = "gpt-4o" # Upgrade to vision-capable model
agent = BrowserAgent(model=agent_model, max_steps=self.max_steps, verbose=self.verbose)
# Reset action history for this run
self._action_history = []
# Start video recording if enabled - use screenshot_dir for output
if self.record_video and self.screenshot_dir:
await self._start_screencast(self.screenshot_dir)
if self.debug:
logger.info(f"[DEBUG] Video recording enabled, output dir: {self.screenshot_dir}")
# Main automation loop
for step in range(self.max_steps):
# Capture video frame (polling mode for reliable capture)
if self._screencast_active:
await self._capture_screencast_frame()
# Get page state
state = await self._get_page_state()
if self.verbose:
if self.verbose or self.debug:
logger.info(f"Step {step}: {state.url}")
if self.debug:
logger.info(f"[DEBUG] Elements found: {len(state.elements)}")
for i, el in enumerate(state.elements[:5]):
logger.info(f"[DEBUG] [{i}] {el.get('tag')} - {el.get('text', '')[:30]} - {el.get('selector', '')}")
# Update session URL
if self.record_session and self._session_manager:
self._session_manager.update_session(self._current_session_id, current_url=state.url)
@@ -453,7 +1089,7 @@ class CDPBrowserAgent:
except Exception as e:
logger.debug(f"Screenshot capture failed: {e}")
# Build observation
# Build observation with action history and overlay info
observation = {
"task": goal,
"original_goal": goal,
@@ -461,19 +1097,53 @@ class CDPBrowserAgent:
"title": state.title,
"elements": state.elements,
"step_number": step,
"action_history": self._action_history[-10:], # Last 10 actions for context
"overlay_info": state.overlay_info, # Pass overlay detection info
"max_steps": self.max_steps,
"steps_remaining": self.max_steps - step,
"mode": "cdp", # Technology mode: cdp or extension
}
# Add last action error if available
if self._action_history and not self._action_history[-1].get("success", True):
observation["last_action_error"] = self._action_history[-1].get("error", "Previous action failed")
# Add screenshot for vision analysis if enabled
if self.enable_vision and screenshot_base64:
observation["screenshot"] = screenshot_base64
# Get action from agent
action = agent.process_observation(observation)
action = normalize_action(action)
# DYNAMIC AGENT: Let LLM handle consent/overlays using visual context
# Only intervene if we have EXPLICIT consent buttons with valid selectors in elements
consent_elements = [el for el in state.elements if el.get("isConsentButton") or el.get("type") == "consent_button"]
if self.verbose:
if consent_elements and consent_elements[0].get("selector"):
# We have real consent buttons with selectors - click the first one
target_btn = consent_elements[0]
if self.debug:
logger.info(f"[DEBUG] Found consent button in elements: {target_btn.get('text', '')[:30]}")
action = {"action": "click", "selector": target_btn["selector"], "thought": f"Clicking consent button: {target_btn.get('text', '')[:30]}"}
# Check if agent is stuck (repeated same-URL actions with failures)
elif self._is_stuck(state.url):
if self.debug:
logger.info("[DEBUG] Agent appears stuck, letting LLM decide with extra context")
# Add stuck context to observation for LLM
observation["stuck_detected"] = True
observation["last_action_error"] = "Agent appears stuck. Try a different approach."
# Let LLM decide
action = agent.process_observation(observation)
action = normalize_action(action)
else:
# DYNAMIC: Let LLM decide based on observation (including screenshot if vision enabled)
action = agent.process_observation(observation)
action = normalize_action(action)
if self.verbose or self.debug:
logger.info(f" Action: {action.get('action')} | Done: {action.get('done')}")
if self.debug:
logger.info(f"[DEBUG] Full action: {action}")
# Check for completion
if action.get("done"):
# Record final step
@@ -508,6 +1178,13 @@ class CDPBrowserAgent:
action_success = False
original_action = action.copy()
# Phase 1: Capture BEFORE screenshot for action verification
screenshot_before = None
if self.enable_vision:
screenshot_before = await self._capture_vision_screenshot(f"before_step_{step}")
if self.debug and screenshot_before:
logger.info(f"[DEBUG] <SCREENSHOT_BEFORE step={step}> captured ({len(screenshot_before)//1024}KB)")
for attempt in range(self.max_retries + 1):
result = await self._execute_action(action)
@@ -561,6 +1238,60 @@ class CDPBrowserAgent:
else:
logger.warning(f"Action failed after {self.max_retries + 1} attempts: {result.get('error')}")
# Phase 1: Capture AFTER screenshot and verify action success
screenshot_after = None
verification_result = None
if self.enable_vision and action_success:
# Wait for page to stabilize before capturing AFTER screenshot
screenshot_after = await self._wait_for_stable_frame(max_wait=1.5, stability_threshold=0.3)
if self.debug and screenshot_after:
logger.info(f"[DEBUG] <SCREENSHOT_AFTER step={step}> captured ({len(screenshot_after)//1024}KB)")
# Save AFTER screenshot to file if directory specified
if screenshot_after and self.screenshot_dir:
try:
import base64
after_path = os.path.join(
self.screenshot_dir,
f"step_{step:03d}_after.png"
)
with open(after_path, "wb") as f:
f.write(base64.b64decode(screenshot_after))
except Exception as e:
logger.debug(f"Failed to save AFTER screenshot: {e}")
# Phase 2: Verify action success using LLM vision
if screenshot_before and screenshot_after:
verification_result = await self._verify_action_success(
action=action,
screenshot_before=screenshot_before,
screenshot_after=screenshot_after,
expected_change=action.get("thought", "")
)
if self.debug:
logger.info(f"[DEBUG] Action verification: {verification_result}")
# If verification fails with high confidence, log warning
if verification_result and not verification_result.get("success") and verification_result.get("confidence", 0) > 0.7:
logger.warning(f"Action may have failed: {verification_result.get('reason', 'Unknown')}")
# Track action in history for stuck detection and reporting
import time as time_module
self._action_history.append({
"action": action.get("action"),
"selector": action.get("selector"),
"text": action.get("text") or action.get("value"), # For type actions
"thought": action.get("thought", ""),
"url": state.url,
"success": action_success,
"step": step,
"timestamp": time_module.time(),
"vision_used": self.enable_vision and screenshot_base64 is not None,
"verified": verification_result.get("success") if verification_result else None,
"verification_confidence": verification_result.get("confidence") if verification_result else None,
})
# Record step
if self.record_session and self._session_manager:
self._session_manager.add_step(
@@ -575,7 +1306,14 @@ class CDPBrowserAgent:
screenshot_path=screenshot_path,
)
await asyncio.sleep(1) # Brief pause between steps
# Brief pause between steps - capture video frames during wait
if self._screencast_active:
# Capture multiple frames for smooth video (5 frames in 1 second)
for _ in range(5):
await self._capture_screencast_frame()
await asyncio.sleep(0.2)
else:
await asyncio.sleep(1) # Normal pause if not recording
# Max steps reached
if self.record_session and self._session_manager:
@@ -604,6 +1342,33 @@ class CDPBrowserAgent:
raise
finally:
# Stop video recording if active
video_path = await self._stop_screencast()
if video_path:
logger.info(f"Video saved: {video_path}")
# Generate session report
if self._action_history and self.debug:
try:
from .report import generate_session_report
import time
generate_session_report(
action_history=self._action_history,
goal=goal,
success=len(self._action_history) > 0 and self._action_history[-1].get("success", False),
final_url=self._last_url,
duration_seconds=time.time() - (self._action_history[0].get("timestamp", time.time()) if self._action_history else time.time()),
tech_flags={
"cdp": True,
"vision": self.enable_vision,
"extension": False,
},
)
except Exception as e:
if self.debug:
logger.warning(f"Report generation failed: {e}")
if self.ws:
await self.ws.close()
@@ -619,6 +1384,9 @@ async def run_cdp_only(
enable_vision: bool = False,
record_session: bool = True,
screenshot_dir: Optional[str] = None,
debug: bool = False,
record_video: bool = False,
video_fps: int = 5,
) -> Dict[str, Any]:
"""Run browser agent using direct CDP (no extension required).
@@ -633,6 +1401,9 @@ async def run_cdp_only(
enable_vision: Enable vision-based analysis
record_session: Log to session database
screenshot_dir: Save screenshots to directory
debug: Enable debug mode with detailed logging
record_video: Enable real-time video recording (requires FFmpeg)
video_fps: Frames per second for video (default 5)
Returns:
Result with success status and summary
@@ -652,6 +1423,9 @@ async def run_cdp_only(
enable_vision=enable_vision,
record_session=record_session,
screenshot_dir=screenshot_dir,
debug=debug,
record_video=record_video,
video_fps=video_fps,
)
return await agent.run(goal, url)
File diff suppressed because it is too large. Load diff
+137
View File
@@ -0,0 +1,137 @@
"""Session report generation for browser automation.
Generates detailed Rich-formatted reports showing step-by-step execution.
"""
import logging
from typing import List, Dict, Any, Optional
from datetime import datetime
logger = logging.getLogger("praisonai.browser.report")
def generate_session_report(
action_history: List[Dict[str, Any]],
goal: str,
success: bool,
final_url: str = "",
duration_seconds: float = 0,
tech_flags: Optional[Dict[str, bool]] = None,
) -> None:
"""Generate and display a Rich-formatted session report.
Args:
action_history: List of step dicts from CDPBrowserAgent._action_history
goal: Original automation goal
success: Whether the task completed successfully
final_url: Final page URL
duration_seconds: Total session duration
tech_flags: Dict of technologies used {cdp: bool, vision: bool, extension: bool}
"""
try:
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.text import Text
from rich import box
except ImportError:
# Fallback: simple text output
logger.info("Rich not installed - using simple output")
_print_simple_report(action_history, goal, success)
return
console = Console()
tech_flags = tech_flags or {}
# Header panel
status_icon = "✅" if success else "❌"
status_text = "COMPLETED" if success else "FAILED"
tech_badges = []
if tech_flags.get("cdp", True):
tech_badges.append("[cyan]🔌 CDP[/cyan]")
if tech_flags.get("vision"):
tech_badges.append("[yellow]👁️ Vision[/yellow]")
if tech_flags.get("extension"):
tech_badges.append("[green]🧩 Extension[/green]")
header = f"""[bold]{status_icon} Session {status_text}[/bold]
[dim]Goal:[/dim] {goal[:80]}{'...' if len(goal) > 80 else ''}
[dim]Final URL:[/dim] {final_url[:60]}{'...' if len(final_url) > 60 else ''}
[dim]Duration:[/dim] {duration_seconds:.1f}s | [dim]Steps:[/dim] {len(action_history)} | [dim]Tech:[/dim] {' '.join(tech_badges) or 'CDP'}"""
console.print(Panel(header, title="📋 Session Report", border_style="blue"))
# Steps table
table = Table(
title="Step-by-Step Execution",
box=box.ROUNDED,
show_header=True,
header_style="bold cyan",
)
table.add_column("#", style="dim", width=4, justify="right")
table.add_column("Action", style="bold", width=10)
table.add_column("Target/Value", width=30, overflow="fold")
table.add_column("Result", width=10, justify="center")
table.add_column("Tech", width=12)
table.add_column("Thought", width=35, overflow="fold")
for i, step in enumerate(action_history):
step_num = str(i + 1)
action_type = step.get("action", "?")
# Target/value
target = step.get("selector", step.get("text", step.get("url", "")))
if len(str(target)) > 28:
target = str(target)[:28] + "..."
# Result
result_success = step.get("success", True)
result = "[green]✓[/green]" if result_success else "[red]✗[/red]"
# Tech used this step
step_tech = []
if step.get("vision_used"):
step_tech.append("👁️")
step_tech.append("🔌") # CDP always used
# Thought (truncated)
thought = step.get("thought", "")
if len(thought) > 33:
thought = thought[:33] + "..."
table.add_row(
step_num,
action_type,
str(target),
result,
" ".join(step_tech),
thought,
)
console.print(table)
console.print()
def _print_simple_report(
action_history: List[Dict[str, Any]],
goal: str,
success: bool,
) -> None:
"""Simple text-based report fallback."""
print("\n" + "=" * 60)
print(f"SESSION REPORT: {'SUCCESS' if success else 'FAILED'}")
print("=" * 60)
print(f"Goal: {goal}")
print(f"Steps: {len(action_history)}")
print("-" * 60)
for i, step in enumerate(action_history):
action = step.get("action", "?")
target = step.get("selector", step.get("text", ""))[:30]
result = "✓" if step.get("success", True) else "✗"
print(f" {i+1}. {action:10} | {target:30} | {result}")
print("=" * 60 + "\n")
+235
View File
@@ -476,3 +476,238 @@ class BrowserServer:
self._agents.clear()
self._connections.clear()
async def run_browser_agent(
goal: str,
url: str = "https://www.google.com",
model: str = "gpt-4o-mini",
max_steps: int = 20,
timeout: float = 120.0,
debug: bool = False,
port: int = 8765,
) -> Dict:
"""Run browser agent via extension bridge server.
This function connects to the bridge server WebSocket and sends
goals to be executed by the Chrome extension.
Args:
goal: Task to accomplish
url: Starting URL
model: LLM model to use
max_steps: Maximum steps
timeout: Timeout in seconds
debug: Enable debug logging
port: Bridge server port
Returns:
Dict with success, summary, steps, etc.
Example:
result = await run_browser_agent(
goal="Search for PraisonAI",
url="https://google.com"
)
"""
import json
import asyncio
import uuid
if debug:
logger.setLevel(logging.DEBUG)
logger.info(f"[DEBUG] run_browser_agent: goal='{goal}', url='{url}', model='{model}'")
session_id = str(uuid.uuid4())
result = {
"success": False,
"summary": "",
"steps": 0,
"error": "",
"session_id": session_id,
"engine": "extension",
}
try:
import websockets
except ImportError:
result["error"] = "websockets package required. Install with: pip install websockets"
logger.error(result["error"])
return result
ws_url = f"ws://localhost:{port}/ws"
if debug:
logger.info(f"[DEBUG] Connecting to bridge server at {ws_url}")
try:
async with websockets.connect(ws_url, close_timeout=10) as ws:
if debug:
logger.info(f"[DEBUG] Connected to bridge server")
# Wait for welcome message
welcome = await asyncio.wait_for(ws.recv(), timeout=5.0)
welcome_data = json.loads(welcome)
if debug:
logger.info(f"[DEBUG] Welcome: {welcome_data}")
# Start session
start_msg = {
"type": "start_session",
"goal": goal,
"url": url,
"model": model,
"max_steps": max_steps,
"session_id": session_id,
}
if debug:
logger.info(f"[DEBUG] Sending start_session: {start_msg}")
await ws.send(json.dumps(start_msg))
# Wait for session response
response = await asyncio.wait_for(ws.recv(), timeout=timeout)
response_data = json.loads(response)
if debug:
logger.info(f"[DEBUG] Response: {response_data}")
if response_data.get("type") == "error":
result["error"] = response_data.get("error", "Unknown error")
return result
# Extract result
result["success"] = response_data.get("status") == "completed"
result["summary"] = response_data.get("summary", "")
result["steps"] = response_data.get("steps", 0)
result["final_url"] = response_data.get("final_url", "")
if debug:
logger.info(f"[DEBUG] Result: success={result['success']}, steps={result['steps']}")
return result
except asyncio.TimeoutError:
result["error"] = f"Timeout after {timeout}s waiting for bridge server response"
logger.error(result["error"])
return result
except ConnectionRefusedError:
result["error"] = f"Cannot connect to bridge server at {ws_url}. Is it running?"
logger.error(result["error"])
return result
except Exception as e:
result["error"] = str(e)
logger.error(f"Extension mode error: {e}")
return result
async def test_extension_mode(
goal: str = "Go to google.com and confirm the page loaded",
url: str = "https://www.google.com",
debug: bool = True,
timeout: float = 60.0,
) -> Dict:
"""Test extension mode end-to-end.
This is a diagnostic function to verify extension mode works:
1. Check if bridge server is running
2. Send a simple goal
3. Report success/failure with detailed debug info
Args:
goal: Test goal to execute
url: Starting URL
debug: Enable debug output
timeout: Timeout in seconds
Returns:
Test result with diagnostics
"""
import aiohttp
result = {
"bridge_server_running": False,
"extension_connected": False,
"goal_executed": False,
"success": False,
"error": "",
"diagnostics": [],
}
# Step 1: Check bridge server
logger.info("[TEST] Checking bridge server...")
result["diagnostics"].append("Checking bridge server on localhost:8765")
try:
async with aiohttp.ClientSession() as session:
async with session.get("http://localhost:8765/health", timeout=aiohttp.ClientTimeout(total=3)) as resp:
if resp.status == 200:
health = await resp.json()
result["bridge_server_running"] = True
result["diagnostics"].append(f"✓ Bridge server running: {health}")
logger.info(f"[TEST] ✓ Bridge server: {health}")
else:
result["diagnostics"].append(f"✗ Bridge server returned {resp.status}")
result["error"] = "Bridge server not healthy"
return result
except Exception as e:
result["diagnostics"].append(f"✗ Cannot connect to bridge server: {e}")
result["error"] = f"Bridge server not reachable: {e}"
logger.error(f"[TEST] ✗ Bridge server: {e}")
return result
# Step 2: Try to execute goal via extension
logger.info(f"[TEST] Executing goal: '{goal}'")
result["diagnostics"].append(f"Sending goal: '{goal}'")
try:
exec_result = await run_browser_agent(
goal=goal,
url=url,
debug=debug,
timeout=timeout,
)
result["goal_executed"] = True
result["success"] = exec_result.get("success", False)
result["steps"] = exec_result.get("steps", 0)
result["summary"] = exec_result.get("summary", "")
result["error"] = exec_result.get("error", "")
if result["success"]:
result["diagnostics"].append(f"✓ Goal completed in {result['steps']} steps")
logger.info(f"[TEST] ✓ Goal completed: {result['summary']}")
else:
result["diagnostics"].append(f"✗ Goal failed: {result['error']}")
logger.warning(f"[TEST] ✗ Goal failed: {result['error']}")
except Exception as e:
result["diagnostics"].append(f"✗ Execution error: {e}")
result["error"] = str(e)
logger.error(f"[TEST] ✗ Execution error: {e}")
return result
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="PraisonAI Browser Server")
parser.add_argument("--host", default="0.0.0.0", help="Host to bind to")
parser.add_argument("--port", type=int, default=8765, help="Port to listen on")
parser.add_argument("--model", default="gpt-4o-mini", help="LLM model")
parser.add_argument("--max-steps", type=int, default=20, help="Max steps per session")
parser.add_argument("--verbose", "-v", action="store_true", help="Verbose logging")
args = parser.parse_args()
server = BrowserServer(
host=args.host,
port=args.port,
model=args.model,
max_steps=args.max_steps,
verbose=args.verbose,
)
server.start()
+333
View File
@@ -0,0 +1,333 @@
"""Video creation utility for browser automation recordings.
Creates videos from screenshot sequences captured during browser automation.
"""
import os
import logging
from pathlib import Path
from typing import Optional, List
logger = logging.getLogger("praisonai.browser.video")
def create_video_from_screenshots(
screenshot_dir: str,
output_path: Optional[str] = None,
fps: int = 2,
duration_per_frame: float = 0.5,
) -> Optional[str]:
"""Create a video/GIF from a directory of screenshots.
Args:
screenshot_dir: Directory containing step_*.png files
output_path: Output video path (default: screenshot_dir/recording.gif)
fps: Frames per second for the output
duration_per_frame: Duration per frame in seconds (for GIF)
Returns:
Path to created video/GIF, or None if failed
"""
screenshot_dir = Path(screenshot_dir)
if not screenshot_dir.exists():
logger.warning(f"Screenshot directory does not exist: {screenshot_dir}")
return None
# Find all step screenshots
screenshots = sorted(screenshot_dir.glob("step_*.png"))
if not screenshots:
logger.warning(f"No screenshots found in {screenshot_dir}")
return None
logger.info(f"Found {len(screenshots)} screenshots to combine")
# Default output path
if output_path is None:
output_path = str(screenshot_dir / "recording.gif")
# Try using Pillow (most common)
try:
from PIL import Image
# Load all frames
frames = []
for screenshot in screenshots:
try:
img = Image.open(screenshot)
# Convert to RGB if necessary (GIF requirement)
if img.mode != 'RGB':
img = img.convert('RGB')
# Resize for smaller file size
max_width = 1280
if img.width > max_width:
ratio = max_width / img.width
new_size = (max_width, int(img.height * ratio))
img = img.resize(new_size, Image.Resampling.LANCZOS)
frames.append(img)
except Exception as e:
logger.warning(f"Failed to load {screenshot}: {e}")
if not frames:
logger.error("No frames could be loaded")
return None
# Save as GIF
duration_ms = int(duration_per_frame * 1000)
frames[0].save(
output_path,
save_all=True,
append_images=frames[1:],
duration=duration_ms,
loop=0, # Infinite loop
optimize=True,
)
logger.info(f"Created GIF: {output_path} ({len(frames)} frames)")
return output_path
except ImportError:
logger.warning("Pillow not installed, trying imageio")
# Fallback to imageio (if available)
try:
import imageio.v3 as iio
frames = []
for screenshot in screenshots:
try:
frame = iio.imread(screenshot)
frames.append(frame)
except Exception as e:
logger.warning(f"Failed to load {screenshot}: {e}")
if frames:
# Create GIF
iio.imwrite(output_path, frames, duration=duration_per_frame, loop=0)
logger.info(f"Created GIF with imageio: {output_path}")
return output_path
except ImportError:
logger.warning("imageio not installed")
# Last resort: just return the screenshot directory path
logger.warning("No image library available for video creation. Screenshots saved to: {screenshot_dir}")
return None
def create_video_ffmpeg(
screenshot_dir: str,
output_path: str,
fps: int = 2,
) -> Optional[str]:
"""Create video using ffmpeg (if available).
Args:
screenshot_dir: Directory containing step_*.png files
output_path: Output video path (e.g. recording.mp4)
fps: Frames per second
Returns:
Path to created video, or None if failed
"""
import subprocess
import shutil
if not shutil.which("ffmpeg"):
logger.warning("ffmpeg not found in PATH")
return None
screenshot_dir = Path(screenshot_dir)
pattern = str(screenshot_dir / "step_%03d.png")
try:
cmd = [
"ffmpeg",
"-y", # Overwrite output
"-framerate", str(fps),
"-i", pattern,
"-vf", "scale=1280:-2", # Scale to 1280 width
"-c:v", "libx264",
"-pix_fmt", "yuv420p",
output_path,
]
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=60,
)
if result.returncode == 0 and Path(output_path).exists():
logger.info(f"Created video: {output_path}")
return output_path
else:
logger.warning(f"ffmpeg failed: {result.stderr}")
return None
except Exception as e:
logger.warning(f"ffmpeg error: {e}")
return None
class FFmpegVideoEncoder:
"""Real-time video encoder using FFmpeg subprocess.
Encodes frames streamed from CDP Page.startScreencast to WebM video.
Uses VP9 codec for good quality and browser compatibility.
Example:
encoder = FFmpegVideoEncoder("/path/to/output.webm")
encoder.start()
for frame_data in cdp_screencast_frames:
encoder.write_frame(frame_data)
encoder.finish()
"""
def __init__(
self,
output_path: str,
fps: int = 10,
width: int = 1280,
height: int = 720,
quality: str = "good", # good, best, realtime
):
"""Initialize encoder.
Args:
output_path: Output video file path (should end in .webm or .mp4)
fps: Target frames per second
width: Video width
height: Video height
quality: Encoding quality preset
"""
import shutil
self.output_path = output_path
self.fps = fps
self.width = width
self.height = height
self.quality = quality
self.process = None
self.frame_count = 0
self._ffmpeg_available = shutil.which("ffmpeg") is not None
self._started = False
@property
def available(self) -> bool:
"""Check if FFmpeg is available."""
return self._ffmpeg_available
def start(self) -> bool:
"""Start the FFmpeg encoding process.
Returns:
True if started successfully, False otherwise
"""
if not self._ffmpeg_available:
logger.warning("FFmpeg not available, video recording disabled")
return False
import subprocess
# Determine output format from extension
is_webm = self.output_path.endswith('.webm')
try:
if is_webm:
# VP9 for WebM (good quality, browser compatible)
cmd = [
"ffmpeg",
"-y", # Overwrite output
"-f", "image2pipe", # Read images from pipe
"-framerate", str(self.fps),
"-i", "-", # Read from stdin
"-vf", f"scale={self.width}:-2", # Scale maintaining aspect
"-c:v", "libvpx-vp9", # VP9 codec
"-b:v", "1M", # Bitrate
"-deadline", self.quality,
"-cpu-used", "4", # Faster encoding
self.output_path,
]
else:
# H.264 for MP4
cmd = [
"ffmpeg",
"-y",
"-f", "image2pipe",
"-framerate", str(self.fps),
"-i", "-",
"-vf", f"scale={self.width}:-2",
"-c:v", "libx264",
"-preset", "fast",
"-pix_fmt", "yuv420p",
self.output_path,
]
self.process = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
)
self._started = True
logger.info(f"Started video recording to {self.output_path}")
return True
except Exception as e:
logger.error(f"Failed to start FFmpeg: {e}")
return False
def write_frame(self, frame_data: bytes) -> bool:
"""Write a frame to the video.
Args:
frame_data: JPEG or PNG image data
Returns:
True if written successfully
"""
if not self._started or not self.process or not self.process.stdin:
return False
try:
self.process.stdin.write(frame_data)
self.frame_count += 1
return True
except Exception as e:
logger.warning(f"Failed to write frame: {e}")
return False
def finish(self) -> Optional[str]:
"""Finish encoding and close the video file.
Returns:
Path to output video if successful, None otherwise
"""
if not self._started or not self.process:
return None
try:
self.process.stdin.close()
self.process.wait(timeout=30)
if self.process.returncode == 0:
logger.info(f"Video saved: {self.output_path} ({self.frame_count} frames)")
return self.output_path
else:
stderr = self.process.stderr.read().decode() if self.process.stderr else ""
logger.error(f"FFmpeg failed: {stderr[:200]}")
return None
except Exception as e:
logger.error(f"Error finishing video: {e}")
return None
finally:
self._started = False
self.process = None
def check_ffmpeg_available() -> bool:
"""Check if FFmpeg is available in PATH."""
import shutil
return shutil.which("ffmpeg") is not None
@@ -7,6 +7,7 @@ Usage:
praisonai docs run # Run all Python blocks from docs
praisonai docs list # List discovered code blocks
praisonai docs run --dry-run # Extract only, no execution
praisonai docs cli run-all # Validate all CLI commands from docs
"""
from datetime import datetime
@@ -17,6 +18,10 @@ import typer
app = typer.Typer(help="Documentation management and code validation")
# CLI subcommand group for validating CLI commands in documentation
cli_app = typer.Typer(help="Validate CLI commands from documentation")
app.add_typer(cli_app, name="cli")
def _get_default_docs_path() -> Path:
"""Get default docs path."""
@@ -710,6 +715,69 @@ def docs_run_all(
'groups': group_summaries,
}, f, indent=2)
# Generate consolidated reports (report.md, report.csv, report.json at root level)
from praisonai.suite_runner import RunReport, RunResult, SuiteReporter
all_results = []
for group_name in all_groups:
group_report_path = output_dir / group_name / 'report.json'
if group_report_path.exists():
try:
with open(group_report_path, 'r') as f:
group_data = json.load(f)
# Parse results from group report
for r in group_data.get('results', []):
result = RunResult(
item_id=r.get('item_id', ''),
suite=r.get('suite', 'docs'),
group=r.get('group', group_name),
source_path=Path(r.get('source_path', '')),
status=r.get('status', 'not_run'),
exit_code=r.get('exit_code'),
duration_seconds=r.get('duration_seconds'),
start_time=r.get('start_time'),
end_time=r.get('end_time'),
error_message=r.get('error_message'),
error_type=r.get('error_type'),
skip_reason=r.get('skip_reason'),
runnable_decision=r.get('runnable_decision'),
code_hash=r.get('code_hash'),
block_index=r.get('block_index'),
line_start=r.get('line_start'),
line_end=r.get('line_end'),
language=r.get('language'),
)
all_results.append(result)
except (json.JSONDecodeError, KeyError) as e:
if not quiet:
if ci:
print(f"Warning: Could not parse {group_report_path}: {e}")
else:
typer.echo(f"⚠️ Could not parse {group_report_path}: {e}")
# Create consolidated report
if all_results:
consolidated_report = RunReport(
results=all_results,
suite='docs',
source_path=docs,
groups_run=all_groups,
)
# Generate consolidated reports at root level
reporter = SuiteReporter(output_dir)
reporter.generate_all(consolidated_report)
if ci:
print(f"Consolidated report.json: {output_dir / 'report.json'}")
print(f"Consolidated report.md: {output_dir / 'report.md'}")
print(f"Consolidated report.csv: {output_dir / 'report.csv'}")
else:
typer.echo(f"📄 Consolidated report.json: {output_dir / 'report.json'}")
typer.echo(f"📄 Consolidated report.md: {output_dir / 'report.md'}")
typer.echo(f"📄 Consolidated report.csv: {output_dir / 'report.csv'}")
if ci:
print(f"Final summary: {summary_path}")
print(f"Group reports: {output_dir}")
@@ -899,3 +967,478 @@ def docs_api_md(
stdout=stdout
)
raise typer.Exit(exit_code)
# =============================================================================
# CLI Subcommand Group - Validate CLI commands from documentation
# =============================================================================
def _get_default_cli_report_dir() -> Path:
"""Get default report directory for CLI validation."""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
return Path.home() / "Downloads" / "reports" / "docs-cli" / timestamp
@cli_app.command("run-all")
def cli_run_all(
docs_path: Optional[Path] = typer.Option(
None,
"--docs-path", "-p",
help="Path to documentation directory (default: ~/PraisonAIDocs/docs/cli)",
),
timeout: int = typer.Option(
10,
"--timeout", "-t",
help="Per-command timeout in seconds",
),
report_dir: Optional[Path] = typer.Option(
None,
"--report-dir", "-r",
help="Directory for reports (default: ~/Downloads/reports/docs-cli/<timestamp>)",
),
parallel: bool = typer.Option(
True,
"--parallel/--sequential",
help="Run commands in parallel (default: parallel)",
),
max_workers: int = typer.Option(
4,
"--workers", "-w",
help="Max parallel workers (default: 4)",
),
max_items: Optional[int] = typer.Option(
None,
"--max-items",
help="Maximum commands to run",
),
group: Optional[List[str]] = typer.Option(
None,
"--group", "-g",
help="Run only specific groups (can be repeated)",
),
quiet: bool = typer.Option(
False,
"--quiet", "-q",
help="Minimal output",
),
ci: bool = typer.Option(
False,
"--ci",
help="CI-friendly output (no colors, proper exit codes)",
),
):
"""
Validate all CLI commands from documentation.
Discovers praisonai CLI commands in bash code blocks and validates
them by running with --help to ensure they exist and work.
Examples:
praisonai docs cli run-all
praisonai docs cli run-all --max-items 10
praisonai docs cli run-all --group cli --workers 8
praisonai docs cli run-all --ci --timeout 5
"""
import subprocess
from concurrent.futures import ThreadPoolExecutor, as_completed
from praisonai.suite_runner import CLIDocsSource, RunResult, RunReport, SuiteReporter
# Resolve paths
if docs_path:
docs = Path(docs_path).resolve()
else:
# Default to CLI docs directory
candidates = [
Path.home() / "PraisonAIDocs" / "docs" / "cli",
Path("/Users/praison/PraisonAIDocs/docs/cli"),
Path.cwd() / "docs" / "cli",
]
docs = None
for candidate in candidates:
if candidate.exists() and candidate.is_dir():
docs = candidate
break
if not docs:
docs = Path.cwd()
output_dir = report_dir or _get_default_cli_report_dir()
if not docs.exists():
if ci:
print(f"ERROR: Docs path not found: {docs}")
else:
typer.echo(f"❌ Docs path not found: {docs}")
raise typer.Exit(2)
# Discover CLI commands
source = CLIDocsSource(
root=docs,
groups=group,
help_only=True,
)
items = source.discover()
if max_items:
items = items[:max_items]
if not items:
if ci:
print("No CLI commands found in documentation")
else:
typer.echo("⚠️ No CLI commands found in documentation")
raise typer.Exit(0)
# Get groups for reporting
all_groups = sorted(set(item.group for item in items))
# Header
if ci:
print("=" * 60)
print("PraisonAI CLI Docs Validator")
print("=" * 60)
print(f"Docs path: {docs}")
print(f"Commands: {len(items)}")
print(f"Groups: {len(all_groups)}")
print(f"Mode: {'parallel' if parallel else 'sequential'}")
print(f"Workers: {max_workers}")
print(f"Timeout: {timeout}s")
print("=" * 60)
else:
typer.echo(f"📚 Docs path: {docs}")
typer.echo(f"🔍 Found {len(items)} CLI commands in {len(all_groups)} groups")
typer.echo(f"⚙️ Mode: {'parallel' if parallel else 'sequential'}, Workers: {max_workers}")
typer.echo("=" * 60)
# Run CLI commands
results = []
def run_cli_command(item) -> RunResult:
"""Run a single CLI command with --help."""
import time
start_time = time.time()
# Get the test command (stored in title field, or add --help)
test_cmd = item.title or (item.code + ' --help')
if item.skip:
return RunResult(
item_id=item.item_id,
suite='cli-docs',
group=item.group,
source_path=item.source_path,
status='skipped',
skip_reason=item.skip_reason,
code_hash=item.code_hash,
)
try:
result = subprocess.run(
test_cmd,
shell=True,
capture_output=True,
text=True,
timeout=timeout,
)
duration = time.time() - start_time
if result.returncode == 0:
return RunResult(
item_id=item.item_id,
suite='cli-docs',
group=item.group,
source_path=item.source_path,
status='passed',
exit_code=result.returncode,
duration_seconds=duration,
stdout=result.stdout[:1000] if result.stdout else None,
code_hash=item.code_hash,
)
else:
return RunResult(
item_id=item.item_id,
suite='cli-docs',
group=item.group,
source_path=item.source_path,
status='failed',
exit_code=result.returncode,
duration_seconds=duration,
error_message=result.stderr[:500] if result.stderr else f"Exit code: {result.returncode}",
stdout=result.stdout[:500] if result.stdout else None,
code_hash=item.code_hash,
)
except subprocess.TimeoutExpired:
return RunResult(
item_id=item.item_id,
suite='cli-docs',
group=item.group,
source_path=item.source_path,
status='timeout',
duration_seconds=timeout,
error_message=f"Command timed out after {timeout}s",
code_hash=item.code_hash,
)
except Exception as e:
return RunResult(
item_id=item.item_id,
suite='cli-docs',
group=item.group,
source_path=item.source_path,
status='failed',
error_message=str(e),
code_hash=item.code_hash,
)
# Execute commands
if parallel and len(items) > 1:
with ThreadPoolExecutor(max_workers=min(max_workers, len(items))) as pool:
futures = {pool.submit(run_cli_command, item): item for item in items}
for i, future in enumerate(as_completed(futures), 1):
item = futures[future]
result = future.result()
results.append(result)
if not quiet:
icon = {'passed': '✅', 'failed': '❌', 'skipped': '⏭️', 'timeout': '⏱️'}.get(result.status, '❓')
if ci:
print(f"[{i}/{len(items)}] {result.status.upper()} {item.code[:60]}...")
else:
typer.echo(f"[{i}/{len(items)}] {icon} {item.code[:60]}...")
else:
for i, item in enumerate(items, 1):
if not quiet:
if ci:
print(f"[{i}/{len(items)}] Running: {item.code[:60]}...")
else:
typer.echo(f"[{i}/{len(items)}] 🔄 {item.code[:60]}...")
result = run_cli_command(item)
results.append(result)
if not quiet:
icon = {'passed': '✅', 'failed': '❌', 'skipped': '⏭️', 'timeout': '⏱️'}.get(result.status, '❓')
if ci:
print(f" {result.status.upper()}")
else:
typer.echo(f" {icon} {result.status}")
# Create report
report = RunReport(
results=results,
suite='cli-docs',
source_path=docs,
groups_run=all_groups,
)
# Generate reports
output_dir.mkdir(parents=True, exist_ok=True)
reporter = SuiteReporter(output_dir)
reporter.save_logs(results)
json_path = reporter.generate_json(report)
md_path = reporter.generate_markdown(report)
csv_path = reporter.generate_csv(report)
# Summary
totals = report.totals
if ci:
print("")
print("=" * 60)
print("SUMMARY")
print("=" * 60)
print(f" PASSED: {totals.get('passed', 0)}")
print(f" FAILED: {totals.get('failed', 0)}")
print(f" SKIPPED: {totals.get('skipped', 0)}")
print(f" TIMEOUT: {totals.get('timeout', 0)}")
print(" -----------------")
print(f" TOTAL: {totals.get('total', 0)}")
print("=" * 60)
print(f"Reports: {output_dir}")
else:
typer.echo("")
typer.echo("=" * 60)
typer.echo("SUMMARY")
typer.echo("=" * 60)
typer.echo(f" ✅ Passed: {totals.get('passed', 0)}")
typer.echo(f" ❌ Failed: {totals.get('failed', 0)}")
typer.echo(f" ⏭️ Skipped: {totals.get('skipped', 0)}")
typer.echo(f" ⏱️ Timeout: {totals.get('timeout', 0)}")
typer.echo(" ─────────────────")
typer.echo(f" Total: {totals.get('total', 0)}")
typer.echo("=" * 60)
typer.echo(f"\n📄 JSON report: {json_path}")
typer.echo(f"📄 Markdown report: {md_path}")
typer.echo(f"📄 CSV report: {csv_path}")
typer.echo(f"📁 Reports saved to: {output_dir}")
# Exit code
if totals.get('failed', 0) > 0 or totals.get('timeout', 0) > 0:
raise typer.Exit(1)
raise typer.Exit(0)
@cli_app.command("list")
def cli_list(
docs_path: Optional[Path] = typer.Option(
None,
"--docs-path", "-p",
help="Path to documentation directory",
),
group: Optional[List[str]] = typer.Option(
None,
"--group", "-g",
help="Filter by group",
),
runnable_only: bool = typer.Option(
False,
"--runnable",
help="Show only runnable commands",
),
show_groups: bool = typer.Option(
False,
"--groups",
help="Show available groups only",
),
):
"""
List discovered CLI commands from documentation.
Examples:
praisonai docs cli list
praisonai docs cli list --groups
praisonai docs cli list --runnable
"""
from praisonai.suite_runner import CLIDocsSource
# Resolve path
if docs_path:
docs = Path(docs_path).resolve()
else:
candidates = [
Path.home() / "PraisonAIDocs" / "docs" / "cli",
Path("/Users/praison/PraisonAIDocs/docs/cli"),
]
docs = None
for candidate in candidates:
if candidate.exists():
docs = candidate
break
if not docs:
docs = Path.cwd()
source = CLIDocsSource(root=docs, groups=group)
if show_groups:
groups = source.get_groups()
typer.echo(f"Available groups ({len(groups)}):")
for g in groups:
typer.echo(f" - {g}")
return
items = source.discover()
if runnable_only:
items = [i for i in items if i.runnable]
typer.echo(f"Found {len(items)} CLI commands:")
typer.echo("")
for item in items:
status = "✅" if item.runnable else "⏭️"
typer.echo(f"{status} [{item.group}] {item.code[:70]}...")
if not item.runnable:
typer.echo(f" Skip: {item.runnable_decision}")
@cli_app.command("stats")
def cli_stats(
docs_path: Optional[Path] = typer.Option(
None,
"--docs-path", "-p",
help="Path to documentation directory",
),
):
"""
Show statistics for CLI commands in documentation.
Examples:
praisonai docs cli stats
"""
from praisonai.suite_runner import CLIDocsSource
# Resolve path
if docs_path:
docs = Path(docs_path).resolve()
else:
candidates = [
Path.home() / "PraisonAIDocs" / "docs" / "cli",
Path("/Users/praison/PraisonAIDocs/docs/cli"),
]
docs = None
for candidate in candidates:
if candidate.exists():
docs = candidate
break
if not docs:
docs = Path.cwd()
source = CLIDocsSource(root=docs)
stats = source.get_stats()
typer.echo("CLI Commands Statistics")
typer.echo("=" * 40)
typer.echo(f"Total commands: {stats['total']}")
typer.echo(f"Runnable: {stats['runnable']}")
typer.echo(f"Skipped: {stats['skipped']}")
typer.echo("")
typer.echo("By Group:")
typer.echo("-" * 40)
for group_name, group_stats in sorted(stats['by_group'].items()):
typer.echo(f" {group_name:<20} {group_stats['runnable']:>4}/{group_stats['total']:<4} runnable")
@cli_app.command("report")
def cli_report(
report_dir: Optional[Path] = typer.Argument(
None,
help="Report directory path (default: auto-detect latest)",
),
base_dir: Optional[Path] = typer.Option(
None,
"--base-dir", "-b",
help="Base directory for auto-detection",
),
limit: int = typer.Option(
20,
"--limit", "-n",
help="Max items to show (0 = unlimited)",
),
output_format: str = typer.Option(
"table",
"--format", "-f",
help="Output format: table or json",
),
):
"""
View CLI validation report.
Examples:
praisonai docs cli report
praisonai docs cli report ./my-report
praisonai docs cli report --format json
"""
from praisonai.suite_runner.report_viewer import view_report
output, exit_code = view_report(
report_dir=report_dir,
suite="docs-cli",
base_dir=base_dir or (Path.home() / "Downloads" / "reports" / "docs-cli"),
limit=limit,
output_format=output_format,
)
typer.echo(output)
raise typer.Exit(exit_code)
+4 -3
View File
@@ -2880,7 +2880,8 @@ class PraisonAI:
action_args: Additional arguments for the action
"""
# Code validation commands - delegate to typer app
code_validation_actions = {'run', 'run-all', 'stats', 'report', 'generate', 'serve'}
# Also handle 'cli' subcommand group for CLI validation
code_validation_actions = {'run', 'run-all', 'stats', 'report', 'generate', 'serve', 'cli'}
if action in code_validation_actions:
from praisonai.cli.commands.docs import app as docs_app
import typer
@@ -2888,8 +2889,8 @@ class PraisonAI:
typer_args = [action] + action_args
try:
typer.main.get_command(docs_app)(typer_args)
except SystemExit:
pass # typer raises SystemExit on completion
except SystemExit as e:
sys.exit(e.code if e.code is not None else 0)
return
try:
+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==3.9.4 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==3.9.7 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
@@ -26,6 +26,7 @@ __all__ = [
"ExamplesSource",
"DocsSource",
"BatchSource",
"CLIDocsSource",
# Executor
"SuiteExecutor",
]
@@ -61,6 +62,10 @@ def __getattr__(name: str):
from .batch_source import BatchSource
return BatchSource
if name == "CLIDocsSource":
from .cli_docs_source import CLIDocsSource
return CLIDocsSource
if name == "SuiteExecutor":
from .executor import SuiteExecutor
return SuiteExecutor
@@ -0,0 +1,300 @@
"""
CLI Docs Source - Extracts CLI commands from documentation for validation.
Discovers and extracts praisonai CLI commands from bash code blocks
in Markdown/MDX documentation files.
"""
from __future__ import annotations
import re
import hashlib
from pathlib import Path
from typing import List, Optional, Dict, Set
from .models import RunItem
from .discovery import FileDiscovery
# Pattern to match bash/shell code blocks
BASH_FENCE_PATTERN = re.compile(
r'^```(?:bash|shell|sh|zsh|console)\s*\n(.*?)^```',
re.MULTILINE | re.DOTALL
)
# Pattern to extract praisonai commands (handles multi-line with \)
PRAISONAI_CMD_PATTERN = re.compile(
r'^(?:#\s*)?(praisonai\s+[^\n\\]+(?:\\\n[^\n]+)*)',
re.MULTILINE
)
# Pattern to detect help-only commands (just --help)
HELP_ONLY_PATTERN = re.compile(r'--help\s*$')
# Pattern to detect commands with placeholders that can't be run
PLACEHOLDER_PATTERNS = [
re.compile(r'<[^>]+>'), # <placeholder>
re.compile(r'\[OPTIONS\]'), # [OPTIONS]
re.compile(r'\[ARGS\]'), # [ARGS]
re.compile(r'\[PROMPT\]'), # [PROMPT]
re.compile(r'\[COMMAND\]'), # [COMMAND]
re.compile(r'your[-_]?'), # your_api_key, your-file
re.compile(r'path/to/'), # path/to/file
re.compile(r'example\.com'), # example.com
]
# Commands that are safe to run with --help
SAFE_HELP_COMMANDS = {
'chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research',
'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools',
'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule', 'skills', 'profile',
'eval', 'agents', 'run', 'thinking', 'compaction', 'output', 'deploy',
'templates', 'recipe', 'endpoints', 'batch', 'examples', 'test',
'config', 'traces', 'env', 'completion', 'version', 'debug', 'lsp',
'diag', 'doctor', 'acp', 'tui', 'queue', 'benchmark', 'registry',
'package', 'standardise', 'standardize', 'background', 'checkpoint',
}
class CLIDocsSource:
"""
Extracts CLI commands from documentation for validation testing.
Parses bash code blocks in .md/.mdx files and extracts praisonai commands.
Commands are validated by running them with --help to ensure they exist.
"""
def __init__(
self,
root: Path,
include_patterns: Optional[List[str]] = None,
exclude_patterns: Optional[List[str]] = None,
groups: Optional[List[str]] = None,
help_only: bool = True,
):
"""
Initialize CLI docs source.
Args:
root: Root documentation directory.
include_patterns: Glob patterns to include.
exclude_patterns: Glob patterns to exclude.
groups: Specific groups (subdirs) to include.
help_only: If True, only test commands with --help (safe mode).
"""
self.root = Path(root).resolve()
self.include_patterns = include_patterns
self.exclude_patterns = exclude_patterns
self.groups = groups
self.help_only = help_only
self._discovery = FileDiscovery(
root=self.root,
extensions=['.md', '.mdx'],
include_patterns=include_patterns,
exclude_patterns=exclude_patterns,
)
def discover(self) -> List[RunItem]:
"""
Discover all CLI commands from documentation.
Returns:
List of RunItem objects for each unique CLI command.
"""
items = []
seen_commands: Set[str] = set()
for doc_path in self._discovery.discover():
# Filter by group if specified
group = self._get_group(doc_path)
if self.groups and group not in self.groups:
continue
# Extract commands from this file
file_items = self._extract_commands(doc_path, group)
# Deduplicate commands
for item in file_items:
cmd_key = self._normalize_command(item.code)
if cmd_key not in seen_commands:
seen_commands.add(cmd_key)
items.append(item)
return items
def _get_group(self, path: Path) -> str:
"""Get group name from path (top-level directory under root)."""
try:
rel = path.relative_to(self.root)
parts = rel.parts
if len(parts) > 1:
return parts[0]
return "root"
except ValueError:
return "root"
def _normalize_command(self, cmd: str) -> str:
"""Normalize command for deduplication."""
# Remove extra whitespace and normalize
cmd = ' '.join(cmd.split())
# Remove --help suffix for dedup purposes
cmd = re.sub(r'\s+--help\s*$', '', cmd)
return cmd
def _extract_commands(self, doc_path: Path, group: str) -> List[RunItem]:
"""Extract CLI commands from a documentation file."""
items = []
try:
content = doc_path.read_text(encoding='utf-8')
except Exception:
return items
# Find all bash code blocks
for match in BASH_FENCE_PATTERN.finditer(content):
block_content = match.group(1)
block_start = content[:match.start()].count('\n') + 1
# Extract praisonai commands from block
for cmd_match in PRAISONAI_CMD_PATTERN.finditer(block_content):
raw_cmd = cmd_match.group(1)
# Clean up multi-line commands
cmd = self._clean_command(raw_cmd)
if not cmd or not cmd.startswith('praisonai'):
continue
# Determine if command is runnable
runnable, decision = self._is_runnable(cmd)
# Create test command (add --help for safe testing)
test_cmd = self._create_test_command(cmd)
# Generate unique ID
cmd_line = block_start + block_content[:cmd_match.start()].count('\n')
item_id = self._generate_id(doc_path, cmd, cmd_line)
item = RunItem(
item_id=item_id,
suite='cli-docs',
group=group,
source_path=doc_path,
block_index=cmd_line,
language='bash',
code=cmd,
script_path=None, # CLI commands don't need script files
runnable=runnable,
runnable_decision=decision,
skip=not runnable,
skip_reason=decision if not runnable else None,
line_start=cmd_line,
line_end=cmd_line,
title=test_cmd, # Store test command in title field
)
items.append(item)
return items
def _clean_command(self, cmd: str) -> str:
"""Clean up a command string."""
# Remove line continuations
cmd = re.sub(r'\\\n\s*', ' ', cmd)
# Remove comments at end
cmd = re.sub(r'\s*#.*$', '', cmd)
# Normalize whitespace
cmd = ' '.join(cmd.split())
return cmd.strip()
def _is_runnable(self, cmd: str) -> tuple:
"""
Determine if a command is runnable for testing.
Returns:
Tuple of (is_runnable, decision_reason)
"""
# Check for placeholders
for pattern in PLACEHOLDER_PATTERNS:
if pattern.search(cmd):
return False, f"Contains placeholder: {pattern.pattern}"
# Extract base command (praisonai <subcommand>)
parts = cmd.split()
if len(parts) < 2:
return False, "No subcommand"
subcommand = parts[1]
# Check if it's a known safe command
if subcommand in SAFE_HELP_COMMANDS:
return True, "Known safe command"
# Check for file arguments that don't exist
if subcommand.endswith('.yaml') or subcommand.endswith('.py'):
return False, "File argument (may not exist)"
# Check for quoted prompts (these are actual executions)
if '"' in cmd or "'" in cmd:
# These would actually run the AI - skip unless help_only is False
if self.help_only:
return False, "Contains prompt (would execute AI)"
return True, "Runnable"
def _create_test_command(self, cmd: str) -> str:
"""Create a safe test command (typically with --help)."""
# If already has --help, use as-is
if '--help' in cmd:
return cmd
# Extract base command without arguments
parts = cmd.split()
if len(parts) >= 2:
# praisonai <subcommand> --help
base_parts = [parts[0], parts[1]]
# Handle nested subcommands (e.g., praisonai batch list)
if len(parts) >= 3 and not parts[2].startswith('-'):
# Check if third part looks like a subcommand
if parts[2] in ('list', 'run', 'stats', 'report', 'show', 'add',
'delete', 'create', 'update', 'get', 'set'):
base_parts.append(parts[2])
return ' '.join(base_parts) + ' --help'
return cmd + ' --help'
def _generate_id(self, path: Path, cmd: str, line: int) -> str:
"""Generate unique ID for a command."""
rel_path = path.relative_to(self.root).as_posix()
# Hash the command for uniqueness
cmd_hash = hashlib.md5(cmd.encode()).hexdigest()[:8]
return f"{rel_path.replace('/', '__').replace('.mdx', '').replace('.md', '')}__{line}__{cmd_hash}"
def get_groups(self) -> List[str]:
"""Get all available groups."""
groups = set()
for doc_path in self._discovery.discover():
groups.add(self._get_group(doc_path))
return sorted(groups)
def get_stats(self) -> Dict[str, int]:
"""Get statistics about discovered commands."""
items = self.discover()
stats = {
'total': len(items),
'runnable': sum(1 for i in items if i.runnable),
'skipped': sum(1 for i in items if i.skip),
'by_group': {},
}
for item in items:
if item.group not in stats['by_group']:
stats['by_group'][item.group] = {'total': 0, 'runnable': 0}
stats['by_group'][item.group]['total'] += 1
if item.runnable:
stats['by_group'][item.group]['runnable'] += 1
return stats
+1 -1
View File
@@ -1 +1 @@
__version__ = "3.9.4"
__version__ = "3.9.7"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.12.3",
"praisonaiagents>=0.12.4",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
+99
View File
@@ -247,3 +247,102 @@ class TestBrowserAgentAsync:
assert action["action"] == "click"
mock_process.assert_called_once_with(observation)
class TestConsentDialogHandling:
"""Tests for cookie consent dialog detection and handling."""
def test_build_prompt_with_overlay_info(self):
"""Test that overlay info is included in prompt when detected."""
agent = BrowserAgent()
observation = {
"task": "Search for AI on Google",
"url": "https://google.com",
"title": "Google",
"step_number": 1,
"elements": [
{"selector": "button.accept", "tag": "button", "text": "Accept all", "type": "consent_button"},
{"selector": "#search", "tag": "input", "text": ""},
],
"overlay_info": {
"detected": True,
"type": "consent_dialog",
"selector": '[role="dialog"]',
},
}
prompt = agent._build_prompt(observation)
# Should contain overlay warning
assert "COOKIE CONSENT" in prompt or "OVERLAY" in prompt
assert "consent_dialog" in prompt or "consent" in prompt.lower()
def test_build_prompt_with_action_history(self):
"""Test that action history is included in prompt."""
agent = BrowserAgent()
observation = {
"task": "Test",
"url": "https://test.com",
"title": "Test",
"step_number": 3,
"elements": [],
"action_history": [
{"action": "navigate", "selector": "", "success": True},
{"action": "click", "selector": "#btn", "success": False},
],
}
prompt = agent._build_prompt(observation)
# Should contain action history section
assert "Recent Actions" in prompt
assert "navigate" in prompt
assert "click" in prompt
def test_build_prompt_with_consent_elements_first(self):
"""Test that consent elements are highlighted when present."""
agent = BrowserAgent()
observation = {
"task": "Test",
"url": "https://google.com",
"title": "Google",
"step_number": 1,
"elements": [
{"selector": "[jsname='abc']", "tag": "button", "text": "Accept all", "type": "consent_button", "isConsentButton": True},
{"selector": "#search", "tag": "input", "text": "", "type": ""},
],
"overlay_info": {"detected": True, "type": "consent_dialog"},
}
prompt = agent._build_prompt(observation)
# Should show consent dialog detected prominently
assert "🚨" in prompt or "COOKIE" in prompt or "CONSENT" in prompt
def test_system_prompt_contains_consent_instructions(self):
"""Test that system prompt includes consent handling instructions."""
assert "consent" in BROWSER_AGENT_SYSTEM_PROMPT.lower() or "cookie" in BROWSER_AGENT_SYSTEM_PROMPT.lower()
assert "Accept" in BROWSER_AGENT_SYSTEM_PROMPT or "accept" in BROWSER_AGENT_SYSTEM_PROMPT.lower()
assert "overlay" in BROWSER_AGENT_SYSTEM_PROMPT.lower() or "dialog" in BROWSER_AGENT_SYSTEM_PROMPT.lower()
def test_build_prompt_with_last_action_error(self):
"""Test that last action error is shown prominently."""
agent = BrowserAgent()
observation = {
"task": "Test",
"url": "https://test.com",
"title": "Test",
"step_number": 2,
"elements": [],
"last_action_error": "Element not found: #missing-button",
}
prompt = agent._build_prompt(observation)
# Should contain the error prominently
assert "Element not found" in prompt or "FAILED" in prompt
+4 -4
View File
@@ -4276,7 +4276,7 @@ requires-dist = [
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.1.0" },
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.1.0" },
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.1.0" },
{ name = "praisonaiagents", specifier = ">=0.12.3" },
{ name = "praisonaiagents", specifier = ">=0.12.4" },
{ 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" },
@@ -4324,7 +4324,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.12.3"
version = "0.12.4"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -4333,9 +4333,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/38/36/f25ff3c9f6de046f4ce01a0ac2c2fa83a62c35e9b5d42c61203d196044ee/praisonaiagents-0.12.3.tar.gz", hash = "sha256:ba42248069703027f1a1114527be0b8d19d81130d38cf32eb3aa8d1156c5a6fa", size = 970485, upload-time = "2026-01-16T06:30:41.082Z" }
sdist = { url = "https://files.pythonhosted.org/packages/d1/c3/72f564eea0a41e33c5fee0cc342faaf88893ca963ed32f9f68f09e089f35/praisonaiagents-0.12.4.tar.gz", hash = "sha256:12b33c177198e3e232f740172cdfebdafe0ae781479432fa8df47e74a5ca10ed", size = 976416, upload-time = "2026-01-16T12:03:43.052Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/44/39/c80a7e942efd53b38f070c17ae16efe458760f69ad1dbb9b04ee60c4691f/praisonaiagents-0.12.3-py3-none-any.whl", hash = "sha256:b3917ff3d6dcf0a2334efdad99299722e1b21875ff7d4c3e39c27d4f2fdd1b0a", size = 951423, upload-time = "2026-01-16T06:30:39.098Z" },
{ url = "https://files.pythonhosted.org/packages/ca/a7/6274356eb38e286c07f88ee10831af36b0675f5ce22e92dbac44b0dcd313/praisonaiagents-0.12.4-py3-none-any.whl", hash = "sha256:c9242b22c6399179ed251abcafccc91e8b51ffec02ac26492a32277d4e0c9ff9", size = 955083, upload-time = "2026-01-16T12:03:40.792Z" },
]
[[package]]