Compare commits

...
3 Commits
Author SHA1 Message Date
Mervin Praison 585b68d02c Release v3.6.2 2026-01-10 09:08:53 +00:00
Mervin Praison 66481cc51c Release v3.6.1 2026-01-09 17:16:27 +00:00
Mervin Praison 8e637daecd Release v3.6.0 2026-01-09 15:45:57 +00:00
40 changed files with 3606 additions and 102 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>=3.5.9" \
"praisonai>=3.6.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>=3.5.9" \
"praisonai>=3.6.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>=3.5.9" \
"praisonai>=3.6.2" \
"praisonai[ui]" \
"praisonai[crewai]"
+16
View File
@@ -23,6 +23,7 @@ examples/
| Category | Examples | Docs |
|----------|----------|------|
| **Consolidated Params** | [consolidated_params/](consolidated_params/) | [📖](https://docs.praison.ai/api/consolidated-params) |
| **Agents** | [python/agents/](python/agents/) | [📖](https://docs.praison.ai/concepts/agents) |
| **Workflows** | [python/workflows/](python/workflows/) | [📖](https://docs.praison.ai/features/workflows) |
| **Model Router** | [python/agents/router-agent-cost-optimization.py](python/agents/router-agent-cost-optimization.py) | [📖](https://docs.praison.ai/features/model-router) |
@@ -35,6 +36,21 @@ examples/
| **Serve** | [serve/](serve/) | [📖](https://docs.praison.ai/cli/serve) |
| **Endpoints** | [serve/](serve/) | [📖](https://docs.praison.ai/cli/endpoints) |
## Consolidated Params Examples
Agent-centric API with unified parameter resolution. Precedence: Instance > Config > Array > Dict > String > Bool > Default
| Example | Description |
|---------|-------------|
| [basic_agent.py](consolidated_params/basic_agent.py) | Minimal agent with memory |
| [basic_agents.py](consolidated_params/basic_agents.py) | Multi-agent with memory+planning |
| [basic_workflow.py](consolidated_params/basic_workflow.py) | Workflow with consolidated params |
| [basic_memory.py](consolidated_params/basic_memory.py) | Memory presets (file, redis, postgres) |
| [basic_guardrails.py](consolidated_params/basic_guardrails.py) | Guardrails with callable or config |
| [basic_workflow_agentlike.py](consolidated_params/basic_workflow_agentlike.py) | Workflow with agent-like params |
| [basic_step_override.py](consolidated_params/basic_step_override.py) | Step-level override of workflow defaults |
| [advanced_workflow_full_features.py](consolidated_params/advanced_workflow_full_features.py) | All consolidated params |
## Serve Examples
| Example | Description | CLI Command |
@@ -0,0 +1,17 @@
"""
Advanced Output + Execution Example - Agent-Centric API
Demonstrates preset + override array syntax.
"""
from praisonaiagents import Agent
# Advanced: Preset with overrides using array syntax
agent = Agent(
instructions="You are a helpful assistant.",
output=["verbose", {"stream": True, "metrics": True}],
execution=["thorough", {"max_iter": 30}],
)
response = agent.start("Explain quantum computing in simple terms.")
print(response)
@@ -0,0 +1,55 @@
"""Advanced Workflow with all consolidated params configured."""
from praisonaiagents import Agent
from praisonaiagents.workflows import Workflow, WorkflowStep
from praisonaiagents.workflows.workflow_configs import (
WorkflowOutputConfig, WorkflowPlanningConfig, WorkflowMemoryConfig,
)
# Full-featured workflow with all consolidated params
workflow = Workflow(
name="FullFeaturedWorkflow",
description="Demonstrates all agent-like consolidated params",
steps=[
WorkflowStep(
name="researcher",
action="Research: {{input}}",
agent=Agent(instructions="You are a researcher."),
knowledge=["research_docs/"], # Step-specific knowledge
web="duckduckgo", # Use DuckDuckGo for this step
),
WorkflowStep(
name="analyzer",
action="Analyze the research",
agent=Agent(instructions="You are an analyst."),
reflection="thorough", # Thorough reflection for analysis
guardrails="strict", # Strict validation
),
WorkflowStep(
name="writer",
action="Write the final report",
agent=Agent(instructions="You are a writer."),
autonomy=True, # Enable autonomy for writing
),
],
# Workflow-level consolidated params
output=WorkflowOutputConfig(verbose=True, stream=True),
planning=WorkflowPlanningConfig(enabled=True, reasoning=True),
memory=WorkflowMemoryConfig(backend="file"),
# Agent-like params
autonomy=True,
knowledge=["shared_docs/"],
guardrails=True,
web=True,
reflection=True,
)
if __name__ == "__main__":
print(f"Workflow: {workflow.name}")
print(f"Output verbose: {workflow._verbose}")
print(f"Planning enabled: {workflow._planning_enabled}")
print(f"Memory config: {workflow._memory_config}")
print(f"Autonomy: {workflow.autonomy}")
print(f"Knowledge: {workflow.knowledge}")
print(f"Guardrails: {workflow.guardrails}")
print(f"Web: {workflow.web}")
print(f"Reflection: {workflow.reflection}")
@@ -0,0 +1,12 @@
"""Basic Agent example with consolidated params."""
from praisonaiagents import Agent
# Minimal agent with memory
agent = Agent(
instructions="You are a helpful assistant.",
memory=True,
)
if __name__ == "__main__":
response = agent.chat("Hello!")
print(response)
@@ -0,0 +1,17 @@
"""Basic Agents (multi-agent) example with consolidated params."""
from praisonaiagents import Agent, Agents
# Create agents
writer = Agent(instructions="You write content.")
editor = Agent(instructions="You edit content.")
# Multi-agent with consolidated params
agents = Agents(
agents=[writer, editor],
memory=True,
planning=True,
)
if __name__ == "__main__":
result = agents.start("Write a haiku about AI")
print(result)
@@ -0,0 +1,16 @@
"""
Basic Execution Example - Agent-Centric API
Demonstrates execution preset usage.
"""
from praisonaiagents import Agent
# Basic: Use fast execution preset
agent = Agent(
instructions="You are a helpful assistant.",
execution="fast", # Presets: fast, balanced, thorough, unlimited
)
response = agent.start("What is the capital of France?")
print(response)
@@ -0,0 +1,26 @@
"""
Basic Guardrails Example - Agent-Centric API
Demonstrates guardrails with callable validator.
"""
from praisonaiagents import Agent
# Define a simple guardrail validator
def validate_response(output):
"""Validate that response doesn't contain forbidden words."""
forbidden = ["error", "fail", "cannot"]
text = str(output).lower()
for word in forbidden:
if word in text:
return (False, f"Response contains forbidden word: {word}")
return (True, output)
# Basic: Use callable guardrail
agent = Agent(
instructions="You are a helpful assistant.",
guardrails=validate_response,
)
response = agent.start("What is 2 + 2?")
print(response)
@@ -0,0 +1,21 @@
"""
Basic Memory Example - Agent-Centric API
Demonstrates the simplest memory usage with consolidated params.
"""
from praisonaiagents import Agent
# Basic: Enable memory with preset
agent = Agent(
instructions="You are a helpful assistant with memory.",
memory="sqlite", # Preset: sqlite, redis, postgres, file
)
# Run a simple task
response = agent.start("Remember that my favorite color is blue.")
print(response)
# Follow-up to test memory
response = agent.start("What is my favorite color?")
print(response)
@@ -0,0 +1,16 @@
"""
Basic Output Example - Agent-Centric API
Demonstrates output preset usage.
"""
from praisonaiagents import Agent
# Basic: Use verbose output preset
agent = Agent(
instructions="You are a helpful assistant.",
output="verbose", # Presets: minimal, normal, verbose, debug, silent
)
response = agent.start("What is 2 + 2?")
print(response)
@@ -0,0 +1,33 @@
"""Basic example showing step-level override of workflow defaults."""
from praisonaiagents import Agent
from praisonaiagents.workflows import Workflow, WorkflowStep
# Workflow with defaults
workflow = Workflow(
name="StepOverrideWorkflow",
steps=[
WorkflowStep(
name="step1",
action="Process with workflow defaults",
agent=Agent(instructions="You are a processor."),
# Uses workflow-level web=True (inherited)
),
WorkflowStep(
name="step2",
action="Process with step override",
agent=Agent(instructions="You are a processor."),
# Override workflow default
web=False, # Disable web for this step
reflection="thorough", # Use thorough reflection preset
),
],
# Workflow defaults
web=True,
reflection=True,
)
if __name__ == "__main__":
print(f"Workflow web default: {workflow.web}")
print(f"Step1 web: {workflow.steps[0].web}") # None (uses workflow default)
print(f"Step2 web: {workflow.steps[1].web}") # False (overridden)
print(f"Step2 reflection: {workflow.steps[1].reflection}") # thorough
@@ -0,0 +1,26 @@
"""
Basic Workflow Example - Agent-Centric API
Demonstrates minimal workflow with consolidated params.
"""
from praisonaiagents import Agent
from praisonaiagents.workflows import Workflow, WorkflowStep
# Create a simple workflow
workflow = Workflow(
name="simple_workflow",
output="verbose", # Workflow-level output preset
)
# Add a step
step = WorkflowStep(
name="writer",
action="Write a haiku about programming",
agent=Agent(instructions="You are a creative poet."),
output="result.txt", # Save output to file
)
# Note: This is a structural example - actual execution requires workflow.run()
print("Workflow created with output preset:", workflow.output)
print("Step output file:", step.output_file)
@@ -0,0 +1,32 @@
"""Basic Workflow with Agent-like consolidated params."""
from praisonaiagents import Agent
from praisonaiagents.workflows import Workflow, WorkflowStep
# Workflow with agent-like params (knowledge, web, guardrails, reflection)
workflow = Workflow(
name="AgentLikeWorkflow",
steps=[
WorkflowStep(
name="researcher",
action="Research the topic: {{input}}",
agent=Agent(instructions="You are a researcher."),
),
WorkflowStep(
name="writer",
action="Write about the research findings",
agent=Agent(instructions="You are a writer."),
),
],
# Agent-like consolidated params
knowledge=["docs/"], # Enable RAG
web=True, # Enable web search
guardrails=True, # Enable guardrails
reflection=True, # Enable self-reflection
)
if __name__ == "__main__":
print(f"Workflow: {workflow.name}")
print(f"Knowledge: {workflow.knowledge}")
print(f"Web: {workflow.web}")
print(f"Guardrails: {workflow.guardrails}")
print(f"Reflection: {workflow.reflection}")
@@ -0,0 +1,23 @@
"""Advanced memory example - with config and overrides."""
from praisonaiagents import Agent, MemoryConfig
# With full config
agent = Agent(
instructions="You are a helpful assistant with memory.",
memory=MemoryConfig(
backend="redis",
user_id="user123",
session_id="session456",
auto_memory=True,
),
)
# With array [preset, overrides]
agent_override = Agent(
instructions="You are a helpful assistant.",
memory=["redis", {"user_id": "custom_user"}],
)
if __name__ == "__main__":
response = agent.chat("Remember that my favorite color is blue.")
print(response)
@@ -0,0 +1,24 @@
"""Basic memory example - minimal usage."""
from praisonaiagents import Agent
# Simple enable (uses default file-based memory)
agent = Agent(
instructions="You are a helpful assistant with memory.",
memory=True,
)
# With preset
agent_redis = Agent(
instructions="You are a helpful assistant.",
memory="redis", # Uses redis preset
)
# With URL
agent_postgres = Agent(
instructions="You are a helpful assistant.",
memory="postgresql://localhost/mydb", # URL auto-detected
)
if __name__ == "__main__":
response = agent.chat("Remember that my favorite color is blue.")
print(response)
@@ -0,0 +1,13 @@
"""Basic output and execution example."""
from praisonaiagents import Agent
# Simple preset usage
agent = Agent(
instructions="You are a helpful assistant.",
output="verbose", # Preset: verbose output
execution="thorough", # Preset: thorough execution
)
if __name__ == "__main__":
response = agent.chat("What is 2+2?")
print(response)
@@ -0,0 +1,22 @@
"""Basic workflow example with consolidated params."""
from praisonaiagents import Agent
from praisonaiagents.workflows import Workflow, WorkflowStep
# Create agents
writer = Agent(instructions="You are a content writer.")
editor = Agent(instructions="You are an editor.")
# Create workflow with consolidated params
workflow = Workflow(
name="Content Pipeline",
steps=[
WorkflowStep(name="write", agent=writer, action="Write about {{topic}}"),
WorkflowStep(name="edit", agent=editor, action="Edit the content", context=["write"]),
],
output="verbose",
planning=True,
)
if __name__ == "__main__":
result = workflow.run(variables={"topic": "AI agents"})
print(result)
@@ -177,6 +177,13 @@ def __getattr__(name):
_lazy_cache["PraisonAIAgents"] = Agents # Also cache alias
return Agents
elif name == "PraisonAIAgents":
import warnings
warnings.warn(
"PraisonAIAgents is deprecated, use Agents instead. "
"Example: from praisonaiagents import Agents",
DeprecationWarning,
stacklevel=2
)
from .agents.agents import Agents
_lazy_cache["Agents"] = Agents
_lazy_cache[name] = Agents
@@ -0,0 +1,49 @@
"""
Resolver helper functions for consolidated parameters.
These helpers wrap the canonical resolver for specific parameter types,
handling special cases like callable validators for guardrails.
"""
from typing import Any, Type
from .config.param_resolver import resolve, ArrayMode
from .config.presets import GUARDRAIL_PRESETS
def resolve_guardrails(value: Any, config_class: Type) -> Any:
"""
Resolve guardrails parameter.
Handles special case of callable validators which have highest precedence.
Args:
value: The guardrails parameter value
config_class: GuardrailConfig class
Returns:
Resolved config, callable, or string
"""
# Handle None
if value is None:
return None
# Handle False (disabled)
if value is False:
return None
# Handle callable (highest precedence after None check)
# This includes functions, lambdas, and callable objects
if callable(value) and not isinstance(value, type):
return value
# Use canonical resolver for everything else
return resolve(
value=value,
param_name="guardrails",
config_class=config_class,
presets=GUARDRAIL_PRESETS,
array_mode=ArrayMode.PRESET_OVERRIDE,
string_mode="llm_prompt",
default=None,
)
@@ -522,34 +522,45 @@ class Agent:
cache, prompt_caching = True, None
# ─────────────────────────────────────────────────────────────────────
# Resolve HOOKS param
# Supports: None, list of callables, HooksConfig
# Resolve HOOKS param using unified resolver
# Supports: None, list of callables, HooksConfig, dict
# ─────────────────────────────────────────────────────────────────────
_hooks_list = []
step_callback = None
if hooks is not None:
if isinstance(hooks, list):
_hooks_list = hooks
elif isinstance(hooks, HooksConfig):
step_callback = hooks.on_step
_hooks_list = hooks.middleware or []
elif isinstance(hooks, dict):
step_callback = hooks.get('on_step')
_hooks_list = hooks.get('middleware', [])
_hooks_config = resolve(
value=hooks,
param_name="hooks",
config_class=HooksConfig,
array_mode=ArrayMode.PASSTHROUGH,
default=None,
)
if _hooks_config is not None:
if isinstance(_hooks_config, list):
_hooks_list = _hooks_config
elif isinstance(_hooks_config, HooksConfig):
step_callback = _hooks_config.on_step
_hooks_list = _hooks_config.middleware or []
# ─────────────────────────────────────────────────────────────────────
# Resolve SKILLS param
# Supports: None, list of paths, SkillsConfig
# Resolve SKILLS param using unified resolver
# Supports: None, list of paths, SkillsConfig, str
# ─────────────────────────────────────────────────────────────────────
_skills = None
if skills is not None:
if isinstance(skills, list):
_skills = skills
elif isinstance(skills, SkillsConfig):
_skills = skills.paths
skills_dirs = skills.dirs
elif isinstance(skills, str):
_skills = [skills]
skills_dirs = None
_skills_config = resolve(
value=skills,
param_name="skills",
config_class=SkillsConfig,
array_mode=ArrayMode.SOURCES,
string_mode="path_as_source",
default=None,
)
if _skills_config is not None:
if isinstance(_skills_config, SkillsConfig):
_skills = _skills_config.paths
skills_dirs = _skills_config.dirs
elif isinstance(_skills_config, list):
_skills = _skills_config
# ─────────────────────────────────────────────────────────────────────
# Resolve MEMORY param using unified resolver
@@ -595,36 +606,42 @@ class Agent:
memory = None
# ─────────────────────────────────────────────────────────────────────
# Resolve KNOWLEDGE param
# Resolve KNOWLEDGE param using unified resolver
# Supports: None, bool, str path, list of sources, KnowledgeConfig, Instance
# ─────────────────────────────────────────────────────────────────────
retrieval_config = None
embedder_config = None
if knowledge is not None and knowledge is not False:
if isinstance(knowledge, list):
# List of sources - use directly
pass
elif hasattr(knowledge, 'search') and hasattr(knowledge, 'add'):
_knowledge_config = resolve(
value=knowledge,
param_name="knowledge",
config_class=KnowledgeConfig,
instance_check=lambda v: hasattr(v, 'search') and hasattr(v, 'add'),
array_mode=ArrayMode.SOURCES_WITH_CONFIG,
string_mode="path_as_source",
default=None,
)
if _knowledge_config is not None:
if hasattr(_knowledge_config, 'search') and hasattr(_knowledge_config, 'add'):
# Knowledge instance - pass through
pass
elif isinstance(knowledge, KnowledgeConfig):
embedder_config = knowledge.embedder_config
if knowledge.config:
retrieval_config = knowledge.config
knowledge = _knowledge_config
elif isinstance(_knowledge_config, KnowledgeConfig):
embedder_config = _knowledge_config.embedder_config
if _knowledge_config.config:
retrieval_config = _knowledge_config.config
else:
retrieval_config = {
'top_k': knowledge.retrieval_k,
'threshold': knowledge.retrieval_threshold,
'rerank': knowledge.rerank,
'rerank_model': knowledge.rerank_model,
'top_k': _knowledge_config.retrieval_k,
'threshold': _knowledge_config.retrieval_threshold,
'rerank': _knowledge_config.rerank,
'rerank_model': _knowledge_config.rerank_model,
}
knowledge = knowledge.sources if knowledge.sources else None
elif isinstance(knowledge, str):
# Single path - convert to list
knowledge = [knowledge]
elif knowledge is True:
knowledge = None
knowledge = _knowledge_config.sources if _knowledge_config.sources else None
elif isinstance(_knowledge_config, list):
knowledge = _knowledge_config
elif knowledge is False:
knowledge = None
# ─────────────────────────────────────────────────────────────────────
# Resolve PLANNING param
@@ -681,22 +698,25 @@ class Agent:
self_reflect = False
# ─────────────────────────────────────────────────────────────────────
# Resolve GUARDRAILS param
# Resolve GUARDRAILS param using unified resolver
# Supports: None, bool, callable, str prompt, GuardrailConfig
# ─────────────────────────────────────────────────────────────────────
if guardrails is not None:
if guardrails is False:
guardrail = None
elif callable(guardrails):
guardrail = guardrails
elif isinstance(guardrails, GuardrailConfig):
guardrail = guardrails.validator or guardrails.llm_validator
max_guardrail_retries = guardrails.max_retries
elif isinstance(guardrails, str):
# Long string = LLM validator prompt
guardrail = guardrails
elif guardrails is True:
pass # No-op, user must provide validator
from .._resolver_helpers import resolve_guardrails as _resolve_guardrails
_guardrails_config = _resolve_guardrails(
value=guardrails,
config_class=GuardrailConfig,
)
if _guardrails_config is not None:
if callable(_guardrails_config) and not isinstance(_guardrails_config, type):
# Callable validator function
guardrail = _guardrails_config
elif isinstance(_guardrails_config, GuardrailConfig):
guardrail = _guardrails_config.validator or _guardrails_config.llm_validator
max_guardrail_retries = _guardrails_config.max_retries
elif isinstance(_guardrails_config, str):
# LLM validator prompt
guardrail = _guardrails_config
# ─────────────────────────────────────────────────────────────────────
# Resolve WEB param using unified resolver
@@ -142,6 +142,12 @@ class Agents:
output: Optional[Any] = None, # Union[str, MultiAgentOutputConfig]
execution: Optional[Any] = None, # Union[str, MultiAgentExecutionConfig]
hooks: Optional[Any] = None, # MultiAgentHooksConfig
# Additional consolidated params for feature parity with Agent
autonomy: Optional[Any] = None, # Union[bool, AutonomyConfig] - agent autonomy
knowledge: Optional[Any] = None, # Union[bool, List[str], KnowledgeConfig] - RAG
guardrails: Optional[Any] = None, # Union[bool, Callable, GuardrailConfig] - validation
web: Optional[Any] = None, # Union[bool, WebConfig] - web search/fetch
reflection: Optional[Any] = None, # Union[bool, ReflectionConfig] - self-reflection
):
"""
Initialize PraisonAIAgents with consolidated feature parameters.
@@ -159,7 +165,18 @@ class Agents:
output: Output configuration (str | MultiAgentOutputConfig)
execution: Execution configuration (str | MultiAgentExecutionConfig)
hooks: Hooks configuration (MultiAgentHooksConfig)
autonomy: Autonomy configuration (bool | AutonomyConfig)
knowledge: Knowledge/RAG configuration (bool | List[str] | KnowledgeConfig)
guardrails: Guardrails configuration (bool | Callable | GuardrailConfig)
web: Web search/fetch configuration (bool | WebConfig)
reflection: Self-reflection configuration (bool | ReflectionConfig)
"""
# Store new params for propagation to agents
self._autonomy = autonomy
self._knowledge = knowledge
self._guardrails = guardrails
self._web = web
self._reflection = reflection
# ─────────────────────────────────────────────────────────────────────
# Extract values from consolidated params using UNIFIED CANONICAL resolver
# Precedence: Instance > Config > Dict > Array > String > Bool > Default
@@ -12,7 +12,8 @@ This is the SINGLE, DRY resolver used by:
Performance: O(1) happy path. Typo suggestions only on error path.
"""
from typing import Any, Callable, Dict, Optional, Type, Union
from typing import Any, Callable, Dict, Optional, Set, Type, Union
import inspect
from .parse_utils import (
detect_url_scheme,
@@ -53,7 +54,7 @@ def resolve(
) -> Any:
"""
Resolve a consolidated parameter following precedence rules:
Instance > Config > Array > String > Bool > Default
Instance > Config > Array > Dict > String > Bool > Default
Args:
value: The parameter value
@@ -91,7 +92,20 @@ def resolve(
return value
# =========================================================================
# 4. Dict -> convert to config (STRICT validation)
# 4. Array handling (BEFORE Dict per AC-1 precedence)
# =========================================================================
if isinstance(value, (list, tuple)):
return _resolve_array(
value=value,
param_name=param_name,
config_class=config_class,
presets=presets,
url_schemes=url_schemes,
array_mode=array_mode,
)
# =========================================================================
# 5. Dict -> convert to config (STRICT validation)
# =========================================================================
if isinstance(value, dict):
if config_class:
@@ -119,19 +133,6 @@ def resolve(
f"Use a supported type: bool, str, or instance."
)
# =========================================================================
# 5. Array handling (before string to respect precedence)
# =========================================================================
if isinstance(value, (list, tuple)):
return _resolve_array(
value=value,
param_name=param_name,
config_class=config_class,
presets=presets,
url_schemes=url_schemes,
array_mode=array_mode,
)
# =========================================================================
# 6. String handling
# =========================================================================
@@ -220,7 +221,7 @@ def _resolve_array(
return config_class(sources=sources, **config_override)
return sources
# Single or list mode
# Single or list mode - SINGLE item only, multiple items raise error
if array_mode == ArrayMode.SINGLE_OR_LIST:
if len(value) == 1:
single_value = value[0]
@@ -232,10 +233,17 @@ def _resolve_array(
# Check if single item is a preset
if isinstance(single_value, str) and presets and single_value in presets:
return _apply_preset(single_value, presets, config_class)
# Multiple items - treat as sources
if config_class:
return config_class(sources=list(value))
return list(value)
# Single non-preset string - try as URL or return as-is
if isinstance(single_value, str):
if config_class:
return config_class()
return single_value
# Multiple items - ERROR (not allowed for SINGLE_OR_LIST mode)
raise TypeError(
f"Multiple values not allowed for {param_name}. "
f"Use a single value: {param_name}='preset' or {param_name}='url://...' "
f"Got {len(value)} items: {value}"
)
# Preset override mode - [preset, {overrides}] or [preset]
if array_mode == ArrayMode.PRESET_OVERRIDE:
@@ -401,16 +409,51 @@ def _get_config_fields(config_class: Type) -> str:
return "unknown"
def _get_valid_keys(config_class: Type) -> Optional[Set[str]]:
"""
Get valid keys from a config class.
Supports:
- Dataclasses (via __dataclass_fields__)
- Regular classes (via __init__ signature)
- Classes with __annotations__
Returns:
Set of valid key names, or None if cannot determine.
"""
# 1. Dataclass fields (most common)
if hasattr(config_class, '__dataclass_fields__'):
return set(config_class.__dataclass_fields__.keys())
# 2. __init__ signature parameters
try:
sig = inspect.signature(config_class.__init__)
params = set(sig.parameters.keys()) - {'self'}
if params:
return params
except (ValueError, TypeError):
pass
# 3. Class annotations
if hasattr(config_class, '__annotations__'):
return set(config_class.__annotations__.keys())
# Cannot determine valid keys
return None
def _validate_dict_keys(value: dict, config_class: Type) -> list:
"""
Validate dict keys against config class fields.
Returns list of unknown keys (empty if all valid).
O(n) where n = number of keys in dict (typically small).
"""
if not hasattr(config_class, '__dataclass_fields__'):
return [] # Not a dataclass, can't validate
valid_keys = set(config_class.__dataclass_fields__.keys())
Works for both dataclasses and regular classes.
"""
valid_keys = _get_valid_keys(config_class)
if valid_keys is None:
return [] # Cannot validate, allow all keys
provided_keys = set(value.keys())
unknown = provided_keys - valid_keys
return list(unknown)
@@ -231,20 +231,29 @@ class WorkflowStep:
loop_over: Optional[str] = None # Variable name to iterate over (e.g., "items")
loop_var: str = "item" # Variable name for current item in loop
# Guardrail (like process="workflow")
guardrail: Optional[Callable[['StepResult'], Tuple[bool, Any]]] = None # Returns (is_valid, feedback)
# Guardrails (standardized plural naming)
# Accepts: callable, str (LLM prompt), GuardrailConfig, or list [preset, overrides]
guardrails: Optional[Any] = None # Union[Callable, str, GuardrailConfig] - validation
# Backward compatibility alias (deprecated - use guardrails instead)
guardrail: Optional[Callable[['StepResult'], Tuple[bool, Any]]] = None # DEPRECATED: use guardrails
# Image handling (migrated from Task)
images: Optional[List[str]] = None # Image paths/URLs for vision tasks
# ============================================================
# CONSOLIDATED FEATURE PARAMS (workflow-centric API)
# Precedence: Instance > Config > String > Bool > Default
# Precedence: Instance > Config > Array > Dict > String > Bool > Default
# ============================================================
context: Optional[Any] = None # Union[bool, List[str], WorkflowStepContextConfig] - context handling
output: Optional[Any] = None # Union[str, WorkflowStepOutputConfig] - output handling
execution: Optional[Any] = None # Union[str, WorkflowStepExecutionConfig] - execution control
routing: Optional[Any] = None # Union[List[str], WorkflowStepRoutingConfig] - branching
# NEW: Agent-like consolidated params for feature parity
autonomy: Optional[Any] = None # Union[bool, AutonomyConfig] - agent autonomy
knowledge: Optional[Any] = None # Union[bool, List[str], KnowledgeConfig] - RAG
web: Optional[Any] = None # Union[bool, WebConfig] - web search/fetch
reflection: Optional[Any] = None # Union[bool, ReflectionConfig] - self-reflection
# Status tracking
status: str = "pending" # pending, running, completed, failed, skipped
@@ -264,6 +273,11 @@ class WorkflowStep:
_on_error: str = field(default="stop", repr=False)
_next_steps: Optional[List[str]] = field(default=None, repr=False)
_branch_condition: Optional[Dict[str, List[str]]] = field(default=None, repr=False)
# NEW: Resolved configs for agent-like params
_autonomy_config: Optional[Any] = field(default=None, repr=False)
_knowledge_config: Optional[Any] = field(default=None, repr=False)
_web_config: Optional[Any] = field(default=None, repr=False)
_reflection_config: Optional[Any] = field(default=None, repr=False)
def __post_init__(self):
"""Resolve consolidated params to internal values."""
@@ -272,6 +286,18 @@ class WorkflowStep:
resolve_step_execution_config, resolve_step_routing_config,
)
# Handle guardrails/guardrail compatibility
# guardrails (plural) takes precedence; guardrail (singular) is deprecated
if self.guardrails is None and self.guardrail is not None:
# Use deprecated guardrail value, map to guardrails
import warnings
warnings.warn(
"WorkflowStep 'guardrail' parameter is deprecated. Use 'guardrails' instead.",
DeprecationWarning,
stacklevel=2,
)
object.__setattr__(self, 'guardrails', self.guardrail)
# Resolve context param
ctx_cfg = resolve_step_context_config(self.context)
if ctx_cfg:
@@ -300,6 +326,50 @@ class WorkflowStep:
if route_cfg:
self._next_steps = route_cfg.next_steps
self._branch_condition = route_cfg.branches
# Resolve NEW agent-like params using canonical resolver
from ..config.param_resolver import resolve, ArrayMode
from ..config.presets import (
WEB_PRESETS, REFLECTION_PRESETS, AUTONOMY_PRESETS, KNOWLEDGE_PRESETS,
)
try:
from ..config.feature_configs import (
AutonomyConfig, KnowledgeConfig, WebConfig, ReflectionConfig,
)
except ImportError:
AutonomyConfig = KnowledgeConfig = WebConfig = ReflectionConfig = None
# Resolve autonomy
if AutonomyConfig and self.autonomy is not None:
self._autonomy_config = resolve(
value=self.autonomy, param_name="autonomy",
config_class=AutonomyConfig, presets=AUTONOMY_PRESETS,
default=None,
)
# Resolve knowledge
if KnowledgeConfig and self.knowledge is not None:
self._knowledge_config = resolve(
value=self.knowledge, param_name="knowledge",
config_class=KnowledgeConfig, presets=KNOWLEDGE_PRESETS,
array_mode=ArrayMode.SOURCES, default=None,
)
# Resolve web
if WebConfig and self.web is not None:
self._web_config = resolve(
value=self.web, param_name="web",
config_class=WebConfig, presets=WEB_PRESETS,
default=None,
)
# Resolve reflection
if ReflectionConfig and self.reflection is not None:
self._reflection_config = resolve(
value=self.reflection, param_name="reflection",
config_class=ReflectionConfig, presets=REFLECTION_PRESETS,
default=None,
)
# Property accessors for backward compatibility
@property
@@ -423,13 +493,20 @@ class Workflow:
# ============================================================
# CONSOLIDATED FEATURE PARAMS (workflow-centric API)
# Precedence: Instance > Config > String > Bool > Default
# Precedence: Instance > Config > Array > Dict > String > Bool > Default
# Workflow is "Agents in an organised way" - supports same params as Agents
# ============================================================
output: Optional[Any] = None # Union[str, WorkflowOutputConfig] - verbose/stream
planning: Optional[Any] = False # Union[bool, str, WorkflowPlanningConfig] - planning mode
memory: Optional[Any] = None # Union[bool, WorkflowMemoryConfig, instance] - memory
hooks: Optional[Any] = None # WorkflowHooksConfig - lifecycle callbacks
context: Optional[Any] = False # Union[bool, ManagerConfig, ContextManager] - context management
# NEW: Agent-like consolidated params for feature parity
autonomy: Optional[Any] = None # Union[bool, AutonomyConfig] - agent autonomy
knowledge: Optional[Any] = None # Union[bool, List[str], KnowledgeConfig] - RAG
guardrails: Optional[Any] = None # Union[bool, Callable, GuardrailConfig] - validation
web: Optional[Any] = None # Union[bool, WebConfig] - web search/fetch
reflection: Optional[Any] = None # Union[bool, ReflectionConfig] - self-reflection
# Status tracking
status: str = "not_started" # not_started, running, completed, failed
@@ -445,6 +522,12 @@ class Workflow:
_hooks_config: Optional[Any] = field(default=None, repr=False)
_context_manager: Optional[Any] = field(default=None, repr=False)
_context_manager_initialized: bool = field(default=False, repr=False)
# NEW: Resolved configs for agent-like params (for propagation to steps/agents)
_autonomy_config: Optional[Any] = field(default=None, repr=False)
_knowledge_config: Optional[Any] = field(default=None, repr=False)
_guardrails_config: Optional[Any] = field(default=None, repr=False)
_web_config: Optional[Any] = field(default=None, repr=False)
_reflection_config: Optional[Any] = field(default=None, repr=False)
def __post_init__(self):
"""Resolve consolidated params to internal values."""
@@ -474,6 +557,62 @@ class Workflow:
# Resolve hooks param
self._hooks_config = resolve_hooks_config(self.hooks)
# Resolve NEW agent-like params using canonical resolver
# These are stored for propagation to steps/agents
from ..config.param_resolver import resolve, ArrayMode
from ..config.presets import (
WEB_PRESETS, REFLECTION_PRESETS, GUARDRAIL_PRESETS,
AUTONOMY_PRESETS, KNOWLEDGE_PRESETS,
)
try:
from ..config.feature_configs import (
AutonomyConfig, KnowledgeConfig, GuardrailConfig,
WebConfig, ReflectionConfig,
)
except ImportError:
AutonomyConfig = KnowledgeConfig = GuardrailConfig = None
WebConfig = ReflectionConfig = None
# Resolve autonomy
if AutonomyConfig and self.autonomy is not None:
self._autonomy_config = resolve(
value=self.autonomy, param_name="autonomy",
config_class=AutonomyConfig, presets=AUTONOMY_PRESETS,
default=None,
)
# Resolve knowledge
if KnowledgeConfig and self.knowledge is not None:
self._knowledge_config = resolve(
value=self.knowledge, param_name="knowledge",
config_class=KnowledgeConfig, presets=KNOWLEDGE_PRESETS,
array_mode=ArrayMode.SOURCES, default=None,
)
# Resolve guardrails
if GuardrailConfig and self.guardrails is not None:
self._guardrails_config = resolve(
value=self.guardrails, param_name="guardrails",
config_class=GuardrailConfig, presets=GUARDRAIL_PRESETS,
default=None,
)
# Resolve web
if WebConfig and self.web is not None:
self._web_config = resolve(
value=self.web, param_name="web",
config_class=WebConfig, presets=WEB_PRESETS,
default=None,
)
# Resolve reflection
if ReflectionConfig and self.reflection is not None:
self._reflection_config = resolve(
value=self.reflection, param_name="reflection",
config_class=ReflectionConfig, presets=REFLECTION_PRESETS,
default=None,
)
def to_dict(self) -> Dict[str, Any]:
return {
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.10.7"
version = "0.10.10"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
@@ -0,0 +1,526 @@
"""
Comprehensive tests for param_resolver.py - TDD for outstanding gaps.
Tests cover:
1. Dict strict validation (dataclass and non-dataclass configs)
2. Array PRESET_OVERRIDE invalid preset error UX
3. Precedence ladder for all consolidated params
4. Agent consolidated params via canonical resolver
5. Workflow/WorkflowStep resolution consistency
"""
import pytest
from dataclasses import dataclass, field
from typing import List
# =============================================================================
# Test Fixtures - Config Classes
# =============================================================================
@dataclass
class MockConfig:
"""Mock config for testing."""
enabled: bool = True
value: str = "default"
count: int = 0
@dataclass
class MockConfigWithSources:
"""Mock config with sources field."""
sources: List[str] = field(default_factory=list)
rerank: bool = False
class NonDataclassConfig:
"""Non-dataclass config for testing validation."""
def __init__(self, enabled: bool = True, value: str = "default"):
self.enabled = enabled
self.value = value
# =============================================================================
# PHASE 1A: Dict Strict Validation Tests
# =============================================================================
class TestDictStrictValidation:
"""Tests for strict dict key validation."""
def test_dict_unknown_keys_error_dataclass(self):
"""Dict with unknown keys should raise TypeError with helpful message."""
from praisonaiagents.config.param_resolver import resolve
with pytest.raises(TypeError) as exc_info:
resolve(
value={"enabled": True, "unknown_key": "bad"},
param_name="test",
config_class=MockConfig,
)
error_msg = str(exc_info.value)
assert "unknown_key" in error_msg.lower() or "Unknown keys" in error_msg
assert "enabled" in error_msg or "Valid keys" in error_msg
def test_dict_valid_keys_accepted(self):
"""Dict with valid keys should create config."""
from praisonaiagents.config.param_resolver import resolve
result = resolve(
value={"enabled": False, "value": "custom"},
param_name="test",
config_class=MockConfig,
)
assert isinstance(result, MockConfig)
assert result.enabled is False
assert result.value == "custom"
def test_dict_no_config_class_error(self):
"""Dict without config_class should raise TypeError."""
from praisonaiagents.config.param_resolver import resolve
with pytest.raises(TypeError) as exc_info:
resolve(
value={"key": "value"},
param_name="test",
config_class=None,
)
assert "not supported" in str(exc_info.value).lower() or "no config class" in str(exc_info.value).lower()
def test_dict_validation_non_dataclass_config(self):
"""Non-dataclass config should still validate keys via signature/annotations."""
from praisonaiagents.config.param_resolver import _get_valid_keys
# This tests the _get_valid_keys helper handles non-dataclass
# The implementation should use inspect.signature or __annotations__
# For now, verify the helper exists and returns something useful
keys = _get_valid_keys(NonDataclassConfig)
# Should return valid keys or empty (graceful degradation)
assert isinstance(keys, (set, list, frozenset)) or keys is None
# =============================================================================
# PHASE 1B: Array PRESET_OVERRIDE Error UX Tests
# =============================================================================
class TestArrayPresetOverrideErrorUX:
"""Tests for array preset error messages with suggestions."""
def test_invalid_preset_in_array_raises_error_with_suggestions(self):
"""Invalid preset string in array should raise error with typo suggestions."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
presets = {
"fast": {"enabled": True, "value": "fast"},
"slow": {"enabled": True, "value": "slow"},
"balanced": {"enabled": True, "value": "balanced"},
}
with pytest.raises((ValueError, TypeError)) as exc_info:
resolve(
value=["fats"], # Typo for "fast"
param_name="execution",
config_class=MockConfig,
presets=presets,
array_mode=ArrayMode.PRESET_OVERRIDE,
)
error_msg = str(exc_info.value).lower()
# Should suggest similar preset
assert "fast" in error_msg or "fats" in error_msg
# Should list valid presets
assert "valid" in error_msg or "available" in error_msg or any(p in error_msg for p in presets.keys())
def test_valid_preset_in_array_works(self):
"""Valid preset string in array should work."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
presets = {
"fast": {"enabled": True, "value": "fast"},
}
result = resolve(
value=["fast"],
param_name="execution",
config_class=MockConfig,
presets=presets,
array_mode=ArrayMode.PRESET_OVERRIDE,
)
assert isinstance(result, MockConfig)
assert result.value == "fast"
def test_preset_with_overrides_in_array(self):
"""Preset with override dict in array should merge correctly."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
presets = {
"fast": {"enabled": True, "value": "fast", "count": 1},
}
result = resolve(
value=["fast", {"count": 99}],
param_name="execution",
config_class=MockConfig,
presets=presets,
array_mode=ArrayMode.PRESET_OVERRIDE,
)
assert isinstance(result, MockConfig)
assert result.value == "fast"
assert result.count == 99
# =============================================================================
# PHASE 1C: Precedence Ladder Tests
# =============================================================================
class TestPrecedenceLadder:
"""Tests for Instance > Config > Dict > Array > String > Bool > Default precedence."""
def test_instance_highest_precedence(self):
"""Instance check should have highest precedence."""
from praisonaiagents.config.param_resolver import resolve
class MockInstance:
def search(self): pass
def add(self): pass
instance = MockInstance()
result = resolve(
value=instance,
param_name="memory",
config_class=MockConfig,
instance_check=lambda v: hasattr(v, 'search') and hasattr(v, 'add'),
)
assert result is instance
def test_config_class_instance_precedence(self):
"""Config class instance should be returned as-is."""
from praisonaiagents.config.param_resolver import resolve
config = MockConfig(enabled=False, value="custom")
result = resolve(
value=config,
param_name="test",
config_class=MockConfig,
)
assert result is config
def test_dict_before_string(self):
"""Dict should be processed before string."""
from praisonaiagents.config.param_resolver import resolve
# Dict input
result = resolve(
value={"enabled": False},
param_name="test",
config_class=MockConfig,
)
assert isinstance(result, MockConfig)
assert result.enabled is False
def test_bool_true_returns_default_config(self):
"""Bool True should return default config instance."""
from praisonaiagents.config.param_resolver import resolve
result = resolve(
value=True,
param_name="test",
config_class=MockConfig,
)
assert isinstance(result, MockConfig)
assert result.enabled is True # Default
def test_bool_false_returns_none(self):
"""Bool False should return None (disabled)."""
from praisonaiagents.config.param_resolver import resolve
result = resolve(
value=False,
param_name="test",
config_class=MockConfig,
)
assert result is None
def test_none_returns_default(self):
"""None should return the default value."""
from praisonaiagents.config.param_resolver import resolve
default = MockConfig(value="default_value")
result = resolve(
value=None,
param_name="test",
config_class=MockConfig,
default=default,
)
assert result is default
# =============================================================================
# PHASE 1D: Agent Consolidated Params Tests
# =============================================================================
class TestAgentConsolidatedParams:
"""Tests for Agent consolidated params using canonical resolver."""
def test_agent_hooks_via_resolver(self):
"""Agent hooks param should use canonical resolver."""
from praisonaiagents.config.feature_configs import HooksConfig
from praisonaiagents.config.param_resolver import resolve, ArrayMode
# Test list passthrough
hooks_list = [lambda x: x]
result = resolve(
value=hooks_list,
param_name="hooks",
config_class=HooksConfig,
array_mode=ArrayMode.PASSTHROUGH,
)
assert result == hooks_list
# Test HooksConfig instance
hooks_config = HooksConfig(on_step=lambda x: x)
result = resolve(
value=hooks_config,
param_name="hooks",
config_class=HooksConfig,
)
assert result is hooks_config
# Test dict
result = resolve(
value={"on_step": None},
param_name="hooks",
config_class=HooksConfig,
)
assert isinstance(result, HooksConfig)
def test_agent_skills_via_resolver(self):
"""Agent skills param should use canonical resolver."""
from praisonaiagents.config.feature_configs import SkillsConfig
from praisonaiagents.config.param_resolver import resolve
# Test SkillsConfig instance directly
config = SkillsConfig(paths=["./skill1", "./skill2"])
result = resolve(
value=config,
param_name="skills",
config_class=SkillsConfig,
)
assert isinstance(result, SkillsConfig)
assert result.paths == ["./skill1", "./skill2"]
# Test dict input
result = resolve(
value={"paths": ["./my-skill"]},
param_name="skills",
config_class=SkillsConfig,
)
assert isinstance(result, SkillsConfig)
assert result.paths == ["./my-skill"]
def test_agent_knowledge_via_resolver(self):
"""Agent knowledge param should use canonical resolver."""
from praisonaiagents.config.feature_configs import KnowledgeConfig
from praisonaiagents.config.param_resolver import resolve, ArrayMode
# Test list of sources
result = resolve(
value=["docs/", "data.pdf"],
param_name="knowledge",
config_class=KnowledgeConfig,
array_mode=ArrayMode.SOURCES_WITH_CONFIG,
)
assert isinstance(result, KnowledgeConfig)
assert "docs/" in result.sources
# Test list with config override
result = resolve(
value=["docs/", {"rerank": True}],
param_name="knowledge",
config_class=KnowledgeConfig,
array_mode=ArrayMode.SOURCES_WITH_CONFIG,
)
assert isinstance(result, KnowledgeConfig)
assert result.rerank is True
def test_agent_guardrails_via_resolver(self):
"""Agent guardrails param should use canonical resolver."""
from praisonaiagents.config.feature_configs import GuardrailConfig
from praisonaiagents.config.param_resolver import resolve, ArrayMode
from praisonaiagents.config.presets import GUARDRAIL_PRESETS
# Test string preset
if GUARDRAIL_PRESETS:
preset_name = list(GUARDRAIL_PRESETS.keys())[0]
result = resolve(
value=preset_name,
param_name="guardrails",
config_class=GuardrailConfig,
presets=GUARDRAIL_PRESETS,
array_mode=ArrayMode.PRESET_OVERRIDE,
)
assert result is not None
# Test callable (should be handled via instance_check)
def validator_fn(x):
return (True, x)
result = resolve(
value=validator_fn,
param_name="guardrails",
config_class=GuardrailConfig,
instance_check=callable,
)
assert result is validator_fn
# Test GuardrailConfig instance
config = GuardrailConfig(max_retries=5)
result = resolve(
value=config,
param_name="guardrails",
config_class=GuardrailConfig,
)
assert result is config
# Test string as LLM prompt
result = resolve(
value="Ensure response is safe and helpful",
param_name="guardrails",
config_class=GuardrailConfig,
string_mode="llm_prompt",
)
assert isinstance(result, GuardrailConfig)
assert result.llm_validator == "Ensure response is safe and helpful"
# =============================================================================
# PHASE 1E: Workflow/WorkflowStep Resolution Tests
# =============================================================================
class TestWorkflowResolution:
"""Tests for Workflow and WorkflowStep resolution consistency."""
def test_workflow_step_context_resolved(self):
"""WorkflowStep context should be resolved to config."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(
name="test",
context=["step1", "step2"],
)
# Should have resolved _context_from
assert step._context_from == ["step1", "step2"]
def test_workflow_step_execution_resolved(self):
"""WorkflowStep execution should be resolved to config."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(
name="test",
execution="fast",
)
# Should have resolved execution config
assert step._async_execution is not None or step._max_retries is not None
def test_workflow_step_output_resolved(self):
"""WorkflowStep output should be resolved to config."""
from praisonaiagents.workflows.workflows import WorkflowStep
from praisonaiagents.workflows.workflow_configs import WorkflowStepOutputConfig
# Test with config object
step = WorkflowStep(
name="test",
output=WorkflowStepOutputConfig(file="result.txt"),
)
# Should have resolved output file
assert step._output_file == "result.txt"
def test_workflow_output_resolved(self):
"""Workflow output should be resolved to config."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(
name="test",
output="verbose",
)
# Should have resolved verbose setting
assert workflow._verbose is True
def test_workflow_planning_resolved(self):
"""Workflow planning should be resolved to config."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(
name="test",
planning=True,
)
# Should have resolved planning enabled
assert workflow._planning_enabled is True
# =============================================================================
# PHASE 1F: Naming/Alias Tests
# =============================================================================
class TestNamingAlias:
"""Tests for Agents vs PraisonAIAgents naming."""
def test_agents_is_primary(self):
"""Agents should be the primary import."""
from praisonaiagents import Agents
assert Agents is not None
def test_praisonaiagents_alias_works(self):
"""PraisonAIAgents alias should still work."""
from praisonaiagents import PraisonAIAgents
assert PraisonAIAgents is not None
def test_both_are_same_class(self):
"""Both names should refer to the same class."""
from praisonaiagents import Agents, PraisonAIAgents
assert Agents is PraisonAIAgents
# =============================================================================
# PHASE 1G: Helper Function Tests
# =============================================================================
class TestHelperFunctions:
"""Tests for resolver helper functions."""
def test_get_valid_keys_dataclass(self):
"""_get_valid_keys should return dataclass fields."""
from praisonaiagents.config.param_resolver import _get_valid_keys
keys = _get_valid_keys(MockConfig)
assert "enabled" in keys
assert "value" in keys
assert "count" in keys
def test_get_valid_keys_non_dataclass(self):
"""_get_valid_keys should handle non-dataclass via signature."""
from praisonaiagents.config.param_resolver import _get_valid_keys
keys = _get_valid_keys(NonDataclassConfig)
# Should return keys from __init__ signature or __annotations__
# or gracefully return empty/None
assert keys is None or isinstance(keys, (set, list, frozenset))
if __name__ == "__main__":
pytest.main([__file__, "-v"])
@@ -0,0 +1,720 @@
"""
TDD Tests for Resolver Standardization.
These tests verify:
1. Precedence order: None → Instance → Config → Array → Dict → String → Bool → Default
2. Memory array semantics: single=ok, multiple=error
3. Caching array support with PRESET_OVERRIDE
4. Workflow/WorkflowStep using canonical resolve
5. CLI resolver integration
6. Naming standardization (guardrails plural)
7. All precedence levels per consolidated param
Run with: pytest tests/unit/config/test_resolver_standardization.py -v
"""
import pytest
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
# =============================================================================
# Test Fixtures - Config Classes
# =============================================================================
@dataclass
class MockConfig:
"""Mock config class for testing."""
enabled: bool = True
value: str = "default"
count: int = 0
@dataclass
class MockMemoryConfig:
"""Mock memory config for testing."""
backend: str = "file"
user_id: Optional[str] = None
config: Optional[Dict[str, Any]] = None
sources: Optional[List[str]] = None
@dataclass
class MockCachingConfig:
"""Mock caching config for testing."""
enabled: bool = True
ttl: int = 3600
prompt_caching: bool = False
# =============================================================================
# TODO 1.1: Tests for Precedence Order (Array BEFORE Dict)
# =============================================================================
class TestPrecedenceOrder:
"""Test that precedence order is: None → Instance → Config → Array → Dict → String → Bool → Default"""
def test_array_takes_precedence_over_dict_interpretation(self):
"""Array input should be processed as array, not as dict."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
# When given an array, it should use array handling, not dict handling
result = resolve(
value=["preset1"],
param_name="test",
config_class=MockConfig,
presets={"preset1": {"enabled": True, "value": "from_preset"}},
array_mode=ArrayMode.PRESET_OVERRIDE,
)
# Should resolve via array path (preset), not dict path
assert isinstance(result, MockConfig)
assert result.value == "from_preset"
def test_precedence_instance_over_config(self):
"""Instance check should take precedence over config class check."""
from praisonaiagents.config.param_resolver import resolve
class CustomInstance:
def __init__(self):
self.is_custom = True
custom = CustomInstance()
result = resolve(
value=custom,
param_name="test",
config_class=MockConfig,
instance_check=lambda v: hasattr(v, 'is_custom'),
)
# Should return instance as-is, not try to convert
assert result is custom
assert result.is_custom is True
def test_precedence_config_over_array(self):
"""Config class instance should take precedence over array handling."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
config = MockConfig(enabled=False, value="config_value")
result = resolve(
value=config,
param_name="test",
config_class=MockConfig,
array_mode=ArrayMode.PRESET_OVERRIDE,
)
# Should return config as-is
assert result is config
assert result.value == "config_value"
def test_precedence_array_over_dict(self):
"""Array should be checked BEFORE dict in precedence."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
# This test verifies the fix: Array must be checked before Dict
# A list input should go through array handling, not dict handling
result = resolve(
value=["verbose", {"count": 5}],
param_name="test",
config_class=MockConfig,
presets={"verbose": {"enabled": True, "value": "verbose_mode"}},
array_mode=ArrayMode.PRESET_OVERRIDE,
)
assert isinstance(result, MockConfig)
assert result.value == "verbose_mode"
assert result.count == 5 # Override applied
def test_precedence_dict_over_string(self):
"""Dict should take precedence over string handling."""
from praisonaiagents.config.param_resolver import resolve
result = resolve(
value={"enabled": False, "value": "dict_value"},
param_name="test",
config_class=MockConfig,
presets={"dict_value": {"enabled": True}}, # Should NOT use this
)
assert isinstance(result, MockConfig)
assert result.value == "dict_value"
assert result.enabled is False
def test_precedence_string_over_bool(self):
"""String preset should take precedence over bool handling."""
from praisonaiagents.config.param_resolver import resolve
result = resolve(
value="verbose",
param_name="test",
config_class=MockConfig,
presets={"verbose": {"enabled": True, "value": "verbose_mode"}},
)
assert isinstance(result, MockConfig)
assert result.value == "verbose_mode"
def test_precedence_bool_over_default(self):
"""Bool True should create default config, not return default value."""
from praisonaiagents.config.param_resolver import resolve
result = resolve(
value=True,
param_name="test",
config_class=MockConfig,
default={"should": "not_use"},
)
assert isinstance(result, MockConfig)
# Should be default MockConfig(), not the default dict
def test_precedence_none_returns_default(self):
"""None should return the default value."""
from praisonaiagents.config.param_resolver import resolve
default_config = MockConfig(value="default_value")
result = resolve(
value=None,
param_name="test",
config_class=MockConfig,
default=default_config,
)
assert result is default_config
# =============================================================================
# TODO 1.2: Tests for Memory Array Semantics
# =============================================================================
class TestMemoryArraySemantics:
"""Test memory array semantics: single=ok, multiple=error."""
def test_memory_single_item_array_resolves_as_scalar(self):
"""memory=["redis"] should resolve as scalar preset."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
result = resolve(
value=["redis"],
param_name="memory",
config_class=MockMemoryConfig,
presets={"redis": {"backend": "redis"}},
array_mode=ArrayMode.SINGLE_OR_LIST,
)
assert isinstance(result, MockMemoryConfig)
assert result.backend == "redis"
def test_memory_multiple_items_raises_error(self):
"""memory=["redis", "file"] should raise TypeError."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
with pytest.raises(TypeError) as exc_info:
resolve(
value=["redis", "file"],
param_name="memory",
config_class=MockMemoryConfig,
presets={"redis": {"backend": "redis"}, "file": {"backend": "file"}},
array_mode=ArrayMode.SINGLE_OR_LIST,
)
# Error message should be helpful
assert "memory" in str(exc_info.value).lower()
assert "multiple" in str(exc_info.value).lower() or "single" in str(exc_info.value).lower()
def test_memory_single_url_in_array_resolves(self):
"""memory=["postgresql://..."] should resolve as URL."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
result = resolve(
value=["postgresql://localhost/db"],
param_name="memory",
config_class=MockMemoryConfig,
url_schemes={"postgresql": "postgres"},
array_mode=ArrayMode.SINGLE_OR_LIST,
)
assert isinstance(result, MockMemoryConfig)
assert result.backend == "postgres"
# =============================================================================
# TODO 1.3: Tests for Caching Array Support
# =============================================================================
class TestCachingArraySupport:
"""Test caching array support with PRESET_OVERRIDE."""
def test_caching_array_preset_override(self):
"""caching=["enabled", {"ttl": 7200}] should work."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
result = resolve(
value=["enabled", {"ttl": 7200}],
param_name="caching",
config_class=MockCachingConfig,
presets={"enabled": {"enabled": True, "ttl": 3600}},
array_mode=ArrayMode.PRESET_OVERRIDE,
)
assert isinstance(result, MockCachingConfig)
assert result.enabled is True
assert result.ttl == 7200 # Override applied
def test_caching_string_preset(self):
"""caching="enabled" should work."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
result = resolve(
value="enabled",
param_name="caching",
config_class=MockCachingConfig,
presets={"enabled": {"enabled": True}},
array_mode=ArrayMode.PRESET_OVERRIDE,
)
assert isinstance(result, MockCachingConfig)
assert result.enabled is True
# =============================================================================
# TODO 1.4: Tests for Workflow/WorkflowStep Using Canonical Resolve
# =============================================================================
class TestWorkflowCanonicalResolve:
"""Test that Workflow and WorkflowStep use canonical resolve directly."""
def test_workflow_output_uses_canonical_resolve(self):
"""Workflow output param should use canonical resolve."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(
name="test",
steps=[],
output="verbose",
)
assert workflow._verbose is True
def test_workflow_output_dict_validation(self):
"""Workflow output dict should have strict validation."""
from praisonaiagents.workflows.workflows import Workflow
with pytest.raises(TypeError) as exc_info:
Workflow(
name="test",
steps=[],
output={"invalid_key": True},
)
assert "invalid_key" in str(exc_info.value).lower() or "unknown" in str(exc_info.value).lower()
def test_workflowstep_context_uses_canonical_resolve(self):
"""WorkflowStep context param should use canonical resolve."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(
name="test",
context=["step1", "step2"],
)
assert step._context_from == ["step1", "step2"]
def test_workflowstep_execution_preset(self):
"""WorkflowStep execution preset should work."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(
name="test",
execution="thorough",
)
# Should have thorough preset values
assert step._max_retries >= 3 # Thorough has more retries
# =============================================================================
# TODO 1.5: Tests for CLI Resolver Integration
# =============================================================================
class TestCLIResolverIntegration:
"""Test CLI flags use canonical resolver."""
def test_cli_memory_flag_true(self):
"""CLI --memory flag (true) should resolve correctly."""
# This tests the integration point - CLI should pass value to canonical resolver
from praisonaiagents.config.param_resolver import resolve, ArrayMode
from praisonaiagents.config.presets import MEMORY_PRESETS, MEMORY_URL_SCHEMES
# Simulate CLI passing True
result = resolve(
value=True,
param_name="memory",
config_class=MockMemoryConfig,
presets=MEMORY_PRESETS,
url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST,
)
assert isinstance(result, MockMemoryConfig)
def test_cli_memory_flag_preset_string(self):
"""CLI --memory=redis should resolve as preset."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
from praisonaiagents.config.presets import MEMORY_PRESETS, MEMORY_URL_SCHEMES
result = resolve(
value="redis",
param_name="memory",
config_class=MockMemoryConfig,
presets=MEMORY_PRESETS,
url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST,
)
assert isinstance(result, MockMemoryConfig)
assert result.backend == "redis"
def test_cli_memory_flag_url(self):
"""CLI --memory=postgresql://... should resolve as URL."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
from praisonaiagents.config.presets import MEMORY_PRESETS, MEMORY_URL_SCHEMES
result = resolve(
value="postgresql://localhost/db",
param_name="memory",
config_class=MockMemoryConfig,
presets=MEMORY_PRESETS,
url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST,
)
assert isinstance(result, MockMemoryConfig)
assert result.backend == "postgres"
# =============================================================================
# TODO 1.6: Tests for Naming Standardization (guardrails plural)
# =============================================================================
class TestNamingStandardization:
"""Test guardrail → guardrails naming standardization."""
def test_workflowstep_accepts_guardrails_plural(self):
"""WorkflowStep should accept guardrails= (plural)."""
from praisonaiagents.workflows.workflows import WorkflowStep
def my_guardrail(result):
return (True, result)
# Should accept guardrails= (plural)
step = WorkflowStep(
name="test",
guardrails=my_guardrail,
)
# Should store in guardrails attribute
assert step.guardrails is my_guardrail or step.guardrail is my_guardrail
def test_workflowstep_guardrail_singular_compat(self):
"""WorkflowStep should accept guardrail= (singular) for compatibility."""
from praisonaiagents.workflows.workflows import WorkflowStep
def my_guardrail(result):
return (True, result)
# Should still accept guardrail= (singular) for backward compat
step = WorkflowStep(
name="test",
guardrail=my_guardrail,
)
# Should work (backward compat)
assert step.guardrail is my_guardrail
# =============================================================================
# TODO 1.7: Tests for All Precedence Levels Per Consolidated Param
# =============================================================================
class TestAllPrecedenceLevels:
"""Test each consolidated param has all 7 precedence levels working."""
def test_output_all_precedence_levels(self):
"""Test output param at all precedence levels."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
from praisonaiagents.config.feature_configs import OutputConfig
from praisonaiagents.config.presets import OUTPUT_PRESETS
# 1. None → Default
result = resolve(value=None, param_name="output", config_class=OutputConfig,
presets=OUTPUT_PRESETS, default=OutputConfig())
assert isinstance(result, OutputConfig)
# 2. Instance (custom object with expected attributes)
class CustomOutput:
verbose = True
stream = True
custom = CustomOutput()
result = resolve(value=custom, param_name="output", config_class=OutputConfig,
presets=OUTPUT_PRESETS, instance_check=lambda v: hasattr(v, 'verbose'))
assert result is custom
# 3. Config class instance
config = OutputConfig(verbose=True, stream=False)
result = resolve(value=config, param_name="output", config_class=OutputConfig,
presets=OUTPUT_PRESETS)
assert result is config
# 4. Array [preset, overrides]
result = resolve(value=["verbose", {"stream": False}], param_name="output",
config_class=OutputConfig, presets=OUTPUT_PRESETS,
array_mode=ArrayMode.PRESET_OVERRIDE)
assert isinstance(result, OutputConfig)
# 5. Dict
result = resolve(value={"verbose": True, "stream": False}, param_name="output",
config_class=OutputConfig, presets=OUTPUT_PRESETS)
assert isinstance(result, OutputConfig)
assert result.verbose is True
# 6. String preset
result = resolve(value="verbose", param_name="output", config_class=OutputConfig,
presets=OUTPUT_PRESETS)
assert isinstance(result, OutputConfig)
# 7. Bool True
result = resolve(value=True, param_name="output", config_class=OutputConfig,
presets=OUTPUT_PRESETS)
assert isinstance(result, OutputConfig)
# 8. Bool False → None
result = resolve(value=False, param_name="output", config_class=OutputConfig,
presets=OUTPUT_PRESETS)
assert result is None
def test_memory_all_precedence_levels(self):
"""Test memory param at all precedence levels."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
from praisonaiagents.config.feature_configs import MemoryConfig
from praisonaiagents.config.presets import MEMORY_PRESETS, MEMORY_URL_SCHEMES
# 1. None → Default
result = resolve(value=None, param_name="memory", config_class=MemoryConfig,
presets=MEMORY_PRESETS, url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST, default=None)
assert result is None
# 2. Config class instance
config = MemoryConfig(backend="redis")
result = resolve(value=config, param_name="memory", config_class=MemoryConfig,
presets=MEMORY_PRESETS, url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST)
assert result is config
# 3. Dict
result = resolve(value={"backend": "sqlite"}, param_name="memory",
config_class=MemoryConfig, presets=MEMORY_PRESETS,
url_schemes=MEMORY_URL_SCHEMES, array_mode=ArrayMode.SINGLE_OR_LIST)
assert isinstance(result, MemoryConfig)
assert result.backend == "sqlite"
# 4. String preset
result = resolve(value="redis", param_name="memory", config_class=MemoryConfig,
presets=MEMORY_PRESETS, url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST)
assert isinstance(result, MemoryConfig)
# 5. String URL
result = resolve(value="postgresql://localhost/db", param_name="memory",
config_class=MemoryConfig, presets=MEMORY_PRESETS,
url_schemes=MEMORY_URL_SCHEMES, array_mode=ArrayMode.SINGLE_OR_LIST)
assert isinstance(result, MemoryConfig)
# 6. Bool True
result = resolve(value=True, param_name="memory", config_class=MemoryConfig,
presets=MEMORY_PRESETS, url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST)
assert isinstance(result, MemoryConfig)
# 7. Bool False → None
result = resolve(value=False, param_name="memory", config_class=MemoryConfig,
presets=MEMORY_PRESETS, url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST)
assert result is None
# =============================================================================
# Tests for Error Messages and Typo Suggestions
# =============================================================================
class TestErrorMessages:
"""Test error messages are helpful with typo suggestions."""
def test_invalid_preset_suggests_similar(self):
"""Invalid preset should suggest similar valid presets."""
from praisonaiagents.config.param_resolver import resolve
with pytest.raises(ValueError) as exc_info:
resolve(
value="verbos", # Typo
param_name="output",
config_class=MockConfig,
presets={"verbose": {}, "minimal": {}, "debug": {}},
)
error_msg = str(exc_info.value).lower()
assert "verbose" in error_msg # Should suggest "verbose"
def test_unknown_dict_keys_lists_valid_keys(self):
"""Unknown dict keys should list valid keys."""
from praisonaiagents.config.param_resolver import resolve
with pytest.raises(TypeError) as exc_info:
resolve(
value={"invalid_key": True},
param_name="test",
config_class=MockConfig,
)
error_msg = str(exc_info.value).lower()
assert "invalid_key" in error_msg or "unknown" in error_msg
assert "enabled" in error_msg or "valid" in error_msg
# =============================================================================
# Tests for Strict Dict Validation
# =============================================================================
class TestStrictDictValidation:
"""Test strict dict validation."""
def test_dict_without_config_class_raises_error(self):
"""Dict input without config_class should raise TypeError."""
from praisonaiagents.config.param_resolver import resolve
with pytest.raises(TypeError) as exc_info:
resolve(
value={"key": "value"},
param_name="test",
config_class=None, # No config class
)
assert "dict" in str(exc_info.value).lower()
def test_dict_with_unknown_keys_raises_error(self):
"""Dict with unknown keys should raise TypeError."""
from praisonaiagents.config.param_resolver import resolve
with pytest.raises(TypeError) as exc_info:
resolve(
value={"enabled": True, "unknown_field": "value"},
param_name="test",
config_class=MockConfig,
)
error_msg = str(exc_info.value)
assert "unknown_field" in error_msg.lower() or "unknown" in error_msg.lower()
# =============================================================================
# PHASE 1 TDD: Tests for Workflow/WorkflowStep New Params (ALL GREEN target)
# =============================================================================
class TestWorkflowNewParams:
"""Test Workflow supports all consolidated params like Agents."""
def test_workflow_has_autonomy_param(self):
"""Workflow should have autonomy param."""
from praisonaiagents.workflows.workflows import Workflow
# Should accept autonomy param without error
workflow = Workflow(steps=[], autonomy=True)
assert hasattr(workflow, 'autonomy') or hasattr(workflow, '_autonomy_config')
def test_workflow_has_knowledge_param(self):
"""Workflow should have knowledge param."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(steps=[], knowledge=["docs/"])
assert hasattr(workflow, 'knowledge') or hasattr(workflow, '_knowledge_config')
def test_workflow_has_guardrails_param(self):
"""Workflow should have guardrails param."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(steps=[], guardrails=True)
assert hasattr(workflow, 'guardrails') or hasattr(workflow, '_guardrails_config')
def test_workflow_has_web_param(self):
"""Workflow should have web param."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(steps=[], web=True)
assert hasattr(workflow, 'web') or hasattr(workflow, '_web_config')
def test_workflow_has_reflection_param(self):
"""Workflow should have reflection param."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(steps=[], reflection=True)
assert hasattr(workflow, 'reflection') or hasattr(workflow, '_reflection_config')
class TestWorkflowStepNewParams:
"""Test WorkflowStep supports all consolidated params."""
def test_workflowstep_has_autonomy_param(self):
"""WorkflowStep should have autonomy param."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(name="test", autonomy=True)
assert hasattr(step, 'autonomy') or hasattr(step, '_autonomy_config')
def test_workflowstep_has_knowledge_param(self):
"""WorkflowStep should have knowledge param."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(name="test", knowledge=["docs/"])
assert hasattr(step, 'knowledge') or hasattr(step, '_knowledge_config')
def test_workflowstep_has_web_param(self):
"""WorkflowStep should have web param."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(name="test", web=True)
assert hasattr(step, 'web') or hasattr(step, '_web_config')
def test_workflowstep_has_reflection_param(self):
"""WorkflowStep should have reflection param."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(name="test", reflection=True)
assert hasattr(step, 'reflection') or hasattr(step, '_reflection_config')
class TestWorkflowPropagation:
"""Test workflow defaults propagate to steps/agents."""
def test_workflow_defaults_available_for_steps(self):
"""Workflow-level configs should be accessible for step propagation."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(
steps=[],
knowledge=["docs/"],
web=True,
guardrails=True,
)
# Workflow should store resolved configs for propagation
assert workflow.knowledge is not None or hasattr(workflow, '_knowledge_config')
assert workflow.web is not None or hasattr(workflow, '_web_config')
assert workflow.guardrails is not None or hasattr(workflow, '_guardrails_config')
if __name__ == "__main__":
pytest.main([__file__, "-v"])
+1 -1
View File
@@ -2422,7 +2422,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.10.7"
version = "0.10.10"
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/v3.5.9.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.5.9.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.6.2.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.6.2.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+71 -3
View File
@@ -5,20 +5,60 @@ Provides terminal-native interactive chat mode.
This command NEVER opens a browser - it runs entirely in the terminal.
"""
from typing import List, Optional
from typing import List, Optional, Union
import typer
app = typer.Typer(help="Terminal-native interactive chat mode")
def _parse_memory_flag(memory: Optional[str], no_memory: bool) -> Union[bool, str, None]:
"""
Parse memory CLI flag to value for Agent.
Precedence: --no-memory > --memory=value > --memory (flag) > None
Args:
memory: Memory flag value (None, "true", preset string, or URL)
no_memory: Whether --no-memory was specified
Returns:
- False if --no-memory
- True if --memory (flag only)
- str if --memory=preset or --memory=URL
- None if neither specified
"""
if no_memory:
return False
if memory is None:
return None
# --memory flag without value sets "true"
if memory.lower() == "true":
return True
# --memory=false explicitly disables
if memory.lower() == "false":
return False
# Otherwise it's a preset or URL string
return memory
@app.callback(invoke_without_command=True)
def chat_main(
ctx: typer.Context,
prompt: Optional[str] = typer.Argument(None, help="Initial prompt for chat"),
model: Optional[str] = typer.Option(None, "--model", "-m", help="LLM model to use"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output"),
memory: bool = typer.Option(False, "--memory", help="Enable memory"),
memory: Optional[str] = typer.Option(
None, "--memory",
help="Enable memory. Use --memory for default, --memory=redis for preset, --memory=postgresql://... for URL",
is_flag=False,
flag_value="true",
),
no_memory: bool = typer.Option(False, "--no-memory", help="Disable memory"),
tools: Optional[str] = typer.Option(None, "--tools", "-t", help="Tools file path"),
user_id: Optional[str] = typer.Option(None, "--user-id", help="User ID for memory isolation"),
session_id: Optional[str] = typer.Option(None, "--session", "-s", help="Session ID to resume"),
@@ -28,6 +68,31 @@ def chat_main(
no_acp: bool = typer.Option(False, "--no-acp", help="Disable ACP tools"),
no_lsp: bool = typer.Option(False, "--no-lsp", help="Disable LSP tools"),
autonomy: bool = typer.Option(True, "--autonomy/--no-autonomy", help="Enable agent autonomy for complex tasks"),
# NEW: Agent-like consolidated params for ALL GREEN consistency
knowledge: Optional[str] = typer.Option(
None, "--knowledge", "-k",
help="Enable knowledge/RAG. Use --knowledge for default, --knowledge=docs/ for sources",
is_flag=False,
flag_value="true",
),
guardrails: Optional[str] = typer.Option(
None, "--guardrails",
help="Enable guardrails. Use --guardrails for default, --guardrails=strict for preset",
is_flag=False,
flag_value="true",
),
web: Optional[str] = typer.Option(
None, "--web",
help="Enable web search. Use --web for default, --web=duckduckgo for preset",
is_flag=False,
flag_value="true",
),
reflection: Optional[str] = typer.Option(
None, "--reflection",
help="Enable self-reflection. Use --reflection for default, --reflection=thorough for preset",
is_flag=False,
flag_value="true",
),
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)"),
):
@@ -69,6 +134,9 @@ def chat_main(
typer.echo("⚠️ Profiling is only supported for single prompt mode.", err=True)
typer.echo(" Use: praisonai chat \"your prompt\" --profile", err=True)
# Parse memory flag: --no-memory takes precedence, then --memory value
memory_value = _parse_memory_flag(memory, no_memory)
# Try InteractiveCore first (preferred terminal-native implementation)
try:
from praisonai.cli.interactive import InteractiveCore, InteractiveConfig
@@ -80,7 +148,7 @@ def chat_main(
continue_session=continue_session,
workspace=workspace or None,
verbose=verbose,
memory=memory,
memory=memory_value,
files=list(file) if file else [],
autonomy=autonomy,
)
@@ -724,3 +724,92 @@ def docs_serve(
pass
finally:
sys.argv = original_argv
@app.command("report")
def docs_report(
report_dir: Optional[Path] = typer.Argument(
None,
help="Report directory path (default: auto-detect latest)",
),
base_dir: Optional[Path] = typer.Option(
None,
"--base-dir", "-b",
help="Base reports directory (default: ~/Downloads/reports/docs)",
),
limit: int = typer.Option(
50,
"--limit", "-n",
help="Limit rows in tables (0 = no limit)",
),
errors: bool = typer.Option(
True,
"--errors/--no-errors",
help="Show error grouping section",
),
wide: bool = typer.Option(
False,
"--wide/--no-wide",
help="Wide output: full error messages and item lists",
),
show_paths: bool = typer.Option(
True,
"--show-paths/--no-show-paths",
help="Show affected item paths in error groups",
),
match: Optional[List[str]] = typer.Option(
None,
"--match", "-m",
help="Filter by status (repeatable: --match failed --match timeout)",
),
group: Optional[List[str]] = typer.Option(
None,
"--group", "-g",
help="Filter by group name (repeatable)",
),
contains: Optional[str] = typer.Option(
None,
"--contains", "-c",
help="Filter items whose error contains text (case-insensitive)",
),
open_dir: bool = typer.Option(
False,
"--open",
help="Show report artifacts listing",
),
output_format: str = typer.Option(
"table",
"--format", "-f",
help="Output format: table or json",
),
):
"""
View execution report with failures and error grouping.
Examples:
praisonai docs report # View latest report
praisonai docs report ./my-report # View specific report
praisonai docs report --wide --limit 0 # Full details
praisonai docs report --match failed # Only failures
praisonai docs report --format json # JSON output
"""
# Lazy import to avoid loading at CLI startup
from praisonai.suite_runner.report_viewer import view_report
output, exit_code = view_report(
report_dir=report_dir,
suite="docs",
base_dir=base_dir,
limit=limit,
show_errors=errors,
wide=wide,
show_paths=show_paths,
match_statuses=match,
match_groups=group,
contains=contains,
show_artifacts=open_dir,
output_format=output_format,
)
typer.echo(output)
raise typer.Exit(exit_code)
@@ -691,5 +691,94 @@ def examples_run_all(
raise typer.Exit(0)
@app.command("report")
def examples_report(
report_dir: Optional[Path] = typer.Argument(
None,
help="Report directory path (default: auto-detect latest)",
),
base_dir: Optional[Path] = typer.Option(
None,
"--base-dir", "-b",
help="Base reports directory (default: ~/Downloads/reports/examples)",
),
limit: int = typer.Option(
50,
"--limit", "-n",
help="Limit rows in tables (0 = no limit)",
),
errors: bool = typer.Option(
True,
"--errors/--no-errors",
help="Show error grouping section",
),
wide: bool = typer.Option(
False,
"--wide/--no-wide",
help="Wide output: full error messages and item lists",
),
show_paths: bool = typer.Option(
True,
"--show-paths/--no-show-paths",
help="Show affected item paths in error groups",
),
match: Optional[List[str]] = typer.Option(
None,
"--match", "-m",
help="Filter by status (repeatable: --match failed --match timeout)",
),
group: Optional[List[str]] = typer.Option(
None,
"--group", "-g",
help="Filter by group name (repeatable)",
),
contains: Optional[str] = typer.Option(
None,
"--contains", "-c",
help="Filter items whose error contains text (case-insensitive)",
),
open_dir: bool = typer.Option(
False,
"--open",
help="Show report artifacts listing",
),
output_format: str = typer.Option(
"table",
"--format", "-f",
help="Output format: table or json",
),
):
"""
View execution report with failures and error grouping.
Examples:
praisonai examples report # View latest report
praisonai examples report ./my-report # View specific report
praisonai examples report --wide --limit 0 # Full details
praisonai examples report --match failed # Only failures
praisonai examples report --format json # JSON output
"""
# Lazy import to avoid loading at CLI startup
from praisonai.suite_runner.report_viewer import view_report
output, exit_code = view_report(
report_dir=report_dir,
suite="examples",
base_dir=base_dir,
limit=limit,
show_errors=errors,
wide=wide,
show_paths=show_paths,
match_statuses=match,
match_groups=group,
contains=contains,
show_artifacts=open_dir,
output_format=output_format,
)
typer.echo(output)
raise typer.Exit(exit_code)
if __name__ == "__main__":
app()
+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.5.9 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==3.6.2 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
@@ -0,0 +1,793 @@
"""
Report Viewer for suite execution reports.
Provides CLI-friendly viewing of JSON/CSV reports with:
- Latest report auto-detection
- Failure tables with filtering
- Error grouping and deduplication
- ASCII table rendering (no external deps)
Used by both `praisonai examples report` and `praisonai docs report`.
"""
from __future__ import annotations
import json
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Tuple
# =============================================================================
# Data Models
# =============================================================================
@dataclass
class ReportItem:
"""Normalized report item from JSON/CSV."""
item_id: str
suite: str
group: str
source_path: str
block_index: int = 0
status: str = "not_run"
duration_seconds: float = 0.0
exit_code: int = 0
error_type: Optional[str] = None
error_message: Optional[str] = None
error_signature: str = "" # Normalized for grouping
skip_reason: Optional[str] = None
stdout_path: Optional[str] = None
stderr_path: Optional[str] = None
start_time: str = ""
end_time: str = ""
runnable_decision: str = ""
@property
def display_name(self) -> str:
"""Human-readable display name."""
path = Path(self.source_path)
if self.suite == "docs" and self.block_index > 0:
return f"{path.name}:{self.block_index}"
return path.name
@dataclass
class ReportMetadata:
"""Report metadata from JSON."""
suite: str = ""
source_path: str = ""
timestamp: str = ""
platform: str = ""
python_version: str = ""
python_executable: str = ""
praisonai_version: str = ""
git_commit: Optional[str] = None
groups_run: List[str] = field(default_factory=list)
totals: Dict[str, int] = field(default_factory=dict)
@dataclass
class ErrorGroup:
"""Group of items with similar errors."""
signature: str
error_type: str
sample_message: str
count: int = 0
items: List[ReportItem] = field(default_factory=list)
@dataclass
class LoadedReport:
"""Complete loaded and normalized report."""
metadata: ReportMetadata
items: List[ReportItem]
report_dir: Path
@property
def failures(self) -> List[ReportItem]:
"""Get failed/timeout/error items."""
return [i for i in self.items if i.status in ("failed", "timeout", "error")]
# =============================================================================
# Error Signature Normalization
# =============================================================================
# Patterns to strip from error messages for grouping
_STRIP_PATTERNS = [
# File paths
(r'File "[^"]+[/\\][^"]+\.py"', 'File "...py"'),
(r'/[^\s:]+\.py', '...py'),
(r'[A-Z]:\\[^\s:]+\.py', '...py'),
# Line numbers
(r', line \d+', ', line N'),
(r'line \d+', 'line N'),
# Memory addresses
(r'0x[0-9a-fA-F]+', '0x...'),
# Request IDs, UUIDs
(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', 'UUID'),
(r'req_[a-zA-Z0-9]+', 'req_...'),
# Timestamps
(r'\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}[^\s]*', 'TIMESTAMP'),
# Numbers in brackets
(r'\[\d+\]', '[N]'),
# Port numbers
(r':\d{4,5}(?=/|$|\s)', ':PORT'),
# Whitespace normalization
(r'\s+', ' '),
]
def normalize_error_signature(error_type: Optional[str], error_message: Optional[str]) -> str:
"""
Create normalized error signature for grouping.
Strips variable parts (paths, line numbers, addresses, timestamps)
to group similar errors together.
"""
if not error_message and not error_type:
return "unknown_error"
# Start with error type
sig_parts = []
if error_type:
sig_parts.append(error_type)
if error_message:
msg = error_message.strip()
# Apply all normalization patterns
for pattern, replacement in _STRIP_PATTERNS:
msg = re.sub(pattern, replacement, msg)
# Take first 100 chars of normalized message
msg = msg[:100].strip()
if msg:
sig_parts.append(msg)
return ": ".join(sig_parts) if sig_parts else "unknown_error"
# =============================================================================
# Report Discovery
# =============================================================================
def find_latest_report_dir(suite: str, base_dir: Optional[Path] = None) -> Optional[Path]:
"""
Find the latest report directory for a suite.
Args:
suite: "examples" or "docs"
base_dir: Base reports directory (default: ~/Downloads/reports/<suite>)
Returns:
Path to latest report directory, or None if not found.
"""
if base_dir is None:
base_dir = Path.home() / "Downloads" / "reports" / suite
if not base_dir.exists():
return None
# Check if base_dir itself contains report.json (user passed specific run dir)
if (base_dir / "report.json").exists():
return base_dir
# Find subdirectories with report.json
candidates = []
for subdir in base_dir.iterdir():
if subdir.is_dir():
# Check for report.json in subdir or in group subdirs
if (subdir / "report.json").exists():
candidates.append(subdir)
else:
# Check if this is a run dir with group subdirs
for group_dir in subdir.iterdir():
if group_dir.is_dir() and (group_dir / "report.json").exists():
candidates.append(subdir)
break
if not candidates:
return None
# Sort by mtime (newest first)
candidates.sort(key=lambda p: p.stat().st_mtime, reverse=True)
return candidates[0]
def find_report_files(report_dir: Path) -> List[Path]:
"""
Find all report.json files in a report directory.
Handles both single-group and multi-group report structures.
"""
report_files = []
# Check for report.json directly in report_dir
direct = report_dir / "report.json"
if direct.exists():
report_files.append(direct)
# Check for group subdirectories
for subdir in report_dir.iterdir():
if subdir.is_dir():
group_report = subdir / "report.json"
if group_report.exists():
report_files.append(group_report)
return report_files
# =============================================================================
# Report Loading
# =============================================================================
def load_report(report_dir: Path) -> LoadedReport:
"""
Load and normalize a report from a directory.
Aggregates all group reports if multiple exist.
Args:
report_dir: Path to report run directory.
Returns:
LoadedReport with normalized items.
Raises:
FileNotFoundError: If no report.json found.
ValueError: If report cannot be parsed.
"""
report_files = find_report_files(report_dir)
if not report_files:
raise FileNotFoundError(f"No report.json found in {report_dir}")
all_items: List[ReportItem] = []
metadata = ReportMetadata()
groups_seen = set()
for report_file in report_files:
try:
data = json.loads(report_file.read_text(encoding='utf-8'))
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON in {report_file}: {e}")
# Parse metadata from first report
if not metadata.suite and "metadata" in data:
meta = data["metadata"]
metadata = ReportMetadata(
suite=meta.get("suite", ""),
source_path=meta.get("source_path", ""),
timestamp=meta.get("timestamp", ""),
platform=meta.get("platform", ""),
python_version=meta.get("python_version", ""),
python_executable=meta.get("python_executable", ""),
praisonai_version=meta.get("praisonai_version", ""),
git_commit=meta.get("git_commit"),
groups_run=meta.get("groups_run", []),
totals=meta.get("totals", {}),
)
# Parse results
for r in data.get("results", []):
error_sig = normalize_error_signature(
r.get("error_type"),
r.get("error_message")
)
item = ReportItem(
item_id=r.get("item_id", ""),
suite=r.get("suite", ""),
group=r.get("group", ""),
source_path=r.get("source_path", ""),
block_index=r.get("block_index", 0),
status=r.get("status", "not_run"),
duration_seconds=r.get("duration_seconds", 0.0),
exit_code=r.get("exit_code", 0),
error_type=r.get("error_type"),
error_message=r.get("error_message"),
error_signature=error_sig,
skip_reason=r.get("skip_reason"),
stdout_path=r.get("stdout_path"),
stderr_path=r.get("stderr_path"),
start_time=r.get("start_time", ""),
end_time=r.get("end_time", ""),
runnable_decision=r.get("runnable_decision", ""),
)
all_items.append(item)
groups_seen.add(item.group)
# Update metadata with aggregated groups
if groups_seen:
metadata.groups_run = sorted(groups_seen)
# Recalculate totals from all items
totals = {"passed": 0, "failed": 0, "skipped": 0, "timeout": 0, "not_run": 0, "xfail": 0, "total": 0}
for item in all_items:
totals["total"] += 1
if item.status in totals:
totals[item.status] += 1
metadata.totals = totals
return LoadedReport(
metadata=metadata,
items=all_items,
report_dir=report_dir,
)
# =============================================================================
# Error Grouping
# =============================================================================
def group_errors(items: List[ReportItem]) -> List[ErrorGroup]:
"""
Group items by error signature.
Returns groups sorted by count (descending), then signature.
"""
groups: Dict[str, ErrorGroup] = {}
for item in items:
if item.status not in ("failed", "timeout", "error"):
continue
sig = item.error_signature or "unknown_error"
if sig not in groups:
groups[sig] = ErrorGroup(
signature=sig,
error_type=item.error_type or "",
sample_message=item.error_message or "",
)
groups[sig].count += 1
groups[sig].items.append(item)
# Sort by count desc, then signature asc
sorted_groups = sorted(
groups.values(),
key=lambda g: (-g.count, g.signature)
)
return sorted_groups
# =============================================================================
# ASCII Table Rendering
# =============================================================================
def _truncate(text: str, max_len: int, ellipsis: str = "...") -> str:
"""Truncate text with ellipsis if too long."""
if not text:
return ""
text = text.replace("\n", " ").replace("\r", "")
if len(text) <= max_len:
return text
return text[:max_len - len(ellipsis)] + ellipsis
def _render_table(
headers: List[str],
rows: List[List[str]],
col_widths: Optional[List[int]] = None,
max_width: int = 120,
) -> str:
"""
Render a simple ASCII table.
Args:
headers: Column headers.
rows: List of row data (list of strings).
col_widths: Optional fixed column widths.
max_width: Maximum total table width.
Returns:
Formatted table string.
"""
if not headers:
return ""
num_cols = len(headers)
# Calculate column widths
if col_widths is None:
col_widths = [len(h) for h in headers]
for row in rows:
for i, cell in enumerate(row[:num_cols]):
col_widths[i] = max(col_widths[i], len(str(cell)))
# Ensure we don't exceed max_width
total_width = sum(col_widths) + (num_cols - 1) * 3 + 4 # separators + borders
if total_width > max_width and col_widths:
# Shrink largest columns proportionally
excess = total_width - max_width
while excess > 0 and max(col_widths) > 10:
max_idx = col_widths.index(max(col_widths))
col_widths[max_idx] -= 1
excess -= 1
lines = []
# Header separator
sep = "+" + "+".join("-" * (w + 2) for w in col_widths) + "+"
lines.append(sep)
# Header row
header_cells = [
f" {_truncate(h, w).ljust(w)} " for h, w in zip(headers, col_widths)
]
lines.append("|" + "|".join(header_cells) + "|")
lines.append(sep)
# Data rows
for row in rows:
cells = []
for i, w in enumerate(col_widths):
cell = str(row[i]) if i < len(row) else ""
cells.append(f" {_truncate(cell, w).ljust(w)} ")
lines.append("|" + "|".join(cells) + "|")
lines.append(sep)
return "\n".join(lines)
# =============================================================================
# Report Rendering
# =============================================================================
STATUS_ICONS = {
"passed": "✅",
"failed": "❌",
"skipped": "⏭️",
"timeout": "⏱️",
"not_run": "📝",
"xfail": "⚠️",
"error": "💥",
}
def render_overview(report: LoadedReport, wide: bool = False) -> str:
"""Render overview section with totals and metadata."""
meta = report.metadata
totals = meta.totals
lines = [
"",
"=" * 60,
f" REPORT: {meta.suite.upper()} EXECUTION",
"=" * 60,
"",
f" Timestamp: {meta.timestamp}",
f" Platform: {meta.platform}",
f" Python: {meta.python_version}",
f" PraisonAI: {meta.praisonai_version}",
]
if meta.git_commit:
lines.append(f" Git Commit: {meta.git_commit}")
if meta.groups_run:
groups_str = ", ".join(meta.groups_run[:10])
if len(meta.groups_run) > 10:
groups_str += f" (+{len(meta.groups_run) - 10} more)"
lines.append(f" Groups: {groups_str}")
lines.append(f" Report Dir: {report.report_dir}")
lines.append("")
# Summary table
lines.append(" SUMMARY")
lines.append(" " + "-" * 40)
total = totals.get("total", 0)
for status in ["passed", "failed", "timeout", "skipped", "not_run", "xfail"]:
count = totals.get(status, 0)
icon = STATUS_ICONS.get(status, "?")
pct = f"({count * 100 / total:.1f}%)" if total > 0 else ""
lines.append(f" {icon} {status.capitalize():12} {count:5} {pct}")
lines.append(" " + "-" * 40)
lines.append(f" {'Total':15} {total:5}")
lines.append("")
return "\n".join(lines)
def render_failures_table(
report: LoadedReport,
limit: int = 50,
match_statuses: Optional[List[str]] = None,
match_groups: Optional[List[str]] = None,
contains: Optional[str] = None,
wide: bool = False,
) -> str:
"""
Render failures table.
Args:
report: Loaded report.
limit: Max rows to show (0 = no limit).
match_statuses: Filter by status(es).
match_groups: Filter by group(s).
contains: Filter by error message substring.
wide: Show full error messages.
Returns:
Formatted table string.
"""
# Default to failure statuses
if match_statuses is None:
match_statuses = ["failed", "timeout", "error"]
# Filter items
items = []
for item in report.items:
if item.status not in match_statuses:
continue
if match_groups and item.group not in match_groups:
continue
if contains:
msg = (item.error_message or "").lower()
if contains.lower() not in msg:
continue
items.append(item)
if not items:
return "\n No matching failures found.\n"
# Apply limit
total_count = len(items)
if limit > 0 and len(items) > limit:
items = items[:limit]
lines = [
"",
f" FAILURES ({total_count} total, showing {len(items)})",
" " + "-" * 50,
"",
]
# Build table
headers = ["Group", "Item", "Status", "Duration", "Error"]
rows = []
error_width = 60 if wide else 30
for item in items:
icon = STATUS_ICONS.get(item.status, "?")
duration = f"{item.duration_seconds:.2f}s" if item.duration_seconds else "-"
error = item.error_message or item.skip_reason or "-"
rows.append([
item.group,
item.display_name,
f"{icon} {item.status}",
duration,
_truncate(error, error_width) if not wide else error[:200],
])
col_widths = [15, 30, 12, 10, error_width]
lines.append(_render_table(headers, rows, col_widths if not wide else None))
if limit > 0 and total_count > limit:
lines.append(f"\n ... and {total_count - limit} more (use --limit 0 to show all)")
lines.append("")
return "\n".join(lines)
def render_error_groups(
report: LoadedReport,
limit: int = 20,
wide: bool = False,
show_paths: bool = True,
) -> str:
"""
Render error grouping section.
Args:
report: Loaded report.
limit: Max groups to show (0 = no limit).
wide: Show full error messages and all paths.
show_paths: Show affected item paths.
Returns:
Formatted error groups string.
"""
groups = group_errors(report.items)
if not groups:
return "\n No errors to group.\n"
total_groups = len(groups)
if limit > 0 and len(groups) > limit:
groups = groups[:limit]
lines = [
"",
f" ERROR GROUPS ({total_groups} unique errors)",
" " + "=" * 50,
"",
]
for i, grp in enumerate(groups, 1):
lines.append(f" [{i}] {grp.error_type or 'Error'} ({grp.count} occurrences)")
lines.append(" " + "-" * 40)
# Show signature/sample message
if wide:
msg = grp.sample_message or grp.signature
lines.append(f" {msg[:500]}")
else:
lines.append(f" {_truncate(grp.signature, 80)}")
# Show affected items
if show_paths:
lines.append("")
lines.append(" Affected items:")
items_to_show = grp.items if wide else grp.items[:5]
for item in items_to_show:
lines.append(f" - {item.group}/{item.display_name}")
if not wide and len(grp.items) > 5:
lines.append(f" ... and {len(grp.items) - 5} more")
lines.append("")
if limit > 0 and total_groups > limit:
lines.append(f" ... and {total_groups - limit} more error groups (use --limit 0 to show all)")
lines.append("")
return "\n".join(lines)
def render_artifacts(report: LoadedReport) -> str:
"""Render list of report artifacts."""
lines = [
"",
" REPORT ARTIFACTS",
" " + "-" * 40,
f" Directory: {report.report_dir}",
"",
]
# List files
for f in sorted(report.report_dir.rglob("*")):
if f.is_file():
rel = f.relative_to(report.report_dir)
size = f.stat().st_size
size_str = f"{size:,} bytes" if size < 1024 else f"{size / 1024:.1f} KB"
lines.append(f" {rel}: {size_str}")
lines.append("")
return "\n".join(lines)
# =============================================================================
# Main View Function
# =============================================================================
def view_report(
report_dir: Optional[Path] = None,
suite: str = "examples",
base_dir: Optional[Path] = None,
limit: int = 50,
show_errors: bool = True,
wide: bool = False,
show_paths: bool = True,
match_statuses: Optional[List[str]] = None,
match_groups: Optional[List[str]] = None,
contains: Optional[str] = None,
show_artifacts: bool = False,
output_format: str = "table",
) -> Tuple[str, int]:
"""
Main entry point for viewing a report.
Args:
report_dir: Explicit report directory (or None for auto-detect).
suite: Suite name for auto-detection.
base_dir: Base reports directory for auto-detection.
limit: Max rows in tables.
show_errors: Show error grouping section.
wide: Wide output mode.
show_paths: Show paths in error groups.
match_statuses: Filter by status(es).
match_groups: Filter by group(s).
contains: Filter by error message substring.
show_artifacts: Show artifact listing.
output_format: "table" or "json".
Returns:
Tuple of (output_string, exit_code).
Exit code is 0 if no failures, 1 if failures exist.
"""
# Find report directory
if report_dir is None:
report_dir = find_latest_report_dir(suite, base_dir)
if report_dir is None:
return f"Error: No reports found for suite '{suite}'", 1
if not report_dir.exists():
return f"Error: Report directory not found: {report_dir}", 1
# Load report
try:
report = load_report(report_dir)
except FileNotFoundError as e:
return f"Error: {e}", 1
except ValueError as e:
return f"Error: {e}", 1
# JSON output
if output_format == "json":
output = {
"metadata": {
"suite": report.metadata.suite,
"timestamp": report.metadata.timestamp,
"totals": report.metadata.totals,
"groups_run": report.metadata.groups_run,
},
"failures": [
{
"item_id": i.item_id,
"group": i.group,
"display_name": i.display_name,
"status": i.status,
"error_type": i.error_type,
"error_message": i.error_message,
}
for i in report.failures
],
"error_groups": [
{
"signature": g.signature,
"error_type": g.error_type,
"count": g.count,
"items": [i.display_name for i in g.items],
}
for g in group_errors(report.items)
],
}
return json.dumps(output, indent=2), 0 if not report.failures else 1
# Table output
parts = []
# Overview
parts.append(render_overview(report, wide=wide))
# Failures table
parts.append(render_failures_table(
report,
limit=limit,
match_statuses=match_statuses,
match_groups=match_groups,
contains=contains,
wide=wide,
))
# Error groups
if show_errors:
parts.append(render_error_groups(
report,
limit=limit,
wide=wide,
show_paths=show_paths,
))
# Artifacts
if show_artifacts:
parts.append(render_artifacts(report))
output = "\n".join(parts)
exit_code = 0 if not report.failures else 1
return output, exit_code
+1 -1
View File
@@ -1 +1 @@
__version__ = "3.5.9"
__version__ = "3.6.2"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.10.7",
"praisonaiagents>=0.10.10",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
+18
View File
@@ -130,6 +130,19 @@ def bump_version(new_version: str, agents_version: Optional[str] = None):
print("\n✨ Version bump complete!")
def validate_dependencies() -> bool:
"""Validate that uv lock will succeed before making any changes."""
praisonai_dir = get_praisonai_dir()
print("\n🔍 Validating dependencies (dry-run)...")
result = run(["uv", "lock", "--dry-run"], cwd=praisonai_dir, check=False)
if result.returncode != 0:
print("\n❌ Dependency validation failed. Fix the issues above before releasing.")
return False
print(" ✅ Dependencies validated successfully")
return True
def release(version: str):
"""Run the release process."""
root = get_project_root()
@@ -215,6 +228,11 @@ def main():
# Run bump version
bump_version(args.version, args.agents)
# Validate dependencies after version bump (before git operations)
if not validate_dependencies():
print("\n💡 Tip: Revert changes with 'git checkout .' if needed")
sys.exit(1)
# Run release
release(args.version)
@@ -0,0 +1,557 @@
"""
Unit tests for report_viewer module.
Tests cover:
- Latest report directory detection
- JSON parsing and loading
- Error signature normalization
- Error grouping
- Table rendering
- CLI integration smoke test
"""
import json
import pytest
from praisonai.suite_runner.report_viewer import (
normalize_error_signature,
find_latest_report_dir,
find_report_files,
load_report,
group_errors,
render_overview,
render_failures_table,
render_error_groups,
view_report,
_truncate,
_render_table,
)
# =============================================================================
# Fixtures
# =============================================================================
@pytest.fixture
def sample_report_json():
"""Sample report JSON structure."""
return {
"metadata": {
"suite": "examples",
"source_path": "/path/to/examples",
"timestamp": "2026-01-09T14:26:07.808316+00:00",
"platform": "Darwin-25.3.0-arm64",
"python_version": "3.12.11",
"python_executable": "/usr/bin/python",
"praisonai_version": "3.5.8",
"git_commit": "abc1234",
"cli_args": [],
"groups_run": ["acp", "tools"],
"totals": {
"passed": 5,
"failed": 2,
"skipped": 1,
"timeout": 1,
"not_run": 0,
"xfail": 0,
"total": 9,
},
},
"results": [
{
"item_id": "acp__test1",
"suite": "examples",
"group": "acp",
"source_path": "/path/to/examples/acp/test1.py",
"block_index": 0,
"language": "python",
"status": "passed",
"duration_seconds": 1.5,
"exit_code": 0,
},
{
"item_id": "acp__test2",
"suite": "examples",
"group": "acp",
"source_path": "/path/to/examples/acp/test2.py",
"block_index": 0,
"language": "python",
"status": "failed",
"duration_seconds": 2.0,
"exit_code": 1,
"error_type": "ImportError",
"error_message": "No module named 'missing_module'",
},
{
"item_id": "tools__test3",
"suite": "examples",
"group": "tools",
"source_path": "/path/to/examples/tools/test3.py",
"block_index": 0,
"language": "python",
"status": "failed",
"duration_seconds": 0.5,
"exit_code": 1,
"error_type": "ImportError",
"error_message": "No module named 'missing_module'",
},
{
"item_id": "tools__test4",
"suite": "examples",
"group": "tools",
"source_path": "/path/to/examples/tools/test4.py",
"block_index": 0,
"language": "python",
"status": "timeout",
"duration_seconds": 60.0,
"exit_code": -1,
"error_type": "TimeoutError",
"error_message": "Execution timed out after 60 seconds",
},
],
}
@pytest.fixture
def report_dir(tmp_path, sample_report_json):
"""Create a temporary report directory with sample data."""
report_path = tmp_path / "report.json"
report_path.write_text(json.dumps(sample_report_json), encoding='utf-8')
return tmp_path
@pytest.fixture
def multi_group_report_dir(tmp_path, sample_report_json):
"""Create a report directory with multiple group subdirs."""
# Create group subdirectories
acp_dir = tmp_path / "acp"
acp_dir.mkdir()
tools_dir = tmp_path / "tools"
tools_dir.mkdir()
# Write separate reports for each group
acp_report = {
"metadata": sample_report_json["metadata"].copy(),
"results": [r for r in sample_report_json["results"] if r["group"] == "acp"],
}
acp_report["metadata"]["groups_run"] = ["acp"]
(acp_dir / "report.json").write_text(json.dumps(acp_report), encoding='utf-8')
tools_report = {
"metadata": sample_report_json["metadata"].copy(),
"results": [r for r in sample_report_json["results"] if r["group"] == "tools"],
}
tools_report["metadata"]["groups_run"] = ["tools"]
(tools_dir / "report.json").write_text(json.dumps(tools_report), encoding='utf-8')
return tmp_path
# =============================================================================
# Error Signature Normalization Tests
# =============================================================================
class TestNormalizeErrorSignature:
"""Tests for error signature normalization."""
def test_basic_error(self):
"""Test basic error type and message."""
sig = normalize_error_signature("ImportError", "No module named 'foo'")
assert "ImportError" in sig
assert "No module named" in sig
def test_strips_file_paths(self):
"""Test that file paths are normalized."""
msg = 'File "/Users/praison/project/test.py", line 42, in foo'
sig = normalize_error_signature("SyntaxError", msg)
assert "/Users/praison" not in sig
assert "...py" in sig or "line N" in sig
def test_strips_line_numbers(self):
"""Test that line numbers are normalized."""
msg = "Error at line 123 in function"
sig = normalize_error_signature("Error", msg)
assert "123" not in sig
assert "line N" in sig
def test_strips_memory_addresses(self):
"""Test that memory addresses are normalized."""
msg = "Object at 0x7f8b1c2d3e4f crashed"
sig = normalize_error_signature("RuntimeError", msg)
assert "0x7f8b1c2d3e4f" not in sig
assert "0x..." in sig
def test_strips_uuids(self):
"""Test that UUIDs are normalized."""
msg = "Request 550e8400-e29b-41d4-a716-446655440000 failed"
sig = normalize_error_signature("Error", msg)
assert "550e8400" not in sig
assert "UUID" in sig
def test_strips_timestamps(self):
"""Test that timestamps are normalized."""
msg = "Error at 2026-01-09T14:30:00Z"
sig = normalize_error_signature("Error", msg)
assert "2026-01-09" not in sig
assert "TIMESTAMP" in sig
def test_empty_message(self):
"""Test with empty message."""
sig = normalize_error_signature("Error", "")
assert sig == "Error"
def test_none_values(self):
"""Test with None values."""
sig = normalize_error_signature(None, None)
assert sig == "unknown_error"
def test_truncates_long_messages(self):
"""Test that long messages are truncated."""
long_msg = "x" * 200
sig = normalize_error_signature("Error", long_msg)
assert len(sig) <= 110 # Error: + 100 chars
# =============================================================================
# Report Discovery Tests
# =============================================================================
class TestFindLatestReportDir:
"""Tests for latest report directory detection."""
def test_finds_direct_report(self, report_dir):
"""Test finding report.json directly in directory."""
result = find_latest_report_dir("examples", report_dir)
assert result == report_dir
def test_finds_latest_by_mtime(self, tmp_path):
"""Test finding latest directory by modification time."""
# Create older directory
old_dir = tmp_path / "20260101_120000"
old_dir.mkdir()
(old_dir / "report.json").write_text('{"metadata": {}, "results": []}')
# Create newer directory
new_dir = tmp_path / "20260109_140000"
new_dir.mkdir()
(new_dir / "report.json").write_text('{"metadata": {}, "results": []}')
result = find_latest_report_dir("examples", tmp_path)
assert result == new_dir
def test_returns_none_for_empty_dir(self, tmp_path):
"""Test returns None when no reports found."""
result = find_latest_report_dir("examples", tmp_path)
assert result is None
def test_returns_none_for_nonexistent_dir(self, tmp_path):
"""Test returns None for nonexistent directory."""
result = find_latest_report_dir("examples", tmp_path / "nonexistent")
assert result is None
class TestFindReportFiles:
"""Tests for finding report files in a directory."""
def test_finds_direct_report(self, report_dir):
"""Test finding report.json directly."""
files = find_report_files(report_dir)
assert len(files) == 1
assert files[0].name == "report.json"
def test_finds_group_reports(self, multi_group_report_dir):
"""Test finding reports in group subdirectories."""
files = find_report_files(multi_group_report_dir)
assert len(files) == 2
names = {f.parent.name for f in files}
assert names == {"acp", "tools"}
# =============================================================================
# Report Loading Tests
# =============================================================================
class TestLoadReport:
"""Tests for loading and parsing reports."""
def test_loads_single_report(self, report_dir):
"""Test loading a single report.json."""
report = load_report(report_dir)
assert report.metadata.suite == "examples"
assert len(report.items) == 4
def test_loads_multi_group_report(self, multi_group_report_dir):
"""Test loading and aggregating multiple group reports."""
report = load_report(multi_group_report_dir)
assert len(report.items) == 4
assert set(report.metadata.groups_run) == {"acp", "tools"}
def test_calculates_totals(self, report_dir):
"""Test that totals are calculated correctly."""
report = load_report(report_dir)
totals = report.metadata.totals
assert totals["passed"] == 1
assert totals["failed"] == 2
assert totals["timeout"] == 1
assert totals["total"] == 4
def test_raises_for_missing_report(self, tmp_path):
"""Test raises FileNotFoundError for missing report."""
with pytest.raises(FileNotFoundError):
load_report(tmp_path)
def test_raises_for_invalid_json(self, tmp_path):
"""Test raises ValueError for invalid JSON."""
(tmp_path / "report.json").write_text("not valid json")
with pytest.raises(ValueError):
load_report(tmp_path)
def test_normalizes_error_signatures(self, report_dir):
"""Test that error signatures are normalized during load."""
report = load_report(report_dir)
failed_items = [i for i in report.items if i.status == "failed"]
assert all(i.error_signature for i in failed_items)
# =============================================================================
# Error Grouping Tests
# =============================================================================
class TestGroupErrors:
"""Tests for error grouping functionality."""
def test_groups_similar_errors(self, report_dir):
"""Test that similar errors are grouped together."""
report = load_report(report_dir)
groups = group_errors(report.items)
# Should have 2 groups: ImportError and TimeoutError
assert len(groups) == 2
def test_counts_occurrences(self, report_dir):
"""Test that occurrence counts are correct."""
report = load_report(report_dir)
groups = group_errors(report.items)
# Find ImportError group (should have 2 occurrences)
import_group = next((g for g in groups if "ImportError" in g.error_type), None)
assert import_group is not None
assert import_group.count == 2
def test_sorts_by_count_desc(self, report_dir):
"""Test that groups are sorted by count descending."""
report = load_report(report_dir)
groups = group_errors(report.items)
counts = [g.count for g in groups]
assert counts == sorted(counts, reverse=True)
def test_ignores_passed_items(self, report_dir):
"""Test that passed items are not grouped."""
report = load_report(report_dir)
groups = group_errors(report.items)
all_items = [i for g in groups for i in g.items]
assert all(i.status in ("failed", "timeout", "error") for i in all_items)
# =============================================================================
# Rendering Tests
# =============================================================================
class TestTruncate:
"""Tests for text truncation."""
def test_short_text_unchanged(self):
"""Test that short text is not truncated."""
assert _truncate("hello", 10) == "hello"
def test_long_text_truncated(self):
"""Test that long text is truncated with ellipsis."""
result = _truncate("hello world", 8)
assert len(result) == 8
assert result.endswith("...")
def test_empty_text(self):
"""Test empty text handling."""
assert _truncate("", 10) == ""
assert _truncate(None, 10) == ""
def test_newlines_removed(self):
"""Test that newlines are removed."""
assert "\n" not in _truncate("hello\nworld", 20)
class TestRenderTable:
"""Tests for ASCII table rendering."""
def test_renders_headers(self):
"""Test that headers are rendered."""
table = _render_table(["A", "B"], [["1", "2"]])
assert "A" in table
assert "B" in table
def test_renders_data(self):
"""Test that data rows are rendered."""
table = _render_table(["Col"], [["value1"], ["value2"]])
assert "value1" in table
assert "value2" in table
def test_empty_table(self):
"""Test empty table handling."""
table = _render_table([], [])
assert table == ""
class TestRenderOverview:
"""Tests for overview rendering."""
def test_includes_metadata(self, report_dir):
"""Test that overview includes metadata."""
report = load_report(report_dir)
output = render_overview(report)
assert "EXAMPLES" in output.upper()
assert "Python" in output or "python" in output
def test_includes_totals(self, report_dir):
"""Test that overview includes totals."""
report = load_report(report_dir)
output = render_overview(report)
assert "Passed" in output or "passed" in output
assert "Failed" in output or "failed" in output
class TestRenderFailuresTable:
"""Tests for failures table rendering."""
def test_shows_failures(self, report_dir):
"""Test that failures are shown."""
report = load_report(report_dir)
output = render_failures_table(report)
assert "test2.py" in output or "test3.py" in output
def test_respects_limit(self, report_dir):
"""Test that limit is respected."""
report = load_report(report_dir)
output = render_failures_table(report, limit=1)
# Should mention there are more
assert "more" in output.lower() or "1" in output
def test_filters_by_status(self, report_dir):
"""Test filtering by status."""
report = load_report(report_dir)
output = render_failures_table(report, match_statuses=["timeout"])
assert "timeout" in output.lower()
class TestRenderErrorGroups:
"""Tests for error groups rendering."""
def test_shows_groups(self, report_dir):
"""Test that error groups are shown."""
report = load_report(report_dir)
output = render_error_groups(report)
assert "ERROR GROUPS" in output
assert "ImportError" in output or "TimeoutError" in output
def test_shows_counts(self, report_dir):
"""Test that occurrence counts are shown."""
report = load_report(report_dir)
output = render_error_groups(report)
assert "occurrence" in output.lower()
# =============================================================================
# Integration Tests
# =============================================================================
class TestViewReport:
"""Integration tests for view_report function."""
def test_table_output(self, report_dir):
"""Test table output format."""
output, exit_code = view_report(
report_dir=report_dir,
suite="examples",
output_format="table",
)
assert "EXAMPLES" in output.upper()
assert exit_code == 1 # Has failures
def test_json_output(self, report_dir):
"""Test JSON output format."""
output, exit_code = view_report(
report_dir=report_dir,
suite="examples",
output_format="json",
)
data = json.loads(output)
assert "metadata" in data
assert "failures" in data
assert "error_groups" in data
def test_exit_code_zero_for_no_failures(self, tmp_path):
"""Test exit code is 0 when no failures."""
report = {
"metadata": {"suite": "examples", "totals": {"passed": 1, "total": 1}},
"results": [
{
"item_id": "test1",
"suite": "examples",
"group": "test",
"source_path": "/test.py",
"status": "passed",
}
],
}
(tmp_path / "report.json").write_text(json.dumps(report))
output, exit_code = view_report(report_dir=tmp_path, suite="examples")
assert exit_code == 0
def test_error_for_missing_report(self, tmp_path):
"""Test error message for missing report."""
output, exit_code = view_report(
report_dir=tmp_path / "nonexistent",
suite="examples",
)
assert "Error" in output
assert exit_code == 1
def test_auto_detect_latest(self, tmp_path):
"""Test auto-detection of latest report."""
# Create a report directory
run_dir = tmp_path / "20260109_140000"
run_dir.mkdir()
report = {
"metadata": {"suite": "examples", "totals": {"passed": 1, "total": 1}},
"results": [
{
"item_id": "test1",
"suite": "examples",
"group": "test",
"source_path": "/test.py",
"status": "passed",
}
],
}
(run_dir / "report.json").write_text(json.dumps(report))
output, exit_code = view_report(
report_dir=None,
suite="examples",
base_dir=tmp_path,
)
assert "EXAMPLES" in output.upper()
assert exit_code == 0
+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.10.7" },
{ name = "praisonaiagents", specifier = ">=0.10.10" },
{ 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.10.7"
version = "0.10.10"
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/44/be/6ad0f124e0e0e8c18c0dc3e10b097fff56fd62a3df3f13f2c880a207f0b7/praisonaiagents-0.10.7.tar.gz", hash = "sha256:ce8af091043c20acbc0a8eb4879f01eb8dc1cbe6037f52a6b20926a29594d1c6", size = 960169, upload-time = "2026-01-09T14:56:47.042Z" }
sdist = { url = "https://files.pythonhosted.org/packages/56/4f/d01c588827a74ddc58a8bf30cb7eb97671d210b7f5ef46f9eb2a7249f919/praisonaiagents-0.10.10.tar.gz", hash = "sha256:b6cfea93fef40b898575d0c1574134fdc9ec49754d2287e9ade90ba4381ca3d7", size = 962453, upload-time = "2026-01-10T08:57:35.82Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/6c/26/c4145c31884e607c58232ed5a2dd685a8a629fc9202112ec664265736be5/praisonaiagents-0.10.7-py3-none-any.whl", hash = "sha256:3c12b2c69f74c7a6bcb5753998982a5b0e4cace6f675dd1fd95f90ca897f9d3f", size = 939197, upload-time = "2026-01-09T14:56:45.254Z" },
{ url = "https://files.pythonhosted.org/packages/07/fe/edd179c792d7f6143d3278c01daf37015fe4a2d98c5dc326fb309b263336/praisonaiagents-0.10.10-py3-none-any.whl", hash = "sha256:fa9bc6db1fb12ae25ea491db8e30eb98ac16c1e9798952cf617594c09999b826", size = 941797, upload-time = "2026-01-10T08:57:34.07Z" },
]
[[package]]