AdvancedResearch：Anthropic 多智能体研究系统的开源实现

pip3 install -U advanced-research

SWARMS_API_KEY="..."   # https://swarms.world/platform/api-keys
EXA_API_KEY="..."      # https://dashboard.exa.ai/api-keys

from advanced_research import AdvancedResearch

research = AdvancedResearch(
    name="AI Research Team",
    director_model_name="gpt-5.5",     # 主导研究员、报告、引用
    worker_model_name="gpt-5.4-mini",  # 搜索子智能体
    max_loops=1,
    verbose=True,                      # 每个智能体的回复一到就打印
)

report = research.run(
    "How do solid-state batteries compare with lithium-ion batteries for electric vehicles in 2026?"
)
print(report)
print(research.get_usage())  # token 数以及 Swarms + Exa 费用

# advanced_research/prompts.py（节选）
STEP 2: WRITE ONE SUBTASK PER SUBAGENT
Each subtask must give the subagent everything it needs, because it cannot ask you questions:
- objective: one precise question to answer, including the time frame when recency matters.
- search_queries: 2 or 3 web search queries. Start short and broad, then one more specific query.
- output_format: what the subagent should return, e.g. "a list of approved drugs with approval dates and sources".
- boundaries: what is out of scope for this subagent, so two subagents never cover the same ground.

STEP 1: JUDGE THE EFFORT THE QUERY NEEDS
- simple: one fact or a narrow question. Use 1 subagent.
- comparison: a direct comparison or a question with a few distinct parts. Use 2 to 4 subagents.
- complex: open-ended research across many angles. Use up to {max_subagents} subagents with clearly divided responsibilities.
Never use more subagents than the query needs.

# advanced_research/main.py，run_agent() 内部
for round_number in range(1, config.worker_search_rounds + 1):
    queries = [q for q in queries if q not in seen_queries][:3]
    if not queries:
        break
    seen_queries.extend(queries)
    with concurrent.futures.ThreadPoolExecutor(max_workers=len(queries)) as pool:
        search_outputs = list(pool.map(lambda q: exa_search(q, ...), queries))
    ...
    queries = [q for q in findings.get("follow_up_queries") or []]
    if findings.get("coverage_assessment") == "comprehensive":
        break

class SourceRegistry:
    """Numbers every source found during a run, shared by all subagents,
    so a citation like [3] means the same page everywhere."""

    def add(self, result: dict[str, Any]) -> int:
        with self._lock:
            url = result["url"]
            if url not in self._ids:
                self.sources.append(result)
                self._ids[url] = len(self.sources)
            return self._ids[url]

{"needs_more_research": false, "gaps": []}

{
  "complexity": "comparison",
  "strategy": "Divide the comparison into four non-overlapping angles: technical performance, safety/lifetime, commercialization status, and cost/supply-chain outlook for EV applications as of November 2026.",
  "subtasks": [
    {
      "objective": "Compare solid-state batteries and conventional lithium-ion batteries for EVs on technical performance metrics as of 2026, including energy density, charging speed, power capability, operating temperature range, and vehicle range implications.",
      "search_queries": [
        "solid-state battery vs lithium-ion EV energy density 2026",
        "solid state batteries electric vehicles charging speed power density 2026",
        "EV solid-state battery performance metrics 2026"
      ],
      "output_format": "A comparison table with metrics, units, typical 2026 values or ranges, and cited sources for each figure.",
      "boundaries": "Focus only on technical performance metrics. Do not cover safety, cost, manufacturing scale, commercial rollout timelines, or company-specific deployment status except where needed to source a metric."
    },
    {
      "objective": "Compare solid-state and lithium-ion EV batteries on safety, degradation, cycle life, calendar life, thermal runaway risk, and real-world reliability as understood in 2026.",
      "search_queries": [
        "solid-state battery safety cycle life lithium-ion EV 2026",
        "solid state battery thermal runaway degradation cycle life 2026",
        "EV battery safety comparison solid-state lithium-ion 2026"
      ],
      "output_format": "A concise evidence summary with separate sections for safety, degradation/cycle life, calendar life, and reliability, noting where evidence is lab-scale versus real-world EV use.",
      "boundaries": "Do not discuss energy density, charging speed, costs, supply chains, or commercialization timelines except where directly tied to safety or lifetime evidence."
    },
    {
      "objective": "Determine the commercialization status of solid-state EV batteries versus conventional lithium-ion EV batteries as of 08 November 2026, including which companies have announced vehicles, pilot production, mass production, or delays.",
      "search_queries": [
        "solid-state battery EV commercialization 2026 Toyota QuantumScape Nissan BMW",
        "solid state battery electric vehicle production timeline 2026",
        "which EVs use solid-state batteries 2026"
      ],
      "output_format": "A company-by-company table listing announced milestones, current status as of November 2026, expected production dates, named vehicle programs if any, and sources.",
      "boundaries": "Focus on commercialization and deployment status only. Do not compare chemistry-level performance, safety, or cost except when an announcement includes those claims."
    },
    {
      "objective": "Compare the 2026 cost, manufacturing readiness, supply-chain requirements, and scalability of solid-state batteries versus lithium-ion batteries for EVs.",
      "search_queries": [
        "solid-state battery cost per kWh 2026 lithium-ion EV",
        "solid state battery manufacturing challenges scalability 2026",
        "EV battery cost forecast solid-state versus lithium-ion 2026"
      ],
      "output_format": "A cost and scalability comparison with estimated $/kWh ranges where available, key manufacturing bottlenecks, material requirements, and short-term outlook through 2030.",
      "boundaries": "Do not cover technical performance, safety, degradation, or company deployment timelines except where they directly affect manufacturing cost or scalability."
    }
  ]
}

def _post_json(service, url, headers, body, timeout=120.0, attempts=3):
    for attempt in range(1, attempts + 1):
        started = time.monotonic()
        try:
            response = httpx.post(url, headers=headers, json=body, timeout=timeout)
        except httpx.TransportError as error:
            if attempt == attempts:
                raise ResearchAPIError(f"{service} request failed: {error}") from error
            time.sleep(2**attempt)
            continue
        elapsed = time.monotonic() - started
        if response.status_code in (503, 504) and elapsed >= GATEWAY_TIMEOUT_SECONDS:
            raise ResearchAPIError(
                f"{service} returned {response.status_code} after {elapsed:.0f} s, the "
                "Swarms API's 30-second request limit. Use a faster model "
                "or a lower max_tokens for this agent."
            )
        if response.status_code in (429, 500, 502, 503) and attempt < attempts:
            time.sleep(2**attempt)
            continue
        if response.status_code >= 400:
            raise ResearchAPIError(f"{service} returned {response.status_code}: {response.text[:500]}")
        return response.json()

research = AdvancedResearch(
    director_model_name="claude-sonnet-5-5",
    worker_model_name="gpt-5.4-mini",
    max_loops=2,
    export_on=True,  # 把历史、来源和用量写入 examples/outputs/<id>.json
)
report = research.run("What are the most effective treatments for Type 2 diabetes as of 2026?")

first = research.step("How is the EU AI Act being enforced in 2026?")
print(first["plan"]["complexity"], len(first["plan"]["subtasks"]))
if first["needs_more_research"]:
    second = research.step(
        "How is the EU AI Act being enforced in 2026?",
        previous_report=first["report"],
        gaps=first["gaps"],
    )

from advanced_research import AdvancedResearch, AdvancedResearchAdditionalConfig

config = AdvancedResearchAdditionalConfig(
    exa_search_num_results=8,
    exa_search_max_characters=1200,
    worker_search_rounds=3,
    max_subagents=4,
)
research = AdvancedResearch(config=config, worker_model_name="claude-haiku-5-5")