Compare commits

..
Author SHA1 Message Date
praisonai-triage-agent[bot]andMervin Praison ed90426d96 docs: enhance CONTRIBUTING.md with setup instructions and testing guidance
- Add detailed installation steps with pip install commands
- Specify PEP 8 as code style standard
- Include pytest test running instructions
- Add guidance to check/create issues for large changes
- Include link to GitHub Issues for support

Addresses feedback from code review bots (CodeRabbit, Gemini, Greptile)

Co-authored-by: Mervin Praison <[email protected]>
2026-05-09 18:34:13 +00:00
Akhil Chalasani 9c32b9d84c docs: improve documentation
Created CONTRIBUTING.md
2026-05-09 13:22:09 -05:00
436 changed files with 9525 additions and 31494 deletions

No files matched your search

-51
View File
@@ -1,51 +0,0 @@
# PraisonAI Environment Variables
# Copy this file to .env and update with your actual values
# === Required ===
OPENAI_API_KEY=sk-... # OpenAI key for LLM calls. Get yours at https://platform.openai.com
# === Required for Claw web-search tool ===
TAVILY_API_KEY=tvly-... # Get yours at https://app.tavily.com (free tier available)
# === Optional: Claw dashboard security ===
PRAISONAI_ALLOW_LOCAL_TOOLS=0 # Set to 1 to enable local shell tools (security risk)
# === Optional: Alternative LLM Providers ===
ANTHROPIC_API_KEY=sk-ant-... # Anthropic Claude models
GOOGLE_API_KEY=... # Google Gemini models
GROQ_API_KEY=gsk_... # Groq (fast inference)
COHERE_API_KEY=... # Cohere models
AZURE_OPENAI_API_KEY=... # Azure OpenAI
AZURE_OPENAI_ENDPOINT=... # Azure OpenAI endpoint
# === Optional: Memory backends ===
MEM0_API_KEY=... # Mem0 memory service
REDIS_URL=redis://localhost:6379 # Redis for state management
# === Optional: Observability ===
LANGFUSE_SECRET_KEY=sk-lf-... # Langfuse tracing
LANGFUSE_PUBLIC_KEY=pk-lf-... # Langfuse public key
# === Optional: Bot platforms ===
TELEGRAM_BOT_TOKEN=... # Telegram bot token
DISCORD_BOT_TOKEN=... # Discord bot token
SLACK_BOT_TOKEN=xoxb-... # Slack bot token
SLACK_APP_TOKEN=xapp-... # Slack app token
WHATSAPP_ACCESS_TOKEN=... # WhatsApp bot access token
WHATSAPP_PHONE_NUMBER_ID=... # WhatsApp phone number ID
# === Optional: Database ===
DATABASE_URL=sqlite:///~/.praison/database.sqlite # Database connection string
# === Optional: Tool Security ===
# ALLOWED_TOOLS=search,send_message # Whitelist specific tools to prevent name collisions
# Useful in multi-environment setups with overlapping tool names
# Leave unset to use all tools (with collision warnings)
# Do NOT set to empty string: that's treated as configuration error
# In CI, unknown tool names in the whitelist cause strict startup failure
# HERMES_ONLY_TOOLS=search,send_message # (Deprecated) Backward compatibility alias for ALLOWED_TOOLS
# === Optional: Development ===
LOGLEVEL=INFO # Logging level (DEBUG, INFO, WARNING, ERROR)
CHAINLIT_AUTH_SECRET=... # Chainlit authentication secret
DEBUG=false # Enable debug mode
+2 -5
View File
@@ -181,15 +181,12 @@ jobs:
- name: Fetch PR branch and Setup Remote
if: github.event.issue.pull_request
env:
PR_BRANCH: ${{ steps.check_fork.outputs.pr_branch }}
IS_FORK: ${{ steps.check_fork.outputs.is_fork }}
run: |
# Fetch PR head from base repo and put it into local branch
git fetch origin pull/${{ github.event.issue.number }}/head:"${PR_BRANCH}"
git fetch origin pull/${{ github.event.issue.number }}/head:${{ steps.check_fork.outputs.pr_branch }}
# If it's a fork, make origin point to local to trick claude-code-action's `git fetch origin <branch>`
if [ "${IS_FORK}" = "true" ]; then
if [ "${{ steps.check_fork.outputs.is_fork }}" == "true" ]; then
git remote set-url origin file://$(pwd)
fi
@@ -1,158 +0,0 @@
name: Cross-Repo Integration Tests
on:
push:
branches: [ main ]
paths:
- 'src/praisonai/praisonai/integration/**'
- 'src/praisonai/tests/integration/test_aiui_*'
- 'src/praisonai/pyproject.toml'
pull_request:
branches: [ main ]
paths:
- 'src/praisonai/praisonai/integration/**'
- 'src/praisonai/tests/integration/test_aiui_*'
- 'src/praisonai/pyproject.toml'
workflow_dispatch:
jobs:
cross-repo-integration:
runs-on: ubuntu-latest
timeout-minutes: 30
steps:
- name: Checkout PraisonAI
uses: actions/checkout@v4
with:
path: PraisonAI
ref: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
- name: Checkout PraisonAIUI
uses: actions/checkout@v4
with:
repository: MervinPraison/PraisonAIUI
path: PraisonAIUI
ref: main
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
# Install PraisonAI packages in development mode
cd PraisonAI/src/praisonai-agents
pip install -e .
cd ../praisonai
pip install -e ".[ui,dev]"
# Install PraisonAIUI in development mode
cd ../../../PraisonAIUI
pip install -e .
- name: Verify versions
run: |
python -c "import praisonaiagents; print(f'praisonaiagents: {praisonaiagents.__version__}')"
python -c "import praisonai; print(f'praisonai: {praisonai.__version__}')"
python -c "import praisonaiui; print(f'praisonaiui: {praisonaiui.__version__}')"
- name: Run PraisonAI integration tests
run: |
cd PraisonAI/src/praisonai
python -m pytest tests/integration/test_aiui_* -v --timeout=300
- name: Run PraisonAIUI integration tests
run: |
cd PraisonAIUI
if [ -f tests/integration/test_agentic_roundtrip.py ]; then
python -m pytest tests/integration/test_agentic_roundtrip.py -v --timeout=300
fi
if [ -f tests/test_feature_sdk_backends.py ]; then
python -m pytest tests/test_feature_sdk_backends.py -v --timeout=300
fi
- name: Test Pattern C CLI
run: |
cd PraisonAI/src/praisonai
# Test that ui-gateway command exists
python -m praisonai serve --help | grep ui-gateway
- name: Test public API exports
run: |
cd PraisonAI/src/praisonai
python -c "from praisonai import run_integrated_gateway; print('✓ run_integrated_gateway imported')"
python -c "from praisonai import configure_host; print('✓ configure_host imported')"
- name: Integration smoke test
env:
PRAISONAI_TEST_MODE: "1"
PYTHONPATH: ${{ github.workspace }}/PraisonAI/src/praisonai:${{ github.workspace }}/PraisonAI/src/praisonai-agents:${{ github.workspace }}/PraisonAIUI
run: |
cd PraisonAI/src/praisonai
python -c "
import sys
sys.path.insert(0, '.')
from praisonai.integration.host_app import configure_host
from praisonai import run_integrated_gateway
print('✓ Integration imports successful')
# Test configure_host with new parameters
try:
configure_host(
style='dashboard',
context_paths=['AGENTS.md'],
title='Test App'
)
print('✓ configure_host with new parameters works')
except Exception as e:
print(f'× configure_host failed: {e}')
sys.exit(1)
"
# Optional job with API key for agentic tests (if secret is available)
agentic-integration:
runs-on: ubuntu-latest
timeout-minutes: 30
if: ${{ vars.RUN_AGENTIC_TESTS == 'true' }}
needs: cross-repo-integration
steps:
- name: Checkout PraisonAI
uses: actions/checkout@v4
with:
path: PraisonAI
ref: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
- name: Checkout PraisonAIUI
uses: actions/checkout@v4
with:
repository: MervinPraison/PraisonAIUI
path: PraisonAIUI
ref: main
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
cd PraisonAI/src/praisonai-agents
pip install -e .
cd ../praisonai
pip install -e ".[ui,dev]"
cd ../../../PraisonAIUI
pip install -e .
- name: Run agentic integration test
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
cd PraisonAI/src/praisonai
if [ -n "$OPENAI_API_KEY" ]; then
python -m pytest tests/integration/test_aiui_host_agentic.py -v --timeout=600
else
echo "OPENAI_API_KEY not available, skipping agentic tests"
fi
+5 -4
View File
@@ -29,6 +29,9 @@ jobs:
uses: actions/checkout@v4
with:
persist-credentials: false
with:
persist-credentials: false
with:
fetch-depth: 0
- name: Generate GitHub App Token
@@ -60,12 +63,10 @@ jobs:
GH_TOKEN: ${{ steps.generate_token.outputs.token }}
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
PRAISONAI_AUTO_APPROVE: "true"
ISSUE_NUMBER_INPUT: ${{ inputs.issue_number }}
ISSUE_NUMBER_EVENT: ${{ github.event.issue.number }}
run: |
if [ "${{ github.event_name }}" = "workflow_dispatch" ]; then
export ISSUE_NUMBER="$ISSUE_NUMBER_INPUT"
export ISSUE_NUMBER="${{ inputs.issue_number }}"
else
export ISSUE_NUMBER="$ISSUE_NUMBER_EVENT"
export ISSUE_NUMBER="${{ github.event.issue.number }}"
fi
praisonai github triage --issue $ISSUE_NUMBER --agent-file .github/praisonai-issue-triage.yaml
+6 -7
View File
@@ -30,23 +30,22 @@ jobs:
steps:
- name: Determine checkout ref
id: dest
env:
ISSUE_NUMBER: ${{ github.event.issue.number }}
PR_NUMBER_INPUT: ${{ inputs.pr_number }}
PR_HEAD_SHA: ${{ github.event.pull_request.head.sha }}
run: |
if [ "${{ github.event_name }}" = "issue_comment" ]; then
echo "ref=refs/pull/$ISSUE_NUMBER/head" >> "$GITHUB_OUTPUT"
echo "ref=refs/pull/${{ github.event.issue.number }}/head" >> "$GITHUB_OUTPUT"
elif [ "${{ github.event_name }}" = "workflow_dispatch" ]; then
echo "ref=refs/pull/$PR_NUMBER_INPUT/head" >> "$GITHUB_OUTPUT"
echo "ref=refs/pull/${{ inputs.pr_number }}/head" >> "$GITHUB_OUTPUT"
else
echo "ref=$PR_HEAD_SHA" >> "$GITHUB_OUTPUT"
echo "ref=${{ github.event.pull_request.head.sha }}" >> "$GITHUB_OUTPUT"
fi
- name: Checkout Repository
uses: actions/checkout@v4
with:
persist-credentials: false
with:
persist-credentials: false
with:
ref: ${{ steps.dest.outputs.ref }}
fetch-depth: 0
+2 -14
View File
@@ -26,8 +26,8 @@ jobs:
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
with:
persist-credentials: false
- name: Set up Python 3.11
uses: actions/setup-python@v5
with:
@@ -176,18 +176,6 @@ jobs:
run: praisonai queue list
continue-on-error: true
# ==================== PACKAGING TESTS ====================
- name: Test packaging/daemon entry points
run: |
echo "Testing both entry point methods..."
python -m praisonai --version
praisonai --version
- name: Test doctor packaging command
run: |
echo "Testing doctor packaging diagnostics..."
praisonai doctor packaging --json
# ==================== REAL API TESTS ====================
- name: Real API test - Agent run
if: env.OPENAI_API_KEY != ''
+29
View File
@@ -0,0 +1,29 @@
# Contributing to PraisonAI
Thank you for your interest! Here's how to get started:
## Getting Started
1. Fork the repository
2. Clone your fork and install dependencies:
```bash
git clone https://github.com/your-username/PraisonAI.git
cd PraisonAI
pip install -e src/praisonai-agents/
pip install -e src/praisonai/
```
3. Create a feature branch: `git checkout -b feature/your-feature`
4. Make your changes
5. Submit a pull request
## Guidelines
- Follow existing code style (PEP 8)
- Add tests for new features and run them locally using pytest:
```bash
cd src/praisonai-agents && pytest tests/
```
- Update documentation as needed
- Check existing issues or open a new one to discuss your idea before implementing large changes
For questions or support, please visit our [GitHub Issues](https://github.com/MervinPraison/PraisonAI/issues).
+3 -40
View File
@@ -294,35 +294,8 @@ def search(query: str) -> str:
@tool
def calculate(expression: str) -> float:
"""Safely evaluate a numeric arithmetic expression."""
import ast
import operator
# Define allowed operations
_OPS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.USub: operator.neg,
ast.UAdd: operator.pos,
}
def _safe_eval(node):
if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
return node.value
elif isinstance(node, ast.BinOp) and type(node.op) in _OPS:
return _OPS[type(node.op)](_safe_eval(node.left), _safe_eval(node.right))
elif isinstance(node, ast.UnaryOp) and type(node.op) in _OPS:
return _OPS[type(node.op)](_safe_eval(node.operand))
else:
raise ValueError("Unsupported expression")
try:
return _safe_eval(ast.parse(expression, mode="eval").body)
except (ValueError, SyntaxError, TypeError, ZeroDivisionError, OverflowError):
raise ValueError("Invalid arithmetic expression")
"""Evaluate a math expression."""
return eval(expression)
agent = Agent(
instructions="You are a helpful assistant",
@@ -331,7 +304,6 @@ agent = Agent(
agent.start("Search for AI news and calculate 15*4")
```
> ⚠️ **Security Note:** Never use `eval()`, `exec()`, or `subprocess` in tool functions that process LLM-generated or user-supplied input. Always validate and sanitize inputs to prevent code injection attacks.
> 📖 [Full tools docs](https://docs.praison.ai/docs/tools/tools) — BaseTool, tool packages, 100+ built-in tools
### 5. Persistence (Databases)
@@ -358,15 +330,6 @@ pip install "praisonai[claw]"
praisonai claw
```
#### Required Environment Variables
Copy `.env.example` to `.env` and configure the following variables:
| Variable | Required | Description |
|----------|----------|-------------|
| `OPENAI_API_KEY` | Yes | OpenAI API key for all LLM calls |
| `TAVILY_API_KEY` | Yes (Claw) | Tavily key for the built-in web-search tool. Get one free at https://app.tavily.com |
Open **http://localhost:8082** — the dashboard comes with 13 built-in pages: Chat, Agents, Memory, Knowledge, Channels, Guardrails, Cron, and more. Add messaging channels directly from the UI.
> 📖 [Full Claw docs](https://docs.praison.ai/docs/concepts/claw) — platform tokens, CLI options, Docker, and YAML agent mode
@@ -382,7 +345,7 @@ praisonai flow
Open **http://localhost:7861** — use the **Agent** and **Agent Team** components to create sequential or parallel workflows. Connect Chat Input → Agent Team → Chat Output for instant multi-agent pipelines.
> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/agentflow) — visual agent building, component reference, and deployment
> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/flow) — visual agent building, component reference, and deployment
### 8. PraisonAI UI 🤖 (Clean Chat)
+1 -33
View File
@@ -40,7 +40,6 @@ Methods:
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">chat_with_context</a>(message: str, context: 'ContextPack', **kwargs) -> str</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">checkpoints</a>() -> Optional[bool]</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">checkpoints</a>(value: Optional[bool]) -> None</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">clone_for_channel</a>() -> 'Agent'</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">close</a>() -> None</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">console</a>() -> Optional[Any]</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">context_manager</a>() -> Optional[Any]</code>
@@ -172,7 +171,6 @@ Methods:
* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">safe_run</a>(**kwargs) -> ToolResult</code>
* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate</a>() -> bool</code>
* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate_class</a>() -> bool</code>
* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate_schema_roundtrip</a>() -> bool</code>
* <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">__call__</a>(*args, **kwargs) -> Any</code>
* <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">check_availability</a>() -> tuple[bool, str]</code>
* <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">injected_params</a>() -> Dict[str, Any]</code>
@@ -185,7 +183,6 @@ Methods:
* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_available_tools</a>(context: Optional[Dict[str, Any]] = None) -> List[Union[BaseTool, Callable]]</code>
* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_base_tools</a>() -> List[BaseTool]</code>
* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_tools</a>() -> List[str]</code>
* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_tools_with_allowed_filter</a>(context: Optional[Dict[str, Any]] = None) -> List[str]</code>
* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">register</a>(tool: Union[BaseTool, Callable], name: Optional[str] = None, overwrite: bool = False) -> None</code>
* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">unregister</a>(name: str) -> bool</code>
* <code title="class Tools">Tools.<a href="./src/praisonai-agents/praisonaiagents/tools/tools.py">internet_search</a>(*args, **kwargs)</code>
@@ -479,10 +476,8 @@ Methods:
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">edit_skill</a>(name: str, content: str) -> dict</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_allowed_tools</a>(name: str) -> List[str]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_available_skills</a>() -> List[SkillMetadata]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_available_skills_by_state</a>(state: SkillState) -> List[LoadedSkill]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_instructions</a>(name: str) -> Optional[str]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_skill</a>(name: str) -> Optional[LoadedSkill]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_skills_diagnostics</a>() -> Dict[str, ValidationResult]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_user_invocable_skills</a>() -> List[LoadedSkill]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">invoke</a>(name: str, raw_args: str = '', session_id: Optional[str] = None, shell_exec: bool = False) -> Optional[str]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">load_resources</a>(name: str) -> bool</code>
@@ -491,7 +486,6 @@ Methods:
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">skill_names</a>() -> List[str]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">skills</a>() -> List[LoadedSkill]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">to_prompt</a>() -> str</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">validate_skill_capabilities</a>(skill_name: str, force_refresh: bool = False) -> ValidationResult</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">write_skill_file</a>(name: str, file_path: str, file_content: str) -> dict</code>
* <code title="class SkillMetadata">SkillMetadata.<a href="./src/praisonai-agents/praisonaiagents/skills/models.py">from_properties</a>(props: SkillProperties) -> 'SkillMetadata'</code>
* <code title="class SkillProperties">SkillProperties.<a href="./src/praisonai-agents/praisonaiagents/skills/models.py">to_dict</a>() -> dict</code>
@@ -607,13 +601,6 @@ Methods:
* <code title="class ContextManager">ContextManager.<a href="./src/praisonai-agents/praisonaiagents/context/manager.py">set_tool_budget</a>(tool_name: str, max_tokens: int, protected: bool = False) -> None</code>
* <code title="class ContextManager">ContextManager.<a href="./src/praisonai-agents/praisonaiagents/context/manager.py">truncate_tool_output</a>(tool_name: str, output: str) -> str</code>
# UI
Types:
```python
from praisonaiagents import A2UI
```
# Config
Types:
@@ -768,7 +755,7 @@ Methods:
Types:
```python
from praisonai import Agent, AgentApp, AgentOS, AnthropicManagedAgent, CloudProvider, DB, Deploy, DeployConfig, DeployType, HostedAgent, HostedAgentConfig, LocalAgent, LocalAgentConfig, LocalManagedAgent, LocalManagedConfig, ManagedAgent, ManagedConfig, PraisonAI, __version__, arun, run, run_integrated_gateway
from praisonai import Agent, AgentApp, AgentOS, AnthropicManagedAgent, CloudProvider, DB, Deploy, DeployConfig, DeployType, HostedAgent, HostedAgentConfig, LocalAgent, LocalAgentConfig, LocalManagedAgent, LocalManagedConfig, ManagedAgent, ManagedConfig, PraisonAI, __version__
```
# CLI
@@ -800,9 +787,6 @@ Methods:
* <code title="cli">praisonai bot email <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot install-daemon <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot linear <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot list <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot purge <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot replay <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot slack <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot start <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot telegram <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
@@ -870,10 +854,8 @@ Methods:
* <code title="cli">praisonai doctor env <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor mcp <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor network <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor packaging <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor performance <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor selftest <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor skills <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor tools <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor troubleshoot <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai endpoints list <a href="./src/praisonai/praisonai/cli/commands/endpoints.py">--help</a></code>
@@ -905,24 +887,11 @@ Methods:
* <code title="cli">praisonai gateway send <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway start <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway status <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway stop <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway uninstall <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai github triage <a href="./src/praisonai/praisonai/cli/commands/github.py">--help</a></code>
* <code title="cli">praisonai hooks add <a href="./src/praisonai/praisonai/cli/commands/hooks.py">--help</a></code>
* <code title="cli">praisonai hooks list <a href="./src/praisonai/praisonai/cli/commands/hooks.py">--help</a></code>
* <code title="cli">praisonai hooks remove <a href="./src/praisonai/praisonai/cli/commands/hooks.py">--help</a></code>
* <code title="cli">praisonai kanban block <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
* <code title="cli">praisonai kanban boards <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
* <code title="cli">praisonai kanban comment <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
* <code title="cli">praisonai kanban complete <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
* <code title="cli">praisonai kanban create <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
* <code title="cli">praisonai kanban dispatch <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
* <code title="cli">praisonai kanban link <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
* <code title="cli">praisonai kanban list <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
* <code title="cli">praisonai kanban move <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
* <code title="cli">praisonai kanban reclaim <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
* <code title="cli">praisonai kanban show <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
* <code title="cli">praisonai kanban unblock <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
* <code title="cli">praisonai knowledge add <a href="./src/praisonai/praisonai/cli/commands/knowledge.py">--help</a></code>
* <code title="cli">praisonai knowledge index <a href="./src/praisonai/praisonai/cli/commands/knowledge.py">--help</a></code>
* <code title="cli">praisonai knowledge list <a href="./src/praisonai/praisonai/cli/commands/knowledge.py">--help</a></code>
@@ -1080,7 +1049,6 @@ Methods:
* <code title="cli">praisonai serve status <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai serve stop <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai serve ui <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai serve ui-gateway <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai serve unified <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai session delete <a href="./src/praisonai/praisonai/cli/commands/session.py">--help</a></code>
* <code title="cli">praisonai session export <a href="./src/praisonai/praisonai/cli/commands/session.py">--help</a></code>
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=4.6.52" \
"praisonai>=4.6.37" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=4.6.52" \
"praisonai>=4.6.37" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=4.6.52" \
"praisonai>=4.6.37" \
"praisonai[ui]" \
"praisonai[crewai]"
+3 -3
View File
@@ -35,7 +35,7 @@ def run_quick_comparison():
prompt="What is 2+2?",
iterations=2,
paths=["openai_sdk", "praisonai_agent"],
verbose=True
output="verbose"
)
# Print comparison table
@@ -56,7 +56,7 @@ def run_full_benchmark():
report = handler.run_full_benchmark(
prompt="Hi",
iterations=3,
verbose=True
output="verbose"
)
# Print full report
@@ -78,7 +78,7 @@ def run_agent_benchmark():
prompt="Explain Python in one sentence",
iterations=3,
paths=["openai_sdk", "praisonai_agent"],
verbose=True
output="verbose"
)
# Show timeline diagrams
+1 -1
View File
@@ -22,7 +22,7 @@ custom_modes = {
}
# Create handler with verbose output
handler = CompareHandler(verbose=True)
handler = CompareHandler(output="verbose")
# Example: Manually create comparison results for analysis
# (useful for testing without API calls)
+1 -1
View File
@@ -22,7 +22,7 @@ modes = parse_modes("basic,tools,planning")
print(f"\nParsed modes: {modes}")
# Create compare handler
handler = CompareHandler(verbose=True)
handler = CompareHandler(output="verbose")
# Compare modes (this will run actual agent calls)
# Uncomment to run:
@@ -15,7 +15,7 @@ agent = Agent(
# With array [preset, overrides]
agent_override = Agent(
instructions="You are a helpful assistant.",
memory=MemoryConfig(backend="redis", user_id="custom_user"),
memory=["redis", {"user_id": "custom_user"}],
)
if __name__ == "__main__":
+1 -13
View File
@@ -15,20 +15,8 @@ workflow = AgentFlow(
],
output="verbose",
planning=True,
variables={"topic": "AI agents"},
)
if __name__ == "__main__":
import os
import sys
if not os.environ.get("OPENAI_API_KEY"):
print("Set OPENAI_API_KEY to run this example.")
sys.exit(0)
if sys.platform == "win32":
try:
sys.stdout.reconfigure(encoding="utf-8")
except Exception:
pass
result = workflow.run("AI agents")
result = workflow.run(variables={"topic": "AI agents"})
print(result)
+2 -2
View File
@@ -24,7 +24,7 @@ if has_api_key:
input_text="What is 2 + 2?",
expected_output="4",
num_iterations=3, # Run 3 times for statistical significance
verbose=True
output="verbose"
)
# Run evaluation
@@ -42,7 +42,7 @@ else:
input_text="What is 2 + 2?",
expected_output="4",
num_iterations=1,
verbose=True
output="verbose"
)
# Test with mock output
+2 -2
View File
@@ -27,7 +27,7 @@ if has_api_key:
scoring_type="numeric", # Score 1-10
threshold=7.0, # Pass if score >= 7
num_iterations=2,
verbose=True
output="verbose"
)
# Run evaluation
@@ -45,7 +45,7 @@ if has_api_key:
agent=agent,
input_text="Tell me a joke",
scoring_type="binary", # Pass or Fail
verbose=True
output="verbose"
)
binary_result = binary_evaluator.run(print_summary=True)
+1 -1
View File
@@ -26,7 +26,7 @@ if has_api_key:
num_iterations=10, # Run 10 benchmark iterations
warmup_runs=2, # 2 warmup runs before measurement
track_memory=True, # Track memory usage
verbose=True
output="verbose"
)
# Run evaluation
+4 -15
View File
@@ -16,19 +16,8 @@ def search_web(query: str) -> str:
def calculate(expression: str) -> str:
"""Calculate a math expression safely."""
import ast, operator
_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
ast.Mod: operator.mod, ast.Pow: operator.pow,
ast.USub: operator.neg, ast.UAdd: operator.pos}
def _ev(n):
if isinstance(n, ast.Expression): return _ev(n.body)
if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
raise ValueError(f"Unsupported: {ast.dump(n)}")
return str(_ev(ast.parse(expression, mode="eval")))
"""Calculate a math expression."""
return str(eval(expression))
# Check if we have an API key
@@ -48,7 +37,7 @@ if has_api_key:
input_text="Search for the weather in Paris and calculate 25 * 4",
expected_tools=["search_web", "calculate"], # Tools that should be called
forbidden_tools=["delete_file"], # Tools that should NOT be called
verbose=True
output="verbose"
)
# Run evaluation
@@ -70,7 +59,7 @@ evaluator = ReliabilityEvaluator(
agent=mock_agent,
expected_tools=["search_web", "calculate"], # Tools that should be called
forbidden_tools=["delete_file"], # Tools that should NOT be called
verbose=True
output="verbose"
)
result2 = evaluator.evaluate_tool_calls(
+1 -1
View File
@@ -184,7 +184,7 @@ Answer questions based ONLY on the provided knowledge context.
When asked about codes, provide the EXACT code from the documents.
Be concise but accurate.""",
knowledge=[temp_dir],
memory={"user_id": "compression_demo_user"},
user_id="compression_demo_user",
output="verbose",
)
@@ -164,7 +164,7 @@ Answer questions based ONLY on the provided knowledge context.
When asked about department codes, provide the EXACT code from the documents.
Be specific about which department you're referring to.""",
knowledge=[temp_dir],
memory={"user_id": "hierarchy_demo_user"},
user_id="hierarchy_demo_user",
output="verbose",
)
@@ -124,7 +124,7 @@ def main():
# Show strategy selection for different corpus sizes
print(f"\nStrategy selection by corpus size:")
for size in [100, 1000, 10000, 50000, 100000]:
strategy = select_strategy(CorpusStats(file_count=max(1, size // 1000), total_tokens=size))
strategy = select_strategy(corpus_tokens=size)
print(f" {size:>6} tokens -> {strategy.value}")
except ImportError as e:
@@ -142,7 +142,7 @@ Answer questions based ONLY on the provided knowledge context.
When asked about verification codes, provide the EXACT code from the documents.
If information is not in the context, say 'Information not found in documents.'""",
knowledge=[temp_dir],
memory={"user_id": "strategy_demo_user"},
user_id="strategy_demo_user",
output="verbose", # Use new consolidated param
)
+1 -1
View File
@@ -65,7 +65,7 @@ def main():
name="PolicyExpert",
instructions="You are a company policy expert. Answer questions based ONLY on the provided knowledge. Quote exact values when asked about codes or numbers.",
knowledge=[temp_dir], # Add directory with documents
memory={"user_id": "example_user"},
user_id="example_user", # Required for mem0 backend
)
print("\n1. Testing Agent.chat() with context-required question:")
@@ -50,7 +50,7 @@ def main():
name="HR_Assistant_User1",
instructions="You are an HR assistant.",
knowledge=[temp_dir],
memory={"user_id": "user_alice"},
user_id="user_alice", # Knowledge scoped to Alice
)
response = user1_agent.chat("What is the policy on remote work?")
@@ -80,7 +80,7 @@ def main():
name="Personal_HR_Bot",
instructions="You are a personal HR assistant.",
knowledge=[temp_dir],
memory={"user_id": "user_bob"},
user_id="user_bob",
agent_id="hr_bot_v2",
)
@@ -49,7 +49,7 @@ Only authorized personnel with code WRENCH-992 may access this project.
name="AgentA_Ingester",
instructions="You ingest and store knowledge.",
knowledge=[temp_dir],
memory={"user_id": "user_alice"},
user_id="user_alice", # Scoped to Alice
output="silent",
)
@@ -103,7 +103,7 @@ Only authorized personnel with code WRENCH-992 may access this project.
name="AgentB_Querier",
instructions="Answer questions based on the provided knowledge. If you don't have the information, say so.",
knowledge=[temp_dir],
memory={"user_id": "user_alice"},
user_id="user_alice", # Same scope as Agent A
output="silent",
)
+8 -31
View File
@@ -19,43 +19,20 @@ import os
def calculate(expression: str) -> str:
"""
A simple calculator tool that evaluates mathematical expressions safely.
A simple calculator tool that evaluates mathematical expressions.
Args:
expression: A mathematical expression to evaluate (e.g., "2 + 2")
Returns:
The result of the calculation
"""
import ast
import operator
_OPS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.FloorDiv: operator.floordiv,
ast.Mod: operator.mod,
ast.Pow: operator.pow,
ast.USub: operator.neg,
ast.UAdd: operator.pos,
}
def _safe_eval(node):
if isinstance(node, ast.Expression):
return _safe_eval(node.body)
if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
return node.value
if isinstance(node, ast.UnaryOp) and type(node.op) in _OPS:
return _OPS[type(node.op)](_safe_eval(node.operand))
if isinstance(node, ast.BinOp) and type(node.op) in _OPS:
return _OPS[type(node.op)](_safe_eval(node.left), _safe_eval(node.right))
raise ValueError(f"Unsupported expression: {ast.dump(node)}")
try:
tree = ast.parse(expression, mode="eval")
result = _safe_eval(tree)
# Safe evaluation of mathematical expressions
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
return "Error: Invalid characters in expression"
result = eval(expression)
return f"Result: {result}"
except Exception as e:
return f"Error: {str(e)}"
+6 -12
View File
@@ -24,18 +24,12 @@ def main():
def calculate(expression: str) -> str:
"""Safely evaluate a mathematical expression."""
try:
import ast, operator
_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
ast.Mod: operator.mod, ast.Pow: operator.pow,
ast.USub: operator.neg, ast.UAdd: operator.pos}
def _ev(n):
if isinstance(n, ast.Expression): return _ev(n.body)
if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
raise ValueError(f"Unsupported: {ast.dump(n)}")
return f"Result: {_ev(ast.parse(expression, mode='eval'))}"
# Only allow safe math operations
allowed = set("0123456789+-*/.(). ")
if not all(c in allowed for c in expression):
return "Error: Invalid characters in expression"
result = eval(expression) # Safe due to character filtering
return f"Result: {result}"
except Exception as e:
return f"Error: {e}"
+3 -5
View File
@@ -50,12 +50,10 @@ print("=== Qdrant Knowledge Store Demo (Agent-First) ===")
agent = Agent(
name="KnowledgeAssistant",
instructions="You are a helpful assistant with access to technical documentation.",
knowledge=[doc_path],
knowledge={
"sources": [doc_path],
"vector_store": {
"provider": "qdrant",
"url": "http://localhost:6333",
},
"vector_store": "qdrant",
"url": "http://localhost:6333"
}
)
+2 -29
View File
@@ -21,36 +21,9 @@ def search_web(query: str) -> str:
return f"Search results for: {query}"
def calculate(expression: str) -> str:
"""Calculate a mathematical expression safely."""
import ast
import operator
_OPS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.FloorDiv: operator.floordiv,
ast.Mod: operator.mod,
ast.Pow: operator.pow,
ast.USub: operator.neg,
ast.UAdd: operator.pos,
}
def _eval(node):
if isinstance(node, ast.Expression):
return _eval(node.body)
if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
return node.value
if isinstance(node, ast.UnaryOp) and type(node.op) in _OPS:
return _OPS[type(node.op)](_eval(node.operand))
if isinstance(node, ast.BinOp) and type(node.op) in _OPS:
return _OPS[type(node.op)](_eval(node.left), _eval(node.right))
raise ValueError(f"Unsupported expression: {ast.dump(node)}")
"""Calculate a mathematical expression."""
try:
tree = ast.parse(expression, mode="eval")
return f"Result: {_eval(tree)}"
return f"Result: {eval(expression)}"
except Exception:
return "Invalid expression"
-27
View File
@@ -1,27 +0,0 @@
#!/usr/bin/env python3
"""Minimal A2UI example — agent + send_a2ui_messages tool.
Install:
pip install praisonaiagents[a2ui]
Run:
python send_surface.py
For live rendering, use PraisonAIUI with AG-UI (/agui) — see integrate-a2ui-frontend docs.
"""
from praisonaiagents import Agent
from praisonaiagents.tools.a2ui_tools import send_a2ui_messages
agent = Agent(
name="ui-assistant",
instructions=(
"When asked for UI, call send_a2ui_messages with A2UI v0.9 JSON "
"(createSurface on surface id 'main')."
),
tools=[send_a2ui_messages],
)
if __name__ == "__main__":
result = agent.start("Show a simple welcome card on surface main")
print(result)
@@ -8,7 +8,7 @@ Docs: https://docs.praison.ai/cli/git-integration
from praisonai.cli.features import GitIntegrationHandler
# Initialize
handler = GitIntegrationHandler(verbose=True)
handler = GitIntegrationHandler(output="verbose")
git = handler.initialize(repo_path=".")
# Check if it's a git repo
+1 -1
View File
@@ -8,7 +8,7 @@ Docs: https://docs.praison.ai/cli/repo-map
from praisonai.cli.features import RepoMapHandler
# Initialize with current directory
handler = RepoMapHandler(verbose=True)
handler = RepoMapHandler(output="verbose")
repo_map = handler.initialize(root=".")
# Get the repository map
@@ -9,7 +9,7 @@ from praisonai.cli.features import SandboxExecutorHandler
from praisonai.cli.features.sandbox_executor import SandboxMode
# Initialize with basic sandbox
handler = SandboxExecutorHandler(verbose=True)
handler = SandboxExecutorHandler(output="verbose")
sandbox = handler.initialize(mode="basic")
print(f"Sandbox mode: {handler.get_mode()}")
+2 -1
View File
@@ -25,7 +25,8 @@ pdf_agent = Agent(
goal="Help users understand and extract information from PDF documents",
backstory="You are an expert at reading, analyzing, and answering questions about PDF documents. You provide accurate, detailed answers based on the document content.",
instructions="Read the provided PDF document and answer questions about its content. Be specific and cite relevant sections when possible.",
knowledge={**config, "sources": ["document.pdf"]}, # Replace with your PDF path
knowledge=["document.pdf"], # Replace with your PDF file path
knowledge=config,
reflection=True,
@@ -1,9 +1,6 @@
from pathlib import Path
from praisonaiagents import Agent, Task, AgentTeam
import logging
import os
import sys
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
@@ -20,18 +17,14 @@ config = {
}
}
_sample_path = Path(__file__).with_name("sample_knowledge.txt")
if not _sample_path.exists():
_sample_path.write_text("Mervin Praison is the creator of PraisonAI.\n", encoding="utf-8")
# Create an agent with knowledge capabilities
knowledge_agent = Agent(
name="KnowledgeAgent",
role="Information Specialist",
goal="Store and retrieve knowledge efficiently",
backstory="Expert in managing and utilizing stored knowledge",
knowledge={**config, "sources": [str(_sample_path)]},
)
knowledge=["sample.pdf"],
knowledge=config)
# Define a task for the agent
knowledge_task = Task(
@@ -46,16 +39,8 @@ agents = AgentTeam(
agents=[knowledge_agent],
tasks=[knowledge_task],
process="sequential",
user_id="user1"
)
if __name__ == "__main__":
if not os.environ.get("OPENAI_API_KEY"):
print("Set OPENAI_API_KEY to run this example.")
sys.exit(0)
if sys.platform == "win32":
try:
sys.stdout.reconfigure(encoding="utf-8")
except Exception:
pass
result = agents.start()
print(result)
# Start execution
result = agents.start()
@@ -138,7 +138,8 @@ def main():
agent = Agent(
name="AI Research Assistant",
instructions="You are an AI research assistant. Use your knowledge base to answer questions about artificial intelligence topics.",
knowledge={**rerank_config, "sources": sample_documents},
knowledge=sample_documents, # Will use default knowledge config
knowledge=rerank_config, # Use reranking config
llm="gpt-4o-mini"
)
+3 -7
View File
@@ -32,12 +32,8 @@ rag_task = Task(
agents = AgentTeam(
agents=[rag_agent],
tasks=[rag_task],
user_id="user1"
)
if __name__ == "__main__":
import os
import sys
if not os.environ.get("OPENAI_API_KEY"):
print("Set OPENAI_API_KEY to run this example.")
sys.exit(0)
agents.start()
# Start Agents
agents.start()
@@ -49,18 +49,7 @@ def calculate_with_tracking(expression: str) -> str:
time.sleep(0.05)
try:
import ast, operator
_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
ast.Mod: operator.mod, ast.Pow: operator.pow,
ast.USub: operator.neg, ast.UAdd: operator.pos}
def _ev(n):
if isinstance(n, ast.Expression): return _ev(n.body)
if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
raise ValueError(f"Unsupported: {ast.dump(n)}")
result = _ev(ast.parse(expression, mode="eval"))
result = eval(expression)
return f"Result: {result}"
except Exception as e:
return f"Error: {str(e)}"
+3 -3
View File
@@ -5,7 +5,7 @@ import os
def main():
# Initialize memory config
memory_config = {
"backend": "sqlite",
"provider": "rag",
"use_embedding": True,
"storage": {
"type": "sqlite",
@@ -105,13 +105,13 @@ def main():
agents=[researcher, retriever],
tasks=[store_task, verify_task, query_task, query_both_task], # Use same verbose level as memory
memory={
"provider": "rag",
"embedder": {
"provider": "openai",
"config": {
"model": "text-embedding-3-small"
}
},
"config": {"provider": "sqlite"},
}
}
)
@@ -23,39 +23,12 @@ print(f"[2] Updated to v{managed.agent_version}")
result = agent.start("What is 123 + 456?", stream=True)
# ── 3. Custom tool ──
def _safe_calc(expr: str) -> str:
import ast
allowed = set("0123456789+-*/.() ")
if not all(c in allowed for c in expr):
return "error"
try:
tree = ast.parse(expr, mode="eval")
for node in ast.walk(tree):
if not isinstance(
node,
(
ast.Expression,
ast.BinOp,
ast.UnaryOp,
ast.Constant,
ast.Add,
ast.Sub,
ast.Mult,
ast.Div,
ast.USub,
ast.UAdd,
),
):
return "error"
val = eval(compile(tree, "<expr>", "eval"), {"__builtins__": {}})
return str(val)
except Exception:
return "error"
def handle_calculator(tool_name, tool_input):
expr = tool_input.get("expression", "0")
val = _safe_calc(str(expr))
try:
val = eval(expr, {"__builtins__": {}})
except Exception:
val = "error"
print(f" [Calculator: {expr} = {val}]")
return str(val)
+4 -12
View File
@@ -36,18 +36,10 @@ def calculate(expression: str) -> dict:
expression: Math expression (e.g., "2 + 2 * 3")
"""
try:
import ast, operator
_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
ast.Mod: operator.mod, ast.Pow: operator.pow,
ast.USub: operator.neg, ast.UAdd: operator.pos}
def _ev(n):
if isinstance(n, ast.Expression): return _ev(n.body)
if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
raise ValueError(f"Unsupported: {ast.dump(n)}")
return {"expression": expression, "result": _ev(ast.parse(expression, mode="eval"))}
allowed = set("0123456789+-*/.() ")
if not all(c in allowed for c in expression):
return {"error": "Invalid characters"}
return {"expression": expression, "result": eval(expression)}
except Exception as e:
return {"error": str(e)}
+6 -12
View File
@@ -47,18 +47,12 @@ def calculate(expression: str) -> Dict[str, Any]:
The result of the calculation
"""
try:
import ast, operator
_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
ast.Mod: operator.mod, ast.Pow: operator.pow,
ast.USub: operator.neg, ast.UAdd: operator.pos}
def _ev(n):
if isinstance(n, ast.Expression): return _ev(n.body)
if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
raise ValueError(f"Unsupported: {ast.dump(n)}")
result = _ev(ast.parse(expression, mode="eval"))
# Safe evaluation of mathematical expressions
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
return {"error": "Invalid characters in expression"}
result = eval(expression) # Use safer alternative in production
return {"expression": expression, "result": result}
except Exception as e:
return {"error": str(e)}
+7 -12
View File
@@ -49,18 +49,13 @@ def calculate(expression: str) -> Dict[str, Any]:
The result of the calculation
"""
try:
import ast, operator
_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
ast.Mod: operator.mod, ast.Pow: operator.pow,
ast.USub: operator.neg, ast.UAdd: operator.pos}
def _ev(n):
if isinstance(n, ast.Expression): return _ev(n.body)
if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
raise ValueError(f"Unsupported: {ast.dump(n)}")
result = _ev(ast.parse(expression, mode="eval"))
# Safe evaluation of mathematical expressions
# In production, use a proper math parser
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
return {"error": "Invalid characters in expression"}
result = eval(expression) # Note: Use safer alternative in production
return {"expression": expression, "result": result}
except Exception as e:
return {"error": str(e)}
@@ -12,8 +12,8 @@ print("=== Advanced Graph Memory Integration Example ===\n")
# Configure simple graph memory
memory_config = {
"provider": "mem0",
"config": {
"provider": "mem0",
"vector_store": {
"provider": "chroma",
"config": {
@@ -60,8 +60,8 @@ query_task = Task(
agents_system = AgentTeam(
agents=[knowledge_agent],
tasks=[build_task, query_task],
memory=memory_config,
output="verbose",
memory=True,
memory=memory_config, output="verbose"
)
print("Starting graph memory demonstration...")
@@ -12,8 +12,8 @@ print("=== Advanced Memory Search and Optimization Example ===\n")
# Configure optimized memory settings
memory_config = {
"provider": "mem0",
"config": {
"provider": "mem0",
"vector_store": {
"provider": "chroma",
"config": {
@@ -61,8 +61,8 @@ search_task = Task(
agents_system = AgentTeam(
agents=[research_agent],
tasks=[build_memory_task, search_task],
memory=memory_config,
output="verbose",
memory=True,
memory=memory_config, output="verbose"
)
print("Starting memory search optimization demonstration...")
@@ -86,13 +86,13 @@ def main():
agents=[researcher, retriever],
tasks=[store_task, retrieve_task],
memory={
"provider": "rag",
"embedder": {
"provider": "gemini",
"config": {
"model": "text-embedding-004",
}
},
"config": {"provider": "sqlite"},
}
}
)
+2 -1
View File
@@ -152,7 +152,8 @@ def main():
agents_system = AgentTeam(
agents=[researcher],
tasks=[build_knowledge_task, query_knowledge_task, expand_knowledge_task],
memory=memory_config,
memory=True,
memory=memory_config
)
print("\n🚀 Starting graph memory demonstration...")
@@ -30,7 +30,7 @@ def main():
try:
# Create memory instance
memory = Memory(config={
"backend": "sqlite",
"provider": "rag",
"use_embedding": False,
"agent_name": "demo_agent", # Used in trace events
})
@@ -32,8 +32,9 @@ def init_agent():
return Agent(
name="Knowledge Agent",
instructions="You answer questions based on the provided knowledge.",
knowledge={**config, "sources": ["kag-research-paper.pdf"]},
memory={"user_id": "user1"},
knowledge=["kag-research-paper.pdf"],
knowledge=config,
user_id="user1",
llm="deepseek-r1"
)
@@ -30,12 +30,9 @@ config = {
agent = Agent(
name="Knowledge Agent",
instructions="You answer questions based on the provided knowledge.",
knowledge={
"sources": ["kag-research-paper.pdf"],
"vector_store": config["vector_store"],
"embedder": config["embedder"],
},
memory={"user_id": "user1"},
knowledge=["kag-research-paper.pdf"], # Indexing
knowledge=config,
user_id="user1",
llm="deepseek-r1"
)
@@ -189,8 +189,8 @@ def main():
business_system = AgentTeam(
agents=[data_manager, knowledge_curator, business_analyst, customer_service],
tasks=business_tasks,
memory=mongodb_memory_config,
output="verbose",
memory=True,
memory=mongodb_memory_config, output="verbose"
)
# Execute the comprehensive business pipeline
@@ -55,10 +55,8 @@ def main():
retrieve information from a comprehensive MongoDB knowledge base. You excel
at finding relevant information, synthesizing knowledge from multiple sources,
and providing accurate, context-aware responses.""",
knowledge={
**mongodb_knowledge_config,
"sources": [os.path.join(os.path.dirname(__file__), "llms.md")],
},
knowledge=mongodb_knowledge_config,
knowledge=[os.path.join(os.path.dirname(__file__), "llms.md")],
memory=True,
llm="gpt-4o-mini"
)
@@ -28,8 +28,8 @@ if not os.environ.get("OPENAI_API_KEY"):
def main():
# MongoDB memory configuration
mongodb_memory_config = {
"provider": "mongodb",
"config": {
"provider": "mongodb",
"connection_string": "mongodb://localhost:27017/", # Replace with your MongoDB connection string
"database": "praisonai_memory",
"use_vector_search": True, # Enable Atlas Vector Search (requires MongoDB Atlas)
@@ -93,8 +93,8 @@ def main():
research_system = AgentTeam(
agents=[research_agent],
tasks=research_tasks,
memory=mongodb_memory_config,
output="verbose",
memory=True,
memory=mongodb_memory_config, output="verbose"
)
# Execute the research pipeline
+1 -1
View File
@@ -50,7 +50,7 @@ agent = Agent(
name="TechSupport",
instructions="You are a technical support assistant.",
knowledge=[sample_doc],
memory={"user_id": "comparison_user"},
user_id="comparison_user",
llm="openai/gpt-4o-mini",
output="minimal",
)
+3 -4
View File
@@ -20,9 +20,8 @@ if not os.environ.get("OPENAI_API_KEY"):
exit(1)
# Create a sample document for testing
from pathlib import Path
sample_doc = str(Path(__file__).with_name("sample_doc.txt"))
with open(sample_doc, "w", encoding="utf-8") as f:
sample_doc = "/tmp/sample_doc.txt"
with open(sample_doc, "w") as f:
f.write("""
PraisonAI Framework Documentation
@@ -45,7 +44,7 @@ agent = Agent(
name="DocBot",
instructions="You are a helpful documentation assistant. Answer questions based on the provided context.",
knowledge=[sample_doc],
memory={"user_id": "example_user"},
user_id="example_user", # Required for RAG retrieval
llm="openai/gpt-4o-mini",
output="minimal",
)
+1 -1
View File
@@ -38,7 +38,7 @@ agent = Agent(
name="PolicyBot",
instructions="You are a customer service assistant.",
knowledge=[sample_doc],
memory={"user_id": "policy_user"},
user_id="policy_user",
llm="openai/gpt-4o-mini",
output="minimal",
)
+1 -1
View File
@@ -44,7 +44,7 @@ agent = Agent(
name="MLExpert",
instructions="You are a machine learning expert. Answer questions based on the provided context.",
knowledge=[sample_doc],
memory={"user_id": "rag_user"},
user_id="rag_user",
llm="openai/gpt-4o-mini",
output="minimal",
)
+1 -1
View File
@@ -59,7 +59,7 @@ agent = Agent(
name="APIDocBot",
instructions="You are an API documentation assistant.",
knowledge=[sample_doc],
memory={"user_id": "hybrid_user"},
user_id="hybrid_user",
llm="openai/gpt-4o-mini",
output="minimal",
)
+1 -1
View File
@@ -44,7 +44,7 @@ agent = Agent(
name="ResearchBot",
instructions="You are a research assistant. Always cite your sources.",
knowledge=[doc1, doc2],
memory={"user_id": "citation_user"},
user_id="citation_user",
llm="openai/gpt-4o-mini",
output="minimal",
)
@@ -172,7 +172,7 @@ print("="*70)
# Create sessions for different users working on the same project
reviewer_session = Session(
session_id="peer_review_001",
user_id="user_reviewer_01",
user_id="user_reviewer_01", # Different user
storage_path=session_dir
)
@@ -14,12 +14,13 @@ def main():
# Initialize memory with quality-based configuration
memory_config = {
"backend": "sqlite",
"user_id": "quality_demo_user",
"provider": "rag",
"use_embedding": True,
"rag_db_path": "quality_memory_db"
}
memory = Memory(config=memory_config, verbose=5)
print(f"📦 Initialized memory system with backend: {memory_config['backend']}")
print(f"📦 Initialized memory system with provider: {memory_config['provider']}")
# Create agent with memory
agent = Agent(
@@ -27,6 +28,7 @@ def main():
role="Information Quality Specialist",
instructions="Store and retrieve information with quality assessment",
memory=memory,
user_id="quality_demo_user"
)
print(f"🤖 Created agent: {agent.name}")
+1 -1
View File
@@ -15,7 +15,7 @@ def main():
# Create a session with persistent storage
session = Session(
session_id="demo_chat_001",
user_id="demo_user",
user_id="demo_user"
)
print(f"📝 Created session: {session}")
@@ -88,8 +88,9 @@ def main():
workflow = AgentTeam(
agents=[researcher, analyzer, writer],
tasks=[research_task, analysis_task, report_task],
memory={"user_id": "research_project_001"},
memory=True,
process="workflow",
user_id="research_project_001",
)
print(f"🤖 Created workflow with {len(workflow.agents)} agents and {len(workflow.tasks)} tasks")
@@ -14,7 +14,7 @@ def main():
goal="Find information about a topic",
backstory="You are an expert researcher who finds comprehensive information.",
llm="gpt-4o-mini", # Use a cost-effective model for examples
output={"metrics": True} # Token metrics via consolidated output config
metrics=True # 🎯 SIMPLIFIED: Just add this parameter!
)
# Create a simple task
@@ -14,7 +14,7 @@ def main():
goal="Analyze data and provide insights",
backstory="You are an experienced data analyst with expertise in market research.",
llm="gpt-4o-mini",
output={"metrics": True}
metrics=True # 🎯 Enable metrics tracking
)
writer_agent = Agent(
@@ -23,7 +23,7 @@ def main():
goal="Create well-structured content",
backstory="You are a skilled technical writer who creates clear, engaging content.",
llm="gpt-4o-mini",
output={"metrics": True}
metrics=True # 🎯 Enable metrics tracking
)
# Create multiple tasks
@@ -14,7 +14,7 @@ def main():
goal="Conduct comprehensive market research",
backstory="You are a seasoned market researcher with 10+ years of experience.",
llm="gpt-4o-mini",
output={"metrics": True}
metrics=True # 🎯 Enable automatic metrics tracking
)
strategy_agent = Agent(
@@ -23,7 +23,7 @@ def main():
goal="Develop actionable business strategies",
backstory="You are a strategic consultant who translates research into business plans.",
llm="gpt-4o-mini",
output={"metrics": True}
metrics=True # 🎯 Enable automatic metrics tracking
)
# Create tasks that will generate substantial token usage
@@ -32,8 +32,9 @@ def init_agent():
return Agent(
name="Knowledge Agent",
instructions="You answer questions based on the provided knowledge.",
knowledge={**config, "sources": ["kag-research-paper.pdf"]},
memory={"user_id": "user1"},
knowledge=["kag-research-paper.pdf"],
knowledge=config,
user_id="user1",
llm="deepseek-r1"
)
@@ -264,12 +264,13 @@ agents = AgentTeam(
tasks=[tool_query_task, pentester_analysis_task, ciso_review_task],
process="hierarchical",
manager_llm="gpt-4o-mini",
memory=True, # Enable shared memory for domain context
memory={
"backend": "sqlite",
"provider": "rag",
"config": {
"collection_name": f"{domain.replace('.', '_')}_security_analysis"
},
},
}
}
)
# Main Execution Workflow
@@ -5,7 +5,7 @@ Demonstrates using should_run to conditionally execute agent steps
based on input or previous results.
"""
from praisonaiagents import Agent, AgentFlow, Task
from praisonaiagents import Agent, Workflow, Task
# Condition functions
def is_technical(ctx) -> bool:
+1 -1
View File
@@ -207,7 +207,7 @@ def api_augmented_rag():
instructions="""You are a financial assistant with access to market APIs.
Provide accurate financial information and context.
Always note that prices are subject to change.""",
memory={"user_id": "api_demo"},
user_id="api_demo"
)
queries = [
-40
View File
@@ -1,40 +0,0 @@
# Scientific paper workflow — YAML equivalent of multi_agent_paper.py
# Run with:
# praisonai agents run --file examples/scientific_writing/agents.yaml
framework: praisonai
topic: "Climate change effects on coral reef biodiversity"
roles:
literature_reviewer:
role: "Literature Reviewer"
goal: "Survey recent academic work on {topic} and produce an APA review."
backstory: "Expert in academic literature surveying and citation hygiene."
llm: "gpt-4o-mini"
tasks:
review_literature:
description: "Produce a concise literature review with 5-8 APA citations."
expected_output: "APA-formatted literature review."
methodology_designer:
role: "Methodology Designer"
goal: "Design a reproducible research methodology for {topic}."
backstory: "Specialises in rigorous experimental design."
llm: "gpt-4o-mini"
tasks:
design_methodology:
description: "Write a reproducible methods section in LaTeX."
expected_output: "LaTeX methods section."
scientific_writer:
role: "Scientific Writer"
goal: "Assemble the final LaTeX paper on {topic}."
backstory: >-
Specialised in LaTeX-formatted academic papers, fine-tuned on
scientific literature (CAJAL-4B checkpoint).
llm: "huggingface/Agnuxo/CAJAL-4B-P2PCLAW"
tasks:
write_paper:
description: >-
Combine the literature review and methodology into a full
scientific paper on {topic}. Use LaTeX formatting.
expected_output: "Complete LaTeX paper."
@@ -1,73 +0,0 @@
"""Multi-agent scientific paper workflow (Researcher -> Methodologist -> Writer).
Demonstrates the agent-centric pattern from AGENTS.md: compose specialised
Agents with ``AgentTeam`` + sequential ``Task``s instead of creating custom
Agent subclasses. The writer uses the CAJAL-4B HuggingFace model; the
reviewer and methodologist use general-purpose LLMs.
Usage:
python examples/scientific_writing/multi_agent_paper.py
"""
from praisonaiagents import Agent, AgentTeam, Task
from scientific_writer import ( # noqa: E402 — example imports a sibling file
build_scientific_writer,
format_citation,
format_latex_section,
)
literature_reviewer = Agent(
name="Literature Reviewer",
instructions=(
"You are an expert at surveying academic literature. "
"Produce a concise literature review with 5-8 key citations in APA."
),
tools=[format_citation],
)
methodology_designer = Agent(
name="Methodology Designer",
instructions=(
"You design rigorous research methodologies. "
"Output a clear, reproducible methods section."
),
tools=[format_latex_section],
)
scientific_writer = build_scientific_writer()
def run(topic: str) -> str:
"""Run the 3-agent workflow for a given research ``topic``."""
t1 = Task(
name="review_literature",
description=f"Review literature on: {topic}",
agent=literature_reviewer,
expected_output="APA literature review.",
)
t2 = Task(
name="design_methodology",
description=f"Design research methodology for: {topic}",
agent=methodology_designer,
expected_output="LaTeX methods section.",
)
t3 = Task(
name="write_paper",
description=(
f"Combine the literature review and methodology into a full "
f"scientific paper on: {topic}. Use LaTeX formatting."
),
agent=scientific_writer,
expected_output="Complete LaTeX paper.",
)
team = AgentTeam(
agents=[literature_reviewer, methodology_designer, scientific_writer],
tasks=[t1, t2, t3],
)
return team.start()
if __name__ == "__main__":
result = run("Transformer architectures for protein folding prediction")
print(result)
@@ -1,85 +0,0 @@
"""Scientific Paper Writer — single-agent example using the CAJAL model.
CAJAL (``Agnuxo/CAJAL-4B-P2PCLAW``) is a 4B-parameter HuggingFace model
specialised for LaTeX-formatted academic writing. This example shows how
to use it with a regular ``Agent`` and ``@tool`` functions — no new Agent
subclass is needed. Run 3 ways: Python (this file), CLI, or YAML.
Requirements:
pip install "praisonaiagents[llm]" # for litellm / HuggingFace
export HUGGINGFACE_API_KEY=... # or run a local inference server
Usage:
python examples/scientific_writing/scientific_writer.py
"""
from praisonaiagents import Agent, tool
@tool
def format_latex_section(title: str, content: str) -> str:
"""Wrap prose content in a LaTeX ``\\section{}`` block.
Args:
title: Section heading (e.g. ``"Introduction"``).
content: The body text.
Returns:
LaTeX-formatted section string.
"""
return f"\\section{{{title}}}\n{content}\n"
@tool
def format_citation(authors: str, year: int, title: str, venue: str) -> str:
"""Render a single APA-style citation line.
Args:
authors: ``"Smith, J. & Jones, A."``
year: Publication year.
title: Paper title.
venue: Journal / conference name.
Returns:
One APA-formatted citation string.
"""
return f"{authors} ({year}). {title}. *{venue}*."
SCIENTIFIC_INSTRUCTIONS = """You are a specialised scientific paper writer.
Produce high-quality academic content with:
- Clear structure (Abstract, Introduction, Methodology, Results, Discussion,
Conclusion, References)
- Rigorous methodology and precise language
- Proper APA citations and LaTeX formatting
- Honest limitations and future work sections
Use ``format_latex_section`` to wrap each section in LaTeX, and
``format_citation`` to build the References list.
"""
def build_scientific_writer(
model: str = "huggingface/Agnuxo/CAJAL-4B-P2PCLAW",
) -> Agent:
"""Construct a scientific-writing Agent.
Override ``model`` for any HuggingFace / local / API model — the default
targets the CAJAL-4B checkpoint for LaTeX-style academic output.
"""
return Agent(
name="Scientific Writer",
instructions=SCIENTIFIC_INSTRUCTIONS,
llm=model,
tools=[format_latex_section, format_citation],
)
if __name__ == "__main__":
agent = build_scientific_writer()
paper = agent.start(
"Write a short scientific paper (Abstract + Introduction + Conclusion) "
"on 'Climate change effects on coral reef biodiversity'. "
"Use LaTeX formatting and include at least 3 APA citations."
)
print(paper)
+5 -26
View File
@@ -44,33 +44,12 @@ def main():
def calculate(expression: str) -> str:
"""Calculate a math expression safely."""
import ast
allowed = set("0123456789+-*/.() ")
if not all(c in allowed for c in expression):
return "Error: Invalid expression"
try:
tree = ast.parse(expression, mode="eval")
for node in ast.walk(tree):
if not isinstance(
node,
(
ast.Expression,
ast.BinOp,
ast.UnaryOp,
ast.Constant,
ast.Add,
ast.Sub,
ast.Mult,
ast.Div,
ast.USub,
ast.UAdd,
),
):
return "Error: Invalid expression"
return str(
eval(compile(tree, "<expr>", "eval"), {"__builtins__": {}})
)
# Safe eval for basic math
allowed = set("0123456789+-*/.() ")
if all(c in allowed for c in expression):
return str(eval(expression))
return "Error: Invalid expression"
except Exception as e:
return f"Error: {e}"
-222
View File
@@ -1,222 +0,0 @@
# Multi-Channel Bot Deployment Guide
This guide explains how to set up multiple bots on the same platform (e.g., multiple Telegram bots) for different roles using PraisonAI, similar to Hermes-style AI workforces.
## Overview
PraisonAI supports running multiple bots on the same platform, each routing to different agents with specialized knowledge and capabilities. This is useful for:
- **Role-based bots**: `@company_cfo_bot`, `@company_ops_bot`, `@company_content_bot`
- **Department-specific agents**: Each with different knowledge brains and tool access
- **Workflow separation**: Different bots handle different types of requests
## Quick Start with Onboard Wizard
The easiest way to set up multiple channels is using the onboard wizard:
```bash
praisonai onboard
```
The wizard now supports:
1. **Multi-channel setup**: Add multiple bots for the same platform
2. **Role-based naming**: Automatically generates `TELEGRAM_<ROLE>_BOT_TOKEN` environment variables
3. **Agent routing**: Each channel routes to a different agent
### Onboard Wizard Flow
1. **Choose platform**: Start with your first platform (e.g., telegram)
2. **Add more channels**: Wizard asks "Add another bot channel?"
3. **Specify role**: Enter role name (e.g., "cfo", "ops", "content")
4. **Configure tokens**: Each channel gets its own environment variable
5. **Test connections**: Wizard tests each bot separately
## Environment Variable Convention
For multiple bots on the same platform, use this naming convention:
| Channel | Environment Variable | Bot Purpose |
|---------|---------------------|-------------|
| `telegram_cfo` | `TELEGRAM_CFO_BOT_TOKEN` | CFO-related tasks |
| `telegram_ops` | `TELEGRAM_OPS_BOT_TOKEN` | Operations tasks |
| `telegram_content` | `TELEGRAM_CONTENT_BOT_TOKEN` | Content creation |
| `telegram_support` | `TELEGRAM_SUPPORT_BOT_TOKEN` | Customer support |
### Setting Up Tokens
1. **Create bots in @BotFather**:
```text
/newbot
Bot name: Company CFO Bot
Username: @company_cfo_bot
-> Copy token
/newbot
Bot name: Company Ops Bot
Username: @company_ops_bot
-> Copy token
```
2. **Set environment variables**:
```bash
export TELEGRAM_CFO_BOT_TOKEN="123456789:ABC..."
export TELEGRAM_OPS_BOT_TOKEN="987654321:DEF..."
export TELEGRAM_CONTENT_BOT_TOKEN="456789123:GHI..."
```
3. **Or use `~/.praisonai/.env`**:
```env
TELEGRAM_CFO_BOT_TOKEN=123456789:ABC...
TELEGRAM_OPS_BOT_TOKEN=987654321:DEF...
TELEGRAM_CONTENT_BOT_TOKEN=456789123:GHI...
TELEGRAM_ALLOWED_USERS=123456789,987654321
```
## Example Configuration
See `multi-telegram-hermes-workforce.yaml` for a complete example.
### Key Features
1. **Multiple Channels**:
```yaml
channels:
telegram_cfo:
platform: telegram
token: ${TELEGRAM_CFO_BOT_TOKEN}
routes:
default: cfo
telegram_ops:
platform: telegram
token: ${TELEGRAM_OPS_BOT_TOKEN}
routes:
default: ops
```
2. **Specialized Agents**:
```yaml
agents:
cfo:
instructions: "You are a CFO agent. Help with financial analysis..."
model: gpt-4o-mini
ops:
instructions: "You are an ops agent. Help with system monitoring..."
model: gpt-4o-mini
```
## Validation with Doctor
Use the doctor command to validate your multi-channel setup:
```bash
praisonai doctor
```
The doctor checks for:
- ✅ **Unique tokens**: Each channel uses a different bot token
- ✅ **Naming convention**: Follows `PLATFORM_ROLE_BOT_TOKEN` pattern
- ✅ **Missing tokens**: All referenced environment variables are set
- ❌ **Duplicate tokens**: Same token used by multiple channels (fails)
## Deployment
### Start Gateway
```bash
# Start the multi-bot gateway
praisonai gateway start --config multi-telegram-hermes-workforce.yaml
# Or install as daemon
praisonai onboard # Wizard installs daemon automatically
praisonai gateway status
```
### Verify Setup
1. **Check health**: `curl http://127.0.0.1:8765/health`
2. **View dashboard**: `praisonai claw` → http://127.0.0.1:8082
3. **Test bots**: Send messages to each bot to verify routing
## Common Patterns
### Basic Multi-Role Setup
```yaml
# Single platform, multiple roles
channels:
telegram_support:
token: ${TELEGRAM_SUPPORT_BOT_TOKEN}
routes:
default: support_agent
telegram_sales:
token: ${TELEGRAM_SALES_BOT_TOKEN}
routes:
default: sales_agent
```
### Cross-Platform Workforce
```yaml
# Multiple platforms, multiple roles
channels:
telegram_cfo:
platform: telegram
token: ${TELEGRAM_CFO_BOT_TOKEN}
routes:
default: cfo
discord_ops:
platform: discord
token: ${DISCORD_OPS_BOT_TOKEN}
routes:
default: ops
slack_content:
platform: slack
token: ${SLACK_CONTENT_BOT_TOKEN}
routes:
default: content
```
## Troubleshooting
### Common Issues
1. **"Token is used by multiple channels"**
- Each bot needs its own unique token from @BotFather
- Check environment variables for duplicates
2. **"Channel skipped - no token"**
- Set the environment variable: `export TELEGRAM_CFO_BOT_TOKEN="..."`
- Or update `~/.praisonai/.env`
3. **"Bot not responding"**
- Verify token is correct: `praisonai doctor`
- Check bot is not already running elsewhere
- Ensure allowlist includes your user ID
### Debugging Commands
```bash
# Check configuration
praisonai doctor
# View logs
praisonai gateway logs
# Test connection
curl -H "Authorization: Bearer $GATEWAY_AUTH_TOKEN" \
http://127.0.0.1:8765/info
```
## Migration from Single-Bot Setup
If you currently have a single bot setup, you can add more channels:
1. **Run onboard wizard**: `praisonai onboard`
2. **Add new channels**: Wizard will preserve existing setup
3. **Update configuration**: New channels added to existing `bot.yaml`
4. **Restart gateway**: `praisonai gateway restart`
The wizard preserves backward compatibility with single-bot configurations.
-19
View File
@@ -1,19 +0,0 @@
# A2UI canvas agent — YAML example
#
# Usage:
# pip install praisonaiagents[a2ui]
# praisonai agent run --file a2ui-canvas.yaml --prompt "Show a welcome card"
#
# For live A2UI rendering, use PraisonAIUI (AG-UI / surfaces):
# pip install praisonaiui
# See examples/python/a2ui/send_surface.py and PraisonAIUI example 29.
agent:
name: "ui-assistant"
instructions: |
You are a helpful assistant. When the user asks for UI elements
(buttons, cards, forms), call send_a2ui_messages with valid A2UI v0.9 JSON.
Use surface id "main" unless told otherwise.
llm: gpt-4o-mini
tools:
- send_a2ui_messages
-40
View File
@@ -1,40 +0,0 @@
# Example: Agent with MCP Tool Integration
#
# This example shows how to configure an agent with MCP (Model Context Protocol)
# servers for additional tool capabilities.
#
# Usage:
# praisonai workflow run --file examples/yaml/agent-with-mcp.yaml
agent:
name: mcp_assistant
instructions: |
You are a helpful assistant with access to filesystem and time tools.
Use the MCP tools when needed to help users with file operations or time queries.
llm: gpt-4o-mini
# MCP server configuration
mcp:
servers:
filesystem:
# Filesystem access for the agent
command: npx
args:
- "-y"
- "@modelcontextprotocol/server-filesystem"
- "."
env: {}
enabled: true
time:
# Time and date utilities
command: npx
args:
- "-y"
- "@modelcontextprotocol/server-time"
env: {}
enabled: true
# NOTE: Per-server tool filtering (tools.include/exclude) is not yet supported
# by the MCP server config schema. Tool filtering should be applied after
# calling mcp.get_tools() in your application code.
@@ -1,66 +0,0 @@
# Multi-Telegram Channel Hermes Workforce Example
# Generated by `praisonai onboard` with multi-channel support
# Start with: praisonai gateway start --config multi-telegram-hermes-workforce.yaml
gateway:
host: "127.0.0.1"
port: 8765
# Legacy single-agent block for backward compatibility
agent:
name: "assistant"
instructions: "You are a helpful AI assistant. You can search the web, manage schedules, and remember past conversations. Use your tools proactively."
model: "gpt-4o-mini"
# Multi-agent workforce with different specialized roles
agents:
cfo:
instructions: "You are a CFO agent. You are a helpful AI assistant. Use your tools proactively to help with CFO-related tasks like financial analysis, budget planning, and cost optimization."
model: gpt-4o-mini
memory: false
ops:
instructions: "You are an ops agent. You are a helpful AI assistant. Use your tools proactively to help with ops-related tasks like system monitoring, deployment, and infrastructure management."
model: gpt-4o-mini
memory: false
content:
instructions: "You are a content agent. You are a helpful AI assistant. Use your tools proactively to help with content-related tasks like writing, editing, and content strategy."
model: gpt-4o-mini
memory: false
# Multiple Telegram channels - each with unique tokens and routing
channels:
telegram_cfo:
platform: telegram
token: ${TELEGRAM_CFO_BOT_TOKEN}
group_policy: mention_only
allowed_users: ${TELEGRAM_ALLOWED_USERS}
# auto_approve_tools: true
# Chat bots cannot show CLI approval prompts; safe tools auto-approved by default.
routes:
default: cfo
telegram_ops:
platform: telegram
token: ${TELEGRAM_OPS_BOT_TOKEN}
group_policy: mention_only
allowed_users: ${TELEGRAM_ALLOWED_USERS}
# auto_approve_tools: true
# Chat bots cannot show CLI approval prompts; safe tools auto-approved by default.
routes:
default: ops
telegram_content:
platform: telegram
token: ${TELEGRAM_CONTENT_BOT_TOKEN}
group_policy: mention_only
allowed_users: ${TELEGRAM_ALLOWED_USERS}
# auto_approve_tools: true
# Chat bots cannot show CLI approval prompts; safe tools auto-approved by default.
routes:
default: content
# Legacy routing block
routing:
default: assistant
@@ -1,10 +1,6 @@
#!/usr/bin/env python3
from __future__ import annotations
"""W1 real agentic smoke test (high-level BotOS API, no manual mgr).
This is not a pytest test file - it's an integration smoke test.
Uses the user-facing API only — ``Bot``, ``BotOS``, ``InMemoryIdentityResolver`` —
and exercises the same cross-platform-continuity scenario as
``smoke_w1_real.py`` but without touching ``BotSessionManager`` directly.
@@ -17,11 +13,10 @@ wire-up from ``BotOS(identity_resolver=...)`` → ``Bot._identity_resolver``
Run::
# From the repository root:
PYTHONPATH=src/praisonai-agents:src/praisonai python src/praisonai/tests/integration/smoke_w1_botos_real.py
PYTHONPATH=src/praisonai-agents:src/praisonai python scripts/smoke_w1_botos_real.py
"""
# Prevent pytest from treating this as a test file
__test__ = False
from __future__ import annotations
import asyncio
import os
@@ -1,18 +1,13 @@
#!/usr/bin/env python3
from __future__ import annotations
"""W1 real agentic smoke test (bypasses pytest conftest gating).
This is not a pytest test file - it's an integration smoke test.
Runs::
cd /path/to/PraisonAI
PYTHONPATH=src/praisonai-agents:src/praisonai python src/praisonai/tests/integration/smoke_w1_real.py
cd src/praisonai
python ../../scripts/smoke_w1_real.py
"""
# Prevent pytest from treating this as a test file
__test__ = False
from __future__ import annotations
import asyncio
import os
@@ -1,10 +1,6 @@
#!/usr/bin/env python3
from __future__ import annotations
"""W1 deep-validation smoke test.
This is not a pytest test file - it's an integration smoke test.
Exercises every robustness dimension of W1 with REAL LLM calls:
T1. 3-platform unification (telegram → discord → slack, one user)
@@ -16,12 +12,11 @@ Exercises every robustness dimension of W1 with REAL LLM calls:
Run::
cd /path/to/PraisonAI
PYTHONPATH=src/praisonai-agents:src/praisonai python src/praisonai/tests/integration/smoke_w1_robust.py
cd /Users/praison/worktrees/hermes-parity
PYTHONPATH=src/praisonai-agents:src/praisonai python scripts/smoke_w1_robust.py
"""
# Prevent pytest from treating this as a test file
__test__ = False
from __future__ import annotations
import asyncio
import os
+1 -2
View File
@@ -95,5 +95,4 @@ chroma_db/
.windsurf
.ruff_cache
*.sfdx
.sfdx/
.cursor/
.sfdx/
@@ -79,20 +79,10 @@ def get_time(timezone: Literal["est", "pst"]) -> str:
def calculate(expression: str) -> str:
"""Calculate a math expression safely."""
"""Calculate a math expression."""
try:
import ast, operator
_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
ast.Mod: operator.mod, ast.Pow: operator.pow,
ast.USub: operator.neg, ast.UAdd: operator.pos}
def _ev(n):
if isinstance(n, ast.Expression): return _ev(n.body)
if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
raise ValueError(f"Unsupported: {ast.dump(n)}")
return f"Result: {_ev(ast.parse(expression, mode='eval'))}"
result = eval(expression)
return f"Result: {result}"
except Exception:
return "Error: Invalid expression"
@@ -343,7 +343,6 @@ _LAZY_IMPORTS = {
# UI (optional)
'AGUI': ('praisonaiagents.ui.agui', 'AGUI'),
'A2A': ('praisonaiagents.ui.a2a', 'A2A'),
'A2UI': ('praisonaiagents.ui.a2ui', 'A2UI'),
# Feature configs
'MemoryConfig': ('praisonaiagents.config.feature_configs', 'MemoryConfig'),
@@ -384,7 +383,6 @@ _LAZY_IMPORTS = {
'MultiAgentExecutionConfig': ('praisonaiagents.config.feature_configs', 'MultiAgentExecutionConfig'),
'MultiAgentPlanningConfig': ('praisonaiagents.config.feature_configs', 'MultiAgentPlanningConfig'),
'MultiAgentMemoryConfig': ('praisonaiagents.config.feature_configs', 'MultiAgentMemoryConfig'),
'ToolSearchConfig': ('praisonaiagents.config.feature_configs', 'ToolSearchConfig'),
# Parameter resolver
'resolve': ('praisonaiagents.config.param_resolver', 'resolve'),
@@ -816,6 +814,6 @@ def __dir__():
# - Workflow, Task, Route, Parallel, Loop, Repeat, Pipeline, etc.
# - MCP, FlowDisplay, track_workflow, FastContext, etc.
# - Plan, PlanStep, TodoList, PlanningAgent, ApprovalCallback, etc.
# - RAG, RAGConfig, RAGResult, AGUI, A2A, A2UI, etc.
# - RAG, RAGConfig, RAGResult, AGUI, A2A, etc.
# - All telemetry, display, and utility functions
# ============================================================================
@@ -21,7 +21,6 @@ from .chat_handler import ChatHandlerMixin
from .session_manager import SessionManagerMixin
from .async_safety import AsyncSafeState
from .unified_execution_mixin import UnifiedExecutionMixin
from .sandbox_mixin import SandboxMixin
# Module-level logger for thread safety errors and debugging
logger = get_logger(__name__)
@@ -256,7 +255,7 @@ if TYPE_CHECKING:
# Import structured error from central errors module
from ..errors import BudgetExceededError
class Agent(SandboxMixin, UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, SessionManagerMixin, ChatMixin, ExecutionMixin, MemoryMixin, AsyncMemoryMixin):
class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, SessionManagerMixin, ChatMixin, ExecutionMixin, MemoryMixin, AsyncMemoryMixin):
# Class-level counter for generating unique display names for nameless agents
_agent_counter = 0
_agent_counter_lock = threading.Lock()
@@ -556,8 +555,6 @@ class Agent(SandboxMixin, UnifiedExecutionMixin, ToolExecutionMixin, ChatHandler
backend: Optional[Any] = None, # External managed agent backend (e.g., ManagedAgentIntegration)
cli_backend: Optional[Union[str, Any]] = None, # CLI backend for delegating turns (e.g., "claude-code")
interrupt_controller: Optional['InterruptController'] = None, # G2: Cooperative cancellation
tool_search: Optional[Union[bool, str, Dict[str, Any], 'ToolSearchConfig']] = False, # Progressive tool disclosure
sandbox: Optional[Union[bool, 'SandboxConfig']] = None, # Sandbox for safe code execution
):
"""Initialize an Agent instance.
@@ -661,14 +658,6 @@ class Agent(SandboxMixin, UnifiedExecutionMixin, ToolExecutionMixin, ChatHandler
When provided, agent delegates entire conversation turns to the CLI tool
instead of using the built-in LLM. Enables session continuity and
tool integration through external AI coding assistants.
tool_search: Progressive tool disclosure configuration. Accepts:
- bool: False=disabled (default), True=auto mode
- str: Mode ("auto", "on", "off")
- Dict[str, Any]: Config overrides (e.g. {"threshold_pct": 15})
- ToolSearchConfig: Custom configuration
When enabled, replaces large tool schemas with bridge tools (tool_search,
tool_describe, tool_call) to save context. Core SDK tools never defer.
Auto mode activates based on token threshold. Opt-in feature.
Raises:
ValueError: If all of name, role, goal, backstory, and instructions are None.
@@ -1439,34 +1428,6 @@ class Agent(SandboxMixin, UnifiedExecutionMixin, ToolExecutionMixin, ChatHandler
web_search = False
web_fetch = False
# ─────────────────────────────────────────────────────────────────────
# Resolve TOOL_SEARCH param
# ─────────────────────────────────────────────────────────────────────
# Fast path: None/False -> disabled (zero overhead)
if tool_search is None or tool_search is False:
self._tool_search_config = None
elif tool_search is True:
# True -> auto mode with defaults
from ..tools.tool_search import ToolSearchConfig as _ToolSearchConfig
self._tool_search_config = _ToolSearchConfig(enabled="auto")
elif isinstance(tool_search, str):
# String mode ("auto", "on", "off")
from ..tools.tool_search import ToolSearchConfig as _ToolSearchConfig
self._tool_search_config = _ToolSearchConfig.from_raw(tool_search)
elif isinstance(tool_search, dict):
# Dict -> config overrides
from ..tools.tool_search import ToolSearchConfig as _ToolSearchConfig
self._tool_search_config = _ToolSearchConfig(**tool_search)
else:
from ..tools.tool_search import ToolSearchConfig as _ToolSearchConfig
if isinstance(tool_search, _ToolSearchConfig):
self._tool_search_config = tool_search
else:
raise TypeError(
"tool_search must be False/None, True, a mode string, "
"a dict of ToolSearchConfig fields, or ToolSearchConfig"
)
# ============================================================
# END CONSOLIDATED PARAMS EXTRACTION
# ============================================================
@@ -1985,133 +1946,6 @@ Your Goal: {self.goal}
# Telemetry - lazy initialized via property for performance
self.__telemetry = None
self.__telemetry_initialized = False
# Sandbox configuration - initialize SandboxMixin
super().__init__(sandbox=sandbox)
def __deepcopy__(self, memo: dict) -> "Agent":
"""Custom deepcopy that creates fresh threading primitives.
threading.RLock (self.__cache_lock) and threading.Lock (self._cost_lock)
cannot be deep-copied on CPython < 3.13. This hook deep-copies every
other attribute normally and replaces the locks with new instances so
that copy.deepcopy(agent) works in any Python version.
"""
import copy
cls = self.__class__
result = cls.__new__(cls)
memo[id(self)] = result
for k, v in self.__dict__.items():
if k in ("_Agent__cache_lock",):
object.__setattr__(result, k, threading.RLock())
elif k == "_cost_lock":
object.__setattr__(result, k, threading.Lock())
else:
object.__setattr__(result, k, copy.deepcopy(v, memo))
return result
def clone_for_channel(self) -> "Agent":
"""Return a fully independent copy of this agent for a gateway channel.
This method safely clones an agent for use in multi-channel scenarios
by avoiding deepcopy issues with tools/handoffs and creating fresh
instances of mutable objects like interrupt_controller.
Returns:
Agent: A new Agent instance with copied configuration but fresh
locks, interrupt controller, and shallow-copied tools.
"""
import copy
from ..tools.base import BaseTool
# Build kwargs for new Agent instance, copying core attributes
clone_kwargs = {
# Core identity
'name': self.name,
'role': self.role,
'goal': self.goal,
'backstory': self.backstory,
'instructions': self.instructions,
# LLM configuration
'llm': self.llm,
'base_url': getattr(self, 'base_url', None),
'api_key': getattr(self, 'api_key', None),
'auth': getattr(self, 'auth', None),
# Shallow copy tools to avoid deepcopy issues with nested objects
'tools': list(self.tools) if self.tools else None,
# Skip handoffs entirely - they shouldn't be shared across channels
# and can contain nested Agent instances that cause RLock issues
'handoffs': None,
# Feature configurations - check for actual stored config objects
'memory': getattr(self, '_memory_config', None),
'knowledge': getattr(self, '_knowledge_config', None),
'planning': getattr(self, '_planning_config', None),
'reflection': getattr(self, '_reflection_config', None),
'guardrails': getattr(self, '_guardrails_config', None),
'web': getattr(self, '_web_config', None),
'context': getattr(self, '_context_config', None),
'autonomy': getattr(self, '_autonomy_config', None),
'output': getattr(self, '_output_config', None),
'execution': getattr(self, '_execution_config', None),
'templates': getattr(self, '_template_config', None),
'caching': getattr(self, '_caching_config', None),
'hooks': getattr(self, '_hooks_config', None),
'skills': getattr(self, '_skills_config', None),
'approval': getattr(self, '_approval_config', None),
'learn': getattr(self, '_learn_config', None),
'tool_search': getattr(self, '_tool_search_config', None),
# Tool configuration
'tool_timeout': getattr(self, '_tool_timeout', None),
'parallel_tool_calls': getattr(self, 'parallel_tool_calls', False),
# CLI backend
'cli_backend': getattr(self, '_cli_backend', None),
# Create fresh interrupt controller to avoid shared state
'interrupt_controller': None, # Let new instance create its own
# Sandbox config
'sandbox': getattr(self, '_sandbox_config', None),
}
# Handle deprecated parameters for backward compatibility
import warnings
# Check for deprecated standalone attributes and emit warnings
deprecated_attrs = [
('allow_code_execution', 'execution'),
('code_execution_mode', 'execution'),
('auto_save', 'memory'),
('rate_limiter', 'execution'),
('allow_delegation', 'handoffs'),
('verification_hooks', 'autonomy')
]
for old_attr, new_param in deprecated_attrs:
if hasattr(self, old_attr):
value = getattr(self, old_attr)
if value is not None and value != False: # Skip None/False defaults
warnings.warn(
f"Deprecated attribute '{old_attr}' found in agent clone. "
f"Use '{new_param}=' parameter instead.",
DeprecationWarning,
stacklevel=2
)
# For basic compatibility, include in clone_kwargs if not already set
if old_attr == 'allow_code_execution' and clone_kwargs['execution'] is None:
from ..config.execution import ExecutionConfig
clone_kwargs['execution'] = ExecutionConfig(code_execution=value)
elif old_attr == 'auto_save' and clone_kwargs['memory'] is None:
from ..config.memory import MemoryConfig
clone_kwargs['memory'] = MemoryConfig(auto_save=value)
# Create new Agent instance
return self.__class__(**{k: v for k, v in clone_kwargs.items() if v is not None})
@property
def _telemetry(self):
@@ -4777,57 +4611,60 @@ Answer:"""
error=f"Agent guardrail validation error: {str(e)}"
)
def _validate_with_guardrail(self, response_text):
"""Validate response with guardrail. Returns (success, result, error)."""
def _apply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
"""Apply guardrail validation with retry logic.
Args:
response_text: The response to validate
prompt: Original prompt for regeneration if needed
temperature: Temperature for regeneration
tools: Tools for regeneration
Returns:
str: The validated response text or None if validation fails after retries
"""
if not self._guardrail_fn:
return True, response_text, None
return response_text
from ..main import TaskOutput
task_output = TaskOutput(
description="Agent response output",
raw=response_text,
agent=self.name
)
guardrail_result = self._process_guardrail(task_output)
if guardrail_result.success:
# Return the potentially modified result
if guardrail_result.result and hasattr(guardrail_result.result, 'raw'):
return True, guardrail_result.result.raw, None
elif guardrail_result.result:
return True, str(guardrail_result.result), None
else:
return True, response_text, None
else:
return False, None, guardrail_result.error
def _apply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
"""Apply guardrail validation with retry logic (sync version)."""
retry_count = 0
current_response = response_text
while retry_count <= self.max_guardrail_retries:
success, result, error = self._validate_with_guardrail(current_response)
# Create TaskOutput object
task_output = TaskOutput(
description="Agent response output",
raw=current_response,
agent=self.name
)
if success:
# Process guardrail
guardrail_result = self._process_guardrail(task_output)
if guardrail_result.success:
logging.info(f"Agent {self.name}: Guardrail validation passed")
return result
# Return the potentially modified result
if guardrail_result.result and hasattr(guardrail_result.result, 'raw'):
return guardrail_result.result.raw
elif guardrail_result.result:
return str(guardrail_result.result)
else:
return current_response
# Guardrail failed
if retry_count >= self.max_guardrail_retries:
raise Exception(
f"Agent {self.name} response failed guardrail validation after {self.max_guardrail_retries} retries. "
f"Last error: {error}"
f"Last error: {guardrail_result.error}"
)
retry_count += 1
logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {error}")
logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {guardrail_result.error}")
# Regenerate response for retry
try:
retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {error}. Please provide an improved response."
retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {guardrail_result.error}. Please provide an improved response."
response = self._chat_completion([{"role": "user", "content": retry_prompt}], temperature, tools, task_name=task_name, task_description=task_description, task_id=task_id)
if response and response.choices:
content = response.choices[0].message.content
@@ -4836,46 +4673,10 @@ Answer:"""
raise Exception("Failed to generate retry response")
except Exception as e:
logging.error(f"Agent {self.name}: Error during guardrail retry: {e}")
raise Exception(f"Agent {self.name} guardrail retry failed: {e}")
return current_response
async def _aapply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
"""Apply guardrail validation with retry logic (async version)."""
retry_count = 0
current_response = response_text
while retry_count <= self.max_guardrail_retries:
success, result, error = self._validate_with_guardrail(current_response)
if success:
logging.info(f"Agent {self.name}: Guardrail validation passed")
return result
# Guardrail failed
if retry_count >= self.max_guardrail_retries:
# If we can't regenerate, fail the guardrail
raise Exception(
f"Agent {self.name} response failed guardrail validation after {self.max_guardrail_retries} retries. "
f"Last error: {error}"
f"Agent {self.name} guardrail retry failed: {e}"
)
retry_count += 1
logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {error}")
# Regenerate response for retry (async version)
try:
retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {error}. Please provide an improved response."
response = await self._execute_unified_achat_completion([{"role": "user", "content": retry_prompt}], temperature, tools, task_name=task_name, task_description=task_description, task_id=task_id)
if response and hasattr(response, 'choices') and response.choices:
content = response.choices[0].message.content
current_response = content.strip() if content else ""
elif isinstance(response, str):
current_response = response.strip()
else:
raise Exception("Failed to generate retry response")
except Exception as e:
logging.error(f"Agent {self.name}: Error during guardrail retry: {e}")
raise Exception(f"Agent {self.name} guardrail retry failed: {e}")
return current_response
@@ -331,7 +331,6 @@ Your Goal: {self.goal}"""
- Callable functions
- String function names
- Objects with to_openai_tool() method
- Tool Search progressive disclosure (if enabled)
Args:
tools: List of tools in various formats or None to use self.tools
@@ -345,14 +344,10 @@ Your Goal: {self.goal}"""
if not tools:
return []
# Check cache first - include tool_search config in cache key for safety
# Check cache first
tools_key = self._get_tools_cache_key(tools)
tool_search_enabled = getattr(self, '_tool_search_config', None) is not None
cache_key = f"{tools_key}:tool_search={tool_search_enabled}"
cached_entry = self._cache_get(self._formatted_tools_cache, cache_key)
if cached_entry is not None:
cached_tools, cached_metadata = cached_entry
self._tool_search_metadata = cached_metadata
cached_tools = self._cache_get(self._formatted_tools_cache, tools_key)
if cached_tools is not None:
return cached_tools
formatted_tools = []
@@ -404,46 +399,9 @@ Your Goal: {self.goal}"""
logging.error(f"Tools are not JSON serializable: {e}")
return []
# Apply tool search assembly if enabled (after formatting, before caching)
if hasattr(self, '_tool_search_config') and self._tool_search_config is not None:
try:
from ..tools.tool_search import assemble_tool_defs
# Get context length from LLM config if available
context_length = getattr(self, '_context_window_size', None)
# Assemble tools with bridge mode check
assembled_tools, metadata = assemble_tool_defs(
tool_defs=formatted_tools,
config=self._tool_search_config,
context_length=context_length
)
# Store metadata for bridge tool dispatch
self._tool_search_metadata = metadata
formatted_tools = assembled_tools
except ImportError:
# Tool search module not available, continue with original tools
logging.warning("Tool search requested but tool_search module not available")
# Strip __praisonai_deferrable__ from provider-facing tool payloads
# Keep the marker only for internal tool classification
cleaned_tools = []
for tool in formatted_tools:
if isinstance(tool, dict) and "__praisonai_deferrable__" in tool:
tool_copy = tool.copy()
tool_copy.pop("__praisonai_deferrable__", None)
cleaned_tools.append(tool_copy)
else:
cleaned_tools.append(tool)
# Cache the formatted tools with LRU eviction, including tool search metadata
self._cache_put(
self._formatted_tools_cache,
cache_key,
(cleaned_tools, getattr(self, "_tool_search_metadata", None)),
)
return cleaned_tools
# Cache the formatted tools with LRU eviction
self._cache_put(self._formatted_tools_cache, tools_key, formatted_tools)
return formatted_tools
def _build_multimodal_prompt(
self,
@@ -549,7 +507,7 @@ Your Goal: {self.goal}"""
emit_events=True
)
def _chat_completion(self, messages, temperature=1.0, tools=None, stream=None, reasoning_steps=False, task_name=None, task_description=None, task_id=None, response_format=None, _retry_depth=0):
def _chat_completion(self, messages, temperature=1.0, tools=None, stream=True, reasoning_steps=False, task_name=None, task_description=None, task_id=None, response_format=None, _retry_depth=0):
start_time = time.time()
# --- Context compaction (opt-in via ExecutionConfig.context_compaction) ---
@@ -619,39 +577,6 @@ Your Goal: {self.goal}"""
# Use the new _format_tools_for_completion helper method
formatted_tools = self._format_tools_for_completion(tools)
# Smart fallback for streaming: try streaming first, fall back to non-streaming if unsupported
if stream is None:
# Auto-detect: prefer streaming for better UX, fallback if adapter doesn't support it
try:
# First attempt: try with streaming enabled for better user experience
stream_callback = self.stream_emitter.emit if hasattr(self, 'stream_emitter') else None
final_response = self._execute_unified_chat_completion(
messages=messages,
temperature=temperature,
tools=formatted_tools,
stream=True, # Try streaming first
reasoning_steps=reasoning_steps,
task_name=task_name,
task_description=task_description,
task_id=task_id,
response_format=response_format,
stream_callback=stream_callback,
emit_events=True
)
return final_response
except ValueError as e:
if "Streaming is not supported" in str(e):
# Fallback: retry with non-streaming for sync adapters
logging.debug(f"{self.name}: Streaming not supported by adapter, falling back to non-streaming")
stream = False # Set for the main execution below
else:
raise # Re-raise if it's a different ValueError
except Exception:
# For any other exception, fall back to non-streaming
logging.debug(f"{self.name}: Streaming attempt failed, falling back to non-streaming")
stream = False # Set for the main execution below
# If stream was explicitly set or fallback occurred, use the specified/fallback value
try:
# NEW: Unified protocol dispatch path (Issue #1304, #1362)
# UNIFIED: Single protocol-driven dispatch path (fixes DRY violation)
@@ -828,11 +753,11 @@ Your Goal: {self.goal}"""
# Create LLMError with contextual metadata
error = LLMError(
str(e),
model_name=model_name,
agent_id=self.name,
str(e),
model_name=model_name,
agent_id=self.name,
is_retryable=is_retryable,
context={"session_id": session_id},
session_id=session_id
)
# Call error hook if available for error interception
@@ -849,7 +774,7 @@ Your Goal: {self.goal}"""
messages,
temperature=1.0,
tools=None,
stream=None,
stream=True,
reasoning_steps=False,
task_name=None,
task_description=None,
@@ -881,47 +806,6 @@ Your Goal: {self.goal}"""
# Cache the dispatcher
self._unified_dispatcher = dispatcher
# Smart fallback for streaming: try streaming first, fall back to non-streaming if unsupported
if stream is None:
# Auto-detect: prefer streaming for better UX, fallback if adapter doesn't support it
try:
# First attempt: try with streaming enabled for better user experience
if stream_callback is None and hasattr(self, 'stream_emitter'):
stream_callback = getattr(self.stream_emitter, 'emit', None)
final_response = self._unified_dispatcher.chat_completion(
messages=messages,
tools=tools,
tool_choice=getattr(self, 'tool_choice', None),
temperature=temperature,
max_tokens=getattr(self, 'max_tokens', None),
stream=True, # Try streaming first
response_format=response_format,
execute_tool_fn=getattr(self, 'execute_tool', None),
console=self.console if (self.verbose or True) else None, # Enable console for streaming
display_fn=self._display_generating if self.verbose else None,
stream_callback=stream_callback,
emit_events=emit_events,
verbose=self.verbose,
max_iterations=10,
reasoning_steps=reasoning_steps,
task_name=task_name,
task_description=task_description,
task_id=task_id,
agent_name=getattr(self, 'name', 'assistant')
)
return final_response
except ValueError as e:
if "Streaming is not supported" in str(e):
# Fallback: retry with non-streaming for sync adapters
logging.debug(f"Agent: Streaming not supported by adapter, falling back to non-streaming")
stream = False # Set for the main execution below
else:
raise # Re-raise if it's a different ValueError
except Exception:
# For any other exception, fall back to non-streaming
logging.debug(f"Agent: Streaming attempt failed, falling back to non-streaming")
stream = False # Set for the main execution below
# Execute unified dispatch with all necessary parameters
# Includes all parameters from both legacy paths to ensure full compatibility
try:
@@ -973,7 +857,7 @@ Your Goal: {self.goal}"""
messages,
temperature=1.0,
tools=None,
stream=True, # Async methods keep stream=True default (async adapters support streaming vs sync smart fallback)
stream=True,
reasoning_steps=False,
task_name=None,
task_description=None,
@@ -1018,7 +902,7 @@ Your Goal: {self.goal}"""
max_tokens=getattr(self, 'max_tokens', None),
stream=stream,
response_format=response_format,
execute_tool_fn=getattr(self, 'execute_tool_async', None),
execute_tool_fn=getattr(self, 'execute_tool', None),
console=self.console if (self.verbose or stream) else None,
display_fn=self._display_generating if self.verbose else None,
stream_callback=stream_callback,
@@ -1051,151 +935,6 @@ Your Goal: {self.goal}"""
logging.error(f"Unified async chat completion failed: {e}")
raise
async def _finalize_unified_achat_response(
self,
response,
original_prompt,
temperature,
tools,
output_json,
output_pydantic,
reasoning_steps,
stream,
messages,
chat_history_length,
start_time,
task_name=None,
task_description=None,
task_id=None,
):
"""Post-process unified achat completion (parity with sync _chat_impl)."""
if not response:
self._truncate_chat_history(chat_history_length)
return None
if hasattr(response, "choices") and response.choices:
msg = response.choices[0].message
content = getattr(msg, "content", None)
response_text = content.strip() if content else ""
if (
reasoning_steps
and getattr(msg, "reasoning_content", None)
):
response_text = msg.reasoning_content.strip()
else:
extracted = self._extract_llm_response_content(response)
response_text = (extracted or "").strip() if isinstance(extracted, str) else str(extracted or "")
if output_json or output_pydantic:
self._append_to_chat_history({"role": "assistant", "content": response_text})
self._persist_message("assistant", response_text)
try:
validated_response = self._apply_guardrail_with_retry(
response_text, original_prompt, temperature, tools,
task_name, task_description, task_id,
)
self._execute_callback_and_display(
original_prompt, validated_response, time.time() - start_time,
task_name, task_description, task_id,
)
return await self._atrigger_after_agent_hook(
original_prompt, validated_response, start_time,
)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for JSON output: {e}")
self._truncate_chat_history(chat_history_length)
return None
if not self.self_reflect:
self._append_to_chat_history({"role": "assistant", "content": response_text})
self._persist_message("assistant", response_text)
try:
validated_response = self._apply_guardrail_with_retry(
response_text, original_prompt, temperature, tools,
task_name, task_description, task_id,
)
self._execute_callback_and_display(
original_prompt, validated_response, time.time() - start_time,
task_name, task_description, task_id,
)
return await self._atrigger_after_agent_hook(
original_prompt, validated_response, start_time,
)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
self._truncate_chat_history(chat_history_length)
return None
# Self-reflection (legacy OpenAI path)
reflection_count = 0
while True:
reflection_prompt = f"""
Reflect on your previous response: '{response_text}'.
{self.reflect_prompt if self.reflect_prompt else "Identify any flaws, improvements, or actions."}
Provide a "satisfactory" status ('yes' or 'no').
Output MUST be JSON with 'reflection' and 'satisfactory'.
"""
reflection_messages = messages + [
{"role": "assistant", "content": response_text},
{"role": "user", "content": reflection_prompt},
]
try:
if self._openai_client is None:
self._append_to_chat_history({"role": "assistant", "content": response_text})
return await self._atrigger_after_agent_hook(
original_prompt, response_text, start_time,
)
reflection_response = await self._openai_client.async_client.beta.chat.completions.parse(
model=self.reflect_llm if self.reflect_llm else self.llm,
messages=reflection_messages,
temperature=temperature,
response_format=_get_display_functions()['ReflectionOutput'],
)
reflection_output = reflection_response.choices[0].message.parsed
if reflection_output.satisfactory == "yes" and reflection_count >= self.min_reflect - 1:
break
if reflection_count >= self.max_reflect - 1:
break
regenerate_messages = reflection_messages + [
{
"role": "assistant",
"content": (
f"Self Reflection: {reflection_output.reflection} "
f"Satisfactory?: {reflection_output.satisfactory}"
),
},
{"role": "user", "content": "Now regenerate your response using the reflection you made"},
]
new_response = await self._openai_client.async_client.chat.completions.create(
model=self.llm,
messages=regenerate_messages,
temperature=temperature,
)
response_text = new_response.choices[0].message.content
reflection_count += 1
except Exception as e:
logging.error("Reflection parsing failed.", exc_info=True)
reflection_count += 1
if reflection_count >= self.max_reflect:
break
try:
validated_response = self._apply_guardrail_with_retry(
response_text, original_prompt, temperature, tools,
task_name, task_description, task_id,
)
self._execute_callback_and_display(
original_prompt, validated_response, time.time() - start_time,
task_name, task_description, task_id,
)
return await self._atrigger_after_agent_hook(
original_prompt, validated_response, start_time,
)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for OpenAI client: {e}")
self._truncate_chat_history(chat_history_length)
return None
def _execute_callback_and_display(self, prompt: str, response: str, generation_time: float, task_name=None, task_description=None, task_id=None):
"""Helper method to execute callbacks and display interaction.
@@ -2195,7 +1934,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
# Apply guardrail validation for custom LLM response
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, prompt, temperature, tools, task_name, task_description, task_id)
validated_response = self._apply_guardrail_with_retry(response_text, prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(normalized_content, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(prompt, validated_response, start_time)
@@ -2303,216 +2042,8 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
task_id=task_id,
response_format=response_format
)
# Process response - mirror sync _chat_impl behavior (lines ~1544-1602)
if not response:
# Rollback chat history on response failure
self._truncate_chat_history(chat_history_length)
return None
# Extract response content using the same method as sync
response_text = self._extract_llm_response_content(response)
if isinstance(response_text, str):
response_text = response_text.strip()
# Handle output_json or output_pydantic if specified
if output_json or output_pydantic:
# Add to chat history and return raw response
# User message already added before LLM call via _build_messages
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
# Apply guardrail validation even for JSON output
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for JSON output: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
# For regular responses (no self-reflection)
if not self.self_reflect:
# User message already added before LLM call via _build_messages
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB (non-reflect path)
self._persist_message("assistant", response_text)
if self.verbose:
logging.debug(f"Agent {self.name} final response: {response_text}")
# Return only reasoning content if reasoning_steps is True
if reasoning_steps and hasattr(response, 'choices') and response.choices and hasattr(response.choices[0].message, 'reasoning_content') and response.choices[0].message.reasoning_content:
# Apply guardrail to reasoning content
try:
validated_reasoning = await self._aapply_guardrail_with_retry(response.choices[0].message.reasoning_content, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_reasoning, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_reasoning, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for reasoning content: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
else:
# Apply guardrail to regular response content
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
# If self-reflection is enabled, implement reflection logic
if self.self_reflect:
# Implement async self-reflection similar to sync path
reflection_prompt = f"""
Reflect on your previous response: '{response_text}'.
{self.reflect_prompt if self.reflect_prompt else "Identify any flaws, improvements, or actions."}
Provide a "satisfactory" status ('yes' or 'no').
Output MUST be JSON with 'reflection' and 'satisfactory'.
"""
# Add reflection prompt to messages
reflection_messages = messages + [
{"role": "assistant", "content": response_text},
{"role": "user", "content": reflection_prompt}
]
reflection_count = 0
while True:
try:
# Check if OpenAI client is available for self-reflection
if self._openai_client is None:
# For custom LLMs, self-reflection with structured output is not supported
if self.verbose:
_get_display_functions()['display_self_reflection'](f"Agent {self.name}: Self-reflection with structured output is not supported for custom LLM providers. Skipping reflection.", console=self.console)
# Return the original response without reflection
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
reflection_response = await self._openai_client.async_client.beta.chat.completions.parse(
model=self.reflect_llm if self.reflect_llm else self.llm,
messages=reflection_messages,
temperature=temperature,
response_format=_get_display_functions()['ReflectionOutput']
)
reflection_output = reflection_response.choices[0].message.parsed
if self.verbose:
_get_display_functions()['display_self_reflection'](f"Agent {self.name} self reflection (using {self.reflect_llm if self.reflect_llm else self.llm}): reflection='{reflection_output.reflection}' satisfactory='{reflection_output.satisfactory}'", console=self.console)
# Only consider satisfactory after minimum reflections
if reflection_output.satisfactory == "yes" and reflection_count >= self.min_reflect - 1:
if self.verbose:
_get_display_functions()['display_self_reflection']("Agent marked the response as satisfactory after meeting minimum reflections", console=self.console)
# Add to chat history and return
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed after reflection: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
# Check if we've hit max reflections
if reflection_count >= self.max_reflect - 1:
if self.verbose:
_get_display_functions()['display_self_reflection']("Maximum reflection count reached, returning current response", console=self.console)
# Add to chat history and return
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed after max reflections: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
# Regenerate response based on reflection
regenerate_messages = reflection_messages + [
{"role": "assistant", "content": f"Self Reflection: {reflection_output.reflection} Satisfactory?: {reflection_output.satisfactory}"},
{"role": "user", "content": "Now regenerate your response using the reflection you made"}
]
new_response = await self._execute_unified_achat_completion(
messages=regenerate_messages,
temperature=temperature,
tools=formatted_tools,
stream=stream,
reasoning_steps=reasoning_steps,
task_name=task_name,
task_description=task_description,
task_id=task_id,
response_format=response_format
)
if new_response:
new_response_text = self._extract_llm_response_content(new_response)
if isinstance(new_response_text, str):
response_text = new_response_text.strip()
# Update reflection_messages to include the new response for next iteration
reflection_messages = regenerate_messages + [
{"role": "assistant", "content": response_text}
]
reflection_count += 1
except Exception as e:
if self.verbose:
_get_display_functions()['display_error'](f"Error in parsing self-reflection json {e}. Retrying", console=self.console)
logging.error("Reflection parsing failed.", exc_info=True)
reflection_count += 1
if reflection_count >= self.max_reflect:
# Return original response after max reflection attempts
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as guard_e:
logging.error(f"Agent {self.name}: Guardrail validation failed after reflection error: {guard_e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
continue
# This should never be reached due to the returns above
# But adding as safety fallback
logging.warning(f"Agent {self.name}: Unexpected code path reached in unified dispatch")
return None
# Continue to handle structured outputs and other processing
# Don't return immediately - fall through to existing logic
else:
# LEGACY: Check if OpenAI client is available
if self._openai_client is None:
@@ -2649,7 +2180,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
# Apply guardrail validation for OpenAI client response
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
validated_response = self._apply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
@@ -164,8 +164,7 @@ Follow best practices and coding standards."""
messages=messages,
**{k: v for k, v in kwargs.items() if k != "model"}
)
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
code = response.choices[0].message.content
self._log("Code generated successfully", "green")
return code
@@ -205,8 +204,7 @@ Follow best practices and coding standards."""
messages=messages,
**{k: v for k, v in kwargs.items() if k != "model"}
)
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
code = response.choices[0].message.content
self._log("Code generated successfully", "green")
return code
@@ -367,8 +365,6 @@ Provide clear, actionable feedback."""
**{k: v for k, v in kwargs.items() if k != "model"}
)
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
review = response.choices[0].message.content
self._log("Review completed", "green")
return review
@@ -409,8 +405,6 @@ Include:
**{k: v for k, v in kwargs.items() if k != "model"}
)
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
explanation = response.choices[0].message.content
self._log("Explanation completed", "green")
return explanation
@@ -457,8 +451,6 @@ Only output the refactored code."""
**{k: v for k, v in kwargs.items() if k != "model"}
)
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
refactored = response.choices[0].message.content
self._log("Refactoring completed", "green")
return refactored
@@ -503,8 +495,6 @@ Ensure the fix:
**{k: v for k, v in kwargs.items() if k != "model"}
)
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
fixed = response.choices[0].message.content
self._log("Fix completed", "green")
return fixed
@@ -403,33 +403,9 @@ class MemoryMixin:
self._session_store.add_user_message(self._session_id, content)
elif role == "assistant":
self._session_store.add_assistant_message(self._session_id, content)
self._persist_session_stats()
except Exception as e:
logging.warning(f"Failed to persist message to session store: {e}")
def _persist_session_stats(self):
"""Flush agent cost/token stats into session JSON metadata."""
store = getattr(self, "_session_store", None)
session_id = getattr(self, "_session_id", None)
if store is None or session_id is None:
return
if not hasattr(store, "update_session_metadata"):
return
try:
summary = getattr(self, "cost_summary", {}) or {}
llm = getattr(self, "llm", None)
model = str(llm) if llm is not None else None
store.update_session_metadata(
session_id,
model=model,
total_tokens=summary.get("tokens_in", 0) + summary.get("tokens_out", 0),
cost=summary.get("cost", 0),
source=getattr(self, "_session_source", "chat"),
agent_id=getattr(self, "agent_id", None) or getattr(self, "_agent_id", None),
)
except Exception as e:
logging.debug(f"Session stats persist skipped: {e}")
def session_id(self) -> Optional[str]:
"""Get the current session ID."""
return self._session_id
@@ -1,231 +0,0 @@
"""
Sandbox execution mixin for Agent.
Provides safe code execution capabilities to agents via the sandbox framework.
"""
import logging
from typing import Optional, Dict, Any, Union
from ..sandbox import (
SandboxConfig,
SandboxManager,
SandboxResult,
check_code_safety,
format_warnings,
)
logger = logging.getLogger(__name__)
class SandboxMixin:
"""Mixin that adds sandbox execution capabilities to Agent.
Provides safe code execution in isolated environments with security pre-checks.
"""
def __init__(self, *args, **kwargs):
# Extract sandbox config from kwargs
self.sandbox_config: Optional[SandboxConfig] = kwargs.pop('sandbox', None)
# Convert boolean True to default config
if self.sandbox_config is True:
self.sandbox_config = SandboxConfig.subprocess()
super().__init__(*args, **kwargs)
self._sandbox_manager: Optional[SandboxManager] = None
@property
def has_sandbox(self) -> bool:
"""Check if sandbox is configured for this agent."""
return self.sandbox_config is not None
def get_sandbox_manager(self) -> Optional[SandboxManager]:
"""Get or create sandbox manager."""
if not self.has_sandbox:
return None
if self._sandbox_manager is None:
self._sandbox_manager = SandboxManager(self.sandbox_config)
return self._sandbox_manager
async def execute_code(
self,
code: str,
language: str = "python",
check_security: bool = True,
**kwargs
) -> SandboxResult:
"""Execute code safely in configured sandbox.
Args:
code: Code to execute
language: Programming language (python, bash, etc.)
check_security: Whether to run security pre-checks
**kwargs: Additional arguments passed to sandbox.execute()
Returns:
SandboxResult with execution details
Raises:
RuntimeError: If no sandbox is configured
"""
if not self.has_sandbox:
raise RuntimeError(
"No sandbox configured. Set sandbox=True or provide SandboxConfig to Agent()"
)
# Security pre-check if enabled
if check_security:
warnings = check_code_safety(code, language)
if warnings:
warning_text = format_warnings(warnings)
if self.verbose:
logger.warning(f"Security warnings for code execution:\n{warning_text}")
# Store warnings in result metadata
kwargs.setdefault('metadata', {})['security_warnings'] = warnings
manager = self.get_sandbox_manager()
return await manager.run_code(code, language=language, **kwargs)
def execute_code_sync(
self,
code: str,
language: str = "python",
check_security: bool = True,
**kwargs
) -> SandboxResult:
"""Synchronous wrapper for execute_code.
Args:
code: Code to execute
language: Programming language
check_security: Whether to run security pre-checks
**kwargs: Additional arguments
Returns:
SandboxResult with execution details
"""
from ..approval.utils import run_coroutine_safely
return run_coroutine_safely(self.execute_code(code, language, check_security, **kwargs))
async def run_shell_command(
self,
command: Union[str, list],
check_security: bool = True,
**kwargs
) -> SandboxResult:
"""Run a shell command in the sandbox.
Args:
command: Shell command to run
check_security: Whether to run security pre-checks
**kwargs: Additional arguments
Returns:
SandboxResult with execution details
"""
if not self.has_sandbox:
raise RuntimeError("No sandbox configured")
# Convert command to string for security checking
if isinstance(command, list):
command_str = " ".join(command)
else:
command_str = command
# Security pre-check
if check_security:
warnings = check_code_safety(command_str, "bash")
if warnings:
warning_text = format_warnings(warnings)
if self.verbose:
logger.warning(f"Security warnings for command:\n{warning_text}")
async with self.get_sandbox_manager() as sandbox:
return await sandbox.run_command(command, **kwargs)
def get_sandbox_status(self) -> Dict[str, Any]:
"""Get sandbox status information.
Returns:
Dictionary with sandbox status details
"""
if not self.has_sandbox:
return {"configured": False}
status = {
"configured": True,
"config": self.sandbox_config.to_dict() if self.sandbox_config else None,
}
manager = self.get_sandbox_manager()
if manager:
# Get available sandbox types
available_types = manager.get_available_types()
status["available_types"] = available_types
status["current_type"] = self.sandbox_config.sandbox_type
return status
async def sandbox_cleanup(self):
"""Clean up sandbox resources."""
if self._sandbox_manager:
# The context manager handles cleanup, but we can force it here
try:
await self._sandbox_manager.__aexit__(None, None, None)
except Exception as e:
logger.warning(f"Error during sandbox cleanup: {e}")
finally:
self._sandbox_manager = None
def _get_code_execution_tools(self):
"""Get sandbox-based code execution tools for this agent.
Returns tools that use the configured sandbox for safe execution.
Returns:
List of tool functions
"""
if not self.has_sandbox:
return []
from ..tools import tool
@tool
def execute_python_code(code: str) -> str:
"""Execute Python code safely in sandbox.
Args:
code: Python code to execute
Returns:
Execution output
"""
result = self.execute_code_sync(code, language="python")
if result.success:
return result.stdout or result.output
else:
return f"Error: {result.error or result.stderr}"
@tool
def execute_shell_command(command: str) -> str:
"""Execute shell command safely in sandbox.
Args:
command: Shell command to execute
Returns:
Command output
"""
from ..approval.utils import run_coroutine_safely
result = run_coroutine_safely(self.run_shell_command(command))
if result.success:
return result.stdout or result.output
else:
return f"Error: {result.error or result.stderr}"
return [execute_python_code, execute_shell_command]
@@ -26,21 +26,6 @@ if TYPE_CHECKING:
class ToolExecutionMixin:
"""Mixin providing toolexecution methods for the Agent class."""
def _get_existing_stream_emitter(self):
"""Return an already-initialized stream emitter without creating one."""
emitter = getattr(self, "_stream_emitter", None)
if emitter is not None:
return emitter
# Support name-mangled private attributes across class renames/inheritance.
for cls in type(self).mro():
mangled = f"_{cls.__name__}__stream_emitter"
if hasattr(self, mangled):
emitter = getattr(self, mangled, None)
if emitter is not None:
return emitter
return None
def _resolve_tool_names(self, tool_names):
"""Resolve tool names to actual tool instances from registry.
@@ -135,12 +120,6 @@ class ToolExecutionMixin:
"""
logging.debug(f"{self.name} executing tool {function_name} with arguments: {arguments}")
# Handle bridge tool unwrapping BEFORE trace/stream/hooks (design invariant #6)
# Only intercept when tool_search is active; otherwise fall through to real tool execution
if (getattr(self, '_tool_search_config', None) is not None and
function_name in ("tool_search", "tool_describe", "tool_call")):
return self._handle_bridge_tool_call(function_name, arguments, tool_call_id)
# NOTE: tool_call callback is triggered by display_tool_call in openai_client.py
# Do NOT call it here to avoid duplicate output
@@ -181,9 +160,8 @@ class ToolExecutionMixin:
# Emit TOOL_CALL_START to stream_emitter (for AIUI/AG-UI consumers)
# Zero overhead when no callbacks registered
_stream_emitter = self._get_existing_stream_emitter()
if _stream_emitter is not None and _stream_emitter.has_callbacks:
_stream_emitter.emit(StreamEvent(
if hasattr(self, '_Agent__stream_emitter') and getattr(self, "_Agent__stream_emitter", None) is not None and getattr(self, "_Agent__stream_emitter", None).has_callbacks:
getattr(self, "_Agent__stream_emitter", None).emit(StreamEvent(
type=StreamEventType.TOOL_CALL_START,
timestamp=_tool_start_perf,
tool_call={
@@ -297,10 +275,10 @@ class ToolExecutionMixin:
# Emit TOOL_CALL_RESULT to stream_emitter (for AIUI/AG-UI consumers)
# Zero overhead when no callbacks registered
if _stream_emitter is not None and _stream_emitter.has_callbacks:
if hasattr(self, '_Agent__stream_emitter') and getattr(self, "_Agent__stream_emitter", None) is not None and getattr(self, "_Agent__stream_emitter", None).has_callbacks:
# Truncate result for stream event (keep it reasonable for UI display)
result_summary = str(result)[:500] if result else None
_stream_emitter.emit(StreamEvent(
getattr(self, "_Agent__stream_emitter", None).emit(StreamEvent(
type=StreamEventType.TOOL_CALL_RESULT,
timestamp=_time.perf_counter(),
tool_call={
@@ -312,18 +290,6 @@ class ToolExecutionMixin:
agent_id=self.name,
metadata={"duration_ms": _duration_ms},
))
_stream_emitter.emit(StreamEvent(
type=StreamEventType.TOOL_CALL_END,
timestamp=_time.perf_counter(),
tool_call={
"name": function_name,
"arguments": arguments,
"result": result_summary,
"id": tool_call_id,
},
agent_id=self.name,
metadata={"duration_ms": _duration_ms},
))
# Trigger AFTER_TOOL hook
from ..hooks import HookEvent, AfterToolInput
@@ -960,97 +926,4 @@ class ToolExecutionMixin:
def pending_approval_count(self) -> int:
"""Number of approval requests still waiting."""
return len(self._pending_approvals)
def _handle_bridge_tool_call(self, function_name: str, arguments: Dict[str, Any], tool_call_id: Optional[str] = None) -> Any:
"""
Handle bridge tool calls (tool_search, tool_describe, tool_call).
This implements the tool search unwrapping logic before trace/stream/hooks
as required by design invariant #6.
Args:
function_name: Bridge tool name
arguments: Arguments passed to bridge tool
tool_call_id: Optional tool call ID
Returns:
Result of bridge tool execution or unwrapped real tool call
"""
# Ensure tool search metadata is available
if not hasattr(self, '_tool_search_metadata') or self._tool_search_metadata is None:
return "Tool search not available or not in bridge mode"
metadata = self._tool_search_metadata
# Check if we're in bridge mode
if not metadata.get("bridge_mode", False):
return "Tool search not in bridge mode"
# Get deferrable tools from metadata
deferrable_tools = metadata.get("deferrable_tools", [])
if function_name == "tool_search":
# Handle tool_search bridge call
try:
from ..tools.tool_search import dispatch_tool_search
query = arguments.get("query", "")
limit = arguments.get("limit", None)
result = dispatch_tool_search(
query=query,
limit=limit,
deferrable_tools=deferrable_tools,
config=self._tool_search_config
)
return json.dumps(result, indent=2)
except ImportError:
return "Tool search module not available"
except Exception as e:
logging.error(f"Error in tool_search: {e}")
return f"Error searching tools: {e}"
elif function_name == "tool_describe":
# Handle tool_describe bridge call
try:
from ..tools.tool_search import dispatch_tool_describe
tool_name = arguments.get("tool_name", "")
result = dispatch_tool_describe(
tool_name=tool_name,
deferrable_tools=deferrable_tools
)
return json.dumps(result, indent=2)
except ImportError:
return "Tool search module not available"
except Exception as e:
logging.error(f"Error in tool_describe: {e}")
return f"Error describing tool: {e}"
elif function_name == "tool_call":
# Handle tool_call bridge - unwrap and recurse with real tool
try:
from ..tools.tool_search import resolve_underlying_call
# Unwrap the real tool call
real_function_name, real_arguments = resolve_underlying_call(function_name, arguments)
# Validate that the real tool is in our deferrable set (security check)
deferrable_names = {
tool_def.get("function", {}).get("name", "")
for tool_def in deferrable_tools
}
if real_function_name not in deferrable_names:
return f"Tool '{real_function_name}' is not available for execution"
# Recursively execute the real tool (this will go through normal execution path)
return self.execute_tool(real_function_name, real_arguments, tool_call_id)
except ImportError:
return "Tool search module not available"
except Exception as e:
logging.error(f"Error in tool_call unwrap: {e}")
return f"Error executing tool: {e}"
else:
return f"Unknown bridge tool: {function_name}"
@@ -15,9 +15,9 @@ from enum import Enum
# Import token tracking
try:
from ..telemetry.token_collector import get_token_collector
from ..telemetry.token_collector import _token_collector
except ImportError:
get_token_collector = None
_token_collector = None
# Import async utility for hot-path usage
try:
@@ -126,24 +126,7 @@ class _AgentServerRegistry:
thread = threading.Thread(target=run_server, daemon=True)
thread.start()
# Check for configurable timeout via environment variable
try:
timeout = float(os.environ.get("PRAISONAI_SERVER_READY_TIMEOUT", "5.0"))
except ValueError:
logger.warning("Invalid PRAISONAI_SERVER_READY_TIMEOUT value. Using default 5.0s.")
timeout = 5.0
became_ready = ready_event.wait(timeout=timeout)
if not became_ready:
logger.warning(
"Agent server on port %s did not become ready within %.1fs. "
"Proceeding, but some features may not work correctly. "
"Check server logs for startup errors.",
port,
timeout,
)
ready_event.wait(timeout=5.0) # Deterministic wait instead of sleep(0.5)
return True
@@ -1609,25 +1592,25 @@ class AgentTeam:
original_on_task_start = self.on_task_start
original_on_task_complete = self.on_task_complete
def composed_on_task_start(task, task_id):
def composed_on_task_start(task):
try:
verbose_task_start_callback(task, task_id)
verbose_task_start_callback(task)
except Exception as e:
logging.debug(f"Error in verbose task start callback: {e}")
if original_on_task_start:
try:
original_on_task_start(task, task_id)
original_on_task_start(task)
except Exception as e:
logging.debug(f"Error in original task start callback: {e}")
def composed_on_task_complete(task, task_output):
def composed_on_task_complete(task):
try:
verbose_task_complete_callback(task, task_output)
verbose_task_complete_callback(task)
except Exception as e:
logging.debug(f"Error in verbose task complete callback: {e}")
if original_on_task_complete:
try:
original_on_task_complete(task, task_output)
original_on_task_complete(task)
except Exception as e:
logging.debug(f"Error in original task complete callback: {e}")
@@ -1826,19 +1809,18 @@ class AgentTeam:
def get_token_usage_summary(self) -> Dict[str, Any]:
"""Get a summary of token usage across all agents and tasks."""
if not get_token_collector:
if not _token_collector:
return {"error": "Token tracking not available"}
return get_token_collector().get_session_summary()
return _token_collector.get_session_summary()
def get_detailed_token_report(self) -> Dict[str, Any]:
"""Get a detailed token usage report."""
if not get_token_collector:
if not _token_collector:
return {"error": "Token tracking not available"}
collector = get_token_collector()
summary = collector.get_session_summary()
recent = collector.get_recent_interactions(limit=20)
summary = _token_collector.get_session_summary()
recent = _token_collector.get_recent_interactions(limit=20)
# Calculate cost estimates (example rates)
cost_per_1k_input = 0.0005 # $0.0005 per 1K input tokens
@@ -1862,11 +1844,11 @@ class AgentTeam:
def display_token_usage(self):
"""Display token usage in a formatted table."""
if not get_token_collector:
if not _token_collector:
print("Token tracking not available")
return
summary = get_token_collector().get_session_summary()
summary = _token_collector.get_session_summary()
print("\n" + "="*50)
print("TOKEN USAGE SUMMARY")
@@ -1,257 +0,0 @@
"""
Tool whitelist filter module with diagnostics.
This module provides a canonical implementation of the ALLOWED_TOOLS filter
that prevents tool name collisions in multi-environment agent systems by
whitelisting only specified tools.
The filter is designed to solve the "tool shadowing" problem where multiple
modules register tools with overlapping or ambiguous names, causing agents
to invoke the wrong implementation.
Environment Variables:
ALLOWED_TOOLS: Primary variable (comma-separated tool names)
HERMES_ONLY_TOOLS: Backward compatibility alias for ALLOWED_TOOLS
Usage:
from praisonaiagents.allowed_tools_filter import AllowedToolsFilter
# Create filter from environment variable
filter = AllowedToolsFilter()
# Filter tools
available_tools = {"search", "send_message", "extract_pdf"}
filtered_tools = filter.filter_tools(available_tools)
# Get diagnostics
filter.log_diagnostics()
"""
import os
import logging
from typing import Any, Dict, List, Optional, Set, Union
logger = logging.getLogger(__name__)
class AllowedToolsFilter:
"""
Canonical tool whitelist filter with diagnostics and backward compatibility.
This filter implements the semantics for tool filtering:
- ALLOWED_TOOLS unset (or HERMES_ONLY_TOOLS): all tools visible (with warning about collisions)
- ALLOWED_TOOLS empty string: error - must specify tools or unset
- ALLOWED_TOOLS with values: only whitelisted tools visible
- Unknown tools in list: warn and strip in dev, strict fail in CI
Backward Compatibility:
- Supports both ALLOWED_TOOLS and HERMES_ONLY_TOOLS
- ALLOWED_TOOLS takes precedence if both are set
"""
def __init__(self, env_var_name: str = "ALLOWED_TOOLS"):
"""
Initialize the filter with backward compatibility support.
Args:
env_var_name: Primary environment variable name (default: "ALLOWED_TOOLS")
"""
# Support both new (ALLOWED_TOOLS) and legacy (HERMES_ONLY_TOOLS) naming
self.primary_var = env_var_name
self.legacy_var = "HERMES_ONLY_TOOLS"
# ALLOWED_TOOLS takes precedence over HERMES_ONLY_TOOLS for backward compatibility
self.env_value = os.environ.get(self.primary_var) or os.environ.get(self.legacy_var)
self.env_var_name = self.primary_var if os.environ.get(self.primary_var) else self.legacy_var
self.is_ci = os.environ.get("CI", "").lower() in ("true", "1", "yes")
self._whitelist: Optional[Set[str]] = self._parse_whitelist()
self._diagnostics: Dict[str, Any] = {}
def _parse_whitelist(self) -> Optional[Set[str]]:
"""
Parse the environment variable value into a whitelist set.
Returns:
None if unset, Set of tool names if set, raises ValueError for empty string
"""
if self.env_value is None:
return None
# Empty string is an error - must be explicit
if self.env_value.strip() == "":
raise ValueError(
f"{self.env_var_name} cannot be empty. Either unset it for all tools "
"or provide a comma-separated list of tool names."
)
# Parse comma-separated values, strip whitespace
tools = {name.strip() for name in self.env_value.split(",") if name.strip()}
if not tools:
raise ValueError(
f"{self.env_var_name} contains no valid tool names. "
"Provide comma-separated tool names or unset the variable."
)
return tools
def filter_tools(self, available_tools: Union[Set[str], List[str], Dict[str, Any]]) -> Set[str]:
"""
Filter tools based on ALLOWED_TOOLS whitelist.
Args:
available_tools: Available tools as set, list, or dict with tool names as keys
Returns:
Filtered set of tool names
Raises:
ValueError: If unknown tools in CI mode or configuration errors
"""
# Convert input to set of tool names
if isinstance(available_tools, dict):
tool_names = set(available_tools.keys())
else:
tool_names = set(available_tools)
# Store for diagnostics
self._diagnostics["registered_before_filter"] = sorted(tool_names)
# If no whitelist, return all tools (with warning)
if self._whitelist is None:
logger.warning(
"Tool whitelist is unset. All %d tools are visible. "
"Consider using ALLOWED_TOOLS to prevent tool name collisions.",
len(tool_names)
)
self._diagnostics["registered_after_filter"] = sorted(tool_names)
self._diagnostics["dropped_tools"] = []
self._diagnostics["unknown_tools"] = []
return tool_names
# Find intersection and unknown tools
available_whitelisted = tool_names.intersection(self._whitelist)
unknown_tools = self._whitelist.difference(tool_names)
dropped_tools = tool_names.difference(self._whitelist)
# Store for diagnostics
self._diagnostics["registered_after_filter"] = sorted(available_whitelisted)
self._diagnostics["dropped_tools"] = sorted(dropped_tools)
self._diagnostics["unknown_tools"] = sorted(unknown_tools)
# Handle unknown tools based on environment
if unknown_tools:
unknown_list = sorted(unknown_tools)
if self.is_ci:
raise ValueError(
f"Unknown tools in {self.env_var_name}: {unknown_list}. "
"All specified tools must be available in CI mode."
)
else:
logger.warning(
"Unknown tools in %s will be ignored: %s",
self.env_var_name,
unknown_list
)
if dropped_tools:
logger.info(
"Dropped %d tools not in %s whitelist: %s",
len(dropped_tools),
self.env_var_name,
sorted(dropped_tools)
)
logger.info(
"%s filtered tools: %d available, %d after filter",
self.env_var_name,
len(tool_names),
len(available_whitelisted)
)
return available_whitelisted
def log_diagnostics(self) -> None:
"""Log startup diagnostics section as specified in the issue."""
if not self._diagnostics:
logger.warning("No diagnostics available. Call filter_tools() first.")
return
logger.info("=" * 50)
logger.info("ALLOWED_TOOLS FILTER DIAGNOSTICS")
logger.info("=" * 50)
logger.info("%s=%s", self.env_var_name, self.env_value or "<unset>")
logger.info("RegisteredBeforeFilter=%s", self._diagnostics.get("registered_before_filter", []))
logger.info("RegisteredAfterFilter=%s", self._diagnostics.get("registered_after_filter", []))
logger.info("DroppedTools=%s", self._diagnostics.get("dropped_tools", []))
logger.info("UnknownTools=%s", self._diagnostics.get("unknown_tools", []))
logger.info("IsCI=%s", self.is_ci)
logger.info("=" * 50)
def get_diagnostics(self) -> Dict[str, Any]:
"""
Get diagnostics data as a dictionary.
Returns:
Dictionary with diagnostics information
"""
return {
"env_var_name": self.env_var_name,
"env_value": self.env_value,
"is_ci": self.is_ci,
"whitelist": list(self._whitelist) if self._whitelist else None,
**self._diagnostics
}
def is_enabled(self) -> bool:
"""
Check if filtering is enabled.
Returns:
True if tool whitelist is set and filtering is active
"""
return self._whitelist is not None
def get_whitelist(self) -> Optional[Set[str]]:
"""
Get the current whitelist.
Returns:
Set of whitelisted tool names, or None if not set
"""
return self._whitelist.copy() if self._whitelist else None
def filter_tools_with_allowed_tools(
available_tools: Union[Set[str], List[str], Dict[str, Any]],
env_var_name: str = "ALLOWED_TOOLS",
log_diagnostics: bool = True
) -> Set[str]:
"""
Convenience function to filter tools with ALLOWED_TOOLS.
Args:
available_tools: Available tools to filter
env_var_name: Environment variable name (default: "ALLOWED_TOOLS")
log_diagnostics: Whether to log diagnostics (default: True)
Returns:
Filtered set of tool names
"""
filter_instance = AllowedToolsFilter(env_var_name)
filtered = filter_instance.filter_tools(available_tools)
if log_diagnostics:
filter_instance.log_diagnostics()
return filtered
# Backward compatibility aliases
filter_tools_with_hermes = filter_tools_with_allowed_tools
hermes_filter = filter_tools_with_allowed_tools
apply_hermes_filter = filter_tools_with_allowed_tools
# Legacy class alias for backward compatibility
HermesToolFilter = AllowedToolsFilter
@@ -39,7 +39,6 @@ __all__ = [
"CachingConfig",
"HooksConfig",
"SkillsConfig",
"ToolSearchConfig",
# Type aliases
"MemoryParam",
"KnowledgeParam",
@@ -123,7 +122,6 @@ _MODULE_MAP = {
"CachingConfig": "feature_configs",
"HooksConfig": "feature_configs",
"SkillsConfig": "feature_configs",
"ToolSearchConfig": "feature_configs",
"MemoryParam": "feature_configs",
"KnowledgeParam": "feature_configs",
"PlanningParam": "feature_configs",
@@ -8,7 +8,6 @@ Provides dataclasses for consolidated feature configuration:
- ReflectionConfig: Self-reflection settings
- GuardrailConfig: Safety and validation
- WebConfig: Web search and fetch
- ToolSearchConfig: Progressive tool disclosure
All configs follow the agent-centric pattern:
- False: Feature disabled (zero overhead)
@@ -17,22 +16,21 @@ All configs follow the agent-centric pattern:
- Instance: Pre-configured manager/engine
Usage:
from praisonaiagents import Agent, MemoryConfig, KnowledgeConfig, ToolSearchConfig
from praisonaiagents import Agent, MemoryConfig, KnowledgeConfig
# Simple enable
agent = Agent(instructions="...", memory=True, tool_search=True)
agent = Agent(instructions="...", memory=True)
# With config
agent = Agent(
instructions="...",
memory=MemoryConfig(backend="redis", user_id="user123"),
knowledge=KnowledgeConfig(sources=["docs/"], rerank=True),
tool_search=ToolSearchConfig(enabled="auto", threshold_pct=15),
)
"""
from dataclasses import dataclass, field
from typing import Dict, List, Any, Optional, Callable, Tuple, Union, FrozenSet
from typing import Dict, List, Any, Optional, Callable, Tuple, Union
from enum import Enum
# Import AutonomyConfig from canonical location (no circular dep)
@@ -44,7 +42,6 @@ class MemoryBackend(str, Enum):
FILE = "file"
SQLITE = "sqlite"
REDIS = "redis"
VALKEY = "valkey"
POSTGRES = "postgres"
MEM0 = "mem0"
MONGODB = "mongodb"
@@ -1074,25 +1071,6 @@ class MultiAgentMemoryConfig:
}
# Import ToolSearchConfig from tools module to avoid duplication
def __get_tool_search_config():
try:
from ..tools.tool_search import ToolSearchConfig as _ToolSearchConfig
return _ToolSearchConfig
except ImportError:
# Fallback minimal config if tools module not available
@dataclass
class FallbackToolSearchConfig:
enabled: Union[bool, str] = "auto"
threshold_pct: float = 10.0
search_default_limit: int = 5
max_search_limit: int = 20
core_tools: Optional[FrozenSet[str]] = None
return FallbackToolSearchConfig
ToolSearchConfig = __get_tool_search_config()
class AutonomyLevel(str, Enum):
"""Autonomy levels for agent behavior."""
SUGGEST = "suggest"
@@ -1113,7 +1091,6 @@ CachingParam = Union[bool, CachingConfig]
HooksParam = Union[List[Any], HooksConfig]
SkillsParam = Union[List[str], SkillsConfig]
AutonomyParam = Union[bool, Dict[str, Any], "AutonomyConfig"]
ToolSearchParam = Union[bool, str, Dict[str, Any], ToolSearchConfig]
# =============================================================================
@@ -1365,21 +1342,6 @@ def resolve_autonomy(value: AutonomyParam) -> Optional[AutonomyConfig]:
return value
def resolve_tool_search(value: ToolSearchParam) -> Optional[ToolSearchConfig]:
"""Resolve tool_search= parameter following precedence ladder."""
if value is None or value is False:
return None
if value is True:
return ToolSearchConfig()
if isinstance(value, str):
return ToolSearchConfig(enabled=value)
if isinstance(value, dict):
return ToolSearchConfig(**value)
if isinstance(value, ToolSearchConfig):
return value
return value
__all__ = [
# Enums
"MemoryBackend",
@@ -1405,7 +1367,6 @@ __all__ = [
"HooksConfig",
"SkillsConfig",
"AutonomyConfig",
"ToolSearchConfig",
# Config classes (Multi-Agent)
"MultiAgentHooksConfig",
"MultiAgentOutputConfig",
@@ -1426,7 +1387,6 @@ __all__ = [
"HooksParam",
"SkillsParam",
"AutonomyParam",
"ToolSearchParam",
# Precedence ladder resolvers
"resolve_memory",
"resolve_knowledge",
@@ -1438,5 +1398,4 @@ __all__ = [
"resolve_execution",
"resolve_caching",
"resolve_autonomy",
"resolve_tool_search",
]
@@ -7,7 +7,7 @@ Inspired by Agno's SessionContextStore pattern.
from dataclasses import dataclass, field, asdict
from typing import Any, Dict, List, Optional
from datetime import datetime, timezone
from datetime import datetime
import json
import logging
from praisonaiagents._logging import get_logger
@@ -97,8 +97,8 @@ class SessionContextTracker:
self._state = SessionState(
session_id=self.session_id,
created_at=datetime.now(timezone.utc).isoformat(),
updated_at=datetime.now(timezone.utc).isoformat(),
created_at=datetime.utcnow().isoformat() + "Z",
updated_at=datetime.utcnow().isoformat() + "Z",
)
def _generate_session_id(self) -> str:
@@ -163,7 +163,7 @@ class SessionContextTracker:
def _mark_updated(self) -> None:
"""Update the updated_at timestamp."""
self._state.updated_at = datetime.now(timezone.utc).isoformat()
self._state.updated_at = datetime.utcnow().isoformat() + "Z"
self._state.turn_count += 1
def to_context_string(self) -> str:
@@ -257,7 +257,7 @@ Current messages take precedence if there's any conflict with this summary.
self._state = SessionState(
session_id=self.session_id,
created_at=self._state.created_at,
updated_at=datetime.now(timezone.utc).isoformat(),
updated_at=datetime.utcnow().isoformat() + "Z",
)
def to_dict(self) -> Dict[str, Any]:
@@ -12,7 +12,7 @@ import os
import logging
from praisonaiagents._logging import get_logger
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from datetime import datetime
from typing import Any, Dict, List, Optional
logger = get_logger(__name__)
@@ -39,7 +39,7 @@ class GradeResult:
input_text: str = ""
output: str = ""
expected_output: Optional[str] = None
timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
@@ -55,7 +55,7 @@ class GradeResult:
input_text=data.get("input_text", ""),
output=data.get("output", ""),
expected_output=data.get("expected_output"),
timestamp=data.get("timestamp", datetime.now(timezone.utc).isoformat()),
timestamp=data.get("timestamp", datetime.utcnow().isoformat()),
)
class BaseLLMGrader:
@@ -1,13 +0,0 @@
"""Hooks query protocol for UI/API consumers."""
from __future__ import annotations
from typing import Dict, List, Protocol, runtime_checkable
@runtime_checkable
class HooksQueryProtocol(Protocol):
"""Read path for registered hooks."""
def list_hooks_for_api(self) -> List[Dict]:
...
@@ -439,13 +439,3 @@ def has_hook(event: Union[str, HookEvent]) -> bool:
)
return get_default_registry().has_hooks(event)
def list_hooks_for_api() -> List[Dict]:
"""Flat hook list suitable for REST/UI consumers."""
registry = get_default_registry()
flat: List[Dict] = []
for event, hooks in registry.list_hooks().items():
for h in hooks:
flat.append({"event": event, "source": "sdk", **h})
return flat
Loaded 100 of 436 files, more files were not shown because too many files have changed in this diff. Show more