# Node-Based Processing Architecture in BettaFish: How the Four Engines Orchestrate AI Pipelines

> Explore BettaFish's node-based processing architecture. Discover how four specialized engines orchestrate AI data pipelines using BaseNode and shared mutable state.

- Repository: [BaiFu/bettafish](https://github.com/666ghj/bettafish)
- Tags: architecture
- Published: 2026-02-23

---

**BettaFish implements a unified node-based processing architecture where specialized engines orchestrate pipelines of nodes inheriting from `BaseNode`, transforming data through a shared mutable state object.**

The BettaFish repository (666ghj/bettafish) is an open-source AI reporting framework that processes complex queries through composable execution pipelines. Its node-based processing architecture enables four specialized engines—Report, Query, Media, and Insight—to execute sophisticated workflows using reusable, testable components that share state across processing stages.

## Core Node Abstractions

At the foundation of every engine lies the **`BaseNode`** class. According to the source code in paths like [`ReportEngine/nodes/base_node.py`](https://github.com/666ghj/bettafish/blob/main/ReportEngine/nodes/base_node.py) and [`QueryEngine/nodes/base_node.py`](https://github.com/666ghj/bettafish/blob/main/QueryEngine/nodes/base_node.py), this abstract base class defines the contract that all processing units must follow.

Each node implements a **`run(input, **kwargs)`** method containing the core business logic. The architecture also provides optional hooks for input validation (`validate_input`), output processing (`process_output`), and unified logging (`log_info`, `log_warning`, `log_error`).

For engines requiring state modification, the **`StateMutationNode`** extends `BaseNode` with an abstract **`mutate_state`** method. This allows nodes to modify engine-specific state objects—such as `ReportState` in the Report Engine or the generic `State` in other engines—enabling downstream nodes to access intermediate results.

## Report Engine Node Pipeline

The Report Engine demonstrates the architecture's modularity through specialized document generation nodes located in `ReportEngine/nodes/`.

Key components include:

- **`ChapterGenerationNode`** ([`ReportEngine/nodes/chapter_generation_node.py`](https://github.com/666ghj/bettafish/blob/main/ReportEngine/nodes/chapter_generation_node.py)): Drafts individual report chapters using LLM clients.
- **`TemplateSelectionNode`**: Selects appropriate document templates based on query analysis.
- **`WordBudgetNode`**: Enforces word-count constraints across document sections.

The engine orchestrator, referenced in [`ReportEngine/utils/dependency_check.py`](https://github.com/666ghj/bettafish/blob/main/ReportEngine/utils/dependency_check.py) and implemented in [`ReportEngine/agent.py`](https://github.com/666ghj/bettafish/blob/main/ReportEngine/agent.py), instantiates these nodes and chains them sequentially. When the CLI entry point [`report_engine_only.py`](https://github.com/666ghj/bettafish/blob/main/report_engine_only.py) executes, it creates a `ReportState` object and passes it through the node chain—typically `TemplateSelectionNode → ChapterGenerationNode → WordBudgetNode`—with each node optionally mutating the shared state.

## Query Engine Node Pipeline

Located in `QueryEngine/`, this engine processes search queries through a reflective research pipeline. The orchestrator in [`QueryEngine/query_engine.py`](https://github.com/666ghj/bettafish/blob/main/QueryEngine/query_engine.py) wires together nodes defined in [`QueryEngine/nodes/search_node.py`](https://github.com/666ghj/bettafish/blob/main/QueryEngine/nodes/search_node.py) and related files.

Concrete implementations include:

- **`FirstSearchNode`**: Generates initial search queries and retrieves raw results.
- **`ReflectionNode`**: Performs reflective re-research to fill knowledge gaps.
- **`ReportStructureNode`**: Builds the hierarchical skeleton of the final report.
- **`ReportFormattingNode`**: Converts structured content into formatted markdown.

The engine maintains a shared **`State`** instance that persists across the pipeline, allowing the `ReflectionNode` to access initial search results and the `ReportFormattingNode` to consume the final structured content.

## Media Engine Node Pipeline

The Media Engine, with its base node class in [`MediaEngine/nodes/base_node.py`](https://github.com/666ghj/bettafish/blob/main/MediaEngine/nodes/base_node.py), adapts the architecture for multimedia processing. Node implementations in [`MediaEngine/nodes/search_node.py`](https://github.com/666ghj/bettafish/blob/main/MediaEngine/nodes/search_node.py) handle media-specific artifacts.

Key nodes include:

- **`FirstSearchNode`**: Conducts media-centric searches across image and video sources.
- **`SummaryNode`**: Generates concise summaries of media transcripts and captions.
- **`ReportStructureNode`** and **`FormattingNode`**: Assemble and format the final media report.

As shown in [`SingleEngineApp/media_engine_streamlit_app.py`](https://github.com/666ghj/bettafish/blob/main/SingleEngineApp/media_engine_streamlit_app.py), the engine executes this pipeline using a `MediaAgent` that manages the shared `State` object capturing media URLs, transcripts, and processed content.

## Insight Engine Node Pipeline

The Insight Engine follows an identical pattern with its base definitions in [`InsightEngine/nodes/base_node.py`](https://github.com/666ghj/bettafish/blob/main/InsightEngine/nodes/base_node.py) and concrete implementations in [`InsightEngine/nodes/search_node.py`](https://github.com/666ghj/bettafish/blob/main/InsightEngine/nodes/search_node.py). This engine focuses on generating insight-driven analytical reports.

The node chain comprises:

- **`FirstSearchNode`** and **`ReflectionNode`**: Generate queries and reflect on findings to deepen analysis.
- **`ReportStructureNode`**: Composes the insight narrative structure.
- **`FormattingNode`**: Produces the final markdown or HTML output.

The entry point in [`SingleEngineApp/insight_engine_streamlit_app.py`](https://github.com/666ghj/bettafish/blob/main/SingleEngineApp/insight_engine_streamlit_app.py) demonstrates how the engine chains these nodes to transform raw queries into structured insights, maintaining state throughout the pipeline execution.

## Practical Implementation Examples

### Running a Report Engine Pipeline

The following example from [`report_engine_only.py`](https://github.com/666ghj/bettafish/blob/main/report_engine_only.py) demonstrates engine initialization:

```python
from ReportEngine.utils.dependency_check import check_pango_available
from ReportEngine.agent import ReportAgent
from ReportEngine.llms.base import LLMClient

# Initialize LLM client

llm = LLMClient(model_name="gpt-4o-mini")

# Create the engine agent with internal node chain

report_agent = ReportAgent(llm_client=llm)

# Execute the pipeline

final_report = report_agent.run(prompt="请生成关于《人工智能伦理》的报告")
print(final_report)

```

### Direct Query Engine Usage

For programmatic access to the Query Engine as implemented in [`QueryEngine/query_engine.py`](https://github.com/666ghj/bettafish/blob/main/QueryEngine/query_engine.py):

```python
from QueryEngine.query_engine import QueryEngine
from QueryEngine.utils.state import State
from QueryEngine.llms.base import LLMClient

llm = LLMClient(model_name="gpt-4o-mini")
engine = QueryEngine(llm_client=llm)

# Initialize empty state

state = State()

# Run the node pipeline

engine.run(state, user_query="2024 年中国新能源政策趋势")
print(state.report)  # Access formatted markdown report

```

### Media Engine Streamlit Integration

Example usage from [`SingleEngineApp/media_engine_streamlit_app.py`](https://github.com/666ghj/bettafish/blob/main/SingleEngineApp/media_engine_streamlit_app.py):

```python
import streamlit as st
from MediaEngine.agent import MediaAgent
from MediaEngine.llms.base import LLMClient

llm = LLMClient(model_name="gpt-4o-mini")
agent = MediaAgent(llm)

query = st.text_input("输入媒体查询")
if st.button("生成报告"):
    report = agent.run(query)
    st.markdown(report)

```

## Summary

- **BettaFish** employs a consistent node-based processing architecture across all four engines (Report, Query, Media, and Insight), enabling modular AI pipeline construction.
- **BaseNode** abstract class in engine-specific paths (e.g., [`ReportEngine/nodes/base_node.py`](https://github.com/666ghj/bettafish/blob/main/ReportEngine/nodes/base_node.py)) defines the `run(input, **kwargs)` interface and logging hooks, while **StateMutationNode** enables state modification.
- **Engine orchestrators** wire nodes sequentially, passing shared state objects that allow downstream nodes to access intermediate processing results.
- **Concrete node implementations** handle specific tasks—such as `ChapterGenerationNode` for document drafting or `ReflectionNode` for iterative research—promoting code reusability and testability.
- **Entry points** range from CLI scripts ([`report_engine_only.py`](https://github.com/666ghj/bettafish/blob/main/report_engine_only.py)) to Streamlit applications ([`SingleEngineApp/media_engine_streamlit_app.py`](https://github.com/666ghj/bettafish/blob/main/SingleEngineApp/media_engine_streamlit_app.py)), all utilizing the same underlying node chaining mechanism.

## Frequently Asked Questions

### How does state management work between nodes in the BettaFish architecture?

Each engine maintains a shared state object—`ReportState` for the Report Engine or `State` for the others—that persists across the node pipeline. Nodes implementing the `mutate_state` method from `StateMutationNode` can modify this object, allowing downstream nodes to access previous results. For example, in [`QueryEngine/query_engine.py`](https://github.com/666ghj/bettafish/blob/main/QueryEngine/query_engine.py), the `State` instance holds raw queries, search results, and intermediate summaries that propagate from `FirstSearchNode` through `ReportFormattingNode`.

### What is the difference between `BaseNode` and `StateMutationNode`?

`BaseNode`, defined in files like [`QueryEngine/nodes/base_node.py`](https://github.com/666ghj/bettafish/blob/main/QueryEngine/nodes/base_node.py), provides the foundational interface including the `run()` method and logging hooks. `StateMutationNode` extends this abstract class specifically for nodes that must modify the engine's mutable state object. While all nodes inherit from `BaseNode`, only those needing to persist data to the shared state implement `StateMutationNode` and its abstract `mutate_state` method.

### Can nodes be reused across different BettaFish engines?

While each engine maintains its own `BaseNode` implementation in engine-specific directories (e.g., [`MediaEngine/nodes/base_node.py`](https://github.com/666ghj/bettafish/blob/main/MediaEngine/nodes/base_node.py)), the architectural pattern remains identical across engines. However, concrete nodes like `ChapterGenerationNode` are typically engine-specific due to specialized state requirements. The shared abstraction allows developers to port logic between engines by adapting the state interface while preserving the `run(input, **kwargs)` contract.

### How does the engine orchestrator handle node execution order?

The engine orchestrator—such as `ReportAgent` in [`ReportEngine/agent.py`](https://github.com/666ghj/bettafish/blob/main/ReportEngine/agent.py) or `QueryEngine` in [`QueryEngine/query_engine.py`](https://github.com/666ghj/bettafish/blob/main/QueryEngine/query_engine.py)—explicitly instantiates nodes in a predefined sequence and executes them sequentially. The output of one node's `run()` method becomes the input to the next, while the shared state object accumulates context throughout the pipeline. This design allows easy reordering or substitution of nodes without modifying the core engine logic.