Multi-Agent
Research System

# agents/search_agent.py from anthropic import Anthropic from mcp_tools import web_search, fetch_url from schemas import SearchResult client = Anthropic() SYSTEM_PROMPT = """You are a specialized web research agent. Your ONLY job: find high-quality, authoritative sources on the given topic. ## Tools - web_search(query, num_results): Returns ranked URLs + snippets - fetch_url(url): Fetches and returns full page content ## Rules 1. Run web_search first, then fetch the top 3-5 most relevant URLs 2. Evaluate source credibility: prefer .edu, .gov, peer-reviewed, established outlets 3. Extract: title, url, date, key_claims[], credibility_score (0-1) 4. Return structured JSON only — no prose commentary 5. If a URL is paywalled or unreachable, skip it silently ## Output schema Return: { results: SearchResult[] } """ class SearchAgent: @staticmethod async def run(query: str, max_results: int = 10, max_turns: int = 5) -> list[SearchResult]: """ Isolated context window — search agent knows nothing about the coordinator or other agents. """ messages = [{ "role": "user", "content": f"Search for: '{query}'. Return up to {max_results} quality sources." }] tools = [ {"name": "web_search", "description": "Search the web", ...}, {"name": "fetch_url", "description": "Fetch page content", ...} ] turn = 0 while turn < max_turns: response = client.messages.create( model="claude-sonnet-4-6", # Use Sonnet for speed max_tokens=4096, system=SYSTEM_PROMPT, tools=tools, messages=messages ) turn += 1 if response.stop_reason == "end_turn": return parse_search_results(response) # Handle tool calls and continue loop messages = append_tool_results(messages, response) # Hit max_turns — return whatever was found return parse_search_results(response)

# agents/analyst_agent.py # Receives raw search results, extracts structured Findings with claim→URL chains. # Runs one instance per source — parallel execution via coordinator. from anthropic import Anthropic from schemas import SearchResult, Finding, Claim client = Anthropic() ANALYST_SYSTEM_PROMPT = """You are a research analyst. You receive a single source document (web page text + metadata) and extract a structured list of findings. ## Your job - Read the source carefully - Extract every factual claim that is relevant to the research topic - For each claim, record the EXACT quote or close paraphrase from the source - Assign a credibility score based on source type (see below) - Do NOT synthesize across sources — analyse this source only ## Credibility scoring 1.0 — Peer-reviewed journal, government data, primary source 0.8 — Established news outlet, official report, industry body 0.6 — General web content, blogs, secondary reporting 0.3 — Forum posts, opinion pieces, unverified claims ## Output — return JSON only, no preamble { "source_url": "...", "source_credibility": 0.0-1.0, "findings": [ { "claim": "exact or close paraphrase of the claim", "quote": "verbatim excerpt supporting the claim, or null", "credibility": 0.0-1.0, "topic_relevance": 0.0-1.0 } ] } ## Rules - NEVER invent claims not present in the source text - If the source is off-topic, return an empty findings array - Do not include duplicate claims - Credibility inherits from source_credibility unless the specific claim has stronger/weaker support than the overall source""" async def run_analyst_agent( result: SearchResult, topic: str, max_turns: int = 3 ) -> list[Finding]: """ Analyse a single search result and extract structured findings. Intentionally low max_turns — analysts read one source, extract, done. """ messages = [{ "role": "user", "content": ( f"Research topic: {topic}\n\n" f"Source URL: {result['url']}\n" f"Source title: {result['title']}\n\n" f"Source content:\n{result['content']}" ) }] turn = 0 while turn < max_turns: response = client.messages.create( model="claude-sonnet-4-6", # Sonnet — fast, parallel max_tokens=2048, system=ANALYST_SYSTEM_PROMPT, messages=messages ) turn += 1 raw = response.content[0].text.strip() try: import json clean = raw.replace("```json", "").replace("```", "").strip() parsed = json.loads(clean) findings = parsed.get("findings", []) source_cred = parsed.get("source_credibility", 0.6) # Attach source metadata to each finding return [ Finding( claim=f["claim"], quote=f.get("quote"), source_url=result["url"], source_title=result["title"], source_credibility=source_cred, claim_credibility=f.get("credibility", source_cred), topic_relevance=f.get("topic_relevance", 0.7) ) for f in findings if f.get("topic_relevance", 1) >= 0.5 # filter low-relevance ] except (json.JSONDecodeError, KeyError) as e: if turn == max_turns: return [] # Fail gracefully — don't crash the pipeline messages += [ {"role": "assistant", "content": raw}, {"role": "user", "content": f"JSON error: {e}. Return valid JSON only."} ] return [] async def run_analyst_batch( results: list[SearchResult], topic: str ) -> list[Finding]: """ Run analyst agents in parallel — one per search result. Uses asyncio.gather with return_exceptions=True so one failure doesn't abort the entire batch. """ import asyncio tasks = [run_analyst_agent(r, topic) for r in results] batches = await asyncio.gather(*tasks, return_exceptions=True) all_findings: list[Finding] = [] for batch in batches: if isinstance(batch, Exception): continue # log and skip failed analysts all_findings.extend(batch) # Deduplicate near-identical claims (simple text similarity check) return deduplicate_findings(all_findings) def deduplicate_findings(findings: list[Finding]) -> list[Finding]: """Remove near-duplicate claims. Keep highest-credibility version.""" seen: dict[str, Finding] = {} for f in findings: key = f.claim[:80].lower().strip() # coarse key if key not in seen or f.claim_credibility > seen[key].claim_credibility: seen[key] = f return list(seen.values())

# agents/synthesis_agent.py # Receives deduplicated Finding arrays (claim-only, no full doc content). # Cross-source reasoning, conflict resolution, structured section output. from anthropic import Anthropic from schemas import Finding, SynthesizedSection, Synthesis import json client = Anthropic() SYNTHESIS_SYSTEM_PROMPT = """You are a research synthesis specialist. You receive a structured list of findings extracted from multiple sources. Your job is to synthesise these into coherent sections for a research report. ## Input format You receive a JSON array of findings, each with: - claim: the factual statement - source_url: origin URL - source_title: origin title - claim_credibility: 0.0-1.0 - topic_relevance: 0.0-1.0 ## Your responsibilities 1. GROUP related findings into thematic sections 2. IDENTIFY agreements — claims supported by multiple sources 3. IDENTIFY conflicts — claims that contradict each other 4. RESOLVE conflicts using credibility weighting: - Higher credibility source wins, IF the gap is > 0.2 - If credibility is similar (gap <= 0.2), mark as CONTESTED — never pick silently 5. Note GAPS — important sub-topics with insufficient evidence ## Conflict resolution rule (critical) When two sources conflict with similar credibility: DO: "Sources disagree on X: [Source A] claims Y while [Source B] claims Z. Given similar source credibility, this remains contested." DON'T: Silently choose one without noting the disagreement. ## Output — return JSON only { "sections": [ { "title": "section heading", "summary": "2-3 sentence synthesis paragraph", "key_claims": [ { "claim": "synthesised claim", "supporting_sources": ["url1", "url2"], "contested": false, "contest_note": null } ], "confidence": 0.0-1.0 } ], "gaps": ["topic area lacking evidence", ...], "overall_confidence": 0.0-1.0 }""" async def run_synthesis_agent( findings: list[Finding], topic: str, max_turns: int = 4 ) -> Synthesis: """ Synthesise findings from all analysts into structured sections. Uses Opus for this step — synthesis requires deeper reasoning. IMPORTANT: Pass claim arrays only, not full document content. This keeps the context window manageable regardless of source count. """ # Serialize only the fields synthesis needs — not raw document content findings_payload = [ { "claim": f.claim, "source_url": f.source_url, "source_title": f.source_title, "claim_credibility": f.claim_credibility, "topic_relevance": f.topic_relevance, } for f in findings if f.topic_relevance >= 0.5 # pre-filter before sending ] messages = [{ "role": "user", "content": ( f"Research topic: {topic}\n\n" f"Findings from {len(findings_payload)} claims across multiple sources:\n\n" + json.dumps(findings_payload, indent=2) ) }] turn = 0 while turn < max_turns: response = client.messages.create( model="claude-opus-4-6", # Opus — synthesis needs deeper reasoning max_tokens=4096, system=SYNTHESIS_SYSTEM_PROMPT, messages=messages ) turn += 1 raw = response.content[0].text.strip() try: clean = raw.replace("```json", "").replace("```", "").strip() parsed = json.loads(clean) return Synthesis( topic=topic, sections=[SynthesizedSection(**s) for s in parsed["sections"]], gaps=parsed.get("gaps", []), overall_confidence=parsed.get("overall_confidence", 0.7) ) except (json.JSONDecodeError, KeyError) as e: if turn == max_turns: raise RuntimeError(f"Synthesis failed after {max_turns} attempts: {e}") messages += [ {"role": "assistant", "content": raw}, {"role": "user", "content": f"JSON error: {e}. Return valid JSON only."} ] raise RuntimeError("Synthesis: max turns exceeded")

# agents/report_agent.py # Formats the synthesis into a polished markdown report. # CRITICAL: This agent formats only — it adds NO new information. # All claims must trace back to the synthesis output. from anthropic import Anthropic from schemas import Synthesis, Report import json, re client = Anthropic() REPORT_SYSTEM_PROMPT = """You are a technical report writer. You receive a structured synthesis JSON and format it into a polished, well-organised markdown research report. ## Critical rule — FORMAT ONLY You MUST NOT add any new factual claims, statistics, or information that is not present in the synthesis input. Every sentence in the report body must correspond to a claim in the synthesis sections or be a structural/transitional phrase. If the synthesis does not contain information on a sub-topic, do NOT fill the gap with your own knowledge. ## Report structure # [Topic] — Research Report ## Executive Summary 2-3 sentence high-level overview of findings. ## [Section Title] (one section per synthesis section) Prose paragraph synthesising the section's key_claims. For contested claims, include the disagreement explicitly. ## Evidence Gaps Bullet list of the synthesis gaps field. ## Sources Numbered list of all unique source URLs referenced. Format: [N] Title — URL ## Confidence Assessment Overall confidence: [score] Note any sections with confidence < 0.7. ## Citation rules - Every claim gets an inline citation: claim text [N] - [N] references the Sources list number - Contested claims must include: "(sources disagree — see gaps)" - Do NOT cite sources not in the synthesis input""" async def run_report_agent( synthesis: Synthesis, max_turns: int = 3 ) -> Report: """ Generate a formatted markdown report from synthesis output. Uses Sonnet — formatting doesn't need Opus-level reasoning. """ # Build numbered source index from all referenced URLs all_urls: list[str] = [] for section in synthesis.sections: for claim in section.key_claims: for url in claim.supporting_sources: if url not in all_urls: all_urls.append(url) source_index = {url: i + 1 for i, url in enumerate(all_urls)} messages = [{ "role": "user", "content": ( f"Write a research report on: {synthesis.topic}\n\n" "Synthesis input (format ONLY this — add nothing new):\n\n" + json.dumps({ "sections": [ { "title": s.title, "summary": s.summary, "key_claims": [ { "claim": c.claim, "sources": [f"[{source_index[u]}]" for u in c.supporting_sources if u in source_index], "contested": c.contested, "contest_note": c.contest_note } for c in s.key_claims ], "confidence": s.confidence } for s in synthesis.sections ], "gaps": synthesis.gaps, "sources": [ {"index": i + 1, "url": url} for i, url in enumerate(all_urls) ], "overall_confidence": synthesis.overall_confidence }, indent=2) ) }] turn = 0 while turn < max_turns: response = client.messages.create( model="claude-sonnet-4-6", # Sonnet — formatting task max_tokens=4096, system=REPORT_SYSTEM_PROMPT, messages=messages ) turn += 1 markdown = response.content[0].text.strip() # Post-process: validate every [N] citation maps to a real source cited_indices = {int(n) for n in re.findall(r'\[(\d+)\]', markdown)} valid_indices = set(source_index.values()) phantom_citations = cited_indices - valid_indices if phantom_citations: if turn == max_turns: # Strip phantom citations rather than fail entirely for idx in phantom_citations: markdown = markdown.replace(f"[{idx}]", "") break messages += [ {"role": "assistant", "content": markdown}, {"role": "user", "content": ( f"Citation error: indices {phantom_citations} do not exist in the sources list. " "Remove or correct these citations and resubmit the full report." )} ] continue return Report( topic=synthesis.topic, markdown=markdown, sources=all_urls, overall_confidence=synthesis.overall_confidence, gaps=synthesis.gaps ) return Report( topic=synthesis.topic, markdown=markdown, sources=all_urls, overall_confidence=synthesis.overall_confidence, gaps=synthesis.gaps ) def save_report(report: Report, output_dir: str = "output/"): """Save report markdown and metadata to disk.""" import os os.makedirs(output_dir, exist_ok=True) slug = report.topic[:40].lower().replace(" ", "_") with open(f"{output_dir}{slug}.md", "w") as f: f.write(report.markdown) with open(f"{output_dir}{slug}_meta.json", "w") as f: json.dump({ "topic": report.topic, "sources": report.sources, "overall_confidence": report.overall_confidence, "gaps": report.gaps }, f, indent=2)

# schemas.py — ALL inter-agent data structures # These are the "contracts" that prevent silent data corruption. from typing import TypedDict, Optional from datetime import datetime # ── Search Agent Output ───────────────────────────────────── class SearchResult(TypedDict): url: str title: str date_published: Optional[str] snippet: str credibility_score: float # 0.0 – 1.0 full_content: Optional[str] # fetched if relevant # ── Analyst Agent Output ──────────────────────────────────── class Claim(TypedDict): text: str source_url: str confidence: float # 0.0 – 1.0 quote: Optional[str] # verbatim excerpt if available class Finding(TypedDict): source: SearchResult claims: list[Claim] relevance_score: float angle_covered: str # which research angle this addresses contradictions: list[str] # conflicts with other known sources # ── Synthesis Agent Output ────────────────────────────────── class SynthesizedSection(TypedDict): angle: str narrative: str supporting_claims: list[Claim] confidence_level: str # "high" | "medium" | "low" unresolved_conflicts: list[str] class Synthesis(TypedDict): topic: str sections: list[SynthesizedSection] total_sources: int gaps_identified: list[str] # topics with insufficient coverage generated_at: str # ── Report Agent Output ───────────────────────────────────── class Report(TypedDict): title: str executive_summary: str sections: list[dict] citations: list[dict] metadata: dict format: str

═══ SEARCH AGENT ═══════════════════════════════════════════ You are a specialized web research agent. Focus: Find authoritative, recent, relevant sources only. Tools: web_search, fetch_url CRITICAL: Do not summarize or analyze. Only retrieve and package. Return structured JSON. Prefer: peer-reviewed > institutional > news. Flag paywalled content but don't skip the URL. ═══ ANALYST AGENT ══════════════════════════════════════════ You are a document analysis specialist. Input: A single source document + research topic. Tools: read_document, extract_claims, score_relevance CRITICAL: Every claim you extract MUST include the source URL. Do not interpolate or hallucinate. Only extract what is explicitly stated. Rate confidence honestly: if the source is ambiguous, say 0.4, not 0.9. Identify contradictions with any explicitly provided prior findings. ═══ SYNTHESIS AGENT ════════════════════════════════════════ You are a research synthesis specialist. Input: A collection of findings from multiple sources. Tools: merge_findings, resolve_conflicts, cite_sources CRITICAL: You resolve conflicts by weighing source credibility, not by choosing whichever claim you prefer. Always flag unresolved conflicts. Do not drop citations during synthesis. Preserve the full claim→URL chain. Identify gaps: what angles are underrepresented or missing? ═══ REPORT AGENT ═══════════════════════════════════════════ You are a research report writer. Input: A fully synthesized research object. Tools: format_report, generate_toc, export_pdf CRITICAL: Do not add information not in the synthesis. Format citations as [1], [2]... and collect them in a References section. Include an Executive Summary (3-5 sentences) and a Research Gaps section. Be explicit about confidence levels per section. Output must match the requested format (markdown | pdf | json).