Compare commits

...
14 Commits
Author SHA1 Message Date
Mervin Praison f991f37cfe Release v2.3.23 2025-12-17 00:41:48 +00:00
Mervin Praison a43b3ac72c Release v2.3.22 2025-12-17 00:25:14 +00:00
Mervin Praison 06f82249e1 Release v2.3.21 2025-12-17 00:05:40 +00:00
Mervin Praison d3aa78241a Release v2.3.20 2025-12-16 23:46:47 +00:00
Mervin Praison dfb77c39f9 Release v2.3.19 2025-12-16 23:26:53 +00:00
Mervin Praison 7e8f9b0574 Release v2.3.18 2025-12-16 23:18:36 +00:00
Mervin Praison 19bcdcd000 Release v2.3.16 2025-12-16 22:59:29 +00:00
Mervin Praison 8f5160f827 feat(workflows): Add branching, loops, callbacks, and per-step variables
- Add branching support with next_steps and branch_condition fields
- Add loop support with loop_over and loop_var for iterating over variables
- Add on_task_start and on_task_complete callbacks to PraisonAIAgents
- Add variables parameter for global variable substitution across tasks
- Add per-step agent creation from agent_config when no agent assigned
- Add task-level variable substitution in task descriptions
- Refactor workflow
2025-12-16 22:58:56 +00:00
Mervin Praison 3574a4d2c0 Release v2.3.15 2025-12-16 22:50:37 +00:00
Mervin Praison b552843fa9 docs: Update README workflow CLI with new features
- Add inline workflow examples (--workflow 'step1,step2')
- Add planning mode example (--planning)
- Add tools and save examples (--tools, --save)
- Update to use global flags instead of workflow-specific flags
2025-12-16 21:46:30 +00:00
Mervin Praison 0461a133df fix(workflow): Actually use planning mode when --planning is enabled
Before: --planning flag was stored but never used, agent.chat() was called
After: --planning flag creates agent with planning=True and calls agent.start()

The agent.start() method checks if planning is enabled and calls
_start_with_planning() which:
1. Creates a plan using PlanningAgent
2. Displays the generated plan
3. Executes each step with context passing
4. Compiles results into final output
2025-12-16 19:12:28 +00:00
Mervin Praison b33857de33 refactor(cli): Use global flags for workflow instead of workflow-specific flags
BREAKING CHANGE: Removed workflow-specific flags in favor of global flags

Before:
  praisonai workflow run 'name' --workflow-tools tavily --workflow-save --workflow-verbose

After:
  praisonai workflow run 'name' --tools tavily --save --verbose

This simplifies the CLI by reusing existing global flags:
- --llm (instead of --workflow-llm)
- --tools (instead of --workflow-tools)
- --planning (instead of --workflow-planning)
- --memory (instead of --workflow-memory)
- --save (instead of --workflow-save)
- --verbose (instead of --workflow-verbose)

Only --workflow-var remains workflow-specific for variable substitution.
2025-12-16 19:09:52 +00:00
Mervin Praison 433c643f6c feat(cli): Add inline workflow support without .md template
Usage: praisonai 'prompt' --workflow 'step1:action1,step2:action2'

Example:
  praisonai 'What is AI?' --workflow 'Research:Research the topic,Summarize:Write a summary' --workflow-save

Features:
- Parse inline workflow steps from --workflow flag
- Sequential execution with context passing
- All workflow options work: --workflow-tools, --workflow-save, etc.
- No .md template file required
2025-12-16 19:05:20 +00:00
Mervin Praison 214119a063 feat(cli): Add complete workflow CLI options
- Add --workflow-llm for LLM model selection
- Add --workflow-tools for tools (comma-separated)
- Add --workflow-planning for planning mode
- Add --workflow-memory for memory support
- Add --workflow-stream for streaming output
- Add --workflow-verbose for verbose output
- Add --workflow-save to save output to file
- Show final output after workflow completion
- Save all step outputs to markdown file
- Update help text with all options
2025-12-16 19:01:57 +00:00
44 changed files with 9079 additions and 287 deletions

No files matched your search

+285 -11
View File
@@ -97,6 +97,26 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
| 🪝 Hooks | [Example](#9-hooks) | [📖](https://docs.praison.ai/features/hooks) |
| 📈 Telemetry | [Example](examples/python/telemetry/production-telemetry-example.py) | [📖](https://docs.praison.ai/features/telemetry) |
| 📹 Camera Integration | [Example](examples/python/camera/) | [📖](https://docs.praison.ai/features/camera-integration) |
| 📄 Project Docs (.praison/docs/) | [Example](#docs-cli) | [📖](https://docs.praison.ai/cli/docs) |
| 🔌 MCP Config Management | [Example](#mcp-config-cli) | [📖](https://docs.praison.ai/cli/mcp) |
| 💬 AI Commit Messages | [Example](#ai-commit-cli) | [📖](https://docs.praison.ai/cli/commit) |
| @ @Mentions in Prompts | [Example](#mentions-in-prompts) | [📖](https://docs.praison.ai/cli/mentions) |
| 💾 Auto-Save Sessions | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
| 📜 History in Context | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
| ⏸️ Workflow Checkpoints | [Example](examples/python/workflows/workflow_checkpoints.py) | [📖](https://docs.praison.ai/cli/session) |
| 🔀 Workflow Branching | [Example](examples/python/workflows/workflow_branching.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🔁 Workflow Loops | [Example](examples/python/workflows/workflow_loops.py) | [📖](https://docs.praison.ai/features/workflows) |
| 📞 Task Callbacks | [Example](examples/python/workflows/task_callbacks.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🛑 Workflow Early Stop | [Example](examples/python/workflows/workflow_early_stop.py) | [📖](https://docs.praison.ai/features/workflows) |
| ⚡ Simple Workflow | [Example](examples/python/workflows/simple_workflow.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🤖 Workflow with Agents | [Example](examples/python/workflows/workflow_with_agents.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🔄 Conditional Steps | [Example](examples/python/workflows/workflow_conditional.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🎭 Mixed Steps | [Example](examples/python/workflows/workflow_mixed_steps.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🔀 Agentic Routing | [Example](examples/python/workflows/workflow_routing.py) | [📖](https://docs.praison.ai/features/workflow-routing) |
| ⚡ Parallel Execution | [Example](examples/python/workflows/workflow_parallel.py) | [📖](https://docs.praison.ai/features/workflow-parallel) |
| 🔁 Loop over List | [Example](examples/python/workflows/workflow_loop_list.py) | [📖](https://docs.praison.ai/features/workflow-loop) |
| 📄 Loop over CSV | [Example](examples/python/workflows/workflow_loop_csv.py) | [📖](https://docs.praison.ai/features/workflow-loop) |
| 🔄 Evaluator-Optimizer | [Example](examples/python/workflows/workflow_repeat.py) | [📖](https://docs.praison.ai/features/workflow-repeat) |
## Supported Providers
@@ -438,6 +458,127 @@ result = asyncio.run(manager.aexecute("deploy", default_llm="gpt-4o-mini"))
- **Per-Step Agents**: Configure different agents with roles, goals, tools for each step
- **Async Execution**: Use `aexecute()` for async workflows
- **Planning Mode**: Enable at workflow level with `planning=True`
- **Branching**: Use `next_steps` and `branch_condition` for conditional routing
- **Loops**: Use `loop_over` and `loop_var` to iterate over data
### Choosing the Right Workflow System
| Use Case | Recommended |
|----------|-------------|
| Simple function pipelines | `Workflow` class ⭐ |
| Agent-only pipelines | `Workflow` class |
| CSV batch processing | `process="workflow"` |
| Complex task routing | `process="workflow"` |
| Markdown templates | `WorkflowManager` |
| Early stop / conditional | `Workflow` class |
### Simple Workflow (Recommended)
The easiest way to create workflows - just pass functions as steps:
```python
from praisonaiagents import Workflow, WorkflowContext, StepResult
# Define simple handler functions
def validate(ctx: WorkflowContext) -> StepResult:
if not ctx.input:
return StepResult(output="No input", stop_workflow=True)
return StepResult(output=f"Valid: {ctx.input}")
def process(ctx: WorkflowContext) -> StepResult:
return StepResult(output=f"Processed: {ctx.previous_result}")
# Create and run workflow
workflow = Workflow(steps=[validate, process])
result = workflow.start("Hello World", verbose=True)
print(result["output"]) # "Processed: Valid: Hello World"
```
**Key Features:**
- **Just pass functions** - No complex configuration needed
- **Early stop** - Return `stop_workflow=True` to stop the workflow
- **Context passing** - Access `ctx.input`, `ctx.previous_result`, `ctx.variables`
- **Verbose mode** - See step-by-step progress
### Workflow Branching & Loops
```python
from praisonaiagents import WorkflowStep
# Branching step
decision_step = WorkflowStep(
name="decide",
action="Evaluate if task is complete",
next_steps=["success_step", "retry_step"],
branch_condition={"success": ["success_step"], "failure": ["retry_step"]}
)
# Loop step
loop_step = WorkflowStep(
name="process_items",
action="Process {{item}}",
loop_over="items", # Variable containing list
loop_var="item" # Current item variable name
)
```
### Workflow Patterns (route, parallel, loop, repeat)
```python
from praisonaiagents import Workflow, WorkflowContext, StepResult
# Or use Pipeline (alias for Workflow)
from praisonaiagents import Pipeline
from praisonaiagents.workflows import route, parallel, loop, repeat
# 1. ROUTING - Decision-based branching
workflow = Workflow(steps=[
classify_request, # Returns "approve" or "reject"
route({
"approve": [approve_handler],
"reject": [reject_handler],
"default": [fallback_handler]
})
])
# 2. PARALLEL - Concurrent execution
workflow = Workflow(steps=[
parallel([research_market, research_competitors, research_customers]),
summarize_results # Gets all parallel outputs
])
# 3. LOOP - Iterate over list or CSV
workflow = Workflow(
steps=[loop(process_item, over="items")],
variables={"items": ["a", "b", "c"]}
)
# Or from CSV file:
workflow = Workflow(steps=[loop(process_row, from_csv="data.csv")])
# 4. REPEAT - Evaluator-Optimizer pattern
workflow = Workflow(steps=[
repeat(
generator,
until=lambda ctx: "done" in ctx.previous_result,
max_iterations=5
)
])
# 5. CALLBACKS - Monitor workflow execution
workflow = Workflow(
steps=[step1, step2],
on_workflow_start=lambda w, i: print(f"Starting: {i}"),
on_step_complete=lambda name, r: print(f"{name}: {r.output[:50]}"),
on_workflow_complete=lambda w, r: print(f"Done: {r['status']}")
)
# 6. GUARDRAILS - Validate and retry
def validate(result):
return ("error" not in result.output, "Fix the error")
workflow = Workflow(steps=[
WorkflowStep(name="gen", handler=generator, guardrail=validate, max_retries=3)
])
```
### 9. Hooks
@@ -573,19 +714,49 @@ praisonai "Task" --include-rules security,testing
# List available workflows
praisonai workflow list
# Execute a workflow
praisonai workflow run deploy
# Execute a workflow with tools and save output
praisonai workflow run "Research Blog" --tools tavily --save
# Execute with variables
praisonai workflow run deploy --workflow-var environment=staging --workflow-var branch=main
# Execute with planning mode (AI creates sub-steps for each workflow step)
praisonai workflow run "Research Blog" --planning --verbose
# Execute with reasoning mode (chain-of-thought)
praisonai workflow run "Analysis" --reasoning --verbose
# Execute with memory enabled
praisonai workflow run "Research" --memory
# Show workflow details
praisonai workflow show deploy
# Create a new workflow template
praisonai workflow create my_workflow
# Inline workflow (no template file needed)
praisonai "What is AI?" --workflow "Research,Summarize" --save
# Inline workflow with step actions
praisonai "GPT-5" --workflow "Research:Search for info,Write:Write blog" --tools tavily
# Workflow CLI help
praisonai workflow help
```
**Workflow CLI Options:**
| Flag | Description |
|------|-------------|
| `--workflow-var key=value` | Set workflow variable (can be repeated) |
| `--llm <model>` | LLM model (e.g., openai/gpt-4o-mini) |
| `--tools <tools>` | Tools (comma-separated, e.g., tavily) |
| `--planning` | Enable planning mode |
| `--reasoning` | Enable reasoning mode |
| `--memory` | Enable memory |
| `--verbose` | Enable verbose output |
| `--save` | Save output to file |
### Hooks CLI:
```bash
# List configured hooks
@@ -685,23 +856,59 @@ praisonai knowledge help
### Session CLI:
```bash
# Start a new session
praisonai session start my-project
# List all sessions
# List all saved sessions
praisonai session list
# Resume a session
praisonai session resume my-project
# Show session details
praisonai session show my-project
# Resume a session (load into memory)
praisonai session resume my-project
# Delete a session
praisonai session delete my-project
# Show all commands
praisonai session help
# Auto-save session after each run
praisonai "Analyze this code" --auto-save my-project
# Load history from last N sessions into context
praisonai "Continue our discussion" --history 5
```
### Session Management (Python):
```python
from praisonaiagents import Agent
# Auto-save session after each run
agent = Agent(
name="Assistant",
memory=True,
auto_save="my-project"
)
# Load history from last 5 sessions
agent = Agent(
name="Assistant",
memory=True,
history_in_context=5
)
```
### Workflow Checkpoints:
```python
from praisonaiagents.memory.workflows import WorkflowManager
manager = WorkflowManager()
# Save checkpoint after each step
result = manager.execute("deploy", checkpoint="deploy-v1")
# Resume from checkpoint
result = manager.execute("deploy", resume="deploy-v1")
# List/delete checkpoints
manager.list_checkpoints()
manager.delete_checkpoint("deploy-v1")
```
### Tools CLI:
@@ -783,6 +990,73 @@ praisonai agents.yaml --flow-display
praisonai "Multi-step task" --planning --flow-display
```
### Docs CLI:
```bash
# List all project docs
praisonai docs list
# Create a new doc
praisonai docs create project-overview "This project is a Python web app..."
# Show a specific doc
praisonai docs show project-overview
# Delete a doc
praisonai docs delete old-doc
# Show all commands
praisonai docs help
```
### MCP Config CLI:
```bash
# List all MCP configurations
praisonai mcp list
# Create a new MCP config
praisonai mcp create filesystem npx -y @modelcontextprotocol/server-filesystem .
# Show a specific config
praisonai mcp show filesystem
# Enable/disable a config
praisonai mcp enable filesystem
praisonai mcp disable filesystem
# Delete a config
praisonai mcp delete filesystem
# Show all commands
praisonai mcp help
```
### AI Commit CLI:
```bash
# Generate AI commit message for staged changes
praisonai commit
# Generate, commit, and push
praisonai commit --push
```
### @Mentions in Prompts:
```bash
# Include file content in prompt
praisonai "@file:src/main.py explain this code"
# Include project doc
praisonai "@doc:project-overview help me add a feature"
# Search the web
praisonai "@web:python best practices give me tips"
# Fetch URL content
praisonai "@url:https://docs.python.org summarize this"
# Combine multiple mentions
praisonai "@file:main.py @doc:coding-standards review this code"
```
## Prompt Expansion
Expand short prompts into detailed, actionable prompts:
+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 \
flask \
"praisonai>=2.3.14" \
"praisonai>=2.3.23" \
"praisonai[api]" \
gunicorn \
markdown
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.14" \
"praisonai>=2.3.23" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.14" \
"praisonai>=2.3.23" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.14" \
"praisonai>=2.3.23" \
"praisonai[ui]" \
"praisonai[crewai]"
@@ -29,7 +29,7 @@
},
{
"cell_type": "code",
"execution_count": 1,
"execution_count": null,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
@@ -71,7 +71,6 @@
"import pandas as pd\n",
"\n",
"print(\"📦 Dependencies installed!\")\n",
"print(\"🔄 Converting from AGNO → PraisonAI\")\n",
"print(\"🎯 Building Job Hunting Agent System\")"
]
},
@@ -0,0 +1,36 @@
"""
Simple Workflow Example
The easiest way to create and run workflows in PraisonAI.
Just pass functions or agents as steps - no complex configuration needed!
"""
from praisonaiagents import Workflow, WorkflowContext, StepResult
# Define simple handler functions
def validate(ctx: WorkflowContext) -> StepResult:
"""Validate the input data."""
if not ctx.input:
return StepResult(output="No input provided", stop_workflow=True)
return StepResult(output=f"Validated: {ctx.input}")
def process(ctx: WorkflowContext) -> StepResult:
"""Process the validated data."""
return StepResult(output=f"Processed: {ctx.previous_result}")
def format_output(ctx: WorkflowContext) -> StepResult:
"""Format the final output."""
return StepResult(output=f"✅ Final: {ctx.previous_result}")
# Create workflow - just list your functions!
workflow = Workflow(
name="Simple Pipeline",
steps=[validate, process, format_output]
)
if __name__ == "__main__":
# Run the workflow - PraisonAI style uses .start()
result = workflow.start("Hello, World!", verbose=True)
print(f"\nFinal output: {result['output']}")
print(f"Steps completed: {len(result['steps'])}")
@@ -0,0 +1,57 @@
"""
Task Callbacks Example
Demonstrates using on_task_start and on_task_complete callbacks
with PraisonAIAgents for workflow monitoring and logging.
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
# Create an agent
researcher = Agent(
name="Researcher",
role="Research Assistant",
goal="Find and summarize information",
llm="gpt-4o-mini"
)
# Create tasks with variables
task1 = Task(
name="research",
description="Research the topic: {{topic}}",
expected_output="A brief summary of the topic",
agent=researcher,
variables={"topic": "artificial intelligence"}
)
task2 = Task(
name="summarize",
description="Create a one-paragraph summary of the research.",
expected_output="A concise summary",
agent=researcher
)
# Define callbacks
def on_start(task, task_id):
"""Called before each task starts."""
print(f"🚀 Starting task: {task.name} (ID: {task_id})")
def on_complete(task, output):
"""Called after each task completes."""
print(f"✅ Completed task: {task.name}")
print(f" Output preview: {str(output)[:100]}...")
if __name__ == "__main__":
# Create agents with callbacks
agents = PraisonAIAgents(
agents=[researcher],
tasks=[task1, task2],
process="workflow",
on_task_start=on_start,
on_task_complete=on_complete,
variables={"global_var": "shared_value"} # Global variables
)
# Run the workflow
result = agents.start()
print("\n🎉 Workflow completed!")
@@ -0,0 +1,48 @@
"""
Workflow Branching Example
Demonstrates conditional branching in workflows where steps can route
to different next steps based on output content.
"""
from praisonaiagents.memory.workflows import WorkflowManager, WorkflowStep, Workflow
# Create a workflow with branching
workflow = Workflow(
name="Decision Workflow",
description="A workflow that branches based on validation result",
steps=[
WorkflowStep(
name="validate",
action="Check if the number 42 is positive. Reply with 'valid' or 'invalid'.",
next_steps=["success_handler", "error_handler"],
branch_condition={
"valid": ["success_handler"],
"invalid": ["error_handler"]
}
),
WorkflowStep(
name="success_handler",
action="The validation passed! Generate a success message."
),
WorkflowStep(
name="error_handler",
action="The validation failed. Generate an error message."
)
]
)
if __name__ == "__main__":
# Create manager and register workflow
manager = WorkflowManager()
manager.workflows["Decision Workflow"] = workflow
# Execute - will branch to success_handler since 42 is positive
result = manager.execute(
"Decision Workflow",
default_llm="gpt-4o-mini"
)
print("Workflow completed!")
for step_result in result["results"]:
print(f" {step_result['step']}: {step_result['status']}")
@@ -0,0 +1,63 @@
"""
Workflow Checkpoints Example
Demonstrates saving and resuming workflow execution using checkpoints.
Useful for long-running workflows that may be interrupted.
"""
from praisonaiagents.memory.workflows import WorkflowManager, WorkflowStep, Workflow
# Create a multi-step workflow
workflow = Workflow(
name="Long Process",
description="A workflow with checkpoints for resumability",
steps=[
WorkflowStep(
name="step1",
action="Initialize the process and prepare data."
),
WorkflowStep(
name="step2",
action="Process the first batch of data."
),
WorkflowStep(
name="step3",
action="Process the second batch of data."
),
WorkflowStep(
name="step4",
action="Finalize and generate report."
)
]
)
if __name__ == "__main__":
manager = WorkflowManager()
manager.workflows["Long Process"] = workflow
# Execute with checkpoint - saves after each step
print("=== Starting workflow with checkpoint ===")
result = manager.execute(
"Long Process",
default_llm="gpt-4o-mini",
checkpoint="my-checkpoint" # Saves progress after each step
)
print(f"Completed {len(result['results'])} steps")
# List available checkpoints
print("\n=== Available checkpoints ===")
checkpoints = manager.list_checkpoints()
for cp in checkpoints:
print(f" {cp['name']}: {cp['completed_steps']} steps completed")
# Resume from checkpoint (if workflow was interrupted)
# result = manager.execute(
# "Long Process",
# default_llm="gpt-4o-mini",
# resume="my-checkpoint" # Resumes from last saved state
# )
# Clean up checkpoint
manager.delete_checkpoint("my-checkpoint")
print("\n✅ Checkpoint deleted")
@@ -0,0 +1,68 @@
"""
Workflow Conditional Execution Example
Demonstrates using should_run to conditionally execute steps
based on input or previous results.
"""
from praisonaiagents import Workflow, WorkflowStep, WorkflowContext, StepResult
# Condition functions - return True to run the step, False to skip
def is_sensitive_content(ctx: WorkflowContext) -> bool:
"""Only run compliance check for sensitive topics."""
sensitive_keywords = ["legal", "medical", "financial", "security"]
return any(keyword in ctx.input.lower() for keyword in sensitive_keywords)
def needs_translation(ctx: WorkflowContext) -> bool:
"""Only translate if content is not in English."""
return ctx.variables.get("language", "en") != "en"
# Handler functions
def process_content(ctx: WorkflowContext) -> StepResult:
return StepResult(output=f"Processed: {ctx.input}")
def compliance_check(ctx: WorkflowContext) -> StepResult:
return StepResult(output=f"✅ Compliance verified for: {ctx.previous_result}")
def translate(ctx: WorkflowContext) -> StepResult:
lang = ctx.variables.get("language", "unknown")
return StepResult(output=f"🌐 Translated to {lang}: {ctx.previous_result}")
def finalize(ctx: WorkflowContext) -> StepResult:
return StepResult(output=f"📄 Final: {ctx.previous_result}")
# Create workflow with conditional steps
workflow = Workflow(
name="Conditional Pipeline",
steps=[
WorkflowStep(name="process", handler=process_content),
WorkflowStep(
name="compliance",
handler=compliance_check,
should_run=is_sensitive_content # Only runs for sensitive content
),
WorkflowStep(
name="translate",
handler=translate,
should_run=needs_translation # Only runs if language != "en"
),
WorkflowStep(name="finalize", handler=finalize)
]
)
if __name__ == "__main__":
# Test 1: Normal content (skips compliance)
print("=== Test 1: Normal content ===")
result = workflow.start("Hello world", verbose=True)
print(f"Output: {result['output']}\n")
# Test 2: Sensitive content (runs compliance)
print("=== Test 2: Legal content ===")
result = workflow.start("Review this legal document", verbose=True)
print(f"Output: {result['output']}\n")
# Test 3: Non-English content (runs translation)
print("=== Test 3: Non-English content ===")
workflow.variables = {"language": "es"}
result = workflow.start("Hola mundo", verbose=True)
print(f"Output: {result['output']}")
@@ -0,0 +1,69 @@
"""
Workflow Early Stop Example
Demonstrates how to stop a workflow early using custom handler
functions that return StepResult with stop_workflow=True.
"""
from praisonaiagents.memory.workflows import (
WorkflowManager, WorkflowStep, Workflow,
WorkflowContext, StepResult
)
# Custom validator that can stop the workflow
def validate_data(context: WorkflowContext) -> StepResult:
"""Check if data is valid and stop workflow if not."""
data = context.variables.get("data", {})
if data.get("value", 0) < 0:
return StepResult(
output="❌ Validation failed: negative value detected. Stopping workflow.",
stop_workflow=True # This stops the entire workflow
)
return StepResult(
output="✅ Validation passed: data is valid.",
stop_workflow=False # Continue to next step
)
# Create workflow with early stop capability
workflow = Workflow(
name="Data Processing",
description="Process data with validation gate",
variables={
"data": {"value": -5} # Invalid data - will trigger early stop
},
steps=[
WorkflowStep(
name="validate",
handler=validate_data # Custom function
),
WorkflowStep(
name="process",
action="Process the validated data." # Won't run if validation fails
),
WorkflowStep(
name="report",
action="Generate final report." # Won't run if validation fails
)
]
)
if __name__ == "__main__":
manager = WorkflowManager()
manager.workflows["Data Processing"] = workflow
# Test with invalid data (will stop early)
print("=== Testing with invalid data ===")
result = manager.execute("Data Processing", default_llm="gpt-4o-mini")
for step_result in result["results"]:
print(f" {step_result['step']}: {step_result['output']}")
# Test with valid data (will complete)
print("\n=== Testing with valid data ===")
workflow.variables["data"] = {"value": 42}
result = manager.execute("Data Processing", default_llm="gpt-4o-mini")
for step_result in result["results"]:
print(f" {step_result['step']}: {step_result['status']}")
@@ -0,0 +1,55 @@
"""
Workflow Loop with CSV Example
Demonstrates iterating over a CSV file, processing each row.
"""
from praisonaiagents import Workflow, WorkflowContext, StepResult
from praisonaiagents.workflows import loop
import tempfile
import os
# Create sample CSV file
csv_content = """name,email,task
Alice,[email protected],Write documentation
Bob,[email protected],Review pull requests
Charlie,[email protected],Fix critical bug
Diana,[email protected],Deploy to production"""
# Processor for each CSV row
def process_task(ctx: WorkflowContext) -> StepResult:
"""Process a single task from CSV."""
item = ctx.variables.get("item", {})
name = item.get("name", "Unknown")
task = item.get("task", "No task")
index = ctx.variables.get("loop_index", 0)
return StepResult(
output=f"[{index+1}] {name}: {task} ✅",
variables={"last_processed": name}
)
if __name__ == "__main__":
# Create temp CSV file
with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as f:
f.write(csv_content)
csv_path = f.name
try:
print("=== Testing Loop with CSV ===\n")
print(f"CSV file: {csv_path}\n")
# Create workflow
workflow = Workflow(
name="CSV Task Processor",
steps=[loop(process_task, from_csv=csv_path)]
)
result = workflow.start("Process all tasks", verbose=True)
print(f"\nAll outputs:\n{result['output']}")
print(f"\nProcessed {len(result['variables']['loop_outputs'])} items")
print(f"Last processed: {result['variables'].get('last_processed')}")
finally:
os.unlink(csv_path)
@@ -0,0 +1,52 @@
"""
Workflow Loop with List Example
Demonstrates iterating over a list of items, processing each one.
"""
from praisonaiagents import Workflow, WorkflowContext, StepResult
from praisonaiagents.workflows import loop
# Sample data
fruits = ["apple", "banana", "cherry", "date", "elderberry"]
# Processor for each item
def process_fruit(ctx: WorkflowContext) -> StepResult:
"""Process a single fruit from the list."""
fruit = ctx.variables.get("item", "unknown")
index = ctx.variables.get("loop_index", 0)
# Simulate processing
result = f"🍎 [{index + 1}] {fruit.upper()} - processed successfully"
return StepResult(
output=result,
variables={"last_fruit": fruit}
)
# Aggregator to summarize results
def summarize(ctx: WorkflowContext) -> StepResult:
"""Summarize all processed fruits."""
outputs = ctx.variables.get("loop_outputs", [])
summary = f"📊 Processed {len(outputs)} fruits:\n"
summary += "\n".join(f" {o}" for o in outputs)
return StepResult(output=summary)
# Create workflow
workflow = Workflow(
name="Fruit Processor",
steps=[
loop(process_fruit, over="fruits"),
summarize
],
variables={"fruits": fruits}
)
if __name__ == "__main__":
print("=== Testing Loop with List ===\n")
result = workflow.start("Process all fruits", verbose=True)
print(f"\nFinal Output:\n{result['output']}")
print(f"\nLoop outputs count: {len(result['variables']['loop_outputs'])}")
print(f"Last processed: {result['variables'].get('last_fruit')}")
@@ -0,0 +1,46 @@
"""
Workflow Loops Example
Demonstrates iterating over a list of items in a workflow step.
Each item is processed individually with the loop variable.
"""
from praisonaiagents.memory.workflows import WorkflowManager, WorkflowStep, Workflow
# Create a workflow with loop
workflow = Workflow(
name="Process Items",
description="Process a list of items using loop",
variables={
"items": ["apple", "banana", "cherry"] # List to iterate over
},
steps=[
WorkflowStep(
name="process_each",
action="Describe the fruit: {{item}}",
loop_over="items", # Variable name containing the list
loop_var="item" # Variable name for current item
),
WorkflowStep(
name="summarize",
action="Summarize all the fruits that were processed."
)
]
)
if __name__ == "__main__":
# Create manager and register workflow
manager = WorkflowManager()
manager.workflows["Process Items"] = workflow
# Execute - will process each fruit in the list
result = manager.execute(
"Process Items",
default_llm="gpt-4o-mini"
)
print("Workflow completed!")
for step_result in result["results"]:
print(f" {step_result['step']}: {step_result['status']}")
if isinstance(step_result.get("output"), list):
print(f" Loop processed {len(step_result['output'])} items")
@@ -0,0 +1,68 @@
"""
Mixed Steps Workflow Example
Demonstrates combining different step types in a single workflow:
- Agent instances
- Handler functions
- WorkflowStep with action strings
"""
from praisonaiagents import Workflow, WorkflowStep, WorkflowContext, StepResult, Agent
# Create an agent
analyzer = Agent(
name="Analyzer",
role="Data Analyst",
goal="Analyze and summarize data",
llm="gpt-4o-mini"
)
# Custom handler functions
def validate_input(ctx: WorkflowContext) -> StepResult:
"""Validate the input before processing."""
if len(ctx.input) < 5:
return StepResult(
output="Input too short. Minimum 5 characters required.",
stop_workflow=True
)
return StepResult(output=f"✅ Valid input: {ctx.input}")
def format_output(ctx: WorkflowContext) -> StepResult:
"""Format the final output."""
return StepResult(
output=f"📊 Report:\n{ctx.previous_result}",
variables={"report_generated": True}
)
# Create workflow with mixed step types
workflow = Workflow(
name="Mixed Pipeline",
steps=[
# Step 1: Function - validates input
validate_input,
# Step 2: Agent - analyzes content
analyzer,
# Step 3: WorkflowStep with action string
WorkflowStep(
name="enhance",
action="Add a conclusion to this analysis: {{previous_output}}"
),
# Step 4: Function - formats output
format_output
],
default_llm="gpt-4o-mini" # Used for action-based steps
)
if __name__ == "__main__":
# Test with valid input
print("=== Running mixed workflow ===")
result = workflow.start(
"Analyze the trends in AI adoption for 2024",
verbose=True
)
print(f"\nFinal output:\n{result['output']}")
print(f"\nVariables: {result['variables']}")
@@ -0,0 +1,53 @@
"""
Workflow Parallel Execution Example
Demonstrates running multiple steps concurrently and
combining their results.
"""
from praisonaiagents import Workflow, WorkflowContext, StepResult
from praisonaiagents.workflows import parallel
import time
# Parallel workers - each does independent work
def research_market(ctx: WorkflowContext) -> StepResult:
"""Research market trends."""
time.sleep(0.1) # Simulate work
return StepResult(output="📊 Market: Growth expected 15% YoY")
def research_competitors(ctx: WorkflowContext) -> StepResult:
"""Research competitor analysis."""
time.sleep(0.1) # Simulate work
return StepResult(output="🏢 Competitors: 3 major players identified")
def research_customers(ctx: WorkflowContext) -> StepResult:
"""Research customer feedback."""
time.sleep(0.1) # Simulate work
return StepResult(output="👥 Customers: 85% satisfaction rate")
# Aggregator - combines parallel results
def summarize_research(ctx: WorkflowContext) -> StepResult:
"""Summarize all research findings."""
outputs = ctx.variables.get("parallel_outputs", [])
summary = "📋 RESEARCH SUMMARY:\n" + "\n".join(f" • {o}" for o in outputs)
return StepResult(output=summary)
# Create workflow with parallel execution
workflow = Workflow(
name="Parallel Research",
steps=[
parallel([research_market, research_competitors, research_customers]),
summarize_research
]
)
if __name__ == "__main__":
print("=== Testing Parallel Workflow ===\n")
start_time = time.time()
result = workflow.start("Analyze the business landscape", verbose=True)
elapsed = time.time() - start_time
print(f"\nFinal Output:\n{result['output']}")
print(f"\nCompleted in {elapsed:.2f}s (parallel execution)")
print(f"Individual outputs: {len(result['variables']['parallel_outputs'])}")
@@ -0,0 +1,69 @@
"""
Workflow Repeat (Evaluator-Optimizer) Example
Demonstrates repeating a step until a condition is met.
This is useful for iterative improvement patterns.
"""
from praisonaiagents import Workflow, WorkflowContext, StepResult
from praisonaiagents.workflows import repeat
# Simulated content generator that improves each iteration
class ContentGenerator:
def __init__(self):
self.iteration = 0
self.content = []
def generate(self, ctx: WorkflowContext) -> StepResult:
"""Generate content, adding more each iteration."""
self.iteration += 1
# Add new content each iteration
new_items = [
f"Point {self.iteration}.1: Key insight about the topic",
f"Point {self.iteration}.2: Supporting evidence"
]
self.content.extend(new_items)
output = f"Iteration {self.iteration}: Generated {len(self.content)} points total"
return StepResult(
output=output,
variables={
"total_points": len(self.content),
"content": self.content.copy()
}
)
# Condition: stop when we have enough content
def has_enough_content(ctx: WorkflowContext) -> bool:
"""Check if we have generated enough content."""
total = ctx.variables.get("total_points", 0)
return total >= 6 # Stop when we have 6+ points
# Create generator instance
generator = ContentGenerator()
# Create workflow with repeat pattern
workflow = Workflow(
name="Content Generator",
steps=[
repeat(
generator.generate,
until=has_enough_content,
max_iterations=10
)
]
)
if __name__ == "__main__":
print("=== Testing Repeat (Evaluator-Optimizer) Pattern ===\n")
result = workflow.start("Generate comprehensive content", verbose=True)
print(f"\nFinal output: {result['output']}")
print(f"Total iterations: {result['variables']['repeat_iterations']}")
print(f"Total points generated: {result['variables']['total_points']}")
print("\nGenerated content:")
for point in result['variables']['content']:
print(f" • {point}")
@@ -0,0 +1,61 @@
"""
Workflow Routing Example
Demonstrates decision-based routing where the workflow
takes different paths based on the output of a decision step.
"""
from praisonaiagents import Workflow, WorkflowContext, StepResult
from praisonaiagents.workflows import route
# Decision maker - determines which route to take
def classify_request(ctx: WorkflowContext) -> StepResult:
"""Classify the input and decide the route."""
input_lower = ctx.input.lower()
if "urgent" in input_lower or "emergency" in input_lower:
return StepResult(output="priority: high")
elif "question" in input_lower or "help" in input_lower:
return StepResult(output="priority: support")
else:
return StepResult(output="priority: normal")
# Route handlers
def handle_high_priority(ctx: WorkflowContext) -> StepResult:
return StepResult(output="🚨 HIGH PRIORITY: Escalating to senior team immediately!")
def handle_support(ctx: WorkflowContext) -> StepResult:
return StepResult(output="💬 SUPPORT: Routing to help desk for assistance.")
def handle_normal(ctx: WorkflowContext) -> StepResult:
return StepResult(output="📋 NORMAL: Added to standard processing queue.")
# Create workflow with routing
workflow = Workflow(
name="Request Router",
steps=[
classify_request,
route({
"high": [handle_high_priority],
"support": [handle_support],
"normal": [handle_normal],
"default": [handle_normal]
})
]
)
if __name__ == "__main__":
# Test different inputs
print("=== Testing Routing Workflow ===\n")
# Test 1: Urgent request
result = workflow.start("This is an URGENT matter!", verbose=True)
print(f"Result: {result['output']}\n")
# Test 2: Support request
result = workflow.start("I have a question about my account", verbose=True)
print(f"Result: {result['output']}\n")
# Test 3: Normal request
result = workflow.start("Please process my order", verbose=True)
print(f"Result: {result['output']}")
@@ -0,0 +1,45 @@
"""
Workflow with Agents Example
Demonstrates using Agent instances directly as workflow steps.
Agents are automatically wrapped and executed in sequence.
"""
from praisonaiagents import Workflow, Agent
# Create agents
researcher = Agent(
name="Researcher",
role="Research Specialist",
goal="Find accurate information",
llm="gpt-4o-mini"
)
writer = Agent(
name="Writer",
role="Content Writer",
goal="Write engaging content",
llm="gpt-4o-mini"
)
editor = Agent(
name="Editor",
role="Content Editor",
goal="Polish and improve content",
llm="gpt-4o-mini"
)
# Create workflow with agents as steps
workflow = Workflow(
name="Content Pipeline",
steps=[researcher, writer, editor]
)
if __name__ == "__main__":
# Run the workflow
result = workflow.start(
"Write a short paragraph about artificial intelligence",
verbose=True
)
print(f"\nFinal output:\n{result['output']}")
@@ -54,6 +54,13 @@ except ImportError:
MCP = None
from .session import Session
from .memory.memory import Memory
# Workflows - import from dedicated workflows module
from .workflows import (
Workflow, WorkflowStep, WorkflowContext, StepResult,
Route, Parallel, Loop, Repeat,
route, parallel, loop, repeat,
Pipeline # Alias for Workflow
)
from .guardrails import GuardrailResult, LLMGuardrail
# Fast Context support (lazy loaded to avoid performance impact)
@@ -248,7 +248,9 @@ class Agent:
fast_context_model: str = "gpt-4o-mini",
fast_context_max_turns: int = 4,
fast_context_parallelism: int = 8,
fast_context_timeout: float = 30.0
fast_context_timeout: float = 30.0,
history_in_context: Optional[int] = None,
auto_save: Optional[str] = None
):
"""Initialize an Agent instance.
@@ -544,6 +546,10 @@ Your Goal: {self.goal}
self.prompt_caching = prompt_caching
self.claude_memory = claude_memory
# Session management
self.history_in_context = history_in_context # Number of past sessions to include
self.auto_save = auto_save # Session name for auto-saving
# Initialize rules manager for persistent context (like Cursor/Windsurf)
self._rules_manager = None
self._init_rules_manager()
@@ -2676,21 +2682,78 @@ Write the complete compiled report:"""
def start(self, prompt: str, **kwargs):
"""Start the agent with a prompt. This is a convenience method that wraps chat()."""
# Load history from past sessions if history_in_context is set
self._load_history_context()
# Check if planning mode is enabled
if self.planning:
return self._start_with_planning(prompt, **kwargs)
# Check if streaming is enabled (either from kwargs or agent's stream attribute)
stream_enabled = kwargs.get('stream', getattr(self, 'stream', False))
if stream_enabled:
result = self._start_with_planning(prompt, **kwargs)
elif kwargs.get('stream', getattr(self, 'stream', False)):
# Return a generator for streaming response
return self._start_stream(prompt, **kwargs)
result = self._start_stream(prompt, **kwargs)
else:
# Return regular chat response for backward compatibility
# Explicitly pass the resolved stream parameter to avoid chat() method default
kwargs['stream'] = stream_enabled
return self.chat(prompt, **kwargs)
kwargs['stream'] = False
result = self.chat(prompt, **kwargs)
# Auto-save session if enabled
self._auto_save_session()
return result
def _load_history_context(self):
"""Load history from past sessions into context if history_in_context is set."""
if not self.history_in_context or not self._memory_instance:
return
try:
sessions = self._memory_instance.list_sessions()
if not sessions:
return
# Get the last N sessions
sessions_to_load = sessions[:self.history_in_context]
for session_info in reversed(sessions_to_load):
try:
session_data = self._memory_instance.resume_session(session_info["name"])
history = session_data.get("conversation_history", [])
# Add history to chat_history with a marker
for msg in history:
if msg not in self.chat_history:
# Mark as from history to avoid duplication
msg_copy = msg.copy()
msg_copy["_from_history"] = True
self.chat_history.append(msg_copy)
except Exception:
continue
if sessions_to_load:
logging.debug(f"Loaded history from {len(sessions_to_load)} past session(s)")
except Exception as e:
logging.debug(f"Error loading history context: {e}")
def _auto_save_session(self):
"""Auto-save session if auto_save is enabled."""
if not self.auto_save or not self._memory_instance:
return
try:
# Filter out history markers before saving
clean_history = [
{k: v for k, v in msg.items() if k != "_from_history"}
for msg in self.chat_history
]
self._memory_instance.save_session(
name=self.auto_save,
conversation_history=clean_history,
metadata={"agent_name": self.name, "user_id": self.user_id}
)
logging.debug(f"Auto-saved session: {self.auto_save}")
except Exception as e:
logging.debug(f"Error auto-saving session: {e}")
def _start_stream(self, prompt: str, **kwargs) -> Generator[str, None, None]:
"""Stream generator for real-time response chunks."""
@@ -2,7 +2,7 @@ import os
import time
import json
import logging
from typing import Any, Dict, Optional, List
from typing import Any, Dict, Optional, List, Callable
from pydantic import BaseModel
from rich.text import Text
from rich.panel import Panel
@@ -127,7 +127,7 @@ def process_task_context(context_item, verbose=0, user_id=None):
return str(context_item) # Fallback for unknown types
class PraisonAIAgents:
def __init__(self, agents, tasks=None, verbose=0, completion_checker=None, max_retries=5, process="sequential", manager_llm=None, memory=False, memory_config=None, embedder=None, user_id=None, max_iter=10, stream=True, name: Optional[str] = None, planning: bool = False, planning_llm: Optional[str] = None, auto_approve_plan: bool = False, planning_tools: Optional[List] = None, planning_reasoning: bool = False):
def __init__(self, agents, tasks=None, verbose=0, completion_checker=None, max_retries=5, process="sequential", manager_llm=None, memory=False, memory_config=None, embedder=None, user_id=None, max_iter=10, stream=True, name: Optional[str] = None, planning: bool = False, planning_llm: Optional[str] = None, auto_approve_plan: bool = False, planning_tools: Optional[List] = None, planning_reasoning: bool = False, on_task_start: Optional[Callable] = None, on_task_complete: Optional[Callable] = None, variables: Optional[Dict[str, Any]] = None):
# Add check at the start if memory is requested
if memory:
try:
@@ -162,6 +162,11 @@ class PraisonAIAgents:
self.stream = stream
self.name = name # Store the name for the Agents collection
# Callbacks for workflow execution
self.on_task_start = on_task_start # Called before each task
self.on_task_complete = on_task_complete # Called after each task
self.variables = variables if variables else {} # Global variables for substitution
# Check for manager_llm in environment variable if not provided
self.manager_llm = manager_llm or os.getenv('OPENAI_MODEL_NAME', 'gpt-5-nano')
@@ -661,13 +666,24 @@ Context:
executor_agent = task.agent
# Create agent from agent_config if provided and no agent assigned
if executor_agent is None and getattr(task, 'agent_config', None):
executor_agent = self._create_agent_from_config(task.agent_config)
task.agent = executor_agent
# Set current agent for token tracking
llm = getattr(executor_agent, 'llm', None) or getattr(executor_agent, 'llm_instance', None)
if llm and hasattr(llm, 'set_current_agent'):
llm.set_current_agent(executor_agent.name)
# Substitute variables in task description if provided
task_description = task.description
if getattr(task, 'variables', None):
for key, value in task.variables.items():
task_description = task_description.replace(f"{{{{{key}}}}}", str(value))
task_prompt = f"""
You need to do the following task: {task.description}.
You need to do the following task: {task_description}.
Expected Output: {task.expected_output}.
"""
if task.context:
@@ -825,6 +841,17 @@ Context:
logger.info(f"Task with ID {task_id} is already completed")
return
# Call on_task_start callback if provided
if self.on_task_start:
try:
self.on_task_start(task, task_id)
except Exception as e:
logger.error(f"Error in on_task_start callback: {e}")
# Apply global variables to task if not already set
if self.variables and not getattr(task, 'variables', None):
task.variables = self.variables
retries = 0
while task.status != "completed" and retries < self.max_retries:
logger.debug(f"Attempt {retries+1} for task {task_id}")
@@ -857,6 +884,14 @@ Context:
logger.exception(e)
self.save_output_to_file(task, task_output)
# Call on_task_complete callback if provided
if self.on_task_complete:
try:
self.on_task_complete(task, task_output)
except Exception as e:
logger.error(f"Error in on_task_complete callback: {e}")
if self.verbose >= 1:
logger.info(f"Task {task_id} completed successfully.")
else:
@@ -1561,6 +1596,29 @@ Context:
"""Get the current todo list."""
return self._todo_list
def _create_agent_from_config(self, agent_config: dict) -> 'Agent':
"""
Create an Agent from a configuration dictionary.
Args:
agent_config: Dict with keys like 'name', 'role', 'goal', 'backstory', 'llm', 'tools'
Returns:
Agent instance
"""
from ..agent.agent import Agent
return Agent(
name=agent_config.get('name', 'TaskAgent'),
role=agent_config.get('role'),
goal=agent_config.get('goal'),
backstory=agent_config.get('backstory'),
llm=agent_config.get('llm'),
tools=agent_config.get('tools'),
verbose=self.verbose >= 1,
memory=self.memory
)
def _get_planning_agent(self):
"""Lazy load PlanningAgent."""
if self._planning_agent is None and self.planning:
@@ -15,6 +15,8 @@ This module provides memory management capabilities including:
- Auto-generated memories (like Windsurf Cascade)
- Workflows (like Windsurf)
- Hooks (like Windsurf Cascade Hooks)
- Docs management (like Cursor docs)
- MCP config management (like Cursor .cursor/mcp/)
Memory Providers:
- FileMemory: Zero-dependency JSON file-based storage (default)
@@ -23,10 +25,14 @@ Memory Providers:
- AutoMemory: Automatic memory extraction from conversations
- WorkflowManager: Multi-step workflow execution
- HooksManager: Pre/post operation hooks
- DocsManager: Documentation context management
- MCPConfigManager: MCP server configuration management
"""
from .file_memory import FileMemory, create_memory
from .rules_manager import RulesManager, Rule, create_rules_manager
from .docs_manager import DocsManager, Doc
from .mcp_config import MCPConfigManager, MCPConfig
# Lazy imports for optional modules to avoid dependency issues and improve startup time
def __getattr__(name):
@@ -42,17 +48,58 @@ def __getattr__(name):
if name == "create_auto_memory":
from .auto_memory import create_auto_memory
return create_auto_memory
# Workflows - now in dedicated module, re-export for backward compatibility
if name == "WorkflowManager":
from .workflows import WorkflowManager
from ..workflows import WorkflowManager
return WorkflowManager
if name == "Workflow":
from .workflows import Workflow
from ..workflows import Workflow
return Workflow
if name == "WorkflowStep":
from .workflows import WorkflowStep
from ..workflows import WorkflowStep
return WorkflowStep
if name == "WorkflowContext":
from ..workflows import WorkflowContext
return WorkflowContext
if name == "StepResult":
from ..workflows import StepResult
return StepResult
if name == "Pipeline":
from ..workflows import Pipeline
return Pipeline
if name == "Route":
from ..workflows import Route
return Route
if name == "Parallel":
from ..workflows import Parallel
return Parallel
if name == "Loop":
from ..workflows import Loop
return Loop
if name == "Repeat":
from ..workflows import Repeat
return Repeat
if name == "route":
from ..workflows import route
return route
if name == "parallel":
from ..workflows import parallel
return parallel
if name == "loop":
from ..workflows import loop
return loop
if name == "repeat":
from ..workflows import repeat
return repeat
# Backward compatibility aliases
if name == "StepInput":
from ..workflows import WorkflowContext
return WorkflowContext
if name == "StepOutput":
from ..workflows import StepResult
return StepResult
if name == "create_workflow_manager":
from .workflows import create_workflow_manager
from ..workflows.workflows import create_workflow_manager
return create_workflow_manager
if name == "HooksManager":
from .hooks import HooksManager
@@ -63,6 +110,7 @@ def __getattr__(name):
if name == "create_hooks_manager":
from .hooks import create_hooks_manager
return create_hooks_manager
# DocsManager and MCPConfigManager are already imported at module level
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
__all__ = [
@@ -78,13 +126,31 @@ __all__ = [
"AutoMemory",
"AutoMemoryExtractor",
"create_auto_memory",
# Workflows
# Workflows (re-exported from praisonaiagents.workflows for backward compatibility)
"WorkflowManager",
"Workflow",
"Pipeline",
"WorkflowStep",
"WorkflowContext",
"StepResult",
"create_workflow_manager",
# Workflow patterns
"Route",
"Parallel",
"Loop",
"Repeat",
"route",
"parallel",
"loop",
"repeat",
# Hooks
"HooksManager",
"HookResult",
"create_hooks_manager",
# Docs management
"DocsManager",
"Doc",
# MCP config management
"MCPConfigManager",
"MCPConfig",
]
@@ -0,0 +1,431 @@
"""
Docs Manager for PraisonAI Agents.
Provides reusable documentation context support similar to:
- Cursor Notepads (deprecated in favor of docs)
- Cursor .cursor/docs/
- Project documentation for AI context
Features:
- Auto-discovery of docs from .praison/docs/
- Support for markdown, text, and code files
- @doc mentions for including specific docs
- Priority ordering for docs
- Character limits for context window management
Storage Structure:
.praison/docs/
├── project-overview.md # Project overview
├── architecture.md # Architecture decisions
├── api-reference.md # API documentation
└── coding-standards.md # Coding guidelines
Doc File Format (YAML frontmatter + Markdown):
---
description: Project architecture overview
priority: 10
tags: [architecture, design]
---
# Architecture Overview
This project uses...
"""
import os
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional, Literal
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class Doc:
"""A single documentation file with metadata and content."""
name: str
content: str
description: str = ""
priority: int = 0
tags: List[str] = field(default_factory=list)
file_path: Optional[str] = None
def matches_tag(self, tag: str) -> bool:
"""Check if this doc has a specific tag."""
return tag.lower() in [t.lower() for t in self.tags]
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"content": self.content,
"description": self.description,
"priority": self.priority,
"tags": self.tags,
"file_path": self.file_path
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'Doc':
return cls(**data)
class DocsManager:
"""
Manages documentation context for AI agents.
Provides:
- Auto-discovery of docs from .praison/docs/
- Global docs from ~/.praison/docs/
- @doc mentions for including specific docs
- Priority-based ordering
"""
DOCS_DIR_NAME = ".praison/docs"
SUPPORTED_EXTENSIONS = [".md", ".txt", ".rst"]
# Maximum characters per doc (context window management)
MAX_DOC_CHARS = 12000
def __init__(
self,
workspace_path: Optional[str] = None,
global_docs_path: Optional[str] = None,
verbose: int = 0
):
"""
Initialize DocsManager.
Args:
workspace_path: Path to workspace/project root
global_docs_path: Path to global docs (default: ~/.praison/docs)
verbose: Verbosity level
"""
self.workspace_path = Path(workspace_path) if workspace_path else Path.cwd()
self.global_docs_path = Path(global_docs_path) if global_docs_path else Path.home() / ".praison" / "docs"
self.verbose = verbose
self._docs: Dict[str, Doc] = {}
self._load_all_docs()
def _log(self, msg: str, level: int = logging.INFO):
"""Log message if verbose."""
if self.verbose >= 1:
logger.log(level, msg)
def _parse_frontmatter(self, content: str) -> tuple:
"""
Parse YAML frontmatter from content.
Returns:
Tuple of (frontmatter_dict, remaining_content)
"""
frontmatter = {}
remaining = content
# Check for YAML frontmatter
if content.startswith("---"):
parts = content.split("---", 2)
if len(parts) >= 3:
try:
import yaml
frontmatter = yaml.safe_load(parts[1]) or {}
remaining = parts[2].strip()
except Exception:
# If YAML parsing fails, use content as-is
pass
return frontmatter, remaining
def _load_doc_file(self, file_path: Path, scope: str = "workspace") -> Optional[Doc]:
"""Load a single doc file."""
try:
content = file_path.read_text(encoding="utf-8")
frontmatter, doc_content = self._parse_frontmatter(content)
name = file_path.stem
# Determine priority based on scope
base_priority = 100 if scope == "workspace" else 0
doc = Doc(
name=name,
content=doc_content,
description=frontmatter.get("description", ""),
priority=frontmatter.get("priority", base_priority),
tags=frontmatter.get("tags", []),
file_path=str(file_path)
)
return doc
except Exception as e:
self._log(f"Failed to load doc {file_path}: {e}", logging.WARNING)
return None
def _load_docs_from_dir(self, docs_dir: Path, scope: str = "workspace"):
"""Load all docs from a directory."""
if not docs_dir.exists():
return
for ext in self.SUPPORTED_EXTENSIONS:
for file_path in docs_dir.glob(f"*{ext}"):
if file_path.is_file():
doc = self._load_doc_file(file_path, scope)
if doc:
key = f"{scope}:{doc.name}"
self._docs[key] = doc
self._log(f"Loaded doc: {key}")
def _load_all_docs(self):
"""Load all docs from global and workspace directories."""
self._docs.clear()
# 1. Load global docs (lowest priority)
if self.global_docs_path.exists():
self._load_docs_from_dir(self.global_docs_path, "global")
for key, doc in self._docs.items():
if key.startswith("global:"):
doc.priority = doc.priority - 1000 # Lower priority for global
# 2. Load workspace docs
workspace_docs_dir = self.workspace_path / self.DOCS_DIR_NAME.replace("/", os.sep)
if workspace_docs_dir.exists():
self._load_docs_from_dir(workspace_docs_dir, "workspace")
self._log(f"Loaded {len(self._docs)} docs total")
def reload(self):
"""Reload all docs from disk."""
self._load_all_docs()
def get_all_docs(self) -> List[Doc]:
"""Get all loaded docs, sorted by priority."""
docs = list(self._docs.values())
docs.sort(key=lambda d: d.priority, reverse=True)
return docs
def get_doc(self, name: str) -> Optional[Doc]:
"""Get a specific doc by name."""
# Try workspace first, then global
for scope in ["workspace", "global"]:
key = f"{scope}:{name}"
if key in self._docs:
return self._docs[key]
return None
def get_docs_by_tag(self, tag: str) -> List[Doc]:
"""Get all docs with a specific tag."""
docs = [d for d in self._docs.values() if d.matches_tag(tag)]
docs.sort(key=lambda d: d.priority, reverse=True)
return docs
def get_docs_for_context(
self,
include_docs: Optional[List[str]] = None,
include_tags: Optional[List[str]] = None,
max_chars: int = 50000
) -> List[Doc]:
"""
Get docs for inclusion in agent context.
Args:
include_docs: Specific doc names to include (via @doc mentions)
include_tags: Tags to filter by
max_chars: Maximum total characters
Returns:
List of docs sorted by priority
"""
active = []
# Include specific docs by name
if include_docs:
for name in include_docs:
doc = self.get_doc(name)
if doc and doc not in active:
active.append(doc)
# Include docs by tag
if include_tags:
for tag in include_tags:
for doc in self.get_docs_by_tag(tag):
if doc not in active:
active.append(doc)
# If no specific docs requested, include all high-priority docs
if not include_docs and not include_tags:
for doc in self.get_all_docs():
if doc.priority >= 100: # Only high-priority docs by default
active.append(doc)
# Sort by priority
active.sort(key=lambda d: d.priority, reverse=True)
return active
def format_docs_for_prompt(
self,
include_docs: Optional[List[str]] = None,
include_tags: Optional[List[str]] = None,
max_chars: int = 50000
) -> str:
"""
Format docs as a string for inclusion in prompts.
Args:
include_docs: Specific doc names to include
include_tags: Tags to filter by
max_chars: Maximum total characters
Returns:
Formatted docs string
"""
active_docs = self.get_docs_for_context(include_docs, include_tags, max_chars)
if not active_docs:
return ""
parts = ["# Project Documentation\n"]
total_chars = len(parts[0])
for doc in active_docs:
if total_chars >= max_chars:
break
doc_text = doc.content.strip()
if not doc_text:
continue
# Add doc with header
if doc.description:
header = f"## {doc.name}: {doc.description}"
else:
header = f"## {doc.name}"
section = f"{header}\n{doc_text}\n"
if total_chars + len(section) <= max_chars:
parts.append(section)
total_chars += len(section)
else:
# Truncate last doc
remaining = max_chars - total_chars
if remaining > 100:
parts.append(section[:remaining] + "\n... (truncated)")
break
return "\n".join(parts)
def create_doc(
self,
name: str,
content: str,
description: str = "",
priority: int = 0,
tags: Optional[List[str]] = None,
scope: Literal["global", "workspace"] = "workspace"
) -> Doc:
"""
Create a new doc file.
Args:
name: Doc name (used as filename)
content: Doc content (markdown)
description: Short description
priority: Priority (higher = applied first)
tags: Tags for categorization
scope: Where to save (global or workspace)
Returns:
Created Doc object
"""
# Determine save path
if scope == "global":
docs_dir = self.global_docs_path
else:
docs_dir = self.workspace_path / self.DOCS_DIR_NAME.replace("/", os.sep)
docs_dir.mkdir(parents=True, exist_ok=True)
file_path = docs_dir / f"{name}.md"
# Build frontmatter
frontmatter_lines = ["---"]
if description:
frontmatter_lines.append(f'description: "{description}"')
if priority != 0:
frontmatter_lines.append(f"priority: {priority}")
if tags:
tags_str = ", ".join(f'"{t}"' for t in tags)
frontmatter_lines.append(f"tags: [{tags_str}]")
frontmatter_lines.append("---")
frontmatter_lines.append("")
# Write file
full_content = "\n".join(frontmatter_lines) + content
file_path.write_text(full_content, encoding="utf-8")
# Create and register doc
doc = Doc(
name=name,
content=content,
description=description,
priority=priority,
tags=tags or [],
file_path=str(file_path)
)
self._docs[f"{scope}:{name}"] = doc
self._log(f"Created doc '{name}' at {file_path}")
return doc
def delete_doc(self, name: str, scope: Optional[str] = None) -> bool:
"""
Delete a doc file.
Args:
name: Doc name to delete
scope: Scope to delete from (None = try both)
Returns:
True if deleted, False if not found
"""
scopes = [scope] if scope else ["workspace", "global"]
for s in scopes:
key = f"{s}:{name}"
if key in self._docs:
doc = self._docs[key]
if doc.file_path:
try:
Path(doc.file_path).unlink()
del self._docs[key]
self._log(f"Deleted doc '{name}' from {s}")
return True
except Exception as e:
self._log(f"Failed to delete doc file: {e}", logging.ERROR)
return False
def list_docs(self) -> List[Dict[str, Any]]:
"""
List all docs with metadata.
Returns:
List of doc info dicts
"""
docs = []
for key, doc in self._docs.items():
scope = key.split(":")[0]
docs.append({
"name": doc.name,
"scope": scope,
"description": doc.description,
"priority": doc.priority,
"tags": doc.tags,
"file_path": doc.file_path,
"content_length": len(doc.content)
})
# Sort by priority
docs.sort(key=lambda d: d["priority"], reverse=True)
return docs
@@ -0,0 +1,369 @@
"""
MCP Config Manager for PraisonAI Agents.
Provides JSON-based MCP server configuration support similar to:
- Cursor .cursor/mcp/*.mcpjson
- Windsurf MCP configuration
Features:
- Auto-discovery of MCP configs from .praison/mcp/
- JSON config files for MCP servers
- Environment variable support
- Easy server management
Storage Structure:
.praison/mcp/
├── filesystem.json # Filesystem MCP server
├── brave-search.json # Brave search MCP server
└── postgres.json # PostgreSQL MCP server
Config File Format (JSON):
{
"name": "filesystem",
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "."],
"env": {
"SOME_VAR": "value"
},
"enabled": true,
"description": "Filesystem access for the agent"
}
"""
import os
import json
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class MCPConfig:
"""A single MCP server configuration."""
name: str
command: str
args: List[str] = field(default_factory=list)
env: Dict[str, str] = field(default_factory=dict)
enabled: bool = True
description: str = ""
file_path: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"command": self.command,
"args": self.args,
"env": self.env,
"enabled": self.enabled,
"description": self.description
}
@classmethod
def from_dict(cls, data: Dict[str, Any], file_path: Optional[str] = None) -> 'MCPConfig':
return cls(
name=data.get("name", ""),
command=data.get("command", ""),
args=data.get("args", []),
env=data.get("env", {}),
enabled=data.get("enabled", True),
description=data.get("description", ""),
file_path=file_path
)
def to_mcp_instance(self):
"""
Convert this config to an MCP instance.
Returns:
MCP instance ready for use with agents
"""
try:
from praisonaiagents.tools.mcp import MCP
# Resolve environment variables
resolved_env = {}
for key, value in self.env.items():
if value.startswith("$"):
# Environment variable reference
env_var = value[1:]
resolved_env[key] = os.environ.get(env_var, value)
else:
resolved_env[key] = value
return MCP(
command=self.command,
args=self.args,
env=resolved_env if resolved_env else None
)
except ImportError:
logger.warning("MCP tools not available")
return None
class MCPConfigManager:
"""
Manages MCP server configurations for AI agents.
Provides:
- Auto-discovery of MCP configs from .praison/mcp/
- Global configs from ~/.praison/mcp/
- Easy server management
"""
MCP_DIR_NAME = ".praison/mcp"
SUPPORTED_EXTENSIONS = [".json", ".mcpjson"]
def __init__(
self,
workspace_path: Optional[str] = None,
global_mcp_path: Optional[str] = None,
verbose: int = 0
):
"""
Initialize MCPConfigManager.
Args:
workspace_path: Path to workspace/project root
global_mcp_path: Path to global MCP configs (default: ~/.praison/mcp)
verbose: Verbosity level
"""
self.workspace_path = Path(workspace_path) if workspace_path else Path.cwd()
self.global_mcp_path = Path(global_mcp_path) if global_mcp_path else Path.home() / ".praison" / "mcp"
self.verbose = verbose
self._configs: Dict[str, MCPConfig] = {}
self._load_all_configs()
def _log(self, msg: str, level: int = logging.INFO):
"""Log message if verbose."""
if self.verbose >= 1:
logger.log(level, msg)
def _load_config_file(self, file_path: Path, scope: str = "workspace") -> Optional[MCPConfig]:
"""Load a single MCP config file."""
try:
content = file_path.read_text(encoding="utf-8")
data = json.loads(content)
# Use filename as name if not specified
if not data.get("name"):
data["name"] = file_path.stem
config = MCPConfig.from_dict(data, str(file_path))
return config
except json.JSONDecodeError as e:
self._log(f"Invalid JSON in {file_path}: {e}", logging.WARNING)
return None
except Exception as e:
self._log(f"Failed to load MCP config {file_path}: {e}", logging.WARNING)
return None
def _load_configs_from_dir(self, mcp_dir: Path, scope: str = "workspace"):
"""Load all MCP configs from a directory."""
if not mcp_dir.exists():
return
for ext in self.SUPPORTED_EXTENSIONS:
for file_path in mcp_dir.glob(f"*{ext}"):
if file_path.is_file():
config = self._load_config_file(file_path, scope)
if config:
key = f"{scope}:{config.name}"
self._configs[key] = config
self._log(f"Loaded MCP config: {key}")
def _load_all_configs(self):
"""Load all MCP configs from global and workspace directories."""
self._configs.clear()
# 1. Load global configs
if self.global_mcp_path.exists():
self._load_configs_from_dir(self.global_mcp_path, "global")
# 2. Load workspace configs (override global)
workspace_mcp_dir = self.workspace_path / self.MCP_DIR_NAME.replace("/", os.sep)
if workspace_mcp_dir.exists():
self._load_configs_from_dir(workspace_mcp_dir, "workspace")
self._log(f"Loaded {len(self._configs)} MCP configs total")
def reload(self):
"""Reload all configs from disk."""
self._load_all_configs()
def get_all_configs(self) -> List[MCPConfig]:
"""Get all loaded MCP configs."""
return list(self._configs.values())
def get_enabled_configs(self) -> List[MCPConfig]:
"""Get all enabled MCP configs."""
return [c for c in self._configs.values() if c.enabled]
def get_config(self, name: str) -> Optional[MCPConfig]:
"""Get a specific MCP config by name."""
# Try workspace first, then global
for scope in ["workspace", "global"]:
key = f"{scope}:{name}"
if key in self._configs:
return self._configs[key]
return None
def get_mcp_tools(self, names: Optional[List[str]] = None) -> List:
"""
Get MCP instances for use with agents.
Args:
names: Specific config names to load (None = all enabled)
Returns:
List of MCP instances
"""
tools = []
if names:
configs = [self.get_config(name) for name in names]
configs = [c for c in configs if c and c.enabled]
else:
configs = self.get_enabled_configs()
for config in configs:
mcp = config.to_mcp_instance()
if mcp:
tools.append(mcp)
return tools
def create_config(
self,
name: str,
command: str,
args: Optional[List[str]] = None,
env: Optional[Dict[str, str]] = None,
enabled: bool = True,
description: str = "",
scope: str = "workspace"
) -> MCPConfig:
"""
Create a new MCP config file.
Args:
name: Config name (used as filename)
command: Command to run (e.g., "npx")
args: Command arguments
env: Environment variables
enabled: Whether the config is enabled
description: Description of the MCP server
scope: Where to save (global or workspace)
Returns:
Created MCPConfig object
"""
# Determine save path
if scope == "global":
mcp_dir = self.global_mcp_path
else:
mcp_dir = self.workspace_path / self.MCP_DIR_NAME.replace("/", os.sep)
mcp_dir.mkdir(parents=True, exist_ok=True)
file_path = mcp_dir / f"{name}.json"
# Build config
config = MCPConfig(
name=name,
command=command,
args=args or [],
env=env or {},
enabled=enabled,
description=description,
file_path=str(file_path)
)
# Write file
file_path.write_text(
json.dumps(config.to_dict(), indent=2),
encoding="utf-8"
)
self._configs[f"{scope}:{name}"] = config
self._log(f"Created MCP config '{name}' at {file_path}")
return config
def delete_config(self, name: str, scope: Optional[str] = None) -> bool:
"""
Delete an MCP config file.
Args:
name: Config name to delete
scope: Scope to delete from (None = try both)
Returns:
True if deleted, False if not found
"""
scopes = [scope] if scope else ["workspace", "global"]
for s in scopes:
key = f"{s}:{name}"
if key in self._configs:
config = self._configs[key]
if config.file_path:
try:
Path(config.file_path).unlink()
del self._configs[key]
self._log(f"Deleted MCP config '{name}' from {s}")
return True
except Exception as e:
self._log(f"Failed to delete config file: {e}", logging.ERROR)
return False
def enable_config(self, name: str) -> bool:
"""Enable an MCP config."""
config = self.get_config(name)
if config:
config.enabled = True
self._save_config(config)
return True
return False
def disable_config(self, name: str) -> bool:
"""Disable an MCP config."""
config = self.get_config(name)
if config:
config.enabled = False
self._save_config(config)
return True
return False
def _save_config(self, config: MCPConfig):
"""Save a config back to its file."""
if config.file_path:
Path(config.file_path).write_text(
json.dumps(config.to_dict(), indent=2),
encoding="utf-8"
)
def list_configs(self) -> List[Dict[str, Any]]:
"""
List all MCP configs with metadata.
Returns:
List of config info dicts
"""
configs = []
for key, config in self._configs.items():
scope = key.split(":")[0]
configs.append({
"name": config.name,
"scope": scope,
"command": config.command,
"args": config.args,
"enabled": config.enabled,
"description": config.description,
"file_path": config.file_path
})
return configs
File diff suppressed because it is too large. Load diff
@@ -826,7 +826,34 @@ Provide a JSON with the structure:
logging.info("Hierarchical task execution finished")
def workflow(self):
"""Synchronous version of workflow method"""
"""Synchronous version of workflow method.
DEPRECATED: This method is deprecated. Use the Workflow class instead:
```python
from praisonaiagents import Workflow, StepResult
from praisonaiagents.workflows import route, parallel, loop, repeat
workflow = Workflow(steps=[step1, step2])
result = workflow.start("input")
```
The Workflow class provides a simpler API with more features:
- route() for decision-based branching
- parallel() for concurrent execution
- loop() for iteration over lists/CSV
- repeat() for evaluator-optimizer patterns
- Callbacks (on_step_start, on_step_complete, etc.)
- Guardrails with validation feedback
- Status tracking
"""
import warnings
warnings.warn(
"process='workflow' is deprecated. Use the Workflow class instead: "
"from praisonaiagents import Workflow; workflow = Workflow(steps=[...]); workflow.start()",
DeprecationWarning,
stacklevel=2
)
current_iter = 0 # Track how many times we've looped
# Build workflow relationships first
for task in self.tasks.values():
@@ -44,7 +44,9 @@ class Task:
retain_full_context: bool = False, # By default, only use previous task output, not all previous tasks
guardrail: Optional[Union[Callable[[TaskOutput], Tuple[bool, Any]], str]] = None,
max_retries: int = 3,
retry_count: int = 0
retry_count: int = 0,
agent_config: Optional[Dict[str, Any]] = None,
variables: Optional[Dict[str, Any]] = None
):
# Add check if memory config is provided
if memory is not None or (config and config.get('memory_config')):
@@ -89,6 +91,8 @@ class Task:
self.retry_count = retry_count
self._guardrail_fn = None
self.validation_feedback = None # Store validation failure feedback for retry attempts
self.agent_config = agent_config # Per-task agent configuration {role, goal, backstory, llm}
self.variables = variables if variables else {} # Variables for substitution in description
# Set logger level based on config verbose level
verbose = self.config.get("verbose", 0)
@@ -0,0 +1,319 @@
"""
Mentions Parser for PraisonAI Agents.
Provides @mention syntax support similar to Cursor IDE:
- @file:path/to/file.py - Include file content
- @web:query - Search the web
- @doc:name - Include doc from .praison/docs/
- @rule:name - Include specific rule
Usage:
from praisonaiagents.tools.mentions import MentionsParser
parser = MentionsParser()
context, cleaned_prompt = parser.process("@file:main.py explain this code")
"""
import re
import logging
from pathlib import Path
from typing import Dict, List, Optional, Tuple
logger = logging.getLogger(__name__)
class MentionsParser:
"""
Parse and process @mentions in prompts.
Supports:
- @file:path - Include file content
- @web:query - Web search (requires web search tool)
- @doc:name - Include doc from .praison/docs/
- @rule:name - Include specific rule
- @url:https://... - Fetch URL content
"""
# Patterns for different mention types
PATTERNS = {
"file": re.compile(r'@file:([^\s]+)'),
"web": re.compile(r'@web:([^\s]+(?:\s+[^\s@]+)*)'),
"doc": re.compile(r'@doc:([^\s]+)'),
"rule": re.compile(r'@rule:([^\s]+)'),
"url": re.compile(r'@url:(https?://[^\s]+)'),
}
def __init__(
self,
workspace_path: Optional[str] = None,
verbose: int = 0
):
"""
Initialize MentionsParser.
Args:
workspace_path: Path to workspace/project root
verbose: Verbosity level
"""
self.workspace_path = Path(workspace_path) if workspace_path else Path.cwd()
self.verbose = verbose
# Lazy-loaded managers
self._docs_manager = None
self._rules_manager = None
def _log(self, msg: str, level: int = logging.INFO):
"""Log message if verbose."""
if self.verbose >= 1:
logger.log(level, msg)
def _get_docs_manager(self):
"""Lazy load DocsManager."""
if self._docs_manager is None:
try:
from praisonaiagents.memory.docs_manager import DocsManager
self._docs_manager = DocsManager(workspace_path=str(self.workspace_path))
except ImportError:
self._log("DocsManager not available", logging.WARNING)
return self._docs_manager
def _get_rules_manager(self):
"""Lazy load RulesManager."""
if self._rules_manager is None:
try:
from praisonaiagents.memory.rules_manager import RulesManager
self._rules_manager = RulesManager(workspace_path=str(self.workspace_path))
except ImportError:
self._log("RulesManager not available", logging.WARNING)
return self._rules_manager
def process(self, prompt: str) -> Tuple[str, str]:
"""
Process a prompt and extract @mentions.
Args:
prompt: The input prompt with @mentions
Returns:
Tuple of (context_string, cleaned_prompt)
"""
context_parts = []
cleaned_prompt = prompt
# Process each mention type
for mention_type, pattern in self.PATTERNS.items():
matches = pattern.findall(prompt)
for match in matches:
context = self._process_mention(mention_type, match)
if context:
context_parts.append(context)
# Remove the mention from the prompt
cleaned_prompt = pattern.sub('', cleaned_prompt, count=1)
# Clean up extra whitespace
cleaned_prompt = ' '.join(cleaned_prompt.split())
# Build context string
context_string = ""
if context_parts:
context_string = "\n\n".join(context_parts) + "\n\n"
return context_string, cleaned_prompt
def _process_mention(self, mention_type: str, value: str) -> Optional[str]:
"""
Process a single mention.
Args:
mention_type: Type of mention (file, web, doc, rule, url)
value: The mention value
Returns:
Context string or None
"""
if mention_type == "file":
return self._process_file_mention(value)
elif mention_type == "web":
return self._process_web_mention(value)
elif mention_type == "doc":
return self._process_doc_mention(value)
elif mention_type == "rule":
return self._process_rule_mention(value)
elif mention_type == "url":
return self._process_url_mention(value)
return None
def _process_file_mention(self, file_path: str) -> Optional[str]:
"""Process @file:path mention."""
try:
# Resolve path relative to workspace
full_path = self.workspace_path / file_path
if not full_path.exists():
# Try as absolute path
full_path = Path(file_path)
if not full_path.exists():
self._log(f"File not found: {file_path}", logging.WARNING)
return f"# File: {file_path}\n[File not found]"
content = full_path.read_text(encoding="utf-8")
# Limit content size
if len(content) > 50000:
content = content[:50000] + "\n... (truncated)"
# Detect language for code fence
ext = full_path.suffix.lower()
lang_map = {
".py": "python",
".js": "javascript",
".ts": "typescript",
".jsx": "jsx",
".tsx": "tsx",
".java": "java",
".cpp": "cpp",
".c": "c",
".go": "go",
".rs": "rust",
".rb": "ruby",
".php": "php",
".swift": "swift",
".kt": "kotlin",
".md": "markdown",
".json": "json",
".yaml": "yaml",
".yml": "yaml",
".xml": "xml",
".html": "html",
".css": "css",
".sql": "sql",
".sh": "bash",
}
lang = lang_map.get(ext, "")
return f"# File: {file_path}\n```{lang}\n{content}\n```"
except Exception as e:
self._log(f"Error reading file {file_path}: {e}", logging.ERROR)
return f"# File: {file_path}\n[Error reading file: {e}]"
def _process_web_mention(self, query: str) -> Optional[str]:
"""Process @web:query mention."""
try:
# Try to use web search
from praisonaiagents.tools import duckduckgo_search
results = duckduckgo_search(query, max_results=3)
if results:
return f"# Web Search: {query}\n{results}"
else:
return f"# Web Search: {query}\n[No results found]"
except ImportError:
self._log("Web search not available", logging.WARNING)
return f"# Web Search: {query}\n[Web search not available. Install with: pip install duckduckgo-search]"
except Exception as e:
self._log(f"Error in web search: {e}", logging.ERROR)
return f"# Web Search: {query}\n[Error: {e}]"
def _process_doc_mention(self, doc_name: str) -> Optional[str]:
"""Process @doc:name mention."""
docs = self._get_docs_manager()
if not docs:
return f"# Doc: {doc_name}\n[DocsManager not available]"
doc = docs.get_doc(doc_name)
if doc:
return f"# Doc: {doc.name}\n{doc.content}"
else:
return f"# Doc: {doc_name}\n[Doc not found]"
def _process_rule_mention(self, rule_name: str) -> Optional[str]:
"""Process @rule:name mention."""
rules = self._get_rules_manager()
if not rules:
return f"# Rule: {rule_name}\n[RulesManager not available]"
rule = rules.get_rule(rule_name)
if rule:
return f"# Rule: {rule.name}\n{rule.content}"
else:
return f"# Rule: {rule_name}\n[Rule not found]"
def _process_url_mention(self, url: str) -> Optional[str]:
"""Process @url:https://... mention."""
try:
import urllib.request
req = urllib.request.Request(
url,
headers={'User-Agent': 'Mozilla/5.0 (compatible; PraisonAI/1.0)'}
)
with urllib.request.urlopen(req, timeout=10) as response:
content = response.read().decode('utf-8', errors='ignore')
# Limit content size
if len(content) > 50000:
content = content[:50000] + "\n... (truncated)"
# Try to extract text from HTML
if '<html' in content.lower():
# Simple HTML text extraction
import re
# Remove scripts and styles
content = re.sub(r'<script[^>]*>.*?</script>', '', content, flags=re.DOTALL | re.IGNORECASE)
content = re.sub(r'<style[^>]*>.*?</style>', '', content, flags=re.DOTALL | re.IGNORECASE)
# Remove tags
content = re.sub(r'<[^>]+>', ' ', content)
# Clean whitespace
content = ' '.join(content.split())
return f"# URL: {url}\n{content[:10000]}"
except Exception as e:
self._log(f"Error fetching URL {url}: {e}", logging.ERROR)
return f"# URL: {url}\n[Error fetching URL: {e}]"
def extract_mentions(self, prompt: str) -> Dict[str, List[str]]:
"""
Extract all mentions from a prompt without processing them.
Args:
prompt: The input prompt
Returns:
Dict mapping mention types to lists of values
"""
mentions = {}
for mention_type, pattern in self.PATTERNS.items():
matches = pattern.findall(prompt)
if matches:
mentions[mention_type] = matches
return mentions
def has_mentions(self, prompt: str) -> bool:
"""Check if a prompt contains any @mentions."""
for pattern in self.PATTERNS.values():
if pattern.search(prompt):
return True
return False
def process_mentions(prompt: str, workspace_path: Optional[str] = None) -> Tuple[str, str]:
"""
Convenience function to process mentions in a prompt.
Args:
prompt: The input prompt with @mentions
workspace_path: Optional workspace path
Returns:
Tuple of (context_string, cleaned_prompt)
"""
parser = MentionsParser(workspace_path=workspace_path)
return parser.process(prompt)
# Export
__all__ = ["MentionsParser", "process_mentions"]
@@ -0,0 +1,265 @@
"""
Rules Tools for PraisonAI Agents.
Provides tools for agents to create, read, and manage rules dynamically.
This enables Agent-Requested rules similar to Cursor's agent-requested rules feature.
Usage:
from praisonaiagents.tools.rules_tools import (
create_rule_tool,
list_rules_tool,
get_rule_tool,
delete_rule_tool
)
agent = Agent(
name="Assistant",
tools=[create_rule_tool, list_rules_tool, get_rule_tool]
)
"""
import logging
from typing import Optional
logger = logging.getLogger(__name__)
# Singleton RulesManager instance
_rules_manager = None
def _get_rules_manager():
"""Get or create the RulesManager singleton."""
global _rules_manager
if _rules_manager is None:
from praisonaiagents.memory.rules_manager import RulesManager
_rules_manager = RulesManager()
return _rules_manager
def create_rule_tool(
name: str,
content: str,
description: str = "",
globs: Optional[str] = None,
activation: str = "always",
priority: int = 0,
scope: str = "workspace"
) -> str:
"""
Create a new rule for the AI agent to follow.
This tool allows the agent to create persistent rules that will be
applied to future conversations and tasks.
Args:
name: Rule name (used as filename, no spaces or special chars)
content: The rule content in markdown format
description: Short description of what the rule does
globs: Comma-separated glob patterns for file matching (e.g., "**/*.py,**/*.js")
activation: When to apply the rule:
- "always": Always apply this rule
- "glob": Apply when working with files matching globs
- "manual": Only apply when explicitly mentioned
- "ai_decision": Let AI decide when to apply
priority: Priority (higher = applied first, default 0)
scope: Where to save ("workspace" or "global")
Returns:
Success message with rule details
Example:
create_rule_tool(
name="python-style",
content="- Use type hints for all functions\\n- Follow PEP 8",
description="Python coding standards",
globs="**/*.py",
activation="glob"
)
"""
try:
manager = _get_rules_manager()
# Parse globs
glob_list = None
if globs:
glob_list = [g.strip() for g in globs.split(",") if g.strip()]
# Validate activation
valid_activations = ["always", "glob", "manual", "ai_decision"]
if activation not in valid_activations:
return f"Error: activation must be one of {valid_activations}"
# Validate scope
if scope not in ["workspace", "global"]:
return "Error: scope must be 'workspace' or 'global'"
# Create the rule
rule = manager.create_rule(
name=name,
content=content,
description=description,
globs=glob_list,
activation=activation,
priority=priority,
scope=scope
)
return f"""Successfully created rule '{name}'
Location: {rule.file_path}
Activation: {activation}
Priority: {priority}
Description: {description or 'None'}
Globs: {glob_list or 'None'}
The rule will be applied according to its activation mode."""
except Exception as e:
logger.error(f"Error creating rule: {e}")
return f"Error creating rule: {str(e)}"
def list_rules_tool() -> str:
"""
List all available rules.
Returns a formatted list of all rules with their metadata.
Returns:
Formatted string with rule information
"""
try:
manager = _get_rules_manager()
rules = manager.get_all_rules()
if not rules:
return "No rules found. Use create_rule_tool to create new rules."
lines = ["# Available Rules\n"]
for rule in rules:
scope = "global" if "global:" in (rule.file_path or "") else "workspace"
lines.append(f"## {rule.name}")
lines.append(f"- **Activation**: {rule.activation}")
lines.append(f"- **Priority**: {rule.priority}")
if rule.description:
lines.append(f"- **Description**: {rule.description}")
if rule.globs:
lines.append(f"- **Globs**: {', '.join(rule.globs)}")
lines.append(f"- **Scope**: {scope}")
lines.append("")
return "\n".join(lines)
except Exception as e:
logger.error(f"Error listing rules: {e}")
return f"Error listing rules: {str(e)}"
def get_rule_tool(name: str) -> str:
"""
Get the content of a specific rule.
Args:
name: The name of the rule to retrieve
Returns:
The rule content and metadata
"""
try:
manager = _get_rules_manager()
rule = manager.get_rule(name)
if not rule:
return f"Rule '{name}' not found. Use list_rules_tool to see available rules."
return f"""# Rule: {rule.name}
**Description**: {rule.description or 'None'}
**Activation**: {rule.activation}
**Priority**: {rule.priority}
**Globs**: {', '.join(rule.globs) if rule.globs else 'None'}
**File**: {rule.file_path}
## Content:
{rule.content}
"""
except Exception as e:
logger.error(f"Error getting rule: {e}")
return f"Error getting rule: {str(e)}"
def delete_rule_tool(name: str, scope: Optional[str] = None) -> str:
"""
Delete a rule.
Args:
name: The name of the rule to delete
scope: Optional scope to delete from ("workspace" or "global")
Returns:
Success or error message
"""
try:
manager = _get_rules_manager()
if scope and scope not in ["workspace", "global"]:
return "Error: scope must be 'workspace' or 'global'"
success = manager.delete_rule(name, scope)
if success:
return f"Successfully deleted rule '{name}'"
else:
return f"Rule '{name}' not found or could not be deleted"
except Exception as e:
logger.error(f"Error deleting rule: {e}")
return f"Error deleting rule: {str(e)}"
def get_active_rules_tool(file_path: Optional[str] = None) -> str:
"""
Get all rules that are currently active for a given file or context.
Args:
file_path: Optional file path to check which rules apply
Returns:
Formatted string with active rules
"""
try:
manager = _get_rules_manager()
if file_path:
rules = manager.get_rules_for_file(file_path)
else:
rules = manager.get_active_rules()
if not rules:
return "No active rules for this context."
lines = ["# Active Rules\n"]
for rule in rules:
lines.append(f"## {rule.name}")
if rule.description:
lines.append(f"*{rule.description}*\n")
lines.append(rule.content)
lines.append("")
return "\n".join(lines)
except Exception as e:
logger.error(f"Error getting active rules: {e}")
return f"Error getting active rules: {str(e)}"
# Export all tools
__all__ = [
"create_rule_tool",
"list_rules_tool",
"get_rule_tool",
"delete_rule_tool",
"get_active_rules_tool"
]
@@ -0,0 +1,55 @@
"""
Workflows module for PraisonAI Agents.
Provides workflow/pipeline patterns for orchestrating agents and functions.
"""
from .workflows import (
# Core classes
Workflow,
Pipeline, # Alias for Workflow
WorkflowStep,
WorkflowContext,
StepResult,
WorkflowManager,
# Pattern helpers
Route,
Parallel,
Loop,
Repeat,
# Convenience functions
route,
parallel,
loop,
repeat,
# Backward compatibility aliases
StepInput,
StepOutput,
)
__all__ = [
# Core
"Workflow",
"Pipeline",
"WorkflowStep",
"WorkflowContext",
"StepResult",
"WorkflowManager",
# Patterns
"Route",
"Parallel",
"Loop",
"Repeat",
"route",
"parallel",
"loop",
"repeat",
# Aliases
"StepInput",
"StepOutput",
]
File diff suppressed because it is too large. Load diff
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.0.192"
version = "0.1.1"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
requires-python = ">=3.10"
authors = [
File diff suppressed because it is too large. Load diff
+1 -1
View File
@@ -2422,7 +2422,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.0.192"
version = "0.1.1"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
+285 -11
View File
@@ -97,6 +97,26 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
| 🪝 Hooks | [Example](#9-hooks) | [📖](https://docs.praison.ai/features/hooks) |
| 📈 Telemetry | [Example](examples/python/telemetry/production-telemetry-example.py) | [📖](https://docs.praison.ai/features/telemetry) |
| 📹 Camera Integration | [Example](examples/python/camera/) | [📖](https://docs.praison.ai/features/camera-integration) |
| 📄 Project Docs (.praison/docs/) | [Example](#docs-cli) | [📖](https://docs.praison.ai/cli/docs) |
| 🔌 MCP Config Management | [Example](#mcp-config-cli) | [📖](https://docs.praison.ai/cli/mcp) |
| 💬 AI Commit Messages | [Example](#ai-commit-cli) | [📖](https://docs.praison.ai/cli/commit) |
| @ @Mentions in Prompts | [Example](#mentions-in-prompts) | [📖](https://docs.praison.ai/cli/mentions) |
| 💾 Auto-Save Sessions | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
| 📜 History in Context | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
| ⏸️ Workflow Checkpoints | [Example](examples/python/workflows/workflow_checkpoints.py) | [📖](https://docs.praison.ai/cli/session) |
| 🔀 Workflow Branching | [Example](examples/python/workflows/workflow_branching.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🔁 Workflow Loops | [Example](examples/python/workflows/workflow_loops.py) | [📖](https://docs.praison.ai/features/workflows) |
| 📞 Task Callbacks | [Example](examples/python/workflows/task_callbacks.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🛑 Workflow Early Stop | [Example](examples/python/workflows/workflow_early_stop.py) | [📖](https://docs.praison.ai/features/workflows) |
| ⚡ Simple Workflow | [Example](examples/python/workflows/simple_workflow.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🤖 Workflow with Agents | [Example](examples/python/workflows/workflow_with_agents.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🔄 Conditional Steps | [Example](examples/python/workflows/workflow_conditional.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🎭 Mixed Steps | [Example](examples/python/workflows/workflow_mixed_steps.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🔀 Agentic Routing | [Example](examples/python/workflows/workflow_routing.py) | [📖](https://docs.praison.ai/features/workflow-routing) |
| ⚡ Parallel Execution | [Example](examples/python/workflows/workflow_parallel.py) | [📖](https://docs.praison.ai/features/workflow-parallel) |
| 🔁 Loop over List | [Example](examples/python/workflows/workflow_loop_list.py) | [📖](https://docs.praison.ai/features/workflow-loop) |
| 📄 Loop over CSV | [Example](examples/python/workflows/workflow_loop_csv.py) | [📖](https://docs.praison.ai/features/workflow-loop) |
| 🔄 Evaluator-Optimizer | [Example](examples/python/workflows/workflow_repeat.py) | [📖](https://docs.praison.ai/features/workflow-repeat) |
## Supported Providers
@@ -438,6 +458,127 @@ result = asyncio.run(manager.aexecute("deploy", default_llm="gpt-4o-mini"))
- **Per-Step Agents**: Configure different agents with roles, goals, tools for each step
- **Async Execution**: Use `aexecute()` for async workflows
- **Planning Mode**: Enable at workflow level with `planning=True`
- **Branching**: Use `next_steps` and `branch_condition` for conditional routing
- **Loops**: Use `loop_over` and `loop_var` to iterate over data
### Choosing the Right Workflow System
| Use Case | Recommended |
|----------|-------------|
| Simple function pipelines | `Workflow` class ⭐ |
| Agent-only pipelines | `Workflow` class |
| CSV batch processing | `process="workflow"` |
| Complex task routing | `process="workflow"` |
| Markdown templates | `WorkflowManager` |
| Early stop / conditional | `Workflow` class |
### Simple Workflow (Recommended)
The easiest way to create workflows - just pass functions as steps:
```python
from praisonaiagents import Workflow, WorkflowContext, StepResult
# Define simple handler functions
def validate(ctx: WorkflowContext) -> StepResult:
if not ctx.input:
return StepResult(output="No input", stop_workflow=True)
return StepResult(output=f"Valid: {ctx.input}")
def process(ctx: WorkflowContext) -> StepResult:
return StepResult(output=f"Processed: {ctx.previous_result}")
# Create and run workflow
workflow = Workflow(steps=[validate, process])
result = workflow.start("Hello World", verbose=True)
print(result["output"]) # "Processed: Valid: Hello World"
```
**Key Features:**
- **Just pass functions** - No complex configuration needed
- **Early stop** - Return `stop_workflow=True` to stop the workflow
- **Context passing** - Access `ctx.input`, `ctx.previous_result`, `ctx.variables`
- **Verbose mode** - See step-by-step progress
### Workflow Branching & Loops
```python
from praisonaiagents import WorkflowStep
# Branching step
decision_step = WorkflowStep(
name="decide",
action="Evaluate if task is complete",
next_steps=["success_step", "retry_step"],
branch_condition={"success": ["success_step"], "failure": ["retry_step"]}
)
# Loop step
loop_step = WorkflowStep(
name="process_items",
action="Process {{item}}",
loop_over="items", # Variable containing list
loop_var="item" # Current item variable name
)
```
### Workflow Patterns (route, parallel, loop, repeat)
```python
from praisonaiagents import Workflow, WorkflowContext, StepResult
# Or use Pipeline (alias for Workflow)
from praisonaiagents import Pipeline
from praisonaiagents.workflows import route, parallel, loop, repeat
# 1. ROUTING - Decision-based branching
workflow = Workflow(steps=[
classify_request, # Returns "approve" or "reject"
route({
"approve": [approve_handler],
"reject": [reject_handler],
"default": [fallback_handler]
})
])
# 2. PARALLEL - Concurrent execution
workflow = Workflow(steps=[
parallel([research_market, research_competitors, research_customers]),
summarize_results # Gets all parallel outputs
])
# 3. LOOP - Iterate over list or CSV
workflow = Workflow(
steps=[loop(process_item, over="items")],
variables={"items": ["a", "b", "c"]}
)
# Or from CSV file:
workflow = Workflow(steps=[loop(process_row, from_csv="data.csv")])
# 4. REPEAT - Evaluator-Optimizer pattern
workflow = Workflow(steps=[
repeat(
generator,
until=lambda ctx: "done" in ctx.previous_result,
max_iterations=5
)
])
# 5. CALLBACKS - Monitor workflow execution
workflow = Workflow(
steps=[step1, step2],
on_workflow_start=lambda w, i: print(f"Starting: {i}"),
on_step_complete=lambda name, r: print(f"{name}: {r.output[:50]}"),
on_workflow_complete=lambda w, r: print(f"Done: {r['status']}")
)
# 6. GUARDRAILS - Validate and retry
def validate(result):
return ("error" not in result.output, "Fix the error")
workflow = Workflow(steps=[
WorkflowStep(name="gen", handler=generator, guardrail=validate, max_retries=3)
])
```
### 9. Hooks
@@ -573,19 +714,49 @@ praisonai "Task" --include-rules security,testing
# List available workflows
praisonai workflow list
# Execute a workflow
praisonai workflow run deploy
# Execute a workflow with tools and save output
praisonai workflow run "Research Blog" --tools tavily --save
# Execute with variables
praisonai workflow run deploy --workflow-var environment=staging --workflow-var branch=main
# Execute with planning mode (AI creates sub-steps for each workflow step)
praisonai workflow run "Research Blog" --planning --verbose
# Execute with reasoning mode (chain-of-thought)
praisonai workflow run "Analysis" --reasoning --verbose
# Execute with memory enabled
praisonai workflow run "Research" --memory
# Show workflow details
praisonai workflow show deploy
# Create a new workflow template
praisonai workflow create my_workflow
# Inline workflow (no template file needed)
praisonai "What is AI?" --workflow "Research,Summarize" --save
# Inline workflow with step actions
praisonai "GPT-5" --workflow "Research:Search for info,Write:Write blog" --tools tavily
# Workflow CLI help
praisonai workflow help
```
**Workflow CLI Options:**
| Flag | Description |
|------|-------------|
| `--workflow-var key=value` | Set workflow variable (can be repeated) |
| `--llm <model>` | LLM model (e.g., openai/gpt-4o-mini) |
| `--tools <tools>` | Tools (comma-separated, e.g., tavily) |
| `--planning` | Enable planning mode |
| `--reasoning` | Enable reasoning mode |
| `--memory` | Enable memory |
| `--verbose` | Enable verbose output |
| `--save` | Save output to file |
### Hooks CLI:
```bash
# List configured hooks
@@ -685,23 +856,59 @@ praisonai knowledge help
### Session CLI:
```bash
# Start a new session
praisonai session start my-project
# List all sessions
# List all saved sessions
praisonai session list
# Resume a session
praisonai session resume my-project
# Show session details
praisonai session show my-project
# Resume a session (load into memory)
praisonai session resume my-project
# Delete a session
praisonai session delete my-project
# Show all commands
praisonai session help
# Auto-save session after each run
praisonai "Analyze this code" --auto-save my-project
# Load history from last N sessions into context
praisonai "Continue our discussion" --history 5
```
### Session Management (Python):
```python
from praisonaiagents import Agent
# Auto-save session after each run
agent = Agent(
name="Assistant",
memory=True,
auto_save="my-project"
)
# Load history from last 5 sessions
agent = Agent(
name="Assistant",
memory=True,
history_in_context=5
)
```
### Workflow Checkpoints:
```python
from praisonaiagents.memory.workflows import WorkflowManager
manager = WorkflowManager()
# Save checkpoint after each step
result = manager.execute("deploy", checkpoint="deploy-v1")
# Resume from checkpoint
result = manager.execute("deploy", resume="deploy-v1")
# List/delete checkpoints
manager.list_checkpoints()
manager.delete_checkpoint("deploy-v1")
```
### Tools CLI:
@@ -783,6 +990,73 @@ praisonai agents.yaml --flow-display
praisonai "Multi-step task" --planning --flow-display
```
### Docs CLI:
```bash
# List all project docs
praisonai docs list
# Create a new doc
praisonai docs create project-overview "This project is a Python web app..."
# Show a specific doc
praisonai docs show project-overview
# Delete a doc
praisonai docs delete old-doc
# Show all commands
praisonai docs help
```
### MCP Config CLI:
```bash
# List all MCP configurations
praisonai mcp list
# Create a new MCP config
praisonai mcp create filesystem npx -y @modelcontextprotocol/server-filesystem .
# Show a specific config
praisonai mcp show filesystem
# Enable/disable a config
praisonai mcp enable filesystem
praisonai mcp disable filesystem
# Delete a config
praisonai mcp delete filesystem
# Show all commands
praisonai mcp help
```
### AI Commit CLI:
```bash
# Generate AI commit message for staged changes
praisonai commit
# Generate, commit, and push
praisonai commit --push
```
### @Mentions in Prompts:
```bash
# Include file content in prompt
praisonai "@file:src/main.py explain this code"
# Include project doc
praisonai "@doc:project-overview help me add a feature"
# Search the web
praisonai "@web:python best practices give me tips"
# Fetch URL content
praisonai "@url:https://docs.python.org summarize this"
# Combine multiple mentions
praisonai "@file:main.py @doc:coding-standards review this code"
```
## Prompt Expansion
Expand short prompts into detailed, actionable prompts:
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.14.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.14.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.23.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.23.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+128 -144
View File
@@ -2,12 +2,12 @@
Session Handler for CLI.
Provides session management for multi-turn conversations.
Usage: praisonai session start "project-x"
praisonai session list
praisonai session resume "project-x"
Usage: praisonai session list
praisonai session show <name>
praisonai session resume <name>
praisonai session delete <name>
"""
import os
from typing import Any, Dict, List
from .base import CommandHandler
@@ -16,151 +16,175 @@ class SessionHandler(CommandHandler):
"""
Handler for session command.
Manages conversation sessions for multi-turn interactions.
Manages conversation sessions using FileMemory.
Example:
praisonai session start my-project
praisonai session list
praisonai session show my-project
praisonai session resume my-project
praisonai session delete my-project
"""
def __init__(self, verbose: bool = False, workspace: str = None):
def __init__(self, verbose: bool = False, user_id: str = "praison"):
super().__init__(verbose)
self.workspace = workspace or os.path.join(os.path.expanduser("~"), ".praison", "sessions")
os.makedirs(self.workspace, exist_ok=True)
self._session = None
self.user_id = user_id
self._memory = None
@property
def feature_name(self) -> str:
return "session"
def get_actions(self) -> List[str]:
return ["start", "list", "resume", "delete", "info", "help"]
return ["list", "show", "resume", "delete", "help"]
def get_help_text(self) -> str:
return """
Session Commands:
praisonai session start <name> - Start a new session
praisonai session list - List all sessions
praisonai session resume <name> - Resume an existing session
praisonai session list - List all saved sessions
praisonai session show <name> - Show session details
praisonai session resume <name> - Resume a session (load into memory)
praisonai session delete <name> - Delete a session
praisonai session info <name> - Show session details
Sessions enable multi-turn conversations with context preservation.
Sessions store conversation history for multi-turn interactions.
Use --auto-save <name> flag when running agents to auto-save sessions.
"""
def _get_session_class(self):
"""Get Session class lazily."""
try:
from praisonaiagents import Session
return Session
except ImportError:
self.print_status(
"Session requires praisonaiagents. Install with: pip install praisonaiagents",
"error"
)
return None
def _get_memory(self):
"""Get FileMemory instance lazily."""
if self._memory is None:
try:
from praisonaiagents.memory.file_memory import FileMemory
self._memory = FileMemory(user_id=self.user_id)
except ImportError:
self.print_status(
"Session requires praisonaiagents. Install with: pip install praisonaiagents",
"error"
)
return None
return self._memory
def _get_session_path(self, name: str) -> str:
"""Get path for a session file."""
return os.path.join(self.workspace, f"{name}.json")
def action_start(self, args: List[str], **kwargs) -> Any:
def action_list(self, args: List[str], **kwargs) -> List[Dict[str, Any]]:
"""
Start a new session.
Args:
args: List containing session name
Returns:
Session instance or None
"""
if not args:
self.print_status("Usage: praisonai session start <name>", "error")
return None
Session = self._get_session_class()
if not Session:
return None
name = args[0]
session_path = self._get_session_path(name)
if os.path.exists(session_path):
self.print_status(f"Session '{name}' already exists. Use 'resume' to continue.", "warning")
return None
try:
session = Session(session_id=name)
self._session = session
self.print_status(f"✅ Session '{name}' started", "success")
return session
except Exception as e:
self.print_status(f"Failed to start session: {e}", "error")
return None
def action_list(self, args: List[str], **kwargs) -> List[str]:
"""
List all sessions.
List all saved sessions.
Returns:
List of session names
List of session info dicts
"""
sessions = []
memory = self._get_memory()
if not memory:
return []
try:
for file in os.listdir(self.workspace):
if file.endswith('.json'):
sessions.append(file[:-5]) # Remove .json extension
sessions = memory.list_sessions()
if sessions:
self.print_status("\n📋 Available Sessions:", "info")
for name in sessions:
self.print_status(f" - {name}", "info")
from rich.table import Table
from rich.console import Console
console = Console()
table = Table(title="Saved Sessions")
table.add_column("Name", style="cyan")
table.add_column("Saved At", style="green")
table.add_column("Messages", style="yellow")
table.add_column("Has History", style="magenta")
for s in sessions:
table.add_row(
s.get("name", ""),
s.get("saved_at", "")[:19] if s.get("saved_at") else "",
str(s.get("short_term_count", 0)),
"✅" if s.get("has_conversation") else "❌"
)
console.print(table)
else:
self.print_status("No sessions found", "warning")
self.print_status("No sessions found. Use --auto-save <name> when running agents.", "warning")
return sessions
except Exception as e:
self.print_status(f"Failed to list sessions: {e}", "error")
return []
def action_resume(self, args: List[str], **kwargs) -> Any:
def action_show(self, args: List[str], **kwargs) -> Dict[str, Any]:
"""
Resume an existing session.
Show session details.
Args:
args: List containing session name
Returns:
Session instance or None
Session data dict
"""
if not args:
self.print_status("Usage: praisonai session show <name>", "error")
return {}
memory = self._get_memory()
if not memory:
return {}
name = args[0]
try:
session_data = memory.resume_session(name)
print(f"\n📊 Session: {name}")
print(f"Saved at: {session_data.get('saved_at_iso', 'Unknown')}")
print(f"User ID: {session_data.get('user_id', 'Unknown')}")
print(f"Short-term items: {len(session_data.get('short_term', []))}")
print(f"Conversation messages: {len(session_data.get('conversation_history', []))}")
# Show last few messages
history = session_data.get('conversation_history', [])
if history:
print(f"\nLast {min(5, len(history))} messages:")
for msg in history[-5:]:
role = msg.get('role', 'unknown')
content = msg.get('content', '')[:100]
if len(msg.get('content', '')) > 100:
content += "..."
print(f" [{role}]: {content}")
return session_data
except FileNotFoundError:
self.print_status(f"Session '{name}' not found", "error")
return {}
except Exception as e:
self.print_status(f"Failed to show session: {e}", "error")
return {}
def action_resume(self, args: List[str], **kwargs) -> Dict[str, Any]:
"""
Resume a session (load into memory).
Args:
args: List containing session name
Returns:
Session data dict
"""
if not args:
self.print_status("Usage: praisonai session resume <name>", "error")
return None
return {}
Session = self._get_session_class()
if not Session:
return None
memory = self._get_memory()
if not memory:
return {}
name = args[0]
session_path = self._get_session_path(name)
if not os.path.exists(session_path):
self.print_status(f"Session '{name}' not found", "error")
return None
try:
session = Session(session_id=name)
if hasattr(session, 'load'):
session.load(session_path)
self._session = session
session_data = memory.resume_session(name)
self.print_status(f"✅ Session '{name}' resumed", "success")
return session
print(f" Loaded {len(session_data.get('short_term', []))} memory items")
print(f" Loaded {len(session_data.get('conversation_history', []))} conversation messages")
return session_data
except FileNotFoundError:
self.print_status(f"Session '{name}' not found", "error")
return {}
except Exception as e:
self.print_status(f"Failed to resume session: {e}", "error")
return None
return {}
def action_delete(self, args: List[str], **kwargs) -> bool:
"""
@@ -176,63 +200,23 @@ Sessions enable multi-turn conversations with context preservation.
self.print_status("Usage: praisonai session delete <name>", "error")
return False
name = args[0]
session_path = self._get_session_path(name)
if not os.path.exists(session_path):
self.print_status(f"Session '{name}' not found", "error")
memory = self._get_memory()
if not memory:
return False
name = args[0]
try:
os.remove(session_path)
self.print_status(f"✅ Session '{name}' deleted", "success")
return True
if memory.delete_session(name):
self.print_status(f"✅ Session '{name}' deleted", "success")
return True
else:
self.print_status(f"Session '{name}' not found", "error")
return False
except Exception as e:
self.print_status(f"Failed to delete session: {e}", "error")
return False
def action_info(self, args: List[str], **kwargs) -> Dict[str, Any]:
"""
Show session info.
Args:
args: List containing session name
Returns:
Dictionary of session info
"""
if not args:
self.print_status("Usage: praisonai session info <name>", "error")
return {}
name = args[0]
session_path = self._get_session_path(name)
if not os.path.exists(session_path):
self.print_status(f"Session '{name}' not found", "error")
return {}
try:
import json
with open(session_path, 'r') as f:
data = json.load(f)
info = {
"name": name,
"path": session_path,
"messages": len(data.get('messages', [])),
"created": data.get('created_at', 'Unknown')
}
self.print_status(f"\n📊 Session '{name}' Info:", "info")
for key, value in info.items():
self.print_status(f" {key}: {value}", "info")
return info
except Exception as e:
self.print_status(f"Failed to get session info: {e}", "error")
return {}
def execute(self, action: str, action_args: List[str], **kwargs) -> Any:
"""Execute session command action."""
return super().execute(action, action_args, **kwargs)
+648 -17
View File
@@ -549,7 +549,7 @@ class PraisonAI:
return default_args
# Define special commands
special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo']
special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit']
parser = argparse.ArgumentParser(prog="praisonai", description="praisonAI command-line interface")
parser.add_argument("--framework", choices=["crewai", "autogen", "praisonai"], help="Specify the framework")
@@ -600,10 +600,15 @@ class PraisonAI:
parser.add_argument("--memory", action="store_true", help="Enable file-based memory for agent")
parser.add_argument("--user-id", type=str, help="User ID for memory isolation")
# Session management arguments
parser.add_argument("--auto-save", type=str, metavar="NAME", help="Auto-save session with given name after each run")
parser.add_argument("--history", type=int, metavar="N", help="Load history from last N sessions into context")
# Rules arguments
parser.add_argument("--include-rules", type=str, help="Include manual rules by name (comma-separated)")
# Workflow arguments
# Workflow arguments (uses global --memory, --save, --verbose, --planning flags)
parser.add_argument("--workflow", type=str, help="Run inline workflow steps (format: 'step1:action1,step2:action2')")
parser.add_argument("--workflow-var", action="append", help="Workflow variable in key=value format (can be used multiple times)")
# Claude Memory Tool arguments
@@ -831,7 +836,7 @@ class PraisonAI:
if '=' in var:
key, value = var.split('=', 1)
workflow_vars[key] = value
self.handle_workflow_command(action, action_args, workflow_vars)
self.handle_workflow_command(action, action_args, workflow_vars, args)
sys.exit(0)
elif args.command == 'hooks':
@@ -893,6 +898,39 @@ class PraisonAI:
self.handle_todo_command(action, action_args)
sys.exit(0)
elif args.command == 'docs':
if not PRAISONAI_AVAILABLE:
print("[red]ERROR: PraisonAI Agents is not installed. Install with:[/red]")
print("\npip install praisonaiagents\n")
sys.exit(1)
# Get action and arguments from remaining args
action = unknown_args[0] if unknown_args else 'list'
action_args = unknown_args[1:] if len(unknown_args) > 1 else []
self.handle_docs_command(action, action_args)
sys.exit(0)
elif args.command == 'mcp':
if not PRAISONAI_AVAILABLE:
print("[red]ERROR: PraisonAI Agents is not installed. Install with:[/red]")
print("\npip install praisonaiagents\n")
sys.exit(1)
# Get action and arguments from remaining args
action = unknown_args[0] if unknown_args else 'list'
action_args = unknown_args[1:] if len(unknown_args) > 1 else []
self.handle_mcp_command(action, action_args)
sys.exit(0)
elif args.command == 'commit':
if not PRAISONAI_AVAILABLE:
print("[red]ERROR: PraisonAI Agents is not installed. Install with:[/red]")
print("\npip install praisonaiagents\n")
sys.exit(1)
self.handle_commit_command(unknown_args)
sys.exit(0)
# Only check framework availability for agent-related operations
if not args.command and (args.init or args.auto or args.framework):
if not CREWAI_AVAILABLE and not AUTOGEN_AVAILABLE and not PRAISONAI_AVAILABLE:
@@ -1403,7 +1441,7 @@ class PraisonAI:
except Exception as e:
print(f"[red]ERROR: Rules command failed: {e}[/red]")
def handle_workflow_command(self, action: str, action_args: list, variables: dict = None):
def handle_workflow_command(self, action: str, action_args: list, variables: dict = None, args=None):
"""
Handle workflow subcommand actions.
@@ -1411,6 +1449,7 @@ class PraisonAI:
action: The workflow action (list, run, create, show)
action_args: Additional arguments for the action
variables: Workflow variables for substitution
args: Parsed command line arguments
"""
try:
from praisonaiagents.memory import WorkflowManager
@@ -1446,27 +1485,90 @@ class PraisonAI:
return
workflow_name = action_args[0]
# Create a simple executor using PraisonAgent
def executor(prompt):
agent = PraisonAgent(
name="WorkflowExecutor",
role="Task Executor",
goal="Execute workflow steps"
)
return agent.chat(prompt)
# Use global flags (--llm, --tools, --planning, --memory, --save, --verbose)
workflow_llm = getattr(args, 'llm', None) if args else None
workflow_tools_str = getattr(args, 'tools', None) if args else None
workflow_planning = getattr(args, 'planning', False) if args else False
workflow_verbose = getattr(args, 'verbose', False) if args else False
workflow_memory = getattr(args, 'memory', False) if args else False
workflow_save = getattr(args, 'save', False) if args else False
# Load tools if specified
workflow_tools = None
if workflow_tools_str:
workflow_tools = self._load_tools(workflow_tools_str)
if workflow_tools:
print(f"[cyan]Loaded {len(workflow_tools)} tool(s) for workflow[/cyan]")
# Initialize memory if enabled
memory = None
if workflow_memory:
try:
from praisonaiagents.memory import Memory
memory = Memory()
print("[cyan]Memory enabled for workflow[/cyan]")
except ImportError:
print("[yellow]Warning: Memory not available[/yellow]")
# Create default agent for steps without agent_config
default_agent = PraisonAgent(
name="WorkflowExecutor",
role="Task Executor",
goal="Execute workflow steps",
llm=workflow_llm,
tools=workflow_tools,
verbose=1 if workflow_verbose else 0
)
print(f"[bold cyan]Running workflow: {workflow_name}[/bold cyan]")
if workflow_planning:
print("[cyan]Planning mode enabled[/cyan]")
result = manager.execute(
workflow_name,
executor=executor,
variables=variables or {}
default_agent=default_agent,
default_llm=workflow_llm,
variables=variables or {},
memory=memory,
planning=workflow_planning,
verbose=1 if workflow_verbose else 0,
on_step=lambda step, i: print(f"[cyan] → Step {i+1}: {step.name}[/cyan]"),
on_result=lambda step, output: print(f"[green] ✓ Completed: {step.name}[/green]")
)
if result.get("success"):
print("[green]✅ Workflow completed successfully![/green]")
for step_result in result.get("results", []):
status = "✅" if step_result.get("status") == "completed" else "❌"
status = "✅" if step_result.get("status") == "success" else "❌"
print(f" {status} {step_result.get('step', 'Unknown step')}")
# Show final output
final_results = result.get("results", [])
if final_results:
last_output = final_results[-1].get("output", "")
if last_output:
print("\n[bold]Final Output:[/bold]")
print(last_output[:2000] + "..." if len(last_output) > 2000 else last_output)
# Save output if requested
if workflow_save and final_results:
import datetime
output_dir = os.path.join(os.getcwd(), "output", "workflows")
os.makedirs(output_dir, exist_ok=True)
timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
safe_name = workflow_name.replace(" ", "_").lower()
output_file = os.path.join(output_dir, f"{timestamp}_{safe_name}.md")
with open(output_file, "w") as f:
f.write(f"# Workflow: {workflow_name}\n\n")
f.write(f"**Executed:** {timestamp}\n\n")
for step_result in final_results:
f.write(f"## {step_result.get('step', 'Unknown')}\n\n")
f.write(f"**Status:** {step_result.get('status', 'unknown')}\n\n")
if step_result.get("output"):
f.write(f"{step_result['output']}\n\n")
print(f"\n[green]✅ Output saved to: {output_file}[/green]")
else:
print(f"[red]❌ Workflow failed: {result.get('error', 'Unknown error')}[/red]")
@@ -1511,10 +1613,16 @@ class PraisonAI:
print(" praisonai workflow run <name> - Execute a workflow")
print(" praisonai workflow show <name> - Show workflow details")
print(" praisonai workflow create <name> - Create a new workflow")
print("\n[bold]Options:[/bold]")
print("\n[bold]Options (uses global flags):[/bold]")
print(" --workflow-var key=value - Set workflow variable (can be repeated)")
print(" --llm <model> - LLM model (e.g., openai/gpt-4o-mini)")
print(" --tools <tools> - Tools (comma-separated, e.g., tavily)")
print(" --planning - Enable planning mode")
print(" --memory - Enable memory")
print(" --verbose - Enable verbose output")
print(" --save - Save output to file")
print("\n[bold]Example:[/bold]")
print(" praisonai workflow run deploy --workflow-var environment=staging")
print(" praisonai workflow run 'Research Blog' --tools tavily --save")
else:
print(f"[red]Unknown workflow action: {action}[/red]")
print("Use 'praisonai workflow help' for available commands")
@@ -1681,6 +1789,369 @@ class PraisonAI:
except Exception as e:
print(f"[red]ERROR: Todo command failed: {e}[/red]")
def handle_docs_command(self, action: str, action_args: list):
"""
Handle docs subcommand actions.
Args:
action: The docs action (list, show, create, delete)
action_args: Additional arguments for the action
"""
try:
from praisonaiagents.memory import DocsManager
from rich import print
from rich.table import Table
from rich.console import Console
console = Console()
docs = DocsManager(workspace_path=os.getcwd())
if action == 'list':
all_docs = docs.list_docs()
if all_docs:
table = Table(title="Project Documentation")
table.add_column("Name", style="cyan")
table.add_column("Description", style="white")
table.add_column("Priority", style="yellow")
table.add_column("Tags", style="green")
table.add_column("Scope", style="magenta")
for doc in all_docs:
table.add_row(
doc["name"],
doc["description"][:40] + "..." if len(doc["description"]) > 40 else doc["description"],
str(doc["priority"]),
", ".join(doc["tags"][:3]) if doc["tags"] else "",
doc["scope"]
)
console.print(table)
else:
print("[yellow]No docs found. Create files in .praison/docs/[/yellow]")
elif action == 'show':
if not action_args:
print("[red]ERROR: Doc name required. Usage: praisonai docs show <name>[/red]")
return
doc_name = action_args[0]
doc = docs.get_doc(doc_name)
if doc:
print(f"[bold cyan]Doc: {doc.name}[/bold cyan]")
print(f"[bold]Description:[/bold] {doc.description}")
print(f"[bold]Priority:[/bold] {doc.priority}")
if doc.tags:
print(f"[bold]Tags:[/bold] {', '.join(doc.tags)}")
print(f"\n[bold]Content:[/bold]\n{doc.content}")
else:
print(f"[red]Doc not found: {doc_name}[/red]")
elif action == 'create':
if len(action_args) < 2:
print("[red]ERROR: Name and content required. Usage: praisonai docs create <name> <content>[/red]")
return
doc_name = action_args[0]
content = ' '.join(action_args[1:])
docs.create_doc(
name=doc_name,
content=content,
description=f"Doc created via CLI: {doc_name}",
scope="workspace"
)
print(f"[green]✅ Doc created: {doc_name}[/green]")
elif action == 'delete':
if not action_args:
print("[red]ERROR: Doc name required. Usage: praisonai docs delete <name>[/red]")
return
doc_name = action_args[0]
if docs.delete_doc(doc_name):
print(f"[green]✅ Doc deleted: {doc_name}[/green]")
else:
print(f"[red]Doc not found: {doc_name}[/red]")
elif action == 'help' or action == '--help':
print("[bold]Docs Commands:[/bold]")
print(" praisonai docs list - List all docs")
print(" praisonai docs show <name> - Show specific doc")
print(" praisonai docs create <name> <content> - Create a new doc")
print(" praisonai docs delete <name> - Delete a doc")
print("\n[bold]Doc Location:[/bold]")
print(" .praison/docs/*.md, ~/.praison/docs/*.md")
else:
print(f"[red]Unknown docs action: {action}[/red]")
print("Use 'praisonai docs help' for available commands")
except ImportError as e:
print(f"[red]ERROR: Failed to import docs module: {e}[/red]")
print("Make sure praisonaiagents is installed: pip install praisonaiagents")
except Exception as e:
print(f"[red]ERROR: Docs command failed: {e}[/red]")
def handle_mcp_command(self, action: str, action_args: list):
"""
Handle mcp subcommand actions.
Args:
action: The mcp action (list, show, create, delete, enable, disable)
action_args: Additional arguments for the action
"""
try:
from praisonaiagents.memory import MCPConfigManager
from rich import print
from rich.table import Table
from rich.console import Console
console = Console()
mcp = MCPConfigManager(workspace_path=os.getcwd())
if action == 'list':
all_configs = mcp.list_configs()
if all_configs:
table = Table(title="MCP Server Configurations")
table.add_column("Name", style="cyan")
table.add_column("Command", style="white")
table.add_column("Enabled", style="green")
table.add_column("Scope", style="magenta")
table.add_column("Description", style="yellow")
for config in all_configs:
table.add_row(
config["name"],
config["command"],
"✅" if config["enabled"] else "❌",
config["scope"],
config["description"][:30] + "..." if len(config["description"]) > 30 else config["description"]
)
console.print(table)
else:
print("[yellow]No MCP configs found. Create files in .praison/mcp/[/yellow]")
elif action == 'show':
if not action_args:
print("[red]ERROR: Config name required. Usage: praisonai mcp show <name>[/red]")
return
config_name = action_args[0]
config = mcp.get_config(config_name)
if config:
print(f"[bold cyan]MCP Config: {config.name}[/bold cyan]")
print(f"[bold]Command:[/bold] {config.command}")
print(f"[bold]Args:[/bold] {' '.join(config.args)}")
print(f"[bold]Enabled:[/bold] {'Yes' if config.enabled else 'No'}")
print(f"[bold]Description:[/bold] {config.description}")
if config.env:
print("[bold]Environment:[/bold]")
for key, value in config.env.items():
# Mask sensitive values
masked = value[:4] + "..." if len(value) > 8 else "***"
print(f" {key}: {masked}")
else:
print(f"[red]MCP config not found: {config_name}[/red]")
elif action == 'create':
if len(action_args) < 2:
print("[red]ERROR: Name and command required. Usage: praisonai mcp create <name> <command> [args...][/red]")
return
config_name = action_args[0]
command = action_args[1]
args = action_args[2:] if len(action_args) > 2 else []
mcp.create_config(
name=config_name,
command=command,
args=args,
description="MCP server created via CLI",
scope="workspace"
)
print(f"[green]✅ MCP config created: {config_name}[/green]")
elif action == 'delete':
if not action_args:
print("[red]ERROR: Config name required. Usage: praisonai mcp delete <name>[/red]")
return
config_name = action_args[0]
if mcp.delete_config(config_name):
print(f"[green]✅ MCP config deleted: {config_name}[/green]")
else:
print(f"[red]MCP config not found: {config_name}[/red]")
elif action == 'enable':
if not action_args:
print("[red]ERROR: Config name required. Usage: praisonai mcp enable <name>[/red]")
return
config_name = action_args[0]
if mcp.enable_config(config_name):
print(f"[green]✅ MCP config enabled: {config_name}[/green]")
else:
print(f"[red]MCP config not found: {config_name}[/red]")
elif action == 'disable':
if not action_args:
print("[red]ERROR: Config name required. Usage: praisonai mcp disable <name>[/red]")
return
config_name = action_args[0]
if mcp.disable_config(config_name):
print(f"[green]✅ MCP config disabled: {config_name}[/green]")
else:
print(f"[red]MCP config not found: {config_name}[/red]")
elif action == 'help' or action == '--help':
print("[bold]MCP Commands:[/bold]")
print(" praisonai mcp list - List all MCP configs")
print(" praisonai mcp show <name> - Show specific config")
print(" praisonai mcp create <name> <cmd> [args...] - Create a new config")
print(" praisonai mcp delete <name> - Delete a config")
print(" praisonai mcp enable <name> - Enable a config")
print(" praisonai mcp disable <name> - Disable a config")
print("\n[bold]Config Location:[/bold]")
print(" .praison/mcp/*.json, ~/.praison/mcp/*.json")
print("\n[bold]Example:[/bold]")
print(" praisonai mcp create filesystem npx -y @modelcontextprotocol/server-filesystem .")
else:
print(f"[red]Unknown mcp action: {action}[/red]")
print("Use 'praisonai mcp help' for available commands")
except ImportError as e:
print(f"[red]ERROR: Failed to import mcp module: {e}[/red]")
print("Make sure praisonaiagents is installed: pip install praisonaiagents")
except Exception as e:
print(f"[red]ERROR: MCP command failed: {e}[/red]")
def handle_commit_command(self, args: list):
"""
Handle AI commit message generation.
Generates a commit message based on staged changes using AI.
Args:
args: Additional arguments (e.g., --push to auto-push)
"""
try:
import subprocess
from rich import print
from praisonaiagents import Agent
# Check if we're in a git repository
try:
subprocess.run(["git", "rev-parse", "--git-dir"], check=True, capture_output=True)
except subprocess.CalledProcessError:
print("[red]ERROR: Not in a git repository[/red]")
return
# Get staged diff
result = subprocess.run(
["git", "diff", "--cached", "--stat"],
capture_output=True,
text=True
)
if not result.stdout.strip():
print("[yellow]No staged changes. Use 'git add' to stage files first.[/yellow]")
return
# Get detailed diff for context
diff_result = subprocess.run(
["git", "diff", "--cached"],
capture_output=True,
text=True
)
# Limit diff size for context
diff_content = diff_result.stdout[:8000] if len(diff_result.stdout) > 8000 else diff_result.stdout
print("[bold]Staged changes:[/bold]")
print(result.stdout)
print("\n[bold]Generating commit message...[/bold]")
# Create agent for commit message generation
agent = Agent(
name="CommitMessageGenerator",
role="Git Commit Message Writer",
goal="Generate clear, concise, and conventional commit messages",
instructions="""You are an expert at writing git commit messages.
Follow the Conventional Commits specification:
- feat: A new feature
- fix: A bug fix
- docs: Documentation changes
- style: Code style changes (formatting, etc.)
- refactor: Code refactoring
- test: Adding or updating tests
- chore: Maintenance tasks
Format:
<type>(<scope>): <short description>
<optional body with more details>
Keep the first line under 72 characters.
Be specific about what changed and why.""",
llm=os.environ.get("OPENAI_MODEL_NAME", "gpt-4o-mini")
)
prompt = f"""Generate a commit message for these changes:
{result.stdout}
Detailed diff:
{diff_content}
Provide ONLY the commit message, no explanations."""
response = agent.chat(prompt)
commit_message = response.strip()
print("\n[bold green]Suggested commit message:[/bold green]")
print(f"[cyan]{commit_message}[/cyan]")
# Ask for confirmation
print("\n[bold]Options:[/bold]")
print(" [y] Use this message and commit")
print(" [e] Edit the message")
print(" [n] Cancel")
choice = input("\nYour choice [y/e/n]: ").strip().lower()
if choice == 'y':
# Commit with the generated message
subprocess.run(["git", "commit", "-m", commit_message], check=True)
print("[green]✅ Committed successfully![/green]")
# Check if --push was passed
if '--push' in args:
subprocess.run(["git", "push"], check=True)
print("[green]✅ Pushed to remote![/green]")
elif choice == 'e':
# Open editor with the message
import tempfile
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
f.write(commit_message)
temp_path = f.name
editor = os.environ.get('EDITOR', 'nano')
subprocess.run([editor, temp_path])
with open(temp_path, 'r') as f:
edited_message = f.read().strip()
os.unlink(temp_path)
if edited_message:
subprocess.run(["git", "commit", "-m", edited_message], check=True)
print("[green]✅ Committed successfully![/green]")
else:
print("[yellow]Empty message, commit cancelled.[/yellow]")
else:
print("[yellow]Commit cancelled.[/yellow]")
except ImportError as e:
print(f"[red]ERROR: Failed to import required module: {e}[/red]")
print("Make sure praisonaiagents is installed: pip install praisonaiagents")
except subprocess.CalledProcessError as e:
print(f"[red]ERROR: Git command failed: {e}[/red]")
except Exception as e:
print(f"[red]ERROR: Commit command failed: {e}[/red]")
def _save_output(self, prompt: str, result: str):
"""
Save output to output/prompts/ folder.
@@ -1710,15 +2181,164 @@ class PraisonAI:
print(f"[green]✅ Output saved to: {filepath}[/green]")
def _run_inline_workflow(self, initial_prompt):
"""
Run an inline workflow without a .md template file.
Format: --workflow "step1:action1,step2:action2"
Or: --workflow "Research,Write Blog" (uses prompt as context)
Args:
initial_prompt: The initial prompt/context for the workflow
Returns:
Final output from the workflow
"""
from rich import print
workflow_str = self.args.workflow
# Parse workflow steps
steps = []
for step_def in workflow_str.split(","):
step_def = step_def.strip()
if ":" in step_def:
name, action = step_def.split(":", 1)
steps.append({"name": name.strip(), "action": action.strip()})
else:
# Just a step name, use it as both name and action
steps.append({"name": step_def, "action": step_def})
if not steps:
print("[red]ERROR: No workflow steps defined[/red]")
return ""
# Use global flags (--llm, --tools, --planning, --memory, --save, --verbose)
workflow_llm = getattr(self.args, 'llm', None)
workflow_tools_str = getattr(self.args, 'tools', None)
workflow_planning = getattr(self.args, 'planning', False)
workflow_verbose = getattr(self.args, 'verbose', False)
workflow_memory = getattr(self.args, 'memory', False)
workflow_save = getattr(self.args, 'save', False)
# Load tools
workflow_tools = None
if workflow_tools_str:
workflow_tools = self._load_tools(workflow_tools_str)
if workflow_tools:
print(f"[cyan]Loaded {len(workflow_tools)} tool(s) for workflow[/cyan]")
# Initialize memory if enabled (memory will be passed to agent when supported)
if workflow_memory:
print("[cyan]Memory enabled for workflow (shared across steps)[/cyan]")
print(f"[bold cyan]Running inline workflow with {len(steps)} steps[/bold cyan]")
if workflow_planning:
print("[cyan]Planning mode enabled[/cyan]")
# Execute workflow steps sequentially with context passing
context = initial_prompt
results = []
for i, step in enumerate(steps):
step_name = step["name"]
step_action = step["action"]
print(f"[cyan] → Step {i+1}: {step_name}[/cyan]")
# Build prompt with context
if i == 0:
full_prompt = f"{step_action}\n\nContext:\n{context}"
else:
full_prompt = f"{step_action}\n\nContext from previous steps:\n{context}"
# Create agent for this step
agent = PraisonAgent(
name=f"{step_name}Agent",
role=step_name,
goal=step_action,
llm=workflow_llm,
tools=workflow_tools,
verbose=1 if workflow_verbose else 0
)
try:
output = agent.chat(full_prompt)
results.append({"step": step_name, "status": "success", "output": output})
context = output # Pass output to next step
print(f"[green] ✓ Completed: {step_name}[/green]")
except Exception as e:
results.append({"step": step_name, "status": "failed", "error": str(e)})
print(f"[red] ✗ Failed: {step_name} - {e}[/red]")
break
# Show final output
if results:
last_output = results[-1].get("output", "")
if last_output:
print("\n[bold]Final Output:[/bold]")
print(last_output[:2000] + "..." if len(last_output) > 2000 else last_output)
# Save if requested
if workflow_save:
import datetime
output_dir = os.path.join(os.getcwd(), "output", "workflows")
os.makedirs(output_dir, exist_ok=True)
timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
output_file = os.path.join(output_dir, f"{timestamp}_inline_workflow.md")
with open(output_file, "w") as f:
f.write("# Inline Workflow\n\n")
f.write(f"**Executed:** {timestamp}\n\n")
f.write(f"**Initial Prompt:** {initial_prompt}\n\n")
for step_result in results:
f.write(f"## {step_result.get('step', 'Unknown')}\n\n")
f.write(f"**Status:** {step_result.get('status', 'unknown')}\n\n")
if step_result.get("output"):
f.write(f"{step_result['output']}\n\n")
print(f"\n[green]✅ Output saved to: {output_file}[/green]")
return results[-1].get("output", "") if results else ""
def handle_direct_prompt(self, prompt):
"""
Handle direct prompt by creating a single agent and running it.
Supports @mentions:
- @file:path/to/file.py - Include file content
- @web:query - Search the web
- @doc:name - Include doc from .praison/docs/
- @rule:name - Include specific rule
- @url:https://... - Fetch URL content
"""
# Check for inline workflow mode
if hasattr(self, 'args') and getattr(self.args, 'workflow', None):
return self._run_inline_workflow(prompt)
# Process @mentions in the prompt
mention_context = ""
try:
from praisonaiagents.tools.mentions import MentionsParser
parser = MentionsParser(workspace_path=os.getcwd())
if parser.has_mentions(prompt):
mention_context, prompt = parser.process(prompt)
if mention_context:
print("[bold cyan]Processing @mentions...[/bold cyan]")
except ImportError:
pass # Mentions not available
except Exception as e:
logging.debug(f"Error processing mentions: {e}")
# Apply query rewriting if enabled
prompt = self._rewrite_query_if_enabled(prompt)
# Apply prompt expansion if enabled
prompt = self._expand_prompt_if_enabled(prompt)
# Prepend mention context to prompt
if mention_context:
prompt = f"{mention_context}# Task:\n{prompt}"
if PRAISONAI_AVAILABLE:
agent_config = {
"name": "DirectAgent",
@@ -1785,6 +2405,17 @@ class PraisonAI:
if getattr(self.args, 'user_id', None):
agent_config["user_id"] = self.args.user_id
# Session management
if getattr(self.args, 'auto_save', None):
agent_config["memory"] = True # Auto-save requires memory
agent_config["auto_save"] = self.args.auto_save
print(f"[bold cyan]Auto-save enabled - session will be saved as '{self.args.auto_save}'[/bold cyan]")
if getattr(self.args, 'history', None):
agent_config["memory"] = True # History requires memory
agent_config["history_in_context"] = self.args.history
print(f"[bold cyan]History enabled - loading context from last {self.args.history} session(s)[/bold cyan]")
# Claude Memory Tool (Anthropic only)
if getattr(self.args, 'claude_memory', False):
llm = getattr(self.args, 'llm', '')
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==2.3.14 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==2.3.23 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+1 -1
View File
@@ -1 +1 @@
__version__ = "2.3.14"
__version__ = "2.3.23"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.0.192",
"praisonaiagents>=0.1.1",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
+4 -4
View File
@@ -4197,7 +4197,7 @@ requires-dist = [
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.22" },
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.0.22" },
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.22" },
{ name = "praisonaiagents", specifier = ">=0.0.192" },
{ name = "praisonaiagents", specifier = ">=0.1.1" },
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
{ name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
{ name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4252,7 +4252,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.0.192"
version = "0.1.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -4261,9 +4261,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/7f/83/5c3a2bc494a23750571ea5dada6eaafd14fe09eca674bc2483d3659b28d4/praisonaiagents-0.0.192.tar.gz", hash = "sha256:f890f2938a29ff20908e0090537623c9127be9cd1ee8355962e3e5c6f4986e2a", size = 503976, upload-time = "2025-12-16T18:50:30.517Z" }
sdist = { url = "https://files.pythonhosted.org/packages/b0/41/14a8e49c644c26ee565511776c67cc978427f36aec68a1442ac666c35401/praisonaiagents-0.1.1.tar.gz", hash = "sha256:76f69188e4893b0e868f33ea02d26cccfa90b99d679c2db4185d8cf624fa740e", size = 550745, upload-time = "2025-12-17T00:37:35.974Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/a6/22/dc629225b2e5d5e3cf8eb8389490428e9d1ea55575256b99978b4aae255e/praisonaiagents-0.0.192-py3-none-any.whl", hash = "sha256:658378c2235ff5644cb7a294f8ce0a98642e8802a7055ba603e32bdf4f004925", size = 478192, upload-time = "2025-12-16T18:50:28.174Z" },
{ url = "https://files.pythonhosted.org/packages/f7/93/2cef070fae77bbef32a986ad7e1787f799442047f5dac831b9c137839353/praisonaiagents-0.1.1-py3-none-any.whl", hash = "sha256:c116551a35df0f8766496ba5102c54633f071735eae2a71b33eb4df5e2728e86", size = 518796, upload-time = "2025-12-17T00:37:33.824Z" },
]
[[package]]