Einen privaten Research-Agenten bauen

Research-Agenten sind nützlich, wenn du mehr willst als ein einzelnes Suchergebnis oder eine schnelle Modellantwort. Ein guter Research-Agent kann ein breites Thema in Suchanfragen verwandeln, Quellen sammeln, die wichtigen Belege extrahieren, Lücken nachgehen und ein zitiertes Briefing schreiben, das du danach prüfen kannst.

In diesem Tutorial bauen wir einen privaten Research-Agenten mit Python und der Venice-API. Am Ende hast du ein CLI, das ein Thema recherchieren, öffentliche Seiten in Markdown scrapen, Quell-Chunks zusammenfassen, gap-bewusste Folgesuchen ausführen und einen zitierten Bericht mit optionalen lokalen JSONL-Artefakten erzeugen kann.

Interesse an der vollständigen Implementierung? Schau in das GitHub-Repo.

Bevor wir loslegen, brauchst du einen Venice API-Schlüssel:

export VENICE_API_KEY=<my-key>

​

Was wir bauen

Die Referenzimplementierung ist ein kleines Python-Projekt mit einigen klar getrennten Teilen:

Teil	Was es tut
CLI	Nimmt Research-Thema, Modell, Provider, Tiefen-Settings, Output-Pfad und Artefakt-Verzeichnis entgegen
Venice-Client	Ruft Chat-Completions, Streaming-Chat-Completions und POST /augment/scrape auf
Such-Layer	Sucht per Default auf DuckDuckGo, optional mit arXiv-Paper-Discovery
Datenmodelle	Verfolgt Quell-URLs, kanonische URLs, Chunks, Belege, Notizen, Fehler und Reports
Research-Agent	Plant Suchen, liest Quellen, extrahiert Belege, analysiert Lücken, erzeugt Folgequeries und schreibt den finalen Report
Artefakt-Writer	Speichert auditierbare JSONL-Records für Queries, Research-Gaps, Ergebnisse, Fetches, Chunks, Source-Notes, Report-Drafts, Fehler und Reports

Der Flow sieht so aus:

1. Venice bitten, vielfältige Suchanfragen zum Thema zu generieren.
2. Im Web mit einem oder mehreren Providern suchen.
3. URLs vor dem Lesen deduplizieren.
4. Mit Venices Scrape-Endpoint jede öffentliche Seite in Markdown verwandeln.
5. Lange Seiten in Chunks splitten.
6. Venice bitten, aus jedem Chunk Belege zu extrahieren.
7. Venice bitten, aus den Chunk-Belegen Source-Notes zu machen.
8. Research-Gaps und Source-Balance-Probleme identifizieren, bevor Folgequeries erzeugt werden.
9. Venice bitten, den finalen Report mit Fußnoten-Zitaten zu synthetisieren.

Das ist „privat” im praktischen Sinn: Der Agent hält Orchestrierung, Source-Notes, Artefakte und finale Reports auf deinem Rechner. Venice übernimmt die Modellaufrufe und das Scraping über seine API. Die Default-Referenzimplementierung schickt Suchanfragen weiterhin an DuckDuckGo oder arXiv, behandle die Provider-Wahl also als Teil deines Privacy-Designs.

​

Projekt einrichten

Das Referenzprojekt nutzt Python 3.13 und uv, derselbe Code funktioniert aber auch mit einer klassischen Virtual Environment.

Neues Projekt anlegen:

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

Abhängigkeiten installieren:

uv add httpx beautifulsoup4 python-dotenv

Wer lieber pip nutzt, legt eine Virtual Environment an und installiert dieselben Pakete:

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

Lege eine .env-Datei für die lokale Entwicklung an:

VENICE_API_KEY=your_venice_api_key_here
VENICE_MODEL=openai-gpt-55

Wir nutzen VENICE_MODEL, damit du das Modell ohne Code-Änderungen wechseln kannst. Die Referenz nimmt aktuell standardmäßig openai-gpt-55, du kannst es aber gegen ein anderes für dein Venice-Konto verfügbares Chat-Modell tauschen.

​

Datenmodelle erstellen

Bevor wir die Agent-Logik schreiben, definieren wir die Objekte, die durch die Pipeline laufen. Diese Modelle machen den Rest des Codes leichter nachvollziehbar, weil jede Quelle Provenienz mit sich trägt: woher sie stammt, welche Query sie gefunden hat, wann sie abgerufen wurde und wie sie gechunkt wurde.

Erstelle 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)

Wichtige Felder sind hier canonical_url, content_hash und chunks.

Mit canonical_url vermeidet der Agent, dieselbe Quelle mehrfach zu lesen, wenn sich Suchergebnisse nur in Tracking-Parametern oder Fragments unterscheiden. content_hash hilft, Duplikate auch dann zu erkennen, wenn sie unter anderen URLs liegen. chunks erlaubt es, lange Seiten in kleineren Stücken zusammenzufassen, statt nützliche Belege an Kontextlimits zu verlieren.

Ergänze die Helferfunktionen unter den 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

Das Chunking ist hier bewusst einfach: feste Zeichen-Chunks mit Überlappung. Das reicht für einen Demo-Research-Agenten, weil Venices Scrape-Endpoint Markdown zurückgibt, das in der Regel viel sauberer ist als rohes HTML. Für Produktions-Research an langen technischen Dokumenten kannst du das verbessern, indem du an Überschriften, Absätzen oder Token-Counts splittest.

​

Den Venice-Client bauen

Als Nächstes erstellen wir einen kleinen Venice-Client. Du könntest das OpenAI-Python-SDK für Chat-Completions verwenden, weil Venice OpenAI-kompatibel ist – die Referenz nutzt aber direkt httpx, damit derselbe Client auch Venices POST /augment/scrape-Endpoint aufrufen kann.

Erstelle 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,
        )

Der from_env()-Helper hält Secrets aus deinem Quellcode raus. Außerdem ist die lokale Entwicklung angenehmer, weil python-dotenv VENICE_API_KEY und VENICE_MODEL aus .env laden kann.

Jetzt Chat-Completions ergänzen:

    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

Für den finalen Report wollen wir Streaming nutzen, weil tiefe Berichte deutlich länger dauern können (sie produzieren viel mehr Text). Das kann zu Timeout-Problemen bei Anfragen führen, deren finale Ausgabe extrem lange dauert. Mit Streaming umgehen wir das und machen den Request robuster gegen Timeout-Fehler:

    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()

Dann Scraping ergänzen:

    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",
        )

Venices Scrape-Endpoint akzeptiert eine öffentlich zugängliche URL und liefert die Seite als Markdown. Damit muss das Modell kein rohes HTML parsen, und deine Source-Extraction-Prompts können mit saubereren Texten arbeiten.

Der verbleibende Helper kümmert sich um Retries und Response-Parsing:

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

Das vollständige Repo enthält außerdem einen robusten _post_chat_stream()-Helper, der Server-Sent Events aus Streaming-Chat-Completions liest. Du kannst zunächst ohne Streaming starten und es ergänzen, sobald der restliche Research-Flow läuft.

​

Such-Provider hinzufügen

Der Such-Layer hat zwei Aufgaben: Quell-URLs finden und diese URLs durch den Venice-Scraper holen. Die Referenzimplementierung nutzt DuckDuckGos HTML-Endpoint für allgemeine Websuche und arXivs Atom-API für Papers.

Erstelle 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

Jetzt DuckDuckGo ergänzen:

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

Und 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

Die WebSearch-Klasse koordiniert Provider und holt Seiten:

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)

Die vollständige Referenz ergänzt Retries, Host-bezogene Request-Delays und freundlichere Fehlermeldungen. Das ist wertvoll, weil Research-Agenten viel Zeit mit Seiten verbringen, die Automatisierung blockieren, unerwartet umleiten oder transiente Fehler liefern.

Ergänze unten die kleinen Provider-Helper:

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

​

Lokale Artefakte schreiben

Für Research-Workflows zählt Auditierbarkeit. Wenn der finale Report etwas Überraschendes sagt, willst du nachvollziehen können, welche Quelle dazu geführt hat.

Erstelle 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

Das schreibt ein JSON-Objekt pro Zeile, was die Artefakte leicht anhängbar, inspizierbar und später per Kommandozeilen-Tools verarbeitbar macht.

​

Den Research-Agenten bauen

Da wir Venice, Suche, Modelle und Artefakte haben, können wir den eigentlichen Agenten bauen.

Erstelle 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

Der System-Prompt ist die zentrale Verhaltens-Leitplanke. Wir wollen keinen beeindruckend klingenden Report aus dem Modellgedächtnis. Wir wollen, dass es das Quellmaterial nutzt und Unsicherheit benennt, wenn die Belege dünn sind.

Wir brauchen außerdem zwei finale Dataclasses in models.py, falls sie noch nicht da sind:

@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

Als Nächstes definieren wir den 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

Die run()-Methode koordiniert die Research-Passes:

    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,
        )

Die zwei seen_*-Sets verhindern, dass der Agent Zeit mit Duplikatquellen verschwendet. URL-Dedupe fängt wiederholte Links. Content-Hash-Dedupe fängt Spiegelseiten, syndizierte Posts und Seiten, die auf denselben Endinhalt umleiten.

​

Initiale und Folge-Suchen planen

Der erste Modell-Aufruf macht aus dem Thema Suchanfragen:

    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])

Nach jedem Research-Pass führt der aktualisierte Agent einen bewussteren Gap-Analyse-Schritt durch. Er schaut auf die aktuellen Notizen, zählt Source-Cluster nach Domain, fragt Venice, was an Coverage fehlt, schreibt diese Gaps in Artefakte und verwendet die resultierenden Queries für den nächsten Pass.

Beginne damit, die Source-Balance zu verfolgen:

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)

Das gibt dem Agent einen einfachen Weg, Source-Cluster-Capture zu erkennen. Wenn alle Quellen aus einer Firma, einem Framework oder einer Domain kommen, sollten Folgequeries die Quellbasis bewusst erweitern, statt mehr vom Gleichen zu sammeln.

Diese Balance-Information jetzt beim Erstellen der Folgesuchen nutzen:

    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])

Die neuere Referenz wickelt das in _gap_follow_up_queries(), das Venice bittet, sowohl Gap-Records als auch Queries zurückzugeben:

    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]

Mit --artifacts werden diese Records nach research_gaps.jsonl geschrieben. Das gibt dir eine nützliche Audit-Spur, warum der Agent in einem zweiten Pass nach einer bestimmten Query gesucht hat.

Der Parser sollte tolerant sein. Liefert das Modell fehlerhaftes JSON, fällt der Agent auf das Originalthema zurück:

    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]

Dieses Muster ist im Agent-Code überall hilfreich: strukturierte Ausgabe verlangen, parsen, und einen einfachen Fallback bereitstellen, wenn die Ausgabe unbrauchbar ist.

​

Quellen lesen und zusammenfassen

Jetzt sammeln wir Source-Notes. Der Agent sucht pro Query, holt jedes Ergebnis durch Venice Scrape, chunkt das Markdown und fasst die nützlichen Belege zusammen.

    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)

Einzelne Such- und Fetch-Fehler sollen nicht den ganzen Lauf stoppen. Das öffentliche Web ist chaotisch. Manche Seiten blockieren Scraping, manche liefern PDFs, manche sind down, manche leiten unerwartet um. Ein Research-Agent sollte weitermachen und festhalten, was fehlgeschlagen ist.

So sieht die Methode zum Lesen einer Quelle aus:

    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

Für jeden Quell-Chunk Venice nach einer kurzen Belegzusammenfassung und exakten Zitaten fragen:

    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)

Dann die Chunk-Zusammenfassungen in eine Source-Note zusammenführen:

    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,
        )

Diese zweistufige Zusammenfassung ist der Teil, der den Agenten verlässlicher wirken lässt als ein einfaches „Fasse diese URLs zusammen”-Skript. Das Modell liest erst die Quell-Chunks, dann schreibt es aus den extrahierten Belegen eine Source-Note.

​

Den finalen Report schreiben

Sobald der Agent Source-Notes hat, kann er den Report schreiben. Starten wir mit einem einfachen Single-Pass-Report-Writer:

    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,
        )

Die Referenz geht bei tiefen Reports weiter: Sie bittet Venice um eine Gliederung, schreibt jeden Abschnitt einzeln und lässt am Ende einen Editor-Pass den fertigen Report zusammenbauen und interne Source-IDs in Fußnoten-Zitate umwandeln.

Dieser gestaffelte Ansatz ist nützlich für Long-Form-Research, weil ein einziger riesiger Prompt oft zu stark komprimiert. Die aktualisierten Prompts drängen den Report außerdem in Richtung breiter, quellenbasierter Übersicht statt eines dünnen Entscheidungsleitfadens. Ist die Quellbasis zu einem Cluster verschoben, weist der Editor-Prompt Venice an, diese Schieflage anzuerkennen und sie nicht als repräsentativ für das gesamte Feld darzustellen.

Ergänze die Digest-Helper:

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]

Zum Abschluss Fehler-Aufzeichnung ergänzen:

    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,
            ),
        )

Damit steht der Kern-Research-Loop.

​

Das CLI ergänzen

Jetzt brauchen wir einen Command-Line-Einstieg. Erstelle 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()

Das CLI macht die Stellschrauben verfügbar, die du beim Recherchieren tatsächlich anpasst:

Option	Was sie steuert
--iterations	Anzahl der Research-Passes
--queries	Pro Pass generierte Suchanfragen
--results	Pro Query und Provider gelesene Ergebnisse
--providers	Suchprovider, z. B. duckduckgo oder duckduckgo,arxiv
--max-sources	Maximale Anzahl nutzbarer Quellen
--chunk-chars	Ungefähre Chunk-Größe vor der Belegextraktion
--max-chunks-per-source	Anzahl der pro Quelle zusammengefassten Chunks
--report-style	Tiefe des finalen Reports: brief, standard oder deep
--artifacts	Verzeichnis für JSONL-Audit-Records
--output	Pfad für den finalen Markdown-Report

Jetzt alles verdrahten:

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())

Damit haben wir ein lauffähiges lokales Research-CLI.

​

Den Agenten ausführen

Schneller Research-Pass:

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

Report in eine Markdown-Datei schreiben:

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

Mehr Quellen und mehrere Provider nutzen:

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

Den finalen Report-Stil wählen:

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

Verwende brief für ein knappes, quellenbasiertes Briefing, standard für eine umfangreichere Übersicht und deep für den gestaffelten Outline-/Section-/Editor-Workflow.

Auditierbare Artefakte speichern:

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

Wenn Artefakte aktiviert sind, siehst du Dateien wie:

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

Diese Dateien sind nützlich, wenn du verstehen willst, wie der Agent zu einer Schlussfolgerung gekommen ist. So zeigt source_notes.jsonl die zusammengefassten Quellbelege, research_gaps.jsonl zeigt, warum Folgequeries entstanden sind, und errors.jsonl zeigt Seiten, die beim Suchen, Scrapen oder Zusammenfassen scheiterten.

​

Privacy- und Reliability-Hinweise

Ein Research-Agent berührt mehrere Systeme – es lohnt sich, präzise zu sein, was wohin geht:

Schicht	Wer sieht die Daten
Lokales CLI	Thema, Konfiguration, Source-Notes, Artefakte und finale Reports bleiben auf deinem Rechner
Suchprovider	Suchanfragen gehen an den gewählten Provider, z. B. DuckDuckGo oder arXiv
Venice Scrape	Öffentliche Quell-URLs werden an Venices Scrape-Endpoint geschickt
Venice Chat Completions	Prompts, Quell-Chunks, Source-Notes und Report-Anweisungen werden an Venice gesendet
Output-Dateien	Markdown-Reports und JSONL-Artefakte werden lokal geschrieben

Willst du mehr vom Suchpfad innerhalb von Venice halten, kannst du den Provider-Layer anpassen, sodass er Venices POST /augment/search-Endpoint nutzt statt direkt DuckDuckGo. Die Referenz nutzt schlanke öffentliche Provider, damit die Demo leicht lauffähig und verständlich bleibt.

Für Reliability die Defaults konservativ halten:

• Retries für Venice-Aufrufe und Web-Requests nutzen.
• Ein kleines --request-delay ergänzen, wenn du viele Seiten vom selben Host liest.
• --max-sources deckeln, damit breite Themen nicht endlos laufen.
• --artifacts für wichtige Reports speichern, um die finale Ausgabe auditieren zu können.
• Den Report als Briefing behandeln, nicht als Wahrheit. Bei wichtigen Aussagen den Zitaten zur Originalquelle folgen.

​

Die Teile testen

Du brauchst keine echten Web-Requests oder Venice-Aufrufe, um den Großteil des Systems zu testen. Die Referenz nutzt Fake-Venice- und Fake-Web-Klassen, um Research-Loop, Dedupe-Verhalten, Artefakte und Report-Prompts zu testen.

Ein guter erster Test ist die URL-Kanonisierung:

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"

Dann testen, dass Duplikat-Content übersprungen wird:

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),
        )

Fakes machen Agent-Tests deutlich schneller und stabiler. Du kannst die Orchestrierungs-Logik prüfen, ohne dich auf Live-Suchergebnisse, Netzwerkbedingungen oder Modell-Output zu verlassen.

​

Benchmarking

Viele KI-Provider haben mittlerweile eigene Deep-Research-Workflows, deshalb enthält die Referenz einen einfachen Benchmark gegen Perplexitys Deep Research. Beide Agenten wurden gebeten, einen Report über die Architektur von KI-Agent-Frameworks zu schreiben, und die erzeugten Reports liegen im GitHub-Repo.

Das ist kein formaler Benchmark, sondern ein praktischer Weg, Report-Struktur, Quellabdeckung, Zitationsqualität und etwaige Fokussierung auf einen Source-Cluster zu inspizieren. Genau deshalb verfolgt die aktualisierte Implementierung research_gaps.jsonl und die Source-Balance vor Folgesuchen.

​

Dieses Beispiel erweitern

Sobald der Baseline-Agent läuft, gibt es praktische Verbesserungen:

• Einen Venice-Suchprovider mit POST /augment/search ergänzen.
• Reports und Artefakte in einer kleinen SQLite-DB statt in JSONL-Dateien ablegen.
• Allow- oder Blocklisten für vertrauenswürdige Research-Domains.
• PDF-Support, indem Venice Scrape mit Document-Parsing für Quellen ohne sauberes HTML kombiniert wird.
• Ein Evaluations-Set aus Themen und erwarteten Quelltypen, um Research-Qualität nach Prompt-Änderungen zu vergleichen.
• Einen Review-Schritt, der Venice bittet, unsupportete Claims im finalen Report zu finden, bevor er gespeichert wird.

Das größte Upgrade ist meist bessere Quellauswahl. Query-Generierung hilft, du kannst die Qualität aber auch verbessern, indem du primäre Quellen, Standards-Dokumente, offizielle Docs, Papers, Changelogs und Datensatz-Seiten gegenüber Low-Signal-Zusammenfassungen bevorzugst.

​

Abschluss

Danke fürs Lesen! Hoffentlich hat dir das geholfen, einen praktischen privaten Research-Agenten mit Python und der Venice-API zu bauen.

Das nützliche Muster hier ist nicht nur „ein Modell etwas recherchieren lassen”. Es ist das Zerlegen der Recherche in auditierbare Schritte: Suchen planen, Quellen sammeln, Belege extrahieren, Source-Notes schreiben, Lücken nachgehen und mit Zitaten synthetisieren. Durch die explizite Trennung dieser Schritte entsteht ein Research-Workflow, der leichter zu inspizieren, zu testen und über die Zeit zu verbessern ist.