The text processing pipeline in bettafish uses a four-stage workflow to clean, extract, repair, and parse LLM outputs into structured JSON across QueryEngine, MediaEngine, and InsightEngine modules.

The bettafish repository implements a deterministic text processing pipeline that transforms raw LLM outputs into clean, structured data for downstream engine nodes. This shared workflow ensures consistent handling of JSON payloads across the QueryEngine, MediaEngine, and InsightEngine modules. The pipeline centers on helper functions located in each engine's utils/text_processing.py file, providing robust text cleaning and JSON repair capabilities that node classes rely on to process search queries, summaries, and reflections.

The Four Stages of the Text Processing Pipeline

The text processing pipeline follows a strict sequence to handle the unpredictable nature of LLM outputs. Each stage addresses specific formatting challenges before the data reaches business logic.

Stage 1: Strip Formatting and Reasoning Text

The pipeline first removes Markdown code fences and explanatory prose that LLMs often prepend to JSON payloads. The clean_json_tags() function in QueryEngine/utils/text_processing.py strips specific ```json and ``` fences, while clean_markdown_tags() handles generic Markdown blocks in the MediaEngine implementation. Simultaneously, remove_reasoning_from_output() locates the first { or [ character that signals the start of a JSON payload, discarding any preceding "reasoning" or "analysis" sentences.

Stage 2: Extract the JSON Payload

After cleaning, the pipeline attempts to isolate the valid JSON structure. The extract_clean_response() function orchestrates this stage by calling the cleaning functions sequentially, then attempting direct parsing with json.loads(). If standard parsing succeeds, the function returns the Python dictionary immediately. This extraction logic ensures that even when LLMs wrap JSON in multiple layers of formatting or explanatory text, the pipeline can isolate the machine-readable payload.

Stage 3: Repair Malformed JSON

When direct parsing fails, the pipeline enters repair mode using two levels of heuristics. The fix_incomplete_json() function performs lightweight fixes: removing stray commas, balancing braces and brackets, and optionally wrapping stray objects in an array. If lightweight repair fails, fix_aggressive_json() extracts every { … } fragment from the text and constructs a valid JSON array from these fragments. These repair functions handle common LLM output errors such as truncated responses, unclosed braces, or concatenated JSON objects without proper array syntax.

Stage 4: Parse to Python Dictionary

The final stage converts the cleaned and repaired JSON string into a native Python dict using json.loads(). The extract_clean_response() function returns this dictionary to the calling node, which then accesses specific fields such as search_query, reasoning, or paragraph_latest_state to perform its business logic. This parsing stage completes the pipeline's transformation from raw LLM text to structured data that the bettafish system can reliably consume.

Key Helper Functions and File Locations

The text processing pipeline relies on a consistent set of utility functions replicated across all three engine modules:

Function	Purpose	Source Location
clean_json_tags(text)	Removes ```json and ``` code fences	QueryEngine/utils/text_processing.py
clean_markdown_tags(text)	Removes generic Markdown code fences	MediaEngine/utils/text_processing.py
remove_reasoning_from_output(text)	Discards explanatory text before JSON payload	QueryEngine/utils/text_processing.py
extract_clean_response(text)	Orchestrates full cleaning and extraction workflow	QueryEngine/utils/text_processing.py
fix_incomplete_json(text)	Lightweight JSON repair (commas, braces)	QueryEngine/utils/text_processing.py
fix_aggressive_json(text)	Aggressive fragment extraction and array building	QueryEngine/utils/text_processing.py
format_search_results_for_prompt(search_results, max_length)	Truncates search results for LLM prompt injection	QueryEngine/utils/text_processing.py
truncate_content(content, max_length)	Cuts text at word boundaries	QueryEngine/utils/text_processing.py

All three engine groups—QueryEngine, MediaEngine, and InsightEngine—maintain identical implementations of these functions within their respective utils/text_processing.py files, ensuring consistent behavior across the bettafish system.

Pipeline Implementation in Engine Nodes

The engine nodes consume the text processing pipeline through direct function calls to handle LLM outputs specific to their domain.

Search Nodes (QueryEngine Example)

In QueryEngine/nodes/search_node.py, classes like FirstSearchNode and ReflectionNode process raw LLM outputs to extract structured search queries:

# QueryEngine/nodes/search_node.py

cleaned_output = remove_reasoning_from_output(output)
cleaned_output = clean_json_tags(cleaned_output)

# Try normal JSON parsing first

try:
    result = json.loads(cleaned_output)
except JSONDecodeError:
    # Fallback to the robust extractor

    result = extract_clean_response(cleaned_output)

If parsing still fails, the node runs fix_incomplete_json() and, on success, re-parses the repaired string. The final result dictionary contains the fields search_query and reasoning that downstream nodes consume.

Summary Nodes (QueryEngine Example)

In QueryEngine/nodes/summary_node.py, classes like FirstSummaryNode and ReflectionSummaryNode handle paragraph generation outputs:

cleaned_output = remove_reasoning_from_output(output)
cleaned_output = clean_json_tags(cleaned_output)

try:
    result = json.loads(cleaned_output)
except JSONDecodeError:
    # Attempt self-repair

    fixed_json = fix_incomplete_json(cleaned_output)
    result = json.loads(fixed_json) if fixed_json else cleaned_output

The node then extracts paragraph_latest_state (or updated_paragraph_latest_state) from the parsed dict and writes it back into the global State object.

MediaEngine and InsightEngine Consistency

Both MediaEngine and InsightEngine follow the exact same pattern—simply importing their local utils.text_processing modules. This design guarantees consistent handling of LLM output across all engines, regardless of the specific prompt or task domain.

Practical Code Example

You can leverage the text processing pipeline directly in custom nodes or scripts:

from QueryEngine.utils.text_processing import extract_clean_response

raw_llm_output = """
Here is the reasoning for the query:

```json
{
  "search_query": "latest advances in quantum computing",
  "reasoning": "The field has seen several breakthroughs in error correction."
}

"""

payload = extract_clean_response(raw_llm_output)

print(payload["search_query"])

→ latest advances in quantum computing

This snippet works identically whether imported from `MediaEngine` or `InsightEngine`—just import the respective module's `text_processing` utilities.

## Summary

- The **text processing pipeline** in bettafish follows a four-stage workflow: strip formatting, extract JSON, repair malformed syntax, and parse to Python dictionaries.
- **Helper functions** like `clean_json_tags()`, `remove_reasoning_from_output()`, and `extract_clean_response()` live in each engine's [`utils/text_processing.py`](https://github.com/666ghj/bettafish/blob/main/utils/text_processing.py) file, ensuring modular consistency.
- **Engine nodes** such as `FirstSearchNode` and `FirstSummaryNode` call these utilities to handle raw LLM outputs, with fallback logic that attempts `fix_incomplete_json()` and `fix_aggressive_json()` when standard parsing fails.
- The identical implementation across **QueryEngine**, **MediaEngine**, and **InsightEngine** guarantees that all LLM interactions produce reliably structured data regardless of the specific task domain.

## Frequently Asked Questions

### What engines use the text processing pipeline?

The text processing pipeline is implemented across three engine modules in the bettafish repository: **QueryEngine**, **MediaEngine**, and **InsightEngine**. Each engine maintains an identical copy of the utility functions in its respective [`utils/text_processing.py`](https://github.com/666ghj/bettafish/blob/main/utils/text_processing.py) file, ensuring consistent JSON cleaning and repair logic whether processing search queries, media analysis, or insight generation tasks.

### How does the pipeline handle malformed JSON from LLMs?

When standard `json.loads()` fails, the pipeline activates a two-tier repair system. First, `fix_incomplete_json()` applies lightweight heuristics to remove stray commas and balance braces or brackets. If that fails, `fix_aggressive_json()` extracts every `{ … }` fragment from the text and constructs a valid JSON array from these fragments. This robust fallback mechanism ensures that even truncated or syntactically broken LLM outputs can be converted into usable Python dictionaries.

### Where are the text processing utilities located in the codebase?

The core text processing utilities reside in [`utils/text_processing.py`](https://github.com/666ghj/bettafish/blob/main/utils/text_processing.py) within each engine module. Specifically, you will find the implementations at [`QueryEngine/utils/text_processing.py`](https://github.com/666ghj/bettafish/blob/main/QueryEngine/utils/text_processing.py), [`MediaEngine/utils/text_processing.py`](https://github.com/666ghj/bettafish/blob/main/MediaEngine/utils/text_processing.py), and [`InsightEngine/utils/text_processing.py`](https://github.com/666ghj/bettafish/blob/main/InsightEngine/utils/text_processing.py). Additionally, the pipeline is consumed by node classes located in files such as [`QueryEngine/nodes/search_node.py`](https://github.com/666ghj/bettafish/blob/main/QueryEngine/nodes/search_node.py) and [`QueryEngine/nodes/summary_node.py`](https://github.com/666ghj/bettafish/blob/main/QueryEngine/nodes/summary_node.py).

### Can I use the pipeline functions outside of the engine nodes?

Yes, the text processing functions are designed as standalone utilities that can be imported and used in any Python script or custom node implementation. Simply import the specific functions you need from the respective engine's utility module, such as `from QueryEngine.utils.text_processing import extract_clean_response`. This allows you to leverage the same JSON cleaning, extraction, and repair logic used by the built-in search and summary nodes for your own LLM output processing tasks.

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