Compare commits

...
14 Commits
Author SHA1 Message Date
Mervin Praison 7a36352196 Release v4.5.39 2026-03-11 00:23:03 +00:00
Mervin Praison 6ae4ebedcb Merge branch 'main' of https://github.com/MervinPraison/PraisonAI 2026-03-11 00:20:17 +00:00
Mervin Praison 3377347087 feat(scheduler): add session ID to DeliveryTarget and scheduling commands
Added an optional session_id parameter to DeliveryTarget and scheduling
functions to preserve conversation context during job execution. This
enhancement allows agents to access prior chat history and improves
overall task management flexibility. Updated relevant documentation and
CLI commands accordingly.
2026-03-11 00:19:32 +00:00
github-actions[bot] f47aa248ea docs: auto-update api.md [skip ci] 2026-03-10 18:00:30 +00:00
Mervin Praison 0379e9f196 Release v4.5.38 2026-03-10 18:00:13 +00:00
Mervin Praison c841d46ae9 Release v4.5.37 2026-03-10 15:27:06 +00:00
Mervin Praison cd6fdd0107 feat(memory): implement MongoDB storage backend and memory protocol aliases
Added a MongoDB backend for storage, which allows document-oriented
data storage. Implemented protocol-compliant aliases for memory operations
to improve consistency. Updated related memory management logic to support
new functionality, enhancing overall performance and user experience.
2026-03-10 15:23:43 +00:00
Mervin Praison d840407b12 Release v4.5.36 2026-03-10 14:05:29 +00:00
Mervin Praison b99ca3f0ae Release v4.5.35 2026-03-10 13:09:43 +00:00
Mervin Praison 23c48f6cd5 Release v4.5.34 2026-03-10 11:16:45 +00:00
Mervin Praison d0fa48b110 Release v4.5.33 2026-03-10 06:07:07 +00:00
Mervin Praison 0d2c1bd961 Release v4.5.32 2026-03-10 05:42:59 +00:00
Mervin Praison ef6b1a8913 Release v4.5.31 2026-03-09 14:43:23 +00:00
github-actions[bot] dc9c250adb docs: auto-update api.md [skip ci] 2026-03-09 14:06:13 +00:00
51 changed files with 4248 additions and 232 deletions

No files matched your search

+5 -1
View File
@@ -381,7 +381,7 @@ from praisonaiagents import GuardrailAction, GuardrailConfig, LLMGuardrail
Methods:
* <code title="class GuardrailConfig">GuardrailConfig.<a href="./src/praisonai-agents/praisonaiagents/config/feature_configs.py">to_dict</a>() -> Dict[str, Any]</code>
* <code title="class LLMGuardrail">LLMGuardrail.<a href="./src/praisonai-agents/praisonaiagents/guardrails/llm_guardrail.py">__call__</a>(task_output: TaskOutput) -> Tuple[bool, Union[str, TaskOutput]]</code>
* <code title="class LLMGuardrail">LLMGuardrail.<a href="./src/praisonai-agents/praisonaiagents/guardrails/llm_guardrail.py">__call__</a>(task_output) -> Tuple[bool, Union[str, 'TaskOutput']]</code>
# Planning
@@ -868,7 +868,9 @@ Methods:
* <code title="cli">praisonai examples run <a href="./src/praisonai/praisonai/cli/commands/examples.py">--help</a></code>
* <code title="cli">praisonai examples run-all <a href="./src/praisonai/praisonai/cli/commands/examples.py">--help</a></code>
* <code title="cli">praisonai examples stats <a href="./src/praisonai/praisonai/cli/commands/examples.py">--help</a></code>
* <code title="cli">praisonai gateway channels <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway gateway-callback <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway send <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway start <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway status <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai hooks add <a href="./src/praisonai/praisonai/cli/commands/hooks.py">--help</a></code>
@@ -906,6 +908,7 @@ Methods:
* <code title="cli">praisonai package install <a href="./src/praisonai/praisonai/cli/commands/package.py">--help</a></code>
* <code title="cli">praisonai package list <a href="./src/praisonai/praisonai/cli/commands/package.py">--help</a></code>
* <code title="cli">praisonai package uninstall <a href="./src/praisonai/praisonai/cli/commands/package.py">--help</a></code>
* <code title="cli">praisonai paths show <a href="./src/praisonai/praisonai/cli/commands/paths.py">--help</a></code>
* <code title="cli">praisonai plugins create <a href="./src/praisonai/praisonai/cli/commands/plugins.py">--help</a></code>
* <code title="cli">praisonai plugins disable <a href="./src/praisonai/praisonai/cli/commands/plugins.py">--help</a></code>
* <code title="cli">praisonai plugins discover <a href="./src/praisonai/praisonai/cli/commands/plugins.py">--help</a></code>
@@ -963,6 +966,7 @@ Methods:
* <code title="cli">praisonai sandbox recreate <a href="./src/praisonai/praisonai/cli/commands/sandbox.py">--help</a></code>
* <code title="cli">praisonai sandbox sandbox-callback <a href="./src/praisonai/praisonai/cli/commands/sandbox.py">--help</a></code>
* <code title="cli">praisonai sandbox status <a href="./src/praisonai/praisonai/cli/commands/sandbox.py">--help</a></code>
* <code title="cli">praisonai schedule add <a href="./src/praisonai/praisonai/cli/commands/schedule.py">--help</a></code>
* <code title="cli">praisonai schedule delete <a href="./src/praisonai/praisonai/cli/commands/schedule.py">--help</a></code>
* <code title="cli">praisonai schedule describe <a href="./src/praisonai/praisonai/cli/commands/schedule.py">--help</a></code>
* <code title="cli">praisonai schedule list <a href="./src/praisonai/praisonai/cli/commands/schedule.py">--help</a></code>
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=4.5.30" \
"praisonai>=4.5.39" \
"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.5.30" \
"praisonai>=4.5.39" \
"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.5.30" \
"praisonai>=4.5.39" \
"praisonai[ui]" \
"praisonai[crewai]"
@@ -344,6 +344,14 @@ _LAZY_IMPORTS = {
'ObservabilityEventType': ('praisonaiagents.escalation.observability', 'EventType'),
'DoomLoopDetector': ('praisonaiagents.escalation.doom_loop', 'DoomLoopDetector'),
'MemoryBackend': ('praisonaiagents.config.feature_configs', 'MemoryBackend'),
'LearnConfig': ('praisonaiagents.config.feature_configs', 'LearnConfig'),
'LearnScope': ('praisonaiagents.config.feature_configs', 'LearnScope'),
'LearnMode': ('praisonaiagents.config.feature_configs', 'LearnMode'),
'LearnBackend': ('praisonaiagents.config.feature_configs', 'LearnBackend'),
'LearnProtocol': ('praisonaiagents.memory.learn.protocols', 'LearnProtocol'),
'AsyncLearnProtocol': ('praisonaiagents.memory.learn.protocols', 'AsyncLearnProtocol'),
'LearnManagerProtocol': ('praisonaiagents.memory.learn.protocols', 'LearnManagerProtocol'),
'LearnManager': ('praisonaiagents.memory.learn', 'LearnManager'),
'ChunkingStrategy': ('praisonaiagents.config.feature_configs', 'ChunkingStrategy'),
'GuardrailAction': ('praisonaiagents.config.feature_configs', 'GuardrailAction'),
'WebSearchProvider': ('praisonaiagents.config.feature_configs', 'WebSearchProvider'),
@@ -439,6 +439,7 @@ class Agent:
skills: Optional[Union[List[str], Any]] = None, # Union[list, SkillsConfig]
approval: Optional[Union[bool, Any]] = None, # Union[bool, ApprovalProtocol backend]
tool_timeout: Optional[int] = None, # P8/G11: Timeout in seconds for each tool call
learn: Optional[Union[bool, Any]] = None, # Union[bool, LearnConfig] - Continuous learning (peer to memory)
):
"""Initialize an Agent instance.
@@ -509,6 +510,11 @@ class Agent:
skills: Agent skills. Accepts:
- List[str]: Skill directory paths
- SkillsConfig: Custom configuration
learn: Continuous learning configuration. Accepts:
- bool: True enables with defaults, False disables
- LearnConfig: Custom configuration
Learning is a first-class citizen, peer to memory. It captures patterns,
preferences, and insights from interactions to improve future responses.
Raises:
ValueError: If all of name, role, goal, backstory, and instructions are None.
@@ -969,6 +975,49 @@ class Agent:
elif memory is False:
memory = None
# ─────────────────────────────────────────────────────────────────────
# Resolve LEARN param - FAST PATH (top-level, peer to memory)
# learn= is a first-class citizen, independent of memory=
# ─────────────────────────────────────────────────────────────────────
from ..config.feature_configs import LearnConfig
_learn_config = None
# Top-level learn= takes precedence over memory.learn
if learn is not None:
# Explicit top-level learn= param
if learn is False:
_learn_config = None
elif learn is True:
# learn=True shorthand enables AGENTIC mode (auto-learning)
from ..memory.learn.protocols import LearnMode
_learn_config = LearnConfig(mode=LearnMode.AGENTIC)
elif isinstance(learn, LearnConfig):
_learn_config = learn
elif isinstance(learn, dict):
_learn_config = LearnConfig(**learn)
else:
_learn_config = learn # Pass through
elif _memory_config is not None and isinstance(_memory_config, MemoryConfig):
# Fallback to memory.learn for backward compatibility
if _memory_config.learn:
if _memory_config.learn is True:
# learn=True shorthand enables AGENTIC mode
from ..memory.learn.protocols import LearnMode
_learn_config = LearnConfig(mode=LearnMode.AGENTIC)
elif isinstance(_memory_config.learn, LearnConfig):
_learn_config = _memory_config.learn
elif isinstance(_memory_config.learn, dict):
_learn_config = LearnConfig(**_memory_config.learn)
# If learn is enabled but memory is not, we need to enable memory for learn to work
if _learn_config is not None and memory is None:
# Enable minimal memory for learn to work
memory = {"learn": _learn_config.to_dict() if hasattr(_learn_config, 'to_dict') else _learn_config}
elif _learn_config is not None and isinstance(memory, dict):
# Merge learn config into memory dict
memory["learn"] = _learn_config.to_dict() if hasattr(_learn_config, 'to_dict') else _learn_config
# ─────────────────────────────────────────────────────────────────────
# Resolve HISTORY from MemoryConfig - FAST PATH
# History is enabled via memory="history" preset or MemoryConfig(history=True)
@@ -1575,6 +1624,10 @@ Your Goal: {self.goal}
self._history_enabled = _history_enabled
self._history_limit = _history_limit
self._history_session_id = _history_session_id
# Learning configuration (top-level, peer to memory)
# Stored for access via agent._learn_config
self._learn_config = _learn_config
# Agent-centric feature instances (lazy loaded for zero performance impact)
self._auto_memory = auto_memory
@@ -3422,10 +3475,17 @@ Summary:"""
if not self._memory_instance:
return
if memory_type == "short_term" and hasattr(self._memory_instance, 'add_short_term'):
self._memory_instance.add_short_term(content, **kwargs)
elif memory_type == "long_term" and hasattr(self._memory_instance, 'add_long_term'):
self._memory_instance.add_long_term(content, **kwargs)
# Use protocol names first (store_*), fallback to legacy names (add_*)
if memory_type == "short_term":
if hasattr(self._memory_instance, 'store_short_term'):
self._memory_instance.store_short_term(content, **kwargs)
elif hasattr(self._memory_instance, 'add_short_term'):
self._memory_instance.add_short_term(content, **kwargs)
elif memory_type == "long_term":
if hasattr(self._memory_instance, 'store_long_term'):
self._memory_instance.store_long_term(content, **kwargs)
elif hasattr(self._memory_instance, 'add_long_term'):
self._memory_instance.add_long_term(content, **kwargs)
elif memory_type == "entity" and hasattr(self._memory_instance, 'add_entity'):
self._memory_instance.add_entity(content, **kwargs)
elif memory_type == "episodic" and hasattr(self._memory_instance, 'add_episodic'):
@@ -4603,6 +4663,36 @@ Your Goal: {self.goal}"""
prompt_str = prompt if isinstance(prompt, str) else str(prompt)
self._process_auto_memory(prompt_str, str(response))
# Auto-learning extraction (opt-in via LearnConfig(mode=LearnMode.AGENTIC))
self._process_auto_learning()
return response
async def _atrigger_after_agent_hook(self, prompt, response, start_time, tools_used=None):
"""Async version: Trigger AFTER_AGENT hook and return response."""
from ..hooks import HookEvent, AfterAgentInput
after_agent_input = AfterAgentInput(
session_id=getattr(self, '_session_id', 'default'),
cwd=os.getcwd(),
event_name=HookEvent.AFTER_AGENT,
timestamp=str(time.time()),
agent_name=self.name,
prompt=prompt if isinstance(prompt, str) else str(prompt),
response=response or "",
tools_used=tools_used or [],
total_tokens=0,
execution_time_ms=(time.time() - start_time) * 1000
)
await self._hook_runner.execute(HookEvent.AFTER_AGENT, after_agent_input)
# Auto-memory extraction (opt-in via MemoryConfig(auto_memory=True))
if response:
prompt_str = prompt if isinstance(prompt, str) else str(prompt)
self._process_auto_memory(prompt_str, str(response))
# Auto-learning extraction (opt-in via LearnConfig(mode=LearnMode.AGENTIC))
self._process_auto_learning()
return response
@@ -4863,7 +4953,14 @@ Your Goal: {self.goal}"""
return {"error": error_msg}
def clear_history(self):
"""Clear all chat history.
Also resets _auto_save_last_index to prevent silent message loss
when auto_save is enabled.
"""
self.chat_history = []
# Reset auto-save index to prevent stale index causing message loss
self._auto_save_last_index = 0
# -------------------------------------------------------------------------
# History Management Methods
@@ -4888,6 +4985,8 @@ Your Goal: {self.goal}"""
deleted_count = len(self.chat_history) - keep_last
self.chat_history = self.chat_history[-keep_last:]
# Reset auto-save index to match new history length
self._auto_save_last_index = len(self.chat_history)
return deleted_count
def delete_history(self, index: int) -> bool:
@@ -4905,6 +5004,12 @@ Your Goal: {self.goal}"""
with self._history_lock:
try:
del self.chat_history[index]
# Adjust auto-save index if deletion affects saved range
if hasattr(self, '_auto_save_last_index'):
self._auto_save_last_index = min(
self._auto_save_last_index,
len(self.chat_history)
)
return True
except IndexError:
return False
@@ -4927,7 +5032,14 @@ Your Goal: {self.goal}"""
msg for msg in self.chat_history
if pattern.lower() not in msg.get("content", "").lower()
]
return original_len - len(self.chat_history)
deleted_count = original_len - len(self.chat_history)
# Adjust auto-save index if deletion affects saved range
if deleted_count > 0 and hasattr(self, '_auto_save_last_index'):
self._auto_save_last_index = min(
self._auto_save_last_index,
len(self.chat_history)
)
return deleted_count
def get_history_size(self) -> int:
"""Get the current number of messages in chat history."""
@@ -5987,7 +6099,7 @@ Your Goal: {self.goal}"""
validated_response = self._apply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return validated_response
return self._trigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for JSON output: {e}")
# Rollback chat history on guardrail failure
@@ -6019,7 +6131,7 @@ Your Goal: {self.goal}"""
validated_response = self._apply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return validated_response
return self._trigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
# Rollback chat history on guardrail failure
@@ -6089,7 +6201,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
self._end_run(validated_response, "completed", {"duration_ms": (time.time() - start_time) * 1000})
return validated_response
return self._trigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed after reflection: {e}")
# Rollback chat history on guardrail failure
@@ -6108,7 +6220,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
validated_response = self._apply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return validated_response
return self._trigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed after max reflections: {e}")
# Rollback chat history on guardrail failure
@@ -6296,7 +6408,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
validated_response = self._apply_guardrail_with_retry(response_text, prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(normalized_content, validated_response, time.time() - start_time, task_name, task_description, task_id)
return validated_response
return await self._atrigger_after_agent_hook(prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for custom LLM: {e}")
# Rollback chat history on guardrail failure
@@ -6374,7 +6486,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
logging.debug(f"Agent.achat completed in {total_time:.2f} seconds")
# Execute callback after tool completion
self._execute_callback_and_display(original_prompt, result, time.time() - start_time, task_name, task_description, task_id)
return result
return await self._atrigger_after_agent_hook(original_prompt, result, start_time)
elif output_json or output_pydantic:
response = await self._openai_client.async_client.chat.completions.create(
model=self.llm,
@@ -6388,7 +6500,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
logging.debug(f"Agent.achat completed in {total_time:.2f} seconds")
# Execute callback after JSON/Pydantic completion
self._execute_callback_and_display(original_prompt, response_text, time.time() - start_time, task_name, task_description, task_id)
return response_text
return await self._atrigger_after_agent_hook(original_prompt, response_text, start_time)
else:
response = await self._openai_client.async_client.chat.completions.create(
model=self.llm,
@@ -6428,7 +6540,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
if logging.getLogger().getEffectiveLevel() == logging.DEBUG:
total_time = time.time() - start_time
logging.debug(f"Agent.achat completed in {total_time:.2f} seconds")
return response_text
return await self._atrigger_after_agent_hook(original_prompt, response_text, start_time)
reflection_response = await self._openai_client.async_client.beta.chat.completions.parse(
model=self.reflect_llm if self.reflect_llm else self.llm,
@@ -6486,7 +6598,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
validated_response = self._apply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return validated_response
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for OpenAI client: {e}")
# Rollback chat history on guardrail failure
@@ -7207,9 +7319,88 @@ Write the complete compiled report:"""
except Exception as e:
logging.debug(f"Auto-memory extraction failed: {e}")
def _process_auto_learning(self):
"""Process auto-learning extraction after agent response.
Called after each agent response when LearnMode.AGENTIC is set.
Uses LearnManager.process_conversation() to extract and store learnings
(persona, insights, patterns) from the conversation. No-op when learning
is disabled or mode is not AGENTIC.
"""
# Quick exit if no learn config or mode is not AGENTIC
if not self._learn_config:
return
# Check if mode is AGENTIC (auto-extract learnings)
from ..memory.learn.protocols import LearnMode
mode = getattr(self._learn_config, 'mode', None)
# Handle PROPOSE mode — extract but store as pending for user approval
if (isinstance(mode, LearnMode) and mode == LearnMode.PROPOSE) or \
(isinstance(mode, str) and mode == 'propose'):
learn_manager = None
if self._memory_instance and hasattr(self._memory_instance, 'learn'):
learn_manager = self._memory_instance.learn
if learn_manager:
try:
recent_messages = self.chat_history[-2:] if len(self.chat_history) >= 2 else self.chat_history
if recent_messages:
# Use same extraction as AGENTIC, but don't auto-store
result = learn_manager.process_conversation(
messages=recent_messages,
llm=getattr(self._learn_config, 'llm', None) or self.llm,
extract_only=True,
)
# Move extracted items to pending queue
if result:
for p in result.get("persona", []):
if p:
learn_manager.add_pending(p, category="persona")
for i in result.get("insights", []):
if i:
learn_manager.add_pending(i, category="insights")
for pt in result.get("patterns", []):
if pt:
learn_manager.add_pending(pt, category="patterns")
except Exception as e:
logging.debug(f"PROPOSE mode learning extraction failed: {e}")
return
if mode is None or (isinstance(mode, LearnMode) and mode != LearnMode.AGENTIC):
return
if isinstance(mode, str) and mode != 'agentic':
return
# Get LearnManager from memory instance
learn_manager = None
if self._memory_instance and hasattr(self._memory_instance, 'learn'):
learn_manager = self._memory_instance.learn
if not learn_manager:
return
try:
# Process recent conversation (last 2 messages: user + assistant)
recent_messages = self.chat_history[-2:] if len(self.chat_history) >= 2 else self.chat_history
if recent_messages:
learn_manager.process_conversation(
messages=recent_messages,
llm=getattr(self._learn_config, 'llm', None) or self.llm,
)
except Exception as e:
logging.debug(f"Auto-learning extraction failed: {e}")
def _auto_save_session(self):
"""Auto-save session if auto_save is enabled."""
if not self.auto_save or not self._memory_instance:
"""Auto-save session if auto_save is enabled.
G-1 FIX: Routes to SessionStore instead of Memory.save_session() to
maintain clean separation between conversation history (SessionStore)
and semantic memory (Memory).
Issue 1 FIX: Track last saved index to avoid duplicate insertion when
called multiple times in the same session.
"""
if not self.auto_save:
return
try:
@@ -7219,12 +7410,36 @@ Write the complete compiled report:"""
for msg in self.chat_history
]
self._memory_instance.save_session(
name=self.auto_save,
conversation_history=clean_history,
metadata={"agent_name": self.name, "user_id": self.user_id}
)
logging.debug(f"Auto-saved session: {self.auto_save}")
# Issue 1 FIX: Only save NEW messages since last save
# Track last saved index to avoid duplicates
last_saved = getattr(self, '_auto_save_last_index', 0)
new_messages = clean_history[last_saved:]
if not new_messages:
return # Nothing new to save
# G-1 FIX: Use SessionStore for conversation history persistence
# This maintains clean separation: Memory = facts, SessionStore = turns
if self._session_store is not None:
# Persist only NEW messages to SessionStore
for msg in new_messages:
self._session_store.add_message(
self.auto_save, # Use auto_save name as session_id
role=msg.get("role", "user"),
content=msg.get("content", ""),
)
# Update last saved index
self._auto_save_last_index = len(clean_history)
logging.debug(f"Auto-saved {len(new_messages)} new messages to SessionStore: {self.auto_save}")
elif self._memory_instance and hasattr(self._memory_instance, 'save_session'):
# Fallback to Memory.save_session() for backward compatibility
# Memory.save_session() replaces entire history, so no duplicate issue
self._memory_instance.save_session(
name=self.auto_save,
conversation_history=clean_history,
metadata={"agent_name": self.name, "user_id": self.user_id}
)
logging.debug(f"Auto-saved session to Memory: {self.auto_save}")
except Exception as e:
logging.debug(f"Error auto-saving session: {e}")
@@ -220,6 +220,7 @@ class AgentTeam:
web: Optional[Any] = None, # Union[bool, WebConfig] - web search/fetch
reflection: Optional[Any] = None, # Union[bool, ReflectionConfig] - self-reflection
caching: Optional[Any] = None, # Union[bool, CachingConfig] - caching
learn: Optional[Any] = None, # Union[bool, LearnConfig] - continuous learning
):
"""
Initialize AgentManager with consolidated feature parameters.
@@ -243,8 +244,10 @@ class AgentTeam:
guardrails: Guardrails configuration (bool | Callable | GuardrailConfig)
web: Web search/fetch configuration (bool | WebConfig)
reflection: Self-reflection configuration (bool | ReflectionConfig)
learn: Continuous learning configuration (bool | LearnConfig)
"""
# Store new params for propagation to agents
self._learn = learn
self.llm = llm # Store default LLM for API consistency
self._autonomy = autonomy
self._knowledge = knowledge
@@ -57,6 +57,34 @@ class LearnScope(str, Enum):
SHARED = "shared" # Shared with all agents
class LearnMode(str, Enum):
"""Learning mode for automatic learning extraction.
DISABLED: No automatic learning (manual capture only)
AGENTIC: Agent autonomously extracts and stores learnings after each conversation
PROPOSE: (Future) Agent proposes learnings for user approval before storing.
Note: PROPOSE mode is defined but not yet implemented. Using it will
behave the same as DISABLED until the approval workflow is added.
"""
DISABLED = "disabled"
AGENTIC = "agentic"
PROPOSE = "propose"
class LearnBackend(str, Enum):
"""Storage backend for learning data.
FILE: JSON files (default, zero dependencies)
SQLITE: SQLite database
REDIS: Redis (requires redis package)
MONGODB: MongoDB (requires pymongo)
"""
FILE = "file"
SQLITE = "sqlite"
REDIS = "redis"
MONGODB = "mongodb"
@dataclass
class LearnConfig:
"""
@@ -66,16 +94,21 @@ class LearnConfig:
to improve future responses. All learning data is stored within the memory system.
Usage:
# Simple enable
Agent(memory=MemoryConfig(learn=True))
# Simple enable (top-level param)
Agent(learn=True)
# With specific capabilities
Agent(memory=MemoryConfig(
learn=LearnConfig(
persona=True, # User preferences
insights=True, # Observations
patterns=True, # Reusable knowledge
)
Agent(learn=LearnConfig(
persona=True, # User preferences
insights=True, # Observations
patterns=True, # Reusable knowledge
mode="agentic", # Auto-extract learnings
))
# With database backend
Agent(learn=LearnConfig(
backend="sqlite",
db_url="sqlite:///learn.db",
))
"""
# Learning capabilities
@@ -87,15 +120,19 @@ class LearnConfig:
feedback: bool = False # Outcome signals
improvements: bool = False # Self-improvement proposals
# Learning mode
mode: Union[str, LearnMode] = LearnMode.DISABLED # How to extract learnings
# Scope configuration
scope: Union[str, LearnScope] = LearnScope.PRIVATE
# Storage configuration
store_path: Optional[str] = None # Custom storage path
# Storage backend configuration
backend: Union[str, LearnBackend] = LearnBackend.FILE # Storage backend
db_url: Optional[str] = None # Database connection URL (for non-file backends)
store_path: Optional[str] = None # Custom storage path (for file backend)
# Maintenance settings
auto_consolidate: bool = True # Auto-consolidate learnings
retention_days: Optional[int] = None # Days to retain (None = forever)
# LLM for learning extraction
llm: Optional[str] = None # LLM model for extracting learnings (defaults to agent's LLM)
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
@@ -107,10 +144,12 @@ class LearnConfig:
"decisions": self.decisions,
"feedback": self.feedback,
"improvements": self.improvements,
"mode": self.mode.value if isinstance(self.mode, LearnMode) else self.mode,
"scope": self.scope.value if isinstance(self.scope, LearnScope) else self.scope,
"backend": self.backend.value if isinstance(self.backend, LearnBackend) else self.backend,
"db_url": self.db_url,
"store_path": self.store_path,
"auto_consolidate": self.auto_consolidate,
"retention_days": self.retention_days,
"llm": self.llm,
}
@@ -67,16 +67,22 @@ class LLMGuardrail:
self.logger.warning(f"Unknown LLM type: {type(llm)}, treating as-is")
return llm
def __call__(self, task_output: TaskOutput) -> Tuple[bool, Union[str, TaskOutput]]:
def __call__(self, task_output) -> Tuple[bool, Union[str, "TaskOutput"]]:
"""Validate the task output using the LLM.
Args:
task_output: The task output to validate
task_output: The task output to validate (TaskOutput or plain str)
Returns:
Tuple of (success, result) where result is the output or error message
"""
try:
# Accept plain str for convenience (e.g. chat-input screening)
if isinstance(task_output, str):
raw_text = task_output
else:
raw_text = task_output.raw
if not self.llm:
self.logger.warning("No LLM provided for guardrail validation")
return True, task_output
@@ -88,7 +94,7 @@ You are a quality assurance validator. Your task is to evaluate the following ou
Validation Criteria: {self.description}
Output to Validate:
{task_output.raw}
{raw_text}
Please evaluate if this output meets the criteria. Respond with:
1. "PASS" if the output meets all criteria
@@ -20,34 +20,59 @@ class Chunking:
@cached_property
def SUPPORTED_CHUNKERS(self) -> Dict[str, Any]:
"""Lazy load chunker classes."""
chunkers = {}
try:
from chonkie import (
TokenChunker,
SentenceChunker,
SemanticChunker,
SDPMChunker,
LateChunker,
RecursiveChunker,
CodeChunker,
NeuralChunker,
SlumberChunker
)
from chonkie import TokenChunker
chunkers['token'] = TokenChunker
except ImportError:
pass
try:
from chonkie import SentenceChunker
chunkers['sentence'] = SentenceChunker
except ImportError:
pass
try:
from chonkie import SemanticChunker
chunkers['semantic'] = SemanticChunker
except ImportError:
pass
try:
from chonkie import SDPMChunker
chunkers['sdpm'] = SDPMChunker
except ImportError:
pass
try:
from chonkie import LateChunker
chunkers['late'] = LateChunker
except ImportError:
pass
try:
from chonkie import RecursiveChunker
chunkers['recursive'] = RecursiveChunker
except ImportError:
pass
try:
from chonkie import CodeChunker
chunkers['code'] = CodeChunker
except ImportError:
pass
try:
from chonkie import NeuralChunker
chunkers['neural'] = NeuralChunker
except ImportError:
pass
try:
from chonkie import SlumberChunker
chunkers['slumber'] = SlumberChunker
except ImportError:
pass
if not chunkers:
raise ImportError(
"chonkie package not found. Please install it using: pip install 'praisonaiagents[knowledge]'"
)
return {
'token': TokenChunker,
'sentence': SentenceChunker,
'semantic': SemanticChunker,
'sdpm': SDPMChunker,
'late': LateChunker,
'recursive': RecursiveChunker,
'code': CodeChunker,
'neural': NeuralChunker,
'slumber': SlumberChunker
}
return chunkers
def __init__(
self,
@@ -116,7 +141,18 @@ class Chunking:
if self._chunker is None:
chunker_cls = self.SUPPORTED_CHUNKERS[self.chunker_type]
common_params = self._get_chunker_params()
self._chunker = chunker_cls(**common_params)
try:
self._chunker = chunker_cls(**common_params)
except TypeError:
# Chonkie API may have changed — retry with just chunk_size
fallback_params = {'chunk_size': self.chunk_size}
if 'embedding_model' in common_params:
fallback_params['embedding_model'] = common_params['embedding_model']
try:
self._chunker = chunker_cls(**fallback_params)
except TypeError:
# Last resort — no params
self._chunker = chunker_cls()
return self._chunker
@@ -1,7 +1,5 @@
import os
import logging
import uuid
import time
from datetime import datetime
from .chunking import Chunking
from functools import cached_property
@@ -414,8 +412,9 @@ class Knowledge:
@cached_property
def config(self):
# Generate unique collection name for each instance (only if not provided in config)
default_collection = f"test_{int(time.time())}_{str(uuid.uuid4())[:8]}"
# Use deterministic collection name for persistence across restarts
# Only generate unique name if explicitly requested via config
default_collection = "praisonai_knowledge"
from ..paths import get_project_data_dir
persist_dir = str(get_project_data_dir())
@@ -245,7 +245,7 @@ class AutoMemory:
def __init__(
self,
memory: "FileMemory",
memory: Any,
enabled: bool = True,
extractor: Optional[AutoMemoryExtractor] = None,
min_importance: float = 0.6,
@@ -1066,6 +1066,55 @@ class FileMemory:
"storage_path": str(self.user_path)
}
# -------------------------------------------------------------------------
# MemoryProtocol Compliance Aliases
# -------------------------------------------------------------------------
def store_short_term(
self,
text: str,
metadata: Optional[Dict[str, Any]] = None,
**kwargs
) -> str:
"""Protocol-compliant alias for add_short_term."""
return self.add_short_term(text, metadata)
def store_long_term(
self,
text: str,
metadata: Optional[Dict[str, Any]] = None,
**kwargs
) -> str:
"""Protocol-compliant alias for add_long_term."""
return self.add_long_term(text, metadata)
def search_short_term(
self,
query: str,
limit: int = 5,
**kwargs
) -> List[Dict[str, Any]]:
"""Protocol-compliant alias for search (short_term only)."""
return self.search(query, memory_types=["short_term"], limit=limit)
def search_long_term(
self,
query: str,
limit: int = 5,
**kwargs
) -> List[Dict[str, Any]]:
"""Protocol-compliant alias for search (long_term only)."""
return self.search(query, memory_types=["long_term"], limit=limit)
def get_all_memories(self, **kwargs) -> List[Dict[str, Any]]:
"""Get all memories from short-term and long-term stores."""
all_memories = []
for item in self._short_term:
all_memories.append({"type": "short_term", **item.to_dict()})
for item in self._long_term:
all_memories.append({"type": "long_term", **item.to_dict()})
return all_memories
def export(self) -> Dict[str, Any]:
"""Export all memory data."""
return {
@@ -5,23 +5,30 @@ Provides learning capabilities for agents to capture patterns, preferences,
and insights from interactions to improve future responses.
Usage:
from praisonaiagents import Agent, MemoryConfig, LearnConfig
from praisonaiagents import Agent, LearnConfig
# Simple enable
agent = Agent(memory=MemoryConfig(learn=True))
# Simple enable (top-level param)
agent = Agent(learn=True)
# With specific capabilities
agent = Agent(memory=MemoryConfig(
learn=LearnConfig(
persona=True,
insights=True,
patterns=True,
)
# With specific capabilities and auto-learning
agent = Agent(learn=LearnConfig(
persona=True,
insights=True,
patterns=True,
mode="agentic", # Auto-extract learnings
))
# With database backend
agent = Agent(learn=LearnConfig(
backend="sqlite",
db_url="sqlite:///learn.db",
))
"""
__all__ = [
# Manager
"LearnManager",
# Stores
"PersonaStore",
"InsightStore",
"ThreadStore",
@@ -29,14 +36,22 @@ __all__ = [
"DecisionStore",
"FeedbackStore",
"ImprovementStore",
# Protocols
"LearnMode",
"LearnProtocol",
"AsyncLearnProtocol",
"LearnManagerProtocol",
]
def __getattr__(name: str):
"""Lazy loading for learn module components."""
# Manager
if name == "LearnManager":
from .manager import LearnManager
return LearnManager
# Stores
if name == "PersonaStore":
from .stores import PersonaStore
return PersonaStore
@@ -58,4 +73,19 @@ def __getattr__(name: str):
if name == "ImprovementStore":
from .stores import ImprovementStore
return ImprovementStore
# Protocols
if name == "LearnMode":
from .protocols import LearnMode
return LearnMode
if name == "LearnProtocol":
from .protocols import LearnProtocol
return LearnProtocol
if name == "AsyncLearnProtocol":
from .protocols import AsyncLearnProtocol
return AsyncLearnProtocol
if name == "LearnManagerProtocol":
from .protocols import LearnManagerProtocol
return LearnManagerProtocol
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -6,8 +6,9 @@ capturing and retrieving learned information.
"""
from typing import Any, Dict, List, Optional
import logging
from ...config.feature_configs import LearnConfig, LearnScope
from ...config.feature_configs import LearnConfig, LearnScope, LearnBackend
from .stores import (
PersonaStore,
InsightStore,
@@ -43,65 +44,163 @@ class LearnManager:
config: Optional[LearnConfig] = None,
user_id: Optional[str] = None,
store_path: Optional[str] = None,
custom_stores: Optional[Dict[str, Any]] = None,
):
"""
Initialize LearnManager.
Args:
config: LearnConfig for enabling/disabling learning capabilities
user_id: User identifier for scoping learnings
store_path: Custom path for storing learning data
custom_stores: Optional dict of custom store implementations.
Keys are store names (e.g., "domain", "skills").
Values must implement LearnProtocol (add, search, list_all, etc.)
Example:
```python
class MyDomainStore:
def add(self, content, metadata=None): ...
def search(self, query, limit=10): ...
def list_all(self, limit=100): ...
def get(self, entry_id): ...
def update(self, entry_id, content, metadata=None): ...
def delete(self, entry_id): ...
def clear(self): ...
manager = LearnManager(
config=LearnConfig(persona=True),
custom_stores={"domain": MyDomainStore()}
)
```
"""
self.config = config or LearnConfig()
self.user_id = user_id or "default"
self.store_path = store_path or self.config.store_path
# Resolve backend from config
self._backend = self._create_backend()
if self._backend is not None:
logging.info(
f"LearnManager using {self.config.backend} backend"
)
scope = self.config.scope
if isinstance(scope, LearnScope):
scope = scope.value
self._stores: Dict[str, Any] = {}
self._custom_stores = custom_stores or {}
if self.config.persona:
# Built-in stores are added first, custom stores can override them
if self.config.persona and "persona" not in self._custom_stores:
self._stores["persona"] = PersonaStore(
store_path=self._get_store_path("persona"),
user_id=self.user_id,
scope=scope,
backend=self._backend,
)
if self.config.insights:
if self.config.insights and "insights" not in self._custom_stores:
self._stores["insights"] = InsightStore(
store_path=self._get_store_path("insights"),
user_id=self.user_id,
scope=scope,
backend=self._backend,
)
if self.config.thread:
if self.config.thread and "threads" not in self._custom_stores:
self._stores["threads"] = ThreadStore(
store_path=self._get_store_path("threads"),
user_id=self.user_id,
scope=scope,
backend=self._backend,
)
if self.config.patterns:
if self.config.patterns and "patterns" not in self._custom_stores:
self._stores["patterns"] = PatternStore(
store_path=self._get_store_path("patterns"),
user_id=self.user_id,
scope=scope,
backend=self._backend,
)
if self.config.decisions:
if self.config.decisions and "decisions" not in self._custom_stores:
self._stores["decisions"] = DecisionStore(
store_path=self._get_store_path("decisions"),
user_id=self.user_id,
scope=scope,
backend=self._backend,
)
if self.config.feedback:
if self.config.feedback and "feedback" not in self._custom_stores:
self._stores["feedback"] = FeedbackStore(
store_path=self._get_store_path("feedback"),
user_id=self.user_id,
scope=scope,
backend=self._backend,
)
if self.config.improvements:
if self.config.improvements and "improvements" not in self._custom_stores:
self._stores["improvements"] = ImprovementStore(
store_path=self._get_store_path("improvements"),
user_id=self.user_id,
scope=scope,
backend=self._backend,
)
# Add custom stores last (they override built-in stores)
if self._custom_stores:
for name, store in self._custom_stores.items():
self._stores[name] = store
# PROPOSE mode: pending learnings queue
self._pending: List[LearnEntry] = []
def _create_backend(self) -> Any:
"""Create a storage backend from config.
Returns backend instance or None for FILE (default) backend.
Falls back to None (FILE) on error.
"""
backend = self.config.backend
backend_str = backend.value if isinstance(backend, LearnBackend) else str(backend)
if backend_str == "file":
return None # FILE uses BaseJSONStore default
if backend_str == "sqlite":
try:
from ...storage.backends import SQLiteBackend
db_path = self.config.db_url or f"{self.store_path or 'learn'}/learn.db"
# Strip sqlite:/// prefix if present
if db_path.startswith("sqlite:///"):
db_path = db_path[len("sqlite:///"):]
return SQLiteBackend(db_path=db_path, table_name="learn_store")
except Exception as e:
logging.warning(f"Failed to create SQLite backend: {e}. Falling back to FILE.")
return None
if backend_str == "redis":
try:
from ...storage.backends import RedisBackend
url = self.config.db_url or "redis://localhost:6379"
return RedisBackend(url=url, prefix="praison:learn:")
except Exception as e:
logging.warning(f"Failed to create Redis backend: {e}. Falling back to FILE.")
return None
if backend_str == "mongodb":
try:
from ...storage.backends import MongoDBBackend
url = self.config.db_url or "mongodb://localhost:27017/"
return MongoDBBackend(url=url, collection="praison_learn")
except Exception as e:
logging.warning(f"Failed to create MongoDB backend: {e}. Falling back to FILE.")
return None
logging.warning(f"Unknown learn backend '{backend_str}'. Falling back to FILE.")
return None
def _get_store_path(self, store_name: str) -> Optional[str]:
"""Get path for a specific store."""
@@ -216,6 +315,88 @@ class LearnManager:
for store_name, store in self._stores.items()
}
@property
def pending_learnings(self) -> List[LearnEntry]:
"""Get list of pending learnings awaiting approval (PROPOSE mode)."""
return list(self._pending)
def add_pending(
self,
content: str,
category: str = "persona",
metadata: Optional[Dict[str, Any]] = None,
) -> LearnEntry:
"""Add a learning to the pending queue for user approval.
Args:
content: The learning content
category: Store category (persona, insights, patterns, etc.)
metadata: Optional metadata
Returns:
The pending LearnEntry
"""
from datetime import datetime
entry_id = f"pending_{category}_{len(self._pending)}_{datetime.utcnow().timestamp()}"
entry = LearnEntry(
id=entry_id,
content=content,
metadata={**(metadata or {}), "category": category, "status": "pending"},
)
self._pending.append(entry)
return entry
def approve_learning(self, entry_id: str) -> bool:
"""Approve a pending learning — move it to the appropriate store.
Args:
entry_id: ID of the pending learning to approve
Returns:
True if approved and stored, False if not found
"""
for i, entry in enumerate(self._pending):
if entry.id == entry_id:
category = entry.metadata.get("category", "persona")
# Route to appropriate store
if category in self._stores:
self._stores[category].add(entry.content, entry.metadata)
elif "persona" in self._stores:
self._stores["persona"].add(entry.content, entry.metadata)
self._pending.pop(i)
return True
return False
def reject_learning(self, entry_id: str) -> bool:
"""Reject a pending learning — discard it.
Args:
entry_id: ID of the pending learning to reject
Returns:
True if found and rejected, False if not found
"""
for i, entry in enumerate(self._pending):
if entry.id == entry_id:
self._pending.pop(i)
return True
return False
def approve_all_learnings(self) -> int:
"""Approve all pending learnings at once.
Returns:
Number of learnings approved
"""
count = 0
while self._pending:
entry = self._pending[0]
if self.approve_learning(entry.id):
count += 1
else:
break # Safety
return count
def to_system_prompt_context(self) -> str:
"""
Generate context suitable for injection into system prompt.
@@ -275,3 +456,211 @@ class LearnManager:
# Note: threads are session-specific and not included in system prompt
return "\n".join(parts) if parts else ""
def process_conversation(
self,
messages: List[Dict[str, Any]],
llm: Optional[str] = None,
extract_only: bool = False,
) -> Dict[str, Any]:
"""
Process a conversation to extract learnings automatically.
Uses LLM to analyze the conversation and extract:
- User preferences (persona)
- Insights and observations
- Patterns in user behavior
Args:
messages: Conversation messages (list of {"role": ..., "content": ...})
llm: Optional LLM model to use for extraction (defaults to gpt-4o-mini)
Returns:
Dictionary with extracted learnings:
{
"persona": [...],
"insights": [...],
"patterns": [...],
"stored": {"persona": N, "insights": N, "patterns": N}
}
"""
if not messages:
return {"persona": [], "insights": [], "patterns": [], "stored": {}}
# Build conversation text
conversation_text = "\n".join([
f"{msg.get('role', 'unknown')}: {msg.get('content', '')}"
for msg in messages
])
# Use LLM to extract learnings
extraction_prompt = f"""Analyze this conversation and extract learnings.
CONVERSATION:
{conversation_text}
Extract the following (if present):
1. USER PREFERENCES: Things the user likes, dislikes, prefers, or their style
2. INSIGHTS: Observations about the user's domain, work, or context
3. PATTERNS: Recurring behaviors or request patterns
Return JSON:
{{
"persona": ["preference 1", "preference 2"],
"insights": ["insight 1", "insight 2"],
"patterns": ["pattern 1", "pattern 2"]
}}
Only include items that are clearly evident from the conversation.
Return empty arrays if nothing is found for a category."""
try:
# Lazy import to avoid circular dependency
from ...llm import LLM
import json
model = llm or "gpt-4o-mini"
llm_instance = LLM(model=model)
response = llm_instance.get_response(
prompt=extraction_prompt,
output_json=True,
)
# Parse response - handle both string and dict responses
if isinstance(response, str):
extracted = json.loads(response)
elif isinstance(response, dict):
extracted = response
else:
extracted = {"persona": [], "insights": [], "patterns": []}
# Store extracted learnings (unless extract_only)
stored = {"persona": 0, "insights": 0, "patterns": 0}
if not extract_only:
for preference in extracted.get("persona", []):
if preference and "persona" in self._stores:
self.capture_persona(preference)
stored["persona"] += 1
for insight in extracted.get("insights", []):
if insight and "insights" in self._stores:
self.capture_insight(insight, source="auto_extraction")
stored["insights"] += 1
for pattern in extracted.get("patterns", []):
if pattern and "patterns" in self._stores:
self.capture_pattern(pattern, pattern_type="auto_extracted")
stored["patterns"] += 1
return {
"persona": extracted.get("persona", []),
"insights": extracted.get("insights", []),
"patterns": extracted.get("patterns", []),
"stored": stored,
}
except Exception as e:
import logging
logging.warning(f"Failed to extract learnings: {e}")
return {"persona": [], "insights": [], "patterns": [], "stored": {}, "error": str(e)}
async def aprocess_conversation(
self,
messages: List[Dict[str, Any]],
llm: Optional[str] = None,
) -> Dict[str, Any]:
"""
Async version of process_conversation.
Args:
messages: Conversation messages
llm: Optional LLM model
Returns:
Dictionary with extracted learnings
"""
if not messages:
return {"persona": [], "insights": [], "patterns": [], "stored": {}}
# Build conversation text
conversation_text = "\n".join([
f"{msg.get('role', 'unknown')}: {msg.get('content', '')}"
for msg in messages
])
extraction_prompt = f"""Analyze this conversation and extract learnings.
CONVERSATION:
{conversation_text}
Extract the following (if present):
1. USER PREFERENCES: Things the user likes, dislikes, prefers, or their style
2. INSIGHTS: Observations about the user's domain, work, or context
3. PATTERNS: Recurring behaviors or request patterns
Return JSON:
{{
"persona": ["preference 1", "preference 2"],
"insights": ["insight 1", "insight 2"],
"patterns": ["pattern 1", "pattern 2"]
}}
Only include items that are clearly evident from the conversation.
Return empty arrays if nothing is found for a category."""
try:
# Lazy import to avoid circular dependency
from ...llm import LLM
import json
import asyncio
model = llm or "gpt-4o-mini"
llm_instance = LLM(model=model)
# Run sync LLM call in executor for async compatibility
loop = asyncio.get_event_loop()
response = await loop.run_in_executor(
None,
lambda: llm_instance.get_response(
prompt=extraction_prompt,
output_json=True,
)
)
# Parse response - handle both string and dict responses
if isinstance(response, str):
extracted = json.loads(response)
elif isinstance(response, dict):
extracted = response
else:
extracted = {"persona": [], "insights": [], "patterns": []}
stored = {"persona": 0, "insights": 0, "patterns": 0}
for preference in extracted.get("persona", []):
if preference and "persona" in self._stores:
self.capture_persona(preference)
stored["persona"] += 1
for insight in extracted.get("insights", []):
if insight and "insights" in self._stores:
self.capture_insight(insight, source="auto_extraction")
stored["insights"] += 1
for pattern in extracted.get("patterns", []):
if pattern and "patterns" in self._stores:
self.capture_pattern(pattern, pattern_type="auto_extracted")
stored["patterns"] += 1
return {
"persona": extracted.get("persona", []),
"insights": extracted.get("insights", []),
"patterns": extracted.get("patterns", []),
"stored": stored,
}
except Exception as e:
import logging
logging.warning(f"Failed to extract learnings: {e}")
return {"persona": [], "insights": [], "patterns": [], "stored": {}, "error": str(e)}
@@ -0,0 +1,266 @@
"""
Learn Protocol Definitions.
Provides Protocol interfaces for learning implementations.
Enables custom learning backends and extensibility.
"""
from typing import Any, Dict, List, Optional, Protocol, runtime_checkable
# Re-export LearnMode from feature_configs (canonical location)
# This avoids duplication while maintaining backward compatibility
from ...config.feature_configs import LearnMode
@runtime_checkable
class LearnProtocol(Protocol):
"""
Protocol for learning store implementations.
Enables custom backends (file, SQLite, Redis, MongoDB, etc.)
while maintaining a consistent interface.
Example:
```python
class RedisLearnStore:
def add(self, content: str, metadata=None) -> Dict[str, Any]:
# Store in Redis
return {"id": "...", "content": content}
def search(self, query: str, limit: int = 10) -> List[Dict[str, Any]]:
# Search Redis
return []
def list_all(self, limit: int = 100) -> List[Dict[str, Any]]:
return []
def delete(self, entry_id: str) -> bool:
return True
def clear(self) -> int:
return 0
```
"""
def add(
self,
content: str,
metadata: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""
Add a new entry to the store.
Args:
content: The content to store
metadata: Optional metadata dictionary
Returns:
Dictionary with entry details including 'id'
"""
...
def search(
self,
query: str,
limit: int = 10,
) -> List[Dict[str, Any]]:
"""
Search entries by query.
Args:
query: Search query string
limit: Maximum results to return
Returns:
List of matching entries
"""
...
def list_all(
self,
limit: int = 100,
) -> List[Dict[str, Any]]:
"""
List all entries.
Args:
limit: Maximum entries to return
Returns:
List of all entries (most recent first)
"""
...
def get(
self,
entry_id: str,
) -> Optional[Dict[str, Any]]:
"""
Get entry by ID.
Args:
entry_id: Entry identifier
Returns:
Entry dict or None if not found
"""
...
def update(
self,
entry_id: str,
content: str,
metadata: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]:
"""
Update an existing entry.
Args:
entry_id: Entry identifier
content: New content
metadata: Optional metadata to merge
Returns:
Updated entry or None if not found
"""
...
def delete(
self,
entry_id: str,
) -> bool:
"""
Delete an entry.
Args:
entry_id: Entry identifier
Returns:
True if deleted, False if not found
"""
...
def clear(self) -> int:
"""
Clear all entries.
Returns:
Number of entries cleared
"""
...
@runtime_checkable
class AsyncLearnProtocol(Protocol):
"""
Async Protocol for learning store implementations.
For backends that support async operations.
"""
async def aadd(
self,
content: str,
metadata: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Async add entry."""
...
async def asearch(
self,
query: str,
limit: int = 10,
) -> List[Dict[str, Any]]:
"""Async search entries."""
...
async def alist_all(
self,
limit: int = 100,
) -> List[Dict[str, Any]]:
"""Async list all entries."""
...
async def aget(
self,
entry_id: str,
) -> Optional[Dict[str, Any]]:
"""Async get entry by ID."""
...
async def aupdate(
self,
entry_id: str,
content: str,
metadata: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]:
"""Async update entry."""
...
async def adelete(
self,
entry_id: str,
) -> bool:
"""Async delete entry."""
...
async def aclear(self) -> int:
"""Async clear all entries."""
...
@runtime_checkable
class LearnManagerProtocol(Protocol):
"""
Protocol for LearnManager implementations.
Defines the interface for managing learning across multiple stores.
"""
def capture_persona(self, content: str, category: str = "general") -> Optional[Dict[str, Any]]:
"""Capture user preference or profile information."""
...
def capture_insight(self, content: str, source: str = "interaction") -> Optional[Dict[str, Any]]:
"""Capture an observation or learning."""
...
def capture_pattern(self, pattern: str, pattern_type: str = "general") -> Optional[Dict[str, Any]]:
"""Capture a reusable knowledge pattern."""
...
def get_learning_context(self, limit_per_store: int = 5) -> Dict[str, List[Dict[str, Any]]]:
"""Get learning context from all enabled stores."""
...
def search(self, query: str, limit: int = 10) -> Dict[str, List[Dict[str, Any]]]:
"""Search across all enabled stores."""
...
def to_system_prompt_context(self) -> str:
"""Generate context for system prompt injection."""
...
def process_conversation(
self,
messages: List[Dict[str, Any]],
llm: Optional[str] = None,
) -> Dict[str, Any]:
"""
Process a conversation to extract learnings automatically.
Args:
messages: Conversation messages
llm: Optional LLM model to use for extraction
Returns:
Dictionary with extracted learnings
"""
...
__all__ = [
'LearnMode',
'LearnProtocol',
'AsyncLearnProtocol',
'LearnManagerProtocol',
]
@@ -73,12 +73,23 @@ class BaseStore(ABC):
self._backend = backend
self.store_path = store_path or self._default_path()
self._entries: Dict[str, LearnEntry] = {}
self._was_updated: bool = False
# DRY: Use BaseJSONStore for thread-safe storage
self._store = BaseJSONStore(
storage_path=self.store_path,
backend=backend,
)
# Fall back to FILE backend if connection fails (e.g., Redis unavailable)
try:
self._store = BaseJSONStore(
storage_path=self.store_path,
backend=backend,
)
except Exception as e:
import logging
logging.warning(f"Failed to connect to backend: {e}. Falling back to FILE.")
self._backend = None
self._store = BaseJSONStore(
storage_path=self.store_path,
backend=None,
)
self._load()
@property
@@ -87,6 +98,15 @@ class BaseStore(ABC):
"""Name of the store (used for file naming)."""
pass
@property
def was_updated(self) -> bool:
"""Whether this store has been modified since last reset."""
return self._was_updated
def reset_updated(self) -> None:
"""Reset the was_updated flag."""
self._was_updated = False
def _default_path(self) -> str:
"""Default storage path (uses centralized paths - DRY)."""
from ...paths import get_learn_dir, ensure_dir
@@ -106,7 +126,24 @@ class BaseStore(ABC):
self._store.save({k: v.to_dict() for k, v in self._entries.items()})
def add(self, content: str, metadata: Optional[Dict[str, Any]] = None) -> LearnEntry:
"""Add a new entry."""
"""Add a new entry with deduplication.
Checks for exact content matches to prevent duplicate learnings.
If a duplicate is found, returns the existing entry instead of creating a new one.
"""
# Deduplication: Check for exact content match
content_normalized = content.strip().lower()
for existing_entry in self._entries.values():
if existing_entry.content.strip().lower() == content_normalized:
# Update metadata if provided, but don't create duplicate
if metadata:
existing_entry.metadata.update(metadata)
existing_entry.updated_at = datetime.utcnow().isoformat()
self._save()
self._was_updated = True
return existing_entry
# No duplicate found, create new entry
entry_id = f"{self.store_name}_{len(self._entries)}_{datetime.utcnow().timestamp()}"
entry = LearnEntry(
id=entry_id,
@@ -115,6 +152,7 @@ class BaseStore(ABC):
)
self._entries[entry_id] = entry
self._save()
self._was_updated = True
return entry
def get(self, entry_id: str) -> Optional[LearnEntry]:
@@ -146,6 +184,7 @@ class BaseStore(ABC):
entry.metadata.update(metadata)
entry.updated_at = datetime.utcnow().isoformat()
self._save()
self._was_updated = True
return entry
def delete(self, entry_id: str) -> bool:
@@ -153,6 +192,7 @@ class BaseStore(ABC):
if entry_id in self._entries:
del self._entries[entry_id]
self._save()
self._was_updated = True
return True
return False
@@ -161,6 +201,7 @@ class BaseStore(ABC):
count = len(self._entries)
self._entries = {}
self._save()
self._was_updated = True
return count
@@ -40,6 +40,9 @@ class MemoryProtocol(Protocol):
def search_long_term(self, query: str, limit: int = 5, **kwargs):
return self._data[:limit]
def get_all_memories(self, **kwargs):
return self._data
# Use in agent
memory: MemoryProtocol = InMemoryStore()
@@ -122,6 +125,15 @@ class MemoryProtocol(Protocol):
"""
...
def get_all_memories(self, **kwargs) -> List[Dict[str, Any]]:
"""
Get all memories from all backends.
Returns:
List of all memory entries (short-term and long-term combined)
"""
...
@runtime_checkable
class ResettableMemoryProtocol(MemoryProtocol, Protocol):
@@ -20,7 +20,7 @@ Default storage: ~/.praisonai/schedules/jobs.json
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .models import Schedule, ScheduleJob
from .models import Schedule, ScheduleJob, DeliveryTarget
from .store import FileScheduleStore
from .parser import parse_schedule
from .runner import ScheduleRunner
@@ -33,10 +33,11 @@ def __getattr__(name: str):
if name in _module_cache:
return _module_cache[name]
if name in ("Schedule", "ScheduleJob"):
from .models import Schedule, ScheduleJob
if name in ("Schedule", "ScheduleJob", "DeliveryTarget"):
from .models import Schedule, ScheduleJob, DeliveryTarget
_module_cache["Schedule"] = Schedule
_module_cache["ScheduleJob"] = ScheduleJob
_module_cache["DeliveryTarget"] = DeliveryTarget
return _module_cache[name]
if name == "FileScheduleStore":
@@ -60,6 +61,7 @@ def __getattr__(name: str):
__all__ = [
"Schedule",
"ScheduleJob",
"DeliveryTarget",
"FileScheduleStore",
"parse_schedule",
"ScheduleRunner",
@@ -51,6 +51,45 @@ class Schedule:
)
@dataclass
class DeliveryTarget:
"""Where to deliver the result when a scheduled job fires.
Attributes:
channel: Platform name (``"telegram"``, ``"discord"``,
``"slack"``, ``"whatsapp"``).
channel_id: Platform-specific chat / channel / group ID.
thread_id: Optional thread ID for threaded delivery.
session_id: Optional session ID to preserve conversation
context when the cron fires.
"""
channel: str = ""
channel_id: str = ""
thread_id: Optional[str] = None
session_id: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
d: Dict[str, Any] = {
"channel": self.channel,
"channel_id": self.channel_id,
}
if self.thread_id is not None:
d["thread_id"] = self.thread_id
if self.session_id is not None:
d["session_id"] = self.session_id
return d
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "DeliveryTarget":
return cls(
channel=d.get("channel", ""),
channel_id=d.get("channel_id", ""),
thread_id=d.get("thread_id"),
session_id=d.get("session_id"),
)
@dataclass
class ScheduleJob:
"""A persisted scheduled job.
@@ -67,6 +106,8 @@ class ScheduleJob:
delete_after_run: Auto-remove after first execution (one-shot).
created_at: Epoch timestamp.
last_run_at: Epoch timestamp of most recent execution.
delivery: Optional delivery target for routing results back
to a channel bot (e.g. Telegram chat).
"""
name: str = ""
@@ -79,11 +120,12 @@ class ScheduleJob:
delete_after_run: bool = False
created_at: float = field(default_factory=time.time)
last_run_at: Optional[float] = None
delivery: Optional[DeliveryTarget] = None
# ── serialisation ────────────────────────────────────────────────
def to_dict(self) -> Dict[str, Any]:
return {
d: Dict[str, Any] = {
"id": self.id,
"name": self.name,
"schedule": self.schedule.to_dict(),
@@ -95,10 +137,14 @@ class ScheduleJob:
"created_at": self.created_at,
"last_run_at": self.last_run_at,
}
if self.delivery is not None:
d["delivery"] = self.delivery.to_dict()
return d
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "ScheduleJob":
sched_data = d.get("schedule", {})
delivery_data = d.get("delivery")
return cls(
id=d.get("id", uuid.uuid4().hex[:12]),
name=d.get("name", ""),
@@ -110,4 +156,5 @@ class ScheduleJob:
delete_after_run=d.get("delete_after_run", False),
created_at=d.get("created_at", time.time()),
last_run_at=d.get("last_run_at"),
delivery=DeliveryTarget.from_dict(delivery_data) if isinstance(delivery_data, dict) else None,
)
+36 -13
View File
@@ -329,7 +329,11 @@ class Session:
def _save_agent_chat_histories(self) -> None:
"""
Save all agent chat histories to memory.
Save all agent chat histories.
G-2 FIX: Routes to SessionStore instead of Memory.store_short_term() to
maintain clean separation between conversation history (SessionStore)
and semantic memory (Memory). Falls back to Memory for backward compat.
"""
if self.is_remote:
return
@@ -346,18 +350,37 @@ class Session:
chat_history = agent_data.get("chat_history")
if chat_history is not None:
# Save to memory
history_text = f"Agent chat history for {agent_key}"
self.memory.store_short_term(
text=history_text,
metadata={
"type": "agent_chat_history",
"session_id": self.session_id,
"user_id": self.user_id,
"agent_key": agent_key,
"chat_history": chat_history
}
)
# G-2 FIX: Try SessionStore first for clean separation
session_store = None
try:
from .session import get_default_session_store
session_store = get_default_session_store()
except ImportError:
pass
if session_store is not None:
# Use SessionStore for conversation history
session_id = f"{self.session_id}_{agent_key}"
for msg in chat_history:
if isinstance(msg, dict):
session_store.add_message(
session_id,
role=msg.get("role", "user"),
content=msg.get("content", ""),
)
else:
# Fallback to Memory.store_short_term() for backward compatibility
history_text = f"Agent chat history for {agent_key}"
self.memory.store_short_term(
text=history_text,
metadata={
"type": "agent_chat_history",
"session_id": self.session_id,
"user_id": self.user_id,
"agent_key": agent_key,
"chat_history": chat_history
}
)
def get_state(self, key: str, default: Any = None) -> Any:
"""Get a specific state value"""
@@ -311,7 +311,12 @@ class Session:
}
def _save_agent_chat_histories(self) -> None:
"""Save all agent chat histories to memory."""
"""Save all agent chat histories.
G-2 FIX: Routes to SessionStore instead of Memory.store_short_term() to
maintain clean separation between conversation history (SessionStore)
and semantic memory (Memory). Falls back to Memory for backward compat.
"""
if self.is_remote:
return
@@ -326,17 +331,37 @@ class Session:
chat_history = agent_data.get("chat_history")
if chat_history is not None:
history_text = f"Agent chat history for {agent_key}"
self.memory.store_short_term(
text=history_text,
metadata={
"type": "agent_chat_history",
"session_id": self.session_id,
"user_id": self.user_id,
"agent_key": agent_key,
"chat_history": chat_history
}
)
# G-2 FIX: Try SessionStore first for clean separation
session_store = None
try:
from . import get_default_session_store
session_store = get_default_session_store()
except ImportError:
pass
if session_store is not None:
# Use SessionStore for conversation history
session_id = f"{self.session_id}_{agent_key}"
for msg in chat_history:
if isinstance(msg, dict):
session_store.add_message(
session_id,
role=msg.get("role", "user"),
content=msg.get("content", ""),
)
else:
# Fallback to Memory.store_short_term() for backward compatibility
history_text = f"Agent chat history for {agent_key}"
self.memory.store_short_term(
text=history_text,
metadata={
"type": "agent_chat_history",
"session_id": self.session_id,
"user_id": self.user_id,
"agent_key": agent_key,
"chat_history": chat_history
}
)
def get_state(self, key: str, default: Any = None) -> Any:
"""Get a specific state value"""
@@ -29,6 +29,7 @@ __all__ = [
"FileBackend",
"SQLiteBackend",
"RedisBackend",
"MongoDBBackend",
"get_backend",
]
@@ -47,6 +48,7 @@ _LAZY_IMPORTS = {
"FileBackend": ("backends", "FileBackend"),
"SQLiteBackend": ("backends", "SQLiteBackend"),
"RedisBackend": ("backends", "RedisBackend"),
"MongoDBBackend": ("backends", "MongoDBBackend"),
"get_backend": ("backends", "get_backend"),
}
@@ -13,6 +13,7 @@ DRY: All backends implement StorageBackendProtocol for consistent interface.
import json
import os
import time
import threading
import tempfile
import logging
@@ -456,6 +457,114 @@ class RedisBackend:
self._client = None
class MongoDBBackend:
"""
MongoDB-based storage backend.
Uses MongoDB for document-oriented data storage.
Requires the `pymongo` package (optional dependency).
Example:
```python
backend = MongoDBBackend(url="mongodb://localhost:27017/")
backend.save("session_123", {"messages": []})
data = backend.load("session_123")
```
"""
def __init__(
self,
url: str = "mongodb://localhost:27017/",
database: str = "praisonai",
collection: str = "praison_storage",
max_pool_size: int = 50,
):
self.url = url
self.database = database
self.collection_name = collection
self.max_pool_size = max_pool_size
self._client = None
self._collection = None
def _get_collection(self):
"""Lazy initialize MongoDB client and collection."""
if self._collection is None:
try:
import pymongo
except ImportError:
raise ImportError(
"MongoDB backend requires the 'pymongo' package. "
"Install with: pip install pymongo"
)
self._client = pymongo.MongoClient(
self.url,
maxPoolSize=self.max_pool_size,
retryWrites=True,
retryReads=True,
serverSelectionTimeoutMS=5000, # 5 second timeout for connection
)
db = self._client[self.database]
self._collection = db[self.collection_name]
return self._collection
def save(self, key: str, data: Dict[str, Any]) -> None:
"""Save data with the given key (upsert)."""
collection = self._get_collection()
json_data = json.dumps(data, default=str, ensure_ascii=False)
collection.replace_one(
{"_id": key},
{"_id": key, "data": json_data, "updated_at": time.time()},
upsert=True,
)
def load(self, key: str) -> Optional[Dict[str, Any]]:
"""Load data by key."""
collection = self._get_collection()
doc = collection.find_one({"_id": key})
if doc and "data" in doc:
try:
return json.loads(doc["data"])
except json.JSONDecodeError:
return None
return None
def delete(self, key: str) -> bool:
"""Delete data by key."""
collection = self._get_collection()
result = collection.delete_one({"_id": key})
return result.deleted_count > 0
def list_keys(self, prefix: str = "") -> List[str]:
"""List all keys, optionally filtered by prefix."""
collection = self._get_collection()
if prefix:
cursor = collection.find(
{"_id": {"$regex": f"^{prefix}"}},
{"_id": 1}
)
else:
cursor = collection.find({}, {"_id": 1})
return sorted([doc["_id"] for doc in cursor])
def exists(self, key: str) -> bool:
"""Check if a key exists."""
collection = self._get_collection()
return collection.count_documents({"_id": key}, limit=1) > 0
def clear(self) -> int:
"""Clear all data. Returns number of items deleted."""
collection = self._get_collection()
result = collection.delete_many({})
return result.deleted_count
def close(self) -> None:
"""Close the MongoDB connection."""
if self._client:
self._client.close()
self._client = None
self._collection = None
def get_backend(
backend_type: str = "file",
**kwargs
@@ -488,6 +597,8 @@ def get_backend(
return SQLiteBackend(**kwargs)
elif backend_type == "redis":
return RedisBackend(**kwargs)
elif backend_type == "mongodb":
return MongoDBBackend(**kwargs)
else:
raise ValueError(f"Unknown backend type: {backend_type}")
@@ -496,5 +607,6 @@ __all__ = [
'FileBackend',
'SQLiteBackend',
'RedisBackend',
'MongoDBBackend',
'get_backend',
]
@@ -41,6 +41,10 @@ def schedule_add(
name: str,
schedule: str,
message: str = "",
channel: str = "",
channel_id: str = "",
agent_id: str = "",
session_id: str = "",
) -> str:
"""Add a new scheduled job.
@@ -53,16 +57,39 @@ def schedule_add(
- "at:2026-03-01T09:00:00" (one-shot ISO timestamp)
- "in 20 minutes" (relative one-shot)
message: The prompt or reminder text to deliver when triggered.
channel: Delivery platform ("telegram", "discord", "slack",
"whatsapp"). If set with channel_id, the result will
be sent back to that chat when the job fires.
channel_id: Target chat/channel/group ID on the platform.
agent_id: Which agent should execute this job. If empty, the
executor uses its default agent.
session_id: Optional session ID to preserve conversation context.
If set, the agent will have access to prior chat history.
Returns:
Confirmation string with the job id.
"""
try:
from ..scheduler.parser import parse_schedule
from ..scheduler.models import ScheduleJob
from ..scheduler.models import ScheduleJob, DeliveryTarget
sched = parse_schedule(schedule)
job = ScheduleJob(name=name, schedule=sched, message=message)
delivery = None
if channel and channel_id:
delivery = DeliveryTarget(
channel=channel,
channel_id=channel_id,
session_id=session_id or None,
)
job = ScheduleJob(
name=name,
schedule=sched,
message=message,
agent_id=agent_id or None,
delivery=delivery,
)
store = _get_store()
@@ -72,7 +99,9 @@ def schedule_add(
return f"A schedule named '{name}' already exists (id: {existing.id}). Remove it first or choose a different name."
store.add(job)
return f"Schedule '{name}' added (id: {job.id}, {schedule})."
delivery_note = f" → deliver to {channel}:{channel_id}" if delivery else ""
agent_note = f" agent={agent_id}" if agent_id else ""
return f"Schedule '{name}' added (id: {job.id}, {schedule}{agent_note}{delivery_note})."
except ValueError as e:
return f"Error adding schedule: {e}"
except Exception as e:
@@ -105,10 +134,15 @@ def schedule_list() -> str:
sched_str = f"at: {sched.at}"
else:
sched_str = str(sched.kind)
lines.append(
f" - {j.name} (id: {j.id}) [{status}] — {sched_str}"
+ (f" — \"{j.message}\"" if j.message else "")
)
parts = [f" - {j.name} (id: {j.id}) [{status}] — {sched_str}"]
if j.agent_id:
parts.append(f" agent={j.agent_id}")
if j.delivery:
parts.append(f" → {j.delivery.channel}:{j.delivery.channel_id}")
if j.message:
parts.append(f' — "{j.message}"')
lines.append("".join(parts))
return "\n".join(lines)
except Exception as e:
logger.error("schedule_list failed: %s", e, exc_info=True)
@@ -569,6 +569,7 @@ class AgentFlow:
reflection: Optional[Any] = None # Union[bool, ReflectionConfig] - self-reflection
execution: Optional[Any] = None # Union[str, ExecutionConfig] - execution control
caching: Optional[Any] = None # Union[bool, CachingConfig] - caching
learn: Optional[Any] = None # Union[bool, LearnConfig] - continuous learning
# ============================================================
# ROBUSTNESS PARAMS (debugging & audit trail)
@@ -1901,14 +1902,46 @@ Create a brief execution plan (2-3 sentences) describing how to best accomplish
# Only pass output_config if it has values
output_json_model = output_config.get('json_model') if output_config else None
output_pydantic_model = output_config.get('pydantic_model') if output_config else None
return Task(
# Extract remaining YAML step-level attributes (previously dead stores)
yaml_guardrail = getattr(step, '_yaml_guardrail', None)
yaml_max_retries = getattr(step, '_yaml_max_retries', None)
yaml_expected_output = getattr(step, '_yaml_expected_output', None)
yaml_output_file = getattr(step, '_yaml_output_file', None)
yaml_skip_on_failure = getattr(step, '_yaml_skip_on_failure', None)
yaml_retry_delay = getattr(step, '_yaml_retry_delay', None)
yaml_create_directory = getattr(step, '_yaml_create_directory', None)
yaml_callback = getattr(step, '_yaml_callback', None)
yaml_async_execution = getattr(step, '_yaml_async_execution', None)
# Build Task kwargs — only include YAML overrides when present
task_kwargs = dict(
name=yaml_step_name or getattr(step, 'name', f'agent_{index+1}'),
agent=step,
tools=agent_tools,
action=action,
output_json=output_json_model,
output_pydantic=output_pydantic_model
output_pydantic=output_pydantic_model,
)
if yaml_guardrail is not None:
task_kwargs['guardrails'] = yaml_guardrail
if yaml_max_retries is not None:
task_kwargs['max_retries'] = yaml_max_retries
if yaml_expected_output is not None:
task_kwargs['expected_output'] = yaml_expected_output
if yaml_output_file is not None:
task_kwargs['output_file'] = yaml_output_file
if yaml_skip_on_failure is not None:
task_kwargs['skip_on_failure'] = yaml_skip_on_failure
if yaml_retry_delay is not None:
task_kwargs['retry_delay'] = yaml_retry_delay
if yaml_create_directory is not None:
task_kwargs['create_directory'] = yaml_create_directory
if yaml_callback is not None:
task_kwargs['callback'] = yaml_callback
if yaml_async_execution is not None:
task_kwargs['async_execution'] = yaml_async_execution
return Task(**task_kwargs)
elif callable(step):
return Task(
name=getattr(step, '__name__', f'step_{index+1}'),
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "1.5.30"
version = "1.5.39"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
@@ -434,7 +434,7 @@ callbacks:
class TestYAMLWorkflowParserGuardrails:
"""Tests for guardrails parsing."""
"""Tests for guardrails parsing and wiring to Task objects."""
def test_parse_step_guardrail(self):
"""Test parsing guardrail on a step."""
@@ -459,6 +459,167 @@ steps:
workflow = parser.parse_string(yaml_content)
assert workflow is not None
def test_guardrail_wired_to_task(self):
"""Test that guardrail string from YAML is wired into the Task object."""
from praisonaiagents.workflows import YAMLWorkflowParser
yaml_content = """
name: Guardrail Wiring Test
agents:
writer:
name: Writer
role: Writer
goal: Write
instructions: "Write content"
steps:
- agent: writer
action: "Write professional content"
guardrail: "Ensure the content is professional and error-free"
"""
parser = YAMLWorkflowParser()
workflow = parser.parse_string(yaml_content)
# Normalize the step to a Task (this is what run() does internally)
task = workflow._normalize_single_step(workflow.steps[0], 0)
# The guardrail string must be set on the Task
assert task.guardrail == "Ensure the content is professional and error-free"
# _setup_guardrail should have created an LLMGuardrail instance
assert task._guardrail_fn is not None
def test_max_retries_wired_to_task(self):
"""Test that max_retries from YAML is wired into the Task object."""
from praisonaiagents.workflows import YAMLWorkflowParser
yaml_content = """
name: Retries Wiring Test
agents:
writer:
name: Writer
role: Writer
goal: Write
instructions: "Write content"
steps:
- agent: writer
action: "Write content"
max_retries: 7
"""
parser = YAMLWorkflowParser()
workflow = parser.parse_string(yaml_content)
task = workflow._normalize_single_step(workflow.steps[0], 0)
assert task.max_retries == 7
def test_expected_output_wired_to_task(self):
"""Test that expected_output from YAML is wired into the Task object."""
from praisonaiagents.workflows import YAMLWorkflowParser
yaml_content = """
name: Expected Output Wiring Test
agents:
writer:
name: Writer
role: Writer
goal: Write
instructions: "Write content"
steps:
- agent: writer
action: "Write a blog post"
expected_output: "A well-structured blog post with introduction, body, and conclusion"
"""
parser = YAMLWorkflowParser()
workflow = parser.parse_string(yaml_content)
task = workflow._normalize_single_step(workflow.steps[0], 0)
assert task.expected_output == "A well-structured blog post with introduction, body, and conclusion"
def test_skip_on_failure_wired_to_task(self):
"""Test that skip_on_failure from YAML is wired into the Task object."""
from praisonaiagents.workflows import YAMLWorkflowParser
yaml_content = """
name: Skip Failure Wiring Test
agents:
writer:
name: Writer
role: Writer
goal: Write
instructions: "Write content"
steps:
- agent: writer
action: "Write content"
skip_on_failure: true
"""
parser = YAMLWorkflowParser()
workflow = parser.parse_string(yaml_content)
task = workflow._normalize_single_step(workflow.steps[0], 0)
assert task.skip_on_failure == True
def test_output_file_wired_to_task(self):
"""Test that output_file from YAML is wired into the Task object."""
from praisonaiagents.workflows import YAMLWorkflowParser
yaml_content = """
name: Output File Wiring Test
agents:
writer:
name: Writer
role: Writer
goal: Write
instructions: "Write content"
steps:
- agent: writer
action: "Write content"
output_file: "output/result.txt"
"""
parser = YAMLWorkflowParser()
workflow = parser.parse_string(yaml_content)
task = workflow._normalize_single_step(workflow.steps[0], 0)
assert task.output_file == "output/result.txt"
def test_all_yaml_step_attrs_wired(self):
"""Test that ALL YAML step-level attributes are wired through to Task."""
from praisonaiagents.workflows import YAMLWorkflowParser
yaml_content = """
name: Full Wiring Test
agents:
writer:
name: Writer
role: Writer
goal: Write
instructions: "Write content"
steps:
- agent: writer
action: "Write content"
guardrail: "Check quality"
max_retries: 5
expected_output: "High quality content"
output_file: "out.txt"
skip_on_failure: true
retry_delay: 2.5
create_directory: true
"""
parser = YAMLWorkflowParser()
workflow = parser.parse_string(yaml_content)
task = workflow._normalize_single_step(workflow.steps[0], 0)
assert task.guardrail == "Check quality"
assert task.max_retries == 5
assert task.expected_output == "High quality content"
assert task.output_file == "out.txt"
assert task.skip_on_failure == True
assert task.retry_delay == 2.5
assert task.create_directory == True
class TestYAMLWorkflowParserVariables:
@@ -0,0 +1,194 @@
"""TDD Tests for top-level learn= parameter on Agent and related classes.
Tests that learn= is a first-class citizen, peer to memory=, not nested inside it.
"""
class TestAgentLearnParam:
"""Test Agent(learn=...) top-level parameter."""
def test_agent_accepts_learn_false(self):
"""Agent should accept learn=False (default, no learning)."""
from praisonaiagents import Agent
agent = Agent(
name="test",
instructions="Test agent",
learn=False,
)
assert agent._learn_config is None or agent._learn_config is False
def test_agent_accepts_learn_true(self):
"""Agent should accept learn=True (enable with defaults)."""
from praisonaiagents import Agent
agent = Agent(
name="test",
instructions="Test agent",
learn=True,
)
# Should have learn config enabled
assert agent._learn_config is not None
assert agent._learn_config is not False
def test_agent_accepts_learn_config(self):
"""Agent should accept learn=LearnConfig(...)."""
from praisonaiagents import Agent
from praisonaiagents.config.feature_configs import LearnConfig
config = LearnConfig(persona=True, insights=True, patterns=False)
agent = Agent(
name="test",
instructions="Test agent",
learn=config,
)
# Should store the config
assert agent._learn_config is not None
if hasattr(agent._learn_config, 'persona'):
assert agent._learn_config.persona is True
assert agent._learn_config.insights is True
def test_learn_param_independent_of_memory(self):
"""learn= should work without memory= being set."""
from praisonaiagents import Agent
# learn=True without memory=True should still work
agent = Agent(
name="test",
instructions="Test agent",
learn=True,
memory=None, # No memory explicitly
)
# Learn should be enabled even without explicit memory
assert agent._learn_config is not None
def test_learn_and_memory_both_work(self):
"""learn= and memory= should work together."""
from praisonaiagents import Agent
agent = Agent(
name="test",
instructions="Test agent",
learn=True,
memory=True,
)
# Both should be enabled
assert agent._learn_config is not None
assert agent._memory_instance is not None
class TestLearnConfigExport:
"""Test that LearnConfig is exported from top-level package."""
def test_learn_config_exported(self):
"""LearnConfig should be importable from praisonaiagents."""
from praisonaiagents import LearnConfig
assert LearnConfig is not None
# Should be able to instantiate
config = LearnConfig()
assert config.persona is True # Default
assert config.insights is True # Default
def test_learn_scope_exported(self):
"""LearnScope should be importable from praisonaiagents."""
from praisonaiagents import LearnScope
assert LearnScope is not None
assert LearnScope.PRIVATE.value == "private"
assert LearnScope.SHARED.value == "shared"
class TestAgentTeamLearnParam:
"""Test AgentTeam(learn=...) parameter."""
def test_agent_team_accepts_learn_param(self):
"""AgentTeam should accept learn= parameter."""
from praisonaiagents import Agent, AgentTeam
agent1 = Agent(name="agent1", instructions="Agent 1")
agent2 = Agent(name="agent2", instructions="Agent 2")
team = AgentTeam(
agents=[agent1, agent2],
learn=True,
)
# Should store learn config
assert hasattr(team, '_learn') or hasattr(team, 'learn')
class TestWorkflowLearnParam:
"""Test Workflow/AgentFlow(learn=...) parameter."""
def test_workflow_accepts_learn_param(self):
"""Workflow should accept learn= parameter."""
from praisonaiagents import Agent
from praisonaiagents.workflows import Workflow
agent = Agent(name="test", instructions="Test")
workflow = Workflow(
steps=[agent], # Workflow uses steps=, not agents=
learn=True,
)
# Should store learn config (dataclass field)
assert workflow.learn is True
class TestLearnParamDocstring:
"""Test that learn= param is documented."""
def test_agent_docstring_mentions_learn(self):
"""Agent.__init__ docstring should mention learn parameter."""
from praisonaiagents import Agent
docstring = Agent.__init__.__doc__ or ""
# Should mention learn parameter
assert "learn" in docstring.lower()
class TestBackwardCompatibility:
"""Test backward compatibility with memory=MemoryConfig(learn=...)."""
def test_memory_config_learn_still_works(self):
"""memory=MemoryConfig(learn=True) should still work."""
from praisonaiagents import Agent
from praisonaiagents.config.feature_configs import MemoryConfig
agent = Agent(
name="test",
instructions="Test agent",
memory=MemoryConfig(learn=True),
)
# Should have memory with learn enabled
assert agent._memory_instance is not None
# Learn context should be accessible
if hasattr(agent._memory_instance, 'get_learn_context'):
context = agent._memory_instance.get_learn_context()
assert isinstance(context, str)
def test_top_level_learn_takes_precedence(self):
"""Top-level learn= should take precedence over memory.learn."""
from praisonaiagents import Agent
from praisonaiagents.config.feature_configs import MemoryConfig
# Top-level learn=False should override memory.learn=True
agent = Agent(
name="test",
instructions="Test agent",
memory=MemoryConfig(learn=True),
learn=False, # This should take precedence
)
# Learn should be disabled due to top-level param
assert agent._learn_config is None or agent._learn_config is False
@@ -16,6 +16,17 @@ requires_knowledge = pytest.mark.skipif(
)
def _make_knowledge_or_skip():
"""Create a Knowledge() instance, skipping if OpenAI API key is unavailable."""
from praisonaiagents import Knowledge
try:
return Knowledge()
except Exception as e:
if "api_key" in str(e).lower() or "openai" in str(e).lower():
pytest.skip(f"Skipping: OpenAI API key required for embeddings: {e}")
raise
@requires_knowledge
class TestDirectoryIngestion:
"""Test that directories are properly processed and file contents stored."""
@@ -42,10 +53,14 @@ class TestDirectoryIngestion:
def test_directory_ingestion_stores_text_not_path(self, temp_knowledge_dir):
"""Test that ingesting a directory stores file contents, not the directory path."""
from praisonaiagents import Knowledge
knowledge = _make_knowledge_or_skip()
knowledge = Knowledge()
result = knowledge.add(temp_knowledge_dir, user_id='test_user')
try:
result = knowledge.add(temp_knowledge_dir, user_id='test_user')
except Exception as e:
if "api_key" in str(e).lower() or "openai" in str(e).lower():
pytest.skip(f"Skipping: OpenAI API key required: {e}")
raise
# Should have results from files
assert 'results' in result
@@ -65,10 +80,14 @@ class TestDirectoryIngestion:
def test_directory_ingestion_processes_multiple_files(self, temp_knowledge_dir):
"""Test that all files in a directory are processed."""
from praisonaiagents import Knowledge
knowledge = _make_knowledge_or_skip()
knowledge = Knowledge()
result = knowledge.add(temp_knowledge_dir, user_id='test_user')
try:
knowledge.add(temp_knowledge_dir, user_id='test_user')
except Exception as e:
if "api_key" in str(e).lower() or "openai" in str(e).lower():
pytest.skip(f"Skipping: OpenAI API key required: {e}")
raise
# Search for content from both files
search1 = knowledge.search('ZEBRA-71', user_id='test_user')
@@ -80,10 +99,14 @@ class TestDirectoryIngestion:
def test_directory_ingestion_sets_metadata(self, temp_knowledge_dir):
"""Test that file metadata is properly set."""
from praisonaiagents import Knowledge
knowledge = _make_knowledge_or_skip()
knowledge = Knowledge()
knowledge.add(temp_knowledge_dir, user_id='test_user')
try:
knowledge.add(temp_knowledge_dir, user_id='test_user')
except Exception as e:
if "api_key" in str(e).lower() or "openai" in str(e).lower():
pytest.skip(f"Skipping: OpenAI API key required: {e}")
raise
search_result = knowledge.search('ZEBRA-71', user_id='test_user')
@@ -114,15 +137,20 @@ class TestContextBuilderUsesText:
"""Test that Agent._get_knowledge_context returns text, not path."""
from praisonaiagents import Agent
agent = Agent(
name='TestAgent',
instructions='Answer based on knowledge.',
knowledge=[temp_knowledge_dir],
output='silent',
)
# Ensure knowledge is processed
agent._ensure_knowledge_processed()
try:
agent = Agent(
name='TestAgent',
instructions='Answer based on knowledge.',
knowledge=[temp_knowledge_dir],
output='silent',
)
# Ensure knowledge is processed
agent._ensure_knowledge_processed()
except Exception as e:
if "api_key" in str(e).lower() or "openai" in str(e).lower():
pytest.skip(f"Skipping: OpenAI API key required: {e}")
raise
# Get context
context, _ = agent._get_knowledge_context('What is the password?', use_rag=True)
@@ -33,8 +33,6 @@ class TestLearnConfig:
assert config.feedback is False
assert config.improvements is False
assert config.scope == LearnScope.PRIVATE
assert config.auto_consolidate is True
assert config.retention_days is None
def test_custom_values(self):
"""Test LearnConfig with custom values."""
@@ -43,12 +41,10 @@ class TestLearnConfig:
insights=True,
patterns=True,
scope=LearnScope.SHARED,
retention_days=30,
)
assert config.persona is False
assert config.patterns is True
assert config.scope == LearnScope.SHARED
assert config.retention_days == 30
def test_to_dict(self):
"""Test LearnConfig to_dict method."""
@@ -271,6 +267,198 @@ class TestLearnManager:
assert "Python" in context
class TestBackendWiring:
"""P3-A: Test LearnBackend wiring — SQLite, Redis, FILE."""
def setup_method(self):
self.temp_dir = tempfile.mkdtemp()
def teardown_method(self):
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_sqlite_backend_round_trip(self):
"""SQLite backend stores and retrieves learnings."""
from praisonaiagents.config.feature_configs import LearnBackend
config = LearnConfig(
persona=True, insights=True, patterns=False,
backend=LearnBackend.SQLITE,
)
manager = LearnManager(config=config, user_id="test", store_path=self.temp_dir)
# Store
entry = manager.capture_persona("Prefers dark mode")
assert entry is not None
assert "dark mode" in entry.content
# Retrieve
context = manager.get_persona_context()
assert any("dark mode" in c for c in context)
def test_file_backend_unchanged(self):
"""FILE backend works as before (default)."""
config = LearnConfig(persona=True)
manager = LearnManager(config=config, user_id="test", store_path=self.temp_dir)
entry = manager.capture_persona("Prefers light mode")
assert entry is not None
# Verify JSON file exists
store_files = list(Path(self.temp_dir).rglob("*.json"))
assert len(store_files) > 0
def test_redis_fallback_when_unavailable(self):
"""Redis backend falls back to FILE when Redis not running."""
import warnings
from praisonaiagents.config.feature_configs import LearnBackend
config = LearnConfig(
persona=True,
backend=LearnBackend.REDIS,
db_url="redis://localhost:59999", # unlikely port
)
with warnings.catch_warnings(record=True):
warnings.simplefilter("always")
manager = LearnManager(config=config, user_id="test", store_path=self.temp_dir)
# Should still work with FILE fallback
entry = manager.capture_persona("Redis fallback test")
assert entry is not None
def test_mongodb_warns_and_falls_back(self):
"""MongoDB warns and falls back to FILE."""
from praisonaiagents.config.feature_configs import LearnBackend
config = LearnConfig(
persona=True,
backend=LearnBackend.MONGODB,
)
# Should not crash, falls back to FILE
manager = LearnManager(config=config, user_id="test", store_path=self.temp_dir)
entry = manager.capture_persona("Mongo fallback test")
assert entry is not None
class TestWasUpdated:
"""P3-B: Test was_updated tracking on BaseStore."""
def setup_method(self):
self.temp_dir = tempfile.mkdtemp()
def teardown_method(self):
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_was_updated_false_initially(self):
"""New store starts with was_updated=False."""
store = PersonaStore(store_path=os.path.join(self.temp_dir, "persona.json"))
assert store.was_updated is False
def test_was_updated_true_after_add(self):
"""was_updated becomes True after add()."""
store = PersonaStore(store_path=os.path.join(self.temp_dir, "persona.json"))
store.add("test preference")
assert store.was_updated is True
def test_was_updated_true_after_update(self):
"""was_updated becomes True after update()."""
store = PersonaStore(store_path=os.path.join(self.temp_dir, "persona.json"))
entry = store.add("preference")
store.reset_updated()
store.update(entry.id, "updated preference")
assert store.was_updated is True
def test_was_updated_true_after_delete(self):
"""was_updated becomes True after delete()."""
store = PersonaStore(store_path=os.path.join(self.temp_dir, "persona.json"))
entry = store.add("to delete")
store.reset_updated()
store.delete(entry.id)
assert store.was_updated is True
def test_was_updated_true_after_clear(self):
"""was_updated becomes True after clear()."""
store = PersonaStore(store_path=os.path.join(self.temp_dir, "persona.json"))
store.add("something")
store.reset_updated()
store.clear()
assert store.was_updated is True
def test_reset_updated(self):
"""reset_updated() sets flag back to False."""
store = PersonaStore(store_path=os.path.join(self.temp_dir, "persona.json"))
store.add("test")
assert store.was_updated is True
store.reset_updated()
assert store.was_updated is False
class TestProposeMode:
"""P3-C: Test PROPOSE mode — extract but don't auto-store."""
def setup_method(self):
self.temp_dir = tempfile.mkdtemp()
def teardown_method(self):
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_pending_learnings_initially_empty(self):
"""Pending learnings list starts empty."""
config = LearnConfig(persona=True, mode="propose")
manager = LearnManager(config=config, user_id="test", store_path=self.temp_dir)
assert manager.pending_learnings == []
def test_add_pending_learning(self):
"""Can add a pending learning manually."""
config = LearnConfig(persona=True, mode="propose")
manager = LearnManager(config=config, user_id="test", store_path=self.temp_dir)
manager.add_pending("User prefers Python", category="persona")
assert len(manager.pending_learnings) == 1
assert "Python" in manager.pending_learnings[0].content
def test_approve_learning_moves_to_store(self):
"""Approving a pending learning moves it to the main store."""
config = LearnConfig(persona=True, mode="propose")
manager = LearnManager(config=config, user_id="test", store_path=self.temp_dir)
manager.add_pending("User likes concise answers", category="persona")
pending_id = manager.pending_learnings[0].id
result = manager.approve_learning(pending_id)
assert result is True
assert len(manager.pending_learnings) == 0
# Should now be in persona store
personas = manager.get_persona_context()
assert any("concise" in p for p in personas)
def test_reject_learning_discards(self):
"""Rejecting a pending learning discards it."""
config = LearnConfig(persona=True, mode="propose")
manager = LearnManager(config=config, user_id="test", store_path=self.temp_dir)
manager.add_pending("Bad learning", category="persona")
pending_id = manager.pending_learnings[0].id
result = manager.reject_learning(pending_id)
assert result is True
assert len(manager.pending_learnings) == 0
# Should NOT be in any store
personas = manager.get_persona_context()
assert not any("Bad learning" in p for p in personas)
def test_approve_all_learnings(self):
"""Bulk approve all pending learnings."""
config = LearnConfig(persona=True, insights=True, mode="propose")
manager = LearnManager(config=config, user_id="test", store_path=self.temp_dir)
manager.add_pending("Preference A", category="persona")
manager.add_pending("Insight B", category="insights")
assert len(manager.pending_learnings) == 2
count = manager.approve_all_learnings()
assert count == 2
assert len(manager.pending_learnings) == 0
def test_reject_nonexistent_learning(self):
"""Rejecting a nonexistent ID returns False."""
config = LearnConfig(persona=True, mode="propose")
manager = LearnManager(config=config, user_id="test", store_path=self.temp_dir)
result = manager.reject_learning("nonexistent_id")
assert result is False
class TestNoCortexParam:
"""Test that cortex parameter does not exist."""
@@ -0,0 +1,341 @@
"""
TDD Tests for Learn Module Gap Fixes.
Tests for:
- Gap 1: process_conversation() auto-learning
- Gap 2: LearnProtocol for extensibility
- Gap 3: Learning modes (AGENTIC, PROPOSE)
- Gap 4: Async in learn module
- Gap 5: DB backend support in LearnConfig
"""
# Tests for Learn Module Gap Fixes
class TestLearnProtocols:
"""Test LearnProtocol and related protocols."""
def test_learn_protocol_exported(self):
"""LearnProtocol should be exported from top-level."""
from praisonaiagents import LearnProtocol
assert LearnProtocol is not None
def test_async_learn_protocol_exported(self):
"""AsyncLearnProtocol should be exported from top-level."""
from praisonaiagents import AsyncLearnProtocol
assert AsyncLearnProtocol is not None
def test_learn_manager_protocol_exported(self):
"""LearnManagerProtocol should be exported from top-level."""
from praisonaiagents import LearnManagerProtocol
assert LearnManagerProtocol is not None
def test_learn_manager_exported(self):
"""LearnManager should be exported from top-level."""
from praisonaiagents import LearnManager
assert LearnManager is not None
def test_custom_store_satisfies_protocol(self):
"""Custom store should satisfy LearnProtocol."""
from praisonaiagents import LearnProtocol
class CustomLearnStore:
def add(self, content, metadata=None):
return {"id": "test", "content": content}
def search(self, query, limit=10):
return []
def list_all(self, limit=100):
return []
def get(self, entry_id):
return None
def update(self, entry_id, content, metadata=None):
return None
def delete(self, entry_id):
return True
def clear(self):
return 0
store = CustomLearnStore()
assert isinstance(store, LearnProtocol)
class TestLearnModes:
"""Test LearnMode enum and configuration."""
def test_learn_mode_exported(self):
"""LearnMode should be exported from top-level."""
from praisonaiagents import LearnMode
assert LearnMode is not None
def test_learn_mode_values(self):
"""LearnMode should have DISABLED, AGENTIC, PROPOSE values."""
from praisonaiagents import LearnMode
assert LearnMode.DISABLED.value == "disabled"
assert LearnMode.AGENTIC.value == "agentic"
assert LearnMode.PROPOSE.value == "propose"
def test_learn_config_has_mode(self):
"""LearnConfig should have mode parameter."""
from praisonaiagents import LearnConfig, LearnMode
config = LearnConfig(mode=LearnMode.AGENTIC)
assert config.mode == LearnMode.AGENTIC
def test_learn_config_mode_default_disabled(self):
"""LearnConfig mode should default to DISABLED."""
from praisonaiagents import LearnConfig, LearnMode
config = LearnConfig()
assert config.mode == LearnMode.DISABLED
def test_learn_config_mode_string(self):
"""LearnConfig should accept mode as string."""
from praisonaiagents import LearnConfig
config = LearnConfig(mode="agentic")
assert config.mode == "agentic"
class TestLearnBackend:
"""Test LearnBackend enum and configuration."""
def test_learn_backend_exported(self):
"""LearnBackend should be exported from top-level."""
from praisonaiagents import LearnBackend
assert LearnBackend is not None
def test_learn_backend_values(self):
"""LearnBackend should have FILE, SQLITE, REDIS, MONGODB values."""
from praisonaiagents import LearnBackend
assert LearnBackend.FILE.value == "file"
assert LearnBackend.SQLITE.value == "sqlite"
assert LearnBackend.REDIS.value == "redis"
assert LearnBackend.MONGODB.value == "mongodb"
def test_learn_config_has_backend(self):
"""LearnConfig should have backend parameter."""
from praisonaiagents import LearnConfig, LearnBackend
config = LearnConfig(backend=LearnBackend.SQLITE)
assert config.backend == LearnBackend.SQLITE
def test_learn_config_has_db_url(self):
"""LearnConfig should have db_url parameter."""
from praisonaiagents import LearnConfig
config = LearnConfig(
backend="sqlite",
db_url="sqlite:///learn.db"
)
assert config.db_url == "sqlite:///learn.db"
def test_learn_config_has_llm(self):
"""LearnConfig should have llm parameter for extraction."""
from praisonaiagents import LearnConfig
config = LearnConfig(llm="gpt-4o")
assert config.llm == "gpt-4o"
class TestProcessConversation:
"""Test process_conversation() auto-learning method."""
def test_learn_manager_has_process_conversation(self):
"""LearnManager should have process_conversation method."""
from praisonaiagents.memory.learn import LearnManager
manager = LearnManager()
assert hasattr(manager, 'process_conversation')
assert callable(manager.process_conversation)
def test_learn_manager_has_aprocess_conversation(self):
"""LearnManager should have async aprocess_conversation method."""
from praisonaiagents.memory.learn import LearnManager
manager = LearnManager()
assert hasattr(manager, 'aprocess_conversation')
assert callable(manager.aprocess_conversation)
def test_process_conversation_empty_messages(self):
"""process_conversation should handle empty messages."""
from praisonaiagents.memory.learn import LearnManager
manager = LearnManager()
result = manager.process_conversation([])
assert result == {"persona": [], "insights": [], "patterns": [], "stored": {}}
def test_process_conversation_signature(self):
"""process_conversation should have correct signature."""
from praisonaiagents.memory.learn import LearnManager
import inspect
manager = LearnManager()
sig = inspect.signature(manager.process_conversation)
params = list(sig.parameters.keys())
assert "messages" in params
assert "llm" in params
class TestLearnConfigToDict:
"""Test LearnConfig.to_dict() includes new fields."""
def test_to_dict_includes_mode(self):
"""to_dict should include mode."""
from praisonaiagents import LearnConfig, LearnMode
config = LearnConfig(mode=LearnMode.AGENTIC)
d = config.to_dict()
assert "mode" in d
assert d["mode"] == "agentic"
def test_to_dict_includes_backend(self):
"""to_dict should include backend."""
from praisonaiagents import LearnConfig, LearnBackend
config = LearnConfig(backend=LearnBackend.SQLITE)
d = config.to_dict()
assert "backend" in d
assert d["backend"] == "sqlite"
def test_to_dict_includes_db_url(self):
"""to_dict should include db_url."""
from praisonaiagents import LearnConfig
config = LearnConfig(db_url="sqlite:///test.db")
d = config.to_dict()
assert "db_url" in d
assert d["db_url"] == "sqlite:///test.db"
def test_to_dict_includes_llm(self):
"""to_dict should include llm."""
from praisonaiagents import LearnConfig
config = LearnConfig(llm="gpt-4o")
d = config.to_dict()
assert "llm" in d
assert d["llm"] == "gpt-4o"
class TestAgentWithLearnMode:
"""Test Agent with learn mode configuration."""
def test_agent_accepts_learn_config_with_mode(self):
"""Agent should accept LearnConfig with mode."""
from praisonaiagents import Agent, LearnConfig, LearnMode
agent = Agent(
name="test",
instructions="Test agent",
learn=LearnConfig(
mode=LearnMode.AGENTIC,
persona=True,
insights=True,
),
)
assert agent._learn_config is not None
assert agent._learn_config.mode == LearnMode.AGENTIC
def test_agent_accepts_learn_config_with_backend(self):
"""Agent should accept LearnConfig with backend."""
from praisonaiagents import Agent, LearnConfig, LearnBackend
agent = Agent(
name="test",
instructions="Test agent",
learn=LearnConfig(
backend=LearnBackend.SQLITE,
db_url="sqlite:///test.db",
),
)
assert agent._learn_config is not None
assert agent._learn_config.backend == LearnBackend.SQLITE
def test_agent_has_process_auto_learning_method(self):
"""Agent should have _process_auto_learning method."""
from praisonaiagents import Agent
agent = Agent(name="test", instructions="Test agent")
assert hasattr(agent, '_process_auto_learning')
assert callable(agent._process_auto_learning)
class TestCustomStores:
"""Test custom_stores support in LearnManager."""
def test_learn_manager_accepts_custom_stores(self):
"""LearnManager should accept custom_stores parameter."""
from praisonaiagents import LearnManager, LearnConfig
class MyCustomStore:
def __init__(self):
self.data = []
def add(self, content, metadata=None):
self.data.append(content)
return {"id": "1", "content": content}
def search(self, query, limit=10):
return []
def list_all(self, limit=100):
return self.data
def get(self, entry_id):
return None
def update(self, entry_id, content, metadata=None):
return None
def delete(self, entry_id):
return True
def clear(self):
return 0
custom_store = MyCustomStore()
manager = LearnManager(
config=LearnConfig(persona=True),
custom_stores={"domain": custom_store}
)
assert "domain" in manager._stores
assert manager._stores["domain"] is custom_store
def test_custom_stores_can_override_builtin(self):
"""Custom stores can override built-in stores."""
from praisonaiagents import LearnManager, LearnConfig
class MyPersonaStore:
def add(self, content, metadata=None):
return {"id": "custom", "content": content}
def search(self, query, limit=10):
return []
def list_all(self, limit=100):
return []
def get(self, entry_id):
return None
def update(self, entry_id, content, metadata=None):
return None
def delete(self, entry_id):
return True
def clear(self):
return 0
custom_persona = MyPersonaStore()
manager = LearnManager(
config=LearnConfig(persona=True),
custom_stores={"persona": custom_persona}
)
# Custom stores should override built-in stores
assert "persona" in manager._stores
assert manager._stores["persona"] is custom_persona
@@ -23,14 +23,18 @@ from unittest.mock import patch, MagicMock, PropertyMock
class TestDoomLoopFiring:
"""G1/G2: Doom loop detection must actually fire in run_autonomous()."""
def test_doom_loop_fires_when_agent_repeats_same_response(self):
@patch("praisonaiagents.agent.agent.Agent.get_recommended_stage", return_value="heuristic")
def test_doom_loop_fires_when_agent_repeats_same_response(self, mock_stage):
"""run_autonomous() must call _record_action() so DoomLoopTracker gets data."""
from praisonaiagents import Agent
agent = Agent(
name="test_doom",
instructions="You are a test agent",
autonomy={"max_iterations": 10, "doom_loop_threshold": 3},
# Need enough iterations for graduated recovery to exhaust all 4 stages:
# threshold=3 means 3 identical actions per doom-loop cycle, 4 recovery
# stages, plus check iterations = ~16 iterations minimum.
autonomy={"max_iterations": 25, "doom_loop_threshold": 3},
)
# Mock chat to return the same response every time
@@ -43,16 +47,16 @@ class TestDoomLoopFiring:
agent.chat = mock_chat
# Use a prompt that triggers autonomous stage (multi-step)
# so the loop runs multiple iterations instead of exiting after 1 (direct)
result = agent.run_autonomous("First refactor the auth module, then test it step by step")
# Doom loop should have been detected before max_iterations
assert result.completion_reason == "doom_loop", (
f"Expected doom_loop but got {result.completion_reason} after {result.iterations} iterations"
# Graduated recovery produces needs_help (stage 3) or doom_loop (stage 4).
# Both are valid doom-loop-family abort outcomes.
assert result.completion_reason in ("doom_loop", "needs_help"), (
f"Expected doom_loop or needs_help but got {result.completion_reason} "
f"after {result.iterations} iterations"
)
assert result.success is False
assert result.iterations < 10 # Should stop well before max
assert result.iterations < 25 # Should stop well before max
def test_doom_loop_does_not_fire_with_varied_responses(self):
"""Varied responses should NOT trigger doom loop."""
@@ -157,7 +157,11 @@ class TestDoomLoopRecovery:
agent = self._make_agent(threshold=2)
result = agent.run_autonomous("Do something")
assert result.success is False
assert result.completion_reason == "doom_loop"
# Graduated recovery produces needs_help (stage 3) or doom_loop (stage 4).
# Both are valid doom-loop-family abort outcomes.
assert result.completion_reason in ("doom_loop", "needs_help"), (
f"Expected doom_loop or needs_help but got {result.completion_reason}"
)
@patch("praisonaiagents.agent.agent.Agent.chat")
@patch("praisonaiagents.agent.agent.Agent.get_recommended_stage", return_value="heuristic")
@@ -232,7 +232,10 @@ class TestMultiChannelGatewayConfig:
"""Test MultiChannelGatewayConfig with defaults."""
from praisonaiagents.gateway import MultiChannelGatewayConfig
config = MultiChannelGatewayConfig()
assert isinstance(config.gateway, GatewayConfig)
# Use qualname check to be robust against module reimport identity issues
assert type(config.gateway).__qualname__ == "GatewayConfig", (
f"Expected GatewayConfig, got {type(config.gateway).__qualname__}"
)
assert config.agents == {}
assert config.channels == {}
@@ -0,0 +1,336 @@
"""TDD Tests for Memory vs History Separation Fixes.
Tests G-1 (auto_save_session routes to SessionStore) and
G-2 (Session._save_agent_chat_histories routes to SessionStore).
These tests verify the clean separation between:
- Memory = semantic facts/entities (injected into system prompt)
- History = conversation turns (injected into message array)
"""
import pytest
from unittest.mock import MagicMock, patch
class TestG1AutoSaveSessionRouting:
"""G-1: _auto_save_session() should route to SessionStore first."""
def test_auto_save_uses_session_store_when_available(self):
"""When SessionStore is available, use it instead of Memory.save_session()."""
from praisonaiagents import Agent
# Create agent with auto_save
agent = Agent(
name="test",
instructions="Test agent",
auto_save="test_session",
)
# Mock session store
mock_session_store = MagicMock()
agent._session_store = mock_session_store
# Add some chat history
agent.chat_history = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"},
]
# Call _auto_save_session
agent._auto_save_session()
# Verify SessionStore.add_message was called
assert mock_session_store.add_message.call_count == 2
def test_auto_save_falls_back_to_memory_when_no_session_store(self):
"""When SessionStore is not available, fall back to Memory.save_session()."""
from praisonaiagents import Agent
# Create agent with auto_save and memory
agent = Agent(
name="test",
instructions="Test agent",
auto_save="test_session",
)
# No session store
agent._session_store = None
# Mock memory instance
mock_memory = MagicMock()
mock_memory.save_session = MagicMock()
agent._memory_instance = mock_memory
# Add some chat history
agent.chat_history = [
{"role": "user", "content": "Hello"},
]
# Call _auto_save_session
agent._auto_save_session()
# Verify Memory.save_session was called as fallback
mock_memory.save_session.assert_called_once()
def test_auto_save_does_nothing_when_disabled(self):
"""When auto_save is not set, do nothing."""
from praisonaiagents import Agent
agent = Agent(
name="test",
instructions="Test agent",
)
# No auto_save set
agent.auto_save = None
# Mock session store
mock_session_store = MagicMock()
agent._session_store = mock_session_store
# Call _auto_save_session
agent._auto_save_session()
# Verify nothing was called
mock_session_store.add_message.assert_not_called()
class TestG2SessionSaveAgentChatHistories:
"""G-2: Session._save_agent_chat_histories() should route to SessionStore first."""
def test_save_agent_histories_routes_to_session_store(self):
"""Verify _save_agent_chat_histories has SessionStore routing logic."""
from praisonaiagents.session import Session
import inspect
# Verify the method exists and has the G-2 fix comment
source = inspect.getsource(Session._save_agent_chat_histories)
# Check that the G-2 fix is present
assert "G-2 FIX" in source, "G-2 fix should be documented in the method"
assert "SessionStore" in source, "Method should reference SessionStore"
assert "get_default_session_store" in source, "Method should try to get SessionStore"
def test_save_agent_histories_behavioral(self):
"""Issue 3 FIX: Behavioral test - verify actual routing to SessionStore."""
from praisonaiagents.session import Session
# Create session
session = Session(session_id="test_behavioral", user_id="test_user")
# Create mock agent with chat history
mock_agent = MagicMock()
mock_agent.chat_history = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi!"},
]
# Add agent to session
session._agents["test_agent"] = {
"agent": mock_agent,
"chat_history": []
}
# Call the method - it should not raise an exception
# This tests the actual behavior, not just source code
session._save_agent_chat_histories()
class TestIssue1DuplicatePrevention:
"""Issue 1: Verify duplicate history insertion is prevented."""
def test_auto_save_tracks_last_saved_index(self):
"""Verify _auto_save_session tracks last saved index to prevent duplicates."""
from praisonaiagents import Agent
agent = Agent(
name="test",
instructions="Test agent",
auto_save="test_session",
)
# Mock session store
mock_session_store = MagicMock()
agent._session_store = mock_session_store
# Add initial chat history
agent.chat_history = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"},
]
# First save - should save 2 messages
agent._auto_save_session()
assert mock_session_store.add_message.call_count == 2
# Second save with same history - should save 0 new messages
mock_session_store.reset_mock()
agent._auto_save_session()
assert mock_session_store.add_message.call_count == 0, "Should not re-save existing messages"
# Add new message
agent.chat_history.append({"role": "user", "content": "How are you?"})
# Third save - should save only 1 new message
mock_session_store.reset_mock()
agent._auto_save_session()
assert mock_session_store.add_message.call_count == 1, "Should only save new message"
class TestEdgeCaseClearHistoryResetIndex:
"""Edge case: clear_history() must reset _auto_save_last_index."""
def test_clear_history_resets_auto_save_index(self):
"""Verify clear_history() resets _auto_save_last_index to prevent silent message loss."""
from praisonaiagents import Agent
agent = Agent(
name="test",
instructions="Test agent",
auto_save="test_session",
)
mock_store = MagicMock()
agent._session_store = mock_store
# Save initial messages
agent.chat_history = [{"role": "user", "content": "msg1"}]
agent._auto_save_session()
assert mock_store.add_message.call_count == 1
assert agent._auto_save_last_index == 1
# Clear history - index should reset
agent.clear_history()
assert agent._auto_save_last_index == 0, "Index should reset to 0 after clear_history()"
# Add new message after clear
agent.chat_history = [{"role": "user", "content": "msg2"}]
mock_store.reset_mock()
agent._auto_save_session()
# Should save the new message (not skip due to stale index)
assert mock_store.add_message.call_count == 1, "New message should be saved after clear"
def test_prune_history_adjusts_auto_save_index(self):
"""Verify prune_history() adjusts _auto_save_last_index."""
from praisonaiagents import Agent
agent = Agent(
name="test",
instructions="Test agent",
auto_save="test_session",
)
mock_store = MagicMock()
agent._session_store = mock_store
# Save 5 messages
agent.chat_history = [
{"role": "user", "content": f"msg{i}"} for i in range(5)
]
agent._auto_save_session()
assert agent._auto_save_last_index == 5
# Prune to keep last 2
agent.prune_history(keep_last=2)
assert len(agent.chat_history) == 2
assert agent._auto_save_last_index == 2, "Index should match pruned history length"
def test_delete_history_adjusts_auto_save_index(self):
"""Verify delete_history() adjusts _auto_save_last_index."""
from praisonaiagents import Agent
agent = Agent(
name="test",
instructions="Test agent",
auto_save="test_session",
)
mock_store = MagicMock()
agent._session_store = mock_store
# Save 3 messages
agent.chat_history = [
{"role": "user", "content": f"msg{i}"} for i in range(3)
]
agent._auto_save_session()
assert agent._auto_save_last_index == 3
# Delete one message
agent.delete_history(0)
assert len(agent.chat_history) == 2
assert agent._auto_save_last_index == 2, "Index should be adjusted after deletion"
class TestMemoryHistorySeparation:
"""Test that Memory and History are properly separated."""
def test_memory_context_does_not_include_conversation_turns(self):
"""Memory.get_context() should return facts/entities, not conversation turns."""
from praisonaiagents.memory import FileMemory
memory = FileMemory(user_id="test_user")
# Store some facts using correct API
memory.add_long_term("User prefers Python programming")
memory.add_entity("Alice", "person", {"role": "developer"})
# Get context
context = memory.get_context()
# Should contain facts and entities
assert "Important Facts" in context or "Known Entities" in context or context == ""
# Should NOT contain conversation turn markers
assert "role" not in context.lower() or "user:" not in context.lower()
def test_session_store_stores_conversation_turns(self):
"""SessionStore should store conversation turns with role/content."""
from praisonaiagents.session import get_default_session_store
store = get_default_session_store()
# Add messages
store.add_message("test_session", role="user", content="Hello")
store.add_message("test_session", role="assistant", content="Hi!")
# Get history
history = store.get_chat_history("test_session")
# Should have conversation turns
assert len(history) >= 2
assert any(msg.get("role") == "user" for msg in history)
assert any(msg.get("role") == "assistant" for msg in history)
class TestG3PerTurnPersist:
"""G-3: Verify per-turn persist exists via _persist_message()."""
def test_persist_message_exists(self):
"""Agent should have _persist_message method for per-turn persistence."""
from praisonaiagents import Agent
agent = Agent(name="test", instructions="Test")
# Verify method exists
assert hasattr(agent, '_persist_message')
assert callable(agent._persist_message)
def test_persist_message_uses_session_store(self):
"""_persist_message should use SessionStore when available."""
from praisonaiagents import Agent
# Create agent without session_id (it's set via memory config)
agent = Agent(name="test", instructions="Test")
# Mock session store
mock_store = MagicMock()
agent._session_store = mock_store
agent._session_id = "test_session"
agent._db = None # No DB adapter
# Call _persist_message
agent._persist_message("user", "Hello")
# Verify SessionStore was used
mock_store.add_user_message.assert_called_once_with("test_session", "Hello")
+1 -1
View File
@@ -2564,7 +2564,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "1.5.30"
version = "1.5.39"
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.5.30.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.5.30.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.5.39.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.5.39.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+152 -3
View File
@@ -51,6 +51,149 @@ def gateway_status(
handler.status(host=host, port=port)
@app.command("channels")
def gateway_channels(
config: str = typer.Option("gateway.yaml", "--config", "-c", help="Path to gateway.yaml"),
json_output: bool = typer.Option(False, "--json", help="Output JSON"),
):
"""List channels configured in a gateway.yaml file.
Examples:
praisonai gateway channels
praisonai gateway channels --config my-gateway.yaml --json
"""
import os
import yaml
if not os.path.exists(config):
print(f"Error: Config file not found: {config}")
raise typer.Exit(1)
with open(config) as f:
cfg = yaml.safe_load(f) or {}
channels = cfg.get("channels", {})
if not channels:
print("No channels configured.")
raise typer.Exit(0)
if json_output:
import json
print(json.dumps(channels, indent=2))
raise typer.Exit(0)
try:
from rich.table import Table
from rich.console import Console
console = Console()
table = Table(title="Configured Channels")
table.add_column("Name", style="cyan")
table.add_column("Platform", style="green")
table.add_column("Token", style="yellow")
table.add_column("Config Keys", style="dim")
for name, ch_cfg in channels.items():
platform = ch_cfg.get("platform", "unknown")
token_val = ch_cfg.get("token", "")
has_token = "✅ set" if token_val else "❌ missing"
keys = ", ".join(
k for k in ch_cfg.keys() if k not in ("platform", "token")
)
table.add_row(name, platform, has_token, keys or "—")
console.print(table)
except ImportError:
print(f"{'Name':<20} {'Platform':<12} {'Token':<12}")
print("-" * 44)
for name, ch_cfg in channels.items():
platform = ch_cfg.get("platform", "unknown")
has_token = "set" if ch_cfg.get("token") else "missing"
print(f"{name:<20} {platform:<12} {has_token:<12}")
@app.command("send")
def gateway_send(
config: str = typer.Option("gateway.yaml", "--config", "-c", help="Path to gateway.yaml"),
channel: str = typer.Option(..., "--channel", help="Channel name from config (e.g. 'telegram')"),
channel_id: str = typer.Option(..., "--channel-id", help="Target chat/channel ID"),
message: str = typer.Option(..., "--message", "-m", help="Message text to send"),
thread_id: Optional[str] = typer.Option(None, "--thread-id", help="Optional thread ID"),
):
"""Send a one-shot test message to a channel bot.
Instantiates the bot from gateway.yaml config, sends the message, then exits.
Useful for testing scheduled delivery targets.
Examples:
praisonai gateway send --config gateway.yaml --channel telegram --channel-id 12345 -m "Hello"
"""
import os
import asyncio
import yaml
if not os.path.exists(config):
print(f"Error: Config file not found: {config}")
raise typer.Exit(1)
with open(config) as f:
cfg = yaml.safe_load(f) or {}
channels_cfg = cfg.get("channels", {})
ch_cfg = channels_cfg.get(channel)
if not ch_cfg:
available = ", ".join(channels_cfg.keys()) if channels_cfg else "(none)"
print(f"Error: Channel '{channel}' not found in config. Available: {available}")
raise typer.Exit(1)
platform = ch_cfg.get("platform", channel)
token = ch_cfg.get("token", "")
# Resolve env vars in token
if token and token.startswith("${") and token.endswith("}"):
env_var = token[2:-1]
token = os.environ.get(env_var, "")
if not token:
print(f"Error: Environment variable {env_var} not set")
raise typer.Exit(1)
async def _send():
try:
from praisonai.gateway.server import WebSocketGateway
bot = WebSocketGateway._create_bot(channel, ch_cfg)
except Exception as e:
print(f"Error creating bot: {e}")
raise typer.Exit(1)
try:
await bot.start()
await asyncio.sleep(1) # Let bot initialise
result = await bot.send_message(
channel_id, message, thread_id=thread_id,
)
print(f"✅ Message sent to {channel}:{channel_id}")
if hasattr(result, "message_id"):
print(f" Message ID: {result.message_id}")
except Exception as e:
print(f"❌ Send failed: {e}")
raise typer.Exit(1)
finally:
try:
await bot.stop()
except Exception:
pass
try:
asyncio.run(_send())
except typer.Exit:
raise
except Exception as e:
print(f"Error: {e}")
raise typer.Exit(1)
@app.callback(invoke_without_command=True)
def gateway_callback(ctx: typer.Context):
"""Show gateway help if no subcommand provided."""
@@ -61,8 +204,10 @@ def gateway_callback(ctx: typer.Context):
Manage the gateway server: praisonai gateway <command>
[bold]Commands:[/bold]
[green]start[/green] Start the gateway server
[green]status[/green] Check gateway status
[green]start[/green] Start the gateway server
[green]status[/green] Check gateway status
[green]channels[/green] List channels from gateway.yaml
[green]send[/green] Send a test message to a channel
[bold]Multi-Bot Mode:[/bold]
praisonai gateway start --config gateway.yaml
@@ -73,11 +218,15 @@ Manage the gateway server: praisonai gateway <command>
[bold]Check Status:[/bold]
praisonai gateway status
[bold]Channel Management:[/bold]
praisonai gateway channels --config gateway.yaml
praisonai gateway send --config gateway.yaml --channel telegram --channel-id 12345 -m "test"
"""
try:
from rich import print as rprint
rprint(help_text)
except ImportError:
import re
plain = re.sub(r'\[/?[^\]]+\]', '', help_text)
plain = re.sub(r'\\[/?[^\\]]+\\]', '', help_text)
print(plain)
@@ -37,6 +37,51 @@ def _run_schedule(args: list) -> int:
return 4
@app.command("add")
def schedule_add_cmd(
name: str = typer.Argument(..., help="Schedule name (e.g. 'morning-hello')"),
schedule: str = typer.Option(..., "--schedule", "-s", help="When to run: 'hourly', 'daily', '*/30m', 'cron:0 9 * * *', 'at:2026-03-01T09:00', 'in 20 minutes'"),
message: str = typer.Option("", "--message", "-m", help="Prompt / reminder text"),
agent: str = typer.Option("", "--agent", "-a", help="Agent ID to execute this job (default: first registered agent)"),
channel: str = typer.Option("", "--channel", help="Delivery platform: telegram, discord, slack, whatsapp"),
channel_id: str = typer.Option("", "--channel-id", help="Target chat/channel ID on the platform"),
session_id: str = typer.Option("", "--session-id", help="Session ID to preserve conversation context"),
json_output: bool = typer.Option(False, "--json", help="Output JSON"),
):
"""Add a job to the schedule store (with optional delivery target).
Examples:
praisonai schedule add "morning-hello" -s "cron:0 9 * * *" -m "say hello"
praisonai schedule add "tg-reminder" -s daily -m "check email" --agent support --channel telegram --channel-id 12345
"""
output = get_output_controller()
try:
from praisonaiagents.tools.schedule_tools import schedule_add as _schedule_add
result = _schedule_add(
name=name,
schedule=schedule,
message=message,
agent_id=agent,
channel=channel,
channel_id=channel_id,
session_id=session_id,
)
if json_output:
import json as _json
print(_json.dumps({"result": result}))
else:
if "Error" in result:
output.print_error(result)
raise typer.Exit(1)
else:
output.print_success(result)
except ImportError as e:
output.print_error(f"Schedule tools not available: {e}")
raise typer.Exit(4)
@app.command("start")
def schedule_start(
agents_file: str = typer.Argument("agents.yaml", help="Agents YAML file"),
+14 -1
View File
@@ -801,7 +801,7 @@ class PraisonAI:
return default_args
# Define special commands
special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule', 'skills', 'profile', 'eval', 'agents', 'run', 'thinking', 'compaction', 'output', 'deploy', 'templates', 'recipe', 'endpoints', 'audio', 'embed', 'embedding', 'images', 'moderate', 'files', 'batches', 'vector-stores', 'rerank', 'ocr', 'assistants', 'fine-tuning', 'completions', 'messages', 'guardrails', 'rag', 'videos', 'a2a', 'containers', 'passthrough', 'responses', 'search', 'realtime-api', 'doctor', 'registry', 'package', 'install', 'uninstall', 'acp', 'debug', 'lsp', 'diag', 'browser', 'replay', 'bot', 'gateway', 'sandbox', 'wizard', 'migrate', 'security', 'persistence']
special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule', 'skills', 'profile', 'eval', 'agents', 'run', 'thinking', 'compaction', 'output', 'deploy', 'templates', 'recipe', 'endpoints', 'audio', 'embed', 'embedding', 'images', 'moderate', 'files', 'batches', 'vector-stores', 'rerank', 'ocr', 'assistants', 'fine-tuning', 'completions', 'messages', 'guardrails', 'rag', 'videos', 'a2a', 'containers', 'passthrough', 'responses', 'search', 'realtime-api', 'doctor', 'registry', 'package', 'install', 'uninstall', 'acp', 'debug', 'lsp', 'diag', 'browser', 'replay', 'bot', 'gateway', 'sandbox', 'wizard', 'migrate', 'security', 'persistence', 'paths']
parser = argparse.ArgumentParser(prog="praisonai", description="praisonAI command-line interface")
parser.add_argument("--framework", choices=["crewai", "autogen", "praisonai"], help="Specify the framework")
@@ -1711,6 +1711,19 @@ class PraisonAI:
exit_code = handle_audit_command(unknown_args)
sys.exit(exit_code)
elif args.command == 'paths':
# Paths command - storage path inspection
# Routes to Typer CLI for paths commands (show)
from .app import app as typer_app, register_commands
register_commands()
import sys as _sys
_sys.argv = ['praisonai', 'paths'] + unknown_args
try:
typer_app()
except SystemExit as e:
sys.exit(e.code if e.code else 0)
sys.exit(0)
# Only check framework availability for agent-related operations
if not args.command and (args.init or args.auto or args.framework):
if not CREWAI_AVAILABLE and not AUTOGEN_AVAILABLE and not PRAISONAI_AVAILABLE:
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==4.5.30 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==4.5.39 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+179 -22
View File
@@ -202,6 +202,9 @@ class WebSocketGateway:
self._channel_bots: Dict[str, Any] = {} # channel_name -> bot instance
self._routing_rules: Dict[str, Dict[str, str]] = {} # channel_name -> {context -> agent_id}
self._channel_tasks: List[asyncio.Task] = []
# Scheduler tick background task
self._scheduler_task: Optional[asyncio.Task] = None
@property
def is_running(self) -> bool:
@@ -722,6 +725,91 @@ class WebSocketGateway:
"channels": channel_status,
}
# ── Scheduled delivery ────────────────────────────────────────────
async def _deliver_scheduled_result(
self, delivery: Any, text: str,
) -> None:
"""Route a scheduled job result to the correct channel bot.
Args:
delivery: A ``DeliveryTarget`` with ``channel`` and ``channel_id``.
text: The agent response to deliver.
"""
channel = getattr(delivery, "channel", "") or ""
channel_id = getattr(delivery, "channel_id", "") or ""
thread_id = getattr(delivery, "thread_id", None)
if not channel or not channel_id:
logger.warning("Delivery target missing channel or channel_id, skipping")
return
bot = self.get_channel_bot(channel)
if bot is None:
# Try case-insensitive lookup
for name, b in self._channel_bots.items():
if name.lower() == channel.lower():
bot = b
break
if bot is None:
logger.warning(
"No channel bot '%s' found for scheduled delivery", channel,
)
return
try:
await bot.send_message(
channel_id, text, thread_id=thread_id,
)
logger.info(
"Delivered scheduled result to %s:%s", channel, channel_id,
)
except Exception as e:
logger.error(
"Failed to deliver to %s:%s: %s", channel, channel_id, e,
)
def _start_scheduler_tick(self, interval: float = 15.0) -> None:
"""Start a background task that polls the scheduler for due jobs.
Creates a ``ScheduledAgentExecutor`` wired to:
- a ``ScheduleRunner`` with a ``FileScheduleStore``
- this gateway's agent registry for resolution
- ``_deliver_scheduled_result`` for outbound delivery
"""
async def _run():
try:
from praisonaiagents.scheduler import (
ScheduleRunner,
FileScheduleStore,
)
from praisonai.scheduler.executor import ScheduledAgentExecutor
except ImportError as e:
logger.warning(
"Scheduler dependencies not available, skipping tick: %s", e,
)
return
store = FileScheduleStore()
runner = ScheduleRunner(store)
def _resolve_agent(agent_id):
if agent_id:
return self._agents.get(agent_id)
# Fallback: first registered agent
return next(iter(self._agents.values()), None) if self._agents else None
executor = ScheduledAgentExecutor(
runner=runner,
agent_resolver=_resolve_agent,
delivery_handler=self._deliver_scheduled_result,
)
await executor.run_loop(interval=interval)
self._scheduler_task = asyncio.create_task(_run())
logger.info("Scheduler tick started (interval=15s)")
# ── Multi-bot lifecycle ───────────────────────────────────────────
@staticmethod
@@ -782,13 +870,15 @@ class WebSocketGateway:
if not isinstance(cdef, dict):
errors.append(f"Channel '{cname}' must be a dictionary")
continue
if cname.lower() not in valid_platforms:
# Resolve platform: explicit field takes priority over key
platform = cdef.get("platform", cname).lower()
if platform not in valid_platforms:
logger.warning(
f"Channel '{cname}' is not a known platform "
f"({', '.join(valid_platforms)})"
f"Channel '{cname}' platform '{platform}' is not "
f"a known platform ({', '.join(sorted(valid_platforms))})"
)
# WhatsApp web mode doesn't require a token
is_wa_web = (cname.lower() == "whatsapp" and
is_wa_web = (platform == "whatsapp" and
cdef.get("mode", "cloud").lower().strip() == "web")
if not cdef.get("token") and not is_wa_web:
errors.append(
@@ -814,9 +904,15 @@ class WebSocketGateway:
return _resolve(raw)
def _create_agents_from_config(self, agents_cfg: Dict[str, Dict[str, Any]]) -> None:
"""Create and register Agent instances from the agents section of gateway.yaml.
def _create_agents_from_config(
self,
agents_cfg: Dict[str, Dict[str, Any]],
*,
default_model: Optional[str] = None,
guardrails_cfg: Optional[Dict[str, Any]] = None,
) -> None:
"""Create and register Agent instances from the agents section of config.
Supports all agent configuration options including:
- tools: List of tool names to resolve via ToolResolver
- tool_choice: Tool selection mode ('auto', 'required', 'none')
@@ -825,10 +921,19 @@ class WebSocketGateway:
- role: Agent role (CrewAI-style)
- goal: Agent goal (CrewAI-style)
- backstory: Agent backstory (CrewAI-style)
- temperature: LLM temperature (optional, uses SDK default)
- base_url: Custom LLM endpoint URL (optional)
- api_key: API key for LLM provider (optional)
Args:
agents_cfg: The ``agents`` section of the config.
default_model: Fallback model when an agent has no ``model`` key
(e.g. from the ``provider.model`` section of UI config).
guardrails_cfg: Optional ``guardrails.registry`` dict to wire per-agent.
"""
from praisonaiagents import Agent
# G1: Resolve tool names to callables (same pattern as agents_generator)
# Resolve tool names to callables (same pattern as agents_generator)
tool_resolver = None
try:
from praisonai.tool_resolver import ToolResolver
@@ -838,15 +943,26 @@ class WebSocketGateway:
for agent_id, agent_def in agents_cfg.items():
instructions = agent_def.get("instructions", "")
model = agent_def.get("model", None)
# G7: Apply provider.model as fallback when agent has no model
model = agent_def.get("model", None) or default_model
memory = agent_def.get("memory", False)
# G4: Support role/goal/backstory for CrewAI-style agents
# G2: Pass temperature through (optional, SDK uses its own default)
temperature = agent_def.get("temperature", None)
# G3/G4: Pass base_url and api_key for multi-provider support
base_url = agent_def.get("base_url", None)
api_key = agent_def.get("api_key", None)
# Support role/goal/backstory for CrewAI-style agents
role = agent_def.get("role", None)
goal = agent_def.get("goal", None)
backstory = agent_def.get("backstory", None)
# G1: Resolve tools from YAML config
# G6: Fall back to system_prompt if instructions is empty
if not instructions:
instructions = agent_def.get("system_prompt", "") or ""
# Resolve tools from YAML config
agent_tools = []
yaml_tool_names = agent_def.get("tools", [])
if yaml_tool_names and tool_resolver:
@@ -860,16 +976,32 @@ class WebSocketGateway:
logger.debug(f"Resolved tool '{tool_name}' for agent '{agent_id}'")
else:
logger.warning(f"Tool '{tool_name}' not found for agent '{agent_id}'")
# Additional agent options from YAML
tool_choice = agent_def.get("tool_choice", None)
reflection = agent_def.get("reflection", True) # Default: enable reflection/interactive mode
reflection = agent_def.get("reflection", True)
allow_delegation = agent_def.get("allow_delegation", False)
# G5: Wire guardrails if config has a matching agent_name or global rule
guardrail_prompt = None
if guardrails_cfg:
for _gr_id, gr_def in guardrails_cfg.items():
gr_agent = gr_def.get("agent_name", "")
if not gr_agent or gr_agent == agent_id:
guardrail_prompt = gr_def.get("description", None)
break # first match wins
# Compose model as dict when temperature is specified
# (Agent accepts temperature via dict model config)
model_cfg = model
if temperature is not None and model and isinstance(model, str):
model_cfg = {"model": model, "temperature": temperature}
# Use model= (preferred) instead of deprecated llm=
agent = Agent(
name=agent_id,
instructions=instructions,
llm=model,
model=model_cfg,
memory=memory,
tools=agent_tools if agent_tools else None,
reflection=reflection,
@@ -877,14 +1009,21 @@ class WebSocketGateway:
role=role,
goal=goal,
backstory=backstory,
base_url=base_url,
api_key=api_key,
guardrails=guardrail_prompt,
)
# Store tool_choice for later use in chat()
if tool_choice:
agent._yaml_tool_choice = tool_choice
self.register_agent(agent, agent_id=agent_id)
logger.info(f"Created agent '{agent_id}' (model={model}, tools={len(agent_tools)}, reflection={reflection})")
logger.info(
f"Created agent '{agent_id}' "
f"(model={model}, tools={len(agent_tools)}, "
f"reflection={reflection})"
)
def _determine_routing_context(
self, channel_type: str, message_metadata: Dict[str, Any]
@@ -934,7 +1073,7 @@ class WebSocketGateway:
from praisonaiagents.bots import BotConfig
for channel_name, ch_cfg in channels_cfg.items():
channel_type = channel_name.lower()
channel_type = ch_cfg.get("platform", channel_name).lower()
token = ch_cfg.get("token", "")
# WhatsApp web mode doesn't require a token
wa_web_mode = (channel_type == "whatsapp" and
@@ -1222,9 +1361,16 @@ class WebSocketGateway:
# Recreate agents
agents_cfg = cfg.get("agents", {})
provider_cfg = cfg.get("provider", {})
default_model = provider_cfg.get("model") if provider_cfg else None
guardrails_cfg = (cfg.get("guardrails") or {}).get("registry")
if agents_cfg:
self._agents.clear()
self._create_agents_from_config(agents_cfg)
self._create_agents_from_config(
agents_cfg,
default_model=default_model,
guardrails_cfg=guardrails_cfg,
)
# Restart channels
channels_cfg = cfg.get("channels", {})
@@ -1273,8 +1419,15 @@ class WebSocketGateway:
# Create agents
agents_cfg = cfg.get("agents", {})
provider_cfg = cfg.get("provider", {})
default_model = provider_cfg.get("model") if provider_cfg else None
guardrails_cfg = (cfg.get("guardrails") or {}).get("registry")
if agents_cfg:
self._create_agents_from_config(agents_cfg)
self._create_agents_from_config(
agents_cfg,
default_model=default_model,
guardrails_cfg=guardrails_cfg,
)
# Start channels + WebSocket server concurrently
channels_cfg = cfg.get("channels", {})
@@ -1289,6 +1442,8 @@ class WebSocketGateway:
self._config_watch_task = asyncio.create_task(
self._watch_config(config_path)
)
# Launch scheduler tick to poll for due jobs
self._start_scheduler_tick()
await self.start()
# Register signal handlers for graceful shutdown using
@@ -1316,4 +1471,6 @@ class WebSocketGateway:
finally:
if self._config_watch_task:
self._config_watch_task.cancel()
if self._scheduler_task:
self._scheduler_task.cancel()
await self.stop_channels()
@@ -17,6 +17,8 @@ __all__ = [
'ScheduleParser',
'ExecutorInterface',
'PraisonAgentExecutor',
'ScheduledAgentExecutor',
'JobResult',
]
# Lazy imports for modules that will be created later
@@ -42,4 +44,7 @@ def __getattr__(name):
elif name == 'create_deployment_scheduler':
from .agent_scheduler import create_agent_scheduler
return create_agent_scheduler
elif name in ('ScheduledAgentExecutor', 'JobResult'):
from .executor import ScheduledAgentExecutor, JobResult
return ScheduledAgentExecutor if name == 'ScheduledAgentExecutor' else JobResult
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -0,0 +1,268 @@
"""
ScheduledAgentExecutor — bridges ScheduleRunner + agent resolution.
This is a thin reusable bridge that any consumer (UI, CLI, custom app)
can use to poll the scheduler and dispatch due jobs to agents, without
reimplementing the same polling + agent-lookup logic each time.
Lives in the **wrapper** layer (not the SDK core) because it composes
two independent core modules: gateway (agent registry) and scheduler
(due-job detection).
Usage::
from praisonaiagents.scheduler import ScheduleRunner, FileScheduleStore
from praisonai.scheduler.executor import ScheduledAgentExecutor
store = FileScheduleStore()
runner = ScheduleRunner(store)
executor = ScheduledAgentExecutor(
runner=runner,
agent_resolver=lambda agent_id: my_gateway.get_agent(agent_id),
)
# In your async event loop:
async for job, result in executor.tick():
print(f"Job {job.name}: {result}")
"""
from __future__ import annotations
import asyncio
import logging
import time
from dataclasses import dataclass, field
from typing import (
Any,
AsyncIterator,
Callable,
Dict,
List,
Optional,
Tuple,
TYPE_CHECKING,
)
if TYPE_CHECKING:
from praisonaiagents.scheduler import ScheduleRunner, ScheduleJob
logger = logging.getLogger(__name__)
@dataclass
class JobResult:
"""Result of a single scheduled job execution.
Attributes:
job: The ScheduleJob that was executed.
result: Agent response (str) or None on failure.
status: ``"succeeded"``, ``"failed"``, or ``"skipped"``.
error: Error message if status is ``"failed"``.
duration: Wall-clock seconds for agent execution.
delivered: Whether the result was delivered to a channel bot.
"""
job: Any # ScheduleJob
result: Optional[str] = None
status: str = "succeeded"
error: Optional[str] = None
duration: float = 0.0
delivered: bool = False
class ScheduledAgentExecutor:
"""Bridges ScheduleRunner + agent resolution for any consumer.
This is a **stateless** helper — it does not own a loop or thread.
Callers decide how and when to call :meth:`tick` or :meth:`run_loop`.
Parameters:
runner: An SDK ``ScheduleRunner`` instance.
agent_resolver: A callable ``(agent_id: str | None) -> Agent``
that returns an agent for the given ID. May return ``None``
if no agent is available.
delivery_handler: Optional async callable
``(delivery: DeliveryTarget, text: str) -> None``.
Called after successful execution when the job has a
delivery target. Typically routes to a channel bot's
``send_message()``.
on_success: Optional callback ``(job, result) -> None``.
on_failure: Optional callback ``(job, error) -> None``.
"""
def __init__(
self,
runner: "ScheduleRunner",
agent_resolver: Callable[[Optional[str]], Any],
*,
delivery_handler: Optional[Callable[..., Any]] = None,
on_success: Optional[Callable[..., None]] = None,
on_failure: Optional[Callable[..., None]] = None,
) -> None:
self._runner = runner
self._resolve = agent_resolver
self._deliver = delivery_handler
self._on_success = on_success
self._on_failure = on_failure
# ── public API ───────────────────────────────────────────────────
async def tick(self) -> AsyncIterator[JobResult]:
"""Check for due jobs and execute them, yielding results.
Each due job is dispatched to the resolved agent via
``agent.chat(job.message)`` in a background thread (so sync
agents work correctly in an async context).
Yields:
:class:`JobResult` for each due job.
"""
due_jobs: List["ScheduleJob"] = self._runner.get_due_jobs()
for job in due_jobs:
result = await self._execute_one(job)
yield result
async def tick_all(self) -> List[JobResult]:
"""Like :meth:`tick` but collects all results into a list."""
results: List[JobResult] = []
async for r in self.tick():
results.append(r)
return results
async def run_loop(
self,
interval: float = 15.0,
*,
max_ticks: Optional[int] = None,
) -> None:
"""Convenience loop that calls :meth:`tick` at a fixed interval.
Args:
interval: Seconds between ticks (default 15).
max_ticks: Stop after this many ticks (``None`` = forever).
"""
tick_count = 0
logger.info(
"ScheduledAgentExecutor loop started (interval=%.1fs)", interval,
)
try:
while max_ticks is None or tick_count < max_ticks:
async for result in self.tick():
logger.info(
"Job '%s' %s (%.1fs)",
getattr(result.job, "name", result.job),
result.status,
result.duration,
)
tick_count += 1
await asyncio.sleep(interval)
except asyncio.CancelledError:
pass
finally:
logger.info("ScheduledAgentExecutor loop stopped after %d ticks", tick_count)
# ── internals ────────────────────────────────────────────────────
async def _execute_one(self, job: "ScheduleJob") -> JobResult:
"""Execute a single job and return the result."""
started = time.time()
message = getattr(job, "message", "") or ""
agent_id = getattr(job, "agent_id", None)
# Skip jobs with no message
if not message:
self._runner.mark_run(job)
return JobResult(
job=job,
status="skipped",
error="No message configured",
duration=time.time() - started,
)
# Resolve the agent
try:
agent = self._resolve(agent_id)
except Exception as e:
logger.warning("Agent resolution failed for job '%s': %s", job.id, e)
err = f"Agent resolution failed: {e}"
if self._on_failure:
self._on_failure(job, err)
return JobResult(
job=job, status="failed", error=err, duration=time.time() - started,
)
if agent is None:
err = f"No agent found for agent_id={agent_id!r}"
logger.warning("Skipping job '%s': %s", job.id, err)
if self._on_failure:
self._on_failure(job, err)
return JobResult(
job=job, status="failed", error=err, duration=time.time() - started,
)
# Execute via agent.chat() in a thread (sync-safe)
# Wire session_target: "main" preserves context, "isolated" is fresh
try:
session_target = getattr(job, "session_target", "isolated")
delivery = getattr(job, "delivery", None)
session_id = (
getattr(delivery, "session_id", None)
if delivery else None
)
# Build chat kwargs
chat_kwargs: Dict[str, Any] = {}
if session_target == "main" and session_id:
chat_kwargs["session_id"] = session_id
else:
# Isolated mode: use a cron-specific session
chat_kwargs["session_id"] = f"cron_{job.id}"
result = await asyncio.to_thread(
agent.chat, message, **chat_kwargs,
)
result_str = str(result)
except Exception as e:
logger.warning("Job '%s' execution failed: %s", job.id, e)
err = str(e)
if self._on_failure:
self._on_failure(job, err)
return JobResult(
job=job, status="failed", error=err, duration=time.time() - started,
)
# Success
self._runner.mark_run(job)
duration = time.time() - started
# Deliver to channel bot if delivery target exists
delivered = False
delivery = getattr(job, "delivery", None)
if delivery and self._deliver:
try:
coro = self._deliver(delivery, result_str)
if asyncio.iscoroutine(coro):
await coro
delivered = True
logger.info(
"Delivered job '%s' result to %s:%s",
job.id, delivery.channel, delivery.channel_id,
)
except Exception as e:
logger.warning(
"Delivery failed for job '%s' to %s:%s: %s",
job.id, delivery.channel, delivery.channel_id, e,
)
if self._on_success:
self._on_success(job, result_str)
return JobResult(
job=job,
result=result_str,
status="succeeded",
duration=duration,
delivered=delivered,
)
@@ -1156,7 +1156,7 @@ print(f'Section 5: {{result5}}')
"1. Follow the template structure exactly",
"2. Use REAL, WORKING code examples (not placeholders)",
"3. Be concise but comprehensive",
"4. Include a mermaid diagram if this is a concept page",
"4. Include exactly TWO mermaid diagrams for concept pages: (1) high-level flow graph LR, (2) detailed flowchart TB with subgraphs. Use colour scheme: classDef agent fill:#8B0000,color:#fff and classDef tool fill:#189AB4,color:#fff - apply :::agent to inputs/outputs/agents, :::tool to processes",
"5. Use Mintlify MDX components (CodeGroup, Tabs, Cards)",
"6. Ensure all imports are correct for praisonaiagents",
"7. Code examples must be complete and runnable",
@@ -28,25 +28,31 @@ icon: "{icon}"
```mermaid
flowchart LR
A[User Request]:::agent --> B[Process]:::tool
B --> C[Validate]:::tool
C --> D[Execute]:::tool
D --> E[Result]:::agent
classDef agent fill:#8B0000,color:#fff
classDef tool fill:#189AB4,color:#fff
```
```mermaid
flowchart TB
subgraph Input
A[User Request]
A[User Request]:::agent
end
subgraph {name}
B[Process] --> C[Validate]
C --> D[Execute]
B[Process]:::tool --> C[Validate]:::tool --> D[Execute]:::tool
end
subgraph Output
E[Result]
E[Result]:::agent
end
A --> B
D --> E
style B fill:#189AB4,color:#fff
style C fill:#2E8B57,color:#fff
style D fill:#8B0000,color:#fff
classDef agent fill:#8B0000,color:#fff
classDef tool fill:#189AB4,color:#fff
```
## Types of {name}
@@ -148,8 +148,10 @@ icon: "lightbulb"
```mermaid
graph LR
A[Input] --> B[{name}]
B --> C[Output]
A[Input]:::agent --> B[{name}]:::tool --> C[Output]:::agent
classDef agent fill:#8B0000,color:#fff
classDef tool fill:#189AB4,color:#fff
```
## When to Use
+1 -1
View File
@@ -1 +1 @@
__version__ = "4.5.30"
__version__ = "4.5.39"
+2 -2
View File
@@ -4,7 +4,7 @@ dynamic = ["version"]
description = "PraisonAI is an AI Agents Framework with Self Reflection. PraisonAI application combines PraisonAI Agents, AutoGen, and CrewAI into a low-code solution for building and managing multi-agent LLM systems, focusing on simplicity, customisation, and efficient human-agent collaboration."
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.10,<3.13"
requires-python = ">=3.10,<3.15"
authors = [
{ name = "Mervin Praison" }
]
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=1.5.30",
"praisonaiagents>=1.5.39",
"python-dotenv>=0.19.0",
"litellm>=1.81.0",
"PyYAML>=6.0",
+754 -17
View File
File diff suppressed because it is too large. Load diff