A Perplexity-Style Local Search Agent

1 START -> search -> filter -> fetch -> summarize -> synthesize -> END

1 $ cd source-code/local_search
2 $ uv sync
3 $ ollama pull qwen3.5:4b

  1 from typing import TypedDict
  2 
  3 import trafilatura
  4 from ddgs import DDGS
  5 from langchain_ollama import ChatOllama
  6 from langgraph.graph import END, START, StateGraph
  7 
  8 model = ChatOllama(model="qwen3.5:4b", temperature=0)
  9 
 10 MAX_RESULTS = 5
 11 MAX_PAGE_CHARS = 6000
 12 
 13 
 14 class State(TypedDict):
 15     query: str
 16     raw_results: list[dict]
 17     filtered_results: list[dict]
 18     pages: list[dict]
 19     summaries: list[str]
 20     final_answer: str
 21 
 22 
 23 def search_node(state: State) -> dict:
 24     try:
 25         results = list(DDGS().text(state["query"], max_results=MAX_RESULTS))
 26     except Exception as exc:
 27         results = []
 28         print(f"  [search failed: {exc}]")
 29     return {"raw_results": results}
 30 
 31 
 32 def filter_node(state: State) -> dict:
 33     kept = []
 34     for r in state["raw_results"]:
 35         snippet = r.get("body", "")
 36         prompt = (
 37             "Reply with a single character, either Y or N. "
 38             f"Is the following snippet relevant to the query {state['query']!r}?\n\n"
 39             f"{snippet}"
 40         )
 41         answer = model.invoke(prompt).content.strip().upper()
 42         if answer.startswith("Y"):
 43             kept.append(r)
 44     return {"filtered_results": kept}
 45 
 46 
 47 def fetch_node(state: State) -> dict:
 48     pages = []
 49     for r in state["filtered_results"]:
 50         url = r.get("href") or r.get("url")
 51         if not url:
 52             continue
 53         try:
 54             downloaded = trafilatura.fetch_url(url)
 55             text = trafilatura.extract(downloaded) or ""
 56         except Exception:
 57             text = ""
 58         if text:
 59             pages.append({"url": url, "title": r.get("title", ""), "text": text[:MAX_PAGE_CHARS]})
 60     return {"pages": pages}
 61 
 62 
 63 def summarize_node(state: State) -> dict:
 64     summaries = []
 65     for p in state["pages"]:
 66         prompt = (
 67             f"Summarize the following text, including only material relevant to the "
 68             f"query {state['query']!r}. Keep it to at most three sentences.\n\n"
 69             f"{p['text']}"
 70         )
 71         summaries.append(model.invoke(prompt).content.strip())
 72     return {"summaries": summaries}
 73 
 74 
 75 def synthesize_node(state: State) -> dict:
 76     if not state["summaries"]:
 77         return {"final_answer": "No usable sources were found."}
 78 
 79     joined = "\n\n---\n\n".join(state["summaries"])
 80     prompt = (
 81         f"Using the following per-source summaries, write a clear, multi-paragraph "
 82         f"answer to the query {state['query']!r}. Do not repeat information across "
 83         f"paragraphs. Do not include a list of sources.\n\n{joined}"
 84     )
 85     return {"final_answer": model.invoke(prompt).content.strip()}
 86 
 87 
 88 def build_pipeline():
 89     graph = StateGraph(State)
 90     graph.add_node("search", search_node)
 91     graph.add_node("filter", filter_node)
 92     graph.add_node("fetch", fetch_node)
 93     graph.add_node("summarize", summarize_node)
 94     graph.add_node("synthesize", synthesize_node)
 95 
 96     graph.add_edge(START, "search")
 97     graph.add_edge("search", "filter")
 98     graph.add_edge("filter", "fetch")
 99     graph.add_edge("fetch", "summarize")
100     graph.add_edge("summarize", "synthesize")
101     graph.add_edge("synthesize", END)
102 
103     return graph.compile()

 1 from _pipeline import build_pipeline
 2 
 3 app = build_pipeline()
 4 
 5 query = "What are the main challenges in running large language models on consumer laptops?"
 6 
 7 initial: dict = {
 8     "query": query,
 9     "raw_results": [],
10     "filtered_results": [],
11     "pages": [],
12     "summaries": [],
13     "final_answer": "",
14 }
15 
16 final = app.invoke(initial)
17 
18 print("=== FINAL ANSWER ===")
19 print(final["final_answer"])

 1 $ uv run 01_search.py
 2 USER: What are the main challenges in running large language models on consumer laptops?
 3 
 4 === FINAL ANSWER ===
 5 Consumer laptops face three main obstacles when running large language models
 6 locally. The first is memory: even quantized models in the 7-13 billion
 7 parameter range typically need 8-16 GB of RAM just to load, and models beyond
 8 30 B parameters are effectively out of reach for anything under 32 GB.
 9 
10 The second is inference speed. On CPU-only laptops or those without a
11 compatible GPU, token generation rates for a mid-sized model can be under
12 five tokens per second, which is too slow for interactive use. Apple Silicon
13 and NVIDIA GPUs bring this into a usable range, but at the cost of hardware
14 compatibility ...
15 
16 ... The third challenge is thermal management. Sustained inference at high
17 GPU utilization pushes laptop cooling systems close to their limits, causing
18 throttling that can cut throughput in half after a few minutes of continuous
19 generation.

 1 for step in app.stream(initial):
 2     for node_name, node_output in step.items():
 3         print(f"=== node: {node_name} ===")
 4         if node_name == "search":
 5             for r in node_output["raw_results"]:
 6                 print(f"  - {r.get('title', '')}")
 7                 print(f"    {r.get('href') or r.get('url')}")
 8         elif node_name == "filter":
 9             print(f"  kept {len(node_output['filtered_results'])} results")
10         elif node_name == "fetch":
11             for p in node_output["pages"]:
12                 print(f"  fetched {len(p['text'])} chars from {p['url']}")
13         elif node_name == "summarize":
14             for i, s in enumerate(node_output["summaries"]):
15                 snippet = s if len(s) < 200 else s[:200] + "..."
16                 print(f"  [{i}] {snippet}")
17         elif node_name == "synthesize":
18             print(node_output["final_answer"])
19         print()