用多智能体集群构建你自己的深度研究 API

# deep_research.py
# A deep research pipeline on the Swarms API: a planner writes sub-questions,
# searchers answer them in parallel with web search, then a synthesizer and a
# critic produce the report. Requires: pip install requests
import json
import os
import sys
import time
from concurrent.futures import ThreadPoolExecutor

import requests

API = "https://api.swarms.world"
HEADERS = {"x-api-key": os.environ["SWARMS_API_KEY"], "Content-Type": "application/json"}


def agent(name: str, system_prompt: str, task: str, model: str = "gpt-5.4-mini",
          tools: list[str] | None = None, max_tokens: int = 900) -> tuple[str, float]:
    """One call to /v1/agent/completions. Returns (text, cost_in_usd)."""
    body = {
        "agent_config": {
            "agent_name": name,
            "system_prompt": system_prompt,
            "model_name": model,
            "max_loops": 1,
            "max_tokens": max_tokens,
        },
        "task": task,
    }
    if tools:
        body["tools_enabled"] = tools
    r = requests.post(f"{API}/v1/agent/completions", headers=HEADERS, json=body, timeout=180)
    r.raise_for_status()
    data = r.json()
    # With tools on, outputs also contain tool-call entries whose content is a
    # list. The agent's own prose is the string content under its name.
    text = "\n".join(
        m["content"] for m in data["outputs"] if m["role"] == name and isinstance(m["content"], str)
    )
    return text.strip(), data["usage"]["total_cost"]


def plan(question: str) -> tuple[list[str], float]:
    text, cost = agent(
        "Research Planner",
        "You are a research planner. Given a research question, return exactly 4 focused "
        "sub-questions, one per line, with no numbering and no commentary. Each should be "
        "answerable with a web search and together they should cover the question.",
        question,
        max_tokens=400,
    )
    return [line.strip() for line in text.splitlines() if line.strip()], cost


def search(sub_question: str) -> tuple[str, float]:
    return agent(
        "Searcher",
        "You are a research assistant with web search and a page scraper. Answer the question "
        "with current, sourced facts in 150-250 words. Cite the URL of each source you rely on. "
        "If sources conflict, say so.",
        sub_question,
        tools=["auto_search", "web_scraper"],
    )


def synthesize(question: str, findings: list[str]) -> tuple[str, float, float]:
    """Synthesizer then critic, as one SequentialWorkflow swarm."""
    task = f"Research question: {question}\n\nFindings from the search stage:\n\n" + "\n\n".join(
        f"[F{i + 1}] {f}" for i, f in enumerate(findings)
    ) + "\n\nWrite the final report."
    body = {
        "name": "Deep Research Synthesis",
        "swarm_type": "SequentialWorkflow",
        "task": task,
        "max_loops": 1,
        "agents": [
            {
                "agent_name": "Synthesizer",
                "system_prompt": "You are a senior research writer. Turn the findings into a report with the sections: Summary, What changed, Who is affected, Deadlines, Recommended actions, Open questions. Tag every factual claim with the finding it comes from, like [F2]. Under 600 words.",
                "model_name": "claude-sonnet-5-5",
                "max_loops": 1,
                "max_tokens": 1400,
            },
            {
                "agent_name": "Critic",
                "system_prompt": "You are a fact-checking editor. Review the report you receive. List any claim not supported by the findings, any missing deadline, and any vague recommendation, then output the corrected report in full. Under 700 words.",
                "model_name": "gpt-5.4-mini",
                "max_loops": 1,
                "max_tokens": 1600,
            },
        ],
    }
    r = requests.post(f"{API}/v1/swarm/completions", headers=HEADERS, json=body, timeout=300)
    r.raise_for_status()
    data = r.json()
    return data["output"][-1]["content"], data["usage"]["total_cost"], data["execution_time"]


def deep_research(question: str) -> dict:
    started = time.perf_counter()
    sub_questions, plan_cost = plan(question)

    with ThreadPoolExecutor(max_workers=len(sub_questions)) as pool:
        results = list(pool.map(search, sub_questions))
    findings = [text for text, _ in results]
    search_cost = sum(cost for _, cost in results)

    report, synth_cost, synth_seconds = synthesize(question, findings)

    return {
        "question": question,
        "sub_questions": sub_questions,
        "report": report,
        "cost_usd": round(plan_cost + search_cost + synth_cost, 4),
        "seconds": round(time.perf_counter() - started, 1),
        "synthesis_seconds": round(synth_seconds, 1),
    }


if __name__ == "__main__":
    q = sys.argv[1] if len(sys.argv) > 1 else "What changed in EU AI Act obligations for general-purpose AI model providers during 2026, and what should a startup shipping an LLM-based product do before the next deadline?"
    print(json.dumps(deep_research(q), indent=2))

pip install requests
export SWARMS_API_KEY=your_key
python deep_research.py "What changed in EU AI Act obligations for general-purpose AI model providers during 2026?"

# research_api.py
# Exposes the pipeline as an HTTP API. Run: uvicorn research_api:app --port 8000
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

from deep_research import deep_research

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


class ResearchRequest(BaseModel):
    question: str = Field(min_length=10, max_length=500)


@app.post("/research")
def research(req: ResearchRequest) -> dict:
    try:
        return deep_research(req.question)
    except Exception as exc:  # surfaces Swarms API errors as a 502 with the message
        raise HTTPException(status_code=502, detail=str(exc)) from exc

pip install fastapi uvicorn
uvicorn research_api:app --port 8000

curl -X POST http://localhost:8000/research \
  -H "Content-Type: application/json" \
  -d '{"question": "What changed in EU AI Act obligations for general-purpose AI model providers during 2026, and what should a startup shipping an LLM-based product do before the next deadline?"}'

curl -X POST https://api.swarms.world/v1/agent/completions \
  -H "x-api-key: $SWARMS_API_KEY" -H "Content-Type: application/json" \
  -d '{"agent_config": {"agent_name": "Researcher", "system_prompt": "Answer with current, sourced facts and cite the URL of each source.", "model_name": "gpt-5.4-mini", "max_loops": 1, "max_tokens": 800}, "task": "What are the current EU AI Act deadlines for general-purpose AI model providers?", "tools_enabled": ["auto_search"]}'