Compare commits

...
7 Commits
Author SHA1 Message Date
Mervin Praison cbe8beeca6 Release v2.3.80 2025-12-23 07:54:07 +00:00
Mervin Praison f7b66a52b7 Release v2.3.79 2025-12-23 07:27:35 +00:00
Mervin Praison 6dbf194312 Release v2.3.78 2025-12-23 06:46:15 +00:00
Mervin Praison 335e8d7e47 Release v2.3.77 2025-12-23 06:28:26 +00:00
Mervin Praison d65840e092 Release v2.3.76 2025-12-23 05:40:52 +00:00
Mervin Praison 5a13571fdd Release v2.3.75 2025-12-22 19:11:32 +00:00
Mervin Praison e3b48c2526 Release v2.3.74 2025-12-22 17:01:50 +00:00
32 changed files with 3247 additions and 399 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.3.73" \
"praisonai>=2.3.80" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.73" \
"praisonai>=2.3.80" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.73" \
"praisonai>=2.3.80" \
"praisonai[ui]" \
"praisonai[crewai]"
+53 -18
View File
@@ -5,30 +5,65 @@ This example demonstrates how to evaluate agent output accuracy
by comparing against expected outputs using LLM-as-judge.
"""
import os
from praisonaiagents import Agent
from praisonaiagents.eval import AccuracyEvaluator
# Create a simple math tutor agent
agent = Agent(
instructions="You are a math tutor. Answer math questions concisely with just the number."
)
# Check if we have an API key
has_api_key = os.getenv("OPENAI_API_KEY") is not None
# Create accuracy evaluator
evaluator = AccuracyEvaluator(
agent=agent,
input_text="What is 2 + 2?",
expected_output="4",
num_iterations=3, # Run 3 times for statistical significance
verbose=True
)
if has_api_key:
# Create a simple math tutor agent
agent = Agent(
instructions="You are a math tutor. Answer math questions concisely with just the number."
)
# Run evaluation
result = evaluator.run(print_summary=True)
# Create accuracy evaluator
evaluator = AccuracyEvaluator(
agent=agent,
input_text="What is 2 + 2?",
expected_output="4",
num_iterations=3, # Run 3 times for statistical significance
verbose=True
)
# Check results
print(f"\nAverage Score: {result.avg_score}/10")
print(f"Passed: {result.passed}")
# Run evaluation
result = evaluator.run(print_summary=True)
# Check results
print(f"\nAverage Score: {result.avg_score}/10")
print(f"Passed: {result.passed}")
else:
print("⚠️ No OPENAI_API_KEY found. Using mock evaluation...")
# Create evaluator without agent for mock testing
evaluator = AccuracyEvaluator(
func=lambda x: "4", # Mock function
input_text="What is 2 + 2?",
expected_output="4",
num_iterations=1,
verbose=True
)
# Test with mock output
try:
result = evaluator.run(print_summary=True)
print(f"\nMock Average Score: {result.avg_score}/10")
print(f"Mock Passed: {result.passed}")
except Exception as e:
print(f"Mock evaluation failed: {e}")
print("This is expected without an API key")
# You can also evaluate pre-generated outputs
print("\n--- Testing pre-generated output evaluation ---")
output = "The answer is 4"
result2 = evaluator.evaluate_output(output, print_summary=True)
if has_api_key:
try:
result2 = evaluator.evaluate_output(output, print_summary=True)
print(f"Pre-generated output score: {result2.avg_score}/10")
except Exception as e:
print(f"Pre-generated evaluation failed: {e}")
else:
print("⚠️ Skipping pre-generated evaluation (no API key)")
print("To run full evaluation, set OPENAI_API_KEY environment variable")
+67 -34
View File
@@ -5,9 +5,13 @@ This example demonstrates how to run batch evaluations
from a JSON test file.
"""
import os
import json
import tempfile
from praisonaiagents.eval import AccuracyEvaluator, CriteriaEvaluator
from praisonaiagents.eval import AccuracyEvaluator
# Check if we have an API key
has_api_key = os.getenv("OPENAI_API_KEY") is not None
# Create a test file with multiple test cases
test_cases = [
@@ -33,40 +37,69 @@ with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
print(f"Test file created: {test_file}")
print(f"Running {len(test_cases)} test cases...")
# You can use the CLI for batch evaluation:
# praisonai eval batch --agent agents.yaml --test-file tests.json --batch-type accuracy
if has_api_key:
# You can use the CLI for batch evaluation:
# praisonai eval batch --agent agents.yaml --test-file tests.json --batch-type accuracy
# Or programmatically:
from praisonaiagents import Agent
# Or programmatically:
from praisonaiagents import Agent
agent = Agent(
instructions="You are a helpful assistant. Answer questions concisely."
)
results = []
for i, test_case in enumerate(test_cases):
print(f"\nTest {i + 1}: {test_case['input']}")
evaluator = AccuracyEvaluator(
agent=agent,
input_text=test_case["input"],
expected_output=test_case["expected"],
num_iterations=1
agent = Agent(
instructions="You are a helpful assistant. Answer questions concisely."
)
result = evaluator.run()
results.append({
"input": test_case["input"],
"expected": test_case["expected"],
"score": result.avg_score,
"passed": result.passed
})
print(f" Score: {result.avg_score}/10")
print(f" Passed: {result.passed}")
# Summary
passed = sum(1 for r in results if r["passed"])
print(f"\n{'='*50}")
print(f"Summary: {passed}/{len(results)} tests passed")
print(f"Pass Rate: {passed/len(results):.1%}")
results = []
for i, test_case in enumerate(test_cases):
print(f"\nTest {i + 1}: {test_case['input']}")
evaluator = AccuracyEvaluator(
agent=agent,
input_text=test_case["input"],
expected_output=test_case["expected"],
num_iterations=1
)
result = evaluator.run()
results.append({
"input": test_case["input"],
"expected": test_case["expected"],
"score": result.avg_score,
"passed": result.passed
})
print(f" Score: {result.avg_score}/10")
print(f" Passed: {result.passed}")
# Summary
passed = sum(1 for r in results if r["passed"])
print(f"\n{'='*50}")
print(f"Summary: {passed}/{len(results)} tests passed")
print(f"Pass Rate: {passed/len(results):.1%}")
else:
print("⚠️ No OPENAI_API_KEY found. Running mock batch evaluation...")
# Mock evaluation without API key
results = []
for i, test_case in enumerate(test_cases):
print(f"\nTest {i + 1}: {test_case['input']}")
# Mock result (all would fail without API key)
mock_score = 0.0
mock_passed = False
results.append({
"input": test_case["input"],
"expected": test_case["expected"],
"score": mock_score,
"passed": mock_passed
})
print(f" Score: {mock_score}/10 (mock)")
print(f" Passed: {mock_passed} (expected without API key)")
# Summary
passed = sum(1 for r in results if r["passed"])
print(f"\n{'='*50}")
print(f"Summary: {passed}/{len(results)} tests passed (mock evaluation)")
print(f"Pass Rate: {passed/len(results):.1%}")
print("\nTo run real batch evaluation, set OPENAI_API_KEY environment variable")
+63 -43
View File
@@ -5,58 +5,78 @@ This example demonstrates how to evaluate agent outputs
against custom criteria using LLM-as-judge.
"""
import os
from praisonaiagents import Agent
from praisonaiagents.eval import CriteriaEvaluator
# Create a customer service agent
agent = Agent(
instructions="You are a friendly customer service agent. Be helpful and empathetic."
)
# Check if we have an API key
has_api_key = os.getenv("OPENAI_API_KEY") is not None
# Create criteria evaluator with numeric scoring
evaluator = CriteriaEvaluator(
criteria="Response is helpful, empathetic, and provides a clear solution",
agent=agent,
input_text="I'm frustrated because my order hasn't arrived yet.",
scoring_type="numeric", # Score 1-10
threshold=7.0, # Pass if score >= 7
num_iterations=2,
verbose=True
)
if has_api_key:
print("--- Testing Criteria Evaluation with Agent ---")
# Create a customer service agent
agent = Agent(
instructions="You are a friendly customer service agent. Be helpful and empathetic."
)
# Run evaluation
result = evaluator.run(print_summary=True)
# Create criteria evaluator with numeric scoring
evaluator = CriteriaEvaluator(
criteria="Response is helpful, empathetic, and provides a clear solution",
agent=agent,
input_text="I'm frustrated because my order hasn't arrived yet.",
scoring_type="numeric", # Score 1-10
threshold=7.0, # Pass if score >= 7
num_iterations=2,
verbose=True
)
print(f"\nAverage Score: {result.avg_score}/10")
print(f"Pass Rate: {result.pass_rate:.1%}")
print(f"Passed: {result.passed}")
# Run evaluation
result = evaluator.run(print_summary=True)
# Binary scoring example (pass/fail)
binary_evaluator = CriteriaEvaluator(
criteria="Response does not contain any offensive language",
agent=agent,
input_text="Tell me a joke",
scoring_type="binary", # Pass or Fail
verbose=True
)
print("\nNumeric Scoring Results:")
print(f" Average Score: {result.avg_score}/10")
print(f" Pass Rate: {result.pass_rate:.1%}")
print(f" Passed: {result.passed}")
binary_result = binary_evaluator.run(print_summary=True)
# Binary scoring example (pass/fail)
print("\n--- Testing Binary Scoring ---")
binary_evaluator = CriteriaEvaluator(
criteria="Response does not contain any offensive language",
agent=agent,
input_text="Tell me a joke",
scoring_type="binary", # Pass or Fail
verbose=True
)
# Evaluate pre-generated output
output = "I understand your frustration. Let me check on your order right away."
result2 = evaluator.evaluate_output(output, print_summary=True)
binary_result = binary_evaluator.run(print_summary=True)
# With on_fail callback
print("\n--- Testing Failure Callback ---")
def handle_failure(score):
print(f"ALERT: Evaluation failed with score {score.score}")
print(f"Reasoning: {score.reasoning}")
# With on_fail callback
def handle_failure(score):
print(f"ALERT: Evaluation failed with score {score.score}")
print(f"Reasoning: {score.reasoning}")
callback_evaluator = CriteriaEvaluator(
criteria="Response is professional and helpful",
agent=agent,
input_text="Help me",
on_fail=handle_failure,
threshold=8.0
)
callback_evaluator = CriteriaEvaluator(
criteria="Response is professional and helpful",
agent=agent,
input_text="Help me",
on_fail=handle_failure,
threshold=8.0
)
callback_evaluator.run()
else:
print("⚠️ No OPENAI_API_KEY found. Skipping agent-based criteria evaluation...")
callback_evaluator.run()
# Test pre-generated output evaluation (doesn't need agent)
print("\n--- Testing Pre-generated Output Evaluation ---")
if has_api_key:
output = "I understand your frustration. Let me check on your order right away."
try:
result2 = evaluator.evaluate_output(output, print_summary=True)
print(f"Pre-generated output score: {result2.avg_score}/10")
except Exception as e:
print(f"Pre-generated evaluation failed: {e}")
else:
print("⚠️ Skipping pre-generated evaluation (no API key)")
print("To run full evaluation, set OPENAI_API_KEY environment variable")
+31 -23
View File
@@ -5,35 +5,43 @@ This example demonstrates how to benchmark agent performance
by measuring runtime and memory usage across multiple iterations.
"""
import os
from praisonaiagents import Agent
from praisonaiagents.eval import PerformanceEvaluator
# Create a simple agent
agent = Agent(
instructions="You are a helpful assistant. Answer questions briefly."
)
# Check if we have an API key
has_api_key = os.getenv("OPENAI_API_KEY") is not None
# Create performance evaluator
evaluator = PerformanceEvaluator(
agent=agent,
input_text="What is the capital of France?",
num_iterations=10, # Run 10 benchmark iterations
warmup_runs=2, # 2 warmup runs before measurement
track_memory=True, # Track memory usage
verbose=True
)
if has_api_key:
print("--- Testing Agent Performance ---")
# Create a simple agent
agent = Agent(
instructions="You are a helpful assistant. Answer questions briefly."
)
# Run evaluation
result = evaluator.run(print_summary=True)
# Create performance evaluator
evaluator = PerformanceEvaluator(
agent=agent,
input_text="What is the capital of France?",
num_iterations=10, # Run 10 benchmark iterations
warmup_runs=2, # 2 warmup runs before measurement
track_memory=True, # Track memory usage
verbose=True
)
# Access detailed metrics
print(f"\nDetailed Results:")
print(f" Average Time: {result.avg_run_time:.4f}s")
print(f" Min Time: {result.min_run_time:.4f}s")
print(f" Max Time: {result.max_run_time:.4f}s")
print(f" Median Time: {result.median_run_time:.4f}s")
print(f" P95 Time: {result.p95_run_time:.4f}s")
print(f" Avg Memory: {result.avg_memory:.2f} MB")
# Run evaluation
result = evaluator.run(print_summary=True)
# Access detailed metrics
print("\nAgent Performance Results:")
print(f" Average Time: {result.avg_run_time:.4f}s")
print(f" Min Time: {result.min_run_time:.4f}s")
print(f" Max Time: {result.max_run_time:.4f}s")
print(f" Median Time: {result.median_run_time:.4f}s")
print(f" P95 Time: {result.p95_run_time:.4f}s")
print(f" Avg Memory: {result.avg_memory:.2f} MB")
else:
print("⚠️ No OPENAI_API_KEY found. Skipping agent performance test...")
# You can also benchmark any function
def my_function():
+40 -16
View File
@@ -5,6 +5,7 @@ This example demonstrates how to verify that agents call
the expected tools during execution.
"""
import os
from praisonaiagents import Agent
from praisonaiagents.eval import ReliabilityEvaluator
@@ -19,30 +20,53 @@ def calculate(expression: str) -> str:
return str(eval(expression))
# Create an agent with tools
agent = Agent(
instructions="You are a helpful assistant with access to search and calculator tools.",
tools=[search_web, calculate]
)
# Check if we have an API key
has_api_key = os.getenv("OPENAI_API_KEY") is not None
# Create reliability evaluator
if has_api_key:
print("--- Testing Agent Tool Call Reliability ---")
# Create an agent with tools
agent = Agent(
instructions="You are a helpful assistant with access to search and calculator tools.",
tools=[search_web, calculate]
)
# Create reliability evaluator
evaluator = ReliabilityEvaluator(
agent=agent,
input_text="Search for the weather in Paris and calculate 25 * 4",
expected_tools=["search_web", "calculate"], # Tools that should be called
forbidden_tools=["delete_file"], # Tools that should NOT be called
verbose=True
)
# Run evaluation
result = evaluator.run(print_summary=True)
# Check results
print("\nAgent Reliability Results:")
print(f" Passed: {result.passed}")
print(f" Pass Rate: {result.pass_rate:.1%}")
else:
print("⚠️ No OPENAI_API_KEY found. Skipping agent reliability test...")
print("Agent would fail to call tools without API key (expected behavior)")
# You can also evaluate pre-recorded tool calls (doesn't need API key)
print("\n--- Testing Pre-recorded Tool Call Evaluation ---")
# Create a mock agent for tool call evaluation
mock_agent = Agent(instructions="Mock agent")
evaluator = ReliabilityEvaluator(
agent=agent,
input_text="Search for the weather in Paris and calculate 25 * 4",
agent=mock_agent,
expected_tools=["search_web", "calculate"], # Tools that should be called
forbidden_tools=["delete_file"], # Tools that should NOT be called
verbose=True
)
# Run evaluation
result = evaluator.run(print_summary=True)
# Check results
print(f"\nPassed: {result.passed}")
print(f"Pass Rate: {result.pass_rate:.1%}")
# You can also evaluate pre-recorded tool calls
result2 = evaluator.evaluate_tool_calls(
actual_tools=["search_web", "calculate"],
print_summary=True
)
print("\nPre-recorded Tool Call Results:")
print(f" Passed: {result2.passed}")
print(f" Pass Rate: {result2.pass_rate:.1%}")
@@ -3,7 +3,7 @@ Praison AI Agents - A package for hierarchical AI agent task execution
"""
# Apply warning patch BEFORE any imports to intercept warnings at the source
from . import _warning_patch
from . import _warning_patch # noqa: F401
# Import centralized logging configuration FIRST
from . import _logging
@@ -302,6 +302,23 @@ __all__ = [
'ExpandResult'
]
# Add workflow exports
__all__.extend([
'Workflow',
'WorkflowStep',
'WorkflowContext',
'StepResult',
'Route',
'Parallel',
'Loop',
'Repeat',
'route',
'parallel',
'loop',
'repeat',
'Pipeline'
])
# Add MCP to __all__ if available
if _mcp_available:
__all__.append('MCP')
@@ -318,6 +335,14 @@ __all__.extend([
'LineRange'
])
# Add Agent Skills exports (lazy loaded)
__all__.extend([
'SkillManager',
'SkillProperties',
'SkillMetadata',
'SkillLoader'
])
# AG-UI support (optional, lazy loaded)
try:
from praisonaiagents.ui.agui import AGUI
@@ -8,9 +8,10 @@ import functools
import sys
# Apply aggressive warning filters first
warnings.filterwarnings("ignore", message=".*There is no current event loop.*", category=DeprecationWarning)
warnings.filterwarnings("ignore", message=".*Pydantic serializer warnings.*")
warnings.filterwarnings("ignore", message=".*PydanticSerializationUnexpectedValue.*")
warnings.filterwarnings("ignore", message=".*Expected 9 fields but got.*")
warnings.filterwarnings("ignore", message=".*Expected \\d+ fields but got.*")
warnings.filterwarnings("ignore", message=".*Expected `StreamingChoices` but got.*")
warnings.filterwarnings("ignore", message=".*serialized value may not be as expected.*")
warnings.filterwarnings("ignore", message=".*Use 'content=<...>' to upload raw bytes/text content.*")
@@ -27,11 +28,12 @@ SUPPRESSED_PATTERNS = [
"The `dict` method is deprecated; use `model_dump` instead",
"Pydantic serializer warnings",
"PydanticSerializationUnexpectedValue",
"Expected 9 fields but got 5 for type `Message`",
"Expected", # Catches all "Expected N fields but got M" patterns
"Expected `StreamingChoices` but got `Choices`",
"serialized value may not be as expected",
"Mixing V1 models and V2 models",
"Please upgrade `Settings` to V2"
"Please upgrade `Settings` to V2",
"There is no current event loop"
]
@functools.wraps(_original_warn)
@@ -43,7 +45,7 @@ def _patched_warn(message, category=None, stacklevel=1, source=None):
if pattern in msg_str:
return
if category == UserWarning and "pydantic" in msg_str.lower():
if category is UserWarning and "pydantic" in msg_str.lower():
return
_original_warn(message, category, stacklevel, source)
@@ -57,7 +59,7 @@ def _patched_warn_explicit(message, category, filename, lineno, module=None, reg
if pattern in msg_str:
return
if category == UserWarning and "pydantic" in msg_str.lower():
if category is UserWarning and "pydantic" in msg_str.lower():
return
if module and "pydantic" in str(module):
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.1.22"
version = "0.1.24"
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.1.22"
version = "0.1.24"
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.3.73.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.73.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.80.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.80.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
@@ -105,6 +105,7 @@ class FileSearchService:
- Fuzzy match file paths
- Cache results with TTL
- Respect .gitignore (basic)
- Support absolute paths (@/Users/...) and home paths (@~/...)
"""
def __init__(
@@ -125,9 +126,10 @@ class FileSearchService:
# Cache: query -> (results, timestamp)
self._cache: Dict[str, Tuple[List[FileSuggestion], float]] = {}
# File list cache
# File list cache per directory
self._file_list: Optional[List[str]] = None
self._file_list_time: float = 0
self._file_list_dir: Optional[str] = None
def _should_ignore(self, name: str) -> bool:
"""Check if file/dir should be ignored."""
@@ -136,20 +138,23 @@ class FileSearchService:
return True
return False
def _get_file_list(self) -> List[str]:
def _get_file_list(self, search_dir: str = None) -> List[str]:
"""Get list of all files (cached)."""
now = time.time()
target_dir = search_dir or self._root
# Return cached if fresh
if self._file_list is not None and (now - self._file_list_time) < self._cache_ttl:
# Return cached if fresh and same directory
if (self._file_list is not None and
self._file_list_dir == target_dir and
(now - self._file_list_time) < self._cache_ttl):
return self._file_list
files = []
try:
for root, dirs, filenames in os.walk(self._root):
for root, dirs, filenames in os.walk(target_dir):
# Calculate depth
rel_root = os.path.relpath(root, self._root)
rel_root = os.path.relpath(root, target_dir)
depth = 0 if rel_root == '.' else rel_root.count(os.sep) + 1
if depth > self._max_depth:
@@ -161,19 +166,20 @@ class FileSearchService:
# Add directories
for d in dirs:
rel_path = os.path.relpath(os.path.join(root, d), self._root)
rel_path = os.path.relpath(os.path.join(root, d), target_dir)
files.append(rel_path + os.sep)
# Add files
for f in filenames:
if not self._should_ignore(f):
rel_path = os.path.relpath(os.path.join(root, f), self._root)
rel_path = os.path.relpath(os.path.join(root, f), target_dir)
files.append(rel_path)
except OSError:
pass
self._file_list = files
self._file_list_time = now
self._file_list_dir = target_dir
return files
def _fuzzy_match(self, text: str, query: str) -> int:
@@ -221,6 +227,11 @@ class FileSearchService:
"""
Search for files matching query.
Supports:
- Relative paths: @file.txt, @src/main.py
- Home paths: @~/Documents, @~/.config
- Absolute paths: @/Users/name/file.txt, @/etc/hosts
Args:
query: Search query (fuzzy matched)
max_results: Maximum results to return
@@ -228,8 +239,51 @@ class FileSearchService:
Returns:
List of FileSuggestion sorted by score
"""
# Determine search directory based on query
search_dir = self._root
search_query = query
prefix = ""
# Handle home directory paths (@~/)
if query.startswith('~'):
expanded = os.path.expanduser(query)
# Find the directory part and filename part
if os.path.isdir(expanded):
search_dir = expanded
search_query = ""
prefix = query.rstrip('/') + '/'
else:
parent = os.path.dirname(expanded)
if os.path.isdir(parent):
search_dir = parent
search_query = os.path.basename(expanded)
prefix = os.path.dirname(query).rstrip('/') + '/'
else:
# Just expand ~ and search from home
search_dir = os.path.expanduser('~')
search_query = query[2:] if query.startswith('~/') else query[1:]
prefix = "~/"
# Handle absolute paths (@/Users/..., @/etc/...)
elif query.startswith('/'):
if os.path.isdir(query):
search_dir = query
search_query = ""
prefix = query.rstrip('/') + '/'
else:
parent = os.path.dirname(query)
if os.path.isdir(parent):
search_dir = parent
search_query = os.path.basename(query)
prefix = parent.rstrip('/') + '/'
else:
# Search from root with the query
search_dir = '/'
search_query = query[1:]
prefix = "/"
# Check cache
cache_key = f"{query}:{max_results}"
cache_key = f"{search_dir}:{search_query}:{max_results}"
now = time.time()
if cache_key in self._cache:
@@ -237,25 +291,26 @@ class FileSearchService:
if (now - cached_time) < self._cache_ttl:
return results
# Get all files
all_files = self._get_file_list()
# Get all files from the appropriate directory
all_files = self._get_file_list(search_dir)
# Score and filter
scored: List[Tuple[int, str]] = []
for path in all_files:
score = self._fuzzy_match(path, query)
score = self._fuzzy_match(path, search_query)
if score > 0:
scored.append((score, path))
# Sort by score descending
scored.sort(key=lambda x: (-x[0], x[1]))
# Build results
# Build results with proper prefix
results = []
for score, path in scored[:max_results]:
file_type = "directory" if path.endswith(os.sep) else "file"
display_path = prefix + path if prefix else path
results.append(FileSuggestion(
path=path,
path=display_path,
file_type=file_type,
score=score
))
@@ -45,8 +45,10 @@ class AutoMemoryHandler(FlagHandler):
def _get_auto_memory(self, user_id: str = None):
"""Get AutoMemory instance lazily."""
try:
from praisonaiagents.memory import AutoMemory
return AutoMemory(user_id=user_id or "default")
from praisonaiagents.memory import AutoMemory, FileMemory
# AutoMemory requires a FileMemory instance
memory = FileMemory(user_id=user_id or "default")
return AutoMemory(memory=memory)
except ImportError:
self.print_status(
"AutoMemory requires praisonaiagents. Install with: pip install praisonaiagents",
@@ -71,33 +73,36 @@ class AutoMemoryHandler(FlagHandler):
config['auto_memory_user_id'] = flag_value.get('user_id', 'default')
return config
def extract_memories(self, text: str, user_id: str = None) -> Dict[str, Any]:
def extract_memories(self, text: str, user_id: str = None, user_message: str = None) -> list:
"""
Extract memories from text.
Extract and store memories from text.
Args:
text: Text to extract memories from
text: Assistant response text to extract memories from
user_id: User ID for memory isolation
user_message: Original user message (for context)
Returns:
Dictionary of extracted memories
List of extracted memories
"""
auto_memory = self._get_auto_memory(user_id)
if not auto_memory:
return {}
return []
try:
if hasattr(auto_memory, 'extract'):
memories = auto_memory.extract(text)
self.print_status(f"🧠 Extracted {len(memories)} memories", "success")
return memories
elif hasattr(auto_memory, 'process'):
result = auto_memory.process(text)
return result
# Use process_interaction which extracts AND stores memories
memories = auto_memory.process_interaction(
user_message=user_message or text,
assistant_response=text if user_message else None,
store=True # Store the memories
)
if memories:
self.print_status(f"🧠 Extracted and stored {len(memories)} memories", "success")
return memories
except Exception as e:
self.log(f"Memory extraction failed: {e}", "error")
return {}
return []
def store_memory(self, content: str, user_id: str = None, importance: float = 0.5) -> bool:
"""
@@ -127,13 +132,14 @@ class AutoMemoryHandler(FlagHandler):
self.log(f"Memory storage failed: {e}", "error")
return False
def post_process_result(self, result: Any, flag_value: Any) -> Any:
def post_process_result(self, result: Any, flag_value: Any, user_message: str = None) -> Any:
"""
Post-process result to extract and store memories.
Args:
result: Agent output
flag_value: Boolean or dict with configuration
user_message: Original user message for context
Returns:
Original result (memories are stored)
@@ -144,15 +150,18 @@ class AutoMemoryHandler(FlagHandler):
user_id = None
if isinstance(flag_value, dict):
user_id = flag_value.get('user_id')
user_message = flag_value.get('user_message', user_message)
# Extract memories from result
# Extract and store memories from result
text = str(result)
memories = self.extract_memories(text, user_id)
memories = self.extract_memories(text, user_id, user_message=user_message)
if memories:
self.print_status("\n🧠 Auto-extracted Memories:", "info")
for key, value in memories.items():
self.print_status(f" • {key}: {value[:50]}...", "info")
for mem in memories:
mem_type = mem.get('type', 'unknown')
content = mem.get('content', str(mem))[:50]
self.print_status(f" • {mem_type}: {content}...", "info")
return result
+76 -10
View File
@@ -16,7 +16,7 @@ logger = logging.getLogger(__name__)
COMPARE_MODES = {
"basic": {},
"tools": {"tools": "internet_search"},
"tools": {"tools": "calculator,internet_search"},
"research": {"research": True},
"planning": {"planning": True},
"memory": {"memory": True},
@@ -177,12 +177,13 @@ def save_compare_result(result: CompareResult, path: str) -> bool:
return False
def format_comparison_table(result: CompareResult) -> str:
def format_comparison_table(result: CompareResult, show_responses: bool = True) -> str:
"""
Format comparison result as a table string.
Format comparison result as a table string with optional response display.
Args:
result: CompareResult to format
show_responses: Whether to show actual responses from each mode
Returns:
Formatted table string
@@ -190,10 +191,12 @@ def format_comparison_table(result: CompareResult) -> str:
try:
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from io import StringIO
console = Console(file=StringIO(), force_terminal=True)
# Main comparison table
table = Table(title=f"Comparison: {result.query[:50]}...")
table.add_column("Mode", style="cyan")
table.add_column("Time (ms)", style="green")
@@ -224,6 +227,26 @@ def format_comparison_table(result: CompareResult) -> str:
)
console.print(table)
# Show responses if enabled
if show_responses:
console.print("\n[bold cyan]📝 Responses:[/bold cyan]")
for c in result.comparisons:
if c.error is None:
# Truncate long responses
response_preview = c.output[:500] + "..." if len(c.output) > 500 else c.output
console.print(Panel(
response_preview,
title=f"[bold]{c.mode}[/bold] ({c.execution_time_ms:.0f}ms)",
border_style="cyan"
))
else:
console.print(Panel(
f"[red]Error: {c.error}[/red]",
title=f"[bold]{c.mode}[/bold]",
border_style="red"
))
return console.file.getvalue()
except ImportError:
lines = [f"Comparison: {result.query}"]
@@ -231,6 +254,8 @@ def format_comparison_table(result: CompareResult) -> str:
for c in result.comparisons:
status = "OK" if c.error is None else f"ERROR: {c.error}"
lines.append(f"{c.mode}: {c.execution_time_ms:.1f}ms - {status}")
if show_responses and c.error is None:
lines.append(f" Response: {c.output[:200]}...")
return "\n".join(lines)
@@ -407,13 +432,7 @@ class CompareHandler:
from praisonaiagents.tools import internet_search
tools.append(internet_search)
elif name == "calculator":
def calculator(expression: str) -> str:
"""Evaluate a mathematical expression."""
try:
return str(eval(expression))
except Exception as e:
return f"Error: {e}"
tools.append(calculator)
tools.append(_calculator_tool)
except ImportError:
logger.warning(f"Could not load tool: {name}")
@@ -466,3 +485,50 @@ class CompareHandler:
print(f"[red]Failed to save results to: {output_path}[/red]")
return result
def _calculator_tool(expression: str) -> str:
"""
Evaluate a mathematical expression safely.
Args:
expression: Mathematical expression to evaluate (e.g., "2+2", "15*3", "100/4")
Returns:
The result of the calculation as a string
"""
import ast
import operator
# Safe operators
operators = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.USub: operator.neg,
ast.UAdd: operator.pos,
ast.Mod: operator.mod,
ast.FloorDiv: operator.floordiv,
}
def _eval(node):
if isinstance(node, ast.Constant):
return node.value
elif isinstance(node, ast.BinOp):
left = _eval(node.left)
right = _eval(node.right)
return operators[type(node.op)](left, right)
elif isinstance(node, ast.UnaryOp):
operand = _eval(node.operand)
return operators[type(node.op)](operand)
else:
raise ValueError(f"Unsupported operation: {type(node)}")
try:
tree = ast.parse(expression, mode='eval')
result = _eval(tree.body)
return str(result)
except Exception as e:
return f"Error evaluating '{expression}': {e}"
+109 -20
View File
@@ -26,42 +26,68 @@ class EvalHandler:
def run_accuracy(
self,
agent_file: str,
input_text: str,
expected_output: str,
agent_file: Optional[str] = None,
input_text: str = "",
expected_output: str = "",
iterations: int = 1,
model: Optional[str] = None,
output_file: Optional[str] = None
output_file: Optional[str] = None,
prompt: Optional[str] = None,
llm: Optional[str] = None
) -> Dict[str, Any]:
"""
Run accuracy evaluation on an agent.
Args:
agent_file: Path to agents.yaml file
agent_file: Path to agents.yaml file (optional if prompt is provided)
input_text: Input to provide to the agent
expected_output: Expected output to compare against
iterations: Number of evaluation iterations
model: LLM model for judging
output_file: Path to save results
prompt: Direct prompt (alternative to agent_file)
llm: LLM model for the agent (when using prompt)
Returns:
Evaluation result dictionary
"""
try:
from praisonaiagents.eval import AccuracyEvaluator
from praisonai.agents_generator import AgentsGenerator
from praisonaiagents import Agent
except ImportError as e:
logger.error(f"Failed to import evaluation modules: {e}")
return {"error": str(e)}
try:
generator = AgentsGenerator(agent_file)
agents = generator.generate_agents()
if not agents:
return {"error": "No agents found in configuration"}
agent = agents[0] if isinstance(agents, list) else agents
# Create agent either from file or from prompt
if prompt:
# Direct prompt mode - create agent on the fly
agent = Agent(
name="EvalAgent",
role="Assistant",
goal="Complete the given task",
backstory="You are a helpful assistant.",
llm=llm or model or "gpt-4o-mini",
verbose=False
)
# Use prompt as input if input_text not provided
if not input_text:
input_text = prompt
elif agent_file:
# Load from agents.yaml
try:
from praisonai.agents_generator import AgentsGenerator
generator = AgentsGenerator(agent_file)
agents = generator.generate_agents()
if not agents:
return {"error": "No agents found in configuration"}
agent = agents[0] if isinstance(agents, list) else agents
except Exception as e:
return {"error": f"Failed to load agents from {agent_file}: {e}"}
else:
return {"error": "Either --agent or --prompt must be provided"}
evaluator = AccuracyEvaluator(
agent=agent,
@@ -321,18 +347,49 @@ class EvalHandler:
return batch_result
def handle_eval_command(args) -> None:
def handle_eval_command(args) -> int:
"""
Handle the eval CLI command.
Args:
args: Parsed command line arguments
args: Command line arguments (list or parsed namespace)
Returns:
Exit code
"""
import argparse
# If args is a list, parse it first
if isinstance(args, list):
parser = argparse.ArgumentParser(prog="praisonai eval")
subparsers = parser.add_subparsers(dest='eval_type')
add_eval_parser_subcommands(subparsers)
try:
args = parser.parse_args(args)
except SystemExit:
return 1
if not args.eval_type:
parser.print_help()
print("\n[bold]Examples:[/bold]")
print(" praisonai eval accuracy --prompt \"What is 2+2?\" --expected \"4\"")
print(" praisonai eval performance --agent agents.yaml --input \"Hello\"")
return 0
handler = EvalHandler(verbose=getattr(args, 'verbose', False))
eval_type = getattr(args, 'eval_type', 'accuracy')
agent_file = getattr(args, 'agent', 'agents.yaml')
agent_file = getattr(args, 'agent', None)
output_file = getattr(args, 'output', None)
prompt = getattr(args, 'prompt', None)
llm = getattr(args, 'llm', None)
# If no agent file and no prompt, check if agents.yaml exists
if not agent_file and not prompt:
import os
if os.path.exists('agents.yaml'):
agent_file = 'agents.yaml'
if eval_type == 'accuracy':
result = handler.run_accuracy(
@@ -341,7 +398,9 @@ def handle_eval_command(args) -> None:
expected_output=getattr(args, 'expected', ''),
iterations=getattr(args, 'iterations', 1),
model=getattr(args, 'model', None),
output_file=output_file
output_file=output_file,
prompt=prompt,
llm=llm
)
elif eval_type == 'performance':
result = handler.run_performance(
@@ -384,12 +443,40 @@ def handle_eval_command(args) -> None:
)
else:
print(f"Unknown evaluation type: {eval_type}")
return
return 1
if 'error' in result:
print(f"Error: {result['error']}")
return 1
elif not getattr(args, 'quiet', False):
print(json.dumps(result, indent=2))
return 0
def add_eval_parser_subcommands(subparsers) -> None:
"""Add eval subcommand parsers to an existing subparsers object."""
accuracy_parser = subparsers.add_parser('accuracy', help='Run accuracy evaluation')
accuracy_parser.add_argument('--agent', '-a', help='Agent config file (optional if --prompt used)')
accuracy_parser.add_argument('--prompt', '-p', type=str, help='Direct prompt (alternative to --agent)')
accuracy_parser.add_argument('--llm', help='LLM model for agent (when using --prompt)')
accuracy_parser.add_argument('--input', '-i', help='Input text (defaults to --prompt if not provided)')
accuracy_parser.add_argument('--expected', '-e', required=True, help='Expected output')
accuracy_parser.add_argument('--iterations', '-n', type=int, default=1, help='Number of iterations')
accuracy_parser.add_argument('--model', '-m', help='Judge model')
accuracy_parser.add_argument('--output', '-o', help='Output file')
accuracy_parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
accuracy_parser.add_argument('--quiet', '-q', action='store_true', help='Suppress JSON output')
perf_parser = subparsers.add_parser('performance', help='Run performance evaluation')
perf_parser.add_argument('--agent', '-a', default='agents.yaml', help='Agent config file')
perf_parser.add_argument('--input', '-i', default='Hello', help='Input text')
perf_parser.add_argument('--iterations', '-n', type=int, default=10, help='Number of iterations')
perf_parser.add_argument('--warmup', '-w', type=int, default=2, help='Warmup runs')
perf_parser.add_argument('--memory', action='store_true', default=True, help='Track memory')
perf_parser.add_argument('--output', '-o', help='Output file')
perf_parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
perf_parser.add_argument('--quiet', '-q', action='store_true', help='Suppress JSON output')
def add_eval_parser(subparsers) -> None:
@@ -410,8 +497,10 @@ def add_eval_parser(subparsers) -> None:
'accuracy',
help='Run accuracy evaluation'
)
accuracy_parser.add_argument('--agent', '-a', default='agents.yaml', help='Agent config file')
accuracy_parser.add_argument('--input', '-i', required=True, help='Input text')
accuracy_parser.add_argument('--agent', '-a', help='Agent config file (optional if --prompt used)')
accuracy_parser.add_argument('--prompt', '-p', help='Direct prompt (alternative to --agent)')
accuracy_parser.add_argument('--llm', help='LLM model for agent (when using --prompt)')
accuracy_parser.add_argument('--input', '-i', help='Input text (defaults to --prompt if not provided)')
accuracy_parser.add_argument('--expected', '-e', required=True, help='Expected output')
accuracy_parser.add_argument('--iterations', '-n', type=int, default=1, help='Number of iterations')
accuracy_parser.add_argument('--model', '-m', help='Judge model')
@@ -3,12 +3,21 @@ Fast Context Handler for CLI.
Provides codebase search capability using FastContext.
Usage: praisonai "Find authentication code" --fast-context ./src
Fast Context Strategy:
1. Extract search keywords from natural language query using LLM
2. Provide folder tree context to help understand codebase structure
3. Use both file name search AND content search (ripgrep-style)
4. Return relevant code snippets for context
"""
import os
import logging
from typing import Any, Dict, Tuple, List
from .base import FlagHandler
logger = logging.getLogger(__name__)
class FastContextHandler(FlagHandler):
"""
@@ -61,14 +70,208 @@ class FastContextHandler(FlagHandler):
return True, ""
def _count_files_quick(self, path: str, max_count: int = 1000) -> int:
"""Quickly count files in directory, stopping at max_count."""
count = 0
ignore_dirs = {'.git', '__pycache__', 'node_modules', '.venv', 'venv', '.cache', 'dist', 'build'}
for root, dirs, files in os.walk(path):
dirs[:] = [d for d in dirs if d not in ignore_dirs and not d.startswith('.')]
count += len(files)
if count >= max_count:
return count
return count
def _get_folder_tree(self, path: str, max_depth: int = 2) -> str:
"""
Get a compact folder tree structure for context.
Optimized for large directories:
- Limits depth to max_depth (default 2)
- Shows max 10 files per directory
- Truncates total output to 50 lines
- Skips common non-code directories
"""
# Quick check for very large directories
file_count = self._count_files_quick(path, 500)
if file_count >= 500:
# For large directories, reduce depth and show summary
max_depth = 1
logger.debug(f"Large directory detected ({file_count}+ files), reducing tree depth")
tree_lines = []
ignore_dirs = {'.git', '__pycache__', 'node_modules', '.venv', 'venv', '.cache', 'dist', 'build', 'coverage', '.pytest_cache', '.mypy_cache', 'eggs', '*.egg-info'}
ignore_exts = {'.pyc', '.pyo', '.so', '.o', '.a', '.dylib', '.class', '.jar'}
total_files = 0
max_total_files = 100 # Stop after seeing this many files
def walk_tree(current_path, prefix="", depth=0):
nonlocal total_files
if depth > max_depth or total_files >= max_total_files:
return
try:
entries = sorted(os.listdir(current_path))
except PermissionError:
return
dirs = []
files = []
for entry in entries:
full_path = os.path.join(current_path, entry)
if os.path.isdir(full_path):
if entry not in ignore_dirs and not entry.startswith('.'):
dirs.append(entry)
else:
ext = os.path.splitext(entry)[1]
if ext not in ignore_exts and not entry.startswith('.'):
files.append(entry)
# Limit directories shown at each level
dirs_to_show = dirs[:15]
for i, d in enumerate(dirs_to_show):
is_last = (i == len(dirs_to_show) - 1) and not files
tree_lines.append(f"{prefix}{'└── ' if is_last else '├── '}{d}/")
walk_tree(os.path.join(current_path, d),
prefix + (' ' if is_last else '│ '), depth + 1)
if len(dirs) > 15:
tree_lines.append(f"{prefix} ... and {len(dirs) - 15} more directories")
# Limit files shown
files_to_show = files[:10]
total_files += len(files)
for i, f in enumerate(files_to_show):
is_last = i == len(files_to_show) - 1
tree_lines.append(f"{prefix}{'└── ' if is_last else '├── '}{f}")
if len(files) > 10:
tree_lines.append(f"{prefix} ... and {len(files) - 10} more files")
tree_lines.append(os.path.basename(path) + "/")
walk_tree(path)
return "\n".join(tree_lines[:50]) # Limit total lines
def _extract_keywords(self, query: str, folder_tree: str) -> List[str]:
"""Extract search keywords from natural language query using LLM."""
try:
from praisonaiagents import Agent
extractor = Agent(
name="KeywordExtractor",
role="Search Query Analyzer",
goal="Extract file search keywords from natural language",
backstory="Expert at understanding code search queries",
llm="gpt-4o-mini",
verbose=False
)
prompt = f"""Given this folder structure:
{folder_tree}
And this search query: "{query}"
Extract 1-5 specific keywords/patterns to search for in file names and content.
Return ONLY the keywords, one per line, no explanations.
Focus on: file names, function names, class names, variable names that might match.
If the query mentions a folder name that exists, include files from that folder."""
response = extractor.chat(prompt, stream=False)
keywords = [k.strip() for k in str(response).strip().split('\n') if k.strip()]
logger.debug(f"Extracted keywords: {keywords}")
return keywords[:5] # Limit to 5 keywords
except Exception as e:
logger.debug(f"Keyword extraction failed: {e}")
# Fallback: extract words from query
return [w for w in query.split() if len(w) > 2][:3]
def _search_files_by_name(self, path: str, keywords: List[str]) -> List[Dict[str, Any]]:
"""Search for files matching keywords in their names."""
matches = []
path = os.path.abspath(path)
ignore_dirs = {'.git', '__pycache__', 'node_modules', '.venv', 'venv'}
for root, dirs, files in os.walk(path):
dirs[:] = [d for d in dirs if d not in ignore_dirs]
for filename in files:
filepath = os.path.join(root, filename)
rel_path = os.path.relpath(filepath, path)
# Check if any keyword matches filename or path
for kw in keywords:
kw_lower = kw.lower()
if kw_lower in filename.lower() or kw_lower in rel_path.lower():
matches.append({
'file': rel_path,
'lines': [(1, 50)], # First 50 lines
'relevance': 1.0,
'match_type': 'filename'
})
break
return matches[:20] # Limit results
def _search_files_by_content(self, path: str, keywords: List[str]) -> List[Dict[str, Any]]:
"""Search for files containing keywords in content."""
matches = []
path = os.path.abspath(path)
ignore_dirs = {'.git', '__pycache__', 'node_modules', '.venv', 'venv'}
code_exts = {'.py', '.js', '.ts', '.jsx', '.tsx', '.java', '.go', '.rs', '.rb', '.sh', '.yaml', '.yml', '.json', '.md', '.txt'}
for root, dirs, files in os.walk(path):
dirs[:] = [d for d in dirs if d not in ignore_dirs]
for filename in files:
ext = os.path.splitext(filename)[1]
if ext not in code_exts:
continue
filepath = os.path.join(root, filename)
rel_path = os.path.relpath(filepath, path)
try:
with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
content = f.read(50000) # Read first 50KB
content_lower = content.lower()
for kw in keywords:
if kw.lower() in content_lower:
# Find line numbers with matches
lines = content.split('\n')
match_lines = []
for i, line in enumerate(lines[:200], 1): # Check first 200 lines
if kw.lower() in line.lower():
match_lines.append((max(1, i-2), min(len(lines), i+2)))
if match_lines:
matches.append({
'file': rel_path,
'lines': match_lines[:5], # Limit line ranges
'relevance': 0.8,
'match_type': 'content'
})
break
except Exception:
continue
return matches[:15] # Limit results
def search_context(self, query: str, path: str, **kwargs) -> List[Dict[str, Any]]:
"""
Search for relevant context in the codebase.
Uses a smart multi-step approach:
1. Get folder tree structure
2. Use LLM to extract search keywords from query
3. Search by filename AND content
4. Return combined results
Args:
query: Search query
query: Natural language search query
path: Path to search
**kwargs: Additional search options
- use_llm_keywords: Use LLM to extract keywords (default: True)
Returns:
List of matching results
@@ -84,32 +287,50 @@ class FastContextHandler(FlagHandler):
return []
try:
from praisonaiagents import FastContext
use_llm_keywords = kwargs.get('use_llm_keywords', True)
fc = FastContext(
path=path,
model=kwargs.get('model', 'gpt-4o-mini'),
max_turns=kwargs.get('max_turns', 4),
parallelism=kwargs.get('parallelism', 8),
timeout=kwargs.get('timeout', 30.0)
)
# Step 1: Get folder tree for context
logger.debug(f"Getting folder tree for: {path}")
folder_tree = self._get_folder_tree(path)
logger.debug(f"Folder tree:\n{folder_tree}")
result = fc.search(query)
# Step 2: Extract keywords
if use_llm_keywords:
logger.debug(f"Extracting keywords from query: {query}")
keywords = self._extract_keywords(query, folder_tree)
else:
keywords = [w for w in query.split() if len(w) > 2][:3]
# Convert to list of dicts
logger.debug(f"Search keywords: {keywords}")
if not keywords:
self.print_status("Could not extract search keywords", "warning")
return []
# Step 3: Search by filename
logger.debug("Searching by filename...")
filename_matches = self._search_files_by_name(path, keywords)
logger.debug(f"Found {len(filename_matches)} filename matches")
# Step 4: Search by content
logger.debug("Searching by content...")
content_matches = self._search_files_by_content(path, keywords)
logger.debug(f"Found {len(content_matches)} content matches")
# Combine and deduplicate
seen_files = set()
matches = []
if hasattr(result, 'files'):
for file_match in result.files:
matches.append({
'file': file_match.path,
'lines': [(lr.start, lr.end) for lr in file_match.line_ranges],
'relevance': getattr(file_match, 'relevance', 1.0)
})
for m in filename_matches + content_matches:
if m['file'] not in seen_files:
seen_files.add(m['file'])
matches.append(m)
self.print_status(f"🔍 Found {len(matches)} relevant files", "success")
return matches
except Exception as e:
logger.error(f"FastContext search error: {e}")
self.log(f"FastContext search error: {e}", "error")
return []
@@ -75,7 +75,7 @@ class GuardrailHandler(FlagHandler):
Args:
description: Natural language description of validation criteria
llm: Optional LLM instance for validation
llm: Optional LLM instance for validation (defaults to gpt-4o-mini)
Returns:
LLMGuardrail instance or None if unavailable
@@ -87,6 +87,12 @@ class GuardrailHandler(FlagHandler):
)
return None
# Provide default LLM if none specified
if llm is None:
import os
# Try to get from environment or use default
llm = os.environ.get('OPENAI_MODEL_NAME', 'gpt-4o-mini')
return LLMGuardrail(description=description, llm=llm)
def validate_output(self, output: str, description: str, llm: Any = None) -> Tuple[bool, str]:
+35 -16
View File
@@ -72,11 +72,11 @@ class HandoffHandler(FlagHandler):
self.print_status(msg, "error")
return []
from praisonaiagents import Agent, Handoff
from praisonaiagents import Agent, handoff
agents = []
# Create agents for each role
# First pass: Create all agents without handoffs
for i, name in enumerate(agent_names):
agent_config = {
"name": name.title().replace("_", " "),
@@ -91,26 +91,45 @@ class HandoffHandler(FlagHandler):
agent = Agent(**agent_config)
agents.append(agent)
# Set up handoffs between agents
for i, agent in enumerate(agents[:-1]):
next_agent = agents[i + 1]
handoff = Handoff(
target=next_agent,
name=f"handoff_to_{agent_names[i + 1]}",
description=f"Hand off task to {agent_names[i + 1]} for further processing"
)
# Second pass: Set up handoffs by recreating agents with handoff config
final_agents = []
for i, name in enumerate(agent_names):
agent_config = {
"name": name.title().replace("_", " "),
"role": name.title().replace("_", " "),
"goal": f"Complete tasks as {name}",
"backstory": f"Expert {name} with specialized skills"
}
# Add handoff to agent
if not hasattr(agent, 'handoffs') or agent.handoffs is None:
agent.handoffs = []
agent.handoffs.append(handoff)
if llm:
agent_config["llm"] = llm
# Add handoff to next agent if not the last agent
if i < len(agent_names) - 1:
next_agent = agents[i + 1]
next_agent_name = agent_names[i + 1]
# Pass handoffs via constructor - this properly registers them as tools
agent_config["handoffs"] = [
handoff(
agent=next_agent,
tool_name_override=f"handoff_to_{next_agent_name}",
tool_description_override=f"Hand off the task to {next_agent_name}. Call this after completing your part."
)
]
# Add instructions to use handoff
agent_config["instructions"] = f"After completing your task, you MUST use the handoff_to_{next_agent_name} tool to pass your work to the next agent."
final_agent = Agent(**agent_config)
final_agents.append(final_agent)
self.print_status(f"🤝 Created {len(agents)} agents with handoff chain", "success")
self.print_status(f"🤝 Created {len(final_agents)} agents with handoff chain", "success")
for i, name in enumerate(agent_names):
arrow = " → " if i < len(agent_names) - 1 else ""
self.print_status(f" {name}{arrow}", "info")
return agents
return final_agents
def apply_to_agent_config(self, config: Dict[str, Any], flag_value: Any) -> Dict[str, Any]:
"""
+257 -88
View File
@@ -1,32 +1,52 @@
"""
Image Handler for CLI.
Provides image processing capabilities using ImageAgent.
Usage: praisonai "Describe this image" --image path/to/image.png
Provides two distinct image capabilities:
1. Image Description (Vision) - Analyze existing images using vision-capable LLMs
Usage: praisonai "Describe this image" --image path/to/image.png
2. Image Generation - Generate new images from text prompts
Usage: praisonai "A sunset over mountains" --image-generate
"""
import os
from typing import Any, Dict, Tuple, List
import base64
import logging
from typing import Any, Dict, Tuple, List, Optional
from .base import FlagHandler
# Supported image extensions
SUPPORTED_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp', '.svg'}
logger = logging.getLogger(__name__)
# Supported image extensions for vision/description
SUPPORTED_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp', '.svg', '.tiff'}
# Default vision-capable models (gpt-4o has better vision support than gpt-4o-mini)
DEFAULT_VISION_MODEL = "gpt-4o"
# Default image generation models
DEFAULT_IMAGE_GEN_MODEL = "dall-e-3"
SUPPORTED_IMAGE_GEN_MODELS = [
'gpt-image-1', 'gpt-image-1-mini', 'gpt-image-1.5',
'chatgpt-image-latest', 'dall-e-2', 'dall-e-3'
]
class ImageHandler(FlagHandler):
"""
Handler for --image flag.
Handler for --image flag (Image Description/Vision).
Processes images using ImageAgent for vision-based tasks.
Describes/analyzes existing images using vision-capable LLMs like GPT-4o.
This is for understanding what's IN an image, not generating new images.
Example:
praisonai "Describe this image" --image photo.png
praisonai "What's in this image?" --image /path/to/image.jpg
praisonai "Count the objects" --image image1.png,image2.png
"""
@property
def feature_name(self) -> str:
return "image"
return "image_describe"
@property
def flag_name(self) -> str:
@@ -34,17 +54,15 @@ class ImageHandler(FlagHandler):
@property
def flag_help(self) -> str:
return "Path to image file for vision-based tasks"
return "Path to image file for vision-based description/analysis"
def check_dependencies(self) -> Tuple[bool, str]:
"""Check if ImageAgent is available."""
"""Check if litellm is available for vision."""
try:
import importlib.util
if importlib.util.find_spec("praisonaiagents") is not None:
return True, ""
return False, "praisonaiagents not installed. Install with: pip install praisonaiagents"
import litellm
return True, ""
except ImportError:
return False, "praisonaiagents not installed. Install with: pip install praisonaiagents"
return False, "litellm not installed. Install with: pip install litellm"
def validate_image_path(self, path: str) -> Tuple[bool, str]:
"""
@@ -71,90 +89,96 @@ class ImageHandler(FlagHandler):
return True, ""
def _encode_image_to_base64(self, image_path: str) -> str:
"""Encode a local image file to base64."""
with open(image_path, 'rb') as f:
return base64.b64encode(f.read()).decode('utf-8')
def _get_image_mime_type(self, image_path: str) -> str:
"""Get MIME type based on file extension."""
ext = os.path.splitext(image_path)[1].lower()
mime_types = {
'.png': 'image/png',
'.jpg': 'image/jpeg',
'.jpeg': 'image/jpeg',
'.gif': 'image/gif',
'.webp': 'image/webp',
'.bmp': 'image/bmp',
'.svg': 'image/svg+xml',
'.tiff': 'image/tiff',
}
return mime_types.get(ext, 'image/png')
def _build_image_content(self, image_path: str) -> Dict[str, Any]:
"""Build image content for vision API."""
if image_path.startswith(('http://', 'https://')):
return {
"type": "image_url",
"image_url": {"url": image_path}
}
else:
mime_type = self._get_image_mime_type(image_path)
base64_data = self._encode_image_to_base64(image_path)
return {
"type": "image_url",
"image_url": {"url": f"data:{mime_type};base64,{base64_data}"}
}
def apply_to_agent_config(self, config: Dict[str, Any], flag_value: Any) -> Dict[str, Any]:
"""
Apply image configuration to agent config.
Args:
config: Agent configuration dictionary
flag_value: Image path or list of paths
Returns:
Modified configuration with image settings
"""
"""Apply image configuration to agent config."""
if flag_value:
# Store image path(s) for use with ImageAgent
if isinstance(flag_value, str):
config["images"] = [flag_value]
elif isinstance(flag_value, list):
config["images"] = flag_value
config["use_image_agent"] = True
config["use_vision"] = True
return config
def create_image_agent(self, llm: str = None, **kwargs) -> Any:
def describe_image(self, prompt: str, image_path: str, llm: str = None) -> str:
"""
Create an ImageAgent instance.
Args:
llm: Optional LLM model name
**kwargs: Additional agent configuration
Returns:
ImageAgent instance or None if unavailable
"""
available, msg = self.check_dependencies()
if not available:
self.print_status(msg, "error")
return None
from praisonaiagents import ImageAgent
agent_config = {
"name": "ImageAnalyzer",
"role": "Image Analyst",
"goal": "Analyze and describe images accurately",
"backstory": "Expert at understanding visual content"
}
if llm:
agent_config["llm"] = llm
agent_config.update(kwargs)
return ImageAgent(**agent_config)
def process_image(self, prompt: str, image_path: str, llm: str = None) -> str:
"""
Process an image with the given prompt.
Describe an image using vision-capable LLM.
Args:
prompt: The prompt/question about the image
image_path: Path to the image file
llm: Optional LLM model name
llm: Optional LLM model name (must be vision-capable)
Returns:
Agent's response about the image
Description of the image
"""
# Validate image path
valid, msg = self.validate_image_path(image_path)
if not valid:
return f"Error: {msg}"
agent = self.create_image_agent(llm=llm)
if not agent:
return "Error: Could not create ImageAgent"
available, dep_msg = self.check_dependencies()
if not available:
return f"Error: {dep_msg}"
import litellm
model = llm or DEFAULT_VISION_MODEL
try:
# ImageAgent expects images as a list
result = agent.chat(prompt, images=[image_path])
return result
# Build message with image
content = [
{"type": "text", "text": prompt},
self._build_image_content(image_path)
]
response = litellm.completion(
model=model,
messages=[{"role": "user", "content": content}]
)
return response.choices[0].message.content
except Exception as e:
self.log(f"Image processing error: {e}", "error")
return f"Error processing image: {e}"
logger.error(f"Error describing image: {e}")
return f"Error describing image: {e}"
def process_multiple_images(self, prompt: str, image_paths: List[str], llm: str = None) -> str:
def describe_multiple_images(self, prompt: str, image_paths: List[str], llm: str = None) -> str:
"""
Process multiple images with the given prompt.
Describe multiple images using vision-capable LLM.
Args:
prompt: The prompt/question about the images
@@ -162,36 +186,49 @@ class ImageHandler(FlagHandler):
llm: Optional LLM model name
Returns:
Agent's response about the images
Description of the images
"""
# Validate all image paths
for path in image_paths:
valid, msg = self.validate_image_path(path)
if not valid:
return f"Error: {msg}"
agent = self.create_image_agent(llm=llm)
if not agent:
return "Error: Could not create ImageAgent"
available, dep_msg = self.check_dependencies()
if not available:
return f"Error: {dep_msg}"
import litellm
model = llm or DEFAULT_VISION_MODEL
try:
result = agent.chat(prompt, images=image_paths)
return result
# Build message with all images
content = [{"type": "text", "text": prompt}]
for path in image_paths:
content.append(self._build_image_content(path))
response = litellm.completion(
model=model,
messages=[{"role": "user", "content": content}]
)
return response.choices[0].message.content
except Exception as e:
self.log(f"Image processing error: {e}", "error")
return f"Error processing images: {e}"
logger.error(f"Error describing images: {e}")
return f"Error describing images: {e}"
def execute(self, prompt: str = None, image_path: str = None, llm: str = None, **kwargs) -> str:
"""
Execute image processing.
Execute image description (vision).
Args:
prompt: The prompt/question about the image
image_path: Path to the image file (or comma-separated paths)
llm: Optional LLM model name
llm: Optional vision-capable LLM model name
Returns:
Agent's response
Description of the image(s)
"""
if not prompt:
return "Error: No prompt provided"
@@ -202,6 +239,138 @@ class ImageHandler(FlagHandler):
# Handle multiple images (comma-separated)
if ',' in image_path:
paths = [p.strip() for p in image_path.split(',')]
return self.process_multiple_images(prompt, paths, llm)
return self.describe_multiple_images(prompt, paths, llm)
return self.process_image(prompt, image_path, llm)
return self.describe_image(prompt, image_path, llm)
class ImageGenerateHandler(FlagHandler):
"""
Handler for --image-generate flag (Image Generation).
Generates NEW images from text prompts using DALL-E or similar models.
This is for CREATING images, not analyzing existing ones.
Example:
praisonai "A sunset over mountains" --image-generate
praisonai "A cat wearing a hat" --image-generate --llm dall-e-3
"""
@property
def feature_name(self) -> str:
return "image_generate"
@property
def flag_name(self) -> str:
return "image_generate"
@property
def flag_help(self) -> str:
return "Generate an image from the text prompt"
def check_dependencies(self) -> Tuple[bool, str]:
"""Check if ImageAgent is available."""
try:
import importlib.util
if importlib.util.find_spec("praisonaiagents") is not None:
return True, ""
return False, "praisonaiagents not installed. Install with: pip install praisonaiagents"
except ImportError:
return False, "praisonaiagents not installed. Install with: pip install praisonaiagents"
def apply_to_agent_config(self, config: Dict[str, Any], flag_value: Any) -> Dict[str, Any]:
"""Apply image generation configuration."""
if flag_value:
config["use_image_generation"] = True
return config
def generate_image(self, prompt: str, llm: str = None, output_path: str = None) -> Dict[str, Any]:
"""
Generate an image from a text prompt.
Args:
prompt: Text description of the image to generate
llm: Image generation model (dall-e-3, dall-e-2, etc.)
output_path: Optional path to save the generated image
Returns:
Dict with image URL or path, or error message
"""
available, msg = self.check_dependencies()
if not available:
return {"error": msg}
from praisonaiagents import ImageAgent
# Use provided model or default
model = llm or DEFAULT_IMAGE_GEN_MODEL
# Validate model is supported for generation
if model not in SUPPORTED_IMAGE_GEN_MODELS:
return {
"error": f"Invalid model: '{model}'. Supported image generation models: {', '.join(SUPPORTED_IMAGE_GEN_MODELS)}"
}
try:
agent = ImageAgent(
name="ImageGenerator",
role="Image Generation Assistant",
goal="Generate high-quality images from text descriptions",
llm=model
)
result = agent.chat(prompt)
# Save image if output path provided
if output_path and isinstance(result, dict) and 'data' in result:
self._save_image(result, output_path)
result['saved_to'] = output_path
return result
except Exception as e:
logger.error(f"Error generating image: {e}")
return {"error": str(e)}
def _save_image(self, result: Dict[str, Any], output_path: str) -> bool:
"""Save generated image to file."""
try:
import requests
if 'data' in result and len(result['data']) > 0:
image_data = result['data'][0]
if 'url' in image_data:
# Download from URL
response = requests.get(image_data['url'])
with open(output_path, 'wb') as f:
f.write(response.content)
return True
elif 'b64_json' in image_data:
# Decode base64
import base64
img_bytes = base64.b64decode(image_data['b64_json'])
with open(output_path, 'wb') as f:
f.write(img_bytes)
return True
return False
except Exception as e:
logger.error(f"Error saving image: {e}")
return False
def execute(self, prompt: str = None, llm: str = None, output_path: str = None, **kwargs) -> Dict[str, Any]:
"""
Execute image generation.
Args:
prompt: Text description of the image to generate
llm: Image generation model name
output_path: Optional path to save the image
Returns:
Dict with generated image info or error
"""
if not prompt:
return {"error": "No prompt provided for image generation"}
return self.generate_image(prompt, llm, output_path)
@@ -107,6 +107,14 @@ class MetricsHandler(FlagHandler):
if hasattr(agent, 'last_token_metrics'):
metrics.update(agent.last_token_metrics or {})
# Check llm_model property for model info
if hasattr(agent, 'llm_model'):
llm_model = agent.llm_model
if hasattr(llm_model, 'model'):
metrics['model'] = llm_model.model
elif isinstance(llm_model, str):
metrics['model'] = llm_model
if hasattr(agent, 'llm') and agent.llm:
llm = agent.llm
if hasattr(llm, 'last_token_metrics'):
@@ -117,6 +125,42 @@ class MetricsHandler(FlagHandler):
if hasattr(agent, 'metrics') and isinstance(agent.metrics, dict):
metrics.update(agent.metrics or {})
# Try to get from litellm's success callback data
try:
import litellm
# Check if there's accumulated usage data
if hasattr(litellm, '_thread_context') and litellm._thread_context:
ctx = litellm._thread_context
if hasattr(ctx, 'usage'):
usage = ctx.usage
if hasattr(usage, 'prompt_tokens'):
metrics['prompt_tokens'] = usage.prompt_tokens
if hasattr(usage, 'completion_tokens'):
metrics['completion_tokens'] = usage.completion_tokens
if hasattr(usage, 'total_tokens'):
metrics['total_tokens'] = usage.total_tokens
except:
pass
# Try to get cost from litellm cost tracking
try:
import litellm
from litellm import completion_cost
# If we have token counts, estimate cost
if 'prompt_tokens' in metrics and 'completion_tokens' in metrics:
model = metrics.get('model', 'gpt-4o-mini')
try:
cost = completion_cost(
model=model,
prompt_tokens=metrics['prompt_tokens'],
completion_tokens=metrics['completion_tokens']
)
metrics['cost'] = cost
except:
pass
except:
pass
return metrics
def post_process_result(self, result: Any, flag_value: Any) -> Any:
@@ -0,0 +1,226 @@
"""
Profile Handler for CLI.
Provides performance profiling and benchmarking capabilities.
Usage: praisonai profile <subcommand> [options]
"""
import argparse
import time
import json
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class ProfileResult:
"""Result from a profiling run."""
prompt: str
response: str
execution_time_ms: float
model: str
tokens: Optional[Dict[str, int]] = None
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
def to_dict(self) -> Dict[str, Any]:
return {
"prompt": self.prompt,
"response": self.response[:200] + "..." if len(self.response) > 200 else self.response,
"execution_time_ms": self.execution_time_ms,
"model": self.model,
"tokens": self.tokens,
"timestamp": self.timestamp
}
def run_benchmark(prompt: str, model: str = None, iterations: int = 1, verbose: bool = False) -> List[ProfileResult]:
"""
Run a benchmark on a prompt.
Args:
prompt: The prompt to benchmark
model: Optional model to use
iterations: Number of iterations to run
verbose: Enable verbose output
Returns:
List of ProfileResult objects
"""
try:
from praisonaiagents import Agent
except ImportError:
print("[red]ERROR: praisonaiagents not installed. Install with: pip install praisonaiagents[/red]")
return []
results = []
agent_config = {
"name": "BenchmarkAgent",
"role": "Benchmark Assistant",
"goal": "Complete tasks efficiently",
"verbose": verbose
}
if model:
agent_config["llm"] = model
agent = Agent(**agent_config)
for i in range(iterations):
if verbose:
print(f"[dim]Running iteration {i + 1}/{iterations}...[/dim]")
start_time = time.time()
response = agent.start(prompt)
end_time = time.time()
execution_time_ms = (end_time - start_time) * 1000
result = ProfileResult(
prompt=prompt,
response=str(response),
execution_time_ms=execution_time_ms,
model=model or "default"
)
results.append(result)
return results
def display_benchmark_results(results: List[ProfileResult], output_format: str = "table"):
"""Display benchmark results."""
if not results:
print("[yellow]No results to display[/yellow]")
return
try:
from rich.console import Console
from rich.table import Table
console = Console()
except ImportError:
# Fallback to simple output
for r in results:
print(f"Time: {r.execution_time_ms:.1f}ms | Model: {r.model}")
return
if output_format == "json":
print(json.dumps([r.to_dict() for r in results], indent=2))
return
# Calculate statistics
times = [r.execution_time_ms for r in results]
avg_time = sum(times) / len(times)
min_time = min(times)
max_time = max(times)
table = Table(title="📊 Benchmark Results")
table.add_column("Metric", style="cyan")
table.add_column("Value", style="green")
table.add_row("Iterations", str(len(results)))
table.add_row("Average Time", f"{avg_time:.1f}ms")
table.add_row("Min Time", f"{min_time:.1f}ms")
table.add_row("Max Time", f"{max_time:.1f}ms")
table.add_row("Model", results[0].model)
console.print(table)
# Show response preview
if results:
console.print("\n[bold]Response Preview:[/bold]")
preview = results[0].response[:300] + "..." if len(results[0].response) > 300 else results[0].response
console.print(f"[dim]{preview}[/dim]")
def handle_profile_command(args: List[str]) -> int:
"""
Handle the profile command.
Args:
args: Command line arguments
Returns:
Exit code
"""
parser = argparse.ArgumentParser(prog="praisonai profile", description="Performance profiling and benchmarking")
subparsers = parser.add_subparsers(dest='subcommand', help='Profile subcommands')
# Benchmark subcommand
bench_parser = subparsers.add_parser('benchmark', aliases=['bench', 'run'], help='Run a benchmark')
bench_parser.add_argument('prompt', nargs='?', help='Prompt to benchmark')
bench_parser.add_argument('--model', '-m', type=str, help='Model to use')
bench_parser.add_argument('--iterations', '-n', type=int, default=1, help='Number of iterations')
bench_parser.add_argument('--output', '-o', type=str, help='Output file path')
bench_parser.add_argument('--format', '-f', type=str, choices=['table', 'json', 'html'], default='table', help='Output format')
bench_parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
# Compare subcommand
compare_parser = subparsers.add_parser('compare', help='Compare models')
compare_parser.add_argument('prompt', nargs='?', help='Prompt to compare')
compare_parser.add_argument('--models', type=str, help='Comma-separated list of models to compare')
# Report subcommand
report_parser = subparsers.add_parser('report', help='Generate a profiling report')
report_parser.add_argument('--input', '-i', type=str, help='Input JSON file with results')
report_parser.add_argument('--output', '-o', type=str, help='Output file path')
report_parser.add_argument('--format', '-f', type=str, choices=['html', 'json', 'markdown'], default='html', help='Report format')
try:
parsed_args = parser.parse_args(args)
except SystemExit:
return 1
if not parsed_args.subcommand:
parser.print_help()
print("\n[bold]Examples:[/bold]")
print(" praisonai profile benchmark \"What is 2+2?\"")
print(" praisonai profile benchmark \"Write a poem\" --model gpt-4o --iterations 3")
print(" praisonai profile compare \"Explain AI\" --models gpt-4o,gpt-4o-mini")
return 0
if parsed_args.subcommand in ['benchmark', 'bench', 'run']:
if not parsed_args.prompt:
print("[red]ERROR: Please provide a prompt to benchmark[/red]")
print("Usage: praisonai profile benchmark \"Your prompt here\"")
return 1
print(f"[bold cyan]🔬 Running benchmark...[/bold cyan]")
results = run_benchmark(
parsed_args.prompt,
model=parsed_args.model,
iterations=parsed_args.iterations,
verbose=parsed_args.verbose
)
display_benchmark_results(results, parsed_args.format)
if parsed_args.output:
with open(parsed_args.output, 'w') as f:
json.dump([r.to_dict() for r in results], f, indent=2)
print(f"[green]Results saved to {parsed_args.output}[/green]")
return 0
elif parsed_args.subcommand == 'compare':
if not parsed_args.prompt:
print("[red]ERROR: Please provide a prompt to compare[/red]")
return 1
models = parsed_args.models.split(',') if parsed_args.models else ['gpt-4o-mini', 'gpt-4o']
print(f"[bold cyan]🔬 Comparing models: {', '.join(models)}[/bold cyan]")
all_results = []
for model in models:
print(f"[dim]Testing {model}...[/dim]")
results = run_benchmark(parsed_args.prompt, model=model, iterations=1)
all_results.extend(results)
display_benchmark_results(all_results)
return 0
elif parsed_args.subcommand == 'report':
print("[yellow]Report generation not yet implemented[/yellow]")
return 0
return 0
@@ -0,0 +1,350 @@
"""
PraisonAI CLI Profiling Commands
Provides CLI commands for profiling agent execution.
Usage:
praisonai profile run "task" --output report.html
praisonai profile report --format json
praisonai profile benchmark --iterations 10
"""
import time
import json
from typing import Optional, Dict, Any
class ProfilingHandler:
"""
Handler for CLI profiling commands.
Features:
- Run tasks with profiling enabled
- Generate reports (console, JSON, HTML)
- Benchmark agent performance
- Export flamegraphs
"""
def __init__(self):
self._profiler = None
@property
def profiler(self):
"""Lazy load profiler."""
if self._profiler is None:
from praisonai.profiler import Profiler
self._profiler = Profiler
return self._profiler
def enable(self) -> None:
"""Enable profiling."""
self.profiler.enable()
def disable(self) -> None:
"""Disable profiling."""
self.profiler.disable()
def clear(self) -> None:
"""Clear profiling data."""
self.profiler.clear()
def run_with_profiling(self, task: str, config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
Run a task with profiling enabled.
Args:
task: Task description or prompt
config: Optional configuration dict
Returns:
Dict with result and profiling data
"""
self.profiler.enable()
self.profiler.clear()
result = None
error = None
try:
# Import and run PraisonAI
from praisonai import PraisonAI
with self.profiler.block("total_execution"):
with self.profiler.block("initialization"):
praison = PraisonAI(auto=task, **(config or {}))
with self.profiler.block("agent_run"):
result = praison.run()
except Exception as e:
error = str(e)
finally:
self.profiler.disable()
return {
'result': result,
'error': error,
'summary': self.profiler.get_summary(),
'statistics': self.profiler.get_statistics()
}
def report(self, format: str = "console", output: Optional[str] = None) -> str:
"""
Generate profiling report.
Args:
format: Output format (console, json, html)
output: Optional file path to save report
Returns:
Report content as string
"""
if format == "console":
content = self.profiler.report(output="console")
elif format == "json":
content = self.profiler.export_json()
elif format == "html":
content = self.profiler.export_html()
else:
raise ValueError(f"Unknown format: {format}")
if output:
with open(output, 'w') as f:
f.write(content)
print(f"Report saved to: {output}")
return content
def benchmark(self, task: str, iterations: int = 5,
warmup: int = 1, config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
Benchmark agent performance.
Args:
task: Task to benchmark
iterations: Number of iterations
warmup: Number of warmup runs
config: Optional configuration
Returns:
Benchmark results with statistics
"""
from praisonai.profiler import Profiler
times = []
errors = []
# Warmup runs
for i in range(warmup):
print(f"Warmup {i + 1}/{warmup}...")
try:
self.run_with_profiling(task, config)
except Exception as e:
print(f"Warmup error: {e}")
# Benchmark runs
for i in range(iterations):
print(f"Iteration {i + 1}/{iterations}...")
Profiler.clear()
start = time.perf_counter()
try:
self.run_with_profiling(task, config)
duration_ms = (time.perf_counter() - start) * 1000
times.append(duration_ms)
except Exception as e:
errors.append(str(e))
duration_ms = (time.perf_counter() - start) * 1000
times.append(duration_ms)
# Calculate statistics
if times:
sorted_times = sorted(times)
n = len(sorted_times)
results = {
'iterations': iterations,
'successful': iterations - len(errors),
'failed': len(errors),
'times_ms': times,
'mean_ms': sum(times) / n,
'min_ms': min(times),
'max_ms': max(times),
'p50_ms': sorted_times[n // 2],
'p95_ms': sorted_times[int(n * 0.95)] if n > 1 else sorted_times[-1],
'p99_ms': sorted_times[int(n * 0.99)] if n > 1 else sorted_times[-1],
'errors': errors
}
else:
results = {
'iterations': iterations,
'successful': 0,
'failed': len(errors),
'errors': errors
}
return results
def export_flamegraph(self, filepath: str) -> None:
"""Export flamegraph to SVG file."""
self.profiler.export_flamegraph(filepath)
print(f"Flamegraph saved to: {filepath}")
def get_summary(self) -> Dict[str, Any]:
"""Get profiling summary."""
return self.profiler.get_summary()
def get_statistics(self, category: Optional[str] = None) -> Dict[str, float]:
"""Get profiling statistics."""
return self.profiler.get_statistics(category)
def print_summary(self) -> None:
"""Print profiling summary to console."""
summary = self.get_summary()
stats = self.get_statistics()
print("\n" + "=" * 60)
print("PROFILING SUMMARY")
print("=" * 60)
print(f"\nTotal Time: {summary['total_time_ms']:.2f}ms")
print(f"Operations: {summary['timing_count']}")
print(f"Imports: {summary['import_count']}")
print(f"Flow Steps: {summary['flow_steps']}")
print("\nStatistics:")
print(f" P50 (Median): {stats['p50']:.2f}ms")
print(f" P95: {stats['p95']:.2f}ms")
print(f" P99: {stats['p99']:.2f}ms")
print(f" Mean: {stats['mean']:.2f}ms")
print(f" Std Dev: {stats['std_dev']:.2f}ms")
print("\nSlowest Operations:")
for name, duration in summary['slowest_operations'][:5]:
print(f" {name}: {duration:.2f}ms")
print("=" * 60)
def handle_profile_command(args) -> int:
"""
Handle profile CLI command.
Args:
args: Parsed command line arguments
Returns:
Exit code (0 for success)
"""
handler = ProfilingHandler()
subcommand = getattr(args, 'profile_command', None)
if subcommand == 'run':
task = getattr(args, 'task', '')
output = getattr(args, 'output', None)
format_type = getattr(args, 'format', 'console')
print(f"Running with profiling: {task[:50]}...")
result = handler.run_with_profiling(task)
if result['error']:
print(f"Error: {result['error']}")
handler.report(format=format_type, output=output)
return 0 if not result['error'] else 1
elif subcommand == 'report':
output = getattr(args, 'output', None)
format_type = getattr(args, 'format', 'console')
handler.report(format=format_type, output=output)
return 0
elif subcommand == 'benchmark':
task = getattr(args, 'task', '')
iterations = getattr(args, 'iterations', 5)
warmup = getattr(args, 'warmup', 1)
output = getattr(args, 'output', None)
print(f"Benchmarking: {task[:50]}...")
results = handler.benchmark(task, iterations=iterations, warmup=warmup)
print("\n" + "=" * 60)
print("BENCHMARK RESULTS")
print("=" * 60)
print(f"Iterations: {results['iterations']}")
print(f"Successful: {results['successful']}")
print(f"Failed: {results['failed']}")
if 'mean_ms' in results:
print("\nTiming:")
print(f" Mean: {results['mean_ms']:.2f}ms")
print(f" Min: {results['min_ms']:.2f}ms")
print(f" Max: {results['max_ms']:.2f}ms")
print(f" P50: {results['p50_ms']:.2f}ms")
print(f" P95: {results['p95_ms']:.2f}ms")
if output:
with open(output, 'w') as f:
json.dump(results, f, indent=2)
print(f"\nResults saved to: {output}")
return 0
elif subcommand == 'flamegraph':
output = getattr(args, 'output', 'profile.svg')
handler.export_flamegraph(output)
return 0
elif subcommand == 'summary':
handler.print_summary()
return 0
else:
print("Usage: praisonai profile <run|report|benchmark|flamegraph|summary>")
return 1
def add_profile_parser(subparsers) -> None:
"""Add profile subcommand to argument parser."""
profile_parser = subparsers.add_parser(
'profile',
help='Profile agent execution'
)
profile_subparsers = profile_parser.add_subparsers(dest='profile_command')
# Run with profiling
run_parser = profile_subparsers.add_parser('run', help='Run task with profiling')
run_parser.add_argument('task', help='Task to run')
run_parser.add_argument('--output', '-o', help='Output file path')
run_parser.add_argument('--format', '-f', choices=['console', 'json', 'html'],
default='console', help='Output format')
# Generate report
report_parser = profile_subparsers.add_parser('report', help='Generate profiling report')
report_parser.add_argument('--output', '-o', help='Output file path')
report_parser.add_argument('--format', '-f', choices=['console', 'json', 'html'],
default='console', help='Output format')
# Benchmark
bench_parser = profile_subparsers.add_parser('benchmark', help='Benchmark agent performance')
bench_parser.add_argument('task', help='Task to benchmark')
bench_parser.add_argument('--iterations', '-n', type=int, default=5, help='Number of iterations')
bench_parser.add_argument('--warmup', '-w', type=int, default=1, help='Warmup runs')
bench_parser.add_argument('--output', '-o', help='Output file for results')
# Flamegraph
flame_parser = profile_subparsers.add_parser('flamegraph', help='Export flamegraph')
flame_parser.add_argument('--output', '-o', default='profile.svg', help='Output SVG file')
# Summary
profile_subparsers.add_parser('summary', help='Print profiling summary') # noqa: F841
# Exports
__all__ = [
'ProfilingHandler',
'handle_profile_command',
'add_profile_parser',
]
+6 -13
View File
@@ -253,18 +253,11 @@ def add_skills_parser(subparsers) -> None:
Args:
subparsers: Subparsers object from argparse
"""
skills_parser = subparsers.add_parser(
'skills',
help='Manage Agent Skills'
)
skills_subparsers = skills_parser.add_subparsers(
dest='skills_command',
help='Skills commands'
)
# subparsers is already the skills subparsers object
# We just need to add the subcommands to it
# list command
list_parser = skills_subparsers.add_parser(
list_parser = subparsers.add_parser(
'list',
help='List available skills'
)
@@ -275,7 +268,7 @@ def add_skills_parser(subparsers) -> None:
)
# validate command
validate_parser = skills_subparsers.add_parser(
validate_parser = subparsers.add_parser(
'validate',
help='Validate a skill directory'
)
@@ -286,7 +279,7 @@ def add_skills_parser(subparsers) -> None:
)
# create command
create_parser = skills_subparsers.add_parser(
create_parser = subparsers.add_parser(
'create',
help='Create a new skill from template'
)
@@ -306,7 +299,7 @@ def add_skills_parser(subparsers) -> None:
)
# prompt command
prompt_parser = skills_subparsers.add_parser(
prompt_parser = subparsers.add_parser(
'prompt',
help='Generate prompt XML for skills'
)
+191 -41
View File
@@ -5,6 +5,42 @@ import argparse
import warnings
import os
# Suppress Pydantic serialization warnings from LiteLLM BEFORE any imports
# These warnings occur when LiteLLM's response objects have field mismatches
# Using both filterwarnings AND patching warnings.warn for complete suppression
warnings.filterwarnings("ignore", message=".*Pydantic serializer warnings.*")
warnings.filterwarnings("ignore", message=".*PydanticSerializationUnexpectedValue.*")
warnings.filterwarnings("ignore", message=".*Expected \\d+ fields but got.*")
warnings.filterwarnings("ignore", message=".*Expected `StreamingChoices`.*")
warnings.filterwarnings("ignore", message=".*Expected `Message`.*")
warnings.filterwarnings("ignore", message=".*serialized value may not be as expected.*")
warnings.filterwarnings("ignore", category=UserWarning, module="pydantic.*")
# Patch warnings.showwarning to intercept ALL warnings including those from crewai's patched warn
# This is the final output function that actually displays warnings
_SUPPRESSED_PATTERNS = [
"Pydantic serializer warnings",
"PydanticSerializationUnexpectedValue",
"Expected", # Catches "Expected N fields but got M"
"StreamingChoices",
"serialized value may not be as expected",
"duckduckgo_search", # Suppress duckduckgo rename warning
]
_original_showwarning = warnings.showwarning
def _patched_showwarning(message, category, filename, lineno, file=None, line=None):
msg_str = str(message)
for pattern in _SUPPRESSED_PATTERNS:
if pattern in msg_str:
return
if category is UserWarning and "pydantic" in filename.lower():
return
_original_showwarning(message, category, filename, lineno, file, line)
warnings.showwarning = _patched_showwarning
# Suppress crewai RuntimeWarning about module loading order (only in non-debug mode)
# This warning is harmless and occurs when running as `python -m praisonai.cli.main`
if os.environ.get('LOGLEVEL', 'INFO').upper() != 'DEBUG':
@@ -277,6 +313,18 @@ class PraisonAI:
elif hasattr(args, 'direct_prompt') and args.direct_prompt:
# Only handle direct prompt if agent_file wasn't explicitly set in constructor
if original_agent_file == "agents.yaml": # Default value, so safe to use direct prompt
# Handle --compare flag for CLI mode comparison
if hasattr(args, 'compare') and args.compare:
from .features.compare import CompareHandler
handler = CompareHandler(verbose=getattr(args, 'verbose', False))
result = handler.execute(
args.direct_prompt,
args.compare,
model=getattr(args, 'llm', None),
output_path=getattr(args, 'compare_output', None)
)
return result
# Combine direct prompt with any available inputs (stdin and/or file)
prompt_parts = [args.direct_prompt]
if stdin_input:
@@ -613,7 +661,7 @@ class PraisonAI:
return default_args
# Define special commands
special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule']
special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule', 'skills', 'profile', 'eval']
parser = argparse.ArgumentParser(prog="praisonai", description="praisonAI command-line interface")
parser.add_argument("--framework", choices=["crewai", "autogen", "praisonai"], help="Specify the framework")
@@ -685,8 +733,12 @@ class PraisonAI:
# Metrics - token/cost tracking
parser.add_argument("--metrics", action="store_true", help="Display token usage and cost metrics")
# Image - vision processing
parser.add_argument("--image", type=str, help="Path to image file for vision-based tasks")
# Image Description (Vision) - analyze existing images
parser.add_argument("--image", type=str, help="Path to image file for vision-based description/analysis")
# Image Generation - create new images from text
parser.add_argument("--image-generate", action="store_true", dest="image_generate",
help="Generate an image from the text prompt (use with --llm for model selection)")
# Telemetry - usage monitoring
parser.add_argument("--telemetry", action="store_true", help="Enable usage monitoring and analytics")
@@ -814,7 +866,7 @@ class PraisonAI:
except ModuleNotFoundError as e:
missing_module = str(e).split("'")[1]
print(f"[red]ERROR: Missing dependency {missing_module}. Install with:[/red]")
print(f"\npip install \"praisonai[chat]\"\n")
print("\npip install \"praisonai[chat]\"\n")
sys.exit(1)
sys.exit(0)
@@ -1042,6 +1094,35 @@ class PraisonAI:
self.handle_commit_command(unknown_args)
sys.exit(0)
elif args.command == 'skills':
if not PRAISONAI_AVAILABLE:
print("[red]ERROR: PraisonAI Agents is not installed. Install with:[/red]")
print("\npip install praisonaiagents\n")
sys.exit(1)
from .features.skills import handle_skills_command, add_skills_parser
# Create a parser for skills command
skills_parser = argparse.ArgumentParser(prog="praisonai skills")
skills_subparsers = skills_parser.add_subparsers(dest='skills_command', help='Skills commands')
add_skills_parser(skills_subparsers)
skills_args = skills_parser.parse_args(unknown_args)
exit_code = handle_skills_command(skills_args)
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)
elif args.command == 'eval':
# Eval command - evaluate model responses against expected outputs
from .features.eval import handle_eval_command
exit_code = handle_eval_command(unknown_args)
sys.exit(exit_code)
# Only check framework availability for agent-related operations
if not args.command and (args.init or args.auto or args.framework):
if not CREWAI_AVAILABLE and not AUTOGEN_AVAILABLE and not PRAISONAI_AVAILABLE:
@@ -1325,12 +1406,47 @@ class PraisonAI:
console.print(table)
# Show recent memories
# Show recent short-term memories
print("\n[bold]Recent Short-term Memories:[/bold]")
short_term = memory.get_short_term(limit=5)
for i, item in enumerate(short_term, 1):
content = item.get('content', str(item))[:100]
print(f" {i}. {content}")
if short_term:
for i, item in enumerate(short_term, 1):
content = item.get('content', str(item))[:100]
print(f" {i}. {content}")
else:
print(" [dim]No short-term memories[/dim]")
# Show long-term memories
print("\n[bold]Long-term Memories:[/bold]")
long_term = memory.get_long_term(limit=10)
if long_term:
for i, item in enumerate(long_term, 1):
# Handle both dict and MemoryItem objects
if hasattr(item, 'content'):
content = str(item.content)[:100]
importance = getattr(item, 'importance', 0)
else:
content = item.get('content', str(item))[:100]
importance = item.get('importance', 0)
print(f" {i}. [{importance:.1f}] {content}")
else:
print(" [dim]No long-term memories[/dim]")
# Show entities
print("\n[bold]Entities:[/bold]")
entities = memory.get_all_entities()
if entities:
for entity in entities[:10]:
# Handle both dict and Entity objects
if hasattr(entity, 'name'):
name = entity.name
entity_type = getattr(entity, 'entity_type', 'unknown')
else:
name = entity.get('name', 'Unknown')
entity_type = entity.get('entity_type', 'unknown')
print(f" • {name} ({entity_type})")
else:
print(" [dim]No entities[/dim]")
elif action == 'add':
if not action_args:
@@ -3090,10 +3206,9 @@ Provide ONLY the commit message, no explanations."""
selected_model = router.select_model(prompt, provider)
agent_config["llm"] = selected_model
# Metrics - Token/cost tracking
# Metrics - Token/cost tracking (display happens AFTER execution)
if getattr(self.args, 'metrics', False):
agent_config["metrics"] = True
print("[bold cyan]Metrics enabled - will display token usage and costs[/bold cyan]")
# Telemetry - Usage monitoring
if getattr(self.args, 'telemetry', False):
@@ -3101,9 +3216,14 @@ Provide ONLY the commit message, no explanations."""
telemetry = TelemetryHandler(verbose=getattr(self.args, 'verbose', False))
telemetry.enable()
# Auto Memory - Automatic memory extraction
# Sandbox - Secure command execution (display only, actual sandboxing is in tool approval)
sandbox_mode = getattr(self.args, 'sandbox', None)
if sandbox_mode and sandbox_mode != 'off':
print(f"[bold green]🔒 Sandbox Mode: {sandbox_mode.upper()}[/bold green]")
print("[dim]Commands will be validated before execution[/dim]")
# Auto Memory - Automatic memory extraction (handled post-processing, not as Agent param)
if getattr(self.args, 'auto_memory', False):
agent_config["auto_memory"] = True
print("[bold cyan]Auto Memory enabled - will extract and store memories[/bold cyan]")
# MCP - Model Context Protocol tools
@@ -3192,7 +3312,7 @@ Provide ONLY the commit message, no explanations."""
return result
# Image processing - Use ImageAgent instead
# Image Description (Vision) - analyze existing images
if hasattr(self, 'args') and getattr(self.args, 'image', None):
from .features.image import ImageHandler
image_handler = ImageHandler(verbose=getattr(self.args, 'verbose', False))
@@ -3214,6 +3334,27 @@ Provide ONLY the commit message, no explanations."""
return result
# Image Generation - create new images from text
if hasattr(self, 'args') and getattr(self.args, 'image_generate', False):
from .features.image import ImageGenerateHandler
image_gen_handler = ImageGenerateHandler(verbose=getattr(self.args, 'verbose', False))
result = image_gen_handler.execute(
prompt=prompt,
llm=agent_config.get('llm')
)
# Format output for display
if isinstance(result, dict):
if 'error' in result:
print(f"[red]Error: {result['error']}[/red]")
elif 'data' in result and len(result['data']) > 0:
image_url = result['data'][0].get('url', result['data'][0].get('b64_json', 'Generated'))
print(f"[green]Image generated successfully![/green]")
if 'url' in result['data'][0]:
print(f"URL: {result['data'][0]['url']}")
return result
# Flow Display - Visual workflow tracking
if hasattr(self, 'args') and getattr(self.args, 'flow_display', False):
from .features.flow_display import FlowDisplayHandler
@@ -3249,7 +3390,10 @@ Provide ONLY the commit message, no explanations."""
if hasattr(self, 'args') and getattr(self.args, 'auto_memory', False):
from .features.auto_memory import AutoMemoryHandler
auto_mem = AutoMemoryHandler(verbose=getattr(self.args, 'verbose', False))
auto_mem.post_process_result(result, {'user_id': getattr(self.args, 'user_id', None)})
auto_mem.post_process_result(
result,
{'user_id': getattr(self.args, 'user_id', None), 'user_message': prompt}
)
# Todo - Generate todo list from response
if hasattr(self, 'args') and getattr(self.args, 'todo', False):
@@ -3257,6 +3401,12 @@ Provide ONLY the commit message, no explanations."""
todo = TodoHandler(verbose=getattr(self.args, 'verbose', False))
todo.post_process_result(result, True)
# Telemetry - Display usage summary after execution
if hasattr(self, 'args') and getattr(self.args, 'telemetry', False):
from .features.telemetry import TelemetryHandler
telemetry = TelemetryHandler(verbose=getattr(self.args, 'verbose', False))
telemetry.post_process_result(result, True)
# Flow Display - End workflow
if hasattr(self, 'args') and getattr(self.args, 'flow_display', False):
from .features.flow_display import FlowDisplayHandler
@@ -4651,16 +4801,14 @@ Provide a concise summary (max 200 words):"""
def _process_interactive_prompt_async(self, prompt, tools_list, console, session_state=None):
"""
Process a prompt asynchronously with live status display.
Process a prompt asynchronously with non-blocking status display.
Shows real-time status (tools, commands) while processing in background.
Shows status updates without blocking user input.
Allows user to queue more messages while processing.
"""
import threading
import time
from rich.live import Live
from rich.panel import Panel
from rich.text import Text
import sys
from praisonai.cli.features.message_queue import ProcessingState
state_manager = session_state['state_manager']
@@ -4672,8 +4820,11 @@ Provide a concise summary (max 200 words):"""
live_status.clear()
live_status.update_status("Thinking...")
# Print initial status (non-blocking - user can still type)
console.print(f"[dim cyan]⏳ {live_status.current_status}[/dim cyan]")
# Result container for thread communication
result_container = {'response': None, 'error': None, 'done': False}
result_container = {'response': None, 'error': None, 'done': False, 'last_status': ''}
def process_in_background():
"""Run agent processing in background thread."""
@@ -4719,37 +4870,36 @@ Provide a concise summary (max 200 words):"""
bg_thread = threading.Thread(target=process_in_background, daemon=True)
bg_thread.start()
# Show live status while processing
try:
with Live(console=console, refresh_per_second=4, transient=True) as live:
while not result_container['done']:
# Build status display
status_text = Text()
status_text.append("⏳ ", style="cyan")
status_text.append(live_status.current_status or "Processing...", style="dim")
# Show queued messages if any
queue_count = message_queue.count
if queue_count > 0:
status_text.append(f"\n📋 Queued: {queue_count}", style="dim yellow")
live.update(Panel(status_text, border_style="dim"))
time.sleep(0.1)
except KeyboardInterrupt:
console.print("\n[dim]Processing interrupted[/dim]")
# Wait for completion with periodic status updates (non-blocking style)
start_time = time.time()
last_status_update = ""
while not result_container['done']:
time.sleep(0.2)
current_status = live_status.current_status
# Only print status if it changed
if current_status != last_status_update:
# Use carriage return to update in place
elapsed = time.time() - start_time
queue_count = message_queue.count
queue_info = f" | 📋 Queued: {queue_count}" if queue_count > 0 else ""
sys.stdout.write(f"\r[dim cyan]⏳ {current_status} ({elapsed:.1f}s){queue_info}[/dim cyan]")
sys.stdout.flush()
last_status_update = current_status
# Clear the status line
sys.stdout.write("\r" + " " * 80 + "\r")
sys.stdout.flush()
# Wait for thread to finish
bg_thread.join(timeout=1.0)
# Handle result
if result_container['error']:
console.print(f"\n[red]Error: {result_container['error']}[/red]")
console.print(f"[red]Error: {result_container['error']}[/red]")
elif result_container['response']:
# Print response with streaming effect
console.print()
response_str = result_container['response']
words = response_str.split()
import sys
for i, word in enumerate(words):
console.print(word + " ", end="")
sys.stdout.flush()
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==2.3.73 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==2.3.80 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+753 -4
View File
@@ -8,7 +8,14 @@ Features:
- Function execution timing
- Flow tracking
- File/module usage tracking
- Memory usage (optional)
- Memory usage (tracemalloc)
- API call profiling (wall-clock time)
- Streaming profiling (TTFT, total time)
- Statistics (p50, p95, p99)
- cProfile integration
- Flamegraph generation
- Line-level profiling
- JSON/HTML export
Usage:
from praisonai.profiler import Profiler, profile, profile_imports
@@ -22,12 +29,27 @@ Usage:
with Profiler.block("my_operation"):
do_something()
# Profile API calls
with Profiler.api_call("https://api.example.com") as call:
response = requests.get(...)
# Profile streaming
with Profiler.streaming("chat") as tracker:
tracker.first_token()
for chunk in stream:
tracker.chunk()
# Profile imports
with profile_imports():
import heavy_module
# Get report
# Get report with statistics
Profiler.report()
stats = Profiler.get_statistics()
# Export
Profiler.export_json()
Profiler.export_html()
"""
import time
@@ -35,9 +57,15 @@ import functools
import threading
import sys
import os
from dataclasses import dataclass, field
import json
import tracemalloc
import cProfile
import pstats
import io
import statistics
from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional, Callable, Any
from contextlib import contextmanager
from contextlib import contextmanager, asynccontextmanager
# ============================================================================
@@ -55,6 +83,38 @@ class TimingRecord:
timestamp: float = field(default_factory=time.time)
@dataclass
class APICallRecord:
"""Record of an API/HTTP call."""
endpoint: str
method: str
duration_ms: float
status_code: int = 0
request_size: int = 0
response_size: int = 0
timestamp: float = field(default_factory=time.time)
@dataclass
class StreamingRecord:
"""Record of streaming operation (LLM responses)."""
name: str
ttft_ms: float # Time to first token
total_ms: float
chunk_count: int = 0
total_tokens: int = 0
timestamp: float = field(default_factory=time.time)
@dataclass
class MemoryRecord:
"""Record of memory usage."""
name: str
current_kb: float
peak_kb: float
timestamp: float = field(default_factory=time.time)
@dataclass
class ImportRecord:
"""Record of a module import."""
@@ -75,6 +135,80 @@ class FlowRecord:
timestamp: float = field(default_factory=time.time)
# ============================================================================
# Streaming Tracker
# ============================================================================
class StreamingTracker:
"""
Track streaming operations (LLM responses).
Usage:
tracker = StreamingTracker("chat")
tracker.start()
tracker.first_token() # Mark TTFT
for chunk in stream:
tracker.chunk()
tracker.end(total_tokens=100)
"""
def __init__(self, name: str):
self.name = name
self._start_time: Optional[float] = None
self._first_token_time: Optional[float] = None
self._end_time: Optional[float] = None
self._chunk_count: int = 0
self._total_tokens: int = 0
def start(self) -> None:
"""Start tracking."""
self._start_time = time.perf_counter()
def first_token(self) -> None:
"""Mark time to first token."""
if self._first_token_time is None:
self._first_token_time = time.perf_counter()
def chunk(self) -> None:
"""Record a chunk received."""
self._chunk_count += 1
def end(self, total_tokens: int = 0) -> None:
"""End tracking and record to Profiler."""
self._end_time = time.perf_counter()
self._total_tokens = total_tokens
if self._start_time is not None:
ttft_ms = 0.0
if self._first_token_time is not None:
ttft_ms = (self._first_token_time - self._start_time) * 1000
total_ms = (self._end_time - self._start_time) * 1000
Profiler.record_streaming(
name=self.name,
ttft_ms=ttft_ms,
total_ms=total_ms,
chunk_count=self._chunk_count,
total_tokens=self._total_tokens
)
@property
def ttft_ms(self) -> float:
"""Get time to first token in ms."""
if self._start_time and self._first_token_time:
return (self._first_token_time - self._start_time) * 1000
return 0.0
@property
def elapsed_ms(self) -> float:
"""Get elapsed time in ms."""
if self._start_time:
end = self._end_time or time.perf_counter()
return (end - self._start_time) * 1000
return 0.0
# ============================================================================
# Profiler Class
# ============================================================================
@@ -84,6 +218,16 @@ class Profiler:
Centralized profiler for performance monitoring.
Thread-safe singleton pattern for global access.
Features:
- Function/block timing
- API call profiling (wall-clock)
- Streaming profiling (TTFT)
- Memory profiling
- Import timing
- Statistics (p50, p95, p99)
- cProfile integration
- Export (JSON, HTML)
"""
_instance: Optional['Profiler'] = None
@@ -93,9 +237,14 @@ class Profiler:
_timings: List[TimingRecord] = []
_imports: List[ImportRecord] = []
_flow: List[FlowRecord] = []
_api_calls: List[APICallRecord] = []
_streaming: List[StreamingRecord] = []
_memory: List[MemoryRecord] = []
_enabled: bool = False
_flow_step: int = 0
_files_accessed: Dict[str, int] = {}
_line_profile_data: Dict[str, Any] = {}
_cprofile_stats: List[Dict[str, Any]] = []
def __new__(cls):
if cls._instance is None:
@@ -126,8 +275,13 @@ class Profiler:
cls._timings.clear()
cls._imports.clear()
cls._flow.clear()
cls._api_calls.clear()
cls._streaming.clear()
cls._memory.clear()
cls._flow_step = 0
cls._files_accessed.clear()
cls._line_profile_data.clear()
cls._cprofile_stats.clear()
@classmethod
def record_timing(cls, name: str, duration_ms: float, category: str = "function",
@@ -295,6 +449,433 @@ class Profiler:
print(report_text)
return report_text
# ========================================================================
# API Call Profiling
# ========================================================================
@classmethod
def record_api_call(cls, endpoint: str, method: str, duration_ms: float,
status_code: int = 0, request_size: int = 0,
response_size: int = 0) -> None:
"""Record an API/HTTP call timing."""
if not cls.is_enabled():
return
with cls._lock:
cls._api_calls.append(APICallRecord(
endpoint=endpoint,
method=method,
duration_ms=duration_ms,
status_code=status_code,
request_size=request_size,
response_size=response_size
))
@classmethod
def get_api_calls(cls) -> List[APICallRecord]:
"""Get API call records."""
with cls._lock:
return cls._api_calls.copy()
@classmethod
@contextmanager
def api_call(cls, endpoint: str, method: str = "GET"):
"""Context manager for profiling API calls."""
start = time.perf_counter()
call_info = {'status_code': 0, 'request_size': 0, 'response_size': 0}
try:
yield call_info
finally:
duration_ms = (time.perf_counter() - start) * 1000
cls.record_api_call(
endpoint=endpoint,
method=method,
duration_ms=duration_ms,
status_code=call_info.get('status_code', 0),
request_size=call_info.get('request_size', 0),
response_size=call_info.get('response_size', 0)
)
# ========================================================================
# Streaming Profiling
# ========================================================================
@classmethod
def record_streaming(cls, name: str, ttft_ms: float, total_ms: float,
chunk_count: int = 0, total_tokens: int = 0) -> None:
"""Record streaming metrics."""
if not cls.is_enabled():
return
with cls._lock:
cls._streaming.append(StreamingRecord(
name=name,
ttft_ms=ttft_ms,
total_ms=total_ms,
chunk_count=chunk_count,
total_tokens=total_tokens
))
@classmethod
def get_streaming_records(cls) -> List[StreamingRecord]:
"""Get streaming records."""
with cls._lock:
return cls._streaming.copy()
@classmethod
@contextmanager
def streaming(cls, name: str):
"""Context manager for profiling streaming operations."""
tracker = StreamingTracker(name)
tracker.start()
try:
yield tracker
finally:
tracker.end()
@classmethod
@asynccontextmanager
async def streaming_async(cls, name: str):
"""Async context manager for profiling streaming operations."""
tracker = StreamingTracker(name)
tracker.start()
try:
yield tracker
finally:
tracker.end()
# ========================================================================
# Memory Profiling
# ========================================================================
@classmethod
def record_memory(cls, name: str, current_kb: float, peak_kb: float) -> None:
"""Record memory usage."""
if not cls.is_enabled():
return
with cls._lock:
cls._memory.append(MemoryRecord(
name=name,
current_kb=current_kb,
peak_kb=peak_kb
))
@classmethod
def get_memory_records(cls) -> List[MemoryRecord]:
"""Get memory records."""
with cls._lock:
return cls._memory.copy()
@classmethod
@contextmanager
def memory(cls, name: str):
"""Context manager for profiling memory usage."""
tracemalloc.start()
try:
yield
finally:
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
cls.record_memory(name, current / 1024, peak / 1024)
@classmethod
def memory_snapshot(cls) -> Dict[str, float]:
"""Take a memory snapshot."""
tracemalloc.start()
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
return {
'current_kb': current / 1024,
'peak_kb': peak / 1024
}
# ========================================================================
# Statistics
# ========================================================================
@classmethod
def get_statistics(cls, category: Optional[str] = None) -> Dict[str, float]:
"""
Get statistical analysis of timing data.
Returns p50, p95, p99, mean, std_dev, min, max.
"""
with cls._lock:
if category:
durations = [t.duration_ms for t in cls._timings if t.category == category]
else:
durations = [t.duration_ms for t in cls._timings]
if not durations:
return {
'p50': 0.0, 'p95': 0.0, 'p99': 0.0,
'mean': 0.0, 'std_dev': 0.0, 'min': 0.0, 'max': 0.0,
'count': 0
}
sorted_durations = sorted(durations)
n = len(sorted_durations)
def percentile(p: float) -> float:
idx = int(n * p / 100)
return sorted_durations[min(idx, n - 1)]
mean = statistics.mean(durations)
std_dev = statistics.stdev(durations) if n > 1 else 0.0
return {
'p50': percentile(50),
'p95': percentile(95),
'p99': percentile(99),
'mean': mean,
'std_dev': std_dev,
'min': min(durations),
'max': max(durations),
'count': n
}
# ========================================================================
# cProfile Integration
# ========================================================================
@classmethod
@contextmanager
def cprofile(cls, name: str):
"""Context manager for cProfile profiling."""
pr = cProfile.Profile()
pr.enable()
try:
yield pr
finally:
pr.disable()
# Store stats
s = io.StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
ps.print_stats(30)
with cls._lock:
cls._cprofile_stats.append({
'name': name,
'stats': s.getvalue(),
'total_calls': ps.total_calls if hasattr(ps, 'total_calls') else 0,
'timestamp': time.time()
})
@classmethod
def get_cprofile_stats(cls) -> List[Dict[str, Any]]:
"""Get cProfile statistics."""
with cls._lock:
return cls._cprofile_stats.copy()
# ========================================================================
# Line-Level Profiling
# ========================================================================
@classmethod
def get_line_profile_data(cls) -> Dict[str, Any]:
"""Get line-level profiling data."""
with cls._lock:
return cls._line_profile_data.copy()
@classmethod
def set_line_profile_data(cls, func_name: str, data: Any) -> None:
"""Store line-level profiling data."""
if not cls.is_enabled():
return
with cls._lock:
cls._line_profile_data[func_name] = data
# ========================================================================
# Flamegraph
# ========================================================================
@classmethod
def get_flamegraph_data(cls) -> List[Dict[str, Any]]:
"""
Generate flamegraph-compatible data from flow records.
Returns list of {name, value, children} for flamegraph visualization.
"""
with cls._lock:
# Convert flow records to flamegraph format
data = []
for record in cls._flow:
data.append({
'name': record.name,
'value': record.duration_ms,
'file': record.file,
'line': record.line
})
return data
@classmethod
def export_flamegraph(cls, filepath: str) -> None:
"""
Export flamegraph to SVG file.
Note: Requires flamegraph data. For full flamegraph support,
use py-spy: py-spy record -o profile.svg -- python script.py
"""
data = cls.get_flamegraph_data()
# Generate simple SVG flamegraph
svg_content = cls._generate_simple_flamegraph_svg(data)
with open(filepath, 'w') as f:
f.write(svg_content)
@classmethod
def _generate_simple_flamegraph_svg(cls, data: List[Dict[str, Any]]) -> str:
"""Generate a simple SVG flamegraph."""
if not data:
return '<svg xmlns="http://www.w3.org/2000/svg" width="800" height="100"><text x="10" y="50">No profiling data</text></svg>'
total_time = sum(d['value'] for d in data)
if total_time == 0:
total_time = 1
width = 800
height = max(100, len(data) * 25 + 50)
svg_parts = [
f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}">',
'<style>rect:hover { opacity: 0.8; } text { font-family: monospace; font-size: 12px; }</style>',
f'<text x="10" y="20">PraisonAI Profiling Flamegraph (Total: {total_time:.2f}ms)</text>'
]
y = 40
for item in sorted(data, key=lambda x: x['value'], reverse=True)[:20]:
bar_width = max(10, (item['value'] / total_time) * (width - 20))
color = f'hsl({int(item["value"] / total_time * 120)}, 70%, 50%)'
svg_parts.append(
f'<rect x="10" y="{y}" width="{bar_width}" height="20" fill="{color}" />'
)
svg_parts.append(
f'<text x="15" y="{y + 15}">{item["name"]}: {item["value"]:.2f}ms</text>'
)
y += 25
svg_parts.append('</svg>')
return '\n'.join(svg_parts)
# ========================================================================
# Export Functions
# ========================================================================
@classmethod
def export_json(cls) -> str:
"""Export profiling data as JSON."""
# Get summary and stats first (they acquire their own locks)
summary = cls.get_summary()
stats = cls.get_statistics()
with cls._lock:
data = {
'summary': summary,
'statistics': stats,
'timings': [asdict(t) for t in cls._timings],
'api_calls': [asdict(a) for a in cls._api_calls],
'streaming': [asdict(s) for s in cls._streaming],
'memory': [asdict(m) for m in cls._memory],
'imports': [asdict(i) for i in cls._imports],
'flow': [asdict(f) for f in cls._flow]
}
return json.dumps(data, indent=2, default=str)
@classmethod
def export_html(cls) -> str:
"""Export profiling data as HTML report."""
summary = cls.get_summary()
stats = cls.get_statistics()
html = f'''<!DOCTYPE html>
<html>
<head>
<title>PraisonAI Profiling Report</title>
<style>
body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; margin: 20px; }}
h1 {{ color: #333; }}
h2 {{ color: #666; border-bottom: 1px solid #ddd; padding-bottom: 5px; }}
table {{ border-collapse: collapse; width: 100%; margin: 10px 0; }}
th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
th {{ background-color: #f5f5f5; }}
.metric {{ display: inline-block; margin: 10px; padding: 15px; background: #f9f9f9; border-radius: 5px; }}
.metric-value {{ font-size: 24px; font-weight: bold; color: #0066cc; }}
.metric-label {{ font-size: 12px; color: #666; }}
</style>
</head>
<body>
<h1>PraisonAI Profiling Report</h1>
<h2>Summary</h2>
<div class="metric">
<div class="metric-value">{summary['total_time_ms']:.2f}ms</div>
<div class="metric-label">Total Time</div>
</div>
<div class="metric">
<div class="metric-value">{summary['timing_count']}</div>
<div class="metric-label">Operations</div>
</div>
<div class="metric">
<div class="metric-value">{len(cls._api_calls)}</div>
<div class="metric-label">API Calls</div>
</div>
<div class="metric">
<div class="metric-value">{len(cls._streaming)}</div>
<div class="metric-label">Streams</div>
</div>
<h2>Statistics</h2>
<table>
<tr><th>Metric</th><th>Value</th></tr>
<tr><td>P50 (Median)</td><td>{stats['p50']:.2f}ms</td></tr>
<tr><td>P95</td><td>{stats['p95']:.2f}ms</td></tr>
<tr><td>P99</td><td>{stats['p99']:.2f}ms</td></tr>
<tr><td>Mean</td><td>{stats['mean']:.2f}ms</td></tr>
<tr><td>Std Dev</td><td>{stats['std_dev']:.2f}ms</td></tr>
<tr><td>Min</td><td>{stats['min']:.2f}ms</td></tr>
<tr><td>Max</td><td>{stats['max']:.2f}ms</td></tr>
</table>
<h2>Slowest Operations</h2>
<table>
<tr><th>Operation</th><th>Duration (ms)</th></tr>
{''.join(f"<tr><td>{name}</td><td>{dur:.2f}</td></tr>" for name, dur in summary['slowest_operations'])}
</table>
<h2>API Calls</h2>
<table>
<tr><th>Endpoint</th><th>Method</th><th>Duration (ms)</th><th>Status</th></tr>
{''.join(f"<tr><td>{a.endpoint}</td><td>{a.method}</td><td>{a.duration_ms:.2f}</td><td>{a.status_code}</td></tr>" for a in cls._api_calls[:20])}
</table>
<h2>Streaming</h2>
<table>
<tr><th>Name</th><th>TTFT (ms)</th><th>Total (ms)</th><th>Chunks</th><th>Tokens</th></tr>
{''.join(f"<tr><td>{s.name}</td><td>{s.ttft_ms:.2f}</td><td>{s.total_ms:.2f}</td><td>{s.chunk_count}</td><td>{s.total_tokens}</td></tr>" for s in cls._streaming[:20])}
</table>
</body>
</html>'''
return html
@classmethod
def export_to_file(cls, filepath: str, format: str = "json") -> None:
"""Export profiling data to file."""
if format == "json":
content = cls.export_json()
elif format == "html":
content = cls.export_html()
else:
raise ValueError(f"Unknown format: {format}")
with open(filepath, 'w') as f:
f.write(content)
# ============================================================================
@@ -422,6 +1003,161 @@ def profile_imports():
return ImportProfiler()
# ============================================================================
# API Profiling Decorators
# ============================================================================
def profile_api(func: Optional[Callable] = None, *, endpoint: str = ""):
"""
Decorator to profile a function as an API call.
Usage:
@profile_api(endpoint="openai/chat")
def call_openai():
pass
"""
def decorator(fn: Callable) -> Callable:
@functools.wraps(fn)
def wrapper(*args, **kwargs):
if not Profiler.is_enabled():
return fn(*args, **kwargs)
ep = endpoint or fn.__name__
start = time.perf_counter()
try:
return fn(*args, **kwargs)
finally:
duration_ms = (time.perf_counter() - start) * 1000
Profiler.record_api_call(ep, "CALL", duration_ms)
return wrapper
if func is not None:
return decorator(func)
return decorator
def profile_api_async(func: Optional[Callable] = None, *, endpoint: str = ""):
"""
Decorator to profile an async function as an API call.
"""
def decorator(fn: Callable) -> Callable:
@functools.wraps(fn)
async def wrapper(*args, **kwargs):
if not Profiler.is_enabled():
return await fn(*args, **kwargs)
ep = endpoint or fn.__name__
start = time.perf_counter()
try:
return await fn(*args, **kwargs)
finally:
duration_ms = (time.perf_counter() - start) * 1000
Profiler.record_api_call(ep, "CALL", duration_ms)
return wrapper
if func is not None:
return decorator(func)
return decorator
# ============================================================================
# cProfile Decorator
# ============================================================================
def profile_detailed(func: Optional[Callable] = None):
"""
Decorator for detailed cProfile profiling.
Usage:
@profile_detailed
def heavy_computation():
pass
"""
def decorator(fn: Callable) -> Callable:
@functools.wraps(fn)
def wrapper(*args, **kwargs):
if not Profiler.is_enabled():
return fn(*args, **kwargs)
pr = cProfile.Profile()
pr.enable()
try:
return fn(*args, **kwargs)
finally:
pr.disable()
s = io.StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
ps.print_stats(20)
Profiler._cprofile_stats.append({
'name': fn.__name__,
'stats': s.getvalue(),
'timestamp': time.time()
})
return wrapper
if func is not None:
return decorator(func)
return decorator
# ============================================================================
# Line-Level Profiling Decorator
# ============================================================================
def profile_lines(func: Optional[Callable] = None):
"""
Decorator for line-level profiling.
Note: Requires line_profiler package for full functionality.
Falls back to basic timing if not available.
Usage:
@profile_lines
def my_function():
pass
"""
def decorator(fn: Callable) -> Callable:
@functools.wraps(fn)
def wrapper(*args, **kwargs):
if not Profiler.is_enabled():
return fn(*args, **kwargs)
# Try to use line_profiler if available
try:
from line_profiler import LineProfiler
lp = LineProfiler()
lp.add_function(fn)
lp.enable()
try:
result = fn(*args, **kwargs)
finally:
lp.disable()
s = io.StringIO()
lp.print_stats(stream=s)
Profiler.set_line_profile_data(fn.__name__, s.getvalue())
return result
except ImportError:
# Fallback to basic timing
start = time.perf_counter()
try:
return fn(*args, **kwargs)
finally:
duration_ms = (time.perf_counter() - start) * 1000
Profiler.set_line_profile_data(fn.__name__, {
'note': 'line_profiler not installed',
'total_ms': duration_ms
})
return wrapper
if func is not None:
return decorator(func)
return decorator
# ============================================================================
# Quick Profiling Functions
# ============================================================================
@@ -452,14 +1188,27 @@ def check_module_available(module_name: str) -> bool:
# ============================================================================
__all__ = [
# Core
'Profiler',
'StreamingTracker',
# Decorators
'profile',
'profile_async',
'profile_api',
'profile_api_async',
'profile_detailed',
'profile_lines',
# Import profiling
'profile_imports',
'ImportProfiler',
# Utilities
'time_import',
'check_module_available',
# Data classes
'TimingRecord',
'ImportRecord',
'FlowRecord',
'APICallRecord',
'StreamingRecord',
'MemoryRecord',
]
+1 -1
View File
@@ -1 +1 @@
__version__ = "2.3.73"
__version__ = "2.3.80"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.1.22",
"praisonaiagents>=0.1.24",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
@@ -0,0 +1,555 @@
"""
Unit tests for PraisonAI Advanced Profiler features.
TDD: Tests written first for new profiling capabilities:
- API/HTTP call profiling (wall-clock time)
- Streaming profiling (TTFT, total time)
- Line-level profiling
- Memory profiling
- Flamegraph generation
- cProfile integration
- Statistics (p50, p95, p99)
"""
import time
import asyncio
import pytest
# ============================================================================
# API Call Profiling Tests
# ============================================================================
class TestAPICallProfiler:
"""Test API/HTTP call profiling with wall-clock time."""
def test_api_call_record(self):
"""Should record API call timing."""
from praisonai.profiler import Profiler, APICallRecord
Profiler.enable()
Profiler.clear()
Profiler.record_api_call(
endpoint="https://api.openai.com/v1/chat/completions",
method="POST",
duration_ms=1500.0,
status_code=200,
request_size=1024,
response_size=2048
)
calls = Profiler.get_api_calls()
assert len(calls) >= 1
assert calls[-1].endpoint == "https://api.openai.com/v1/chat/completions"
assert calls[-1].duration_ms == 1500.0
assert calls[-1].status_code == 200
Profiler.disable()
Profiler.clear()
def test_api_call_context_manager(self):
"""Should profile API call with context manager."""
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
with Profiler.api_call("https://api.example.com/test", method="GET") as call:
time.sleep(0.01) # Simulate API latency
calls = Profiler.get_api_calls()
assert len(calls) >= 1
assert calls[-1].duration_ms >= 10
Profiler.disable()
Profiler.clear()
def test_api_call_decorator(self):
"""Should profile function as API call."""
from praisonai.profiler import Profiler, profile_api
Profiler.enable()
Profiler.clear()
@profile_api(endpoint="test_endpoint")
def mock_api_call():
time.sleep(0.01)
return {"status": "ok"}
result = mock_api_call()
assert result == {"status": "ok"}
calls = Profiler.get_api_calls()
assert len(calls) >= 1
Profiler.disable()
Profiler.clear()
# ============================================================================
# Streaming Profiling Tests
# ============================================================================
class TestStreamingProfiler:
"""Test streaming profiling (TTFT, chunk times, total time)."""
def test_streaming_record(self):
"""Should record streaming metrics."""
from praisonai.profiler import Profiler, StreamingRecord
Profiler.enable()
Profiler.clear()
Profiler.record_streaming(
name="chat_completion",
ttft_ms=150.0, # Time to first token
total_ms=2000.0,
chunk_count=50,
total_tokens=500
)
streams = Profiler.get_streaming_records()
assert len(streams) >= 1
assert streams[-1].ttft_ms == 150.0
assert streams[-1].total_ms == 2000.0
assert streams[-1].chunk_count == 50
Profiler.disable()
Profiler.clear()
def test_streaming_tracker(self):
"""Should track streaming with context manager."""
from praisonai.profiler import Profiler, StreamingTracker
Profiler.enable()
Profiler.clear()
tracker = StreamingTracker("test_stream")
tracker.start()
time.sleep(0.01)
tracker.first_token() # Mark TTFT
time.sleep(0.02)
tracker.chunk() # Record chunk
tracker.chunk()
tracker.end(total_tokens=100)
streams = Profiler.get_streaming_records()
assert len(streams) >= 1
assert streams[-1].ttft_ms >= 10
assert streams[-1].chunk_count == 2
Profiler.disable()
Profiler.clear()
def test_streaming_context_manager(self):
"""Should profile streaming with context manager."""
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
with Profiler.streaming("test_stream") as tracker:
time.sleep(0.01)
tracker.first_token()
for _ in range(3):
tracker.chunk()
time.sleep(0.005)
streams = Profiler.get_streaming_records()
assert len(streams) >= 1
assert streams[-1].chunk_count == 3
Profiler.disable()
Profiler.clear()
# ============================================================================
# Memory Profiling Tests
# ============================================================================
class TestMemoryProfiler:
"""Test memory profiling capabilities."""
def test_memory_record(self):
"""Should record memory usage."""
from praisonai.profiler import Profiler, MemoryRecord
Profiler.enable()
Profiler.clear()
Profiler.record_memory(
name="agent_init",
current_kb=1024.0,
peak_kb=2048.0
)
memories = Profiler.get_memory_records()
assert len(memories) >= 1
assert memories[-1].current_kb == 1024.0
assert memories[-1].peak_kb == 2048.0
Profiler.disable()
Profiler.clear()
def test_memory_context_manager(self):
"""Should profile memory with context manager."""
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
with Profiler.memory("test_operation"):
# Allocate some memory
data = [i for i in range(10000)]
memories = Profiler.get_memory_records()
assert len(memories) >= 1
assert memories[-1].peak_kb > 0
Profiler.disable()
Profiler.clear()
def test_memory_snapshot(self):
"""Should take memory snapshot."""
from praisonai.profiler import Profiler
Profiler.enable()
snapshot = Profiler.memory_snapshot()
assert 'current_kb' in snapshot
assert 'peak_kb' in snapshot
Profiler.disable()
# ============================================================================
# Statistics Tests
# ============================================================================
class TestProfilerStatistics:
"""Test statistical analysis of profiling data."""
def test_percentiles(self):
"""Should calculate percentiles (p50, p95, p99)."""
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
# Add multiple timing records
for i in range(100):
Profiler.record_timing(f"op_{i}", float(i), "test")
stats = Profiler.get_statistics(category="test")
assert 'p50' in stats
assert 'p95' in stats
assert 'p99' in stats
assert 'mean' in stats
assert 'std_dev' in stats
assert 'min' in stats
assert 'max' in stats
Profiler.disable()
Profiler.clear()
def test_statistics_by_category(self):
"""Should calculate statistics per category."""
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
for i in range(10):
Profiler.record_timing(f"api_{i}", float(i * 100), "api")
Profiler.record_timing(f"func_{i}", float(i * 10), "function")
api_stats = Profiler.get_statistics(category="api")
func_stats = Profiler.get_statistics(category="function")
assert api_stats['mean'] > func_stats['mean']
Profiler.disable()
Profiler.clear()
# ============================================================================
# cProfile Integration Tests
# ============================================================================
class TestCProfileIntegration:
"""Test cProfile integration for detailed function profiling."""
def test_cprofile_context_manager(self):
"""Should profile with cProfile."""
from praisonai.profiler import Profiler
Profiler.enable()
with Profiler.cprofile("test_operation") as stats:
# Do some work
result = sum(i * i for i in range(1000))
assert stats is not None
assert hasattr(stats, 'total_calls') or hasattr(stats, 'stats')
Profiler.disable()
def test_cprofile_decorator(self):
"""Should profile function with cProfile decorator."""
from praisonai.profiler import profile_detailed
@profile_detailed
def compute_heavy():
return sum(i * i for i in range(10000))
result = compute_heavy()
assert result > 0
# ============================================================================
# Flamegraph Tests
# ============================================================================
class TestFlamegraph:
"""Test flamegraph generation."""
def test_flamegraph_data_generation(self):
"""Should generate flamegraph-compatible data."""
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
# Record some nested operations
with Profiler.block("outer"):
time.sleep(0.01)
with Profiler.block("inner"):
time.sleep(0.01)
flamegraph_data = Profiler.get_flamegraph_data()
assert flamegraph_data is not None
assert isinstance(flamegraph_data, (str, list, dict))
Profiler.disable()
Profiler.clear()
def test_flamegraph_export(self):
"""Should export flamegraph to file."""
import tempfile
import os
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
with Profiler.block("test_op"):
time.sleep(0.01)
with tempfile.NamedTemporaryFile(mode='w', suffix='.svg', delete=False) as f:
filepath = f.name
try:
Profiler.export_flamegraph(filepath)
assert os.path.exists(filepath)
finally:
if os.path.exists(filepath):
os.unlink(filepath)
Profiler.disable()
Profiler.clear()
# ============================================================================
# Line-Level Profiling Tests
# ============================================================================
class TestLineProfiler:
"""Test line-level profiling integration."""
def test_line_profile_decorator(self):
"""Should profile function line by line."""
from praisonai.profiler import profile_lines, Profiler
Profiler.enable()
@profile_lines
def sample_function():
a = 1
b = 2
c = a + b
return c
result = sample_function()
assert result == 3
# Line profiling data should be available
line_data = Profiler.get_line_profile_data()
# May be empty if line_profiler not installed, but shouldn't error
assert line_data is not None
Profiler.disable()
# ============================================================================
# Report Export Tests
# ============================================================================
class TestReportExport:
"""Test report export in various formats."""
def test_json_export(self):
"""Should export report as JSON."""
import json
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
Profiler.record_timing("test_op", 100.0, "test")
json_report = Profiler.export_json()
data = json.loads(json_report)
assert 'timings' in data or 'summary' in data
Profiler.disable()
Profiler.clear()
def test_html_export(self):
"""Should export report as HTML."""
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
Profiler.record_timing("test_op", 100.0, "test")
html_report = Profiler.export_html()
assert '<html>' in html_report or '<!DOCTYPE' in html_report.lower()
Profiler.disable()
Profiler.clear()
def test_file_export(self):
"""Should export report to file."""
import tempfile
import os
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
Profiler.record_timing("test_op", 100.0, "test")
with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
filepath = f.name
try:
Profiler.export_to_file(filepath, format="json")
assert os.path.exists(filepath)
with open(filepath) as f:
content = f.read()
assert len(content) > 0
finally:
if os.path.exists(filepath):
os.unlink(filepath)
Profiler.disable()
Profiler.clear()
# ============================================================================
# Zero Performance Impact Tests
# ============================================================================
class TestZeroPerformanceImpact:
"""Test that profiling has zero impact when disabled."""
def test_disabled_no_overhead(self):
"""Should have minimal overhead when disabled."""
from praisonai.profiler import Profiler, profile
Profiler.disable()
@profile
def fast_function():
return 1 + 1
# Measure overhead
iterations = 10000
start = time.perf_counter()
for _ in range(iterations):
fast_function()
duration = time.perf_counter() - start
# Should complete very fast (< 1ms per call on average)
avg_us = (duration / iterations) * 1_000_000
assert avg_us < 100 # Less than 100 microseconds per call
def test_no_data_when_disabled(self):
"""Should not record data when disabled."""
from praisonai.profiler import Profiler
Profiler.disable()
Profiler.clear()
Profiler.record_timing("test", 100.0, "test")
Profiler.record_api_call("endpoint", "GET", 100.0)
assert len(Profiler.get_timings()) == 0
assert len(Profiler.get_api_calls()) == 0
# ============================================================================
# Async Profiling Tests
# ============================================================================
class TestAsyncProfiling:
"""Test async function profiling."""
@pytest.mark.asyncio
async def test_async_api_call(self):
"""Should profile async API calls."""
from praisonai.profiler import Profiler, profile_api_async
Profiler.enable()
Profiler.clear()
@profile_api_async(endpoint="async_test")
async def async_api_call():
await asyncio.sleep(0.01)
return {"status": "ok"}
result = await async_api_call()
assert result == {"status": "ok"}
calls = Profiler.get_api_calls()
assert len(calls) >= 1
Profiler.disable()
Profiler.clear()
@pytest.mark.asyncio
async def test_async_streaming(self):
"""Should profile async streaming."""
from praisonai.profiler import Profiler
Profiler.enable()
Profiler.clear()
async with Profiler.streaming_async("async_stream") as tracker:
await asyncio.sleep(0.01)
tracker.first_token()
for _ in range(3):
tracker.chunk()
await asyncio.sleep(0.005)
streams = Profiler.get_streaming_records()
assert len(streams) >= 1
Profiler.disable()
Profiler.clear()
+4 -4
View File
@@ -4197,7 +4197,7 @@ requires-dist = [
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.22" },
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.0.22" },
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.22" },
{ name = "praisonaiagents", specifier = ">=0.1.22" },
{ name = "praisonaiagents", specifier = ">=0.1.24" },
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
{ name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
{ name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4252,7 +4252,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.1.22"
version = "0.1.24"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -4261,9 +4261,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/c5/fd/ddc0a0dc7d3d67d49a6e941aa7fbd1fce9c88d27c91f872679b694a3f75a/praisonaiagents-0.1.22.tar.gz", hash = "sha256:673dd4ffc5cc4910642e85da50eb238ae532a179e377fc6d7f0b98a1f8656a54", size = 677941, upload-time = "2025-12-22T15:46:13.8Z" }
sdist = { url = "https://files.pythonhosted.org/packages/31/3e/9dec08dd8107b963cdb3ea3a77b5530e7f68b239c38c80f713be27920f1b/praisonaiagents-0.1.24.tar.gz", hash = "sha256:8a6d17c2df22f2171df11b44b44b817e72a6f34f1f5b4e389691bedbbff06621", size = 678110, upload-time = "2025-12-23T05:39:18.104Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/47/e0/0008cd7b6c8739abfb754986cec83d10a2e3030deffa39dcc19f7689aa3c/praisonaiagents-0.1.22-py3-none-any.whl", hash = "sha256:9a938116fc9312f12e739b5a9a5a5971cd545082b01512b43af4f30d4d5fec87", size = 616650, upload-time = "2025-12-22T15:46:11.677Z" },
{ url = "https://files.pythonhosted.org/packages/37/33/3b45e341433a01408f59ced6eb39da3ecb17067f0ff3c954fa38e07aaffb/praisonaiagents-0.1.24-py3-none-any.whl", hash = "sha256:03a1c476bb5287ad9e519b85a17a93c42ce11b01db87009fcde17a1901e4828a", size = 616792, upload-time = "2025-12-23T05:39:16.12Z" },
]
[[package]]