Build: Research & Analysis Agent -- סוכן שחוקר, מנתח, ומייצר דוחות מקיפים

סקירת הפרויקט וארכיטקטורה

מה אנחנו בונים

בפרק הזה תבנו Research & Analysis Agent -- סוכן שמקבל שאלת מחקר ומחזיר דוח מקיף. לא דוח שטחי שמבוסס על מקור אחד. דוח אמיתי -- עם מקורות מרובים, הצלבת מידע, זיהוי סתירות, fact-checking, ו-citations.

הנה מה שהסוכן עושה:

סך הכל: 1-3 דקות לדוח מקיף. אנליסט אנושי לוקח שעות עד ימים לאותו תוצר.

הארכיטקטורה

הארכיטקטורה משתמשת ב-multi-agent pattern עם orchestrator מרכזי:

                    ┌─────────────────┐
                    │   Orchestrator   │
                    │  (LangGraph)     │
                    └────────┬────────┘
                             │
              ┌──────────────┼──────────────┐
              │              │              │
     ┌────────▼──┐   ┌──────▼───┐   ┌──────▼──────┐
     │  Search    │   │  Search  │   │  Search     │
     │  Agent 1   │   │  Agent 2 │   │  Agent 3    │
     │  (Web)     │   │  (News)  │   │  (Academic) │
     └────────┬──┘   └──────┬───┘   └──────┬──────┘
              │              │              │
              └──────────────┼──────────────┘
                             │
                    ┌────────▼────────┐
                    │  Analysis Agent │
                    │  (Synthesis)    │
                    └────────┬────────┘
                             │
                    ┌────────▼────────┐
                    │  Fact-Checker   │
                    │  Agent          │
                    └────────┬────────┘
                             │
                    ┌────────▼────────┐
                    │  Writing Agent  │
                    │  (Report Gen)   │
                    └─────────────────┘

Tech Stack

Search Tools -- כלי חיפוש ואיסוף מידע

הבסיס של כל research agent הוא כלי חיפוש איכותיים. אנחנו צריכים כלים שמדברים עם APIs שונים ומחזירים תוצאות בפורמט אחיד. ההבדל בין סוכן מחקר טוב לבינוני הוא לא ב-LLM -- הוא באיכות ובמגוון של כלי החיפוש.

פורמט תוצאות אחיד -- The Normalized Result

לפני שבונים את הכלים, נגדיר את הפורמט שכולם צריכים להחזיר:

# Python -- models.py
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
from enum import Enum

class SourceType(str, Enum):
    WEB = "web"
    ACADEMIC = "academic"
    NEWS = "news"
    SOCIAL = "social"
    DOCUMENT = "document"

class SearchResult(BaseModel):
    """Normalized search result -- same format from every source."""
    title: str
    url: str
    snippet: str = Field(description="Relevant excerpt, 100-300 chars")
    full_content: Optional[str] = Field(default=None, description="Full text if available")
    source_type: SourceType
    source_name: str = Field(description="e.g., 'Tavily', 'Semantic Scholar'")
    published_date: Optional[datetime] = None
    relevance_score: float = Field(ge=0.0, le=1.0, description="0-1 relevance")
    author: Optional[str] = None
    metadata: dict = Field(default_factory=dict)

class SearchResults(BaseModel):
    """Collection of results from a single search source."""
    query: str
    source_type: SourceType
    results: list[SearchResult]
    total_found: int
    search_time_seconds: float

כלי 1: Web Search עם Tavily

Tavily הוא ה-search API המוביל לסוכני AI. הוא לא רק מחפש -- הוא מחזיר תוכן נקי שמתאים לעיבוד על ידי LLM. ההבדל מ-Google Search API הוא משמעותי: Tavily מסיר פרסומות, navigation, ו-boilerplate ומחזיר רק את התוכן הרלוונטי.

# Python -- tools/web_search.py
import time
from tavily import TavilyClient
from models import SearchResult, SearchResults, SourceType

class WebSearchTool:
    """Web search using Tavily API -- optimized for AI agents."""

    def __init__(self, api_key: str, max_results: int = 10):
        self.client = TavilyClient(api_key=api_key)
        self.max_results = max_results

    def search(self, query: str, search_depth: str = "advanced") -> SearchResults:
        """
        Search the web for a query.
        search_depth: "basic" (fast, cheaper) or "advanced" (thorough, better for research)
        """
        start = time.time()

        response = self.client.search(
            query=query,
            search_depth=search_depth,
            max_results=self.max_results,
            include_raw_content=True,  # Get full page content
            include_answer=False,       # We do our own analysis
        )

        results = []
        for r in response.get("results", []):
            results.append(SearchResult(
                title=r.get("title", ""),
                url=r.get("url", ""),
                snippet=r.get("content", "")[:300],
                full_content=r.get("raw_content"),
                source_type=SourceType.WEB,
                source_name="Tavily",
                relevance_score=r.get("score", 0.5),
            ))

        return SearchResults(
            query=query,
            source_type=SourceType.WEB,
            results=results,
            total_found=len(results),
            search_time_seconds=round(time.time() - start, 2),
        )

כלי 2: Academic Search עם Semantic Scholar

Semantic Scholar (של AI2) נותן גישה חינמית ל-200M+ מאמרים אקדמיים. ה-API מחזיר metadata עשיר: citation count, abstract, authors, year -- מצוין למחקר שדורש מקורות אקדמיים.

# Python -- tools/academic_search.py
import httpx
import time
from models import SearchResult, SearchResults, SourceType

class AcademicSearchTool:
    """Academic paper search using Semantic Scholar API (free, no key needed)."""

    BASE_URL = "https://api.semanticscholar.org/graph/v1"

    def __init__(self, max_results: int = 10):
        self.max_results = max_results

    def search(self, query: str) -> SearchResults:
        start = time.time()

        response = httpx.get(
            f"{self.BASE_URL}/paper/search",
            params={
                "query": query,
                "limit": self.max_results,
                "fields": "title,abstract,url,year,authors,citationCount,externalIds",
            },
            timeout=30,
        )
        response.raise_for_status()
        data = response.json()

        results = []
        for paper in data.get("data", []):
            citation_count = paper.get("citationCount", 0)
            # Higher citations = higher relevance (simple heuristic)
            relevance = min(1.0, citation_count / 100) if citation_count else 0.3

            authors = ", ".join(
                a.get("name", "") for a in (paper.get("authors") or [])[:3]
            )

            results.append(SearchResult(
                title=paper.get("title", ""),
                url=paper.get("url") or f"https://api.semanticscholar.org/paper/{paper.get('paperId', '')}",
                snippet=paper.get("abstract", "")[:300] if paper.get("abstract") else "",
                source_type=SourceType.ACADEMIC,
                source_name="Semantic Scholar",
                relevance_score=relevance,
                author=authors,
                metadata={
                    "citation_count": citation_count,
                    "year": paper.get("year"),
                    "paper_id": paper.get("paperId"),
                },
            ))

        return SearchResults(
            query=query,
            source_type=SourceType.ACADEMIC,
            results=results,
            total_found=data.get("total", len(results)),
            search_time_seconds=round(time.time() - start, 2),
        )

כלי 3: News Search

# Python -- tools/news_search.py
import httpx
import time
from datetime import datetime
from models import SearchResult, SearchResults, SourceType

class NewsSearchTool:
    """News search using Brave Search API."""

    def __init__(self, api_key: str, max_results: int = 10):
        self.api_key = api_key
        self.max_results = max_results

    def search(self, query: str, freshness: str = "pw") -> SearchResults:
        """
        freshness: pd (past day), pw (past week), pm (past month), py (past year)
        """
        start = time.time()

        response = httpx.get(
            "https://api.search.brave.com/res/v1/news/search",
            headers={"X-Subscription-Token": self.api_key},
            params={
                "q": query,
                "count": self.max_results,
                "freshness": freshness,
            },
            timeout=30,
        )
        response.raise_for_status()
        data = response.json()

        results = []
        for item in data.get("results", []):
            pub_date = None
            if item.get("age"):
                # Brave returns relative age, we approximate
                pub_date = datetime.now()

            results.append(SearchResult(
                title=item.get("title", ""),
                url=item.get("url", ""),
                snippet=item.get("description", "")[:300],
                source_type=SourceType.NEWS,
                source_name=f"Brave News ({item.get('meta_url', {}).get('hostname', 'unknown')})",
                relevance_score=0.7,  # News is generally relevant if returned
                published_date=pub_date,
                metadata={"age": item.get("age", "unknown")},
            ))

        return SearchResults(
            query=query,
            source_type=SourceType.NEWS,
            results=results,
            total_found=len(results),
            search_time_seconds=round(time.time() - start, 2),
        )

כלי 4: Social/Community Search

# Python -- tools/social_search.py
import httpx
import time
from models import SearchResult, SearchResults, SourceType

class SocialSearchTool:
    """Search Reddit and Hacker News for community discussions."""

    def search(self, query: str, max_results: int = 10) -> SearchResults:
        """Search Reddit via its public JSON API."""
        start = time.time()
        results = []

        # Reddit search (no API key needed for public search)
        try:
            response = httpx.get(
                "https://www.reddit.com/search.json",
                params={"q": query, "limit": max_results, "sort": "relevance", "t": "year"},
                headers={"User-Agent": "ResearchAgent/1.0"},
                timeout=15,
            )
            response.raise_for_status()
            data = response.json()

            for post in data.get("data", {}).get("children", []):
                d = post.get("data", {})
                results.append(SearchResult(
                    title=d.get("title", ""),
                    url=f"https://reddit.com{d.get('permalink', '')}",
                    snippet=d.get("selftext", "")[:300] or d.get("title", ""),
                    source_type=SourceType.SOCIAL,
                    source_name=f"Reddit r/{d.get('subreddit', 'unknown')}",
                    relevance_score=min(1.0, d.get("score", 0) / 500),
                    author=d.get("author"),
                    metadata={
                        "subreddit": d.get("subreddit"),
                        "score": d.get("score", 0),
                        "num_comments": d.get("num_comments", 0),
                    },
                ))
        except httpx.HTTPError:
            pass  # Reddit can be flaky, continue without

        return SearchResults(
            query=query,
            source_type=SourceType.SOCIAL,
            results=results,
            total_found=len(results),
            search_time_seconds=round(time.time() - start, 2),
        )

# Test all 4 search tools
import os
from tools.web_search import WebSearchTool
from tools.academic_search import AcademicSearchTool
from tools.news_search import NewsSearchTool
from tools.social_search import SocialSearchTool

query = "AI agents in production 2026"

web = WebSearchTool(os.environ["TAVILY_API_KEY"])
print(f"Web: {web.search(query).total_found} results")

academic = AcademicSearchTool()
print(f"Academic: {academic.search(query).total_found} results")

# Brave requires API key -- skip if not available
if os.environ.get("BRAVE_API_KEY"):
    news = NewsSearchTool(os.environ["BRAVE_API_KEY"])
    print(f"News: {news.search(query).total_found} results")

social = SocialSearchTool()
print(f"Social: {social.search(query).total_found} results")

Parallel Research Pipeline -- חיפוש מקבילי ב-LangGraph

הנה הבעיה: אם מריצים 4 חיפושים אחד אחרי השני, כל אחד לוקח 3-5 שניות, מקבלים 12-20 שניות. עם fan-out pattern ב-LangGraph, כולם רצים במקביל ומסיימים ב-5 שניות. זה 4x speedup עם שורות קוד בודדות.

The Fan-Out / Fan-In Pattern

State Definition

# Python -- graph/state.py
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from models import SearchResults

def merge_search_results(existing: list, new: list) -> list:
    """Merge search results, avoiding duplicates by URL."""
    seen_urls = {r.url for result_set in existing for r in result_set.results}
    merged = list(existing)
    for result_set in new:
        filtered_results = [r for r in result_set.results if r.url not in seen_urls]
        for r in filtered_results:
            seen_urls.add(r.url)
        merged.append(SearchResults(
            query=result_set.query,
            source_type=result_set.source_type,
            results=filtered_results,
            total_found=len(filtered_results),
            search_time_seconds=result_set.search_time_seconds,
        ))
    return merged

class ResearchState(TypedDict):
    """State for the research pipeline."""
    # Input
    research_question: str
    research_depth: int  # 1-3, how many iterations

    # Query planning
    sub_queries: list[str]
    selected_sources: list[str]

    # Search results (fan-in with dedup)
    search_results: Annotated[list[SearchResults], merge_search_results]

    # Analysis
    findings: list[dict]        # extracted claims with confidence
    contradictions: list[dict]  # conflicting claims
    gaps: list[str]             # identified knowledge gaps

    # Fact-checking
    verified_findings: list[dict]
    flagged_claims: list[dict]  # claims that failed verification

    # Report
    report_markdown: str
    report_metadata: dict

    # Control
    current_iteration: int
    messages: Annotated[list, add_messages]

Building the Graph

# Python -- graph/research_graph.py
import os
from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from graph.state import ResearchState
from tools.web_search import WebSearchTool
from tools.academic_search import AcademicSearchTool
from tools.news_search import NewsSearchTool
from tools.social_search import SocialSearchTool

# Initialize LLMs
planner_llm = ChatAnthropic(model="claude-haiku-3.5", temperature=0)
analysis_llm = ChatAnthropic(model="claude-sonnet-4-6-20260325", temperature=0)
writer_llm = ChatAnthropic(model="claude-sonnet-4-6-20260325", temperature=0.3)

# Initialize search tools
web_search = WebSearchTool(os.environ["TAVILY_API_KEY"])
academic_search = AcademicSearchTool()
news_search = NewsSearchTool(os.environ.get("BRAVE_API_KEY", ""))
social_search = SocialSearchTool()


# ---- Node Functions ----

def plan_research(state: ResearchState) -> dict:
    """Break down the research question into sub-queries and select sources."""
    response = planner_llm.invoke([
        {"role": "system", "content": """You are a research planner.
Given a research question, output:
1. 3-5 sub-queries that together cover the topic comprehensively
2. Which sources to search (web, academic, news, social)

Respond in JSON: {"sub_queries": [...], "sources": [...]}"""},
        {"role": "user", "content": state["research_question"]},
    ])

    import json
    plan = json.loads(response.content)
    return {
        "sub_queries": plan["sub_queries"],
        "selected_sources": plan["sources"],
    }


def search_web(state: ResearchState) -> dict:
    """Search the web for all sub-queries."""
    all_results = []
    for query in state["sub_queries"]:
        results = web_search.search(query)
        all_results.append(results)
    return {"search_results": all_results}


def search_academic(state: ResearchState) -> dict:
    """Search academic papers for all sub-queries."""
    all_results = []
    for query in state["sub_queries"]:
        results = academic_search.search(query)
        all_results.append(results)
    return {"search_results": all_results}


def search_news(state: ResearchState) -> dict:
    """Search news for all sub-queries."""
    all_results = []
    for query in state["sub_queries"]:
        results = news_search.search(query)
        all_results.append(results)
    return {"search_results": all_results}


def search_social(state: ResearchState) -> dict:
    """Search social media for all sub-queries."""
    all_results = []
    for query in state["sub_queries"]:
        results = social_search.search(query)
        all_results.append(results)
    return {"search_results": all_results}


def should_search_source(source: str):
    """Create a conditional function for source selection."""
    def check(state: ResearchState) -> bool:
        return source in state.get("selected_sources", [])
    return check


# ---- Build the Graph ----

def build_research_graph():
    graph = StateGraph(ResearchState)

    # Add nodes
    graph.add_node("plan", plan_research)
    graph.add_node("search_web", search_web)
    graph.add_node("search_academic", search_academic)
    graph.add_node("search_news", search_news)
    graph.add_node("search_social", search_social)
    graph.add_node("analyze", analyze_results)        # defined in next section
    graph.add_node("fact_check", fact_check_findings)  # defined later
    graph.add_node("write_report", generate_report)    # defined later

    # Edges: START --> plan
    graph.add_edge(START, "plan")

    # Fan-out: plan --> all search nodes (parallel)
    graph.add_conditional_edges(
        "plan",
        # Route to selected sources
        lambda state: state.get("selected_sources", ["web"]),
        {
            "web": "search_web",
            "academic": "search_academic",
            "news": "search_news",
            "social": "search_social",
        },
    )

    # Fan-in: all searches --> analyze
    graph.add_edge("search_web", "analyze")
    graph.add_edge("search_academic", "analyze")
    graph.add_edge("search_news", "analyze")
    graph.add_edge("search_social", "analyze")

    # analyze --> fact_check --> write_report --> END
    graph.add_edge("analyze", "fact_check")
    graph.add_edge("fact_check", "write_report")
    graph.add_edge("write_report", END)

    return graph.compile()

TypeScript Alternative -- Vercel AI SDK

// TypeScript -- research-agent.ts
import { generateText, tool } from "ai";
import { anthropic } from "@ai-sdk/anthropic";
import { z } from "zod";

// Define search tools for the Vercel AI SDK
const webSearchTool = tool({
  description: "Search the web using Tavily API",
  parameters: z.object({
    query: z.string().describe("Search query"),
    depth: z.enum(["basic", "advanced"]).default("advanced"),
  }),
  execute: async ({ query, depth }) => {
    const response = await fetch("https://api.tavily.com/search", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        api_key: process.env.TAVILY_API_KEY,
        query,
        search_depth: depth,
        max_results: 10,
        include_raw_content: true,
      }),
    });
    const data = await response.json();
    return data.results.map((r: any) => ({
      title: r.title,
      url: r.url,
      snippet: r.content?.slice(0, 300),
      source: "tavily",
      score: r.score,
    }));
  },
});

const academicSearchTool = tool({
  description: "Search academic papers via Semantic Scholar",
  parameters: z.object({
    query: z.string().describe("Academic search query"),
  }),
  execute: async ({ query }) => {
    const url = new URL("https://api.semanticscholar.org/graph/v1/paper/search");
    url.searchParams.set("query", query);
    url.searchParams.set("limit", "10");
    url.searchParams.set("fields", "title,abstract,url,year,citationCount,authors");

    const response = await fetch(url.toString());
    const data = await response.json();
    return (data.data || []).map((p: any) => ({
      title: p.title,
      url: p.url || `https://api.semanticscholar.org/paper/${p.paperId}`,
      snippet: p.abstract?.slice(0, 300) || "",
      source: "semantic_scholar",
      citations: p.citationCount,
      year: p.year,
    }));
  },
});

// Run the research agent
async function runResearch(question: string) {
  // Step 1: Plan the research
  const planResult = await generateText({
    model: anthropic("claude-haiku-3.5"),
    system: `You are a research planner. Break the question into 3-5 sub-queries.
Return JSON: {"sub_queries": [...], "sources": ["web", "academic"]}`,
    prompt: question,
  });
  const plan = JSON.parse(planResult.text);

  // Step 2: Run searches in parallel
  const searchPromises = plan.sub_queries.flatMap((q: string) => [
    webSearchTool.execute({ query: q, depth: "advanced" }, {} as any),
    academicSearchTool.execute({ query: q }, {} as any),
  ]);
  const allResults = await Promise.all(searchPromises);
  const flatResults = allResults.flat();

  // Step 3: Analyze and write report
  const { text: report } = await generateText({
    model: anthropic("claude-sonnet-4-6-20260325"),
    system: `You are a research analyst. Analyze the search results and write
a comprehensive report with: Executive Summary, Key Findings, Analysis,
Recommendations, and Sources. Use inline citations [1], [2], etc.`,
    prompt: `Research question: ${question}

Search results:
${JSON.stringify(flatResults, null, 2)}

Write a comprehensive research report in Markdown.`,
  });

  return report;
}

graph = build_research_graph()

result = graph.invoke({
    "research_question": "What are the most effective AI agent architectures in production in 2026?",
    "research_depth": 1,
    "current_iteration": 0,
})

# Check results
for sr in result["search_results"]:
    print(f"{sr.source_type}: {sr.total_found} results ({sr.search_time_seconds}s)")

De-duplication -- הסרת כפילויות

כשמחפשים ב-4 מקורות שונים, אותו מאמר או אתר יכול להופיע פעמיים ושלוש. de-duplication מבוסס על שתי שכבות:

# Python -- utils/dedup.py
from urllib.parse import urlparse
from models import SearchResult

def deduplicate_results(results: list[SearchResult]) -> list[SearchResult]:
    """Remove duplicate results using URL matching + title similarity."""
    seen_urls = set()
    seen_titles = set()
    unique = []

    for result in sorted(results, key=lambda r: r.relevance_score, reverse=True):
        # Normalize URL (remove query params, trailing slashes)
        parsed = urlparse(result.url)
        normalized_url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}".rstrip("/")

        # Normalize title (lowercase, strip whitespace)
        normalized_title = result.title.lower().strip()

        # Skip if URL or title already seen
        if normalized_url in seen_urls:
            continue
        if normalized_title in seen_titles and len(normalized_title) > 20:
            continue

        seen_urls.add(normalized_url)
        seen_titles.add(normalized_title)
        unique.append(result)

    return unique

Analysis & Synthesis -- ניתוח וסינתזה

חיפוש הוא 30% מהעבודה. ה-70% הקשים הם ניתוח: מה המידע הזה אומר? מה עקבי ומה סותר? מה אמין ומה מפוקפק? כאן הסוכן הופך מ"מחפש" ל-"אנליסט".

Claim Extraction

הצעד הראשון הוא חילוץ טענות ספציפיות מהתוצאות. לא "המצב השתפר" -- אלא "מכירות SaaS עלו ב-23% ברבעון השלישי של 2025" עם ציון המקור.

# Python -- agents/analysis_agent.py
from pydantic import BaseModel, Field
from langchain_anthropic import ChatAnthropic

class Claim(BaseModel):
    """A specific, verifiable claim extracted from search results."""
    statement: str = Field(description="The specific claim or fact")
    source_urls: list[str] = Field(description="URLs where this claim was found")
    source_count: int = Field(description="Number of independent sources")
    confidence: float = Field(ge=0.0, le=1.0, description="Confidence in this claim")
    category: str = Field(description="e.g., 'statistic', 'trend', 'opinion', 'fact'")
    date_relevance: str = Field(description="How current: 'current', 'recent', 'dated'")

class Contradiction(BaseModel):
    """Two claims that contradict each other."""
    claim_a: str
    claim_b: str
    source_a: str
    source_b: str
    explanation: str = Field(description="Why these claims contradict")

class AnalysisResult(BaseModel):
    """Structured output from the analysis agent."""
    key_findings: list[Claim]
    contradictions: list[Contradiction]
    gaps: list[str] = Field(description="Topics that need more research")
    overall_confidence: float = Field(description="0-1 confidence in the analysis")
    summary: str = Field(description="2-3 sentence summary of findings")


def analyze_results(state: dict) -> dict:
    """Analyze all search results -- extract claims, find contradictions, identify gaps."""
    analysis_llm = ChatAnthropic(
        model="claude-sonnet-4-6-20260325",
        temperature=0,
    )

    # Flatten all results into a single context
    all_snippets = []
    for result_set in state["search_results"]:
        for r in result_set.results:
            all_snippets.append(
                f"[{r.source_name}] {r.title}\n{r.snippet}\nURL: {r.url}"
            )

    context = "\n\n---\n\n".join(all_snippets[:30])  # Limit to top 30 results

    structured_llm = analysis_llm.with_structured_output(AnalysisResult)

    analysis = structured_llm.invoke([
        {"role": "system", "content": """You are a research analyst.
Your job is to analyze search results and extract:
1. KEY FINDINGS: Specific, verifiable claims with source URLs and confidence scores
   - confidence 0.9+: claim appears in 3+ reliable sources
   - confidence 0.7-0.9: claim appears in 2 sources or 1 very reliable source
   - confidence 0.5-0.7: claim appears in 1 source, seems plausible
   - confidence <0.5: unverified, uncertain, or from unreliable source
2. CONTRADICTIONS: Claims that conflict with each other
3. GAPS: Topics that need more research (not enough data)

Be specific. "Revenue grew" is bad. "Revenue grew 23% YoY to $4.2B" is good.
Always cite source URLs."""},
        {"role": "user", "content": f"""Research question: {state['research_question']}

Search results:
{context}

Analyze these results thoroughly."""},
    ])

    return {
        "findings": [f.model_dump() for f in analysis.key_findings],
        "contradictions": [c.model_dump() for c in analysis.contradictions],
        "gaps": analysis.gaps,
    }

Confidence Scoring Matrix

Report Generation -- הפקת דוחות

ניתוח טוב בלי דוח קריא הוא כמו נתונים בלי dashboard. ה-Writing Agent לוקח את הממצאים, הסתירות, וה-gaps ומייצר דוח מקצועי שמנהל יכול לקרוא ולקבל ממנו החלטות.

מבנה הדוח

כל דוח מכיל את הסעיפים הבאים:

The Report Generator

# Python -- agents/report_writer.py
from langchain_anthropic import ChatAnthropic

def generate_report(state: dict) -> dict:
    """Generate a structured research report from analysis results."""
    writer_llm = ChatAnthropic(
        model="claude-sonnet-4-6-20260325",
        temperature=0.3,  # Slight creativity for better prose
        max_tokens=4096,
    )

    # Build source list for citations
    all_sources = []
    source_idx = 1
    source_map = {}  # url -> citation number

    for result_set in state["search_results"]:
        for r in result_set.results:
            if r.url not in source_map:
                source_map[r.url] = source_idx
                all_sources.append(f"[{source_idx}] {r.title} - {r.url}")
                source_idx += 1

    findings_text = "\n".join(
        f"- {f['statement']} (confidence: {f['confidence']}, sources: {f['source_count']})"
        for f in state.get("verified_findings", state.get("findings", []))
    )

    contradictions_text = "\n".join(
        f"- {c['claim_a']} vs {c['claim_b']}: {c['explanation']}"
        for c in state.get("contradictions", [])
    )

    gaps_text = "\n".join(f"- {g}" for g in state.get("gaps", []))
    flagged_text = "\n".join(
        f"- {c['statement']}: {c.get('flag_reason', 'unverified')}"
        for c in state.get("flagged_claims", [])
    )

    response = writer_llm.invoke([
        {"role": "system", "content": """You are a professional research report writer.
Write a comprehensive, well-structured report in Markdown format.

Report structure:
# Research Report: [Topic]

## Executive Summary
(2-3 paragraphs: what was researched, key findings, top recommendation)

## Methodology
(What sources were searched, how many results, what analysis was performed)

## Key Findings
(Bulleted list of main findings with inline citations [1], [2])

## Detailed Analysis
(3-5 paragraphs of deeper analysis, trends, patterns)

## Contradictions & Uncertainties
(What conflicting info was found, what remains unclear)

## Recommendations
(3-5 actionable recommendations based on findings)

## Sources
(Numbered list of all sources)

RULES:
- Use inline citations [1], [2] etc. linking to the sources list
- Be specific: include numbers, dates, names when available
- Clearly mark unverified claims as such
- Write in professional but accessible English
- Flag low-confidence findings explicitly"""},
        {"role": "user", "content": f"""Research question: {state['research_question']}

VERIFIED FINDINGS:
{findings_text}

CONTRADICTIONS:
{contradictions_text}

KNOWLEDGE GAPS:
{gaps_text}

FLAGGED/UNVERIFIED CLAIMS:
{flagged_text}

SOURCES:
{chr(10).join(all_sources)}

Write the complete research report."""},
    ])

    return {
        "report_markdown": response.content,
        "report_metadata": {
            "total_sources": len(all_sources),
            "total_findings": len(state.get("verified_findings", state.get("findings", []))),
            "contradictions_found": len(state.get("contradictions", [])),
            "flagged_claims": len(state.get("flagged_claims", [])),
            "gaps_identified": len(state.get("gaps", [])),
        },
    }

TypeScript Report Generator

// TypeScript -- report-writer.ts
import { generateText } from "ai";
import { anthropic } from "@ai-sdk/anthropic";

interface Finding {
  statement: string;
  confidence: number;
  source_count: number;
  source_urls: string[];
}

interface ReportInput {
  question: string;
  findings: Finding[];
  contradictions: Array<{ claim_a: string; claim_b: string; explanation: string }>;
  gaps: string[];
  sources: Array<{ title: string; url: string }>;
}

async function generateReport(input: ReportInput): Promise {
  const sourcesText = input.sources
    .map((s, i) => `[${i + 1}] ${s.title} - ${s.url}`)
    .join("\n");

  const findingsText = input.findings
    .map((f) => `- ${f.statement} (confidence: ${f.confidence})`)
    .join("\n");

  const { text } = await generateText({
    model: anthropic("claude-sonnet-4-6-20260325"),
    temperature: 0.3,
    system: `You are a professional research report writer.
Write a comprehensive Markdown report with: Executive Summary, Methodology,
Key Findings (with [1] [2] citations), Detailed Analysis, Contradictions,
Recommendations, and Sources. Be specific and cite everything.`,
    prompt: `Question: ${input.question}

Findings:
${findingsText}

Sources:
${sourcesText}

Write the full report.`,
  });

  return text;
}

Fact-Checking Layer -- שכבת אימות עובדות

הנה הבעיה הכי גדולה עם research agents: הם ממציאים דברים. LLM יכול לקחת snippet מאתר אחד, "לשדרג" אותו עם מספרים שהמציא, ולהציג את זה כעובדה מאומתת. Fact-checking layer הוא ההבדל בין דוח שאפשר לסמוך עליו לדוח שעלול להזיק.

שלושה סוגי בדיקות

The Fact-Checker Agent

# Python -- agents/fact_checker.py
from pydantic import BaseModel, Field
from langchain_anthropic import ChatAnthropic

class FactCheckResult(BaseModel):
    """Result of fact-checking a single claim."""
    original_claim: str
    verified: bool
    confidence_after_check: float = Field(ge=0.0, le=1.0)
    flag_reason: str = Field(default="", description="Why this was flagged, if flagged")
    corrected_claim: str = Field(default="", description="Corrected version if needed")
    verification_method: str = Field(description="How this was verified")

class FactCheckResults(BaseModel):
    verified_claims: list[FactCheckResult]
    flagged_claims: list[FactCheckResult]
    overall_reliability: float = Field(ge=0.0, le=1.0)


def fact_check_findings(state: dict) -> dict:
    """Fact-check all findings using dual-model verification."""
    checker_llm = ChatAnthropic(
        model="claude-sonnet-4-6-20260325",
        temperature=0,
    )

    findings = state.get("findings", [])

    # Build the context from search results for verification
    source_context = {}
    for result_set in state["search_results"]:
        for r in result_set.results:
            content = r.full_content or r.snippet
            source_context[r.url] = {
                "title": r.title,
                "content": content[:2000],  # Limit per source
            }

    # Create source context string for the checker
    sources_for_checker = "\n\n".join(
        f"SOURCE [{url}]:\n{info['title']}\n{info['content']}"
        for url, info in list(source_context.items())[:20]  # Limit to 20 sources
    )

    structured_checker = checker_llm.with_structured_output(FactCheckResults)

    findings_text = "\n".join(
        f"CLAIM {i+1}: {f['statement']} (cited sources: {', '.join(f.get('source_urls', [])[:3])})"
        for i, f in enumerate(findings)
    )

    result = structured_checker.invoke([
        {"role": "system", "content": """You are a skeptical fact-checker.
For each claim, verify it against the source material provided.

Verification rules:
1. SOURCE VERIFICATION: Does the cited source actually say this? Look for the specific text.
2. PLAUSIBILITY: Are the numbers/dates/facts plausible? Flag obviously wrong stats.
3. SPECIFICITY: Vague claims without evidence get lower confidence.
4. RECENCY: Old data presented as current gets flagged.

Mark a claim as "verified: true" only if you can confirm it from the source material.
Mark as "verified: false" and explain why if:
- The source doesn't actually say this (hallucination)
- The numbers are implausible
- The claim contradicts other verified information
- The source is unreliable for this type of claim"""},
        {"role": "user", "content": f"""CLAIMS TO VERIFY:
{findings_text}

SOURCE MATERIAL:
{sources_for_checker}

Fact-check each claim against the source material."""},
    ])

    return {
        "verified_findings": [c.model_dump() for c in result.verified_claims],
        "flagged_claims": [c.model_dump() for c in result.flagged_claims],
    }

Iterative Research -- מחקר איטרטיבי

מחקר טוב הוא לעולם לא one-shot. אנליסט אנושי קורא את התוצאות הראשונות, מזהה gaps, ושואל שאלות המשך. Iterative research נותן לסוכן את אותה יכולת: חפש --> נתח --> זהה gaps --> חפש שוב --> עדכן.

The Deep Dive Pattern

# Python -- graph/iterative.py
def identify_gaps(state: dict) -> dict:
    """Identify knowledge gaps and generate follow-up queries."""
    gap_llm = ChatAnthropic(model="claude-haiku-3.5", temperature=0)

    findings_summary = "\n".join(
        f"- {f['statement']} (confidence: {f['confidence']})"
        for f in state.get("findings", [])
    )

    response = gap_llm.invoke([
        {"role": "system", "content": """You are a research quality analyst.
Given the current findings for a research question, identify:
1. What important aspects haven't been covered?
2. What claims need more evidence?
3. What follow-up questions would deepen the research?

Return JSON: {
    "gaps": ["gap1", "gap2"],
    "follow_up_queries": ["query1", "query2", "query3"],
    "low_confidence_topics": ["topic that needs more sources"]
}"""},
        {"role": "user", "content": f"""Research question: {state['research_question']}

Current findings:
{findings_summary}

Contradictions found: {len(state.get('contradictions', []))}
Sources searched: {sum(len(rs.results) for rs in state['search_results'])}

What gaps exist? What follow-up research is needed?"""},
    ])

    import json
    gaps_analysis = json.loads(response.content)

    return {
        "gaps": gaps_analysis.get("gaps", []),
        "sub_queries": gaps_analysis.get("follow_up_queries", []),
        "current_iteration": state.get("current_iteration", 0) + 1,
    }


def should_continue_research(state: dict) -> str:
    """Decide whether to do another research iteration."""
    current = state.get("current_iteration", 0)
    max_depth = state.get("research_depth", 1)

    if current >= max_depth:
        return "write_report"

    # Also stop if we have enough high-confidence findings
    high_confidence = sum(
        1 for f in state.get("findings", [])
        if f.get("confidence", 0) >= 0.8
    )
    if high_confidence >= 10:
        return "write_report"

    return "search_web"  # Do another round of searches

להוספת iterative research ל-graph:

# Add to graph builder
graph.add_node("identify_gaps", identify_gaps)
graph.add_edge("fact_check", "identify_gaps")
graph.add_conditional_edges(
    "identify_gaps",
    should_continue_research,
    {
        "search_web": "search_web",     # Another iteration
        "write_report": "write_report",  # Done, generate report
    },
)

Templates -- תבניות מחקר מותאמות

research agent גנרי הוא טוב, אבל research agent עם תבניות מותאמות הוא מצוין. כל סוג מחקר דורש מקורות שונים, שאלות שונות, ומבנה דוח שונה.

שלוש תבניות עיקריות

# Python -- templates/research_templates.py
from dataclasses import dataclass, field

@dataclass
class ResearchTemplate:
    """Template that configures the research agent for specific use cases."""
    name: str
    description: str
    preferred_sources: list[str]
    search_depth: str  # "basic" or "advanced"
    research_depth: int  # 1-3 iterations
    report_sections: list[str]
    analysis_focus: str  # What to prioritize in analysis
    custom_instructions: str  # Domain-specific guidance

# Template 1: Market Research
MARKET_RESEARCH = ResearchTemplate(
    name="Market Research",
    description="Analyze a market: size, trends, players, opportunities",
    preferred_sources=["web", "news", "social"],
    search_depth="advanced",
    research_depth=2,
    report_sections=[
        "Executive Summary",
        "Market Size & Growth",
        "Key Players & Market Share",
        "Trends & Drivers",
        "Challenges & Risks",
        "Opportunities",
        "Competitive Landscape",
        "Recommendations",
        "Sources",
    ],
    analysis_focus="market data, growth rates, competitive dynamics",
    custom_instructions="""Focus on:
- Specific market size numbers (TAM, SAM, SOM if available)
- Growth rates (YoY, CAGR)
- Key players and their market share
- Recent funding rounds and M&A activity
- Pricing trends
Always distinguish between global and regional (Israeli) data.""",
)

# Template 2: Competitor Analysis
COMPETITOR_ANALYSIS = ResearchTemplate(
    name="Competitor Analysis",
    description="Deep dive on a specific competitor or set of competitors",
    preferred_sources=["web", "news", "social"],
    search_depth="advanced",
    research_depth=2,
    report_sections=[
        "Executive Summary",
        "Company Overview",
        "Products & Services",
        "Pricing & Business Model",
        "Strengths & Weaknesses",
        "Recent Activity",
        "Customer Sentiment",
        "Strategic Positioning",
        "Recommendations",
        "Sources",
    ],
    analysis_focus="competitive positioning, product features, pricing, customer feedback",
    custom_instructions="""Focus on:
- Product features and recent launches
- Pricing (tiers, changes)
- Customer reviews and sentiment (G2, Capterra, Reddit)
- Team size and hiring (LinkedIn, job boards)
- Funding and financial health
- Marketing strategy (content, ads, social presence)
Search in Hebrew too for Israeli competitors.""",
)

# Template 3: Technology Assessment
TECHNOLOGY_ASSESSMENT = ResearchTemplate(
    name="Technology Assessment",
    description="Evaluate a technology: maturity, adoption, pros/cons",
    preferred_sources=["web", "academic", "social"],
    search_depth="advanced",
    research_depth=2,
    report_sections=[
        "Executive Summary",
        "Technology Overview",
        "Current State & Maturity",
        "Adoption & Use Cases",
        "Technical Architecture",
        "Pros & Cons",
        "Alternatives Comparison",
        "Community & Ecosystem",
        "Recommendations",
        "Sources",
    ],
    analysis_focus="technical capabilities, maturity, adoption rate, community strength",
    custom_instructions="""Focus on:
- Technical capabilities and limitations
- Maturity level (experimental, production-ready, mainstream)
- Adoption statistics (GitHub stars, npm downloads, companies using it)
- Performance benchmarks
- Developer experience and documentation quality
- Community activity (GitHub issues, Stack Overflow, Discord)
Prioritize 2025-2026 data. Older benchmarks may be outdated.""",
)

TEMPLATES = {
    "market_research": MARKET_RESEARCH,
    "competitor_analysis": COMPETITOR_ANALYSIS,
    "tech_assessment": TECHNOLOGY_ASSESSMENT,
}

שימוש בתבניות -- Israeli Context

# Python -- example: Israeli market research
result = graph.invoke({
    "research_question": "מה גודל שוק ה-SaaS בישראל ב-2026? מי השחקנים המובילים?",
    "research_depth": 2,
    "current_iteration": 0,
    "template": "market_research",
    # Template overrides:
    "custom_instructions": """
    Focus on Israeli SaaS market specifically:
    - Search in both Hebrew and English
    - Include Israeli sources: Calcalist, Geektime, CTech, Globes
    - Look for IVC Research Center data
    - Include Israeli-founded companies (headquartered anywhere)
    - Note companies that are Israeli but marketed as US companies
    """,
})

Deployment & Automation -- פריסה ואוטומציה

research agent שרץ רק מה-terminal שלכם הוא כלי שימושי אבל מוגבל. בחלק הזה נהפוך אותו לשירות -- REST API שאפשר לקרוא לו מכל מקום, עם scheduling לדוחות אוטומטיים.

REST API עם FastAPI

# Python -- api/main.py
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, Field
from typing import Optional
import uuid
import asyncio

app = FastAPI(title="Research Agent API")

# In-memory storage (use Redis/DB in production)
research_jobs: dict = {}

class ResearchRequest(BaseModel):
    question: str = Field(description="The research question")
    template: str = Field(default="market_research", description="Research template")
    depth: int = Field(default=1, ge=1, le=3, description="Research depth 1-3")
    custom_instructions: Optional[str] = None

class ResearchResponse(BaseModel):
    job_id: str
    status: str
    report: Optional[str] = None
    metadata: Optional[dict] = None


async def run_research_background(job_id: str, request: ResearchRequest):
    """Run research in the background."""
    research_jobs[job_id]["status"] = "running"

    try:
        graph = build_research_graph()
        result = await asyncio.to_thread(
            graph.invoke,
            {
                "research_question": request.question,
                "research_depth": request.depth,
                "current_iteration": 0,
            },
        )

        research_jobs[job_id]["status"] = "completed"
        research_jobs[job_id]["report"] = result["report_markdown"]
        research_jobs[job_id]["metadata"] = result["report_metadata"]
    except Exception as e:
        research_jobs[job_id]["status"] = "failed"
        research_jobs[job_id]["error"] = str(e)


@app.post("/research", response_model=ResearchResponse)
async def start_research(request: ResearchRequest, bg: BackgroundTasks):
    """Start a new research job (runs in background)."""
    job_id = str(uuid.uuid4())
    research_jobs[job_id] = {"status": "queued"}

    bg.add_task(run_research_background, job_id, request)

    return ResearchResponse(job_id=job_id, status="queued")


@app.get("/research/{job_id}", response_model=ResearchResponse)
async def get_research(job_id: str):
    """Check status of a research job."""
    if job_id not in research_jobs:
        return ResearchResponse(job_id=job_id, status="not_found")

    job = research_jobs[job_id]
    return ResearchResponse(
        job_id=job_id,
        status=job["status"],
        report=job.get("report"),
        metadata=job.get("metadata"),
    )

Scheduled Research -- דוחות אוטומטיים

# Python -- scheduling/scheduler.py
import schedule
import time
from datetime import datetime

class ResearchScheduler:
    """Schedule recurring research reports."""

    def __init__(self, graph, delivery_fn):
        self.graph = graph
        self.deliver = delivery_fn  # Function to send the report (email, Slack, etc.)

    def add_daily_digest(self, topic: str, time_str: str = "08:00"):
        """Run research daily and deliver results."""
        def job():
            print(f"[{datetime.now()}] Running daily research: {topic}")
            result = self.graph.invoke({
                "research_question": f"{topic} -- latest news and developments today",
                "research_depth": 1,
                "current_iteration": 0,
            })
            self.deliver(
                subject=f"Daily Research Digest: {topic}",
                body=result["report_markdown"],
            )

        schedule.every().day.at(time_str).do(job)
        print(f"Scheduled daily research for '{topic}' at {time_str}")

    def add_weekly_report(self, topic: str, day: str = "monday", time_str: str = "09:00"):
        """Run in-depth research weekly."""
        def job():
            print(f"[{datetime.now()}] Running weekly research: {topic}")
            result = self.graph.invoke({
                "research_question": f"{topic} -- comprehensive weekly analysis",
                "research_depth": 2,  # Deeper for weekly
                "current_iteration": 0,
            })
            self.deliver(
                subject=f"Weekly Research Report: {topic}",
                body=result["report_markdown"],
            )

        getattr(schedule.every(), day).at(time_str).do(job)
        print(f"Scheduled weekly research for '{topic}' on {day} at {time_str}")

    def run(self):
        """Start the scheduler loop."""
        print("Research scheduler started. Press Ctrl+C to stop.")
        while True:
            schedule.run_pending()
            time.sleep(60)


# Example usage:
# scheduler = ResearchScheduler(graph, send_email)
# scheduler.add_daily_digest("AI agents market", "07:30")
# scheduler.add_weekly_report("Israeli SaaS competitive landscape", "sunday", "09:00")
# scheduler.run()

Slack/Email Delivery

# Python -- delivery/slack_delivery.py
import httpx

def send_to_slack(webhook_url: str, subject: str, body: str):
    """Send research report to Slack channel."""
    # Slack has a 40,000 char limit, truncate if needed
    if len(body) > 35000:
        body = body[:35000] + "\n\n... (report truncated, full version attached)"

    httpx.post(webhook_url, json={
        "blocks": [
            {
                "type": "header",
                "text": {"type": "plain_text", "text": subject},
            },
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": body[:3000]},
            },
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": "_Full report in thread._"},
            },
        ],
    })

Cost Estimation

לשם השוואה: אנליסט אנושי שעושה את אותו מחקר עולה $50-200 לשעה ולוקח 4-8 שעות. הסוכן עושה את זה ב-$2-5 ו-2-3 דקות. גם אם הסוכן מפספס דברים ודורש review אנושי של 30 דקות -- עדיין חסכתם 80% מהזמן.

טעויות נפוצות -- ואיך להימנע מהן

תרגילים

צ'קליסט -- סיכום פרק 16

• מבין/ה את ארכיטקטורת ה-research agent -- orchestrator, search agents, analysis, fact-checker, writer
• בנית 4 כלי חיפוש עם פורמט תוצאות אחיד (SearchResult)
• מימשת fan-out / fan-in pattern ב-LangGraph -- חיפושים מקביליים ממקורות שונים
• מימשת de-duplication -- URL matching ו-title similarity
• בנית analysis agent שמחלץ claims עם confidence scores
• מבין/ה את Confidence Scoring Matrix -- 4 גורמים ומשקלים
• מימשת fact-checking layer עם source verification, cross-reference, ו-plausibility
• בנית report generator שמייצר דוחות עם Executive Summary ו-citations
• מימשת iterative research עם gap identification ו-configurable depth
• יצרת 3 תבניות מחקר -- market research, competitor analysis, tech assessment
• יודע/ת להתאים את הסוכן לIsraeli context -- מקורות בעברית, אתרים ישראליים
• פרסת את הסוכן כ-REST API עם FastAPI
• הגדרת scheduling -- דוחות אוטומטיים יומיים/שבועיים
• מבין/ה את מבנה העלויות -- $0.50-5 לדוח, $20-50 לחודש
• בנית Research & Analysis Agent מלא שמייצר דוחות מקיפים, מאומתים, ועם citations