Merge pull request #1297 from MervinPraison/claude/issue-1288-20260408-0723

fix: resolve three architectural gaps (DRY, error protocol, streaming parity)

Author: MervinPraison

src/praisonai-agents/praisonaiagents/__init__.py

@@ -177,10 +177,10 @@ def _get_lazy_cache():
     'HandoffResult': ('praisonaiagents.agent.handoff', 'HandoffResult'),
     'HandoffInputData': ('praisonaiagents.agent.handoff', 'HandoffInputData'),
     'ContextPolicy': ('praisonaiagents.agent.handoff', 'ContextPolicy'),
-    'HandoffError': ('praisonaiagents.agent.handoff', 'HandoffError'),
-    'HandoffCycleError': ('praisonaiagents.agent.handoff', 'HandoffCycleError'),
-    'HandoffDepthError': ('praisonaiagents.agent.handoff', 'HandoffDepthError'),
-    'HandoffTimeoutError': ('praisonaiagents.agent.handoff', 'HandoffTimeoutError'),
+    'HandoffError': ('praisonaiagents.errors', 'HandoffError'),
+    'HandoffCycleError': ('praisonaiagents.errors', 'HandoffCycleError'),
+    'HandoffDepthError': ('praisonaiagents.errors', 'HandoffDepthError'),
+    'HandoffTimeoutError': ('praisonaiagents.errors', 'HandoffTimeoutError'),
 
     # Embedding API (Note: embedding/embeddings handled in custom_handler to override subpackage)
     'aembedding': ('praisonaiagents.embedding.embed', 'aembedding'),
@@ -204,7 +204,15 @@ def _get_lazy_cache():
 
     # Agent classes
     'Agent': ('praisonaiagents.agent.agent', 'Agent'),
-    'BudgetExceededError': ('praisonaiagents.agent.agent', 'BudgetExceededError'),
+    'BudgetExceededError': ('praisonaiagents.errors', 'BudgetExceededError'),
+    
+    # Error hierarchy - structured exception handling
+    'PraisonAIError': ('praisonaiagents.errors', 'PraisonAIError'),
+    'ToolExecutionError': ('praisonaiagents.errors', 'ToolExecutionError'),
+    'LLMError': ('praisonaiagents.errors', 'LLMError'),
+    'ValidationError': ('praisonaiagents.errors', 'ValidationError'),
+    'NetworkError': ('praisonaiagents.errors', 'NetworkError'),
+    'ErrorContextProtocol': ('praisonaiagents.errors', 'ErrorContextProtocol'),
     'Heartbeat': ('praisonaiagents.agent.heartbeat', 'Heartbeat'),
     'HeartbeatConfig': ('praisonaiagents.agent.heartbeat', 'HeartbeatConfig'),
     'ImageAgent': ('praisonaiagents.agent.image_agent', 'ImageAgent'),

src/praisonai-agents/praisonaiagents/agent/agent.py

@@ -187,23 +187,8 @@ def _is_file_path(value: str) -> bool:
     from ..rag.models import RAGResult, ContextPack
     from ..eval.results import EvaluationLoopResult
 
-class BudgetExceededError(Exception):
-    """Raised when an agent exceeds its max_budget.
-
-    Usage:
-        try:
-            agent.start("...")
-        except BudgetExceededError as e:
-            print(f"Agent '{e.agent_name}' spent ${e.total_cost:.4f} of ${e.max_budget:.4f}")
-    """
-    def __init__(self, agent_name: str, total_cost: float, max_budget: float):
-        self.agent_name = agent_name
-        self.total_cost = total_cost
-        self.max_budget = max_budget
-        super().__init__(
-            f"Agent '{agent_name}' exceeded budget: "
-            f"${total_cost:.4f} >= ${max_budget:.4f}"
-        )
+# Import structured error from central errors module
+from ..errors import BudgetExceededError
 
 class Agent(ToolExecutionMixin, ChatHandlerMixin, SessionManagerMixin, ChatMixin, ExecutionMixin, MemoryMixin):
     # Class-level counter for generating unique display names for nameless agents

src/praisonai-agents/praisonaiagents/agent/chat_mixin.py

@@ -13,6 +13,7 @@
 from praisonaiagents._logging import get_logger
 import asyncio
 import threading
+from ..errors import BudgetExceededError
 
 # Fallback helpers to avoid circular imports
 def _get_console():
@@ -666,7 +667,13 @@ def _chat_completion(self, messages, temperature=1.0, tools=None, stream=True, r
             self._llm_call_count += 1
             if self._max_budget and self._total_cost >= self._max_budget:
                 if self._on_budget_exceeded == "stop":
-                    raise BudgetExceededError(self.name, self._total_cost, self._max_budget)
+                    raise BudgetExceededError(
+                        f"Agent '{self.name}' exceeded budget: ${self._total_cost:.4f} >= ${self._max_budget:.4f}",
+                        budget_type="cost",
+                        limit=self._max_budget,
+                        used=self._total_cost,
+                        agent_id=self.name
+                    )
                 elif self._on_budget_exceeded == "warn":
                     logging.warning(
                         f"[budget] {self.name}: ${self._total_cost:.4f} exceeded "
@@ -692,6 +699,8 @@ def _chat_completion(self, messages, temperature=1.0, tools=None, stream=True, r
 
             return final_response
 
+        except BudgetExceededError:
+            raise
         except Exception as e:
             error_str = str(e).lower()
 

src/praisonai-agents/praisonaiagents/agent/handoff.py

@@ -100,29 +100,13 @@ def from_dict(cls, data: Dict[str, Any]) -> 'HandoffConfig':
             data["context_policy"] = ContextPolicy(data["context_policy"])
         return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})
 
-class HandoffError(Exception):
-    """Base exception for handoff errors."""
-    pass
-
-class HandoffCycleError(HandoffError):
-    """Raised when a cycle is detected in handoff chain."""
-    def __init__(self, chain: List[str]):
-        self.chain = chain
-        super().__init__(f"Handoff cycle detected: {' -> '.join(chain)}")
-
-class HandoffDepthError(HandoffError):
-    """Raised when max handoff depth is exceeded."""
-    def __init__(self, depth: int, max_depth: int):
-        self.depth = depth
-        self.max_depth = max_depth
-        super().__init__(f"Max handoff depth exceeded: {depth} > {max_depth}")
-
-class HandoffTimeoutError(HandoffError):
-    """Raised when handoff times out."""
-    def __init__(self, timeout: float, agent_name: str):
-        self.timeout = timeout
-        self.agent_name = agent_name
-        super().__init__(f"Handoff to {agent_name} timed out after {timeout}s")
+# Import structured error hierarchy from central errors module
+from ..errors import (
+    HandoffError, 
+    HandoffCycleError, 
+    HandoffDepthError, 
+    HandoffTimeoutError
+)
 
 # Thread-local storage for tracking handoff chains
 _handoff_context = threading.local()
@@ -270,12 +254,26 @@ def _check_safety(self, source_agent: 'Agent') -> None:
         if self.config.detect_cycles:
             chain = _get_handoff_chain()
             if target_name in chain:
-                raise HandoffCycleError(chain + [target_name])
+                cycle_path = chain + [target_name]
+                raise HandoffCycleError(
+                    f"Handoff cycle detected: {' -> '.join(cycle_path)}",
+                    source_agent=source_agent.name if hasattr(source_agent, 'name') else 'unknown',
+                    target_agent=target_name,
+                    agent_id=source_agent.name if hasattr(source_agent, 'name') else 'unknown',
+                    cycle_path=cycle_path
+                )
 
         # Check depth
         current_depth = _get_handoff_depth()
         if current_depth >= self.config.max_depth:
-            raise HandoffDepthError(current_depth + 1, self.config.max_depth)
+            raise HandoffDepthError(
+                f"Max handoff depth exceeded: {current_depth + 1} > {self.config.max_depth}",
+                source_agent=source_agent.name if hasattr(source_agent, 'name') else 'unknown',
+                target_agent=target_name,
+                agent_id=source_agent.name if hasattr(source_agent, 'name') else 'unknown',
+                max_depth=self.config.max_depth,
+                current_depth=current_depth + 1
+            )
 
     def _prepare_context(self, source_agent: 'Agent', kwargs: Dict[str, Any]) -> HandoffInputData:
         """
@@ -541,7 +539,13 @@ async def _execute():
                 handoff_depth=_get_handoff_depth(),
             )
             self._execute_callback(self.config.on_error, source_agent, kwargs, result)
-            raise HandoffTimeoutError(self.config.timeout_seconds, self.agent.name)
+            raise HandoffTimeoutError(
+                f"Handoff to {self.agent.name} timed out after {self.config.timeout_seconds}s",
+                timeout_seconds=self.config.timeout_seconds,
+                source_agent=source_agent.name if hasattr(source_agent, "name") else "unknown",
+                target_agent=self.agent.name,
+                agent_id=source_agent.name if hasattr(source_agent, "name") else "unknown"
+            )
         except Exception as e:
             result = HandoffResult(
                 success=False,

src/praisonai-agents/praisonaiagents/agents/agents.py

@@ -108,6 +108,93 @@ def get_multimodal_message(text_prompt: str, images: list) -> list:
             })
     return content
 
+def _prepare_task_prompt(task, task_description, context_text=None):
+    """
+    Prepare task prompt with context and memory (DRY helper).
+    """
+    # Build task prompt - only use "User Input/Topic" format if there's actual content
+    if context_text and context_text.strip():
+        task_prompt = f"""
+User Input/Topic: {context_text}
+
+Task: {task_description}
+Expected Output: {task.expected_output}
+
+IMPORTANT: Your response must be about the user's input/topic above. Incorporate it into your task."""
+    else:
+        task_prompt = f"""
+You need to do the following task: {task_description}.
+Expected Output: {task.expected_output}."""
+
+    # Add memory context if available
+    if task.memory:
+        try:
+            memory_context = task.memory.build_context_for_task(task.description)
+            if memory_context:
+                # Log detailed memory context for debugging
+                logger.debug(f"Memory context for task '{task.description}': {memory_context}")
+                # Include actual memory content without verbose headers (essential for AI agent functionality)
+                task_prompt += f"\n\n{memory_context}"
+        except Exception as e:
+            logger.error(f"Error getting memory context: {e}")
+
+    task_prompt += "\nPlease provide only the final result of your work. Do not add any conversation or extra explanation."
+    return task_prompt
+
+def _execute_with_agent_sync(executor_agent, task_prompt, task, tools, stream=None):
+    """
+    Execute task with agent synchronously (DRY helper).
+    """
+    if task.images:
+        return executor_agent.chat(
+            get_multimodal_message(task_prompt, task.images),
+            tools=tools,
+            output_json=task.output_json,
+            output_pydantic=task.output_pydantic,
+            stream=stream,
+            task_name=task.name,
+            task_description=task.description,
+            task_id=task.id
+        )
+    else:
+        return executor_agent.chat(
+            task_prompt,
+            tools=tools,
+            output_json=task.output_json,
+            output_pydantic=task.output_pydantic,
+            stream=stream,
+            task_name=task.name,
+            task_description=task.description,
+            task_id=task.id
+        )
+
+async def _execute_with_agent_async(executor_agent, task_prompt, task, tools, stream=None):
+    """
+    Execute task with agent asynchronously (DRY helper).
+    """
+    if task.images:
+        return await executor_agent.achat(
+            get_multimodal_message(task_prompt, task.images),
+            tools=tools,
+            output_json=task.output_json,
+            output_pydantic=task.output_pydantic,
+            stream=stream,
+            task_name=task.name,
+            task_description=task.description,
+            task_id=task.id
+        )
+    else:
+        return await executor_agent.achat(
+            task_prompt,
+            tools=tools,
+            output_json=task.output_json,
+            output_pydantic=task.output_pydantic,
+            stream=stream,
+            task_name=task.name,
+            task_description=task.description,
+            task_id=task.id
+        )
+
 def process_task_context(context_item, verbose=0, user_id=None):
     """
     Process a single context item for task execution.
@@ -703,59 +790,17 @@ async def aexecute_task(self, task_id):
             context_separator = '\n\n'
             context_text = context_separator.join(unique_contexts)
 
-        # Build task prompt - only use "User Input/Topic" format if there's actual content
-        if context_text and context_text.strip():
-            task_prompt = f"""
-User Input/Topic: {context_text}
-
-Task: {task_description}
-Expected Output: {task.expected_output}
-
-IMPORTANT: Your response must be about the user's input/topic above. Incorporate it into your task."""
-        else:
-            task_prompt = f"""
-You need to do the following task: {task_description}.
-Expected Output: {task.expected_output}."""
-
-        # Add memory context if available
-        if task.memory:
-            try:
-                memory_context = task.memory.build_context_for_task(task.description)
-                if memory_context:
-                    # Log detailed memory context for debugging
-                    logger.debug(f"Memory context for task '{task.description}': {memory_context}")
-                    # Include actual memory content without verbose headers (essential for AI agent functionality)
-                    task_prompt += f"\n\n{memory_context}"
-            except Exception as e:
-                logger.error(f"Error getting memory context: {e}")
-
-        task_prompt += "\nPlease provide only the final result of your work. Do not add any conversation or extra explanation."
+        # Build task prompt using DRY helper
+        task_prompt = _prepare_task_prompt(task, task_description, context_text)
 
         if self.verbose >= 2:
             logger.info(f"Executing task {task_id}: {task_description} using {executor_agent.display_name}")
         logger.debug(f"Starting execution of task {task_id} with prompt:\n{task_prompt}")
 
-        if task.images:
-            # Use shared multimodal helper (DRY - defined at module level)
-            agent_output = await executor_agent.achat(
-                get_multimodal_message(task_prompt, task.images),
-                tools=tools,
-                output_json=task.output_json,
-                output_pydantic=task.output_pydantic,
-                task_name=task.name,
-                task_description=task.description,
-                task_id=task.id
-            )
-        else:
-            agent_output = await executor_agent.achat(
-                task_prompt,
-                tools=tools,
-                output_json=task.output_json,
-                output_pydantic=task.output_pydantic,
-                task_name=task.name,
-                task_description=task.description,
-                task_id=task.id
-            )
+        # Execute with agent using DRY helper (fixes missing stream parameter)
+        agent_output = await _execute_with_agent_async(
+            executor_agent, task_prompt, task, tools, stream=self.stream
+        )
 
         if agent_output:
             # Store the response in memory
@@ -1066,60 +1111,17 @@ def execute_task(self, task_id):
             context_separator = '\n\n'
             context_text = context_separator.join(unique_contexts)
 
-        # Build task prompt - only use "User Input/Topic" format if there's actual content
-        if context_text and context_text.strip():
-            task_prompt = f"""
-User Input/Topic: {context_text}
-
-Task: {task_description}
-Expected Output: {task.expected_output}
-
-IMPORTANT: Your response must be about the user's input/topic above. Incorporate it into your task."""
-        else:
-            task_prompt = f"""
-You need to do the following task: {task_description}.
-Expected Output: {task.expected_output}."""
-
-        # Add memory context if available
-        if task.memory:
-            try:
-                memory_context = task.memory.build_context_for_task(task.description)
-                if memory_context:
-                    # Log detailed memory context for debugging
-                    logger.debug(f"Memory context for task '{task.description}': {memory_context}")
-                    # Include actual memory content without verbose headers (essential for AI agent functionality)
-                    task_prompt += f"\n\n{memory_context}"
-            except Exception as e:
-                logger.error(f"Error getting memory context: {e}")
-
-        task_prompt += "\nPlease provide only the final result of your work. Do not add any conversation or extra explanation."
+        # Build task prompt using DRY helper
+        task_prompt = _prepare_task_prompt(task, task_description, context_text)
 
         if self.verbose >= 2:
             logger.info(f"Executing task {task_id}: {task.description} using {executor_agent.display_name}")
         logger.debug(f"Starting execution of task {task_id} with prompt:\n{task_prompt}")
 
-        if task.images:
-            # Use shared multimodal helper (DRY - defined at module level)
-            agent_output = executor_agent.chat(
-                get_multimodal_message(task_prompt, task.images),
-                tools=tools,
-                output_json=task.output_json,
-                output_pydantic=task.output_pydantic,
-                task_name=task.name,
-                task_description=task.description,
-                task_id=task_id
-            )
-        else:
-            agent_output = executor_agent.chat(
-                task_prompt,
-                tools=tools,
-                output_json=task.output_json,
-                output_pydantic=task.output_pydantic,
-                stream=self.stream,
-                task_name=task.name,
-                task_description=task.description,
-                task_id=task_id
-            )
+        # Execute with agent using DRY helper
+        agent_output = _execute_with_agent_sync(
+            executor_agent, task_prompt, task, tools, stream=self.stream
+        )
 
         if agent_output:
             # Store the response in memory