为什么你的 Python 智能体框架要花 2 秒导入（以及我们如何把它降到 0.28 秒）

"""Time an import in fresh Python processes and report the median."""
import statistics
import subprocess
import sys

MODULE = "swarms"  # change to any module you want to measure
RUNS = 9

# Runs inside each child process: time only the import statement.
CHILD = (
    "import time\n"
    "t0 = time.perf_counter()\n"
    f"import {MODULE}\n"
    "print(time.perf_counter() - t0)\n"
)


def import_seconds() -> float:
    """Start a new interpreter, import MODULE, return the seconds it took."""
    result = subprocess.run(
        [sys.executable, "-c", CHILD],
        capture_output=True,
        text=True,
        check=True,
    )
    return float(result.stdout.strip().splitlines()[-1])


import_seconds()  # warm-up: the first run pays for cold disk caches and .pyc writes
samples = sorted(import_seconds() for _ in range(RUNS))

print(f"import {MODULE}: median {statistics.median(samples):.3f} s over {RUNS} fresh processes")
print(f"min {samples[0]:.3f} s, max {samples[-1]:.3f} s")

python -X importtime -c "import swarms" 2> importtime.log

# the 20 largest cumulative times (microseconds, second column)
sort -t'|' -k2 -n importtime.log | tail -20

pip install tuna
tuna importtime.log

"""Check what `import swarms` and building an Agent load and write."""
import json
import os
import subprocess
import sys
import tempfile

HEAVY = [
    "litellm", "routehub", "mcp", "networkx", "rustworkx", "graphviz",
    "opentelemetry", "starlette", "anyio", "toml", "sqlite3",
]

CHILD = r"""
import json, sys, time

t0 = time.perf_counter()
import swarms
t1 = time.perf_counter()

from swarms import Agent

agent = Agent(
    agent_name="Cold-Start-Probe",
    model_name="gpt-5.4-mini",
    max_loops=1,
    context_length=200_000,  # set both limits, so the model catalog is never consulted
    max_tokens=4_096,
)
t2 = time.perf_counter()

heavy = json.loads(sys.argv[1])
loaded = sorted(h for h in heavy if any(m == h or m.startswith(h + ".") for m in sys.modules))
print(json.dumps({"import_s": t1 - t0, "agent_s": t2 - t1, "loaded": loaded}))
"""

with tempfile.TemporaryDirectory() as workdir:
    env = {k: v for k, v in os.environ.items() if k != "WORKSPACE_DIR"}
    result = subprocess.run(
        [sys.executable, "-c", CHILD, json.dumps(HEAVY)],
        cwd=workdir, env=env, capture_output=True, text=True, check=True,
    )
    report = json.loads(result.stdout.strip().splitlines()[-1])
    written = sorted(os.listdir(workdir))

print(f"import swarms:   {report['import_s']:.3f} s")
print(f"Agent(...):      {report['agent_s'] * 1000:.1f} ms")
print(f"heavy libraries: {report['loaded'] or 'none'}")
print(f"files written:   {written or 'none'}")

nx = _LazyModule("networkx")
rx = _LazyModule("rustworkx") if RUSTWORKX_AVAILABLE else None

fastpkg/
    __init__.py
    client.py
    store.py
demo.py

"""fastpkg: cheap to import, loads its heavy parts on first use (PEP 562)."""
import importlib
from typing import TYPE_CHECKING, Any

from fastpkg.client import Client  # light: imported eagerly

# Public name -> the submodule that defines it.
_LAZY = {
    "ReportStore": "fastpkg.store",
}

__all__ = ["Client", *_LAZY]

if TYPE_CHECKING:  # type checkers and IDEs still see the real classes
    from fastpkg.store import ReportStore


def __getattr__(name: str) -> Any:
    """Called only when `name` is not already a module global."""
    module_name = _LAZY.get(name)
    if module_name is None:
        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
    value = getattr(importlib.import_module(module_name), name)
    globals()[name] = value  # cache it: later lookups never reach __getattr__
    return value


def __dir__() -> list:
    return sorted(set(globals()) | set(__all__))

# fastpkg/client.py
class Client:
    def __init__(self, name: str):
        self.name = name

# fastpkg/store.py
import sqlite3  # stands in for any expensive dependency


class ReportStore:
    def __init__(self, path: str = ":memory:"):
        self.conn = sqlite3.connect(path)
        self.conn.execute("CREATE TABLE IF NOT EXISTS reports (body TEXT)")

    def add(self, body: str) -> None:
        self.conn.execute("INSERT INTO reports VALUES (?)", (body,))

    def count(self) -> int:
        return self.conn.execute("SELECT COUNT(*) FROM reports").fetchone()[0]

import sys

import fastpkg

print("after import:      store loaded =", "fastpkg.store" in sys.modules)

client = fastpkg.Client("cli")
print("after Client:      store loaded =", "fastpkg.store" in sys.modules)

store = fastpkg.ReportStore()  # first access runs __getattr__ and imports the module
store.add("nightly run ok")
print("after ReportStore: store loaded =", "fastpkg.store" in sys.modules, "| rows:", store.count())

from fastpkg import ReportStore  # `from` imports go through __getattr__ too
print(ReportStore is fastpkg.ReportStore)

"""A module-level stand-in that imports the real module on first attribute access."""
import importlib
import sys
from typing import Any


class LazyModule:
    def __init__(self, name: str):
        self._name = name
        self._module = None

    def __getattr__(self, attr: str) -> Any:
        if self._module is None:
            self._module = importlib.import_module(self._name)
        return getattr(self._module, attr)


sqlite3 = LazyModule("sqlite3")  # swap in networkx, pandas, torch, ...


def count_rows() -> int:
    conn = sqlite3.connect(":memory:")  # the real import happens here, once
    conn.execute("CREATE TABLE t (x INTEGER)")
    conn.executemany("INSERT INTO t VALUES (?)", [(1,), (2,), (3,)])
    return conn.execute("SELECT COUNT(*) FROM t").fetchone()[0]


print("before first use:", "sqlite3" in sys.modules)
print("rows:", count_rows())
print("after first use: ", "sqlite3" in sys.modules)

"""test_import_budget.py: fail CI when an import-time regression lands."""
import json
import subprocess
import sys

PACKAGE = "swarms"
MUST_STAY_LAZY = ["litellm", "routehub", "mcp", "networkx", "starlette"]


def modules_loaded_by_import() -> set:
    code = f"import json, sys; import {PACKAGE}; print(json.dumps(sorted(sys.modules)))"
    out = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, check=True)
    return set(json.loads(out.stdout))


def test_heavy_dependencies_stay_lazy():
    loaded = modules_loaded_by_import()
    eager = [name for name in MUST_STAY_LAZY if name in loaded]
    assert not eager, f"import {PACKAGE} now loads {eager}"


if __name__ == "__main__":
    test_heavy_dependencies_stay_lazy()
    print("ok: none of", MUST_STAY_LAZY, "load at import")

"""A serverless-style handler: import once per container, one fresh Agent per request."""
import json
import time

from swarms import Agent  # module scope: paid once per cold start


def build_agent() -> Agent:
    return Agent(
        agent_name="Ticket-Triage",
        model_name="anthropic/claude-haiku-5-5",
        system_prompt=(
            "Classify the support ticket as billing, bug or how-to. "
            "Reply with the label only."
        ),
        max_loops=1,
        context_length=100_000,  # limits you set yourself: no model catalog lookup at build time
        max_tokens=256,
        output_type="final",     # return the reply, not the conversation
        print_on=False,          # no console panels in a function log
    )


def handler(event, context=None):
    started = time.perf_counter()
    agent = build_agent()  # cheap in v17, and no history leaks between requests
    label = agent.run(event["ticket"])
    return {
        "statusCode": 200,
        "body": json.dumps({
            "label": label,
            "agent_build_and_run_ms": round((time.perf_counter() - started) * 1000, 1),
        }),
    }


if __name__ == "__main__":
    for ticket in ["I was charged twice this month.", "The export button does nothing."]:
        print(handler({"ticket": ticket}))

pip install -U swarms

# RouteHub for model calls and rustworkx for graphs
pip install -U "swarms[fast]"