research_loop

research_loop

Agentic research loop over the hybrid-search tool.

A small, self-contained planner-executor loop:

• the planner is supplied by the caller (the web endpoint resolves it from config, falling back to gemma4:31b on Ollama for legacy installs),
• it can call :meth:HybridSearch.search over the corpus and — when the caller wires one — a web_search over the public web (22 Sept 2026: « Fais-moi une recherche [sur] des sites web qui proposent des jeux de programmation » under Deep Research got « your corpus has nothing », then a list from the model's memory with an invented site; the follow-up « Donne-moi les liens de ces sites » got links pulled from the user's e-mails — the request was stateless and web-less),
• it receives the previous turns of the conversation (history) so a follow-up like « ces sites » has a referent,
• it gets up to max_iterations tool calls,
• tool results are trimmed before re-entering the context window, and
• the final reply must cite specific hits.

The loop is deliberately decoupled from the rest of the agent scaffolding (ToolUsingAgent, EventBus, AgentContext, etc.) so the surface stays small. Anything that wants tracing or registry integration can wrap it.

Classes

WebHit dataclass

WebHit(
    title: str,
    url: str,
    site: str = "",
    date: str = "",
    snippet: str = "",
)

One public-web result, as the router's web_search callable returns it.

ToolInvocation dataclass

ToolInvocation(
    arguments: Dict[str, Any],
    num_results: int = 0,
    top_titles: List[str] = list(),
    raw_hits: List[SearchHit] = list(),
    tool_name: str = "search",
    response: str = "",
    web_hits: List[WebHit] = list(),
    error: str = "",
)

One tool call together with what the planner asked for and got.

tool_name is "search", "web_search" or "clarify". For search calls, num_results, top_titles and raw_hits are populated (web_hits for the web); for clarify calls, response holds the user's answer.

ResearchAgent

ResearchAgent(
    engine: InferenceEngine,
    search: HybridSearch,
    *,
    model: str = DEFAULT_PLANNER_MODEL,
    max_iterations: int = 5,
    temperature: float = 0.3,
    max_tokens: int = 1500,
    num_ctx: int = 16384,
    clarify_handler: Optional[Callable[[str], str]] = None,
    on_event: Optional[
        Callable[[Dict[str, Any]], None]
    ] = None,
    available_sources: Optional[List[str]] = None,
    web_search: Optional[WebSearcher] = None,
)

Planner + executor loop over a single hybrid-search tool.

PARAMETER	DESCRIPTION
engine	An InferenceEngine that supports OpenAI-style tools in generate (Ollama with a tool-capable model). TYPE: InferenceEngine
search	The HybridSearch instance the planner can call. TYPE: HybridSearch
model	Planner model tag (default gemma4:31b). TYPE: str DEFAULT: DEFAULT_PLANNER_MODEL
max_iterations	Hard ceiling on tool calls before the loop is forced into synthesis. TYPE: int DEFAULT: 5
temperature	Generation parameters passed through to engine.generate. TYPE: float DEFAULT: 0.3
max_tokens	Generation parameters passed through to engine.generate. TYPE: float DEFAULT: 0.3
num_ctx	Generation parameters passed through to engine.generate. TYPE: float DEFAULT: 0.3
web_search	Optional (query, recency, news) -> [WebHit]. When given, the planner gets a web_search tool next to the corpus search. TYPE: Optional[WebSearcher] DEFAULT: None
on_event	Optional callback fired at loop milestones so callers (e.g. the SSE research router) can stream progress without rewriting the loop. Receives a dict in one of these shapes: - {"type": "search_call", "arguments": {...}} — about to call search - {"type": "search_result", "num_hits": N, "top_titles": [...], "sources": [{"ref": 1, "title": ..., "sender": ..., "date": ..., "source_id": ..., "url": ...}, ...]} — search returned - {"type": "clarify_call", "question": "..."} — about to ask for clarification - {"type": "clarify_response", "response": "..."} — clarification received - {"type": "final_answer", "text": "..."} — synthesis ready The callback runs on the same thread as run and must be non-blocking. TYPE: Optional[Callable[[Dict[str, Any]], None]] DEFAULT: None

Source code in src/diapason/agents/research_loop.py

def __init__(
    self,
    engine: InferenceEngine,
    search: HybridSearch,
    *,
    model: str = DEFAULT_PLANNER_MODEL,
    max_iterations: int = 5,
    temperature: float = 0.3,
    max_tokens: int = 1500,
    num_ctx: int = 16384,
    clarify_handler: Optional[Callable[[str], str]] = None,
    on_event: Optional[Callable[[Dict[str, Any]], None]] = None,
    available_sources: Optional[List[str]] = None,
    web_search: Optional[WebSearcher] = None,
) -> None:
    self._engine = engine
    self._search = search
    # ``web_search(query, recency, news) -> [WebHit]``; None keeps the
    # agent corpus-only (CLI presets, tests).
    self._web_search = web_search
    self._model = model
    self._max_iterations = int(max_iterations)
    self._temperature = float(temperature)
    self._max_tokens = int(max_tokens)
    self._num_ctx = int(num_ctx)
    self._clarify_handler = clarify_handler or _default_clarify_handler
    self._on_event = on_event
    # Explicit list wins; otherwise we'll discover sources from the
    # KnowledgeStore on each run() call so the prompt stays accurate
    # even as the user connects new connectors mid-session.
    self._available_sources_override = available_sources

Methods:

run

run(
    query: str,
    *,
    history: Optional[Sequence[Message]] = None,
    hints: Optional[Sequence[str]] = None,
) -> ResearchResult

Run the loop end-to-end and return the synthesis plus a trace.

history — the previous user/assistant turns, placed before the query so a follow-up has its referent. hints — system lines the caller established about THIS query (what « ces sites » refers to, that the user asked for the web), placed right after it.

Source code in src/diapason/agents/research_loop.py

def run(
    self,
    query: str,
    *,
    history: Optional[Sequence[Message]] = None,
    hints: Optional[Sequence[str]] = None,
) -> ResearchResult:
    """Run the loop end-to-end and return the synthesis plus a trace.

    ``history`` — the previous user/assistant turns, placed before the
    query so a follow-up has its referent. ``hints`` — system lines the
    caller established about THIS query (what « ces sites » refers to,
    that the user asked for the web), placed right after it.
    """
    sources_list = self._resolve_available_sources()
    if sources_list:
        sources_blurb = ", ".join(sources_list)
    else:
        sources_blurb = (
            "(no connected sources — tell the user to connect a "
            "connector before searching)"
        )
    with_web = self._web_search is not None
    sys_msg = Message(
        role=Role.SYSTEM,
        content=SYSTEM_PROMPT.format(
            # .astimezone() : sans le décalage dans la chaîne, le modèle
            # relit cet instant comme de l'UTC et décale tout ce qu'il en
            # déduit — des réunions terminées ressortaient « à venir ».
            today=datetime.now().astimezone().isoformat(timespec="minutes"),
            available_sources=sources_blurb,
            and_the_web=" and to the public web" if with_web else "",
            web_search_line=(
                "    web_search(query, recency=None, news=False)\n"
                if with_web
                else ""
            ),
            web_search_rules=WEB_SEARCH_RULES if with_web else "",
        ),
    )
    messages: List[Message] = [sys_msg]
    for prior in history or ():
        if (
            prior.role in (Role.USER, Role.ASSISTANT)
            and (prior.content or "").strip()
        ):
            messages.append(Message(role=prior.role, content=prior.content))
    messages.append(Message(role=Role.USER, content=query))
    for hint in hints or ():
        if hint and hint.strip():
            messages.append(Message(role=Role.SYSTEM, content=hint.strip()))
    specs: List[Dict[str, Any]] = [SEARCH_TOOL_SPEC]
    if with_web:
        specs.append(WEB_SEARCH_TOOL_SPEC)
    specs.append(CLARIFY_TOOL_SPEC)

    invocations: List[ToolInvocation] = []
    total_usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}

    # Global ref counter: each search increments by the number of hits
    # it returned so the planner sees unique refs across calls. The
    # accumulator lets us renumber whatever the synthesis cites at the
    # end into a single deduped client-facing sources list.
    next_ref: int = 1
    ref_to_source: Dict[int, Dict[str, Any]] = {}

    def _finalize(text: str) -> Tuple[str, List[Dict[str, Any]]]:
        return renumber_citations(text, ref_to_source)

    iterations = 0
    for _ in range(self._max_iterations + 1):
        iterations += 1
        tools_arg = specs if len(invocations) < self._max_iterations else None
        result = self._engine.generate(
            messages,
            model=self._model,
            temperature=self._temperature,
            max_tokens=self._max_tokens,
            num_ctx=self._num_ctx,
            tools=tools_arg,
        )
        for k in total_usage:
            total_usage[k] += int(result.get("usage", {}).get(k, 0))

        content = result.get("content", "") or ""
        tool_calls_raw = result.get("tool_calls", []) or []

        if not tool_calls_raw:
            if content.strip():
                answer, final_sources = _finalize(content.strip())
                self._emit(
                    {
                        "type": "final_answer",
                        "text": answer,
                        "sources": final_sources,
                    }
                )
                return ResearchResult(
                    answer=answer,
                    iterations=iterations,
                    tool_calls=invocations,
                    usage=total_usage,
                )
            # Empty content with no tool call — push a synthesis prod
            if invocations:
                messages.append(Message(role=Role.ASSISTANT, content=content))
                messages.append(
                    Message(
                        role=Role.USER,
                        content=(
                            "Write your final answer now based on the search "
                            "results above. Cite sources as [1], [2], etc."
                        ),
                    )
                )
                continue
            fallback = "(model returned no content and no tool calls)"
            self._emit({"type": "final_answer", "text": fallback, "sources": []})
            return ResearchResult(
                answer=fallback,
                iterations=iterations,
                tool_calls=invocations,
                usage=total_usage,
            )

        assistant_msg = Message(
            role=Role.ASSISTANT,
            content=content,
            tool_calls=[
                ToolCall(
                    id=tc.get("id", f"call_{i}"),
                    name=tc.get("name", "search"),
                    arguments=tc.get("arguments", "{}") or "{}",
                )
                for i, tc in enumerate(tool_calls_raw)
            ],
        )
        messages.append(assistant_msg)

        for tc in tool_calls_raw:
            name = tc.get("name", "")
            raw_args = tc.get("arguments", "{}") or "{}"
            try:
                args = (
                    json.loads(raw_args)
                    if isinstance(raw_args, str)
                    else dict(raw_args)
                )
            except json.JSONDecodeError:
                args = {}

            if name == "search":
                # Guard against the planner pre-empting clarify before any
                # search has run — silently accept; the rule lives in the
                # system prompt as guidance, not enforcement.
                self._emit({"type": "search_call", "arguments": args})
                inv = self._execute_search(args)
                invocations.append(inv)
                offset = next_ref - 1
                sources_for_search = build_sources_for_client(
                    inv.raw_hits, ref_offset=offset
                )
                self._emit(
                    {
                        "type": "search_result",
                        "num_hits": inv.num_results,
                        "top_titles": inv.top_titles,
                        "sources": sources_for_search,
                    }
                )
                for src in sources_for_search:
                    ref_to_source[int(src["ref"])] = src
                next_ref += len(sources_for_search)
                tool_output = json.dumps(
                    shape_results_for_model(inv.raw_hits, ref_offset=offset),
                    ensure_ascii=False,
                )
            elif name == "web_search" and with_web:
                if not str(args.get("query", "") or "").strip():
                    # No call, no event, no budget: an empty query (a
                    # malformed arguments blob, review of 22 Sept 2026)
                    # would otherwise read as "the web has nothing".
                    tool_output = json.dumps(
                        {"error": "web_search needs a non-empty query"}
                    )
                else:
                    self._emit(
                        {
                            "type": "search_call",
                            "arguments": {"tool": "web_search", **args},
                        }
                    )
                    inv = self._execute_web_search(args)
                    if inv.error:
                        # A search that could not be made: the planner
                        # reads an error, the client a failed step, and
                        # the budget is not charged.
                        self._emit(
                            {
                                "type": "search_result",
                                "tool": "web_search",
                                "error": inv.error,
                                "sources": [],
                            }
                        )
                        tool_output = json.dumps(
                            {
                                "error": (
                                    f"web_search failed: {inv.error}. The web "
                                    "was NOT searched — say so if you cannot "
                                    "answer from the corpus; never say the "
                                    "web has nothing."
                                )
                            }
                        )
                    else:
                        invocations.append(inv)
                        offset = next_ref - 1
                        sources_for_web = build_web_sources_for_client(
                            inv.web_hits, ref_offset=offset
                        )
                        self._emit(
                            {
                                "type": "search_result",
                                "tool": "web_search",
                                "num_hits": inv.num_results,
                                "top_titles": inv.top_titles,
                                "sources": sources_for_web,
                            }
                        )
                        for src in sources_for_web:
                            ref_to_source[int(src["ref"])] = src
                        next_ref += len(sources_for_web)
                        tool_output = json.dumps(
                            shape_web_results_for_model(
                                inv.web_hits, ref_offset=offset
                            ),
                            ensure_ascii=False,
                        )
            elif name == "clarify":
                # Enforce the "search first" rule at runtime so we don't
                # surprise the user with a clarification before showing any
                # work. If the planner jumps to clarify with no searches
                # behind it, return an error and let the loop try again.
                if not any(
                    i.tool_name in ("search", "web_search") for i in invocations
                ):
                    tool_output = json.dumps(
                        {
                            "error": (
                                "clarify is only available after at least "
                                "one search call. Run search first, then "
                                "use clarify if the results are ambiguous "
                                "or empty."
                            )
                        }
                    )
                else:
                    self._emit(
                        {
                            "type": "clarify_call",
                            "question": str(args.get("question", "")),
                        }
                    )
                    inv = self._execute_clarify(args)
                    invocations.append(inv)
                    self._emit(
                        {"type": "clarify_response", "response": inv.response}
                    )
                    tool_output = json.dumps(
                        {
                            "question": inv.arguments.get("question", ""),
                            "user_response": inv.response,
                        }
                    )
            else:
                disponibles = ", ".join(
                    repr(sp["function"]["name"]) for sp in specs
                )
                tool_output = json.dumps(
                    {
                        "error": (
                            f"unknown tool {name!r}; available tools are "
                            f"{disponibles}"
                        )
                    }
                )

            messages.append(
                Message(
                    role=Role.TOOL,
                    content=tool_output,
                    tool_call_id=tc.get("id", ""),
                    name=name,
                )
            )

        if len(invocations) >= self._max_iterations:
            messages.append(
                Message(
                    role=Role.USER,
                    content=(
                        "You have used your tool-call budget (search + "
                        "clarify combined). Write the final synthesis now "
                        "using only the search results and clarifications "
                        "above. Cite sources as [1], [2], etc."
                    ),
                )
            )

    # Loop fell through without the model producing a text response.
    # Force one final tool-less synthesis call so the caller always gets
    # an answer — bailing out with a sentinel string is never useful to
    # the user, who already paid for the searches.
    messages.append(
        Message(
            role=Role.USER,
            content=(
                "You've used all your search attempts. Synthesize your "
                "findings now from whatever you've found so far. Do not "
                "request more tool calls — write the final answer as "
                "plain text, citing sources as [1], [2], etc. where you can. "
                "If the searches returned nothing usable, say so plainly."
            ),
        )
    )
    iterations += 1
    final = self._engine.generate(
        messages,
        model=self._model,
        temperature=self._temperature,
        max_tokens=self._max_tokens,
        num_ctx=self._num_ctx,
        tools=None,
    )
    for k in total_usage:
        total_usage[k] += int(final.get("usage", {}).get(k, 0))
    answer = (final.get("content", "") or "").strip()
    if not answer:
        answer = (
            "(no synthesis available — the search budget was exhausted "
            "and the model returned no text response)"
        )
    answer, final_sources = _finalize(answer)
    self._emit({"type": "final_answer", "text": answer, "sources": final_sources})
    return ResearchResult(
        answer=answer,
        iterations=iterations,
        tool_calls=invocations,
        usage=total_usage,
    )

Functions:

shape_results_for_model

shape_results_for_model(
    hits: List[SearchHit],
    *,
    detailed_top: int = 5,
    thread_ctx_per_hit: int = 3,
    total_cap: int = 20,
    ref_offset: int = 0,
) -> Dict[str, Any]

Compact a hit list into a JSON payload the planner can chew through.

The first detailed_top rows keep their content snippet and trimmed thread context; the remainder are summarised to title + sender + date so the planner still sees the breadth of what's available without blowing the context window. Each hit gets a numeric ref (1-indexed, plus ref_offset) so the synthesis can cite it as [N]. The offset lets multi-search runs hand the planner globally unique refs across calls so a later renumbering pass can dedupe by first appearance.

Source code in src/diapason/agents/research_loop.py

def shape_results_for_model(
    hits: List[SearchHit],
    *,
    detailed_top: int = 5,
    thread_ctx_per_hit: int = 3,
    total_cap: int = 20,
    ref_offset: int = 0,
) -> Dict[str, Any]:
    """Compact a hit list into a JSON payload the planner can chew through.

    The first ``detailed_top`` rows keep their content snippet and trimmed
    thread context; the remainder are summarised to title + sender + date so
    the planner still sees the breadth of what's available without blowing the
    context window. Each hit gets a numeric ``ref`` (1-indexed, plus
    ``ref_offset``) so the synthesis can cite it as ``[N]``. The offset lets
    multi-search runs hand the planner globally unique refs across calls so
    a later renumbering pass can dedupe by first appearance.
    """
    out_hits: List[Dict[str, Any]] = []
    visible = hits[:total_cap]
    for i, h in enumerate(visible):
        sender = h.participants[0] if h.participants else ""
        base = {
            "ref": i + 1 + ref_offset,
            "title": h.title,
            "sender": sender,
            "timestamp": h.timestamp,
            "source": h.source,
            "score": round(h.score, 4),
        }
        if i < detailed_top:
            base["snippet"] = h.content_snippet
            if h.thread_context:
                base["thread"] = _trim_thread_context(
                    h.thread_context, thread_ctx_per_hit
                )
        out_hits.append(base)
    return {
        "num_results": len(hits),
        "shown": len(visible),
        "truncated": len(hits) > total_cap,
        "hits": out_hits,
    }

renumber_citations

renumber_citations(
    text: str, ref_to_source: Dict[int, Dict[str, Any]]
) -> Tuple[str, List[Dict[str, Any]]]

Renumber [N] citations in text by first-appearance order.

The planner sees globally-offset refs across multiple search calls (search 1 returns 1..20, search 2 returns 21..40, …). When the synthesis arrives, the first ref the model actually cited becomes [1], the second unique one becomes [2], and so on. Repeats map to the same new ref. Refs the synthesis never cites are dropped from the returned sources list — only the ones the user can actually click on get carried through.

PARAMETER	DESCRIPTION
text	Synthesis text containing inline [N] references. TYPE: str
ref_to_source	Mapping from the original (offset) ref to the source dict that build_sources_for_client produced for that hit. TYPE: Dict[int, Dict[str, Any]]

RETURNS	DESCRIPTION
(new_text, ordered_sources)	new_text has every cited [N] rewritten to its new sequence number. ordered_sources is the deduped list of source dicts in the order they appear in the synthesis, each with its ref field set to the new sequence number.

Source code in src/diapason/agents/research_loop.py

def renumber_citations(
    text: str,
    ref_to_source: Dict[int, Dict[str, Any]],
) -> Tuple[str, List[Dict[str, Any]]]:
    """Renumber ``[N]`` citations in ``text`` by first-appearance order.

    The planner sees globally-offset refs across multiple search calls
    (search 1 returns 1..20, search 2 returns 21..40, …). When the
    synthesis arrives, the first ref the model actually cited becomes
    ``[1]``, the second unique one becomes ``[2]``, and so on. Repeats
    map to the same new ref. Refs the synthesis never cites are dropped
    from the returned ``sources`` list — only the ones the user can
    actually click on get carried through.

    Parameters
    ----------
    text:
        Synthesis text containing inline ``[N]`` references.
    ref_to_source:
        Mapping from the original (offset) ref to the source dict that
        ``build_sources_for_client`` produced for that hit.

    Returns
    -------
    (new_text, ordered_sources)
        ``new_text`` has every cited ``[N]`` rewritten to its new
        sequence number. ``ordered_sources`` is the deduped list of
        source dicts in the order they appear in the synthesis, each
        with its ``ref`` field set to the new sequence number.
    """
    old_to_new: Dict[int, int] = {}
    ordered: List[Dict[str, Any]] = []
    for m in _CITE_RE.finditer(text):
        try:
            old = int(m.group(1))
        except ValueError:
            continue
        if old in old_to_new:
            continue
        src = ref_to_source.get(old)
        if src is None:
            # Synthesis cited a ref that doesn't exist in the corpus —
            # leave the literal text alone, drop the source entry.
            continue
        new_ref = len(ordered) + 1
        old_to_new[old] = new_ref
        renumbered_src = dict(src)
        renumbered_src["ref"] = new_ref
        ordered.append(renumbered_src)

    def _replace(match: "re.Match[str]") -> str:
        try:
            old = int(match.group(1))
        except ValueError:
            return match.group(0)
        new = old_to_new.get(old)
        return f"[{new}]" if new is not None else match.group(0)

    new_text = _CITE_RE.sub(_replace, text)
    return new_text, ordered

build_sources_for_client

build_sources_for_client(
    hits: List[SearchHit],
    *,
    total_cap: int = 20,
    ref_offset: int = 0,
) -> List[Dict[str, Any]]

Produce the citation-friendly sources list streamed to the frontend.

One entry per hit, in the same order the planner sees them — so a [N] citation in the synthesis maps to sources[N - 1] on the client. We don't deduplicate by document_id: separate chunks of the same email each get their own citation slot since the planner may quote different parts.

Source code in src/diapason/agents/research_loop.py

def build_sources_for_client(
    hits: List[SearchHit],
    *,
    total_cap: int = 20,
    ref_offset: int = 0,
) -> List[Dict[str, Any]]:
    """Produce the citation-friendly sources list streamed to the frontend.

    One entry per hit, in the same order the planner sees them — so a
    ``[N]`` citation in the synthesis maps to ``sources[N - 1]`` on the
    client. We don't deduplicate by ``document_id``: separate chunks of the
    same email each get their own citation slot since the planner may quote
    different parts.
    """
    out: List[Dict[str, Any]] = []
    for i, h in enumerate(hits[:total_cap]):
        sender = h.participants[0] if h.participants else ""
        # Prefer the URL the connector stored at ingest time (Granola's
        # ``web_url``, Notion's page URL, etc.) — it's the only reliable
        # link for sources whose web URL doesn't derive from the doc_id.
        # Fall back to the doc_id-based reconstruction for sources where
        # that still works (Slack, Gmail).
        url = h.url or _hit_url(h.source, h.document_id)
        out.append(
            {
                "ref": i + 1 + ref_offset,
                "title": h.title,
                "sender": sender,
                "date": _hit_date(h.timestamp),
                "source": h.source,
                "source_id": _bare_doc_id(h.source, h.document_id),
                "url": url,
            }
        )
    return out

build_web_sources_for_client

build_web_sources_for_client(
    hits: List[WebHit],
    *,
    ref_offset: int = 0,
    total_cap: int = 10,
) -> List[Dict[str, Any]]

Same shape as :func:build_sources_for_client; source is web and sender the site, so the client renders a web hit like any other.

Source code in src/diapason/agents/research_loop.py

def build_web_sources_for_client(
    hits: List[WebHit], *, ref_offset: int = 0, total_cap: int = 10
) -> List[Dict[str, Any]]:
    """Same shape as :func:`build_sources_for_client`; ``source`` is ``web``
    and ``sender`` the site, so the client renders a web hit like any other."""
    return [
        {
            "ref": i + 1 + ref_offset,
            "title": h.title,
            "sender": h.site,
            "date": h.date,
            "source": "web",
            "source_id": "",
            "url": h.url,
        }
        for i, h in enumerate(hits[:total_cap])
    ]