Compare commits

...
2 Commits
Author SHA1 Message Date
Mervin Praison a795248cf7 Release v4.4.2 2026-02-03 09:15:50 +00:00
Mervin Praison 92542b7c3c Release v4.4.1 2026-02-02 20:26:43 +00:00
25 changed files with 2078 additions and 19 deletions

No files matched your search

+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=4.4.0" \
"praisonai>=4.4.2" \
"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>=4.4.0" \
"praisonai>=4.4.2" \
"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>=4.4.0" \
"praisonai>=4.4.2" \
"praisonai[ui]" \
"praisonai[crewai]"
+43
View File
@@ -0,0 +1,43 @@
# EvaluationLoop Examples
Examples demonstrating iterative evaluation and improvement patterns.
## Quick Start
```bash
pip install praisonaiagents
export OPENAI_API_KEY=your-key
```
## Examples
| File | Description |
|------|-------------|
| `basic_evaluation_loop.py` | Simple loop: Run → Judge → Improve |
| `multi_step_task_loop.py` | Sequential tasks with context passing |
| `recipe_optimization_loop.py` | CLI-based recipe optimization |
## Pattern
```
┌─────────┐ ┌─────────┐ ┌──────────┐
│ Agent │ ──► │ Judge │ ──► │ Feedback │
│ runs │ │ scores │ │ improves │
└─────────┘ └─────────┘ └──────────┘
▲ │
└───────────────────────────────┘
(repeat until 8.0+)
```
## Run Examples
```bash
# Basic loop
python basic_evaluation_loop.py
# Multi-step with context
python multi_step_task_loop.py
# CLI recipe loop
python recipe_optimization_loop.py "research AI trends"
```
@@ -0,0 +1,73 @@
"""
Basic EvaluationLoop Example
Demonstrates the iterative evaluation loop pattern:
Run agent → Judge output → Improve → Repeat until threshold met.
Usage:
pip install praisonaiagents
export OPENAI_API_KEY=your-key
python basic_evaluation_loop.py
"""
from praisonaiagents import Agent
from praisonaiagents.eval import EvaluationLoop
def on_iteration(result):
"""Callback showing progress after each iteration."""
status = "✅" if result.score >= 8.0 else "⚠️" if result.score >= 6.0 else "❌"
print(f"{status} Iteration {result.iteration}: {result.score:.1f}/10")
if result.findings:
print(f" → {result.findings[0][:60]}...")
def main():
# Create an agent
agent = Agent(
name="writer",
instructions="You are a technical content writer. Write clear, accurate explanations."
)
# Run evaluation loop
loop = EvaluationLoop(
agent=agent,
criteria="""
Evaluate the response on:
1. CLARITY: Is the explanation easy to understand?
2. ACCURACY: Is the technical content correct?
3. COMPLETENESS: Does it cover the key aspects?
4. EXAMPLES: Does it include practical examples?
Score 1-10 based on all criteria.
""",
threshold=8.0,
max_iterations=5,
on_iteration=on_iteration,
verbose=False,
)
print("╔════════════════════════════════════════════════════════════╗")
print("║ Basic EvaluationLoop Example ║")
print("║ Pattern: Run → Judge → Improve → Repeat ║")
print("║ Target: 8.0/10 score ║")
print("╚════════════════════════════════════════════════════════════╝\n")
result = loop.run("Explain how API rate limiting works and why it matters")
print("\n" + "─" * 60)
print("RESULTS:")
print("─" * 60)
print(f"Success: {result.success}")
print(f"Final Score: {result.final_score:.1f}/10")
print(f"Iterations: {result.num_iterations}")
print(f"Score History: {[f'{s:.1f}' for s in result.score_history]}")
print(f"Duration: {result.total_duration_seconds:.2f}s")
print("\n" + "─" * 60)
print("FINAL OUTPUT (first 500 chars):")
print("─" * 60)
print(result.final_output[:500] + "..." if len(result.final_output) > 500 else result.final_output)
if __name__ == "__main__":
main()
@@ -0,0 +1,144 @@
"""
Multi-Step Task Loop Example
Demonstrates evaluating sequential task steps where each step can
reference output from previous steps.
Pattern:
1. Research → Evaluate → Improve
2. Analyze (using research) → Evaluate → Improve
3. Report (using analysis) → Evaluate → Improve
Usage:
pip install praisonaiagents
export OPENAI_API_KEY=your-key
python multi_step_task_loop.py
"""
from praisonaiagents import Agent
from praisonaiagents.eval import EvaluationLoop
def on_iteration(result):
"""Show iteration progress."""
status = "✅" if result.score >= 8.0 else "⚠️" if result.score >= 6.0 else "❌"
print(f" {status} Iteration {result.iteration}: {result.score:.1f}/10")
def run_step(step_name: str, agent: Agent, task: str, criteria: str, context: str = "") -> dict:
"""Run a single step through an evaluation loop."""
print(f"\n{'─'*60}")
print(f"STEP: {step_name}")
print(f"{'─'*60}")
# Include context from previous step if provided
full_task = task
if context:
full_task = f"{task}\n\nContext from previous step:\n{context[:1000]}"
loop = EvaluationLoop(
agent=agent,
criteria=criteria,
threshold=8.0,
max_iterations=3,
on_iteration=on_iteration,
verbose=False,
)
result = loop.run(full_task)
print(f" → Final: {result.final_score:.1f}/10 ({result.num_iterations} iterations)")
return {
"name": step_name,
"output": result.final_output,
"score": result.final_score,
"success": result.success,
"iterations": result.num_iterations,
}
def main():
print("╔════════════════════════════════════════════════════════════╗")
print("║ Multi-Step Task Loop Example ║")
print("║ Pattern: Step1 → Step2 → Step3 (with context passing) ║")
print("║ Each step: Run → Judge → Improve until 8.0+ ║")
print("╚════════════════════════════════════════════════════════════╝")
# Create specialized agents
researcher = Agent(
name="researcher",
instructions="You are a research specialist. Find and summarize key information."
)
analyst = Agent(
name="analyst",
instructions="You are an analyst. Identify patterns, insights, and implications."
)
writer = Agent(
name="writer",
instructions="You are a report writer. Create clear, actionable summaries."
)
# Define task steps
steps = [
{
"name": "Research",
"agent": researcher,
"task": "Research the top 3 trends in AI for 2024-2025. Provide brief descriptions.",
"criteria": "Covers 3 distinct trends with accurate descriptions.",
},
{
"name": "Analysis",
"agent": analyst,
"task": "Analyze the implications of these AI trends for software development.",
"criteria": "Identifies actionable insights and potential challenges.",
},
{
"name": "Report",
"agent": writer,
"task": "Create a brief executive summary (3-4 bullet points) for a CTO.",
"criteria": "Concise, actionable, appropriate for executive audience.",
},
]
# Run all steps with context passing
results = []
context = ""
for step in steps:
result = run_step(
step_name=step["name"],
agent=step["agent"],
task=step["task"],
criteria=step["criteria"],
context=context,
)
results.append(result)
context = result["output"] # Pass output as context to next step
# Summary
print("\n" + "═"*60)
print("SUMMARY")
print("═"*60)
total_iterations = sum(r["iterations"] for r in results)
avg_score = sum(r["score"] for r in results) / len(results)
all_success = all(r["success"] for r in results)
for r in results:
status = "✅" if r["success"] else "❌"
print(f" {status} {r['name']}: {r['score']:.1f}/10 ({r['iterations']} iterations)")
print(f"\n Total Iterations: {total_iterations}")
print(f" Average Score: {avg_score:.1f}/10")
print(f" All Steps Passed: {'Yes' if all_success else 'No'}")
print("\n" + "─"*60)
print("FINAL REPORT OUTPUT:")
print("─"*60)
print(results[-1]["output"][:600])
if __name__ == "__main__":
main()
@@ -0,0 +1,183 @@
"""
Recipe Optimization Loop (CLI-based)
Demonstrates the 5-cycle recipe improvement pattern using CLI commands:
Create (once) → Run → Judge → Optimize (loop 5x)
Usage:
pip install praisonai
export OPENAI_API_KEY=your-key
export TAVILY_API_KEY=your-key
python recipe_optimization_loop.py "research AI trends in healthcare"
"""
import subprocess
import sys
import os
import json
from pathlib import Path
def run_cli(cmd: str, cwd: str = None) -> tuple[int, str, str]:
"""Run a CLI command and return (exit_code, stdout, stderr)."""
result = subprocess.run(
cmd,
shell=True,
capture_output=True,
text=True,
cwd=cwd,
)
return result.returncode, result.stdout, result.stderr
def create_recipe(goal: str, work_dir: str) -> str:
"""Create a recipe once. Returns recipe name."""
print(f"\n{'═'*60}")
print("STEP 1: CREATE RECIPE (once)")
print(f"{'═'*60}")
print(f"Goal: {goal}")
cmd = f'praisonai recipe create "{goal}" --no-optimize'
code, stdout, stderr = run_cli(cmd, cwd=work_dir)
if code != 0:
print(f"❌ Failed to create recipe: {stderr}")
return None
# Extract recipe name from output
for line in stdout.split('\n'):
if 'Created recipe:' in line:
recipe_name = line.split('Created recipe:')[1].strip()
print(f"✅ Created: {recipe_name}")
return recipe_name
print(f"✅ Recipe created")
return None
def run_recipe(recipe_name: str, trace_name: str, work_dir: str) -> tuple[bool, str]:
"""Run the recipe and save trace. Returns (success, output)."""
cmd = f'praisonai recipe run {recipe_name} --save --name {trace_name}'
code, stdout, stderr = run_cli(cmd, cwd=work_dir)
success = code == 0 and 'status: completed' in stdout.lower()
return success, stdout
def judge_recipe(trace_name: str, work_dir: str) -> tuple[float, str]:
"""Judge the trace. Returns (score, feedback)."""
cmd = f'praisonai recipe judge {trace_name}'
code, stdout, stderr = run_cli(cmd, cwd=work_dir)
# Extract score from output
score = 0.0
for line in stdout.split('\n'):
if 'Overall Score:' in line:
try:
score = float(line.split(':')[1].strip().split('/')[0])
except:
pass
return score, stdout
def optimize_recipe(recipe_name: str, iterations: int, work_dir: str) -> tuple[float, str]:
"""Optimize the recipe. Returns (final_score, output)."""
cmd = f'praisonai recipe optimize {recipe_name} --iterations {iterations}'
code, stdout, stderr = run_cli(cmd, cwd=work_dir)
# Extract final score
score = 0.0
for line in stdout.split('\n'):
if 'Final score:' in line or 'score:' in line.lower():
try:
score = float(line.split(':')[1].strip().split('/')[0])
except:
pass
return score, stdout
def run_optimization_loop(goal: str, num_cycles: int = 5, threshold: float = 8.0):
"""Run the full Create → Run → Judge → Optimize loop."""
print("╔════════════════════════════════════════════════════════════╗")
print("║ Recipe Optimization Loop (CLI-based) ║")
print("║ Pattern: Create (once) → Run → Judge → Optimize (loop) ║")
print(f"║ Cycles: {num_cycles} | Threshold: {threshold}/10 ║")
print("╚════════════════════════════════════════════════════════════╝")
# Create work directory
work_dir = Path("/tmp/recipe-optimization-loop")
work_dir.mkdir(parents=True, exist_ok=True)
os.chdir(work_dir)
# Step 1: Create recipe (once)
recipe_name = create_recipe(goal, str(work_dir))
if not recipe_name:
print("Failed to create recipe. Exiting.")
return
# Step 2-N: Run → Judge → Optimize loop
results = []
for cycle in range(1, num_cycles + 1):
print(f"\n{'─'*60}")
print(f"CYCLE {cycle}/{num_cycles}")
print(f"{'─'*60}")
trace_name = f"cycle-{cycle}"
# Run
print(f" 📝 Running recipe...")
success, run_output = run_recipe(recipe_name, trace_name, str(work_dir))
run_status = "✅" if success else "⚠️"
print(f" {run_status} Run complete")
# Judge
print(f" ⚖️ Judging output...")
score, judge_output = judge_recipe(trace_name, str(work_dir))
score_status = "✅" if score >= threshold else "⚠️" if score >= 6.0 else "❌"
print(f" {score_status} Score: {score}/10")
# Check if threshold met
if score >= threshold:
print(f" 🎉 Threshold {threshold}/10 reached!")
results.append({"cycle": cycle, "score": score, "status": "passed"})
break
# Optimize
print(f" 🔧 Optimizing recipe...")
opt_score, opt_output = optimize_recipe(recipe_name, 1, str(work_dir))
print(f" → Optimized score: {opt_score}/10")
results.append({
"cycle": cycle,
"initial_score": score,
"optimized_score": opt_score,
"status": "optimized" if opt_score > score else "no_change"
})
# Summary
print(f"\n{'═'*60}")
print("SUMMARY")
print(f"{'═'*60}")
for r in results:
if "initial_score" in r:
print(f" Cycle {r['cycle']}: {r['initial_score']} → {r['optimized_score']}/10")
else:
print(f" Cycle {r['cycle']}: {r['score']}/10 ✅ PASSED")
final_score = results[-1].get("optimized_score", results[-1].get("score", 0))
success = final_score >= threshold
print(f"\n Final Score: {final_score}/10")
print(f" Success: {'Yes ✅' if success else 'No ❌'}")
print(f" Recipe: {work_dir}/{recipe_name}")
return results
if __name__ == "__main__":
goal = sys.argv[1] if len(sys.argv) > 1 else "research latest AI trends and create a summary report"
run_optimization_loop(goal, num_cycles=5, threshold=8.0)
+3 -1
View File
@@ -89,4 +89,6 @@ graph.py
chroma_db/
.qodo
*_COMPARISON_ANALYSIS.md
*_REPORT.md
*_REPORT.md
.praisonai
@@ -457,6 +457,19 @@ _LAZY_IMPORTS = {
'get_default_plugin_dirs': ('praisonaiagents.plugins', 'get_default_plugin_dirs'),
'get_plugin_template': ('praisonaiagents.plugins', 'get_plugin_template'),
'ensure_plugin_dir': ('praisonaiagents.plugins', 'ensure_plugin_dir'),
# Config loader - for config-driven defaults
'get_config': ('praisonaiagents.config.loader', 'get_config'),
'get_default': ('praisonaiagents.config.loader', 'get_default'),
'get_plugins_config': ('praisonaiagents.config.loader', 'get_plugins_config'),
'get_defaults_config': ('praisonaiagents.config.loader', 'get_defaults_config'),
'apply_config_defaults': ('praisonaiagents.config.loader', 'apply_config_defaults'),
'validate_config': ('praisonaiagents.config.loader', 'validate_config'),
'get_config_path': ('praisonaiagents.config.loader', 'get_config_path'),
'ConfigValidationError': ('praisonaiagents.config.loader', 'ConfigValidationError'),
'PraisonConfig': ('praisonaiagents.config.loader', 'PraisonConfig'),
'PluginsConfig': ('praisonaiagents.config.loader', 'PluginsConfig'),
'DefaultsConfig': ('praisonaiagents.config.loader', 'DefaultsConfig'),
}
@@ -9,6 +9,10 @@ Environment Variables:
PRAISONAI_TELEMETRY_ENABLED: Enable telemetry (default: false, opt-in)
PRAISONAI_TELEMETRY_DISABLED: Disable telemetry (takes precedence)
DO_NOT_TRACK: Standard opt-out flag (takes precedence)
PRAISONAI_PLUGINS: Enable plugins (default: false)
- "true" or "1": Enable all discovered plugins
- "logging,metrics": Enable specific plugins (comma-separated)
- "false" or "0": Disable plugins
"""
import os
from typing import Optional
@@ -77,6 +81,38 @@ AUTO_INSTRUMENT = _str_to_bool(
)
# Plugins: Enable background plugins (hooks, metrics, logging)
# Note: Tools and guardrails work WITHOUT this flag - they are explicit
def _get_plugins_enabled() -> bool:
"""Determine if plugins should be auto-enabled based on env var."""
env_value = os.environ.get('PRAISONAI_PLUGINS', '').lower()
if not env_value:
return False
if env_value in ('true', '1', 'yes', 'on'):
return True
if env_value in ('false', '0', 'no', 'off'):
return False
# Treat as comma-separated list of plugin names (implies enabled)
return True
def _get_plugins_list() -> list:
"""Get list of specific plugins to enable from env var.
Returns:
List of plugin names, or empty list if all plugins enabled.
"""
env_value = os.environ.get('PRAISONAI_PLUGINS', '').lower()
if not env_value or env_value in ('true', '1', 'yes', 'on', 'false', '0', 'no', 'off'):
return []
# Parse comma-separated list
return [p.strip() for p in env_value.split(',') if p.strip()]
PLUGINS_ENABLED = _get_plugins_enabled()
PLUGINS_LIST = _get_plugins_list()
def missing_dependency_error(
package_name: str,
extra_name: Optional[str] = None,
@@ -541,6 +541,49 @@ class Agent:
Agent._configure_logging()
Agent._logging_configured = True
# ============================================================
# CONFIG-DRIVEN DEFAULTS (apply before parameter resolution)
# Precedence: Explicit params > Config file > Built-in defaults
# Only applies when param is None (not explicitly set)
# ============================================================
from ..config.loader import apply_config_defaults, get_default
# Apply config defaults for LLM if not explicitly set
if llm is None and model is None:
config_model = get_default("model")
if config_model:
llm = config_model
# Apply config defaults for base_url if not explicitly set
if base_url is None:
config_base_url = get_default("base_url")
if config_base_url:
base_url = config_base_url
# Apply config defaults for feature params (memory, knowledge, etc.)
# These use apply_config_defaults which handles enabled=True/False logic
if memory is None:
memory = apply_config_defaults("memory", memory, MemoryConfig)
if knowledge is None:
knowledge = apply_config_defaults("knowledge", knowledge, KnowledgeConfig)
if planning is False: # Only override if explicitly False (default)
config_planning = apply_config_defaults("planning", None, PlanningConfig)
if config_planning:
planning = config_planning
if reflection is None:
reflection = apply_config_defaults("reflection", reflection, ReflectionConfig)
if web is None:
web = apply_config_defaults("web", web, WebConfig)
if output is None:
output = apply_config_defaults("output", output, OutputConfig)
if execution is None:
execution = apply_config_defaults("execution", execution, ExecutionConfig)
if caching is None:
caching = apply_config_defaults("caching", caching, CachingConfig)
if autonomy is None:
# AutonomyConfig is in agent/autonomy.py - use dict for config defaults
autonomy = apply_config_defaults("autonomy", autonomy, None)
# ============================================================
# CONSOLIDATED PARAMS EXTRACTION (agent-centric API)
# Uses unified resolver: Instance > Config > Array > String > Bool > Default
@@ -840,7 +883,10 @@ class Agent:
backend = _memory_config.backend
if hasattr(backend, 'value'):
backend = backend.value
if backend == "file":
# If learn is enabled, pass as dict to trigger full Memory class
if _memory_config.learn:
memory = _memory_config.to_dict()
elif backend == "file":
memory = True
elif _memory_config.config:
memory = _memory_config.config
@@ -2712,8 +2758,19 @@ Summary:"""
self._memory_instance = FileMemory(user_id=mem_user_id)
elif isinstance(memory, dict):
# Configuration dict
provider = memory.get("provider", "file")
if provider == "file":
provider = memory.get("provider", memory.get("backend", "file"))
learn_enabled = memory.get("learn", False)
# Use full Memory class if learn is enabled (requires LearnManager)
if learn_enabled:
try:
from ..memory.memory import Memory
self._memory_instance = Memory(memory)
except ImportError:
logging.warning("Memory with learn requires additional dependencies. Falling back to FileMemory (learn disabled).")
from ..memory.file_memory import FileMemory
self._memory_instance = FileMemory(user_id=memory.get("user_id", mem_user_id))
elif provider == "file":
from ..memory.file_memory import FileMemory
self._memory_instance = FileMemory(
user_id=memory.get("user_id", mem_user_id),
@@ -2749,6 +2806,24 @@ Summary:"""
return ""
def get_learn_context(self) -> str:
"""
Get learning context for injection into system prompt.
Returns learned preferences, insights, and patterns when memory="learn"
is enabled. Returns empty string when learn is not enabled (zero overhead).
Returns:
Formatted learning context string, or empty string if learn not enabled
"""
if not self._memory_instance:
return ""
if hasattr(self._memory_instance, 'get_learn_context'):
return self._memory_instance.get_learn_context()
return ""
def store_memory(self, content: str, memory_type: str = "short_term", **kwargs):
"""
Store content in memory.
@@ -3389,6 +3464,11 @@ Your Goal: {self.goal}"""
# Display memory info to user if verbose
if self.verbose:
self._display_memory_info()
# Add learn context if learn is enabled (auto-inject when memory="learn")
learn_context = self.get_learn_context()
if learn_context:
system_prompt += f"\n\n## Learned Context (Patterns and insights from past interactions)\n{learn_context}"
# Add skills prompt if skills are configured
if self._skills or self._skills_dirs:
@@ -176,6 +176,9 @@ class MemoryConfig:
learn_dict = LearnConfig().to_dict()
elif isinstance(self.learn, LearnConfig):
learn_dict = self.learn.to_dict()
elif isinstance(self.learn, dict):
# Pass through dict directly (user provided learn config as dict)
learn_dict = self.learn
return {
"backend": self.backend.value if isinstance(self.backend, MemoryBackend) else self.backend,
@@ -0,0 +1,609 @@
"""
Configuration Loader for PraisonAI Agents.
Loads configuration from multiple sources with precedence:
1. Explicit parameters (highest)
2. Environment variables
3. Config file (.praisonai/config.toml or praisonai.toml)
4. Defaults (lowest)
Zero Performance Impact:
- Config is loaded lazily on first access
- No file I/O at import time
- Cached after first load
Usage:
from praisonaiagents.config.loader import get_config, get_default
# Get entire config
config = get_config()
# Get specific default with fallback
model = get_default("model", "gpt-4o-mini")
memory_config = get_default("memory", {})
# Validate config file
errors = validate_config(config.to_dict())
"""
import os
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
from dataclasses import dataclass, field
# ============================================================================
# CONFIG VALIDATION SCHEMA
# ============================================================================
# Defines valid keys and types for config file validation
VALID_PLUGINS_KEYS = {
"enabled": (bool, list, str), # bool, list of plugin names, or "true"/"false"
"auto_discover": (bool,),
"directories": (list,),
}
VALID_DEFAULTS_KEYS = {
# LLM settings
"model": (str,),
"base_url": (str,),
"api_key": (str,),
# Feature flags
"allow_delegation": (bool,),
"allow_code_execution": (bool,),
"code_execution_mode": (str,),
# Feature configs (nested dicts with 'enabled' key)
"memory": (dict, bool),
"knowledge": (dict, bool),
"planning": (dict, bool),
"reflection": (dict, bool),
"guardrails": (dict, bool),
"web": (dict, bool),
"output": (dict, str), # dict or preset string
"execution": (dict, str), # dict or preset string
"caching": (dict, bool),
"autonomy": (dict, bool),
"skills": (dict, list),
"context": (dict, bool),
"hooks": (dict, list),
"templates": (dict,),
}
VALID_ROOT_KEYS = {"plugins", "defaults"}
class ConfigValidationError(Exception):
"""Raised when config file has validation errors."""
def __init__(self, errors: List[str]):
self.errors = errors
message = "Config validation failed:\n" + "\n".join(f" - {e}" for e in errors)
super().__init__(message)
# Global config cache (lazy loaded)
_config_cache: Optional[Dict[str, Any]] = None
_config_loaded: bool = False
@dataclass
class PluginsConfig:
"""Configuration for the plugin system."""
enabled: Union[bool, List[str]] = False # True, False, or list of plugin names
auto_discover: bool = True
directories: List[str] = field(default_factory=lambda: [
"./.praisonai/plugins/",
"~/.praisonai/plugins/"
])
def to_dict(self) -> Dict[str, Any]:
return {
"enabled": self.enabled,
"auto_discover": self.auto_discover,
"directories": self.directories,
}
@dataclass
class DefaultsConfig:
"""Default configuration for Agent parameters."""
# LLM defaults
model: Optional[str] = None
base_url: Optional[str] = None
# Feature flags
allow_delegation: bool = False
allow_code_execution: bool = False
code_execution_mode: str = "safe"
# Feature configs (nested dicts)
memory: Optional[Dict[str, Any]] = None
knowledge: Optional[Dict[str, Any]] = None
planning: Optional[Dict[str, Any]] = None
reflection: Optional[Dict[str, Any]] = None
web: Optional[Dict[str, Any]] = None
output: Optional[Dict[str, Any]] = None
execution: Optional[Dict[str, Any]] = None
caching: Optional[Dict[str, Any]] = None
autonomy: Optional[Dict[str, Any]] = None
skills: Optional[Dict[str, Any]] = None
def to_dict(self) -> Dict[str, Any]:
return {
"model": self.model,
"base_url": self.base_url,
"allow_delegation": self.allow_delegation,
"allow_code_execution": self.allow_code_execution,
"code_execution_mode": self.code_execution_mode,
"memory": self.memory,
"knowledge": self.knowledge,
"planning": self.planning,
"reflection": self.reflection,
"web": self.web,
"output": self.output,
"execution": self.execution,
"caching": self.caching,
"autonomy": self.autonomy,
"skills": self.skills,
}
@dataclass
class PraisonConfig:
"""Root configuration for PraisonAI Agents."""
plugins: PluginsConfig = field(default_factory=PluginsConfig)
defaults: DefaultsConfig = field(default_factory=DefaultsConfig)
def to_dict(self) -> Dict[str, Any]:
return {
"plugins": self.plugins.to_dict(),
"defaults": self.defaults.to_dict(),
}
def _find_config_file() -> Optional[Path]:
"""Find config file in standard locations.
Search order:
1. .praisonai/config.toml (project-local)
2. praisonai.toml (project root)
3. ~/.praisonai/config.toml (user global)
Returns:
Path to config file if found, None otherwise
"""
# Project-local locations
cwd = Path.cwd()
local_paths = [
cwd / ".praisonai" / "config.toml",
cwd / "praisonai.toml",
]
for path in local_paths:
if path.exists():
return path
# User global location
home = Path.home()
global_path = home / ".praisonai" / "config.toml"
if global_path.exists():
return global_path
return None
def _parse_toml(path: Path) -> Dict[str, Any]:
"""Parse TOML file with fallback for Python < 3.11.
Args:
path: Path to TOML file
Returns:
Parsed config dict
"""
try:
# Python 3.11+ has built-in tomllib
import tomllib
with open(path, "rb") as f:
return tomllib.load(f)
except ImportError:
# Fallback for Python < 3.11
try:
import tomli
with open(path, "rb") as f:
return tomli.load(f)
except ImportError:
# No TOML parser available - return empty config
import logging
logging.debug(
"TOML parser not available. Install tomli for Python < 3.11: "
"pip install tomli"
)
return {}
def _load_config() -> Dict[str, Any]:
"""Load configuration from file.
Returns:
Config dict (empty if no config file found)
"""
config_path = _find_config_file()
if config_path is None:
return {}
try:
return _parse_toml(config_path)
except Exception as e:
import logging
logging.warning(f"Failed to load config from {config_path}: {e}")
return {}
def _dict_to_plugins_config(data: Dict[str, Any]) -> PluginsConfig:
"""Convert dict to PluginsConfig."""
return PluginsConfig(
enabled=data.get("enabled", False),
auto_discover=data.get("auto_discover", True),
directories=data.get("directories", [
"./.praisonai/plugins/",
"~/.praisonai/plugins/"
]),
)
def _dict_to_defaults_config(data: Dict[str, Any]) -> DefaultsConfig:
"""Convert dict to DefaultsConfig."""
return DefaultsConfig(
model=data.get("model"),
base_url=data.get("base_url"),
allow_delegation=data.get("allow_delegation", False),
allow_code_execution=data.get("allow_code_execution", False),
code_execution_mode=data.get("code_execution_mode", "safe"),
memory=data.get("memory"),
knowledge=data.get("knowledge"),
planning=data.get("planning"),
reflection=data.get("reflection"),
web=data.get("web"),
output=data.get("output"),
execution=data.get("execution"),
caching=data.get("caching"),
autonomy=data.get("autonomy"),
skills=data.get("skills"),
)
def _dict_to_praison_config(data: Dict[str, Any]) -> PraisonConfig:
"""Convert dict to PraisonConfig."""
plugins_data = data.get("plugins", {})
defaults_data = data.get("defaults", {})
return PraisonConfig(
plugins=_dict_to_plugins_config(plugins_data),
defaults=_dict_to_defaults_config(defaults_data),
)
def get_config(reload: bool = False) -> PraisonConfig:
"""Get the global configuration.
Loads config lazily on first access and caches it.
Args:
reload: Force reload from file
Returns:
PraisonConfig instance
"""
global _config_cache, _config_loaded
if not _config_loaded or reload:
raw_config = _load_config()
_config_cache = _dict_to_praison_config(raw_config)
_config_loaded = True
return _config_cache
def get_plugins_config() -> PluginsConfig:
"""Get plugins configuration.
Returns:
PluginsConfig instance
"""
return get_config().plugins
def get_defaults_config() -> DefaultsConfig:
"""Get defaults configuration.
Returns:
DefaultsConfig instance
"""
return get_config().defaults
def get_default(key: str, fallback: Any = None) -> Any:
"""Get a specific default value.
Args:
key: Config key (e.g., "model", "memory", "memory.backend")
fallback: Value to return if key not found
Returns:
Config value or fallback
"""
defaults = get_defaults_config()
# Handle nested keys like "memory.backend"
if "." in key:
parts = key.split(".")
value = getattr(defaults, parts[0], None)
if value is None:
return fallback
for part in parts[1:]:
if isinstance(value, dict):
value = value.get(part)
else:
return fallback
if value is None:
return fallback
return value
# Simple key
value = getattr(defaults, key, None)
return value if value is not None else fallback
def is_plugins_enabled() -> bool:
"""Check if plugins are enabled via config or env var.
Returns:
True if plugins should be enabled
"""
# Check env var first (highest precedence)
env_value = os.environ.get("PRAISONAI_PLUGINS", "").lower()
if env_value:
if env_value in ("true", "1", "yes", "on"):
return True
if env_value in ("false", "0", "no", "off"):
return False
# Treat as comma-separated list of plugin names
return True
# Check config file
plugins_config = get_plugins_config()
if isinstance(plugins_config.enabled, bool):
return plugins_config.enabled
if isinstance(plugins_config.enabled, list):
return len(plugins_config.enabled) > 0
return False
def get_enabled_plugins() -> Optional[List[str]]:
"""Get list of enabled plugins (if specific list provided).
Returns:
List of plugin names, or None if all plugins enabled
"""
# Check env var first
env_value = os.environ.get("PRAISONAI_PLUGINS", "").lower()
if env_value and env_value not in ("true", "1", "yes", "on", "false", "0", "no", "off"):
# Treat as comma-separated list
return [p.strip() for p in env_value.split(",") if p.strip()]
# Check config file
plugins_config = get_plugins_config()
if isinstance(plugins_config.enabled, list):
return plugins_config.enabled
return None # All plugins enabled
def clear_config_cache() -> None:
"""Clear the config cache (for testing)."""
global _config_cache, _config_loaded
_config_cache = None
_config_loaded = False
def validate_config(config: Dict[str, Any], raise_on_error: bool = False) -> List[str]:
"""Validate config file structure and types.
Args:
config: Raw config dict (from TOML parsing)
raise_on_error: If True, raise ConfigValidationError on errors
Returns:
List of validation error messages (empty if valid)
Example:
config = {"plugins": {"enabled": True}, "defaults": {"model": "gpt-4o"}}
errors = validate_config(config)
if errors:
print("Config errors:", errors)
"""
errors = []
# Check root keys
for key in config.keys():
if key not in VALID_ROOT_KEYS:
similar = _suggest_similar_key(key, VALID_ROOT_KEYS)
if similar:
errors.append(f"Unknown root key '{key}'. Did you mean '{similar}'?")
else:
errors.append(f"Unknown root key '{key}'. Valid keys: {', '.join(sorted(VALID_ROOT_KEYS))}")
# Validate [plugins] section
if "plugins" in config:
plugins = config["plugins"]
if not isinstance(plugins, dict):
errors.append(f"[plugins] must be a table, got {type(plugins).__name__}")
else:
for key, value in plugins.items():
if key not in VALID_PLUGINS_KEYS:
similar = _suggest_similar_key(key, VALID_PLUGINS_KEYS.keys())
if similar:
errors.append(f"[plugins] Unknown key '{key}'. Did you mean '{similar}'?")
else:
errors.append(f"[plugins] Unknown key '{key}'. Valid keys: {', '.join(sorted(VALID_PLUGINS_KEYS.keys()))}")
else:
valid_types = VALID_PLUGINS_KEYS[key]
if not isinstance(value, valid_types):
errors.append(f"[plugins.{key}] Expected {_format_types(valid_types)}, got {type(value).__name__}")
# Validate [defaults] section
if "defaults" in config:
defaults = config["defaults"]
if not isinstance(defaults, dict):
errors.append(f"[defaults] must be a table, got {type(defaults).__name__}")
else:
for key, value in defaults.items():
if key not in VALID_DEFAULTS_KEYS:
similar = _suggest_similar_key(key, VALID_DEFAULTS_KEYS.keys())
if similar:
errors.append(f"[defaults] Unknown key '{key}'. Did you mean '{similar}'?")
else:
errors.append(f"[defaults] Unknown key '{key}'. Valid keys: {', '.join(sorted(VALID_DEFAULTS_KEYS.keys()))}")
else:
valid_types = VALID_DEFAULTS_KEYS[key]
if not isinstance(value, valid_types):
errors.append(f"[defaults.{key}] Expected {_format_types(valid_types)}, got {type(value).__name__}")
# Validate nested feature configs
if isinstance(value, dict) and key in ("memory", "knowledge", "planning", "reflection",
"guardrails", "web", "output", "execution",
"caching", "autonomy", "context"):
nested_errors = _validate_feature_config(key, value)
errors.extend(nested_errors)
if raise_on_error and errors:
raise ConfigValidationError(errors)
return errors
def _suggest_similar_key(key: str, valid_keys) -> Optional[str]:
"""Suggest a similar valid key using simple string matching."""
key_lower = key.lower()
for valid in valid_keys:
# Exact match after lowercasing
if valid.lower() == key_lower:
return valid
# Prefix match
if valid.lower().startswith(key_lower) or key_lower.startswith(valid.lower()):
return valid
# Substring match
if key_lower in valid.lower() or valid.lower() in key_lower:
return valid
return None
def _format_types(types: tuple) -> str:
"""Format type tuple for error messages."""
type_names = []
for t in types:
if t is bool:
type_names.append("bool")
elif t is str:
type_names.append("str")
elif t is dict:
type_names.append("table")
elif t is list:
type_names.append("array")
else:
type_names.append(t.__name__)
return " or ".join(type_names)
def _validate_feature_config(feature: str, config: Dict[str, Any]) -> List[str]:
"""Validate a nested feature config (memory, knowledge, etc.)."""
errors = []
# Common valid keys for feature configs
common_keys = {"enabled", "backend", "provider", "preset", "verbose"}
# Feature-specific valid keys
feature_keys = {
"memory": {"enabled", "backend", "provider", "use_long_term", "use_short_term",
"user_id", "session_id", "db_path", "collection_name"},
"knowledge": {"enabled", "sources", "provider", "chunk_size", "chunk_overlap"},
"planning": {"enabled", "max_steps", "reasoning"},
"reflection": {"enabled", "max_reflect", "min_reflect", "llm"},
"guardrails": {"enabled", "max_retries", "validator"},
"web": {"enabled", "search", "fetch", "provider"},
"output": {"enabled", "verbose", "markdown", "stream", "metrics", "preset"},
"execution": {"enabled", "max_iter", "max_rpm", "max_execution_time", "preset"},
"caching": {"enabled", "prompt_caching", "ttl"},
"autonomy": {"enabled", "max_steps", "approval_required"},
"context": {"enabled", "max_tokens", "strategy"},
}
valid_keys = feature_keys.get(feature, common_keys)
for key in config.keys():
if key not in valid_keys:
similar = _suggest_similar_key(key, valid_keys)
if similar:
errors.append(f"[defaults.{feature}.{key}] Unknown key. Did you mean '{similar}'?")
# Don't error on unknown nested keys - allow flexibility
return errors
def get_config_path() -> Optional[Path]:
"""Get the path to the config file if it exists.
Returns:
Path to config file, or None if not found
"""
return _find_config_file()
# Convenience function to merge config defaults with explicit params
def apply_config_defaults(
param_name: str,
explicit_value: Any,
config_class: Optional[type] = None,
) -> Any:
"""Apply config defaults to a parameter if not explicitly set.
Args:
param_name: Name of the parameter (e.g., "memory", "knowledge")
explicit_value: Value explicitly passed by user (None if not set)
config_class: Optional config class to instantiate
Returns:
Resolved value with config defaults applied
Example:
# User passes memory=True, config has memory.backend="postgres"
# Result: MemoryConfig with backend="postgres"
memory = apply_config_defaults("memory", True, MemoryConfig)
"""
# If user explicitly passed a value, respect it
if explicit_value is not None:
# If True and we have config defaults, merge them
if explicit_value is True:
config_defaults = get_default(param_name)
if config_defaults and isinstance(config_defaults, dict):
if config_class:
# Check if enabled key exists and is True
if config_defaults.get("enabled", True):
return config_class(**{k: v for k, v in config_defaults.items() if k != "enabled"})
return config_defaults
return explicit_value
# Check if config has defaults for this param
config_defaults = get_default(param_name)
if config_defaults is None:
return None
# If config has enabled=True, return the config
if isinstance(config_defaults, dict):
if config_defaults.get("enabled", False):
if config_class:
return config_class(**{k: v for k, v in config_defaults.items() if k != "enabled"})
return True
return None
@@ -0,0 +1,156 @@
# PraisonAI Agents Configuration File
# Copy this file to .praisonai/config.toml in your project directory
# or to ~/.praisonai/config.toml for global settings
# ============================================================================
# PLUGINS CONFIGURATION
# ============================================================================
# Enable background plugins (hooks, metrics, logging)
# Note: Tools and guardrails work WITHOUT enabling plugins
[plugins]
# Enable all plugins: true
# Enable specific plugins: ["logging", "metrics"]
# Disable: false
enabled = false
# Auto-discover plugins from default directories
auto_discover = true
# Plugin directories to scan
directories = [
"./.praisonai/plugins/",
"~/.praisonai/plugins/"
]
# ============================================================================
# DEFAULT SETTINGS
# ============================================================================
# These defaults are used when you pass True to a feature parameter
# Example: Agent(memory=True) will use [defaults.memory] settings
[defaults]
# Default LLM model
model = "gpt-4o-mini"
# Custom LLM endpoint (optional)
# base_url = "http://localhost:11434/v1"
# Feature flags
allow_delegation = false
allow_code_execution = false
code_execution_mode = "safe"
# ============================================================================
# MEMORY DEFAULTS
# ============================================================================
# Used when: Agent(memory=True)
[defaults.memory]
enabled = false
backend = "file" # file, sqlite, redis, postgres, mongodb
# user_id = "" # Auto-generated if empty
auto_memory = false
history = false
history_limit = 10
# Learning configuration
[defaults.memory.learn]
enabled = false
persona = true
insights = true
patterns = false
# ============================================================================
# KNOWLEDGE DEFAULTS
# ============================================================================
# Used when: Agent(knowledge=True)
[defaults.knowledge]
enabled = false
embedder = "openai"
chunking_strategy = "semantic"
chunk_size = 1000
chunk_overlap = 200
retrieval_k = 5
retrieval_threshold = 0.0
rerank = false
# ============================================================================
# PLANNING DEFAULTS
# ============================================================================
# Used when: Agent(planning=True)
[defaults.planning]
enabled = false
reasoning = false
auto_approve = false
read_only = false
# ============================================================================
# REFLECTION DEFAULTS
# ============================================================================
# Used when: Agent(reflection=True)
[defaults.reflection]
enabled = false
min_iterations = 1
max_iterations = 3
# ============================================================================
# WEB DEFAULTS
# ============================================================================
# Used when: Agent(web=True)
[defaults.web]
enabled = false
search = true
fetch = true
search_provider = "duckduckgo"
max_results = 5
# ============================================================================
# OUTPUT DEFAULTS
# ============================================================================
# Used when: Agent(output=True) or default output mode
[defaults.output]
preset = "silent" # silent, actions, verbose, json, stream, status
verbose = false
markdown = false
stream = false
metrics = false
# ============================================================================
# EXECUTION DEFAULTS
# ============================================================================
# Used when: Agent(execution=True) or default execution limits
[defaults.execution]
max_iter = 20
max_retry_limit = 2
# max_rpm = null
# max_execution_time = null
# ============================================================================
# CACHING DEFAULTS
# ============================================================================
# Used when: Agent(caching=True)
[defaults.caching]
enabled = true
prompt_caching = false
# ============================================================================
# AUTONOMY DEFAULTS
# ============================================================================
# Used when: Agent(autonomy=True)
[defaults.autonomy]
level = "suggest" # suggest, auto_edit, full_auto
escalation_enabled = true
escalation_threshold = 3
doom_loop_detection = true
max_consecutive_failures = 3
require_approval_for_writes = true
require_approval_for_shell = true
@@ -220,10 +220,11 @@ class LearnManager:
"""
Generate context suitable for injection into system prompt.
Returns a formatted string with relevant learnings.
Returns a formatted string with relevant learnings from all enabled stores.
"""
parts = []
# 1. Persona - User preferences and profile
if "persona" in self._stores:
personas = self.get_persona_context(limit=5)
if personas:
@@ -231,6 +232,7 @@ class LearnManager:
for p in personas:
parts.append(f"- {p}")
# 2. Insights - Observations and learnings
if "insights" in self._stores:
insights = self.get_insights_context(limit=5)
if insights:
@@ -238,4 +240,38 @@ class LearnManager:
for i in insights:
parts.append(f"- {i}")
# 3. Patterns - Reusable knowledge patterns
if "patterns" in self._stores:
entries = self._stores["patterns"].list_all(limit=3)
if entries:
parts.append("\nKnown Patterns:")
for e in entries:
parts.append(f"- {e.content}")
# 4. Decisions - Past decision records (for consistency)
if "decisions" in self._stores:
entries = self._stores["decisions"].list_all(limit=3)
if entries:
parts.append("\nPast Decisions:")
for e in entries:
parts.append(f"- {e.content}")
# 5. Feedback - User feedback signals
if "feedback" in self._stores:
entries = self._stores["feedback"].list_all(limit=3)
if entries:
parts.append("\nUser Feedback:")
for e in entries:
parts.append(f"- {e.content}")
# 6. Improvements - Self-improvement proposals
if "improvements" in self._stores:
entries = self._stores["improvements"].list_all(limit=3)
if entries:
parts.append("\nImprovement Areas:")
for e in entries:
parts.append(f"- {e.content}")
# Note: threads are session-specific and not included in system prompt
return "\n".join(parts) if parts else ""
@@ -37,6 +37,11 @@ Usage:
"""
__all__ = [
# Easy enable API
"enable",
"disable",
"list_plugins",
"is_enabled",
# Core
"PluginManager",
"Plugin",
@@ -65,6 +70,170 @@ __all__ = [
]
# ============================================================================
# EASY ENABLE API
# ============================================================================
# These functions provide a simple way to enable/disable plugins globally.
# Zero performance impact when not called - plugins only load on enable().
# ============================================================================
# Global state for plugin system (lazy initialized)
_plugins_enabled: bool = False
_enabled_plugin_names: list = None # None = all, list = specific
def enable(plugins: list = None) -> None:
"""Enable the plugin system.
Discovers and enables plugins from default directories.
This is the main entry point for using plugins.
Args:
plugins: Optional list of plugin names to enable.
If None, enables all discovered plugins.
Examples:
# Enable all discovered plugins
from praisonaiagents import plugins
plugins.enable()
# Enable specific plugins only
plugins.enable(["logging", "metrics"])
Note:
- Tools and guardrails work WITHOUT calling enable()
- Only background plugins (hooks, metrics, logging) need enable()
- Can also be enabled via PRAISONAI_PLUGINS env var
- Can also be enabled via .praisonai/config.toml
"""
global _plugins_enabled, _enabled_plugin_names
_plugins_enabled = True
_enabled_plugin_names = plugins # None = all, list = specific
# Get plugin manager and auto-discover
from .manager import get_plugin_manager
manager = get_plugin_manager()
# Auto-discover plugins from default directories
manager.auto_discover_plugins()
# Enable specific plugins or all
if plugins is not None:
# Enable only specified plugins
for name in plugins:
manager.enable(name)
else:
# Enable all discovered plugins
for plugin_info in manager.list_plugins():
manager.enable(plugin_info.get("name", ""))
import logging
logging.debug(f"Plugins enabled: {plugins if plugins else 'all'}")
def disable(plugins: list = None) -> None:
"""Disable plugins.
Args:
plugins: Optional list of plugin names to disable.
If None, disables all plugins.
Examples:
# Disable all plugins
plugins.disable()
# Disable specific plugins
plugins.disable(["logging"])
"""
global _plugins_enabled, _enabled_plugin_names
from .manager import get_plugin_manager
manager = get_plugin_manager()
if plugins is not None:
# Disable specific plugins
for name in plugins:
manager.disable(name)
else:
# Disable all plugins
_plugins_enabled = False
_enabled_plugin_names = None
for plugin_info in manager.list_plugins():
manager.disable(plugin_info.get("name", ""))
def list_plugins() -> list:
"""List all discovered plugins.
Returns:
List of plugin info dicts with name, version, enabled status.
Example:
from praisonaiagents import plugins
all_plugins = plugins.list_plugins()
for p in all_plugins:
print(f"{p['name']} v{p['version']} - {'enabled' if p['enabled'] else 'disabled'}")
"""
from .manager import get_plugin_manager
manager = get_plugin_manager()
# Auto-discover if not already done
if not manager._plugins and not manager._single_file_plugins:
manager.auto_discover_plugins()
# Combine registered plugins and single-file plugins
result = []
# Add registered plugins (Plugin class instances)
for info in manager.list_plugins():
if hasattr(info, 'name'):
# PluginInfo object
result.append({
"name": info.name,
"version": getattr(info, 'version', '1.0.0'),
"description": getattr(info, 'description', ''),
"enabled": manager.is_enabled(info.name),
"type": "registered",
})
elif isinstance(info, dict):
# Already a dict
info["enabled"] = manager.is_enabled(info.get("name", ""))
info["type"] = "registered"
result.append(info)
# Add single-file plugins
for name, meta in manager._single_file_plugins.items():
result.append({
"name": name,
"version": meta.get("version", "1.0.0"),
"description": meta.get("description", ""),
"enabled": manager.is_enabled(name),
"type": "single_file",
})
return result
def is_enabled(name: str = None) -> bool:
"""Check if plugins are enabled.
Args:
name: Optional plugin name to check. If None, checks if system is enabled.
Returns:
True if enabled, False otherwise.
"""
global _plugins_enabled
if name is None:
return _plugins_enabled
from .manager import get_plugin_manager
manager = get_plugin_manager()
return manager.is_enabled(name)
def __getattr__(name: str):
"""Lazy load module components."""
# Core classes
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "1.4.0"
version = "1.4.2"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
@@ -0,0 +1,229 @@
"""
TDD Tests for Learn Context Injection.
Tests that when memory="learn" is used, the learn context is automatically
injected into the system prompt without requiring a new Agent parameter.
Key behaviors:
1. memory="learn" preset enables LearnManager
2. Learn context is auto-injected into system prompt
3. No new Agent params needed (uses existing memory= param)
4. Zero overhead when learn is not enabled
"""
import pytest # noqa: F401 - used for test markers
class TestLearnPresetResolution:
"""Test that memory='learn' resolves correctly."""
def test_learn_preset_exists_in_memory_presets(self):
"""MEMORY_PRESETS should have 'learn' preset with learn=True."""
from praisonaiagents.config.presets import MEMORY_PRESETS
assert "learn" in MEMORY_PRESETS
assert MEMORY_PRESETS["learn"]["learn"] is True
assert MEMORY_PRESETS["learn"]["backend"] == "file"
def test_memory_learn_string_resolves_to_config_with_learn_true(self):
"""memory='learn' should resolve to MemoryConfig with learn=True."""
from praisonaiagents.config.param_resolver import resolve
from praisonaiagents.config.feature_configs import MemoryConfig
from praisonaiagents.config.presets import MEMORY_PRESETS, MEMORY_URL_SCHEMES
result = resolve(
value="learn",
param_name="memory",
config_class=MemoryConfig,
presets=MEMORY_PRESETS,
url_schemes=MEMORY_URL_SCHEMES,
)
assert isinstance(result, MemoryConfig)
assert result.learn is True
class TestAgentLearnContextInjection:
"""Test that Agent injects learn context into system prompt."""
def test_agent_with_memory_learn_has_learn_manager(self):
"""Agent with memory='learn' should have access to LearnManager."""
from praisonaiagents import Agent
# Create agent with memory="learn"
agent = Agent(
name="test_agent",
instructions="Test instructions",
memory="learn",
)
# Memory instance should exist
assert agent._memory_instance is not None
# Memory should have learn property that returns LearnManager
if hasattr(agent._memory_instance, 'learn'):
learn_manager = agent._memory_instance.learn
assert learn_manager is not None
def test_get_learn_context_returns_empty_when_no_learnings(self):
"""get_learn_context should return empty string when no learnings exist."""
from praisonaiagents import Agent
import uuid
# Use unique user_id to ensure clean state
unique_user_id = f"test_user_{uuid.uuid4().hex[:8]}"
agent = Agent(
name="test_agent",
instructions="Test instructions",
memory={"backend": "file", "learn": True, "user_id": unique_user_id},
)
# Should have get_learn_context method
assert hasattr(agent, 'get_learn_context')
# Should return empty string when no learnings (fresh user)
context = agent.get_learn_context()
assert context == "" or context is None or (isinstance(context, str) and len(context) == 0)
def test_get_learn_context_returns_learnings_when_present(self):
"""get_learn_context should return formatted learnings when present."""
from praisonaiagents import Agent
agent = Agent(
name="test_agent",
instructions="Test instructions",
memory="learn",
)
# Add some learnings via the memory's learn manager
if agent._memory_instance and hasattr(agent._memory_instance, 'learn'):
learn_manager = agent._memory_instance.learn
if learn_manager:
learn_manager.capture_persona("User prefers concise responses")
learn_manager.capture_insight("User works in data science")
# get_learn_context should now return formatted context
context = agent.get_learn_context()
# Should contain the captured learnings
if context:
assert "concise" in context.lower() or "User Preferences" in context
def test_build_system_prompt_includes_learn_context(self):
"""_build_system_prompt should include learn context when memory='learn'."""
from praisonaiagents import Agent
agent = Agent(
name="test_agent",
role="Test Role",
goal="Test Goal",
backstory="Test Backstory",
instructions="Test instructions",
memory="learn",
)
# Add some learnings
if agent._memory_instance and hasattr(agent._memory_instance, 'learn'):
learn_manager = agent._memory_instance.learn
if learn_manager:
learn_manager.capture_persona("User prefers detailed explanations")
# Build system prompt
system_prompt = agent._build_system_prompt()
# System prompt should include learn context section
# Note: This test will fail until we implement the fix
if agent._memory_instance and hasattr(agent._memory_instance, 'learn'):
learn_manager = agent._memory_instance.learn
if learn_manager and learn_manager.get_persona_context():
assert "Learned" in system_prompt or "User Preferences" in system_prompt or "detailed explanations" in system_prompt.lower()
class TestLearnContextZeroOverhead:
"""Test that learn context has zero overhead when not enabled."""
def test_no_learn_context_when_memory_true(self):
"""memory=True should NOT enable learn context."""
from praisonaiagents import Agent
agent = Agent(
name="test_agent",
instructions="Test instructions",
memory=True,
)
# get_learn_context should return empty or None
context = agent.get_learn_context()
assert context == "" or context is None
def test_no_learn_context_when_memory_false(self):
"""memory=False should NOT enable learn context."""
from praisonaiagents import Agent
agent = Agent(
name="test_agent",
instructions="Test instructions",
memory=False,
)
# get_learn_context should return empty or None
context = agent.get_learn_context()
assert context == "" or context is None
def test_no_learn_context_when_memory_none(self):
"""memory=None should NOT enable learn context."""
from praisonaiagents import Agent
agent = Agent(
name="test_agent",
instructions="Test instructions",
)
# get_learn_context should return empty or None
context = agent.get_learn_context()
assert context == "" or context is None
class TestMemoryLearnProperty:
"""Test Memory.learn property and get_learn_context method."""
def test_memory_learn_property_returns_learn_manager_when_enabled(self):
"""Memory.learn should return LearnManager when learn=True in config."""
from praisonaiagents.memory.memory import Memory
memory = Memory({"learn": True})
# learn property should return LearnManager
learn_manager = memory.learn
assert learn_manager is not None
# Should have capture methods
assert hasattr(learn_manager, 'capture_persona')
assert hasattr(learn_manager, 'capture_insight')
assert hasattr(learn_manager, 'to_system_prompt_context')
def test_memory_learn_property_returns_none_when_disabled(self):
"""Memory.learn should return None when learn is not enabled."""
from praisonaiagents.memory.memory import Memory
memory = Memory({}) # No learn config
# learn property should return None
assert memory.learn is None
def test_memory_get_learn_context_returns_formatted_string(self):
"""Memory.get_learn_context should return formatted context string."""
from praisonaiagents.memory.memory import Memory
memory = Memory({"learn": True})
# Add some learnings
memory.learn.capture_persona("User prefers Python")
memory.learn.capture_insight("User is a developer")
# get_learn_context should return formatted string
context = memory.get_learn_context()
assert isinstance(context, str)
assert "Python" in context or "User Preferences" in context
@@ -0,0 +1,283 @@
"""
Unit tests for plugins.enable() API and config-driven defaults.
Tests:
- plugins.enable() / plugins.disable()
- PRAISONAI_PLUGINS env var
- Config file loading
- Config-driven defaults for Agent parameters
"""
import os
import pytest
import tempfile
from pathlib import Path
from unittest.mock import patch
class TestPluginsEnableAPI:
"""Test the plugins.enable() API."""
def test_enable_all_plugins(self):
"""Test enabling all plugins."""
from praisonaiagents import plugins
# Reset state
plugins._plugins_enabled = False
plugins._enabled_plugin_names = None
# Enable all
plugins.enable()
assert plugins.is_enabled() is True
assert plugins._enabled_plugin_names is None # None means all
def test_enable_specific_plugins(self):
"""Test enabling specific plugins by name."""
from praisonaiagents import plugins
# Reset state
plugins._plugins_enabled = False
plugins._enabled_plugin_names = None
# Enable specific plugins
plugins.enable(["logging", "metrics"])
assert plugins.is_enabled() is True
assert plugins._enabled_plugin_names == ["logging", "metrics"]
def test_disable_all_plugins(self):
"""Test disabling all plugins."""
from praisonaiagents import plugins
# First enable
plugins._plugins_enabled = True
plugins._enabled_plugin_names = None
# Then disable
plugins.disable()
assert plugins.is_enabled() is False
def test_list_plugins_returns_list(self):
"""Test that list_plugins returns a list."""
from praisonaiagents import plugins
result = plugins.list_plugins()
assert isinstance(result, list)
def test_is_enabled_default_false(self):
"""Test that plugins are disabled by default."""
from praisonaiagents import plugins
# Reset state
plugins._plugins_enabled = False
assert plugins.is_enabled() is False
class TestPluginsEnvVar:
"""Test PRAISONAI_PLUGINS environment variable."""
def test_env_var_true_enables_plugins(self):
"""Test that PRAISONAI_PLUGINS=true enables plugins."""
from praisonaiagents._config import _get_plugins_enabled
with patch.dict(os.environ, {"PRAISONAI_PLUGINS": "true"}):
assert _get_plugins_enabled() is True
def test_env_var_false_disables_plugins(self):
"""Test that PRAISONAI_PLUGINS=false disables plugins."""
from praisonaiagents._config import _get_plugins_enabled
with patch.dict(os.environ, {"PRAISONAI_PLUGINS": "false"}):
assert _get_plugins_enabled() is False
def test_env_var_list_enables_specific(self):
"""Test that PRAISONAI_PLUGINS=logging,metrics enables specific plugins."""
from praisonaiagents._config import _get_plugins_enabled, _get_plugins_list
with patch.dict(os.environ, {"PRAISONAI_PLUGINS": "logging,metrics"}):
assert _get_plugins_enabled() is True
assert _get_plugins_list() == ["logging", "metrics"]
def test_env_var_empty_disables(self):
"""Test that empty PRAISONAI_PLUGINS disables plugins."""
from praisonaiagents._config import _get_plugins_enabled
with patch.dict(os.environ, {"PRAISONAI_PLUGINS": ""}):
assert _get_plugins_enabled() is False
class TestConfigLoader:
"""Test config file loading."""
def test_get_config_returns_praison_config(self):
"""Test that get_config returns a PraisonConfig."""
from praisonaiagents.config.loader import get_config, PraisonConfig, clear_config_cache
clear_config_cache()
config = get_config()
assert isinstance(config, PraisonConfig)
def test_get_plugins_config(self):
"""Test getting plugins config."""
from praisonaiagents.config.loader import get_plugins_config, PluginsConfig, clear_config_cache
clear_config_cache()
plugins_config = get_plugins_config()
assert isinstance(plugins_config, PluginsConfig)
def test_get_defaults_config(self):
"""Test getting defaults config."""
from praisonaiagents.config.loader import get_defaults_config, DefaultsConfig, clear_config_cache
clear_config_cache()
defaults_config = get_defaults_config()
assert isinstance(defaults_config, DefaultsConfig)
def test_get_default_with_fallback(self):
"""Test get_default returns fallback when key not found."""
from praisonaiagents.config.loader import get_default, clear_config_cache
clear_config_cache()
result = get_default("nonexistent_key", "fallback_value")
assert result == "fallback_value"
def test_is_plugins_enabled_from_config(self):
"""Test is_plugins_enabled checks config file."""
from praisonaiagents.config.loader import is_plugins_enabled, clear_config_cache
clear_config_cache()
# Without env var, should check config (which defaults to False)
with patch.dict(os.environ, {}, clear=True):
# Remove PRAISONAI_PLUGINS if present
os.environ.pop("PRAISONAI_PLUGINS", None)
result = is_plugins_enabled()
assert isinstance(result, bool)
def test_config_file_parsing(self):
"""Test parsing a TOML config file."""
from praisonaiagents.config.loader import _parse_toml, clear_config_cache
# Create a temp TOML file
with tempfile.NamedTemporaryFile(mode='w', suffix='.toml', delete=False) as f:
f.write("""
[plugins]
enabled = true
auto_discover = true
[defaults]
model = "gpt-4o-mini"
[defaults.memory]
enabled = true
backend = "file"
""")
f.flush()
try:
result = _parse_toml(Path(f.name))
assert result.get("plugins", {}).get("enabled") is True
assert result.get("defaults", {}).get("model") == "gpt-4o-mini"
assert result.get("defaults", {}).get("memory", {}).get("backend") == "file"
finally:
os.unlink(f.name)
class TestConfigDrivenDefaults:
"""Test config-driven defaults for Agent parameters."""
def test_apply_config_defaults_with_true(self):
"""Test that True uses config defaults."""
from praisonaiagents.config.loader import apply_config_defaults, clear_config_cache
from praisonaiagents.config.feature_configs import MemoryConfig
clear_config_cache()
# When user passes True, should use config defaults
# Without config file, should just return True
result = apply_config_defaults("memory", True, MemoryConfig)
# Should return True or a MemoryConfig
assert result is True or isinstance(result, MemoryConfig)
def test_apply_config_defaults_with_none(self):
"""Test that None checks config for enabled."""
from praisonaiagents.config.loader import apply_config_defaults, clear_config_cache
clear_config_cache()
# When user passes None, should check config
result = apply_config_defaults("memory", None, None)
# Without config, should return None
assert result is None
def test_apply_config_defaults_explicit_value(self):
"""Test that explicit values are preserved."""
from praisonaiagents.config.loader import apply_config_defaults, clear_config_cache
from praisonaiagents.config.feature_configs import MemoryConfig
clear_config_cache()
# When user passes explicit config, should preserve it
explicit_config = MemoryConfig(backend="postgres")
result = apply_config_defaults("memory", explicit_config, MemoryConfig)
assert result == explicit_config
class TestToolsGuardrailsExempt:
"""Test that tools and guardrails work without plugins.enable()."""
def test_tools_work_without_enable(self):
"""Test that tools work without calling plugins.enable()."""
from praisonaiagents import plugins
# Ensure plugins are disabled
plugins._plugins_enabled = False
# Tools should still work - this is tested by Agent creation
# Just verify the flag is false
assert plugins.is_enabled() is False
def test_guardrails_work_without_enable(self):
"""Test that guardrails work without calling plugins.enable()."""
from praisonaiagents import plugins
# Ensure plugins are disabled
plugins._plugins_enabled = False
# Guardrails should still work - this is tested by Agent creation
# Just verify the flag is false
assert plugins.is_enabled() is False
class TestPluginManagerIntegration:
"""Test PluginManager integration with enable API."""
def test_get_plugin_manager_singleton(self):
"""Test that get_plugin_manager returns singleton."""
from praisonaiagents.plugins import get_plugin_manager
manager1 = get_plugin_manager()
manager2 = get_plugin_manager()
assert manager1 is manager2
def test_plugin_manager_has_required_methods(self):
"""Test that PluginManager has all required methods."""
from praisonaiagents.plugins import get_plugin_manager
manager = get_plugin_manager()
# Check required methods exist
assert hasattr(manager, 'register')
assert hasattr(manager, 'unregister')
assert hasattr(manager, 'enable')
assert hasattr(manager, 'disable')
assert hasattr(manager, 'is_enabled')
assert hasattr(manager, 'list_plugins')
assert hasattr(manager, 'auto_discover_plugins')
assert hasattr(manager, 'execute_hook')
if __name__ == "__main__":
pytest.main([__file__, "-v"])
+1 -1
View File
@@ -2549,7 +2549,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "1.4.0"
version = "1.4.2"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.4.0.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.4.0.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.4.2.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.4.2.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+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==4.4.0 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==4.4.2 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+1 -1
View File
@@ -1 +1 @@
__version__ = "4.4.0"
__version__ = "4.4.2"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=1.4.0",
"praisonaiagents>=1.4.2",
"python-dotenv>=0.19.0",
"litellm>=1.81.0",
"PyYAML>=6.0",
+4 -4
View File
@@ -4054,7 +4054,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 = ">=1.4.0" },
{ name = "praisonaiagents", specifier = ">=1.4.2" },
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
{ name = "pyjwt", marker = "extra == 'os'", specifier = ">=2.8.0" },
{ name = "pyngrok", marker = "extra == 'call'", specifier = ">=1.4.0" },
@@ -4110,7 +4110,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "1.4.0"
version = "1.4.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -4119,9 +4119,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/44/82/ba80875a378f6443b46e69af55b76d89fbff9200ed5fd1ed657efcea98ec/praisonaiagents-1.4.0.tar.gz", hash = "sha256:2fafeacfcd796737a6e3ace447a005d09080dad8ccc2f2cab35cd15b229d0559", size = 1177142, upload-time = "2026-02-02T17:53:09.827Z" }
sdist = { url = "https://files.pythonhosted.org/packages/f8/54/ec17e7e4331cd191511d073f236e797856ab5a6998761117b6e51ee6d2f8/praisonaiagents-1.4.2.tar.gz", hash = "sha256:53760024b034d3ff166b6ee499e74fc29f41a301f6054608f68d13be310fa14f", size = 1184243, upload-time = "2026-02-03T08:42:10.738Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/39/b9/6f023c9d13009a8ba47f47d7588ecee1436b3c8b79ca47b2584a18577eb0/praisonaiagents-1.4.0-py3-none-any.whl", hash = "sha256:2d145b0c39782486246d90b5d6984dc646e539b08884adad36ea1f74b407ef8f", size = 1171050, upload-time = "2026-02-02T17:53:05.123Z" },
{ url = "https://files.pythonhosted.org/packages/47/a1/de39d2bd2a25387c513c3bc87a87fb578bea15620bd11e6133f89561a820/praisonaiagents-1.4.2-py3-none-any.whl", hash = "sha256:3d5b47c83f7b0f013b9eda85d2d870f5ef8a7e03a1326ea57d0ab3576d65f201", size = 1179296, upload-time = "2026-02-03T08:42:08.118Z" },
]
[[package]]