Compare commits

...
1 Commits
Author SHA1 Message Date
Mervin Praison 911483c92c Release v3.0.0 2025-12-31 18:55:59 +00:00
32 changed files with 4966 additions and 27 deletions

No files matched your search

+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.9.2" \
"praisonai>=3.0.0" \
"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.9.2" \
"praisonai>=3.0.0" \
"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.9.2" \
"praisonai>=3.0.0" \
"praisonai[ui]" \
"praisonai[crewai]"
@@ -0,0 +1,147 @@
"""
Interactive Default Tools Example
This example demonstrates how PraisonAI's interactive modes (TUI and prompt)
include ACP and LSP tools by default.
Tool Groups:
- ACP: acp_create_file, acp_edit_file, acp_delete_file, acp_execute_command
- LSP: lsp_list_symbols, lsp_find_definition, lsp_find_references, lsp_get_diagnostics
- Basic: read_file, write_file, list_files, execute_command, internet_search
Usage:
python interactive_tools_example.py
"""
import os
import sys
# Add paths for development
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../../src/praisonai'))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../../src/praisonai-agents'))
def example_get_default_tools():
"""Example: Get all default interactive tools."""
from praisonai.cli.features.interactive_tools import (
get_interactive_tools,
ToolConfig,
TOOL_GROUPS,
)
print("=== Default Interactive Tools ===\n")
# Show tool groups
print("Tool Groups:")
for group, tools in TOOL_GROUPS.items():
if group != "interactive":
print(f" {group}: {tools}")
print()
# Get all tools
config = ToolConfig(workspace=os.getcwd())
tools = get_interactive_tools(config=config)
print(f"Total tools loaded: {len(tools)}")
print(f"Tool names: {[t.__name__ for t in tools]}")
print()
return tools
def example_disable_acp():
"""Example: Disable ACP tools."""
from praisonai.cli.features.interactive_tools import (
get_interactive_tools,
ToolConfig,
)
print("=== Disable ACP Tools (--no-acp) ===\n")
config = ToolConfig(workspace=os.getcwd(), enable_acp=False)
tools = get_interactive_tools(config=config, disable=["acp"])
acp_tools = [t for t in tools if t.__name__.startswith("acp_")]
print(f"ACP tools after --no-acp: {len(acp_tools)}")
print(f"Remaining tools: {[t.__name__ for t in tools]}")
print()
return tools
def example_disable_lsp():
"""Example: Disable LSP tools."""
from praisonai.cli.features.interactive_tools import (
get_interactive_tools,
ToolConfig,
)
print("=== Disable LSP Tools (--no-lsp) ===\n")
config = ToolConfig(workspace=os.getcwd(), enable_lsp=False)
tools = get_interactive_tools(config=config, disable=["lsp"])
lsp_tools = [t for t in tools if t.__name__.startswith("lsp_")]
print(f"LSP tools after --no-lsp: {len(lsp_tools)}")
print(f"Remaining tools: {[t.__name__ for t in tools]}")
print()
return tools
def example_agent_with_default_tools():
"""Example: Create an Agent with default interactive tools."""
from praisonai.cli.features.interactive_tools import get_interactive_tools, ToolConfig
from praisonaiagents import Agent
print("=== Agent with Default Interactive Tools ===\n")
# Get default tools
config = ToolConfig(workspace=os.getcwd())
tools = get_interactive_tools(config=config)
# Create agent with tools
agent = Agent(
name="InteractiveAgent",
instructions="You are a helpful coding assistant with file and code analysis capabilities.",
tools=tools,
verbose=False
)
print(f"Agent created with {len(tools)} tools")
print(f"Tools: {[t.__name__ for t in tools]}")
print()
# Simple test - list files
print("Testing agent with list_files tool...")
result = agent.start("List the Python files in the current directory using list_files tool.")
print(f"Result: {result[:200]}..." if len(str(result)) > 200 else f"Result: {result}")
print()
return agent
if __name__ == "__main__":
print("=" * 60)
print("PraisonAI Interactive Default Tools Example")
print("=" * 60)
print()
# Example 1: Get default tools
example_get_default_tools()
# Example 2: Disable ACP
example_disable_acp()
# Example 3: Disable LSP
example_disable_lsp()
# Example 4: Agent with default tools (requires API key)
if os.environ.get("OPENAI_API_KEY"):
example_agent_with_default_tools()
else:
print("=== Agent Example Skipped (no OPENAI_API_KEY) ===")
print("=" * 60)
print("Examples Complete")
print("=" * 60)
+83
View File
@@ -0,0 +1,83 @@
# PraisonAI Profiling Examples
This directory contains examples for profiling PraisonAI agent performance.
## Prerequisites
```bash
pip install praisonai
export OPENAI_API_KEY=your_key_here
```
## Examples
### 1. Basic Profiling (`basic_profiling.py`)
Demonstrates programmatic profiling of a single query:
```bash
python basic_profiling.py
```
### 2. Suite Profiling (`suite_profiling.py`)
Runs a comprehensive profiling suite with multiple scenarios:
```bash
python suite_profiling.py
```
### 3. Optimization Demo (`optimization_example.py`)
Demonstrates Tier 0/1/2 performance optimizations:
```bash
python optimization_example.py
```
## CLI Commands
You can also profile directly from the command line:
```bash
# Profile a query
praisonai profile query "What is 2+2?"
# Profile with file grouping
praisonai profile query "Hello" --show-files --limit 20
# Profile imports
praisonai profile imports
# Profile startup time
praisonai profile startup
# Run comprehensive suite
praisonai profile suite --quick
# Create performance baseline
praisonai profile snapshot --baseline
# Compare against baseline
praisonai profile snapshot current --compare
# Show optimization status
praisonai profile optimize --show
```
## Output
- **Text output**: Human-readable timing breakdown
- **JSON output**: Machine-readable for CI/CD (`--format json`)
- **Artifacts**: Binary cProfile data and reports (`--save ./results`)
- **Snapshots**: Performance baselines for regression detection
## Performance Optimizations
Enable opt-in optimizations via environment variables:
```bash
export PRAISONAI_LITE_MODE=1
export PRAISONAI_SKIP_TYPE_VALIDATION=1
export PRAISONAI_MINIMAL_IMPORTS=1
```
@@ -0,0 +1,66 @@
"""
Basic Profiling Example for PraisonAI
This example demonstrates how to use the profiler programmatically
to measure agent performance.
Run with:
python basic_profiling.py
"""
import os
# Ensure API key is set
if not os.environ.get("OPENAI_API_KEY"):
print("Please set OPENAI_API_KEY environment variable")
exit(1)
from praisonai.cli.features.profiler import (
ProfilerConfig,
QueryProfiler,
format_profile_report,
)
def main():
"""Run a profiled query and display results."""
# Configure profiler
config = ProfilerConfig(
deep=False, # Set True for detailed call tracing (slower)
limit=15, # Show top 15 functions
show_files=True, # Group by file
first_token=False, # Track time to first token (for streaming)
)
# Create profiler
profiler = QueryProfiler(config)
# Run profiled query
print("🔬 Running profiled query...")
result = profiler.profile_query(
prompt="What is 2+2?",
model=None, # Use default model
stream=False,
)
# Print formatted report
print("\n" + format_profile_report(result, config))
# Access timing data programmatically
print("\n📊 Timing Summary:")
print(f" Imports: {result.timing.imports_ms:>10.2f} ms")
print(f" Agent Construct:{result.timing.agent_construction_ms:>10.2f} ms")
print(f" Total Run: {result.timing.total_run_ms:>10.2f} ms")
# Show top 5 functions
print("\n🔥 Top 5 Functions by Cumulative Time:")
for i, func in enumerate(result.top_functions[:5], 1):
print(f" {i}. {func.name}: {func.cumtime_ms:.2f}ms")
# Show response preview
print(f"\n💬 Response: {result.response[:100]}...")
if __name__ == "__main__":
main()
@@ -0,0 +1,223 @@
"""
Performance Optimization Example for PraisonAI
This example demonstrates how to use the Tier 0/1/2 optimization features
programmatically.
Run with:
python optimization_example.py
"""
import os
# Ensure API key is set
if not os.environ.get("OPENAI_API_KEY"):
print("Please set OPENAI_API_KEY environment variable")
exit(1)
def demo_provider_cache():
"""Demonstrate provider caching (Tier 0)."""
print("\n" + "=" * 60)
print("Tier 0: Provider Cache Demo")
print("=" * 60)
from praisonai.cli.features.profiler import (
get_provider_cache,
resolve_model_provider,
)
cache = get_provider_cache()
cache.invalidate() # Start fresh
# Resolve some models
models = ['gpt-4o', 'claude-3-opus', 'gemini-pro', 'gpt-4o-mini']
print("\nResolving models:")
for model in models:
provider, name = resolve_model_provider(model)
print(f" {model} -> {provider}/{name}")
# Check cache stats
stats = cache.get_stats()
print("\nCache Stats:")
print(f" Hits: {stats['hits']}")
print(f" Misses: {stats['misses']}")
print(f" Size: {stats['size']}")
def demo_lazy_importer():
"""Demonstrate lazy imports (Tier 0)."""
print("\n" + "=" * 60)
print("Tier 0: Lazy Importer Demo")
print("=" * 60)
from praisonai.cli.features.profiler import LazyImporter
import time
LazyImporter.clear()
# Time lazy import
start = time.perf_counter()
json_module = LazyImporter.get('json')
first_import = (time.perf_counter() - start) * 1000
# Second import (cached)
start = time.perf_counter()
json_module2 = LazyImporter.get('json')
cached_import = (time.perf_counter() - start) * 1000
print(f"\nFirst import: {first_import:.3f}ms")
print(f"Cached import: {cached_import:.3f}ms")
print(f"Same object: {json_module is json_module2}")
def demo_client_pool():
"""Demonstrate client pooling (Tier 1)."""
print("\n" + "=" * 60)
print("Tier 1: Client Pool Demo")
print("=" * 60)
from praisonai.cli.features.profiler import ClientPool
pool = ClientPool(max_size=3)
# Create some mock clients
def create_client(name):
return {'name': name, 'created': True}
# Get or create clients
client1 = pool.get_or_create('openai', lambda: create_client('openai'))
_ = pool.get_or_create('anthropic', lambda: create_client('anthropic'))
# Reuse existing client
client1_again = pool.get_or_create('openai', lambda: create_client('openai_new'))
print(f"\nPool size: {pool.size()}")
print(f"Client reused: {client1 is client1_again}")
print(f"Client1 name: {client1['name']}")
def demo_prewarm():
"""Demonstrate pre-warming (Tier 1)."""
print("\n" + "=" * 60)
print("Tier 1: Pre-warming Demo")
print("=" * 60)
from praisonai.cli.features.profiler import PrewarmManager
import time
PrewarmManager.reset()
print(f"\nPre-warming enabled: {PrewarmManager.is_enabled()}")
# Enable and prewarm
PrewarmManager.enable()
print(f"After enable: {PrewarmManager.is_enabled()}")
# Prewarm OpenAI (runs in background)
PrewarmManager.prewarm_provider('openai')
print("Pre-warming OpenAI in background...")
# Wait a bit for background thread
time.sleep(0.5)
print(f"OpenAI prewarmed: {PrewarmManager.is_prewarmed('openai')}")
PrewarmManager.reset()
def demo_lite_mode():
"""Demonstrate lite mode (Tier 2)."""
print("\n" + "=" * 60)
print("Tier 2: Lite Mode Demo")
print("=" * 60)
from praisonai.cli.features.profiler import (
LiteModeConfig,
is_lite_mode,
)
# Default config
config = LiteModeConfig()
print(f"\nDefault lite mode: {config.enabled}")
# Config from env
print("\nTo enable lite mode, set environment variables:")
print(" export PRAISONAI_LITE_MODE=1")
print(" export PRAISONAI_SKIP_TYPE_VALIDATION=1")
print(f"\nCurrent is_lite_mode(): {is_lite_mode()}")
def demo_perf_snapshot():
"""Demonstrate performance snapshots (Tier 2)."""
print("\n" + "=" * 60)
print("Tier 2: Performance Snapshot Demo")
print("=" * 60)
from praisonai.cli.features.profiler import (
PerfSnapshot,
PerfComparison,
PerfSnapshotManager,
)
import tempfile
from pathlib import Path
# Create manager with temp directory
with tempfile.TemporaryDirectory() as tmpdir:
manager = PerfSnapshotManager(snapshot_dir=Path(tmpdir))
# Create baseline
baseline = PerfSnapshot(
timestamp='2025-01-01T00:00:00Z',
name='baseline',
startup_cold_ms=100.0,
startup_warm_ms=80.0,
import_time_ms=500.0,
query_time_ms=2000.0,
first_token_ms=300.0,
)
manager.save_baseline(baseline)
print("\nBaseline saved")
# Create current snapshot (simulated improvement)
current = PerfSnapshot(
timestamp='2025-01-02T00:00:00Z',
name='current',
startup_cold_ms=95.0, # 5% faster
startup_warm_ms=78.0,
import_time_ms=480.0, # 4% faster
query_time_ms=1900.0, # 5% faster
first_token_ms=280.0,
)
# Compare
comparison = PerfComparison(baseline=baseline, current=current)
print(f"\nStartup improvement: {-comparison.startup_cold_diff_pct:.1f}%")
print(f"Import improvement: {-comparison.import_time_diff_pct:.1f}%")
print(f"Query improvement: {-comparison.query_time_diff_pct:.1f}%")
print(f"Is regression: {comparison.is_regression()}")
def main():
"""Run all demos."""
print("=" * 60)
print("PraisonAI Performance Optimization Demo")
print("=" * 60)
demo_provider_cache()
demo_lazy_importer()
demo_client_pool()
demo_prewarm()
demo_lite_mode()
demo_perf_snapshot()
print("\n" + "=" * 60)
print("All demos complete!")
print("=" * 60)
if __name__ == "__main__":
main()
@@ -0,0 +1,227 @@
{
"timestamp": "2025-12-31T18:45:38.187539Z",
"metadata": {
"python_version": "3.12.11",
"platform": "macOS-15.7.4-arm64-arm-64bit",
"praisonai_version": "2.9.2",
"timestamp": "2025-12-31T18:45:38.226558Z"
},
"startup": {
"cold_ms": 80.9305000002496,
"warm_ms": 55.266541996388696
},
"import_analysis": [
{
"module": "praisonaiagents",
"self_us": 2419,
"cumulative_us": 1645390,
"self_ms": 2.419,
"cumulative_ms": 1645.39
},
{
"module": "praisonaiagents.workflows",
"self_us": 670,
"cumulative_us": 1375919,
"self_ms": 0.67,
"cumulative_ms": 1375.919
},
{
"module": "praisonaiagents.workflows.yaml_parser",
"self_us": 775,
"cumulative_us": 1369435,
"self_ms": 0.775,
"cumulative_ms": 1369.435
},
{
"module": "praisonaiagents.agent.agent",
"self_us": 19,
"cumulative_us": 1349174,
"self_ms": 0.019,
"cumulative_ms": 1349.174
},
{
"module": "praisonaiagents.agent",
"self_us": 537,
"cumulative_us": 1349155,
"self_ms": 0.537,
"cumulative_ms": 1349.155
},
{
"module": "praisonaiagents.agent.agent",
"self_us": 1453,
"cumulative_us": 1285299,
"self_ms": 1.453,
"cumulative_ms": 1285.299
},
{
"module": "praisonaiagents.llm.openai_client",
"self_us": 7887,
"cumulative_us": 1123578,
"self_ms": 7.887,
"cumulative_ms": 1123.578
},
{
"module": "openai",
"self_us": 1083,
"cumulative_us": 1115692,
"self_ms": 1.083,
"cumulative_ms": 1115.692
},
{
"module": "openai.types",
"self_us": 2606,
"cumulative_us": 957990,
"self_ms": 2.606,
"cumulative_ms": 957.99
},
{
"module": "openai.types.batch",
"self_us": 1787,
"cumulative_us": 485667,
"self_ms": 1.787,
"cumulative_ms": 485.667
},
{
"module": "openai._models",
"self_us": 13423,
"cumulative_us": 450798,
"self_ms": 13.423,
"cumulative_ms": 450.798
},
{
"module": "openai.types.eval_create_params",
"self_us": 644,
"cumulative_us": 393275,
"self_ms": 0.644,
"cumulative_ms": 393.275
},
{
"module": "openai.types.graders.grader_inputs_param",
"self_us": 17,
"cumulative_us": 359515,
"self_ms": 0.017,
"cumulative_ms": 359.515
},
{
"module": "openai.types.graders",
"self_us": 392,
"cumulative_us": 359498,
"self_ms": 0.392,
"cumulative_ms": 359.498
},
{
"module": "openai.types.graders.multi_grader",
"self_us": 3847,
"cumulative_us": 351251,
"self_ms": 3.847,
"cumulative_ms": 351.251
},
{
"module": "openai.types.graders.label_model_grader",
"self_us": 5450,
"cumulative_us": 337291,
"self_ms": 5.45,
"cumulative_ms": 337.291
},
{
"module": "openai.types.graders.grader_inputs",
"self_us": 545,
"cumulative_us": 331841,
"self_ms": 0.545,
"cumulative_ms": 331.841
},
{
"module": "openai.types.responses.response_input_text",
"self_us": 21,
"cumulative_us": 331297,
"self_ms": 0.021,
"cumulative_ms": 331.297
},
{
"module": "openai.types.responses",
"self_us": 1301,
"cumulative_us": 331277,
"self_ms": 1.301,
"cumulative_ms": 331.277
},
{
"module": "rich.logging",
"self_us": 1066,
"cumulative_us": 238256,
"self_ms": 1.066,
"cumulative_ms": 238.256
}
],
"scenarios": [
{
"name": "simple_query",
"config": {
"prompt": "What is 2+2?",
"model": "default",
"stream": false,
"iterations": 2
},
"import_time_stats": {
"mean": 0.008833496394800022,
"min": 0.005916997906751931,
"max": 0.011749994882848114,
"stdev": 0.0041245517164382375,
"median": 0.008833496394800022
},
"total_time_stats": {
"mean": 1727.4090415012324,
"min": 1367.2264170018025,
"max": 2087.5916660006624,
"stdev": 509.3751524982296,
"median": 1727.4090415012324
},
"first_token_stats": null,
"raw_import_times_ms": [
0.011749994882848114,
0.005916997906751931
],
"raw_total_times_ms": [
1367.2264170018025,
2087.5916660006624
]
},
{
"name": "simple_stream",
"config": {
"prompt": "What is 2+2?",
"model": "default",
"stream": true,
"iterations": 2
},
"import_time_stats": {
"mean": 0.008792001608526334,
"min": 0.006375004886649549,
"max": 0.01120899833040312,
"stdev": 0.0034181495442894613,
"median": 0.008792001608526334
},
"total_time_stats": {
"mean": 2321.0982290001994,
"min": 1697.1089580038097,
"max": 2945.087499996589,
"stdev": 882.4540898183949,
"median": 2321.0982290001994
},
"first_token_stats": {
"mean": 2312.8511880022415,
"min": 1681.2440420035273,
"max": 2944.458334000956,
"stdev": 893.2273919631452,
"median": 2312.8511880022415
},
"raw_import_times_ms": [
0.01120899833040312,
0.006375004886649549
],
"raw_total_times_ms": [
2945.087499996589,
1697.1089580038097
]
}
]
}
@@ -0,0 +1,80 @@
"""
Profile Suite Example for PraisonAI
This example demonstrates how to run a comprehensive profiling suite
programmatically with multiple scenarios.
Run with:
python suite_profiling.py
"""
import os
# Ensure API key is set
if not os.environ.get("OPENAI_API_KEY"):
print("Please set OPENAI_API_KEY environment variable")
exit(1)
from praisonai.cli.features.profiler import (
ScenarioConfig,
run_profile_suite,
)
def main():
"""Run a profiling suite with custom scenarios."""
# Define custom scenarios
scenarios = [
ScenarioConfig(
name="simple_query",
prompt="What is 2+2?",
stream=False,
iterations=2,
warmup=1,
),
ScenarioConfig(
name="simple_stream",
prompt="What is 2+2?",
stream=True,
iterations=2,
warmup=1,
),
]
# Run suite with custom output directory
print("🔬 Running Profile Suite...")
result = run_profile_suite(
output_dir="./profile_results",
scenarios=scenarios,
verbose=True,
)
# Print summary
print("\n" + "=" * 60)
print("Profile Suite Results")
print("=" * 60)
print("\n📊 Startup Times:")
print(f" Cold Start: {result.startup_cold_ms:.2f}ms")
print(f" Warm Start: {result.startup_warm_ms:.2f}ms")
if result.import_analysis:
print("\n📦 Top Import:")
print(f" {result.import_analysis[0]['module']}: {result.import_analysis[0]['cumulative_ms']:.2f}ms")
print("\n🎯 Scenario Results:")
for scenario in result.scenarios:
stats = scenario.get_stats(scenario.total_times)
print(f" {scenario.name}:")
print(f" Mean: {stats['mean']:.2f}ms")
print(f" Min: {stats['min']:.2f}ms")
print(f" Max: {stats['max']:.2f}ms")
print("\n✅ Results saved to: ./profile_results/")
print(" - suite_results.json (machine-readable)")
print(" - suite_report.txt (human-readable)")
if __name__ == "__main__":
main()
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.5.3"
version = "0.6.0"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
+1 -1
View File
@@ -2422,7 +2422,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.5.3"
version = "0.6.0"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.9.2.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.9.2.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.0.0.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.0.0.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+2
View File
@@ -176,6 +176,7 @@ def register_commands():
from .commands.serve import app as serve_app
from .commands.schedule import app as schedule_app
from .commands.run import app as run_app
from .commands.profile import app as profile_app
# Import TUI and queue commands
from .features.tui.debug import create_debug_app as create_tui_debug_app
@@ -197,6 +198,7 @@ def register_commands():
app.add_typer(serve_app, name="serve", help="API server management")
app.add_typer(schedule_app, name="schedule", help="Scheduler management")
app.add_typer(run_app, name="run", help="Run agents")
app.add_typer(profile_app, name="profile", help="Performance profiling and diagnostics")
# Register TUI and queue commands
tui_app = create_tui_debug_app()
@@ -0,0 +1,457 @@
"""
Profile command group for PraisonAI CLI.
Provides detailed cProfile-based profiling for query execution.
"""
import sys
from typing import Optional
import typer
app = typer.Typer(help="Performance profiling and diagnostics")
@app.command("query")
def profile_query(
prompt: str = typer.Argument(..., help="Prompt to profile"),
model: Optional[str] = typer.Option(None, "--model", "-m", help="Model to use"),
stream: bool = typer.Option(False, "--stream/--no-stream", help="Use streaming mode"),
deep: bool = typer.Option(False, "--deep", help="Enable deep call tracing (higher overhead)"),
limit: int = typer.Option(30, "--limit", "-n", help="Top N functions to show"),
sort: str = typer.Option("cumulative", "--sort", "-s", help="Sort by: cumulative or tottime"),
show_files: bool = typer.Option(False, "--show-files", help="Group timing by file/module"),
show_callers: bool = typer.Option(False, "--show-callers", help="Show caller functions"),
show_callees: bool = typer.Option(False, "--show-callees", help="Show callee functions"),
importtime: bool = typer.Option(False, "--importtime", help="Show module import times"),
first_token: bool = typer.Option(False, "--first-token", help="Track time to first token (streaming)"),
save: Optional[str] = typer.Option(None, "--save", help="Save artifacts to path (creates .prof, .txt)"),
output_format: str = typer.Option("text", "--format", "-f", help="Output format: text or json"),
):
"""
Profile a query execution with detailed timing breakdown.
Shows per-function and per-file timing, call graphs, and latency metrics.
Examples:
praisonai profile query "What is 2+2?"
praisonai profile query "Hello" --show-files --limit 20
praisonai profile query "Test" --stream --first-token
praisonai profile query "Test" --deep --show-callers --show-callees
praisonai profile query "Test" --save ./profile_results
"""
# Lazy import to avoid startup overhead
try:
from ..features.profiler import (
ProfilerConfig,
QueryProfiler,
format_profile_report,
)
except ImportError as e:
typer.echo(f"Error: Profiler module not available: {e}", err=True)
raise typer.Exit(1)
# Warn about deep tracing overhead
if deep:
typer.echo("⚠️ Deep call tracing enabled - this adds significant overhead", err=True)
# Create config
config = ProfilerConfig(
deep=deep,
limit=limit,
sort_by=sort,
show_files=show_files,
show_callers=show_callers,
show_callees=show_callees,
importtime=importtime,
first_token=first_token,
save_path=save,
output_format=output_format,
stream=stream,
)
# Run profiler
typer.echo("🔬 Starting profiled execution...", err=True)
try:
profiler = QueryProfiler(config)
result = profiler.profile_query(prompt, model=model, stream=stream)
except ImportError as e:
typer.echo(f"Error: {e}", err=True)
typer.echo("Install praisonaiagents: pip install praisonaiagents", err=True)
raise typer.Exit(1)
except Exception as e:
typer.echo(f"Error during profiling: {e}", err=True)
raise typer.Exit(1)
# Output results
if output_format == "json":
import json
typer.echo(json.dumps(result.to_dict(), indent=2, default=str))
else:
report = format_profile_report(result, config)
typer.echo(report)
# Save artifacts if requested
if save:
try:
prof_path, txt_path = profiler.save_artifacts(result, save)
typer.echo("\n✅ Artifacts saved:", err=True)
typer.echo(f" {prof_path}", err=True)
typer.echo(f" {txt_path}", err=True)
except Exception as e:
typer.echo(f"Warning: Failed to save artifacts: {e}", err=True)
@app.command("imports")
def profile_imports():
"""
Profile module import times.
Shows which modules take longest to import, useful for optimizing startup time.
"""
import subprocess
import re
typer.echo("🔬 Profiling import times...", err=True)
try:
result = subprocess.run(
[sys.executable, "-X", "importtime", "-c", "import praisonaiagents"],
capture_output=True,
text=True,
timeout=60,
)
# Parse import times
import_times = []
for line in result.stderr.split('\n'):
if 'import time:' in line:
match = re.search(r'import time:\s+(\d+)\s+\|\s+(\d+)\s+\|\s+(.+)', line)
if match:
self_time = int(match.group(1))
cumulative = int(match.group(2))
module = match.group(3).strip()
import_times.append((module, self_time, cumulative))
# Sort by cumulative time
import_times.sort(key=lambda x: x[2], reverse=True)
# Display
typer.echo("\n" + "=" * 70)
typer.echo("Import Time Analysis")
typer.echo("=" * 70)
typer.echo(f"{'Module':<45} {'Self (μs)':>10} {'Cumul (μs)':>12}")
typer.echo("-" * 70)
for module, self_time, cumulative in import_times[:30]:
module_name = module
if len(module_name) > 43:
module_name = "..." + module_name[-40:]
typer.echo(f"{module_name:<45} {self_time:>10} {cumulative:>12}")
typer.echo("-" * 70)
if import_times:
total_time = max(t[2] for t in import_times)
typer.echo(f"Total import time: {total_time / 1000:.2f} ms")
except subprocess.TimeoutExpired:
typer.echo("Error: Import profiling timed out", err=True)
raise typer.Exit(1)
except Exception as e:
typer.echo(f"Error: {e}", err=True)
raise typer.Exit(1)
@app.command("startup")
def profile_startup():
"""
Profile CLI startup time.
Measures time from CLI invocation to ready state.
"""
import time
import subprocess
typer.echo("🔬 Profiling startup time...", err=True)
# Measure cold start (new process)
start = time.perf_counter()
subprocess.run(
[sys.executable, "-c", "import praisonai; import praisonai.cli"],
capture_output=True,
text=True,
)
cold_start = (time.perf_counter() - start) * 1000
# Measure warm start (imports cached)
start = time.perf_counter()
subprocess.run(
[sys.executable, "-c", "import praisonai; import praisonai.cli"],
capture_output=True,
text=True,
)
warm_start = (time.perf_counter() - start) * 1000
typer.echo("\n" + "=" * 50)
typer.echo("Startup Time Analysis")
typer.echo("=" * 50)
typer.echo(f"Cold Start: {cold_start:>10.2f} ms")
typer.echo(f"Warm Start: {warm_start:>10.2f} ms")
typer.echo("=" * 50)
if cold_start > 1000:
typer.echo("\n⚠️ Cold start > 1s - consider lazy imports", err=True)
@app.command("suite")
def profile_suite(
output_dir: Optional[str] = typer.Option("/tmp/praisonai_profile_suite", "--output", "-o", help="Output directory for results"),
iterations: int = typer.Option(3, "--iterations", "-n", help="Iterations per scenario"),
quick: bool = typer.Option(False, "--quick", help="Quick mode (fewer iterations)"),
):
"""
Run a comprehensive profiling suite.
Runs multiple scenarios (streaming/non-streaming, simple/complex prompts)
and produces aggregated statistics and reports.
Examples:
praisonai profile suite
praisonai profile suite --output ./my_profile_results
praisonai profile suite --quick
"""
try:
from ..features.profiler import run_profile_suite, ScenarioConfig
except ImportError as e:
typer.echo(f"Error: Profiler module not available: {e}", err=True)
raise typer.Exit(1)
# Define scenarios
iters = 1 if quick else iterations
scenarios = [
ScenarioConfig(name="simple_non_stream", prompt="hi", stream=False, iterations=iters, warmup=0 if quick else 1),
ScenarioConfig(name="simple_stream", prompt="hi", stream=True, iterations=iters, warmup=0 if quick else 1),
]
if not quick:
scenarios.extend([
ScenarioConfig(name="medium_non_stream", prompt="Explain Python in 2 sentences", stream=False, iterations=max(1, iters-1)),
ScenarioConfig(name="medium_stream", prompt="Explain Python in 2 sentences", stream=True, iterations=max(1, iters-1)),
])
typer.echo("🔬 Running Profile Suite...", err=True)
typer.echo(f" Output: {output_dir}", err=True)
typer.echo(f" Scenarios: {len(scenarios)}", err=True)
typer.echo(f" Iterations: {iterations}", err=True)
try:
result = run_profile_suite(
output_dir=output_dir,
scenarios=scenarios,
verbose=True,
)
# Print summary
typer.echo("\n" + "=" * 60)
typer.echo("Profile Suite Summary")
typer.echo("=" * 60)
typer.echo(f"Startup Cold: {result.startup_cold_ms:.2f}ms")
typer.echo(f"Startup Warm: {result.startup_warm_ms:.2f}ms")
if result.import_analysis:
typer.echo(f"\nTop Import: {result.import_analysis[0]['module']}")
typer.echo(f" Time: {result.import_analysis[0]['cumulative_ms']:.2f}ms")
typer.echo("\nScenario Results:")
for scenario in result.scenarios:
stats = scenario.get_stats(scenario.total_times)
typer.echo(f" {scenario.name}: {stats['mean']:.2f}ms (±{stats['stdev']:.2f}ms)")
typer.echo(f"\n✅ Full results saved to: {output_dir}")
except Exception as e:
typer.echo(f"Error running suite: {e}", err=True)
raise typer.Exit(1)
@app.command("snapshot")
def profile_snapshot(
name: str = typer.Argument("current", help="Name for the snapshot"),
baseline: bool = typer.Option(False, "--baseline", "-b", help="Save as baseline for comparison"),
compare: bool = typer.Option(False, "--compare", "-c", help="Compare against baseline"),
output_format: str = typer.Option("text", "--format", "-f", help="Output format: text or json"),
):
"""
Create or compare performance snapshots.
Snapshots record timing data for baseline comparison.
Use --baseline to save a reference point, then --compare to check for regressions.
Examples:
praisonai profile snapshot --baseline
praisonai profile snapshot current --compare
praisonai profile snapshot v2.0 --format json
"""
try:
from ..features.profiler import (
run_profile_suite,
ScenarioConfig,
PerfSnapshotManager,
create_snapshot_from_suite,
format_comparison_report,
)
except ImportError as e:
typer.echo(f"Error: Profiler module not available: {e}", err=True)
raise typer.Exit(1)
manager = PerfSnapshotManager()
if compare:
# Load baseline and compare
baseline_snap = manager.load_baseline()
if not baseline_snap:
typer.echo("❌ No baseline found. Create one with: praisonai profile snapshot --baseline", err=True)
raise typer.Exit(1)
typer.echo("🔬 Running current profile for comparison...", err=True)
# Run quick suite
scenarios = [
ScenarioConfig(name="simple", prompt="hi", stream=False, iterations=2, warmup=1),
]
result = run_profile_suite(
output_dir="/tmp/praisonai_snapshot_compare",
scenarios=scenarios,
verbose=False,
)
current_snap = create_snapshot_from_suite(result, name)
comparison = manager.compare(current_snap)
if output_format == "json":
import json
typer.echo(json.dumps(comparison.to_dict(), indent=2))
else:
typer.echo(format_comparison_report(comparison))
if comparison.is_regression():
raise typer.Exit(1) # Exit with error on regression
else:
# Create new snapshot
typer.echo(f"🔬 Creating performance snapshot '{name}'...", err=True)
scenarios = [
ScenarioConfig(name="simple", prompt="hi", stream=False, iterations=3, warmup=1),
ScenarioConfig(name="stream", prompt="hi", stream=True, iterations=2, warmup=1),
]
result = run_profile_suite(
output_dir="/tmp/praisonai_snapshot",
scenarios=scenarios,
verbose=True,
)
snapshot = create_snapshot_from_suite(result, name)
if baseline:
path = manager.save_baseline(snapshot)
typer.echo(f"\n✅ Baseline saved to: {path}")
else:
path = manager.save(snapshot)
typer.echo(f"\n✅ Snapshot saved to: {path}")
if output_format == "json":
import json
typer.echo(json.dumps(snapshot.to_dict(), indent=2))
else:
typer.echo("\nSnapshot Summary:")
typer.echo(f" Startup Cold: {snapshot.startup_cold_ms:.2f}ms")
typer.echo(f" Import Time: {snapshot.import_time_ms:.2f}ms")
typer.echo(f" Query Time: {snapshot.query_time_ms:.2f}ms")
@app.command("optimize")
def profile_optimize(
prewarm: bool = typer.Option(False, "--prewarm", help="Enable provider pre-warming"),
lite: bool = typer.Option(False, "--lite", help="Enable lite mode (skip type validation)"),
show_config: bool = typer.Option(False, "--show", help="Show current optimization config"),
):
"""
Configure performance optimizations.
All optimizations are opt-in and safe by default.
Examples:
praisonai profile optimize --show
praisonai profile optimize --prewarm
praisonai profile optimize --lite
"""
try:
from ..features.profiler import (
PrewarmManager,
get_lite_mode_config,
get_provider_cache,
)
except ImportError as e:
typer.echo(f"Error: Profiler module not available: {e}", err=True)
raise typer.Exit(1)
if show_config:
typer.echo("=" * 50)
typer.echo("Performance Optimization Status")
typer.echo("=" * 50)
lite_config = get_lite_mode_config()
typer.echo(f"\nLite Mode: {'Enabled' if lite_config.enabled else 'Disabled'}")
if lite_config.enabled:
typer.echo(f" Skip Type Validation: {lite_config.skip_type_validation}")
typer.echo(f" Skip Model Validation: {lite_config.skip_model_validation}")
typer.echo(f" Minimal Imports: {lite_config.minimal_imports}")
typer.echo(f"\nPre-warming: {'Enabled' if PrewarmManager.is_enabled() else 'Disabled'}")
cache = get_provider_cache()
stats = cache.get_stats()
typer.echo("\nProvider Cache:")
typer.echo(f" Hits: {stats['hits']}")
typer.echo(f" Misses: {stats['misses']}")
typer.echo(f" Size: {stats['size']}")
typer.echo("\n" + "=" * 50)
typer.echo("\nTo enable optimizations, set environment variables:")
typer.echo(" PRAISONAI_LITE_MODE=1")
typer.echo(" PRAISONAI_SKIP_TYPE_VALIDATION=1")
typer.echo(" PRAISONAI_MINIMAL_IMPORTS=1")
return
if prewarm:
typer.echo("🔥 Enabling provider pre-warming...")
PrewarmManager.enable()
import os
if os.environ.get('OPENAI_API_KEY'):
typer.echo(" Pre-warming OpenAI...")
PrewarmManager.prewarm_provider('openai')
if os.environ.get('ANTHROPIC_API_KEY'):
typer.echo(" Pre-warming Anthropic...")
PrewarmManager.prewarm_provider('anthropic')
typer.echo("✅ Pre-warming initiated (runs in background)")
if lite:
typer.echo("⚡ Lite mode configuration:")
typer.echo(" Set these environment variables to enable:")
typer.echo(" export PRAISONAI_LITE_MODE=1")
typer.echo(" export PRAISONAI_SKIP_TYPE_VALIDATION=1")
typer.echo(" export PRAISONAI_MINIMAL_IMPORTS=1")
if not prewarm and not lite and not show_config:
typer.echo("Use --show to see current config, or --prewarm/--lite to enable optimizations")
@app.callback(invoke_without_command=True)
def profile_callback(ctx: typer.Context):
"""Show profile help if no subcommand."""
if ctx.invoked_subcommand is None:
typer.echo(ctx.get_help())
@@ -102,6 +102,11 @@ __all__ = [
'RuntimeConfig',
'CodeIntelligenceRouter',
'ActionOrchestrator',
# Interactive tools provider (canonical source)
'get_interactive_tools',
'resolve_tool_groups',
'ToolConfig',
'TOOL_GROUPS',
]
def __getattr__(name):
@@ -247,4 +252,17 @@ def __getattr__(name):
elif name == 'ActionOrchestrator':
from .action_orchestrator import ActionOrchestrator
return ActionOrchestrator
# Interactive tools provider (canonical source)
elif name == 'get_interactive_tools':
from .interactive_tools import get_interactive_tools
return get_interactive_tools
elif name == 'resolve_tool_groups':
from .interactive_tools import resolve_tool_groups
return resolve_tool_groups
elif name == 'ToolConfig':
from .interactive_tools import ToolConfig
return ToolConfig
elif name == 'TOOL_GROUPS':
from .interactive_tools import TOOL_GROUPS
return TOOL_GROUPS
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -0,0 +1,380 @@
"""
Interactive Default Tools Provider for PraisonAI.
This module provides the canonical source of truth for default tools
in interactive modes (TUI and `praison "prompt"`).
Tool Groups:
- `acp`: ACP-powered file operations (create, edit, delete, execute)
- `lsp`: LSP-powered code intelligence (symbols, definitions, references)
- `basic`: Basic file tools (read, list, search)
- `interactive`: Union of all groups (default)
Usage:
from praisonai.cli.features.interactive_tools import (
get_interactive_tools,
resolve_tool_groups,
TOOL_GROUPS,
)
# Get all interactive tools (default)
tools = get_interactive_tools()
# Get specific groups
tools = get_interactive_tools(groups=["acp", "basic"])
# Disable specific groups
tools = get_interactive_tools(disable=["lsp"])
"""
import logging
import os
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional, Set
logger = logging.getLogger(__name__)
# Tool group definitions
TOOL_GROUPS = {
"acp": [
"acp_create_file",
"acp_edit_file",
"acp_delete_file",
"acp_execute_command",
],
"lsp": [
"lsp_list_symbols",
"lsp_find_definition",
"lsp_find_references",
"lsp_get_diagnostics",
],
"basic": [
"read_file",
"write_file",
"list_files",
"execute_command",
"internet_search",
],
}
# Default interactive group includes all
TOOL_GROUPS["interactive"] = (
TOOL_GROUPS["acp"] +
TOOL_GROUPS["lsp"] +
TOOL_GROUPS["basic"]
)
@dataclass
class ToolConfig:
"""Configuration for tool loading."""
workspace: str = field(default_factory=os.getcwd)
enable_acp: bool = True
enable_lsp: bool = True
enable_basic: bool = True
approval_mode: str = "auto" # auto, manual, scoped
lsp_enabled: bool = True
acp_enabled: bool = True
@classmethod
def from_env(cls) -> "ToolConfig":
"""Create config from environment variables."""
config = cls()
# Check env vars for disabling groups
disable_str = os.environ.get("PRAISON_TOOLS_DISABLE", "")
if disable_str:
disabled = [g.strip().lower() for g in disable_str.split(",")]
if "acp" in disabled:
config.enable_acp = False
if "lsp" in disabled:
config.enable_lsp = False
if "basic" in disabled:
config.enable_basic = False
# Check for workspace
workspace = os.environ.get("PRAISON_WORKSPACE", "")
if workspace:
config.workspace = workspace
# Check approval mode
approval = os.environ.get("PRAISON_APPROVAL_MODE", "")
if approval in ("auto", "manual", "scoped"):
config.approval_mode = approval
return config
def resolve_tool_groups(
groups: Optional[List[str]] = None,
disable: Optional[List[str]] = None,
config: Optional[ToolConfig] = None,
) -> Set[str]:
"""
Resolve which tool names should be included based on groups and config.
Args:
groups: Specific groups to include (None = all enabled)
disable: Groups to explicitly disable
config: Tool configuration
Returns:
Set of tool names to include
"""
if config is None:
config = ToolConfig.from_env()
disable = disable or []
# Start with requested groups or all enabled
if groups:
tool_names = set()
for group in groups:
if group in TOOL_GROUPS:
tool_names.update(TOOL_GROUPS[group])
else:
tool_names = set()
if config.enable_acp:
tool_names.update(TOOL_GROUPS["acp"])
if config.enable_lsp:
tool_names.update(TOOL_GROUPS["lsp"])
if config.enable_basic:
tool_names.update(TOOL_GROUPS["basic"])
# Remove disabled groups
for group in disable:
if group in TOOL_GROUPS:
tool_names -= set(TOOL_GROUPS[group])
return tool_names
def _load_basic_tools() -> Dict[str, Callable]:
"""Lazy load basic tools from praisonaiagents."""
tools = {}
try:
from praisonaiagents.tools import read_file
tools["read_file"] = read_file
except ImportError:
logger.debug("read_file not available")
try:
from praisonaiagents.tools import write_file
tools["write_file"] = write_file
except ImportError:
logger.debug("write_file not available")
try:
from praisonaiagents.tools import list_files
tools["list_files"] = list_files
except ImportError:
logger.debug("list_files not available")
try:
from praisonaiagents.tools import execute_command
tools["execute_command"] = execute_command
except ImportError:
logger.debug("execute_command not available")
try:
from praisonaiagents.tools import internet_search
tools["internet_search"] = internet_search
except ImportError:
logger.debug("internet_search not available")
return tools
def _load_acp_tools(config: ToolConfig) -> Dict[str, Callable]:
"""Lazy load ACP tools with runtime initialization."""
tools = {}
try:
from .interactive_runtime import InteractiveRuntime, RuntimeConfig
from .agent_tools import create_agent_centric_tools
# Create runtime config
runtime_config = RuntimeConfig(
workspace=config.workspace,
lsp_enabled=config.lsp_enabled,
acp_enabled=config.acp_enabled,
approval_mode=config.approval_mode,
)
# Create runtime (lazy - doesn't start until needed)
runtime = InteractiveRuntime(runtime_config)
# Get all agent-centric tools
all_tools = create_agent_centric_tools(runtime)
# Map by name
for tool in all_tools:
name = tool.__name__
if name.startswith("acp_"):
tools[name] = tool
logger.debug(f"Loaded {len(tools)} ACP tools")
except ImportError as e:
logger.debug(f"ACP tools not available: {e}")
except Exception as e:
logger.warning(f"Error loading ACP tools: {e}")
return tools
def _load_lsp_tools(config: ToolConfig) -> Dict[str, Callable]:
"""Lazy load LSP tools with runtime initialization."""
tools = {}
try:
from .interactive_runtime import InteractiveRuntime, RuntimeConfig
from .agent_tools import create_agent_centric_tools
# Create runtime config
runtime_config = RuntimeConfig(
workspace=config.workspace,
lsp_enabled=config.lsp_enabled,
acp_enabled=config.acp_enabled,
approval_mode=config.approval_mode,
)
# Create runtime
runtime = InteractiveRuntime(runtime_config)
# Get all agent-centric tools
all_tools = create_agent_centric_tools(runtime)
# Map by name
for tool in all_tools:
name = tool.__name__
if name.startswith("lsp_"):
tools[name] = tool
logger.debug(f"Loaded {len(tools)} LSP tools")
except ImportError as e:
logger.debug(f"LSP tools not available: {e}")
except Exception as e:
logger.warning(f"Error loading LSP tools: {e}")
return tools
def get_interactive_tools(
groups: Optional[List[str]] = None,
disable: Optional[List[str]] = None,
config: Optional[ToolConfig] = None,
workspace: Optional[str] = None,
) -> List[Callable]:
"""
Get the default interactive tools for TUI and prompt modes.
This is the canonical source of truth for interactive tool defaults.
Both `praisonai tui launch` and `praison "prompt"` should use this.
Args:
groups: Specific tool groups to include (default: all enabled)
disable: Tool groups to disable (e.g., ["acp", "lsp"])
config: Tool configuration (created from env if not provided)
workspace: Override workspace path
Returns:
List of tool callables
Example:
# Default (all tools)
tools = get_interactive_tools()
# Disable ACP tools
tools = get_interactive_tools(disable=["acp"])
# Only basic tools
tools = get_interactive_tools(groups=["basic"])
"""
if config is None:
config = ToolConfig.from_env()
if workspace:
config.workspace = workspace
# Resolve which tools to include
tool_names = resolve_tool_groups(groups, disable, config)
# Load tools by group
all_tools: Dict[str, Callable] = {}
# Load basic tools (always available, no runtime needed)
if config.enable_basic and not (disable and "basic" in disable):
basic_tools = _load_basic_tools()
all_tools.update(basic_tools)
# Load ACP tools (requires runtime)
if config.enable_acp and not (disable and "acp" in disable):
acp_tools = _load_acp_tools(config)
all_tools.update(acp_tools)
# Load LSP tools (requires runtime)
if config.enable_lsp and not (disable and "lsp" in disable):
lsp_tools = _load_lsp_tools(config)
all_tools.update(lsp_tools)
# Filter to requested tools
result = []
for name in sorted(tool_names): # Deterministic ordering
if name in all_tools:
result.append(all_tools[name])
logger.debug(f"Loaded {len(result)} interactive tools: {[t.__name__ for t in result]}")
return result
def merge_tool_overrides(
default_tools: List[Callable],
add_tools: Optional[List[Callable]] = None,
remove_names: Optional[List[str]] = None,
) -> List[Callable]:
"""
Merge tool overrides with default tools.
Args:
default_tools: Base list of tools
add_tools: Tools to add
remove_names: Tool names to remove
Returns:
Merged tool list
"""
# Build name -> tool map
tool_map = {t.__name__: t for t in default_tools}
# Remove specified tools
if remove_names:
for name in remove_names:
tool_map.pop(name, None)
# Add new tools
if add_tools:
for tool in add_tools:
tool_map[tool.__name__] = tool
return list(tool_map.values())
def get_tool_group_names() -> Dict[str, List[str]]:
"""Get available tool groups and their tool names."""
return TOOL_GROUPS.copy()
def print_tool_summary(tools: List[Callable]) -> None:
"""Print a summary of loaded tools (for debugging)."""
acp_count = sum(1 for t in tools if t.__name__.startswith("acp_"))
lsp_count = sum(1 for t in tools if t.__name__.startswith("lsp_"))
basic_count = len(tools) - acp_count - lsp_count
print(f"Interactive tools loaded: {len(tools)} total")
print(f" - ACP tools: {acp_count}")
print(f" - LSP tools: {lsp_count}")
print(f" - Basic tools: {basic_count}")
@@ -0,0 +1,81 @@
"""
PraisonAI CLI Profiler Module.
Provides detailed cProfile-based profiling for CLI commands and agent execution.
Supports per-file/per-function timing, call graphs, and import time analysis.
Includes Tier 0/1/2 optimizations:
- Tier 0: Lazy imports, provider caching, CLI startup optimization
- Tier 1: Connection pooling, prewarm hooks (opt-in)
- Tier 2: Lite mode, async init, perf snapshot
"""
from .core import (
ProfilerConfig,
ProfilerResult,
QueryProfiler,
run_profiled_query,
format_profile_report,
)
from .suite import (
ScenarioConfig,
ScenarioResult,
SuiteResult,
ProfileSuiteRunner,
run_profile_suite,
)
from .optimizations import (
# Tier 0
ProviderCache,
get_provider_cache,
resolve_model_provider,
LazyImporter,
# Tier 1
ClientPool,
get_client_pool,
PrewarmManager,
# Tier 2
LiteModeConfig,
get_lite_mode_config,
is_lite_mode,
PerfSnapshot,
PerfComparison,
PerfSnapshotManager,
create_snapshot_from_suite,
format_comparison_report,
)
__all__ = [
# Core profiler
'ProfilerConfig',
'ProfilerResult',
'QueryProfiler',
'run_profiled_query',
'format_profile_report',
# Suite runner
'ScenarioConfig',
'ScenarioResult',
'SuiteResult',
'ProfileSuiteRunner',
'run_profile_suite',
# Tier 0 optimizations
'ProviderCache',
'get_provider_cache',
'resolve_model_provider',
'LazyImporter',
# Tier 1 optimizations
'ClientPool',
'get_client_pool',
'PrewarmManager',
# Tier 2 optimizations
'LiteModeConfig',
'get_lite_mode_config',
'is_lite_mode',
'PerfSnapshot',
'PerfComparison',
'PerfSnapshotManager',
'create_snapshot_from_suite',
'format_comparison_report',
]
@@ -0,0 +1,558 @@
"""
Core profiler implementation for PraisonAI CLI.
Provides cProfile-based profiling with detailed per-file/per-function timing,
call graphs, and import time analysis.
"""
import cProfile
import io
import json
import os
import pstats
import re
import subprocess
import sys
import time
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
# Patterns for secrets to redact
SECRET_PATTERNS = [
r'sk-[a-zA-Z0-9]{20,}', # OpenAI keys
r'sk-ant-[a-zA-Z0-9-]{20,}', # Anthropic keys
r'AIza[a-zA-Z0-9_-]{35}', # Google API keys
r'tvly-[a-zA-Z0-9-]{20,}', # Tavily keys
r'[a-zA-Z0-9_-]{32,}', # Generic long tokens (be careful)
]
def redact_secrets(text: str) -> str:
"""Redact potential secrets from text."""
result = text
for pattern in SECRET_PATTERNS[:4]: # Skip generic pattern for safety
result = re.sub(pattern, '[REDACTED]', result)
return result
@dataclass
class ProfilerConfig:
"""Configuration for profiler."""
deep: bool = False # Enable deep call tracing
limit: int = 30 # Top N functions to show
sort_by: str = "cumulative" # cumulative or tottime
show_files: bool = False # Group by file
show_callers: bool = False # Show callers
show_callees: bool = False # Show callees
importtime: bool = False # Show import timing
first_token: bool = False # Track time to first token
save_path: Optional[str] = None # Path to save artifacts
output_format: str = "text" # text or json
stream: bool = False # Streaming mode
@dataclass
class TimingBreakdown:
"""Timing breakdown for profiling."""
cli_parse_ms: float = 0.0
imports_ms: float = 0.0
agent_construction_ms: float = 0.0
model_init_ms: float = 0.0
first_token_ms: float = 0.0
total_run_ms: float = 0.0
@dataclass
class FunctionStats:
"""Statistics for a single function."""
name: str
filename: str
lineno: int
calls: int
tottime: float # Time in function excluding subcalls
cumtime: float # Time in function including subcalls
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"filename": self.filename,
"lineno": self.lineno,
"calls": self.calls,
"tottime_ms": self.tottime * 1000,
"cumtime_ms": self.cumtime * 1000,
}
@dataclass
class FileStats:
"""Statistics aggregated by file."""
filename: str
total_time: float
functions: List[FunctionStats] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
return {
"filename": self.filename,
"total_time_ms": self.total_time * 1000,
"function_count": len(self.functions),
}
@dataclass
class ProfilerResult:
"""Result from a profiling run."""
prompt: str
response: str
timing: TimingBreakdown
function_stats: List[FunctionStats]
file_stats: List[FileStats]
callers: Dict[str, List[str]] = field(default_factory=dict)
callees: Dict[str, List[str]] = field(default_factory=dict)
import_times: List[Tuple[str, float]] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
def to_dict(self) -> Dict[str, Any]:
return {
"timestamp": self.timestamp,
"metadata": self.metadata,
"prompt": self.prompt[:200] + "..." if len(self.prompt) > 200 else self.prompt,
"response_preview": self.response[:200] + "..." if len(self.response) > 200 else self.response,
"timing": {
"cli_parse_ms": self.timing.cli_parse_ms,
"imports_ms": self.timing.imports_ms,
"agent_construction_ms": self.timing.agent_construction_ms,
"model_init_ms": self.timing.model_init_ms,
"first_token_ms": self.timing.first_token_ms,
"total_run_ms": self.timing.total_run_ms,
},
"top_functions": [f.to_dict() for f in self.function_stats[:20]],
"top_files": [f.to_dict() for f in self.file_stats[:10]],
}
class QueryProfiler:
"""
Profiler for query execution.
Uses cProfile for function-level profiling and optional sys.setprofile
for deep call tracing.
"""
def __init__(self, config: Optional[ProfilerConfig] = None):
self.config = config or ProfilerConfig()
self.profiler = cProfile.Profile()
self._first_token_time: Optional[float] = None
self._start_time: float = 0.0
self._deep_trace_data: Dict[str, Dict] = {}
def _on_first_token(self):
"""Called when first token is received (for streaming)."""
if self._first_token_time is None:
self._first_token_time = time.perf_counter()
def _deep_trace_callback(self, frame, event, arg):
"""Callback for sys.setprofile deep tracing."""
if event not in ('call', 'return'):
return
filename = frame.f_code.co_filename
funcname = frame.f_code.co_name
lineno = frame.f_lineno
key = f"{filename}:{funcname}:{lineno}"
if key not in self._deep_trace_data:
self._deep_trace_data[key] = {
'filename': filename,
'funcname': funcname,
'lineno': lineno,
'calls': 0,
'total_time': 0.0,
'start_times': [],
}
if event == 'call':
self._deep_trace_data[key]['calls'] += 1
self._deep_trace_data[key]['start_times'].append(time.perf_counter())
elif event == 'return' and self._deep_trace_data[key]['start_times']:
start = self._deep_trace_data[key]['start_times'].pop()
self._deep_trace_data[key]['total_time'] += time.perf_counter() - start
def profile_query(
self,
prompt: str,
model: Optional[str] = None,
stream: bool = False,
) -> ProfilerResult:
"""
Profile a query execution.
Args:
prompt: The prompt to execute
model: Optional model to use
stream: Whether to use streaming mode
Returns:
ProfilerResult with detailed timing and statistics
"""
timing = TimingBreakdown()
response = ""
# Metadata
metadata = self._collect_metadata(model)
# Time CLI parsing (simulated - actual parsing happens before this)
cli_start = time.perf_counter()
timing.cli_parse_ms = (time.perf_counter() - cli_start) * 1000
# Time imports
import_start = time.perf_counter()
try:
from praisonaiagents import Agent
except ImportError:
raise ImportError("praisonaiagents not installed")
timing.imports_ms = (time.perf_counter() - import_start) * 1000
# Time agent construction
construct_start = time.perf_counter()
agent_config = {
"name": "ProfilerAgent",
"role": "Assistant",
"goal": "Complete the task",
"verbose": False,
}
if model:
agent_config["llm"] = model
agent = Agent(**agent_config)
timing.agent_construction_ms = (time.perf_counter() - construct_start) * 1000
# Time model initialization (first call)
model_init_start = time.perf_counter()
timing.model_init_ms = (time.perf_counter() - model_init_start) * 1000
# Profile the actual execution
self._start_time = time.perf_counter()
self._first_token_time = None
if self.config.deep:
sys.setprofile(self._deep_trace_callback)
self.profiler.enable()
try:
if stream and hasattr(agent, '_start_stream'):
# Streaming mode with first token tracking
chunks = []
for chunk in agent._start_stream(prompt):
if not chunks:
self._on_first_token()
chunks.append(chunk)
response = ''.join(chunks)
else:
# Non-streaming mode
response = agent.start(prompt)
if response is None:
response = ""
finally:
self.profiler.disable()
if self.config.deep:
sys.setprofile(None)
end_time = time.perf_counter()
timing.total_run_ms = (end_time - self._start_time) * 1000
if self._first_token_time:
timing.first_token_ms = (self._first_token_time - self._start_time) * 1000
# Extract statistics
function_stats = self._extract_function_stats()
file_stats = self._extract_file_stats(function_stats)
callers, callees = self._extract_call_graph()
# Get import times if requested
import_times = []
if self.config.importtime:
import_times = self._get_import_times()
return ProfilerResult(
prompt=prompt,
response=str(response),
timing=timing,
function_stats=function_stats,
file_stats=file_stats,
callers=callers if self.config.show_callers else {},
callees=callees if self.config.show_callees else {},
import_times=import_times,
metadata=metadata,
)
def _collect_metadata(self, model: Optional[str]) -> Dict[str, Any]:
"""Collect system metadata."""
import platform
try:
from praisonai.version import __version__ as praisonai_version
except ImportError:
praisonai_version = "unknown"
return {
"python_version": platform.python_version(),
"platform": platform.platform(),
"praisonai_version": praisonai_version,
"model": model or "default",
"timestamp": datetime.utcnow().isoformat() + "Z",
}
def _extract_function_stats(self) -> List[FunctionStats]:
"""Extract function statistics from profiler."""
stats_stream = io.StringIO()
stats = pstats.Stats(self.profiler, stream=stats_stream)
# Sort by configured key
sort_key = pstats.SortKey.CUMULATIVE if self.config.sort_by == "cumulative" else pstats.SortKey.TIME
stats.sort_stats(sort_key)
function_stats = []
for (filename, lineno, funcname), (cc, nc, tt, ct, callers) in stats.stats.items():
function_stats.append(FunctionStats(
name=funcname,
filename=filename,
lineno=lineno,
calls=nc,
tottime=tt,
cumtime=ct,
))
# Sort and limit
if self.config.sort_by == "cumulative":
function_stats.sort(key=lambda x: x.cumtime, reverse=True)
else:
function_stats.sort(key=lambda x: x.tottime, reverse=True)
return function_stats[:self.config.limit]
def _extract_file_stats(self, function_stats: List[FunctionStats]) -> List[FileStats]:
"""Aggregate statistics by file."""
file_map: Dict[str, FileStats] = {}
for func in function_stats:
if func.filename not in file_map:
file_map[func.filename] = FileStats(
filename=func.filename,
total_time=0.0,
functions=[],
)
file_map[func.filename].total_time += func.cumtime
file_map[func.filename].functions.append(func)
file_stats = list(file_map.values())
file_stats.sort(key=lambda x: x.total_time, reverse=True)
return file_stats[:self.config.limit]
def _extract_call_graph(self) -> Tuple[Dict[str, List[str]], Dict[str, List[str]]]:
"""Extract caller/callee relationships."""
callers: Dict[str, List[str]] = {}
callees: Dict[str, List[str]] = {}
stats = pstats.Stats(self.profiler)
for (filename, lineno, funcname), (cc, nc, tt, ct, caller_dict) in stats.stats.items():
func_key = f"{funcname} ({os.path.basename(filename)}:{lineno})"
if caller_dict:
callers[func_key] = []
for (caller_file, caller_line, caller_name), _ in caller_dict.items():
caller_key = f"{caller_name} ({os.path.basename(caller_file)}:{caller_line})"
callers[func_key].append(caller_key)
if caller_key not in callees:
callees[caller_key] = []
callees[caller_key].append(func_key)
return callers, callees
def _get_import_times(self) -> List[Tuple[str, float]]:
"""Get import timing by running subprocess with -X importtime."""
try:
result = subprocess.run(
[sys.executable, "-X", "importtime", "-c", "import praisonaiagents"],
capture_output=True,
text=True,
timeout=30,
)
import_times = []
for line in result.stderr.split('\n'):
if 'import time:' in line:
# Parse: "import time: 123 | 456 | module_name"
match = re.search(r'import time:\s+(\d+)\s+\|\s+(\d+)\s+\|\s+(.+)', line)
if match:
self_time = int(match.group(1))
cumulative = int(match.group(2))
module = match.group(3).strip()
import_times.append((module, cumulative / 1000000)) # Convert to seconds
# Sort by time descending
import_times.sort(key=lambda x: x[1], reverse=True)
return import_times[:20] # Top 20
except Exception:
return []
def save_artifacts(self, result: ProfilerResult, base_path: str):
"""Save profiling artifacts to files."""
base = Path(base_path)
base.parent.mkdir(parents=True, exist_ok=True)
# Save .prof binary
prof_path = str(base) + ".prof"
self.profiler.dump_stats(prof_path)
# Save .txt report
txt_path = str(base) + ".txt"
with open(txt_path, 'w') as f:
f.write(format_profile_report(result, self.config))
# Save .json if requested
if self.config.output_format == "json":
json_path = str(base) + ".json"
with open(json_path, 'w') as f:
json.dump(result.to_dict(), f, indent=2, default=str)
return prof_path, txt_path
def run_profiled_query(
prompt: str,
config: Optional[ProfilerConfig] = None,
model: Optional[str] = None,
) -> ProfilerResult:
"""
Run a profiled query.
Args:
prompt: The prompt to execute
config: Profiler configuration
model: Optional model to use
Returns:
ProfilerResult with detailed timing and statistics
"""
profiler = QueryProfiler(config)
return profiler.profile_query(prompt, model=model, stream=config.stream if config else False)
def format_profile_report(result: ProfilerResult, config: Optional[ProfilerConfig] = None) -> str:
"""
Format a profile result as a text report.
Args:
result: ProfilerResult to format
config: Optional profiler config
Returns:
Formatted text report
"""
config = config or ProfilerConfig()
lines = []
# Header
lines.append("=" * 70)
lines.append("PraisonAI Profile Report")
lines.append("=" * 70)
lines.append("")
# Metadata
lines.append("## System Information")
lines.append(f" Timestamp: {result.timestamp}")
lines.append(f" Python Version: {result.metadata.get('python_version', 'N/A')}")
lines.append(f" Platform: {result.metadata.get('platform', 'N/A')}")
lines.append(f" PraisonAI: {result.metadata.get('praisonai_version', 'N/A')}")
lines.append(f" Model: {result.metadata.get('model', 'N/A')}")
lines.append("")
# Timing breakdown
lines.append("## Timing Breakdown")
lines.append(f" CLI Parse: {result.timing.cli_parse_ms:>10.2f} ms")
lines.append(f" Imports: {result.timing.imports_ms:>10.2f} ms")
lines.append(f" Agent Construct: {result.timing.agent_construction_ms:>10.2f} ms")
lines.append(f" Model Init: {result.timing.model_init_ms:>10.2f} ms")
if result.timing.first_token_ms > 0:
lines.append(f" First Token: {result.timing.first_token_ms:>10.2f} ms")
lines.append(f" Total Run: {result.timing.total_run_ms:>10.2f} ms")
lines.append("")
# Per-file timing (if requested)
if config.show_files and result.file_stats:
lines.append("## Per-File Timing (Top Files)")
lines.append("-" * 70)
lines.append(f"{'File':<50} {'Time (ms)':>15}")
lines.append("-" * 70)
for fs in result.file_stats[:15]:
filename = os.path.basename(fs.filename)
if len(filename) > 48:
filename = "..." + filename[-45:]
lines.append(f"{filename:<50} {fs.total_time * 1000:>15.2f}")
lines.append("")
# Per-function timing
lines.append("## Per-Function Timing (Top Functions)")
lines.append("-" * 70)
sort_label = "Cumulative" if config.sort_by == "cumulative" else "Total"
lines.append(f"{'Function':<35} {'Calls':>8} {sort_label + ' (ms)':>12} {'Self (ms)':>12}")
lines.append("-" * 70)
for fs in result.function_stats[:config.limit]:
funcname = fs.name
if len(funcname) > 33:
funcname = funcname[:30] + "..."
lines.append(f"{funcname:<35} {fs.calls:>8} {fs.cumtime * 1000:>12.2f} {fs.tottime * 1000:>12.2f}")
lines.append("")
# Callers (if requested)
if config.show_callers and result.callers:
lines.append("## Callers (Who called each function)")
lines.append("-" * 70)
for func, caller_list in list(result.callers.items())[:10]:
lines.append(f" {func}:")
for caller in caller_list[:5]:
lines.append(f" <- {caller}")
lines.append("")
# Callees (if requested)
if config.show_callees and result.callees:
lines.append("## Callees (What each function called)")
lines.append("-" * 70)
for func, callee_list in list(result.callees.items())[:10]:
lines.append(f" {func}:")
for callee in callee_list[:5]:
lines.append(f" -> {callee}")
lines.append("")
# Import times (if requested)
if config.importtime and result.import_times:
lines.append("## Import Times (Top Modules)")
lines.append("-" * 70)
lines.append(f"{'Module':<50} {'Time (ms)':>15}")
lines.append("-" * 70)
for module, time_sec in result.import_times[:15]:
module_name = module
if len(module_name) > 48:
module_name = "..." + module_name[-45:]
lines.append(f"{module_name:<50} {time_sec * 1000:>15.2f}")
lines.append("")
# Response preview
lines.append("## Response Preview")
lines.append("-" * 70)
preview = result.response[:500] + "..." if len(result.response) > 500 else result.response
lines.append(preview)
lines.append("")
lines.append("=" * 70)
return "\n".join(lines)
@@ -0,0 +1,652 @@
"""
Performance Optimizations for PraisonAI CLI.
Implements Tier 0/1/2 optimizations:
- Tier 0: Lazy imports, provider caching, CLI startup optimization
- Tier 1: Connection pooling, prewarm hooks (opt-in)
- Tier 2: Lite mode, async init, perf snapshot
All optimizations are safe-by-default and opt-in where behavior changes.
Multi-agent safe: no global mutable state that affects concurrent runs.
"""
import os
import json
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple
from functools import lru_cache
# =============================================================================
# TIER 0: Safe Fast Wins (No behavior change)
# =============================================================================
class ProviderCache:
"""
Thread-safe cache for provider/model resolution.
Caches resolved provider configurations to avoid repeated lookups.
Per-process cache with clear invalidation rules.
Multi-agent safe: uses thread-local storage for mutable state.
"""
_instance = None
_lock = threading.Lock()
def __new__(cls):
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._cache = {}
cls._instance._cache_lock = threading.Lock()
cls._instance._hits = 0
cls._instance._misses = 0
return cls._instance
def get(self, key: str) -> Optional[Any]:
"""Get cached value (thread-safe)."""
with self._cache_lock:
if key in self._cache:
self._hits += 1
return self._cache[key]
self._misses += 1
return None
def set(self, key: str, value: Any, ttl_seconds: int = 300) -> None:
"""Set cached value with TTL (thread-safe)."""
with self._cache_lock:
self._cache[key] = {
'value': value,
'expires': time.time() + ttl_seconds,
}
def invalidate(self, key: Optional[str] = None) -> None:
"""Invalidate cache entry or entire cache."""
with self._cache_lock:
if key:
self._cache.pop(key, None)
else:
self._cache.clear()
def get_stats(self) -> Dict[str, int]:
"""Get cache statistics."""
with self._cache_lock:
return {
'hits': self._hits,
'misses': self._misses,
'size': len(self._cache),
}
def cleanup_expired(self) -> int:
"""Remove expired entries. Returns count of removed entries."""
now = time.time()
removed = 0
with self._cache_lock:
expired_keys = [
k for k, v in self._cache.items()
if isinstance(v, dict) and v.get('expires', float('inf')) < now
]
for k in expired_keys:
del self._cache[k]
removed += 1
return removed
# Global provider cache instance (singleton, thread-safe)
_provider_cache = None
def get_provider_cache() -> ProviderCache:
"""Get the global provider cache instance."""
global _provider_cache
if _provider_cache is None:
_provider_cache = ProviderCache()
return _provider_cache
@lru_cache(maxsize=32)
def resolve_model_provider(model: str) -> Tuple[str, str]:
"""
Resolve model string to (provider, model_name) tuple.
Cached using LRU cache for repeated lookups.
"""
if '/' in model:
parts = model.split('/', 1)
return parts[0], parts[1]
# Default provider mappings
if model.startswith('gpt-') or model.startswith('o1-') or model.startswith('o3-'):
return 'openai', model
elif model.startswith('claude-'):
return 'anthropic', model
elif model.startswith('gemini-'):
return 'google', model
elif model.startswith('llama') or model.startswith('mistral'):
return 'ollama', model
return 'openai', model # Default to OpenAI
class LazyImporter:
"""
Lazy importer for heavy modules.
Defers import until first use, reducing CLI startup time.
Thread-safe and caches imported modules.
"""
_imports: Dict[str, Any] = {}
_lock = threading.Lock()
@classmethod
def get(cls, module_path: str, attr: Optional[str] = None) -> Any:
"""
Lazily import a module or attribute.
Args:
module_path: Full module path (e.g., 'openai.OpenAI')
attr: Optional attribute to get from module
Returns:
Imported module or attribute
"""
cache_key = f"{module_path}.{attr}" if attr else module_path
with cls._lock:
if cache_key in cls._imports:
return cls._imports[cache_key]
# Import outside lock to avoid deadlocks
import importlib
if attr:
module = importlib.import_module(module_path)
result = getattr(module, attr)
else:
result = importlib.import_module(module_path)
with cls._lock:
cls._imports[cache_key] = result
return result
@classmethod
def clear(cls) -> None:
"""Clear import cache."""
with cls._lock:
cls._imports.clear()
# =============================================================================
# TIER 1: Medium Effort (Guarded/opt-in)
# =============================================================================
class ClientPool:
"""
Connection pool for API clients.
Reuses client instances for repeated calls to the same provider.
Multi-agent safe: clients are isolated by key.
Does not leak API keys between different configurations.
"""
def __init__(self, max_size: int = 10):
self._pool: Dict[str, Any] = {}
self._lock = threading.Lock()
self._max_size = max_size
self._access_times: Dict[str, float] = {}
def get_or_create(
self,
key: str,
factory: Callable[[], Any],
) -> Any:
"""
Get existing client or create new one.
Args:
key: Unique key for this client configuration
factory: Function to create new client if needed
Returns:
Client instance
"""
with self._lock:
if key in self._pool:
self._access_times[key] = time.time()
return self._pool[key]
# Evict oldest if at capacity
if len(self._pool) >= self._max_size:
oldest_key = min(self._access_times, key=self._access_times.get)
del self._pool[oldest_key]
del self._access_times[oldest_key]
# Create outside lock
client = factory()
with self._lock:
self._pool[key] = client
self._access_times[key] = time.time()
return client
def remove(self, key: str) -> None:
"""Remove client from pool."""
with self._lock:
self._pool.pop(key, None)
self._access_times.pop(key, None)
def clear(self) -> None:
"""Clear all clients from pool."""
with self._lock:
self._pool.clear()
self._access_times.clear()
def size(self) -> int:
"""Get current pool size."""
with self._lock:
return len(self._pool)
# Global client pool (opt-in usage)
_client_pool: Optional[ClientPool] = None
def get_client_pool() -> ClientPool:
"""Get the global client pool instance."""
global _client_pool
if _client_pool is None:
_client_pool = ClientPool()
return _client_pool
class PrewarmManager:
"""
Manager for pre-warming provider connections.
OPT-IN ONLY: Must be explicitly enabled.
Pre-initializes provider clients in background to reduce first-call latency.
"""
_enabled = False
_prewarmed: Dict[str, bool] = {}
_lock = threading.Lock()
_background_thread: Optional[threading.Thread] = None
@classmethod
def enable(cls) -> None:
"""Enable pre-warming (opt-in)."""
cls._enabled = True
@classmethod
def disable(cls) -> None:
"""Disable pre-warming."""
cls._enabled = False
@classmethod
def is_enabled(cls) -> bool:
"""Check if pre-warming is enabled."""
return cls._enabled
@classmethod
def prewarm_provider(cls, provider: str, api_key: Optional[str] = None) -> None:
"""
Pre-warm a provider connection in background.
Only runs if pre-warming is enabled.
"""
if not cls._enabled:
return
with cls._lock:
if provider in cls._prewarmed:
return
cls._prewarmed[provider] = False
def _prewarm():
try:
if provider == 'openai':
# Just import and create client - don't make API call
from openai import OpenAI
key = api_key or os.environ.get('OPENAI_API_KEY')
if key:
_ = OpenAI(api_key=key)
elif provider == 'anthropic':
from anthropic import Anthropic
key = api_key or os.environ.get('ANTHROPIC_API_KEY')
if key:
_ = Anthropic(api_key=key)
with cls._lock:
cls._prewarmed[provider] = True
except Exception:
pass # Silently fail - pre-warming is optional
# Run in background thread
thread = threading.Thread(target=_prewarm, daemon=True)
thread.start()
@classmethod
def is_prewarmed(cls, provider: str) -> bool:
"""Check if provider is pre-warmed."""
with cls._lock:
return cls._prewarmed.get(provider, False)
@classmethod
def reset(cls) -> None:
"""Reset pre-warm state."""
with cls._lock:
cls._prewarmed.clear()
cls._enabled = False
# =============================================================================
# TIER 2: Architectural (Safe and non-breaking)
# =============================================================================
@dataclass
class LiteModeConfig:
"""
Configuration for lite runtime mode.
OPT-IN ONLY: Defaults to OFF.
Avoids expensive type/model loading when not required.
"""
enabled: bool = False
skip_type_validation: bool = False
skip_model_validation: bool = False
minimal_imports: bool = False
@classmethod
def from_env(cls) -> 'LiteModeConfig':
"""Create config from environment variables."""
return cls(
enabled=os.environ.get('PRAISONAI_LITE_MODE', '').lower() in ('1', 'true', 'yes'),
skip_type_validation=os.environ.get('PRAISONAI_SKIP_TYPE_VALIDATION', '').lower() in ('1', 'true'),
skip_model_validation=os.environ.get('PRAISONAI_SKIP_MODEL_VALIDATION', '').lower() in ('1', 'true'),
minimal_imports=os.environ.get('PRAISONAI_MINIMAL_IMPORTS', '').lower() in ('1', 'true'),
)
# Global lite mode config
_lite_mode_config: Optional[LiteModeConfig] = None
def get_lite_mode_config() -> LiteModeConfig:
"""Get lite mode configuration."""
global _lite_mode_config
if _lite_mode_config is None:
_lite_mode_config = LiteModeConfig.from_env()
return _lite_mode_config
def is_lite_mode() -> bool:
"""Check if lite mode is enabled."""
return get_lite_mode_config().enabled
@dataclass
class PerfSnapshot:
"""
Performance snapshot for baseline comparison.
Records timing data that can be compared against later runs.
"""
timestamp: str
name: str
startup_cold_ms: float
startup_warm_ms: float
import_time_ms: float
query_time_ms: float
first_token_ms: float
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
return {
'timestamp': self.timestamp,
'name': self.name,
'startup_cold_ms': self.startup_cold_ms,
'startup_warm_ms': self.startup_warm_ms,
'import_time_ms': self.import_time_ms,
'query_time_ms': self.query_time_ms,
'first_token_ms': self.first_token_ms,
'metadata': self.metadata,
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'PerfSnapshot':
return cls(
timestamp=data['timestamp'],
name=data['name'],
startup_cold_ms=data['startup_cold_ms'],
startup_warm_ms=data['startup_warm_ms'],
import_time_ms=data['import_time_ms'],
query_time_ms=data['query_time_ms'],
first_token_ms=data.get('first_token_ms', 0.0),
metadata=data.get('metadata', {}),
)
@dataclass
class PerfComparison:
"""Comparison between two performance snapshots."""
baseline: PerfSnapshot
current: PerfSnapshot
@property
def startup_cold_diff_ms(self) -> float:
return self.current.startup_cold_ms - self.baseline.startup_cold_ms
@property
def startup_cold_diff_pct(self) -> float:
if self.baseline.startup_cold_ms == 0:
return 0.0
return (self.startup_cold_diff_ms / self.baseline.startup_cold_ms) * 100
@property
def import_time_diff_ms(self) -> float:
return self.current.import_time_ms - self.baseline.import_time_ms
@property
def import_time_diff_pct(self) -> float:
if self.baseline.import_time_ms == 0:
return 0.0
return (self.import_time_diff_ms / self.baseline.import_time_ms) * 100
@property
def query_time_diff_ms(self) -> float:
return self.current.query_time_ms - self.baseline.query_time_ms
@property
def query_time_diff_pct(self) -> float:
if self.baseline.query_time_ms == 0:
return 0.0
return (self.query_time_diff_ms / self.baseline.query_time_ms) * 100
def is_regression(self, threshold_pct: float = 10.0) -> bool:
"""Check if there's a performance regression above threshold."""
return (
self.startup_cold_diff_pct > threshold_pct or
self.import_time_diff_pct > threshold_pct or
self.query_time_diff_pct > threshold_pct
)
def to_dict(self) -> Dict[str, Any]:
return {
'baseline': self.baseline.to_dict(),
'current': self.current.to_dict(),
'diffs': {
'startup_cold_ms': self.startup_cold_diff_ms,
'startup_cold_pct': self.startup_cold_diff_pct,
'import_time_ms': self.import_time_diff_ms,
'import_time_pct': self.import_time_diff_pct,
'query_time_ms': self.query_time_diff_ms,
'query_time_pct': self.query_time_diff_pct,
},
'is_regression': self.is_regression(),
}
class PerfSnapshotManager:
"""
Manager for performance snapshots.
Stores snapshots locally for baseline comparison.
OPT-IN: User must explicitly invoke snapshot commands.
"""
DEFAULT_DIR = Path.home() / '.praisonai' / 'perf_snapshots'
def __init__(self, snapshot_dir: Optional[Path] = None):
self.snapshot_dir = snapshot_dir or self.DEFAULT_DIR
self.snapshot_dir.mkdir(parents=True, exist_ok=True)
def save(self, snapshot: PerfSnapshot) -> Path:
"""Save a snapshot to disk."""
filename = f"{snapshot.name}_{snapshot.timestamp.replace(':', '-').replace('.', '-')}.json"
path = self.snapshot_dir / filename
with open(path, 'w') as f:
json.dump(snapshot.to_dict(), f, indent=2)
return path
def load(self, name: str) -> Optional[PerfSnapshot]:
"""Load the most recent snapshot with given name."""
pattern = f"{name}_*.json"
files = sorted(self.snapshot_dir.glob(pattern), reverse=True)
if not files:
return None
with open(files[0]) as f:
data = json.load(f)
return PerfSnapshot.from_dict(data)
def load_baseline(self, name: str = 'baseline') -> Optional[PerfSnapshot]:
"""Load the baseline snapshot."""
return self.load(name)
def save_baseline(self, snapshot: PerfSnapshot) -> Path:
"""Save as baseline (overwrites previous baseline)."""
snapshot.name = 'baseline'
# Remove old baselines
for f in self.snapshot_dir.glob('baseline_*.json'):
f.unlink()
return self.save(snapshot)
def compare(self, current: PerfSnapshot, baseline_name: str = 'baseline') -> Optional[PerfComparison]:
"""Compare current snapshot against baseline."""
baseline = self.load(baseline_name)
if not baseline:
return None
return PerfComparison(baseline=baseline, current=current)
def list_snapshots(self) -> List[str]:
"""List all saved snapshot names."""
files = self.snapshot_dir.glob('*.json')
names = set()
for f in files:
# Extract name from filename (before first underscore)
name = f.stem.rsplit('_', 1)[0] if '_' in f.stem else f.stem
names.add(name)
return sorted(names)
# =============================================================================
# Utility Functions
# =============================================================================
def create_snapshot_from_suite(suite_result: Any, name: str = 'current') -> PerfSnapshot:
"""
Create a PerfSnapshot from a SuiteResult.
Args:
suite_result: SuiteResult from profile suite
name: Name for the snapshot
Returns:
PerfSnapshot instance
"""
# Get average query time from scenarios
query_times = []
first_token_times = []
for scenario in suite_result.scenarios:
query_times.extend(scenario.total_times)
first_token_times.extend(scenario.first_token_times)
avg_query_time = sum(query_times) / len(query_times) if query_times else 0.0
avg_first_token = sum(first_token_times) / len(first_token_times) if first_token_times else 0.0
# Get import time from analysis
import_time = 0.0
if suite_result.import_analysis:
import_time = suite_result.import_analysis[0].get('cumulative_ms', 0.0)
return PerfSnapshot(
timestamp=suite_result.timestamp,
name=name,
startup_cold_ms=suite_result.startup_cold_ms,
startup_warm_ms=suite_result.startup_warm_ms,
import_time_ms=import_time,
query_time_ms=avg_query_time,
first_token_ms=avg_first_token,
metadata=suite_result.metadata,
)
def format_comparison_report(comparison: PerfComparison) -> str:
"""Format a comparison report as text."""
lines = []
lines.append("=" * 70)
lines.append("Performance Comparison Report")
lines.append("=" * 70)
lines.append("")
lines.append(f"Baseline: {comparison.baseline.name} ({comparison.baseline.timestamp})")
lines.append(f"Current: {comparison.current.name} ({comparison.current.timestamp})")
lines.append("")
lines.append("-" * 70)
lines.append(f"{'Metric':<25} {'Baseline':>12} {'Current':>12} {'Diff':>12} {'%':>8}")
lines.append("-" * 70)
def format_row(name: str, baseline: float, current: float, diff: float, pct: float) -> str:
sign = '+' if diff > 0 else ''
pct_sign = '+' if pct > 0 else ''
return f"{name:<25} {baseline:>12.2f} {current:>12.2f} {sign}{diff:>11.2f} {pct_sign}{pct:>7.1f}%"
lines.append(format_row(
"Startup Cold (ms)",
comparison.baseline.startup_cold_ms,
comparison.current.startup_cold_ms,
comparison.startup_cold_diff_ms,
comparison.startup_cold_diff_pct,
))
lines.append(format_row(
"Import Time (ms)",
comparison.baseline.import_time_ms,
comparison.current.import_time_ms,
comparison.import_time_diff_ms,
comparison.import_time_diff_pct,
))
lines.append(format_row(
"Query Time (ms)",
comparison.baseline.query_time_ms,
comparison.current.query_time_ms,
comparison.query_time_diff_ms,
comparison.query_time_diff_pct,
))
lines.append("-" * 70)
lines.append("")
if comparison.is_regression():
lines.append("⚠️ REGRESSION DETECTED (>10% slower)")
else:
lines.append("✅ No significant regression")
lines.append("=" * 70)
return "\n".join(lines)
@@ -0,0 +1,386 @@
"""
Profile Suite Runner for PraisonAI CLI.
Runs a matrix of profiling scenarios and produces aggregated reports.
Supports cold/warm start, streaming/non-streaming, and multiple iterations.
"""
import json
import statistics
import subprocess
import sys
import time
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional
from .core import ProfilerConfig, ProfilerResult, QueryProfiler
@dataclass
class ScenarioConfig:
"""Configuration for a single profiling scenario."""
name: str
prompt: str
model: Optional[str] = None
stream: bool = False
iterations: int = 3
warmup: int = 1
show_files: bool = True
limit: int = 20
@dataclass
class ScenarioResult:
"""Result from running a scenario."""
name: str
config: ScenarioConfig
results: List[ProfilerResult] = field(default_factory=list)
@property
def import_times(self) -> List[float]:
return [r.timing.imports_ms for r in self.results]
@property
def total_times(self) -> List[float]:
return [r.timing.total_run_ms for r in self.results]
@property
def first_token_times(self) -> List[float]:
return [r.timing.first_token_ms for r in self.results if r.timing.first_token_ms > 0]
def get_stats(self, values: List[float]) -> Dict[str, float]:
if not values:
return {"mean": 0, "min": 0, "max": 0, "stdev": 0, "median": 0}
return {
"mean": statistics.mean(values),
"min": min(values),
"max": max(values),
"stdev": statistics.stdev(values) if len(values) > 1 else 0,
"median": statistics.median(values),
}
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"config": {
"prompt": self.config.prompt[:50] + "..." if len(self.config.prompt) > 50 else self.config.prompt,
"model": self.config.model or "default",
"stream": self.config.stream,
"iterations": self.config.iterations,
},
"import_time_stats": self.get_stats(self.import_times),
"total_time_stats": self.get_stats(self.total_times),
"first_token_stats": self.get_stats(self.first_token_times) if self.first_token_times else None,
"raw_import_times_ms": self.import_times,
"raw_total_times_ms": self.total_times,
}
@dataclass
class SuiteResult:
"""Result from running a full profiling suite."""
scenarios: List[ScenarioResult] = field(default_factory=list)
startup_cold_ms: float = 0.0
startup_warm_ms: float = 0.0
import_analysis: List[Dict[str, Any]] = field(default_factory=list)
timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
return {
"timestamp": self.timestamp,
"metadata": self.metadata,
"startup": {
"cold_ms": self.startup_cold_ms,
"warm_ms": self.startup_warm_ms,
},
"import_analysis": self.import_analysis[:20],
"scenarios": [s.to_dict() for s in self.scenarios],
}
class ProfileSuiteRunner:
"""
Runs a suite of profiling scenarios.
Collects comprehensive timing data across multiple runs and scenarios.
"""
DEFAULT_SCENARIOS = [
ScenarioConfig(name="simple_non_stream", prompt="hi", stream=False, iterations=3),
ScenarioConfig(name="simple_stream", prompt="hi", stream=True, iterations=3),
ScenarioConfig(name="medium_non_stream", prompt="Explain what Python is in 2 sentences", stream=False, iterations=2),
ScenarioConfig(name="medium_stream", prompt="Explain what Python is in 2 sentences", stream=True, iterations=2),
]
def __init__(
self,
scenarios: Optional[List[ScenarioConfig]] = None,
output_dir: Optional[str] = None,
verbose: bool = False,
):
self.scenarios = scenarios or self.DEFAULT_SCENARIOS
self.output_dir = Path(output_dir) if output_dir else Path("/tmp/praisonai_profile_suite")
self.verbose = verbose
def run(self) -> SuiteResult:
"""Run the full profiling suite."""
result = SuiteResult()
result.metadata = self._collect_metadata()
# Create output directory
self.output_dir.mkdir(parents=True, exist_ok=True)
if self.verbose:
print("🔬 Starting Profile Suite...")
print(f" Output: {self.output_dir}")
# 1. Measure startup times
if self.verbose:
print("\n📊 Measuring startup times...")
result.startup_cold_ms, result.startup_warm_ms = self._measure_startup()
if self.verbose:
print(f" Cold: {result.startup_cold_ms:.2f}ms, Warm: {result.startup_warm_ms:.2f}ms")
# 2. Analyze imports
if self.verbose:
print("\n📊 Analyzing imports...")
result.import_analysis = self._analyze_imports()
if self.verbose and result.import_analysis:
print(f" Top import: {result.import_analysis[0]['module']} ({result.import_analysis[0]['cumulative_ms']:.2f}ms)")
# 3. Run scenarios
for scenario in self.scenarios:
if self.verbose:
print(f"\n📊 Running scenario: {scenario.name}")
scenario_result = self._run_scenario(scenario)
result.scenarios.append(scenario_result)
if self.verbose:
stats = scenario_result.get_stats(scenario_result.total_times)
print(f" Total time: {stats['mean']:.2f}ms (±{stats['stdev']:.2f}ms)")
# 4. Save results
self._save_results(result)
if self.verbose:
print(f"\n✅ Suite complete. Results saved to {self.output_dir}")
return result
def _collect_metadata(self) -> Dict[str, Any]:
"""Collect system metadata."""
import platform
try:
from praisonai.version import __version__ as praisonai_version
except ImportError:
praisonai_version = "unknown"
return {
"python_version": platform.python_version(),
"platform": platform.platform(),
"praisonai_version": praisonai_version,
"timestamp": datetime.utcnow().isoformat() + "Z",
}
def _measure_startup(self) -> tuple:
"""Measure cold and warm startup times."""
# Cold start
start = time.perf_counter()
subprocess.run(
[sys.executable, "-c", "import praisonai; import praisonai.cli"],
capture_output=True,
text=True,
)
cold_ms = (time.perf_counter() - start) * 1000
# Warm start
start = time.perf_counter()
subprocess.run(
[sys.executable, "-c", "import praisonai; import praisonai.cli"],
capture_output=True,
text=True,
)
warm_ms = (time.perf_counter() - start) * 1000
return cold_ms, warm_ms
def _analyze_imports(self) -> List[Dict[str, Any]]:
"""Analyze import times."""
import re
try:
result = subprocess.run(
[sys.executable, "-X", "importtime", "-c", "import praisonaiagents"],
capture_output=True,
text=True,
timeout=60,
)
import_times = []
for line in result.stderr.split('\n'):
if 'import time:' in line:
match = re.search(r'import time:\s+(\d+)\s+\|\s+(\d+)\s+\|\s+(.+)', line)
if match:
self_time = int(match.group(1))
cumulative = int(match.group(2))
module = match.group(3).strip()
import_times.append({
"module": module,
"self_us": self_time,
"cumulative_us": cumulative,
"self_ms": self_time / 1000,
"cumulative_ms": cumulative / 1000,
})
import_times.sort(key=lambda x: x["cumulative_us"], reverse=True)
return import_times[:30]
except Exception:
return []
def _run_scenario(self, scenario: ScenarioConfig) -> ScenarioResult:
"""Run a single scenario with multiple iterations."""
result = ScenarioResult(name=scenario.name, config=scenario)
config = ProfilerConfig(
stream=scenario.stream,
show_files=scenario.show_files,
limit=scenario.limit,
first_token=scenario.stream,
)
# Warmup runs (discarded)
for _ in range(scenario.warmup):
try:
profiler = QueryProfiler(config)
profiler.profile_query(scenario.prompt, model=scenario.model, stream=scenario.stream)
except Exception:
pass
# Actual runs
for i in range(scenario.iterations):
try:
profiler = QueryProfiler(config)
profile_result = profiler.profile_query(
scenario.prompt,
model=scenario.model,
stream=scenario.stream,
)
result.results.append(profile_result)
if self.verbose:
print(f" Iteration {i+1}: {profile_result.timing.total_run_ms:.2f}ms")
except Exception as e:
if self.verbose:
print(f" Iteration {i+1}: ERROR - {e}")
return result
def _save_results(self, result: SuiteResult):
"""Save results to files."""
# Save JSON summary
json_path = self.output_dir / "suite_results.json"
with open(json_path, 'w') as f:
json.dump(result.to_dict(), f, indent=2, default=str)
# Save human-readable report
report_path = self.output_dir / "suite_report.txt"
with open(report_path, 'w') as f:
f.write(self._format_report(result))
def _format_report(self, result: SuiteResult) -> str:
"""Format a human-readable report."""
lines = []
lines.append("=" * 70)
lines.append("PraisonAI Profile Suite Report")
lines.append("=" * 70)
lines.append("")
# Metadata
lines.append("## System Information")
lines.append(f" Timestamp: {result.timestamp}")
lines.append(f" Python Version: {result.metadata.get('python_version', 'N/A')}")
lines.append(f" Platform: {result.metadata.get('platform', 'N/A')}")
lines.append(f" PraisonAI: {result.metadata.get('praisonai_version', 'N/A')}")
lines.append("")
# Startup times
lines.append("## Startup Times")
lines.append(f" Cold Start: {result.startup_cold_ms:>10.2f} ms")
lines.append(f" Warm Start: {result.startup_warm_ms:>10.2f} ms")
lines.append("")
# Import analysis
if result.import_analysis:
lines.append("## Top Import Times")
lines.append("-" * 70)
lines.append(f"{'Module':<45} {'Self (ms)':>10} {'Cumul (ms)':>12}")
lines.append("-" * 70)
for imp in result.import_analysis[:15]:
module = imp["module"]
if len(module) > 43:
module = "..." + module[-40:]
lines.append(f"{module:<45} {imp['self_ms']:>10.2f} {imp['cumulative_ms']:>12.2f}")
lines.append("")
# Scenario results
lines.append("## Scenario Results")
lines.append("-" * 70)
lines.append(f"{'Scenario':<25} {'Stream':>8} {'Iters':>6} {'Mean (ms)':>12} {'StdDev':>10}")
lines.append("-" * 70)
for scenario in result.scenarios:
stats = scenario.get_stats(scenario.total_times)
stream_str = "Yes" if scenario.config.stream else "No"
lines.append(
f"{scenario.name:<25} {stream_str:>8} {len(scenario.results):>6} "
f"{stats['mean']:>12.2f} {stats['stdev']:>10.2f}"
)
lines.append("")
# Summary statistics
all_import_times = []
all_total_times = []
for scenario in result.scenarios:
all_import_times.extend(scenario.import_times)
all_total_times.extend(scenario.total_times)
if all_import_times:
lines.append("## Overall Statistics")
lines.append(f" Import Time (mean): {statistics.mean(all_import_times):>10.2f} ms")
lines.append(f" Total Time (mean): {statistics.mean(all_total_times):>10.2f} ms")
if len(all_total_times) > 1:
lines.append(f" Total Time (stdev): {statistics.stdev(all_total_times):>10.2f} ms")
lines.append("")
lines.append("=" * 70)
return "\n".join(lines)
def run_profile_suite(
output_dir: Optional[str] = None,
scenarios: Optional[List[ScenarioConfig]] = None,
verbose: bool = True,
) -> SuiteResult:
"""
Run a profile suite and return results.
Args:
output_dir: Directory to save results
scenarios: Custom scenarios to run
verbose: Print progress
Returns:
SuiteResult with all profiling data
"""
runner = ProfileSuiteRunner(
scenarios=scenarios,
output_dir=output_dir,
verbose=verbose,
)
return runner.run()
@@ -73,6 +73,8 @@ if TEXTUAL_AVAILABLE:
model: Optional[str] = None,
agent_config: Optional[Dict[str, Any]] = None,
queue_config: Optional[QueueConfig] = None,
enable_acp: bool = True,
enable_lsp: bool = True,
):
super().__init__()
@@ -80,6 +82,12 @@ if TEXTUAL_AVAILABLE:
self.session_id = session_id or str(uuid.uuid4())[:8]
self.model = model or os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
self.agent_config = agent_config or {}
self.enable_acp = enable_acp
self.enable_lsp = enable_lsp
# Load default interactive tools (ACP + LSP) if not provided
if "tools" not in self.agent_config or not self.agent_config["tools"]:
self.agent_config["tools"] = self._load_default_tools()
# Queue manager
self.queue_config = queue_config or QueueConfig()
@@ -95,6 +103,43 @@ if TEXTUAL_AVAILABLE:
self._command_handlers: Dict[str, callable] = {}
self._setup_commands()
def _load_default_tools(self) -> list:
"""
Load default interactive tools (ACP + LSP) for TUI.
Uses the canonical interactive_tools provider.
"""
try:
from ..interactive_tools import get_interactive_tools, ToolConfig
# Build disable list based on flags
disable = []
if not self.enable_acp:
disable.append('acp')
if not self.enable_lsp:
disable.append('lsp')
config = ToolConfig(
workspace=self.workspace,
enable_acp=self.enable_acp,
enable_lsp=self.enable_lsp,
)
tools = get_interactive_tools(
config=config,
disable=disable if disable else None,
)
logger.info(f"TUI loaded {len(tools)} default tools (ACP: {self.enable_acp}, LSP: {self.enable_lsp})")
return tools
except ImportError as e:
logger.warning(f"Could not load interactive tools: {e}")
return []
except Exception as e:
logger.error(f"Error loading default tools: {e}")
return []
def _setup_commands(self) -> None:
"""Setup slash command handlers."""
self._command_handlers = {
@@ -505,6 +550,8 @@ def run_tui(
session_id: Optional[str] = None,
model: Optional[str] = None,
agent_config: Optional[Dict[str, Any]] = None,
enable_acp: bool = True,
enable_lsp: bool = True,
) -> None:
"""
Run the PraisonAI TUI.
@@ -514,6 +561,8 @@ def run_tui(
session_id: Session ID to resume.
model: Default model.
agent_config: Agent configuration.
enable_acp: Enable ACP tools (default: True).
enable_lsp: Enable LSP tools (default: True).
"""
if not TEXTUAL_AVAILABLE:
raise ImportError(
@@ -525,5 +574,7 @@ def run_tui(
session_id=session_id,
model=model,
agent_config=agent_config,
enable_acp=enable_acp,
enable_lsp=enable_lsp,
)
app.run()
@@ -56,8 +56,16 @@ def create_tui_app():
None, "--agent", "-a",
help="Agent config file (YAML)"
),
no_acp: bool = typer.Option(
False, "--no-acp",
help="Disable ACP tools (file operations with plan/approve/apply)"
),
no_lsp: bool = typer.Option(
False, "--no-lsp",
help="Disable LSP tools (code intelligence: symbols, definitions)"
),
):
"""Launch the interactive TUI."""
"""Launch the interactive TUI with ACP + LSP tools enabled by default."""
try:
from .app import run_tui
except ImportError:
@@ -79,6 +87,8 @@ def create_tui_app():
session_id=session,
model=model,
agent_config=agent_config,
enable_acp=not no_acp,
enable_lsp=not no_lsp,
)
return app
+1 -1
View File
@@ -27,7 +27,7 @@ LEGACY_COMMANDS = {
TYPER_COMMANDS = {
'config', 'traces', 'env', 'session', 'completion', 'version',
'debug', 'lsp', 'diag', 'doctor', 'acp', 'mcp', 'serve', 'schedule', 'run',
'tui', 'queue', # TUI and queue management commands
'tui', 'queue', 'profile', # TUI, queue management, and profiling commands
}
+64 -11
View File
@@ -329,8 +329,7 @@ class PraisonAI:
if combined_inputs:
combined_prompt = f"{args.command} {' '.join(combined_inputs)}"
result = self.handle_direct_prompt(combined_prompt)
if result is not None:
print(result)
# Result already printed by handle_direct_prompt, don't print again
return result
else:
self.agent_file = args.command
@@ -357,8 +356,7 @@ class PraisonAI:
prompt_parts.append(file_input)
prompt = ' '.join(prompt_parts)
result = self.handle_direct_prompt(prompt)
if result is not None:
print(result)
# Result already printed by handle_direct_prompt, don't print again
return result
else:
# Agent file was explicitly set, ignore direct prompt and use the file
@@ -374,8 +372,7 @@ class PraisonAI:
inputs.append(file_input)
combined_input = ' '.join(inputs)
result = self.handle_direct_prompt(combined_input)
if result is not None:
print(result)
# Result already printed by handle_direct_prompt, don't print again
return result
# If no command or direct_prompt, preserve agent_file from constructor (don't overwrite)
@@ -737,6 +734,8 @@ class PraisonAI:
parser.add_argument("--expand-tools", type=str, help="Tools for prompt expander (e.g., 'internet_search' or path to tools.py)")
parser.add_argument("--tools", "-t", type=str, help="Path to tools.py file for research agent")
parser.add_argument("--no-tools", action="store_true", help="Disable default built-in tools (for models that don't support tool calling)")
parser.add_argument("--no-acp", action="store_true", help="Disable ACP tools (agentic file operations with plan/approve/apply)")
parser.add_argument("--no-lsp", action="store_true", help="Disable LSP tools (code intelligence: symbols, definitions, references)")
parser.add_argument("--save", "-s", action="store_true", help="Save research output to file (output/research/)")
parser.add_argument("--verbose", "-v", action="store_true", help="Enable verbose output for research")
parser.add_argument("--web-search", action="store_true", help="Enable native web search (OpenAI, Gemini, Anthropic, xAI, Perplexity)")
@@ -1201,10 +1200,13 @@ class PraisonAI:
sys.exit(exit_code)
elif args.command == 'profile':
# Profile command - performance profiling and benchmarking
from .features.profile import handle_profile_command
exit_code = handle_profile_command(unknown_args)
sys.exit(exit_code)
# Profile command - delegate to Typer CLI for new profiler
# This routes to commands/profile.py which has query, imports, startup subcommands
from .app import app as typer_app
import sys as _sys
_sys.argv = ['praisonai', 'profile'] + unknown_args
typer_app()
sys.exit(0)
elif args.command == 'eval':
# Eval command - evaluate model responses against expected outputs
@@ -4883,7 +4885,58 @@ Now, {final_instruction.lower()}:"""
sys.exit(1)
def _load_interactive_tools(self):
"""Load tools for interactive mode."""
"""
Load tools for interactive mode using the canonical provider.
This method uses the centralized interactive_tools module which provides:
- ACP tools (acp_create_file, acp_edit_file, etc.) for safe file operations
- LSP tools (lsp_list_symbols, lsp_find_definition, etc.) for code intelligence
- Basic tools (read_file, write_file, etc.) for standard operations
Tool groups can be disabled via:
- CLI flags: --no-acp, --no-lsp
- Env vars: PRAISON_TOOLS_DISABLE=acp,lsp
"""
# Determine which groups to disable based on CLI args
disable_groups = []
if hasattr(self, 'args'):
if getattr(self.args, 'no_acp', False):
disable_groups.append('acp')
if getattr(self.args, 'no_lsp', False):
disable_groups.append('lsp')
# Get workspace
workspace = os.getcwd()
try:
from .features.interactive_tools import get_interactive_tools, ToolConfig
# Create config
config = ToolConfig.from_env()
config.workspace = workspace
# Apply CLI overrides
if 'acp' in disable_groups:
config.enable_acp = False
if 'lsp' in disable_groups:
config.enable_lsp = False
# Get tools from canonical provider
tools_list = get_interactive_tools(
config=config,
disable=disable_groups if disable_groups else None,
)
logging.debug(f"Loaded {len(tools_list)} interactive tools (ACP: {config.enable_acp}, LSP: {config.enable_lsp})")
return tools_list
except ImportError as e:
logging.debug(f"Interactive tools provider not available: {e}")
# Fallback to basic tools only
return self._load_basic_tools_fallback()
def _load_basic_tools_fallback(self):
"""Fallback to load basic tools when interactive_tools module unavailable."""
tools_list = []
try:
from praisonaiagents.tools import (
+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.9.2 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==3.0.0 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+1 -1
View File
@@ -1 +1 @@
__version__ = "2.9.2"
__version__ = "3.0.0"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.5.3",
"praisonaiagents>=0.6.0",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
@@ -0,0 +1,287 @@
"""
Unit tests for the interactive tools provider.
Tests the canonical source of truth for default tools in interactive modes.
"""
import os
import pytest
from unittest.mock import patch, MagicMock
class TestToolGroups:
"""Tests for tool group definitions."""
def test_tool_groups_defined(self):
"""Test that all expected tool groups are defined."""
from praisonai.cli.features.interactive_tools import TOOL_GROUPS
assert "acp" in TOOL_GROUPS
assert "lsp" in TOOL_GROUPS
assert "basic" in TOOL_GROUPS
assert "interactive" in TOOL_GROUPS
def test_acp_tools_list(self):
"""Test ACP tool group contains expected tools."""
from praisonai.cli.features.interactive_tools import TOOL_GROUPS
expected = {
"acp_create_file",
"acp_edit_file",
"acp_delete_file",
"acp_execute_command",
}
assert set(TOOL_GROUPS["acp"]) == expected
def test_lsp_tools_list(self):
"""Test LSP tool group contains expected tools."""
from praisonai.cli.features.interactive_tools import TOOL_GROUPS
expected = {
"lsp_list_symbols",
"lsp_find_definition",
"lsp_find_references",
"lsp_get_diagnostics",
}
assert set(TOOL_GROUPS["lsp"]) == expected
def test_basic_tools_list(self):
"""Test basic tool group contains expected tools."""
from praisonai.cli.features.interactive_tools import TOOL_GROUPS
expected = {
"read_file",
"write_file",
"list_files",
"execute_command",
"internet_search",
}
assert set(TOOL_GROUPS["basic"]) == expected
def test_interactive_group_is_union(self):
"""Test interactive group is union of all groups."""
from praisonai.cli.features.interactive_tools import TOOL_GROUPS
expected = set(TOOL_GROUPS["acp"]) | set(TOOL_GROUPS["lsp"]) | set(TOOL_GROUPS["basic"])
assert set(TOOL_GROUPS["interactive"]) == expected
class TestToolConfig:
"""Tests for ToolConfig dataclass."""
def test_default_config(self):
"""Test default configuration values."""
from praisonai.cli.features.interactive_tools import ToolConfig
config = ToolConfig()
assert config.enable_acp is True
assert config.enable_lsp is True
assert config.enable_basic is True
assert config.approval_mode == "auto"
def test_config_from_env_disable_acp(self):
"""Test config from env with ACP disabled."""
from praisonai.cli.features.interactive_tools import ToolConfig
with patch.dict(os.environ, {"PRAISON_TOOLS_DISABLE": "acp"}):
config = ToolConfig.from_env()
assert config.enable_acp is False
assert config.enable_lsp is True
def test_config_from_env_disable_lsp(self):
"""Test config from env with LSP disabled."""
from praisonai.cli.features.interactive_tools import ToolConfig
with patch.dict(os.environ, {"PRAISON_TOOLS_DISABLE": "lsp"}):
config = ToolConfig.from_env()
assert config.enable_acp is True
assert config.enable_lsp is False
def test_config_from_env_disable_multiple(self):
"""Test config from env with multiple groups disabled."""
from praisonai.cli.features.interactive_tools import ToolConfig
with patch.dict(os.environ, {"PRAISON_TOOLS_DISABLE": "acp,lsp"}):
config = ToolConfig.from_env()
assert config.enable_acp is False
assert config.enable_lsp is False
assert config.enable_basic is True
def test_config_from_env_workspace(self):
"""Test config from env with workspace override."""
from praisonai.cli.features.interactive_tools import ToolConfig
with patch.dict(os.environ, {"PRAISON_WORKSPACE": "/custom/path"}):
config = ToolConfig.from_env()
assert config.workspace == "/custom/path"
class TestResolveToolGroups:
"""Tests for resolve_tool_groups function."""
def test_resolve_all_enabled(self):
"""Test resolving with all groups enabled."""
from praisonai.cli.features.interactive_tools import resolve_tool_groups, ToolConfig, TOOL_GROUPS
config = ToolConfig()
tool_names = resolve_tool_groups(config=config)
expected = set(TOOL_GROUPS["interactive"])
assert tool_names == expected
def test_resolve_disable_acp(self):
"""Test resolving with ACP disabled."""
from praisonai.cli.features.interactive_tools import resolve_tool_groups, ToolConfig, TOOL_GROUPS
config = ToolConfig(enable_acp=False)
tool_names = resolve_tool_groups(disable=["acp"], config=config)
# Should not contain ACP tools
for acp_tool in TOOL_GROUPS["acp"]:
assert acp_tool not in tool_names
# Should contain LSP and basic tools
for lsp_tool in TOOL_GROUPS["lsp"]:
assert lsp_tool in tool_names
def test_resolve_disable_lsp(self):
"""Test resolving with LSP disabled."""
from praisonai.cli.features.interactive_tools import resolve_tool_groups, ToolConfig, TOOL_GROUPS
config = ToolConfig(enable_lsp=False)
tool_names = resolve_tool_groups(disable=["lsp"], config=config)
# Should not contain LSP tools
for lsp_tool in TOOL_GROUPS["lsp"]:
assert lsp_tool not in tool_names
# Should contain ACP and basic tools
for acp_tool in TOOL_GROUPS["acp"]:
assert acp_tool in tool_names
def test_resolve_specific_groups(self):
"""Test resolving specific groups only."""
from praisonai.cli.features.interactive_tools import resolve_tool_groups, TOOL_GROUPS
tool_names = resolve_tool_groups(groups=["basic"])
assert tool_names == set(TOOL_GROUPS["basic"])
class TestGetInteractiveTools:
"""Tests for get_interactive_tools function."""
def test_get_tools_returns_list(self):
"""Test that get_interactive_tools returns a list."""
from praisonai.cli.features.interactive_tools import get_interactive_tools, ToolConfig
config = ToolConfig(workspace="/tmp")
tools = get_interactive_tools(config=config)
assert isinstance(tools, list)
def test_get_tools_deterministic_order(self):
"""Test that tool order is deterministic."""
from praisonai.cli.features.interactive_tools import get_interactive_tools, ToolConfig
config = ToolConfig(workspace="/tmp")
tools1 = get_interactive_tools(config=config)
tools2 = get_interactive_tools(config=config)
names1 = [t.__name__ for t in tools1]
names2 = [t.__name__ for t in tools2]
assert names1 == names2
def test_get_tools_with_disable_acp(self):
"""Test get_interactive_tools with ACP disabled."""
from praisonai.cli.features.interactive_tools import get_interactive_tools, ToolConfig
config = ToolConfig(workspace="/tmp", enable_acp=False)
tools = get_interactive_tools(config=config, disable=["acp"])
tool_names = [t.__name__ for t in tools]
acp_tools = [n for n in tool_names if n.startswith("acp_")]
assert len(acp_tools) == 0
def test_get_tools_with_disable_lsp(self):
"""Test get_interactive_tools with LSP disabled."""
from praisonai.cli.features.interactive_tools import get_interactive_tools, ToolConfig
config = ToolConfig(workspace="/tmp", enable_lsp=False)
tools = get_interactive_tools(config=config, disable=["lsp"])
tool_names = [t.__name__ for t in tools]
lsp_tools = [n for n in tool_names if n.startswith("lsp_")]
assert len(lsp_tools) == 0
class TestMergeToolOverrides:
"""Tests for merge_tool_overrides function."""
def test_merge_add_tools(self):
"""Test adding tools to default list."""
from praisonai.cli.features.interactive_tools import merge_tool_overrides
def tool_a():
pass
def tool_b():
pass
def tool_c():
pass
default = [tool_a, tool_b]
result = merge_tool_overrides(default, add_tools=[tool_c])
names = [t.__name__ for t in result]
assert "tool_c" in names
def test_merge_remove_tools(self):
"""Test removing tools from default list."""
from praisonai.cli.features.interactive_tools import merge_tool_overrides
def tool_a():
pass
def tool_b():
pass
default = [tool_a, tool_b]
result = merge_tool_overrides(default, remove_names=["tool_a"])
names = [t.__name__ for t in result]
assert "tool_a" not in names
assert "tool_b" in names
class TestLazyImports:
"""Tests for lazy import behavior."""
def test_import_time_fast(self):
"""Test that importing the module is fast (no heavy deps)."""
import time
start = time.time()
# Force reimport
import importlib
import praisonai.cli.features.interactive_tools as module
importlib.reload(module)
duration = time.time() - start
# Should be under 100ms
assert duration < 0.1, f"Import took {duration:.3f}s, expected < 0.1s"
def test_tool_groups_available_without_runtime(self):
"""Test that TOOL_GROUPS is available without runtime initialization."""
from praisonai.cli.features.interactive_tools import TOOL_GROUPS
# Should be accessible immediately
assert len(TOOL_GROUPS) > 0
if __name__ == "__main__":
pytest.main([__file__, "-v"])
@@ -0,0 +1,366 @@
"""
Unit tests for CLI profiler module.
"""
class TestProfilerConfig:
"""Tests for ProfilerConfig dataclass."""
def test_default_config(self):
"""Test default configuration values."""
from praisonai.cli.features.profiler import ProfilerConfig
config = ProfilerConfig()
assert config.deep is False
assert config.limit == 30
assert config.sort_by == "cumulative"
assert config.show_files is False
assert config.show_callers is False
assert config.show_callees is False
assert config.importtime is False
assert config.first_token is False
assert config.save_path is None
assert config.output_format == "text"
assert config.stream is False
def test_custom_config(self):
"""Test custom configuration."""
from praisonai.cli.features.profiler import ProfilerConfig
config = ProfilerConfig(
deep=True,
limit=50,
sort_by="tottime",
show_files=True,
)
assert config.deep is True
assert config.limit == 50
assert config.sort_by == "tottime"
assert config.show_files is True
class TestTimingBreakdown:
"""Tests for TimingBreakdown dataclass."""
def test_default_timing(self):
"""Test default timing values."""
from praisonai.cli.features.profiler.core import TimingBreakdown
timing = TimingBreakdown()
assert timing.cli_parse_ms == 0.0
assert timing.imports_ms == 0.0
assert timing.agent_construction_ms == 0.0
assert timing.model_init_ms == 0.0
assert timing.first_token_ms == 0.0
assert timing.total_run_ms == 0.0
class TestFunctionStats:
"""Tests for FunctionStats dataclass."""
def test_to_dict(self):
"""Test conversion to dictionary."""
from praisonai.cli.features.profiler.core import FunctionStats
stats = FunctionStats(
name="test_func",
filename="/path/to/file.py",
lineno=42,
calls=10,
tottime=0.5,
cumtime=1.0,
)
result = stats.to_dict()
assert result["name"] == "test_func"
assert result["filename"] == "/path/to/file.py"
assert result["lineno"] == 42
assert result["calls"] == 10
assert result["tottime_ms"] == 500.0
assert result["cumtime_ms"] == 1000.0
class TestFileStats:
"""Tests for FileStats dataclass."""
def test_to_dict(self):
"""Test conversion to dictionary."""
from praisonai.cli.features.profiler.core import FileStats, FunctionStats
func_stats = FunctionStats(
name="func1",
filename="/path/to/file.py",
lineno=10,
calls=5,
tottime=0.1,
cumtime=0.2,
)
file_stats = FileStats(
filename="/path/to/file.py",
total_time=0.5,
functions=[func_stats],
)
result = file_stats.to_dict()
assert result["filename"] == "/path/to/file.py"
assert result["total_time_ms"] == 500.0
assert result["function_count"] == 1
class TestProfilerResult:
"""Tests for ProfilerResult dataclass."""
def test_to_dict(self):
"""Test conversion to dictionary."""
from praisonai.cli.features.profiler.core import (
ProfilerResult,
TimingBreakdown,
FunctionStats,
FileStats,
)
timing = TimingBreakdown(
cli_parse_ms=1.0,
imports_ms=100.0,
total_run_ms=500.0,
)
func_stats = [FunctionStats(
name="test",
filename="test.py",
lineno=1,
calls=1,
tottime=0.1,
cumtime=0.2,
)]
file_stats = [FileStats(
filename="test.py",
total_time=0.2,
)]
result = ProfilerResult(
prompt="test prompt",
response="test response",
timing=timing,
function_stats=func_stats,
file_stats=file_stats,
metadata={"model": "test"},
)
data = result.to_dict()
assert "timestamp" in data
assert data["prompt"] == "test prompt"
assert data["response_preview"] == "test response"
assert data["timing"]["cli_parse_ms"] == 1.0
assert data["timing"]["imports_ms"] == 100.0
assert len(data["top_functions"]) == 1
assert len(data["top_files"]) == 1
class TestQueryProfiler:
"""Tests for QueryProfiler class."""
def test_init_default_config(self):
"""Test initialization with default config."""
from praisonai.cli.features.profiler import QueryProfiler, ProfilerConfig
profiler = QueryProfiler()
assert profiler.config is not None
assert isinstance(profiler.config, ProfilerConfig)
def test_init_custom_config(self):
"""Test initialization with custom config."""
from praisonai.cli.features.profiler import QueryProfiler, ProfilerConfig
config = ProfilerConfig(deep=True, limit=50)
profiler = QueryProfiler(config)
assert profiler.config.deep is True
assert profiler.config.limit == 50
def test_collect_metadata(self):
"""Test metadata collection."""
from praisonai.cli.features.profiler import QueryProfiler
profiler = QueryProfiler()
metadata = profiler._collect_metadata("gpt-4")
assert "python_version" in metadata
assert "platform" in metadata
assert "praisonai_version" in metadata
assert metadata["model"] == "gpt-4"
assert "timestamp" in metadata
class TestFormatProfileReport:
"""Tests for format_profile_report function."""
def test_basic_report(self):
"""Test basic report formatting."""
from praisonai.cli.features.profiler import (
format_profile_report,
ProfilerConfig,
)
from praisonai.cli.features.profiler.core import (
ProfilerResult,
TimingBreakdown,
FunctionStats,
FileStats,
)
timing = TimingBreakdown(
cli_parse_ms=1.0,
imports_ms=100.0,
total_run_ms=500.0,
)
func_stats = [FunctionStats(
name="test_func",
filename="test.py",
lineno=1,
calls=10,
tottime=0.1,
cumtime=0.2,
)]
file_stats = [FileStats(
filename="test.py",
total_time=0.2,
)]
result = ProfilerResult(
prompt="test",
response="response",
timing=timing,
function_stats=func_stats,
file_stats=file_stats,
metadata={"python_version": "3.12", "platform": "test", "praisonai_version": "1.0", "model": "test"},
)
report = format_profile_report(result)
assert "PraisonAI Profile Report" in report
assert "System Information" in report
assert "Timing Breakdown" in report
assert "Per-Function Timing" in report
assert "test_func" in report
def test_report_with_files(self):
"""Test report with file grouping."""
from praisonai.cli.features.profiler import (
format_profile_report,
ProfilerConfig,
)
from praisonai.cli.features.profiler.core import (
ProfilerResult,
TimingBreakdown,
FunctionStats,
FileStats,
)
config = ProfilerConfig(show_files=True)
timing = TimingBreakdown()
func_stats = []
file_stats = [FileStats(filename="test.py", total_time=0.5)]
result = ProfilerResult(
prompt="test",
response="response",
timing=timing,
function_stats=func_stats,
file_stats=file_stats,
metadata={},
)
report = format_profile_report(result, config)
assert "Per-File Timing" in report
class TestRedactSecrets:
"""Tests for secret redaction."""
def test_redact_openai_key(self):
"""Test redacting OpenAI API keys."""
from praisonai.cli.features.profiler.core import redact_secrets
text = "API key: sk-1234567890abcdefghijklmnopqrstuvwxyz"
result = redact_secrets(text)
assert "sk-1234567890" not in result
assert "[REDACTED]" in result
def test_redact_anthropic_key(self):
"""Test redacting Anthropic API keys."""
from praisonai.cli.features.profiler.core import redact_secrets
text = "Key: sk-ant-api03-abcdefghijklmnopqrstuvwxyz"
result = redact_secrets(text)
assert "sk-ant-api03" not in result
assert "[REDACTED]" in result
def test_preserve_normal_text(self):
"""Test that normal text is preserved."""
from praisonai.cli.features.profiler.core import redact_secrets
text = "This is normal text without secrets"
result = redact_secrets(text)
assert result == text
class TestCLIImports:
"""Tests for CLI import functionality."""
def test_cli_app_imports(self):
"""Test that CLI app imports without errors."""
# This tests the fix for ModuleNotFoundError
from praisonai.cli.app import app
assert app is not None
def test_output_module_imports(self):
"""Test that output module imports correctly."""
from praisonai.cli.output import OutputController, OutputMode
assert OutputController is not None
assert OutputMode is not None
def test_profiler_module_imports(self):
"""Test that profiler module imports correctly."""
from praisonai.cli.features.profiler import (
ProfilerConfig,
ProfilerResult,
QueryProfiler,
run_profiled_query,
format_profile_report,
)
assert ProfilerConfig is not None
assert ProfilerResult is not None
assert QueryProfiler is not None
assert run_profiled_query is not None
assert format_profile_report is not None
class TestOutputController:
"""Tests for OutputController."""
def test_default_mode(self):
"""Test default output mode."""
from praisonai.cli.output import OutputController, OutputMode
controller = OutputController()
assert controller.mode == OutputMode.TEXT
assert controller.is_json_mode is False
assert controller.is_quiet is False
def test_json_mode(self):
"""Test JSON output mode."""
from praisonai.cli.output import OutputController, OutputMode
controller = OutputController(mode=OutputMode.JSON)
assert controller.is_json_mode is True
def test_quiet_mode(self):
"""Test quiet output mode."""
from praisonai.cli.output import OutputController, OutputMode
controller = OutputController(mode=OutputMode.QUIET)
assert controller.is_quiet is True
@@ -0,0 +1,584 @@
"""
Unit tests for profiler optimizations module.
Tests Tier 0/1/2 optimization features:
- Provider caching
- Lazy imports
- Client pooling
- Prewarm manager
- Lite mode
- Performance snapshots
"""
import os
import tempfile
import threading
from pathlib import Path
from unittest.mock import patch
class TestProviderCache:
"""Tests for ProviderCache (Tier 0)."""
def test_singleton_pattern(self):
"""Test that ProviderCache is a singleton."""
from praisonai.cli.features.profiler import ProviderCache
cache1 = ProviderCache()
cache2 = ProviderCache()
assert cache1 is cache2
def test_get_set(self):
"""Test basic get/set operations."""
from praisonai.cli.features.profiler import get_provider_cache
cache = get_provider_cache()
cache.invalidate() # Clear cache
# Miss
result = cache.get('test_key')
assert result is None
# Set
cache.set('test_key', {'value': 'test'})
# Hit - cache stores as {'value': ..., 'expires': ...}
result = cache.get('test_key')
assert result is not None
assert result['value']['value'] == 'test'
def test_invalidate(self):
"""Test cache invalidation."""
from praisonai.cli.features.profiler import get_provider_cache
cache = get_provider_cache()
cache.set('key1', 'value1')
cache.set('key2', 'value2')
# Invalidate single key
cache.invalidate('key1')
assert cache.get('key1') is None
assert cache.get('key2') is not None
# Invalidate all
cache.invalidate()
assert cache.get('key2') is None
def test_stats(self):
"""Test cache statistics."""
from praisonai.cli.features.profiler import get_provider_cache
cache = get_provider_cache()
cache.invalidate()
# Generate some hits and misses
cache.get('miss1') # miss
cache.set('hit1', 'value')
cache.get('hit1') # hit
cache.get('miss2') # miss
stats = cache.get_stats()
assert 'hits' in stats
assert 'misses' in stats
assert 'size' in stats
def test_thread_safety(self):
"""Test thread-safe operations."""
from praisonai.cli.features.profiler import get_provider_cache
cache = get_provider_cache()
cache.invalidate()
errors = []
def worker(thread_id):
try:
for i in range(100):
key = f"thread_{thread_id}_key_{i}"
cache.set(key, f"value_{i}")
cache.get(key)
except Exception as e:
errors.append(e)
threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()
assert len(errors) == 0
class TestResolveModelProvider:
"""Tests for resolve_model_provider (Tier 0)."""
def test_openai_models(self):
"""Test OpenAI model resolution."""
from praisonai.cli.features.profiler import resolve_model_provider
assert resolve_model_provider('gpt-4o') == ('openai', 'gpt-4o')
assert resolve_model_provider('gpt-4o-mini') == ('openai', 'gpt-4o-mini')
assert resolve_model_provider('o1-preview') == ('openai', 'o1-preview')
def test_anthropic_models(self):
"""Test Anthropic model resolution."""
from praisonai.cli.features.profiler import resolve_model_provider
assert resolve_model_provider('claude-3-opus') == ('anthropic', 'claude-3-opus')
assert resolve_model_provider('claude-3-sonnet') == ('anthropic', 'claude-3-sonnet')
def test_google_models(self):
"""Test Google model resolution."""
from praisonai.cli.features.profiler import resolve_model_provider
assert resolve_model_provider('gemini-pro') == ('google', 'gemini-pro')
assert resolve_model_provider('gemini-1.5-flash') == ('google', 'gemini-1.5-flash')
def test_explicit_provider(self):
"""Test explicit provider/model format."""
from praisonai.cli.features.profiler import resolve_model_provider
assert resolve_model_provider('openai/gpt-4') == ('openai', 'gpt-4')
assert resolve_model_provider('anthropic/claude-3') == ('anthropic', 'claude-3')
def test_caching(self):
"""Test that results are cached."""
from praisonai.cli.features.profiler import resolve_model_provider
# Clear cache
resolve_model_provider.cache_clear()
# First call
result1 = resolve_model_provider('gpt-4o')
# Second call (should be cached)
result2 = resolve_model_provider('gpt-4o')
assert result1 == result2
# Check cache info
info = resolve_model_provider.cache_info()
assert info.hits >= 1
class TestLazyImporter:
"""Tests for LazyImporter (Tier 0)."""
def test_import_module(self):
"""Test lazy module import."""
from praisonai.cli.features.profiler import LazyImporter
LazyImporter.clear()
# Import a standard library module
json_module = LazyImporter.get('json')
assert json_module is not None
assert hasattr(json_module, 'dumps')
def test_import_attribute(self):
"""Test lazy attribute import."""
from praisonai.cli.features.profiler import LazyImporter
LazyImporter.clear()
# Import a specific attribute
Path = LazyImporter.get('pathlib', 'Path')
assert Path is not None
assert callable(Path)
def test_caching(self):
"""Test that imports are cached."""
from praisonai.cli.features.profiler import LazyImporter
LazyImporter.clear()
# First import
json1 = LazyImporter.get('json')
# Second import (should be cached)
json2 = LazyImporter.get('json')
assert json1 is json2
class TestClientPool:
"""Tests for ClientPool (Tier 1)."""
def test_get_or_create(self):
"""Test client creation and reuse."""
from praisonai.cli.features.profiler import ClientPool
pool = ClientPool(max_size=5)
# Create client
client1 = pool.get_or_create('key1', lambda: {'id': 1})
assert client1 == {'id': 1}
# Reuse client
client2 = pool.get_or_create('key1', lambda: {'id': 2})
assert client2 == {'id': 1} # Same as first
def test_eviction(self):
"""Test LRU eviction when pool is full."""
from praisonai.cli.features.profiler import ClientPool
pool = ClientPool(max_size=2)
pool.get_or_create('key1', lambda: 'client1')
pool.get_or_create('key2', lambda: 'client2')
assert pool.size() == 2
# This should evict key1 (oldest)
pool.get_or_create('key3', lambda: 'client3')
assert pool.size() == 2
def test_remove(self):
"""Test explicit client removal."""
from praisonai.cli.features.profiler import ClientPool
pool = ClientPool()
pool.get_or_create('key1', lambda: 'client1')
assert pool.size() == 1
pool.remove('key1')
assert pool.size() == 0
def test_clear(self):
"""Test clearing all clients."""
from praisonai.cli.features.profiler import ClientPool
pool = ClientPool()
pool.get_or_create('key1', lambda: 'client1')
pool.get_or_create('key2', lambda: 'client2')
pool.clear()
assert pool.size() == 0
class TestPrewarmManager:
"""Tests for PrewarmManager (Tier 1)."""
def test_enable_disable(self):
"""Test enabling and disabling prewarm."""
from praisonai.cli.features.profiler import PrewarmManager
PrewarmManager.reset()
assert not PrewarmManager.is_enabled()
PrewarmManager.enable()
assert PrewarmManager.is_enabled()
PrewarmManager.disable()
assert not PrewarmManager.is_enabled()
def test_prewarm_disabled(self):
"""Test that prewarm does nothing when disabled."""
from praisonai.cli.features.profiler import PrewarmManager
PrewarmManager.reset()
# Should not raise even without API key
PrewarmManager.prewarm_provider('openai')
assert not PrewarmManager.is_prewarmed('openai')
def test_reset(self):
"""Test reset clears state."""
from praisonai.cli.features.profiler import PrewarmManager
PrewarmManager.enable()
PrewarmManager.reset()
assert not PrewarmManager.is_enabled()
class TestLiteModeConfig:
"""Tests for LiteModeConfig (Tier 2)."""
def test_default_disabled(self):
"""Test that lite mode is disabled by default."""
from praisonai.cli.features.profiler import LiteModeConfig
config = LiteModeConfig()
assert not config.enabled
assert not config.skip_type_validation
assert not config.skip_model_validation
assert not config.minimal_imports
def test_from_env(self):
"""Test loading config from environment."""
from praisonai.cli.features.profiler import LiteModeConfig
with patch.dict(os.environ, {
'PRAISONAI_LITE_MODE': '1',
'PRAISONAI_SKIP_TYPE_VALIDATION': 'true',
}):
config = LiteModeConfig.from_env()
assert config.enabled
assert config.skip_type_validation
assert not config.skip_model_validation
def test_is_lite_mode(self):
"""Test is_lite_mode helper."""
from praisonai.cli.features.profiler import is_lite_mode
# Reset global config
import praisonai.cli.features.profiler.optimizations as opt
opt._lite_mode_config = None
with patch.dict(os.environ, {'PRAISONAI_LITE_MODE': ''}, clear=False):
# Force reload
opt._lite_mode_config = None
assert not is_lite_mode()
class TestPerfSnapshot:
"""Tests for PerfSnapshot (Tier 2)."""
def test_to_dict(self):
"""Test snapshot serialization."""
from praisonai.cli.features.profiler import PerfSnapshot
snapshot = PerfSnapshot(
timestamp='2025-01-01T00:00:00Z',
name='test',
startup_cold_ms=100.0,
startup_warm_ms=80.0,
import_time_ms=500.0,
query_time_ms=2000.0,
first_token_ms=300.0,
)
data = snapshot.to_dict()
assert data['name'] == 'test'
assert data['startup_cold_ms'] == 100.0
assert data['query_time_ms'] == 2000.0
def test_from_dict(self):
"""Test snapshot deserialization."""
from praisonai.cli.features.profiler import PerfSnapshot
data = {
'timestamp': '2025-01-01T00:00:00Z',
'name': 'test',
'startup_cold_ms': 100.0,
'startup_warm_ms': 80.0,
'import_time_ms': 500.0,
'query_time_ms': 2000.0,
'first_token_ms': 300.0,
}
snapshot = PerfSnapshot.from_dict(data)
assert snapshot.name == 'test'
assert snapshot.startup_cold_ms == 100.0
class TestPerfComparison:
"""Tests for PerfComparison (Tier 2)."""
def test_diff_calculation(self):
"""Test difference calculations."""
from praisonai.cli.features.profiler import PerfSnapshot, PerfComparison
baseline = PerfSnapshot(
timestamp='2025-01-01T00:00:00Z',
name='baseline',
startup_cold_ms=100.0,
startup_warm_ms=80.0,
import_time_ms=500.0,
query_time_ms=2000.0,
first_token_ms=300.0,
)
current = PerfSnapshot(
timestamp='2025-01-02T00:00:00Z',
name='current',
startup_cold_ms=110.0, # 10% slower
startup_warm_ms=85.0,
import_time_ms=550.0, # 10% slower
query_time_ms=2200.0, # 10% slower
first_token_ms=330.0,
)
comparison = PerfComparison(baseline=baseline, current=current)
assert comparison.startup_cold_diff_ms == 10.0
assert comparison.startup_cold_diff_pct == 10.0
assert comparison.import_time_diff_ms == 50.0
assert comparison.query_time_diff_ms == 200.0
def test_regression_detection(self):
"""Test regression detection."""
from praisonai.cli.features.profiler import PerfSnapshot, PerfComparison
baseline = PerfSnapshot(
timestamp='2025-01-01T00:00:00Z',
name='baseline',
startup_cold_ms=100.0,
startup_warm_ms=80.0,
import_time_ms=500.0,
query_time_ms=2000.0,
first_token_ms=300.0,
)
# No regression (within 10%)
current_ok = PerfSnapshot(
timestamp='2025-01-02T00:00:00Z',
name='current',
startup_cold_ms=105.0,
startup_warm_ms=82.0,
import_time_ms=520.0,
query_time_ms=2100.0,
first_token_ms=310.0,
)
comparison_ok = PerfComparison(baseline=baseline, current=current_ok)
assert not comparison_ok.is_regression()
# Regression (>10%)
current_bad = PerfSnapshot(
timestamp='2025-01-02T00:00:00Z',
name='current',
startup_cold_ms=150.0, # 50% slower
startup_warm_ms=82.0,
import_time_ms=520.0,
query_time_ms=2100.0,
first_token_ms=310.0,
)
comparison_bad = PerfComparison(baseline=baseline, current=current_bad)
assert comparison_bad.is_regression()
class TestPerfSnapshotManager:
"""Tests for PerfSnapshotManager (Tier 2)."""
def test_save_and_load(self):
"""Test saving and loading snapshots."""
from praisonai.cli.features.profiler import PerfSnapshot, PerfSnapshotManager
with tempfile.TemporaryDirectory() as tmpdir:
manager = PerfSnapshotManager(snapshot_dir=Path(tmpdir))
snapshot = PerfSnapshot(
timestamp='2025-01-01T00:00:00Z',
name='test',
startup_cold_ms=100.0,
startup_warm_ms=80.0,
import_time_ms=500.0,
query_time_ms=2000.0,
first_token_ms=300.0,
)
path = manager.save(snapshot)
assert path.exists()
loaded = manager.load('test')
assert loaded is not None
assert loaded.name == 'test'
assert loaded.startup_cold_ms == 100.0
def test_baseline(self):
"""Test baseline save and load."""
from praisonai.cli.features.profiler import PerfSnapshot, PerfSnapshotManager
with tempfile.TemporaryDirectory() as tmpdir:
manager = PerfSnapshotManager(snapshot_dir=Path(tmpdir))
snapshot = PerfSnapshot(
timestamp='2025-01-01T00:00:00Z',
name='original',
startup_cold_ms=100.0,
startup_warm_ms=80.0,
import_time_ms=500.0,
query_time_ms=2000.0,
first_token_ms=300.0,
)
manager.save_baseline(snapshot)
baseline = manager.load_baseline()
assert baseline is not None
assert baseline.name == 'baseline'
def test_list_snapshots(self):
"""Test listing snapshots."""
from praisonai.cli.features.profiler import PerfSnapshot, PerfSnapshotManager
with tempfile.TemporaryDirectory() as tmpdir:
manager = PerfSnapshotManager(snapshot_dir=Path(tmpdir))
for name in ['snap1', 'snap2', 'snap3']:
snapshot = PerfSnapshot(
timestamp='2025-01-01T00:00:00Z',
name=name,
startup_cold_ms=100.0,
startup_warm_ms=80.0,
import_time_ms=500.0,
query_time_ms=2000.0,
first_token_ms=300.0,
)
manager.save(snapshot)
names = manager.list_snapshots()
assert 'snap1' in names
assert 'snap2' in names
assert 'snap3' in names
class TestFormatComparisonReport:
"""Tests for format_comparison_report."""
def test_format_report(self):
"""Test report formatting."""
from praisonai.cli.features.profiler import (
PerfSnapshot, PerfComparison, format_comparison_report
)
baseline = PerfSnapshot(
timestamp='2025-01-01T00:00:00Z',
name='baseline',
startup_cold_ms=100.0,
startup_warm_ms=80.0,
import_time_ms=500.0,
query_time_ms=2000.0,
first_token_ms=300.0,
)
current = PerfSnapshot(
timestamp='2025-01-02T00:00:00Z',
name='current',
startup_cold_ms=105.0,
startup_warm_ms=82.0,
import_time_ms=520.0,
query_time_ms=2100.0,
first_token_ms=310.0,
)
comparison = PerfComparison(baseline=baseline, current=current)
report = format_comparison_report(comparison)
assert 'Performance Comparison Report' in report
assert 'baseline' in report
assert 'current' in report
assert 'No significant regression' in report
class TestModuleImports:
"""Test that all exports are available."""
def test_all_exports_available(self):
"""Test that all __all__ exports are importable."""
from praisonai.cli.features.profiler import __all__
import praisonai.cli.features.profiler as profiler
for name in __all__:
assert hasattr(profiler, name), f"Missing export: {name}"
@@ -0,0 +1,228 @@
"""
Unit tests for CLI profiler suite module.
"""
class TestScenarioConfig:
"""Tests for ScenarioConfig dataclass."""
def test_default_config(self):
"""Test default configuration values."""
from praisonai.cli.features.profiler import ScenarioConfig
config = ScenarioConfig(name="test", prompt="hi")
assert config.name == "test"
assert config.prompt == "hi"
assert config.model is None
assert config.stream is False
assert config.iterations == 3
assert config.warmup == 1
assert config.show_files is True
assert config.limit == 20
def test_custom_config(self):
"""Test custom configuration."""
from praisonai.cli.features.profiler import ScenarioConfig
config = ScenarioConfig(
name="custom",
prompt="test prompt",
model="gpt-4o",
stream=True,
iterations=5,
)
assert config.name == "custom"
assert config.model == "gpt-4o"
assert config.stream is True
assert config.iterations == 5
class TestScenarioResult:
"""Tests for ScenarioResult dataclass."""
def test_empty_result(self):
"""Test empty result."""
from praisonai.cli.features.profiler import ScenarioConfig, ScenarioResult
config = ScenarioConfig(name="test", prompt="hi")
result = ScenarioResult(name="test", config=config)
assert result.name == "test"
assert len(result.results) == 0
assert result.import_times == []
assert result.total_times == []
def test_get_stats_empty(self):
"""Test stats with empty values."""
from praisonai.cli.features.profiler import ScenarioConfig, ScenarioResult
config = ScenarioConfig(name="test", prompt="hi")
result = ScenarioResult(name="test", config=config)
stats = result.get_stats([])
assert stats["mean"] == 0
assert stats["min"] == 0
assert stats["max"] == 0
def test_get_stats_single_value(self):
"""Test stats with single value."""
from praisonai.cli.features.profiler import ScenarioConfig, ScenarioResult
config = ScenarioConfig(name="test", prompt="hi")
result = ScenarioResult(name="test", config=config)
stats = result.get_stats([100.0])
assert stats["mean"] == 100.0
assert stats["min"] == 100.0
assert stats["max"] == 100.0
assert stats["stdev"] == 0
def test_get_stats_multiple_values(self):
"""Test stats with multiple values."""
from praisonai.cli.features.profiler import ScenarioConfig, ScenarioResult
config = ScenarioConfig(name="test", prompt="hi")
result = ScenarioResult(name="test", config=config)
stats = result.get_stats([100.0, 200.0, 300.0])
assert stats["mean"] == 200.0
assert stats["min"] == 100.0
assert stats["max"] == 300.0
assert stats["median"] == 200.0
def test_to_dict(self):
"""Test conversion to dictionary."""
from praisonai.cli.features.profiler import ScenarioConfig, ScenarioResult
config = ScenarioConfig(name="test", prompt="hi", iterations=2)
result = ScenarioResult(name="test", config=config)
data = result.to_dict()
assert data["name"] == "test"
assert "config" in data
assert data["config"]["prompt"] == "hi"
assert "import_time_stats" in data
assert "total_time_stats" in data
class TestSuiteResult:
"""Tests for SuiteResult dataclass."""
def test_default_result(self):
"""Test default result."""
from praisonai.cli.features.profiler import SuiteResult
result = SuiteResult()
assert len(result.scenarios) == 0
assert result.startup_cold_ms == 0.0
assert result.startup_warm_ms == 0.0
assert len(result.import_analysis) == 0
assert result.timestamp.endswith("Z") # ISO format with Z suffix
def test_to_dict(self):
"""Test conversion to dictionary."""
from praisonai.cli.features.profiler import SuiteResult
result = SuiteResult(
startup_cold_ms=100.0,
startup_warm_ms=50.0,
)
data = result.to_dict()
assert "timestamp" in data
assert "metadata" in data
assert data["startup"]["cold_ms"] == 100.0
assert data["startup"]["warm_ms"] == 50.0
class TestProfileSuiteRunner:
"""Tests for ProfileSuiteRunner class."""
def test_init_default(self):
"""Test initialization with defaults."""
from praisonai.cli.features.profiler import ProfileSuiteRunner
runner = ProfileSuiteRunner()
assert len(runner.scenarios) == 4 # DEFAULT_SCENARIOS
assert runner.verbose is False
def test_init_custom_scenarios(self):
"""Test initialization with custom scenarios."""
from praisonai.cli.features.profiler import ProfileSuiteRunner, ScenarioConfig
scenarios = [
ScenarioConfig(name="custom1", prompt="test1"),
ScenarioConfig(name="custom2", prompt="test2"),
]
runner = ProfileSuiteRunner(scenarios=scenarios)
assert len(runner.scenarios) == 2
assert runner.scenarios[0].name == "custom1"
def test_collect_metadata(self):
"""Test metadata collection."""
from praisonai.cli.features.profiler import ProfileSuiteRunner
runner = ProfileSuiteRunner()
metadata = runner._collect_metadata()
assert "python_version" in metadata
assert "platform" in metadata
assert "praisonai_version" in metadata
assert "timestamp" in metadata
def test_measure_startup(self):
"""Test startup measurement."""
from praisonai.cli.features.profiler import ProfileSuiteRunner
runner = ProfileSuiteRunner()
cold, warm = runner._measure_startup()
# Both should be positive numbers
assert cold > 0
assert warm > 0
def test_analyze_imports(self):
"""Test import analysis."""
from praisonai.cli.features.profiler import ProfileSuiteRunner
runner = ProfileSuiteRunner()
imports = runner._analyze_imports()
# Should return a list (may be empty if subprocess fails)
assert isinstance(imports, list)
# If we got results, check structure
if imports:
assert "module" in imports[0]
assert "cumulative_ms" in imports[0]
class TestProfilerModuleImports:
"""Tests for profiler module imports."""
def test_all_exports_available(self):
"""Test that all exports are available."""
from praisonai.cli.features.profiler import (
ProfilerConfig,
ProfilerResult,
QueryProfiler,
run_profiled_query,
format_profile_report,
ScenarioConfig,
ScenarioResult,
SuiteResult,
ProfileSuiteRunner,
run_profile_suite,
)
assert ProfilerConfig is not None
assert ProfilerResult is not None
assert QueryProfiler is not None
assert run_profiled_query is not None
assert format_profile_report is not None
assert ScenarioConfig is not None
assert ScenarioResult is not None
assert SuiteResult is not None
assert ProfileSuiteRunner is not None
assert run_profile_suite is not None
+4 -4
View File
@@ -4276,7 +4276,7 @@ requires-dist = [
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.1.0" },
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.1.0" },
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.1.0" },
{ name = "praisonaiagents", specifier = ">=0.5.3" },
{ name = "praisonaiagents", specifier = ">=0.6.0" },
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
{ name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
{ name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4324,7 +4324,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.5.3"
version = "0.6.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -4333,9 +4333,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/75/33/6091503673fb3c143293e9639bafe3acb6e7406e0ace1f893c34cbe22912/praisonaiagents-0.5.3.tar.gz", hash = "sha256:4163b00c28ef5e5d1a75411eb4b72fc7749790e52cdaa4bfc98c9a747fe42140", size = 780945, upload-time = "2025-12-31T16:36:45.335Z" }
sdist = { url = "https://files.pythonhosted.org/packages/4b/3c/a25740d7e3da4ebea0d83319d412650a016d91b6eb2f2653df1b28a3882f/praisonaiagents-0.6.0.tar.gz", hash = "sha256:f83528f6f1c0adea17cea25a1ea93f660e46ba43002e354717501e82112d9a96", size = 780850, upload-time = "2025-12-31T18:50:35.35Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/32/a2/459f111873f1751a34bc780e1b3a2cd264cf580f35acff868fa35aa9d917/praisonaiagents-0.5.3-py3-none-any.whl", hash = "sha256:35bd6c26dfff6c9e77575f72a2f077cdbcc0544d0d45b0ed29fb00ca96c3719c", size = 724879, upload-time = "2025-12-31T16:36:43.383Z" },
{ url = "https://files.pythonhosted.org/packages/8e/18/f4f8360024ef26806831dde4ba3dad23950a4d94dde86085dde510ef1833/praisonaiagents-0.6.0-py3-none-any.whl", hash = "sha256:416c4550b6460b0866c1d59b284aacb86f23d849b38e75542f7056e530e84d80", size = 724880, upload-time = "2025-12-31T18:50:33.21Z" },
]
[[package]]