如何用 Python 追踪每个智能体的 LLM token 用量与成本

pip install -U swarms
# or
uv pip install -U swarms

export OPENAI_API_KEY="sk-..."
export TYPESAFE_API_KEY="..."

python -c "import swarms; print(swarms.__version__)"

from swarms import Agent

agent = Agent(
    agent_name="Usage-Demo",
    system_prompt="You answer in two sentences.",
    model_name="gpt-5.4-mini",
    max_loops=1,
    output_type="final",
    print_on=False,
)

answer = agent.run("What is prompt caching, and why does it lower LLM costs?")
print(answer)
print(agent.usage)

import litellm
from swarms import Agent

MODEL = "gpt-5.4-mini"


def usage_cost(model: str, usage: dict) -> float:
    """Dollar cost of a usage dict, priced from litellm's model price table."""
    input_cost, output_cost = litellm.cost_per_token(
        model=model,
        prompt_tokens=usage["input_tokens"],
        completion_tokens=usage["output_tokens"],
        cache_read_input_tokens=usage["cached_tokens"],
    )
    return input_cost + output_cost


def usage_delta(before: dict, after: dict) -> dict:
    """Usage between two snapshots of agent.usage."""
    return {key: after[key] - before[key] for key in after}


agent = Agent(
    agent_name="Cost-Demo",
    system_prompt="You are a concise database consultant.",
    model_name=MODEL,
    max_loops=1,
    output_type="final",
    print_on=False,
)

agent.run("Name three workloads that suit a vector database.")

before = agent.usage
agent.run("Which of those three needs the lowest latency, and why?")
this_run = usage_delta(before, agent.usage)

print(f"second run: {this_run}  ${usage_cost(MODEL, this_run):.6f}")
print(f"lifetime:   {agent.usage}  ${usage_cost(MODEL, agent.usage):.6f}")

from swarms import Agent

# A stand-in for a long, fixed system prompt: a policy manual or tool guide.
POLICY = "\n".join(
    f"Rule {i}: state every figure with its unit and the table it came from."
    for i in range(1, 151)
)

agent = Agent(
    agent_name="Cache-Demo",
    system_prompt=f"You are a careful analyst.\n\n{POLICY}",
    model_name="gpt-5.4-mini",
    reasoning_effort="medium",
    max_loops=1,
    output_type="final",
    print_on=False,
)

tasks = [
    "A train leaves at 09:40 and arrives at 13:05. How long is the trip?",
    "A tank fills at 12 litres per minute. How long does 300 litres take?",
]

for task in tasks:
    before = agent.usage
    agent.run(task)
    run = {key: agent.usage[key] - before[key] for key in before}
    print(
        f"input={run['input_tokens']} cached={run['cached_tokens']} "
        f"output={run['output_tokens']} reasoning={run['reasoning_tokens']}"
    )

from swarms import Agent

agent = Agent(
    agent_name="Stream-Demo",
    model_name="gpt-5.4-mini",
    max_loops=1,
    print_on=False,
)

for token in agent.run_stream("Write a haiku about token budgets."):
    print(token, end="", flush=True)

print()
print(agent.usage)

import litellm
from swarms import Agent, count_tokens

MODEL = "gpt-5.4-mini"

agent = Agent(
    agent_name="Budget-Demo",
    system_prompt="You summarise documents in three bullets.",
    model_name=MODEL,
    max_loops=1,
    output_type="final",
    print_on=False,
)

document = "\n".join(
    f"Q{q} revenue grew {q * 3}% while support tickets fell {q * 2}%."
    for q in range(1, 41)
)
task = f"Summarise this report:\n\n{document}"

window = litellm.get_model_info(MODEL)["max_input_tokens"]
planned = agent.input_tokens + count_tokens(task, model=MODEL)
print(f"planned input: {planned} of {window} tokens")

if planned > 0.9 * window:
    raise SystemExit("Too large: split the document first.")

agent.run(task)
print(f"billed input:  {agent.usage['input_tokens']} tokens")
print(f"next request:  {agent.input_tokens} tokens")

from swarms import Agent, SwarmRouter

workers = [
    Agent(
        agent_name=name,
        system_prompt=prompt,
        model_name="gpt-5.4-mini",
        max_loops=1,
        print_on=False,
    )
    for name, prompt in [
        ("Researcher", "Gather the key facts. Be brief."),
        ("Writer", "Write a three-sentence brief. Be brief."),
    ]
]

router = SwarmRouter(
    agents=workers,
    swarm_type="HierarchicalSwarm",
    director_model_name="gpt-5.4-mini",
    max_loops=1,
)
router.run("Should a five-person startup self-host its LLM? Recommend one option.")

worker_total = sum(agent.usage["total_tokens"] for agent in workers)
print("workers:", worker_total)
print("router: ", router.usage["total_tokens"])
print("director share:", router.usage["total_tokens"] - worker_total)

from swarms import Agent, GraphWorkflow

def make(name: str, prompt: str) -> Agent:
    return Agent(
        agent_name=name,
        system_prompt=prompt,
        model_name="gpt-5.4-mini",
        max_loops=1,
        print_on=False,
    )

planner = make("Planner", "List two angles to analyse. One line each.")
bull = make("Bull", "Give the strongest case for. Two sentences.")
bear = make("Bear", "Give the strongest case against. Two sentences.")
judge = make("Judge", "Weigh both cases and decide. Two sentences.")

graph = GraphWorkflow(name="usage-graph")
for agent in (planner, bull, bear, judge):
    graph.add_node(agent)
graph.add_edge("Planner", "Bull")
graph.add_edge("Planner", "Bear")
graph.add_edge("Bull", "Judge")
graph.add_edge("Bear", "Judge")

graph.run("Should a mid-size retailer move its data warehouse to the cloud?")

for agent in (planner, bull, bear, judge):
    print(f"{agent.agent_name:<8} {agent.usage['total_tokens']}")
print(f"graph    {graph.usage['total_tokens']}")

from swarms import HeavySwarm

swarm = HeavySwarm(
    question_agent_model_name="gpt-5.4-mini",
    worker_model_name="gpt-5.4-mini",
    max_loops=1,
)
swarm.run("Is a four-day work week viable for a 40-person agency? Be brief.")

agent_total = sum(agent.usage["total_tokens"] for agent in swarm.agents.values())
print("agents:", agent_total)
print("swarm: ", swarm.usage["total_tokens"])
print("question generation:", swarm.usage["total_tokens"] - agent_total)

from swarms import DecisionModel

model = DecisionModel()  # TypeSafe jev-latest, reads TYPESAFE_API_KEY

result = model.run(
    state={"message": "I was charged twice for my subscription this month."},
    questions={
        "team": {
            "type": "choice",
            "instructions": "Which team should handle this?",
            "criteria": {"billing": "Payments and refunds", "technical": "Bugs and outages"},
        },
        "urgent": {"type": "noul", "instructions": "Is this urgent?"},
    },
)
print(result["answers"]["team"]["choice"], result["usage"])
print(model.calculate_cost(result["usage"]))

model.noul(state="The site is down for everyone.", instructions="Is this urgent?")
print(model.usage)
print(model.calculate_cost())