Créer un agent de recherche privé avec Venice AI

Les agents de recherche sont utiles lorsque vous voulez plus qu’un simple résultat de recherche ou qu’une réponse rapide d’un modèle. Un bon agent de recherche peut transformer un sujet large en requêtes de recherche, collecter des sources, extraire les preuves importantes, suivre les lacunes et rédiger un briefing sourcé que vous pouvez ensuite inspecter. Dans ce tutoriel, nous allons créer un agent de recherche privé en utilisant Python et l’API Venice. À la fin, vous disposerez d’une CLI capable de rechercher un sujet, de scraper des pages publiques en Markdown, de résumer des morceaux de source, d’exécuter des passes de recherche complémentaires conscientes des lacunes et de générer un rapport sourcé avec des artefacts JSONL locaux optionnels. Intéressé par l’implémentation complète du code ? Consultez le dépôt GitHub. Avant de continuer, vous aurez besoin d’une clé API Venice :

export VENICE_API_KEY=<my-key>

Ce que nous allons construire

L’implémentation de référence est un petit projet Python avec quelques parties clairement définies :

Partie	Ce qu’elle fait
CLI	Accepte un sujet de recherche, un modèle, des fournisseurs, des paramètres de profondeur, un chemin de sortie et un répertoire d’artefacts
Client Venice	Appelle les chat completions, les chat completions en streaming et `POST /augment/scrape`
Couche de recherche	Recherche sur DuckDuckGo par défaut, avec une découverte optionnelle de papiers arXiv
Modèles de données	Suit les URL sources, les URL canoniques, les chunks, les preuves, les notes, les erreurs et les rapports
Agent de recherche	Planifie les recherches, lit les sources, extrait les preuves, analyse les lacunes, génère des requêtes complémentaires et rédige le rapport final
Rédacteur d’artefacts	Stocke des enregistrements JSONL auditables pour les requêtes, lacunes de recherche, résultats, fetches, chunks, notes de sources, brouillons de rapports, erreurs et rapports

Le flux ressemble à ceci :

Demander à Venice de générer des requêtes de recherche diversifiées pour le sujet.
Rechercher sur le web avec un ou plusieurs fournisseurs.
Dédupliquer les URL avant de les lire.
Utiliser l’endpoint de scraping de Venice pour transformer chaque page source publique en Markdown.
Diviser les longues pages en chunks.
Demander à Venice d’extraire les preuves de chaque chunk.
Demander à Venice de transformer les preuves des chunks en notes de source.
Identifier les lacunes de recherche et les problèmes d’équilibre des sources avant de générer des requêtes complémentaires.
Demander à Venice de synthétiser le rapport final avec des citations de style notes de bas de page.

C’est « privé » au sens pratique où l’agent conserve l’orchestration, les notes de source, les artefacts et les rapports finaux sur votre machine. Venice gère les appels au modèle et le scraping via son API. L’implémentation de référence par défaut envoie toujours les requêtes de recherche à DuckDuckGo ou arXiv, donc considérez le choix du fournisseur comme partie intégrante de votre conception de confidentialité.

Configuration du projet

Le projet de référence utilise Python 3.13 et uv, mais le même code fonctionne aussi avec un environnement virtuel normal. Créez un nouveau projet :

mkdir venice-research-agent
cd venice-research-agent
uv init

Installez les dépendances :

uv add httpx beautifulsoup4 python-dotenv

Si vous préférez pip, créez un environnement virtuel et installez les mêmes paquets :

python -m venv .venv
source .venv/bin/activate
pip install "httpx>=0.28.0" "beautifulsoup4>=4.13.0" "python-dotenv>=1.0.0"

Créez un fichier .env pour le développement local :

VENICE_API_KEY=your_venice_api_key_here
VENICE_MODEL=openai-gpt-55

Nous utilisons VENICE_MODEL afin que vous puissiez changer de modèle sans modifier le code. L’implémentation de référence utilise actuellement openai-gpt-55 par défaut, mais vous pouvez le remplacer par un autre modèle de chat disponible pour votre compte Venice.

Création des modèles de données

Avant d’écrire la logique de l’agent, nous allons définir les objets qui circulent dans le pipeline. Ces modèles facilitent la compréhension du reste du code car chaque source porte sa provenance : d’où elle vient, quelle requête l’a trouvée, quand elle a été récupérée et comment elle a été découpée en chunks. Créez research_agent/models.py :

from __future__ import annotations

import hashlib
from dataclasses import dataclass, field
from datetime import UTC, datetime
from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse

TRACKING_PARAMS = {
    "fbclid",
    "gclid",
    "igshid",
    "mc_cid",
    "mc_eid",
    "msclkid",
    "ref",
    "ref_src",
}


@dataclass(frozen=True)
class SearchResult:
    title: str
    url: str
    snippet: str
    query: str = ""
    rank: int = 0
    provider: str = "duckduckgo"
    canonical_url: str = ""

    def __post_init__(self) -> None:
        if not self.canonical_url:
            object.__setattr__(self, "canonical_url", canonicalize_url(self.url))


@dataclass(frozen=True)
class ScrapeResult:
    url: str
    content: str
    title: str = ""
    final_url: str = ""
    content_type: str = "text/markdown"


@dataclass(frozen=True)
class TextChunk:
    chunk_id: str
    text: str
    start: int
    end: int
    content_hash: str


@dataclass(frozen=True)
class WebPage:
    title: str
    url: str
    text: str
    final_url: str = ""
    canonical_url: str = ""
    content_type: str = ""
    retrieved_at: str = ""
    content_hash: str = ""
    chunks: tuple[TextChunk, ...] = field(default_factory=tuple)

    def __post_init__(self) -> None:
        final_url = self.final_url or self.url
        object.__setattr__(self, "final_url", final_url)
        if not self.canonical_url:
            object.__setattr__(self, "canonical_url", canonicalize_url(final_url))
        if not self.retrieved_at:
            object.__setattr__(self, "retrieved_at", utc_now())
        if not self.content_hash:
            object.__setattr__(self, "content_hash", content_hash(self.text))


@dataclass(frozen=True)
class EvidenceChunk:
    chunk_id: str
    text: str
    summary: str
    quotes: tuple[str, ...] = field(default_factory=tuple)


@dataclass(frozen=True)
class SourceNote:
    source_id: str
    title: str
    url: str
    query: str
    summary: str
    canonical_url: str = ""
    final_url: str = ""
    rank: int = 0
    snippet: str = ""
    provider: str = "duckduckgo"
    retrieved_at: str = ""
    content_type: str = ""
    content_hash: str = ""
    chunks: tuple[EvidenceChunk, ...] = field(default_factory=tuple)

Les champs importants ici sont canonical_url, content_hash et chunks. canonical_url permet à l’agent d’éviter de lire la même source plusieurs fois lorsque les résultats de recherche ne diffèrent que par des paramètres de suivi ou des fragments. content_hash aide à détecter les pages dupliquées même lorsqu’elles se trouvent à des URL différentes. chunks nous permet de résumer de longues pages en petits morceaux au lieu de perdre des preuves utiles à cause des limites de contexte. Ajoutez les fonctions utilitaires sous les dataclasses :

def utc_now() -> str:
    return datetime.now(UTC).isoformat()


def content_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def canonicalize_url(raw_url: str) -> str:
    if not raw_url:
        return ""

    parsed = urlparse(raw_url.strip())
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        return ""

    scheme = parsed.scheme.lower()
    netloc = parsed.netloc.lower()
    path = parsed.path or "/"
    if path != "/":
        path = path.rstrip("/")

    query_pairs = [
        (key, value)
        for key, value in parse_qsl(parsed.query, keep_blank_values=True)
        if not _is_tracking_param(key)
    ]
    query = urlencode(sorted(query_pairs), doseq=True)
    return urlunparse((scheme, netloc, path, "", query, ""))


def chunk_text(text: str, *, chunk_chars: int = 3000, overlap: int = 250) -> tuple[TextChunk, ...]:
    clean = text.strip()
    if not clean:
        return ()
    if chunk_chars <= 0:
        raise ValueError("chunk_chars must be greater than 0")
    if overlap < 0 or overlap >= chunk_chars:
        raise ValueError("overlap must be at least 0 and smaller than chunk_chars")

    chunks: list[TextChunk] = []
    start = 0
    index = 1
    while start < len(clean):
        end = min(len(clean), start + chunk_chars)
        chunk = clean[start:end].strip()
        if chunk:
            chunks.append(
                TextChunk(
                    chunk_id=f"C{index}",
                    text=chunk,
                    start=start,
                    end=end,
                    content_hash=content_hash(chunk),
                )
            )
            index += 1
        if end == len(clean):
            break
        start = end - overlap

    return tuple(chunks)


def _is_tracking_param(key: str) -> bool:
    lowered = key.lower()
    return lowered.startswith("utm_") or lowered in TRACKING_PARAMS

Le chunking est délibérément simple ici : des chunks de taille fixe en caractères avec chevauchement. C’est suffisant pour un agent de recherche de démonstration car l’endpoint de scraping de Venice renvoie du Markdown, qui est généralement beaucoup plus propre que du HTML brut. Pour la recherche en production sur de longs documents techniques, vous pouvez améliorer cela en découpant selon les titres, les paragraphes ou le nombre de tokens.

Construction du client Venice

Ensuite, nous allons créer un petit client Venice. Vous pourriez utiliser le SDK Python d’OpenAI pour les chat completions car Venice est compatible OpenAI, mais l’implémentation de référence utilise directement httpx afin que le même client puisse appeler l’endpoint POST /augment/scrape de Venice. Créez research_agent/venice.py :

from __future__ import annotations

import json
import os
import time
from dataclasses import dataclass
from typing import Any

import httpx

from .models import ScrapeResult

DEFAULT_BASE_URL = "https://api.venice.ai/api/v1"
DEFAULT_MODEL = "openai-gpt-55"
RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}


class VeniceError(RuntimeError):
    """Raised when the Venice API returns an unusable response."""


@dataclass(frozen=True)
class VeniceClient:
    api_key: str
    model: str = DEFAULT_MODEL
    base_url: str = DEFAULT_BASE_URL
    timeout: float = 60.0
    max_retries: int = 2
    backoff_seconds: float = 1.0

    @classmethod
    def from_env(cls, model: str | None = None, *, max_retries: int = 2) -> "VeniceClient":
        api_key = os.getenv("VENICE_API_KEY")
        if not api_key:
            raise VeniceError("VENICE_API_KEY is required.")

        return cls(
            api_key=api_key,
            model=model or os.getenv("VENICE_MODEL", DEFAULT_MODEL),
            base_url=os.getenv("VENICE_BASE_URL", DEFAULT_BASE_URL).rstrip("/"),
            max_retries=max_retries,
        )

L’utilitaire from_env() empêche les secrets de figurer dans votre code source. Il rend également le développement local pratique car python-dotenv peut charger VENICE_API_KEY et VENICE_MODEL depuis .env. Maintenant, ajoutez les chat completions :

    def chat(
        self,
        messages: list[dict[str, str]],
        *,
        temperature: float = 0.2,
        max_tokens: int = 1600,
    ) -> str:
        payload: dict[str, Any] = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
        }

        data = self._post_json("/chat/completions", payload)
        try:
            return data["choices"][0]["message"]["content"].strip()
        except (KeyError, IndexError, TypeError) as exc:
            raise VeniceError(f"Unexpected Venice API response: {data}") from exc

Pour le rapport final, nous voulons utiliser le streaming car les rapports approfondis peuvent prendre beaucoup plus de temps (car ils produiront beaucoup plus de texte). Cela peut entraîner des problèmes de timeout pour les requêtes où la production de la sortie finale peut prendre extrêmement longtemps. En utilisant le streaming, nous pouvons éliminer ce problème et rendre la requête plus résistante aux échecs de timeout :

    def chat_stream(
        self,
        messages: list[dict[str, str]],
        *,
        temperature: float = 0.2,
        max_tokens: int = 1600,
    ) -> str:
        payload: dict[str, Any] = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": True,
        }
        return self._post_chat_stream("/chat/completions", payload).strip()

Ensuite, ajoutez le scraping :

    def scrape(self, url: str) -> ScrapeResult:
        data = self._post_json("/augment/scrape", {"url": url})
        content = _first_string(data, "content", "markdown", "text")
        if not content:
            raise VeniceError(f"Unexpected Venice scrape response: {data}")

        return ScrapeResult(
            url=url,
            final_url=_first_string(data, "final_url", "url", "source_url") or url,
            title=_first_string(data, "title"),
            content=content,
            content_type="text/markdown",
        )

L’endpoint de scraping de Venice accepte une URL accessible publiquement et renvoie la page en Markdown. Cela signifie que le modèle n’a pas besoin d’analyser du HTML brut, et vos prompts d’extraction de source peuvent travailler avec du texte plus propre. L’utilitaire restant gère les retries et le parsing des réponses :

    def _post_json(self, path: str, payload: dict[str, Any]) -> dict[str, Any]:
        for attempt in range(self.max_retries + 1):
            try:
                response = httpx.post(
                    f"{self.base_url}{path}",
                    headers={
                        "Authorization": f"Bearer {self.api_key}",
                        "Content-Type": "application/json",
                    },
                    json=payload,
                    timeout=self.timeout,
                )
                if response.status_code in RETRYABLE_STATUS_CODES and attempt < self.max_retries:
                    time.sleep(self.backoff_seconds * (2**attempt))
                    continue
                response.raise_for_status()
                data = response.json()
                if not isinstance(data, dict):
                    raise VeniceError(f"Unexpected Venice API response: {data}")
                return data
            except httpx.HTTPError as exc:
                if attempt < self.max_retries:
                    time.sleep(self.backoff_seconds * (2**attempt))
                    continue
                raise VeniceError(f"Could not reach Venice API: {exc}") from exc

        raise VeniceError("Could not reach Venice API")


def _first_string(data: dict[str, Any], *keys: str) -> str:
    for key in keys:
        value = data.get(key)
        if isinstance(value, str) and value.strip():
            return value.strip()

    for nested_key in ("data", "result", "scrape"):
        nested = data.get(nested_key)
        if isinstance(nested, dict):
            value = _first_string(nested, *keys)
            if value:
                return value

    return ""

Le dépôt complet inclut également un utilitaire robuste _post_chat_stream() qui lit les server-sent events des chat completions en streaming. Vous pouvez commencer sans streaming, puis l’ajouter une fois que le reste du flux de recherche fonctionne.

Ajout des fournisseurs de recherche

La couche de recherche a deux tâches : trouver les URL sources et récupérer ces URL via le scraper Venice. L’implémentation de référence utilise l’endpoint HTML de DuckDuckGo pour la recherche web générale et l’API Atom d’arXiv pour les papiers. Créez research_agent/web.py :

from __future__ import annotations

import re
import xml.etree.ElementTree as ET
from collections.abc import Callable, Iterable
from urllib.parse import parse_qs, unquote, urlparse

import httpx
from bs4 import BeautifulSoup

from .models import ScrapeResult, SearchResult, TextChunk, WebPage, canonicalize_url, chunk_text, content_hash, utc_now

USER_AGENT = "venice-research-agent-demo/0.1 (+https://venice.ai)"


class SearchProvider:
    name = "provider"

    def search(self, web: "WebSearch", query: str, limit: int) -> list[SearchResult]:
        raise NotImplementedError

Maintenant, ajoutez DuckDuckGo :

class DuckDuckGoProvider(SearchProvider):
    name = "duckduckgo"

    def search(self, web: "WebSearch", query: str, limit: int) -> list[SearchResult]:
        response = web.get("https://duckduckgo.com/html/", params={"q": query})
        soup = BeautifulSoup(response.text, "html.parser")
        results: list[SearchResult] = []
        seen_urls: set[str] = set()

        for node in soup.select(".result"):
            link = node.select_one(".result__a")
            if link is None:
                continue

            url = _normalize_duckduckgo_url(link.get("href", ""))
            canonical_url = canonicalize_url(url)
            if not canonical_url or canonical_url in seen_urls:
                continue

            snippet = node.select_one(".result__snippet")
            results.append(
                SearchResult(
                    title=_clean_text(link.get_text(" ", strip=True)),
                    url=url,
                    snippet=_clean_text(snippet.get_text(" ", strip=True) if snippet else ""),
                    query=query,
                    rank=len(results) + 1,
                    provider=self.name,
                    canonical_url=canonical_url,
                )
            )
            seen_urls.add(canonical_url)

            if len(results) >= limit:
                break

        return results

Et arXiv :

class ArxivProvider(SearchProvider):
    name = "arxiv"

    def search(self, web: "WebSearch", query: str, limit: int) -> list[SearchResult]:
        response = web.get(
            "https://export.arxiv.org/api/query",
            params={
                "search_query": f"all:{query}",
                "start": 0,
                "max_results": limit,
                "sortBy": "relevance",
            },
        )
        namespace = {"atom": "http://www.w3.org/2005/Atom"}
        root = ET.fromstring(response.text)
        results: list[SearchResult] = []

        for entry in root.findall("atom:entry", namespace):
            title = _clean_text(_xml_text(entry.find("atom:title", namespace)))
            summary = _clean_text(_xml_text(entry.find("atom:summary", namespace)))
            url = _xml_text(entry.find("atom:id", namespace)).strip()
            canonical_url = canonicalize_url(url)
            if not url or not canonical_url:
                continue

            results.append(
                SearchResult(
                    title=title or url,
                    url=url,
                    snippet=summary,
                    query=query,
                    rank=len(results) + 1,
                    provider=self.name,
                    canonical_url=canonical_url,
                )
            )

            if len(results) >= limit:
                break

        return results

La classe WebSearch coordonne les fournisseurs et récupère les pages :

class WebSearch:
    def __init__(
        self,
        timeout: float = 15.0,
        *,
        providers: Iterable[SearchProvider] | None = None,
        chunk_chars: int = 3000,
        scraper: Callable[[str], ScrapeResult] | None = None,
    ) -> None:
        self._client = httpx.Client(
            timeout=timeout,
            follow_redirects=True,
            headers={"User-Agent": USER_AGENT},
        )
        self.providers = tuple(providers or (DuckDuckGoProvider(),))
        self.chunk_chars = chunk_chars
        self.scraper = scraper

    @classmethod
    def from_provider_names(cls, provider_names: Iterable[str], **kwargs: object) -> "WebSearch":
        providers = [_provider_from_name(name) for name in provider_names]
        return cls(providers=providers, **kwargs)

    def search(self, query: str, limit: int = 5) -> list[SearchResult]:
        results: list[SearchResult] = []
        seen_urls: set[str] = set()

        for provider in self.providers:
            for result in provider.search(self, query, limit):
                if result.canonical_url in seen_urls:
                    continue
                results.append(result)
                seen_urls.add(result.canonical_url)

        return results

    def fetch(self, result: SearchResult) -> WebPage:
        if self.scraper is None:
            raise RuntimeError("WebSearch.fetch requires a Venice scrape function.")

        scraped = self.scraper(result.url)
        text = scraped.content.strip() or result.snippet
        chunks = self._chunk_text(text)
        return WebPage(
            title=scraped.title or result.title,
            url=result.url,
            final_url=scraped.final_url or scraped.url or result.url,
            canonical_url=canonicalize_url(scraped.final_url or result.url),
            text=text,
            content_type=scraped.content_type or "text/markdown",
            retrieved_at=utc_now(),
            content_hash=content_hash(text),
            chunks=chunks,
        )

    def get(self, url: str, *, params: dict[str, object] | None = None) -> httpx.Response:
        response = self._client.get(url, params=params)
        response.raise_for_status()
        return response

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> "WebSearch":
        return self

    def __exit__(self, *_: object) -> None:
        self.close()

    def _chunk_text(self, text: str) -> tuple[TextChunk, ...]:
        overlap = min(250, max(0, self.chunk_chars // 10))
        return chunk_text(text, chunk_chars=self.chunk_chars, overlap=overlap)

L’implémentation de référence complète ajoute des retries, des délais de requête au niveau de l’hôte et des erreurs plus claires. Cela vaut la peine d’être conservé car les agents de recherche passent beaucoup de temps à gérer des pages qui bloquent l’automatisation, redirigent de manière inattendue ou renvoient des erreurs transitoires. Ajoutez les petits utilitaires de fournisseurs en bas :

def _normalize_duckduckgo_url(raw_url: str) -> str:
    if not raw_url:
        return ""

    parsed = urlparse(raw_url)
    if parsed.netloc.endswith("duckduckgo.com") and parsed.path == "/l/":
        target = parse_qs(parsed.query).get("uddg", [""])[0]
        return unquote(target)

    if parsed.scheme in {"http", "https"}:
        return raw_url

    return ""


def _provider_from_name(name: str) -> SearchProvider:
    normalized = name.strip().lower()
    if normalized in {"duckduckgo", "ddg", "web"}:
        return DuckDuckGoProvider()
    if normalized == "arxiv":
        return ArxivProvider()
    raise ValueError(f"Unknown source provider: {name}")


def _clean_text(value: str) -> str:
    return re.sub(r"\s+", " ", value).strip()


def _xml_text(node: ET.Element | None) -> str:
    return "" if node is None or node.text is None else node.text

Écriture des artefacts locaux

Pour les workflows de recherche, l’auditabilité est importante. Si le rapport final dit quelque chose de surprenant, vous devriez pouvoir inspecter quelle source y a mené. Créez research_agent/artifacts.py :

from __future__ import annotations

import json
from dataclasses import asdict, is_dataclass
from pathlib import Path
from typing import Any


class ArtifactWriter:
    def __init__(self, root: Path | None = None) -> None:
        self.root = root
        if self.root is not None:
            self.root.mkdir(parents=True, exist_ok=True)

    @property
    def enabled(self) -> bool:
        return self.root is not None

    def write(self, kind: str, record: object) -> None:
        if self.root is None:
            return

        path = self.root / f"{kind}.jsonl"
        payload = json.dumps(_to_jsonable(record), ensure_ascii=False, sort_keys=True)
        with path.open("a", encoding="utf-8") as file:
            file.write(f"{payload}\n")


def _to_jsonable(value: object) -> Any:
    if is_dataclass(value):
        return _to_jsonable(asdict(value))
    if isinstance(value, Path):
        return str(value)
    if isinstance(value, dict):
        return {str(key): _to_jsonable(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [_to_jsonable(item) for item in value]
    return value

Cela écrit un objet JSON par ligne, ce qui rend les artefacts faciles à ajouter, à inspecter et à traiter avec des outils en ligne de commande par la suite.

Construction de l’agent de recherche

Maintenant que nous avons Venice, la recherche, les modèles et les artefacts, nous pouvons construire l’agent lui-même. Créez research_agent/agent.py :

from __future__ import annotations

import json
from collections.abc import Callable
from textwrap import dedent

from .artifacts import ArtifactWriter
from .models import CollectionError, EvidenceChunk, ResearchReport, SearchResult, SourceNote, WebPage, utc_now
from .venice import VeniceClient, VeniceError
from .web import WebSearch

SYSTEM_PROMPT = """You are a careful research assistant.
Use the supplied source material only when making factual claims.
Flag uncertainty, contradictions, and missing context instead of filling gaps."""

ProgressCallback = Callable[[str], None]

DEFAULT_ITERATIONS = 3
DEFAULT_QUERY_COUNT = 6
DEFAULT_RESULTS_PER_QUERY = 4
DEFAULT_MAX_SOURCES = 40
DEFAULT_MAX_CHUNKS_PER_SOURCE = 6

Le system prompt est le garde-fou comportemental principal. Nous ne voulons pas que le modèle produise un rapport au son impressionnant à partir de sa mémoire. Nous voulons qu’il utilise le matériel source et signale les incertitudes lorsque les preuves sont minces. Nous avons également besoin de deux dataclasses finales dans models.py si vous ne les avez pas encore ajoutées :

@dataclass(frozen=True)
class CollectionError:
    stage: str
    message: str
    query: str = ""
    url: str = ""
    source_id: str = ""
    provider: str = ""


@dataclass(frozen=True)
class ResearchReport:
    topic: str
    markdown: str
    sources: list[SourceNote]
    artifacts_dir: str | None = None

Ensuite, définissez le ResearchAgent :

class ResearchAgent:
    def __init__(
        self,
        venice: VeniceClient,
        web: WebSearch | None = None,
        artifacts: ArtifactWriter | None = None,
        progress: ProgressCallback | None = None,
        max_sources: int | None = DEFAULT_MAX_SOURCES,
        max_chunks_per_source: int = DEFAULT_MAX_CHUNKS_PER_SOURCE,
    ) -> None:
        self.venice = venice
        self.web = web or WebSearch(scraper=venice.scrape)
        self.artifacts = artifacts or ArtifactWriter()
        self.progress = progress or (lambda _: None)
        self.max_sources = max_sources
        self.max_chunks_per_source = max_chunks_per_source

La méthode run() coordonne les passes de recherche :

    def run(
        self,
        topic: str,
        *,
        iterations: int = DEFAULT_ITERATIONS,
        query_count: int = DEFAULT_QUERY_COUNT,
        results_per_query: int = DEFAULT_RESULTS_PER_QUERY,
    ) -> ResearchReport:
        notes: list[SourceNote] = []
        seen_source_keys: set[str] = set()
        seen_content_hashes: set[str] = set()
        queries = self._initial_queries(topic, query_count)

        self.artifacts.write("queries", {"stage": "initial", "topic": topic, "queries": queries})

        for iteration in range(1, iterations + 1):
            self.progress(f"Research pass {iteration}/{iterations}: {', '.join(queries)}")
            self._collect_notes(
                topic,
                queries,
                results_per_query,
                seen_source_keys,
                seen_content_hashes,
                notes,
                iteration,
            )

            if iteration < iterations:
                gaps, queries = self._gap_follow_up_queries(topic, notes, query_count)
                self.artifacts.write(
                    "research_gaps",
                    {
                        "topic": topic,
                        "after_iteration": iteration,
                        "source_balance": _source_cluster_counts(notes),
                        "gaps": gaps,
                        "queries": queries,
                    },
                )
                self.artifacts.write(
                    "queries",
                    {
                        "stage": "follow_up",
                        "topic": topic,
                        "iteration": iteration + 1,
                        "gap_count": len(gaps),
                        "queries": queries,
                    },
                )

        report = self._write_report(topic, notes)
        self.artifacts.write(
            "reports",
            {
                "topic": topic,
                "source_count": len(notes),
                "generated_at": utc_now(),
                "markdown": report,
            },
        )

        return ResearchReport(
            topic=topic,
            markdown=report,
            sources=notes,
            artifacts_dir=str(self.artifacts.root) if self.artifacts.root is not None else None,
        )

Les deux ensembles seen_* sont ce qui empêche l’agent de perdre du temps sur des sources dupliquées. La déduplication d’URL détecte les liens répétés. La déduplication par hachage de contenu détecte les miroirs, les articles syndiqués et les pages qui redirigent vers le même contenu final.

Planification des recherches initiales et complémentaires

Le premier appel au modèle transforme le sujet en requêtes de recherche :

    def _initial_queries(self, topic: str, count: int) -> list[str]:
        prompt = dedent(
            f"""
            Create {count} diverse web search queries for researching this topic:
            {topic}

            Cover background, recent developments, primary sources, criticism, and data.
            Include at least one query likely to find primary sources or datasets.
            Return JSON only in this shape: {{"queries": ["..."]}}
            """
        ).strip()
        return self._query_list(prompt, count, fallback=[topic])

Après chaque passe de recherche, l’agent mis à jour effectue une étape d’analyse des lacunes plus délibérée. Il examine les notes actuelles, compte les clusters de sources par domaine, demande à Venice quelle couverture est manquante, écrit ces lacunes dans les artefacts, puis utilise les requêtes résultantes pour la passe suivante.

Commencez par suivre l’équilibre des sources :

from urllib.parse import urlparse


def _source_cluster_counts(notes: list[SourceNote]) -> list[dict[str, object]]:
    total = len(notes)
    if total == 0:
        return []

    clusters: dict[str, list[str]] = {}
    for note in notes:
        cluster = _source_cluster(note)
        clusters.setdefault(cluster, []).append(note.source_id)

    return [
        {
            "cluster": cluster,
            "source_count": len(source_ids),
            "source_share": round(len(source_ids) / total, 3),
            "source_ids": source_ids,
        }
        for cluster, source_ids in sorted(
            clusters.items(), key=lambda item: (-len(item[1]), item[0])
        )
    ]


def _source_cluster(note: SourceNote) -> str:
    url = note.canonical_url or note.final_url or note.url
    host = urlparse(url).netloc.lower()
    if host.startswith("www."):
        host = host[4:]
    return host or "unknown"


def _source_balance_digest(notes: list[SourceNote], limit: int = 8) -> str:
    clusters = _source_cluster_counts(notes)
    if not clusters:
        return "No source clusters yet."

    total = len(notes)
    lines = [
        f"- {cluster['cluster']}: {cluster['source_count']}/{total} sources "
        f"({cluster['source_share']:.0%}); IDs: {', '.join(cluster['source_ids'])}"
        for cluster in clusters[:limit]
    ]
    return "\n".join(lines)

Cela donne à l’agent un moyen simple de remarquer la capture par cluster de sources. Si toutes les sources proviennent d’une seule entreprise, d’un seul framework ou d’un seul domaine, les requêtes complémentaires devraient délibérément élargir l’ensemble des sources au lieu d’en collecter davantage du même type. Utilisez maintenant ces informations d’équilibre lors de la création des recherches complémentaires :

    def _follow_up_queries(self, topic: str, notes: list[SourceNote], count: int) -> list[str]:
        digest = _source_digest(notes, max_chars=9000)
        source_balance = _source_balance_digest(notes)
        prompt = dedent(
            f"""
            We are researching: {topic}

            Current notes:
            {digest}

            Source balance:
            {source_balance}

            Create {count} follow-up web search queries that fill gaps, verify important claims,
            find primary evidence, and look for dissenting evidence.
            If one source domain, vendor, framework, product, or perspective is overrepresented,
            deliberately broaden beyond it unless the topic explicitly asks for that focus.
            Return JSON only in this shape: {{"queries": ["..."]}}
            """
        ).strip()
        return self._query_list(prompt, count, fallback=[topic])

La nouvelle implémentation de référence enveloppe cela dans _gap_follow_up_queries(), qui demande à Venice de renvoyer à la fois les enregistrements de lacunes et les requêtes :

    def _gap_follow_up_queries(
        self, topic: str, notes: list[SourceNote], count: int
    ) -> tuple[list[dict[str, str]], list[str]]:
        if not notes:
            return [], [topic]

        digest = _source_digest(notes, max_chars=12000)
        source_balance = _source_balance_digest(notes)
        prompt = dedent(
            f"""
            Identify coverage gaps before the next research pass.

            Research topic:
            {topic}

            Current source notes:
            {digest}

            Source balance:
            {source_balance}

            Find important missing coverage that would improve a deep research report.
            Look specifically for primary sources, technical concepts, dissenting views,
            overrepresented source clusters, and claims that need verification.

            Return JSON only in this shape:
            {{"gaps": [{{"missing": "...", "why_it_matters": "...", "query": "..."}}],
              "queries": ["targeted web search query"]}}
            """
        ).strip()
        response = self.venice.chat(
            [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": prompt},
            ],
            temperature=0.3,
            max_tokens=900,
        )

        data = json.loads(response)
        gaps = _clean_gap_records(data.get("gaps"))
        queries = _clean_string_list(data.get("queries"))
        if not queries:
            queries = [gap["query"] for gap in gaps if gap.get("query")]
        return gaps, queries[:count]

Lorsque --artifacts est activé, ces enregistrements sont écrits dans research_gaps.jsonl. Cela vous donne une trace d’audit utile pour comprendre pourquoi l’agent a recherché une requête particulière en deuxième passe. Le parseur doit être tolérant. Si le modèle renvoie un JSON mal formé, l’agent revient au sujet original :

    def _query_list(self, prompt: str, count: int, fallback: list[str]) -> list[str]:
        response = self.venice.chat(
            [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": prompt},
            ],
            temperature=0.4,
            max_tokens=500,
        )
        try:
            data = json.loads(response)
            queries = data.get("queries", [])
        except (json.JSONDecodeError, AttributeError):
            queries = []

        clean_queries = [
            query.strip()
            for query in queries
            if isinstance(query, str) and query.strip()
        ]
        return (clean_queries or fallback)[:count]

Ce modèle vaut la peine d’être utilisé partout dans le code d’agent : demander une sortie structurée, la parser et fournir un fallback simple lorsque la sortie n’est pas utilisable.

Lecture et résumé des sources

Maintenant, nous collectons les notes de source. L’agent recherche chaque requête, récupère chaque résultat via le scraping Venice, découpe le Markdown en chunks et résume les preuves utiles.

    def _collect_notes(
        self,
        topic: str,
        queries: list[str],
        results_per_query: int,
        seen_source_keys: set[str],
        seen_content_hashes: set[str],
        notes: list[SourceNote],
        iteration: int,
    ) -> None:
        for query in queries:
            if self.max_sources is not None and len(notes) >= self.max_sources:
                return

            self.progress(f"Searching: {query}")
            try:
                results = self.web.search(query, limit=results_per_query)
            except Exception as exc:
                self._record_error("search", exc, query=query)
                continue

            self.artifacts.write(
                "search_results",
                {"iteration": iteration, "query": query, "results": results},
            )

            for result in results:
                if self.max_sources is not None and len(notes) >= self.max_sources:
                    return

                source_key = result.canonical_url or result.url
                if source_key in seen_source_keys:
                    self.artifacts.write("dedupe", {"reason": "canonical_url", "url": result.url})
                    continue

                seen_source_keys.add(source_key)
                source_id = f"S{len(notes) + 1}"
                note = self._read_source(topic, query, source_id, result, seen_source_keys, seen_content_hashes)
                if note is not None:
                    notes.append(note)

Les échecs individuels de recherche et de récupération ne devraient pas arrêter l’exécution entière. Le web public est désordonné. Certaines pages bloquent le scraping, certaines renvoient des PDF, certaines sont indisponibles et certaines redirigent vers des endroits inattendus. Un agent de recherche devrait continuer à avancer et enregistrer ce qui a échoué. Voici la méthode de lecture de source :

    def _read_source(
        self,
        topic: str,
        query: str,
        source_id: str,
        result: SearchResult,
        seen_source_keys: set[str],
        seen_content_hashes: set[str],
    ) -> SourceNote | None:
        self.progress(f"Reading {source_id}: {result.title}")
        try:
            page = self.web.fetch(result)
        except Exception as exc:
            self._record_error("fetch", exc, query=query, url=result.url, source_id=source_id)
            return None

        if page.content_hash in seen_content_hashes:
            self.artifacts.write(
                "dedupe",
                {"reason": "content_hash", "source_id": source_id, "url": result.url},
            )
            return None
        seen_content_hashes.add(page.content_hash)

        chunks = self._summarize_chunks(topic, query, source_id, page)
        if not chunks:
            self._record_error("summarize_chunk", VeniceError("no chunks could be summarized"), url=result.url)
            return None

        summary = self._summarize_source(topic, query, source_id, page, chunks)
        note = SourceNote(
            source_id=source_id,
            title=page.title,
            url=result.url,
            canonical_url=page.canonical_url,
            final_url=page.final_url,
            query=query,
            rank=result.rank,
            snippet=result.snippet,
            provider=result.provider,
            retrieved_at=page.retrieved_at,
            content_type=page.content_type,
            content_hash=page.content_hash,
            chunks=chunks,
            summary=summary,
        )
        self.artifacts.write("source_notes", note)
        return note

Pour chaque chunk de source, demandez à Venice un court résumé des preuves et des citations exactes :

    def _summarize_chunks(
        self,
        topic: str,
        query: str,
        source_id: str,
        page: WebPage,
    ) -> tuple[EvidenceChunk, ...]:
        evidence: list[EvidenceChunk] = []
        for chunk in page.chunks[: self.max_chunks_per_source]:
            prompt = dedent(
                f"""
                Topic: {topic}
                Search query: {query}
                Source ID: {source_id}
                Chunk ID: {chunk.chunk_id}
                Source title: {page.title}
                Source URL: {page.final_url}

                Source chunk:
                {chunk.text}

                Extract only evidence relevant to the topic.
                Return JSON only in this shape:
                {{"summary": "...", "quotes": ["short exact quote", "..."]}}
                """
            ).strip()

            try:
                response = self.venice.chat(
                    [
                        {"role": "system", "content": SYSTEM_PROMPT},
                        {"role": "user", "content": prompt},
                    ],
                    temperature=0.1,
                    max_tokens=600,
                )
                data = json.loads(response)
                evidence.append(
                    EvidenceChunk(
                        chunk_id=chunk.chunk_id,
                        text=chunk.text,
                        summary=str(data.get("summary", "")).strip(),
                        quotes=tuple(
                            quote.strip()
                            for quote in data.get("quotes", [])
                            if isinstance(quote, str) and quote.strip()
                        ),
                    )
                )
            except Exception as exc:
                self._record_error("summarize_chunk", exc, query=query, url=page.final_url, source_id=source_id)
                continue

        return tuple(evidence)

Puis condensez les résumés de chunks en une note de source :

    def _summarize_source(
        self,
        topic: str,
        query: str,
        source_id: str,
        page: WebPage,
        chunks: tuple[EvidenceChunk, ...],
    ) -> str:
        chunk_digest = _chunk_digest(chunks, max_chars=9000)
        prompt = dedent(
            f"""
            Topic: {topic}
            Search query: {query}
            Source ID: {source_id}
            Source title: {page.title}
            Source URL: {page.final_url}

            Chunk evidence:
            {chunk_digest}

            Synthesize a source note using only the chunk evidence. Include:
            - key facts with dates/numbers where present
            - any limitations or bias in the source
            - useful exact wording from quotes if it is short

            Keep the note under 180 words and refer to the source as [{source_id}].
            """
        ).strip()
        return self.venice.chat(
            [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": prompt},
            ],
            temperature=0.1,
            max_tokens=500,
        )

Cette résumé en deux étapes est ce qui rend l’agent plus fiable qu’un simple script « résumer ces URL ». Le modèle lit d’abord les chunks de la source, puis écrit une note au niveau de la source à partir de ces éléments de preuve extraits.

Rédaction du rapport final

Une fois que l’agent dispose des notes de source, il peut rédiger le rapport. Commencez par un rédacteur de rapport en une seule passe :

    def _write_report(self, topic: str, notes: list[SourceNote]) -> str:
        if not notes:
            return (
                f"# Research report: {topic}\n\n"
                "No usable web sources were collected. Check your network connection or try a narrower topic."
            )

        prompt = dedent(
            f"""
            Research topic:
            {topic}

            Source notes:
            {_source_digest(notes, max_chars=45000)}

            Write a detailed source-backed Markdown research survey.

            Requirements:
            - Start with a precise H1 title.
            - Open with "## Overview".
            - Use topic-specific sections.
            - Use footnote-style citation markers like [^1] and [^2].
            - Do not cite with internal source IDs like [S1] in the report body.
            - Do not include uncited factual claims.
            - Avoid source-cluster capture from one vendor, domain, framework, or viewpoint.
            - Include uncertainty, contradictions, and missing context where relevant.
            - End with "## References" as a numbered list ordered by first citation.
            """
        ).strip()

        return self.venice.chat_stream(
            [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": prompt},
            ],
            temperature=0.2,
            max_tokens=7000,
        )

L’implémentation de référence va plus loin pour les rapports approfondis : elle demande à Venice un plan, rédige chaque section du rapport séparément, puis demande une passe d’édition finale pour assembler le rapport terminé et convertir les ID de source internes en citations de style notes de bas de page. Cette approche par étapes est utile lorsque vous voulez une sortie de recherche en long format car un seul prompt géant compresse souvent trop d’informations. Les prompts mis à jour orientent également le rapport vers une étude large et sourcée plutôt qu’un mince guide de décision. Si la base de sources est biaisée vers un cluster, le prompt de l’éditeur indique à Venice de reconnaître ce biais et de ne pas le présenter comme représentatif de l’ensemble du domaine. Ajoutez les utilitaires de digest :

def _chunk_digest(chunks: tuple[EvidenceChunk, ...], max_chars: int) -> str:
    parts = []
    for chunk in chunks:
        quote_text = "; ".join(chunk.quotes)
        parts.append(
            f"{chunk.chunk_id}: {chunk.summary}"
            + (f"\nQuotes: {quote_text}" if quote_text else "")
        )
    return "\n\n".join(parts)[:max_chars]


def _source_digest(notes: list[SourceNote], max_chars: int) -> str:
    chunks = [
        "\n".join(
            [
                f"[{note.source_id}] {note.title}",
                f"URL: {note.final_url or note.url}",
                f"Canonical URL: {note.canonical_url}",
                f"Found via: {note.query}",
                f"Provider/rank: {note.provider}/{note.rank}",
                f"Retrieved: {note.retrieved_at}",
                f"Content hash: {note.content_hash}",
                f"Note: {note.summary}",
                f"Chunk evidence: {_chunk_digest(note.chunks, max_chars=1000)}",
            ]
        )
        for note in notes
    ]
    return "\n\n".join(chunks)[:max_chars]

Enfin, ajoutez l’enregistrement des erreurs :

    def _record_error(
        self,
        stage: str,
        exc: Exception,
        *,
        query: str = "",
        url: str = "",
        source_id: str = "",
        provider: str = "",
    ) -> None:
        message = str(exc)
        self.progress(f"{stage.replace('_', ' ').title()} failed: {message}")
        self.artifacts.write(
            "errors",
            CollectionError(
                stage=stage,
                message=message,
                query=query,
                url=url,
                source_id=source_id,
                provider=provider,
            ),
        )

À ce stade, la boucle de recherche principale est en place.

Ajout de la CLI

Nous avons maintenant besoin d’un point d’entrée en ligne de commande. Créez main.py :

from __future__ import annotations

import argparse
from pathlib import Path

from dotenv import load_dotenv

from research_agent.agent import (
    DEFAULT_ITERATIONS,
    DEFAULT_MAX_CHUNKS_PER_SOURCE,
    DEFAULT_MAX_SOURCES,
    DEFAULT_QUERY_COUNT,
    DEFAULT_REPORT_STYLE,
    DEFAULT_RESULTS_PER_QUERY,
    ResearchAgent,
)
from research_agent.artifacts import ArtifactWriter
from research_agent.venice import VeniceClient, VeniceError
from research_agent.web import WebSearch


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Run a minimal deep research agent powered by Venice AI.",
    )
    parser.add_argument("topic", nargs="+", help="Research topic, wrapped in quotes for best results.")
    parser.add_argument("--model", help="Venice model name. Defaults to VENICE_MODEL or openai-gpt-55.")
    parser.add_argument("--iterations", type=int, default=DEFAULT_ITERATIONS)
    parser.add_argument("--queries", type=int, default=DEFAULT_QUERY_COUNT)
    parser.add_argument("--results", type=int, default=DEFAULT_RESULTS_PER_QUERY)
    parser.add_argument("--output", "--markdown-output", dest="output", type=Path)
    parser.add_argument("--artifacts", type=Path, help="Optional directory for JSONL research artifacts.")
    parser.add_argument("--providers", default="duckduckgo", help="Comma-separated providers: duckduckgo, arxiv.")
    parser.add_argument("--max-sources", type=int, default=DEFAULT_MAX_SOURCES)
    parser.add_argument("--chunk-chars", type=int, default=3000)
    parser.add_argument("--max-chunks-per-source", type=int, default=DEFAULT_MAX_CHUNKS_PER_SOURCE)
    parser.add_argument(
        "--report-style",
        choices=["brief", "standard", "deep"],
        default=DEFAULT_REPORT_STYLE,
        help=f"Final report depth. Default: {DEFAULT_REPORT_STYLE}.",
    )
    parser.add_argument("--quiet", action="store_true", help="Hide progress messages.")
    return parser.parse_args()

La CLI expose les paramètres que vous ajusterez réellement pendant la recherche :

Option	Ce qu’elle contrôle
`--iterations`	Nombre de passes de recherche
`--queries`	Requêtes de recherche générées par passe
`--results`	Résultats lus par fournisseur pour chaque requête
`--providers`	Fournisseurs de recherche, comme `duckduckgo` ou `duckduckgo,arxiv`
`--max-sources`	Nombre maximal de sources utilisables à collecter
`--chunk-chars`	Taille approximative des chunks avant l’extraction des preuves de source
`--max-chunks-per-source`	Nombre de chunks résumés par source
`--report-style`	Profondeur du rapport final : `brief`, `standard` ou `deep`
`--artifacts`	Répertoire pour les enregistrements d’audit JSONL
`--output`	Chemin pour le rapport Markdown final

Maintenant, reliez le tout :

def main() -> int:
    load_dotenv()
    args = parse_args()
    topic = " ".join(args.topic)

    try:
        venice = VeniceClient.from_env(model=args.model)
        progress = None if args.quiet else lambda message: print(f"[agent] {message}")
        provider_names = [name.strip() for name in args.providers.split(",") if name.strip()]

        with WebSearch.from_provider_names(
            provider_names,
            chunk_chars=args.chunk_chars,
            scraper=venice.scrape,
        ) as web:
            agent = ResearchAgent(
                venice=venice,
                web=web,
                artifacts=ArtifactWriter(args.artifacts),
                progress=progress,
                max_sources=args.max_sources,
                max_chunks_per_source=args.max_chunks_per_source,
                report_style=args.report_style,
            )
            report = agent.run(
                topic,
                iterations=args.iterations,
                query_count=args.queries,
                results_per_query=args.results,
            )
    except ValueError as exc:
        print(f"Configuration error: {exc}")
        return 1
    except VeniceError as exc:
        print(f"Venice API error: {exc}")
        return 1

    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(report.markdown, encoding="utf-8")
        print(f"\nSaved report to {args.output}")
    else:
        print()
        print(report.markdown)

    if report.artifacts_dir:
        print(f"Saved research artifacts to {report.artifacts_dir}")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Cela nous donne une CLI de recherche locale fonctionnelle.

Exécution de l’agent

Exécutez une passe de recherche rapide :

uv run python main.py "How are AI agents changing software engineering workflows?"

Écrivez le rapport dans un fichier Markdown :

uv run python main.py "state of open source LLM inference in 2026" \
  --output reports/inference.md

Utilisez plus de sources et plusieurs fournisseurs :

uv run python main.py "agentic coding research" \
  --providers duckduckgo,arxiv \
  --iterations 3 \
  --queries 5 \
  --results 4 \
  --max-sources 12

Choisissez le style du rapport final :

uv run python main.py "AI agents in software engineering" --report-style deep

Utilisez brief pour un briefing concis sourcé, standard pour une étude plus complète et deep pour le workflow par étapes plan/section/éditeur. Enregistrez des artefacts auditables :

uv run python main.py "privacy tradeoffs in hosted LLM APIs" \
  --output reports/privacy.md \
  --artifacts runs/privacy

Lorsque les artefacts sont activés, vous verrez des fichiers comme :

runs/privacy/
  queries.jsonl
  research_gaps.jsonl
  search_results.jsonl
  fetches.jsonl
  source_chunks.jsonl
  chunk_summaries.jsonl
  source_notes.jsonl
  dedupe.jsonl
  errors.jsonl
  report_outline.jsonl
  report_sections.jsonl
  report_editor.jsonl
  reports.jsonl

Ces fichiers sont utiles lorsque vous voulez comprendre comment l’agent est arrivé à une conclusion. Par exemple, source_notes.jsonl montre les preuves de source résumées, research_gaps.jsonl montre pourquoi des recherches complémentaires ont été générées, et errors.jsonl montre les pages qui ont échoué pendant la recherche, le scraping ou le résumé.

Notes sur la confidentialité et la fiabilité

Un agent de recherche touche à plusieurs systèmes, il est donc utile d’être précis sur ce qui va où :

Limites de données de l'agent de recherche privé

Couche	Ce qui voit les données
CLI locale	Le sujet, la configuration, les notes de source, les artefacts et les rapports finaux restent sur votre machine
Fournisseur de recherche	Les requêtes de recherche sont envoyées au fournisseur que vous choisissez, comme DuckDuckGo ou arXiv
Scraping Venice	Les URL des sources publiques sont envoyées à l’endpoint de scraping de Venice
Chat completions Venice	Les prompts, les chunks de source, les notes de source et les instructions de génération de rapport sont envoyés à Venice
Fichiers de sortie	Les rapports Markdown et les artefacts JSONL sont écrits localement

Si vous voulez garder une plus grande partie du chemin de recherche à l’intérieur de Venice, vous pouvez adapter la couche de fournisseurs pour appeler l’endpoint POST /augment/search de Venice au lieu d’interroger directement DuckDuckGo. L’implémentation de référence utilise des fournisseurs publics légers afin que la démo reste facile à exécuter et à comprendre. Pour la fiabilité, gardez ces valeurs par défaut prudentes :

Utilisez des retries pour les appels à Venice et les requêtes web.
Ajoutez un petit --request-delay si vous lisez de nombreuses pages du même hôte.
Plafonnez --max-sources pour que les sujets larges ne s’exécutent pas indéfiniment.
Enregistrez --artifacts pour les rapports importants afin de pouvoir auditer la sortie finale.
Traitez le rapport comme un briefing, pas comme une vérité absolue. Suivez les citations jusqu’à la source originale lorsque la précision est importante.

Tester les composants

Vous n’avez pas besoin de requêtes web en direct ni d’appels Venice pour tester la plupart du système. Le dépôt de référence utilise de fausses classes Venice et web pour tester la boucle de recherche, le comportement de déduplication, les artefacts et les prompts de rapport. Un premier test utile est la canonicalisation des URL :

from research_agent.models import canonicalize_url


def test_canonicalize_url_removes_tracking_params():
    url = "https://example.com/post?utm_source=x&b=2&a=1#section"
    assert canonicalize_url(url) == "https://example.com/post?a=1&b=2"

Testez ensuite que le contenu dupliqué est ignoré :

from research_agent.models import SearchResult, WebPage, chunk_text


class FakeWeb:
    def search(self, query: str, limit: int = 5) -> list[SearchResult]:
        return [
            SearchResult(title="First source", url="https://example.com/a", snippet="snippet"),
            SearchResult(title="Mirror", url="https://example.com/b", snippet="snippet"),
        ]

    def fetch(self, result: SearchResult) -> WebPage:
        text = "This page contains relevant evidence. " * 5
        return WebPage(
            title=result.title,
            url=result.url,
            final_url=result.url,
            text=text,
            content_hash="same-content",
            chunks=chunk_text(text, chunk_chars=80, overlap=10),
        )

Les faux rendent les tests d’agent beaucoup plus rapides et moins instables. Vous pouvez vérifier la logique d’orchestration sans dépendre de résultats de recherche en direct, des conditions du réseau ou de la sortie du modèle.

Benchmarking

De nombreux fournisseurs d’IA ont maintenant leurs propres workflows de recherche approfondie, donc le dépôt de référence inclut un benchmark simple comparé à l’outil Deep Research de Perplexity. Les deux agents ont été chargés de rédiger un rapport sur l’architecture des frameworks d’agents IA, puis les rapports générés ont été commités dans le dépôt GitHub. Ceci n’est pas censé être un benchmark formel. C’est une façon pratique d’inspecter la structure du rapport, la couverture des sources, la qualité des citations et de voir si l’agent se concentre trop sur un cluster de sources. C’est aussi pour cela que l’implémentation mise à jour suit research_gaps.jsonl et l’équilibre des sources avant les recherches complémentaires.

Étendre cet exemple

Une fois que l’agent de base fonctionne, voici des moyens pratiques de l’améliorer :

Ajouter un fournisseur de recherche Venice en utilisant POST /augment/search.
Stocker les rapports et les artefacts dans une petite base de données SQLite au lieu de fichiers JSONL.
Ajouter des allowlists ou blocklists de sources pour les domaines de recherche de confiance.
Ajouter le support PDF en combinant le scraping Venice avec l’analyse de documents pour les sources qui n’exposent pas de HTML propre.
Ajouter un ensemble d’évaluation de sujets et de types de sources attendus afin de pouvoir comparer la qualité de la recherche après des modifications de prompts.
Ajouter une étape de relecture qui demande à Venice de trouver les affirmations non étayées dans le rapport final avant de l’enregistrer.

La plus grosse amélioration concerne généralement la sélection des sources. La génération de requêtes aide, mais vous pouvez également améliorer la qualité en privilégiant les sources primaires, les documents de normes, les docs officielles, les papiers, les changelogs et les pages de jeux de données plutôt que les résumés à faible signal.

Pour finir

Merci d’avoir lu ! J’espère que cela vous a aidé à créer un agent de recherche privé pratique avec Python et l’API Venice. Le modèle utile ici n’est pas seulement « demander à un modèle de rechercher quelque chose ». C’est de décomposer la recherche en étapes auditables : planifier les recherches, collecter les sources, extraire les preuves, écrire des notes de source, suivre les lacunes et synthétiser avec des citations. En gardant ces étapes explicites, nous obtenons un workflow de recherche plus facile à inspecter, à tester et à améliorer au fil du temps.

​Ce que nous allons construire

​Configuration du projet

​Création des modèles de données

​Construction du client Venice

​Ajout des fournisseurs de recherche

​Écriture des artefacts locaux

​Construction de l’agent de recherche

​Planification des recherches initiales et complémentaires

​Lecture et résumé des sources

​Rédaction du rapport final

​Ajout de la CLI

​Exécution de l’agent

​Notes sur la confidentialité et la fiabilité

​Tester les composants

​Benchmarking

​Étendre cet exemple

​Pour finir