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

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 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 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 to spawn the respective engine’s entry point—SingleEngineApp/query_engine_streamlit_app.py, media_engine_streamlit_app.py, or 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, MediaEngine/agent.py, and 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) 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:

{
  "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:

python app.py

In a separate terminal, initialize each engine:

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:

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:

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:

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) 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 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. 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 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) defaults to Tavily web search, MediaAgent (in MediaEngine/agent.py) initializes BochaMultimodalSearch, and InsightAgent (in InsightEngine/agent.py) connects to MediaCrawlerDB. These dependencies are configured via the shared 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.

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