Compare commits

...
Author SHA1 Message Date
praisonai-triage-agent[bot]andMervin Praison 3e9826bda7 fix: resolve missing AgentTeam import and RouterAgent kwargs forwarding
- Fix missing AgentTeam import in self-reflection-optimization-patterns.py
- Forward kwargs properly in RouterAgent._build_system_prompt to super()
- Addresses critical issues identified by Gemini code reviewer

Co-authored-by: Mervin Praison <[email protected]>
2026-06-12 23:25:32 +00:00
praisonai-triage-agent[bot]andMervin Praison fc298049e2 fix: resolve critical RouterAgent **kwargs bug and add missing if __name__ guards
- Fixed TypeError in RouterAgent._build_system_prompt() by removing **kwargs from super() call
- Added if __name__ == '__main__' guards to 6 example files to prevent auto-execution on import
- Addresses Gemini Code Assist review feedback

Co-authored-by: Mervin Praison <[email protected]>
2026-06-12 23:25:32 +00:00
Dhivya-Bharathy 0b7bf10820 fix(examples): phase-4 API alignment for 40 validation failures on v4.6.52 2026-06-12 23:25:32 +00:00
Dhivya-Bharathy cbb7c726aa fix(examples): phase-3 API alignment for consolidated params on v4.6.52 2026-06-12 23:25:12 +00:00
Dhivya-Bharathy 21d1986d0c fix(examples): align AgentTeam memory and knowledge configs with v4.6.52 API 2026-06-12 23:25:12 +00:00
praisonai-triage-agent[bot]andMervin Praison 26b5792192 fix: correct TypeError in 3 example files by using proper parameter names
- Fix MemoryConfig(memory=...) → user_id=... in agent_centric_api.py
- Fix Session(memory=...) → user_id=... in session-example.py
- Fix build_context_for_task(memory=...) → user_id=... in memory-quality-example.py

Co-authored-by: Mervin Praison <[email protected]>
2026-06-12 23:25:12 +00:00
Dhivya-Bharathy afcea5800e fix(examples): resolve syntax errors and align with consolidated-params API 2026-06-12 23:25:12 +00:00
98cfdcc9ad fix: resolve three architectural gaps in wrapper layer (#1867)
* fix: resolve three architectural gaps in wrapper layer (fixes #1866)

Gap 1: resolve_llm_endpoint() now maps model prefixes to correct provider API keys
- Added provider mapping for anthropic/, groq/, google/, openrouter/, etc.
- Stops hardcoding OPENAI_API_KEY for all providers
- Maintains fallback to OPENAI_API_KEY for default OpenAI models

Gap 2: SubprocessSandbox now enforces SecurityPolicy and ResourceLimits
- Builds child environment from policy instead of copying host environment
- Applies resource limits via POSIX setrlimit where supported
- Enforces max_output_size to prevent OOM attacks
- Uses process groups for proper timeout cleanup
- Provides warnings when isolation is limited (Windows, missing modules)

Gap 3: Remove root logger mutation from inc/models.py
- Removed logging.basicConfig() call that violated project logging policy
- CLI logging configuration remains through _logging.py as intended
- Embedders no longer have their root logger hijacked at import time

Co-authored-by: Mervin Praison <[email protected]>

* fix: resolve critical bugs in architectural gaps implementation

- Fix inverted boolean condition in llm/env.py causing wrong API key fallback
- Add proper provider base URLs to prevent misrouting to OpenAI endpoints
- Add Windows compatibility for subprocess sandbox (preexec_fn, os.killpg)
- Remove unused variables and improve comments
- Add defensive documentation for hardcoded /tmp fallback

Co-authored-by: Mervin Praison <[email protected]>

* test: add comprehensive unit tests for security fixes

- Add provider mapping tests in test_env_resolver.py
- Add sandbox isolation tests in test_subprocess_security.py
- Add logging regression tests in test_logging_regression.py

Tests validate:
- Provider-specific API key routing (no cross-contamination)
- Environment isolation with SecurityPolicy
- Resource limits enforcement via setrlimit
- Cross-platform compatibility (Windows/POSIX)
- Logging.basicConfig() removal from inc/models.py

All critical security gaps now have test coverage.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: Mervin Praison <[email protected]>

* fix: replace hardcoded secrets in tests with clearly fake placeholders

- Replace 'secret123', 'dbpass456', 'token789' with descriptive fake values
- Preserves test functionality while passing GitGuardian security checks
- Test logic checks for env var keys, not values, so functionality unchanged

Co-authored-by: Mervin Praison <[email protected]>

* fix: replace hardcoded secrets in tests with clearly fake placeholders

Replace test placeholders that GitGuardian was flagging:
- 'fake-aws-key-for-testing' → 'not-a-real-aws-secret'
- 'fake-db-password-for-testing' → 'not-a-real-db-password'
- 'fake-api-token-for-testing' → 'not-a-real-api-token'

Test functionality is preserved since the regression test checks for
environment variable keys, not their values.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

* fix: replace hardcoded secrets in tests with clearly fake placeholders

- Updated test values in test_subprocess_security.py to use obviously fake placeholders
- Changed 'not-a-real-*' format to 'FAKE_*_FOR_TESTING_ONLY' format
- Fixes GitGuardian security scanner detection of Generic Password patterns
- Test functionality preserved (tests environment variable keys, not values)

Co-authored-by: Mervin Praison <[email protected]>

---------

Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: Mervin Praison <[email protected]>
Co-authored-by: Claude <[email protected]>
2026-06-13 00:23:05 +01:00
AkhiChalasaniandAkhil Chalasani 6f2df7db2a docs: improve documentation (#1640)
Created CONTRIBUTING.md

Co-authored-by: Akhil Chalasani <[email protected]>
2026-06-12 23:44:01 +01:00
praisonai-triage-agent[bot]andMervin Praison 495d4604fe feat: remove deprecated tool_timeout and parallel_tool_calls parameters (#1902)
* feat: remove deprecated tool_timeout and parallel_tool_calls parameters

- Remove tool_timeout and parallel_tool_calls parameters from Agent constructor
- These are now fully consolidated into tool_config=ToolConfig() pattern
- Update tests to use new ToolConfig pattern
- Simplify _resolve_tool_config method
- Remove deprecation warning logic as parameters are fully removed

This completes the parameter consolidation effort following AGENTS.md §5.3

Co-authored-by: Mervin Praison <[email protected]>

* fix: address reviewer feedback - restore ExecutionConfig.parallel_tool_calls support and update docs

- Fix critical regression where ExecutionConfig.parallel_tool_calls was ignored
- Update ToolConfig documentation examples to use tool_config parameter
- Fix test expectations for Agent.__init__ parameters (tool_timeout -> tool_config)
- Consolidate ToolConfig imports to module level in test files

Co-authored-by: Mervin Praison <[email protected]>

---------

Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: Mervin Praison <[email protected]>
2026-06-12 23:43:31 +01:00
MervinPraison a4386796c9 Release v4.6.55 2026-06-12 23:39:35 +01:00
e1f9ae5b6b fix: prevent duplicate session messages from persist+auto_save and save_state (#1897)
* fix: prevent duplicate session messages from persist+auto_save and save_state

When memory=history and auto_save share a session_id, per-turn _persist_message
wrote messages that _auto_save_session then re-appended because
_auto_save_last_index was not updated.

Session.save_state() also re-appended the full chat history on every call
instead of replacing it atomically.

- Sync _auto_save_last_index in _persist_message when sessions match
- Use set_chat_history() in Session._save_agent_chat_histories for idempotent saves
- Add regression tests for both duplicate paths

Co-authored-by: Mervin Praison <[email protected]>

* fix: address thread safety and fallback logging issues

- Wrap _auto_save_last_index increment in _history_lock for thread safety
- Add warning when session store lacks set_chat_history (duplicate risk)
- Add debug logging for empty message lists

Co-authored-by: Mervin Praison <[email protected]>

---------

Co-authored-by: Cursor Agent <[email protected]>
Co-authored-by: Mervin Praison <[email protected]>
Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
2026-06-12 23:21:15 +01:00
aa5315b90e fix: implement retry policy for tool execution with exponential backoff (fixes #1840) (#1852)
* fix: implement retry policy for tool execution with exponential backoff (fixes #1840)

- Add RetryPolicy integration to tool execution pipeline
- Implement retry logic in _execute_tool_with_circuit_breaker() and execute_tool_async()
- Add tool-level and agent-level retry policy support to Agent constructor
- Extend OnRetryInput hook event with tool-specific fields (tool_name, attempt, delay_ms, error_type)
- Add helper methods for retry policy resolution and error classification
- Support exponential backoff with jitter for transient failures (timeout, rate_limit, connection_error)
- Emit ON_RETRY hook events during retry attempts for observability
- Maintain backward compatibility with existing OnRetryInput fields

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

* fix: add missing circuit_open guard and optimize retry policy caching

- Add missing circuit_open guard to async retry path (addresses Greptile P1 issue)
- Move inline import os to module level for consistency
- Cache default RetryPolicy instance to avoid repeated allocation

Co-authored-by: Mervin Praison <[email protected]>

* feat: comprehensive tool retry policy improvements and testing

- Fix RetryPolicy import in TYPE_CHECKING block
- Fix clone_for_channel to propagate tool_retry_policy
- Fix async hook emission to use executor instead of blocking event loop
- Add exception chaining (raise ... from e) for better error tracebacks
- Fix iterable guard in _get_tool_retry_policy for MCP tools
- Wire @tool(retry_policy=...) support on FunctionTool decorator
- Add comprehensive unit tests for sync/async retry functionality
- Add real agentic tests with flaky tools (satisfies AGENTS.md §9.4)
- Add CLI support with --tool-retry-* flags
- Add YAML configuration support for tool_retry_policy
- Test tool-level policy precedence and error classification

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: Claude <[email protected]>

* feat: implement ToolConfig consolidation pattern to reduce Agent class bloat

- Add ToolConfig class following AGENTS.md §5.3 pattern (False=disabled, True=defaults, Config=custom)
- Consolidate tool_timeout, tool_retry_policy, parallel_tool_calls into single tool_config parameter
- Maintain full backward compatibility with deprecation warnings for legacy parameters
- Add ToolParam type alias and resolve_tools() function for type safety
- Update Agent constructor and clone_for_channel method to support consolidated config
- Tested with both new consolidated and legacy deprecated approaches

Follows AGENTS.md §5.3 parameter consolidation guidelines. Reduces Agent constructor bloat
while maintaining backward compatibility and following consistent SDK patterns.

Co-authored-by: Mervin Praison <[email protected]>

---------

Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: Claude <[email protected]>
Co-authored-by: Mervin Praison <[email protected]>
2026-06-12 23:19:23 +01:00
b9d3ede110 fix: defer framework adapter creation until YAML is loaded (fixes #1877) (#1881)
* fix: defer framework adapter creation until YAML is loaded

- Fixes issue where AgentsGenerator crashes with empty framework before YAML is read
- Defers framework_adapter creation from __init__ to generate_crew_and_kickoff()
- Changes default YAML framework from 'crewai' to 'praisonai' for better onboarding
- Enables 'praisonai agents.yaml' to work without --framework flag

Fixes #1877

Co-authored-by: praisonai-triage-agent[bot] <praisonai-triage-agent[bot]@users.noreply.github.com>

* test: add mandatory tests for framework default change and async path parity

- Add test verifying YAML files without framework key default to praisonai (fixes #1877)
- Add test ensuring async path has same adapter.resolve() + init_observability() pattern as sync
- Add FRAMEWORK_DEFAULTS_CHANGE.md documenting breaking change migration

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: Mervin Praison <[email protected]>

---------

Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: praisonai-triage-agent[bot] <praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: Mervin Praison <[email protected]>
2026-06-12 23:12:36 +01:00
111 changed files with 2800 additions and 609 deletions

No files matched your search

+14
View File
@@ -0,0 +1,14 @@
# Contributing to PraisonAI
Thank you for your interest! Here's how to get started:
1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Submit a pull request
## Guidelines
- Follow existing code style
- Add tests for new features
- Update documentation as needed
+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.54" \
"praisonai>=4.6.55" \
"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.54" \
"praisonai>=4.6.55" \
"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.54" \
"praisonai>=4.6.55" \
"praisonai[ui]" \
"praisonai[crewai]"
+1 -1
View File
@@ -52,7 +52,7 @@ def example_with_config():
instructions="You are a research assistant",
memory=MemoryConfig(
backend="file",
memory={"user_id": "researcher_001"},
user_id="researcher_001",
auto_memory=True,
),
reflection=ReflectionConfig(
+6 -1
View File
@@ -12,10 +12,15 @@ from praisonaiagents import Agent
from praisonaiagents.approval import AgentApproval
from praisonaiagents.tools.shell_tools import execute_command
reviewer = AgentApproval(
reviewer_agent = Agent(
name="CommandReviewer",
instructions="Only approve read-only commands like 'ls' or 'cat'. Deny destructive commands.",
)
reviewer = AgentApproval(
approver_agent=reviewer_agent,
)
agent = Agent(
name="DevOps",
instructions="You are a DevOps assistant. Use shell tools when asked.",
@@ -2,7 +2,7 @@
from praisonaiagents import Agent
from praisonaiagents import AgentFlow, Task
from praisonaiagents.workflows.workflow_configs import (
WorkflowOutputConfig, WorkflowPlanningConfig, WorkflowMemoryConfig,
WorkflowPlanningConfig, WorkflowMemoryConfig,
)
# Full-featured workflow with all consolidated params
@@ -32,7 +32,7 @@ workflow = AgentFlow(
),
],
# Workflow-level consolidated params
output=WorkflowOutputConfig(output="verbose", stream=True),
output="verbose",
planning=WorkflowPlanningConfig(enabled=True, reasoning=True),
memory=WorkflowMemoryConfig(backend="file"),
# Agent-like params
+1 -1
View File
@@ -10,7 +10,7 @@ from praisonaiagents import Agent
# Basic: Enable web search with preset
agent = Agent(
instructions="You are a research assistant with web access.",
web="duckduckgo", # Presets: duckduckgo, tavily, google, bing, serper
web="search_only", # Compatible across models without web_fetch
)
if __name__ == "__main__":
@@ -80,7 +80,7 @@ Be precise and quote exact values when asked about codes or numbers.""",
print(f"\nAnswer: {response}\n")
# Verify the answer contains the unique code
if "ZEBRA-71" in response.upper():
if "ZEBRA-71" in str(response or "").upper():
print("✅ VERIFIED: Agent correctly retrieved the unique code from context!\n")
else:
print("❌ WARNING: Agent did not find the code - check retrieval\n")
@@ -104,7 +104,7 @@ Be precise and quote exact values when asked about codes or numbers.""",
response = agent.chat("What is the security incident code?")
print(f"\nAnswer: {response}\n")
if "HAWK-88" in response.upper():
if "HAWK-88" in str(response or "").upper():
print("✅ VERIFIED: Agent correctly retrieved the security code!\n")
else:
print("❌ WARNING: Agent did not find the code - check retrieval\n")
@@ -17,8 +17,10 @@ from praisonaiagents import Agent
agent = Agent(
name="CockroachDB Agent",
instructions="You are a helpful assistant.",
db={"database_url": os.getenv("COCKROACHDB_URL")},
session_id="crdb-demo-session",
memory={
"session_id": "crdb-demo-session",
"db": {"database_url": os.getenv("COCKROACHDB_URL")},
},
)
result = agent.start("What is distributed SQL? Explain in 2 sentences.")
@@ -18,8 +18,10 @@ from praisonaiagents import Agent
agent = Agent(
name="Neon Agent",
instructions="You are a helpful assistant.",
db={"database_url": os.getenv("NEON_DATABASE_URL")},
session_id="neon-demo-session",
memory={
"session_id": "neon-demo-session",
"db": {"database_url": os.getenv("NEON_DATABASE_URL")},
},
)
# Option 2: Direct URL
@@ -20,8 +20,10 @@ from praisonaiagents import Agent
agent = Agent(
name="Supabase Agent",
instructions="You are a helpful assistant.",
db={"database_url": os.getenv("SUPABASE_DATABASE_URL")},
session_id="supabase-demo-session",
memory={
"session_id": "supabase-demo-session",
"db": {"database_url": os.getenv("SUPABASE_DATABASE_URL")},
},
)
result = agent.start("What is Supabase? Explain in 2 sentences.")
@@ -18,11 +18,13 @@ from praisonaiagents import Agent
agent = Agent(
name="Turso Agent",
instructions="You are a helpful assistant.",
db={
"database_url": os.getenv("TURSO_DATABASE_URL"),
"auth_token": os.getenv("TURSO_AUTH_TOKEN"),
memory={
"session_id": "turso-demo-session",
"db": {
"database_url": os.getenv("TURSO_DATABASE_URL"),
"auth_token": os.getenv("TURSO_AUTH_TOKEN"),
},
},
session_id="turso-demo-session",
)
result = agent.start("What is edge computing? Explain in 2 sentences.")
+1 -2
View File
@@ -10,8 +10,7 @@ from praisonaiagents import Agent
agent = Agent(
name="FlexBot",
instructions="You are a helpful assistant.",
llm="gpt-4o-mini",
llm_config={"configurable": True}
llm={"model": "gpt-4o-mini", "configurable": True},
)
if __name__ == "__main__":
+1 -1
View File
@@ -21,7 +21,7 @@ agent = Agent(
name="ContextBot",
instructions="You help show context information.",
tools=[show_context],
session_id="my-session-123"
memory={"session_id": "my-session-123"},
)
if __name__ == "__main__":
+13 -3
View File
@@ -19,6 +19,16 @@ from praisonaiagents import Agent, AgentTeam
# Initialize observability
obs.init()
def set_span_attr(span, key, value):
"""Set span attribute safely across observability implementations."""
if hasattr(span, "set_attribute"):
span.set_attribute(key, value)
return
attrs = getattr(span, "attributes", None)
if attrs is not None:
attrs[key] = value
# Create agents
researcher = Agent(
name="Researcher",
@@ -37,13 +47,13 @@ with obs.trace("content-creation", metadata={"workflow": "research-write"}):
# Research phase
with obs.span("research-phase", kind=SpanKind.AGENT) as research_span:
research_span.attributes["agent"] = "Researcher"
set_span_attr(research_span, "agent", "Researcher")
research_result = researcher.chat("Research the benefits of AI in healthcare")
# Writing phase
with obs.span("writing-phase", kind=SpanKind.AGENT) as write_span:
write_span.attributes["agent"] = "Writer"
write_span.attributes["input_length"] = len(str(research_result))
set_span_attr(write_span, "agent", "Writer")
set_span_attr(write_span, "input_length", len(str(research_result)))
final_content = writer.chat(f"Write a blog post based on: {research_result}")
print("Final Content:")
+4 -2
View File
@@ -52,8 +52,10 @@ agent = Agent(
instructions="You are a helpful assistant with access to technical documentation.",
knowledge={
"sources": [doc_path],
"vector_store": "qdrant",
"url": "http://localhost:6333",
"vector_store": {
"provider": "qdrant",
"url": "http://localhost:6333",
},
}
)
+8 -5
View File
@@ -13,7 +13,8 @@ Expected output:
- Session can be resumed later with same session_id
"""
from praisonaiagents import Agent, db
from praisonaiagents import Agent
from praisonaiagents.db import db
# Create database connection (PostgreSQL + Redis)
my_db = db(
@@ -25,10 +26,12 @@ my_db = db(
agent = Agent(
name="Assistant",
instructions="You are a helpful assistant. Be concise.",
db=my_db,
# session_id is optional - defaults to per-hour ID (YYYYMMDDHH-hash)
# Set explicitly for session continuity across runs:
session_id="my-persistent-session"
memory={
"db": my_db,
# session_id is optional - defaults to per-hour ID (YYYYMMDDHH-hash)
# Set explicitly for session continuity across runs:
"session_id": "my-persistent-session",
},
)
# Chat - messages are automatically persisted
+6 -3
View File
@@ -12,7 +12,8 @@ Expected output:
- Export shows full session data
"""
from praisonaiagents import Agent, db
from praisonaiagents import Agent
from praisonaiagents.db import db
def get_weather(city: str) -> str:
"""Get weather for a city."""
@@ -29,8 +30,10 @@ agent = Agent(
name="WeatherBot",
instructions="You help with weather. Use the get_weather tool when asked.",
tools=[get_weather],
db=my_db,
session_id="weather-session-001"
memory={
"db": my_db,
"session_id": "weather-session-001",
},
)
# First chat - triggers run tracking
+6 -3
View File
@@ -13,7 +13,8 @@ Expected output:
Agent responds with state persisted to Redis
"""
from praisonaiagents import Agent, db
from praisonaiagents import Agent
from praisonaiagents.db import db
print("=== Redis State Store (Agent-First) ===")
@@ -26,8 +27,10 @@ my_db = db(
agent = Agent(
name="Assistant",
instructions="You are a helpful assistant.",
db=my_db,
session_id="redis-state-session"
memory={
"db": my_db,
"session_id": "redis-state-session",
},
)
# Chat - state is automatically managed via Redis
+6 -3
View File
@@ -11,7 +11,8 @@ Expected output:
- Second run: Resumes with history
"""
from praisonaiagents import Agent, db
from praisonaiagents import Agent
from praisonaiagents.db import db
# Create database
my_db = db(
@@ -25,8 +26,10 @@ SESSION_ID = "resume-demo-session"
agent = Agent(
name="MemoryBot",
instructions="You remember our conversation. Be helpful and concise.",
db=my_db,
session_id=SESSION_ID
memory={
"db": my_db,
"session_id": SESSION_ID,
},
)
# Check if this is a new or resumed session
+6 -3
View File
@@ -26,7 +26,8 @@ Expected Output:
Messages persisted: 2
"""
from praisonaiagents import Agent, db
from praisonaiagents import Agent
from praisonaiagents.db import db
# Create database adapter (simplified import)
db_instance = db(
@@ -37,8 +38,10 @@ db_instance = db(
agent = Agent(
name="Assistant",
instructions="You are a helpful assistant. Keep responses brief.",
db=db_instance,
session_id="demo-session-001", # Same session_id = resume conversation
memory={
"db": db_instance,
"session_id": "demo-session-001", # Same session_id = resume conversation
},
output="silent"
)
+6 -3
View File
@@ -11,7 +11,8 @@ Expected output:
- Data persisted to local SQLite file
"""
from praisonaiagents import Agent, db
from praisonaiagents import Agent
from praisonaiagents.db import db
import tempfile
import os
@@ -25,8 +26,10 @@ print(f"Database: {db_path}")
agent = Agent(
name="LocalBot",
instructions="You are a helpful assistant.",
db=my_db,
session_id="local-session"
memory={
"db": my_db,
"session_id": "local-session",
},
)
# Chat
+1 -1
View File
@@ -57,8 +57,8 @@ def main():
agent = Agent(
name="SecureAgent",
instructions="You are a file management assistant.",
policy=engine
)
agent.policy = engine
print("\n--- Agent with Policy Enforcement Created ---")
print(f"Agent: {agent.name}")
@@ -60,9 +60,10 @@ agents_system = AgentTeam(
process="sequential", output="verbose"
)
print("Starting hierarchical agent generation...")
result = agents_system.start()
if __name__ == "__main__":
print("Starting hierarchical agent generation...")
result = agents_system.start()
print(f"\nHierarchical Result: {result[:200]}...")
print("\n✅ AutoAgents hierarchical generation complete!")
print("Demonstrated coordinated hierarchical workflow between manager and specialist agents.")
print(f"\nHierarchical Result: {str(result)[:200]}...")
print("\n✅ AutoAgents hierarchical generation complete!")
print("Demonstrated coordinated hierarchical workflow between manager and specialist agents.")
+3 -2
View File
@@ -1,4 +1,5 @@
from praisonaiagents import Agent
agent = Agent(name="TweetAgent", instructions="Create a Tweet based on the topic provided")
agent.launch(port=8080, protocol="mcp")
if __name__ == "__main__":
agent = Agent(name="TweetAgent", instructions="Create a Tweet based on the topic provided")
agent.launch(port=8080, protocol="mcp", path="/")
+11 -2
View File
@@ -90,9 +90,18 @@ def main():
if result:
print("\n🔍 Analysis Results:")
for task_id, task_result in result["task_results"].items():
task_results = result.get("task_results") if isinstance(result, dict) else result
if isinstance(task_results, dict):
iterable = task_results.items()
elif isinstance(task_results, list):
iterable = enumerate(task_results)
else:
iterable = [("result", task_results)]
for task_id, task_result in iterable:
print(f"\nTask {task_id}:")
print(task_result.raw)
raw_value = getattr(task_result, "raw", None)
print(raw_value if raw_value is not None else str(task_result))
else:
print("❌ Failed to capture and analyze camera feed")
+1 -1
View File
@@ -75,7 +75,7 @@ class ContinuousCameraMonitor:
agents=[self.vision_agent],
tasks=[task],
process="sequential",
verbose=0 # Reduced verbosity for continuous monitoring
output="silent", # Reduced verbosity for continuous monitoring
)
return agents.start()
+11 -2
View File
@@ -148,10 +148,19 @@ def main():
"scene_analysis": "🌍 SCENE ANALYSIS"
}
for task_id, task_result in result["task_results"].items():
task_results = result.get("task_results") if isinstance(result, dict) else result
if isinstance(task_results, dict):
iterable = task_results.items()
elif isinstance(task_results, list):
iterable = enumerate(task_results)
else:
iterable = [("result", task_results)]
for task_id, task_result in iterable:
print(f"\n{agent_names.get(task_id, f'Task {task_id}')}:")
print("-" * 40)
print(task_result.raw)
raw_value = getattr(task_result, "raw", None)
print(raw_value if raw_value is not None else str(task_result))
print("\n" + "="*60)
print("✅ Multi-agent analysis complete!")
@@ -34,5 +34,5 @@ print(f"\nChanged to: {handler.get_mode()}")
print("File edits now auto-approved, shell commands still require approval")
# Get statistics
stats = handler.get_stats()
stats = manager.get_stats()
print(f"\nStats: {stats}")
+1 -1
View File
@@ -31,4 +31,4 @@ print(f"Total cost: ${summary['total_cost']:.4f}")
print(f"Avg cost/request: ${summary['avg_cost_per_request']:.4f}")
# Display formatted (uses Rich if available)
handler.display_summary()
handler.print_summary()
@@ -41,9 +41,9 @@ def my_custom_handler(args, context):
custom_cmd = SlashCommand(
name="mycommand",
description="My custom command",
handler=my_custom_handler,
kind=CommandKind.ACTION,
aliases=["mc"]
action=my_custom_handler,
kind=CommandKind.CUSTOM,
alt_names=["mc"]
)
handler.register(custom_cmd)
@@ -16,7 +16,8 @@ agents = AgentTeam(
agents=[agent],
tasks=[task],
process="workflow",
max_iter=30
execution={"max_iter": 30},
)
agents.start()
if __name__ == "__main__":
agents.start()
@@ -5,7 +5,8 @@ This example demonstrates self-reflection capabilities using PraisonAI's
built-in reflection features for iterative improvement and quality optimization.
"""
from praisonaiagents import Agent, Task, AgentTeam
from praisonaiagents import Agent, Task
from praisonaiagents.agents import AgentTeam
from praisonaiagents.tools import internet_search
print("=== Self-Reflection Optimization Patterns Example ===\n")
@@ -33,9 +34,16 @@ reflection_task = Task(
)
# Run with self-reflection optimization
print("Starting self-reflection optimization demonstration...")
result = reflection_agent.execute_task(reflection_task)
if __name__ == "__main__":
print("Starting self-reflection optimization demonstration...")
agents = AgentTeam(
agents=[reflection_agent],
tasks=[reflection_task],
process="sequential",
output="verbose",
)
result = agents.start()
print(f"\nSelf-Reflection Result: {result[:200]}...")
print("\n✅ Self-reflection optimization complete!")
print("Agent demonstrated iterative improvement through built-in self-reflection capabilities.")
print(f"\nSelf-Reflection Result: {str(result)[:200]}...")
print("\n✅ Self-reflection optimization complete!")
print("Agent demonstrated iterative improvement through built-in self-reflection capabilities.")
@@ -61,7 +61,6 @@ def example_with_wrappers():
role="Data Analyst",
goal="Analyze data and provide insights",
llm="gpt-4o-mini",
request_id="request_2" # Unique request ID
)
# Operations are automatically tracked
+4 -2
View File
@@ -132,7 +132,9 @@ agents = AgentTeam(
agents=[qa_generator, total_questions_evaluator, cot_generator, upload_to_huggingface],
tasks=[generate_task, evaluate_total_questions_task, generate_cot_task, upload_to_huggingface_task],
process="workflow",
max_iter=30, output="minimal"
execution={"max_iter": 30},
output="minimal",
)
agents.start()
if __name__ == "__main__":
agents.start()
+3 -3
View File
@@ -18,7 +18,7 @@ openai_profile = AuthProfile(
provider="openai",
api_key=os.environ.get("OPENAI_API_KEY", "sk-..."),
priority=1, # Highest priority
rate_limit=100,
rate_limit_rpm=100,
)
anthropic_profile = AuthProfile(
@@ -26,7 +26,7 @@ anthropic_profile = AuthProfile(
provider="anthropic",
api_key=os.environ.get("ANTHROPIC_API_KEY", "sk-ant-..."),
priority=2, # Second priority
rate_limit=50,
rate_limit_rpm=50,
)
groq_profile = AuthProfile(
@@ -34,7 +34,7 @@ groq_profile = AuthProfile(
provider="groq",
api_key=os.environ.get("GROQ_API_KEY", "gsk-..."),
priority=3, # Third priority (fast but limited)
rate_limit=30,
rate_limit_rpm=30,
)
# Configure failover behavior
+2 -2
View File
@@ -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={
"backend": "sqlite",
"embedder": {
"provider": "openai",
"config": {
"model": "text-embedding-3-small"
}
}
},
"config": {"provider": "sqlite"},
}
)
+16 -6
View File
@@ -49,10 +49,20 @@ agents = AgentTeam(
process="sequential"
)
# Run all tasks
result = agents.start()
if __name__ == "__main__":
# Run all tasks
result = agents.start()
# Print results
for task_id, task_result in result["task_results"].items():
print(f"\nTask {task_id} Result:")
print(task_result.raw)
# Print results
task_results = result.get("task_results") if isinstance(result, dict) else result
if isinstance(task_results, dict):
iterable = task_results.items()
elif isinstance(task_results, list):
iterable = enumerate(task_results)
else:
iterable = [("result", task_results)]
for task_id, task_result in iterable:
print(f"\nTask {task_id} Result:")
raw_value = getattr(task_result, "raw", None)
print(raw_value if raw_value is not None else str(task_result))
@@ -6,6 +6,7 @@ built-in validation and safety mechanisms for secure agent operations.
"""
from praisonaiagents import Agent, Task, AgentTeam
from praisonaiagents.config.feature_configs import GuardrailConfig
from praisonaiagents.tools import internet_search
print("=== Production Guardrails Patterns Example ===\n")
@@ -38,8 +39,8 @@ production_agent = Agent(
goal="Generate safe, compliant content for production use",
backstory="Production-ready agent with built-in safety and compliance checks",
tools=[internet_search],
guardrails=content_guardrail,
max_retries=3)
guardrails=GuardrailConfig(validator=content_guardrail, max_retries=3),
)
# Create secure task with validation
secure_task = Task(
+1 -1
View File
@@ -40,7 +40,7 @@ agent = Agent(
"execute_command"
],
memory=True,
web_search=True,
web=True,
auto_approve_tools=True,
)
@@ -14,30 +14,33 @@ agent = Agent(name="env-mgmt", backend=managed)
# First call creates agent + environment + session
agent.start("Say hello briefly")
# Access the underlying Anthropic client for environment management
client = managed._get_client()
env_id = managed.environment_id
print(f"Environment ID: {env_id}")
if hasattr(managed, "_get_client"):
# Access the underlying Anthropic client for environment management
client = managed._get_client()
env_id = managed.environment_id
print(f"Environment ID: {env_id}")
# List all environments
environments = client.beta.environments.list()
print(f"\nTotal environments: {len(environments.data)}")
for env in environments.data[:5]:
print(f" {env.id} | {env.name} | {getattr(env, 'status', 'active')}")
# List all environments
environments = client.beta.environments.list()
print(f"\nTotal environments: {len(environments.data)}")
for env in environments.data[:5]:
print(f" {env.id} | {env.name} | {getattr(env, 'status', 'active')}")
# Retrieve a specific environment
env = client.beta.environments.retrieve(env_id)
print(f"\nRetrieved: {env.id} | {env.name}")
# Retrieve a specific environment
env = client.beta.environments.retrieve(env_id)
print(f"\nRetrieved: {env.id} | {env.name}")
# Archive (read-only, existing sessions continue)
archived = client.beta.environments.archive(env_id)
print(f"Archived: {archived.id} | status: {getattr(archived, 'status', 'archived')}")
# Archive (read-only, existing sessions continue)
archived = client.beta.environments.archive(env_id)
print(f"Archived: {archived.id} | status: {getattr(archived, 'status', 'archived')}")
# Delete (only if no sessions reference it — may fail if sessions exist)
try:
client.beta.environments.delete(env_id)
print(f"Deleted: {env_id}")
except Exception as e:
print(f"Delete skipped (expected if sessions exist): {e}")
# Delete (only if no sessions reference it — may fail if sessions exist)
try:
client.beta.environments.delete(env_id)
print(f"Deleted: {env_id}")
except Exception as e:
print(f"Delete skipped (expected if sessions exist): {e}")
else:
print("Environment management demo skipped: backend does not expose _get_client().")
print("\nEnvironment management complete.")
+6 -7
View File
@@ -15,13 +15,6 @@ Protocol: MCP 2025-11-25
from praisonaiagents import Agent, MCP
# Agent with resumability support (automatic)
agent = Agent(
name="Resumable Assistant",
instructions="You are a helpful assistant with resumable connections.",
tools=MCP("https://api.example.com/mcp")
)
# Resumability is automatic - the MCP client:
# 1. Tracks event IDs from SSE streams (id: field)
# 2. Stores last event ID per stream
@@ -29,6 +22,12 @@ agent = Agent(
# 4. Respects retry delay from server (retry: field)
if __name__ == "__main__":
agent = Agent(
name="Resumable Assistant",
instructions="You are a helpful assistant with resumable connections.",
tools=MCP("https://api.example.com/mcp"),
)
print("MCP Resumability Example")
print("=" * 40)
print()
@@ -15,13 +15,6 @@ Protocol: MCP 2025-11-25
from praisonaiagents import Agent, MCP
# Basic agent with automatic session management
agent = Agent(
name="Session Assistant",
instructions="You are a helpful assistant with session support.",
tools=MCP("https://api.example.com/mcp")
)
# Session management is automatic - the MCP client:
# 1. Receives Mcp-Session-Id from server during initialization
# 2. Includes session ID in all subsequent requests
@@ -29,6 +22,12 @@ agent = Agent(
# 4. Can terminate session explicitly when done
if __name__ == "__main__":
agent = Agent(
name="Session Assistant",
instructions="You are a helpful assistant with session support.",
tools=MCP("https://api.example.com/mcp"),
)
print("MCP Session Management Example")
print("=" * 40)
print()
+24 -27
View File
@@ -16,34 +16,31 @@ Protocol: MCP 2025-11-25
from praisonaiagents import Agent, MCP
# Basic Streamable HTTP connection
agent = Agent(
name="HTTP Stream Assistant",
instructions="You are a helpful assistant connected via Streamable HTTP.",
tools=MCP("https://api.example.com/mcp")
)
# With authentication headers
agent_auth = Agent(
name="Authenticated Assistant",
instructions="You are a helpful assistant with authenticated connection.",
tools=MCP(
"https://api.example.com/mcp",
headers={"Authorization": "Bearer your-token"}
)
)
# With timeout and debug
agent_debug = Agent(
name="Debug Assistant",
tools=MCP(
"https://api.example.com/mcp",
timeout=120,
debug=True
)
)
if __name__ == "__main__":
agent = Agent(
name="HTTP Stream Assistant",
instructions="You are a helpful assistant connected via Streamable HTTP.",
tools=MCP("https://api.example.com/mcp"),
)
agent_auth = Agent(
name="Authenticated Assistant",
instructions="You are a helpful assistant with authenticated connection.",
tools=MCP(
"https://api.example.com/mcp",
headers={"Authorization": "Bearer your-token"},
),
)
agent_debug = Agent(
name="Debug Assistant",
tools=MCP(
"https://api.example.com/mcp",
timeout=120,
debug=True,
),
)
print("Streamable HTTP MCP Transport Example")
print("=" * 40)
print("This example requires a Streamable HTTP MCP server running.")
+26 -29
View File
@@ -16,36 +16,33 @@ Protocol: SEP-1288 (WebSocket Transport)
from praisonaiagents import Agent, MCP
# Basic WebSocket connection
agent = Agent(
name="WebSocket Assistant",
instructions="You are a helpful assistant connected via WebSocket.",
tools=MCP("ws://localhost:8080/mcp")
)
# Secure WebSocket with authentication
agent_secure = Agent(
name="Secure WebSocket Assistant",
instructions="You are a helpful assistant with secure WebSocket connection.",
tools=MCP(
"wss://api.example.com/mcp",
auth_token="Bearer your-secret-token",
timeout=60
)
)
# WebSocket with custom options
agent_custom = Agent(
name="Custom WebSocket Assistant",
tools=MCP(
"wss://api.example.com/mcp",
auth_token="your-token",
timeout=120,
debug=True
)
)
if __name__ == "__main__":
agent = Agent(
name="WebSocket Assistant",
instructions="You are a helpful assistant connected via WebSocket.",
tools=MCP("ws://localhost:8080/mcp"),
)
agent_secure = Agent(
name="Secure WebSocket Assistant",
instructions="You are a helpful assistant with secure WebSocket connection.",
tools=MCP(
"wss://api.example.com/mcp",
auth_token="Bearer your-secret-token",
timeout=60,
),
)
agent_custom = Agent(
name="Custom WebSocket Assistant",
tools=MCP(
"wss://api.example.com/mcp",
auth_token="your-token",
timeout=120,
debug=True,
),
)
# Note: Requires a running WebSocket MCP server
# Example server: https://github.com/modelcontextprotocol/servers
@@ -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": {
@@ -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": {
@@ -86,13 +86,13 @@ def main():
agents=[researcher, retriever],
tasks=[store_task, retrieve_task],
memory={
"backend": "sqlite",
"embedder": {
"provider": "gemini",
"config": {
"model": "text-embedding-004",
}
}
},
"config": {"provider": "sqlite"},
}
)
@@ -30,7 +30,11 @@ config = {
agent = Agent(
name="Knowledge Agent",
instructions="You answer questions based on the provided knowledge.",
knowledge={**config, "sources": ["kag-research-paper.pdf"]},
knowledge={
"sources": ["kag-research-paper.pdf"],
"vector_store": config["vector_store"],
"embedder": config["embedder"],
},
memory={"user_id": "user1"},
llm="deepseek-r1"
)
@@ -31,10 +31,9 @@ llm_config = {
agent = Agent(
instructions="You are a helpful Assistant specialized in scientific explanations. "
"Provide clear, accurate, and engaging responses.",
llm=llm_config, # Pass the detailed configuration
# Enable detailed output
markdown=True, # Format responses in markdown
reflection=True # Enable self-reflection
llm=llm_config,
output={"markdown": True},
reflection=True,
)
# Test the agent
@@ -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)
@@ -4,7 +4,7 @@ Example 3: Save Agent Output Using Workflow output_file
Workflow step output is automatically saved with variable substitution.
"""
from praisonaiagents import AgentFlow
from praisonaiagents import Agent, AgentFlow, Task
# Define workflow with output_file
workflow_config = {
@@ -33,9 +33,30 @@ workflow_config = {
]
}
# Run workflow
workflow = AgentFlow(config=workflow_config)
result = workflow.run()
# Build agent and workflow from config dict
writer_cfg = workflow_config["agents"]["writer"]
writer = Agent(
name="writer",
role=writer_cfg["role"],
goal=writer_cfg["goal"],
llm=writer_cfg["llm"],
)
step_cfg = workflow_config["steps"][0]
workflow = AgentFlow(
name=workflow_config["metadata"]["name"],
variables=workflow_config["variables"],
steps=[
Task(
name="write_tutorial",
description=step_cfg["action"],
expected_output=step_cfg["expected_output"],
output_file=step_cfg["output_file"],
agent=writer,
)
],
)
result = workflow.start()
print("✅ Workflow completed!")
print("Output saved to: generated/tutorial.md")
@@ -28,7 +28,7 @@ def example_basic_persistence():
agent = Agent(
name="Assistant",
instructions="You are a helpful assistant. Be very brief.",
session_id=session_id,
memory={"session_id": session_id},
output="minimal",
)
@@ -59,7 +59,7 @@ def example_session_restoration():
agent1 = Agent(
name="MemoryBot",
instructions="You are a helpful assistant. Be very brief.",
session_id=session_id,
memory={"session_id": session_id},
output="minimal",
)
@@ -72,7 +72,7 @@ def example_session_restoration():
agent2 = Agent(
name="MemoryBot",
instructions="You are a helpful assistant. Be very brief.",
session_id=session_id,
memory={"session_id": session_id},
output="minimal",
)
@@ -69,9 +69,10 @@ agents_system = AgentTeam(
memory=True, output="verbose"
)
print("Starting state management demonstration...")
result = agents_system.start()
if __name__ == "__main__":
print("Starting state management demonstration...")
result = agents_system.start()
print(f"\nState Management Result: {result[:200]}...")
print("\n✅ State management patterns complete!")
print("Agent demonstrated persistent state tracking across multiple tasks.")
print(f"\nState Management Result: {str(result)[:200]}...")
print("\n✅ State management patterns complete!")
print("Agent demonstrated persistent state tracking across multiple tasks.")
@@ -179,7 +179,7 @@ def main():
context = memory.build_context_for_task(
task_descr="Explain AI alignment challenges",
memory={"user_id": "quality_demo_user"},
user_id="quality_demo_user",
max_items=3
)
+1 -1
View File
@@ -15,7 +15,7 @@ def main():
# Create a session with persistent storage
session = Session(
session_id="demo_chat_001",
memory={"user_id": "demo_user"},
user_id="demo_user",
)
print(f"📝 Created session: {session}")
@@ -1,284 +1,55 @@
"""
Production Telemetry Example
This example demonstrates comprehensive telemetry and monitoring setup for
production AI agent systems, including metrics collection, performance tracking,
and observability.
Features demonstrated:
- OpenTelemetry integration for distributed tracing
- Custom metrics collection and monitoring
- Performance tracking and analytics
- Error tracking and alerting
- Usage analytics and reporting
Demonstrates compatible telemetry setup with `enable_telemetry()` and
`TelemetryCollector` patterns.
"""
from contextlib import nullcontext
from praisonaiagents import Agent, Task, AgentTeam
from praisonaiagents.tools import duckduckgo
from praisonaiagents.telemetry import enable_telemetry
import time
import random
from datetime import datetime
from praisonaiagents.telemetry import enable_telemetry, TelemetryCollector
# Initialize telemetry for production monitoring
print("="*80)
print("PRODUCTION TELEMETRY SETUP")
print("="*80)
# Configure telemetry (in production, you'd use actual telemetry backends)
telemetry = enable_telemetry(
backend="opentelemetry",
config={
"service_name": "praisonai_production_agents",
"service_version": "1.0.0",
"environment": "production",
"export_console": True, # For demo purposes
"export_file": True, # Export to file for analysis
"sample_rate": 1.0, # 100% sampling for demo
"attributes": {
"deployment": "cloud",
"region": "us-east-1",
"cluster": "main"
}
}
)
def main() -> None:
print("=" * 80)
print("PRODUCTION TELEMETRY SETUP")
print("=" * 80)
# Create agents for production workflow with telemetry
customer_service_agent = Agent(
name="CustomerServiceAgent",
role="Customer Service Representative",
goal="Provide excellent customer service with fast response times",
backstory="You are a customer service representative focused on resolving customer issues quickly and effectively.",
tools=[duckduckgo],
instructions="Provide helpful, accurate customer service responses. Research solutions when needed."
)
# New API: no args
enable_telemetry()
collector = TelemetryCollector()
technical_support_agent = Agent(
name="TechnicalSupportAgent",
role="Technical Support Specialist",
goal="Resolve technical issues with detailed solutions",
backstory="You are a technical support specialist who provides detailed technical solutions and troubleshooting.",
instructions="Provide detailed technical solutions with step-by-step instructions."
)
customer_service_agent = Agent(
name="CustomerServiceAgent",
role="Customer Service Representative",
goal="Provide excellent customer service with fast response times",
backstory="Customer service expert focused on quick issue resolution.",
tools=[duckduckgo],
instructions="Provide accurate customer-service responses.",
)
quality_assurance_agent = Agent(
name="QualityAssuranceAgent",
role="Quality Assurance Specialist",
goal="Ensure all customer interactions meet quality standards",
backstory="You review and assess the quality of customer service interactions to ensure high standards.",
instructions="Review customer service interactions and provide quality scores with detailed feedback."
)
# Production workflow simulation with telemetry tracking
def simulate_customer_requests():
"""Simulate various customer requests for telemetry demonstration"""
customer_requests = [
{
"type": "billing_inquiry",
"complexity": "low",
"description": "Customer asking about their monthly bill charges",
"expected_duration": 30
},
{
"type": "technical_issue",
"complexity": "high",
"description": "Customer experiencing connectivity issues with the service",
"expected_duration": 120
},
{
"type": "account_setup",
"complexity": "medium",
"description": "New customer needs help setting up their account",
"expected_duration": 60
},
{
"type": "feature_question",
"complexity": "low",
"description": "Customer asking about product features and capabilities",
"expected_duration": 45
}
requests = [
"Customer asking about monthly bill charges",
"Customer experiencing connectivity issues",
]
return customer_requests
# Execute production workflow with comprehensive telemetry
print("Starting production workflow with telemetry tracking...")
for idx, description in enumerate(requests, start=1):
trace_ctx = getattr(collector, "trace", None)
trace_cm = trace_ctx(f"customer_request_{idx}") if callable(trace_ctx) else nullcontext()
with trace_cm:
task = Task(
name=f"customer_request_{idx}",
description=description,
expected_output="Professional customer service response with solution",
agent=customer_service_agent,
)
agents = AgentTeam(agents=[customer_service_agent], tasks=[task], output="minimal")
result = agents.start()
print(f"✅ Request {idx} completed, response length={len(str(result))}")
customer_requests = simulate_customer_requests()
print("\nTelemetry demonstration completed.")
for i, request in enumerate(customer_requests):
print(f"\n{'='*60}")
print(f"PROCESSING CUSTOMER REQUEST {i+1}: {request['type'].upper()}")
print(f"Complexity: {request['complexity']} | Expected Duration: {request['expected_duration']}s")
print(f"{'='*60}")
# Start telemetry trace for this customer interaction
with telemetry.trace_operation(f"customer_request_{request['type']}") as trace:
# Add custom attributes to the trace
trace.set_attributes({
"request.type": request['type'],
"request.complexity": request['complexity'],
"request.expected_duration": request['expected_duration'],
"customer.id": f"customer_{random.randint(1000, 9999)}",
"agent.assigned": "CustomerServiceAgent"
})
start_time = time.time()
# Create customer service task with telemetry
customer_task = Task(
name=f"customer_request_{i+1}",
description=request['description'],
expected_output="Professional customer service response with solution",
agent=customer_service_agent
)
# Execute with telemetry tracking
try:
# Customer service response
with telemetry.trace_agent_execution("CustomerServiceAgent") as agent_trace:
agents = AgentTeam(
agents=[customer_service_agent],
tasks=[customer_task],
output="minimal" # Reduce verbosity for telemetry demo
)
cs_result = agents.start()
agent_trace.set_attributes({
"agent.response_length": len(str(cs_result)),
"agent.tools_used": len(customer_service_agent.tools) if customer_service_agent.tools else 0
})
# Technical support (if needed for high complexity)
if request['complexity'] == 'high':
with telemetry.trace_agent_execution("TechnicalSupportAgent") as tech_trace:
tech_task = Task(
name=f"technical_support_{i+1}",
description=f"Provide detailed technical support for: {request['description']}",
expected_output="Detailed technical solution with troubleshooting steps",
agent=technical_support_agent,
context=[customer_task]
)
tech_agents = AgentTeam(
agents=[technical_support_agent],
tasks=[tech_task], output="minimal"
)
tech_result = tech_agents.start()
tech_trace.set_attributes({
"agent.response_length": len(str(tech_result)),
"agent.escalation": True
})
# Quality assurance check
with telemetry.trace_agent_execution("QualityAssuranceAgent") as qa_trace:
qa_task = Task(
name=f"quality_check_{i+1}",
description=f"Review the quality of customer service provided for: {request['description']}",
expected_output="Quality assessment score and feedback",
agent=quality_assurance_agent
)
qa_agents = AgentTeam(
agents=[quality_assurance_agent],
tasks=[qa_task], output="minimal"
)
qa_result = qa_agents.start()
# Extract quality score (simulated)
quality_score = random.uniform(0.8, 1.0) # Simulated score
qa_trace.set_attributes({
"qa.score": quality_score,
"qa.passed": quality_score >= 0.85
})
# Calculate performance metrics
end_time = time.time()
duration = end_time - start_time
# Record custom metrics
telemetry.record_metric("request_duration", duration, {
"request_type": request['type'],
"complexity": request['complexity']
})
telemetry.record_metric("request_success", 1, {
"request_type": request['type']
})
# Set final trace attributes
trace.set_attributes({
"request.duration_seconds": duration,
"request.status": "completed",
"request.quality_score": quality_score,
"request.escalated": request['complexity'] == 'high'
})
print(f"✅ Request completed successfully in {duration:.2f}s (Quality: {quality_score:.2f})")
except Exception as e:
# Record error metrics
telemetry.record_metric("request_error", 1, {
"request_type": request['type'],
"error_type": type(e).__name__
})
trace.set_attributes({
"request.status": "error",
"request.error": str(e)
})
print(f"❌ Request failed: {str(e)}")
# Add some delay between requests
time.sleep(2)
# Generate telemetry summary
print(f"\n{'='*80}")
print("TELEMETRY SUMMARY AND ANALYTICS")
print(f"{'='*80}")
# Simulated telemetry analytics (in production, this would come from your telemetry backend)
print("📊 Performance Metrics:")
print("- Total Requests Processed: 4")
print("- Average Response Time: 3.2s")
print("- Success Rate: 100%")
print("- Quality Score Average: 0.91")
print("- High Complexity Requests: 25%")
print("\n📈 Usage Analytics:")
print("- Most Common Request Type: technical_issue")
print("- Peak Performance Window: All requests within SLA")
print("- Agent Utilization: CustomerService (100%), TechnicalSupport (25%), QA (100%)")
print("\n🔍 Observability Features Demonstrated:")
print("- Distributed tracing across agent interactions")
print("- Custom metrics for business KPIs")
print("- Performance monitoring and SLA tracking")
print("- Error tracking and alerting")
print("- Quality assurance metrics")
print("- User journey tracking")
print("- Resource utilization monitoring")
print("\n⚙️ Production Monitoring Setup:")
print("- OpenTelemetry integration for standardized observability")
print("- Custom metrics for business-specific KPIs")
print("- Trace correlation across multi-agent workflows")
print("- Performance benchmarking against expected durations")
print("- Quality scoring and automated validation")
print("- Error categorization and alerting")
print(f"\n{'='*80}")
print("TELEMETRY DEMONSTRATION COMPLETED")
print(f"{'='*80}")
print("This example demonstrated comprehensive production telemetry including:")
print("- Request tracing and performance monitoring")
print("- Custom business metrics collection")
print("- Quality assurance tracking")
print("- Error handling and alerting")
print("- Multi-agent workflow observability")
print("- Production-ready monitoring patterns")
if __name__ == "__main__":
main()
@@ -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
metrics=True # 🎯 SIMPLIFIED: Just add this parameter!
output={"metrics": True} # Token metrics via consolidated output config
)
# 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",
metrics=True # 🎯 Enable metrics tracking
output={"metrics": True}
)
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",
metrics=True # 🎯 Enable metrics tracking
output={"metrics": True}
)
# 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",
metrics=True # 🎯 Enable automatic metrics tracking
output={"metrics": True}
)
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",
metrics=True # 🎯 Enable automatic metrics tracking
output={"metrics": True}
)
# Create tasks that will generate substantial token usage
@@ -22,7 +22,8 @@ rag_agent = Agent(
You can use the internet_search function to search the web when needed.""",
llm="gpt-4o",
markdown=True)
output={"markdown": True},
)
if __name__ == "__main__":
print("🤖 Thai Recipe RAG Agent is ready!")
@@ -5,11 +5,17 @@ from dotenv import load_dotenv
load_dotenv()
os.environ['SERPER_API_KEY'] = os.getenv('SERPER_API_KEY')
serper_api_key = os.getenv("SERPER_API_KEY")
if serper_api_key is not None:
os.environ["SERPER_API_KEY"] = serper_api_key
search = GoogleSerperAPIWrapper()
data_agent = Agent(instructions="Suggest me top 5 most visited websites for Dosa Recipe", tools=[search])
editor_agent = Agent(instructions="List out the websites with their url and a short description")
agents = AgentTeam(agents=[data_agent, editor_agent])
agents.start()
if __name__ == "__main__":
if not serper_api_key:
print("SERPER_API_KEY is not set. Skipping Google Serper example.")
else:
search = GoogleSerperAPIWrapper()
data_agent = Agent(instructions="Suggest me top 5 most visited websites for Dosa Recipe", tools=[search])
editor_agent = Agent(instructions="List out the websites with their url and a short description")
agents = AgentTeam(agents=[data_agent, editor_agent])
agents.start()
@@ -2,6 +2,7 @@ from praisonaiagents import Agent, Task, AgentTeam
from pydantic import BaseModel
from typing import List, Dict
from gitingest import ingest
import sys
class CodeMetrics(BaseModel):
category: str
@@ -60,7 +61,10 @@ def analyze_code(code_source: str) -> CodeAnalysisReport:
Analyze code from directory path or GitHub URL
"""
# Ingest code content
summary, tree, content = ingest(code_source)
try:
summary, tree, content = ingest(code_source)
except Exception as exc:
raise RuntimeError(f"Failed to ingest code source: {exc}") from exc
# Concatenate context into structured format
context_text = f"""
@@ -86,10 +90,22 @@ def analyze_code(code_source: str) -> CodeAnalysisReport:
tasks=[code_analysis_task]
)
return agents.start(context_text)
result = agents.start(context_text)
if result is None:
raise RuntimeError("No analysis result returned by AgentTeam.start()")
return result
if __name__ == "__main__":
def main() -> None:
# Example usage
code_source = "https://github.com/openai/openai-python/tree/main/src/openai/cli/_api/chat" # GitHub URL or local directory
result = analyze_code(code_source)
print(result)
try:
result = analyze_code(code_source)
except RuntimeError as exc:
print(f"Code analysis failed: {exc}")
sys.exit(1)
print(result)
if __name__ == "__main__":
main()
@@ -77,25 +77,44 @@ agents = AgentTeam(
process="sequential", output="verbose"
)
def _extract_pydantic_result(result_obj):
"""Extract pydantic payload from AgentTeam.start() output shapes."""
if hasattr(result_obj, "pydantic"):
return result_obj.pydantic
if isinstance(result_obj, dict):
task_results = result_obj.get("task_results")
if isinstance(task_results, dict):
values = list(task_results.values())
if values and hasattr(values[-1], "pydantic"):
return values[-1].pydantic
if isinstance(task_results, list) and task_results and hasattr(task_results[-1], "pydantic"):
return task_results[-1].pydantic
return None
# Start the analysis
result = agents.start()
pydantic_result = _extract_pydantic_result(result)
if pydantic_result is None:
raise ValueError("Could not extract structured CVAnalysisReport from workflow result")
# Access the structured results
print("\nCV Analysis Results:")
print(f"\nOverall Score: {result.pydantic.overall_score}/100")
print(f"\nOverall Score: {pydantic_result.overall_score}/100")
print("\nKey Strengths:")
for strength in result.pydantic.key_strengths:
for strength in pydantic_result.key_strengths:
print(f"- {strength}")
print("\nSkill Analysis:")
print("\nTechnical Skills:")
for skill in result.pydantic.skill_analysis.technical_skills:
for skill in pydantic_result.skill_analysis.technical_skills:
print(f"- {skill}")
print("\nExperience Highlights:")
for achievement in result.pydantic.experience_analysis.key_achievements:
for achievement in pydantic_result.experience_analysis.key_achievements:
print(f"- {achievement}")
print("\nRecommendations:")
for recommendation in result.pydantic.recommendations:
for recommendation in pydantic_result.recommendations:
print(f"- {recommendation}")
@@ -100,6 +100,8 @@ def evaluate_project(video_path: str) -> ProjectEvaluation:
evaluation_data = response
print(f"Debug - Parsed evaluation_data: {evaluation_data}")
if not evaluation_data or not isinstance(evaluation_data, dict):
raise ValueError("No structured evaluation data returned by agent")
# Create and return ProjectEvaluation instance
return ProjectEvaluation(
+9 -3
View File
@@ -1,5 +1,11 @@
from praisonaiagents import VideoAgent
import os
agent = VideoAgent(llm="azure/sora-2")
video = agent.generate("Mountain landscape")
print(video.id)
if __name__ == "__main__":
if not os.getenv("AZURE_API_BASE"):
print("AZURE_API_BASE is not configured. Skipping Azure video example.")
else:
agent = VideoAgent(llm="azure/sora-2")
video = agent.generate("Mountain landscape")
print(video.id)
+1 -1
View File
@@ -5,7 +5,7 @@ Demonstrates using Agent instances directly as workflow steps.
Each agent processes the input and passes output to the next agent.
"""
from praisonaiagents import Agent, Workflow
from praisonaiagents import Agent, AgentFlow
# Create agents with specific roles
researcher = Agent(
+5 -2
View File
@@ -6,6 +6,7 @@ with Agents for workflow monitoring and logging.
"""
from praisonaiagents import Agent, Task, AgentTeam
from praisonaiagents.config.feature_configs import MultiAgentHooksConfig
# Create an agent
researcher = Agent(
@@ -47,8 +48,10 @@ if __name__ == "__main__":
agents=[researcher],
tasks=[task1, task2],
process="workflow",
on_task_start=on_start,
on_task_complete=on_complete,
hooks=MultiAgentHooksConfig(
on_task_start=on_start,
on_task_complete=on_complete,
),
variables={"global_var": "shared_value"} # Global variables
)
@@ -54,11 +54,13 @@ workflow = AgentFlow(
analyzer, # Always runs
Task(
name="tech_review",
description="Review technical aspects when technical keywords are detected.",
agent=tech_reviewer,
should_run=is_technical # Only runs for technical content
),
Task(
name="creative_enhance",
description="Enhance narrative quality when creative keywords are detected.",
agent=creative_enhancer,
should_run=needs_creative # Only runs for creative content
),
@@ -53,7 +53,7 @@ workflow = AgentFlow(
# Step 4: Function - formats output
format_output
],
default_llm="gpt-4o-mini" # Used for action-based steps
llm="gpt-4o-mini", # Used for action-based steps
)
if __name__ == "__main__":
@@ -42,6 +42,7 @@ def main():
# Create tasks with robustness features
research_task = Task(
name="research",
description="Research the latest trends in AI agents for 2025",
expected_output="A summary of key AI agent trends",
agent=researcher,
@@ -51,6 +52,7 @@ def main():
# Optional enrichment step - workflow continues even if this fails
enrich_task = Task(
name="enrich",
description="Enrich the research with real-world examples",
expected_output="Research with added examples",
agent=enricher,
@@ -60,6 +62,7 @@ def main():
)
write_task = Task(
name="write",
description="Write a brief summary based on the research",
expected_output="A well-written summary paragraph",
agent=writer,
+9 -7
View File
@@ -51,13 +51,15 @@ print("=" * 60)
agent = Agent(
name="Knowledge Agent",
instructions="You are a helpful assistant that answers questions based on the provided knowledge. Be concise.",
knowledge=[".praison/test_docs/overview.txt"],
retrieval_config={
"policy": "always", # Always retrieve for this demo
"top_k": 3, # Get top 3 chunks
"citations": True, # Include citations
"max_context_tokens": 2000,
}
knowledge={
"sources": [".praison/test_docs/overview.txt"],
"config": {
"policy": "always",
"top_k": 3,
"citations": True,
"max_context_tokens": 2000,
},
},
)
# Test 1: Basic chat with retrieval
+1 -1
View File
@@ -60,7 +60,7 @@ print("=" * 50)
agent_with_discovery = Agent(
name="Multi-Skill Agent",
instructions="You are a versatile assistant with multiple skills.",
skills_dirs=["./"], # Scan current directory for skill subdirectories
skills=["./"], # Scan current directory for skill subdirectories
)
if agent_with_discovery.skill_manager:
+1 -1
View File
@@ -190,7 +190,7 @@ def example_run_history_with_backend():
# Use with RunHistory
history = RunHistory(backend=backend)
runs = history.list()
runs = history.list_runs()
print(f"RunHistory created with SQLite backend, {len(runs)} runs")
print("Recipe runs will be stored in database instead of files")
+1 -1
View File
@@ -257,7 +257,7 @@ if __name__ == "__main__":
# Check dependencies
try:
import praisonaiagents
print(f"✅ PraisonAI version: {praisonaiagents.__version__}")
print(f"✅ PraisonAI version: {getattr(praisonaiagents, '__version__', None)}")
except ImportError:
print("❌ PraisonAI not installed: pip install praisonaiagents")
sys.exit(1)
+1 -1
View File
@@ -12,7 +12,7 @@ assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY must be set"
# Enable web search and fetch
agent = Agent(
instructions="You are a research assistant with web access.",
web=True,
web="search_only",
)
# Disable web (default)
+3 -11
View File
@@ -4,7 +4,7 @@ WebConfig Example
Demonstrates using WebConfig for fine-grained control.
"""
import os
from praisonaiagents import Agent, WebConfig
from praisonaiagents import Agent
# Ensure API key is set from environment
assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY must be set"
@@ -12,21 +12,13 @@ assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY must be set"
# Custom web configuration
agent = Agent(
instructions="You are a research assistant.",
web=WebConfig(
search=True,
fetch=True,
search_provider="duckduckgo",
max_results=5,
),
web="search_only",
)
# Search only (no fetch)
agent_search_only = Agent(
instructions="You are a research assistant.",
web=WebConfig(
search=True,
fetch=False,
),
web="search_only",
)
if __name__ == "__main__":
+1 -1
View File
@@ -93,7 +93,7 @@ def example_3_execute_yaml_directly():
result = manager.execute_yaml(
yaml_file,
input_data="The future of AI",
output="verbose"
verbose=True,
)
print(f"Status: {result['status']}")
@@ -243,7 +243,7 @@ def _get_default_server_registry() -> ServerRegistry:
if TYPE_CHECKING:
from ..approval.protocols import ApprovalConfig, ApprovalProtocol
from ..config.feature_configs import LearnConfig, MemoryConfig
from ..config.feature_configs import LearnConfig, MemoryConfig, ToolConfig
from ..context.models import ContextConfig
from ..context.manager import ContextManager
from ..knowledge.knowledge import Knowledge
@@ -253,6 +253,7 @@ if TYPE_CHECKING:
from .handoff import Handoff, HandoffConfig, HandoffResult
from ..rag.models import RAGResult, ContextPack
from ..eval.results import EvaluationLoopResult
from ..tools.retry import RetryPolicy
# Import structured error from central errors module
from ..errors import BudgetExceededError
@@ -552,8 +553,7 @@ class Agent(SteeringMixin, SandboxMixin, UnifiedExecutionMixin, ToolExecutionMix
hooks: Optional[Union[List[Any], Dict[str, Any], 'HooksConfig']] = None,
skills: Optional[Union[List[str], str, Dict[str, Any], 'SkillsConfig']] = None,
approval: Optional[Union[bool, str, Dict[str, Any], 'ApprovalConfig', 'ApprovalProtocol']] = None,
tool_timeout: Optional[int] = None, # P8/G11: Timeout in seconds for each tool call
parallel_tool_calls: bool = False, # Gap 2: Enable parallel execution of batched LLM tool calls
tool_config: Optional[Union[bool, 'ToolConfig']] = None, # Tool execution configuration (timeout, retry, parallel)
learn: Optional[Union[bool, str, Dict[str, Any], 'LearnConfig']] = None, # Continuous learning (peer to memory)
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")
@@ -648,10 +648,10 @@ class Agent(SteeringMixin, SandboxMixin, UnifiedExecutionMixin, ToolExecutionMix
- LearnConfig: Custom configuration
Learning is a first-class citizen, peer to memory. It captures patterns,
preferences, and insights from interactions to improve future responses.
parallel_tool_calls: Enable parallel execution of batched LLM tool calls (default False).
When True and LLM returns multiple tool calls in a single response, they execute
concurrently instead of sequentially. Provides ~3x speedup for I/O-bound tools.
Maintains backward compatibility with False default.
tool_config: Tool execution configuration. Accepts:
- bool: True enables defaults, False disables
- ToolConfig: Custom configuration for timeout, retry policy, parallel execution
Consolidates tool timeout, retry policy, and parallel execution settings.
backend: External managed agent backend for hybrid execution. Accepts:
- ManagedAgentIntegration: External managed agent service
- None: Use local execution (default)
@@ -691,6 +691,7 @@ class Agent(SteeringMixin, SandboxMixin, UnifiedExecutionMixin, ToolExecutionMix
- auto_save → memory=MemoryConfig(auto_save=)
- rate_limiter → execution=ExecutionConfig(rate_limiter=)
- verification_hooks → autonomy=AutonomyConfig(verification_hooks=)
- Use tool_config=ToolConfig(...) for tool execution configuration
"""
# Add check at start if memory is requested
if memory is not None:
@@ -801,8 +802,7 @@ class Agent(SteeringMixin, SandboxMixin, UnifiedExecutionMixin, ToolExecutionMix
alternative="use 'execution=ExecutionConfig(rate_limiter=obj)' instead",
stacklevel=3
)
# Note: parallel_tool_calls is NOT deprecated - it's a new Gap 2 feature
# Both direct parameter and ExecutionConfig.parallel_tool_calls are supported
# Note: parallel_tool_calls moved to tool_config pattern
if verification_hooks is not None:
warn_deprecated_param(
"verification_hooks",
@@ -978,8 +978,8 @@ class Agent(SteeringMixin, SandboxMixin, UnifiedExecutionMixin, ToolExecutionMix
allow_code_execution = True
if _exec_config.code_mode != "safe":
code_execution_mode = _exec_config.code_mode
# Get parallel_tool_calls from ExecutionConfig, fall back to parameter
parallel_tool_calls = getattr(_exec_config, 'parallel_tool_calls', parallel_tool_calls)
# Get parallel_tool_calls from ExecutionConfig
parallel_tool_calls = getattr(_exec_config, 'parallel_tool_calls', False)
# Budget guard extraction
_max_budget = getattr(_exec_config, 'max_budget', None)
_on_budget_exceeded = getattr(_exec_config, 'on_budget_exceeded', 'stop') or 'stop'
@@ -987,7 +987,8 @@ class Agent(SteeringMixin, SandboxMixin, UnifiedExecutionMixin, ToolExecutionMix
max_iter, max_rpm, max_execution_time, max_retry_limit = 20, None, None, 2
_max_budget = None
_on_budget_exceeded = 'stop'
# Keep parallel_tool_calls parameter value when no ExecutionConfig provided
# Default to False when no ExecutionConfig provided
parallel_tool_calls = False
# (already set from parameter, no need to override)
# ─────────────────────────────────────────────────────────────────────
@@ -1514,8 +1515,14 @@ class Agent(SteeringMixin, SandboxMixin, UnifiedExecutionMixin, ToolExecutionMix
self.self_reflect = True if self_reflect is None else self_reflect
self.instructions = instructions
# Resolve tool_config for tool execution settings
_tool_config = Agent._resolve_tool_config(
tool_config=tool_config
)
# Gap 2: Store parallel tool calls setting for ToolCallExecutor selection
self.parallel_tool_calls = parallel_tool_calls
self.parallel_tool_calls = _tool_config.parallel if _tool_config else parallel_tool_calls
# G2: Store interrupt controller for cooperative cancellation
self.interrupt_controller = interrupt_controller
# Check for model name in environment variable if not provided
@@ -1914,8 +1921,14 @@ Your Goal: {self.goal}
self._pending_approvals = {}
self._approvals_lock = asyncio.Lock()
# Store the resolved tool_config for cloning (resolved earlier in __init__)
self._tool_config = _tool_config
# P8/G11: Tool timeout - prevent slow tools from blocking
self._tool_timeout = tool_timeout
self._tool_timeout = _tool_config.timeout if _tool_config else None
# Store tool retry policy for tool execution with exponential backoff
self._tool_retry_policy = _tool_config.retry_policy if _tool_config else None
# Cache for system prompts and formatted tools with eager thread-safe lock
# Use OrderedDict for LRU behavior
@@ -2031,6 +2044,19 @@ Your Goal: {self.goal}
# Sandbox configuration - initialize SandboxMixin
super().__init__(sandbox=sandbox)
@staticmethod
def _resolve_tool_config(tool_config):
"""Resolve tool_config parameter with backward compatibility."""
# Import here to avoid circular imports
from ..config.feature_configs import ToolConfig, resolve_tools
# Handle the new consolidated parameter
if tool_config is not None:
resolved_config = resolve_tools(tool_config)
return resolved_config
return None
def __deepcopy__(self, memo: dict) -> "Agent":
"""Custom deepcopy that creates fresh threading primitives.
@@ -2107,9 +2133,8 @@ Your Goal: {self.goal}
'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),
# Tool configuration - use consolidated config when available
'tool_config': getattr(self, '_tool_config', None),
# CLI backend
'cli_backend': getattr(self, '_cli_backend', None),
@@ -1064,7 +1064,6 @@ Your Goal: {self.goal}"""
# Use structured error classification for all error types (replaces legacy heuristic checks)
from ..llm.error_classifier import classify_llm_error
from ..llm.retry_utils import jittered_backoff
import time
model_name = self.llm if isinstance(self.llm, str) else "unknown"
session_id = getattr(self, '_session_id', 'unknown')
@@ -9,6 +9,7 @@ import time
import json
import logging
import inspect
import os
import asyncio
import threading
@@ -965,7 +966,80 @@ Write the complete compiled report:"""
return await self.achat(prompt, task_name=task_name, task_description=task_description, task_id=task_id)
async def execute_tool_async(self, function_name: str, arguments: Dict[str, Any], tool_call_id: Optional[str] = None, tools_override: Optional[List] = None) -> Any:
"""Async version of execute_tool"""
"""Async version of execute_tool with retry policy support"""
# Get retry policy (tool-level > agent-level > default)
retry_policy = self._get_tool_retry_policy(function_name)
last_exception = None
# Retry loop with exponential backoff
for attempt in range(retry_policy.max_attempts):
try:
result = await self._execute_tool_async_impl(function_name, arguments, tool_call_id, tools_override)
# Check if result is an error that should be retried
if isinstance(result, dict) and result.get("error"):
# Skip retry for non-retryable errors (approval, permission, etc.)
if (result.get("approval_denied") or
result.get("permission_denied") or
result.get("approval_error") or
result.get("circuit_open")):
return result
# Determine error type for retry policy
error_type = self._classify_error_type(result, last_exception)
# Check if we should retry
if not retry_policy.should_retry(error_type, attempt):
return result
# Don't retry on last attempt
if attempt == retry_policy.max_attempts - 1:
return result
# Emit retry hook event
delay_ms = retry_policy.get_delay_ms(attempt)
await self._emit_retry_hook_async(function_name, attempt + 1, delay_ms, result.get("error", "Unknown error"), retry_policy.max_attempts, error_type)
# Wait before retry (async)
await asyncio.sleep(delay_ms / 1000.0)
continue
else:
# Success - return result
return result
except Exception as e:
last_exception = e
# Determine if retryable
is_retryable = not isinstance(e, (ValueError, TypeError, AttributeError))
# Check if retryable and not last attempt
if not is_retryable or attempt == retry_policy.max_attempts - 1:
# Return error dict instead of raising for consistency with sync version
return {"error": f"Error in execute_tool_async: {str(e)}"}
# Determine error type
error_type = self._classify_error_type(None, e)
if not retry_policy.should_retry(error_type, attempt):
return {"error": f"Error in execute_tool_async: {str(e)}"}
# Emit retry hook event
delay_ms = retry_policy.get_delay_ms(attempt)
await self._emit_retry_hook_async(function_name, attempt + 1, delay_ms, str(e), retry_policy.max_attempts, error_type)
# Wait before retry (async)
await asyncio.sleep(delay_ms / 1000.0)
continue
# Should never reach here due to loop logic, but safety fallback
if last_exception:
return {"error": f"Error in execute_tool_async: {str(last_exception)}"}
return {"error": "Maximum retry attempts exceeded"}
async def _execute_tool_async_impl(self, function_name: str, arguments: Dict[str, Any], tool_call_id: Optional[str] = None, tools_override: Optional[List] = None) -> Any:
"""Internal async tool execution implementation"""
try:
logging.info(f"Executing async tool: {function_name} with arguments: {arguments}")
@@ -1304,3 +1378,51 @@ Write the complete compiled report:"""
print(f"\nTo add MCP capabilities, install: pip install {missing_module}")
return None
async def _emit_retry_hook_async(self, tool_name, attempt, delay_ms, error, max_attempts, error_type):
"""Emit ON_RETRY hook event (async version).
Args:
tool_name: Name of the tool being retried
attempt: Current attempt number (1-based)
delay_ms: Delay before retry in milliseconds
error: Error message or description
max_attempts: Maximum number of attempts configured
error_type: Classified error type
"""
try:
from ..hooks import HookEvent, OnRetryInput
# Only emit if we have a hook runner
hook_runner = getattr(self, '_hook_runner', None)
if hook_runner is None:
return
retry_input = OnRetryInput(
session_id=getattr(self, '_session_id', 'default'),
cwd=os.getcwd(),
event_name=HookEvent.ON_RETRY,
timestamp=str(time.time()),
agent_name=self.name,
tool_name=tool_name,
attempt=attempt,
delay_ms=delay_ms,
error=error,
max_attempts=max_attempts,
error_type=error_type
)
# Execute hook asynchronously if supported, otherwise use sync
if hasattr(hook_runner, 'execute_async'):
await hook_runner.execute_async(HookEvent.ON_RETRY, retry_input, target=tool_name)
else:
# Fallback to sync execution in executor to avoid blocking event loop
loop = asyncio.get_running_loop()
await loop.run_in_executor(
None,
lambda: hook_runner.execute_sync(HookEvent.ON_RETRY, retry_input, target=tool_name)
)
except Exception as e:
# Don't let hook failures break retry logic
logging.debug(f"Failed to emit async retry hook: {e}")
@@ -403,6 +403,12 @@ 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)
# Keep auto_save index in sync when per-turn persist shares session_id
if self.auto_save and self.auto_save == self._session_id:
with self._history_lock:
self._auto_save_last_index = (
getattr(self, "_auto_save_last_index", 0) + 1
)
self._persist_session_stats()
except Exception as e:
logging.warning(f"Failed to persist message to session store: {e}")
@@ -353,9 +353,9 @@ class RouterAgent(Agent):
return report
def _build_system_prompt(self) -> str:
def _build_system_prompt(self, tools=None, **kwargs) -> str:
"""Build system prompt (inherited from parent but can be customized)."""
base_prompt = super()._build_system_prompt()
base_prompt = super()._build_system_prompt(tools=tools, **kwargs)
# Add multi-model context if needed
if self.routing_strategy == "auto":
@@ -713,7 +713,111 @@ class ToolExecutionMixin:
return None, arguments
def _execute_tool_with_circuit_breaker(self, function_name, arguments):
"""Execute tool with circuit breaker protection.
"""Execute tool with retry policy and circuit breaker protection.
Args:
function_name: Name of the tool to execute
arguments: Arguments for the tool
Returns:
Tool execution result or circuit breaker error
"""
# Get retry policy (tool-level > agent-level > default)
retry_policy = self._get_tool_retry_policy(function_name)
last_exception = None
# Retry loop with exponential backoff
for attempt in range(retry_policy.max_attempts):
try:
result = self._execute_tool_with_circuit_breaker_impl(function_name, arguments)
# Check if result is an error that should be retried
if isinstance(result, dict) and result.get("error"):
# Skip retry for non-retryable errors (approval, permission, etc.)
if (result.get("approval_denied") or
result.get("permission_denied") or
result.get("approval_error") or
result.get("circuit_open")):
return result
# Determine error type for retry policy
error_type = self._classify_error_type(result, last_exception)
# Check if we should retry
if not retry_policy.should_retry(error_type, attempt):
return result
# Don't retry on last attempt
if attempt == retry_policy.max_attempts - 1:
return result
# Emit retry hook event
delay_ms = retry_policy.get_delay_ms(attempt)
self._emit_retry_hook(function_name, attempt + 1, delay_ms, result.get("error", "Unknown error"), retry_policy.max_attempts, error_type)
# Wait before retry
time.sleep(delay_ms / 1000.0)
continue
else:
# Success - return result
return result
except ToolExecutionError as e:
last_exception = e
# Check if the error is retryable
if not e.is_retryable or attempt == retry_policy.max_attempts - 1:
raise
# Determine error type
error_type = self._classify_error_type(None, e)
if not retry_policy.should_retry(error_type, attempt):
raise
# Emit retry hook event
delay_ms = retry_policy.get_delay_ms(attempt)
self._emit_retry_hook(function_name, attempt + 1, delay_ms, str(e), retry_policy.max_attempts, error_type)
# Wait before retry
time.sleep(delay_ms / 1000.0)
continue
except Exception as e:
# Wrap in ToolExecutionError for consistency
is_retryable = not isinstance(e, (ValueError, TypeError, AttributeError))
wrapped_error = ToolExecutionError(
f"Tool '{function_name}' failed: {e}",
tool_name=function_name,
agent_id=self.name,
is_retryable=is_retryable,
)
# Check if retryable and not last attempt
if not is_retryable or attempt == retry_policy.max_attempts - 1:
raise wrapped_error from e
# Determine error type
error_type = self._classify_error_type(None, wrapped_error)
if not retry_policy.should_retry(error_type, attempt):
raise wrapped_error from e
# Emit retry hook event
delay_ms = retry_policy.get_delay_ms(attempt)
self._emit_retry_hook(function_name, attempt + 1, delay_ms, str(e), retry_policy.max_attempts, error_type)
# Wait before retry
time.sleep(delay_ms / 1000.0)
continue
# Should never reach here due to loop logic, but safety fallback
if last_exception:
raise last_exception
return {"error": "Maximum retry attempts exceeded"}
def _execute_tool_with_circuit_breaker_impl(self, function_name, arguments):
"""Execute tool with circuit breaker protection (internal implementation).
Args:
function_name: Name of the tool to execute
@@ -1123,3 +1227,110 @@ class ToolExecutionMixin:
else:
return f"Unknown bridge tool: {function_name}"
def _get_tool_retry_policy(self, tool_name):
"""Get retry policy for a tool (tool-level > agent-level > default).
Args:
tool_name: Name of the tool to get retry policy for
Returns:
RetryPolicy instance
"""
from ..tools.retry import RetryPolicy
# Check for tool-level retry policy first
tools = getattr(self, 'tools', [])
# Handle non-iterable tools (e.g., single MCP instance)
if not isinstance(tools, (list, tuple)):
tools = [] # MCP or single tool instance - no tool-level policy lookup
for tool in tools:
if (callable(tool) and
getattr(tool, '__name__', '') == tool_name and
hasattr(tool, 'retry_policy')):
return tool.retry_policy
# Check for agent-level retry policy
agent_policy = getattr(self, '_tool_retry_policy', None)
if agent_policy is not None:
return agent_policy
# Return default retry policy (cached class-level instance)
if not hasattr(ToolExecutionMixin, '_default_retry_policy'):
ToolExecutionMixin._default_retry_policy = RetryPolicy()
return ToolExecutionMixin._default_retry_policy
def _classify_error_type(self, error_dict, exception):
"""Classify error type for retry policy matching.
Args:
error_dict: Error dictionary from tool execution (if any)
exception: Exception that was raised (if any)
Returns:
String error type for retry policy checking
"""
# Check error dict first
if error_dict and isinstance(error_dict, dict):
error_msg = error_dict.get("error", "").lower()
if "timeout" in error_msg or "timed out" in error_msg:
return "timeout"
elif "rate" in error_msg and "limit" in error_msg:
return "rate_limit"
elif "connection" in error_msg or "network" in error_msg:
return "connection_error"
# Check exception type
if exception:
exc_msg = str(exception).lower()
exc_type = type(exception).__name__.lower()
if "timeout" in exc_msg or "timeout" in exc_type:
return "timeout"
elif "rate" in exc_msg and "limit" in exc_msg:
return "rate_limit"
elif ("connection" in exc_msg or "network" in exc_msg or
"connection" in exc_type):
return "connection_error"
return "unknown"
def _emit_retry_hook(self, tool_name, attempt, delay_ms, error, max_attempts, error_type):
"""Emit ON_RETRY hook event.
Args:
tool_name: Name of the tool being retried
attempt: Current attempt number (1-based)
delay_ms: Delay before retry in milliseconds
error: Error message or description
max_attempts: Maximum number of attempts configured
error_type: Classified error type
"""
try:
from ..hooks import HookEvent, OnRetryInput
# Only emit if we have a hook runner
hook_runner = getattr(self, '_hook_runner', None)
if hook_runner is None:
return
retry_input = OnRetryInput(
session_id=getattr(self, '_session_id', 'default'),
cwd=os.getcwd(),
event_name=HookEvent.ON_RETRY,
timestamp=str(time.time()),
agent_name=self.name,
tool_name=tool_name,
attempt=attempt,
delay_ms=delay_ms,
error=error,
max_attempts=max_attempts,
error_type=error_type
)
# Execute hook synchronously
hook_runner.execute_sync(HookEvent.ON_RETRY, retry_input, target=tool_name)
except Exception as e:
# Don't let hook failures break retry logic
logging.debug(f"Failed to emit retry hook: {e}")
@@ -843,6 +843,67 @@ class ExecutionConfig:
)
@dataclass
class ToolConfig:
"""
Configuration for tool execution behavior.
Configuration for tool execution behavior including timeout, retry policy, and parallel execution.
Usage:
# Simple enable with defaults
Agent(tool_config=ToolConfig())
# With custom settings
Agent(tool_config=ToolConfig(
timeout=60,
retry_policy=RetryPolicy(max_attempts=5),
parallel=True,
))
# With timeout only
Agent(tool_config=ToolConfig(timeout=30))
"""
# Tool execution timeout in seconds
timeout: Optional[int] = None
# Retry policy for tool execution with exponential backoff
retry_policy: Optional[Any] = None # RetryPolicy instance
# Enable parallel execution of batched LLM tool calls
parallel: bool = False
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
return {
"timeout": self.timeout,
"retry_policy": (
self.retry_policy.to_dict()
if hasattr(self.retry_policy, 'to_dict')
else self.retry_policy
),
"parallel": self.parallel,
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "ToolConfig":
"""Create ToolConfig from dictionary."""
retry_policy = data.get("retry_policy")
if isinstance(retry_policy, dict):
try:
from ..tools.retry import RetryPolicy
retry_policy = RetryPolicy.from_dict(retry_policy)
except ImportError:
# Keep as dict if RetryPolicy not available
pass
return cls(
timeout=data.get("timeout"),
retry_policy=retry_policy,
parallel=data.get("parallel", False),
)
@dataclass
class TemplateConfig:
"""
@@ -1188,6 +1249,7 @@ 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]
ToolParam = Union[bool, ToolConfig] # bool = defaults, ToolConfig = custom
# =============================================================================
@@ -1454,6 +1516,19 @@ def resolve_tool_search(value: ToolSearchParam) -> Optional[ToolSearchConfig]:
return value
def resolve_tools(value: ToolParam) -> Optional[ToolConfig]:
"""Resolve tools= parameter following precedence ladder."""
if value is None or value is False:
return None
if value is True:
return ToolConfig()
if isinstance(value, dict):
return ToolConfig.from_dict(value)
if isinstance(value, ToolConfig):
return value
return value
__all__ = [
# Enums
"MemoryBackend",
@@ -1474,6 +1549,7 @@ __all__ = [
"WebConfig",
"OutputConfig",
"ExecutionConfig",
"ToolConfig",
"TemplateConfig",
"CachingConfig",
"HooksConfig",
@@ -1500,7 +1576,8 @@ __all__ = [
"HooksParam",
"SkillsParam",
"AutonomyParam",
"ToolSearchParam",
"ToolSearchParam",
"ToolParam",
# Precedence ladder resolvers
"resolve_memory",
"resolve_knowledge",
@@ -1513,4 +1590,5 @@ __all__ = [
"resolve_caching",
"resolve_autonomy",
"resolve_tool_search",
"resolve_tools",
]
@@ -178,7 +178,14 @@ class OnErrorInput(HookInput):
@dataclass
class OnRetryInput(HookInput):
"""Input for OnRetry hooks."""
"""Input for OnRetry hooks fired during tool execution retries."""
tool_name: str = ""
attempt: int = 1
delay_ms: int = 0
error: str = ""
max_attempts: int = 0
error_type: str = "unknown"
# Legacy fields for backward compatibility
retry_count: int = 0
max_retries: int = 3
error_message: str = ""
@@ -187,6 +194,13 @@ class OnRetryInput(HookInput):
def to_dict(self) -> Dict[str, Any]:
base = super().to_dict()
base.update({
"tool_name": self.tool_name,
"attempt": self.attempt,
"delay_ms": self.delay_ms,
"error": self.error,
"max_attempts": self.max_attempts,
"error_type": self.error_type,
# Legacy fields
"retry_count": self.retry_count,
"max_retries": self.max_retries,
"error_message": self.error_message,
@@ -11,6 +11,7 @@ from __future__ import annotations
import os
import uuid
import json
import logging
from typing import Any, Dict, List, Optional, TYPE_CHECKING
if TYPE_CHECKING:
@@ -340,14 +341,31 @@ class Session:
pass
if session_store is not None:
# Use SessionStore for conversation history
# Replace atomically to avoid duplicate appends on repeated save_state()
session_id = f"{self.session_id}_{agent_key}"
for msg in chat_history:
if isinstance(msg, dict):
messages = [
{
"role": msg.get("role", "user"),
"content": msg.get("content", ""),
}
for msg in chat_history
if isinstance(msg, dict)
]
if not messages:
logging.debug(f"No chat history to persist for session {session_id}")
elif hasattr(session_store, "set_chat_history"):
session_store.set_chat_history(session_id, messages)
else:
# Fallback to add_message - may create duplicates on repeated calls
logging.warning(
f"Session store lacks 'set_chat_history' method. Using fallback "
f"'add_message' which may create duplicates on repeated save_state() calls."
)
for msg in messages:
session_store.add_message(
session_id,
role=msg.get("role", "user"),
content=msg.get("content", ""),
role=msg["role"],
content=msg["content"],
)
else:
# Fallback to Memory.store_short_term() for backward compatibility
@@ -70,7 +70,8 @@ class FunctionTool(BaseTool):
description: Optional[str] = None,
version: str = "1.0.0",
availability: Optional[Callable[[], tuple[bool, str]]] = None,
dynamic_schema_overrides: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None
dynamic_schema_overrides: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None,
retry_policy: Optional[Any] = None
):
self._func = func
self.name = name or func.__name__
@@ -78,6 +79,7 @@ class FunctionTool(BaseTool):
self.version = version
self._availability = availability
self._schema_override = dynamic_schema_overrides
self.retry_policy = retry_policy
# Detect injected parameters
self._injected_params = get_injected_params(func)
@@ -205,7 +207,8 @@ def tool(
description: Optional[str] = None,
version: str = "1.0.0",
availability: Optional[Callable[[], tuple[bool, str]]] = None,
dynamic_schema_overrides: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None
dynamic_schema_overrides: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None,
retry_policy: Optional[Any] = None
) -> Union[FunctionTool, Callable[[Callable], FunctionTool]]:
"""Decorator to convert a function into a tool.
@@ -223,6 +226,10 @@ def tool(
@tool(availability=lambda: (bool(os.getenv("API_KEY")), "API_KEY missing"))
def my_func(x: str) -> str:
return x
@tool(retry_policy=RetryPolicy(max_attempts=5))
def my_func(x: str) -> str:
return x
Args:
func: The function to wrap (when used without parentheses)
@@ -231,6 +238,7 @@ def tool(
version: Tool version (default: "1.0.0")
availability: Function that returns (is_available, reason) tuple
dynamic_schema_overrides: Function to dynamically modify tool schema at runtime
retry_policy: RetryPolicy for tool execution with exponential backoff
Returns:
FunctionTool instance that wraps the function
@@ -242,7 +250,8 @@ def tool(
description=description,
version=version,
availability=availability,
dynamic_schema_overrides=dynamic_schema_overrides
dynamic_schema_overrides=dynamic_schema_overrides,
retry_policy=retry_policy
)
# Validate the tool at creation time for early error detection
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "1.6.54"
version = "1.6.55"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
@@ -19,7 +19,7 @@ from typing import Any, Dict, Optional
from praisonaiagents import Agent, tool
from praisonaiagents.config.feature_configs import (
OutputConfig, ExecutionConfig, ReflectionConfig,
TemplateConfig, CachingConfig,
TemplateConfig, CachingConfig, ToolConfig,
)
@@ -179,7 +179,7 @@ class TestAgentParamSmoke:
agent = Agent(name="AP", approval=False)
def test_param_tool_timeout(self):
agent = Agent(name="TT", tool_timeout=30)
agent = Agent(name="TT", tool_config=ToolConfig(timeout=30))
def test_param_learn(self):
agent = Agent(name="LN", learn=False)
@@ -349,12 +349,12 @@ class TestAgentParamReal:
# 33: tool_timeout
def test_real_tool_timeout(self, api_key_check):
"""Tests: tool_timeout=30"""
"""Tests: tool_config=ToolConfig(timeout=30)"""
agent = Agent(
name="TimeoutAgent",
instructions="Use echo_tool.",
tools=[echo_tool],
tool_timeout=30,
tool_config=ToolConfig(timeout=30),
llm="gpt-4o-mini",
)
result = agent.start("Echo the phrase 'timeout_test'.")
@@ -548,7 +548,7 @@ class TestMethodExistence:
'memory', 'knowledge', 'planning', 'reflection', 'guardrails',
'web', 'context', 'autonomy', 'verification_hooks',
'output', 'execution', 'templates', 'caching', 'hooks', 'skills',
'approval', 'tool_timeout', 'learn',
'approval', 'tool_config', 'learn',
]
missing = [p for p in expected if p not in params]
@@ -18,6 +18,7 @@ import asyncio
import time
from unittest.mock import Mock, patch, MagicMock, AsyncMock
from praisonaiagents import Agent, tool
from praisonaiagents.config.feature_configs import ToolConfig
from praisonaiagents.agent.unified_execution_mixin import UnifiedExecutionMixin
from praisonaiagents.llm.streaming_protocol import (
get_streaming_adapter,
@@ -220,7 +221,7 @@ class TestGap2ParallelToolExecution:
agent = Agent(
name="parallel_test",
instructions="You are a test agent",
parallel_tool_calls=True
tool_config=ToolConfig(parallel=True)
)
assert hasattr(agent, 'parallel_tool_calls')
@@ -370,7 +371,7 @@ class TestRealAgenticIntegration:
name="integration_test",
instructions="You are a helpful assistant. Use tools when needed.",
tools=[get_weather, get_time, get_news],
parallel_tool_calls=True, # Gap 2: Enable parallel execution
tool_config=ToolConfig(parallel=True), # Gap 2: Enable parallel execution
llm="gpt-4o-mini", # Use a real model
stream=True # Gap 3: Enable streaming
)
@@ -303,6 +303,63 @@ class TestMemoryHistorySeparation:
assert any(msg.get("role") == "assistant" for msg in history)
class TestPersistAutoSaveNoDuplicates:
"""Per-turn _persist_message must not double-save via _auto_save_session."""
def test_persist_then_auto_save_does_not_duplicate(self, tmp_path, monkeypatch):
"""memory=history + auto_save: one turn should persist exactly two messages."""
monkeypatch.setenv("PRAISONAI_SESSIONS_DIR", str(tmp_path))
import praisonaiagents.session as session_mod
session_mod._default_store = None
from praisonaiagents import Agent
from praisonaiagents.session import get_default_session_store
agent = Agent(
name="test",
instructions="Test agent",
memory="history",
auto_save="dup-session",
)
agent._init_session_store()
agent.chat_history = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi"},
]
agent._persist_message("user", "Hello")
agent._persist_message("assistant", "Hi")
agent._auto_save_session()
history = get_default_session_store().get_chat_history("dup-session")
assert len(history) == 2
class TestSessionSaveStateNoDuplicates:
"""Repeated save_state() must not append duplicate chat history."""
def test_repeated_save_agent_histories_is_idempotent(self, tmp_path, monkeypatch):
monkeypatch.setenv("PRAISONAI_SESSIONS_DIR", str(tmp_path))
import praisonaiagents.session as session_mod
session_mod._default_store = None
from unittest.mock import MagicMock
from praisonaiagents.session import Session, get_default_session_store
session = Session(session_id="sess1", user_id="user1")
mock_agent = MagicMock()
mock_agent.chat_history = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi"},
]
session._agents["agent1"] = {"agent": mock_agent, "chat_history": []}
session._save_agent_chat_histories()
session._save_agent_chat_histories()
history = get_default_session_store().get_chat_history("sess1_agent1")
assert len(history) == 2
class TestG3PerTurnPersist:
"""G-3: Verify per-turn persist exists via _persist_message()."""
@@ -0,0 +1,97 @@
"""
Tests for @tool(retry_policy=...) decorator functionality.
"""
from praisonaiagents import tool, Agent
from praisonaiagents.tools.retry import RetryPolicy
class TestToolDecoratorRetryPolicy:
"""Test @tool decorator with retry_policy parameter."""
def test_tool_decorator_with_retry_policy(self):
"""Test @tool decorator accepts retry_policy parameter."""
retry_policy = RetryPolicy(max_attempts=5, initial_delay_ms=500)
@tool(retry_policy=retry_policy)
def test_tool(query: str) -> str:
"""A test tool with retry policy."""
return f"Result for {query}"
assert hasattr(test_tool, 'retry_policy')
assert test_tool.retry_policy == retry_policy
assert test_tool.retry_policy.max_attempts == 5
assert test_tool.retry_policy.initial_delay_ms == 500
def test_tool_decorator_without_retry_policy(self):
"""Test @tool decorator works without retry_policy (default None)."""
@tool
def test_tool(query: str) -> str:
"""A test tool without retry policy."""
return f"Result for {query}"
assert hasattr(test_tool, 'retry_policy')
assert test_tool.retry_policy is None
def test_tool_with_retry_policy_used_by_agent(self):
"""Test that agent respects tool-level retry policy."""
tool_retry_policy = RetryPolicy(max_attempts=7, initial_delay_ms=300)
@tool(retry_policy=tool_retry_policy)
def special_tool(query: str) -> str:
"""Special tool with custom retry policy."""
return f"Special result for {query}"
# Agent with different retry policy
agent_retry_policy = RetryPolicy(max_attempts=2, initial_delay_ms=1000)
agent = Agent(
name="test_agent",
instructions="Test agent",
tools=[special_tool],
tool_retry_policy=agent_retry_policy
)
# Should get tool-level policy (higher precedence)
resolved_policy = agent._get_tool_retry_policy("special_tool")
assert resolved_policy == tool_retry_policy
assert resolved_policy.max_attempts == 7
assert resolved_policy.initial_delay_ms == 300
def test_mixed_tools_retry_policies(self):
"""Test agent with mix of tools - some with retry policy, some without."""
policy_a = RetryPolicy(max_attempts=3)
policy_b = RetryPolicy(max_attempts=6)
@tool(retry_policy=policy_a)
def tool_a(query: str) -> str:
return f"A: {query}"
@tool(retry_policy=policy_b)
def tool_b(query: str) -> str:
return f"B: {query}"
@tool # No retry policy
def tool_c(query: str) -> str:
return f"C: {query}"
agent_policy = RetryPolicy(max_attempts=4)
agent = Agent(
name="test_agent",
instructions="Test agent",
tools=[tool_a, tool_b, tool_c],
tool_retry_policy=agent_policy
)
# Tool A should use its own policy
policy_a_resolved = agent._get_tool_retry_policy("tool_a")
assert policy_a_resolved == policy_a
assert policy_a_resolved.max_attempts == 3
# Tool B should use its own policy
policy_b_resolved = agent._get_tool_retry_policy("tool_b")
assert policy_b_resolved == policy_b
assert policy_b_resolved.max_attempts == 6
# Tool C should use agent policy (fallback)
policy_c_resolved = agent._get_tool_retry_policy("tool_c")
assert policy_c_resolved == agent_policy
assert policy_c_resolved.max_attempts == 4
@@ -0,0 +1,200 @@
"""
Real agentic test for tool retry policy - agent actually runs and calls LLM with retrying tools.
This satisfies the AGENTS.md requirement (§9.4) for real agentic testing.
"""
import pytest
import time
from unittest.mock import patch
from praisonaiagents import Agent, tool
from praisonaiagents.tools.retry import RetryPolicy
class TestRetryPolicyAgentic:
"""Real agentic tests where the agent calls LLM and uses retrying tools."""
def test_agent_with_flaky_tool_real_llm(self):
"""
✅ REAL agentic test — agent runs end-to-end with flaky tool that retries.
Agent MUST call agent.start() with a real prompt and produce LLM output.
"""
call_count = [0]
@tool
def flaky_web_search(query: str) -> str:
"""Search the web for information. May fail due to network issues."""
call_count[0] += 1
# Fail on first two attempts, succeed on third
if call_count[0] <= 2:
raise Exception(f"Connection timeout on attempt {call_count[0]}")
# Return realistic search results
return f"""Found search results for "{query}":
1. OpenAI GPT-4 is a large language model released in 2023
2. It demonstrates improved reasoning capabilities over GPT-3.5
3. Available via API and ChatGPT Plus subscription"""
# Configure retry policy for this flaky tool
retry_policy = RetryPolicy(
max_attempts=3,
initial_delay_ms=50, # Fast for testing
retry_on={"timeout", "connection_error"},
backoff_factor=1.5
)
# Create agent with retry policy and flaky tool
agent = Agent(
name="research_assistant",
instructions="You are a helpful research assistant. When asked to search for information, use the flaky_web_search tool. Be concise in your response.",
tools=[flaky_web_search],
tool_retry_policy=retry_policy
)
# Mock time.sleep to speed up test
with patch('time.sleep'):
# ✅ REAL agent call with actual LLM interaction
result = agent.start("Search for information about GPT-4 and tell me one key fact.")
# Verify the agent got a real LLM response
assert isinstance(result, str), f"Expected string response, got {type(result)}"
assert len(result) > 10, "Response should be substantial"
# Verify the flaky tool was called and retried as expected
assert call_count[0] == 3, f"Tool should have been called 3 times, was called {call_count[0]} times"
# Print the full output so developers can verify end-to-end behavior
print(f"\n🤖 Agent Response: {result}")
print(f"🔄 Tool was retried {call_count[0]} times before success")
def test_agent_with_non_retryable_tool_real_llm(self):
"""
✅ REAL agentic test — agent with tool that fails non-retryably.
Verifies agent handles permission denied gracefully without infinite retries.
"""
call_count = [0]
@tool
def restricted_database_query(query: str) -> str:
"""Query a restricted database. Requires special permissions."""
call_count[0] += 1
return {
"error": "Access denied: insufficient permissions for database query",
"permission_denied": True
}
retry_policy = RetryPolicy(max_attempts=5) # High attempts to test non-retry
agent = Agent(
name="data_analyst",
instructions="You are a data analyst. If you cannot access the database, explain what happened and suggest alternatives. Keep it brief.",
tools=[restricted_database_query],
tool_retry_policy=retry_policy
)
# ✅ REAL agent call with actual LLM interaction
result = agent.start("Query the database for user statistics from last month.")
# Verify the agent got a real LLM response
assert isinstance(result, str), f"Expected string response, got {type(result)}"
assert len(result) > 10, "Response should be substantial"
# Verify tool was called only once (no retries for permission denied)
assert call_count[0] == 1, f"Tool should have been called only once, was called {call_count[0]} times"
# Print the full output for verification
print(f"\n🤖 Agent Response: {result}")
print(f"🚫 Tool was called {call_count[0]} times (no retries for permission denied)")
def test_agent_with_tool_level_retry_policy_real_llm(self):
"""
✅ REAL agentic test — tool-level retry policy overrides agent-level policy.
"""
primary_calls = [0]
fallback_calls = [0]
@tool
def primary_api(query: str) -> str:
"""Primary API that fails frequently."""
primary_calls[0] += 1
if primary_calls[0] <= 1: # Fail once, succeed twice
raise Exception("Primary API rate limit exceeded")
return f"Primary API result for: {query}"
# Tool-level policy: only 2 attempts
primary_api.retry_policy = RetryPolicy(
max_attempts=2,
initial_delay_ms=30,
retry_on={"rate_limit"}
)
@tool
def fallback_api(query: str) -> str:
"""Fallback API that's more reliable."""
fallback_calls[0] += 1
return f"Fallback API result for: {query}"
# Agent-level policy: 5 attempts (should be overridden by tool-level)
agent_retry_policy = RetryPolicy(max_attempts=5)
agent = Agent(
name="api_client",
instructions="You help users get information. Try the primary_api first. If it fails completely, you can try fallback_api. Be brief.",
tools=[primary_api, fallback_api],
tool_retry_policy=agent_retry_policy
)
with patch('time.sleep'):
# ✅ REAL agent call
result = agent.start("Get information about Python programming")
# Verify real LLM response
assert isinstance(result, str) and len(result) > 10
# Verify tool-level policy was respected (2 attempts max)
assert primary_calls[0] == 2, f"Primary API should have been called 2 times, was called {primary_calls[0]} times"
print(f"\n🤖 Agent Response: {result}")
print(f"🔄 Primary API called {primary_calls[0]} times (tool-level policy: max 2)")
print(f"🔄 Fallback API called {fallback_calls[0]} times")
@pytest.mark.asyncio
async def test_async_agent_with_retry_real_llm(self):
"""
✅ REAL agentic test — async agent with retrying tool.
"""
call_count = [0]
@tool
async def async_weather_api(location: str) -> str:
"""Get weather information for a location."""
call_count[0] += 1
if call_count[0] <= 1: # Fail once
raise Exception("Weather service connection timeout")
return f"Weather in {location}: Sunny, 72°F"
retry_policy = RetryPolicy(
max_attempts=3,
initial_delay_ms=25,
retry_on={"timeout"}
)
agent = Agent(
name="weather_assistant",
instructions="You provide weather information using the async_weather_api tool. Keep responses concise.",
tools=[async_weather_api],
tool_retry_policy=retry_policy
)
with patch('time.sleep'), patch('asyncio.sleep'):
# ✅ REAL async agent call
result = await agent.astart("What's the weather like in San Francisco?")
# Verify real LLM response
assert isinstance(result, str) and len(result) > 10
assert call_count[0] == 2, f"Async tool should have been called 2 times, was called {call_count[0]} times"
print(f"\n🤖 Async Agent Response: {result}")
print(f"🔄 Async tool retried {call_count[0]} times before success")
@@ -0,0 +1,396 @@
"""
Unit tests for tool retry policy integration in Agent class.
Tests sync and async retry, hook emission, clone_for_channel propagation, and non-retryable errors.
"""
import asyncio
import pytest
import time
from unittest.mock import Mock, patch, AsyncMock, call
from praisonaiagents import Agent, tool
from praisonaiagents.tools.retry import RetryPolicy
from praisonaiagents.hooks.types import HookEvent
class TestAgentRetryPolicyIntegration:
"""Test agent-level retry policy integration."""
def test_agent_retry_policy_initialization(self):
"""Test agent initialization with retry policy."""
retry_policy = RetryPolicy(max_attempts=5, initial_delay_ms=500)
agent = Agent(
name="test_agent",
instructions="Test agent",
tool_retry_policy=retry_policy
)
assert agent._tool_retry_policy == retry_policy
def test_agent_clone_propagates_retry_policy(self):
"""Test that clone_for_channel preserves tool_retry_policy."""
retry_policy = RetryPolicy(max_attempts=5, initial_delay_ms=500)
agent = Agent(
name="test_agent",
instructions="Test agent",
tool_retry_policy=retry_policy
)
cloned = agent.clone_for_channel()
assert cloned._tool_retry_policy == retry_policy
assert cloned._tool_retry_policy is not agent._tool_retry_policy # Different instance
def test_agent_clone_none_retry_policy(self):
"""Test clone works when retry_policy is None."""
agent = Agent(name="test_agent", instructions="Test agent")
cloned = agent.clone_for_channel()
assert cloned._tool_retry_policy is None
class TestSyncRetryIntegration:
"""Test synchronous retry integration."""
def test_sync_retry_with_timeout_error(self):
"""Test sync tool execution with retryable timeout error."""
call_count = [0]
@tool
def flaky_tool(query: str) -> str:
"""A tool that fails with timeout on first calls."""
call_count[0] += 1
if call_count[0] < 3:
raise Exception("Connection timeout")
return f"Success on attempt {call_count[0]}"
retry_policy = RetryPolicy(
max_attempts=3,
initial_delay_ms=10, # Fast for testing
retry_on={"timeout", "connection_error"}
)
agent = Agent(
name="test_agent",
instructions="Test agent",
tools=[flaky_tool],
tool_retry_policy=retry_policy
)
# Mock time.sleep to avoid actual delays
with patch('time.sleep'):
# This should succeed after retries
result = agent._execute_tool_with_circuit_breaker("flaky_tool", {"query": "test"})
assert not result.get("error")
assert "Success on attempt 3" in str(result)
assert call_count[0] == 3
def test_sync_retry_non_retryable_error(self):
"""Test sync tool execution with non-retryable error."""
call_count = [0]
@tool
def permission_denied_tool(query: str) -> str:
"""A tool that fails with permission denied."""
call_count[0] += 1
return {
"error": "Permission denied",
"permission_denied": True
}
retry_policy = RetryPolicy(max_attempts=3)
agent = Agent(
name="test_agent",
instructions="Test agent",
tools=[permission_denied_tool],
tool_retry_policy=retry_policy
)
result = agent._execute_tool_with_circuit_breaker("permission_denied_tool", {"query": "test"})
assert result.get("error")
assert result.get("permission_denied")
assert call_count[0] == 1 # Should not retry
def test_sync_retry_hook_emission(self):
"""Test that ON_RETRY hooks are emitted during sync retries."""
call_count = [0]
hook_calls = []
@tool
def failing_tool(query: str) -> str:
call_count[0] += 1
if call_count[0] < 3:
raise Exception("Rate limit exceeded")
return "Success"
def retry_hook(event_data):
hook_calls.append(event_data)
retry_policy = RetryPolicy(max_attempts=3, initial_delay_ms=10)
agent = Agent(
name="test_agent",
instructions="Test agent",
tools=[failing_tool],
tool_retry_policy=retry_policy
)
# Add hook manually
agent.hook_runner.register_handler(HookEvent.ON_RETRY, retry_hook)
with patch('time.sleep'):
result = agent._execute_tool_with_circuit_breaker("failing_tool", {"query": "test"})
assert len(hook_calls) == 2 # Two retries before success
assert all(call.tool_name == "failing_tool" for call in hook_calls)
assert hook_calls[0].attempt == 2 # Second attempt (1-based)
assert hook_calls[1].attempt == 3 # Third attempt
class TestAsyncRetryIntegration:
"""Test asynchronous retry integration."""
@pytest.mark.asyncio
async def test_async_retry_with_rate_limit_error(self):
"""Test async tool execution with retryable rate limit error."""
call_count = [0]
async def mock_async_impl(function_name, arguments, tool_call_id, tools_override):
call_count[0] += 1
if call_count[0] < 3:
return {
"error": "Rate limit exceeded. Try again later.",
"tool_name": function_name
}
return {
"result": f"Success on attempt {call_count[0]}",
"tool_name": function_name
}
retry_policy = RetryPolicy(
max_attempts=3,
initial_delay_ms=10,
retry_on={"rate_limit"}
)
agent = Agent(
name="test_agent",
instructions="Test agent",
tool_retry_policy=retry_policy
)
# Mock the internal async implementation
with patch.object(agent, '_execute_tool_async_impl', side_effect=mock_async_impl), \
patch('asyncio.sleep'): # Mock async sleep
result = await agent.execute_tool_async("test_tool", {"query": "test"})
assert not result.get("error")
assert "Success on attempt 3" in str(result)
assert call_count[0] == 3
@pytest.mark.asyncio
async def test_async_retry_circuit_open_guard(self):
"""Test async retry respects circuit_open flag."""
call_count = [0]
async def mock_async_impl(function_name, arguments, tool_call_id, tools_override):
call_count[0] += 1
return {
"error": "Circuit breaker is open",
"circuit_open": True,
"tool_name": function_name
}
retry_policy = RetryPolicy(max_attempts=3)
agent = Agent(
name="test_agent",
instructions="Test agent",
tool_retry_policy=retry_policy
)
with patch.object(agent, '_execute_tool_async_impl', side_effect=mock_async_impl):
result = await agent.execute_tool_async("test_tool", {"query": "test"})
assert result.get("error")
assert result.get("circuit_open")
assert call_count[0] == 1 # Should not retry
@pytest.mark.asyncio
async def test_async_retry_hook_emission(self):
"""Test that ON_RETRY hooks are emitted during async retries."""
call_count = [0]
hook_calls = []
async def mock_async_impl(function_name, arguments, tool_call_id, tools_override):
call_count[0] += 1
if call_count[0] < 3:
return {
"error": "Connection error occurred",
"tool_name": function_name
}
return {
"result": "Success",
"tool_name": function_name
}
async def async_retry_hook(event_data):
hook_calls.append(event_data)
retry_policy = RetryPolicy(max_attempts=3, initial_delay_ms=10)
agent = Agent(
name="test_agent",
instructions="Test agent",
tool_retry_policy=retry_policy
)
# Mock hook runner to capture calls
agent.hook_runner = Mock()
agent.hook_runner.execute_async = AsyncMock()
agent.hook_runner.execute_sync = Mock()
with patch.object(agent, '_execute_tool_async_impl', side_effect=mock_async_impl), \
patch('asyncio.sleep'):
result = await agent.execute_tool_async("test_tool", {"query": "test"})
# Verify hook was called for each retry
assert agent.hook_runner.execute_async.call_count == 2 # Two retries
# Verify hook calls have correct structure
calls = agent.hook_runner.execute_async.call_args_list
for i, call in enumerate(calls):
event, retry_input, _ = call[0]
assert event == HookEvent.ON_RETRY
assert retry_input.tool_name == "test_tool"
assert retry_input.attempt == i + 2 # 1-based, starts at 2nd attempt
class TestRetryPolicyPrecedence:
"""Test retry policy precedence (tool-level > agent-level > default)."""
def test_tool_level_policy_precedence(self):
"""Test that tool-level retry policy takes precedence."""
# Create a tool with its own retry policy
tool_policy = RetryPolicy(max_attempts=1)
@tool
def test_tool(query: str) -> str:
return "result"
# Add retry policy to the tool
test_tool.retry_policy = tool_policy
agent_policy = RetryPolicy(max_attempts=5)
agent = Agent(
name="test_agent",
instructions="Test agent",
tools=[test_tool],
tool_retry_policy=agent_policy
)
# Get the resolved policy for this tool
resolved_policy = agent._get_tool_retry_policy("test_tool")
assert resolved_policy == tool_policy
assert resolved_policy.max_attempts == 1
def test_agent_level_policy_precedence(self):
"""Test that agent-level policy is used when tool-level is absent."""
@tool
def test_tool(query: str) -> str:
return "result"
agent_policy = RetryPolicy(max_attempts=5)
agent = Agent(
name="test_agent",
instructions="Test agent",
tools=[test_tool],
tool_retry_policy=agent_policy
)
resolved_policy = agent._get_tool_retry_policy("test_tool")
assert resolved_policy == agent_policy
assert resolved_policy.max_attempts == 5
def test_default_policy_fallback(self):
"""Test that default policy is used when no other policy is set."""
@tool
def test_tool(query: str) -> str:
return "result"
agent = Agent(
name="test_agent",
instructions="Test agent",
tools=[test_tool]
)
resolved_policy = agent._get_tool_retry_policy("test_tool")
assert resolved_policy is not None
assert resolved_policy.max_attempts == 3 # Default value
def test_mcp_tools_no_crash(self):
"""Test that MCP tools (non-iterable) don't crash retry policy lookup."""
# Mock an MCP instance (non-iterable)
mock_mcp = Mock()
mock_mcp.__name__ = "mcp_instance"
agent = Agent(
name="test_agent",
instructions="Test agent",
tools=mock_mcp # Non-iterable single tool
)
# This should not crash
resolved_policy = agent._get_tool_retry_policy("some_tool")
assert resolved_policy is not None
assert resolved_policy.max_attempts == 3 # Default
class TestErrorClassification:
"""Test error type classification for retry decisions."""
def test_classify_timeout_error(self):
"""Test classification of timeout errors."""
agent = Agent(name="test", instructions="test")
# Test various timeout patterns
timeout_msgs = [
"Connection timeout",
"Request timed out",
"timeout occurred",
"TIMEOUT ERROR"
]
for msg in timeout_msgs:
error_type = agent._classify_error_type({"error": msg}, None)
assert error_type == "timeout", f"Failed to classify: {msg}"
def test_classify_rate_limit_error(self):
"""Test classification of rate limit errors."""
agent = Agent(name="test", instructions="test")
rate_limit_msgs = [
"Rate limit exceeded",
"rate limited",
"Too many requests",
"RATE_LIMIT_ERROR"
]
for msg in rate_limit_msgs:
error_type = agent._classify_error_type({"error": msg}, None)
assert error_type == "rate_limit", f"Failed to classify: {msg}"
def test_classify_connection_error(self):
"""Test classification of connection errors."""
agent = Agent(name="test", instructions="test")
connection_msgs = [
"Connection failed",
"Network error",
"connection refused",
"CONNECTION_ERROR"
]
for msg in connection_msgs:
error_type = agent._classify_error_type({"error": msg}, None)
assert error_type == "connection_error", f"Failed to classify: {msg}"
def test_classify_unknown_error(self):
"""Test classification of unknown errors."""
agent = Agent(name="test", instructions="test")
error_type = agent._classify_error_type({"error": "Something weird happened"}, None)
assert error_type == "unknown"
+1 -1
View File
@@ -3125,7 +3125,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "1.6.54"
version = "1.6.55"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.54.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.54.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.55.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.55.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+7 -6
View File
@@ -235,8 +235,9 @@ class AgentsGenerator:
# CLI users get the process default.
self._adapter_registry = adapter_registry or get_default_registry()
# Get framework adapter (availability already validated at CLI entry)
self.framework_adapter = self._get_framework_adapter(framework)
# Defer framework adapter creation until YAML is loaded
# This fixes the issue where empty framework string fails before YAML framework is read
self.framework_adapter = None
def _get_framework_adapter(self, framework: str) -> FrameworkAdapter:
"""
@@ -281,7 +282,7 @@ class AgentsGenerator:
self.logger.debug(f"CLI override: lsp = {cli_config['lsp']}")
# Handle agent-level overrides using unified approach
agent_level_fields = ['tool_timeout', 'planning_tools', 'autonomy', 'planning', 'web', 'web_fetch']
agent_level_fields = ['tool_timeout', 'tool_retry_policy', 'planning_tools', 'autonomy', 'planning', 'web', 'web_fetch']
agent_overrides = {k: v for k, v in cli_config.items() if k in agent_level_fields}
# Handle handoff configuration - convert CLI flags into handoff dict
@@ -377,7 +378,7 @@ class AgentsGenerator:
# Select framework with AutoGen version logic
framework = self._select_autogen_version(
self.framework or config.get('framework', 'crewai'),
self.framework or config.get('framework', 'praisonai'),
config,
)
@@ -537,8 +538,8 @@ class AgentsGenerator:
'role', 'goal', 'instructions', 'backstory', 'tools', 'toolsets', 'tasks', 'llm',
'function_calling_llm', 'allow_delegation', 'max_iter', 'max_rpm',
'max_execution_time', 'verbose', 'cache', 'system_template',
'prompt_template', 'response_template', 'tool_timeout', 'planning_tools',
'planning', 'autonomy', 'guardrails', 'streaming', 'stream',
'prompt_template', 'response_template', 'tool_timeout', 'tool_retry_policy',
'planning_tools', 'planning', 'autonomy', 'guardrails', 'streaming', 'stream',
'approval', 'skills', 'cli_backend', 'reflection', 'handoff', 'web', 'web_fetch'
}
+42
View File
@@ -1096,6 +1096,16 @@ class PraisonAI:
parser.add_argument("--tool-timeout", type=int, default=60,
help="Timeout in seconds for each tool call (default: 60)")
# Tool retry policy - handle transient failures with exponential backoff
parser.add_argument("--tool-retry-attempts", type=int, default=3,
help="Maximum retry attempts for tool failures (default: 3)")
parser.add_argument("--tool-retry-delay", type=int, default=1000,
help="Initial retry delay in milliseconds (default: 1000)")
parser.add_argument("--tool-retry-backoff", type=float, default=2.0,
help="Retry backoff multiplier (default: 2.0)")
parser.add_argument("--tool-retry-on", type=str, default="timeout,rate_limit,connection_error",
help="Error types to retry (comma-separated, default: timeout,rate_limit,connection_error)")
# Tool Approval - control tool execution approval
parser.add_argument("--trust", action="store_true", help="Auto-approve all tool executions (skip approval prompts)")
parser.add_argument("--approve-level", type=str, choices=["low", "medium", "high", "critical"],
@@ -4138,6 +4148,22 @@ Do NOT add any explanations or formatting."""
tool_timeout = getattr(self.args, 'tool_timeout', None)
if tool_timeout is not None:
cli_config['tool_timeout'] = tool_timeout
# Extract --tool-retry-* flags for retry policy
retry_attempts = getattr(self.args, 'tool_retry_attempts', 3)
retry_delay = getattr(self.args, 'tool_retry_delay', 1000)
retry_backoff = getattr(self.args, 'tool_retry_backoff', 2.0)
retry_on_str = getattr(self.args, 'tool_retry_on', "timeout,rate_limit,connection_error")
retry_on = set(error_type.strip() for error_type in retry_on_str.split(',') if error_type.strip())
if retry_attempts > 1: # Only create retry policy if retries are enabled
from praisonaiagents.tools.retry import RetryPolicy
cli_config['tool_retry_policy'] = RetryPolicy(
max_attempts=retry_attempts,
initial_delay_ms=retry_delay,
backoff_factor=retry_backoff,
retry_on=retry_on
)
# Extract --planning-tools flag
planning_tools = getattr(self.args, 'planning_tools', None)
@@ -4396,6 +4422,22 @@ Do NOT add any explanations or formatting."""
if tool_timeout and tool_timeout > 0:
agent_config["tool_timeout"] = tool_timeout
# Tool retry policy - handle transient failures with exponential backoff
retry_attempts = getattr(self.args, 'tool_retry_attempts', 3)
retry_delay = getattr(self.args, 'tool_retry_delay', 1000)
retry_backoff = getattr(self.args, 'tool_retry_backoff', 2.0)
retry_on_str = getattr(self.args, 'tool_retry_on', "timeout,rate_limit,connection_error")
retry_on = set(error_type.strip() for error_type in retry_on_str.split(',') if error_type.strip())
if retry_attempts > 1: # Only create retry policy if retries are enabled
from praisonaiagents.tools.retry import RetryPolicy
agent_config["tool_retry_policy"] = RetryPolicy(
max_attempts=retry_attempts,
initial_delay_ms=retry_delay,
backoff_factor=retry_backoff,
retry_on=retry_on
)
# Memory
if getattr(self.args, 'memory', False):
memory_kwargs = {}
Loaded 100 of 111 files, more files were not shown because too many files have changed in this diff. Show more