Compare commits

...
6 Commits
Author SHA1 Message Date
Mervin Praison 99ab421ce2 Release v2.3.54 2025-12-19 16:04:43 +00:00
Mervin Praison c1f334905e Release v2.3.53 2025-12-19 15:38:30 +00:00
Mervin Praison 6d3ba7cdf8 Release v2.3.52 2025-12-19 15:34:19 +00:00
Mervin Praison 96616498d2 Release v2.3.51 2025-12-19 15:25:30 +00:00
Mervin Praison 1cad345415 Release v2.3.50 2025-12-19 15:19:29 +00:00
Mervin Praison 802e7e52f9 fix(readme): correct CLI docs URLs to include /docs/ prefix
Changed all https://docs.praison.ai/cli/* URLs to https://docs.praison.ai/docs/cli/*
Fixed 39 URLs total
2025-12-19 13:01:53 +00:00
21 changed files with 4015 additions and 201 deletions

No files matched your search

+65 -39
View File
@@ -252,20 +252,20 @@ npm install praisonai
| 🪝 Hooks | [Example](#9-hooks) | [📖](https://docs.praison.ai/features/hooks) |
| 📈 Telemetry | [Example](examples/python/telemetry/production-telemetry-example.py) | [📖](https://docs.praison.ai/features/telemetry) |
| 📹 Camera Integration | [Example](examples/python/camera/) | [📖](https://docs.praison.ai/features/camera-integration) |
| 📄 Project Docs (.praison/docs/) | [Example](#docs-cli) | [📖](https://docs.praison.ai/cli/docs) |
| 🔌 MCP Config Management | [Example](#mcp-config-cli) | [📖](https://docs.praison.ai/cli/mcp) |
| 💬 AI Commit Messages | [Example](#ai-commit-cli) | [📖](https://docs.praison.ai/cli/commit) |
| @ @Mentions in Prompts | [Example](#mentions-in-prompts) | [📖](https://docs.praison.ai/cli/mentions) |
| 💾 Auto-Save Sessions | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
| 📜 History in Context | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
| 📄 Project Docs (.praison/docs/) | [Example](#docs-cli) | [📖](https://docs.praison.ai/docs/cli/docs) |
| 🔌 MCP Config Management | [Example](#mcp-config-cli) | [📖](https://docs.praison.ai/docs/cli/mcp) |
| 💬 AI Commit Messages | [Example](#ai-commit-cli) | [📖](https://docs.praison.ai/docs/cli/commit) |
| @ @Mentions in Prompts | [Example](#mentions-in-prompts) | [📖](https://docs.praison.ai/docs/cli/mentions) |
| 💾 Auto-Save Sessions | [Example](#session-management-python) | [📖](https://docs.praison.ai/docs/cli/session) |
| 📜 History in Context | [Example](#session-management-python) | [📖](https://docs.praison.ai/docs/cli/session) |
| **🖥️ CLI Features** | | |
| ↳ Slash Commands | [Example](examples/python/cli/slash_commands_example.py) | [📖](https://docs.praison.ai/cli/slash-commands) |
| ↳ Autonomy Modes | [Example](examples/python/cli/autonomy_modes_example.py) | [📖](https://docs.praison.ai/cli/autonomy-modes) |
| ↳ Cost Tracking | [Example](examples/python/cli/cost_tracking_example.py) | [📖](https://docs.praison.ai/cli/cost-tracking) |
| ↳ Repository Map | [Example](examples/python/cli/repo_map_example.py) | [📖](https://docs.praison.ai/cli/repo-map) |
| ↳ Interactive TUI | [Example](examples/python/cli/interactive_tui_example.py) | [📖](https://docs.praison.ai/cli/interactive-tui) |
| ↳ Git Integration | [Example](examples/python/cli/git_integration_example.py) | [📖](https://docs.praison.ai/cli/git-integration) |
| ↳ Sandbox Execution | [Example](examples/python/cli/sandbox_execution_example.py) | [📖](https://docs.praison.ai/cli/sandbox-execution) |
| ↳ Slash Commands | [Example](examples/python/cli/slash_commands_example.py) | [📖](https://docs.praison.ai/docs/cli/slash-commands) |
| ↳ Autonomy Modes | [Example](examples/python/cli/autonomy_modes_example.py) | [📖](https://docs.praison.ai/docs/cli/autonomy-modes) |
| ↳ Cost Tracking | [Example](examples/python/cli/cost_tracking_example.py) | [📖](https://docs.praison.ai/docs/cli/cost-tracking) |
| ↳ Repository Map | [Example](examples/python/cli/repo_map_example.py) | [📖](https://docs.praison.ai/docs/cli/repo-map) |
| ↳ Interactive TUI | [Example](examples/python/cli/interactive_tui_example.py) | [📖](https://docs.praison.ai/docs/cli/interactive-tui) |
| ↳ Git Integration | [Example](examples/python/cli/git_integration_example.py) | [📖](https://docs.praison.ai/docs/cli/git-integration) |
| ↳ Sandbox Execution | [Example](examples/python/cli/sandbox_execution_example.py) | [📖](https://docs.praison.ai/docs/cli/sandbox-execution) |
---
@@ -972,6 +972,32 @@ export OPENAI_API_KEY=xxxxxxxxxxxxxxxxxxxxxx
praisonai --auto create a movie script about Robots in Mars
```
### Interactive Mode CLI:
```bash
# Start interactive terminal mode (inspired by Gemini CLI, Codex CLI, Claude Code)
praisonai --interactive
praisonai -i
# Features:
# - Streaming responses (no boxes)
# - Built-in tools: read_file, write_file, list_files, execute_command, internet_search
# - Slash commands: /help, /exit, /tools, /clear
# Chat mode - single prompt with interactive style (for testing/scripting)
# Use --chat (or --chat-mode for backward compatibility)
praisonai "list files in current folder" --chat
praisonai "search the web for AI news" --chat
praisonai "read README.md" --chat
```
### Chat UI (Web Interface):
```bash
# Start web-based Chainlit chat interface (requires praisonai[chat])
pip install "praisonai[chat]"
praisonai chat
# Opens browser at http://localhost:8084
```
### Query Rewriting (works with any command):
```bash
# Rewrite query for better results (uses QueryRewriterAgent)
@@ -1775,32 +1801,32 @@ app.include_router(a2a.get_router())
| Feature | Docs |
|---------|:----:|
| 🔄 Query Rewrite - RAG optimization | [📖](https://docs.praison.ai/cli/query-rewrite) |
| 🔬 Deep Research - Automated research | [📖](https://docs.praison.ai/cli/deep-research) |
| 📋 Planning - Step-by-step execution | [📖](https://docs.praison.ai/cli/planning) |
| 💾 Memory - Persistent agent memory | [📖](https://docs.praison.ai/cli/memory) |
| 📜 Rules - Auto-discovered instructions | [📖](https://docs.praison.ai/cli/rules) |
| 🔄 Workflow - Multi-step workflows | [📖](https://docs.praison.ai/cli/workflow) |
| 🪝 Hooks - Event-driven actions | [📖](https://docs.praison.ai/cli/hooks) |
| 🧠 Claude Memory - Anthropic memory tool | [📖](https://docs.praison.ai/cli/claude-memory) |
| 🛡️ Guardrail - Output validation | [📖](https://docs.praison.ai/cli/guardrail) |
| 📊 Metrics - Token usage tracking | [📖](https://docs.praison.ai/cli/metrics) |
| 🖼️ Image - Vision processing | [📖](https://docs.praison.ai/cli/image) |
| 📡 Telemetry - Usage monitoring | [📖](https://docs.praison.ai/cli/telemetry) |
| 🔌 MCP - Model Context Protocol | [📖](https://docs.praison.ai/cli/mcp) |
| ⚡ Fast Context - Codebase search | [📖](https://docs.praison.ai/cli/fast-context) |
| 📚 Knowledge - RAG management | [📖](https://docs.praison.ai/cli/knowledge) |
| 💬 Session - Conversation management | [📖](https://docs.praison.ai/cli/session) |
| 🔧 Tools - Tool discovery | [📖](https://docs.praison.ai/cli/tools) |
| 🤝 Handoff - Agent delegation | [📖](https://docs.praison.ai/cli/handoff) |
| 🧠 Auto Memory - Memory extraction | [📖](https://docs.praison.ai/cli/auto-memory) |
| 📋 Todo - Task management | [📖](https://docs.praison.ai/cli/todo) |
| 🎯 Router - Smart model selection | [📖](https://docs.praison.ai/cli/router) |
| 📈 Flow Display - Visual workflow | [📖](https://docs.praison.ai/cli/flow-display) |
| ✨ Prompt Expansion - Detailed prompts | [📖](https://docs.praison.ai/cli/prompt-expansion) |
| 🌐 Web Search - Real-time search | [📖](https://docs.praison.ai/cli/web-search) |
| 📥 Web Fetch - URL content retrieval | [📖](https://docs.praison.ai/cli/web-fetch) |
| 💾 Prompt Caching - Cost reduction | [📖](https://docs.praison.ai/cli/prompt-caching) |
| 🔄 Query Rewrite - RAG optimization | [📖](https://docs.praison.ai/docs/cli/query-rewrite) |
| 🔬 Deep Research - Automated research | [📖](https://docs.praison.ai/docs/cli/deep-research) |
| 📋 Planning - Step-by-step execution | [📖](https://docs.praison.ai/docs/cli/planning) |
| 💾 Memory - Persistent agent memory | [📖](https://docs.praison.ai/docs/cli/memory) |
| 📜 Rules - Auto-discovered instructions | [📖](https://docs.praison.ai/docs/cli/rules) |
| 🔄 Workflow - Multi-step workflows | [📖](https://docs.praison.ai/docs/cli/workflow) |
| 🪝 Hooks - Event-driven actions | [📖](https://docs.praison.ai/docs/cli/hooks) |
| 🧠 Claude Memory - Anthropic memory tool | [📖](https://docs.praison.ai/docs/cli/claude-memory) |
| 🛡️ Guardrail - Output validation | [📖](https://docs.praison.ai/docs/cli/guardrail) |
| 📊 Metrics - Token usage tracking | [📖](https://docs.praison.ai/docs/cli/metrics) |
| 🖼️ Image - Vision processing | [📖](https://docs.praison.ai/docs/cli/image) |
| 📡 Telemetry - Usage monitoring | [📖](https://docs.praison.ai/docs/cli/telemetry) |
| 🔌 MCP - Model Context Protocol | [📖](https://docs.praison.ai/docs/cli/mcp) |
| ⚡ Fast Context - Codebase search | [📖](https://docs.praison.ai/docs/cli/fast-context) |
| 📚 Knowledge - RAG management | [📖](https://docs.praison.ai/docs/cli/knowledge) |
| 💬 Session - Conversation management | [📖](https://docs.praison.ai/docs/cli/session) |
| 🔧 Tools - Tool discovery | [📖](https://docs.praison.ai/docs/cli/tools) |
| 🤝 Handoff - Agent delegation | [📖](https://docs.praison.ai/docs/cli/handoff) |
| 🧠 Auto Memory - Memory extraction | [📖](https://docs.praison.ai/docs/cli/auto-memory) |
| 📋 Todo - Task management | [📖](https://docs.praison.ai/docs/cli/todo) |
| 🎯 Router - Smart model selection | [📖](https://docs.praison.ai/docs/cli/router) |
| 📈 Flow Display - Visual workflow | [📖](https://docs.praison.ai/docs/cli/flow-display) |
| ✨ Prompt Expansion - Detailed prompts | [📖](https://docs.praison.ai/docs/cli/prompt-expansion) |
| 🌐 Web Search - Real-time search | [📖](https://docs.praison.ai/docs/cli/web-search) |
| 📥 Web Fetch - URL content retrieval | [📖](https://docs.praison.ai/docs/cli/web-fetch) |
| 💾 Prompt Caching - Cost reduction | [📖](https://docs.praison.ai/docs/cli/prompt-caching) |
---
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.49" \
"praisonai>=2.3.54" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.49" \
"praisonai>=2.3.54" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.49" \
"praisonai>=2.3.54" \
"praisonai[ui]" \
"praisonai[crewai]"
+65 -39
View File
@@ -252,20 +252,20 @@ npm install praisonai
| 🪝 Hooks | [Example](#9-hooks) | [📖](https://docs.praison.ai/features/hooks) |
| 📈 Telemetry | [Example](examples/python/telemetry/production-telemetry-example.py) | [📖](https://docs.praison.ai/features/telemetry) |
| 📹 Camera Integration | [Example](examples/python/camera/) | [📖](https://docs.praison.ai/features/camera-integration) |
| 📄 Project Docs (.praison/docs/) | [Example](#docs-cli) | [📖](https://docs.praison.ai/cli/docs) |
| 🔌 MCP Config Management | [Example](#mcp-config-cli) | [📖](https://docs.praison.ai/cli/mcp) |
| 💬 AI Commit Messages | [Example](#ai-commit-cli) | [📖](https://docs.praison.ai/cli/commit) |
| @ @Mentions in Prompts | [Example](#mentions-in-prompts) | [📖](https://docs.praison.ai/cli/mentions) |
| 💾 Auto-Save Sessions | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
| 📜 History in Context | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
| 📄 Project Docs (.praison/docs/) | [Example](#docs-cli) | [📖](https://docs.praison.ai/docs/cli/docs) |
| 🔌 MCP Config Management | [Example](#mcp-config-cli) | [📖](https://docs.praison.ai/docs/cli/mcp) |
| 💬 AI Commit Messages | [Example](#ai-commit-cli) | [📖](https://docs.praison.ai/docs/cli/commit) |
| @ @Mentions in Prompts | [Example](#mentions-in-prompts) | [📖](https://docs.praison.ai/docs/cli/mentions) |
| 💾 Auto-Save Sessions | [Example](#session-management-python) | [📖](https://docs.praison.ai/docs/cli/session) |
| 📜 History in Context | [Example](#session-management-python) | [📖](https://docs.praison.ai/docs/cli/session) |
| **🖥️ CLI Features** | | |
| ↳ Slash Commands | [Example](examples/python/cli/slash_commands_example.py) | [📖](https://docs.praison.ai/cli/slash-commands) |
| ↳ Autonomy Modes | [Example](examples/python/cli/autonomy_modes_example.py) | [📖](https://docs.praison.ai/cli/autonomy-modes) |
| ↳ Cost Tracking | [Example](examples/python/cli/cost_tracking_example.py) | [📖](https://docs.praison.ai/cli/cost-tracking) |
| ↳ Repository Map | [Example](examples/python/cli/repo_map_example.py) | [📖](https://docs.praison.ai/cli/repo-map) |
| ↳ Interactive TUI | [Example](examples/python/cli/interactive_tui_example.py) | [📖](https://docs.praison.ai/cli/interactive-tui) |
| ↳ Git Integration | [Example](examples/python/cli/git_integration_example.py) | [📖](https://docs.praison.ai/cli/git-integration) |
| ↳ Sandbox Execution | [Example](examples/python/cli/sandbox_execution_example.py) | [📖](https://docs.praison.ai/cli/sandbox-execution) |
| ↳ Slash Commands | [Example](examples/python/cli/slash_commands_example.py) | [📖](https://docs.praison.ai/docs/cli/slash-commands) |
| ↳ Autonomy Modes | [Example](examples/python/cli/autonomy_modes_example.py) | [📖](https://docs.praison.ai/docs/cli/autonomy-modes) |
| ↳ Cost Tracking | [Example](examples/python/cli/cost_tracking_example.py) | [📖](https://docs.praison.ai/docs/cli/cost-tracking) |
| ↳ Repository Map | [Example](examples/python/cli/repo_map_example.py) | [📖](https://docs.praison.ai/docs/cli/repo-map) |
| ↳ Interactive TUI | [Example](examples/python/cli/interactive_tui_example.py) | [📖](https://docs.praison.ai/docs/cli/interactive-tui) |
| ↳ Git Integration | [Example](examples/python/cli/git_integration_example.py) | [📖](https://docs.praison.ai/docs/cli/git-integration) |
| ↳ Sandbox Execution | [Example](examples/python/cli/sandbox_execution_example.py) | [📖](https://docs.praison.ai/docs/cli/sandbox-execution) |
---
@@ -972,6 +972,32 @@ export OPENAI_API_KEY=xxxxxxxxxxxxxxxxxxxxxx
praisonai --auto create a movie script about Robots in Mars
```
### Interactive Mode CLI:
```bash
# Start interactive terminal mode (inspired by Gemini CLI, Codex CLI, Claude Code)
praisonai --interactive
praisonai -i
# Features:
# - Streaming responses (no boxes)
# - Built-in tools: read_file, write_file, list_files, execute_command, internet_search
# - Slash commands: /help, /exit, /tools, /clear
# Chat mode - single prompt with interactive style (for testing/scripting)
# Use --chat (or --chat-mode for backward compatibility)
praisonai "list files in current folder" --chat
praisonai "search the web for AI news" --chat
praisonai "read README.md" --chat
```
### Chat UI (Web Interface):
```bash
# Start web-based Chainlit chat interface (requires praisonai[chat])
pip install "praisonai[chat]"
praisonai chat
# Opens browser at http://localhost:8084
```
### Query Rewriting (works with any command):
```bash
# Rewrite query for better results (uses QueryRewriterAgent)
@@ -1775,32 +1801,32 @@ app.include_router(a2a.get_router())
| Feature | Docs |
|---------|:----:|
| 🔄 Query Rewrite - RAG optimization | [📖](https://docs.praison.ai/cli/query-rewrite) |
| 🔬 Deep Research - Automated research | [📖](https://docs.praison.ai/cli/deep-research) |
| 📋 Planning - Step-by-step execution | [📖](https://docs.praison.ai/cli/planning) |
| 💾 Memory - Persistent agent memory | [📖](https://docs.praison.ai/cli/memory) |
| 📜 Rules - Auto-discovered instructions | [📖](https://docs.praison.ai/cli/rules) |
| 🔄 Workflow - Multi-step workflows | [📖](https://docs.praison.ai/cli/workflow) |
| 🪝 Hooks - Event-driven actions | [📖](https://docs.praison.ai/cli/hooks) |
| 🧠 Claude Memory - Anthropic memory tool | [📖](https://docs.praison.ai/cli/claude-memory) |
| 🛡️ Guardrail - Output validation | [📖](https://docs.praison.ai/cli/guardrail) |
| 📊 Metrics - Token usage tracking | [📖](https://docs.praison.ai/cli/metrics) |
| 🖼️ Image - Vision processing | [📖](https://docs.praison.ai/cli/image) |
| 📡 Telemetry - Usage monitoring | [📖](https://docs.praison.ai/cli/telemetry) |
| 🔌 MCP - Model Context Protocol | [📖](https://docs.praison.ai/cli/mcp) |
| ⚡ Fast Context - Codebase search | [📖](https://docs.praison.ai/cli/fast-context) |
| 📚 Knowledge - RAG management | [📖](https://docs.praison.ai/cli/knowledge) |
| 💬 Session - Conversation management | [📖](https://docs.praison.ai/cli/session) |
| 🔧 Tools - Tool discovery | [📖](https://docs.praison.ai/cli/tools) |
| 🤝 Handoff - Agent delegation | [📖](https://docs.praison.ai/cli/handoff) |
| 🧠 Auto Memory - Memory extraction | [📖](https://docs.praison.ai/cli/auto-memory) |
| 📋 Todo - Task management | [📖](https://docs.praison.ai/cli/todo) |
| 🎯 Router - Smart model selection | [📖](https://docs.praison.ai/cli/router) |
| 📈 Flow Display - Visual workflow | [📖](https://docs.praison.ai/cli/flow-display) |
| ✨ Prompt Expansion - Detailed prompts | [📖](https://docs.praison.ai/cli/prompt-expansion) |
| 🌐 Web Search - Real-time search | [📖](https://docs.praison.ai/cli/web-search) |
| 📥 Web Fetch - URL content retrieval | [📖](https://docs.praison.ai/cli/web-fetch) |
| 💾 Prompt Caching - Cost reduction | [📖](https://docs.praison.ai/cli/prompt-caching) |
| 🔄 Query Rewrite - RAG optimization | [📖](https://docs.praison.ai/docs/cli/query-rewrite) |
| 🔬 Deep Research - Automated research | [📖](https://docs.praison.ai/docs/cli/deep-research) |
| 📋 Planning - Step-by-step execution | [📖](https://docs.praison.ai/docs/cli/planning) |
| 💾 Memory - Persistent agent memory | [📖](https://docs.praison.ai/docs/cli/memory) |
| 📜 Rules - Auto-discovered instructions | [📖](https://docs.praison.ai/docs/cli/rules) |
| 🔄 Workflow - Multi-step workflows | [📖](https://docs.praison.ai/docs/cli/workflow) |
| 🪝 Hooks - Event-driven actions | [📖](https://docs.praison.ai/docs/cli/hooks) |
| 🧠 Claude Memory - Anthropic memory tool | [📖](https://docs.praison.ai/docs/cli/claude-memory) |
| 🛡️ Guardrail - Output validation | [📖](https://docs.praison.ai/docs/cli/guardrail) |
| 📊 Metrics - Token usage tracking | [📖](https://docs.praison.ai/docs/cli/metrics) |
| 🖼️ Image - Vision processing | [📖](https://docs.praison.ai/docs/cli/image) |
| 📡 Telemetry - Usage monitoring | [📖](https://docs.praison.ai/docs/cli/telemetry) |
| 🔌 MCP - Model Context Protocol | [📖](https://docs.praison.ai/docs/cli/mcp) |
| ⚡ Fast Context - Codebase search | [📖](https://docs.praison.ai/docs/cli/fast-context) |
| 📚 Knowledge - RAG management | [📖](https://docs.praison.ai/docs/cli/knowledge) |
| 💬 Session - Conversation management | [📖](https://docs.praison.ai/docs/cli/session) |
| 🔧 Tools - Tool discovery | [📖](https://docs.praison.ai/docs/cli/tools) |
| 🤝 Handoff - Agent delegation | [📖](https://docs.praison.ai/docs/cli/handoff) |
| 🧠 Auto Memory - Memory extraction | [📖](https://docs.praison.ai/docs/cli/auto-memory) |
| 📋 Todo - Task management | [📖](https://docs.praison.ai/docs/cli/todo) |
| 🎯 Router - Smart model selection | [📖](https://docs.praison.ai/docs/cli/router) |
| 📈 Flow Display - Visual workflow | [📖](https://docs.praison.ai/docs/cli/flow-display) |
| ✨ Prompt Expansion - Detailed prompts | [📖](https://docs.praison.ai/docs/cli/prompt-expansion) |
| 🌐 Web Search - Real-time search | [📖](https://docs.praison.ai/docs/cli/web-search) |
| 📥 Web Fetch - URL content retrieval | [📖](https://docs.praison.ai/docs/cli/web-fetch) |
| 💾 Prompt Caching - Cost reduction | [📖](https://docs.praison.ai/docs/cli/prompt-caching) |
---
@@ -0,0 +1,33 @@
# PraisonAI CLI - Benchmark Results
**Generated:** 2025-12-19 16:04:40
## Import Times
| Module | Time (ms) | Target | Status |
|--------|-----------|--------|--------|
| praisonai | 4.11 | <10 | ✅ |
| message_queue | 6.87 | <50 | ✅ |
| at_mentions | 0.58 | <50 | ✅ |
| profiler | 0.90 | <10 | ✅ |
## Instantiation Times
| Class | Time (μs) | Target | Status |
|-------|-----------|--------|--------|
| MessageQueue | 0.26 | <100 | ✅ |
| StateManager | 0.30 | <100 | ✅ |
| FileSearchService | 8.42 | <100 | ✅ |
## Lazy Loading
| Operation | Time (ms) |
|-----------|----------|
| PraisonAI lazy load | 3886.47 |
## How to Reproduce
```bash
cd praisonai
python benchmarks/simple_benchmark.py
```
@@ -0,0 +1,231 @@
#!/usr/bin/env python3
"""
PraisonAI CLI - Performance Benchmark
Measures import times and module loading performance.
Usage:
python benchmarks/simple_benchmark.py
python benchmarks/simple_benchmark.py --no-save
"""
import time
import sys
import os
import argparse
from datetime import datetime
ITERATIONS = 100
def measure_import(module_name: str) -> float:
"""Measure import time in milliseconds."""
# Clear from cache if present
modules_to_remove = [k for k in sys.modules.keys() if k.startswith(module_name)]
for mod in modules_to_remove:
del sys.modules[mod]
start = time.perf_counter()
__import__(module_name)
return (time.perf_counter() - start) * 1000
def measure_instantiation(create_fn, iterations=ITERATIONS) -> float:
"""Measure average instantiation time in microseconds."""
times = []
for _ in range(iterations):
start = time.perf_counter()
create_fn()
times.append((time.perf_counter() - start) * 1_000_000)
return sum(times) / len(times)
def run_benchmark():
"""Run the benchmark."""
results = {}
print('=' * 60)
print('PraisonAI CLI - Performance Benchmark')
print('=' * 60)
print(f'\nIterations: {ITERATIONS}')
print('Metrics: Import time (ms), Instantiation time (μs)\n')
# 1. Package import time
print("1. Testing package imports...")
# Clear cache
modules_to_remove = [k for k in sys.modules.keys() if 'praisonai' in k]
for mod in modules_to_remove:
del sys.modules[mod]
start = time.perf_counter()
import praisonai
results['praisonai_import_ms'] = (time.perf_counter() - start) * 1000
print(f" import praisonai: {results['praisonai_import_ms']:.2f}ms")
# 2. CLI features import
print("\n2. Testing CLI features imports...")
start = time.perf_counter()
from praisonai.cli.features.message_queue import MessageQueue
results['message_queue_import_ms'] = (time.perf_counter() - start) * 1000
print(f" message_queue: {results['message_queue_import_ms']:.2f}ms")
start = time.perf_counter()
from praisonai.cli.features.at_mentions import CombinedCompleter
results['at_mentions_import_ms'] = (time.perf_counter() - start) * 1000
print(f" at_mentions: {results['at_mentions_import_ms']:.2f}ms")
# 3. Profiler import (should be fast, standalone)
print("\n3. Testing profiler import...")
start = time.perf_counter()
from praisonai.profiler import Profiler
results['profiler_import_ms'] = (time.perf_counter() - start) * 1000
print(f" profiler: {results['profiler_import_ms']:.2f}ms")
# 4. Instantiation benchmarks
print(f"\n4. Testing instantiation ({ITERATIONS} iterations)...")
from praisonai.cli.features.message_queue import MessageQueue, StateManager
from praisonai.cli.features.at_mentions import FileSearchService
results['MessageQueue_us'] = measure_instantiation(lambda: MessageQueue())
print(f" MessageQueue: {results['MessageQueue_us']:.2f}μs")
results['StateManager_us'] = measure_instantiation(lambda: StateManager())
print(f" StateManager: {results['StateManager_us']:.2f}μs")
results['FileSearchService_us'] = measure_instantiation(
lambda: FileSearchService(root_dir=".")
)
print(f" FileSearchService: {results['FileSearchService_us']:.2f}μs")
# 5. PraisonAI class (lazy load)
print("\n5. Testing PraisonAI class access (lazy load)...")
start = time.perf_counter()
PraisonAI = praisonai.PraisonAI
results['PraisonAI_lazy_ms'] = (time.perf_counter() - start) * 1000
print(f" PraisonAI lazy load: {results['PraisonAI_lazy_ms']:.2f}ms")
return results
def print_summary(results: dict):
"""Print benchmark summary."""
print('\n' + '=' * 60)
print('SUMMARY')
print('=' * 60)
print('\nImport Times (target: <50ms for features, <10ms for package):')
print(f" {'Module':<25} {'Time':>10} {'Status':>10}")
print(f" {'-'*25} {'-'*10} {'-'*10}")
import_checks = [
('praisonai', results['praisonai_import_ms'], 10),
('message_queue', results['message_queue_import_ms'], 50),
('at_mentions', results['at_mentions_import_ms'], 50),
('profiler', results['profiler_import_ms'], 10),
]
all_pass = True
for name, time_ms, threshold in import_checks:
status = '✅ PASS' if time_ms < threshold else '❌ SLOW'
if time_ms >= threshold:
all_pass = False
print(f" {name:<25} {time_ms:>8.2f}ms {status:>10}")
print('\nInstantiation Times (target: <100μs):')
print(f" {'Class':<25} {'Time':>10} {'Status':>10}")
print(f" {'-'*25} {'-'*10} {'-'*10}")
inst_checks = [
('MessageQueue', results['MessageQueue_us'], 100),
('StateManager', results['StateManager_us'], 100),
('FileSearchService', results['FileSearchService_us'], 100),
]
for name, time_us, threshold in inst_checks:
status = '✅ PASS' if time_us < threshold else '❌ SLOW'
if time_us >= threshold:
all_pass = False
print(f" {name:<25} {time_us:>8.2f}μs {status:>10}")
print('\n' + '=' * 60)
if all_pass:
print('✅ All benchmarks PASSED')
else:
print('❌ Some benchmarks FAILED')
print('=' * 60)
return all_pass
def save_results(results: dict):
"""Save benchmark results to markdown file."""
filepath = os.path.join(os.path.dirname(__file__), 'BENCHMARK_RESULTS.md')
with open(filepath, 'w') as f:
f.write('# PraisonAI CLI - Benchmark Results\n\n')
f.write(f'**Generated:** {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}\n\n')
f.write('## Import Times\n\n')
f.write('| Module | Time (ms) | Target | Status |\n')
f.write('|--------|-----------|--------|--------|\n')
import_data = [
('praisonai', results['praisonai_import_ms'], 10),
('message_queue', results['message_queue_import_ms'], 50),
('at_mentions', results['at_mentions_import_ms'], 50),
('profiler', results['profiler_import_ms'], 10),
]
for name, time_ms, threshold in import_data:
status = '✅' if time_ms < threshold else '❌'
f.write(f'| {name} | {time_ms:.2f} | <{threshold} | {status} |\n')
f.write('\n## Instantiation Times\n\n')
f.write('| Class | Time (μs) | Target | Status |\n')
f.write('|-------|-----------|--------|--------|\n')
inst_data = [
('MessageQueue', results['MessageQueue_us'], 100),
('StateManager', results['StateManager_us'], 100),
('FileSearchService', results['FileSearchService_us'], 100),
]
for name, time_us, threshold in inst_data:
status = '✅' if time_us < threshold else '❌'
f.write(f'| {name} | {time_us:.2f} | <{threshold} | {status} |\n')
f.write('\n## Lazy Loading\n\n')
f.write(f'| Operation | Time (ms) |\n')
f.write(f'|-----------|----------|\n')
f.write(f'| PraisonAI lazy load | {results["PraisonAI_lazy_ms"]:.2f} |\n')
f.write('\n## How to Reproduce\n\n')
f.write('```bash\n')
f.write('cd praisonai\n')
f.write('python benchmarks/simple_benchmark.py\n')
f.write('```\n')
print(f'\nResults saved to: {filepath}')
return filepath
def main():
parser = argparse.ArgumentParser(description='PraisonAI CLI Performance Benchmark')
parser.add_argument('--no-save', action='store_true', help='Do not save results to file')
args = parser.parse_args()
results = run_benchmark()
all_pass = print_summary(results)
if not args.no_save:
save_results(results)
return 0 if all_pass else 1
if __name__ == '__main__':
sys.exit(main())
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.49.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.49.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.54.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.54.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+19 -20
View File
@@ -6,30 +6,29 @@ logging.getLogger('crewai.cli.config').setLevel(logging.ERROR)
import os
os.environ["OTEL_SDK_DISABLED"] = "true"
os.environ["EC_TELEMETRY"] = "false"
from .cli import PraisonAI
# Version is lightweight, import directly
from .version import __version__
# Re-export all classes from praisonaiagents to enable:
# from praisonai import Agent, Task, PraisonAIAgents
try:
import praisonaiagents
# Import all symbols from praisonaiagents using * import
from praisonaiagents import *
except ImportError:
# If praisonaiagents is not available, these imports will fail gracefully
pass
# Define __all__ to include both PraisonAI core classes and praisonaiagents exports
# Define __all__ for lazy loading
__all__ = [
# Core PraisonAI classes
'PraisonAI',
'__version__',
]
# Dynamically extend __all__ with praisonaiagents exports
try:
import praisonaiagents
__all__.extend(praisonaiagents.__all__)
except (ImportError, AttributeError):
# If praisonaiagents is not available or doesn't have __all__, fail gracefully
pass
# Lazy loading for heavy imports
def __getattr__(name):
"""Lazy load heavy modules to improve import time."""
if name == 'PraisonAI':
from .cli import PraisonAI
return PraisonAI
# Try praisonaiagents exports
try:
import praisonaiagents
if hasattr(praisonaiagents, name):
return getattr(praisonaiagents, name)
except ImportError:
pass
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+9 -2
View File
@@ -8,12 +8,19 @@ Structure:
- features/: Feature handlers for CLI flags and commands
"""
from .main import PraisonAI
__all__ = ["PraisonAI"]
def __getattr__(name):
"""Lazy load PraisonAI to avoid slow imports."""
if name == "PraisonAI":
from .main import PraisonAI
return PraisonAI
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
def main():
"""CLI entry point function."""
from .main import PraisonAI
praison_ai = PraisonAI()
praison_ai.main()
@@ -25,6 +25,8 @@ Features:
- slash_commands: Interactive slash commands (/help, /cost, /plan, etc.)
- autonomy_mode: Autonomy levels (suggest, auto-edit, full-auto)
- cost_tracker: Real-time cost and token tracking
- message_queue: Message queuing while agent is processing
- at_mentions: @ mention autocomplete for files and directories
"""
# Type hints available for IDE support
@@ -56,6 +58,11 @@ __all__ = [
'InteractiveTUIHandler',
'GitIntegrationHandler',
'SandboxExecutorHandler',
'MessageQueueHandler',
# @ mention autocomplete
'AtMentionCompleter',
'CombinedCompleter',
'FileSearchService',
]
def __getattr__(name):
@@ -133,4 +140,17 @@ def __getattr__(name):
elif name == 'SandboxExecutorHandler':
from .sandbox_executor import SandboxExecutorHandler
return SandboxExecutorHandler
elif name == 'MessageQueueHandler':
from .message_queue import MessageQueueHandler
return MessageQueueHandler
# @ mention autocomplete
elif name == 'AtMentionCompleter':
from .at_mentions import AtMentionCompleter
return AtMentionCompleter
elif name == 'CombinedCompleter':
from .at_mentions import CombinedCompleter
return CombinedCompleter
elif name == 'FileSearchService':
from .at_mentions import FileSearchService
return FileSearchService
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -0,0 +1,416 @@
"""
@ Mention Autocomplete for PraisonAI CLI.
Provides file/directory autocomplete when user types @.
Uses only stdlib imports for zero performance impact.
Features:
- Detect @ trigger in input
- Fuzzy search files/directories
- Cache results for performance
- Integrate with prompt_toolkit
"""
import os
import time
import fnmatch
from dataclasses import dataclass
from typing import List, Optional, Dict, Tuple
# ============================================================================
# Data Classes
# ============================================================================
@dataclass
class AtMentionContext:
"""Context for an @ mention in user input."""
is_active: bool
query: str
start_pos: int
mention_type: str = "file" # file, directory, web, etc.
@dataclass
class FileSuggestion:
"""A file or directory suggestion."""
path: str
file_type: str # "file" or "directory"
score: int = 100
display: str = ""
def __post_init__(self):
if not self.display:
self.display = self.path
# ============================================================================
# @ Mention Detection
# ============================================================================
def detect_at_mention(text: str, cursor_pos: int) -> Optional[AtMentionContext]:
"""
Detect if cursor is within an @ mention context.
Scans backward from cursor to find @ symbol.
Returns None if not in @ context.
Args:
text: Full input text
cursor_pos: Current cursor position
Returns:
AtMentionContext if in @ mention, None otherwise
"""
if cursor_pos <= 0 or cursor_pos > len(text):
return None
# Scan backward from cursor to find @
at_pos = -1
for i in range(cursor_pos - 1, -1, -1):
char = text[i]
# Found @
if char == '@':
at_pos = i
break
# Space/newline breaks @ context
if char in ' \t\n\r':
return None
if at_pos == -1:
return None
# Extract query between @ and cursor
query = text[at_pos + 1:cursor_pos]
return AtMentionContext(
is_active=True,
query=query,
start_pos=at_pos
)
# ============================================================================
# File Search Service
# ============================================================================
class FileSearchService:
"""
Service for searching files in a directory.
Features:
- Walk directory tree
- Fuzzy match file paths
- Cache results with TTL
- Respect .gitignore (basic)
"""
def __init__(
self,
root_dir: str,
cache_ttl: int = 30,
max_depth: int = 5,
ignore_patterns: Optional[List[str]] = None
):
self._root = os.path.abspath(root_dir)
self._cache_ttl = cache_ttl
self._max_depth = max_depth
self._ignore_patterns = ignore_patterns or [
'.git', '__pycache__', 'node_modules', '.venv', 'venv',
'*.pyc', '*.pyo', '.DS_Store', '*.egg-info'
]
# Cache: query -> (results, timestamp)
self._cache: Dict[str, Tuple[List[FileSuggestion], float]] = {}
# File list cache
self._file_list: Optional[List[str]] = None
self._file_list_time: float = 0
def _should_ignore(self, name: str) -> bool:
"""Check if file/dir should be ignored."""
for pattern in self._ignore_patterns:
if fnmatch.fnmatch(name, pattern):
return True
return False
def _get_file_list(self) -> List[str]:
"""Get list of all files (cached)."""
now = time.time()
# Return cached if fresh
if self._file_list is not None and (now - self._file_list_time) < self._cache_ttl:
return self._file_list
files = []
try:
for root, dirs, filenames in os.walk(self._root):
# Calculate depth
rel_root = os.path.relpath(root, self._root)
depth = 0 if rel_root == '.' else rel_root.count(os.sep) + 1
if depth > self._max_depth:
dirs.clear() # Don't descend further
continue
# Filter ignored directories
dirs[:] = [d for d in dirs if not self._should_ignore(d)]
# Add directories
for d in dirs:
rel_path = os.path.relpath(os.path.join(root, d), self._root)
files.append(rel_path + os.sep)
# Add files
for f in filenames:
if not self._should_ignore(f):
rel_path = os.path.relpath(os.path.join(root, f), self._root)
files.append(rel_path)
except OSError:
pass
self._file_list = files
self._file_list_time = now
return files
def _fuzzy_match(self, text: str, query: str) -> int:
"""
Simple fuzzy match scoring.
Returns score 0-100, higher is better.
"""
if not query:
return 50 # Default score for empty query
text_lower = text.lower()
query_lower = query.lower()
# Exact match
if text_lower == query_lower:
return 100
# Starts with
if text_lower.startswith(query_lower):
return 90
# Contains
if query_lower in text_lower:
return 80
# Fuzzy: all query chars appear in order
text_idx = 0
query_idx = 0
matches = 0
while text_idx < len(text_lower) and query_idx < len(query_lower):
if text_lower[text_idx] == query_lower[query_idx]:
matches += 1
query_idx += 1
text_idx += 1
if query_idx == len(query_lower):
# All query chars found
return 50 + int(30 * matches / len(text))
return 0
def search(self, query: str, max_results: int = 20) -> List[FileSuggestion]:
"""
Search for files matching query.
Args:
query: Search query (fuzzy matched)
max_results: Maximum results to return
Returns:
List of FileSuggestion sorted by score
"""
# Check cache
cache_key = f"{query}:{max_results}"
now = time.time()
if cache_key in self._cache:
results, cached_time = self._cache[cache_key]
if (now - cached_time) < self._cache_ttl:
return results
# Get all files
all_files = self._get_file_list()
# Score and filter
scored: List[Tuple[int, str]] = []
for path in all_files:
score = self._fuzzy_match(path, query)
if score > 0:
scored.append((score, path))
# Sort by score descending
scored.sort(key=lambda x: (-x[0], x[1]))
# Build results
results = []
for score, path in scored[:max_results]:
file_type = "directory" if path.endswith(os.sep) else "file"
results.append(FileSuggestion(
path=path,
file_type=file_type,
score=score
))
# Cache results
self._cache[cache_key] = (results, now)
return results
def clear_cache(self) -> None:
"""Clear all caches."""
self._cache.clear()
self._file_list = None
# ============================================================================
# Completers for prompt_toolkit
# ============================================================================
# Lazy import to avoid performance impact
_Completer = None
_Completion = None
def _get_prompt_toolkit():
"""Lazy import prompt_toolkit."""
global _Completer, _Completion
if _Completer is None:
from prompt_toolkit.completion import Completer, Completion
_Completer = Completer
_Completion = Completion
return _Completer, _Completion
class AtMentionCompleter:
"""
Completer for @ mentions.
Shows file suggestions when user types @.
Implements prompt_toolkit Completer interface.
"""
def __init__(self, root_dir: str):
self._file_service = FileSearchService(root_dir)
def get_completions(self, document, complete_event):
"""Get completions for @ mentions."""
_, Completion = _get_prompt_toolkit()
text = document.text_before_cursor
cursor_pos = len(text)
# Detect @ context
context = detect_at_mention(text, cursor_pos)
if context is None or not context.is_active:
return
# Search files
results = self._file_service.search(context.query, max_results=15)
# Yield completions
for suggestion in results:
# Icon based on type
icon = "📁 " if suggestion.file_type == "directory" else "📄 "
display = f"{icon}{suggestion.path}"
# Calculate replacement position
# Replace from @ to cursor
start_position = -(cursor_pos - context.start_pos)
yield Completion(
text=f"@{suggestion.path}",
start_position=start_position,
display=display
)
def create_combined_completer(commands: List[str], root_dir: str):
"""
Factory function to create a CombinedCompleter.
Returns a proper Completer subclass that handles both / and @.
"""
Completer, Completion = _get_prompt_toolkit()
file_service = FileSearchService(root_dir)
class CombinedCompleter(Completer):
"""
Combined completer for both / commands and @ mentions.
Properly inherits from prompt_toolkit Completer.
"""
def get_completions(self, document, complete_event):
"""Get completions for / or @."""
text = document.text_before_cursor.lstrip()
# Check for slash command
if text.startswith('/'):
cmd_text = text[1:].lower()
for cmd in commands:
if cmd.lower().startswith(cmd_text):
yield Completion(
f'/{cmd}',
start_position=-len(text),
display=f'/{cmd}'
)
return
# Check for @ mention
full_text = document.text_before_cursor
cursor_pos = len(full_text)
context = detect_at_mention(full_text, cursor_pos)
if context is None or not context.is_active:
return
# Search files
results = file_service.search(context.query, max_results=15)
# Yield completions
for suggestion in results:
icon = "📁 " if suggestion.file_type == "directory" else "📄 "
display = f"{icon}{suggestion.path}"
start_position = -(cursor_pos - context.start_pos)
yield Completion(
text=f"@{suggestion.path}",
start_position=start_position,
display=display
)
return CombinedCompleter()
# Backward compatible alias
class CombinedCompleter:
"""
Wrapper class for backward compatibility.
Use create_combined_completer() for proper Completer subclass.
"""
def __new__(cls, commands: List[str], root_dir: str):
return create_combined_completer(commands, root_dir)
# ============================================================================
# Exports
# ============================================================================
__all__ = [
'AtMentionContext',
'FileSuggestion',
'detect_at_mention',
'FileSearchService',
'AtMentionCompleter',
'CombinedCompleter',
]
@@ -0,0 +1,587 @@
"""
Message Queue for Interactive Mode.
Provides FIFO message queuing while the AI agent is processing,
inspired by Claude Code, Windsurf Cascade, Cursor, and Gemini CLI.
Features:
- Queue messages while agent is processing
- Auto-process queue when agent becomes idle
- Visual indicators for queue status
- Thread-safe operations
"""
import threading
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, List, Optional
# ============================================================================
# Enums
# ============================================================================
class ProcessingState(Enum):
"""State of the agent processing."""
IDLE = "idle"
PROCESSING = "processing"
WAITING_APPROVAL = "waiting_approval"
class MessagePriority(Enum):
"""Priority levels for queued messages."""
LOW = "low"
NORMAL = "normal"
HIGH = "high"
# ============================================================================
# Data Classes
# ============================================================================
@dataclass
class QueuedMessage:
"""A message in the queue with metadata."""
content: str
timestamp: float = field(default_factory=time.time)
priority: MessagePriority = MessagePriority.NORMAL
# ============================================================================
# MessageQueue - Thread-safe FIFO queue
# ============================================================================
class MessageQueue:
"""
Thread-safe FIFO message queue.
Provides basic queue operations with thread safety for concurrent access.
"""
def __init__(self):
self._messages: List[str] = []
self._lock = threading.Lock()
@property
def is_empty(self) -> bool:
"""Check if queue is empty."""
with self._lock:
return len(self._messages) == 0
@property
def count(self) -> int:
"""Get number of messages in queue."""
with self._lock:
return len(self._messages)
def add(self, message: str) -> bool:
"""
Add a message to the end of the queue.
Args:
message: The message to add
Returns:
True if message was added, False if empty/whitespace
"""
trimmed = message.strip() if message else ""
if not trimmed:
return False
with self._lock:
self._messages.append(trimmed)
return True
def pop(self) -> Optional[str]:
"""
Remove and return the first message (FIFO).
Returns:
The first message, or None if queue is empty
"""
with self._lock:
if self._messages:
return self._messages.pop(0)
return None
def peek(self) -> Optional[str]:
"""
View the first message without removing it.
Returns:
The first message, or None if queue is empty
"""
with self._lock:
if self._messages:
return self._messages[0]
return None
def clear(self) -> None:
"""Remove all messages from the queue."""
with self._lock:
self._messages.clear()
def get_all(self) -> List[str]:
"""
Get all messages as a list (does not remove them).
Returns:
Copy of all messages in queue order
"""
with self._lock:
return self._messages.copy()
def remove_at(self, index: int) -> Optional[str]:
"""
Remove message at specific index.
Args:
index: The index to remove (0-based)
Returns:
The removed message, or None if index is invalid
"""
with self._lock:
if 0 <= index < len(self._messages):
return self._messages.pop(index)
return None
# ============================================================================
# StateManager - Processing state management
# ============================================================================
class StateManager:
"""
Manages the processing state with optional callbacks.
Tracks whether the agent is idle, processing, or waiting for approval.
"""
def __init__(self, on_state_change: Optional[Callable[[ProcessingState, ProcessingState], None]] = None):
self._state = ProcessingState.IDLE
self._lock = threading.Lock()
self._on_state_change = on_state_change
@property
def current_state(self) -> ProcessingState:
"""Get current processing state."""
with self._lock:
return self._state
@property
def is_idle(self) -> bool:
"""Check if currently idle."""
return self.current_state == ProcessingState.IDLE
@property
def is_processing(self) -> bool:
"""Check if currently processing."""
return self.current_state == ProcessingState.PROCESSING
def set_state(self, new_state: ProcessingState) -> None:
"""
Set the processing state.
Args:
new_state: The new state to set
"""
with self._lock:
old_state = self._state
self._state = new_state
# Call callback outside lock to prevent deadlock
if self._on_state_change and old_state != new_state:
self._on_state_change(old_state, new_state)
# ============================================================================
# QueueDisplay - Visual formatting
# ============================================================================
class QueueDisplay:
"""
Formats queue and status for display.
Provides visual indicators for queue status, inspired by Codex CLI.
"""
def __init__(
self,
queue: MessageQueue,
state_manager: Optional[StateManager] = None,
max_message_length: int = 50
):
self._queue = queue
self._state_manager = state_manager
self._max_length = max_message_length
def format_queue(self) -> str:
"""
Format queued messages for display.
Returns:
Formatted string with queued messages
"""
messages = self._queue.get_all()
if not messages:
return ""
lines = []
for msg in messages:
# Truncate long messages
display_msg = msg
if len(msg) > self._max_length:
display_msg = msg[:self._max_length - 3] + "..."
lines.append(f" ↳ {display_msg}")
return "\n".join(lines)
def format_status(self) -> str:
"""
Format current processing status.
Returns:
Status string (e.g., "⏳ Processing..." or "")
"""
if not self._state_manager:
return ""
state = self._state_manager.current_state
if state == ProcessingState.PROCESSING:
return "⏳ Processing..."
elif state == ProcessingState.WAITING_APPROVAL:
return "🔒 Waiting for approval..."
return ""
def format_queue_count(self) -> str:
"""
Format queue count indicator.
Returns:
Count string (e.g., "📋 Queued (2)")
"""
count = self._queue.count
if count == 0:
return ""
return f"📋 Queued ({count})"
# ============================================================================
# MessageQueueHandler - Main handler class
# ============================================================================
class MessageQueueHandler:
"""
Main handler for message queue functionality.
Coordinates queue, state, and processing.
"""
def __init__(
self,
processor: Optional[Callable[[str], str]] = None,
on_response: Optional[Callable[[str], None]] = None
):
self._processor = processor
self._on_response = on_response
self._queue = MessageQueue()
self._state_manager = StateManager(on_state_change=self._on_state_change)
self._processing_lock = threading.Lock()
@property
def queue(self) -> MessageQueue:
"""Get the message queue."""
return self._queue
@property
def state_manager(self) -> StateManager:
"""Get the state manager."""
return self._state_manager
def submit(self, message: str) -> bool:
"""
Submit a message for processing.
If idle, processes immediately. If processing, queues the message.
Args:
message: The message to submit
Returns:
True if processed/queued, False if empty message
"""
trimmed = message.strip() if message else ""
if not trimmed:
return False
if self._state_manager.is_idle:
# Process immediately
self._process_message(trimmed)
else:
# Queue for later
self._queue.add(trimmed)
return True
def on_processing_complete(self) -> None:
"""
Called when current processing completes.
Sets state to idle and processes next queued message if any.
"""
self._state_manager.set_state(ProcessingState.IDLE)
self._process_next_in_queue()
def get_status(self) -> dict:
"""
Get current status.
Returns:
Dict with queue_count and state
"""
return {
'queue_count': self._queue.count,
'state': self._state_manager.current_state.value,
'messages': self._queue.get_all()
}
def clear_queue(self) -> None:
"""Clear all queued messages."""
self._queue.clear()
def _process_message(self, message: str) -> None:
"""Process a single message."""
if not self._processor:
return
with self._processing_lock:
self._state_manager.set_state(ProcessingState.PROCESSING)
try:
response = self._processor(message)
if self._on_response and response:
self._on_response(response)
finally:
self._state_manager.set_state(ProcessingState.IDLE)
def _process_next_in_queue(self) -> None:
"""Process the next message in queue if any."""
next_msg = self._queue.pop()
if next_msg:
self._process_message(next_msg)
def _on_state_change(self, old_state: ProcessingState, new_state: ProcessingState) -> None:
"""Handle state changes."""
# Auto-process queue when becoming idle
if new_state == ProcessingState.IDLE and old_state == ProcessingState.PROCESSING:
# Don't auto-process here to avoid recursion
# The on_processing_complete method handles this
pass
# ============================================================================
# AsyncProcessor - Background processing
# ============================================================================
class AsyncProcessor:
"""
Runs work functions in background threads.
Allows agent processing to happen asynchronously while
the main thread remains responsive for user input.
"""
def __init__(
self,
on_complete: Optional[Callable[[str], None]] = None,
on_status: Optional[Callable[[str], None]] = None,
on_error: Optional[Callable[[Exception], None]] = None
):
self._on_complete = on_complete
self._on_status = on_status
self._on_error = on_error
self._thread: Optional[threading.Thread] = None
self._running = False
self._lock = threading.Lock()
@property
def is_running(self) -> bool:
"""Check if currently processing."""
with self._lock:
return self._running
def start(self, work_fn: Callable[[], str]) -> None:
"""
Start processing work in background thread.
Args:
work_fn: Function to execute (should return result string)
"""
def _worker():
try:
with self._lock:
self._running = True
result = work_fn()
with self._lock:
self._running = False
if self._on_complete:
self._on_complete(result)
except Exception as e:
with self._lock:
self._running = False
if self._on_error:
self._on_error(e)
self._thread = threading.Thread(target=_worker, daemon=True)
self._thread.start()
def update_status(self, status: str) -> None:
"""Update status (can be called from work function)."""
if self._on_status:
self._on_status(status)
# ============================================================================
# LiveStatusDisplay - Real-time status display
# ============================================================================
class LiveStatusDisplay:
"""
Tracks and displays real-time status during processing.
Shows tool calls, command executions, and current status.
"""
def __init__(self):
self._status = ""
self._tool_calls: List[dict] = []
self._commands: List[str] = []
self._lock = threading.Lock()
@property
def current_status(self) -> str:
"""Get current status message."""
with self._lock:
return self._status
@property
def tool_calls(self) -> List[dict]:
"""Get list of tool calls."""
with self._lock:
return self._tool_calls.copy()
@property
def commands(self) -> List[str]:
"""Get list of commands executed."""
with self._lock:
return self._commands.copy()
def update_status(self, status: str) -> None:
"""Update current status message."""
with self._lock:
self._status = status
def add_tool_call(self, name: str, args: dict) -> None:
"""Add a tool call to the display."""
with self._lock:
self._tool_calls.append({
'name': name,
'args': args,
'timestamp': time.time()
})
def add_command(self, command: str) -> None:
"""Add a command execution to the display."""
with self._lock:
self._commands.append(command)
def format(self) -> str:
"""Format current status for display."""
lines = []
with self._lock:
if self._status:
lines.append(f"⏳ {self._status}")
for tool in self._tool_calls[-3:]: # Show last 3 tools
lines.append(f" 🔧 {tool['name']}")
for cmd in self._commands[-2:]: # Show last 2 commands
display_cmd = cmd[:40] + "..." if len(cmd) > 40 else cmd
lines.append(f" 💻 {display_cmd}")
return "\n".join(lines)
def clear(self) -> None:
"""Clear all status."""
with self._lock:
self._status = ""
self._tool_calls.clear()
self._commands.clear()
# ============================================================================
# NonBlockingInput - Async input handling
# ============================================================================
class NonBlockingInput:
"""
Handles user input asynchronously.
Allows users to type new messages while processing is ongoing.
"""
def __init__(self):
self._queue = MessageQueue()
@property
def has_pending(self) -> bool:
"""Check if there are pending inputs."""
return not self._queue.is_empty
@property
def pending_count(self) -> int:
"""Get number of pending inputs."""
return self._queue.count
def submit(self, message: str) -> bool:
"""Submit a new input message."""
return self._queue.add(message)
def pop(self) -> Optional[str]:
"""Get next pending input."""
return self._queue.pop()
def peek(self) -> Optional[str]:
"""View next pending input without removing."""
return self._queue.peek()
def get_all(self) -> List[str]:
"""Get all pending inputs."""
return self._queue.get_all()
def clear(self) -> None:
"""Clear all pending inputs."""
self._queue.clear()
# ============================================================================
# Exports
# ============================================================================
__all__ = [
'ProcessingState',
'MessagePriority',
'QueuedMessage',
'MessageQueue',
'StateManager',
'QueueDisplay',
'MessageQueueHandler',
'AsyncProcessor',
'LiveStatusDisplay',
'NonBlockingInput',
]
+738 -94
View File
@@ -2,9 +2,20 @@
import sys
import argparse
import warnings
import os
# Suppress crewai RuntimeWarning about module loading order (only in non-debug mode)
# This warning is harmless and occurs when running as `python -m praisonai.cli.main`
if os.environ.get('LOGLEVEL', 'INFO').upper() != 'DEBUG':
warnings.filterwarnings(
"ignore",
message=".*found in sys.modules after import of package.*",
category=RuntimeWarning
)
from praisonai.version import __version__
import yaml
import os
import time
from rich import print
from dotenv import load_dotenv
@@ -48,41 +59,14 @@ def _get_chainlit_run():
from chainlit.cli import chainlit_run
return chainlit_run
try:
import gradio as gr
GRADIO_AVAILABLE = True
except ImportError:
pass
try:
import praisonai.api.call as call_module
CALL_MODULE_AVAILABLE = True
except ImportError:
pass
try:
import crewai
CREWAI_AVAILABLE = True
except ImportError:
pass
try:
import autogen
AUTOGEN_AVAILABLE = True
except ImportError:
pass
try:
from praisonaiagents import Agent as PraisonAgent, Task as PraisonTask, PraisonAIAgents
PRAISONAI_AVAILABLE = True
except ImportError:
pass
try:
from unsloth import FastLanguageModel
TRAIN_AVAILABLE = True
except ImportError:
pass
# Use find_spec for fast availability checks (no actual import)
import importlib.util
GRADIO_AVAILABLE = importlib.util.find_spec("gradio") is not None
CALL_MODULE_AVAILABLE = importlib.util.find_spec("praisonai.api.call") is not None
CREWAI_AVAILABLE = importlib.util.find_spec("crewai") is not None
AUTOGEN_AVAILABLE = importlib.util.find_spec("autogen") is not None
PRAISONAI_AVAILABLE = importlib.util.find_spec("praisonaiagents") is not None
TRAIN_AVAILABLE = importlib.util.find_spec("unsloth") is not None
logging.basicConfig(level=os.environ.get('LOGLEVEL', 'INFO'), format='%(asctime)s - %(levelname)s - %(message)s')
logging.getLogger('alembic').setLevel(logging.ERROR)
@@ -704,7 +688,7 @@ class PraisonAI:
# Interactive TUI - terminal interface with slash commands
parser.add_argument("--interactive", "-i", action="store_true", help="Start interactive terminal mode with slash commands")
parser.add_argument("--chat-mode", action="store_true", dest="chat_mode", help="Run single prompt in interactive style (non-interactive, for testing)")
parser.add_argument("--chat-mode", "--chat", action="store_true", dest="chat_mode", help="Run single prompt in interactive style (non-interactive, for testing)")
# Autonomy Mode - control AI action approval
parser.add_argument("--autonomy", type=str, choices=["suggest", "auto_edit", "full_auto"], help="Set autonomy mode for AI actions")
@@ -1521,6 +1505,7 @@ class PraisonAI:
args: Parsed command line arguments
"""
try:
from praisonaiagents import Agent as PraisonAgent
from praisonaiagents.memory import WorkflowManager
from rich import print
from rich.table import Table
@@ -2908,6 +2893,8 @@ Provide ONLY the commit message, no explanations."""
prompt = f"{mention_context}# Task:\n{prompt}"
if PRAISONAI_AVAILABLE:
from praisonaiagents import Agent as PraisonAgent
agent_config = {
"name": "DirectAgent",
"role": "Assistant",
@@ -2915,8 +2902,11 @@ Provide ONLY the commit message, no explanations."""
"backstory": "You are a helpful AI assistant"
}
# Suppress verbose output in interactive mode (no Agent Info, Task, Response panels)
if getattr(self, '_interactive_mode', False):
# Set verbose=False by default, only enable with --verbose flag
# This shows minimal "Generating..." status instead of full Agent Info panels
if hasattr(self, 'args') and getattr(self.args, 'verbose', False):
agent_config["verbose"] = True
else:
agent_config["verbose"] = False
# Add llm if specified
@@ -3101,7 +3091,13 @@ Provide ONLY the commit message, no explanations."""
flow.display_workflow_start("Direct Prompt", ["DirectAgent"])
agent = PraisonAgent(**agent_config)
result = agent.start(prompt)
# Run with minimal status display when verbose=False
is_verbose = agent_config.get("verbose", False)
if not is_verbose:
result = self._run_with_status_display(agent, prompt)
else:
result = agent.start(prompt)
# ===== POST-PROCESSING WITH NEW FEATURES =====
@@ -3225,6 +3221,98 @@ Now, {final_instruction.lower()}:"""
print("pip install praisonaiagents # For PraisonAIAgents\n")
sys.exit(1)
def _run_with_status_display(self, agent, prompt):
"""
Run agent with minimal status display (spinner + tool/handoff updates).
Shows:
- "Generating..." with spinner while processing
- Real-time tool call notifications
- Agent handoff notifications
"""
import threading
import time
from rich.console import Console
from rich.live import Live
from rich.text import Text
console = Console()
status_info = {
'status': 'Generating...',
'tool_calls': [],
'handoffs': [],
'done': False,
'result': None,
'error': None,
'start_time': time.time()
}
def build_status_display():
"""Build the status display text."""
elapsed = time.time() - status_info['start_time']
# Main status with spinner
text = Text()
text.append("⏳ ", style="cyan")
text.append(f"{status_info['status']} ", style="bold")
text.append(f"({elapsed:.1f}s)", style="dim")
# Show recent tool calls
if status_info['tool_calls']:
text.append("\n")
for tool in status_info['tool_calls'][-3:]: # Show last 3
text.append(f" ⚙ {tool}", style="dim yellow")
text.append("\n")
# Show handoffs
if status_info['handoffs']:
for handoff in status_info['handoffs'][-2:]: # Show last 2
text.append(f" → {handoff}", style="dim cyan")
text.append("\n")
return text
def run_agent():
"""Run the agent in background thread."""
try:
# Monkey-patch agent's execute_tool to capture tool calls
original_execute_tool = agent.execute_tool
def patched_execute_tool(tool_name, *args, **kwargs):
status_info['tool_calls'].append(tool_name)
status_info['status'] = f"Using {tool_name}..."
return original_execute_tool(tool_name, *args, **kwargs)
agent.execute_tool = patched_execute_tool
status_info['result'] = agent.start(prompt)
except Exception as e:
status_info['error'] = e
finally:
status_info['done'] = True
# Start agent in background thread
thread = threading.Thread(target=run_agent, daemon=True)
thread.start()
# Show live status while processing
try:
with Live(build_status_display(), console=console, refresh_per_second=4, transient=True) as live:
while not status_info['done']:
live.update(build_status_display())
time.sleep(0.1)
except KeyboardInterrupt:
console.print("\n[dim]Interrupted[/dim]")
return None
# Wait for thread to complete
thread.join(timeout=1.0)
# Handle result
if status_info['error']:
console.print(f"[red]Error: {status_info['error']}[/red]")
return None
return status_info['result']
def _handle_serve_command(self, args, unknown_args):
"""
Handle the serve command - start API server for agents.
@@ -3780,10 +3868,15 @@ Now, {final_instruction.lower()}:"""
- Streaming text output (no boxes)
- Tool status indicators
- Built-in tools (file ops, shell, web search)
- @ mentions for file inclusion
- Autocomplete for slash commands and @file mentions
- /compact for context compression
- /model for model switching
- /stats for token/cost tracking
- /undo for reverting changes
"""
try:
from rich.console import Console
import os
console = Console()
@@ -3793,23 +3886,87 @@ Now, {final_instruction.lower()}:"""
# Load interactive tools
tools_list = self._load_interactive_tools()
# Import message queue components
from praisonai.cli.features.message_queue import (
MessageQueue, StateManager, QueueDisplay, ProcessingState,
AsyncProcessor, LiveStatusDisplay
)
import threading
# Create message queue and state manager
message_queue = MessageQueue()
state_manager = StateManager()
queue_display = QueueDisplay(message_queue, state_manager)
live_status = LiveStatusDisplay()
processing_lock = threading.Lock()
# Session state
session_state = {
'show_profiling': False,
'current_model': getattr(args, 'llm', None) or os.environ.get('OPENAI_MODEL_NAME', 'gpt-4o-mini'),
'conversation_history': [],
'total_input_tokens': 0,
'total_output_tokens': 0,
'total_cost': 0.0,
'request_count': 0,
'undo_stack': [], # Stack of (prompt, response) for undo
'message_queue': message_queue,
'state_manager': state_manager,
'queue_display': queue_display,
'live_status': live_status,
'processing_lock': processing_lock,
'async_processor': None, # Will be created per-request
}
# Try to use prompt_toolkit for autocomplete
prompt_session = None
try:
from prompt_toolkit import PromptSession
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
from prompt_toolkit.history import InMemoryHistory
from praisonai.cli.features.at_mentions import CombinedCompleter
# Create combined completer for / commands and @ mentions
commands = ['help', 'exit', 'quit', 'clear', 'tools', 'profile', 'model', 'stats', 'compact', 'undo', 'queue', 'q']
combined_completer = CombinedCompleter(
commands=commands,
root_dir=os.getcwd()
)
prompt_session = PromptSession(
message="❯ ",
completer=combined_completer,
auto_suggest=AutoSuggestFromHistory(),
history=InMemoryHistory(),
complete_while_typing=True
)
except ImportError:
pass # Fall back to simple input
# Print welcome message
console.print("\n[bold cyan]PraisonAI Interactive Mode[/bold cyan]")
console.print("[dim]Type your prompt, use /help for commands, /exit to quit[/dim]")
console.print("[dim]Tools: file read/write, shell, web search[/dim]\n")
console.print(f"[dim]Model: {session_state['current_model']} | Tools: {len(tools_list)}[/dim]")
console.print("[dim]Use @file.txt to include file content | Queue messages while processing[/dim]\n")
running = True
while running:
try:
# Get user input
user_input = input("❯ ").strip()
# Get user input (with autocomplete if available)
if prompt_session:
user_input = prompt_session.prompt().strip()
else:
user_input = input("❯ ").strip()
if not user_input:
continue
# Handle slash commands
if user_input.startswith("/"):
cmd = user_input[1:].lower().split()[0] if user_input[1:] else ""
cmd_parts = user_input[1:].split(maxsplit=1)
cmd = cmd_parts[0].lower() if cmd_parts else ""
cmd_args = cmd_parts[1] if len(cmd_parts) > 1 else ""
if cmd in ["exit", "quit", "q"]:
console.print("[dim]Goodbye![/dim]")
running = False
@@ -3826,12 +3983,47 @@ Now, {final_instruction.lower()}:"""
name = getattr(tool, '__name__', str(tool))
console.print(f" • {name}")
continue
elif cmd == "profile":
session_state['show_profiling'] = not session_state['show_profiling']
status = "enabled" if session_state['show_profiling'] else "disabled"
console.print(f"[cyan]Profiling {status}[/cyan]")
continue
elif cmd == "model":
self._handle_model_command(console, cmd_args, session_state)
continue
elif cmd == "stats":
self._handle_stats_command(console, session_state)
continue
elif cmd == "compact":
self._handle_compact_command(console, session_state)
continue
elif cmd == "undo":
self._handle_undo_command(console, session_state)
continue
elif cmd == "queue":
self._handle_queue_command(console, cmd_args, session_state)
continue
else:
console.print(f"[yellow]Unknown command: /{cmd}. Type /help for available commands.[/yellow]")
continue
# Process the prompt with streaming
self._process_interactive_prompt(user_input, tools_list, console)
# Process @file mentions before sending to LLM
processed_input = self._process_at_mentions(user_input, console)
# Check if already processing - queue if so
if state_manager.is_processing:
message_queue.add(processed_input)
queue_count = message_queue.count
console.print(f"[dim]📋 Queued ({queue_count}) - processing will continue in order[/dim]")
continue
# Start async processing
self._process_interactive_prompt_async(
processed_input,
tools_list,
console,
session_state=session_state
)
except KeyboardInterrupt:
console.print("\n[dim]Use /exit to quit[/dim]")
@@ -3892,23 +4084,312 @@ Now, {final_instruction.lower()}:"""
def _print_interactive_help(self, console):
"""Print help for interactive mode."""
console.print("\n[bold]Commands:[/bold]")
console.print(" /help - Show this help")
console.print(" /exit - Exit interactive mode")
console.print(" /clear - Clear screen")
console.print(" /tools - List available tools")
console.print(" /help - Show this help")
console.print(" /exit - Exit interactive mode")
console.print(" /clear - Clear screen")
console.print(" /tools - List available tools")
console.print(" /profile - Toggle profiling (show timing breakdown)")
console.print(" /model [name] - Show or change current model")
console.print(" /stats - Show session statistics (tokens, cost)")
console.print(" /compact - Compress conversation history")
console.print(" /undo - Undo last response")
console.print(" /queue - Show queued messages")
console.print(" /queue clear - Clear message queue")
console.print("\n[bold]@ Mentions:[/bold]")
console.print(" @file.txt - Include file content in prompt")
console.print(" @src/ - Include directory listing")
console.print("\n[bold]Features:[/bold]")
console.print(" • Streaming responses")
console.print(" • File operations (read, write, list)")
console.print(" • Shell command execution")
console.print(" • Web search")
console.print(" • @file:path to include file content")
console.print(" • Context compression for long sessions")
console.print(" • Queue messages while agent is processing")
console.print("")
def _process_at_mentions(self, user_input, console):
"""
Process @file mentions in user input.
Supports:
- @file.txt - Include file content
- @path/to/file - Include file content
- @directory/ - Include directory listing
Inspired by Gemini CLI and Claude Code @file syntax.
"""
import re
import os
# Pattern to match @path (but not email addresses)
# Must be at start or preceded by whitespace
pattern = r'(?:^|\s)@([^\s@]+)'
matches = re.findall(pattern, user_input)
if not matches:
return user_input
processed_input = user_input
file_contents = []
for path in matches:
# Clean up the path
clean_path = path.strip()
# Expand ~ to home directory
if clean_path.startswith('~'):
clean_path = os.path.expanduser(clean_path)
# Make absolute if relative
if not os.path.isabs(clean_path):
clean_path = os.path.abspath(clean_path)
try:
if os.path.isfile(clean_path):
# Read file content
with open(clean_path, 'r', encoding='utf-8', errors='ignore') as f:
content = f.read()
# Truncate if too large (>50KB)
if len(content) > 50000:
content = content[:50000] + "\n... [truncated, file too large]"
file_contents.append(f"\n--- Content of {path} ---\n{content}\n--- End of {path} ---\n")
console.print(f"[dim]📄 Included: {path} ({len(content)} chars)[/dim]")
elif os.path.isdir(clean_path):
# List directory contents
try:
entries = os.listdir(clean_path)
# Filter out hidden files and common ignore patterns
entries = [e for e in entries if not e.startswith('.') and e not in ['node_modules', '__pycache__', 'venv', '.git']]
entries.sort()
listing = "\n".join(f" {e}" for e in entries[:50])
if len(entries) > 50:
listing += f"\n ... and {len(entries) - 50} more files"
file_contents.append(f"\n--- Directory listing of {path} ---\n{listing}\n--- End of {path} ---\n")
console.print(f"[dim]📁 Listed: {path} ({len(entries)} items)[/dim]")
except PermissionError:
console.print(f"[yellow]⚠ Permission denied: {path}[/yellow]")
else:
console.print(f"[yellow]⚠ Not found: {path}[/yellow]")
except Exception as e:
console.print(f"[yellow]⚠ Error reading {path}: {e}[/yellow]")
# Remove @mentions from the original input and append file contents
for path in matches:
processed_input = processed_input.replace(f"@{path}", "")
processed_input = processed_input.strip()
if file_contents:
processed_input = processed_input + "\n" + "\n".join(file_contents)
return processed_input
def _handle_model_command(self, console, args, session_state):
"""Handle /model command - show or change current model."""
if not args:
# Show current model
console.print(f"[cyan]Current model: {session_state['current_model']}[/cyan]")
console.print("\n[dim]Available models (examples):[/dim]")
console.print(" • gpt-4o, gpt-4o-mini")
console.print(" • claude-3-5-sonnet, claude-3-haiku")
console.print(" • gemini-2.0-flash, gemini-1.5-pro")
console.print("\n[dim]Usage: /model <model-name>[/dim]")
else:
# Change model
new_model = args.strip()
old_model = session_state['current_model']
session_state['current_model'] = new_model
console.print(f"[green]✓ Model changed: {old_model} → {new_model}[/green]")
def _handle_stats_command(self, console, session_state):
"""Handle /stats command - show session statistics."""
from praisonai.cli.features.cost_tracker import get_pricing
console.print("\n[bold cyan]Session Statistics[/bold cyan]")
console.print(f" Model: {session_state['current_model']}")
console.print(f" Requests: {session_state['request_count']}")
console.print(f" Input tokens: {session_state['total_input_tokens']:,}")
console.print(f" Output tokens: {session_state['total_output_tokens']:,}")
console.print(f" Total tokens: {session_state['total_input_tokens'] + session_state['total_output_tokens']:,}")
# Calculate estimated cost
try:
pricing = get_pricing(session_state['current_model'])
cost = pricing.calculate_cost(
session_state['total_input_tokens'],
session_state['total_output_tokens']
)
console.print(f" Estimated cost: ${cost:.4f}")
except Exception:
pass
# Show conversation history size
history_len = len(session_state['conversation_history'])
console.print(f" History turns: {history_len}")
console.print("")
def _handle_compact_command(self, console, session_state):
"""
Handle /compact command - compress conversation history.
Inspired by Claude Code's /compact and Gemini CLI's /compress.
Uses LLM to summarize older conversation turns while keeping recent ones.
"""
history = session_state['conversation_history']
if len(history) < 4:
console.print("[yellow]Not enough conversation history to compact (need at least 4 turns)[/yellow]")
return
console.print("[dim]Compacting conversation history...[/dim]")
try:
from praisonaiagents import Agent
# Keep the last 2 turns (4 messages: 2 user + 2 assistant)
keep_count = 4
to_compress = history[:-keep_count] if len(history) > keep_count else []
to_keep = history[-keep_count:] if len(history) > keep_count else history
if not to_compress:
console.print("[yellow]Not enough old history to compress[/yellow]")
return
# Format history for summarization
history_text = ""
for i, msg in enumerate(to_compress):
role = msg.get('role', 'unknown')
content = msg.get('content', '')
if len(content) > 500:
content = content[:500] + "..."
history_text += f"{role}: {content}\n\n"
# Create summarization prompt
summary_prompt = f"""Summarize the following conversation history into a concise state snapshot.
Focus on:
- Key facts and decisions made
- Important context that should be remembered
- Any ongoing tasks or goals
Conversation to summarize:
{history_text}
Provide a concise summary (max 200 words):"""
# Use a lightweight agent for summarization
summarizer = Agent(
name="Summarizer",
role="Conversation Summarizer",
goal="Create concise summaries of conversations",
backstory="You summarize conversations while preserving key information.",
verbose=False,
llm=session_state['current_model']
)
summary = summarizer.chat(summary_prompt, stream=False)
# Replace old history with summary + recent turns
new_history = [
{"role": "system", "content": f"[Previous conversation summary]: {summary}"},
*to_keep
]
old_count = len(history)
session_state['conversation_history'] = new_history
new_count = len(new_history)
console.print(f"[green]✓ Compacted {old_count} turns → {new_count} turns[/green]")
console.print(f"[dim]Summary: {str(summary)[:100]}...[/dim]")
except Exception as e:
console.print(f"[red]Error compacting: {e}[/red]")
def _handle_undo_command(self, console, session_state):
"""
Handle /undo command - undo the last response.
Inspired by Codex CLI's /undo and Gemini CLI's /restore.
Removes the last user prompt and assistant response from history.
"""
history = session_state['conversation_history']
undo_stack = session_state['undo_stack']
if len(history) < 2:
console.print("[yellow]Nothing to undo[/yellow]")
return
# Remove last assistant response and user prompt
if len(history) >= 2:
removed_assistant = history.pop()
removed_user = history.pop()
# Store in undo stack for potential redo
undo_stack.append((removed_user, removed_assistant))
console.print("[green]✓ Undone last turn[/green]")
console.print(f"[dim]Removed: {str(removed_user.get('content', ''))[:50]}...[/dim]")
else:
console.print("[yellow]Not enough history to undo[/yellow]")
def _handle_queue_command(self, console, args, session_state):
"""
Handle /queue command - show or manage message queue.
Usage:
- /queue - Show queued messages
- /queue clear - Clear the queue
- /queue remove N - Remove message at index N
"""
message_queue = session_state.get('message_queue')
state_manager = session_state.get('state_manager')
queue_display = session_state.get('queue_display')
if not message_queue:
console.print("[yellow]Message queue not initialized[/yellow]")
return
args = args.strip().lower() if args else ""
if args == "clear":
count = message_queue.count
message_queue.clear()
console.print(f"[green]✓ Cleared {count} queued message(s)[/green]")
elif args.startswith("remove "):
try:
index = int(args.split()[1])
removed = message_queue.remove_at(index)
if removed:
console.print(f"[green]✓ Removed: {removed[:50]}...[/green]")
else:
console.print(f"[yellow]Invalid index: {index}[/yellow]")
except (ValueError, IndexError):
console.print("[yellow]Usage: /queue remove <index>[/yellow]")
else:
# Show queue status
if state_manager:
status = queue_display.format_status() if queue_display else ""
if status:
console.print(f"[cyan]{status}[/cyan]")
if message_queue.is_empty:
console.print("[dim]No messages in queue[/dim]")
else:
console.print(f"\n[bold cyan]Queued Messages ({message_queue.count}):[/bold cyan]")
for i, msg in enumerate(message_queue.get_all()):
display_msg = msg[:60] + "..." if len(msg) > 60 else msg
console.print(f" {i}. ↳ {display_msg}")
console.print("\n[dim]Use /queue clear to clear, /queue remove N to remove[/dim]")
def _run_chat_mode(self, prompt, args):
"""
Run a single prompt in interactive style (non-interactive mode for testing).
Usage: praisonai "your prompt" --chat
praisonai "your prompt" --chat-mode (alias)
This runs the prompt using the same agent/tools as interactive mode
but exits after one response (useful for testing and scripting).
@@ -3924,18 +4405,150 @@ Now, {final_instruction.lower()}:"""
console.print(f"[dim]Chat mode: {len(tools_list)} tools available[/dim]")
console.print(f"[dim]Prompt: {prompt[:50]}{'...' if len(prompt) > 50 else ''}[/dim]\n")
# Process the prompt
self._process_interactive_prompt(prompt, tools_list, console)
# Process the prompt with profiling enabled for testing
self._process_interactive_prompt(prompt, tools_list, console, show_profiling=True)
return None
def _process_interactive_prompt(self, prompt, tools_list, console):
def _process_interactive_prompt_async(self, prompt, tools_list, console, session_state=None):
"""
Process a prompt asynchronously with live status display.
Shows real-time status (tools, commands) while processing in background.
Allows user to queue more messages while processing.
"""
import threading
import time
from rich.live import Live
from rich.panel import Panel
from rich.text import Text
from praisonai.cli.features.message_queue import ProcessingState
state_manager = session_state['state_manager']
message_queue = session_state['message_queue']
live_status = session_state['live_status']
# Set processing state
state_manager.set_state(ProcessingState.PROCESSING)
live_status.clear()
live_status.update_status("Thinking...")
# Result container for thread communication
result_container = {'response': None, 'error': None, 'done': False}
def process_in_background():
"""Run agent processing in background thread."""
try:
from praisonaiagents import Agent
import logging
import warnings
# Suppress noisy loggers
for logger_name in ["httpx", "httpcore", "duckduckgo_search", "crawl4ai"]:
logging.getLogger(logger_name).setLevel(logging.WARNING)
with warnings.catch_warnings():
warnings.filterwarnings("ignore")
model = session_state.get('current_model')
live_status.update_status("Creating agent...")
# Create agent with tools
agent = Agent(
name="Assistant",
role="Helpful AI Assistant",
goal="Help the user with their tasks",
backstory="You are a helpful AI assistant with access to tools.",
tools=tools_list if tools_list else None,
verbose=False,
llm=model
)
live_status.update_status("Calling LLM...")
# Get response
response = agent.chat(prompt, stream=False)
result_container['response'] = str(response) if response else ""
except Exception as e:
result_container['error'] = e
finally:
result_container['done'] = True
# Start background thread
bg_thread = threading.Thread(target=process_in_background, daemon=True)
bg_thread.start()
# Show live status while processing
try:
with Live(console=console, refresh_per_second=4, transient=True) as live:
while not result_container['done']:
# Build status display
status_text = Text()
status_text.append("⏳ ", style="cyan")
status_text.append(live_status.current_status or "Processing...", style="dim")
# Show queued messages if any
queue_count = message_queue.count
if queue_count > 0:
status_text.append(f"\n📋 Queued: {queue_count}", style="dim yellow")
live.update(Panel(status_text, border_style="dim"))
time.sleep(0.1)
except KeyboardInterrupt:
console.print("\n[dim]Processing interrupted[/dim]")
# Wait for thread to finish
bg_thread.join(timeout=1.0)
# Handle result
if result_container['error']:
console.print(f"\n[red]Error: {result_container['error']}[/red]")
elif result_container['response']:
# Print response with streaming effect
console.print()
response_str = result_container['response']
words = response_str.split()
import sys
for i, word in enumerate(words):
console.print(word + " ", end="")
sys.stdout.flush()
if i % 15 == 14:
time.sleep(0.005)
console.print()
# Update session state
input_tokens = len(prompt) // 4
output_tokens = len(response_str) // 4
session_state['total_input_tokens'] += input_tokens
session_state['total_output_tokens'] += output_tokens
session_state['request_count'] += 1
session_state['conversation_history'].append({'role': 'user', 'content': prompt})
session_state['conversation_history'].append({'role': 'assistant', 'content': response_str})
# Set state back to idle
state_manager.set_state(ProcessingState.IDLE)
live_status.clear()
# Process next queued message if any
next_msg = message_queue.pop()
if next_msg:
console.print(f"\n[dim cyan]Processing queued message...[/dim cyan]")
self._process_interactive_prompt_async(next_msg, tools_list, console, session_state)
def _process_interactive_prompt(self, prompt, tools_list, console, show_profiling=False, session_state=None):
"""Process a prompt in interactive mode with streaming."""
from rich.live import Live
from rich.spinner import Spinner
import sys
import warnings
import logging
import time
# Profiling timestamps
timings = {}
timings['start'] = time.time()
# Store original log levels to restore later (no global impact)
original_levels = {}
@@ -3952,21 +4565,19 @@ Now, {final_instruction.lower()}:"""
try:
# Import agent
timings['import_start'] = time.time()
from praisonaiagents import Agent
timings['import_end'] = time.time()
# Custom callback to show tool usage
tool_calls_made = []
# Determine model to use
model = None
if session_state and session_state.get('current_model'):
model = session_state['current_model']
elif hasattr(self, 'args') and getattr(self.args, 'llm', None):
model = self.args.llm
def on_tool_start(tool_name, tool_input):
"""Called when a tool starts executing."""
tool_calls_made.append(tool_name)
console.print(f"[dim]⚙ Using {tool_name}...[/dim]", end="\r")
def on_tool_end(tool_name, tool_output):
"""Called when a tool finishes."""
console.print(f"[dim]✓ {tool_name} complete[/dim] ")
# Show thinking indicator
# Show thinking indicator and create agent
timings['agent_create_start'] = time.time()
with Live(Spinner("dots", text="Thinking...", style="cyan"), console=console, refresh_per_second=10, transient=True):
# Create agent with tools
agent = Agent(
@@ -3976,40 +4587,73 @@ Now, {final_instruction.lower()}:"""
backstory="You are a helpful AI assistant with access to tools for file operations, shell commands, and web search. Use tools when needed to complete tasks.",
tools=tools_list if tools_list else None,
verbose=False, # Suppress verbose panels
llm=getattr(self.args, 'llm', None) if hasattr(self, 'args') else None
llm=model
)
timings['agent_create_end'] = time.time()
# Get response with streaming
# Get response
console.print() # New line before response
# Try to use streaming if available
try:
# Use stream method for real streaming
response_text = ""
for chunk in agent.stream(prompt):
if chunk:
chunk_str = str(chunk)
console.print(chunk_str, end="")
sys.stdout.flush()
response_text += chunk_str
console.print("\n") # Final newline
except (AttributeError, TypeError):
# Fallback to chat method with simulated streaming
response = agent.chat(prompt)
timings['llm_start'] = time.time()
# Use chat method (streaming is handled internally by verbose mode)
response = agent.chat(prompt, stream=False)
timings['llm_end'] = time.time()
# Check if tools were used by looking at agent's tool execution history
if hasattr(agent, '_tool_calls') and agent._tool_calls:
for tool_call in agent._tool_calls:
tool_name = tool_call.get('name', 'unknown')
console.print(f"[dim]⚙ Used tool: {tool_name}[/dim]")
# Print response with simulated streaming effect
timings['display_start'] = time.time()
response_str = str(response) if response else ""
if response_str:
words = response_str.split()
for i, word in enumerate(words):
console.print(word + " ", end="")
sys.stdout.flush()
if i % 15 == 14: # Small pause every 15 words for streaming effect
time.sleep(0.005)
console.print() # Final newline
timings['display_end'] = time.time()
# Update session state with token estimates and history
if session_state is not None:
# Estimate tokens (rough: ~4 chars per token)
input_tokens = len(prompt) // 4
output_tokens = len(response_str) // 4
# Print response with word-by-word streaming effect
if response:
words = str(response).split()
for i, word in enumerate(words):
console.print(word + " ", end="")
sys.stdout.flush()
if i % 10 == 9: # Small pause every 10 words
import time
time.sleep(0.01)
console.print("\n") # Final newline
session_state['total_input_tokens'] += input_tokens
session_state['total_output_tokens'] += output_tokens
session_state['request_count'] += 1
# Add to conversation history
session_state['conversation_history'].append({
'role': 'user',
'content': prompt
})
session_state['conversation_history'].append({
'role': 'assistant',
'content': response_str
})
# Show profiling if enabled
if show_profiling:
timings['total'] = time.time() - timings['start']
console.print("\n[dim]─── Profiling ───[/dim]")
console.print(f"[dim]Import: {(timings['import_end'] - timings['import_start'])*1000:.1f}ms[/dim]")
console.print(f"[dim]Agent setup: {(timings['agent_create_end'] - timings['agent_create_start'])*1000:.1f}ms[/dim]")
console.print(f"[dim]LLM call: {(timings['llm_end'] - timings['llm_start'])*1000:.1f}ms[/dim]")
console.print(f"[dim]Display: {(timings['display_end'] - timings['display_start'])*1000:.1f}ms[/dim]")
console.print(f"[dim]Total: {timings['total']*1000:.1f}ms[/dim]")
except Exception as e:
console.print(f"\n[red]Error: {e}[/red]")
import traceback
traceback.print_exc()
finally:
# Restore original log levels (no global impact on main package)
for logger_name, level in original_levels.items():
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==2.3.49 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==2.3.54 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+465
View File
@@ -0,0 +1,465 @@
"""
PraisonAI Profiler Module
Standardized profiling for performance monitoring across praisonai and praisonai-agents.
Features:
- Import timing
- Function execution timing
- Flow tracking
- File/module usage tracking
- Memory usage (optional)
Usage:
from praisonai.profiler import Profiler, profile, profile_imports
# Profile a function
@profile
def my_function():
pass
# Profile a block
with Profiler.block("my_operation"):
do_something()
# Profile imports
with profile_imports():
import heavy_module
# Get report
Profiler.report()
"""
import time
import functools
import threading
import sys
import os
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable, Any
from contextlib import contextmanager
# ============================================================================
# Data Classes
# ============================================================================
@dataclass
class TimingRecord:
"""Record of a single timing measurement."""
name: str
duration_ms: float
category: str = "function"
file: str = ""
line: int = 0
timestamp: float = field(default_factory=time.time)
@dataclass
class ImportRecord:
"""Record of a module import."""
module: str
duration_ms: float
parent: str = ""
timestamp: float = field(default_factory=time.time)
@dataclass
class FlowRecord:
"""Record of execution flow."""
step: int
name: str
file: str
line: int
duration_ms: float
timestamp: float = field(default_factory=time.time)
# ============================================================================
# Profiler Class
# ============================================================================
class Profiler:
"""
Centralized profiler for performance monitoring.
Thread-safe singleton pattern for global access.
"""
_instance: Optional['Profiler'] = None
_lock = threading.Lock()
# Class-level storage
_timings: List[TimingRecord] = []
_imports: List[ImportRecord] = []
_flow: List[FlowRecord] = []
_enabled: bool = False
_flow_step: int = 0
_files_accessed: Dict[str, int] = {}
def __new__(cls):
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
@classmethod
def enable(cls) -> None:
"""Enable profiling."""
cls._enabled = True
@classmethod
def disable(cls) -> None:
"""Disable profiling."""
cls._enabled = False
@classmethod
def is_enabled(cls) -> bool:
"""Check if profiling is enabled."""
return cls._enabled or os.environ.get('PRAISONAI_PROFILE', '').lower() in ('1', 'true', 'yes')
@classmethod
def clear(cls) -> None:
"""Clear all profiling data."""
with cls._lock:
cls._timings.clear()
cls._imports.clear()
cls._flow.clear()
cls._flow_step = 0
cls._files_accessed.clear()
@classmethod
def record_timing(cls, name: str, duration_ms: float, category: str = "function",
file: str = "", line: int = 0) -> None:
"""Record a timing measurement."""
if not cls.is_enabled():
return
with cls._lock:
cls._timings.append(TimingRecord(
name=name,
duration_ms=duration_ms,
category=category,
file=file,
line=line
))
# Track file access
if file:
cls._files_accessed[file] = cls._files_accessed.get(file, 0) + 1
@classmethod
def record_import(cls, module: str, duration_ms: float, parent: str = "") -> None:
"""Record an import timing."""
if not cls.is_enabled():
return
with cls._lock:
cls._imports.append(ImportRecord(
module=module,
duration_ms=duration_ms,
parent=parent
))
@classmethod
def record_flow(cls, name: str, duration_ms: float, file: str = "", line: int = 0) -> None:
"""Record a flow step."""
if not cls.is_enabled():
return
with cls._lock:
cls._flow_step += 1
cls._flow.append(FlowRecord(
step=cls._flow_step,
name=name,
file=file,
line=line,
duration_ms=duration_ms
))
@classmethod
@contextmanager
def block(cls, name: str, category: str = "block"):
"""Context manager for profiling a block of code."""
start = time.time()
frame = sys._getframe(2) if hasattr(sys, '_getframe') else None
file = frame.f_code.co_filename if frame else ""
line = frame.f_lineno if frame else 0
try:
yield
finally:
duration_ms = (time.time() - start) * 1000
cls.record_timing(name, duration_ms, category, file, line)
cls.record_flow(name, duration_ms, file, line)
@classmethod
def get_timings(cls, category: Optional[str] = None) -> List[TimingRecord]:
"""Get timing records, optionally filtered by category."""
with cls._lock:
if category:
return [t for t in cls._timings if t.category == category]
return cls._timings.copy()
@classmethod
def get_imports(cls, min_duration_ms: float = 0) -> List[ImportRecord]:
"""Get import records, optionally filtered by minimum duration."""
with cls._lock:
if min_duration_ms > 0:
return [i for i in cls._imports if i.duration_ms >= min_duration_ms]
return cls._imports.copy()
@classmethod
def get_flow(cls) -> List[FlowRecord]:
"""Get flow records."""
with cls._lock:
return cls._flow.copy()
@classmethod
def get_files_accessed(cls) -> Dict[str, int]:
"""Get files accessed with counts."""
with cls._lock:
return cls._files_accessed.copy()
@classmethod
def get_summary(cls) -> Dict[str, Any]:
"""Get profiling summary."""
with cls._lock:
total_time = sum(t.duration_ms for t in cls._timings)
import_time = sum(i.duration_ms for i in cls._imports)
# Group by category
by_category: Dict[str, float] = {}
for t in cls._timings:
by_category[t.category] = by_category.get(t.category, 0) + t.duration_ms
# Top slowest
slowest = sorted(cls._timings, key=lambda x: x.duration_ms, reverse=True)[:10]
slowest_imports = sorted(cls._imports, key=lambda x: x.duration_ms, reverse=True)[:10]
return {
'total_time_ms': total_time,
'import_time_ms': import_time,
'timing_count': len(cls._timings),
'import_count': len(cls._imports),
'flow_steps': len(cls._flow),
'files_accessed': len(cls._files_accessed),
'by_category': by_category,
'slowest_operations': [(s.name, s.duration_ms) for s in slowest],
'slowest_imports': [(s.module, s.duration_ms) for s in slowest_imports],
}
@classmethod
def report(cls, output: str = "console") -> str:
"""Generate and output profiling report."""
summary = cls.get_summary()
lines = [
"=" * 60,
"PraisonAI Profiling Report",
"=" * 60,
"",
f"Total Time: {summary['total_time_ms']:.2f}ms",
f"Import Time: {summary['import_time_ms']:.2f}ms",
f"Timing Records: {summary['timing_count']}",
f"Import Records: {summary['import_count']}",
f"Flow Steps: {summary['flow_steps']}",
f"Files Accessed: {summary['files_accessed']}",
"",
"By Category:",
]
for cat, time_ms in summary['by_category'].items():
lines.append(f" {cat}: {time_ms:.2f}ms")
lines.extend([
"",
"Slowest Operations:",
])
for name, time_ms in summary['slowest_operations']:
lines.append(f" {name}: {time_ms:.2f}ms")
lines.extend([
"",
"Slowest Imports:",
])
for module, time_ms in summary['slowest_imports']:
lines.append(f" {module}: {time_ms:.2f}ms")
lines.append("=" * 60)
report_text = "\n".join(lines)
if output == "console":
print(report_text)
return report_text
# ============================================================================
# Decorators
# ============================================================================
def profile(func: Optional[Callable] = None, *, category: str = "function"):
"""
Decorator to profile a function.
Usage:
@profile
def my_function():
pass
@profile(category="api")
def api_call():
pass
"""
def decorator(fn: Callable) -> Callable:
@functools.wraps(fn)
def wrapper(*args, **kwargs):
if not Profiler.is_enabled():
return fn(*args, **kwargs)
start = time.time()
try:
return fn(*args, **kwargs)
finally:
duration_ms = (time.time() - start) * 1000
file = fn.__code__.co_filename if hasattr(fn, '__code__') else ""
line = fn.__code__.co_firstlineno if hasattr(fn, '__code__') else 0
Profiler.record_timing(fn.__name__, duration_ms, category, file, line)
Profiler.record_flow(fn.__name__, duration_ms, file, line)
return wrapper
if func is not None:
return decorator(func)
return decorator
def profile_async(func: Optional[Callable] = None, *, category: str = "async"):
"""
Decorator to profile an async function.
"""
def decorator(fn: Callable) -> Callable:
@functools.wraps(fn)
async def wrapper(*args, **kwargs):
if not Profiler.is_enabled():
return await fn(*args, **kwargs)
start = time.time()
try:
return await fn(*args, **kwargs)
finally:
duration_ms = (time.time() - start) * 1000
file = fn.__code__.co_filename if hasattr(fn, '__code__') else ""
line = fn.__code__.co_firstlineno if hasattr(fn, '__code__') else 0
Profiler.record_timing(fn.__name__, duration_ms, category, file, line)
Profiler.record_flow(fn.__name__, duration_ms, file, line)
return wrapper
if func is not None:
return decorator(func)
return decorator
# ============================================================================
# Import Profiling
# ============================================================================
class ImportProfiler:
"""
Context manager to profile imports.
Usage:
with profile_imports() as profiler:
import heavy_module
print(profiler.get_imports())
"""
def __init__(self):
self._original_import = None
self._imports: List[ImportRecord] = []
def __enter__(self):
import builtins
self._original_import = builtins.__import__
def profiled_import(name, globals=None, locals=None, fromlist=(), level=0):
start = time.time()
try:
return self._original_import(name, globals, locals, fromlist, level)
finally:
duration_ms = (time.time() - start) * 1000
if duration_ms > 1: # Only record imports > 1ms
record = ImportRecord(module=name, duration_ms=duration_ms)
self._imports.append(record)
Profiler.record_import(name, duration_ms)
builtins.__import__ = profiled_import
return self
def __exit__(self, exc_type, exc_val, exc_tb):
import builtins
builtins.__import__ = self._original_import
return False
def get_imports(self, min_duration_ms: float = 0) -> List[ImportRecord]:
"""Get recorded imports."""
if min_duration_ms > 0:
return [i for i in self._imports if i.duration_ms >= min_duration_ms]
return self._imports.copy()
def get_slowest(self, n: int = 10) -> List[ImportRecord]:
"""Get N slowest imports."""
return sorted(self._imports, key=lambda x: x.duration_ms, reverse=True)[:n]
def profile_imports():
"""Create an import profiler context manager."""
return ImportProfiler()
# ============================================================================
# Quick Profiling Functions
# ============================================================================
def time_import(module_name: str) -> float:
"""
Time how long it takes to import a module.
Returns duration in milliseconds.
"""
start = time.time()
__import__(module_name)
return (time.time() - start) * 1000
def check_module_available(module_name: str) -> bool:
"""
Check if a module is available without importing it.
Uses importlib.util.find_spec which is fast.
"""
import importlib.util
return importlib.util.find_spec(module_name) is not None
# ============================================================================
# Exports
# ============================================================================
__all__ = [
'Profiler',
'profile',
'profile_async',
'profile_imports',
'ImportProfiler',
'time_import',
'check_module_available',
'TimingRecord',
'ImportRecord',
'FlowRecord',
]
+1 -1
View File
@@ -1 +1 @@
__version__ = "2.3.49"
__version__ = "2.3.54"
@@ -0,0 +1,289 @@
"""
Unit tests for @ mention autocomplete feature.
Tests for:
1. AtMentionContext - Context detection
2. FileSearchService - File search with fuzzy matching
3. AtMentionCompleter - Autocomplete integration
"""
import os
import tempfile
import time
# ============================================================================
# 1. AtMentionContext Tests
# ============================================================================
class TestAtMentionContext:
"""Test @ mention context detection."""
def test_detect_at_start(self):
"""Should detect @ at start of input."""
from praisonai.cli.features.at_mentions import detect_at_mention
result = detect_at_mention("@", cursor_pos=1)
assert result is not None
assert result.is_active
assert result.query == ""
assert result.start_pos == 0
def test_detect_at_with_query(self):
"""Should detect @ with partial query."""
from praisonai.cli.features.at_mentions import detect_at_mention
result = detect_at_mention("@main", cursor_pos=5)
assert result is not None
assert result.is_active
assert result.query == "main"
assert result.start_pos == 0
def test_detect_at_mid_text(self):
"""Should detect @ in middle of text."""
from praisonai.cli.features.at_mentions import detect_at_mention
result = detect_at_mention("read @src/", cursor_pos=10)
assert result is not None
assert result.is_active
assert result.query == "src/"
assert result.start_pos == 5
def test_no_at_symbol(self):
"""Should return None when no @ present."""
from praisonai.cli.features.at_mentions import detect_at_mention
result = detect_at_mention("hello world", cursor_pos=11)
assert result is None
def test_at_after_space_breaks_context(self):
"""Should not detect @ if space after query."""
from praisonai.cli.features.at_mentions import detect_at_mention
result = detect_at_mention("@file.txt ", cursor_pos=10)
assert result is None # Space breaks the @ context
def test_cursor_before_at(self):
"""Should not detect if cursor is before @."""
from praisonai.cli.features.at_mentions import detect_at_mention
result = detect_at_mention("hello @file", cursor_pos=3)
assert result is None
# ============================================================================
# 2. FileSuggestion Tests
# ============================================================================
class TestFileSuggestion:
"""Test FileSuggestion dataclass."""
def test_create_file_suggestion(self):
"""Should create file suggestion."""
from praisonai.cli.features.at_mentions import FileSuggestion
suggestion = FileSuggestion(
path="src/main.py",
file_type="file",
score=100
)
assert suggestion.path == "src/main.py"
assert suggestion.file_type == "file"
assert suggestion.score == 100
def test_directory_suggestion(self):
"""Should create directory suggestion."""
from praisonai.cli.features.at_mentions import FileSuggestion
suggestion = FileSuggestion(
path="src/",
file_type="directory",
score=90
)
assert suggestion.file_type == "directory"
# ============================================================================
# 3. FileSearchService Tests
# ============================================================================
class TestFileSearchService:
"""Test FileSearchService for file discovery."""
def test_service_creation(self):
"""Should create service."""
from praisonai.cli.features.at_mentions import FileSearchService
service = FileSearchService(root_dir=".")
assert service is not None
def test_search_files(self):
"""Should search files in directory."""
from praisonai.cli.features.at_mentions import FileSearchService
with tempfile.TemporaryDirectory() as tmpdir:
# Create test files
open(os.path.join(tmpdir, "main.py"), "w").close()
open(os.path.join(tmpdir, "test.py"), "w").close()
os.makedirs(os.path.join(tmpdir, "src"))
open(os.path.join(tmpdir, "src", "app.py"), "w").close()
service = FileSearchService(root_dir=tmpdir)
results = service.search("main")
assert len(results) >= 1
assert any("main.py" in r.path for r in results)
def test_fuzzy_match(self):
"""Should fuzzy match file names."""
from praisonai.cli.features.at_mentions import FileSearchService
with tempfile.TemporaryDirectory() as tmpdir:
open(os.path.join(tmpdir, "my_long_filename.py"), "w").close()
service = FileSearchService(root_dir=tmpdir)
results = service.search("mlf") # Fuzzy match
assert len(results) >= 1
def test_empty_query_returns_all(self):
"""Should return files when query is empty."""
from praisonai.cli.features.at_mentions import FileSearchService
with tempfile.TemporaryDirectory() as tmpdir:
open(os.path.join(tmpdir, "a.py"), "w").close()
open(os.path.join(tmpdir, "b.py"), "w").close()
service = FileSearchService(root_dir=tmpdir)
results = service.search("")
assert len(results) >= 2
def test_respects_max_results(self):
"""Should limit results."""
from praisonai.cli.features.at_mentions import FileSearchService
with tempfile.TemporaryDirectory() as tmpdir:
for i in range(20):
open(os.path.join(tmpdir, f"file{i}.py"), "w").close()
service = FileSearchService(root_dir=tmpdir)
results = service.search("", max_results=5)
assert len(results) <= 5
def test_cache_works(self):
"""Should cache results."""
from praisonai.cli.features.at_mentions import FileSearchService
with tempfile.TemporaryDirectory() as tmpdir:
open(os.path.join(tmpdir, "test.py"), "w").close()
service = FileSearchService(root_dir=tmpdir, cache_ttl=10)
# First search
start = time.time()
results1 = service.search("test")
first_time = time.time() - start
# Second search (cached)
start = time.time()
results2 = service.search("test")
second_time = time.time() - start
assert len(results1) == len(results2)
# Cache should be faster (or at least not slower)
# ============================================================================
# 4. AtMentionCompleter Tests
# ============================================================================
class TestAtMentionCompleter:
"""Test AtMentionCompleter for prompt_toolkit integration."""
def test_completer_creation(self):
"""Should create completer."""
from praisonai.cli.features.at_mentions import AtMentionCompleter
completer = AtMentionCompleter(root_dir=".")
assert completer is not None
def test_no_completions_without_at(self):
"""Should not complete without @."""
from praisonai.cli.features.at_mentions import AtMentionCompleter
from prompt_toolkit.document import Document
completer = AtMentionCompleter(root_dir=".")
doc = Document("hello world")
completions = list(completer.get_completions(doc, None))
assert len(completions) == 0
def test_completions_with_at(self):
"""Should complete with @."""
from praisonai.cli.features.at_mentions import AtMentionCompleter
from prompt_toolkit.document import Document
with tempfile.TemporaryDirectory() as tmpdir:
open(os.path.join(tmpdir, "main.py"), "w").close()
completer = AtMentionCompleter(root_dir=tmpdir)
doc = Document("@main")
completions = list(completer.get_completions(doc, None))
assert len(completions) >= 1
# ============================================================================
# 5. CombinedCompleter Tests
# ============================================================================
class TestCombinedCompleter:
"""Test CombinedCompleter that handles both / and @."""
def test_slash_commands(self):
"""Should complete slash commands."""
from praisonai.cli.features.at_mentions import CombinedCompleter
from prompt_toolkit.document import Document
completer = CombinedCompleter(
commands=["help", "exit", "queue"],
root_dir="."
)
doc = Document("/he")
completions = list(completer.get_completions(doc, None))
assert len(completions) >= 1
assert any("/help" in c.text for c in completions)
def test_at_mentions(self):
"""Should complete @ mentions."""
from praisonai.cli.features.at_mentions import CombinedCompleter
from prompt_toolkit.document import Document
with tempfile.TemporaryDirectory() as tmpdir:
open(os.path.join(tmpdir, "test.py"), "w").close()
completer = CombinedCompleter(
commands=["help"],
root_dir=tmpdir
)
doc = Document("@test")
completions = list(completer.get_completions(doc, None))
assert len(completions) >= 1
def test_no_completions_for_regular_text(self):
"""Should not complete regular text."""
from praisonai.cli.features.at_mentions import CombinedCompleter
from prompt_toolkit.document import Document
completer = CombinedCompleter(
commands=["help"],
root_dir="."
)
doc = Document("hello world")
completions = list(completer.get_completions(doc, None))
assert len(completions) == 0
@@ -0,0 +1,151 @@
"""
Unit tests for --chat flag (alias for --chat-mode).
Tests verify that:
1. --chat flag works as an alias for --chat-mode
2. --chat flag does not conflict with 'chat' subcommand
3. Both flags set chat_mode=True correctly
4. Backward compatibility with --chat-mode is maintained
"""
import sys
import os
import argparse
# Add parent directory to path for imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))))
class TestChatFlagParsing:
"""Test argument parsing for --chat and --chat-mode flags."""
def test_chat_mode_flag_sets_chat_mode_true(self):
"""--chat-mode should set chat_mode=True."""
parser = argparse.ArgumentParser()
parser.add_argument("--chat-mode", "--chat", action="store_true", dest="chat_mode")
parser.add_argument("command", nargs="?")
args, _ = parser.parse_known_args(["--chat-mode", "test prompt"])
assert args.chat_mode is True
assert args.command == "test prompt"
def test_chat_flag_sets_chat_mode_true(self):
"""--chat should set chat_mode=True (alias for --chat-mode)."""
parser = argparse.ArgumentParser()
parser.add_argument("--chat-mode", "--chat", action="store_true", dest="chat_mode")
parser.add_argument("command", nargs="?")
args, _ = parser.parse_known_args(["--chat", "test prompt"])
assert args.chat_mode is True
assert args.command == "test prompt"
def test_chat_subcommand_does_not_set_chat_mode(self):
"""'chat' as positional command should NOT set chat_mode."""
parser = argparse.ArgumentParser()
parser.add_argument("--chat-mode", "--chat", action="store_true", dest="chat_mode")
parser.add_argument("command", nargs="?")
args, _ = parser.parse_known_args(["chat"])
assert args.chat_mode is False
assert args.command == "chat"
def test_chat_flag_after_prompt(self):
"""--chat after prompt should work."""
parser = argparse.ArgumentParser()
parser.add_argument("--chat-mode", "--chat", action="store_true", dest="chat_mode")
parser.add_argument("command", nargs="?")
args, _ = parser.parse_known_args(["What is 2+2?", "--chat"])
assert args.chat_mode is True
assert args.command == "What is 2+2?"
def test_chat_mode_flag_after_prompt(self):
"""--chat-mode after prompt should work (backward compat)."""
parser = argparse.ArgumentParser()
parser.add_argument("--chat-mode", "--chat", action="store_true", dest="chat_mode")
parser.add_argument("command", nargs="?")
args, _ = parser.parse_known_args(["What is 2+2?", "--chat-mode"])
assert args.chat_mode is True
assert args.command == "What is 2+2?"
def test_no_flag_defaults_to_false(self):
"""Without --chat or --chat-mode, chat_mode should be False."""
parser = argparse.ArgumentParser()
parser.add_argument("--chat-mode", "--chat", action="store_true", dest="chat_mode")
parser.add_argument("command", nargs="?")
args, _ = parser.parse_known_args(["some prompt"])
assert args.chat_mode is False
assert args.command == "some prompt"
class TestChatFlagNoConflict:
"""Test that --chat flag does not conflict with other features."""
def test_chat_flag_with_other_flags(self):
"""--chat should work alongside other flags."""
parser = argparse.ArgumentParser()
parser.add_argument("--chat-mode", "--chat", action="store_true", dest="chat_mode")
parser.add_argument("--verbose", "-v", action="store_true")
parser.add_argument("--model", type=str)
parser.add_argument("command", nargs="?")
args, _ = parser.parse_known_args(["--chat", "--verbose", "--model", "gpt-4o", "test"])
assert args.chat_mode is True
assert args.verbose is True
assert args.model == "gpt-4o"
assert args.command == "test"
def test_chat_subcommand_with_other_flags(self):
"""'chat' subcommand should work with other flags."""
parser = argparse.ArgumentParser()
parser.add_argument("--chat-mode", "--chat", action="store_true", dest="chat_mode")
parser.add_argument("--verbose", "-v", action="store_true")
parser.add_argument("command", nargs="?")
args, _ = parser.parse_known_args(["chat", "--verbose"])
assert args.chat_mode is False # subcommand, not flag
assert args.command == "chat"
assert args.verbose is True
class TestChatFlagIntegration:
"""Integration tests with actual PraisonAI CLI parser."""
def test_praisonai_parser_has_chat_flag(self):
"""PraisonAI CLI should have --chat as alias for --chat-mode."""
# This test will fail until we implement the change
from praisonai.cli.main import PraisonAI
# Create instance and get parser
pai = PraisonAI()
# Mock sys.argv to avoid test environment issues
import sys
original_argv = sys.argv
try:
sys.argv = ["praisonai", "--chat", "test prompt"]
args = pai.parse_args()
assert hasattr(args, 'chat_mode')
# After implementation, this should be True
# For now, it may work due to argparse prefix matching
finally:
sys.argv = original_argv
def test_praisonai_chat_subcommand_still_works(self):
"""'praisonai chat' subcommand should still launch Chainlit.
Note: In test environment, parse_args returns default args to avoid
pytest interference. This test verifies the parser structure instead.
"""
# Test the parser structure directly without going through PraisonAI
# which has test environment detection
parser = argparse.ArgumentParser()
parser.add_argument("--chat-mode", "--chat", action="store_true", dest="chat_mode")
parser.add_argument("command", nargs="?")
# Simulate 'praisonai chat' - should set command='chat', not chat_mode
args, _ = parser.parse_known_args(["chat"])
assert args.command == "chat"
assert args.chat_mode is False # subcommand, not flag
@@ -0,0 +1,661 @@
"""
Unit tests for Message Queue feature.
Tests for:
1. MessageQueue - FIFO queue for messages
2. ProcessingState - State management
3. QueuedMessage - Message with metadata
4. MessageQueueHandler - Main handler class
"""
import time
import threading
# ============================================================================
# 1. MessageQueue Tests - FIFO operations
# ============================================================================
class TestMessageQueue:
"""Test MessageQueue FIFO operations."""
def test_create_empty_queue(self):
"""Queue should start empty."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
assert queue.is_empty
assert queue.count == 0
def test_add_message(self):
"""Should add message to queue."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
queue.add("Hello")
assert not queue.is_empty
assert queue.count == 1
def test_add_multiple_messages(self):
"""Should add multiple messages in order."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
queue.add("First")
queue.add("Second")
queue.add("Third")
assert queue.count == 3
def test_pop_message_fifo(self):
"""Should pop messages in FIFO order."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
queue.add("First")
queue.add("Second")
msg = queue.pop()
assert msg == "First"
assert queue.count == 1
def test_pop_empty_queue(self):
"""Should return None when popping empty queue."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
msg = queue.pop()
assert msg is None
def test_peek_message(self):
"""Should peek without removing."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
queue.add("First")
queue.add("Second")
msg = queue.peek()
assert msg == "First"
assert queue.count == 2 # Not removed
def test_peek_empty_queue(self):
"""Should return None when peeking empty queue."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
msg = queue.peek()
assert msg is None
def test_clear_queue(self):
"""Should clear all messages."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
queue.add("First")
queue.add("Second")
queue.clear()
assert queue.is_empty
assert queue.count == 0
def test_get_all_messages(self):
"""Should get all messages as list."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
queue.add("First")
queue.add("Second")
messages = queue.get_all()
assert messages == ["First", "Second"]
assert queue.count == 2 # Not removed
def test_remove_at_index(self):
"""Should remove message at specific index."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
queue.add("First")
queue.add("Second")
queue.add("Third")
removed = queue.remove_at(1)
assert removed == "Second"
assert queue.count == 2
assert queue.get_all() == ["First", "Third"]
def test_remove_at_invalid_index(self):
"""Should return None for invalid index."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
queue.add("First")
removed = queue.remove_at(5)
assert removed is None
assert queue.count == 1
def test_add_empty_message_ignored(self):
"""Should ignore empty messages."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
queue.add("")
queue.add(" ")
assert queue.is_empty
def test_thread_safety(self):
"""Queue should be thread-safe."""
from praisonai.cli.features.message_queue import MessageQueue
queue = MessageQueue()
results = []
def add_messages():
for i in range(100):
queue.add(f"msg-{i}")
def pop_messages():
for _ in range(100):
msg = queue.pop()
if msg:
results.append(msg)
t1 = threading.Thread(target=add_messages)
t2 = threading.Thread(target=pop_messages)
t1.start()
t2.start()
t1.join()
t2.join()
# All messages should be accounted for
total = len(results) + queue.count
assert total == 100
# ============================================================================
# 2. ProcessingState Tests
# ============================================================================
class TestProcessingState:
"""Test ProcessingState enum and state management."""
def test_state_enum_values(self):
"""Should have correct state values."""
from praisonai.cli.features.message_queue import ProcessingState
assert ProcessingState.IDLE.value == "idle"
assert ProcessingState.PROCESSING.value == "processing"
assert ProcessingState.WAITING_APPROVAL.value == "waiting_approval"
def test_state_manager_initial_state(self):
"""Should start in IDLE state."""
from praisonai.cli.features.message_queue import StateManager, ProcessingState
manager = StateManager()
assert manager.current_state == ProcessingState.IDLE
def test_state_transition(self):
"""Should transition between states."""
from praisonai.cli.features.message_queue import StateManager, ProcessingState
manager = StateManager()
manager.set_state(ProcessingState.PROCESSING)
assert manager.current_state == ProcessingState.PROCESSING
def test_is_idle(self):
"""Should correctly report idle state."""
from praisonai.cli.features.message_queue import StateManager, ProcessingState
manager = StateManager()
assert manager.is_idle
manager.set_state(ProcessingState.PROCESSING)
assert not manager.is_idle
def test_is_processing(self):
"""Should correctly report processing state."""
from praisonai.cli.features.message_queue import StateManager, ProcessingState
manager = StateManager()
assert not manager.is_processing
manager.set_state(ProcessingState.PROCESSING)
assert manager.is_processing
def test_state_callback(self):
"""Should call callback on state change."""
from praisonai.cli.features.message_queue import StateManager, ProcessingState
callback_called = []
def on_state_change(old_state, new_state):
callback_called.append((old_state, new_state))
manager = StateManager(on_state_change=on_state_change)
manager.set_state(ProcessingState.PROCESSING)
assert len(callback_called) == 1
assert callback_called[0] == (ProcessingState.IDLE, ProcessingState.PROCESSING)
# ============================================================================
# 3. QueuedMessage Tests
# ============================================================================
class TestQueuedMessage:
"""Test QueuedMessage dataclass."""
def test_create_queued_message(self):
"""Should create message with content."""
from praisonai.cli.features.message_queue import QueuedMessage
msg = QueuedMessage(content="Hello")
assert msg.content == "Hello"
assert msg.timestamp is not None
def test_message_has_timestamp(self):
"""Should have creation timestamp."""
from praisonai.cli.features.message_queue import QueuedMessage
before = time.time()
msg = QueuedMessage(content="Hello")
after = time.time()
assert before <= msg.timestamp <= after
def test_message_priority(self):
"""Should support priority levels."""
from praisonai.cli.features.message_queue import QueuedMessage, MessagePriority
msg = QueuedMessage(content="Hello", priority=MessagePriority.HIGH)
assert msg.priority == MessagePriority.HIGH
def test_default_priority_normal(self):
"""Should default to NORMAL priority."""
from praisonai.cli.features.message_queue import QueuedMessage, MessagePriority
msg = QueuedMessage(content="Hello")
assert msg.priority == MessagePriority.NORMAL
# ============================================================================
# 4. MessageQueueHandler Tests
# ============================================================================
class TestMessageQueueHandler:
"""Test MessageQueueHandler main class."""
def test_handler_creation(self):
"""Should create handler with default settings."""
from praisonai.cli.features.message_queue import MessageQueueHandler
handler = MessageQueueHandler()
assert handler.queue.is_empty
assert handler.state_manager.is_idle
def test_submit_when_idle(self):
"""Should process immediately when idle."""
from praisonai.cli.features.message_queue import MessageQueueHandler
processed = []
def processor(msg):
processed.append(msg)
return f"Response to: {msg}"
handler = MessageQueueHandler(processor=processor)
handler.submit("Hello")
# Should process immediately
assert "Hello" in processed
def test_queue_when_processing(self):
"""Should queue message when already processing."""
from praisonai.cli.features.message_queue import MessageQueueHandler, ProcessingState
handler = MessageQueueHandler()
handler.state_manager.set_state(ProcessingState.PROCESSING)
handler.submit("Hello")
# Should be queued, not processed
assert handler.queue.count == 1
def test_auto_process_queue_on_idle(self):
"""Should auto-process queue when becoming idle."""
from praisonai.cli.features.message_queue import MessageQueueHandler, ProcessingState
processed = []
def processor(msg):
processed.append(msg)
return f"Response: {msg}"
handler = MessageQueueHandler(processor=processor)
# Simulate processing state
handler.state_manager.set_state(ProcessingState.PROCESSING)
handler.submit("Queued1")
handler.submit("Queued2")
assert handler.queue.count == 2
# Simulate becoming idle - should trigger queue processing
handler.on_processing_complete()
# Queue should be processed
assert "Queued1" in processed or handler.queue.count < 2
def test_get_queue_status(self):
"""Should return queue status."""
from praisonai.cli.features.message_queue import MessageQueueHandler, ProcessingState
handler = MessageQueueHandler()
handler.state_manager.set_state(ProcessingState.PROCESSING)
handler.submit("Hello")
handler.submit("World")
status = handler.get_status()
assert status['queue_count'] == 2
assert status['state'] == 'processing'
def test_clear_queue(self):
"""Should clear queue via handler."""
from praisonai.cli.features.message_queue import MessageQueueHandler, ProcessingState
handler = MessageQueueHandler()
handler.state_manager.set_state(ProcessingState.PROCESSING)
handler.submit("Hello")
handler.clear_queue()
assert handler.queue.is_empty
# ============================================================================
# 5. QueueDisplay Tests
# ============================================================================
class TestQueueDisplay:
"""Test QueueDisplay for visual indicators."""
def test_format_empty_queue(self):
"""Should return empty string for empty queue."""
from praisonai.cli.features.message_queue import QueueDisplay, MessageQueue
queue = MessageQueue()
display = QueueDisplay(queue)
assert display.format_queue() == ""
def test_format_single_message(self):
"""Should format single message."""
from praisonai.cli.features.message_queue import QueueDisplay, MessageQueue
queue = MessageQueue()
queue.add("Hello world")
display = QueueDisplay(queue)
formatted = display.format_queue()
assert "↳" in formatted
assert "Hello world" in formatted
def test_format_multiple_messages(self):
"""Should format multiple messages."""
from praisonai.cli.features.message_queue import QueueDisplay, MessageQueue
queue = MessageQueue()
queue.add("First")
queue.add("Second")
display = QueueDisplay(queue)
formatted = display.format_queue()
assert "First" in formatted
assert "Second" in formatted
def test_truncate_long_message(self):
"""Should truncate long messages."""
from praisonai.cli.features.message_queue import QueueDisplay, MessageQueue
queue = MessageQueue()
queue.add("A" * 100)
display = QueueDisplay(queue, max_message_length=20)
formatted = display.format_queue()
assert "..." in formatted
def test_format_status_idle(self):
"""Should show idle status."""
from praisonai.cli.features.message_queue import QueueDisplay, MessageQueue, StateManager
queue = MessageQueue()
state = StateManager()
display = QueueDisplay(queue, state_manager=state)
status = display.format_status()
assert "idle" in status.lower() or status == ""
def test_format_status_processing(self):
"""Should show processing status."""
from praisonai.cli.features.message_queue import (
QueueDisplay, MessageQueue, StateManager, ProcessingState
)
queue = MessageQueue()
state = StateManager()
state.set_state(ProcessingState.PROCESSING)
display = QueueDisplay(queue, state_manager=state)
status = display.format_status()
assert "processing" in status.lower() or "⏳" in status
def test_format_queue_count(self):
"""Should show queue count."""
from praisonai.cli.features.message_queue import QueueDisplay, MessageQueue, ProcessingState, StateManager
queue = MessageQueue()
queue.add("First")
queue.add("Second")
state = StateManager()
state.set_state(ProcessingState.PROCESSING)
display = QueueDisplay(queue, state_manager=state)
count_str = display.format_queue_count()
assert "2" in count_str
# ============================================================================
# 6. Integration Tests
# ============================================================================
class TestMessageQueueIntegration:
"""Integration tests for message queue system."""
def test_full_workflow(self):
"""Test complete queue workflow."""
from praisonai.cli.features.message_queue import (
MessageQueueHandler, ProcessingState
)
results = []
def processor(msg):
results.append(msg)
return f"Processed: {msg}"
handler = MessageQueueHandler(processor=processor)
# 1. Submit when idle - processes immediately
handler.submit("First")
assert "First" in results
# 2. Simulate processing
handler.state_manager.set_state(ProcessingState.PROCESSING)
# 3. Queue messages while processing
handler.submit("Second")
handler.submit("Third")
assert handler.queue.count == 2
# 4. Complete processing - should auto-process queue
handler.on_processing_complete()
# Queue should be processed
assert handler.queue.count < 2 or "Second" in results
def test_queue_with_display(self):
"""Test queue with display formatting."""
from praisonai.cli.features.message_queue import (
MessageQueueHandler, QueueDisplay, ProcessingState
)
handler = MessageQueueHandler()
handler.state_manager.set_state(ProcessingState.PROCESSING)
handler.submit("Task 1")
handler.submit("Task 2")
display = QueueDisplay(handler.queue, state_manager=handler.state_manager)
# Should show processing status
status = display.format_status()
assert "processing" in status.lower() or "⏳" in status
# Should show queued messages
queue_str = display.format_queue()
assert "Task 1" in queue_str
assert "Task 2" in queue_str
# ============================================================================
# 9. AsyncProcessor Tests - Background processing
# ============================================================================
class TestAsyncProcessor:
"""Test AsyncProcessor for background agent execution."""
def test_processor_creation(self):
"""Should create processor with callback."""
from praisonai.cli.features.message_queue import AsyncProcessor
results = []
def on_complete(result):
results.append(result)
processor = AsyncProcessor(on_complete=on_complete)
assert processor is not None
def test_start_processing(self):
"""Should start processing in background thread."""
from praisonai.cli.features.message_queue import AsyncProcessor
import time
results = []
def work_fn():
time.sleep(0.05)
return "done"
def on_complete(result):
results.append(result)
processor = AsyncProcessor(on_complete=on_complete)
processor.start(work_fn)
# Should be processing
assert processor.is_running
# Wait for completion
time.sleep(0.15)
assert not processor.is_running
assert "done" in results
def test_is_running_property(self):
"""Should track running state."""
from praisonai.cli.features.message_queue import AsyncProcessor
import time
def slow_work():
time.sleep(0.1)
return "result"
processor = AsyncProcessor()
assert not processor.is_running
processor.start(slow_work)
assert processor.is_running
time.sleep(0.15)
assert not processor.is_running
def test_on_status_callback(self):
"""Should call status callback during processing."""
from praisonai.cli.features.message_queue import AsyncProcessor
import time
statuses = []
def on_status(status):
statuses.append(status)
def work_with_status(status_callback):
status_callback("Starting...")
time.sleep(0.02)
status_callback("Working...")
return "done"
processor = AsyncProcessor(on_status=on_status)
processor.start(lambda: work_with_status(on_status))
time.sleep(0.1)
assert len(statuses) >= 1
# ============================================================================
# 10. LiveStatusDisplay Tests - Real-time status
# ============================================================================
class TestLiveStatusDisplay:
"""Test LiveStatusDisplay for real-time tool/command status."""
def test_display_creation(self):
"""Should create display."""
from praisonai.cli.features.message_queue import LiveStatusDisplay
display = LiveStatusDisplay()
assert display is not None
def test_update_status(self):
"""Should update current status."""
from praisonai.cli.features.message_queue import LiveStatusDisplay
display = LiveStatusDisplay()
display.update_status("Processing...")
assert display.current_status == "Processing..."
def test_add_tool_call(self):
"""Should track tool calls."""
from praisonai.cli.features.message_queue import LiveStatusDisplay
display = LiveStatusDisplay()
display.add_tool_call("read_file", {"path": "test.py"})
assert len(display.tool_calls) == 1
assert display.tool_calls[0]['name'] == "read_file"
def test_add_command_execution(self):
"""Should track command executions."""
from praisonai.cli.features.message_queue import LiveStatusDisplay
display = LiveStatusDisplay()
display.add_command("ls -la")
assert len(display.commands) == 1
assert display.commands[0] == "ls -la"
def test_format_live_status(self):
"""Should format live status for display."""
from praisonai.cli.features.message_queue import LiveStatusDisplay
display = LiveStatusDisplay()
display.update_status("Thinking...")
display.add_tool_call("read_file", {"path": "main.py"})
formatted = display.format()
assert "Thinking" in formatted or "read_file" in formatted
def test_clear_status(self):
"""Should clear all status."""
from praisonai.cli.features.message_queue import LiveStatusDisplay
display = LiveStatusDisplay()
display.update_status("Working")
display.add_tool_call("test", {})
display.add_command("echo hi")
display.clear()
assert display.current_status == ""
assert len(display.tool_calls) == 0
assert len(display.commands) == 0
# ============================================================================
# 11. NonBlockingInput Tests - Async input handling
# ============================================================================
class TestNonBlockingInput:
"""Test NonBlockingInput for async user input."""
def test_input_handler_creation(self):
"""Should create input handler."""
from praisonai.cli.features.message_queue import NonBlockingInput
handler = NonBlockingInput()
assert handler is not None
def test_submit_input(self):
"""Should accept input while processing."""
from praisonai.cli.features.message_queue import NonBlockingInput
handler = NonBlockingInput()
handler.submit("new message")
assert handler.has_pending
assert handler.pop() == "new message"
def test_multiple_inputs(self):
"""Should queue multiple inputs."""
from praisonai.cli.features.message_queue import NonBlockingInput
handler = NonBlockingInput()
handler.submit("first")
handler.submit("second")
assert handler.pending_count == 2
assert handler.pop() == "first"
assert handler.pop() == "second"
+259
View File
@@ -0,0 +1,259 @@
"""
Unit tests for PraisonAI Profiler module.
"""
import time
class TestProfiler:
"""Test Profiler class."""
def test_profiler_singleton(self):
"""Should return same instance."""
from praisonai.profiler import Profiler
p1 = Profiler()
p2 = Profiler()
assert p1 is p2
def test_enable_disable(self):
"""Should enable and disable profiling."""
from praisonai.profiler import Profiler
Profiler.disable()
assert not Profiler.is_enabled()
Profiler.enable()
assert Profiler.is_enabled()
Profiler.disable()
def test_record_timing(self):
"""Should record timing."""
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
Profiler.record_timing("test_op", 100.5, "test")
timings = Profiler.get_timings()
assert len(timings) >= 1
assert timings[-1].name == "test_op"
assert timings[-1].duration_ms == 100.5
Profiler.disable()
Profiler.clear()
def test_record_import(self):
"""Should record import timing."""
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
Profiler.record_import("test_module", 50.0)
imports = Profiler.get_imports()
assert len(imports) >= 1
assert imports[-1].module == "test_module"
Profiler.disable()
Profiler.clear()
def test_block_context_manager(self):
"""Should profile a block of code."""
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
with Profiler.block("test_block"):
time.sleep(0.01)
timings = Profiler.get_timings(category="block")
assert len(timings) >= 1
assert timings[-1].name == "test_block"
assert timings[-1].duration_ms >= 10
Profiler.disable()
Profiler.clear()
def test_get_summary(self):
"""Should return summary."""
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
Profiler.record_timing("op1", 100, "function")
Profiler.record_timing("op2", 200, "function")
Profiler.record_import("mod1", 50)
summary = Profiler.get_summary()
assert summary['total_time_ms'] == 300
assert summary['import_time_ms'] == 50
assert summary['timing_count'] == 2
assert summary['import_count'] == 1
Profiler.disable()
Profiler.clear()
class TestProfileDecorator:
"""Test @profile decorator."""
def test_profile_function(self):
"""Should profile a function."""
from praisonai.profiler import Profiler, profile
Profiler.enable()
Profiler.clear()
@profile
def test_func():
time.sleep(0.01)
return "result"
result = test_func()
assert result == "result"
timings = Profiler.get_timings(category="function")
assert len(timings) >= 1
assert timings[-1].name == "test_func"
assert timings[-1].duration_ms >= 10
Profiler.disable()
Profiler.clear()
def test_profile_with_category(self):
"""Should profile with custom category."""
from praisonai.profiler import Profiler, profile
Profiler.enable()
Profiler.clear()
@profile(category="api")
def api_call():
return "response"
api_call()
timings = Profiler.get_timings(category="api")
assert len(timings) >= 1
assert timings[-1].name == "api_call"
Profiler.disable()
Profiler.clear()
def test_profile_disabled(self):
"""Should not record when disabled."""
from praisonai.profiler import Profiler, profile
Profiler.disable()
Profiler.clear()
@profile
def test_func():
return "result"
test_func()
timings = Profiler.get_timings()
assert len(timings) == 0
class TestImportProfiler:
"""Test ImportProfiler."""
def test_profile_imports(self):
"""Should profile imports."""
from praisonai.profiler import profile_imports
with profile_imports() as profiler:
import json
import re
imports = profiler.get_imports()
# May or may not have imports depending on cache
assert isinstance(imports, list)
def test_get_slowest(self):
"""Should get slowest imports."""
from praisonai.profiler import profile_imports
with profile_imports() as profiler:
import collections
slowest = profiler.get_slowest(5)
assert isinstance(slowest, list)
assert len(slowest) <= 5
class TestUtilityFunctions:
"""Test utility functions."""
def test_check_module_available(self):
"""Should check module availability."""
from praisonai.profiler import check_module_available
assert check_module_available("os") is True
assert check_module_available("nonexistent_module_xyz") is False
def test_time_import(self):
"""Should time import."""
from praisonai.profiler import time_import
duration = time_import("json")
assert isinstance(duration, float)
assert duration >= 0
class TestDataClasses:
"""Test data classes."""
def test_timing_record(self):
"""Should create TimingRecord."""
from praisonai.profiler import TimingRecord
record = TimingRecord(
name="test",
duration_ms=100.0,
category="function",
file="test.py",
line=10
)
assert record.name == "test"
assert record.duration_ms == 100.0
assert record.category == "function"
def test_import_record(self):
"""Should create ImportRecord."""
from praisonai.profiler import ImportRecord
record = ImportRecord(
module="test_module",
duration_ms=50.0,
parent="parent_module"
)
assert record.module == "test_module"
assert record.duration_ms == 50.0
def test_flow_record(self):
"""Should create FlowRecord."""
from praisonai.profiler import FlowRecord
record = FlowRecord(
step=1,
name="step1",
file="test.py",
line=20,
duration_ms=30.0
)
assert record.step == 1
assert record.name == "step1"