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

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 and 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): 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 and implemented in ReportEngine/agent.py, instantiates these nodes and chains them sequentially. When the CLI entry point 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 wires together nodes defined in 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, adapts the architecture for multimedia processing. Node implementations in 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, 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 and concrete implementations in 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 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 demonstrates engine initialization:

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:

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:

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) 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) to Streamlit applications (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, 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, 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), 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 or QueryEngine in 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}