Compare commits

...
7 Commits
47 changed files with 5577 additions and 57 deletions

No files matched your search

+57
View File
@@ -0,0 +1,57 @@
name: Update API Reference
on:
push:
branches: [ "main" ]
paths:
- 'src/praisonai-agents/praisonaiagents/**/*.py'
- 'src/praisonai/praisonai/**/*.py'
- 'src/praisonai-ts/src/**/*.ts'
- '.github/workflows/update-api-md.yml'
permissions:
contents: write
jobs:
update-api-md:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v4
with:
ref: main
persist-credentials: true
fetch-depth: 0
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Generate api.md
run: |
cd $GITHUB_WORKSPACE
python3 src/praisonai/scripts/generate_api_md.py
- name: Check for changes
id: check_changes
run: |
if git diff --quiet api.md; then
echo "changes=false" >> $GITHUB_OUTPUT
echo "No changes to api.md"
else
echo "changes=true" >> $GITHUB_OUTPUT
echo "api.md has been updated"
git diff api.md | head -50
fi
- name: Commit and push changes
if: steps.check_changes.outputs.changes == 'true'
uses: stefanzweifel/git-auto-commit-action@v5
with:
commit_message: "docs: auto-update api.md [skip ci]"
file_pattern: 'api.md'
commit_user_name: "github-actions[bot]"
commit_user_email: "41898282+github-actions[bot]@users.noreply.github.com"
commit_author: "github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>"
+6 -1
View File
@@ -552,7 +552,7 @@ Methods:
Types:
```python
from praisonaiagents import AutoRagConfig, Citation, CodeExecutionStep, DeepResearchResponse, ExpandResult, ExpandStrategy, FileSearchCall, MCPCall, Provider, RagRetrievalPolicy, ReasoningStep, RewriteResult, RewriteStrategy, Task, WebSearchCall
from praisonaiagents import AutoRagConfig, Citation, CodeExecutionStep, DeepResearchResponse, ExpandResult, ExpandStrategy, FileSearchCall, MCPCall, Provider, RagRetrievalPolicy, ReasoningStep, RewriteResult, RewriteStrategy, Task, WebSearchCall, warmup
```
Methods:
@@ -569,6 +569,7 @@ Methods:
* <code title="class Task">Task.<a href="./src/praisonai-agents/praisonaiagents/task/task.py">execute_callback_sync</a>(task_output: TaskOutput) -> None</code>
* <code title="class Task">Task.<a href="./src/praisonai-agents/praisonaiagents/task/task.py">initialize_memory</a>()</code>
* <code title="class Task">Task.<a href="./src/praisonai-agents/praisonaiagents/task/task.py">store_in_memory</a>(content: str, agent_name: str = None, task_id: str = None)</code>
* <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/__init__.py">warmup</a>(include_litellm: bool = False, include_openai: bool = True) -> dict</code>
# Wrapper (praisonai)
@@ -731,6 +732,10 @@ Methods:
* <code title="cli">praisonai skills install <a href="./src/praisonai/praisonai/cli/commands/skills.py">--help</a></code>
* <code title="cli">praisonai skills list <a href="./src/praisonai/praisonai/cli/commands/skills.py">--help</a></code>
* <code title="cli">praisonai skills validate <a href="./src/praisonai/praisonai/cli/commands/skills.py">--help</a></code>
* <code title="cli">praisonai standardise check <a href="./src/praisonai/praisonai/cli/commands/standardise.py">--help</a></code>
* <code title="cli">praisonai standardise fix <a href="./src/praisonai/praisonai/cli/commands/standardise.py">--help</a></code>
* <code title="cli">praisonai standardise init <a href="./src/praisonai/praisonai/cli/commands/standardise.py">--help</a></code>
* <code title="cli">praisonai standardise report <a href="./src/praisonai/praisonai/cli/commands/standardise.py">--help</a></code>
* <code title="cli">praisonai templates create <a href="./src/praisonai/praisonai/cli/commands/templates.py">--help</a></code>
* <code title="cli">praisonai templates list <a href="./src/praisonai/praisonai/cli/commands/templates.py">--help</a></code>
* <code title="cli">praisonai test info <a href="./src/praisonai/praisonai/cli/commands/test.py">--help</a></code>
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=3.7.2" \
"praisonai>=3.7.4" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=3.7.2" \
"praisonai>=3.7.4" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=3.7.2" \
"praisonai>=3.7.4" \
"praisonai[ui]" \
"praisonai[crewai]"
+63
View File
@@ -0,0 +1,63 @@
"""
Streaming Example with Profiling
Key Points:
1. output="stream" preset enables streaming by default
2. stream=True in start() is redundant when using output="stream" (but doesn't hurt)
3. Profiler.enable() MUST be called before profiling works
4. Use Profiler.streaming() context manager for TTFT metrics
"""
from praisonai.profiler import Profiler, profile, profile_imports
# IMPORTANT: Enable profiler BEFORE any profiling operations
Profiler.enable()
# Profile imports
with profile_imports():
from praisonaiagents import Agent
@profile
def main():
agent = Agent(
instructions="You are a helpful assistant",
output="stream" # This already enables streaming
)
print("\nStreaming response:")
print("-" * 40)
# Use Profiler.streaming() to track TTFT (Time To First Token)
with Profiler.streaming("llm_stream") as tracker:
first_chunk = True
for chunk in agent.start("Write a short paragraph about the history of computing"):
if first_chunk:
tracker.first_token() # Mark TTFT
first_chunk = False
tracker.chunk() # Count chunks
print(chunk, end="", flush=True)
print("\n" + "-" * 40)
# Run with profiling blocks
with Profiler.block("full_workflow"):
with Profiler.block("setup"):
pass
with Profiler.block("execution"):
result = main()
with Profiler.block("cleanup"):
pass
# Generate report
Profiler.report()
# Show streaming-specific stats
streaming_records = Profiler.get_streaming_records()
if streaming_records:
print("\nStreaming Metrics:")
for record in streaming_records:
print(f" {record.name}: TTFT={record.ttft_ms:.2f}ms, Total={record.total_ms:.2f}ms, Chunks={record.chunk_count}")
stats = Profiler.get_statistics()
print(f"\nMedian execution time: {stats['p50']:.2f}ms")
Whitespace-only changes.
@@ -476,6 +476,62 @@ def _init_telemetry():
_init_telemetry()
def warmup(include_litellm: bool = False, include_openai: bool = True) -> dict:
"""
Pre-import heavy dependencies to reduce first-call latency.
NOTE: For default OpenAI usage (llm="gpt-4o-mini"), warmup is NOT needed.
The default path uses the native OpenAI SDK which is fast (~100ms import).
Warmup is only beneficial when using LiteLLM backend, which is triggered by:
- Using "/" in model name (e.g., llm="openai/gpt-4o-mini")
- Passing a dict config (e.g., llm={"model": "gpt-4o-mini"})
- Using base_url parameter
Args:
include_litellm: Pre-import LiteLLM (~2-3s). Only needed for multi-provider support.
include_openai: Pre-import OpenAI SDK (~100ms). Default path, usually fast.
Returns:
dict: Timing information for each component warmed up
Example:
# For LiteLLM multi-provider usage:
from praisonaiagents import warmup
warmup(include_litellm=True) # Pre-load LiteLLM
agent = Agent(llm="anthropic/claude-3-sonnet") # Now faster
# For default OpenAI usage, no warmup needed:
agent = Agent(llm="gpt-4o-mini") # Already fast!
"""
import time
timings = {}
if include_openai:
start = time.perf_counter()
try:
import openai
timings['openai'] = (time.perf_counter() - start) * 1000
except ImportError:
timings['openai'] = -1 # Not available
if include_litellm:
start = time.perf_counter()
try:
import litellm
# Also configure litellm to avoid first-call overhead
litellm.telemetry = False
litellm.set_verbose = False
litellm.drop_params = True
litellm.modify_params = True
timings['litellm'] = (time.perf_counter() - start) * 1000
except ImportError:
timings['litellm'] = -1 # Not available
return timings
__all__ = [
# Core classes (lazy loaded)
'Agent',
@@ -646,4 +702,6 @@ __all__ = [
# Parse Utilities
'is_policy_string',
'parse_policy_string',
# Performance utilities
'warmup',
]
@@ -1430,6 +1430,8 @@ Now provide your final answer using this result."""
task_id: Optional[str] = None,
execute_tool_fn: Optional[Callable] = None,
stream: bool = True,
stream_callback: Optional[Callable] = None,
emit_events: bool = False,
**kwargs
) -> str:
"""Enhanced get_response with all OpenAI-like features"""
@@ -1489,6 +1491,19 @@ Now provide your final answer using this result."""
# Disable litellm debug messages
litellm.set_verbose = False
# Setup StreamEvent emission if enabled
_emit = emit_events and stream_callback is not None
if _emit:
from ..streaming.events import StreamEvent, StreamEventType
_first_token_emitted = False
_last_content_time = None
_request_start_perf = time.perf_counter()
stream_callback(StreamEvent(
type=StreamEventType.REQUEST_START,
timestamp=_request_start_perf,
metadata={"model": self.model, "provider": "litellm"}
))
# Format tools if provided
formatted_tools = self._format_tools_for_litellm(tools)
@@ -582,6 +582,8 @@ class OpenAIClient:
console: Optional[Console] = None,
display_fn: Optional[Callable] = None,
reasoning_steps: bool = False,
stream_callback: Optional[Callable] = None,
emit_events: bool = False,
**kwargs
) -> Optional[ChatCompletion]:
"""
@@ -596,11 +598,18 @@ class OpenAIClient:
console: Console for output
display_fn: Display function for live updates
reasoning_steps: Whether to show reasoning steps
stream_callback: Optional callback for StreamEvent emission
emit_events: Whether to emit StreamEvents (requires stream_callback)
**kwargs: Additional parameters for the API
Returns:
ChatCompletion object or None if error
"""
# Lazy import StreamEvent types only when needed
_emit = emit_events and stream_callback is not None
if _emit:
from ..streaming.events import StreamEvent, StreamEventType
try:
# Default start time and console if not provided
if start_time is None:
@@ -608,6 +617,15 @@ class OpenAIClient:
if console is None:
console = self.console
# Emit REQUEST_START event
request_start_perf = time.perf_counter()
if _emit:
stream_callback(StreamEvent(
type=StreamEventType.REQUEST_START,
timestamp=request_start_perf,
metadata={"model": model, "message_count": len(messages)}
))
# Create the response stream
response_stream = self.sync_client.chat.completions.create(
model=model,
@@ -615,29 +633,74 @@ class OpenAIClient:
temperature=temperature,
tools=tools if tools else None,
stream=True,
stream_options={"include_usage": True} if _emit else None,
**kwargs
)
# Emit HEADERS_RECEIVED event (stream object created means headers received)
if _emit:
stream_callback(StreamEvent(
type=StreamEventType.HEADERS_RECEIVED,
timestamp=time.perf_counter()
))
full_response_text = ""
reasoning_content = ""
chunks = []
first_token_emitted = False
last_content_time = None
# If display function provided, use Live display
if display_fn:
with Live(
display_fn("", start_time),
console=console,
refresh_per_second=4,
refresh_per_second=10, # Increased for more responsive streaming
transient=True,
vertical_overflow="ellipsis",
auto_refresh=True
) as live:
for chunk in response_stream:
chunks.append(chunk)
if chunk.choices[0].delta.content:
full_response_text += chunk.choices[0].delta.content
# Check for content delta
if chunk.choices and chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
full_response_text += content
last_content_time = time.perf_counter()
# Emit FIRST_TOKEN on first content
if _emit and not first_token_emitted:
stream_callback(StreamEvent(
type=StreamEventType.FIRST_TOKEN,
timestamp=last_content_time,
content=content
))
first_token_emitted = True
elif _emit:
# Emit DELTA_TEXT for subsequent tokens
stream_callback(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=last_content_time,
content=content
))
live.update(display_fn(full_response_text, start_time))
# Handle tool calls streaming
if _emit and chunk.choices and hasattr(chunk.choices[0].delta, 'tool_calls') and chunk.choices[0].delta.tool_calls:
for tc in chunk.choices[0].delta.tool_calls:
stream_callback(StreamEvent(
type=StreamEventType.DELTA_TOOL_CALL,
timestamp=time.perf_counter(),
tool_call={
"index": tc.index,
"id": getattr(tc, 'id', None),
"name": getattr(tc.function, 'name', None) if tc.function else None,
"arguments": getattr(tc.function, 'arguments', None) if tc.function else None
}
))
# Update live display with reasoning content if enabled
if reasoning_steps and hasattr(chunk.choices[0].delta, "reasoning_content"):
rc = chunk.choices[0].delta.reasoning_content
@@ -648,9 +711,54 @@ class OpenAIClient:
# Clear the last generating display with a blank line
console.print()
else:
# Just collect chunks without display
# Just collect chunks without display, but still emit events
for chunk in response_stream:
chunks.append(chunk)
if _emit and chunk.choices and chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
last_content_time = time.perf_counter()
if not first_token_emitted:
stream_callback(StreamEvent(
type=StreamEventType.FIRST_TOKEN,
timestamp=last_content_time,
content=content
))
first_token_emitted = True
else:
stream_callback(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=last_content_time,
content=content
))
# Handle tool calls streaming
if _emit and chunk.choices and hasattr(chunk.choices[0].delta, 'tool_calls') and chunk.choices[0].delta.tool_calls:
for tc in chunk.choices[0].delta.tool_calls:
stream_callback(StreamEvent(
type=StreamEventType.DELTA_TOOL_CALL,
timestamp=time.perf_counter(),
tool_call={
"index": tc.index,
"id": getattr(tc, 'id', None),
"name": getattr(tc.function, 'name', None) if tc.function else None,
"arguments": getattr(tc.function, 'arguments', None) if tc.function else None
}
))
# Emit LAST_TOKEN and STREAM_END events
if _emit:
if last_content_time:
stream_callback(StreamEvent(
type=StreamEventType.LAST_TOKEN,
timestamp=last_content_time
))
stream_callback(StreamEvent(
type=StreamEventType.STREAM_END,
timestamp=time.perf_counter(),
metadata={"chunk_count": len(chunks)}
))
final_response = process_stream_chunks(chunks)
return final_response
@@ -669,6 +777,8 @@ class OpenAIClient:
console: Optional[Console] = None,
display_fn: Optional[Callable] = None,
reasoning_steps: bool = False,
stream_callback: Optional[Callable] = None,
emit_events: bool = False,
**kwargs
) -> Optional[ChatCompletion]:
"""
@@ -683,11 +793,18 @@ class OpenAIClient:
console: Console for output
display_fn: Display function for live updates
reasoning_steps: Whether to show reasoning steps
stream_callback: Optional callback for StreamEvent emission (can be sync or async)
emit_events: Whether to emit StreamEvents (requires stream_callback)
**kwargs: Additional parameters for the API
Returns:
ChatCompletion object or None if error
"""
# Lazy import StreamEvent types only when needed
_emit = emit_events and stream_callback is not None
if _emit:
from ..streaming.events import StreamEvent, StreamEventType
try:
# Default start time and console if not provided
if start_time is None:
@@ -695,6 +812,19 @@ class OpenAIClient:
if console is None:
console = self.console
# Emit REQUEST_START event
request_start_perf = time.perf_counter()
if _emit:
event = StreamEvent(
type=StreamEventType.REQUEST_START,
timestamp=request_start_perf,
metadata={"model": model, "message_count": len(messages)}
)
if asyncio.iscoroutinefunction(stream_callback):
await stream_callback(event)
else:
stream_callback(event)
# Create the response stream
response_stream = await self.async_client.chat.completions.create(
model=model,
@@ -702,29 +832,81 @@ class OpenAIClient:
temperature=temperature,
tools=tools if tools else None,
stream=True,
stream_options={"include_usage": True} if _emit else None,
**kwargs
)
# Emit HEADERS_RECEIVED event
if _emit:
event = StreamEvent(
type=StreamEventType.HEADERS_RECEIVED,
timestamp=time.perf_counter()
)
if asyncio.iscoroutinefunction(stream_callback):
await stream_callback(event)
else:
stream_callback(event)
full_response_text = ""
reasoning_content = ""
chunks = []
first_token_emitted = False
last_content_time = None
async def emit_event(event: 'StreamEvent'):
if asyncio.iscoroutinefunction(stream_callback):
await stream_callback(event)
else:
stream_callback(event)
# If display function provided, use Live display
if display_fn:
with Live(
display_fn("", start_time),
console=console,
refresh_per_second=4,
refresh_per_second=10,
transient=True,
vertical_overflow="ellipsis",
auto_refresh=True
) as live:
async for chunk in response_stream:
chunks.append(chunk)
if chunk.choices[0].delta.content:
full_response_text += chunk.choices[0].delta.content
if chunk.choices and chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
full_response_text += content
last_content_time = time.perf_counter()
if _emit and not first_token_emitted:
await emit_event(StreamEvent(
type=StreamEventType.FIRST_TOKEN,
timestamp=last_content_time,
content=content
))
first_token_emitted = True
elif _emit:
await emit_event(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=last_content_time,
content=content
))
live.update(display_fn(full_response_text, start_time))
# Handle tool calls streaming
if _emit and chunk.choices and hasattr(chunk.choices[0].delta, 'tool_calls') and chunk.choices[0].delta.tool_calls:
for tc in chunk.choices[0].delta.tool_calls:
await emit_event(StreamEvent(
type=StreamEventType.DELTA_TOOL_CALL,
timestamp=time.perf_counter(),
tool_call={
"index": tc.index,
"id": getattr(tc, 'id', None),
"name": getattr(tc.function, 'name', None) if tc.function else None,
"arguments": getattr(tc.function, 'arguments', None) if tc.function else None
}
))
# Update live display with reasoning content if enabled
if reasoning_steps and hasattr(chunk.choices[0].delta, "reasoning_content"):
rc = chunk.choices[0].delta.reasoning_content
@@ -738,6 +920,50 @@ class OpenAIClient:
# Just collect chunks without display
async for chunk in response_stream:
chunks.append(chunk)
if _emit and chunk.choices and chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
last_content_time = time.perf_counter()
if not first_token_emitted:
await emit_event(StreamEvent(
type=StreamEventType.FIRST_TOKEN,
timestamp=last_content_time,
content=content
))
first_token_emitted = True
else:
await emit_event(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=last_content_time,
content=content
))
if _emit and chunk.choices and hasattr(chunk.choices[0].delta, 'tool_calls') and chunk.choices[0].delta.tool_calls:
for tc in chunk.choices[0].delta.tool_calls:
await emit_event(StreamEvent(
type=StreamEventType.DELTA_TOOL_CALL,
timestamp=time.perf_counter(),
tool_call={
"index": tc.index,
"id": getattr(tc, 'id', None),
"name": getattr(tc.function, 'name', None) if tc.function else None,
"arguments": getattr(tc.function, 'arguments', None) if tc.function else None
}
))
# Emit LAST_TOKEN and STREAM_END events
if _emit:
if last_content_time:
await emit_event(StreamEvent(
type=StreamEventType.LAST_TOKEN,
timestamp=last_content_time
))
await emit_event(StreamEvent(
type=StreamEventType.STREAM_END,
timestamp=time.perf_counter(),
metadata={"chunk_count": len(chunks)}
))
final_response = process_stream_chunks(chunks)
return final_response
@@ -0,0 +1,479 @@
"""
PraisonAI Agents Core Profiling Module
Lightweight, protocol-based profiling for the core SDK.
Zero overhead when disabled. Opt-in only.
Features:
- Pre-LLM latency breakdown
- LLM client initialization timing
- Tool resolution timing
- Message serialization timing
- Streaming TTFT tracking
- Import timing
- Async-safe
Usage:
from praisonaiagents.profiling import Profiler, profile_block
# Enable profiling
Profiler.enable()
# Profile a block
with profile_block("my_operation"):
do_something()
# Get report
Profiler.report()
"""
import time
import os
import threading
from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional, Any, Callable
from contextlib import contextmanager
import functools
import json
# ============================================================================
# Data Classes
# ============================================================================
@dataclass
class TimingRecord:
"""Record of a single timing measurement."""
name: str
duration_ms: float
category: str = "block"
parent: str = ""
timestamp: float = field(default_factory=time.time)
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class StreamingRecord:
"""Record of streaming operation."""
name: str
ttft_ms: float # Time to first token
total_ms: float
chunk_count: int = 0
timestamp: float = field(default_factory=time.time)
@dataclass
class PreLLMBreakdown:
"""Breakdown of pre-LLM latency phases."""
agent_init_ms: float = 0.0
llm_client_init_ms: float = 0.0
tool_resolution_ms: float = 0.0
message_build_ms: float = 0.0
context_management_ms: float = 0.0
knowledge_retrieval_ms: float = 0.0
guardrail_check_ms: float = 0.0
total_pre_llm_ms: float = 0.0
def to_dict(self) -> Dict[str, float]:
return asdict(self)
# ============================================================================
# Streaming Tracker
# ============================================================================
class StreamingTracker:
"""Track streaming operations for TTFT metrics."""
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
def start(self) -> None:
self._start_time = time.perf_counter()
def first_token(self) -> None:
if self._first_token_time is None:
self._first_token_time = time.perf_counter()
def chunk(self) -> None:
self._chunk_count += 1
def end(self) -> None:
self._end_time = time.perf_counter()
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
)
@property
def ttft_ms(self) -> float:
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:
if self._start_time:
end = self._end_time or time.perf_counter()
return (end - self._start_time) * 1000
return 0.0
# ============================================================================
# Core Profiler
# ============================================================================
class Profiler:
"""
Lightweight profiler for PraisonAI Agents core SDK.
Thread-safe singleton with zero overhead when disabled.
"""
_lock = threading.Lock()
_enabled: bool = False
_timings: List[TimingRecord] = []
_streaming: List[StreamingRecord] = []
_pre_llm_breakdown: Optional[PreLLMBreakdown] = None
_current_phases: Dict[str, float] = {} # Track in-progress phases
_phase_stack: List[str] = [] # Track nested phases
@classmethod
def enable(cls) -> None:
"""Enable profiling."""
cls._enabled = True
@classmethod
def disable(cls) -> None:
"""Disable profiling."""
cls._enabled = False
@classmethod
def is_enabled(cls) -> bool:
"""Check if profiling is enabled."""
return cls._enabled or os.environ.get('PRAISONAI_PROFILE', '').lower() in ('1', 'true', 'yes')
@classmethod
def clear(cls) -> None:
"""Clear all profiling data."""
with cls._lock:
cls._timings.clear()
cls._streaming.clear()
cls._pre_llm_breakdown = None
cls._current_phases.clear()
cls._phase_stack.clear()
@classmethod
def record_timing(cls, name: str, duration_ms: float, category: str = "block",
parent: str = "", metadata: Optional[Dict[str, Any]] = None) -> None:
"""Record a timing measurement."""
if not cls.is_enabled():
return
with cls._lock:
cls._timings.append(TimingRecord(
name=name,
duration_ms=duration_ms,
category=category,
parent=parent,
metadata=metadata or {}
))
@classmethod
def record_streaming(cls, name: str, ttft_ms: float, total_ms: float,
chunk_count: 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
))
@classmethod
def start_phase(cls, phase_name: str) -> None:
"""Start timing a phase (for pre-LLM breakdown)."""
if not cls.is_enabled():
return
with cls._lock:
cls._current_phases[phase_name] = time.perf_counter()
cls._phase_stack.append(phase_name)
@classmethod
def end_phase(cls, phase_name: str) -> float:
"""End timing a phase and return duration in ms."""
if not cls.is_enabled():
return 0.0
with cls._lock:
start = cls._current_phases.pop(phase_name, None)
if phase_name in cls._phase_stack:
cls._phase_stack.remove(phase_name)
if start is None:
return 0.0
duration_ms = (time.perf_counter() - start) * 1000
# Record as timing
parent = cls._phase_stack[-1] if cls._phase_stack else ""
cls._timings.append(TimingRecord(
name=phase_name,
duration_ms=duration_ms,
category="phase",
parent=parent
))
return duration_ms
@classmethod
@contextmanager
def phase(cls, name: str, category: str = "phase"):
"""Context manager for timing a phase."""
if not cls.is_enabled():
yield
return
start = time.perf_counter()
parent = ""
with cls._lock:
parent = cls._phase_stack[-1] if cls._phase_stack else ""
cls._phase_stack.append(name)
try:
yield
finally:
duration_ms = (time.perf_counter() - start) * 1000
with cls._lock:
if name in cls._phase_stack:
cls._phase_stack.remove(name)
cls._timings.append(TimingRecord(
name=name,
duration_ms=duration_ms,
category=category,
parent=parent
))
@classmethod
@contextmanager
def streaming(cls, name: str):
"""Context manager for streaming profiling."""
tracker = StreamingTracker(name)
tracker.start()
try:
yield tracker
finally:
tracker.end()
@classmethod
def set_pre_llm_breakdown(cls, breakdown: PreLLMBreakdown) -> None:
"""Set the pre-LLM breakdown data."""
if not cls.is_enabled():
return
with cls._lock:
cls._pre_llm_breakdown = breakdown
@classmethod
def get_timings(cls, category: Optional[str] = None) -> List[TimingRecord]:
"""Get timing records."""
with cls._lock:
if category:
return [t for t in cls._timings if t.category == category]
return cls._timings.copy()
@classmethod
def get_streaming_records(cls) -> List[StreamingRecord]:
"""Get streaming records."""
with cls._lock:
return cls._streaming.copy()
@classmethod
def get_pre_llm_breakdown(cls) -> Optional[PreLLMBreakdown]:
"""Get pre-LLM breakdown."""
with cls._lock:
return cls._pre_llm_breakdown
@classmethod
def get_summary(cls) -> Dict[str, Any]:
"""Get profiling summary."""
with cls._lock:
total_time = sum(t.duration_ms for t in cls._timings)
# Group by category
by_category: Dict[str, float] = {}
for t in cls._timings:
by_category[t.category] = by_category.get(t.category, 0) + t.duration_ms
# Top slowest
slowest = sorted(cls._timings, key=lambda x: x.duration_ms, reverse=True)[:15]
# Pre-LLM breakdown
pre_llm = cls._pre_llm_breakdown.to_dict() if cls._pre_llm_breakdown else {}
# Streaming stats
streaming_stats = []
for s in cls._streaming:
streaming_stats.append({
'name': s.name,
'ttft_ms': s.ttft_ms,
'total_ms': s.total_ms,
'chunk_count': s.chunk_count
})
return {
'total_time_ms': total_time,
'timing_count': len(cls._timings),
'by_category': by_category,
'slowest_operations': [(s.name, s.duration_ms, s.category) for s in slowest],
'pre_llm_breakdown': pre_llm,
'streaming': streaming_stats
}
@classmethod
def report(cls, output: str = "console") -> str:
"""Generate and output profiling report."""
summary = cls.get_summary()
lines = [
"=" * 70,
"PraisonAI Agents Core Profiling Report",
"=" * 70,
"",
f"Total Time: {summary['total_time_ms']:.2f}ms",
f"Timing Records: {summary['timing_count']}",
"",
]
# Pre-LLM Breakdown (key for debugging the 7s delay)
if summary['pre_llm_breakdown']:
lines.append("## Pre-LLM Latency Breakdown")
lines.append("-" * 50)
for phase, ms in summary['pre_llm_breakdown'].items():
if ms > 0:
lines.append(f" {phase:<30}: {ms:>10.2f} ms")
lines.append("")
# By Category
lines.append("## By Category")
lines.append("-" * 50)
for cat, time_ms in summary['by_category'].items():
lines.append(f" {cat}: {time_ms:.2f}ms")
lines.append("")
# Slowest Operations
lines.append("## Slowest Operations")
lines.append("-" * 50)
for name, time_ms, cat in summary['slowest_operations']:
lines.append(f" {name:<40}: {time_ms:>10.2f}ms [{cat}]")
lines.append("")
# Streaming Stats
if summary['streaming']:
lines.append("## Streaming Metrics")
lines.append("-" * 50)
for s in summary['streaming']:
lines.append(f" {s['name']}: TTFT={s['ttft_ms']:.2f}ms, Total={s['total_ms']:.2f}ms, Chunks={s['chunk_count']}")
lines.append("")
lines.append("=" * 70)
report_text = "\n".join(lines)
if output == "console":
print(report_text)
return report_text
@classmethod
def export_json(cls) -> str:
"""Export profiling data as JSON."""
summary = cls.get_summary()
with cls._lock:
data = {
'summary': summary,
'timings': [asdict(t) for t in cls._timings],
'streaming': [asdict(s) for s in cls._streaming],
}
return json.dumps(data, indent=2, default=str)
# ============================================================================
# Decorators
# ============================================================================
def profile_func(func: Optional[Callable] = None, *, category: str = "function"):
"""Decorator to profile a function."""
def decorator(fn: Callable) -> Callable:
@functools.wraps(fn)
def wrapper(*args, **kwargs):
if not Profiler.is_enabled():
return fn(*args, **kwargs)
start = time.perf_counter()
try:
return fn(*args, **kwargs)
finally:
duration_ms = (time.perf_counter() - start) * 1000
Profiler.record_timing(fn.__name__, duration_ms, category)
return wrapper
if func is not None:
return decorator(func)
return decorator
def profile_async(func: Optional[Callable] = None, *, category: str = "async"):
"""Decorator to profile an async function."""
def decorator(fn: Callable) -> Callable:
@functools.wraps(fn)
async def wrapper(*args, **kwargs):
if not Profiler.is_enabled():
return await fn(*args, **kwargs)
start = time.perf_counter()
try:
return await fn(*args, **kwargs)
finally:
duration_ms = (time.perf_counter() - start) * 1000
Profiler.record_timing(fn.__name__, duration_ms, category)
return wrapper
if func is not None:
return decorator(func)
return decorator
@contextmanager
def profile_block(name: str, category: str = "block"):
"""Context manager for profiling a block of code."""
if not Profiler.is_enabled():
yield
return
start = time.perf_counter()
try:
yield
finally:
duration_ms = (time.perf_counter() - start) * 1000
Profiler.record_timing(name, duration_ms, category)
# ============================================================================
# Exports
# ============================================================================
__all__ = [
'Profiler',
'StreamingTracker',
'TimingRecord',
'StreamingRecord',
'PreLLMBreakdown',
'profile_func',
'profile_async',
'profile_block',
]
@@ -0,0 +1,75 @@
"""
PraisonAI Streaming Module.
Provides the StreamEvent protocol for pass-through streaming from LLM providers.
This module is lazily loaded to ensure zero performance impact when streaming
features are not used.
Usage:
from praisonaiagents.streaming import (
StreamEvent,
StreamEventType,
StreamMetrics,
StreamEventEmitter,
create_text_printer_callback,
)
"""
__all__ = [
"StreamEvent",
"StreamEventType",
"StreamMetrics",
"StreamEventEmitter",
"StreamCallback",
"AsyncStreamCallback",
"create_text_printer_callback",
"create_metrics_callback",
"StreamLogger",
"create_logging_callback",
]
def __getattr__(name: str):
"""Lazy loading of streaming components."""
if name in __all__:
# Events module exports
if name in (
"StreamEvent", "StreamEventType", "StreamMetrics",
"StreamEventEmitter", "StreamCallback", "AsyncStreamCallback",
"create_text_printer_callback", "create_metrics_callback"
):
from .events import (
StreamEvent,
StreamEventType,
StreamMetrics,
StreamEventEmitter,
StreamCallback,
AsyncStreamCallback,
create_text_printer_callback,
create_metrics_callback,
)
_events_exports = {
"StreamEvent": StreamEvent,
"StreamEventType": StreamEventType,
"StreamMetrics": StreamMetrics,
"StreamEventEmitter": StreamEventEmitter,
"StreamCallback": StreamCallback,
"AsyncStreamCallback": AsyncStreamCallback,
"create_text_printer_callback": create_text_printer_callback,
"create_metrics_callback": create_metrics_callback,
}
return _events_exports.get(name)
# Logging module exports
if name in ("StreamLogger", "create_logging_callback"):
from .logging import StreamLogger, create_logging_callback
_logging_exports = {
"StreamLogger": StreamLogger,
"create_logging_callback": create_logging_callback,
}
return _logging_exports.get(name)
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -0,0 +1,300 @@
"""
StreamEvent protocol for PraisonAI streaming.
This module defines the minimal streaming event protocol that both OpenAI SDK
and LiteLLM streaming paths map their native chunks/events into.
Design principles:
- Zero overhead when not used (lazy imports, no global state)
- Multi-agent safe (per-agent callbacks, no shared mutable state)
- Async-safe (separate sync/async callback interfaces)
- Provider-agnostic (same events for OpenAI, LiteLLM, etc.)
"""
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional, Protocol
import time
class StreamEventType(Enum):
"""Types of streaming events emitted during LLM response streaming."""
REQUEST_START = "request_start" # Before API call is made
HEADERS_RECEIVED = "headers_received" # When HTTP 200 headers arrive
FIRST_TOKEN = "first_token" # First content delta received (TTFT marker)
DELTA_TEXT = "delta_text" # Text content delta
DELTA_TOOL_CALL = "delta_tool_call" # Tool call delta (function name, args chunk)
TOOL_CALL_END = "tool_call_end" # Tool call complete
LAST_TOKEN = "last_token" # Final content delta
STREAM_END = "stream_end" # Stream completed successfully
ERROR = "error" # Error during streaming
@dataclass
class StreamEvent:
"""
A single streaming event emitted during LLM response streaming.
Attributes:
type: The type of event (see StreamEventType)
timestamp: High-precision timestamp (time.perf_counter())
content: Text content for DELTA_TEXT events
tool_call: Tool call data for DELTA_TOOL_CALL events
metadata: Additional event-specific metadata
error: Error message for ERROR events
"""
type: StreamEventType
timestamp: float = field(default_factory=time.perf_counter)
content: Optional[str] = None
tool_call: Optional[Dict[str, Any]] = None
metadata: Optional[Dict[str, Any]] = None
error: Optional[str] = None
def __repr__(self) -> str:
parts = [f"StreamEvent({self.type.value}"]
if self.content:
preview = self.content[:20] + "..." if len(self.content) > 20 else self.content
parts.append(f", content={preview!r}")
if self.tool_call:
parts.append(f", tool_call={self.tool_call.get('name', 'unknown')}")
if self.error:
parts.append(f", error={self.error!r}")
parts.append(")")
return "".join(parts)
class StreamCallback(Protocol):
"""Protocol for synchronous stream event callbacks."""
def __call__(self, event: StreamEvent) -> None:
"""Called when a stream event is emitted."""
...
class AsyncStreamCallback(Protocol):
"""Protocol for asynchronous stream event callbacks."""
async def __call__(self, event: StreamEvent) -> None:
"""Called when a stream event is emitted (async)."""
...
@dataclass
class StreamMetrics:
"""
Timing metrics for a streaming response.
Tracks key timestamps to measure:
- TTFT (Time To First Token): How long until first token arrives
- Stream Duration: How long the token streaming takes
- Total Time: End-to-end request time
All times are in seconds, measured with time.perf_counter() for precision.
"""
request_start: float = 0.0
headers_received: float = 0.0
first_token: float = 0.0
last_token: float = 0.0
stream_end: float = 0.0
token_count: int = 0
@property
def ttft(self) -> float:
"""Time To First Token in seconds (first_token - request_start)."""
if self.first_token and self.request_start:
return self.first_token - self.request_start
return 0.0
@property
def stream_duration(self) -> float:
"""Duration of token streaming in seconds (last_token - first_token)."""
if self.last_token and self.first_token:
return self.last_token - self.first_token
return 0.0
@property
def total_time(self) -> float:
"""Total request time in seconds (stream_end - request_start)."""
if self.stream_end and self.request_start:
return self.stream_end - self.request_start
return 0.0
@property
def tokens_per_second(self) -> float:
"""Token generation rate (tokens / stream_duration)."""
if self.stream_duration > 0 and self.token_count > 0:
return self.token_count / self.stream_duration
return 0.0
def update_from_event(self, event: StreamEvent) -> None:
"""Update metrics from a StreamEvent."""
if event.type == StreamEventType.REQUEST_START:
self.request_start = event.timestamp
elif event.type == StreamEventType.HEADERS_RECEIVED:
self.headers_received = event.timestamp
elif event.type == StreamEventType.FIRST_TOKEN:
self.first_token = event.timestamp
self.token_count = 1
elif event.type == StreamEventType.DELTA_TEXT:
self.token_count += 1
elif event.type == StreamEventType.LAST_TOKEN:
self.last_token = event.timestamp
elif event.type == StreamEventType.STREAM_END:
self.stream_end = event.timestamp
def to_dict(self) -> Dict[str, Any]:
"""Convert metrics to dictionary for logging/display."""
return {
"ttft_ms": round(self.ttft * 1000, 1),
"stream_duration_ms": round(self.stream_duration * 1000, 1),
"total_time_ms": round(self.total_time * 1000, 1),
"token_count": self.token_count,
"tokens_per_second": round(self.tokens_per_second, 1),
}
def format_summary(self) -> str:
"""Format metrics as a human-readable summary string."""
return (
f"TTFT: {self.ttft*1000:.0f}ms | "
f"Stream: {self.stream_duration*1000:.0f}ms | "
f"Total: {self.total_time*1000:.0f}ms | "
f"Tokens: {self.token_count} ({self.tokens_per_second:.1f}/s)"
)
class StreamEventEmitter:
"""
Manages stream event callbacks for an agent or streaming session.
Thread-safe and multi-agent safe - each agent has its own emitter instance.
Supports both sync and async callbacks.
Usage:
emitter = StreamEventEmitter()
emitter.add_callback(my_callback)
# During streaming:
emitter.emit(StreamEvent(type=StreamEventType.DELTA_TEXT, content="Hello"))
"""
def __init__(self) -> None:
self._callbacks: List[StreamCallback] = []
self._async_callbacks: List[AsyncStreamCallback] = []
self._metrics: Optional[StreamMetrics] = None
self._collect_metrics: bool = False
def add_callback(self, callback: StreamCallback) -> None:
"""Add a synchronous callback to receive stream events."""
self._callbacks.append(callback)
def remove_callback(self, callback: StreamCallback) -> None:
"""Remove a synchronous callback."""
if callback in self._callbacks:
self._callbacks.remove(callback)
def add_async_callback(self, callback: AsyncStreamCallback) -> None:
"""Add an asynchronous callback to receive stream events."""
self._async_callbacks.append(callback)
def remove_async_callback(self, callback: AsyncStreamCallback) -> None:
"""Remove an asynchronous callback."""
if callback in self._async_callbacks:
self._async_callbacks.remove(callback)
def enable_metrics(self) -> None:
"""Enable metrics collection for this streaming session."""
self._collect_metrics = True
self._metrics = StreamMetrics()
def get_metrics(self) -> Optional[StreamMetrics]:
"""Get collected metrics (None if metrics not enabled)."""
return self._metrics
def reset_metrics(self) -> None:
"""Reset metrics for a new streaming session."""
if self._collect_metrics:
self._metrics = StreamMetrics()
def emit(self, event: StreamEvent) -> None:
"""
Emit a stream event to all registered callbacks.
This is the synchronous emission path. For async streaming,
use emit_async().
"""
# Update metrics if enabled
if self._collect_metrics and self._metrics:
self._metrics.update_from_event(event)
# Call all sync callbacks
for callback in self._callbacks:
try:
callback(event)
except Exception:
# Don't let callback errors break streaming
pass
async def emit_async(self, event: StreamEvent) -> None:
"""
Emit a stream event to all registered callbacks (async version).
Calls both sync and async callbacks.
"""
# Update metrics if enabled
if self._collect_metrics and self._metrics:
self._metrics.update_from_event(event)
# Call sync callbacks
for callback in self._callbacks:
try:
callback(event)
except Exception:
pass
# Call async callbacks
for callback in self._async_callbacks:
try:
await callback(event)
except Exception:
pass
@property
def has_callbacks(self) -> bool:
"""Check if any callbacks are registered."""
return bool(self._callbacks or self._async_callbacks)
def create_text_printer_callback(flush: bool = True) -> StreamCallback:
"""
Create a simple callback that prints text deltas to stdout.
Args:
flush: Whether to flush stdout after each print (default True)
Returns:
A StreamCallback that prints DELTA_TEXT content
"""
def printer(event: StreamEvent) -> None:
if event.type == StreamEventType.DELTA_TEXT and event.content:
print(event.content, end="", flush=flush)
elif event.type == StreamEventType.STREAM_END:
print() # Newline at end
return printer
def create_metrics_callback(metrics: StreamMetrics) -> StreamCallback:
"""
Create a callback that updates a StreamMetrics instance.
Args:
metrics: The StreamMetrics instance to update
Returns:
A StreamCallback that updates the metrics
"""
def updater(event: StreamEvent) -> None:
metrics.update_from_event(event)
return updater
@@ -0,0 +1,207 @@
"""
Streaming logging utilities for PraisonAI.
Provides guarded logging for streaming events with clear timestamps
for request_start, headers_received, first_token, last_token, and stream_end.
Usage:
from praisonaiagents.streaming.logging import StreamLogger
logger = StreamLogger(verbose=True, metrics=True)
logger.log_request_start(model="gpt-4o-mini")
# ... streaming ...
logger.log_first_token()
# ... more tokens ...
logger.log_stream_end()
print(logger.get_metrics_summary())
"""
import logging
import time
from typing import Optional, Dict, Any
from .events import StreamMetrics, StreamEvent, StreamEventType
class StreamLogger:
"""
Guarded logging for streaming events with timing metrics.
Only logs when verbose=True or metrics=True to ensure zero overhead
when logging is disabled.
Attributes:
verbose: Whether to log detailed streaming events
metrics: Whether to collect and display timing metrics
"""
def __init__(
self,
verbose: bool = False,
metrics: bool = False,
logger: Optional[logging.Logger] = None
):
self.verbose = verbose
self.metrics = metrics
self._logger = logger or logging.getLogger(__name__)
self._metrics = StreamMetrics() if metrics else None
self._enabled = verbose or metrics
def log_request_start(
self,
model: Optional[str] = None,
message_count: Optional[int] = None
) -> None:
"""Log when the streaming request is initiated."""
if not self._enabled:
return
timestamp = time.perf_counter()
if self._metrics:
self._metrics.request_start = timestamp
if self.verbose:
extra = f" model={model}" if model else ""
extra += f" messages={message_count}" if message_count else ""
self._logger.info(f"[STREAM] REQUEST_START{extra}")
def log_headers_received(self) -> None:
"""Log when HTTP response headers are received."""
if not self._enabled:
return
timestamp = time.perf_counter()
if self._metrics:
self._metrics.headers_received = timestamp
if self.verbose:
self._logger.info("[STREAM] HEADERS_RECEIVED")
def log_first_token(self, content: Optional[str] = None) -> None:
"""Log when the first content token is received (TTFT marker)."""
if not self._enabled:
return
timestamp = time.perf_counter()
if self._metrics:
self._metrics.first_token = timestamp
self._metrics.token_count = 1
if self.verbose:
preview = content[:20] + "..." if content and len(content) > 20 else content
self._logger.info(f"[STREAM] FIRST_TOKEN content={preview!r}")
def log_delta_text(self, content: Optional[str] = None) -> None:
"""Log a text delta (only when verbose, not for every token)."""
if not self._enabled:
return
if self._metrics:
self._metrics.token_count += 1
# Don't log every delta to avoid noise - just track metrics
def log_last_token(self) -> None:
"""Log when the last content token is received."""
if not self._enabled:
return
timestamp = time.perf_counter()
if self._metrics:
self._metrics.last_token = timestamp
if self.verbose:
self._logger.info("[STREAM] LAST_TOKEN")
def log_stream_end(self, chunk_count: Optional[int] = None) -> None:
"""Log when the stream ends."""
if not self._enabled:
return
timestamp = time.perf_counter()
if self._metrics:
self._metrics.stream_end = timestamp
if self.verbose:
extra = f" chunks={chunk_count}" if chunk_count else ""
self._logger.info(f"[STREAM] STREAM_END{extra}")
def log_error(self, error: str) -> None:
"""Log a streaming error."""
if self.verbose:
self._logger.error(f"[STREAM] ERROR: {error}")
def get_metrics(self) -> Optional[StreamMetrics]:
"""Get the collected metrics (None if metrics not enabled)."""
return self._metrics
def get_metrics_summary(self) -> str:
"""Get a formatted summary of streaming metrics."""
if not self._metrics:
return "Metrics not enabled"
return self._metrics.format_summary()
def get_metrics_dict(self) -> Optional[Dict[str, Any]]:
"""Get metrics as a dictionary."""
if not self._metrics:
return None
return self._metrics.to_dict()
def update_from_event(self, event: StreamEvent) -> None:
"""Update logger state from a StreamEvent."""
if not self._enabled:
return
if self._metrics:
self._metrics.update_from_event(event)
if self.verbose:
if event.type == StreamEventType.REQUEST_START:
self.log_request_start()
elif event.type == StreamEventType.HEADERS_RECEIVED:
self.log_headers_received()
elif event.type == StreamEventType.FIRST_TOKEN:
self.log_first_token(event.content)
elif event.type == StreamEventType.LAST_TOKEN:
self.log_last_token()
elif event.type == StreamEventType.STREAM_END:
self.log_stream_end()
elif event.type == StreamEventType.ERROR:
self.log_error(event.error or "Unknown error")
def create_logging_callback(
verbose: bool = False,
metrics: bool = False,
logger: Optional[logging.Logger] = None
) -> tuple:
"""
Create a StreamLogger and a callback function for use with streaming.
Args:
verbose: Whether to log detailed events
metrics: Whether to collect timing metrics
logger: Optional custom logger
Returns:
Tuple of (StreamLogger, callback_function)
Usage:
stream_logger, callback = create_logging_callback(verbose=True, metrics=True)
# Pass callback to streaming method
response = client.process_stream_response(
...,
stream_callback=callback,
emit_events=True
)
# After streaming, get metrics
print(stream_logger.get_metrics_summary())
"""
stream_logger = StreamLogger(verbose=verbose, metrics=metrics, logger=logger)
def callback(event: StreamEvent) -> None:
stream_logger.update_from_event(event)
return stream_logger, callback
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.11.1"
version = "0.11.2"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
@@ -0,0 +1,24 @@
"""
Streaming Example with TTFT Metrics
This example demonstrates true pass-through streaming with timing metrics.
When verbose output is enabled, tokens are displayed as they arrive from the provider.
Key timing metrics:
- TTFT (Time To First Token): Time from request to first token received
- Stream Duration: Time from first to last token
- Total Time: End-to-end request time
"""
from praisonaiagents import Agent
from praisonaiagents.config import OutputConfig
# Create agent with verbose output (shows streaming progress)
agent = Agent(
instructions="You are a helpful assistant",
llm="gpt-4o-mini",
output=OutputConfig(verbose=True) # Show generation progress
)
# Run the agent - verbose mode shows streaming progress via Rich Live display
result = agent.start("Write a short paragraph about the history of computing")
print(f"\n--- Response ---\n{result}")
-11
View File
@@ -1,11 +0,0 @@
from praisonaiagents import Agent
agent = Agent(
instructions="You are a helpful assistant",
llm="gpt-4o-mini"
)
# Use chat method with stream=True for streaming responses
for chunk in agent.chat("Write a short paragraph about the history of computing", stream=True):
print(chunk, end="", flush=True)
print() # Final newline
@@ -0,0 +1 @@
"""Unit tests for streaming module."""
@@ -0,0 +1,387 @@
"""
Unit tests for streaming events module.
Tests cover:
- StreamEvent creation and representation
- StreamEventType enum values
- StreamMetrics calculations (TTFT, stream duration, total time)
- StreamEventEmitter callback management
- create_text_printer_callback and create_metrics_callback helpers
"""
import time
from praisonaiagents.streaming.events import (
StreamEvent,
StreamEventType,
StreamMetrics,
StreamEventEmitter,
create_text_printer_callback,
create_metrics_callback,
)
class TestStreamEventType:
"""Tests for StreamEventType enum."""
def test_event_types_exist(self):
"""All required event types should exist."""
assert StreamEventType.REQUEST_START.value == "request_start"
assert StreamEventType.HEADERS_RECEIVED.value == "headers_received"
assert StreamEventType.FIRST_TOKEN.value == "first_token"
assert StreamEventType.DELTA_TEXT.value == "delta_text"
assert StreamEventType.DELTA_TOOL_CALL.value == "delta_tool_call"
assert StreamEventType.TOOL_CALL_END.value == "tool_call_end"
assert StreamEventType.LAST_TOKEN.value == "last_token"
assert StreamEventType.STREAM_END.value == "stream_end"
assert StreamEventType.ERROR.value == "error"
class TestStreamEvent:
"""Tests for StreamEvent dataclass."""
def test_create_basic_event(self):
"""Create a basic event with type only."""
event = StreamEvent(type=StreamEventType.REQUEST_START)
assert event.type == StreamEventType.REQUEST_START
assert event.timestamp > 0
assert event.content is None
assert event.tool_call is None
assert event.metadata is None
assert event.error is None
def test_create_event_with_content(self):
"""Create an event with content."""
event = StreamEvent(
type=StreamEventType.DELTA_TEXT,
content="Hello"
)
assert event.type == StreamEventType.DELTA_TEXT
assert event.content == "Hello"
def test_create_event_with_tool_call(self):
"""Create an event with tool call data."""
tool_data = {"name": "calculator", "arguments": '{"x": 1}'}
event = StreamEvent(
type=StreamEventType.DELTA_TOOL_CALL,
tool_call=tool_data
)
assert event.type == StreamEventType.DELTA_TOOL_CALL
assert event.tool_call == tool_data
def test_create_event_with_metadata(self):
"""Create an event with metadata."""
event = StreamEvent(
type=StreamEventType.REQUEST_START,
metadata={"model": "gpt-4o-mini", "message_count": 2}
)
assert event.metadata["model"] == "gpt-4o-mini"
assert event.metadata["message_count"] == 2
def test_create_error_event(self):
"""Create an error event."""
event = StreamEvent(
type=StreamEventType.ERROR,
error="Connection timeout"
)
assert event.type == StreamEventType.ERROR
assert event.error == "Connection timeout"
def test_event_repr(self):
"""Event repr should be informative."""
event = StreamEvent(
type=StreamEventType.DELTA_TEXT,
content="Hello world"
)
repr_str = repr(event)
assert "DELTA_TEXT" in repr_str or "delta_text" in repr_str
assert "Hello world" in repr_str
class TestStreamMetrics:
"""Tests for StreamMetrics dataclass."""
def test_default_metrics(self):
"""Default metrics should be zero."""
metrics = StreamMetrics()
assert metrics.request_start == 0.0
assert metrics.headers_received == 0.0
assert metrics.first_token == 0.0
assert metrics.last_token == 0.0
assert metrics.stream_end == 0.0
assert metrics.token_count == 0
def test_ttft_calculation(self):
"""TTFT should be first_token - request_start."""
metrics = StreamMetrics()
metrics.request_start = 100.0
metrics.first_token = 100.5
assert metrics.ttft == 0.5
def test_ttft_zero_when_no_first_token(self):
"""TTFT should be 0 when first_token not set."""
metrics = StreamMetrics()
metrics.request_start = 100.0
assert metrics.ttft == 0.0
def test_stream_duration_calculation(self):
"""Stream duration should be last_token - first_token."""
metrics = StreamMetrics()
metrics.first_token = 100.0
metrics.last_token = 102.0
assert metrics.stream_duration == 2.0
def test_total_time_calculation(self):
"""Total time should be stream_end - request_start."""
metrics = StreamMetrics()
metrics.request_start = 100.0
metrics.stream_end = 103.0
assert metrics.total_time == 3.0
def test_tokens_per_second(self):
"""Tokens per second calculation."""
metrics = StreamMetrics()
metrics.first_token = 100.0
metrics.last_token = 102.0 # 2 second duration
metrics.token_count = 100
assert metrics.tokens_per_second == 50.0
def test_tokens_per_second_zero_duration(self):
"""Tokens per second should be 0 when duration is 0."""
metrics = StreamMetrics()
assert metrics.tokens_per_second == 0.0
def test_update_from_event(self):
"""Metrics should update from events."""
metrics = StreamMetrics()
# Request start
event1 = StreamEvent(type=StreamEventType.REQUEST_START, timestamp=100.0)
metrics.update_from_event(event1)
assert metrics.request_start == 100.0
# Headers received
event2 = StreamEvent(type=StreamEventType.HEADERS_RECEIVED, timestamp=100.2)
metrics.update_from_event(event2)
assert metrics.headers_received == 100.2
# First token
event3 = StreamEvent(type=StreamEventType.FIRST_TOKEN, timestamp=100.5)
metrics.update_from_event(event3)
assert metrics.first_token == 100.5
assert metrics.token_count == 1
# Delta text
event4 = StreamEvent(type=StreamEventType.DELTA_TEXT, timestamp=100.6)
metrics.update_from_event(event4)
assert metrics.token_count == 2
# Last token
event5 = StreamEvent(type=StreamEventType.LAST_TOKEN, timestamp=102.0)
metrics.update_from_event(event5)
assert metrics.last_token == 102.0
# Stream end
event6 = StreamEvent(type=StreamEventType.STREAM_END, timestamp=102.1)
metrics.update_from_event(event6)
assert metrics.stream_end == 102.1
def test_to_dict(self):
"""Metrics should convert to dict."""
metrics = StreamMetrics()
metrics.request_start = 100.0
metrics.first_token = 100.5
metrics.last_token = 102.0
metrics.stream_end = 102.1
metrics.token_count = 50
d = metrics.to_dict()
assert "ttft_ms" in d
assert "stream_duration_ms" in d
assert "total_time_ms" in d
assert "token_count" in d
assert "tokens_per_second" in d
def test_format_summary(self):
"""Metrics should format as summary string."""
metrics = StreamMetrics()
metrics.request_start = 100.0
metrics.first_token = 100.5
metrics.last_token = 102.0
metrics.stream_end = 102.1
metrics.token_count = 50
summary = metrics.format_summary()
assert "TTFT" in summary
assert "Stream" in summary
assert "Total" in summary
assert "Tokens" in summary
class TestStreamEventEmitter:
"""Tests for StreamEventEmitter."""
def test_create_emitter(self):
"""Create an emitter."""
emitter = StreamEventEmitter()
assert not emitter.has_callbacks
def test_add_callback(self):
"""Add a callback."""
emitter = StreamEventEmitter()
events = []
def callback(event):
events.append(event)
emitter.add_callback(callback)
assert emitter.has_callbacks
def test_remove_callback(self):
"""Remove a callback."""
emitter = StreamEventEmitter()
events = []
def callback(event):
events.append(event)
emitter.add_callback(callback)
emitter.remove_callback(callback)
assert not emitter.has_callbacks
def test_emit_calls_callbacks(self):
"""Emit should call all callbacks."""
emitter = StreamEventEmitter()
events1 = []
events2 = []
emitter.add_callback(lambda e: events1.append(e))
emitter.add_callback(lambda e: events2.append(e))
event = StreamEvent(type=StreamEventType.DELTA_TEXT, content="Hello")
emitter.emit(event)
assert len(events1) == 1
assert len(events2) == 1
assert events1[0].content == "Hello"
def test_emit_with_metrics(self):
"""Emit should update metrics when enabled."""
emitter = StreamEventEmitter()
emitter.enable_metrics()
emitter.emit(StreamEvent(type=StreamEventType.REQUEST_START, timestamp=100.0))
emitter.emit(StreamEvent(type=StreamEventType.FIRST_TOKEN, timestamp=100.5))
emitter.emit(StreamEvent(type=StreamEventType.STREAM_END, timestamp=102.0))
metrics = emitter.get_metrics()
assert metrics is not None
assert metrics.request_start == 100.0
assert metrics.first_token == 100.5
assert metrics.stream_end == 102.0
def test_callback_error_does_not_break_emission(self):
"""Callback errors should not break other callbacks."""
emitter = StreamEventEmitter()
events = []
def bad_callback(event):
raise ValueError("Test error")
def good_callback(event):
events.append(event)
emitter.add_callback(bad_callback)
emitter.add_callback(good_callback)
event = StreamEvent(type=StreamEventType.DELTA_TEXT)
emitter.emit(event) # Should not raise
assert len(events) == 1
def test_reset_metrics(self):
"""Reset metrics should create new metrics."""
emitter = StreamEventEmitter()
emitter.enable_metrics()
emitter.emit(StreamEvent(type=StreamEventType.REQUEST_START, timestamp=100.0))
emitter.reset_metrics()
metrics = emitter.get_metrics()
assert metrics.request_start == 0.0
class TestHelperFunctions:
"""Tests for helper functions."""
def test_create_text_printer_callback(self, capsys):
"""Text printer callback should print content."""
callback = create_text_printer_callback(flush=True)
callback(StreamEvent(type=StreamEventType.DELTA_TEXT, content="Hello"))
callback(StreamEvent(type=StreamEventType.DELTA_TEXT, content=" World"))
callback(StreamEvent(type=StreamEventType.STREAM_END))
captured = capsys.readouterr()
assert "Hello World" in captured.out
def test_create_metrics_callback(self):
"""Metrics callback should update metrics."""
metrics = StreamMetrics()
callback = create_metrics_callback(metrics)
callback(StreamEvent(type=StreamEventType.REQUEST_START, timestamp=100.0))
callback(StreamEvent(type=StreamEventType.FIRST_TOKEN, timestamp=100.5))
assert metrics.request_start == 100.0
assert metrics.first_token == 100.5
class TestStreamingEventOrdering:
"""Tests for event ordering and pass-through behavior."""
def test_events_emitted_immediately(self):
"""Events should be emitted immediately, not buffered."""
emitter = StreamEventEmitter()
timestamps = []
def callback(event):
timestamps.append((event.type, time.perf_counter()))
emitter.add_callback(callback)
# Simulate streaming with small delays
emitter.emit(StreamEvent(type=StreamEventType.REQUEST_START))
time.sleep(0.01)
emitter.emit(StreamEvent(type=StreamEventType.FIRST_TOKEN))
time.sleep(0.01)
emitter.emit(StreamEvent(type=StreamEventType.DELTA_TEXT))
# Verify events were captured in order
assert len(timestamps) == 3
assert timestamps[0][0] == StreamEventType.REQUEST_START
assert timestamps[1][0] == StreamEventType.FIRST_TOKEN
assert timestamps[2][0] == StreamEventType.DELTA_TEXT
# Verify timestamps are increasing (no buffering)
assert timestamps[0][1] < timestamps[1][1] < timestamps[2][1]
def test_no_buffering_before_first_token(self):
"""Events should not be buffered before first token."""
emitter = StreamEventEmitter()
emitter.enable_metrics()
# Emit request start
t1 = time.perf_counter()
emitter.emit(StreamEvent(type=StreamEventType.REQUEST_START, timestamp=t1))
# Small delay (simulating network)
time.sleep(0.05)
# Emit first token
t2 = time.perf_counter()
emitter.emit(StreamEvent(type=StreamEventType.FIRST_TOKEN, timestamp=t2))
metrics = emitter.get_metrics()
# TTFT should reflect the actual delay
assert metrics.ttft >= 0.04 # Allow some tolerance
@@ -0,0 +1,371 @@
"""
Unit tests for the core SDK profiling module.
Tests cover:
- Profiler enable/disable
- Timing recording
- Streaming tracking
- Phase tracking
- Report generation
- Zero overhead when disabled
"""
import pytest
import time
from unittest.mock import patch
class TestProfilerBasics:
"""Test basic Profiler functionality."""
def test_profiler_disabled_by_default(self):
"""Profiler should be disabled by default."""
from praisonaiagents.profiling import Profiler
Profiler.disable() # Ensure clean state
assert not Profiler.is_enabled()
def test_profiler_enable_disable(self):
"""Test enable/disable functionality."""
from praisonaiagents.profiling import Profiler
Profiler.disable()
assert not Profiler.is_enabled()
Profiler.enable()
assert Profiler.is_enabled()
Profiler.disable()
assert not Profiler.is_enabled()
def test_profiler_env_var_enable(self):
"""Test enabling via environment variable."""
from praisonaiagents.profiling import Profiler
Profiler.disable()
with patch.dict('os.environ', {'PRAISONAI_PROFILE': '1'}):
assert Profiler.is_enabled()
with patch.dict('os.environ', {'PRAISONAI_PROFILE': 'true'}):
assert Profiler.is_enabled()
with patch.dict('os.environ', {'PRAISONAI_PROFILE': 'yes'}):
assert Profiler.is_enabled()
def test_profiler_clear(self):
"""Test clearing profiling data."""
from praisonaiagents.profiling import Profiler
Profiler.enable()
Profiler.record_timing("test", 100.0)
assert len(Profiler.get_timings()) > 0
Profiler.clear()
assert len(Profiler.get_timings()) == 0
Profiler.disable()
class TestTimingRecording:
"""Test timing recording functionality."""
def test_record_timing_when_enabled(self):
"""Timing should be recorded when profiler is enabled."""
from praisonaiagents.profiling import Profiler
Profiler.enable()
Profiler.clear()
Profiler.record_timing("test_op", 123.45, category="test")
timings = Profiler.get_timings()
assert len(timings) == 1
assert timings[0].name == "test_op"
assert timings[0].duration_ms == 123.45
assert timings[0].category == "test"
Profiler.disable()
def test_record_timing_when_disabled(self):
"""Timing should NOT be recorded when profiler is disabled."""
from praisonaiagents.profiling import Profiler
Profiler.disable()
Profiler.clear()
Profiler.record_timing("test_op", 123.45)
timings = Profiler.get_timings()
assert len(timings) == 0
def test_record_timing_with_metadata(self):
"""Timing should include metadata."""
from praisonaiagents.profiling import Profiler
Profiler.enable()
Profiler.clear()
Profiler.record_timing("test_op", 100.0, metadata={"key": "value"})
timings = Profiler.get_timings()
assert timings[0].metadata == {"key": "value"}
Profiler.disable()
class TestPhaseTracking:
"""Test phase tracking functionality."""
def test_phase_context_manager(self):
"""Phase context manager should record timing."""
from praisonaiagents.profiling import Profiler
Profiler.enable()
Profiler.clear()
with Profiler.phase("test_phase"):
time.sleep(0.01) # 10ms
timings = Profiler.get_timings(category="phase")
assert len(timings) == 1
assert timings[0].name == "test_phase"
assert timings[0].duration_ms >= 10 # At least 10ms
Profiler.disable()
def test_nested_phases(self):
"""Nested phases should track parent."""
from praisonaiagents.profiling import Profiler
Profiler.enable()
Profiler.clear()
with Profiler.phase("outer"):
with Profiler.phase("inner"):
time.sleep(0.005)
timings = Profiler.get_timings(category="phase")
assert len(timings) == 2
inner = next(t for t in timings if t.name == "inner")
assert inner.parent == "outer"
Profiler.disable()
class TestStreamingTracker:
"""Test streaming tracker functionality."""
def test_streaming_ttft(self):
"""Streaming tracker should measure TTFT."""
from praisonaiagents.profiling import Profiler, StreamingTracker
Profiler.enable()
Profiler.clear()
tracker = StreamingTracker("test_stream")
tracker.start()
time.sleep(0.01) # 10ms to first token
tracker.first_token()
time.sleep(0.02) # 20ms more
tracker.end()
assert tracker.ttft_ms >= 10
assert tracker.elapsed_ms >= 30
records = Profiler.get_streaming_records()
assert len(records) == 1
assert records[0].name == "test_stream"
assert records[0].ttft_ms >= 10
Profiler.disable()
def test_streaming_context_manager(self):
"""Streaming context manager should work."""
from praisonaiagents.profiling import Profiler
Profiler.enable()
Profiler.clear()
with Profiler.streaming("test_stream") as tracker:
time.sleep(0.01)
tracker.first_token()
tracker.chunk()
tracker.chunk()
records = Profiler.get_streaming_records()
assert len(records) == 1
assert records[0].chunk_count == 2
Profiler.disable()
class TestProfileDecorators:
"""Test profiling decorators."""
def test_profile_func_decorator(self):
"""profile_func decorator should record function timing."""
from praisonaiagents.profiling import Profiler, profile_func
Profiler.enable()
Profiler.clear()
@profile_func
def slow_function():
time.sleep(0.01)
return "done"
result = slow_function()
assert result == "done"
timings = Profiler.get_timings(category="function")
assert len(timings) == 1
assert timings[0].name == "slow_function"
assert timings[0].duration_ms >= 10
Profiler.disable()
def test_profile_func_disabled(self):
"""profile_func should have zero overhead when disabled."""
from praisonaiagents.profiling import Profiler, profile_func
Profiler.disable()
Profiler.clear()
@profile_func
def fast_function():
return "done"
result = fast_function()
assert result == "done"
timings = Profiler.get_timings()
assert len(timings) == 0
class TestProfileBlock:
"""Test profile_block context manager."""
def test_profile_block(self):
"""profile_block should record block timing."""
from praisonaiagents.profiling import Profiler, profile_block
Profiler.enable()
Profiler.clear()
with profile_block("my_block", category="custom"):
time.sleep(0.01)
timings = Profiler.get_timings(category="custom")
assert len(timings) == 1
assert timings[0].name == "my_block"
Profiler.disable()
def test_profile_block_disabled(self):
"""profile_block should have zero overhead when disabled."""
from praisonaiagents.profiling import Profiler, profile_block
Profiler.disable()
Profiler.clear()
with profile_block("my_block"):
pass
timings = Profiler.get_timings()
assert len(timings) == 0
class TestReporting:
"""Test report generation."""
def test_get_summary(self):
"""get_summary should return structured data."""
from praisonaiagents.profiling import Profiler
Profiler.enable()
Profiler.clear()
Profiler.record_timing("op1", 100.0, category="cat1")
Profiler.record_timing("op2", 200.0, category="cat2")
summary = Profiler.get_summary()
assert summary['total_time_ms'] == 300.0
assert summary['timing_count'] == 2
assert 'cat1' in summary['by_category']
assert 'cat2' in summary['by_category']
assert len(summary['slowest_operations']) == 2
Profiler.disable()
def test_export_json(self):
"""export_json should return valid JSON."""
from praisonaiagents.profiling import Profiler
import json
Profiler.enable()
Profiler.clear()
Profiler.record_timing("test", 100.0)
json_str = Profiler.export_json()
data = json.loads(json_str)
assert 'summary' in data
assert 'timings' in data
assert len(data['timings']) == 1
Profiler.disable()
class TestWarmup:
"""Test warmup function."""
def test_warmup_returns_timings(self):
"""warmup should return timing information."""
from praisonaiagents import warmup
result = warmup(include_litellm=False, include_openai=False)
assert isinstance(result, dict)
def test_warmup_with_litellm(self):
"""warmup with litellm should return litellm timing."""
from praisonaiagents import warmup
result = warmup(include_litellm=True, include_openai=False)
assert 'litellm' in result
# Should be positive (imported) or -1 (not available)
assert result['litellm'] > 0 or result['litellm'] == -1
class TestZeroOverhead:
"""Test that profiling has zero overhead when disabled."""
def test_record_timing_overhead(self):
"""record_timing should be fast when disabled."""
from praisonaiagents.profiling import Profiler
Profiler.disable()
start = time.perf_counter()
for _ in range(10000):
Profiler.record_timing("test", 100.0)
elapsed = (time.perf_counter() - start) * 1000
# Should complete 10k calls in under 10ms when disabled
assert elapsed < 10, f"record_timing took {elapsed}ms for 10k calls"
def test_phase_context_overhead(self):
"""phase context manager should be fast when disabled."""
from praisonaiagents.profiling import Profiler
Profiler.disable()
start = time.perf_counter()
for _ in range(1000):
with Profiler.phase("test"):
pass
elapsed = (time.perf_counter() - start) * 1000
# Should complete 1k calls in under 10ms when disabled
assert elapsed < 10, f"phase took {elapsed}ms for 1k calls"
Whitespace-only changes.
Whitespace-only changes.
View File
Whitespace-only changes.
Whitespace-only changes.
Whitespace-only changes.
Whitespace-only changes.
+1 -1
View File
@@ -2422,7 +2422,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.11.1"
version = "0.11.2"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
@@ -0,0 +1,65 @@
"""
Guardrails - Advanced Example
Demonstrates advanced guardrails patterns including:
- Multi-agent configuration
- Custom callbacks/handlers
- Error handling
- All available options
Expected Output:
Comprehensive demonstration of guardrails with multiple agents.
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
# ============================================================
# Section 1: Custom Configuration
# ============================================================
# Define custom guardrails configuration
# config = ...
# ============================================================
# Section 2: Multi-Agent Setup
# ============================================================
agent1 = Agent(
name="Agent1",
instructions="You are the first agent",
# guardrails=config,
)
agent2 = Agent(
name="Agent2",
instructions="You are the second agent",
# guardrails=config,
)
# ============================================================
# Section 3: Task Definition
# ============================================================
task = Task(
description="Demonstrate guardrails with multiple agents",
agent=agent1,
)
# ============================================================
# Section 4: Execution
# ============================================================
agents = PraisonAIAgents(agents=[agent1, agent2], tasks=[task])
result = agents.start()
print(result)
# ============================================================
# Section 5: All Options Reference
# ============================================================
"""
Guardrails Options:
──────────────────
| Option | Type | Default | Description |
|-----------------|----------|---------|----------------------|
| option1 | str | None | Description here |
| option2 | bool | False | Description here |
"""
@@ -0,0 +1,19 @@
"""
Guardrails - Basic Example
Demonstrates minimal usage of guardrails with a single agent.
Expected Output:
Agent response demonstrating guardrails functionality.
"""
from praisonaiagents import Agent
# Basic guardrails usage
agent = Agent(
instructions="You are a helpful assistant",
# guardrails=... # Configure guardrails here
)
# Run the agent
result = agent.start("Hello! Demonstrate guardrails functionality.")
print(result)
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.2.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.2.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.4.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.4.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+109
View File
@@ -266,6 +266,115 @@ def register_commands():
app.add_typer(test_app, name="test", help="Run test suite with tier and provider options")
app.add_typer(examples_app, name="examples", help="Run and manage example files")
# Register standardise command
try:
standardise_app = typer.Typer(name="standardise", help="Documentation and examples standardisation (FDEP)")
@standardise_app.command("check")
def standardise_check(
path: str = typer.Option(".", "--path", "-p", help="Project root path"),
feature: str = typer.Option(None, "--feature", help="Specific feature slug"),
scope: str = typer.Option("all", "--scope", help="Scope: all, docs, examples, sdk, cli"),
ci: bool = typer.Option(False, "--ci", help="CI mode"),
):
"""Check for standardisation issues."""
from .commands.standardise import _run_check
import argparse
args = argparse.Namespace(path=path, feature=feature, scope=scope, ci=ci, dry_run=True)
_run_check(args)
@standardise_app.command("report")
def standardise_report(
path: str = typer.Option(".", "--path", "-p", help="Project root path"),
format: str = typer.Option("text", "--format", "-f", help="Format: text, json, markdown"),
output: str = typer.Option(None, "--output", "-o", help="Output file"),
ci: bool = typer.Option(False, "--ci", help="CI mode"),
):
"""Generate detailed report."""
from .commands.standardise import _run_report
import argparse
args = argparse.Namespace(path=path, format=format, output=output, ci=ci, feature=None, scope="all", dry_run=True)
_run_report(args)
@standardise_app.command("fix")
def standardise_fix(
path: str = typer.Option(".", "--path", "-p", help="Project root path"),
feature: str = typer.Option(None, "--feature", help="Specific feature slug"),
apply: bool = typer.Option(False, "--apply", help="Actually apply changes"),
no_backup: bool = typer.Option(False, "--no-backup", help="Don't create backups"),
):
"""Fix standardisation issues."""
from .commands.standardise import _run_fix
import argparse
args = argparse.Namespace(path=path, feature=feature, apply=apply, no_backup=no_backup, scope="all", ci=False, dry_run=not apply)
_run_fix(args)
@standardise_app.command("init")
def standardise_init(
feature: str = typer.Argument(..., help="Feature slug to initialise"),
path: str = typer.Option(".", "--path", "-p", help="Project root path"),
apply: bool = typer.Option(False, "--apply", help="Actually create files"),
):
"""Initialise a new feature with all required artifacts."""
from .commands.standardise import _run_init
import argparse
args = argparse.Namespace(feature=feature, path=path, apply=apply, scope="all", ci=False, dry_run=not apply)
_run_init(args)
@standardise_app.command("ai")
def standardise_ai(
feature: str = typer.Argument(..., help="Feature slug to generate content for"),
gen_type: str = typer.Option("all", "--type", "-t", help="Type: docs, examples, all"),
apply: bool = typer.Option(False, "--apply", help="Actually create files"),
verify: bool = typer.Option(False, "--verify", help="Verify with AI"),
model: str = typer.Option("gpt-4o-mini", "--model", help="LLM model"),
path: str = typer.Option(".", "--path", "-p", help="Project root path"),
):
"""AI-powered generation of docs/examples."""
from .commands.standardise import _run_ai
import argparse
args = argparse.Namespace(feature=feature, type=gen_type, apply=apply, verify=verify, model=model, path=path, scope="all", ci=False, dry_run=not apply)
_run_ai(args)
@standardise_app.command("checkpoint")
def standardise_checkpoint(
message: str = typer.Option(None, "--message", "-m", help="Checkpoint message"),
path: str = typer.Option(".", "--path", "-p", help="Repository path"),
):
"""Create an undo checkpoint."""
from .commands.standardise import _run_checkpoint
import argparse
args = argparse.Namespace(message=message, path=path)
_run_checkpoint(args)
@standardise_app.command("undo")
def standardise_undo(
checkpoint: str = typer.Option(None, "--checkpoint", help="Checkpoint ID"),
list_checkpoints: bool = typer.Option(False, "--list", help="List checkpoints"),
path: str = typer.Option(".", "--path", "-p", help="Repository path"),
):
"""Undo to a previous checkpoint."""
from .commands.standardise import _run_undo
import argparse
args = argparse.Namespace(checkpoint=checkpoint, list=list_checkpoints, path=path)
_run_undo(args)
@standardise_app.command("redo")
def standardise_redo(
path: str = typer.Option(".", "--path", "-p", help="Repository path"),
):
"""Redo after an undo."""
from .commands.standardise import _run_redo
import argparse
args = argparse.Namespace(path=path)
_run_redo(args)
app.add_typer(standardise_app, name="standardise", help="Documentation and examples standardisation (FDEP)")
# Also register as 'standardize' for US spelling
app.add_typer(standardise_app, name="standardize", help="Documentation and examples standardisation (FDEP)")
except Exception:
pass # Graceful degradation if standardise module not available
# Register TUI and queue commands
tui_app = create_tui_debug_app()
queue_app = create_queue_app()
+2 -3
View File
@@ -127,7 +127,7 @@ def chat_main(
profile: bool = typer.Option(False, "--profile", help="Enable CLI profiling (timing breakdown)"),
profile_deep: bool = typer.Option(False, "--profile-deep", help="Enable deep profiling (cProfile stats, higher overhead)"),
# UI backend selection
ui: str = typer.Option("auto", "--ui", help="UI backend: auto, plain, rich, mg (middle-ground)"),
ui_backend: str = typer.Option("auto", "--ui-backend", help="UI backend: auto, plain, rich, mg (middle-ground)"),
json_output: bool = typer.Option(False, "--json", help="Output JSON (forces plain backend)"),
no_color: bool = typer.Option(False, "--no-color", help="Disable colors"),
theme: str = typer.Option("default", "--theme", help="UI theme: default, dark, light, minimal"),
@@ -159,7 +159,7 @@ def chat_main(
# Configure UI backend
from praisonai.cli.ui import select_backend, UIConfig
ui_config = UIConfig(
ui_backend=ui,
ui_backend=ui_backend,
json_output=json_output,
no_color=no_color,
theme=theme,
@@ -275,7 +275,6 @@ def _run_profiled_chat(
"name": "ChatAgent",
"role": "Assistant",
"goal": "Help the user",
"verbose": verbose,
}
if model:
agent_config["llm"] = model
@@ -2,10 +2,14 @@
CLI commands for the FDEP standardisation system.
Commands:
- praisonai standardise check - Check for standardisation issues
- praisonai standardise report - Generate detailed report
- praisonai standardise fix - Fix issues (with --apply)
- praisonai standardise init - Initialise a new feature
- praisonai standardise check - Check for standardisation issues
- praisonai standardise report - Generate detailed report
- praisonai standardise fix - Fix issues (with --apply)
- praisonai standardise init - Initialise a new feature
- praisonai standardise ai - AI-powered generation
- praisonai standardise checkpoint - Create undo checkpoint
- praisonai standardise undo - Undo to previous checkpoint
- praisonai standardise redo - Redo after undo
"""
import sys
@@ -88,6 +92,91 @@ def standardise_command(args=None):
help="Actually create files (default: dry-run)",
)
# ai subcommand
ai_parser = subparsers.add_parser(
"ai",
help="AI-powered generation of docs/examples",
)
ai_parser.add_argument(
"feature",
type=str,
help="Feature slug to generate content for",
)
ai_parser.add_argument(
"--type", "-t",
choices=["docs", "examples", "all"],
default="all",
help="Type of content to generate",
)
ai_parser.add_argument(
"--apply",
action="store_true",
help="Actually create files (default: dry-run preview)",
)
ai_parser.add_argument(
"--verify",
action="store_true",
help="Verify generated content with AI",
)
ai_parser.add_argument(
"--model",
type=str,
default="gpt-4o-mini",
help="LLM model to use (default: gpt-4o-mini)",
)
_add_common_args(ai_parser)
# checkpoint subcommand
checkpoint_parser = subparsers.add_parser(
"checkpoint",
help="Create an undo checkpoint",
)
checkpoint_parser.add_argument(
"--message", "-m",
type=str,
help="Checkpoint message",
)
checkpoint_parser.add_argument(
"--path", "-p",
type=str,
default=".",
help="Repository path",
)
# undo subcommand
undo_parser = subparsers.add_parser(
"undo",
help="Undo to a previous checkpoint",
)
undo_parser.add_argument(
"--checkpoint",
type=str,
help="Specific checkpoint ID to restore",
)
undo_parser.add_argument(
"--list",
action="store_true",
help="List available checkpoints",
)
undo_parser.add_argument(
"--path", "-p",
type=str,
default=".",
help="Repository path",
)
# redo subcommand
redo_parser = subparsers.add_parser(
"redo",
help="Redo after an undo",
)
redo_parser.add_argument(
"--path", "-p",
type=str,
default=".",
help="Repository path",
)
# Parse args
parsed = parser.parse_args(args)
@@ -105,6 +194,14 @@ def standardise_command(args=None):
return _run_fix(parsed)
elif parsed.subcommand == "init":
return _run_init(parsed)
elif parsed.subcommand == "ai":
return _run_ai(parsed)
elif parsed.subcommand == "checkpoint":
return _run_checkpoint(parsed)
elif parsed.subcommand == "undo":
return _run_undo(parsed)
elif parsed.subcommand == "redo":
return _run_redo(parsed)
return 0
@@ -258,6 +355,210 @@ def _run_init(parsed) -> int:
return 0
def _run_ai(parsed) -> int:
"""Run the AI generation subcommand with ACP/LSP runtime support."""
import asyncio
from praisonai.standardise.ai_generator import AIGenerator
from praisonai.standardise.models import ArtifactType, FeatureSlug
config = _create_config(parsed)
slug = FeatureSlug.from_string(parsed.feature)
if not slug.is_valid:
print(f"Error: Invalid feature slug: {slug.validation_error}")
return 2
# Start ACP/LSP runtime for context gathering
runtime = None
try:
from praisonai.cli.features.interactive_runtime import create_runtime
runtime = create_runtime(
workspace=str(config.sdk_root or config.project_root or "."),
lsp=True,
acp=False # ACP not needed for generation
)
# Start runtime in background
asyncio.get_event_loop().run_until_complete(runtime.start())
if runtime.lsp_ready:
print("🔧 LSP server ready for code intelligence")
except Exception:
# Runtime is optional, continue without it
pass
generator = AIGenerator(
model=parsed.model,
sdk_root=config.sdk_root,
docs_root=config.docs_root,
examples_root=config.examples_root,
use_fast_context=True,
)
# Determine which artifacts to generate
artifact_types = []
gen_type = getattr(parsed, "type", "all")
if gen_type in ["docs", "all"]:
artifact_types.extend([
ArtifactType.DOCS_CONCEPT,
ArtifactType.DOCS_FEATURE,
ArtifactType.DOCS_CLI,
ArtifactType.DOCS_SDK,
])
if gen_type in ["examples", "all"]:
artifact_types.extend([
ArtifactType.EXAMPLE_BASIC,
ArtifactType.EXAMPLE_ADVANCED,
])
print(f"🤖 AI Generation for '{slug.normalised}'")
print("=" * 50)
results = []
for artifact_type in artifact_types:
print(f"\n📝 Generating {artifact_type.value}...")
try:
# Generate with execution verification for examples
content, output_path, verification_info = generator.generate(
slug, artifact_type,
dry_run=not parsed.apply,
verify_execution=True,
)
# Show verification results for examples
if verification_info:
if verification_info.get("execution_passed"):
print(" ✅ Code verified: runs successfully")
elif verification_info.get("requires_external"):
libs = ", ".join(verification_info.get("missing_libraries", []))
print(f" ⚠️ Requires external libraries: {libs}")
elif not verification_info.get("syntax_valid"):
print(f" ❌ Syntax error: {verification_info.get('error', 'Unknown')[:100]}")
else:
print(f" ❌ Execution failed after {verification_info.get('attempt', 1)} attempts")
if verification_info.get("error"):
print(f" Error: {verification_info['error'][:150]}")
# AI content verification (optional)
if parsed.verify:
print(" 🔍 AI content review...")
is_valid, summary, issues = generator.verify(content, artifact_type, slug)
if is_valid:
print(f" ✅ {summary}")
else:
print(f" ⚠️ {summary}")
for issue in issues[:3]:
print(f" - {issue}")
# Check if we should write
can_write = True
if verification_info and not verification_info.get("success"):
if not verification_info.get("requires_external"):
can_write = False
print(" ⛔ Not writing: code doesn't run")
if parsed.apply and output_path and can_write:
print(f" ✓ Created: {output_path}")
results.append((artifact_type.value, True, output_path))
elif not parsed.apply:
print(f" 📋 Preview ({len(content)} chars)")
# Show first few lines
preview_lines = content.split("\n")[:10]
for line in preview_lines:
print(f" {line[:60]}")
if len(content.split("\n")) > 10:
print(" ...")
results.append((artifact_type.value, True, None))
else:
results.append((artifact_type.value, False, "Verification failed"))
except Exception as e:
print(f" ❌ Error: {e}")
results.append((artifact_type.value, False, str(e)))
print("\n" + "=" * 50)
print("Summary:")
success_count = sum(1 for _, success, _ in results if success)
print(f" Generated: {success_count}/{len(results)}")
if not parsed.apply:
print("\nRun with --apply to create these files.")
# Cleanup runtime
if runtime:
try:
asyncio.get_event_loop().run_until_complete(runtime.stop())
except Exception:
pass
return 0
def _run_checkpoint(parsed) -> int:
"""Run the checkpoint subcommand."""
from praisonai.standardise.undo_redo import UndoRedoManager
manager = UndoRedoManager(Path(parsed.path).resolve())
success, result = manager.create_checkpoint(parsed.message)
if success:
print(f"✓ Checkpoint created: {result}")
else:
print(f"❌ Failed: {result}")
return 1
return 0
def _run_undo(parsed) -> int:
"""Run the undo subcommand."""
from praisonai.standardise.undo_redo import UndoRedoManager
manager = UndoRedoManager(Path(parsed.path).resolve())
# List checkpoints if requested
if getattr(parsed, "list", False):
checkpoints = manager.list_checkpoints()
if not checkpoints:
print("No checkpoints found.")
return 0
print("Available checkpoints:")
for cp_id, date, message in checkpoints:
print(f" • {cp_id} ({date}): {message[:50]}")
return 0
# Undo to checkpoint
success, result = manager.undo(parsed.checkpoint)
if success:
print(f"✓ {result}")
else:
print(f"❌ {result}")
return 1
return 0
def _run_redo(parsed) -> int:
"""Run the redo subcommand."""
from praisonai.standardise.undo_redo import UndoRedoManager
manager = UndoRedoManager(Path(parsed.path).resolve())
success, result = manager.redo()
if success:
print(f"✓ {result}")
else:
print(f"❌ {result}")
return 1
return 0
# Click-based alternative for integration with main CLI
def register_click_commands(app):
"""Register click commands with the main CLI app."""
+2
View File
@@ -34,6 +34,8 @@ TYPER_COMMANDS = {
'memory', 'workflow', 'tools', 'knowledge', 'deploy', 'agents', 'skills',
'eval', 'templates', 'recipe', 'todo', 'docs', 'commit', 'hooks', 'rules',
'registry', 'package', 'endpoints',
# Standardisation commands
'standardise', 'standardize',
}
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==3.7.2 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==3.7.4 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
@@ -25,6 +25,9 @@ __all__ = [
"ReportGenerator",
# Config
"StandardiseConfig",
# AI Generation
"AIGenerator",
"ExampleVerifier",
]
@@ -70,4 +73,12 @@ def __getattr__(name: str):
from .config import StandardiseConfig
return StandardiseConfig
if name == "AIGenerator":
from .ai_generator import AIGenerator
return AIGenerator
if name == "ExampleVerifier":
from .example_verifier import ExampleVerifier
return ExampleVerifier
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
File diff suppressed because it is too large. Load diff
+28 -16
View File
@@ -117,25 +117,37 @@ class StandardiseConfig:
root = self.project_root
# Docs root detection
# Docs root detection - default: ~/PraisonAIDocs/docs
if self.docs_root is None:
for candidate in ["docs", "PraisonAIDocs/docs"]:
path = root / candidate
if path.exists():
self.docs_root = path
break
# Check parent for PraisonAIDocs
parent_docs = root.parent / "PraisonAIDocs"
if parent_docs.exists() and self.docs_root is None:
self.docs_root = parent_docs / "docs"
# First check home directory for PraisonAIDocs
home_docs = Path.home() / "PraisonAIDocs" / "docs"
if home_docs.exists():
self.docs_root = home_docs
else:
# Fallback to project-relative paths
for candidate in ["docs", "PraisonAIDocs/docs"]:
path = root / candidate
if path.exists():
self.docs_root = path
break
# Check parent for PraisonAIDocs
parent_docs = root.parent / "PraisonAIDocs"
if parent_docs.exists() and self.docs_root is None:
self.docs_root = parent_docs / "docs"
# Examples root detection
# Examples root detection - default: ~/praisonai-package/examples
if self.examples_root is None:
for candidate in ["examples/python", "examples"]:
path = root / candidate
if path.exists():
self.examples_root = path
break
# First check home directory for praisonai-package/examples
home_examples = Path.home() / "praisonai-package" / "examples"
if home_examples.exists():
self.examples_root = home_examples
else:
# Fallback to project-relative paths
for candidate in ["examples/python", "examples"]:
path = root / candidate
if path.exists():
self.examples_root = path
break
# SDK root detection
if self.sdk_root is None:
@@ -0,0 +1,674 @@
"""
Enhanced templates for the FDEP standardisation system.
Based on analysis of existing PraisonAI documentation best practices:
- Progressive disclosure (simple first, advanced later)
- Mermaid diagrams for concepts
- CodeGroup/Tabs for multiple examples
- Consistent structure across page types
"""
from pathlib import Path
from typing import Optional
from .models import ArtifactType, FeatureSlug
# Template constants based on existing doc analysis
CONCEPT_TEMPLATE = '''---
title: "{name}"
sidebarTitle: "{name}"
description: "{description}"
icon: "{icon}"
---
## Overview
{name} provides [brief description of what this feature does and why it exists].
```mermaid
flowchart LR
subgraph Input
A[User Request]
end
subgraph {name}
B[Process] --> C[Validate]
C --> D[Execute]
end
subgraph Output
E[Result]
end
A --> B
D --> E
style B fill:#189AB4,color:#fff
style C fill:#2E8B57,color:#fff
style D fill:#8B0000,color:#fff
```
## Types of {name}
<CardGroup cols={{2}}>
<Card title="Type 1" icon="code">
Description of first type
</Card>
<Card title="Type 2" icon="brain">
Description of second type
</Card>
</CardGroup>
## Quick Start
```python
from praisonaiagents import Agent, Task
# Basic {slug} usage
agent = Agent(
instructions="You are a helpful assistant",
# Configure {slug} here
)
result = agent.start("Your prompt here")
print(result)
```
## When to Use
| Use Case | Recommendation |
|----------|----------------|
| Simple tasks | Use basic configuration |
| Complex workflows | Use advanced patterns |
| Production | Enable all safety features |
## Related Features
- [Related Feature 1](/docs/concepts/related1)
- [Related Feature 2](/docs/concepts/related2)
'''
FEATURE_TEMPLATE = '''---
title: "{name}"
description: "How to use {slug} in PraisonAI"
icon: "code"
---
## Quick Start
<CodeGroup>
```python Basic Usage
from praisonaiagents import Agent, Task
# Simple {slug} example
agent = Agent(
instructions="You are a helpful assistant",
# {slug} configuration
)
result = agent.start("Hello!")
print(result)
```
```python Advanced Usage
from praisonaiagents import Agent, Task, Agents
# Advanced {slug} with multiple agents
agent1 = Agent(name="Agent1", instructions="First agent")
agent2 = Agent(name="Agent2", instructions="Second agent")
task = Task(
description="Complex task using {slug}",
agent=agent1,
)
agents = Agents(agents=[agent1, agent2], tasks=[task])
result = agents.start()
```
</CodeGroup>
## Configuration Options
<Tabs>
<Tab title="Basic">
```python
# Minimal configuration
agent = Agent(
instructions="Your instructions",
# {slug}=True, # Enable feature
)
```
</Tab>
<Tab title="Advanced">
```python
# Full configuration
agent = Agent(
instructions="Your instructions",
# {slug}={{
# "option1": "value1",
# "option2": "value2",
# }},
)
```
</Tab>
</Tabs>
## API Reference
### Parameters
<ParamField body="param1" type="string" required>
Description of the first parameter.
</ParamField>
<ParamField body="param2" type="boolean" default="false">
Description of the second parameter.
</ParamField>
<ParamField body="param3" type="object">
Advanced configuration object.
</ParamField>
## Examples
### Example 1: Basic Usage
```python
from praisonaiagents import Agent
agent = Agent(instructions="You are helpful")
result = agent.start("Hello")
```
### Example 2: With Callbacks
```python
from praisonaiagents import Agent
def on_complete(result):
print(f"Completed: {{result}}")
agent = Agent(
instructions="You are helpful",
# callbacks=[on_complete],
)
```
## Troubleshooting
<Accordion title="Common Issue 1">
**Problem**: Description of the issue
**Solution**: How to fix it
</Accordion>
<Accordion title="Common Issue 2">
**Problem**: Another common issue
**Solution**: The fix for this issue
</Accordion>
## See Also
- [Concept Guide](/docs/concepts/{slug})
- [CLI Reference](/docs/cli/{slug})
- [API Reference](/docs/sdk/praisonaiagents/{slug}/{slug})
'''
CLI_TEMPLATE = '''---
title: "{name}"
sidebarTitle: "{name}"
description: "CLI commands for {slug}"
icon: "terminal"
---
## Quick Start
```bash
praisonai {slug} [OPTIONS]
```
## Commands
### Basic Command
```bash
praisonai {slug}
```
**Expected Output:**
```
[Output example here]
```
### With Options
```bash
praisonai {slug} --option value
```
## Options
| Option | Type | Default | Description |
|--------|------|---------|-------------|
| `--help` | flag | - | Show help message |
| `--verbose` | flag | false | Enable verbose output |
| `--config` | string | - | Path to config file |
## Environment Variables
| Variable | Description | Default |
|----------|-------------|---------|
| `PRAISONAI_{slug_upper}` | Configure {slug} | None |
## Examples
### Example 1: Basic
```bash
praisonai {slug}
```
### Example 2: With Config
```bash
praisonai {slug} --config config.yaml
```
### Example 3: Verbose Mode
```bash
praisonai {slug} --verbose
```
## Integration with Python
You can also use {slug} programmatically:
```python
from praisonaiagents import Agent
# Equivalent to CLI command
agent = Agent(instructions="Your instructions")
result = agent.start("Your prompt")
```
## Troubleshooting
<Accordion title="Command not found">
Ensure PraisonAI is installed: `pip install praisonai`
</Accordion>
<Accordion title="Permission denied">
Check file permissions and try with `sudo` if needed.
</Accordion>
'''
SDK_TEMPLATE = '''---
title: "{name} Module"
description: "API reference for {slug}"
icon: "code"
---
# {name}
The {name} module provides [brief description].
## Installation
```bash
pip install praisonaiagents
```
## Quick Start
```python
from praisonaiagents import {class_name}
# Basic usage
instance = {class_name}()
result = instance.execute()
```
## Classes
### `{class_name}`
Main class for {slug} functionality.
#### Constructor
```python
{class_name}(
param1: str,
param2: bool = False,
**kwargs
)
```
#### Parameters
<ParamField body="param1" type="string" required>
Primary configuration parameter.
</ParamField>
<ParamField body="param2" type="boolean" default="false">
Enable advanced features.
</ParamField>
#### Methods
##### `execute()`
Execute the main functionality.
```python
result = instance.execute(input_data)
```
**Parameters:**
- `input_data` (str): The input to process
**Returns:** `str` - The processed result
##### `configure()`
Update configuration.
```python
instance.configure(option="value")
```
## Type Definitions
```python
from typing import TypedDict, Optional
class {class_name}Config(TypedDict):
param1: str
param2: Optional[bool]
```
## Examples
### Basic Example
```python
from praisonaiagents import {class_name}
instance = {class_name}(param1="value")
result = instance.execute()
print(result)
```
### Advanced Example
```python
from praisonaiagents import {class_name}
# With full configuration
instance = {class_name}(
param1="value",
param2=True,
)
# Custom processing
result = instance.execute()
```
## See Also
- [Concept Guide](/docs/concepts/{slug})
- [Feature Guide](/docs/features/{slug})
'''
EXAMPLE_BASIC_TEMPLATE = '''"""
{name} - Basic Example
Demonstrates minimal usage of {slug} with a single agent.
Features:
- Simple configuration
- Single agent setup
- Basic execution
Expected Output:
Agent response demonstrating {slug} functionality.
Usage:
python {slug}-basic.py
"""
from praisonaiagents import Agent
# ============================================================
# Basic {name} Configuration
# ============================================================
# Create an agent with {slug} enabled
agent = Agent(
name="{name}Agent",
role="{name} Specialist",
goal="Demonstrate basic {slug} functionality",
backstory="You are an expert in {slug} operations.",
instructions="Provide clear, helpful responses demonstrating {slug}.",
# {slug}=True, # Enable {slug} feature
)
# ============================================================
# Execute
# ============================================================
if __name__ == "__main__":
# Run the agent
result = agent.start("Demonstrate {slug} functionality")
print(result)
'''
EXAMPLE_ADVANCED_TEMPLATE = '''"""
{name} - Advanced Example
Demonstrates advanced {slug} patterns including:
- Multi-agent configuration
- Custom callbacks/handlers
- Error handling
- All available options
Features:
- Multiple agents working together
- Custom configuration
- Production-ready patterns
Expected Output:
Comprehensive demonstration of {slug} with multiple agents.
Usage:
python {slug}-advanced.py
"""
from praisonaiagents import Agent, Task, Agents
from typing import Any
# ============================================================
# Section 1: Custom Configuration
# ============================================================
# Define custom {slug} configuration
# config = {{
# "option1": "value1",
# "option2": True,
# }}
# ============================================================
# Section 2: Custom Callbacks
# ============================================================
def on_task_complete(task_output: Any) -> None:
"""Callback when a task completes."""
print(f"Task completed: {{task_output}}")
def on_error(error: Exception) -> None:
"""Callback when an error occurs."""
print(f"Error occurred: {{error}}")
# ============================================================
# Section 3: Multi-Agent Setup
# ============================================================
# Primary agent
primary_agent = Agent(
name="Primary{name}Agent",
role="Primary {name} Handler",
goal="Handle primary {slug} operations",
backstory="You are the primary handler for {slug} tasks.",
instructions="Process {slug} requests thoroughly and accurately.",
# {slug}=config,
)
# Secondary agent
secondary_agent = Agent(
name="Secondary{name}Agent",
role="Secondary {name} Handler",
goal="Support and verify {slug} operations",
backstory="You verify and enhance {slug} results.",
instructions="Review and improve {slug} outputs.",
# {slug}=config,
)
# ============================================================
# Section 4: Task Definition
# ============================================================
primary_task = Task(
description="Demonstrate advanced {slug} with primary processing",
agent=primary_agent,
expected_output="Detailed {slug} demonstration",
)
verification_task = Task(
description="Verify and enhance the {slug} results",
agent=secondary_agent,
expected_output="Verified and enhanced {slug} output",
)
# ============================================================
# Section 5: Execution
# ============================================================
if __name__ == "__main__":
# Create agents team
agents = Agents(
agents=[primary_agent, secondary_agent],
tasks=[primary_task, verification_task],
verbose=True,
)
# Execute
result = agents.start()
print("\\n" + "=" * 60)
print("FINAL RESULT:")
print("=" * 60)
print(result)
# ============================================================
# Section 6: All Options Reference
# ============================================================
"""
{name} Options:
──────────────────────────────────────────────────────────────
| Option | Type | Default | Description |
|-----------------|----------|---------|----------------------|
| option1 | str | None | Primary option |
| option2 | bool | False | Enable advanced mode |
| option3 | int | 10 | Iteration count |
| option4 | list | [] | Additional handlers |
──────────────────────────────────────────────────────────────
"""
'''
class EnhancedTemplateGenerator:
"""Enhanced template generator with best-practice templates."""
# Icon mapping for features
ICON_MAP = {
"guardrails": "shield-halved",
"memory": "brain",
"knowledge": "book",
"tools": "wrench",
"agents": "robot",
"tasks": "list-check",
"workflows": "diagram-project",
"handoffs": "arrow-right-arrow-left",
"sessions": "clock",
"hooks": "plug",
"callbacks": "phone",
"telemetry": "chart-line",
"default": "code",
}
def get_icon(self, slug: str) -> str:
"""Get appropriate icon for a feature."""
return self.ICON_MAP.get(slug, self.ICON_MAP["default"])
def generate(self, slug: FeatureSlug, artifact_type: ArtifactType) -> str:
"""Generate template content for an artifact type."""
slug_str = slug.normalised
name = slug_str.replace("-", " ").title()
class_name = "".join(word.title() for word in slug_str.split("-"))
icon = self.get_icon(slug_str)
slug_upper = slug_str.upper().replace("-", "_")
# Select template
if artifact_type == ArtifactType.DOCS_CONCEPT:
template = CONCEPT_TEMPLATE
elif artifact_type == ArtifactType.DOCS_FEATURE:
template = FEATURE_TEMPLATE
elif artifact_type == ArtifactType.DOCS_CLI:
template = CLI_TEMPLATE
elif artifact_type == ArtifactType.DOCS_SDK:
template = SDK_TEMPLATE
elif artifact_type == ArtifactType.EXAMPLE_BASIC:
template = EXAMPLE_BASIC_TEMPLATE
elif artifact_type == ArtifactType.EXAMPLE_ADVANCED:
template = EXAMPLE_ADVANCED_TEMPLATE
else:
return ""
# Fill template
return template.format(
name=name,
slug=slug_str,
class_name=class_name,
icon=icon,
slug_upper=slug_upper,
description=f"{name} functionality for PraisonAI agents",
)
def get_expected_path(self, slug: FeatureSlug, artifact_type: ArtifactType,
docs_root: Optional[Path] = None,
examples_root: Optional[Path] = None) -> Optional[Path]:
"""Get the expected path for an artifact."""
slug_str = slug.normalised
if artifact_type == ArtifactType.EXAMPLE_BASIC:
if examples_root:
return examples_root / slug_str / f"{slug_str}-basic.py"
elif artifact_type == ArtifactType.EXAMPLE_ADVANCED:
if examples_root:
return examples_root / slug_str / f"{slug_str}-advanced.py"
elif artifact_type == ArtifactType.DOCS_CONCEPT:
if docs_root:
return docs_root / "concepts" / f"{slug_str}.mdx"
elif artifact_type == ArtifactType.DOCS_FEATURE:
if docs_root:
return docs_root / "features" / f"{slug_str}.mdx"
elif artifact_type == ArtifactType.DOCS_CLI:
if docs_root:
return docs_root / "cli" / f"{slug_str}.mdx"
elif artifact_type == ArtifactType.DOCS_SDK:
if docs_root:
return docs_root / "sdk" / "praisonaiagents" / slug_str / f"{slug_str}.mdx"
return None
@@ -0,0 +1,229 @@
"""
Example verification module for the FDEP standardisation system.
Verifies that generated Python examples are:
1. Syntactically correct (can be parsed)
2. Runnable (can be executed without errors)
3. Handle external library dependencies gracefully
Features:
- Syntax validation via ast.parse
- Execution in subprocess with timeout
- Detection of missing external libraries
- Detailed error reporting
"""
import ast
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import List, Tuple
@dataclass
class VerificationResult:
"""Result of example verification."""
success: bool
syntax_valid: bool
execution_passed: bool
output: str
error: str
missing_libraries: List[str]
requires_external: bool
@property
def can_write(self) -> bool:
"""Whether the example is safe to write."""
return self.syntax_valid and (self.execution_passed or self.requires_external)
class ExampleVerifier:
"""Verifies that generated examples are runnable."""
# Known external libraries that may not be installed
EXTERNAL_LIBRARIES = {
"requests", "httpx", "aiohttp", # HTTP
"pandas", "numpy", "scipy", # Data science
"torch", "tensorflow", "transformers", # ML
"redis", "pymongo", "psycopg2", # Databases
"boto3", "google-cloud", # Cloud
"playwright", "selenium", # Browser automation
"beautifulsoup4", "scrapy", # Web scraping
}
# Imports that indicate external dependencies
EXTERNAL_IMPORT_PATTERNS = [
"import requests",
"import httpx",
"import pandas",
"import numpy",
"import torch",
"import tensorflow",
"from transformers",
"import redis",
"import pymongo",
"import boto3",
]
# Patterns that indicate LLM API calls (will timeout during verification)
LLM_CALL_PATTERNS = [
"agent.start(",
"agents.start(",
".chat(",
".run(",
]
def __init__(self, timeout: int = 30):
self.timeout = timeout
def verify(self, code: str) -> VerificationResult:
"""
Verify that the code is valid and runnable.
Args:
code: Python source code to verify
Returns:
VerificationResult with details
"""
# Step 1: Syntax validation
syntax_valid, syntax_error = self._check_syntax(code)
if not syntax_valid:
return VerificationResult(
success=False,
syntax_valid=False,
execution_passed=False,
output="",
error=f"Syntax error: {syntax_error}",
missing_libraries=[],
requires_external=False,
)
# Step 2: Check for external dependencies
requires_external, external_libs = self._detect_external_deps(code)
# Step 3: Skip execution for examples with LLM calls (they will timeout)
if "LLM_API_CALL" in external_libs:
return VerificationResult(
success=True,
syntax_valid=True,
execution_passed=False,
output="",
error="Skipped: Example makes LLM API calls",
missing_libraries=["praisonaiagents (LLM API)"],
requires_external=True,
)
# Step 4: Execute the code
execution_passed, output, error, missing_libs = self._execute(code)
# Combine missing libraries
all_missing = list(set(missing_libs + external_libs))
return VerificationResult(
success=execution_passed or requires_external,
syntax_valid=True,
execution_passed=execution_passed,
output=output,
error=error,
missing_libraries=all_missing,
requires_external=requires_external and not execution_passed,
)
def _check_syntax(self, code: str) -> Tuple[bool, str]:
"""Check if code is syntactically valid."""
try:
ast.parse(code)
return True, ""
except SyntaxError as e:
return False, f"Line {e.lineno}: {e.msg}"
def _detect_external_deps(self, code: str) -> Tuple[bool, List[str]]:
"""Detect if code requires external libraries or makes LLM calls."""
external_libs = []
for pattern in self.EXTERNAL_IMPORT_PATTERNS:
if pattern in code:
lib_name = pattern.split()[-1]
external_libs.append(lib_name)
# Check for LLM API calls that will timeout during verification
for pattern in self.LLM_CALL_PATTERNS:
if pattern in code:
external_libs.append("LLM_API_CALL")
break
return len(external_libs) > 0, external_libs
def _execute(self, code: str) -> Tuple[bool, str, str, List[str]]:
"""Execute code in a subprocess."""
missing_libs = []
# Write code to temp file
with tempfile.NamedTemporaryFile(
mode="w", suffix=".py", delete=False, encoding="utf-8"
) as f:
f.write(code)
temp_path = f.name
try:
# Run with timeout
result = subprocess.run(
[sys.executable, temp_path],
capture_output=True,
text=True,
timeout=self.timeout,
env={**subprocess.os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
)
output = result.stdout
error = result.stderr
# Check for import errors
if result.returncode != 0:
if "ModuleNotFoundError" in error or "ImportError" in error:
# Extract module name
for line in error.split("\n"):
if "No module named" in line:
# Extract module name from error
parts = line.split("'")
if len(parts) >= 2:
missing_libs.append(parts[1].split(".")[0])
return False, output, error, missing_libs
return True, output, error, missing_libs
except subprocess.TimeoutExpired:
return False, "", f"Execution timed out after {self.timeout}s", missing_libs
except Exception as e:
return False, "", str(e), missing_libs
finally:
# Clean up temp file
Path(temp_path).unlink(missing_ok=True)
def format_result(self, result: VerificationResult) -> str:
"""Format verification result for display."""
lines = []
if result.success:
lines.append("✅ Verification PASSED")
else:
lines.append("❌ Verification FAILED")
lines.append(f" Syntax valid: {'✓' if result.syntax_valid else '✗'}")
lines.append(f" Execution passed: {'✓' if result.execution_passed else '✗'}")
if result.requires_external:
lines.append(f" ⚠️ Requires external libraries: {', '.join(result.missing_libraries)}")
lines.append(" (Will be written with warning)")
if result.error and not result.requires_external:
lines.append(f" Error: {result.error[:200]}")
if result.output:
lines.append(f" Output preview: {result.output[:100]}...")
return "\n".join(lines)
@@ -0,0 +1,232 @@
"""
Git-based undo/redo system for the FDEP standardisation system.
Provides:
- Checkpoint creation before changes
- Undo to previous checkpoint
- Redo to next checkpoint
- List checkpoints
"""
import subprocess
from datetime import datetime
from pathlib import Path
from typing import List, Optional, Tuple
class UndoRedoManager:
"""Git-based undo/redo for standardisation changes."""
CHECKPOINT_PREFIX = "standardise-checkpoint-"
def __init__(self, repo_path: Optional[Path] = None):
self.repo_path = repo_path or Path.cwd()
self._verify_git_repo()
def _verify_git_repo(self) -> bool:
"""Verify we're in a git repository."""
try:
result = subprocess.run(
["git", "rev-parse", "--git-dir"],
cwd=self.repo_path,
capture_output=True,
text=True,
)
return result.returncode == 0
except Exception:
return False
def _run_git(self, *args) -> Tuple[bool, str]:
"""Run a git command."""
try:
result = subprocess.run(
["git"] + list(args),
cwd=self.repo_path,
capture_output=True,
text=True,
)
return result.returncode == 0, result.stdout + result.stderr
except Exception as e:
return False, str(e)
def create_checkpoint(self, message: Optional[str] = None) -> Tuple[bool, str]:
"""
Create a checkpoint before making changes.
Returns:
Tuple of (success, checkpoint_id or error message)
"""
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
checkpoint_id = f"{self.CHECKPOINT_PREFIX}{timestamp}"
# Stage all changes
success, output = self._run_git("add", "-A")
if not success:
return False, f"Failed to stage changes: {output}"
# Check if there are changes to commit
success, output = self._run_git("diff", "--cached", "--quiet")
if success:
# No changes to commit, just create a tag at current HEAD
pass
else:
# Commit changes
commit_msg = message or f"Standardise checkpoint: {timestamp}"
success, output = self._run_git("commit", "-m", commit_msg)
if not success and "nothing to commit" not in output:
return False, f"Failed to commit: {output}"
# Create tag
tag_msg = message or f"Standardise checkpoint at {timestamp}"
success, output = self._run_git("tag", "-a", checkpoint_id, "-m", tag_msg)
if not success:
return False, f"Failed to create tag: {output}"
return True, checkpoint_id
def undo(self, checkpoint_id: Optional[str] = None) -> Tuple[bool, str]:
"""
Undo to a previous checkpoint.
Args:
checkpoint_id: Specific checkpoint to restore (default: previous)
Returns:
Tuple of (success, message)
"""
if checkpoint_id is None:
# Find the most recent checkpoint
checkpoints = self.list_checkpoints()
if len(checkpoints) < 2:
return False, "No previous checkpoint to undo to"
checkpoint_id = checkpoints[1][0] # Second most recent
# Verify checkpoint exists
success, output = self._run_git("rev-parse", checkpoint_id)
if not success:
return False, f"Checkpoint not found: {checkpoint_id}"
# Create a safety checkpoint before undoing
self.create_checkpoint("Auto-checkpoint before undo")
# Reset to checkpoint
success, output = self._run_git("reset", "--hard", checkpoint_id)
if not success:
return False, f"Failed to reset: {output}"
return True, f"Restored to checkpoint: {checkpoint_id}"
def redo(self) -> Tuple[bool, str]:
"""
Redo to the next checkpoint (after an undo).
Returns:
Tuple of (success, message)
"""
# Get reflog to find the previous HEAD
success, output = self._run_git("reflog", "-n", "10", "--format=%H %s")
if not success:
return False, "Failed to read reflog"
# Find the commit before the last reset
lines = output.strip().split("\n")
for i, line in enumerate(lines):
if "reset: moving to" in line and i > 0:
# The previous line has the commit we want to redo to
target_hash = lines[i - 1].split()[0]
success, output = self._run_git("reset", "--hard", target_hash)
if success:
return True, "Redo successful"
return False, f"Failed to redo: {output}"
return False, "No redo available"
def list_checkpoints(self) -> List[Tuple[str, str, str]]:
"""
List all standardise checkpoints.
Returns:
List of (checkpoint_id, date, message) tuples
"""
success, output = self._run_git(
"tag", "-l", f"{self.CHECKPOINT_PREFIX}*",
"--format=%(refname:short)|%(creatordate:short)|%(subject)"
)
if not success:
return []
checkpoints = []
for line in output.strip().split("\n"):
if line and "|" in line:
parts = line.split("|", 2)
if len(parts) >= 3:
checkpoints.append((parts[0], parts[1], parts[2]))
elif len(parts) == 2:
checkpoints.append((parts[0], parts[1], ""))
# Sort by date descending
checkpoints.sort(key=lambda x: x[1], reverse=True)
return checkpoints
def delete_checkpoint(self, checkpoint_id: str) -> Tuple[bool, str]:
"""Delete a checkpoint."""
success, output = self._run_git("tag", "-d", checkpoint_id)
if success:
return True, f"Deleted checkpoint: {checkpoint_id}"
return False, f"Failed to delete: {output}"
def get_changes_since(self, checkpoint_id: str) -> List[str]:
"""Get list of files changed since a checkpoint."""
success, output = self._run_git(
"diff", "--name-only", checkpoint_id, "HEAD"
)
if not success:
return []
return [f for f in output.strip().split("\n") if f]
def preview_undo(self, checkpoint_id: str) -> List[str]:
"""Preview what files would change if we undo to a checkpoint."""
return self.get_changes_since(checkpoint_id)
class FileBackupManager:
"""Simple file-based backup for non-git scenarios."""
def __init__(self, backup_dir: Optional[Path] = None):
self.backup_dir = backup_dir or Path(".praison/backups")
self.backup_dir.mkdir(parents=True, exist_ok=True)
def backup_file(self, file_path: Path) -> Optional[Path]:
"""Create a backup of a file."""
if not file_path.exists():
return None
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_name = f"{file_path.stem}_{timestamp}{file_path.suffix}"
backup_path = self.backup_dir / backup_name
import shutil
shutil.copy2(file_path, backup_path)
return backup_path
def restore_file(self, backup_path: Path, target_path: Path) -> bool:
"""Restore a file from backup."""
if not backup_path.exists():
return False
import shutil
shutil.copy2(backup_path, target_path)
return True
def list_backups(self, file_stem: Optional[str] = None) -> List[Path]:
"""List all backups, optionally filtered by file stem."""
backups = list(self.backup_dir.glob("*"))
if file_stem:
backups = [b for b in backups if b.stem.startswith(file_stem)]
backups.sort(key=lambda x: x.stat().st_mtime, reverse=True)
return backups
+1 -1
View File
@@ -1 +1 @@
__version__ = "3.7.2"
__version__ = "3.7.4"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.11.1",
"praisonaiagents>=0.11.2",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
@@ -140,7 +140,11 @@ class TestFeatureDiscovery:
cli_root.mkdir()
(cli_root / "features").mkdir()
config = StandardiseConfig(sdk_root=sdk_root, docs_root=docs_root, cli_root=cli_root)
# Create empty examples root to ensure no examples found
examples_root = tmp_path / "examples"
examples_root.mkdir()
config = StandardiseConfig(sdk_root=sdk_root, docs_root=docs_root, cli_root=cli_root, examples_root=examples_root)
discovery = FeatureDiscovery(config)
slug = FeatureSlug.from_string("guardrails")
@@ -0,0 +1,121 @@
"""
Unit tests for the ExampleVerifier.
"""
from praisonai.standardise.example_verifier import ExampleVerifier, VerificationResult
class TestExampleVerifier:
"""Tests for ExampleVerifier."""
def test_valid_simple_code(self):
"""Test verification of valid simple code."""
verifier = ExampleVerifier(timeout=10)
code = '''
print("Hello, World!")
'''
result = verifier.verify(code)
assert result.syntax_valid
assert result.execution_passed
assert result.success
assert result.can_write
def test_syntax_error(self):
"""Test detection of syntax errors."""
verifier = ExampleVerifier(timeout=10)
code = '''
def broken(
print("missing closing paren"
'''
result = verifier.verify(code)
assert not result.syntax_valid
assert not result.success
assert not result.can_write
assert "Syntax error" in result.error
def test_runtime_error(self):
"""Test detection of runtime errors."""
verifier = ExampleVerifier(timeout=10)
code = '''
x = 1 / 0 # Division by zero
'''
result = verifier.verify(code)
assert result.syntax_valid
assert not result.execution_passed
assert not result.success
assert "ZeroDivisionError" in result.error
def test_missing_import(self):
"""Test detection of missing imports."""
verifier = ExampleVerifier(timeout=10)
code = '''
import nonexistent_module_xyz
'''
result = verifier.verify(code)
assert result.syntax_valid
assert not result.execution_passed
assert "nonexistent_module_xyz" in result.missing_libraries
def test_external_library_detection(self):
"""Test detection of external library requirements."""
verifier = ExampleVerifier(timeout=10)
code = '''
import pandas
df = pandas.DataFrame()
'''
result = verifier.verify(code)
assert result.syntax_valid
# May or may not pass depending on if pandas is installed
# But should detect pandas as external
assert "pandas" in result.missing_libraries or result.execution_passed
def test_can_write_with_external_libs(self):
"""Test that code requiring external libs can still be written."""
# Create a result that simulates external lib requirement
result = VerificationResult(
success=True,
syntax_valid=True,
execution_passed=False,
output="",
error="ModuleNotFoundError: No module named 'requests'",
missing_libraries=["requests"],
requires_external=True,
)
assert result.can_write # Should allow writing despite execution failure
def test_timeout_handling(self):
"""Test that infinite loops are handled."""
verifier = ExampleVerifier(timeout=2)
code = '''
import time
time.sleep(10) # Will timeout
'''
result = verifier.verify(code)
assert result.syntax_valid
assert not result.execution_passed
assert "timed out" in result.error.lower()
def test_format_result(self):
"""Test result formatting."""
verifier = ExampleVerifier()
result = VerificationResult(
success=True,
syntax_valid=True,
execution_passed=True,
output="Hello",
error="",
missing_libraries=[],
requires_external=False,
)
formatted = verifier.format_result(result)
assert "PASSED" in formatted
assert "Syntax valid: ✓" in formatted
@@ -0,0 +1,207 @@
"""
Unit tests for the undo/redo system.
"""
from unittest.mock import patch
from praisonai.standardise.undo_redo import UndoRedoManager, FileBackupManager
class TestUndoRedoManager:
"""Tests for UndoRedoManager."""
def test_init_with_path(self, tmp_path):
"""Test initialization with a path."""
manager = UndoRedoManager(tmp_path)
assert manager.repo_path == tmp_path
def test_checkpoint_prefix(self):
"""Test checkpoint prefix constant."""
assert UndoRedoManager.CHECKPOINT_PREFIX == "standardise-checkpoint-"
@patch.object(UndoRedoManager, '_run_git')
def test_create_checkpoint_success(self, mock_git, tmp_path):
"""Test successful checkpoint creation."""
mock_git.side_effect = [
(True, ""), # git add
(False, "changes"), # git diff (has changes)
(True, ""), # git commit
(True, ""), # git tag
]
manager = UndoRedoManager(tmp_path)
success, result = manager.create_checkpoint("Test checkpoint")
assert success
assert result.startswith("standardise-checkpoint-")
@patch.object(UndoRedoManager, '_run_git')
def test_create_checkpoint_no_changes(self, mock_git, tmp_path):
"""Test checkpoint creation with no changes."""
mock_git.side_effect = [
(True, ""), # git add
(True, ""), # git diff (no changes)
(True, ""), # git tag
]
manager = UndoRedoManager(tmp_path)
success, result = manager.create_checkpoint()
assert success
@patch.object(UndoRedoManager, '_run_git')
def test_list_checkpoints(self, mock_git, tmp_path):
"""Test listing checkpoints."""
mock_git.return_value = (True, "standardise-checkpoint-20240101|2024-01-01|Test message\n")
manager = UndoRedoManager(tmp_path)
checkpoints = manager.list_checkpoints()
assert len(checkpoints) == 1
assert checkpoints[0][0] == "standardise-checkpoint-20240101"
assert checkpoints[0][1] == "2024-01-01"
assert checkpoints[0][2] == "Test message"
@patch.object(UndoRedoManager, '_run_git')
def test_list_checkpoints_empty(self, mock_git, tmp_path):
"""Test listing checkpoints when none exist."""
mock_git.return_value = (True, "")
manager = UndoRedoManager(tmp_path)
checkpoints = manager.list_checkpoints()
assert checkpoints == []
@patch.object(UndoRedoManager, '_run_git')
def test_undo_success(self, mock_git, tmp_path):
"""Test successful undo."""
mock_git.side_effect = [
(True, ""), # rev-parse (checkpoint exists)
(True, ""), # git add (safety checkpoint)
(True, ""), # git diff
(True, ""), # git tag
(True, ""), # git reset
]
manager = UndoRedoManager(tmp_path)
success, result = manager.undo("standardise-checkpoint-test")
assert success
assert "Restored" in result
@patch.object(UndoRedoManager, '_run_git')
def test_undo_checkpoint_not_found(self, mock_git, tmp_path):
"""Test undo with non-existent checkpoint."""
mock_git.return_value = (False, "not found")
manager = UndoRedoManager(tmp_path)
success, result = manager.undo("nonexistent")
assert not success
assert "not found" in result
@patch.object(UndoRedoManager, '_run_git')
def test_redo_no_redo_available(self, mock_git, tmp_path):
"""Test redo when no redo is available."""
mock_git.return_value = (True, "abc123 HEAD@{0}: commit: test\n")
manager = UndoRedoManager(tmp_path)
success, result = manager.redo()
assert not success
assert "No redo available" in result
@patch.object(UndoRedoManager, '_run_git')
def test_delete_checkpoint(self, mock_git, tmp_path):
"""Test deleting a checkpoint."""
mock_git.return_value = (True, "")
manager = UndoRedoManager(tmp_path)
success, result = manager.delete_checkpoint("test-checkpoint")
assert success
assert "Deleted" in result
@patch.object(UndoRedoManager, '_run_git')
def test_get_changes_since(self, mock_git, tmp_path):
"""Test getting changes since a checkpoint."""
mock_git.return_value = (True, "file1.py\nfile2.py\n")
manager = UndoRedoManager(tmp_path)
changes = manager.get_changes_since("test-checkpoint")
assert len(changes) == 2
assert "file1.py" in changes
assert "file2.py" in changes
class TestFileBackupManager:
"""Tests for FileBackupManager."""
def test_init_creates_backup_dir(self, tmp_path):
"""Test that init creates backup directory."""
backup_dir = tmp_path / "backups"
_ = FileBackupManager(backup_dir) # Creates dir on init
assert backup_dir.exists()
def test_backup_file(self, tmp_path):
"""Test backing up a file."""
# Create a test file
test_file = tmp_path / "test.txt"
test_file.write_text("test content")
backup_dir = tmp_path / "backups"
manager = FileBackupManager(backup_dir)
backup_path = manager.backup_file(test_file)
assert backup_path is not None
assert backup_path.exists()
assert backup_path.read_text() == "test content"
def test_backup_nonexistent_file(self, tmp_path):
"""Test backing up a non-existent file."""
manager = FileBackupManager(tmp_path / "backups")
result = manager.backup_file(tmp_path / "nonexistent.txt")
assert result is None
def test_restore_file(self, tmp_path):
"""Test restoring a file from backup."""
# Create backup
backup_dir = tmp_path / "backups"
backup_dir.mkdir()
backup_file = backup_dir / "backup.txt"
backup_file.write_text("backup content")
# Target file
target_file = tmp_path / "restored.txt"
manager = FileBackupManager(backup_dir)
result = manager.restore_file(backup_file, target_file)
assert result
assert target_file.exists()
assert target_file.read_text() == "backup content"
def test_list_backups(self, tmp_path):
"""Test listing backups."""
backup_dir = tmp_path / "backups"
backup_dir.mkdir()
# Create some backup files
(backup_dir / "file1_20240101.txt").write_text("1")
(backup_dir / "file1_20240102.txt").write_text("2")
(backup_dir / "file2_20240101.txt").write_text("3")
backup_manager = FileBackupManager(backup_dir)
# List all
all_backups = backup_manager.list_backups()
assert len(all_backups) == 3
# List filtered
file1_backups = backup_manager.list_backups("file1")
assert len(file1_backups) == 2
+4 -4
View File
@@ -4276,7 +4276,7 @@ requires-dist = [
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.1.0" },
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.1.0" },
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.1.0" },
{ name = "praisonaiagents", specifier = ">=0.11.1" },
{ name = "praisonaiagents", specifier = ">=0.11.2" },
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
{ name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
{ name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4324,7 +4324,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.11.1"
version = "0.11.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -4333,9 +4333,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/40/42/8fe814f73b19a7decd12bca9e3c720a97906658bf993fb4c3063505531fb/praisonaiagents-0.11.1.tar.gz", hash = "sha256:b0f3720d7d5334fb9aaeb8b8e62cc1a20b9ce9c1cad489031414326479a3c23f", size = 963761, upload-time = "2026-01-10T14:23:57.185Z" }
sdist = { url = "https://files.pythonhosted.org/packages/a4/d0/84583da326312957eb887e32d6df4549889494b736b2f3925f3b80fc7e70/praisonaiagents-0.11.2.tar.gz", hash = "sha256:a32e1063404d61cac71be4ee447f6888f7b9566a298ac63793aeaec41eeeeab1", size = 972927, upload-time = "2026-01-10T19:35:37.032Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/0f/97/d7d465eec9443b80fd79daa57bbac7292b680a8702b5d6ef3fb82899bf7d/praisonaiagents-0.11.1-py3-none-any.whl", hash = "sha256:3792747bd2f449995bb4f77eb9086c1b9a88bc2f89414dd447e3d687449c389b", size = 943019, upload-time = "2026-01-10T14:23:54.691Z" },
{ url = "https://files.pythonhosted.org/packages/43/e8/3c13e6f5a3bcd4b01c0a373a2d3fdd797eeea3f46f673fc6a14ba081e1f2/praisonaiagents-0.11.2-py3-none-any.whl", hash = "sha256:e5d666d23d59e53b498beefc30edef3d28ca42fda7b0cd862d1859e59410067c", size = 954207, upload-time = "2026-01-10T19:35:35.034Z" },
]
[[package]]