Opus 4.6 scored #1 on BrowseComp — the benchmark for complex, multi-source research tasks. This lesson teaches you to build research pipelines that exploit that capability: gathering information from multiple sources, synthesizing findings, verifying facts, and generating properly cited outputs.
Research Pipeline Architecture
┌──────────┐ ┌──────────────┐ ┌───────────┐ ┌──────────┐
│ Query │───▶│ Source │───▶│ Synthesis │───▶│ Output │
│ Planning│ │ Retrieval │ │ & Verify │ │ + Cites │
└──────────┘ └──────────────┘ └───────────┘ └──────────┘
│ │ │ │
▼ ▼ ▼ ▼
Decompose Search APIs Cross-ref Markdown/
question Documents Sources PDF/JSON
into sub- Web pages Flag gaps
queries DatabasesStep 1: Query Decomposition
Complex research questions need to be broken into searchable sub-queries:
from anthropic import Anthropic
import json
client = Anthropic()
def decompose_research_question(question: str) -> list[dict]:
"""Break a complex question into searchable sub-queries."""
response = client.messages.create(
model="claude-opus-4-6-20260205",
max_tokens=2048,
thinking={"type": "adaptive"},
system="""You decompose complex research questions into specific,
searchable sub-queries. Output JSON array with:
[{"sub_query": "...", "purpose": "...", "source_type": "academic|web|database|expert"}]
Rules:
- Each sub-query should be independently searchable
- Cover all aspects of the original question
- Order from foundational to advanced
- Include at least one verification query""",
messages=[{"role": "user", "content": question}]
)
text = next(b.text for b in response.content if b.type == "text")
return json.loads(text)
# Example
sub_queries = decompose_research_question(
"What are the most effective techniques for reducing hallucination "
"in large language models, and how do they compare in terms of "
"computational cost and accuracy improvement?"
)Step 2: Multi-Source Retrieval
from dataclasses import dataclass
@dataclass
class Source:
title: str
content: str
url: str
source_type: str # "academic", "web", "database"
reliability: float # 0.0-1.0
class SourceRetriever:
"""Retrieve information from multiple source types."""
def __init__(self):
self.client = Anthropic()
self.sources: list[Source] = []
def add_document(self, title: str, content: str,
url: str = "", source_type: str = "document",
reliability: float = 0.8):
"""Add a document source to the retriever."""
self.sources.append(Source(
title=title, content=content, url=url,
source_type=source_type, reliability=reliability
))
def retrieve_for_query(self, query: str,
max_sources: int = 10) -> list[Source]:
"""Find the most relevant sources for a query."""
if not self.sources:
return []
source_descriptions = "\n".join(
f"[{i}] {s.title} ({s.source_type}, reliability: {s.reliability})"
for i, s in enumerate(self.sources)
)
response = self.client.messages.create(
model="claude-opus-4-6-20260205",
max_tokens=1024,
thinking={"type": "adaptive"},
messages=[{
"role": "user",
"content": f"""Given this research query: "{query}"
Which of these sources are most relevant? Return a JSON array of
indices (0-based), ordered by relevance. Maximum {max_sources} sources.
Available sources:
{source_descriptions}"""
}]
)
text = next(b.text for b in response.content if b.type == "text")
indices = json.loads(text)
return [self.sources[i] for i in indices if i < len(self.sources)]Step 3: Synthesis with Cross-Referencing
class ResearchSynthesizer:
"""Synthesize findings from multiple sources with cross-referencing."""
SYSTEM_PROMPT = """You are a research analyst. When synthesizing
information from multiple sources:
1. Cross-reference claims across sources
2. Note agreement and disagreement between sources
3. Flag claims that appear in only one source
4. Rate confidence for each finding (HIGH/MEDIUM/LOW)
5. Identify gaps where no source provides information
6. Use inline citations: [Source Title, Section]
Output structure:
## Key Findings
(numbered findings with confidence and citations)
## Agreements Across Sources
(claims confirmed by multiple sources)
## Contradictions
(where sources disagree, with both positions cited)
## Gaps
(questions not answered by available sources)
## Recommended Next Steps
(what additional research is needed)"""
def __init__(self):
self.client = Anthropic()
def synthesize(self, question: str,
sources: list[Source]) -> str:
source_content = ""
for i, src in enumerate(sources):
source_content += f"""
### Source {i+1}: {src.title}
Type: {src.source_type} | Reliability: {src.reliability}
URL: {src.url}
{src.content[:5000]} # Truncate very long sources
---
"""
response = self.client.messages.create(
model="claude-opus-4-6-20260205",
max_tokens=8192,
thinking={"type": "adaptive", "effort": "maximum"},
system=self.SYSTEM_PROMPT,
messages=[{
"role": "user",
"content": f"""Research question: {question}
Sources:
{source_content}
Synthesize findings, cross-reference claims, and identify gaps."""
}]
)
return next(b.text for b in response.content if b.type == "text")Step 4: Fact-Checking Pipeline
class FactChecker:
"""Verify claims extracted from research synthesis."""
def __init__(self):
self.client = Anthropic()
def extract_claims(self, synthesis: str) -> list[dict]:
"""Extract verifiable claims from a research synthesis."""
response = self.client.messages.create(
model="claude-opus-4-6-20260205",
max_tokens=4096,
messages=[{
"role": "user",
"content": f"""Extract all factual claims from this text.
For each claim, output JSON:
[{{"claim": "...", "category": "statistical|causal|comparative|temporal",
"verifiable": true/false, "cited_source": "..."}}]
Text:
{synthesis}"""
}]
)
text = next(b.text for b in response.content if b.type == "text")
return json.loads(text)
def verify_claim(self, claim: dict,
available_sources: list[Source]) -> dict:
"""Verify a single claim against available sources."""
source_text = "\n\n".join(
f"[{s.title}]: {s.content[:2000]}"
for s in available_sources
)
response = self.client.messages.create(
model="claude-opus-4-6-20260205",
max_tokens=1024,
thinking={"type": "adaptive", "effort": "deep"},
messages=[{
"role": "user",
"content": f"""Verify this claim against the provided sources.
Claim: {claim['claim']}
Category: {claim['category']}
Sources:
{source_text}
Output JSON:
{{"verdict": "CONFIRMED|PARTIALLY_CONFIRMED|UNVERIFIABLE|CONTRADICTED",
"confidence": 0.0-1.0,
"supporting_evidence": "...",
"contradicting_evidence": "...",
"notes": "..."}}"""
}]
)
text = next(b.text for b in response.content if b.type == "text")
return json.loads(text)
def verify_all(self, synthesis: str,
sources: list[Source]) -> list[dict]:
"""Extract and verify all claims in a synthesis."""
claims = self.extract_claims(synthesis)
results = []
for claim in claims:
if claim.get("verifiable"):
result = self.verify_claim(claim, sources)
results.append({**claim, **result})
return resultsStep 5: Citation Generation
class CitationFormatter:
"""Generate properly formatted citations."""
def format_citations(self, sources: list[Source],
style: str = "APA") -> str:
"""Generate formatted citation list."""
response = Anthropic().messages.create(
model="claude-opus-4-6-20260205",
max_tokens=2048,
messages=[{
"role": "user",
"content": f"""Format these sources as {style} citations:
{json.dumps([{
"title": s.title,
"url": s.url,
"type": s.source_type
} for s in sources], indent=2)}
Output a numbered reference list in proper {style} format."""
}]
)
return next(b.text for b in response.content if b.type == "text")Complete Pipeline Orchestration
class ResearchPipeline:
"""End-to-end research pipeline."""
def __init__(self):
self.retriever = SourceRetriever()
self.synthesizer = ResearchSynthesizer()
self.fact_checker = FactChecker()
self.citation_formatter = CitationFormatter()
def research(self, question: str) -> dict:
# Step 1: Decompose
sub_queries = decompose_research_question(question)
# Step 2: Retrieve sources for each sub-query
all_sources = []
for sq in sub_queries:
sources = self.retriever.retrieve_for_query(sq["sub_query"])
all_sources.extend(sources)
# Deduplicate
seen = set()
unique_sources = []
for s in all_sources:
if s.title not in seen:
seen.add(s.title)
unique_sources.append(s)
# Step 3: Synthesize
synthesis = self.synthesizer.synthesize(question, unique_sources)
# Step 4: Verify
verification = self.fact_checker.verify_all(synthesis, unique_sources)
# Step 5: Citations
citations = self.citation_formatter.format_citations(unique_sources)
return {
"question": question,
"sub_queries": sub_queries,
"sources_used": len(unique_sources),
"synthesis": synthesis,
"verification": verification,
"citations": citations,
"confidence": self._overall_confidence(verification),
}
def _overall_confidence(self, verification: list[dict]) -> float:
if not verification:
return 0.0
confidences = [v.get("confidence", 0.5) for v in verification]
return sum(confidences) / len(confidences)The research pipeline pattern scales from simple literature reviews to complex multi-source investigations. In the next lesson, you will learn how to combine this with RAG patterns for even more powerful information retrieval.