Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/JasonHonKL/spy-search/llms.txt

Use this file to discover all available pages before exploring further.

Spy Search’s search pipeline is built on an abstract Agent framework where each agent has a clearly defined role. A central Server drives the loop, passing messages through lightweight Router wrappers until the pipeline reaches a TERMINATE signal and a final report is returned to the caller.

​
The Agent Abstract Base Class

Every agent in the system extends src/agent/agent.py::Agent. The class enforces three abstract methods and provides one shared utility:

​
run(response, data) — main async entry point

Each agent’s core logic lives here. The method receives the current task string as response and a data list that accumulates results across the pipeline. It must return a routing dict (see Data flow format below). 
@abstractmethod
async def run(self, response, data=None):
    pass

​
get_recv_format() and get_send_format()

These methods define the Pydantic BaseModel schemas that describe what data an agent expects to receive and what it promises to return. Concrete agents override them when strict schema validation is needed. 
@abstractmethod
def get_recv_format(self) -> BaseModel:
    pass

@abstractmethod
def get_send_format(self) -> BaseModel:
    pass

​
_extract_response(res) — LLM output parser

A shared utility that reliably extracts a JSON object or Python literal from raw LLM output, regardless of whether the model wrapped it in a markdown code block or emitted plain text. The algorithm:
  1. Searches for ```json … ``` or ``` … ``` blocks first.
  2. Falls back to bracket-matching on {…} and […] substrings.
  3. Tries json.loads first, then ast.literal_eval for single-quoted Python dicts.
  4. Returns the largest valid candidate, or None if nothing parsed.

​
name and description attributes

Every agent instance carries a name (used as the router key) and a description (included in the Planner’s system prompt so the LLM knows which agents are available).

​
Agent Implementations

Planner

File: src/agent/planner.pyThe Planner is the pipeline’s brain. On its first call it builds a planning prompt that includes the user’s query and the name → description map of all registered agents, then asks the LLM to return an ordered task list:
[
  {"task": "search for recent AI benchmarks", "agent": "quick-searcher"},
  {"task": "write final report", "agent": "reporter"}
]
The tasks are loaded into an internal deque. On each subsequent call the Planner pops the next task and routes to the corresponding agent. When the queue is empty it emits {"agent": "TERMINATE", "task": "TERMINATE", "data": [...]}.

Quick_searcher

File: src/agent/quick_searcher.pyProvides fast, low-latency web search by delegating to DuckSearch (DuckDuckGo). For each result it normalises the fields into the shared data schema:
{
    "title": ele["title"],
    "summary": ele.get("full_content", ""),
    "brief_summary": ele["snippet"],
    "keywords": [],
    "url": ele["link"],
}
Results are appended to the shared data list and the agent routes back to planner for the next task.

Search_agent

File: src/agent/search.pyA more thorough research agent that uses the Crawl (crawl4ai / Playwright) backend for JavaScript-rendered pages. Its internal workflow:
  1. _plan(task) — asks the LLM for a tool-call list (url_search then page_content steps).
  2. url_search — calls crawl.get_url_llm(url, query) to extract relevant links from a search engine results page.
  3. page_content — calls crawl.get_summary(urls, query) to fetch each URL and produce a structured {title, summary, brief_summary, keywords, url} object.
Collected summaries accumulate in self.db and are returned to planner when all steps are complete.

Reporter

File: src/agent/reporter.pyTurns accumulated search data into a polished report. Its workflow:
  1. data_handler(data) — assigns each data item a random short ID and extracts summary fields into a compact reference list.
  2. _planner(query, db) — asks the LLM to outline report sections, each referencing the IDs of relevant sources.
  3. _task_handler(tasks) — iterates over sections, fetches the full source records for each, and calls report_task() to write that section with a focused LLM prompt.
  4. Concatenates all sections and returns {"agent": "TERMINATE", "data": <final_report>, "task": ""}.

​
RAG_agent

File: src/agent/retrival.py The local retrieval agent ingests files from a configured directory and answers queries against them using ChromaDB. Workflow:
  1. Walks self.filelist recursively; converts each file to Markdown with markitdown.
  2. Chunks the Markdown into 1 500-character segments and upserts each chunk into ChromaDB with the file path as metadata.
  3. Queries the vector store for the top-2 most relevant chunks.
  4. Passes each chunk through the LLM using retrieval_prompt to generate a structured summary.
  5. Appends results to data and routes back to planner.

​
The Server and Router

The Server (src/router/server.py) and Router (src/router/router.py) are the coordination layer that connects agents without coupling them directly.
  • Server holds a dict of named Router objects. It tracks the current active router, the accumulated data list, and drives the main while True loop in server.start().
  • Router wraps a single Agent. Its recv_response(message, data) method calls agent.run(message, data) and returns the result back to the Server.
The loop in Server.start(): 
async def start(self, query: str):
    self.next_router = self.routers[self.initial_router]
    while True:
        response = await self.next_router.recv_response(query, self.data)
        if self.check_response(response):   # response["agent"] == "TERMINATE"
            return response
        self.next_router, query, self.data = self.query_handler(response)
query_handler extracts the next router from response["agent"], the new task string from response["task"], and the updated data list from response["data"]. The task string — not the full response dict — becomes the query passed to the next agent’s run().

​
Data Flow Format

Every agent must return a dict with exactly three keys:
KeyTypePurpose
agentstrName of the next router to call, or "TERMINATE"
taskstrThe task string forwarded to the next agent’s run()
datalistAccumulated result objects passed through the full pipeline

​
Factory Class

src/factory/factory.py provides two static factory methods that decouple agent and model construction from the rest of the codebase: 
Factory.get_agent(name: str, model: Model) -> Agent
Factory.get_model(provider: str, model: str) -> Model
get_agent maps names to their concrete classes:
NameClass
"planner"Planner
"reporter"Reporter
"searcher"Quick_searcher
"local-retrieval"RAG_agent
get_model maps provider strings to the corresponding Model subclass:
Provider string(s)Class
"deepseek"Deepseek
"google" or "gemini"Gemini
"ollama"Ollama
"xAI" or "gork"Gork
"openai" or "gpt"OpenAI

​
Constructing a Pipeline

The helper function generate_report() in src/generate_report.py shows the canonical pattern for assembling a pipeline: 
from src.generate_report import generate_report
from src.factory.factory import Factory
from src.agent import Planner

model = Factory.get_model("google", "gemini-2.0-flash")

planner = Planner(model)
agents = [
    Factory.get_agent("reporter", Factory.get_model("google", "gemini-2.0-flash")),
    Factory.get_agent("searcher", Factory.get_model("google", "gemini-2.0-flash")),
]

report = await generate_report(query, planner, agents)
Internally generate_report creates one Router per agent (including the planner), registers them all on a shared Server, points the Server at the Planner as the initial router, and calls server.start(query). The function returns the "data" field of the final TERMINATE response — the finished report string.

Build docs developers (and LLMs) love

Get started for freeTalk to us