Merge pull request #629 from phenobarbital/orchestrator

Orchestrator

Author: phenobarbital

parrot/bots/abstract.py

@@ -27,8 +27,7 @@
     DEFAULT_ROLE,
     DEFAULT_CAPABILITIES,
     DEFAULT_BACKHISTORY,
-    DEFAULT_RATIONALE,
-    OUTPUT_SYSTEM_PROMPT
+    DEFAULT_RATIONALE
 )
 from ..clients.base import (
     LLM_PRESETS,
@@ -61,7 +60,7 @@
 except ImportError:
     from ..security.prompt_injection import PromptInjectionDetector
     PYTECTOR_ENABLED = False
-from ..mcp import MCPEnabledMixin, MCPServerConfig
+from ..mcp import MCPEnabledMixin
 from ..security import (
     SecurityEventLogger,
     ThreatLevel,
@@ -70,11 +69,12 @@
 from .stores import LocalKBMixin
 from ..interfaces import ToolInterface, VectorInterface
 if TYPE_CHECKING:
-    from ..stores import AbstractStore, supported_stores
+    from ..stores import AbstractStore
     from ..stores.kb import AbstractKnowledgeBase
     from ..stores.models import StoreConfig
 from ..models.status import AgentStatus
 from .dynamic_values import dynamic_values
+from .middleware import PromptPipeline
 
 
 logging.getLogger(name='primp').setLevel(logging.INFO)
@@ -83,10 +83,19 @@
 logging.getLogger('markdown_it').setLevel(logging.CRITICAL)
 
 # LLM parser regex:
-_LLM_PATTERN = re.compile(r'^([a-zA-Z0-9_-]+):(.+)$')
+_LLM_PATTERN = re.compile(
+    r'^([a-zA-Z0-9_-]+):(.+)$'
+)
 
 
-class AbstractBot(MCPEnabledMixin, DBInterface, LocalKBMixin, ToolInterface, VectorInterface, ABC):
+class AbstractBot(
+    MCPEnabledMixin,
+    DBInterface,
+    LocalKBMixin,
+    ToolInterface,
+    VectorInterface,
+    ABC
+):
     """AbstractBot.
 
     This class is an abstract representation a base abstraction for all Chatbots.
@@ -97,6 +106,7 @@ class AbstractBot(MCPEnabledMixin, DBInterface, LocalKBMixin, ToolInterface, Vec
         '_llm',
         '_llm_config',
         '_llm_kwargs',
+        '_prompt_pipeline'
     )
     # Define system prompt template
     system_prompt_template = BASIC_SYSTEM_PROMPT
@@ -175,6 +185,8 @@ def __init__(
             'description',
             self.description or f"{self.name} Chatbot"
         )
+        # Prompt Pipeline:
+        self._prompt_pipeline: PromptPipeline = None
 
 
         # Status and Events
@@ -332,6 +344,14 @@ def __init__(
             logger=self.logger
         )
 
+    @property
+    def prompt_pipeline(self) -> Optional['PromptPipeline']:
+        return self._prompt_pipeline
+
+    @prompt_pipeline.setter
+    def prompt_pipeline(self, pipeline: 'PromptPipeline'):
+        self._prompt_pipeline = pipeline
+
     def _parse_llm_string(self, llm: str) -> Tuple[str, Optional[str]]:
         """Parse 'provider:model' or plain provider string."""
         return match.groups() if (match := _LLM_PATTERN.match(llm)) else (llm, None)

parrot/bots/middleware.py

@@ -0,0 +1,50 @@
+"""Prompt middleware pipeline for query transformation."""
+from typing import Callable, Awaitable, Dict, Any, List
+from dataclasses import dataclass
+import logging
+
+
+@dataclass
+class PromptMiddleware:
+    """Single transformation step in the prompt pipeline."""
+    name: str
+    priority: int = 0  # Lower = runs first
+    transform: Callable[
+        [str, Dict[str, Any]], Awaitable[str]
+    ] = None
+    enabled: bool = True
+
+    async def process(self, query: str, context: Dict[str, Any]) -> str:
+        if not self.enabled or not self.transform:
+            return query
+        return await self.transform(query, context)
+
+
+class PromptPipeline:
+    """Ordered chain of prompt transformations applied before LLM call."""
+
+    def __init__(self):
+        self._middlewares: List[PromptMiddleware] = []
+        self.logger = logging.getLogger(__name__)
+
+    def add(self, middleware: PromptMiddleware) -> None:
+        self._middlewares.append(middleware)
+        self._middlewares.sort(key=lambda m: m.priority)
+
+    def remove(self, name: str) -> None:
+        self._middlewares = [m for m in self._middlewares if m.name != name]
+
+    async def process(self, query: str, context: Dict[str, Any] = None) -> str:
+        context = context or {}
+        for mw in self._middlewares:
+            try:
+                query = await mw.process(query, context)
+            except Exception as e:
+                self.logger.warning(
+                    f"Middleware '{mw.name}' failed: {e}, skipping"
+                )
+        return query
+
+    @property
+    def has_middlewares(self) -> bool:
+        return bool(self._middlewares)

parrot/clients/google/generation.py

@@ -1642,8 +1642,9 @@ async def generate_video_reel(
         Generates a complete video reel from a high-level request.
         Orchestrates:
         1. Scene breakdown (if not provided)
-        2. Parallel generation of Music and Scenes (Image -> Video, Audio)
-        3. Assembly using MoviePy
+        2. Apply user-provided speech texts to scenes (if provided; otherwise no narration)
+        3. Parallel generation of Music and Scenes (Image -> Video, Audio)
+        4. Assembly using MoviePy
         """
         self.logger.info(f"Starting Video Reel Generation: {request.prompt}")
         start_time = time.time()
@@ -1659,7 +1660,21 @@ async def generate_video_reel(
             self.logger.info("Breaking down prompt into scenes...")
             request.scenes = await self._breakdown_prompt_to_scenes(request.prompt)
 
-        # 2. Parallel Generation
+        # 2. Apply user-provided speech texts to scenes (if provided)
+        # This overrides any narration_text that might exist in scenes
+        if request.speech:
+            for i, scene in enumerate(request.scenes):
+                if i < len(request.speech):
+                    scene.narration_text = request.speech[i]
+                else:
+                    # No speech provided for this scene
+                    scene.narration_text = None
+        else:
+            # No speech provided at all - clear all narration
+            for scene in request.scenes:
+                scene.narration_text = None
+
+        # 3. Parallel Generation
         # Task 1: Music
         music_task = asyncio.create_task(
             self._generate_reel_music(request, output_directory)
@@ -1685,7 +1700,7 @@ async def generate_video_reel(
         if not valid_scene_outputs:
             raise RuntimeError("All scene generations failed.")
 
-        # 3. Assembly
+        # 4. Assembly
         final_video_path = await self._create_reel_assembly(
             valid_scene_outputs,
             music_path,
@@ -1718,9 +1733,10 @@ async def _breakdown_prompt_to_scenes(self, prompt: str) -> List[VideoReelScene]
         - `background_prompt`: Detailed visual description for the background image.
         - `foreground_prompt`: (Optional) Text describing a chart, KPI, or specific object to overlay. If not needed, omit.
         - `video_prompt`: Instructions for animating the scene (e.g., "Slow pan up", "Cinematic zoom").
-        - `narration_text`: (Optional) A short sentence for the narrator to read.
         - `duration`: Duration in seconds (usually 3-5s).
 
+        Note: Do NOT generate narration text. Narration/speech is provided separately by the user.
+
         Return the result as a JSON array of objects matching this schema.
         """
 
@@ -1733,7 +1749,6 @@ async def _breakdown_prompt_to_scenes(self, prompt: str) -> List[VideoReelScene]
                     "background_prompt": {"type": "string"},
                     "foreground_prompt": {"type": "string"},
                     "video_prompt": {"type": "string"},
-                    "narration_text": {"type": "string"},
                     "duration": {"type": "number"}
                 },
                 "required": ["background_prompt", "video_prompt", "duration"]

parrot/handlers/agent.py

@@ -751,8 +751,8 @@ async def post(self):
         with contextlib.suppress(AttributeError):
             request_session = self.request.session or await get_session(self.request)
 
-        # conversation (session_id)
-        session_id = data.pop('session_id', None) or qs.get('session_id') or uuid.uuid4().hex
+        # conversation (session_id) — already extracted by _get_user_session()
+        session_id = user_session
         # Support method invocation via body or query parameter in addition to the
         # /{agent_id}/{method_name} route so clients don't need to construct a
         # different URL for maintenance operations like refresh_data.

parrot/models/google.py

@@ -461,6 +461,14 @@ class VideoReelRequest(BaseModel):
         None,
         description="List of scenes. If not provided, they will be generated from the prompt."
     )
+    speech: Optional[List[str]] = Field(
+        None,
+        description=(
+            "List of speech/narration texts, one per scene. "
+            "If provided, each text will be used as narration for the corresponding scene. "
+            "If not provided, no narration will be added to the video reel."
+        )
+    )
     music_prompt: Optional[str] = Field(
         None,
         description="Description for the background music."

parrot/version.py

@@ -5,7 +5,7 @@
     "Complete Framework for AI Chatbots and Agents, "
     "Supporting A2A, MCP, RAG and more."
 )
-__version__ = "0.23.5"
+__version__ = "0.23.6"
 __author__ = "Jesus Lara"
 __author_email__ = "jesuslarag@gmail.com"
 __license__ = "MIT"