title: "Исследовательские агенты: веб-поиск, синтез из множества источников и цепочки цитирования" slug: research-agents-web-synthesis-2026-ru date: 2026-02-22 lang: ru

Исследовательские агенты: веб-поиск, синтез из множества источников и цепочки цитирования

Ключевые факты

• tavily-python: Последняя версия 0.7.22 (PyPI на 2026-03-01)
• langchain: Последняя версия 1.2.10
• langgraph: Последняя версия 1.0.10
• duckduckgo-search: Последняя версия 8.1.1
• newspaper3k: Последняя версия 0.2.8
• Tavily Search API: Возвращает до 10 результатов на запрос по умолчанию; max_results настраивается 1-20
• Tavily API время ответа: Обычно 800-1500ms для базового поиска, 1500-3000ms для продвинутого
• Tavily ценообразование: $0.001 на поиск для базового тарифа; объёмные тарифы доступны свыше 10K поисков/месяц
• Perplexity API модели: sonar, sonar-pro, sonar-reasoning (Q1 2026)
• Perplexity API стоимость: Приблизительно $0.005-0.015 за запрос в зависимости от модели и длины контекста
• Brave Search API: Бесплатный тариф 2000 запросов/месяц; платный тариф от $5/месяц
• Ограничение частоты при веб-скрейпинге: Стандартная практика 1-2 запроса/секунду на домен; используйте задержку между запросами
• robots.txt соответствие: Требуется перед скрейпингом; проверьте /robots.txt и уважайте директиву Crawl-delay
• Формат цитирования: Сохраняйте (source_url, claim_text, extraction_timestamp) для каждого утверждения
• Отслеживание цитирований: Поддерживайте DAG (directed acyclic graph) связывающий утверждения → источники → ссылки
• Окно контекста для синтеза: Выделите 16K-32K токенов для синтеза из нескольких источников с Claude; зарезервируйте 4K на вывод
• Типичное разложение исследовательского запроса: Сложный вопрос → 3-8 подвопросов → 15-40 отдельных поисковых запросов
• Пороги качества источников: Оценка ≥0.7 для первичных источников, ≥0.4 для вспомогательного материала
• Параллельное выполнение поиска: Запускайте поиски в Tavily, Brave, Serper одновременно используя asyncio.gather()
• Дедупликация: Используйте канонизацию URL (удалите параметры запроса, фрагменты) перед дедупликацией по URL

Что такое исследовательский агент

Исследовательский агент — это автономная система, которая выполняет итеративный сбор и синтез информации, отличаясь от статических RAG систем, которые запрашивают фиксированный корпус. Основной цикл следует этому паттерну:

Разложение запроса → Параллельный поиск → Оценка качества источников → Извлечение утверждений → Кросс-источниковая верификация → Обнаружение пробелов → Дополнительные запросы → Синтез

В отличие от чат-бота или RAG системы, исследовательский агент:

• Динамически планирует многошаговые стратегии поиска на основе начальных результатов
• Оценивает надёжность источников и проверяет перекрёстные ссылки на утверждения
• Обнаруживает противоречия между источниками и явно их выявляет
• Рекурсивно следует цепочкам цитирования для обнаружения первичных источников
• Производит структурированные отчёты с проверяемыми цитированиями

Агент работает циклами: выполняет поиски, оценивает найденное, выявляет пробелы в охвате, формулирует новые запросы для заполнения пробелов, повторяет до достижения достаточной глубины. Эта итеративная доработка отличает исследовательские агенты от однопроходных систем извлечения.

Каркас принятия решений

Выберите правильный подход на основе потребностей в информации:

Когда Perplexity API подходит: Вам нужны ответы, обоснованные веб-данными, без построения собственной инфраструктуры, не требуется кастомизация цитирования и вы готовы к проприетарной модели.

Когда RAG подходит: Ваша информационная вселенная ограничена и управляема, не меняется быстро, и вам нужна предсказуемая latency и стоимость.

Когда исследовательский агент подходит: Вам нужен анализ публикационного качества с явными цепочками цитирования, обнаружением противоречий и рекурсивной глубиной, и у вас есть инженерные ресурсы для поддержки pipeline.

Таблица справочников параметров

Распространённые ошибки

Ошибка 1: Не дедупликация по canonical URL

Влияние: Один источник считается несколько раз, раздутый подсчёт источников, потраченная обработка.

❌ Неправильно:

all_results = []
for results in results_per_source:
    all_results.extend(results)  # No deduplication

return all_results

✅ Правильно:

from urllib.parse import urlparse, urlunparse

def canonicalize_url(url: str) -> str:
    parsed = urlparse(url)
    # Remove query params and fragment
    canonical = urlunparse((parsed.scheme, parsed.netloc, parsed.path, '', '', ''))
    return canonical.lower()

seen_urls = set()
unique_results = []
for r in all_results:
    canonical = canonicalize_url(r.url)
    if canonical not in seen_urls:
        seen_urls.add(canonical)
        unique_results.append(r)

Ошибка 2: Выполнение поисков последовательно вместо параллельного

Влияние: Исследовательский цикл в 3 раза медленнее; latency увеличивается пропорционально.

❌ Неправильно:

results_tavily = await search_tavily(query)
results_brave = await search_brave(query)
results_serper = await search_serper(query)
# 3 sequential awaits = 3x latency

✅ Правильно:

tasks = [
    search_tavily(query),
    search_brave(query),
    search_serper(query)
]
results_per_source = await asyncio.gather(*tasks, return_exceptions=True)
# All execute concurrently; latency = max(individual latencies)

Ошибка 3: Неграмотная обработка ошибок API

Влияние: Сбой одного API разрушает весь pipeline исследования; не возвращаются частичные результаты.

❌ Неправильно:

results_per_source = await asyncio.gather(*tasks)
# Raises exception if any task fails

✅ Правильно:

results_per_source = await asyncio.gather(*tasks, return_exceptions=True)

all_results = []
for results in results_per_source:
    if isinstance(results, Exception):
        continue  # Skip failed source, continue with others
    all_results.extend(results)

Ошибка 4: Извлечение утверждений без атрибуции источника

Влияние: Невозможно проверить утверждения, нет цепочки цитирования, отчёт неверифицируем.

❌ Неправильно:

claims = [
    {"claim": "LLMs achieve 94% accuracy", "subject": "LLM performance"}
]
# No source URL attached

✅ Правильно:

for claim in claims:
    claim["source_url"] = source_url
    claim["extraction_timestamp"] = datetime.now(timezone.utc).isoformat()
# Every claim tied to origin

Ошибка 5: Игнорирование robots.txt перед веб-скрейпингом

Влияние: Юридические/этические нарушения, блокировка IP, скрейпер заблокирован сайтом.

❌ Неправильно:

async with session.get(url) as response:
    html = await response.text()
# No robots.txt check

✅ Правильно:

import urllib.robotparser

rp = urllib.robotparser.RobotFileParser()
rp.set_url(f"{parsed_url.scheme}://{parsed_url.netloc}/robots.txt")
rp.read()

if rp.can_fetch("*", url):
    async with session.get(url) as response:
        html = await response.text()
else:
    # Skip this URL; robots.txt disallows
    pass

Интеграция поисковых инструментов

Исследовательские агенты интегрируют множество поисковых API для максимизации охвата и снижения систематической погрешности. Различные поисковые движки индексируют разный контент и ранжируют результаты по-разному.

Tavily Search API

Оптимизирован для AI исследовательских случаев. Возвращает структурированный JSON с полным извлечением содержимого страницы.

import aiohttp
from dataclasses import dataclass
from typing import Optional

@dataclass
class SearchResult:
    url: str
    title: str
    snippet: str
    content: str
    source: str
    published_date: Optional[str]
    relevance_score: float

async def search_tavily(
    api_key: str,
    query: str,
    max_results: int = 5,
    search_depth: str = "advanced"
) -> list[SearchResult]:
    """
    Tavily search with full content extraction.

    Args:
        api_key: Tavily API key
        query: Search query string
        max_results: Number of results (1-20)
        search_depth: "basic" or "advanced"
    """
    async with aiohttp.ClientSession() as session:
        payload = {
            "api_key": api_key,
            "query": query,
            "search_depth": search_depth,
            "include_answer": True,
            "include_raw_content": True,
            "max_results": max_results
        }

        async with session.post(
            "https://api.tavily.com/search",
            json=payload
        ) as response:
            data = await response.json()

            results = []
            for r in data.get("results", []):
                results.append(SearchResult(
                    url=r["url"],
                    title=r["title"],
                    snippet=r.get("content", "")[:500],
                    content=r.get("raw_content", r.get("content", "")),
                    source="tavily",
                    published_date=r.get("published_date"),
                    relevance_score=r.get("score", 0.5)
                ))

            return results

DuckDuckGo Search (пакет duckduckgo-search)

Бесплатный API без требования ключа. Хорош для широких запросов, менее структурированный вывод.

from duckduckgo_search import DDGS

def search_duckduckgo(query: str, max_results: int = 10) -> list[SearchResult]:
    """
    DuckDuckGo search via duckduckgo-search package.
    No API key required; rate-limited.
    """
    results = []
    with DDGS() as ddgs:
        for r in ddgs.text(query, max_results=max_results):
            results.append(SearchResult(
                url=r["link"],
                title=r["title"],
                snippet=r["body"],
                content=r["body"],  # DDG returns snippets, not full content
                source="duckduckgo",
                published_date=None,
                relevance_score=0.5  # DDG doesn't expose scores
            ))
    return results

Google Search через Serper API

Программный доступ к результатам Google. Хорош для авторитетных источников и академического контента.

async def search_serper(
    api_key: str,
    query: str,
    max_results: int = 5,
    search_type: str = "search"
) -> list[SearchResult]:
    """
    Google search via Serper API.

    Args:
        search_type: "search" for web, "scholar" for academic
    """
    async with aiohttp.ClientSession() as session:
        payload = {
            "q": query,
            "num": max_results,
            "gl": "us",
            "hl": "en"
        }

        headers = {
            "X-API-KEY": api_key,
            "Content-Type": "application/json"
        }

        endpoint = f"https://google.serper.dev/{search_type}"

        async with session.post(
            endpoint,
            json=payload,
            headers=headers
        ) as response:
            data = await response.json()

            results = []
            for r in data.get("organic", []):
                results.append(SearchResult(
                    url=r["link"],
                    title=r["title"],
                    snippet=r.get("snippet", ""),
                    content=r.get("snippet", ""),
                    source="serper",
                    published_date=r.get("date"),
                    relevance_score=1.0 / (r.get("position", 10))
                ))

            return results

Параллельное выполнение по нескольким источникам

Комбинируйте несколько поисковых API для максимизации охвата:

async def search_all_sources(
    query: str,
    tavily_key: str,
    serper_key: str,
    max_per_source: int = 5
) -> list[SearchResult]:
    """Execute searches across all sources in parallel."""
    tasks = [
        search_tavily(tavily_key, query, max_per_source),
        search_serper(serper_key, query, max_per_source)
    ]

    results_per_source = await asyncio.gather(*tasks, return_exceptions=True)

    all_results = []
    for results in results_per_source:
        if isinstance(results, Exception):
            continue  # Skip failed sources
        all_results.extend(results)

    # Deduplicate by canonical URL
    from urllib.parse import urlparse, urlunparse

    def canonicalize_url(url: str) -> str:
        parsed = urlparse(url)
        return urlunparse((parsed.scheme, parsed.netloc, parsed.path, '', '', '')).lower()

    seen_urls = set()
    unique_results = []
    for r in all_results:
        canonical = canonicalize_url(r.url)
        if canonical not in seen_urls:
            seen_urls.add(canonical)
            unique_results.append(r)

    return unique_results

Синтез из нескольких источников

Синтезирование информации из нескольких источников требует параллельного извлечения, дедупликации, оценки качества источников и отслеживания цитирований.

Оценка качества источников

Не все веб-источники имеют одинаковую надёжность. Присвойте оценки качества на основе авторитета домена и сигналов контента.

class SourceQualityAssessor:
    def __init__(self):
        self.high_authority_domains = {
            ".gov", ".edu", ".org",
            "nature.com", "science.org", "arxiv.org",
            "who.int", "cdc.gov", "nist.gov"
        }

        self.low_credibility_signals = {
            "wordpress.com", "blogspot.com", "medium.com"
        }

    def assess(self, result: SearchResult) -> dict:
        """Quick rule-based quality assessment."""
        score = 0.5
        flags = []

        from urllib.parse import urlparse
        domain = urlparse(result.url).netloc.lower()

        # Domain authority
        for auth_domain in self.high_authority_domains:
            if auth_domain in domain:
                score += 0.3
                flags.append("authoritative_domain")
                break

        for low_domain in self.low_credibility_signals:
            if low_domain in domain:
                score -= 0.2
                flags.append("user_generated_platform")
                break

        # Content length signal
        if len(result.content) > 2000:
            score += 0.1
            flags.append("substantial_content")
        elif len(result.content) < 200:
            score -= 0.15
            flags.append("thin_content")

        return {
            "score": max(0.0, min(1.0, score)),
            "flags": flags,
            "domain": domain
        }

Извлечение утверждений с атрибуцией источника

Извлекайте проверяемые утверждения и поддерживайте связь цитирования:

import anthropic
import json
from datetime import datetime, timezone

class ClaimExtractor:
    def __init__(self, client: anthropic.Anthropic):
        self.client = client

    def extract_claims(
        self,
        content: str,
        source_url: str,
        research_question: str
    ) -> list[dict]:
        """Extract verifiable claims with source attribution."""

        prompt = f"""Extract factual claims from this source relevant to the research question.

Research question: {research_question}
Source URL: {source_url}
Content: {content[:3000]}

Extract claims that are:
- Specific and verifiable
- Relevant to the research question
- Attributed (who says what)

Return JSON array:
[
  {{
    "claim": "specific factual statement",
    "claim_type": "statistic/finding/opinion/definition",
    "subject": "what the claim is about",
    "confidence_in_claim": "high/medium/low",
    "quote": "exact quote from source if available"
  }}
]"""

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=800,
            messages=[{"role": "user", "content": prompt}]
        )

        claims = json.loads(response.content[0].text)

        # Attach source metadata to every claim
        timestamp = datetime.now(timezone.utc).isoformat()
        for claim in claims:
            claim["source_url"] = source_url
            claim["extraction_timestamp"] = timestamp

        return claims

Кросс-источниковая верификация

Проверяйте утверждения путём проверки на подтверждение или противоречие между источниками:

def verify_claim_across_sources(
    client: anthropic.Anthropic,
    claim: dict,
    all_claims: list[dict]
) -> dict:
    """Check if claim is corroborated or contradicted by other sources."""

    # Find claims from other sources on same subject
    other_sources_claims = [
        c for c in all_claims
        if c["source_url"] != claim["source_url"]
        and c["subject"] == claim.get("subject")
    ]

    if not other_sources_claims:
        return {
            "verification_status": "unverified",
            "corroborating_sources": [],
            "contradicting_sources": []
        }

    claims_text = "\n".join([
        f"Source {c['source_url']}: {c['claim']}"
        for c in other_sources_claims[:5]
    ])

    prompt = f"""Compare this claim against claims from other sources:

Target claim: "{claim['claim']}"
Source: {claim['source_url']}

Other sources' claims on same topic:
{claims_text}

Determine:
1. Which sources CORROBORATE (agree with) this claim?
2. Which sources CONTRADICT this claim?

Return JSON:
{{
  "verification_status": "verified/partially_verified/contradicted/unverified",
  "corroborating_sources": ["url1", "url2"],
  "contradicting_sources": ["url3"],
  "contradiction_details": "if contradicted, explain discrepancy"
}}"""

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=400,
        messages=[{"role": "user", "content": prompt}]
    )

    return json.loads(response.content[0].text)

Построение DAG цитирований

Поддерживайте направленный ациклический граф, связывающий утверждения с источниками и ссылками:

from typing import Dict, List, Set

class CitationGraph:
    def __init__(self):
        self.nodes: Dict[str, dict] = {}  # url -> node metadata
        self.edges: List[tuple] = []  # (from_url, to_url, relation_type)

    def add_source(self, url: str, metadata: dict):
        """Add a source node to the graph."""
        self.nodes[url] = {
            **metadata,
            "node_type": "source"
        }

    def add_claim(self, claim_id: str, claim_data: dict, source_url: str):
        """Add a claim node linked to its source."""
        self.nodes[claim_id] = {
            **claim_data,
            "node_type": "claim"
        }
        self.edges.append((claim_id, source_url, "extracted_from"))

    def add_citation(self, citing_url: str, cited_url: str):
        """Add citation relationship between sources."""
        self.edges.append((citing_url, cited_url, "cites"))

    def get_citation_chain(self, claim_id: str) -> List[str]:
        """Get full citation chain from claim to primary sources."""
        chain = []
        visited = set()

        def traverse(node_id):
            if node_id in visited:
                return
            visited.add(node_id)
            chain.append(node_id)

            for from_id, to_id, rel in self.edges:
                if from_id == node_id:
                    traverse(to_id)

        traverse(claim_id)
        return chain

Цикл итеративной доработки

Исследовательские агенты улучшают результаты благодаря итеративной доработке запросов на основе выявленных пробелов.

Разложение запроса

Разбейте сложные вопросы на подвопросы:

class ResearchPlanner:
    def __init__(self, client: anthropic.Anthropic):
        self.client = client

    def decompose_query(self, research_question: str) -> dict:
        """Break complex research question into structured search plan."""

        prompt = f"""Decompose this research question into a structured search plan.

Research question: {research_question}

Create a research plan with:
1. Core sub-questions (main aspects to research)
2. Specific search queries for each sub-question
3. Source types needed (academic, news, official docs)
4. Time scope (how recent must sources be?)

Return JSON:
{{
  "core_aspects": [
    {{
      "aspect": "description of research dimension",
      "priority": "critical/important/supplementary",
      "search_queries": ["query 1", "query 2"],
      "preferred_sources": ["academic", "industry", "news"],
      "requires_recency": true/false,
      "recency_window_days": 90
    }}
  ],
  "total_estimated_queries": <number>,
  "research_depth": "shallow/moderate/deep"
}}"""

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1000,
            messages=[{"role": "user", "content": prompt}]
        )

        return json.loads(response.content[0].text)

Обнаружение пробелов

Выявляйте что именно отсутствует после начального исследования:

def detect_gaps(
    client: anthropic.Anthropic,
    research_question: str,
    current_claims: list[dict]
) -> list[str]:
    """Identify gaps in research coverage."""

    claims_summary = "\n".join([
        f"- {c['claim']} (from: {c['source_url']})"
        for c in current_claims[:20]
    ])

    prompt = f"""Analyze current research findings and identify gaps.

Research question: {research_question}

Current findings:
{claims_summary}

What important aspects of the question are NOT covered by current findings?
List specific gaps that need additional research.

Return JSON array of gap descriptions:
["gap 1 description", "gap 2 description", ...]"""

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=400,
        messages=[{"role": "user", "content": prompt}]
    )

    return json.loads(response.content[0].text)

Генерация дополнительных запросов

Генерируйте целенаправленные дополнительные запросы для заполнения пробелов:

def generate_follow_up_queries(
    client: anthropic.Anthropic,
    research_question: str,
    identified_gaps: list[str]
) -> list[str]:
    """Generate follow-up search queries to fill identified gaps."""

    prompt = f"""Generate targeted search queries to fill research gaps.

Original question: {research_question}

Identified gaps:
{chr(10).join(f'- {g}' for g in identified_gaps)}

Generate 3-5 specific search queries that would address these gaps.

Return JSON array: ["query1", "query2", ...]"""

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}]
    )

    return json.loads(response.content[0].text)

Полный цикл итеративного исследования

async def iterative_research(
    client: anthropic.Anthropic,
    search_client,
    research_question: str,
    max_iterations: int = 3
) -> dict:
    """Execute iterative research with gap detection and refinement."""

    planner = ResearchPlanner(client)
    extractor = ClaimExtractor(client)

    # Initial decomposition
    research_plan = planner.decompose_query(research_question)

    all_claims = []
    iteration = 0

    while iteration < max_iterations:
        # Execute current round of queries
        if iteration == 0:
            queries = [
                q for aspect in research_plan["core_aspects"]
                for q in aspect["search_queries"][:2]
            ]
        else:
            # Detect gaps and generate follow-ups
            gaps = detect_gaps(client, research_question, all_claims)
            if not gaps:
                break  # No more gaps, research complete
            queries = generate_follow_up_queries(client, research_question, gaps)

        # Execute searches
        for query in queries:
            results = await search_all_sources(query, tavily_key="...", serper_key="...")

            for result in results:
                claims = extractor.extract_claims(
                    result.content,
                    result.url,
                    research_question
                )
                all_claims.extend(claims)

        iteration += 1

    return {
        "total_claims": len(all_claims),
        "iterations": iteration,
        "claims": all_claims
    }

Производительность и бенчмарки

Примечание: Приведённые ниже цифры являются иллюстративными оценками на основе типичных production-конфигураций, а не измерениями конкретной системы.

Latency поисковых API

Исследовательские агенты чувствительны к latency. Параллельное выполнение поисковых API критично:

• Tavily базовый поиск: 800-1500ms за запрос
• Tavily продвинутый поиск: 1500-3000ms за запрос
• Brave Search API: 400-800ms за запрос
• Serper (Google): 600-1200ms за запрос
• DuckDuckGo (бесплатный): 1000-2000ms за запрос

Влияние параллельного выполнения: Запуск 3 источников параллельно даёт общую latency приблизительно равную самому медленному источнику, а не сумме всех источников. Для Tavily (1500ms) + Brave (600ms) + Serper (900ms), последовательное выполнение занимает ~3000ms, тогда как параллельное выполнение занимает ~1500ms.

Объём LLM вызовов

Типичная комплексная исследовательская задача включает:

• Разложение запроса: 1 LLM вызов
• Извлечение утверждений: 10-30 LLM вызовов (1 на источник)
• Кросс-источниковая верификация: 5-15 LLM вызовов
• Обнаружение противоречий: 3-8 LLM вызовов
• Обнаружение пробелов: 2-4 LLM вызова за итерацию
• Финальный синтез: 1 LLM вызов

Всего для глубокого исследования: 25-60 LLM вызовов. При ~$0.003 за вызов (Claude Sonnet, 1K входа/500 выхода токенов), стоимость за исследовательскую задачу составляет ~$0.08-$0.18.

Потребление токенов

Разбивка использования токенов для синтеза из нескольких источников:

• Содержимое результатов поиска (10 источников × 3000 символов): ~15K токенов
• Извлечённые утверждения (30 утверждений × 100 токенов): ~3K токенов
• Метаданные источников и цитирования: ~2K токенов
• Prompt синтеза и инструкции: ~1K токенов
• Всего входа: ~21K токенов
• Выход (комплексный отчёт): ~4K токенов

Выделите 16K-32K входных токенов для шага синтеза в зависимости от количества источников и глубины. Используйте Claude Sonnet 4 или Opus для качества синтеза.

Преимущества кэширования

Стратегии кэширования значительно снижают стоимость для повторного исследования:

• Кэширование результатов поиска: Кэшируйте результаты поиска по хешу запроса; снижает API стоимость при повторных запросах
• Кэширование извлечения утверждений: Кэшируйте извлечённые утверждения по хешу контента; избегайте переизвлечения
• Кэширование качества источников: Кэшируйте оценки качества по домену/URL; амортизируйте стоимость оценки

Оценённые сбережения: Кэширование может сократить объём LLM вызовов на 40-60% для исследования перекрывающихся тем.

Сквозное время исследования

Для сложного исследовательского вопроса требующего глубокого анализа:

• Разложение запроса: 2-3 секунды
• Выполнение начального поиска (3 источника, 5 запросов): 8-12 секунд (параллельно)
• Извлечение утверждений (15 источников): 30-45 секунд (параллельная батчевка)
• Кросс-источниковая верификация: 10-15 секунд
• Обнаружение пробелов + итерация дополнения: 15-20 секунд
• Финальный синтез: 8-12 секунд

Всего: 70-110 секунд для комплексного исследования. Поверхностное исследование (одна итерация, меньше источников) завершается за 30-50 секунд.

Эксперт-человек потребовал бы 30-90 минут для сравнимой глубины и качества цитирования.