vitepress-template

Appearance

Workflow Orchestration and Pipeline ​

**Referenced Files in This Document** - [[main_flow.py]](file/bi-chat/bi-chat/src/workflow/main-flow.py) - [[execution_engine.py]](file/bi-chat/bi-chat/src/workflow/execution-engine.py) - [[context.py]](file/bi-chat/bi-chat/src/workflow/context.py) - [[parallel_executor.py]](file/bi-chat/bi-chat/src/workflow/parallel-executor.py) - [[conversation_manager.py]](file/bi-chat/bi-chat/src/agents/conversation-manager.py) - [[progress_reporter.py]](file/bi-chat/bi-chat/src/utils/progress-reporter.py) - [[schemas.py]](file/bi-chat/bi-chat/src/core/schemas.py) - [[manager_agent.py]](file/bi-chat/bi-chat/src/agents/manager-agent.py) - [[complex_task_handler.py]](file/bi-chat/bi-chat/src/agents/complex-task-handler.py) - [[chat.py]](file/bi-chat/bi-chat/src/routes/chat.py)

Table of Contents ​

  1. Introduction
  2. Project Structure
  3. Core Components
  4. Architecture Overview
  5. Detailed Component Analysis
  6. Dependency Analysis
  7. Performance Considerations
  8. Troubleshooting Guide
  9. Conclusion

Introduction ​

This document explains the MainPipeline workflow orchestration system that powers intelligent, multi-agent analytical workflows. It covers how the pipeline coordinates agent interactions, manages conversation flow, and executes complex multi-step tasks. It also documents the conversation management integration that preserves context and history, the dynamic tool injection mechanism, and the SSE streaming implementation for real-time progress updates. Practical execution patterns, error handling strategies, and performance optimization techniques are included, along with integration points to the conversation manager and agent communication protocols.

Project Structure ​

The MainPipeline orchestrates a set of specialized agents (Manager, Data, General, Knowledge, Calculation, Summary, and a Clarification Guard) around a Plan-Execute-Reflect-Loop (PERL) workflow. The system integrates tightly with a conversation manager that maintains session state and memory, and with a progress reporting subsystem that streams updates via SSE.

Diagram sources

Section sources

Core Components ​

  • MainPipeline: Orchestrates the PERL loop, coordinates agents, manages artifacts and history, and drives DAG execution and reflection.
  • DAGExecutor: Executes tasks in parallel batches respecting dependencies and agent availability, with robust error handling and progress reporting.
  • WorkflowContext: Shared state container for plan, history, artifacts, and completion tracking across iterations.
  • ManagerAgent: Generates and refines task plans, enforces strict internal vs external routing rules, and ensures task purity.
  • ComplexTaskHandler: Alternative route for multi-step tasks, supporting iterative planning, execution, reflection, and summarization.
  • ConversationManager: Creates and caches session-specific agents, injects environment perception tools, and loads/persists memory.
  • ProgressReporter: Emits structured events for SSE streaming, enabling real-time UI updates and A2UI rendering.
  • SSE Streaming: Converts incremental chunks from the pipeline into OpenAI-compatible SSE responses.

Section sources

Architecture Overview ​

The system follows a modular, asynchronous architecture:

  • API layer accepts requests and streams responses via SSE.
  • ConversationManager builds or retrieves session-scoped agents and injects environment-aware tools.
  • MainPipeline runs the PERL loop, delegating planning to ManagerAgent and execution to DAGExecutor.
  • ProgressReporter emits structured events consumed by the UI and SSE generator.
  • Reflection steers replanning until satisfaction criteria are met.
  • Final synthesis is performed by SummaryAgent.

Diagram sources

Detailed Component Analysis ​

MainPipeline: PERL Loop Orchestration ​

MainPipeline coordinates the entire workflow:

  • Initializes agents and registers them in a shared registry.
  • Enforces safety via a ClarificationGuard that can reject or request clarification.
  • Builds a WorkflowContext from the original query and history.
  • Uses ManagerAgent to generate an initial TaskPlan.
  • Executes tasks via DAGExecutor, collecting results and artifacts.
  • Reflects on outcomes and replans until satisfied.
  • Summarizes final results using SummaryAgent.

Diagram sources

Section sources

DAGExecutor: Parallel Execution with Dependency Control ​

DAGExecutor performs:

  • Topological batching respecting dependencies and agent availability.
  • Parallel execution of ready tasks with concurrency limits per agent.
  • Robust error handling returning error Msg objects with task_id metadata.
  • Progress reporting for subtask lifecycle.

Diagram sources

Section sources

WorkflowContext: Shared State and Artifacts ​

WorkflowContext encapsulates:

  • original_query and history for cross-turn continuity.
  • plan, current_step_id, and completed_steps for iteration control.
  • artifacts dictionary for inter-task data exchange.
  • convenience methods to manage history, artifacts, and next step selection.

Section sources

ManagerAgent: Planning and Routing ​

ManagerAgent:

  • Generates initial TaskPlans with execution_mode and subtasks.
  • Enforces strict routing rules: internal business data → UnifiedDataAgent; external info → GeneralAgent.
  • Ensures task purity and dependency correctness.
  • Provides replan capability to adjust plans based on reflection feedback.

Section sources

ComplexTaskHandler: Iterative Multi-Step Execution ​

ComplexTaskHandler:

  • Alternative route for complex tasks outside the main pipeline.
  • Supports iterative planning, DAG execution, reflection, and summarization.
  • Emits A2UI render events for plan and reflection cards.

Section sources

ConversationManager: Session and Agent Lifecycle ​

ConversationManager:

  • Creates or retrieves session-scoped agents with memory and optional long-term memory.
  • Dynamically injects environment perception tools into each agent’s toolkit.
  • Loads conversation history intelligently, sanitizing incomplete tool call sequences.
  • Provides build_workflow_agents to assemble the agent set for the pipeline.

Section sources

ProgressReporter and SSE Streaming ​

ProgressReporter:

  • Emits structured events for agent lifecycle, tool calls, LLM calls, results, and UI rendering.
  • Enables real-time SSE updates via a queue.

SSE Streaming:

  • Converts incremental pipeline chunks into OpenAI-compatible SSE responses.
  • Handles tool_use/tool_result interleaving and ensures [DONE] termination.

Section sources

Dependency Analysis ​

Key dependencies and relationships:

  • MainPipeline depends on ManagerAgent, DAGExecutor, Reflector, and ProgressReporter.
  • DAGExecutor depends on agent_map and ProgressReporter; agents are resolved by name.
  • ManagerAgent depends on agent registry and schemas for planning.
  • ConversationManager constructs agents and injects tools; it also loads memory.
  • SSE streaming depends on ProgressReporter and pipeline output.

Diagram sources

Section sources

Performance Considerations ​

  • Concurrency control: DAGExecutor serializes tasks per agent within a batch to avoid contention, while allowing parallelism across agents.
  • Deadlock detection: If no ready tasks are found, the executor forces execution of remaining tasks to prevent stalls.
  • Memory hygiene: WorkflowContext artifacts are scoped to the current execution; results are keyed by task_id to avoid collisions.
  • Caching: ManagerAgent caches LLM decisions to reduce repeated planning overhead.
  • Streaming efficiency: SSE generator minimizes latency by yielding incremental chunks and sending a single [DONE] terminator.
  • Memory loading: ConversationManager loads only recent rounds up to a compression threshold to avoid oversized contexts.

[No sources needed since this section provides general guidance]

Troubleshooting Guide ​

Common issues and strategies:

  • Guard rejection or clarification: MainPipeline returns explicit messages via SSE when GuardAction is REJECT or CLARIFY. Inspect guard metadata and reasons.
  • Missing dependencies: DAGExecutor logs warnings when dependent results are missing; ensure upstream tasks are scheduled before downstream ones.
  • Incomplete tool call sequences: ConversationManager sanitizes histories to remove orphaned tool messages and strip incomplete tool_calls to avoid model errors.
  • Reflection failures: Reflector falls back to satisfied=true to prevent infinite loops; review reflection feedback for actionable improvements.
  • SSE termination: The SSE generator guarantees [DONE]; if missing, verify the generator’s finally block and error handling.

Section sources

Conclusion ​

The MainPipeline workflow orchestration system provides a robust, extensible framework for multi-agent analytics. It enforces strict routing policies, supports iterative planning and reflection, and delivers real-time progress via SSE. The integration with ConversationManager ensures persistent, context-rich sessions, while the DAGExecutor maximizes throughput under dependency constraints. Together, these components enable complex, reliable, and transparent analytical workflows.