# How QueryAgent, MediaAgent, and InsightAgent Coordinate in BettaFish’s Deep-Search Pipeline

> Discover how BettaFish coordinates QueryAgent MediaAgent and InsightAgent using a Flask orchestrator and Streamlit subprocesses. Learn about log aggregation and JSON merging in this deep search pipeline.

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

---

**BettaFish coordinates these three agents through a Flask orchestrator that launches each engine as an independent Streamlit subprocess, aggregates their real-time logs via Socket.IO, and merges their JSON responses through a unified `/api/search` endpoint.**

The **BettaFish** repository implements a multi-agent deep-search architecture where **QueryAgent** (QueryEngine), **MediaAgent** (MediaEngine), and **InsightAgent** (InsightEngine) execute identical reasoning pipelines but operate as isolated workers. Understanding their coordination requires examining the Flask-based orchestration layer in [`app.py`](https://github.com/666ghj/bettafish/blob/main/app.py) and the shared node-based pipeline defined in each agent’s core module.

## Architecture Overview

The system decouples agent execution from coordination logic. While each agent runs a standalone Streamlit process, the Flask server manages their lifecycle, collects their outputs, and presents a single API surface to clients.

### The Flask Orchestrator Role

The Flask application in [`app.py`](https://github.com/666ghj/bettafish/blob/main/app.py) serves as the central nervous system. It exposes endpoints to start, stop, and query the three agents, ensuring they behave as a cohesive unit despite running in separate Python processes. The orchestrator handles three critical duties: subprocess management, log aggregation, and request fan-out.

### Independent Streamlit Workers

Each agent runs inside its own Streamlit environment on a dedicated port (typically 8501–8503). When the Flask server receives a start command, it invokes `start_streamlit_app` in [`app.py`](https://github.com/666ghj/bettafish/blob/main/app.py) to spawn the respective engine’s entry point—[`SingleEngineApp/query_engine_streamlit_app.py`](https://github.com/666ghj/bettafish/blob/main/SingleEngineApp/query_engine_streamlit_app.py), [`media_engine_streamlit_app.py`](https://github.com/666ghj/bettafish/blob/main/media_engine_streamlit_app.py), or [`insight_engine_streamlit_app.py`](https://github.com/666ghj/bettafish/blob/main/insight_engine_streamlit_app.py). These wrappers expose the core `DeepSearchAgent.research()` method via a local HTTP API that the orchestrator can call.

## The Shared Four-Stage Pipeline

All three agents implement the identical `DeepSearchAgent` class pattern found in [`QueryEngine/agent.py`](https://github.com/666ghj/bettafish/blob/main/QueryEngine/agent.py), [`MediaEngine/agent.py`](https://github.com/666ghj/bettafish/blob/main/MediaEngine/agent.py), and [`InsightEngine/agent.py`](https://github.com/666ghj/bettafish/blob/main/InsightEngine/agent.py). They follow a five-step reasoning flow (mapped to four conceptual stages) that processes a user query into a structured markdown report.

### Stage 1: Report Structure Generation

The pipeline begins with `DeepSearchAgent._generate_report_structure` (QueryEngine lines 182–191; Media/Insight lines 74–84). This method instantiates a **ReportStructureNode** that queries the LLM to generate a skeleton of paragraphs for the final report, establishing the analytical framework before any search occurs.

### Stage 2: Paragraph Processing and Iteration

Next, `_process_paragraphs` (lines 199–215) iterates over the generated structure. For each paragraph, it triggers the initial search and summary phase, then enters the reflection loop. This iteration happens sequentially within each agent but runs in parallel across the three independent agent processes.

### Stage 3: Search Execution and Summarization

The `_initial_search_and_summary` method (lines 217–277) orchestrates the first evidence-gathering phase:

1. **FirstSearchNode** crafts a search query and optionally selects a tool name.
2. `execute_search_tool` dispatches the query to the agent-specific backend (Tavily for QueryAgent, Bocha for MediaAgent, or MediaCrawlerDB for InsightAgent).
3. `format_search_results_for_prompt` sanitizes raw results for the LLM context window.
4. **FirstSummaryNode** generates the initial paragraph draft based on the retrieved evidence.

### Stage 4: Reflection and Refinement

The `_reflection_loop` (lines 311–398) implements iterative refinement. It prompts the LLM to critique the current summary, generates a **ReflectionNode** query, executes another search, and passes new findings to **ReflectionSummaryNode**. This cycle repeats up to `MAX_REFLECTIONS` times (configured in [`config.py`](https://github.com/666ghj/bettafish/blob/main/config.py)) until the content meets quality thresholds or the iteration limit is reached.

### Stage 5: Final Report Assembly

Finally, `_generate_final_report` (lines 399–425) invokes **ReportFormattingNode** to concatenate all paragraph summaries into a cohesive markdown document, which is then returned to the Flask orchestrator.

## Cross-Agent Coordination Mechanisms

Coordination happens exclusively at the Flask layer; the agents themselves are agnostic of each other’s existence. The orchestrator employs three specific mechanisms to synchronize their operations.

### Subprocess Management and Health Monitoring

The `/api/start/<app_name>` endpoint triggers `start_streamlit_app`, which launches the target engine as a subprocess. The Flask server maintains process handles for each agent, enabling health checks and graceful shutdowns. Because each agent writes to its own log stream, crashes in one engine do not cascade to the others.

### Unified Logging and Real-Time Monitoring

All three agents write human-readable status lines (e.g., "`[步骤 1] 生成报告结构...`") to `logs/forum.log` with prefixed identifiers. The Flask server runs `monitor_forum_log` in a background thread, which tails this file and invokes `parse_forum_log_line` to detect lines starting with `[QUERY]`, `[MEDIA]`, or `[INSIGHT]`. Parsed messages are pushed to the front-end via Socket.IO, providing a unified live view of all agents’ progress in a single UI panel.

### Request Fan-Out and Response Aggregation

When the front-end calls `/api/search` with a user query, the Flask route forwards the request via `requests.post` to every running engine’s local API (ports 8501–8503). Each agent executes its full pipeline independently and returns a JSON payload. The orchestrator merges these into a single response object:

```json
{
  "success": true,
  "query": "用户查询内容",
  "results": {
    "query": {"success": true, "report": "..."},
    "media": {"success": true, "report": "..."},
    "insight": {"success": true, "report": "..."}
  }
}

```

This pattern allows the client to receive a consolidated view of textual analysis (QueryAgent), multimodal content (MediaAgent), and database insights (InsightAgent) without managing three separate connections.

## Tool Specialization per Agent

Despite sharing identical pipeline logic in `DeepSearchAgent`, the three engines diverge in their search tool selection:

- **QueryAgent** leverages **Tavily** for general web and news search.
- **MediaAgent** utilizes **BochaMultimodalSearch** to retrieve images and video alongside text.
- **InsightAgent** queries **MediaCrawlerDB** for structured local data, optionally applying clustering and sentiment analysis nodes.

This specialization is injected via the `execute_search_tool` dispatcher, which selects the appropriate backend based on the agent module’s configuration while preserving the same node orchestration flow.

## Practical Code Examples

### Starting the Coordination Layer

Launch the Flask orchestrator to manage all three agents:

```bash
python app.py

```

In a separate terminal, initialize each engine:

```bash
curl http://localhost:5000/api/start/query
curl http://localhost:5000/api/start/media
curl http://localhost:5000/api/start/insight

```

### Running a Single Agent Directly

For debugging or isolated analysis, instantiate an agent directly without the Flask layer:

```python
from QueryEngine.agent import create_agent

# Initialize QueryAgent (loads settings from .env)

query_agent = create_agent()

# Execute deep-search pipeline

report_md = query_agent.research("全球气候变化的最新进展")
print(report_md[:500])  # Preview first 500 characters

```

The same pattern applies to `MediaEngine.agent.create_agent()` and `InsightEngine.agent.create_agent()`.

### Sending Unified Search Requests

Submit a query to all running agents simultaneously:

```bash
curl -X POST http://localhost:5000/api/search \
     -H "Content-Type: application/json" \
     -d '{"query":"2025 年中国人工智能政策"}'

```

### Real-Time Log Streaming

Listen to coordinated agent progress via Socket.IO:

```javascript
const socket = io('http://localhost:5000');

socket.on('forum_message', (msg) => {
  // msg contains: {type:"agent", sender:"Query Engine", content:"[步骤 2] ..."}
  console.log(`${msg.sender}: ${msg.content}`);
});

```

## Summary

- **BettaFish** coordinates **QueryAgent**, **MediaAgent**, and **InsightAgent** through a Flask orchestrator ([`app.py`](https://github.com/666ghj/bettafish/blob/main/app.py)) that manages subprocess lifecycle and aggregates outputs.
- All three agents implement the identical `DeepSearchAgent` pipeline (`_generate_report_structure`, `_process_paragraphs`, `_initial_search_and_summary`, `_reflection_loop`, `_generate_final_report`) found in their respective [`agent.py`](https://github.com/666ghj/bettafish/blob/main/agent.py) files.
- The orchestrator uses **Socket.IO** to stream prefixed log lines (`[QUERY]`, `[MEDIA]`, `[INSIGHT]`) from `logs/forum.log` to the UI, providing real-time visibility.
- The `/api/search` endpoint fans out requests to each agent’s Streamlit API (ports 8501–8503) and merges JSON responses into a unified payload.
- Agents differ only in their search tool backends (Tavily, BochaMultimodalSearch, MediaCrawlerDB), while sharing node implementations like `ReportStructureNode` and `ReflectionNode`.

## Frequently Asked Questions

### How does BettaFish handle failures in one agent without affecting the others?

Each agent runs as an isolated subprocess managed by `start_streamlit_app` in [`app.py`](https://github.com/666ghj/bettafish/blob/main/app.py). If one engine crashes or times out, the Flask orchestrator detects the failed HTTP connection during the `/api/search` request fan-out and includes an error state for that specific agent in the aggregated JSON response, allowing the other two agents’ results to return successfully.

### Can the three agents share intermediate search results to avoid duplicate API calls?

No, the current architecture in `666ghj/bettafish` keeps agents completely isolated. Each maintains its own state in `DeepSearchAgent` and executes searches independently. The coordination layer only merges final reports, not intermediate `FirstSearchNode` or `ReflectionNode` outputs. Shared caching would require modifications to [`utils/knowledge_logger.py`](https://github.com/666ghj/bettafish/blob/main/utils/knowledge_logger.py) or a shared Redis layer.

### What determines which search tools each agent uses?

The tool selection is hardcoded in each engine’s `execute_search_tool` implementation. **QueryAgent** (in [`QueryEngine/agent.py`](https://github.com/666ghj/bettafish/blob/main/QueryEngine/agent.py)) defaults to Tavily web search, **MediaAgent** (in [`MediaEngine/agent.py`](https://github.com/666ghj/bettafish/blob/main/MediaEngine/agent.py)) initializes **BochaMultimodalSearch**, and **InsightAgent** (in [`InsightEngine/agent.py`](https://github.com/666ghj/bettafish/blob/main/InsightEngine/agent.py)) connects to **MediaCrawlerDB**. These dependencies are configured via the shared [`config.py`](https://github.com/666ghj/bettafish/blob/main/config.py) Pydantic `Settings` class but resolve to different API keys and endpoints.

### Is it possible to run only specific agents instead of all three?

Yes. The `/api/start/<app_name>` endpoint allows selective initialization. You can start only `query` and `insight` while omitting `media`. Subsequent calls to `/api/search` will only forward requests to the running agents; inactive ports are skipped in the fan-out loop defined in the Flask `search` route.