Multi-Agent Research System.

from typing import List

def decompose_query(query: str) -> List[str]:
    """Semantic decomposition. Enumerate all relevant sub-domains.

    The exam-question distractor is to blame subagents for narrow
    coverage when the coordinator's decomposition was the bug.
    """
    q = query.lower()
    if "creative industries" in q:
        domains = [
            "visual arts (digital art, graphic design, photography)",
            "music production and composition",
            "writing (novels, journalism, screenwriting)",
            "film and video production",
            "performing arts (theater, dance)",
        ]
    elif "healthcare" in q:
        domains = [
            "clinical decision support",
            "medical imaging and diagnostics",
            "drug discovery and trials",
            "patient-facing communication",
            "administrative + revenue cycle",
        ]
    else:
        # Generic fallback. STILL decompose, never single-shot
        domains = [
            f"{query}. Recent academic literature",
            f"{query}. Industry case studies",
            f"{query}. Empirical adoption data",
        ]
    return [f"Find AI impact on {d}" for d in domains]

RESEARCH_SUBAGENT_SYSTEM = """You are a research specialist.
1. Find authoritative sources on the assigned sub-domain.
2. Extract key claims with evidence.
3. Return structured JSON ONLY:
   {"findings": [
       {"claim": "...", "sources": [{"url": "...", "date": "YYYY-MM-DD", "confidence": 0.0-1.0}]}
   ]}
Do NOT synthesize, editorialize, or pick winners between conflicting sources.
Report facts as stated. Preserve contradictions for verification."""

VERIFICATION_SUBAGENT_SYSTEM = """You are a fact-checker.
1. Read pooled claims from research subagents.
2. Cross-check each claim against its sources.
3. When sources conflict, PRESERVE BOTH with attribution + context.
4. Return JSON:
   {"verifications": [
       {"claim": "...", "verified": true|false, "confidence": 0.0-1.0,
        "sources_reconciled": [...], "notes": "context about conflicts"}
   ]}"""

SYNTHESIS_SUBAGENT_SYSTEM = """You are a synthesis specialist.
1. Read verified findings (JSON).
2. Write a cohesive Markdown narrative with inline [n] citations.
3. Acknowledge data gaps transparently when verifications flag them.
4. Do NOT conduct new research. You have ONLY the Read tool."""

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

async def spawn_research(task: str) -> dict:
    """One isolated research subagent. No inherited history."""
    resp = await client.messages.create(
        model="claude-sonnet-4.5",
        max_tokens=2048,
        system=RESEARCH_SUBAGENT_SYSTEM,
        tools=RESEARCH_TOOLS,  # [Read, WebSearch, Bash]
        messages=[{"role": "user", "content": task}],
    )
    return {
        "task": task,
        "stop_reason": resp.stop_reason,
        "result": extract_json(resp),
    }

async def research_in_parallel(tasks: list[str]) -> list[dict]:
    """Fan out N subagents at once. Latency = max, not sum."""
    return await asyncio.gather(*(spawn_research(t) for t in tasks))

# Coordinator usage
tasks = decompose_query("impact of AI on creative industries")
results = asyncio.run(research_in_parallel(tasks))

def handle_subagent_error(error_type: str, query: str, partial: list | None = None) -> dict:
    """Structured error context. Coordinator consumes status_code to decide."""
    if error_type == "timeout":
        return {
            "status": "timeout",
            "query": query,
            "partial_results": partial or [],
            "alternatives": [
                "Narrow the query to a single sub-aspect",
                "Try synonyms or domain-specific terms",
                "Reduce the time horizon (e.g., last 12 months)",
            ],
        }
    if error_type == "no_results":
        return {
            "status": "no_results",
            "query": query,
            "alternatives": ["Broaden keywords", "Check spelling", "Try sister terms"],
        }
    if error_type == "rate_limited":
        return {"status": "rate_limited", "query": query, "retry_after_s": 60}
    return {"status": "unknown_error", "query": query}

# Coordinator inspects status_code
def coordinator_handle(result: dict, original_task: str):
    status = result.get("status")
    if status == "timeout":
        # Either retry with the first alternative or include a transparent gap
        return retry_with_narrower_query(result["alternatives"][0])
    if status == "no_results":
        return mark_data_unavailable(original_task)
    return result  # success

def build_verification_task(pooled_claims: list[dict]) -> str:
    """Pool all research-subagent claims for fact-checking."""
    body = "\n".join(
        f"- Claim: {c['claim']}\n  Sources: {c['sources']}"
        for c in pooled_claims
    )
    return f"""Verify the following claims pooled from research subagents.

For each claim:
1. Check that sources are credible and dated.
2. If two or more sources CONFLICT, do NOT pick a winner.
   Preserve both with their context (timeframe, definition, population)
   and emit both in sources_reconciled with notes explaining the apparent
   contradiction. The reader gets BOTH numbers + WHY they differ.
3. Return JSON:
   {{"verifications": [
       {{"claim": "...", "verified": true|false, "confidence": 0.0-1.0,
         "sources_reconciled": [...], "notes": "..." }}
   ]}}

CLAIMS TO VERIFY:
{body}"""

# Coordinator pools claims and dispatches
all_claims = [c for r in research_results for c in r.get("findings", [])]
verification_task = build_verification_task(all_claims)
verified = client.messages.create(
    model="claude-sonnet-4.5",
    max_tokens=4096,
    system=VERIFICATION_SUBAGENT_SYSTEM,
    tools=VERIFICATION_TOOLS,
    messages=[{"role": "user", "content": verification_task}],
)

def synthesize(verified_claims: list[dict], narrative_prompt: str) -> str:
    """Read-only narrative generation. No re-research."""
    task = f"""Narrative prompt from coordinator:
{narrative_prompt}

Verified findings (JSON):
{json.dumps(verified_claims, indent=2)}

Write a Markdown report that:
1. Flows logically from finding to finding
2. Uses inline citations [1], [2], etc. matching sources_reconciled order
3. ACKNOWLEDGES data gaps transparently where verifications.notes flagged them
4. Does NOT re-research or speculate beyond the verified findings

You have ONLY the Read tool. No WebSearch, no Bash."""

    resp = client.messages.create(
        model="claude-sonnet-4.5",
        max_tokens=3000,
        system=SYNTHESIS_SUBAGENT_SYSTEM,
        tools=[READ_TOOL_ONLY],  # Critical: read-only
        messages=[{"role": "user", "content": task}],
    )
    return extract_text(resp)

# WRONG. Direct subagent-to-subagent communication
# class ResearcherA:
#     def __init__(self, researcher_b):
#         self.b = researcher_b  # ❌ creates hidden coupling
#     async def find_papers(self, topic):
#         finding = await self._search(topic)
#         await self.b.search_web(finding)  # ❌ breaks isolation

# RIGHT. Coordinator owns all routing
async def coordinated_dependent_research(topic: str):
    # Phase 1: A runs alone
    a_finding = await spawn_research(f"Find academic papers on {topic}")

    # Phase 2: coordinator passes A's finding into B's TASK PROMPT
    b_task = (
        f"Topic: {topic}. "
        f"Key research direction surfaced by paper search: "
        f"{a_finding['result']['top_direction']}. "
        f"Now search the web for industry coverage of that direction."
    )
    b_result = await spawn_research(b_task)

    # Phase 3: coordinator continues with both
    return {"a": a_finding, "b": b_result}

import asyncio

MAX_PARALLEL = 5
RETRY_BUDGET = 2

semaphore = asyncio.Semaphore(MAX_PARALLEL)

async def spawn_with_retry(task: str, attempt: int = 0) -> dict:
    """Bounded-concurrency research with structured-error retry."""
    async with semaphore:
        result = await spawn_research(task)
        status = result.get("result", {}).get("status")
        if status == "timeout" and attempt < RETRY_BUDGET:
            narrower = narrow_query(result["result"].get("alternatives", [task])[0])
            telemetry.increment("subagent.retry", labels={"reason": "timeout"})
            return await spawn_with_retry(narrower, attempt + 1)
        return result

# Coordinator with bounded fan-out + retries
results = await asyncio.gather(*(spawn_with_retry(t) for t in tasks))